perf: reuse zstd context and pre-encode IPC schema in shuffle blocks [experimental]#5005
Draft
andygrove wants to merge 1 commit into
Draft
perf: reuse zstd context and pre-encode IPC schema in shuffle blocks [experimental]#5005andygrove wants to merge 1 commit into
andygrove wants to merge 1 commit into
Conversation
ShuffleBlockWriter::write_batch allocated a fresh zstd context (ZSTD_CCtx plus workspace) and re-serialized the schema flatbuffer for every block via StreamWriter. read_ipc_compressed likewise allocated a fresh ZSTD_DCtx per block. - Reuse the zstd compression/decompression context across blocks on each thread via a thread-local, streaming (not buffering) the data so large blocks are not copied an extra time. - Pre-encode the IPC schema message once in try_new and write it verbatim per block. Schemas containing dictionary types keep using StreamWriter, whose dictionary-id bookkeeping ties schema and batch encoding together. Output is unchanged: blocks remain standalone Arrow IPC streams in the same length-prefixed framing. A large-block zstd write microbenchmark is ~2% faster (context setup is a small fraction there); the win grows as blocks get smaller (high partition counts), which is where per-block context allocation dominates. Part of apache#5002.
Contributor
|
Does this incorporate the evaluation done in #4655? |
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?
Addresses two of the "high impact" items in #5002: the per-block zstd compression-context allocation and the per-block IPC schema re-encoding. That issue is an umbrella tracking several shuffle-writer optimizations, so this PR does not close it.
Rationale for this change
ShuffleBlockWriter::write_batchallocated a fresh zstd context (aZSTD_CCtxplus its workspace) and re-serialized the schema flatbuffer (viaStreamWriter::try_new) for every block.read_ipc_compressedlikewise allocated a freshZSTD_DCtxper block. With small blocks (high partition counts), per-block context setup is a meaningful fraction of compression time.What changes are included in this PR?
try_newand write it verbatim per block, instead of re-serializing it per block. Schemas that contain dictionary types keep usingStreamWriter, whose dictionary-id bookkeeping ties schema and batch encoding together.Output framing is unchanged: blocks remain standalone Arrow IPC streams in the same length-prefixed format, so the read side and the wire format are unaffected.
Preliminary benchmark (local, zstd, 5M-row input;
shuffle_bench)Higher partition count means smaller blocks, where per-block context setup dominates.
No regression at large block sizes; the win grows as blocks shrink. Larger-scale benchmarks are still to be run, which is why this is a draft.
How are these changes tested?
Existing
datafusion-comet-shuffleanddatafusion-cometshuffle tests pass. New tests were added for a dictionary-typed column roundtrip (exercising theStreamWriterfallback and dictionary-id matching) and for reading multiple distinct zstd blocks on one thread (exercising cross-frame reuse of the thread-local compression/decompression contexts, which a dirty context would corrupt).