Multi-column sort fails with ResourcesExhausted instead of spilling when the sort keys are most of each row

Describe the bug

A multi-column sort under a memory limit can fail with Failed to allocate additional ... for ExternalSorterMerge[N] instead of spilling, when the sort keys make up most of each row.

ExternalSorter reserves about twice each buffered batch:

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

Lines 1172 to 1180 in 49c67c2

    
           pub(crate) fn get_reserved_bytes_for_record_batch_size( 
        
               record_batch_size: usize, 
        
               sliced_size: usize, 
        
           ) -> usize { 
        
               // Even 2x may not be enough for some cases, but it's a good enough estimation as a baseline. 
        
               // If 2x is not enough, user can set a larger value for `sort_spill_reservation_bytes` 
        
               // to compensate for the extra memory needed. 
        
               record_batch_size + sliced_size 
        
           }

To spill, or to produce its output without spilling, it sorts each buffered batch into its own run and merges the runs:

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

Lines 732 to 742 in 49c67c2

    
           // For single-column sorts, coalesce the buffered batches into fewer, 
        
           // larger runs to cut the merge fan-in (where the cheap per-key compare is 
        
           // dominated by per-stream cursor/merge overhead). Multi-column sorts are 
        
           // left as one run per batch: the row-format merge of many small runs 
        
           // beats sorting a few large runs with the lexicographic comparator. 
        
           let batches = std::mem::take(&mut self.in_mem_batches); 
        
           let runs = if coalesce_runs && self.expr.len() == 1 { 
        
               self.coalesce_in_mem_batches_into_runs(batches)? 
        
           } else { 
        
               batches 
        
           };

With more than one sort key, the merge uses RowCursorStream, which converts the keys of each run's current batch to the row format. Every run is a single batch, so the merge holds all sorted runs and all of their encoded keys at once. That is (1 + k) times the buffered bytes, where k is the size of the encoded keys over the size of the batch.

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

Lines 203 to 227 in 49c67c2

    
           fn convert_batch( 
        
               &mut self, 
        
               batch: &RecordBatch, 
        
               stream_idx: usize, 
        
           ) -> Result<RowValues> { 
        
               let cols = evaluate_expressions_to_arrays(&self.column_expressions, batch)?; 
        
               // At this point, ownership should of this Rows should be unique 
        
               let mut rows = self.rows.take_next(stream_idx, &self.converter)?; 
        
               rows.clear(); 
        
               self.converter.append(&mut rows, &cols)?; 
        
               self.reservation.try_resize(self.converter.size())?; 
        
               let rows = Arc::new(rows); 
        
               self.rows.save(stream_idx, &rows)?; 
        
               // `self.rows` now holds the reservation for this buffer unconditionally, and 
        
               // holds it for at least as long as the cursor does. `take_next` cannot reclaim 
        
               // the slot while the consumer still owns the `Arc`, so the cursor is handed an 
        
               // empty reservation rather than accounting for the same bytes a second time. 
        
               Ok(RowValues::new(rows, self.reservation.new_empty())) 
        
           }

The row format adds a null byte to each fixed-width value and keeps an 8-byte offset per row, so k > 1 when the keys are most of the row:

Sort keys	Arrow bytes/row	Encoded keys bytes/row	Merge holds
2 × Int64, no other columns	16	26	2.6x
2 × nullable Int32, no other columns	8.25	18	3.2x
85 nullable Int64 and 2 short strings, plus 3 other columns	748	811	2.08x

Beyond the 2x, only sort_spill_reservation_bytes (10 MiB by default) is set aside for the merge. On a full pool the merge fails once k - 1 times the buffered bytes exceeds that. For the third shape with 240 buffered 8192-row batches (1.4 GiB), a counting allocator showed the sorter reserving 1.99x the buffered bytes, and the merge then holding 2.08x until it finished.

To Reproduce

datafusion-cli -m 512M -c "SET datafusion.execution.target_partitions = 1" \
  "EXPLAIN ANALYZE SELECT value % 1000003 AS a, value AS b FROM range(30000000) ORDER BY a, b"

On main (49c67c2):

Error: Failed to allocate additional 128.0 KB for ExternalSorterMerge[0] with 421.6 MB already allocated for this reservation - 11.2 KB remain available for the total memory pool: greedy(used: 512.0 MB, pool_size: 512.0 MB)

With the default number of partitions, the same query fails at both -m 512M and -m 1G, in ExternalSorterMerge[N] or in the SortPreservingMergeExec above it. Sorts that fit without spilling fail the same way in their final merge, when the buffered batches fit at 2 times their size but not at 1 + k times.

Expected behavior

The sort spills and completes.

Additional context

The shapes in Sort-tpch Q5 fails when memory-limit is intermediate while succeeds with smaller memory #16979 (sort-tpch Q5: three keys and no other columns, fails at 100M but passes at 30M) and A memory-limited sort query fails #14143 (two Int64 keys) match this. I have not rerun them.
More accurate memory accounting in external sort #14748 proposes a reservation that accounts for the row format. Improve memory accounting for cursor in external sort #17163 implemented one from the first batch's ratio and was closed as stale. ExternalSorter memory accounting: fixes missing from branch-55 and remaining gaps on main #25804 lists the related gap in the merge of spill files (finding 8).
Proposed fix: sort the runs that feed a row-format merge into smaller batches, for example batch_size / 8 rows, so the merge holds the encoded keys of one small batch per run. The merge output keeps batch_size. A prototype passes the reproducer above, and lowers the peak heap while spilling the third shape from 2.08x to 1.28x the buffered bytes. It does not help input batches smaller than the chunk, which the coalescing in perf: coalesce small input batches before sorting #26931 would address.

Willingness to contribute

I plan to contribute a fix for this bug independently

fix: split multi-column sort runs into smaller batches under a memory…

Navigation Menu

Repository navigation

Multi-column sort fails with ResourcesExhausted instead of spilling when the sort keys are most of each row #26940

Describe the bug

To Reproduce

Expected behavior

Additional context

Willingness to contribute

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	pub(crate) fn get_reserved_bytes_for_record_batch_size(
	record_batch_size: usize,
	sliced_size: usize,
	) -> usize {
	// Even 2x may not be enough for some cases, but it's a good enough estimation as a baseline.
	// If 2x is not enough, user can set a larger value for `sort_spill_reservation_bytes`
	// to compensate for the extra memory needed.
	record_batch_size + sliced_size
	}

	// For single-column sorts, coalesce the buffered batches into fewer,
	// larger runs to cut the merge fan-in (where the cheap per-key compare is
	// dominated by per-stream cursor/merge overhead). Multi-column sorts are
	// left as one run per batch: the row-format merge of many small runs
	// beats sorting a few large runs with the lexicographic comparator.
	let batches = std::mem::take(&mut self.in_mem_batches);
	let runs = if coalesce_runs && self.expr.len() == 1 {
	self.coalesce_in_mem_batches_into_runs(batches)?
	} else {
	batches
	};

	fn convert_batch(
	&mut self,
	batch: &RecordBatch,
	stream_idx: usize,
	) -> Result<RowValues> {
	let cols = evaluate_expressions_to_arrays(&self.column_expressions, batch)?;

	// At this point, ownership should of this Rows should be unique
	let mut rows = self.rows.take_next(stream_idx, &self.converter)?;

	rows.clear();

	self.converter.append(&mut rows, &cols)?;
	self.reservation.try_resize(self.converter.size())?;

	let rows = Arc::new(rows);

	self.rows.save(stream_idx, &rows)?;

	// `self.rows` now holds the reservation for this buffer unconditionally, and
	// holds it for at least as long as the cursor does. `take_next` cannot reclaim
	// the slot while the consumer still owns the `Arc`, so the cursor is handed an
	// empty reservation rather than accounting for the same bytes a second time.
	Ok(RowValues::new(rows, self.reservation.new_empty()))
	}

Multi-column sort fails with ResourcesExhausted instead of spilling when the sort keys are most of each row #26940

Description

Describe the bug

To Reproduce

Expected behavior

Additional context

Willingness to contribute

Activity

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions