andygrove opened a new issue, #5678:
URL: https://github.com/apache/datafusion-comet/issues/5678
## Summary
apache/datafusion-comet#3882 recorded a real problem — Comet shuffle files
can be substantially
larger than Spark's — but its framing has been overtaken by what we have
measured since, and two
of the three fixes it proposed have already been investigated and closed.
Rather than keep editing
it, I am closing it and restating the current position here, along with what
still needs
measuring.
**Nothing in the "what this asks for" section below has been benchmarked.**
This is a request for
investigation, not a proposal for an agreed change.
## Current behavior
### Block format
Every shuffle block is a self-contained Arrow IPC stream.
`ShuffleBlockWriter::write_batch`
(`native/shuffle/src/writers/shuffle_block_writer.rs:232`)
emits, per block:
- a 20-byte header — 8-byte compressed length, 8-byte field count, 4-byte
codec tag
(`LZ4_` / `ZSTD` / `SNAP` / `NONE`), see `try_new_inner:136`
- a compressed payload holding a complete Arrow IPC stream: schema message,
dictionary messages,
one record batch message, and the 8-byte end-of-stream marker
So the Arrow schema flatbuffer is on the wire once per *block*, not once per
partition or once per
file. That is structural rather than an oversight: Spark shuffle is
block-based, and a reducer
fetches blocks from many mappers in arbitrary order, so a block cannot
reference a schema written
earlier in a stream. Both apache/datafusion-comet#1186 and
apache/datafusion-comet#2928 closed on
that reasoning.
Since apache/datafusion-comet#5006 the schema message is pre-encoded once in
`ShuffleBlockWriter::try_new` and copied verbatim into each block
(`SchemaEncoding::Precoded`).
That removed the CPU cost of re-serializing the flatbuffer per block; it did
not remove the bytes.
Schemas containing dictionary types still fall back to a real `StreamWriter`
per block
(`SchemaEncoding::Fallback`), because dictionary-id bookkeeping ties the
schema and record-batch
encoding together, and remote (RSS/Celeborn) writers always take that
fallback path.
### Block sizing
Block boundaries are row counts, not byte counts. Native shuffle coalesces to
`session_config().batch_size()` — i.e. `spark.comet.batchSize`, default 8192
rows — through
`BufBatchWriter`'s `BatchCoalescer`
(`native/shuffle/src/writers/buf_batch_writer.rs:85`), with a
zero-copy passthrough for batches that already exceed it. The JVM columnar
path uses
`spark.comet.shuffle.jvm.batchSize`, also 8192. A narrow schema therefore
produces small blocks and
a wide schema produces large ones, while the fixed per-block overhead stays
the same.
### Compression
Compression is applied per block and wraps the entire IPC stream, so the
schema message is inside
the compressed region. The default codec is lz4
(`spark.comet.shuffle.compression.codec`).
Spark, by contrast, compresses the shuffle stream at the block level with an
lz4 block size of
`spark.io.compression.lz4.blockSize` (32 KB by default). At 8192 rows over a
few narrow columns a
Comet block can be well below that, so each block becomes its own
compression window and the codec
has little history to work with.
## What we already know
- The original report (karuppayya, on apache/datafusion-comet#3882): 204M
rows, 8 columns
(7 string, 1 timestamp), scan → repartition → write. Comet wrote 4.77 GB
against Spark's
1.58 GB — 25.1 vs 8.3 bytes per record.
- **The repro is strongly data-dependent.** My first attempt produced Comet
files *smaller* than
Spark's. It only reproduced once I switched to short, unique strings, i.e.
when the payload per
block is small relative to the fixed overhead.
- **Batch size dominates in the one workload we measured.** At the default
batch size, Comet files
were 50% larger than Spark's and the query was 10% slower; doubling the
batch size took that to
8% larger and 15% faster than Spark.
- **The per-block schema is mostly not the cause.**
apache/datafusion-comet#2928 measured exactly
this: reusing a single encoder and stream writer across all batches gave a
~25% size reduction
for a single-column schema, dropping to **~1% at 100 columns of mixed
types**. The schema
flatbuffer grows with column count, but the data per batch grows faster,
so the relative
overhead shrinks as schemas widen. It was closed as not worth doing for
size.
- **Possible exception: deeply nested schemas.** comphead reported ~1.5x on
nested data and
observed that the schema can exceed the data there. That case is tracked
separately in
apache/datafusion-comet#5355 and was not covered by the #2928 measurement,
which used flat wide
schemas.
The residual hypothesis, then, is not "we repeat the schema too often". It
is "row-count-based
block sizing produces blocks that are too small for the compressor on narrow
schemas."
## What this issue asks for
That hypothesis has never been measured directly, and neither has the cost
of the alternative.
Specific things worth doing:
1. **Attribute the bytes.** Take a repro that shows the gap and break a
shuffle file down into
header bytes, schema-message bytes, record-batch metadata, and buffer
payload — both
compressed and uncompressed. We are currently arguing about which term
dominates by reading the
format rather than by measuring it. Note that the schema sits inside the
compressed region, so
its uncompressed size overstates its real cost, and repeating
near-identical schema bytes in
every block is exactly the pattern a compressor handles well within a
block but cannot exploit
across blocks.
2. **Measure block size against compression ratio.** Sweep
`spark.comet.batchSize` and the three
codecs across at least one narrow-schema and one wide-schema dataset, and
report compressed
bytes per row. The "doubling the batch size" result above is a single
data point from a single
workload, and it is currently carrying more weight than it should.
3. **Decide whether byte-based block sizing is worth building.** This was
the main proposal in
apache/datafusion-comet#3882 and it remains unimplemented. The hard part
is estimating a
batch's serialized size before encoding it; worth evaluating whether
`RecordBatch::get_array_memory_size` is a good enough proxy, or whether
an encode-and-split
approach is needed. It also interacts with the reader:
`spark.comet.batchSize` is what
downstream native operators expect to receive, so decoupling block size
from batch size may
require the reader to re-slice.
4. **Check whether the deeply-nested case is genuinely different**, or
whether it is
apache/datafusion-comet#5355 in another guise. If schema bytes really can
exceed data bytes
there, that is the one place where revisiting the format might pay off.
5. **Quantify what tuning already buys.** If the practical answer for most
users turns out to be
"raise `spark.comet.batchSize`", that belongs in the tuning guide with
numbers attached, and
this becomes a documentation change rather than a format change.
## Tooling
- `native/shuffle/src/bin/shuffle_bench.rs` — standalone shuffle write
benchmark, added in
apache/datafusion-comet#3752
- `native/shuffle/benches/shuffle_writer.rs` and
`native/shuffle/benches/row_columnar.rs`
-
`spark/src/test/scala/org/apache/spark/sql/benchmark/CometShuffleBenchmark.scala`
- `benchmarks/pyspark/benchmarks/shuffle.py`
- apache/datafusion-comet#3909 was a shuffle size comparison benchmark I
opened for this and closed
unmerged. Worth resurrecting as the starting point for item 1.
## Related
Replaces apache/datafusion-comet#3882.
- apache/datafusion-comet#5002 — native shuffle writer optimizations
(umbrella)
- apache/datafusion-comet#5198 — shuffle performance audit
- apache/datafusion-comet#5355 — scan + shuffle-write slower than Spark on
deeply nested schemas
- apache/datafusion-comet#5006 — pre-encoded schema per writer (CPU only,
landed)
- apache/datafusion-comet#2928, apache/datafusion-comet#1186 — earlier
attempts at removing the
per-block schema, both closed
Carried over from apache/datafusion-comet#3882: kazuyukitanimura noted that
`WriteDistributionAndOrderingSuite` is disabled on Spark 3.5+ by
apache/datafusion-comet#834 and
should be re-enabled once the underlying issue is addressed.
--
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]
