Repository navigation
Conversation
Add an adaptive query execution (AQE) prototype to `dfbench`. With `dfbench tpch --adaptive`, every partitioned hash join whose probe side is hash repartitioned is wrapped in an `AdaptiveHashJoinExec`. At runtime it collects the build side up to a threshold, then runs the join in `CollectLeft` mode when the build side is small, without repartitioning the probe side, or in `Partitioned` mode otherwise, streaming the rest of the build side behind the batches collected so far. The rest of the plan, its execution order and its dynamic filters are unchanged.
andygrove
changed the title
Move the adaptive hash join out of the benchmarks crate. With `datafusion.optimizer.enable_adaptive_hash_join` (off by default), the new `AdaptiveHashJoin` physical optimizer rule wraps each partitioned hash join whose probe side is hash repartitioned in an `AdaptiveHashJoinExec`. A hash join reads its whole build side before probing, so the new operator adds no wait: once the build side is read, it runs the join in `CollectLeft` mode, without repartitioning the probe side, if the build side is within `adaptive_hash_join_threshold_rows`, `adaptive_hash_join_threshold` and `adaptive_hash_join_max_probe_ratio`, and in `Partitioned` mode as planned otherwise. It keeps the wrapped join's properties and dynamic filter, and accounts for the build batches it collects in the memory pool, falling back to `Partitioned` mode when the pool is full. `dfbench tpch --adaptive` now sets the option.
andygrove
changed the title
- Count collected build batches with `RecordBatchMemoryCounter`, check the byte threshold as they arrive, and keep their memory reserved until the join takes each batch - Keep the inputs needed to create the join to run apart from the runtime state, so the shared future creating it no longer keeps that state alive - Hand collected batches over through `StreamingTableExec`, releasing each batch as soon as the join takes it - Share the eligibility check between the rule and the operator, derive the planned partitionings from the wrapped join, and reuse `need_produce_result_in_final`, `execute_stream_partitioned` and `validate_child_count!` - Cancel abandoned partitions outside the lock, and only compute probe statistics when the probe ratio applies - Share the test setup, and include the same queries for each setting in the sqllogictest
andygrove
marked this pull request as ready for review
October 14, 2026 21:42
andygrove
requested review from
Dandandan,
alamb,
avantgardnerio,
comphead and
gabotechs
October 14, 2026 21:42
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #26328 +/- ##
==========================================
- Coverage 82.80% 82.79% -0.01%
==========================================
Files 1149 1151 +2
Lines 454146 454604 +458
Branches 454146 454604 +458
==========================================
+ Hits 376035 376386 +351
- Misses 55260 55336 +76
- Partials 22851 22882 +31 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing andygrove/aqe-datafusion-tpc-h-prototype (47b94c4) to 0d79252 (merge-base) diff Run configurationrun benchmark tpchResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing andygrove/aqe-datafusion-tpc-h-prototype (47b94c4) to 0d79252 (merge-base) diff Run configurationrun benchmark clickbench_partitionedResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing andygrove/aqe-datafusion-tpc-h-prototype (47b94c4) to 0d79252 (merge-base) diff Run configurationrun benchmark tpcdsResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing andygrove/aqe-datafusion-tpc-h-prototype (47b94c4) to 0d79252 (merge-base) diff Run configurationrun benchmark tpchCPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing andygrove/aqe-datafusion-tpc-h-prototype (47b94c4) to 0d79252 (merge-base) diff Run configurationrun benchmark tpcdsCPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing andygrove/aqe-datafusion-tpc-h-prototype (47b94c4) to 0d79252 (merge-base) diff Run configurationrun benchmark clickbench_partitionedCPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
Adaptive query execution (AQE) is often read as Spark-style staged execution: materialize a stage, then re-plan the rest of the query. This PR shows that DataFusion can get the benefit without any of that, and without becoming a staged engine.
A hash join already waits for its entire build side before it probes anything. That pipeline breaker is already in every plan. Once the build side is complete, its exact size is known for free, and the join can still choose how to run its probe side. This PR does exactly that, and nothing more:
Today the choice between
CollectLeft(one shared hash table, probe side read as is) andPartitioned(both sides hash-repartitioned) is made at planning time, from estimates. Only joins whose build side is estimated under 4 MiB / 128K rows getCollectLeft. At TPC-H SF100 the estimates for filtered scans, joins and aggregates are rough, so almost every join is plannedPartitioned. Queries then hash-repartition hundreds of millions of probe rows to join against build sides that turn out to be tiny. For example, Q9 repartitions all 600Mlineitemrows to join 1.09M parts, and Q11 repartitions 80Mpartsupprows to join 1M suppliers.What changes are included in this PR?
New options, with the feature off by default:
datafusion.optimizer.enable_adaptive_hash_joinfalsedatafusion.optimizer.adaptive_hash_join_threshold_rowsCollectLeftdatafusion.optimizer.adaptive_hash_join_thresholddatafusion.optimizer.adaptive_hash_join_max_probe_ratioWhen it is on, the new
AdaptiveHashJoinphysical optimizer rule wraps eachPartitionedhash join whose probe side is hash-repartitioned in a newAdaptiveHashJoinExec. The rule runs after dynamic filters are pushed down, and does nothing when the option is off. When the operator starts, it collects the build side, as the join would anyway. Then:CollectLeftmode and reads the probe side as it is, without the repartition.Partitionedmode exactly as planned. The rest of the build side streams in behind the batches collected so far.So it never waits longer than the join itself would, and never holds more of the build side than the hash table will. Everything else stays as the optimizer made it. The wrapped join keeps its dynamic filter, which is already pushed into the probe-side scans, and the plan's properties are unchanged.
dfbench tpch --adaptiveturns the option on.Why not stages
I tried the Spark-style approach first: run each build side bottom-up as its own stage, then re-optimize the rest of the plan with exact statistics. It made Q18 2x slower.
DataFusion builds hash tables top-down along the probe side. In Q18, the semi join at the top finishes its build side first, and its dynamic filter then prunes the
ordersscan that feeds the build side of the join below it. Running that lower build side as an earlier, separate stage lost the filter, so its hash table was built over 150M rows instead of 85.7M. Keeping DataFusion's own execution order is not just less invasive, it is faster.Details
Left,LeftSemi,LeftAnti,LeftMarkandFull) share one bitmap across all probe threads inCollectLeftmode. They only switch for build sides within the existinghash_join_single_partition_threshold_rows(128K rows).Partitioned.Partitionedhash join waits for every partition to report its build side before it publishes its dynamic filter. Partitions that a parent never polls are therefore still created and dropped, so that they report as canceled.datafusion-protoyet.Results
All runs use the same binary with and without
--adaptive. The SF1000 and M3 numbers were measured with the earlier benchmarks-only version of this code, which makes the same decision for every join.TPC-H SF100 on a Ryzen 9 7950X3D (16 cores, 124 GB): total time 109.1 s → 95.9 s, 1.14x faster. The biggest gains are Q11 2.20x, Q8 1.62x, Q2 1.48x, Q9 1.45x, Q7 1.26x, Q17 1.18x, Q12 1.15x. No query is more than 2% slower.
Per-query times (median of 3)
TPC-H SF100 on an M3 Ultra (28 cores, 96 GB): 1.08x, with the same wins. Q21 (0.92x) and Q10 (0.93x) are slower there. Q21 switches a 1M-row supplier join, which pays off on the 7950X3D, whose 128 MB L3 holds the hash table, but not on the M3. A row threshold is a crude cost model.
TPC-H SF1000 on the 7950X3D: 1.08x overall. At this scale the data (about 380 GB) is three times the machine's memory, so most queries wait on the disk, and the clear wins are the joins that no longer repartition
lineitem: Q8, Q17, Q19 and Q2. Each query ran in its own process, in the order main, AQE, AQE, main. Main's first run is always the coldest, which flatters AQE; comparing only the second run of each side, which both follow a run of the same query, gives 1.04x. Q18 ran out of memory both with and without AQE.Per-query times (mean of 2 runs each)
Could a static threshold do the same? On TPC-H, mostly. In a separate run on the 7950X3D, raising the planning-time thresholds to about the same level (4M rows / 128 MiB) got 1.14x, about the same total as AQE (1.07x on the M3, against 1.08x for AQE). TPC-H's estimates for filtered dimension tables are good enough to find the same wins, and the static thresholds also win Q10 (1.09x), where AQE does not switch. But they switch some joins on estimates that turn out wrong: Q22 is 0.81x, Q13 0.90x and Q4 and Q15 0.96x, while the slowest query with AQE is Q4 at 0.97x. Deciding from the real build side gets the same gains without depending on the estimates being right.
What is the testing strategy for this PR?
datafusion/core/tests/physical_optimizer/adaptive_hash_join.rscovers five cases, each checking the results against the static plan:CollectLeft;Partitioned, and the rest of it, after the batches collected up to the threshold, still reaches the join;hash_join_single_partition_threshold_rows;adaptive_hash_join.sltchecks that the option is off by default and the plan when it is on, and that inner, semi, anti, left and full joins return the same results with it off, on, and on with every join keptPartitioned.Are there any user-facing changes?
Four new configuration options. The feature is off by default, so nothing changes unless it is turned on.