andygrove opened a new issue, #7067: URL: https://github.com/apache/datafusion-comet/issues/7067
### What is the problem the feature request solves? Reading a native shuffle block builds a new decompressor for every block, then writes the decompressed bytes twice. For small blocks that fixed cost dominates the decode. `read_ipc_compressed_impl` in `native/shuffle/src/ipc.rs` constructs `zstd::Decoder::with_buffer`, `lz4_flex::frame::FrameDecoder::new` or `snap::read::FrameDecoder::new` for each block. A zstd decoder allocates a decompression context and window buffers (about 1 MB of malloc, memset and free per block at level 1). The lz4_flex decoder allocates source and destination buffers of up to twice the block size, and snappy zero-fills about 140 KB. Then `Streamed::body` calls `MutableBuffer::from_len_zeroed(len)` and `read_exact`s into it, so the decompressor writes into its own window and the bytes are copied again into the body buffer. The writer never pledges a content size, so the decoder cannot size the output up front. The schema parse that used to share this cost is already cached (#5809, #6098). What remains is the decoder construction and the double write. The decode-side fixed cost has not been measured separately, and `shuffle_reader.rs` covers codecs none and lz4 but not zstd or snappy. On the encode side the matching per-block setup costs 4.4 µs (LZ4), 6.2 µs (Snappy) and 13 µs (Zstd(1)) over codec None for a 128-row mixed block. Blocks per reduce task is roughly mappers x blocks per mapper, so small blocks pay this thousands of times. ### Describe the potential solution Step one, no format change: keep one decoder context per thread (like the schema cache) or per `ShuffleScanExec` and per remote decoder handle, and reset it per block instead of constructing it. Add zstd and snappy cases to `native/shuffle/benches/shuffle_reader.rs` first (it has none and lz4 today) so the win is visible on small blocks. Step two, needs a format change: record the uncompressed IPC length in the block header, decompress once into an exact-size `MutableBuffer` (skipping the zero fill), and decode with `RecordBatchDecoder` over that buffer. Celeborn frames would need versioning. ### Additional context Finding R1 in #5905. R2 (schema cache) is done. Related: #5791 (read local shuffle blocks directly in native code). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
