apache / apache/datafusion-ballista

Broadcast the build side of a SortMergeJoinExec without converting to a hash join

Open
#1,922 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

## 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

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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.