andygrove opened a new issue, #2804: URL: https://github.com/apache/datafusion-ballista/issues/2804
**Is your feature request related to a problem or challenge? Please describe what you are trying to do.** Shuffle data is written as Arrow IPC with per-buffer compression (`IpcWriteOptions::try_with_compression`). For variable-width columns (`Utf8`, `Binary`, lists and maps), Arrow's offsets buffer stores running sums. Running sums never repeat, so zstd and LZ4 compress them poorly, even when every value has the same length. In many shuffles the offsets cost about as much as the string bytes themselves. Ballista turns off `schema_force_view_types`, so shuffled strings are plain `Utf8` with `i32` offsets. DataFusion Comet hit the same problem and fixed it in apache/datafusion-comet#6943. Before compressing a block with zstd, its shuffle writer rewrites each offsets buffer as value lengths split into byte planes. The reader rebuilds the offsets before Arrow decodes the body. Comet reports string, binary, array and map shuffle blocks 9–40% smaller, with no slowdown in zstd. It skips this for LZ4, because lz4_flex, which arrow-ipc also uses, decodes the long runs that lengths compress into slowly (PSeitz/lz4_flex#227). Measured on TPC-H SF100 columns: a 1/16 random sample of rows, approximating one hash partition, in 8,192-row batches, compressed with zstd level 3 (arrow-ipc's default): | Columns | Offsets as-is | Offsets as lengths (byte planes) | Smaller by | |---|---:|---:|---:| | customer (`c_custkey`, `c_name`, `c_address`, `c_phone`, `c_acctbal`, `c_comment`) | 29.2 MB | 23.5 MB | 20% | | orders (`o_orderkey`, `o_custkey`, `o_orderstatus`, `o_orderpriority`, `o_clerk`, `o_comment`) | 64.5 MB | 41.0 MB | 36% | | part (`p_partkey`, `p_brand`, `p_type`, `p_container`) | 12.3 MB | 5.5 MB | 55% | In every case the offsets themselves shrink to almost nothing. Most of TPC-DS's shuffles carry string columns, and so do several TPC-H joins (q2, q10, q16, q18, q20, q21). **Describe the solution you'd like** The same transform for Ballista's zstd shuffle blocks. It cannot be done through arrow-ipc's built-in compression, which compresses each buffer as it is and has no way to mark a buffer as transformed. One way to do it: - The writer encodes each batch as uncompressed IPC, rewrites the offsets buffers of the message body as lengths in byte planes, and compresses the body itself. A flag in a small block header, or a new value of the shuffle codec, records that this was done. - Readers reverse it before handing the body to Arrow's decoder. That covers the local sort-shuffle reader (`MultiStreamPartitionStream`), the block-transfer remote reader, and the Flight serve path, which today decodes and re-encodes with LZ4. - Mixed-version clusters need care: a reader must recognise blocks it cannot decode instead of misreading them, or the feature stays off until every executor supports it. Comet's writer checks once per writer, on an empty batch, that arrow-ipc lays out the buffers the way the rewrite expects, and falls back to plain offsets if not. That guards against Arrow layout changes. **Describe alternatives you've considered** - Upstream: the Arrow IPC format has no notion of transformed buffers, so it would need a spec change. That is unlikely to happen soon. - Splitting fixed-width values (`Int64`, `Decimal128`) into byte planes before zstd, which Parquet calls `BYTE_STREAM_SPLIT`. On TPC-H lineitem's shuffled columns this saves only about 10% overall: `l_discount` 32%, `l_suppkey` 15%, `l_extendedprice` 7%, and `l_orderkey` 6% larger. If the block format changes for offsets anyway, it could be an option there. **Additional context** zstd is proposed as the default shuffle codec in a separate PR. It already halves shuffle bytes against LZ4 on numeric-heavy shuffles. This issue is the next step for string-heavy ones. -- 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]
