Dandandan opened a new pull request, #2339: URL: https://github.com/apache/datafusion-ballista/pull/2339
## Which issue does this PR close? None filed. Found while comparing Ballista's TPC-H plans against Spark with AQE: Spark's `BroadcastExchange` materializes a build side once and every task on an executor shares it, where Ballista re-reads it per task. ## Rationale for this change `ShuffleReaderExec` with `broadcast: true` serves partition 0 — every upstream location — on each `execute()` call, and `execute()` runs once per task. An executor running N tasks of the consuming stage therefore fetches and decodes the same build side N times. Shuffle output is immutable once written, so those reads are identical and can be shared. How much that is worth depends on task density, which is why the numbers below are per-executor rather than per-partition: - **`max_partitions_per_task=1` (the default)**: one task per partition, so a stage with 8 tasks on an executor performs 8 identical fetches. This is the case the change is aimed at. - **packing on** (`=0`, a task per free vcore): DataFusion's `CollectLeft` join already shares the build across the partitions inside one task via `OnceAsync`, so the redundancy is only the number of tasks per executor — 2 in my local setup, and close to 1 at the SF1000 benchmark shape (256 partitions over 32 executors x 8 vcores). So this is a smaller lever with packing enabled than the per-partition framing suggests. It still removes the repeat where a stage runs several waves on one executor, and where two consuming stages broadcast the same side. ## What changes are included in this PR? A `broadcast_cache` module holding batches per `(job_id, stage_id, locations fingerprint)` behind a `tokio::sync::OnceCell`, so concurrent tasks share one fetch rather than racing to fill the entry. Two things are worth reviewing: - **Staleness.** The fingerprint hashes the executor ids, map partition ids and shuffle file ids behind the read, so a re-run or replanned stage — different files under the same stage id — keys to a different slot and cannot serve the previous attempt's batches. - **Memory.** Only reads whose upstream byte size is known and within `ballista.optimizer.broadcast_join_threshold_bytes` are cached; that is the budget that made the side broadcastable, so it bounds what a cached entry can hold. Unknown or larger reads stream exactly as before. Entries are dropped alongside the shuffle files they came from, in `remove_job_data`, so both cleanup paths (push and poll) release them. ## Are these changes tested? Three unit tests: repeated lookups share a slot, a re-run's different file ids produce a different fingerprint, and eviction drops only the named stages of the named job. `ballista-core` (319 + 26) and `ballista-executor` (53) suites pass; clippy clean on both. Verified on a live cluster (SF10, 1 scheduler + 2 executors x 4 vcores, AQE, packing on) by instrumenting slot lookups and actual fetches on TPC-H q19: ``` executor1: slot lookups=2 fetches=1 executor2: slot lookups=2 fetches=1 ``` Full 22-query suite returns byte-identical row counts to the pre-change run. **Draft:** the end-to-end timing win at SF10 is inside the noise floor of a single machine — the fetch it removes is a local-disk read there, whereas on a cluster it is a network round trip. I would want a cluster run, and a look at whether the same sharing belongs at the level of the built hash table rather than the batches, before calling this ready. ## Are there any user-facing changes? No API or configuration change. Broadcast reads on the same executor now share memory for the duration of a job's stage, bounded by the broadcast threshold. -- 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]
