You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/06/26 15:17:02 UTC
[PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
andygrove opened a new pull request, #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900
# Which issue does this PR close?
Part of #1770.
This PR makes the default-value change suggested in #1770 (restore non-zero
thresholds). The other part of #1770 — having the *static* job planner respect
`datafusion.optimizer.hash_join_single_partition_threshold(_rows)` (it currently
uses `ballista.optimizer.broadcast_join_threshold_bytes`) — is not included here,
so #1770 is left open.
# Rationale for this change
Ballista pins `datafusion.optimizer.hash_join_single_partition_threshold` and
`..._threshold_rows` to `0`, which disables `CollectLeft` (broadcast) hash joins.
That was a deliberate workaround for #1055 ("Left/full outer join incorrect for
CollectLeft / broadcast"), which is now fixed (closed COMPLETED).
While the threshold is `0`, the AQE join resolver (`DynamicJoinSelectionExec` /
`to_actual_join`) never promotes a small build side to a broadcast `CollectLeft`
join — every join is repartitioned through a shuffle, even single-row dimension
tables. Restoring the thresholds (now safe) lets the resolver broadcast small
build sides.
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 `0` vs.
the values in this PR:
| | threshold = 0 | threshold = 10 MB / 1M rows |
|---|---:|---:|
| full 22-query total | 32.2 s | 16.2 s |
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?
- Set `datafusion.optimizer.hash_join_single_partition_threshold` to 10 MB and
`datafusion.optimizer.hash_join_single_partition_threshold_rows` to 1,000,000
in the Ballista session defaults (previously both `0`).
- Update the client settings test to assert the new defaults (and rename it to
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
is executed as a broadcast (`CollectLeft`) join instead of being repartitioned.
No SQL semantics change. The static (non-AQE) planner uses a separate config
(`ballista.optimizer.broadcast_join_threshold_bytes`) and is unaffected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#discussion_r3488588116
##########
ballista/core/src/extension.rs:
##########
@@ -760,18 +760,15 @@ impl SessionConfigHelperExt for SessionConfig {
// same like previous comment
.set_bool("datafusion.sql_parser.map_string_types_to_utf8view", false)
//
- // As mentioned in https://github.com/apache/datafusion-ballista/issues/1055
- // "Left/full outer join incorrect for CollectLeft / broadcast"
- //
- // In order to make correct results (decreasing performance) CollectLeft
- // has been disabled until fixed
+ // A build side smaller than these thresholds is collected into a
+ // CollectLeft (broadcast) hash join rather than being repartitioned.
.set_u64(
"datafusion.optimizer.hash_join_single_partition_threshold",
- 0,
+ 10 * 1024 * 1024,
Review Comment:
I hit a regression when running benchmarks locally with AQE enabled, so moving this to draft. I also created https://github.com/apache/datafusion-ballista/pull/1913 so that we have CI tests for AQE to avoid regressions. Looking into feedback now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#issuecomment-4844214395
@Dandandan This PR has expanded a lot since the last review. Could you take another look? Also, fyi @avantgardnerio
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#issuecomment-4848473303
Moving to draft until https://github.com/apache/datafusion-ballista/pull/1917 is merged. I want to make sure we are not introducing any regressions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#discussion_r3488478444
##########
ballista/core/src/extension.rs:
##########
@@ -760,18 +760,15 @@ impl SessionConfigHelperExt for SessionConfig {
// same like previous comment
.set_bool("datafusion.sql_parser.map_string_types_to_utf8view", false)
//
- // As mentioned in https://github.com/apache/datafusion-ballista/issues/1055
- // "Left/full outer join incorrect for CollectLeft / broadcast"
- //
- // In order to make correct results (decreasing performance) CollectLeft
- // has been disabled until fixed
+ // A build side smaller than these thresholds is collected into a
+ // CollectLeft (broadcast) hash join rather than being repartitioned.
.set_u64(
"datafusion.optimizer.hash_join_single_partition_threshold",
- 0,
+ 10 * 1024 * 1024,
Review 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? :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#issuecomment-4846393463
@avantgardnerio found a real gap in the broadcast-safety guard via a stacked test PR (andygrove/datafusion-ballista#89), now fixed here in b40a6701.
DataFusion's own `JoinSelection` runs during `create_physical_plan` and, with the non-zero `hash_join_single_partition_threshold` this PR sets, stamps `CollectLeft` without restricting by join type. The guard only covered the `Partitioned -> CollectLeft` promotion path, so a pre-promoted unsafe join (e.g. `small LEFT JOIN big` with the small side already on the left) bypassed it and reached the broadcast-lowering branch, replicating the outer side across every probe task.
The fix moves the broadcast-safety check to the front of `maybe_promote_to_broadcast`: any `CollectLeft` hash join whose join type isn't `collect_left_broadcast_safe` is demoted back to a `Partitioned` (shuffle) join, re-hash-partitioning both inputs on the join keys. It runs ahead of the threshold check so the correctness guard still fires even if Ballista's own broadcast promotion is disabled while DF's threshold is non-zero. avantgardnerio's reproducer test is pulled in verbatim (with the explanatory comment swapped for a current-state one to match the repo convention) and now passes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#issuecomment-4847047476
> 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#issuecomment-4847651420
The new AQE-on TPC-H CI job was hanging (not the AQE-off job). Root-caused and fixed in 382b8e1e.
## Symptom
Under 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 `-c 4`/4, `-c 8`/8, `-c 16`/16).
## Root cause: DataFusion 54 dynamic filter pushdown
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/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 (`CollectLeft`) hash joins create the cross-stage producer/consumer split (the probe-side scan carries `predicate=DynamicFilter [ empty ]`). The static planner escapes it only because it plans sort-merge joins, which carry no join dynamic filter — which is why the AQE-off suite passed and AQE-on hung.
The regression entered when this branch merged `main`: the DataFusion 54 upgrade (#1906) and the AQE-on CI (#1913) landed in the same merge, and AQE-on had never been exercised in CI before.
## How it was confirmed
- `main` (same DF54 + same AQE-on CI) passes the TPC-H job in ~24 min; only this branch hung → the un-zeroed thresholds (broadcast under AQE) are what trigger it.
- DF53-vs-DF54 bisect: on the pre-merge DF53 commit, Q2/Q5/Q8/Q9 all run fine (Q2 0.16 s, 100 rows). On DF54 they all hang.
- `enable_dynamic_filter_pushdown=false` makes Q2 complete and removes the `DynamicFilter` from the plan; the per-type sub-flags (join/topk/aggregate) each individually leave the hang.
## Fix
Pin `datafusion.optimizer.enable_dynamic_filter_pushdown=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 (e.g. `enable_physical_uncorrelated_scalar_subquery`). Long-term, carrying dynamic filters across stage boundaries is tracked by #1375.
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#discussion_r3488190215
##########
ballista/core/src/extension.rs:
##########
@@ -760,18 +760,15 @@ impl SessionConfigHelperExt for SessionConfig {
// same like previous comment
.set_bool("datafusion.sql_parser.map_string_types_to_utf8view", false)
//
- // As mentioned in https://github.com/apache/datafusion-ballista/issues/1055
- // "Left/full outer join incorrect for CollectLeft / broadcast"
- //
- // In order to make correct results (decreasing performance) CollectLeft
- // has been disabled until fixed
+ // A build side smaller than these thresholds is collected into a
+ // CollectLeft (broadcast) hash join rather than being repartitioned.
.set_u64(
"datafusion.optimizer.hash_join_single_partition_threshold",
- 0,
+ 10 * 1024 * 1024,
Review Comment:
I don't really understand yet why https://github.com/apache/datafusion-ballista/pull/1647 fixed this for left/outer joins?
E.g. see
https://github.com/apache/datafusion/issues/12454 for discussion ot it.
It doesn't really change reporting on unmatched rows (for which it needs to produce null rows)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#discussion_r3488163771
##########
ballista/core/src/extension.rs:
##########
@@ -760,18 +760,15 @@ impl SessionConfigHelperExt for SessionConfig {
// same like previous comment
.set_bool("datafusion.sql_parser.map_string_types_to_utf8view", false)
//
- // As mentioned in https://github.com/apache/datafusion-ballista/issues/1055
- // "Left/full outer join incorrect for CollectLeft / broadcast"
- //
- // In order to make correct results (decreasing performance) CollectLeft
- // has been disabled until fixed
+ // A build side smaller than these thresholds is collected into a
+ // CollectLeft (broadcast) hash join rather than being repartitioned.
.set_u64(
"datafusion.optimizer.hash_join_single_partition_threshold",
- 0,
+ 10 * 1024 * 1024,
Review Comment:
https://github.com/apache/datafusion-ballista/issues/1055 is fixed now, so we can revert the workaround
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#issuecomment-4845857342
I'm not super familiar with joins yet, so this could be complete AI slop, but here's a failing test: https://github.com/andygrove/datafusion-ballista/pull/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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: broadcast small build sides via non-zero CollectLeft thresholds [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #1900:
URL: https://github.com/apache/datafusion-ballista/pull/1900#issuecomment-4846798865
The TPC-H CI job appears to be hanging. This is likely catching a real issue. I will investigate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org