[EPIC] A collection of tickets for improving sorting larger than memory datasets / spilling sorts

[alamb]

There is a lot of work to support plans like SELECT * .. ORDER BY x,y,z when that data doesn't all fit in RAM.

I believe the core usecase is "compaction" where multiple small files are resorted/merged/written as larger fiiles

• feat: Add config max_temp_directory_size to limit max disk usage for spilling queries #14975
• Benchmark / program to test Spilling Sorts #15664
• limit max disk usage for spilling queries #15358
• A complete solution for stable and safe sort with spill #14692
• External sort failing with modest memory limit when writing parquet files #15028
• External sorting not working for (maybe only for string columns??) #12136
• Optimized spill file format #14078
• feat: Fix multi-lines printing issue for datafusion-cli and add the streaming printing feature back #14954
• Support --memory-limit for all benchmarking tools #14641
• Internal error in ExternalSorter when running with memory limit #15675
• Improve datafusion-cli memory usage and considering reserve memory for the result batches #14751
• Further improve datafusion-cli memory usage if we setting huge number for maxrow size. #14810
• Set DataFusion runtime configurations through SQL interface #15552
• Improve spill performance: Disable re-validation of spilled files #15320
• Improve Spill Performance: mmap the spill files #15321
• Reduce number of tokio blocking threads in SortExec spill #15323
• Support spilling in TopK queries #15538
• Improve time for SortPreservingMerge stream / uninitiated_partitions VecDeque<usize> #15573

FYI @2010YOUY01 and @Kontinuation I was trying to find some other organizing ticket for improvements in this area and I could not, so I filed this one

Related:

• more generic: [Epic] A collection of items related to processing larger than memory datasets (via spilling, externalized algorithm, etc) #14077
• Make DiskManagerBuilder to construct DiskManagers #15319

[alamb]

I also think that by collecting the related items we may be able to find some more review capacity (as I think this is an important capability for Spark / Comet. FYI @comphead / @andygrove / @kazuyukitanimura )

[alamb]

I filed some more tickets about ways to improve performance with external spill files

• Improve spill performance: Disable re-validation of spilled files #15320
• Improve Spill Performance: mmap the spill files #15321

[andygrove]

I added #15323

[xudong963]

@alamb Thank you for summarizing, I'm also interested in this topic and may have more time to join the game in May, but I will keep an eye on the progress.

[rluvaton]

I saw while debugging some performance issue in AggregateExec I see that we keep all spilled files open (RefCountedTempFile as it keep tempfile which hold File).

and also when merging we read at least 1 batch from every spill file:

datafusion/datafusion/physical-plan/src/aggregates/row_hash.rs

Lines 1059 to 1062 in 7317198

	for spill in self.spill_state.spills.drain(..) {
	let stream = self.spill_state.spill_manager.read_spill_as_stream(spill)?;
	streams.push(stream);
	}

so If I have a lot of spill files or if every batch is really huge (contains very large lists - like result for array_agg on large dataset) we have all of this in memory.

[alamb]

so If I have a lot of spill files or if every batch is really huge (contains very large lists - like result for array_agg on large dataset) we have all of this in memory.

I think this is the same underlying cause as

• Reduce number of tokio blocking threads in SortExec spill #15323