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/07/13 13:26:33 UTC
[I] AQE regresses join-heavy TPC-H queries at SF1000 instead of promoting broadcasts [datafusion-ballista]
andygrove opened a new issue, #2024:
URL: https://github.com/apache/datafusion-ballista/issues/2024
## Describe the bug
At TPC-H SF1000 with sort-merge joins, enabling the adaptive planner (`ballista.planner.adaptive.enabled=true`) makes the join-heavy queries **slower**, not faster, and costs two queries entirely.
These are exactly the queries that dynamic join selection (#1752 / `DynamicJoinSelectionExec`) is supposed to rescue by promoting the small side to a broadcast. Instead they regress 10–36%, which is what you would expect if the machinery is adding runtime overhead **without actually firing the broadcast promotion**.
| Query | AQE off (s) | AQE on (s) | Delta |
|-------|------------:|-----------:|-------|
| 3 | 189.3 | 248.3 | **+31%** |
| 5 | 446.0 | 605.6 | **+36%** |
| 7 | 481.5 | 545.7 | **+13%** |
| 8 | 647.4 | 733.6 | **+13%** |
| 9 | 900.2 | 994.7 | **+10%** |
| 19 | 28.4 | 81.9 | **+188%** |
| 17 | 670.9 | 348.9 | −48% |
| 18 | 587.9 | **deadlock** (#2023) | — |
| 21 | 1012.9 | **execution error** | — |
Totals over the 20 queries the adaptive path completed: **AQE off 4173.9s vs AQE on 4344.0s** — about 4% slower overall, plus 2 failures. The static path completed 22/22.
Q17 (−48%) shows the promotion machinery *can* fire. It just doesn't fire on most of the joins that dominate runtime at this scale.
## Context: this is where the gap to Spark lives
Same hardware (2 executors × 8 cores / 56Gi), same SF1000 node-local parquet, single iteration:
| Engine / config | Total, 22 queries |
|---|---:|
| Spark 3.5 (AQE on) | **4621.5s** |
| Ballista, sort-merge joins (default: `prefer_hash_join=false`), AQE off | **5774.7s** |
Spark is 1.25× faster than stock Ballista, and the deficit is concentrated almost entirely in Q7/Q9/Q21. Inspecting the Spark plans, Spark's AQE emits **36 `BroadcastHashJoin`s alongside 93 `SortMergeJoin`s** — i.e. Spark's "sort-merge default" is not really all sort-merge; roughly a quarter of its joins get promoted to broadcasts at runtime.
Ballista's static planner promotes **zero**, because broadcast promotion only fires for `HashJoinExec` and `prefer_hash_join=false` makes every join an SMJ (see #1922). The adaptive planner is the mechanism that is supposed to recover those broadcasts, and at SF1000 it is not doing so.
For comparison, forcing hash joins (`prefer_hash_join=true`) on the same cluster is **1.28× faster than Spark** over Q1–17 (2683.6s vs 3441.4s), versus 4039.1s for the SMJ default — so the join strategy alone swings Ballista by ~1.5×. (That config currently cannot complete the suite; see the CollectLeft OOM issue filed alongside this one.)
## To Reproduce
Cluster: 2 executors, 8 CPU / 56Gi each, `--memory-pool-size=48GB`, 1 scheduler. TPC-H SF1000 parquet on node-local disk. Ballista `54.0.0-rc2`. Each query run as its own job, 1 iteration.
```sh
cargo run --release --bin tpch -- benchmark ballista \
--host <scheduler> --port 50050 \
--query 9 --path /mnt/bigdata/tpch/sf1000 --format parquet \
--partitions 32 --iterations 1 \
-c datafusion.optimizer.prefer_hash_join=false \
-c datafusion.optimizer.enable_dynamic_filter_pushdown=false \
-c ballista.planner.adaptive.enabled=true # vs =false
```
## Expected behavior
With the adaptive planner enabled, join-heavy queries whose build side fits the broadcast threshold should be promoted to a broadcast and get *faster* — approaching what Spark's AQE achieves — rather than regressing.
## Additional context
- Single iteration per query, so small deltas are noisy; the 10–36% regressions on Q3/Q5/Q7/Q8/Q9 are consistent in direction across all of them, and Q19's +188% and Q17's −48% are far outside noise.
- Related: #1922 (broadcast an SMJ build side without converting to a hash join), #1905 (broadcast lowering still hash-repartitions build/probe), #1359 (AQE epic).
- Worth checking whether the broadcast threshold (`ballista.optimizer.broadcast_join_threshold_bytes`) is simply being exceeded by every relevant build side at SF1000, in which case the question is whether the threshold or the estimate is 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.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: [I] AQE regresses join-heavy TPC-H queries at SF1000 instead of promoting broadcasts [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #2024:
URL: https://github.com/apache/datafusion-ballista/issues/2024#issuecomment-5004188223
Re-measured on `main` @ `26b29391`, same cluster shape as the original report (2 executors x 8 cores / 56 GiB, SF1000 on node-local disk, `target_partitions=32`, `prefer_hash_join=false`, dynamic-filter pushdown off, 1 iteration, full-suite runs).
**The regression reproduces, but the stated cause does not hold.**
| Query | AQE off | AQE on | Delta | Originally reported |
|-------|--------:|-------:|------:|--------------------:|
| 3 | 206.2 | 255.0 | +24% | +31% |
| 5 | 542.7 | 700.5 | +29% | +36% |
| 7 | 575.5 | 531.4 | **−8%** | +13% |
| 8 | 657.1 | 801.2 | +22% | +13% |
| 9 | 891.8 | 1025.2 | +15% | +10% |
| 17 | 697.8 | 400.4 | −43% | −48% |
| 18 | 620.7 | 820.9 | +32% | — |
| 19 | 39.6 | 42.1 | **+6%** | +188% |
| **Total** | **6378.2** | **6320.7** | **−0.9%** | |
Two specifics from the original report have changed: **all 22 queries now complete under the adaptive planner** (it no longer costs two queries), and **Q19's +188% is gone**. Q7 has flipped from a regression to an improvement. Q3/Q5/Q8/Q9/Q18 still regress, so the core complaint stands.
Note the totals are a wash (−0.9%, inside this cluster's noise): the adaptive planner's wins on Q17/Q2/Q1 roughly cancel its losses on the join-heavy queries. The aggregate hides movement in both directions rather than showing nothing happening.
**On the hypothesis.** The description reasons that the regressions are "what you would expect if the machinery is adding runtime overhead **without actually firing the broadcast promotion**". That is not what happens — the promotion does fire. Running the scheduler with `RUST_LOG=info,ballista_scheduler::state::aqe=debug` prints every resolver decision; Q2 at SF1000 gives:
```
to_actual_join - plan_id: 1, decision: CollectLeft left: (Inexact(799140) | Absent), right: (Exact(10000000) | Absent)
to_actual_join - plan_id: 2, decision: CollectLeft left: (Inexact(3196560) | Absent), right: (Exact(25) | Absent)
to_actual_join - plan_id: 3, decision: CollectLeft left: (Inexact(3196560) | Absent), right: (Inexact(1) | Absent)
to_actual_join - plan_id: 4, decision: Partitioned left: (Exact(800000000) | Exact(25628311552)), right: (Exact(10000000) | Exact(160374784))
to_actual_join - plan_id: 5, decision: CollectLeft left: (Inexact(10000000) | Inexact(320000000)), right: (Exact(25) | Exact(400))
to_actual_join - plan_id: 7, decision: CollectLeft left: (Inexact(639312) | Absent), right: (Inexact(159969600) | Inexact(3839270400))
```
Six of ten decisions are `CollectLeft`. Broadcast promotion is working; this is not dead machinery.
`hash_join_single_partition_threshold` is also no longer zero — #1900 set it to 10 MB — so the older "both broadcast mechanisms are effectively dead" reading is out of date.
**What the decisions do show** is a different defect, now filed as #2081: when `total_byte_size` is `Absent`, the resolver falls back to comparing a row count against `hash_join_single_partition_threshold_rows` (1,000,000), which is size-blind. `plan_id: 1` and `plan_id: 7` above are broadcast at 799,140 and 639,312 rows of *unknown* size — `part`/`supplier`-derived rows carrying string columns. Absent sizes are the norm here, since DataFusion discards `total_byte_size` on every join and one `Utf8` column stops `Statistics::calculate_total_byte_size` rebuilding it.
Worth being clear that this does **not** yet explain the regressions. #2084 makes the decision size-aware, and while it demonstrably changes the plan (Q2's 799k-row broadcast becomes `Partitioned` while `nation`/`region` still broadcast), it produced **no measured speedup** at SF1000 — the one query whose plan it changes got marginally slower. So the cause of the Q3/Q5/Q8/Q9/Q18 regressions is still open. Since broadcast promotion is firing, the remaining candidates are the cost of the promotions themselves, the extra shuffles from the `Unknown` -> `Repartition` -> re-decide path, or replanning overhead — not an absence of promotion.
Method and current numbers: https://datafusion.apache.org/ballista/contributors-guide/benchmarking.html
--
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