Sort In Place, or Merge? How DataFusion Decides Between One Big Sort and a Fan-In of Sorted Runs

TL;DR: A merge of 1,000 sorted runs is expensive even if each run is tiny — the per-stream cursor overhead dominates the actual comparisons. DataFusion’s sorter therefore makes two decisions: sort everything in one concatenated batch when the input is small enough (no merge at all), and otherwise coalesce the per-batch runs into fewer, larger runs before merging. The coalescing only helps for single-column sorts, because only there is the comparison cheap enough for the overhead to dominate.

Sorting is the operation where the shape of the algorithm matters as much as the complexity. A merge of k sorted streams is O(N log k) — so you want k small. But each stream costs a cursor, a buffer, and a comparison per element, and for very short streams that fixed cost dominates.

DataFusion’s external sorter (datafusion/physical-plan/src/sorts/sort.rs) makes this trade-off explicitly, with two thresholds and one size-dependent heuristic.

Mode one: sort it all in one go

The first branch:

// If less than sort_in_place_threshold_bytes, concatenate and sort in place
if self.reservation.size() < self.sort_in_place_threshold_bytes {
    // Concatenation can grow the ordinary sort reservation, which cannot
    // borrow merge workspace. Return idle workspace to the execution pool
    // so that growth can use it.
    self.merge_pool.release_unused();
    // Concatenate memory batches together and sort
    let batch = concat_batches(&self.schema, &self.in_mem_batches)?;
    self.in_mem_batches.clear();
    self.reservation
        .try_resize(get_reserved_bytes_for_record_batch(&batch)?)
        .map_err(Self::err_with_oom_context)?;
    let reservation = self.reservation.take();
    let sorted_stream = self.sort_batch_stream(batch, reservation, None)?;
    return Ok(self.observe_if_output(sorted_stream, is_output_stream));
}

If the buffered input is under sort_in_place_threshold_bytes (1 MB by default), DataFusion concatenates everything into one RecordBatch and sorts it. There is no merge at all. One sort, one output stream, no cursors.

The default threshold is small — 1 MB — because the win is not about the size of the data, it is about the number of runs. A megabyte’s worth of 8 KB batches is 128 runs, and 128 cursors in a merge is a lot of overhead for a megabyte.

The single-batch case gets its own branch even earlier:

if self.in_mem_batches.len() == 1 {
    let batch = self.in_mem_batches.swap_remove(0);
    ...
}

which is the degenerate case of the same idea.

Mode two: runs, coalesced

Above the threshold, the input is sorted per batch and the results are merged. But there is a middle step, and its condition is the interesting part:

// 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
};

The reasoning is a comparison of two costs:

• Single-column sort. The merge compares one key per element. That comparison is cheap — a single primitive compare, or in the row format case, a memcmp on a short byte string. With such a cheap comparison, the per-stream overhead (cursor advance, buffer refill, branch) dominates. So fewer, bigger runs win.

• Multi-column sort. The merge compares a tuple of keys, using the lexicographic comparator — which is expensive. Here the comparison cost dominates the overhead, so the optimum shifts: many small runs, whose merge is cheap per element because the row format lets the comparison be a memcmp, beat a few large runs that have to be sorted with the full lexicographic comparator.

That is a genuinely subtle trade-off, and it is the kind of thing you only get right by measuring.

The coalescing itself:

/// Concatenates `batches` into fewer, larger runs, each bounded by
/// `sort_in_place_threshold_bytes`, to reduce merge fan-in. `self.reservation`
/// is resized to the coalesced footprint so the caller's per-run splits stay
/// exact.
fn coalesce_in_mem_batches_into_runs(
    &mut self,
    batches: Vec<RecordBatch>,
) -> Result<Vec<RecordBatch>> {
    let target = self.sort_in_place_threshold_bytes.max(1);
    ...
    // Flush a group into a run, skipping the copy for a single-batch group.
    let flush = |group: &mut Vec<RecordBatch>, ...| {
        match group.len() {
            0 => {}
            1 => runs.push(group.pop().unwrap()),
            _ => {
                runs.push(concat_batches(schema, group.iter())?);
                group.clear();
            }
        }
        Ok(())
    };

Notice the 1 => runs.push(group.pop().unwrap()) arm: a group with a single batch is pushed without concatenating, so the common case of “the batch is already exactly the target size” costs nothing.

The reservation dance

There is a subtlety in the “sort in place” branch that is worth calling out, because it is the kind of thing that causes subtle memory bugs:

// Concatenation can grow the ordinary sort reservation, which cannot
// borrow merge workspace. Return idle workspace to the execution pool
// so that growth can use it.
self.merge_pool.release_unused();

The sorter has two memory sources: the ordinary reservation (for the buffered input) and a merge reservation (for the merge workspace). The concatenation can need more memory than the buffered input occupied, because concatenating N batches produces one batch that may need a contiguous allocation.

The ordinary reservation cannot borrow from the merge pool, so the solution is to give the idle merge workspace back to the pool before the growth happens. That way the growth can use it.

And the comment above the whole block explains a broader policy:

// Please pay attention that any operation inside of `in_mem_sort_stream` will
// not perform any memory reservation. This is for avoiding the need of handling
// reservation failure and spilling in the middle of the sort/merge. The memory
// space for batches produced by the resulting stream will be reserved by the
// consumer of the stream.

Nothing inside the sort/merge path does a reservation. That is deliberate: if the sort could fail its reservation mid-merge, it would need to be able to spill at that point, and spilling a partially-merged stream is a much harder problem than spilling sorted runs. So the design pushes all the reservation to the edges: before the sort and after it.

The catch

• The threshold is a heuristic in both directions. 1 MB might be too small for a machine with fast memory and a wide merge, or too large for one with a slow disk. It is configurable.
• Coalescing copies. Concatenating batches into a run copies their buffers. That is real work, done to avoid more expensive work in the merge. The break-even point depends on the sort’s key width — which is exactly why the single-column / multi-column split exists.
• A large single batch is not coalesced. If one input batch is already bigger than the target, it becomes its own run, and the merge fan-in does not go down.
• The “no reservation inside the sort” rule is a constraint on everything downstream. Any operator that wants to add a reservation inside the sort path has to work around it.
• The sort’s memory is not the sort’s cost. The merge’s fan-in matters, but so does the number of runs the spill produces, and the multi-level merge machinery we looked at earlier is what handles the case where there are too many.

Why it matters

The two-mode structure is a nice illustration of a general point: the same asymptotic complexity can hide a very different constant, and the constant is often dominated by something that has nothing to do with the data — here, the number of streams in a merge.

The single-column/multi-column split is the most interesting part, because it says: the optimal sort strategy depends on how expensive your comparisons are. If comparisons are cheap, amortise them over bigger units of work. If comparisons are expensive, make the units of work smaller and the comparisons cheaper. Same algorithm, opposite advice, decided by one self.expr.len() == 1 check.