apache / apache/datafusion-ballista
Broadcast lowering still hash-repartitions the build and probe sides of a CollectLeft join
- Dominant language
- Rust
- Stars
- 2.1k
- Forks
- 320
- Avg merge
- 1d 22h
- Merged PRs (30d)
- 66
Description
## Is your feature request related to a problem or challenge?
When the static distributed planner promotes a join to a broadcast
`HashJoinExec(CollectLeft)` (`maybe_promote_to_broadcast` + the broadcast lowering
in `plan_query_stages_internal`), the join's inputs still carry the join-key
`RepartitionExec(Hash)` that `EnforceDistribution` inserted for the original
partitioned join. As a result:
- **Build side** is hash-repartitioned into a shuffle stage and *then* written
again as a broadcast stage — two shuffles where broadcast should need one (just
replicate the build's natural partitions).
- **Probe side** still reshuffles on the join key, even though a `CollectLeft`
join replicates the build to every probe task and does not require the probe to
be partitioned on the join key.
So the broadcast promotion adds a broadcast without removing the reshuffles it
was meant to avoid. On TPC-H SF10 (AQE off) the SMJ-broadcast path (#1904) still
gives ~16% because eliminating the sort + collecting a small side helps, but a
large part of the intended benefit — skipping the join-key shuffles entirely — is
left on the table.
This affects **both** broadcast paths:
- the hash-join broadcast from #1647, and
- the sort-merge-join broadcast from #1904 (#1679).
## Describe the solution you'd like
During broadcast lowering, strip the redundant join-key `RepartitionExec(Hash)`
from the converted join's inputs:
- broadcast the build side from its natural (upstream) partitions instead of
hash-repartitioning then broadcasting;
- keep the probe side at its upstream partitioning rather than reshuffling on the
join key.
## Additional context
Correctness to verify: a `CollectLeft` join's output partitioning follows the
probe side. Removing the probe's join-key repartition changes the probe (and
therefore the join output) partitioning, so downstream operators that assume
hash-on-join-key partitioning must be re-checked / re-satisfied by
`EnforceDistribution`/`EnforceSorting`. Build-side broadcast must still replicate
all build partitions to every probe task.
Follow-up to #1904 (#1679). Related: #342, #348.
Contributor guide
Research direction
Start in maybe_promote_to_broadcast and the broadcast lowering in plan_query_stages_internal; trace the HashJoinExec(CollectLeft) inputs and the RepartitionExec(Hash) inserted by EnforceDistribution. Verify that all natural build partitions are broadcast and the probe stays upstream, then re-check downstream distribution and sorting requirements. Use the TPC-H SF10 context to compare the intended shuffle reduction.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- databases, distributed-systems
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 55/100