apache / apache/datafusion-ballista
Broadcast the build side of a SortMergeJoinExec without converting to a hash join
- Dominant language
- Rust
- Stars
- 2.1k
- Forks
- 320
- Avg merge
- 1d 22h
- Merged PRs (30d)
- 66
Description
## Background
PR #1904 (closing #1679) added a static-planner optimization that, when one side of a `SortMergeJoinExec` fits under `ballista.optimizer.broadcast_join_threshold_bytes`, **converts the join to a `HashJoinExec(CollectLeft)`** and reuses the existing broadcast lowering. This avoids the two hash-partitioned shuffles a sort-merge join would otherwise require.
That approach changes the join operator (SMJ → hash join). A distinct, complementary optimization is to **keep the `SortMergeJoinExec`** and instead broadcast (replicate) its small build side to every probe task — the probe side stays in its natural partitioning and the merge-join semantics are preserved. This is closer to what Spark does: as @milenkovicm noted on PR #1904, Spark broadcasts the build side of a sort-merge join (observed in TPC-DS Q72 at SF10).
## Motivation
- The convert-to-hash-join path forgoes sort-merge's spill safety (the reason Ballista defaults `prefer_hash_join = false`). It is sound only because the build side is small enough to fit in memory by definition — but the probe side still runs as a hash join. Broadcasting the build side of an SMJ directly would let large joins keep sort-merge execution while still eliminating the small side's shuffle.
- Neither planner currently does this: the static planner converts to a hash join (#1904), and the AQE `DynamicJoinSelectionExec` path promotes to `CollectLeft` hash join as well — AQE does **not** broadcast the build side of an SMJ (per @milenkovicm on #1904).
## Proposed work
1. Spike: determine what a broadcast/replicated build input looks like for `SortMergeJoinExec` — the build side must still be sorted on the join key and materialized on every probe task, while the probe side keeps its natural partitioning (no join-key repartition).
2. Wire it into both planners (static `DefaultDistributedPlanner` and the AQE join-selection path), gated by config, and compare against the convert-to-hash-join path from #1904.
3. Benchmark under **both AQE on and AQE off** — TPC-H SF10 and the TPC-DS Q72 case @milenkovicm called out — with row-count verification.
## References
- PR #1904 — converts a small-side SMJ to a broadcast `CollectLeft` hash join (the alternative this issue proposes to complement)
- #1679 — extend broadcast/colocated rules to `SortMergeJoinExec`
- #1905 — broadcast lowering still hash-repartitions both sides of a `CollectLeft` join (related shuffle-elision follow-up)
Contributor guide
Research direction
Start by reading SortMergeJoinExec and the DefaultDistributedPlanner, then trace the AQE DynamicJoinSelectionExec path and the existing CollectLeft lowering from PR #1904. Determine how a sorted build input could be materialized on every probe task without repartitioning the probe side. Compare both planner paths with AQE enabled and disabled using TPC-H SF10 and TPC-DS Q72, including row-count verification.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100