Optimized spill file format

@andygrove

Is your feature request related to a problem or challenge?

DataFusion spills data to local disk for processing datasets that do not fit in available memory, as illustrated in this comment:

datafusion/datafusion/physical-plan/src/sorts/sort.rs

Lines 88 to 205 in 918ac1b

    
           /// Sorts an arbitrary sized, unsorted, stream of [`RecordBatch`]es to 
        
           /// a total order. Depending on the input size and memory manager 
        
           /// configuration, writes intermediate results to disk ("spills") 
        
           /// using Arrow IPC format. 
        
           /// 
        
           /// # Algorithm 
        
           /// 
        
           /// 1. get a non-empty new batch from input 
        
           /// 
        
           /// 2. check with the memory manager there is sufficient space to 
        
           ///    buffer the batch in memory 2.1 if memory sufficient, buffer 
        
           ///    batch in memory, go to 1. 
        
           /// 
        
           /// 2.2 if no more memory is available, sort all buffered batches and 
        
           ///     spill to file.  buffer the next batch in memory, go to 1. 
        
           /// 
        
           /// 3. when input is exhausted, merge all in memory batches and spills 
        
           ///    to get a total order. 
        
           /// 
        
           /// # When data fits in available memory 
        
           /// 
        
           /// If there is sufficient memory, data is sorted in memory to produce the output 
        
           /// 
        
           /// ```text 
        
           ///    ┌─────┐ 
        
           ///    │  2  │ 
        
           ///    │  3  │ 
        
           ///    │  1  │─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ 
        
           ///    │  4  │ 
        
           ///    │  2  │                  │ 
        
           ///    └─────┘                  ▼ 
        
           ///    ┌─────┐ 
        
           ///    │  1  │              In memory 
        
           ///    │  4  │─ ─ ─ ─ ─ ─▶ sort/merge  ─ ─ ─ ─ ─▶  total sorted output 
        
           ///    │  1  │ 
        
           ///    └─────┘                  ▲ 
        
           ///      ...                    │ 
        
           /// 
        
           ///    ┌─────┐                  │ 
        
           ///    │  4  │ 
        
           ///    │  3  │─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ 
        
           ///    └─────┘ 
        
           /// 
        
           /// in_mem_batches 
        
           /// 
        
           /// ``` 
        
           /// 
        
           /// # When data does not fit in available memory 
        
           /// 
        
           ///  When memory is exhausted, data is first sorted and written to one 
        
           ///  or more spill files on disk: 
        
           /// 
        
           /// ```text 
        
           ///    ┌─────┐                               .─────────────────. 
        
           ///    │  2  │                              (                   ) 
        
           ///    │  3  │                              │`─────────────────'│ 
        
           ///    │  1  │─ ─ ─ ─ ─ ─ ─                 │  ┌────┐           │ 
        
           ///    │  4  │             │                │  │ 1  │░          │ 
        
           ///    │  2  │                              │  │... │░          │ 
        
           ///    └─────┘             ▼                │  │ 4  │░  ┌ ─ ─   │ 
        
           ///    ┌─────┐                              │  └────┘░    1  │░ │ 
        
           ///    │  1  │         In memory            │   ░░░░░░  │    ░░ │ 
        
           ///    │  4  │─ ─ ▶   sort/merge    ─ ─ ─ ─ ┼ ─ ─ ─ ─ ─▶ ... │░ │ 
        
           ///    │  1  │     and write to file        │           │    ░░ │ 
        
           ///    └─────┘                              │             4  │░ │ 
        
           ///      ...               ▲                │           └░─░─░░ │ 
        
           ///                        │                │            ░░░░░░ │ 
        
           ///    ┌─────┐                              │.─────────────────.│ 
        
           ///    │  4  │             │                (                   ) 
        
           ///    │  3  │─ ─ ─ ─ ─ ─ ─                  `─────────────────' 
        
           ///    └─────┘ 
        
           /// 
        
           /// in_mem_batches                                  spills 
        
           ///                                         (file on disk in Arrow 
        
           ///                                               IPC format) 
        
           /// ``` 
        
           /// 
        
           /// Once the input is completely read, the spill files are read and 
        
           /// merged with any in memory batches to produce a single total sorted 
        
           /// output: 
        
           /// 
        
           /// ```text 
        
           ///   .─────────────────. 
        
           ///  (                   ) 
        
           ///  │`─────────────────'│ 
        
           ///  │  ┌────┐           │ 
        
           ///  │  │ 1  │░          │ 
        
           ///  │  │... │─ ─ ─ ─ ─ ─│─ ─ ─ ─ ─ ─ 
        
           ///  │  │ 4  │░ ┌────┐   │           │ 
        
           ///  │  └────┘░ │ 1  │░  │           ▼ 
        
           ///  │   ░░░░░░ │    │░  │ 
        
           ///  │          │... │─ ─│─ ─ ─ ▶ merge  ─ ─ ─▶  total sorted output 
        
           ///  │          │    │░  │ 
        
           ///  │          │ 4  │░  │           ▲ 
        
           ///  │          └────┘░  │           │ 
        
           ///  │           ░░░░░░  │ 
        
           ///  │.─────────────────.│           │ 
        
           ///  (                   ) 
        
           ///   `─────────────────'            │ 
        
           ///         spills 
        
           ///                                  │ 
        
           /// 
        
           ///                                  │ 
        
           /// 
        
           ///     ┌─────┐                      │ 
        
           ///     │  1  │ 
        
           ///     │  4  │─ ─ ─ ─               │ 
        
           ///     └─────┘       │ 
        
           ///       ...                   In memory 
        
           ///                   └ ─ ─ ─▶  sort/merge 
        
           ///     ┌─────┐ 
        
           ///     │  4  │                      ▲ 
        
           ///     │  3  │─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ 
        
           ///     └─────┘ 
        
           /// 
        
           ///  in_mem_batches 
        
           /// ``` 
        
           struct ExternalSorter {

Here is the code that handles spilling in sort and hash aggregates.

The current version of DataFusion spills data to disk using the Arrow IPC format, which is correct and was easy to get working, but comes with non trivial overhead, as @andygrove found in Comet:

Implement faster single batch encoding/decoding for use in shuffle datafusion-comet#1189

Some potential sources of overhead are due to the validation applied to arrow IPC files (which may in general have come from untrusted sources) which is unecessary if the data was valid when written by DataFusion

Describe the solution you'd like

As part of improving DataFuison's performance of larger than memory datasets, I would like to consider adding a new optimized serialization format for use in spill files. This is similar to how we have added optimized (non Arrow) in memory storage formats for intermediate results in hash aggregation and others

At a high level this would look like:

Add a benchmark for spilling files (maybe write/read data to/from a spill file of various sizes)
Add an optimized Reader/Writer
Update the SortExec and GroupByHashExec to use this new Reader/Writer

Describe alternatives you've considered

Add a customized Reader/Writer

@andygrove has a PR to add a customized BatchReader to comet that seems to offer significant performance improvements for shuffle reading/writing:

feat: Implement custom RecordBatch serde for shuffle for improved performance datafusion-comet#1190

We could potentially upstream this code into DataFusion

Optimize the Arrow IPC reader

Another option would be to continue using the Arrow IPC format, but disable validation:

@totoroyyb actually has a recently PR to arrow-rs proposing this approach:

feat(arrow-ipc) Add: Support FileReader and StreamReader skip array data validation arrow-rs#6938

Additional context

No response

@totoroyyb

As a data point, @totoroyyb reports a 100x faster reading of Arrow IPC data without validation on apache/arrow-rs#6933

Although we're currently spilling column-wise record batches, I think this will change to row-wise batches in the future. It would be better to benchmark and optimize spilling the Arrow Row format in this issue as well.

The reason is that the spilling operation involves sorting, spilling sorted runs, and reading back those runs for merging. Both sorting and merging benefit from the row format. The current implementation performs several unnecessary conversions between row and column formats, which could become inefficient.
The preferred way should be:

Convert to row format and do sorting
Maintaining the row format until the final output (I believe Sort will benefit more from it, because it will do 2 phase or merging, Aggregate will only do 1 phase)
This is tracked by Use Row Format in SortExec #7053

Spilling the row format makes some sense to me, although I suspect IPC will outperform it, presuming a fast enough disk.

I feel I ought to point out though that in order for it to be sound to read a file without validation, DF needs to be sure nobody else could have written/modified it. This may be possible on Unix OSes using some shenanigans with unlinked file descriptors, but I suspect isn't generally possible.

I feel I also ought to point out the mmap use-case is slightly different, as it is effectively in memory already, the performance benefit of skipping validation may be lessened when there are other overheads, e.g. reading the data from disk.

I feel I ought to point out though that in order for it to be sound to read a file without validation, DF needs to be sure nobody else could have written/modified it.

In my opinion, DataFusion itself should not go to any significant trouble / effort to protect against the threat model of someone having enough control over the local file system to make arbitrary changes to spill files. If an adversary already has access to the local file system, the amount of extra safety gained by validating spill files is pretty small in my opinion.

@2010YOUY01

@2010YOUY01 the idea of avoiding Row->Column->Row conversions is a (very) good one

DataFusion itself should not go to any significant trouble / effort to protect against the threat model of someone having enough control over the local file system to make arbitrary changes to spill files

Agreed, the more reasonable scenario would be a bug in DF causes it to trample its files, maybe some sort of concurrency bug where it opens the spill file before it finishes writing it. TBC I am not making a judgement of whether or not DF should care about this, someone would need to collect concrete benchmarks of what portion of the cost of spilling actually is validation as a starting point, I am merely stating that what is being proposed is unsound.

It would be nice to have the option (likely not enabled by default) for the spill files to be compressed. It's almost trivial I think with the current implementation.

It would be nice to have the option (likely not enabled by default) for the spill files to be compressed. It's almost trivial I think with the current implementation.

We support this in Comet with the Arrow IPC Writer and it is trivial. We support LZ4, Snappy, and ZSTD.

@totoroyyb

FYI I am working with @@totoroyyb on the arrow IPC work, in case anyone is interested or has time to help:

feat(arrow-ipc) Add: Support FileReader and StreamReader skip array data validation arrow-rs#6938 (comment)

BTW Add with_skip_validation flag to IPC StreamReader, FileReader and FileDecoder arrow-rs#7120 is complete so will be able to disable validation for spill files with the next arrow release

https://www.cs.cit.tum.de/fileadmin/w00cfj/dis/papers/umami.pdf
Here are some advanced German techniques: this paper discussed

Implementation of a hash-based spilling operator
Its spill file format.

I think the spill format-related optimization is applicable. Here is a TLDR of the idea.

Tiered encoding/compression

The optimized spill format supports several encoding/compression schemes with different CPU/IO tradeoffs. On one extreme of the spectrum, we get very little CPU overhead but worse compression ratio; while on the other extreme, we get compression implementations with better compression ratio but higher CPU overhead. For example:

tier0 - Plain encoding (fast to encode, larger file size)  
tier1 - Plain encoding + LZ4 (medium encoding speed, medium compression ratio)  
tier2 - REE/Dictionary encoding + ZSTD (slow encoding speed, high compression ratio)

Spilling operator integration

The configuration for the spilling format supports auto mode in addition to all available tiers. In this case, the operator will use the default scheme for the first couple of batches and collect related metrics (compute time, IO time):

When compute time >> IO time (e.g., running on a machine with fast SSDs)
- CPU-bound: choose light encoding scheme (tier0)
When compute time << IO time (e.g., running on a machine with spinning disks or with noisy neighbors that consume lots of IO bandwidth)
- IO-bound: choose heavier encoding scheme (tier2)

The paper refers to it as self-regulating compression, and we don't have to manually set this option in average cases.

The tricky part to implement is array encoding like REE or bit-packing for integer arrays. Maybe we can find some reusable code in Arrow Parquet writer implementation or use something like https://github.com/spiraldb/vortex. But it's okay to start without those encodings.

The tricky part to implement is array encoding like REE or bit-packing for integer arrays. Maybe we can find some reusable code in Arrow Parquet writer implementation or use something like https://github.com/spiraldb/vortex. But it's okay to start without those encodings.

I think we might get pretty far by simply compressing the arrow stream zstd or snappy (which would be far simpler to implement as the arrow writer already supports this)

update : skip validation is applied when read spill files. by #15454
and SortExec, GroupedHashAggregateStream applied.

benchmark (only read phase)

spill_io/StreamReader/read_100/
                        time:   [10.725 ms 10.805 ms 10.887 ms]
                        change: [-19.544% -17.944% -16.465%] (p = 0.00 < 0.05)
                        Performance has improved.

20% better -- not bad!

It seems like comet has removed their customed BatchReader/Writer and switched back to arrow IPC reader/writer (see PR#1703).

It seems like comet has removed their customed BatchReader/Writer and switched back to arrow IPC reader/writer (see PR#1703).

So shall we close this issue as complete now?

@2010YOUY01

So shall we close this issue as complete now?

Yes, I think so. Of course, there's still room to seek further performance optimizations, but for now:

Validation skipping is supported,
more benchmarks for spill and compression support have been added, and
Based on Comet’s case, introducing a custom reader/writer or adopting a completely new format instead of Arrow IPC doesn’t seem like a priority at this point.

So I think it makes sense to close this issue for now.

In the future, potential areas to explore might include:

Row conversion during sorting,
Tuning how many batches we accumulate before spilling (as long as it doesn’t put too much memory pressure), and
Maybe supporting compression level tuning or adaptive compression. (what @2010YOUY01 mentioned)

We can always open separate issues for those if/when needed.

Yes, I think so. Of course, there's still room to seek further performance optimizations, but for now:

Indeed -we can always make the code better :)

@ding-young

Thanks again @ding-young

Navigation Menu

Repository navigation

Optimized spill file format #14078

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Add a customized Reader/Writer

Optimize the Arrow IPC reader

Additional context

Tiered encoding/compression

Spilling operator integration

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	/// Sorts an arbitrary sized, unsorted, stream of [`RecordBatch`]es to
	/// a total order. Depending on the input size and memory manager
	/// configuration, writes intermediate results to disk ("spills")
	/// using Arrow IPC format.
	///
	/// # Algorithm
	///
	/// 1. get a non-empty new batch from input
	///
	/// 2. check with the memory manager there is sufficient space to
	/// buffer the batch in memory 2.1 if memory sufficient, buffer
	/// batch in memory, go to 1.
	///
	/// 2.2 if no more memory is available, sort all buffered batches and
	/// spill to file. buffer the next batch in memory, go to 1.
	///
	/// 3. when input is exhausted, merge all in memory batches and spills
	/// to get a total order.
	///
	/// # When data fits in available memory
	///
	/// If there is sufficient memory, data is sorted in memory to produce the output
	///
	/// ```text
	/// ┌─────┐
	/// │ 2 │
	/// │ 3 │
	/// │ 1 │─ ─ ─ ─ ─ ─ ─ ─ ─ ┐
	/// │ 4 │
	/// │ 2 │ │
	/// └─────┘ ▼
	/// ┌─────┐
	/// │ 1 │ In memory
	/// │ 4 │─ ─ ─ ─ ─ ─▶ sort/merge ─ ─ ─ ─ ─▶ total sorted output
	/// │ 1 │
	/// └─────┘ ▲
	/// ... │
	///
	/// ┌─────┐ │
	/// │ 4 │
	/// │ 3 │─ ─ ─ ─ ─ ─ ─ ─ ─ ┘
	/// └─────┘
	///
	/// in_mem_batches
	///
	/// ```
	///
	/// # When data does not fit in available memory
	///
	/// When memory is exhausted, data is first sorted and written to one
	/// or more spill files on disk:
	///
	/// ```text
	/// ┌─────┐ .─────────────────.
	/// │ 2 │ ( )
	/// │ 3 │ │`─────────────────'│
	/// │ 1 │─ ─ ─ ─ ─ ─ ─ │ ┌────┐ │
	/// │ 4 │ │ │ │ 1 │░ │
	/// │ 2 │ │ │... │░ │
	/// └─────┘ ▼ │ │ 4 │░ ┌ ─ ─ │
	/// ┌─────┐ │ └────┘░ 1 │░ │
	/// │ 1 │ In memory │ ░░░░░░ │ ░░ │
	/// │ 4 │─ ─ ▶ sort/merge ─ ─ ─ ─ ┼ ─ ─ ─ ─ ─▶ ... │░ │
	/// │ 1 │ and write to file │ │ ░░ │
	/// └─────┘ │ 4 │░ │
	/// ... ▲ │ └░─░─░░ │
	/// │ │ ░░░░░░ │
	/// ┌─────┐ │.─────────────────.│
	/// │ 4 │ │ ( )
	/// │ 3 │─ ─ ─ ─ ─ ─ ─ `─────────────────'
	/// └─────┘
	///
	/// in_mem_batches spills
	/// (file on disk in Arrow
	/// IPC format)
	/// ```
	///
	/// Once the input is completely read, the spill files are read and
	/// merged with any in memory batches to produce a single total sorted
	/// output:
	///
	/// ```text
	/// .─────────────────.
	/// ( )
	/// │`─────────────────'│
	/// │ ┌────┐ │
	/// │ │ 1 │░ │
	/// │ │... │─ ─ ─ ─ ─ ─│─ ─ ─ ─ ─ ─
	/// │ │ 4 │░ ┌────┐ │ │
	/// │ └────┘░ │ 1 │░ │ ▼
	/// │ ░░░░░░ │ │░ │
	/// │ │... │─ ─│─ ─ ─ ▶ merge ─ ─ ─▶ total sorted output
	/// │ │ │░ │
	/// │ │ 4 │░ │ ▲
	/// │ └────┘░ │ │
	/// │ ░░░░░░ │
	/// │.─────────────────.│ │
	/// ( )
	/// `─────────────────' │
	/// spills
	/// │
	///
	/// │
	///
	/// ┌─────┐ │
	/// │ 1 │
	/// │ 4 │─ ─ ─ ─ │
	/// └─────┘ │
	/// ... In memory
	/// └ ─ ─ ─▶ sort/merge
	/// ┌─────┐
	/// │ 4 │ ▲
	/// │ 3 │─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘
	/// └─────┘
	///
	/// in_mem_batches
	/// ```
	struct ExternalSorter {

Optimized spill file format #14078

Description

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Add a customized Reader/Writer

Optimize the Arrow IPC reader

Additional context

Activity

alamb commented on Jan 10, 2025

2010YOUY01 commented on Jan 11, 2025

tustvold commented on Jan 11, 2025

alamb commented on Jan 11, 2025

alamb commented on Jan 11, 2025

tustvold commented on Jan 11, 2025

Omega359 commented on Jan 11, 2025

andygrove commented on Jan 11, 2025

alamb commented on Jan 12, 2025

alamb commented on Feb 28, 2025

2010YOUY01 commented on Apr 15, 2025

Tiered encoding/compression

Spilling operator integration

alamb commented on Apr 16, 2025

getChan commented on Apr 25, 2025

alamb commented on Apr 28, 2025

ding-young commented on Jul 9, 2025

alamb commented on Jul 9, 2025

ding-young commented on Jul 11, 2025

alamb commented on Jul 11, 2025

alamb commented on Jul 11, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions