Multi-Level Merge: How DataFusion Sorts More Data Than It Can Hold, One Pass at a Time

TL;DR: Sorting more data than fits in memory is usually done by merging runs. But a k-way merge needs memory for k streams at once, and if you have 1000 runs, you cannot merge them all in one pass. DataFusion’s MultiLevelMerge merges as many as it can afford, writes the result to a spill file, adds that file back to the set, and repeats. The really clever part is the skew handling: if a single run’s widest batch is so wide that even two streams will not fit, the run is re-spilled with its batches sliced in half, and the merge is retried.

External merge sort is the standard answer to “sort more than fits in memory”. You sort chunks that fit, write them out as runs, then merge the runs. The merge is a k-way merge: one cursor per run, always taking the smallest head.

The k is the problem. A k-way merge needs one buffered batch per run. If you have 1,000 runs of 10 MB each, merging them all at once needs 10 GB of buffers — which is exactly the memory you were trying to avoid.

The algorithm

datafusion/physical-plan/src/sorts/multi_level_merge.rs (1,897 lines) describes the approach:

1. Get the maximum amount of sorted in-memory streams and spill files we can merge with the available memory
2. Sort them to a sorted stream
3. Do we have more spill files to merge?
   • Yes: write that sorted stream to a spill file, add that spill file back to the spill files to merge and repeat the process
   • No: return that sorted stream as the final output stream

So it is a tree of merges, but built lazily. The ASCII diagram in the source shows three phases:

 Initial State: Multiple sorted streams + spill files
      ┌───────────┐
      │  Phase 1  │
      └───────────┘
 ┌──Can hold in memory─┐
 │   ┌──────────────┐  │
 │   │  In-memory   │  │
 │   │sorted stream │──┼────────┐
 │   │      1       │  │        │
     └──────────────┘  │        │
 │   ┌──────────────┐  │        │
 │   │  In-memory   │           │
 │   │sorted stream │──┼────────┤
 │   │      2       │  │        │
     └──────────────┘  │        │
 │   ┌──────────────┐  │           │
 │   │  In-memory   │           │            ┌───────────┐
 │   │sorted stream │──┼────────┤            │  Phase 2  │
 │   │      3       │  │        │            └───────────┘
     └──────────────┘  │        │
 │   ┌──────────────┐  │        │       ┌──Can hold in memory─┐
 │   │ Sorted Spill │           │       │                     │
 │   │    file 1    │──┼────────┤       │   ┌──────────────┐  │
 ...

Phase 1 merges the three in-memory streams and the first two spill files into one sorted stream. That stream is written to a spill file. Phase 2 merges the remaining spill files plus the new one. And so on until one pass remains.

Note that the output is not produced until the final pass. All the intermediate passes are pure overhead in terms of I/O — and the code has a flag to avoid some of it:

/// Avoid merging more runs than needed before the final replay pass.
size_intermediate_merges: bool,

The memory strategy

The doc is unusually explicit about the memory management, and it reads like a list of hard-won lessons:

This multi-level merge make sure that we can handle any amount of data to sort as long as we have enough memory to merge at least 2 streams at a time, even when individual record batches are skewed (very wide).

1. Worst-Case Memory Reservation: Reserves memory based on the largest batch size encountered in each spill file to merge, ensuring sufficient memory is always available during merge operations.
2. Adaptive Buffer Sizing: Reduces buffer sizes when memory is constrained
3. Spill-to-Disk: Spill to disk when we cannot merge all files in memory
4. Re-spilling Skewed Runs: …

Four strategies, and the fourth is the interesting one.

Worst-case reservation. The merge must buffer one batch from each run. If a run’s batches vary in width, you must reserve for the widest one, or the reservation will be wrong at the worst possible moment. So the reservation is computed from the maximum batch size in each file.

Adaptive buffer sizing. When memory is tight, the merge reads smaller buffers from each run. Smaller buffers mean a smaller per-stream reservation, which means more streams fit. It is a knob that trades I/O (more, smaller reads) for fan-in (more runs merged per pass).

Spill to disk. The obvious fallback.

Re-spilling skewed runs. And then the case that none of the above solves.

The skew problem

Consider a run whose batches are usually 8 MB, but which contains one batch of 400 MB — a single row with a huge string, or a wide non-compressible payload.

The merge must be able to hold one batch from each run. If it is merging 2 runs, and one of them has a 400 MB batch, the reservation is 400 MB + 8 MB. If the budget is 100 MB, the merge cannot even start with two streams.

And you cannot fix it by shrinking the read buffer: the batch is already that wide on disk. Reading it in pieces would require slicing it, and the widest batch is a single row that cannot be split.

The solution in the source is remarkable:

Re-spilling Skewed Runs: If even at the smallest read-buffer size we still cannot reserve memory for the minimum of 2 streams - because a single run’s largest batch is so wide that two streams’ worth of reservation exceeds the budget - the larger of the two runs is re-spilled with each batch sliced in half. This shrinks its largest batch, lowering the per-stream reservation, and the merge pass is retried.

So the operator rewrites the offending run on disk, slicing its batches in half. A 400 MB batch becomes two 200 MB batches. The widest batch shrinks, the per-stream reservation shrinks, and the merge can proceed.

And the re-spilled run is tracked so the problem does not come back:

The re-spilled run is tracked alongside a per-run batch-size limit equal to half the batch size it was written with, so any later merge that includes it caps its output batch size to match - otherwise the merged run could rebuild a full-size batch and reintroduce the skew.

That is the subtle part. If you re-spill a run with halved batches but do not remember that, a later merge pass could re-merge those halves back into a full-size batch, and the skew would return — possibly at a point where there is no memory left to re-spill it again. So the limit travels with the run:

/// Sorted runs still to be merged. Each run is paired with the batch-size limit a
/// merge consuming it must cap its output at. Runs written at the full batch size
/// carry `batch_size`. A run re-spilled smaller to resolve skew carries its halved
/// limit (see `Self::split_spill_file_in_half`). Tracking it here keeps this limit
/// out of the public `SortedSpillFile`, so no external caller has to set it.
sorted_spill_files: Vec<(SortedSpillFile, usize)>,

And the last sentence of the doc explains the termination condition:

Crucially the global merge batch size is not lowered, so re-spilling more than one run does not compound the reduction. If a batch cannot be split any further (a single row wider than the budget), the merge surfaces ResourcesExhausted instead of looping forever.

Two guarantees:

1. The global batch size is not reduced, so the halving does not compound across multiple runs. Each re-spill is relative to the original batch size, not to the current one.
2. If a batch cannot be split further — a single row that is wider than the whole budget — the merge gives up with a clear error rather than looping forever trying to halve a 1-row batch.

That second point is the difference between a clever fallback and an infinite loop. Halving is only possible down to the row level, and the code knows where the floor is.

The round-robin tie-breaker

One more detail in the struct:

enable_round_robin_tie_breaker: bool,

When two runs have rows that compare equal under the sort ordering, which one comes first is arbitrary. A round-robin tie-breaker makes the output deterministic — a stable merge. It is off by default, presumably because it costs something and most queries do not care.

The catch

• The intermediate passes are pure I/O. Every extra phase writes the whole data set to disk and reads it back. The size_intermediate_merges flag exists to limit how many runs get merged before the final pass.
• Re-spilling is expensive. Rewriting a run to disk to halve its batches costs a full read and write of that run. It is a last resort, used only when the adaptive buffer sizing has run out of room.
• The memory model is pessimistic. Reserving for the widest batch in each file means a single outlier batch makes every merge pass conservative. That is the safe choice, but it costs.
• The whole thing only matters at scale. For a dataset that fits in memory, the plain SortPreservingMerge is used and none of this code runs.

Why it matters

This is the part of DataFusion that nobody sees until it saves them. A sort that does not fit in memory is a common failure mode in real deployments, and the difference between “the query fails with OOM” and “the query takes four times longer” is a lot of careful work.

The re-spill trick is the part I would point at. It is a genuinely creative answer to a genuinely awkward problem — one pathological batch can make an entire merge impossible — and the answer is to go back to disk and rewrite the data so the problem goes away. And the way it avoids the obvious follow-up bug (the skew coming back in a later pass) shows how carefully it was thought through.