Repository navigation
perf: default the shuffle compression codec to zstd - #2805
Merged
Merged
Conversation
added 2 commits
November 7, 2026 14:35
The existing multi-input test never spilled, so nothing checked that each output partition's range stitches every input's spill and in-memory bytes together in order. Run it again with a tiny buffer budget so every input spills several times and keeps an in-memory remainder.
The spill manager kept one IPC writer open per spilled output partition until the task finished, each holding a file descriptor and its own compression context outside the memory pool. With hundreds of output partitions that is hundreds of open files per input partition and, with zstd, hundreds of megabytes of compressor state. Write each spill event as one complete stream appended to the partition's file and close it; readers already cross concatenated streams.
github-actions
Bot
added
the
documentation
NoahKusaba
suggested changes
Nov 8, 2026
Comment on lines
+563
to
+566
| "Compression codec used for shuffle data written to disk and \ | ||
| sent between executors. Possible values: none, lz4, zstd. \ | ||
| zstd writes shuffle files about half the size of lz4 for little \ | ||
| extra CPU, which matters most when shuffles cross the network".to_string(), |
Contributor
There was a problem hiding this comment.
Comet's shuffle defaults to lz4. Was the ~14.8 bytes/row figure Comet with lz4? If so, this compares codecs rather than systems; Comet with zstd would be like-for-like.
Member
Author
There was a problem hiding this comment.
Yes, that was Comet with its default lz4, so it was comparing codecs and not systems. I reworded the description to say so.
| extra CPU, which matters most when shuffles cross the network".to_string(), | ||
| DataType::Utf8, | ||
| Some("lz4".to_string()), | ||
| Some("zstd".to_string()), |
Contributor
Member
Author
There was a problem hiding this comment.
Good call. I compared them with shuffle_bench on 60M lineitem rows into 200 partitions: level 1 gave 2.55 GB, level 3 gave 2.50 GB, and level 1 wrote 7-10% faster. So level 1 keeps almost all of the size win. 4980ebb sets level 1 explicitly for all shuffle writes, including the pipelined Flight path, and I added the numbers to the description.
added 3 commits
November 8, 2026 02:28
Address review feedback: spill and the in-memory encode now call one write_ipc_stream helper, SpillManager builds its IpcWriteOptions once, derives spill paths from partition_counters instead of a duplicate map, and the single-batch spill() wrapper is removed (tests call spill_batches).
zstd writes shuffle data at roughly half the size of lz4 (about 8 bytes per row against 16 for TPC-H lineitem's shuffled columns) for little extra CPU. Shuffles that cross the network get correspondingly faster, and smaller files also cut local disk and page-cache traffic. lz4 stays available through ballista.shuffle.compression.codec.
Arrow defaults zstd to level 3. Level 1 keeps nearly all of the size reduction over lz4 for noticeably less CPU: writing 60M TPC-H lineitem rows into 200 sort shuffle partitions produced 2.55 GB at level 1 and 2.50 GB at level 3 (lz4: 4.29 GB), with level 1 writing 7-10% faster. The pipelined Flight serve path now builds its options through create_write_options so it uses the same level.
andygrove
force-pushed
the
perf/zstd-shuffle-default
branch
from
November 8, 2026 09:51
db6b52b to
4980ebb
Compare
NoahKusaba
approved these changes
Nov 8, 2026
NoahKusaba
left a comment
NoahKusaba
left a comment
Contributor
There was a problem hiding this comment.
Approving!
I do wonder why comet is still using lz4 as default though.
Member
Author
Yes, I'm going to look into that tomorrow. |
andygrove
enabled auto-merge
November 8, 2026 17:02
…ault # Conflicts: # ballista/core/src/execution_plans/sort_shuffle/spill.rs
This was referenced Nov 8, 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?
No issue. Split out of #2792. Follow-up for string-heavy shuffles: #2804.
Rationale for this change
Shuffles are often bound by how many bytes they move: across the network between executors, or through local disk and the page cache when they spill. For the columns TPC-H q5 shuffles from lineitem (
l_orderkey,l_suppkey,l_extendedprice,l_discount), Arrow IPC with LZ4 stores about 16 bytes per row, and with zstd about 8.2. For reference, Spark + Comet's shuffle with its default codec (also LZ4) stores the same rows at about 14.8, so this is a codec comparison rather than a system one. zstd costs a little more CPU to compress, but that is small next to the bytes it saves.Measured with the other changes in #2792 applied, LZ4 against zstd, one iteration each:
At SF1000 the biggest changes are q5 274 to 170 s, q3 113 to 76 s, q9 156 to 117 s, q12 50 to 35 s, q18 183 to 145 s and q21 199 to 161 s. No query got slower beyond noise: q11 went from 5.4 to 5.7 s. zstd is faster even on one machine, where the shuffle never crosses a network, because smaller files mean less disk and page-cache traffic.
#2802 is a prerequisite. Without it, the sort shuffle spill manager keeps one writer open per spilled output partition, and with zstd each open writer holds its own compression context outside the memory pool.
What changes are included in this PR?
ballista.shuffle.compression.codecdefaults tozstdinstead oflz4, with a description saying why, andconfigs.mdis regenerated.zstd is written at level 1 (
SHUFFLE_ZSTD_COMPRESSION_LEVEL) instead of arrow's default of 3, matching Comet's shuffle default. Writing 60M TPC-H lineitem rows into 200 sort shuffle partitions withshuffle_bench:Level 3 saves under 2% more bytes for 7-10% more write time. The pipelined Flight serve path now builds its write options through the same helper, so it uses the same level. The TPC-H timings above were measured at level 3. Readers take the codec from the IPC stream metadata, so mixed old and new files read fine, and
lz4andnoneremain available.The Flight serve path (
ballista.shuffle.remote_read_prefer_flight, off by default) still re-encodes with LZ4 and is unchanged here.Are there any user-facing changes?
Yes. Shuffle files and block-transfer remote reads use zstd unless
ballista.shuffle.compression.codecis set. Set it tolz4to keep the previous behavior.