Repository navigation
Potential revamp of broadcast compression policy #2216
Description
What is the problem the feature request solves?
Currently, broadcast batches are compressed twice by the same codec:
It doesn't cause statistically significant improvement on the standard benchmarks (typically broadcasts are just tiny parts of the whole plan in terms of data movements), but it might hurt workloads where 100MB-1GB data chunks are explicitly targeted for being broadcast via hint or configuration.
Describe the potential solution
So, the point is checking two assumptions:
- The query performance is notably regressed on queries where larger data portions are broadcast.
- The double compression brings no benefits in terms of inter-node traffic volume.
If both are true, maybe it is worth turning off/making configurable broadcast compression on the comet side.
Additional context
No response
Activity
parthchandra commentedon Sep 1, 2025
Isn't it the case that either the Spark Broadcast Exchange or the Comet Boadcast exchange will be used. In which case the compression will be either the Spark code path or the Comet code path?
akupchinskiy commentedon Sep 1, 2025
@parthchandra This is initiated for both Comet and Spark broadcast exchanges. Basically, it is a vanilla spark broadcast protocol which is used by Comet Boadcast exchange too. In a nutshell, comet specifies what should be broadcasted (compressed arrow batches) but it doesn't change how it should be broadcasted compare to Spark.
parthchandra commentedon Sep 2, 2025
I see. Thanks for clarifying!
comphead commentedon Nov 4, 2026
I measured both assumptions on
main(7bf572efb7).Setup. A throwaway Scala probe on
local-cluster[2,4,12288](two executor JVMs with 4 cores each), Spark 4.1.3,JavaSerializerand AQE off. The query joins a 4M-row fact table to a dimension with aBROADCASThint. Each row ofthe table is the median of 4 runs. "On" is Spark's default
spark.broadcast.compress=true, and "off" isfalse, whichleaves only Comet's own compression.
lz4zstdlevel 1zstdlevel 5lz4lz4"Executor read" is how long an executor takes to get the value of a fresh broadcast of the same bytes: fetching,
decompressing and deserializing. Running only the torrent layer on the driver over the same bytes, for
lz4,snappyand
zstdlevels 1, 3 and 5, gives the same picture. The second pass saves 6 to 11% with every codec when the buildside has a sequential column, and 0% (
lz4,snappy) to 6% (zstd) when it does not.Conclusions
100 MiB with
lz4and about 300 ms per 100 MiB withzstdat level 5, plus up to 150 ms on each executor withlz4.Since only the first assumption holds, I don't think a Comet change is worth it. The natural one is to write the
coalesced broadcast uncompressed and let the torrent layer compress it once. From the same probe, I estimate that saves
0.2 to 0.25 s of driver time per 300 to 460 MiB broadcast with
lz4and at most about 0.6 s withzstdlevel 5. Inexchange, each executor would keep the uncompressed Arrow IPC bytes on heap, 1.1 to 3 times as much as now depending on
the data and codec. Users who want to skip the second pass can already set
spark.broadcast.compress=false. Thecaveat is that it also stops compressing Spark's own broadcasts. #7492 documents this in the tuning guide and closes
this issue.