andygrove opened a new issue, #6941: URL: https://github.com/apache/datafusion-comet/issues/6941
## Summary - Comet's shuffle files are larger than Spark's on short string columns because LZ4 cannot compress Arrow's offsets buffers at all. apache/datafusion-comet#5678 has the investigation. - Rewriting each offsets buffer as lengths, split into byte planes, before compression makes shuffle files 9% to 43% smaller and closes the gap to Spark on short strings. I have a working prototype on a local branch (not pushed yet). It passes the native and JVM shuffle tests. - With zstd the prototype also encodes and decodes faster. With LZ4, Comet's default, decoding gets 41% to 184% slower, because lz4_flex copies long repeated runs two bytes at a time. PSeitz/lz4_flex#227 fixes that, but it has been waiting on review since June. - Suggested order: help land the lz4_flex fix, then ship the format change behind a config, on by default for zstd at first. ## Background From apache/datafusion-comet#5678: with LZ4, Comet wrote 1.6x Spark's shuffle bytes for seven short ID strings plus a timestamp, and 8% to 28% less than Spark on everything else measured. Block size and the per-block schema are not the cause. The offsets are: seven string columns carry 28 bytes per row of offsets, and LZ4 keeps all 28. A running sum never repeats a 4-byte sequence, and 4 bytes is LZ4's shortest match. ## The change - **Writer.** After Arrow encodes a record batch, every 32- and 64-bit offsets buffer in the IPC body is rewritten in place as lengths, with byte `b` of length `i` moved to position `b * count + i`. Buffer sizes don't change, so the IPC message stays valid. - **Why byte planes.** Interleaved lengths make LZ4 and snappy emit one short match per value, which made encoding 25% (LZ4) and 100% (snappy) slower on variable-length strings. In planes, the high bytes become a few long runs of zeros. - **Flag.** The high bit of the first codec-tag byte marks a block that carries lengths. The current tags are ASCII, and the JVM never reads the tag. - **Reader.** `read_ipc_compressed_impl`, which every read path goes through (local, native shuffle scan, Celeborn), rebuilds the offsets with a prefix sum before Arrow decodes the body, and bounds-checks every buffer it touches. - **Scope.** Only the dictionary-free local writer path. Dictionary schemas, Celeborn writers, types whose layout isn't mapped (views, unions, run-end encoding) and schemas without offsets keep today's format byte for byte. ## Results `shuffle_bench` on `main` (474b5ece84), 7 datasets, one map task, 16 to 4,000 partitions. CPU is per block on real shuffle blocks, the minimum of 21 interleaved runs on an Apple M3 Max. | Codec | File size | Encode CPU | Decode CPU | | --- | --- | --- | --- | | zstd | −9% to −40% | −13% to −2% | −15% to +1% | | LZ4, lz4_flex 0.14 (pinned today) | −12% to −38% | −0.5% to +5.5% | +41% to +184% | | LZ4, with the lz4_flex fix | −12% to −38% | −0.5% to +5.5% | +7% to +26% | | snappy | −14% to −43% | +5% to +44% | +9% to +48% | On the short-ID dataset with LZ4 at 16 partitions, Comet goes from 73.1 to 45.2 bytes per row, against Spark's 45.9. Correctness: the shuffle crate's 186 tests and 193 tests across the JVM shuffle suites pass. A build whose reader ignores the flag fails 26 of the 73 `CometNativeShuffleSuite` tests, so the suites do exercise the new blocks. For 112 combinations of dataset, codec and partition count, the files from `main` and from the prototype decode to identical batches (496M rows). ## The lz4_flex problem Comet builds lz4_flex without `safe-decode`, so it uses the unsafe decoder, whose `duplicate_overlapping` copies two bytes per loop iteration. Lengths of repeated sizes compress into long overlapping matches: a 32 KB buffer of constant lengths decodes at 1.9 GB/s, against 17 GB/s for incompressible offsets. A copy that doubles the copied span with `memcpy` brings it to 17.5 GB/s. It also speeds up decoding of today's format (19% on the nested dataset). PSeitz/lz4_flex#227 does this and keeps short matches on the existing path. Its author reports general data within about 1% and runs 10x to 40x faster. It is waiting on the maintainer to re-run benchmarks, and it now conflicts with the base branch. ## Suggested actions 1. Help land PSeitz/lz4_flex#227: post Comet's numbers there and offer to rebase it. Then bump lz4_flex. 2. Open the format change as a PR behind a config, plumbed through the native shuffle writer's protobuf and the JVM columnar shuffle's JNI call. Turn it on by default for zstd now, and for LZ4 once the lz4_flex fix is released. 3. Before queueing that PR, run the Spark SQL and Iceberg suites with a `run-*` label, since it changes native shuffle. 4. Correct `docs/source/user-guide/latest/tuning/shuffle.md`, which says Spark's row format "generally compresses less well than Comet's Arrow IPC format". That holds for LZ4 except on short, low-entropy strings. Also mention zstd as the lever for shuffle size: Comet with zstd wrote less than Spark with LZ4 on every dataset measured. 5. Later, consider the same rewrite for Celeborn writes and dictionary schemas. ## Not covered Spills and stages with several map tasks (the partition sweep stands in for their small blocks), Celeborn, end-to-end query time, and x86. -- 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]
