andygrove opened a new pull request, #2807:
URL: https://github.com/apache/datafusion-ballista/pull/2807
# Which issue does this PR close?
No issue. Split out of #2792.
# Rationale for this change
The build side of a broadcast hash join is a broadcast `ShuffleReaderExec`
that reads every output partition of the upstream stage. Each task of the
joining stage runs its own copy of the plan, so every task on an executor
fetches and decodes the same data. Spark broadcasts a relation once per
executor.
Each task fetches every upstream output (M map outputs × N partitions), so
the cost grows with partition count. In TPC-H SF100 q2 at 128 partitions on two
executors, a 16-task broadcast stage made 16,384 fetch requests and decoded the
same 318K-row table 16 times. The stage took 3.0 s of task time.
# What changes are included in this PR?
- A new executor-wide `BroadcastReadCache`
(`execution_plans/broadcast_cache.rs`). The first task on an executor to run a
broadcast read fetches it. Tasks that ask while that fetch is running wait for
it, and later tasks reuse the decoded batches.
- Entries are keyed by job, upstream stage, and a hash of the exact
locations read. A re-run upstream stage therefore gets a new entry.
- A failed fetch is not cached. Every task that asks retries it and
reports its own error, so fetch-failure recovery works as before.
- `ShuffleReaderExec::with_broadcast_cache` attaches the cache. `execute`
uses it only for broadcast readers that have locations, when
`ballista.shuffle.cache_broadcast_reads` is on. The existing fetch body moves
into an inherent `fetch` method unchanged.
- `DefaultExecutionEngine` owns one cache and attaches it to every broadcast
reader in the stages it creates. `ExecutionEngine` gains `remove_job`, with a
default that does nothing, which the executor calls when it removes a job's
data.
- New config `ballista.shuffle.cache_broadcast_reads`, default `true`. Six
unit tests cover:
- concurrent waiters sharing one fetch
- reuse after the fetch completes
- failures not cached
- idle expiry
- oversized reads not kept
- `remove_job`
Results:
- q2's broadcast stage at 128 partitions: 3.0 s to 0.7 s of task time, and
16,384 to 2,048 fetches.
- TPC-H SF1000 on two nodes at 256 partitions, with the other changes in
#2792: the total went from 1,744 s to 1,724 s, about 1%. At this scale the
build sides are small next to the fact-table work.
- All 22 TPC-H queries return the same results at SF10 with 16 and with 128
partitions.
**Memory.** The cached batches are not charged to the memory pool. That
matches the fetched batches of a single read, which never were; the pool sees
each task's hash table built from them. What changes is how long they live.
Without the cache, a task's decoded build side is freed once its hash table is
built. With the cache, one decoded copy per broadcast read is kept until it has
been unused for 30 s, or until the executor removes the job's data. Idle
entries are dropped lazily, on the next cache lookup, and reads above 1 GiB are
not kept. So an executor holds at most one extra copy of each recent broadcast
read, in exchange for no longer decoding a copy per task. There is no overall
cap across entries. Setting `ballista.shuffle.cache_broadcast_reads=false`
restores the previous behavior.
# Are there any user-facing changes?
- New config key `ballista.shuffle.cache_broadcast_reads`, enabled by
default.
- Additive API: `ExecutionEngine::remove_job` (with a default
implementation), `DefaultExecutionEngine::broadcast_cache` and
`ShuffleReaderExec::with_broadcast_cache`.
--
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]
