Repository navigation
fix: [branch-56] size aggregate spill batches by bytes as well as rows (backport #25853) - #27243
Merged
Merged
Conversation
) - Closes apache#25851. A grouped aggregate whose groups carry large state, such as `array_agg` over a low-cardinality key, fails with `ResourcesExhausted` after it spills, even though each group fits in memory and the same data spread over many small groups succeeds. Spilling writes the aggregate state in batches of at most `batch_size` rows, whatever their size in bytes. With fewer groups than `batch_size`, every group lands in one batch, and everything after the spill has to hold that batch at once: - With a single spill file, replay reads the batch directly and its table must hold every group at once. In the reproducer from the issue, that is 225 MB of state in a 128 MB pool, although each group is about 28 MB. - With several spill files, the merge reserves twice the largest batch of each file. When that does not fit, it tries to split the file, which first reserves twice the batch it is about to split. Even with small spilled batches, the merge combines rows until it has `batch_size` of them, rebuilding a batch of large groups. Aggregate spilling and replay now bound their batches by bytes as well as rows, with a target of 1 MiB (`SPILL_BATCH_TARGET_BYTES`): - `IncrementalSortIterator::with_max_batch_bytes` cuts the sorted state into chunks by the size each row takes once spilled, so a batch only exceeds the target when it holds a single larger group. Row sizes come from the new `SpilledRowSizes`, which counts view values as they are written after compaction and nested values by their elements. - `StreamingMergeBuilder::with_max_batch_bytes` makes the merge emit a batch once its rows reach the target, estimated from the average row size of the batches they come from, as well as at `batch_size` rows. It applies to every pass of the multi-level merge, so rows of small groups still merge in batches of `batch_size` rows. - Dictionaries whose values are mostly unique (fewer than two keys per value) count the values their keys reference, and each chunk holds only those values, since `take` keeps every value of a dictionary. Dictionaries whose values many keys share count only their keys and stay shared, as on `main`, so that each chunk does not write its own copy of them. - Replay compacts the views and dictionaries of the nested columns of merged batches (`gc_nested_arrays`). A merged batch references the buffers of every batch its rows come from, and `array_agg` keeps the lists it merges until its group completes. Both new options are crate-private and only used by aggregate spilling and replay; other merges are unchanged. The intermediate merge sizing from apache#25428 now receives the largest row count of the batches each run actually wrote. This applies to every stream built on `AggregateSpill` (hash and ordered, final and single-stage). The legacy `GroupedHashAggregateStream`, used only when `datafusion.execution.enable_migration_aggregate` is `false`, is unchanged. - New Case I in `aggregate_memory_spill.slt`, with a 24 MB limit, where `EXPLAIN ANALYZE` assertions check that each query spilled. All of these fail on `main` with `ResourcesExhausted`: - `array_agg` over 32 groups of about 2 MB each, with two partitions (one spill file per partition) and with one partition (several spill files, merged). - 14 large groups, adjacent in key order, spilled together with 62,500 small groups. - Runs of large groups followed by a run of small groups. - `array_agg` over dictionary-encoded values, with two partitions and with one. - `string_agg` with one partition. - `array_agg` over dictionary values that each repeat within one group, at 8 MB. - Unit tests for `SpilledRowSizes` (including views that share bytes, nested views, fixed-width rows, and dictionaries of unique, shared and no values), for byte-limited chunks in `IncrementalSortIterator` and their dictionaries, for byte-limited merges of spill files, and for dictionary compaction. With `datafusion-cli -m 128M`, the reproducer from the issue now returns 16 groups and 3,000,000 values, and so does the variant from the issue's additional context. The reproducer with 3,000 one-value groups added, `array_agg` over dictionary-encoded Parquet strings, and the reproducer with its values cast to `Dictionary(Int32, Utf8)`, which fail on `main`, also succeed. Performance compared with `main`: - `external_aggr` (TPC-H SF1, integer keys): no case is slower; 7 are unchanged and 1 is 1.06x faster. - A spilling `GROUP BY` over 5M string keys at 256 MB is unchanged (0.61 s against 0.62 s). - 8M small groups plus 200 large `array_agg` groups take 0.80 s at 128 MB, against 2.99 s on `main`. There `main` spills 26M rows (1 GB) while merging its runs, and this PR spills each of the 8M rows once. At 512 MB, where both spill twice, this PR takes 1.00 s against 0.96 s, from measuring the rows of the spilled batch that holds the large groups. - `sort_preserving_merge` shows no change beyond run-to-run noise. Merges without a byte limit only check that none is set. - `array_agg` over a dictionary of 100 distinct values (2M rows, 100K groups, 24 MB) spills the same 10.5 MB (20-byte values) and 14.7 MB (200-byte values) as `main`, in the same time. Aggregations whose groups have large state can now complete under memory limits where they previously failed after spilling. No API or configuration changes. (cherry picked from commit 32dd1d2)
github-actions
Bot
added
sqllogictest
35 of 57 tasks
andygrove
changed the title
Codecov Report❌ Patch coverage is
Flags with carried forward coverage won't be shown. Click here to find out more. @@ Coverage Diff @@
## branch-56 #27243 +/- ##
=============================================
+ Coverage 83.27% 83.28% +0.01%
=============================================
Files 1169 1170 +1
Lines 474577 475237 +660
Branches 474577 475237 +660
=============================================
+ Hits 395194 395804 +610
- Misses 55458 55481 +23
- Partials 23925 23952 +27
... and 8 files with indirect coverage changes 🚀 New features to boost your workflow:
|
xudong963
approved these changes
Nov 9, 2026
Contributor
|
🚀 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
branch-5656.0.0(Oct/Nov 2026) #24461Rationale for this change
Aggregate spilling cuts batches by row count, so a row of aggregate state that holds a whole group (an
array_agglist, astring_aggstring) can make a spilled batch far larger than its row count suggests. Replay then fails withFailed to allocate additional .... #25853 bounds the batches by bytes as well, and was merged tomainon 2026-11-08.What changes are included in this PR?
A cherry-pick of #25853 (32dd1d2), with two changes for
branch-56:aggregate_memory_spill.slt: the new case conflicted with cases that are only onmain(byte array keys in compressed spill files from perf: spill byte arrays as lengths and coalesce spill writes #26960, and native DISTINCT counts from perf: add opt-in native grouped integer DISTINCT counts #25988). The backport keeps only this PR's case, which is Case I here (it is Case J onmain), and thetarget_partitionsreset after it.spill/mod.rs:main'sclippy.tomldisallowsDictionaryArray::with_values, which is why fix: size aggregate spill batches by bytes as well as rows #25853 has#[expect(clippy::disallowed_methods)]on the call ingc_array.branch-56does not disallow it, so the unfulfilledexpectwould fail clippy. The backport drops the attribute.The Rust sources otherwise cherry-picked cleanly.
What is the testing strategy for this PR?
The same tests as #25853: the new
aggregate_memory_spill.sltcase, and the unit tests inspilled_row_sizes.rs,spill/mod.rs,sorts/stream.rsandsorts/streaming_merge.rs. Run on this branch:(The clippy command also allowed
clippy::from_iter_instead_of_collect, which fails on untouched code with the stable 1.97 toolchain used here.)Are there any user-facing changes?
No API changes. Aggregates that spill state with large rows now spill and replay it in batches bounded by bytes, so queries that failed with
Failed to allocate additional ...complete under the same memory limit.