AQE regresses join-heavy TPC-H queries at SF1000 instead of promoting broadcasts

[andygrove]

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 BroadcastHashJoins alongside 93 SortMergeJoins — 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.

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: Broadcast the build side of a SortMergeJoinExec without converting to a hash join #1922 (broadcast an SMJ build side without converting to a hash join), Broadcast lowering still hash-repartitions the build and probe sides of a CollectLeft join #1905 (broadcast lowering still hash-repartitions build/probe), [EPIC] Adaptive Query Execution (AQE) #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.

[andygrove]

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