Opensource internals

How DataFusion Spills a Hash Aggregation to Disk Without Breaking GROUP BY

12 min read Jay Hubbert

DataFusion sorts each aggregation state run to disk, merges the runs with a bounded-memory multi-level merge, and replays them through a clustered final aggregate.

TL;DR: DataFusion does not “spill the hash table”. It drains the hash table into intermediate aggregate states, sorts those states by the full group key, writes them as Arrow IPC spill files, then does a bounded-memory multi-level merge of the sorted runs and replays the merged stream through a clustered final aggregate. The sort is the whole trick: it makes equal groups adjacent in the merged output so the final aggregate can merge them with O(1) state per group instead of O(distinct groups).

Why “spill the hash table” is not a thing you can do

A hash aggregation’s state is a hash map from group key → accumulator state. The obvious way to spill it is: serialize the map, write it to disk, keep hashing new input into a fresh map. That works until you need the answer, because now you have K on-disk hash maps and you must merge them — and merging hash maps is exactly the aggregation you were trying to avoid, now with random I/O instead of RAM access.

The other obvious way is the classic “hash-partitioned two-phase aggregation”: partition rows by hash(group_key) % K, aggregate each partition independently, then re-aggregate the K partial results. This is what Spark and Flink do. It has a nasty property in DataFusion’s execution model: it is a blocking reshuffle in the middle of a pipelined plan, and the partial results of a partition may still be enormous (all rows with the same hash land in one partition). You have traded a memory problem for a skew problem.

DataFusion’s actual answer is neither. It keeps a single hash table, and when the table can no longer grow, it converts the table’s contents into sorted runs on disk. Sorting is the key move, for two reasons:

  1. Mergeability. Sorted runs can be merged with a k-way merge that holds one batch per run in memory. The memory cost of merging is O(k * batch_size), not O(total data). The number of runs k can be bounded by the merge fan-in, and if k is too large, the merge itself spills intermediate runs and repeats (a multi-level merge).
  2. Semantics. Once the merged output is sorted by the group key, all rows belonging to the same group are contiguous. A streaming aggregate that only needs to compare the current group key with the previous one can merge them in order, keeping exactly one group’s state live at a time.

So spilling a GROUP BY is really “re-implement the GROUP BY as a sort followed by a streaming aggregate”. That is a lot of machinery, and DataFusion has it.

The pieces

AggregateSpill: the per-partition spill context

datafusion/physical-plan/src/aggregates/spill.rs is 301 lines and holds the entire strategy. The struct doc comment at spill.rs:49-136 contains a worked example:

text 
SELECT k, SUM(v) FROM t GROUP BY k

Step 1: OOM round 1
Buffered batch        Spill file 1 (sorted)
k  partial_sum        k  partial_sum
1            3        1            3
3            4   ->   2            5
2            5        3            4

Step 3: Global sort and final aggregation
SPM output            Final aggregate output
k  partial_sum        k  SUM(v)
1            3        1       5
1            2   ->   2       5
2            5        3      10
3            4
3            6

Three fields do the real work:

rust 
// datafusion/physical-plan/src/aggregates/spill.rs:50-140
/// Spilled rows already contain evaluated group keys and intermediate
/// aggregate states. Replay must therefore use final aggregation semantics
/// and column-based group expressions rather than evaluating the raw input
/// expressions a second time, so single-stage aggregates are rewritten to
/// their final counterpart here.
replay_agg: AggregateExec,
...
/// Full group-key ordering kept by every spill file and the merged input.
spill_expr: LexOrdering,
...
/// Spill runs waiting to be merged, all sorted by `spill_expr`.
spills: Vec<SortedSpillFile>,

That first comment is the “reproduce the grouping semantics” problem in one paragraph. The rows on disk are not input rows. They are (evaluated group key columns, partial accumulator states). Replaying them through the original AggregateExec would re-evaluate the group expressions against the wrong input and double-count. So AggregateSpill::try_new rewrites the plan:

rust 
// datafusion/physical-plan/src/aggregates/spill.rs:148-168
let mut replay_agg = agg.clone();
replay_agg.group_clustering_mode = GroupClusteringMode::Full;
let group_schema = match agg.mode {
    AggregateMode::Final | AggregateMode::FinalPartitioned => {
        agg.group_by().group_schema(spill_schema)?
    }
    AggregateMode::Single | AggregateMode::SinglePartitioned => {
        replay_agg.mode = if agg.mode == AggregateMode::Single {
            AggregateMode::Final
        } else {
            AggregateMode::FinalPartitioned
        };
        *replay_agg.group_by_mut() = Arc::new(agg.group_by().as_final());
        agg.group_by().group_schema(&agg.input().schema())?
    }
    mode => {
        return internal_err!("{label}: cannot replay aggregate mode {mode:?}");
    }
};

Single becomes Final; group_by() becomes group_by().as_final(); the group expressions become plain Column references into the spilled schema. Only Final/FinalPartitioned can be replayed — a Partial aggregate has nowhere to merge to.

sort_and_spill: drain, sort, write

rust 
// datafusion/physical-plan/src/aggregates/spill.rs:218-251
pub(super) fn sort_and_spill(&mut self, state_batch: Option<RecordBatch>) -> Result<()> {
    let Some(state_batch) = state_batch else { return Ok(()); };

    let max_batch_rows = state_batch.num_rows().min(self.batch_size);
    let sorted_iter = IncrementalSortIterator::new(
        state_batch,
        self.spill_expr.clone(),
        self.batch_size,
    );
    let spill_file = self.spill_manager
        .spill_record_batch_iter_and_return_max_batch_memory(sorted_iter, self.label)?;
    ...
    self.spills.push(SortedSpillFile { file, max_record_batch_memory });
    self.min_spill_batch_rows = self.min_spill_batch_rows.min(max_batch_rows);
    Ok(())
}

Note IncrementalSortIterator rather than a full sort: it sorts the batch and then yields it back in batch_size chunks (sorts/stream.rs:353-375). This keeps the output batch size bounded so the spill file does not contain one enormous batch, which matters because the merge sizes its buffers from the largest batch it sees.

The state_batch comes from take_state_batch on the aggregate table:

rust 
// datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs:369-411
/// Takes every intermediate aggregate state and resets the table so it can
/// continue accumulating raw input.
///
/// Unlike normal single aggregation output, this materializes intermediate
/// states rather than final values. The states can therefore be merged after
/// spilling without finalizing the same group more than once.
pub(in crate::aggregates) fn take_state_batch(&mut self) -> Result<Option<RecordBatch>> {
    ...
    state.group_values.clear_shrink(0);
    state.batch_group_indices.clear();
    state.batch_group_indices.shrink_to_fit();
    Ok(Some(batch))
}

The clear_shrink(0) is load-bearing: it shrinks the hash table’s own capacity to zero so the memory can actually be returned to the pool, rather than sitting in a Vec that has grown but no longer needs to. Skipping that step is the classic “we spilled but the RSS did not go down” bug.

into_replay_stream: merge then re-aggregate

rust 
// datafusion/physical-plan/src/aggregates/spill.rs:254-300
pub(super) fn into_replay_stream(
    self,
    baseline_metrics: &BaselineMetrics,
    metrics: ClusteredAggregateTableMetrics,
    reservation: MemoryReservation,
) -> Result<SendableRecordBatchStream> {
    ...
    // The merge and replay table are two components of the same aggregate
    // operator. Keep them under one consumer registration so a fair memory
    // pool does not divide this operator's quota between its own phases.
    let merge_reservation = reservation.new_empty();
    let merged = StreamingMergeBuilder::new()
        .with_schema(spill_schema)
        .with_spill_manager(spill_manager)
        .with_sorted_spill_files(spills)
        .with_expressions(&spill_expr)
        .with_metrics(baseline_metrics.intermediate())
        .with_batch_size(batch_size)
        .with_reservation(merge_reservation)
        .with_replay_headroom()
        .with_intermediate_merge_sizing(Some(min_spill_batch_rows))
        .build()?;
    let replay = ClusteredFinalAggregateStream::new_with_input_and_metrics(
        &replay_agg, &context, partition, merged,
        &GroupClusteringMode::Full,
        baseline_metrics.clone(), metrics, None, reservation,
    )?;
    Ok(replay.into_stream())
}

Three subtleties in twenty lines:

  • reservation.new_empty() creates a sibling reservation sharing the same MemoryConsumer. The merge buffers and the replay table are two phases of one operator; if they had separate consumers, FairSpillPool would split the operator’s own quota between them and each would get half of what it needs.
  • with_replay_headroom() (sorts/streaming_merge.rs:167-171) makes the merge reserve twice what it needs — the merge buffers plus an equal amount of headroom for the aggregate that is about to consume the merged stream. Without this, the merge happily eats the entire budget and then the replay cannot allocate a single group.
  • with_intermediate_merge_sizing(Some(min_spill_batch_rows)) passes the minimum across runs of each run’s largest batch row count (sorts/streaming_merge.rs:174-185). A run without a full batch disables sizing for the whole replay, since merging short batches can increase the intermediate run’s budget.

The multi-level merge

datafusion/physical-plan/src/sorts/multi_level_merge.rs is the machinery that makes “merge arbitrarily many sorted runs in bounded memory” possible.

text 
Initial State: Multiple sorted streams + spill files
     ┌───────────┐
     │  Phase 1  │
     └───────────┘
   (merge as many as fit in memory)
                │
                ▼
     ┌───────────┐
     │  Phase 2  │  ← the intermediate merged run is itself spilled
     └───────────┘

The algorithm, from the module doc (multi_level_merge.rs:51-59):

  1. Take as many sorted runs as the memory budget allows.
  2. Merge them into one sorted stream.
  3. If there are more runs left, write that stream to a spill file, add it back to the pile, and repeat.

The fan-in is bounded by max_spill_merge_fan_in on the DiskManager (multi_level_merge.rs:578-582), so a query that spilled 500 times does not try to open 500 files at once.

The memory strategy is the subtle part:

  • Worst-case reservation: buffers are sized from the largest batch in each run, not the average (multi_level_merge.rs:127-133).
  • Adaptive shrinking: if the budget can’t fit the fan-in, the per-stream read buffer shrinks (get_sorted_spill_files_to_merge, multi_level_merge.rs:568-605).
  • Re-spilling skewed runs: if even a 2-stream merge does not fit because one run has a pathologically wide batch, the larger run is re-spilled with each batch sliced in half, which lowers its largest-batch metric and hence its per-stream reservation. The merge is retried. If a single row is wider than the entire budget, the merge surfaces ResourcesExhausted rather than looping forever (multi_level_merge.rs:139-152).

That last one is the sort of thing you only discover in production.

The spill file format

SpillManager (datafusion/physical-plan/src/spill/spill_manager.rs) is a thin, deliberately dumb layer: it writes RecordBatches to a file and reads them back. It does not interpret the contents — “all records within the same spill file are ordered according to a specific order” is the caller’s contract (spill_manager.rs:38-40).

The format is Arrow IPC stream format, which supports dictionary replacement (spill/mod.rs:469-476). Two details are worth knowing:

View-array GC. Before writing, StringView/BinaryView arrays are garbage-collected, because a sliced view array keeps the entire backing buffer alive. The comment gives the numbers:

rust 
// datafusion/physical-plan/src/spill/mod.rs:752-762
/// For example, in the ClickBench benchmark (issue #19414), repeated slicing of StringView
/// arrays resulted in 820MB of spill files that could be reduced to just 33MB after GC -
/// a 96% reduction in size.

96% is not a rounding error — without GC, a spilled query can write an order of magnitude more data than it holds.

Exact-size framing. The reader reassembles IPC messages so that each decoded batch pins only its own message, not the 128 KB read chunk it arrived in (spill/mod.rs:96-156). The alternative is subtle and nasty: a 5 KB batch that lands inside a 128 KB chunk keeps all 128 KB alive, and the batch’s accounted size (5 KB) is a lie by a factor of 25. The multi-level merge sizes its buffers from these numbers, so a lie here becomes an OOM.

The DiskManager side

DiskManager (datafusion/execution/src/disk_manager.rs) enforces a disk quota (max_temp_directory_size) and hands out temp files from a randomized choice of local directories (create_tmp_file, disk_manager.rs:342-386). Files are reference counted so the parent directory outlives them, and the disk usage is returned when the last reference drops (RefCountedTempFile::drop, disk_manager.rs:440-453).

Crucially, tmp_files_enabled() (disk_manager.rs:334-336) is the switch that turns “can spill” into a real capability. Aggregates check it before registering as spillable:

rust 
// datafusion/physical-plan/src/aggregates/hash_stream.rs:578-596
let can_spill = context.runtime_env().disk_manager.tmp_files_enabled();
let spill_context = if can_spill { ... } else { None };
let reservation = MemoryConsumer::new(...)
    .with_can_spill(can_spill)
    .register(context.memory_pool());

So a deployment with temp files disabled produces a pool with no spillable consumers, and FairSpillPool gives the entire budget to whoever asks first.

The trigger

The modern hash aggregate (aggregates/hash_stream.rs:684-760) checks the reservation after each input batch:

rust 
// datafusion/physical-plan/src/aggregates/hash_stream.rs:705-750
let resize_result = self.reservation.try_resize(
    Self::reservation_size_for_table(hash_table, spill_context.as_deref()));

match resize_result {
    Ok(()) => {}
    Err(e @ DataFusionError::ResourcesExhausted(_)) => {
        let spill_context = spill_context.as_mut().ok_or_else(|| e.context(
            "Final hash aggregate cannot spill because temporary files are not enabled in the DiskManager",
        ))?;
        ...
        let result = hash_table
            .take_state_batch()
            .and_then(|batch| spill_context.sort_and_spill(batch));

        // Spilling shrinks the aggregate table and releases its accumulated
        // memory. Update the reservation accordingly.
        self.reservation.try_resize(hash_table.memory_size())?;
        result?;
        // One sorted run was written; resume reading the original input.
    }
    Err(e) => return Err(e),
}

Three things to notice:

  1. The reservation size includes extra headroom when a spill context exists: table_size + group_count * size_of::<u32>() (hash_stream.rs:661-679) — the index materialization needed to sort the state.
  2. The spill happens after the failed try_resize, and the reservation is then shrunk to the (now much smaller) table size. The failure is not fatal; it is the trigger.
  3. The legacy GroupedHashAggregateStream does the same thing but computes the sort memory budget explicitly, including the “2× worst case” rule:
rust 
// datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs:1249-1267
// The maximum worst case for a sort is 2X the original underlying buffers(regardless of slicing)
let sort_memory = (batch_memory
    + (emit.get_sliced_size()? as f32 * batch_size_ratio) as usize)
    .min(batch_memory * 2);
self.reservation.try_grow(sort_memory).map_err(|err| {
    resources_datafusion_err!("Failed to reserve memory for sort during spill: {err}")
})?;

And it frees the table before reserving the sort memory — “This must happen before reserving sort memory so the pool has room” (grouped_hash_stream.rs:1242-1248). Order matters: if you reserved first, you would need the memory twice.

Trade-offs and failure modes

  • Spilling is expensive and it is paid in latency, not just I/O. Each spill event sorts the entire accumulated state: O(n log n) per run, n runs. A query that spills 100 times pays 100 sorts.
  • The number of runs drives the merge cost. 100 runs means a multi-level merge: merge 16, write the result, merge 16 more… Each level rewrites the whole dataset. This is why max_spill_merge_fan_in exists and why the merge tries hard to fit as many runs as possible in one pass.
  • Partial aggregates cannot be spilled. The replay needs a “final” target to merge into. A Partial-only plan that OOMs has to error out; the mode rewrite in AggregateSpill::try_new is what makes the whole scheme possible, and it is the part most likely to surprise you when reading the code.
  • GroupClusteringMode::Full aggregates never spill (spill.rs:169-172 returns an internal error). If groups are already contiguous in the input, spilling would not help — you would read the same data back in the same order and hit the same problem.
  • Skewed group keys. One hot key means one group’s state never leaves memory. The sort-based spill does not fix that; the group’s state simply stays in the hash table across the whole run.
  • Disk is not infinite. max_temp_directory_size is enforced, and a query that spills more than the quota gets an I/O error, not an OOM. The failure mode moves, it does not disappear.
  • The 2× sort-memory rule is a heuristic. It is derived from get_sliced_size and can be wrong for adversarial layouts; the .min(batch_memory * 2) cap is the safety net, and exceeding it produces "Failed to reserve memory for sort during spill".

The overall shape is worth internalizing: DataFusion’s spill implementation is not “write the hash table to disk”. It is “convert an unordered, unbounded in-memory structure into a sequence of sorted runs, then re-derive the aggregation as a streaming operation over the sorted result”. Every hard part — the mode rewrite, the group-key ordering, the merged-stream replay, the bounded-memory multi-level merge — comes from that one decision.

DatafusionMemoryPerformanceInternals