feat: broadcast small build sides via non-zero CollectLeft thresholds - #1900
Conversation
| .set_u64( | ||
| "datafusion.optimizer.hash_join_single_partition_threshold", | ||
| 0, | ||
| 10 * 1024 * 1024, |
There was a problem hiding this comment.
#1055 is fixed now, so we can revert the workaround
There was a problem hiding this comment.
I don't really understand yet why #1647 fixed this for left/outer joins?
E.g. see
apache/datafusion#12454 for discussion ot it.
It doesn't really change reporting on unmatched rows (for which it needs to produce null rows)
There was a problem hiding this comment.
It probably needs some test cases for those ("duplicated unmatched keys across executors") to make sure they don't regress and then see if we finally solve it / bring it back? :)
There was a problem hiding this comment.
I hit a regression when running benchmarks locally with AQE enabled, so moving this to draft. I also created #1913 so that we have CI tests for AQE to avoid regressions. Looking into feedback now.
…sholds Set hash_join_single_partition_threshold to 10MB and hash_join_single_partition_threshold_rows to 1M so the AQE join resolver collects small build sides into a broadcast (CollectLeft) hash join instead of repartitioning them. Update the client settings test to assert the new default values.
The adaptive join resolver chose CollectLeft purely on size thresholds. A CollectLeft join broadcasts the build (left) side to every probe task, where each task sees only its slice of the probe side, so join types that emit rows on behalf of the build side (Left, Full, LeftSemi, LeftAnti, LeftMark) would have every task emit them independently and produce duplicate or spurious rows. Restrict CollectLeft to join types whose output is driven entirely by the probe side (Inner, Right, RightSemi, RightAnti, RightMark), evaluated on the post-swap join type using the same swap decision as the resolver, and keep unsafe join types repartitioned. Add resolver tests covering left/left-anti (repartitioned) and right (still collected) joins.
6c1b31a to
79651d7
Compare
… matrix The static DefaultDistributedPlanner promoted small-side hash joins to CollectLeft on size alone, with no join-type check, so a LEFT/FULL/semi/anti/ mark join whose build side fit under broadcast_join_threshold_bytes was broadcast and produced duplicate or spurious rows — the same hazard already guarded on the adaptive path. Move the broadcast-safety predicate into physical_optimizer::join_selection so both planners share one definition, and apply it in maybe_promote_to_broadcast using the post-swap join type. Add a test matrix covering both planners, both build-side orientations, every join type, prefer_hash_join on/off, and the post-swap decision, plus a unit test that locks down the broadcast-safe join-type set.
…eshold apache#1902 added tests asserting the Ballista default hash_join_single_partition_threshold is 0; this PR raises that default to 10 MB (and the row threshold to 1M), so should_apply_defaults_when_upgrading_plain_config now asserts the new values. Give should_preserve_user_overrides_on_upgrade an override distinct from the default so it still proves the user value survives.
|
@Dandandan This PR has expanded a lot since the last review. Could you take another look? Also, fyi @avantgardnerio |
DataFusion 54 removed ExecutionPlan::as_any in favor of the inherent downcast_ref on dyn ExecutionPlan, which auto-derefs through Arc. The new broadcast-safety tests still used as_any().downcast_ref(), which no longer compiles. Drop the as_any() hop to match the rest of the file.
avantgardnerio
left a comment
avantgardnerio
left a comment
There was a problem hiding this comment.
From claude:
The static path's safety guard is bypassed when DataFusion's own JoinSelection promotes the join before Ballista sees it. With the PR's session default of
hash_join_single_partition_threshold = 10 MB, DF's try_collect_left (datafusion/physical-optimizer/src/join_selection.rs:200) will set PartitionMode::CollectLeft on any
small-build-side join — it only swaps when both sides are collectable; for small LEFT JOIN big with the small side already on the left, it falls into the (true, false) arm and just
stamps CollectLeft on the LEFT join. No swap, no join-type check.
Test in a stacked PR incoming
|
I'm not super familiar with joins yet, so this could be complete AI slop, but here's a failing test: andygrove#89 Sorry if it's a false alarm, I figured better to share than let a bug hit main. I'll flip my review to approve now, merge if I'm wrong! |
avantgardnerio
left a comment
avantgardnerio
left a comment
There was a problem hiding this comment.
Approved pending review of stacked PR
DataFusion's JoinSelection runs during create_physical_plan and, with the non-zero hash_join_single_partition_threshold default this PR sets, can stamp CollectLeft on a join without restricting by join type. The distributed planner's broadcast-safety guard only covered the Partitioned -> CollectLeft promotion path, so a pre-promoted unsafe join (e.g. a LEFT join with the small side already on the left) reached the broadcast-lowering branch and replicated the outer side across every probe task, producing duplicate/spurious rows. In maybe_promote_to_broadcast, demote any CollectLeft hash join whose join type is not broadcast-safe back to a Partitioned (shuffle) join, hash-partitioning both inputs on the join keys. This is a correctness guard, so it runs regardless of the Ballista broadcast threshold. Add a test that mirrors the production session config so DataFusion's own JoinSelection performs the promotion.
|
@avantgardnerio found a real gap in the broadcast-safety guard via a stacked test PR (andygrove#89), now fixed here in b40a670. DataFusion's own The fix moves the broadcast-safety check to the front of |
|
The TPC-H CI job appears to be hanging. This is likely catching a real issue. I will investigate. |
Benchmark hangs on some queries when AQE is enabled due to changes in this PR |
DataFusion 54 populates dynamic filters at runtime from an upstream operator (a hash join build side, a TopK heap, a partial aggregate) and reads them in a downstream scan within the same plan. Ballista splits a plan into stages at shuffle and broadcast boundaries that execute as independent tasks, so when the producing operator and the consuming scan land in different stages the filter is never populated across the boundary and the consuming scan blocks forever. This deadlocks every multi-join TPC-H query under the adaptive (AQE) planner, where broadcast (CollectLeft) hash joins create the cross-stage producer/ consumer split — the build stages complete but the probe stages hang. The static planner avoids it only because it plans sort-merge joins, which carry no join dynamic filter. Pin datafusion.optimizer.enable_dynamic_filter_pushdown to false in the Ballista session defaults, alongside the other DataFusion options Ballista already disables because they rely on in-process state that cannot cross stage boundaries. Tracked by apache#1375.
|
The new AQE-on TPC-H CI job was hanging (not the AQE-off job). Root-caused and fixed in 382b8e1. SymptomUnder the adaptive (AQE) planner, the job completed the AQE-off suite, then hung on the first multi-join query (Q2) and never recovered. It reproduces locally on a 1-executor / 4-slot / 16-partition cluster and is independent of slot/partition counts (hangs at Root cause: DataFusion 54 dynamic filter pushdownDataFusion 54 populates dynamic filters at runtime from an upstream operator (a hash-join build side, a TopK heap, a partial aggregate) and reads them in a downstream scan within the same plan. Ballista splits a plan into stages at shuffle/broadcast boundaries that run as independent tasks, so when the producing operator and the consuming scan land in different stages, the filter is never populated across the boundary and the consuming scan blocks forever. This deadlocks every multi-join query under AQE, because broadcast ( The regression entered when this branch merged How it was confirmed
FixPin Verified locally: all 21 AQE-on queries (Q16 omitted, as in CI) complete with the default config at the CI topology. Orthogonal note: Q11 and Q15 return 0 rows under both AQE on and off at SF10 — a pre-existing correctness issue, not introduced here and not caught by CI (which doesn't validate row counts). Worth a separate issue. |
|
Moving to draft until #1917 is merged. I want to make sure we are not introducing any regressions |
…apache#1900) * feat(core): broadcast small build sides via non-zero CollectLeft thresholds Set hash_join_single_partition_threshold to 10MB and hash_join_single_partition_threshold_rows to 1M so the AQE join resolver collects small build sides into a broadcast (CollectLeft) hash join instead of repartitioning them. Update the client settings test to assert the new default values. * fix(scheduler): restrict CollectLeft broadcast to safe join types The adaptive join resolver chose CollectLeft purely on size thresholds. A CollectLeft join broadcasts the build (left) side to every probe task, where each task sees only its slice of the probe side, so join types that emit rows on behalf of the build side (Left, Full, LeftSemi, LeftAnti, LeftMark) would have every task emit them independently and produce duplicate or spurious rows. Restrict CollectLeft to join types whose output is driven entirely by the probe side (Inner, Right, RightSemi, RightAnti, RightMark), evaluated on the post-swap join type using the same swap decision as the resolver, and keep unsafe join types repartitioned. Add resolver tests covering left/left-anti (repartitioned) and right (still collected) joins. * fix(scheduler): guard static planner broadcast and add join-type test matrix The static DefaultDistributedPlanner promoted small-side hash joins to CollectLeft on size alone, with no join-type check, so a LEFT/FULL/semi/anti/ mark join whose build side fit under broadcast_join_threshold_bytes was broadcast and produced duplicate or spurious rows — the same hazard already guarded on the adaptive path. Move the broadcast-safety predicate into physical_optimizer::join_selection so both planners share one definition, and apply it in maybe_promote_to_broadcast using the post-swap join type. Add a test matrix covering both planners, both build-side orientations, every join type, prefer_hash_join on/off, and the post-swap decision, plus a unit test that locks down the broadcast-safe join-type set. * test(core): update upgrade defaults test for non-zero CollectLeft threshold apache#1902 added tests asserting the Ballista default hash_join_single_partition_threshold is 0; this PR raises that default to 10 MB (and the row threshold to 1M), so should_apply_defaults_when_upgrading_plain_config now asserts the new values. Give should_preserve_user_overrides_on_upgrade an override distinct from the default so it still proves the user value survives. * fix(scheduler): use dyn ExecutionPlan::downcast_ref in broadcast tests DataFusion 54 removed ExecutionPlan::as_any in favor of the inherent downcast_ref on dyn ExecutionPlan, which auto-derefs through Arc. The new broadcast-safety tests still used as_any().downcast_ref(), which no longer compiles. Drop the as_any() hop to match the rest of the file. * fix: demote DataFusion-promoted unsafe CollectLeft joins to partitioned DataFusion's JoinSelection runs during create_physical_plan and, with the non-zero hash_join_single_partition_threshold default this PR sets, can stamp CollectLeft on a join without restricting by join type. The distributed planner's broadcast-safety guard only covered the Partitioned -> CollectLeft promotion path, so a pre-promoted unsafe join (e.g. a LEFT join with the small side already on the left) reached the broadcast-lowering branch and replicated the outer side across every probe task, producing duplicate/spurious rows. In maybe_promote_to_broadcast, demote any CollectLeft hash join whose join type is not broadcast-safe back to a Partitioned (shuffle) join, hash-partitioning both inputs on the join keys. This is a correctness guard, so it runs regardless of the Ballista broadcast threshold. Add a test that mirrors the production session config so DataFusion's own JoinSelection performs the promotion. * fix: disable DataFusion dynamic filter pushdown in session defaults DataFusion 54 populates dynamic filters at runtime from an upstream operator (a hash join build side, a TopK heap, a partial aggregate) and reads them in a downstream scan within the same plan. Ballista splits a plan into stages at shuffle and broadcast boundaries that execute as independent tasks, so when the producing operator and the consuming scan land in different stages the filter is never populated across the boundary and the consuming scan blocks forever. This deadlocks every multi-join TPC-H query under the adaptive (AQE) planner, where broadcast (CollectLeft) hash joins create the cross-stage producer/ consumer split — the build stages complete but the probe stages hang. The static planner avoids it only because it plans sort-merge joins, which carry no join dynamic filter. Pin datafusion.optimizer.enable_dynamic_filter_pushdown to false in the Ballista session defaults, alongside the other DataFusion options Ballista already disables because they rely on in-process state that cannot cross stage boundaries. Tracked by apache#1375.
…dopt clobbered 53->54 features
The 54.0.0 merge suffered a delete/modify pathology: files the fork had
deleted relative to base 53.0.0 silently stayed deleted whenever upstream
had not touched them during the 53->54 cycle, and several conflict
resolutions kept stale fork-side file versions. This commit repairs it.
Tree repairs (restored from upstream 54.0.0):
- ballista-cli: tui/domain/{mod,executors}.rs, ui/main/jobs/dot_parser.rs,
exec.rs — the crate did not compile with default features
- examples/standalone-broadcast-join.rs (Cargo.toml still declared it)
- scheduler/src/api/ (routes.rs + handlers.rs + mod.rs) and display.rs
- docs python/ dir, example notebooks, test_jupyter.py, TUI screenshots,
benchmarks/tpch-gen.sh, vendored datafusion*.proto stubs
Re-adopted upstream features the merge clobbered:
- apache#1911 partition pruning (under disable-stage-plan-cache) + tests
- apache#1968 shuffle-read metrics, mapped onto the Spice fetch pipeline, and
child-operator metrics in writer plan displays
- apache#1995 session-keyed runtime cache wired into the executor (plus the
upstream --memory-pool-size FairSpillPool option)
- apache#1982 selective intermediate-stage shuffle cleanup end-to-end (also
applied to the in-memory shuffle manager)
- apache#1949 failed-task REST surfacing + apache#1818 CORS/--disable-rest-api,
TaskManager::get_all_jobs / get_job_config, JobState::get_all_jobs,
ExecutorManager::get_executors_state (upstream names)
- apache#1900/apache#1904 static-planner broadcast promotion (maybe_promote_to_
broadcast, CollectLeft demotion guard, SMJ->hash conversion, broadcast
stage lowering, broadcast_join_threshold_bytes config) + 12 tests
- apache#1999 task duration in finished-task log
- execute_physical_plan client entry point (apache#1924/apache#1941)
- ballista.client.io_retries_times / io_retry_wait_time_ms, wired into
the evict-and-retry fetch loop (fork-preserving defaults: 1 retry, 0ms)
Cleanups and adaptations:
- deleted dead executor/src/client_pool.rs, orphaned AQE optimizer files,
committed .pending-snap artifacts
- shuffle.remote_read_prefer_flight now defaults to true: the block-IO
transport cannot serve sort-based shuffle (default-on), so the old
default pair was incoherent; removed the block-IO sort-shuffle test
cases accordingly
- regenerated TPC-H plan-stability goldens (TopK single-stage merge from
the fork's #28.1 was not reflected in the goldens taken from upstream)
- SPICE_FORK_CHANGES.md updated to match reality (#61/#62a rows, apache#1951
disposition, repair-commit inventory)
3 participants
Which issue does this PR close?
Part of #1770.
This PR makes the default-value change suggested in #1770 (restore non-zero
thresholds), and adds the join-type safety guard that broadcasting those build
sides turns out to require. The other part of #1770 — having the static job
planner respect
datafusion.optimizer.hash_join_single_partition_threshold(_rows)instead of
ballista.optimizer.broadcast_join_threshold_bytes— is not includedhere, so #1770 is left open.
Rationale for this change
Ballista pins
datafusion.optimizer.hash_join_single_partition_thresholdand..._threshold_rowsto0, which disablesCollectLeft(broadcast) hash joins.While the threshold is
0, the AQE join resolver (DynamicJoinSelectionExec/to_actual_join) never promotes a small build side to a broadcastCollectLeftjoin — every join is repartitioned through a shuffle, even single-row dimension
tables. Raising the thresholds lets the resolver broadcast small build sides,
which is a large TPC-H win (see numbers below).
The
0pin was originally a workaround for #1055 ("Left/full outer joinincorrect for CollectLeft / broadcast"). Review on this PR established that #1647
never actually fixed the underlying problem for distributed execution: a
CollectLeftjoin replicates the build (left) side to every probe task, andeach task processes only its slice of the probe (right) side. That is correct
only for join types whose output is driven entirely by the probe side. Any join
type that emits rows on behalf of the build side — unmatched outer rows, or
semi/anti/mark rows that depend on a build row's global match status — would have
those rows emitted independently by every probe task, producing duplicate or
spurious results.
So we cannot simply un-zero the thresholds. We un-zero them and restrict
CollectLeftpromotion to the join types that are safe to broadcast. This guardis needed on both planning paths, including the static planner, where the same
unsafe promotion was a latent bug (the #1647 path) that could fire whenever
prefer_hash_join=true.Measured on TPC-H SF10 (1 scheduler + 1 executor, 8 task slots, 8 partitions,
3 iterations, AQE enabled), comparing the resolver with the threshold at
0vs.the values in this PR:
Per-query, broadcast fires for small dimension joins and is left off for
big-vs-big joins. Examples: Q2 0.82 s → 0.12 s, Q8 3.13 s → 0.46 s,
Q9 4.61 s → 1.09 s, Q11 0.59 s → 0.09 s, Q21 5.17 s → 1.49 s. Row counts are
unchanged for all 22 queries.
What changes are included in this PR?
datafusion.optimizer.hash_join_single_partition_thresholdto 10 MB anddatafusion.optimizer.hash_join_single_partition_threshold_rowsto 1,000,000in the Ballista session defaults (previously both
0).collect_left_broadcast_safe(JoinType)predicate that allowsCollectLeftbroadcast only for probe-driven join types:Inner,Right,RightSemi,RightAnti,RightMark. Safety is evaluated on the join typeafter any build-side swap, since the build side is always the left input of
the resulting
HashJoinExec.to_actual_joinindynamic_join.rs):a small build side is collected into a
CollectLeftjoin only when thepost-swap join type is broadcast-safe, otherwise it is repartitioned. The swap
decision reuses
SelectJoinRule::supports_swap_join_order, the same call theresolver later uses to perform the swap, so the predicted and actual join types
stay in sync.
maybe_promote_to_broadcastinplanner.rs), closing the latent feat(scheduler): broadcast-style hash join for small-side joins #1647 path so it can no longer promote anunsafe join type to
CollectLeft.JoinTypematrix for the predicate,resolver tests covering both swap orientations and
prefer_hash_joinon/off,static-planner tests across all join types and build-side orientations, a
LEFT joinsmall-build "must not broadcast" case, and a sort-merge-join"no broadcast" case.
reflect that it now checks the thresholds are set).
Are there any user-facing changes?
Yes. With AQE enabled, a hash join whose build side fits under these thresholds
and whose join type is broadcast-safe is executed as a broadcast (
CollectLeft)join instead of being repartitioned. No SQL semantics change, and results
(including row counts) are unchanged. Join types that emit rows on behalf of the
build side continue to be repartitioned.
The static (non-AQE) planner still uses a separate config
(
ballista.optimizer.broadcast_join_threshold_bytes) to decide whether tobroadcast, so its broadcast frequency is unchanged. Its promotion path now also
applies the broadcast-safety guard, which only removes previously-unsafe
promotions (reachable with
prefer_hash_join=true).