apache / apache/datafusion-ballista

Broadcast lowering still hash-repartitions the build and probe sides of a CollectLeft join

Open
#1,905 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.