fix: leave memory for aggregate spill replay

[sunchao]

Which issue does this PR close?

Prerequisite for #25172, which fixes FairSpillPool accounting across reservations belonging to the same memory consumer. This PR prepares aggregate spill replay to work within that shared allowance and should merge first.

Rationale for this change

Spilling an aggregation to disk only helps if there is enough memory to read it back and finish the aggregation. During spill replay, DataFusion merges sorted spill files and feeds their rows into an aggregate that combines the intermediate states into final results. The merge needs buffers for its inputs and output, while the aggregate needs memory for group keys and accumulators. Those allocations coexist: producing the next batch and processing it are both part of the same operator's memory budget.

For example, consider:

SELECT customer_id, ARRAY_AGG(event_id)
FROM events
GROUP BY customer_id;

With many customers and a small memory limit, the aggregate can spill several times. A customer's events may then be spread across several spill files. Merging those files brings the customer's intermediate states together, but the aggregate still has to build the final array. Even if the merge delivers one row at a time, that array grows until the customer's group is complete. Small input batches therefore do not eliminate the need for replay memory.

The current merge tries to merge as many spill files as its reservation permits, without leaving room for this downstream work. As a simplified example, suppose the operator has a 100 MiB allowance and no other reservations. An 80 MiB merge fits by itself, but if the aggregate then needs 24 MiB to process its output, their combined 104 MiB does not fit. A pool enforcing the shared allowance must reject that growth, even though using a smaller merge could let the same aggregation make progress.

This is particularly relevant to #25172. The merge and aggregate use separate reservations under the same memory consumer. Today, FairSpillPool checks those reservations individually, which can hide their combined usage exceeding the consumer's allowance. Once #25172 enforces that allowance across both reservations, replay must already leave space for the aggregate. This PR addresses that requirement before the accounting fix lands.

What changes are included in this PR?

Aggregate spill merges now leave room for the aggregate consuming their output. When choosing how many spill files to merge and how much to buffer, the merge asks the configured memory pool to admit its buffers plus an equal amount of spare capacity for replay. Once admitted, it releases the spare reservation before replay starts, retaining only the reservation for the merge buffers. This uses the pool's existing allocation checks, so the decision follows its admission policy, including any fair-share restrictions, without adding a public memory-pool API.

In the example above, the 80 MiB merge would first need approval for 160 MiB and would be rejected. A smaller 40 MiB merge would need approval for 80 MiB; after returning the 40 MiB of temporary headroom, it would leave 60 MiB available, enough for the illustrative 24 MiB of aggregate state. The equal-sized headroom is a practical budgeting policy, not an estimate of the exact accumulator size or a guarantee that every aggregation will fit.

The existing multi-level merge adapts by merging fewer files at once, reducing read-ahead, or splitting oversized spill batches into smaller batches. This PR makes those adjustments account for replay as well. It also handles the point where a batch cannot shrink further: a single wide row may still fit the real pool even when equal headroom does not. In that case, only the minimum merge may retry without the extra headroom, with read-ahead disabled and all merge memory still subject to the pool's checks. Inspecting decoded batches before rewriting a spill file avoids unnecessary writes when the rows already fit or cannot be split, including when the original spill files fill the disk quota.

All supported aggregate replay implementations use this policy, including hash and ordered aggregation and the legacy implementation selected by enable_migration_aggregate=false. The legacy path also releases unused initial grouping capacity before replay. The headroom policy is private to aggregate replay; ordinary sort callers retain their existing admission behavior.

Are these changes tested?

Regression tests exercise replay across the supported aggregate implementations; the migrated hash and ordered paths also run with another registered spilling consumer. In particular, the legacy ARRAY_AGG test uses 64 groups with 64 values per group, one-row batches, and an 8 KiB pool to check that accumulator state can grow across replay batches. The tests verify exact aggregate results, reservation bounds, and release of memory and spill files.

Merge-level coverage checks oversized batches, short and odd-sized batches, indivisible rows, decoded string-view sizes, and spill files that already fill the disk quota. It also checks that temporary headroom is released after a candidate merge is rejected and that the fallback still fails when the actual merge cannot fit.

Validation results and environment

Validated replay alone, with the existing FairSpillPool implementation:

• 2,263 physical-plan tests and 2,281 core/CLI tests passed.
• The existing permanent-pressure regression passed with its original success expectation.
• All 520 SQL logic test files passed.
• cargo fmt --all and strict all-target/all-feature Clippy passed.
• Full ./dev/rust_lint.sh passed, including private Rust documentation and local Markdown links.

Local validation uses Rust 1.98.1 and upstream revision 22651d24 with its unchanged dependency lockfile; newer main's dependency versions are unavailable in the local registry. Merge compatibility with current main is checked separately, and GitHub CI validates its merged revision.

Are there any user-facing changes?

Aggregations that spill leave memory available for processing the merged rows, reducing avoidable replay failures under constrained memory. Achieving this can require smaller batches or additional merge passes. SQL semantics and public APIs are unchanged.

The temporary headroom reservations can increase recorded reservation peaks without allocating additional data buffers. That headroom is released before replay, so other concurrent allocations or aggregate state that outgrows the available memory can still cause ResourcesExhausted.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 91.94313% with 34 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.02%. Comparing base (22651d2) to head (1ff4126).
⚠️ Report is 20 commits behind head on main.

Files with missing lines	Patch %	Lines
...usion/physical-plan/src/sorts/multi_level_merge.rs	92.18%	9 Missing and 11 partials ⚠️
datafusion/physical-plan/src/aggregates/mod.rs	90.90%	6 Missing and 8 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25383      +/-   ##
==========================================
+ Coverage   81.92%   82.02%   +0.09%     
==========================================
  Files        1135     1136       +1     
  Lines      427772   439539   +11767     
  Branches   427772   439539   +11767     
==========================================
+ Hits       350456   360522   +10066     
- Misses      56373    58242    +1869     
+ Partials    20943    20775     -168     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[jayzhan211]

Thanks @sunchao! One suggestion below, non-blocking:

check_headroom is enforced on every selection, but only the pass that drains sorted_spill_files feeds replay; intermediate passes spill back to disk and don't need headroom. With many runs this halves fan-in on every pass, roughly doubling the number of intermediate re-spills and peak temp-disk usage. Fine to handle in a follow-up: "Skip replay headroom for intermediate multi-level merge passes".

Sketch: let the caller control headroom per call and cap the number of admitted files. Probe with headroom; if that doesn't admit every remaining run, re-select without headroom while holding one run back so the following pass is still the final one (otherwise a full-pool selection could become the final pass with no headroom).

     fn get_sorted_spill_files_to_merge(
         &mut self,
         buffer_len: usize,
         minimum_number_of_required_streams: usize,
         reservation: &mut MemoryReservation,
         allow_minimum_without_headroom: bool,
+        headroom: bool,
+        max_files: Option<usize>,
     ) -> Result<SpillFilesToMerge> {
-        let max_spill_files = effective_spill_merge_fan_in(configured_fan_in);
+        let max_spill_files = effective_spill_merge_fan_in(configured_fan_in)
+            .min(max_files.unwrap_or(usize::MAX));
         ...
-            let check_headroom = self.reserve_replay_headroom && !skip_headroom;
+            let check_headroom = headroom && !skip_headroom;
         ...
-        if self.reserve_replay_headroom {
+        if headroom {
             reservation.shrink(reservation.size() - accepted_memory);
         }

The recursive buffer_len - 1 call passes headroom and max_files through unchanged. Then in merge_sorted_runs_within_mem_limit:

let total = self.sorted_spill_files.len();
let selection = self.get_sorted_spill_files_to_merge(
    2,
    minimum_number_of_required_streams,
    &mut memory_reservation,
    allow_minimum_without_headroom,
    self.reserve_replay_headroom,
    None,
)?;
let selection = match selection {
    SpillFilesToMerge::Ready(spills, _)
        if self.reserve_replay_headroom && spills.len() < total =>
    {
        // Intermediate pass: no replay follows, so use the full pool but
        // keep one run back so the next pass remains the final one.
        self.sorted_spill_files.splice(0..0, spills);
        memory_reservation.free();
        self.get_sorted_spill_files_to_merge(
            2,
            minimum_number_of_required_streams,
            &mut memory_reservation,
            allow_minimum_without_headroom,
            false,
            Some(total - 1),
        )?
    }
    other => other,
};