You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
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.
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).
Background
PR #1904 (closing #1679) added a static-planner optimization that, when one side of a
SortMergeJoinExecfits underballista.optimizer.broadcast_join_threshold_bytes, converts the join to aHashJoinExec(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
SortMergeJoinExecand 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
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.DynamicJoinSelectionExecpath promotes toCollectLefthash join as well — AQE does not broadcast the build side of an SMJ (per @milenkovicm on feat: broadcast small build side of SortMergeJoinExec in the static planner #1904).Proposed work
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).DefaultDistributedPlannerand the AQE join-selection path), gated by config, and compare against the convert-to-hash-join path from feat: broadcast small build side of SortMergeJoinExec in the static planner #1904.References
CollectLefthash join (the alternative this issue proposes to complement)SortMergeJoinExecCollectLeftjoin (related shuffle-elision follow-up)