Repository navigation
perf(core): split sort-shuffle rows by partition on arrival when it fits - #2774
Draft
NoahKusaba wants to merge 12 commits into
Draft
NoahKusaba wants to merge 12 commits into
NoahKusaba wants to merge 12 commits into
Conversation
The sort-shuffle writer buffers whole input batches plus a `(batch_idx, row_idx)` pair per row, and builds each output batch at spill or final-write time with `interleave_record_batch` (apache#1598). On large shuffles that gather reads every row from a different place in a buffer of hundreds of input batches, so it misses the cache on nearly every row, and pushing the index pairs costs another 8 bytes and a store per row. Inputs that move enough rows now split them as they arrive instead: each output partition's rows are `take`n into that partition's `BatchCoalescer` while the input batch is still hot in cache, and the input batch is dropped. Every coalescer holds a partly filled batch with room for `batch_size` rows, so this costs `partitions x batch_size` rows of memory however few rows arrive, which is why apache#1598 avoided it. So every input starts in the existing deferred layout, unchanged, and switches only once it has buffered `partitions x batch_size` rows, enough to fill a batch per partition on average, and only if that many rows at the measured width, plus a 1 MiB string block per partition for each view column with long values (the coalescer's string blocks grow to that size and stay there), fit the split budget: a quarter of `memory_limit_per_task_bytes`, a quarter of an even share of the pool's free memory between the task's inputs, and 64 MiB, whichever is smallest. The buffered rows are gathered into the coalescers at the switch. A split input whose spill is mostly partly filled batches, which is what memory pressure looks like, goes back to deferred for good, so it never spills a small batch per partition again. Small shuffles, and ones with many partitions, wide rows or a small pool, keep the deferred layout. Shuffle statistics (`num_bytes`, and spill bytes) now count the bytes the rows take rather than `get_array_memory_size`, which counts buffer capacity: a partly filled 8192-row batch would otherwise report the size of a full one, and AQE plans joins from these numbers. View columns count only the bytes their views reference. Both layouts drain through `BufferedBatches::drain`, one partition at a time, so encode and spill don't depend on the layout. New writer metrics: `buffer_time` (copying rows into the buffers) and `split_inputs` (inputs that switched to split). `BufferedBatches::new` takes a split budget, and `push_batch` returns a `Result`.
Adds `ballista.shuffle.sort_based.split_on_arrival_max_bytes` (default 64 MiB), the most memory an input's partly filled batches may take for the writer to split rows by partition on arrival. It replaces the fixed 64 MiB cap; the budget is still held to a quarter of the per-task budget and of the input's share of the pool's free memory. 0 never splits. The value reaches the executor through a new optional field on `SortShuffleWriterExecNode`, so older plans decode to the default.
…sted arrays - Split output keeps the input batches' schema, as the deferred gather does, instead of the stream's declared one, which can differ in nullability or metadata. - The pool must grant the switch to split, covering the buffered rows held twice while they move, or the input stays deferred for good. While split, the reservation keeps the partly filled batches' worst-case size, so inputs sharing a pool cannot each claim the same free memory. - An input whose buffered rows pass the split budget before a decision stays deferred, instead of measuring every batch. - used_bytes counts only the rows nested arrays reach: list values within the offsets, and view data in struct, list and map children. - Docs: reported partition sizes and spill_bytes count used bytes whatever split_on_arrival_max_bytes is set to.
push_batch_with_indices ran a take per output partition, so each input batch cost one take and one index array per partition, plus a second copy into the coalescer. One take now groups the batch's rows by partition, and each partition's rows go to its coalescer as a zero-copy slice.
Tests now compare the buffer's whole layout (deciding, split, deferred, or deferred after split) rather than whether it is split, which could not tell a buffer still deciding from one that stays deferred. The small-pool writer test asserts its exact spill count.
…l refuses the switch A test pool advertises more memory than it grants, as a pool looks to a task once other tasks have claimed memory since the task measured it. The input's budget allows splitting, but the pool refuses the switch, so it stays deferred and writes what a writer that never splits does.
…deciding A spill before the buffer had enough rows to decide reset its row and byte counts, so it stayed undecided and kept measuring every batch. A drain before deciding means memory ran short before splitting could pay off, so the buffer now stays deferred for good.
At the end of input the writer encodes its buffered rows into one IPC buffer per output partition. A deferred buffer keeps its input batches until the drain ends, so input and encoded output are live together, but only the input was reserved, and the reservation kept the input's size until the shuffle file was written. Reserve each partition's encoded bytes as they are produced, give back a split buffer's batches as each is written, and spill the encoded bytes and every later partition's rows when the pool refuses them. Afterwards the reservation holds only the encoded bytes. SpillManager::spill_encoded appends an encoded stream to a partition's spill file, and spill() now appends to a file the manager wrote earlier instead of truncating it.
…plit-on-arrival Resolves the conflict with apache#2796 in sort_shuffle/buffer.rs by moving its RecordBatchMemoryCounter into DeferredBuffer, with its slice test. stays_deferred_when_the_pool_refuses_the_switch now grants 12 MiB rather than 16: its inputs reach the writer as slices of shared arrays, so the deferred rows now count 4 MiB instead of 10, and 16 MiB fit both layouts.
This was referenced Nov 7, 2026
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?
Closes #2578. Second of two PRs, after #2579.
Rationale for this change
The sort-shuffle writer gathers every output batch with
interleave_record_batchat spill or write time. On large shuffles thatgather reads rows that have left the cache, so nearly every row is a miss. On
TPC-H SF10 it is the largest part of the shuffle write. #2578 has the
measurements and the design, including how Gluten handles the same trade-off.
What changes are included in this PR?
BufferedBatchesgets a second layout, split on arrival. Each partition'srows are
taken into that partition'sBatchCoalescerwhile the input batchis still in cache, and the batch is dropped.
Utf8VieworBinaryView. The coalescer copies those types as theyarrive. For any other type it keeps slices of the input and concatenates
them later, which would hold the input batches anyway, so those schemas
always stay deferred.
has buffered
partitions × batch_sizerows, it decides once. It splits ifone partly filled batch per partition fits a budget, at the measured row
width plus a 1 MiB string block per long-string column. Otherwise it stays
deferred for good. A buffer that has to drain (spill) before it decides
also stays deferred.
split_on_arrival_max_bytes(default 64 MiB);memory_limit_per_task_bytes;that are held twice while they move, or the input stays deferred. While
split, the reservation keeps the partly filled batches' worst-case size, so
inputs sharing a pool cannot each claim the same free memory.
compacted with
gc()when their buffers are larger than their strings. Thecoalescer otherwise keeps the input's string buffers, or 1 MiB copy blocks
sized for far more rows than one batch.
a sign of memory pressure, goes back to deferred for good.
num_bytesand spill bytes count the bytes rows use,not allocated capacity. Otherwise a partly filled batch reports the size of
a full one to AQE. For deferred output this is about the same number as
before.
buffer_timeandsplit_inputsare new writer metrics.reserved, and spilled when the pool refuses them. Before, a deferred buffer
held its input batches and the encoded output at once with only the input
reserved. This bug is also on
main, so this commit is also going into itsown PR. I'll drop it from here once that one merges.
The commits:
perf(core): split sort-shuffle rows by partition on arrival when it fits:the layout, the switching rules, the stats and the metrics.
feat(core): make the sort-shuffle split-on-arrival cap configurable:the setting, a new optional proto field, and the docs.
takeper input batch;drain;
fix(core): reserve the sort-shuffle writer's encoded output(see"Final write" above).
main, which ports fix: count buffers shared by sliced batches once in shuffle buffering #2796's shared-buffer counting into thedeferred layout.
Commits 3 fix this PR's own code, so I can squash them into 1 if that's
easier to review.
Are there any user-facing changes?
ballista.shuffle.sort_based.split_on_arrival_max_bytes.0 turns splitting off and restores today's behaviour.
configs.mdand the tuning guide.SortShuffleWriterExecNode. Older plansdecode to the default. This needs the
proto-changelabel.SortShuffleWriterExecshows two more metrics.BufferedBatchesis crate-private.How was this tested?
refuses;
bytes;
num_bytesexcludes spare capacity;spills included;
does;
heap with the pool reservation. Counting requested bytes, live memory
matches the reservation to within 0.5 MiB at the end of input. At the peak,
during the final write, it stays within 2.7 MiB of it, in both layouts,
with and without a bounded pool. The remaining 5–10% of real memory is
mimalloc size-class rounding, which no reservation sees.
Performance, from a release-build probe with mimalloc on a laptop. Output
digests are identical to
mainin every case.mainOn TPC-H SF10 with one executor, the prototype was 8.5% faster overall at 12
partitions, and 22% faster on q1–q8 at 200 partitions. All 22 queries pass
--verifywith the same AQE join decisions.Still to run: the criterion
sort_shufflebench (#1598's 200-partition case)against
main, and TPC-H--verifyon the current branch.