Skip to content

perf(core): split sort-shuffle rows by partition on arrival when it fits - #2774

Draft
NoahKusaba wants to merge 12 commits into
apache:mainfrom
NoahKusaba:perf/sort-shuffle-split-on-arrival-v2
Draft

NoahKusaba wants to merge 12 commits into
apache:mainfrom
NoahKusaba:perf/sort-shuffle-split-on-arrival-v2

Conversation

@NoahKusaba

@NoahKusaba NoahKusaba commented Nov 4, 2026 •

Copy link
Copy Markdown
Contributor

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_batch at spill or write time. On large shuffles that
gather 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?

BufferedBatches gets a second layout, split on arrival. Each partition's
rows are taken into that partition's BatchCoalescer while the input batch
is still in cache, and the batch is dropped.

  • Which schemas can split. Only schemas whose columns are all primitive,
    Utf8View or BinaryView. The coalescer copies those types as they
    arrive. 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.
  • When it splits. Every input starts in today's deferred layout. Once it
    has buffered partitions × batch_size rows, it decides once. It splits if
    one 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.
  • The budget is the smallest of:
    • the new split_on_arrival_max_bytes (default 64 MiB);
    • ¼ of memory_limit_per_task_bytes;
    • ¼ of the input's share of the pool's free memory.
  • Memory pool. The pool must grant the switch, covering the buffered rows
    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.
  • String buffers. Each completed split batch's view columns are
    compacted with gc() when their buffers are larger than their strings. The
    coalescer otherwise keeps the input's string buffers, or 1 MiB copy blocks
    sized for far more rows than one batch.
  • Going back. A split input whose spill is mostly partly filled batches,
    a sign of memory pressure, goes back to deferred for good.
  • Stats. Partition num_bytes and 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.
  • Metrics. buffer_time and split_inputs are new writer metrics.
  • Final write. The IPC bytes encoded at the end of input are now
    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 its
    own PR. I'll drop it from here once that one merges.

The commits:

  1. perf(core): split sort-shuffle rows by partition on arrival when it fits:
    the layout, the switching rules, the stats and the metrics.
  2. feat(core): make the sort-shuffle split-on-arrival cap configurable:
    the setting, a new optional proto field, and the docs.
  3. Fixes and refinements from review and from a memory-accounting probe:
    • pool reservation for the switch, and sizing of nested arrays;
    • one take per input batch;
    • exact test assertions;
    • staying deferred when the pool refuses the switch, or after an early
      drain;
    • splitting only primitive and view columns;
    • compacting string buffers.
  4. fix(core): reserve the sort-shuffle writer's encoded output (see
    "Final write" above).
  5. A merge of main, which ports fix: count buffers shared by sliced batches once in shuffle buffering #2796's shared-buffer counting into the
    deferred 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?

  • New setting: ballista.shuffle.sort_based.split_on_arrival_max_bytes.
    0 turns splitting off and restores today's behaviour.
  • Docs: documented in configs.md and the tuning guide.
  • Proto: adds an optional field to SortShuffleWriterExecNode. Older plans
    decode to the default. This needs the proto-change label.
  • Explain output: SortShuffleWriterExec shows two more metrics.
  • API: no public API changes; BufferedBatches is crate-private.

How was this tested?

  • Buffer unit tests:
    • when the buffer switches, and when it doesn't, including when the pool
      refuses;
    • staying deferred after an early drain;
    • going back to deferred, and staying split;
    • which column types can split;
    • inline and sparse strings, and string buffer compaction;
    • the reservation while split;
    • a check of the coalescer's 1 MiB string block.
  • Writer tests:
    • split and deferred runs write identical partitions and report identical
      bytes;
    • a cap of 0 never splits;
    • columns the coalescer slices never split;
    • num_bytes excludes spare capacity;
    • a small memory pool behaves exactly like a writer that never splits,
      spills included;
    • a pool that refuses the switch writes what a writer that never splits
      does;
    • the budget follows the pool's free memory;
    • the final write spills the partitions the pool cannot hold.
  • Memory accounting: a probe with a counting allocator compared live
    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 main in every case.

Case Layout vs main
16 partitions, strings split 22% faster
16 partitions, strings, spilling split 21% faster
200 partitions, ints split 4% faster
200 partitions, strings deferred unchanged

On 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
--verify with the same AQE join decisions.

Still to run: the criterion sort_shuffle bench (#1598's 200-partition case)
against main, and TPC-H --verify on the current branch.

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.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

documentation Improvements or additions to documentation

Projects

None yet

Development

Successfully merging this pull request may close these issues.

perf: hybrid sort-shuffle buffering: split rows by partition on arrival when memory allows

1 participant