andygrove opened a new pull request, #26328:
URL: https://github.com/apache/datafusion/pull/26328
## Which issue does this PR close?
- Related to #23194 (this does not close it).
## Rationale for this change
DataFusion picks the mode of each hash join at planning time from estimated
statistics. It only collects the build side into a single hash table
(`CollectLeft`) when the estimate is under
`hash_join_single_partition_threshold` (4 MiB / 128K rows). Otherwise it
hash-repartitions both inputs (`Partitioned`). At TPC-H SF100 almost every join
comes out `Partitioned`, because estimates for filtered scans, joins and
aggregates are rough. Many of those joins have tiny build sides, so the query
hash-repartitions hundreds of millions of probe rows for nothing. For example:
- Q9 repartitions all 600M `lineitem` rows to join 1.09M parts.
- Q17 repartitions all 600M `lineitem` rows for a 20K-row part filter.
- Q11 repartitions 80M `partsupp` rows to join 1M suppliers.
This draft is a prototype to show what adaptive query execution (AQE) could
gain here, and to feed the design discussion in #23194. It is not proposed for
merging as is.
## What changes are included in this PR?
Everything is in the benchmarks crate and uses only public APIs:
- `benchmarks/src/util/adaptive.rs`: `make_adaptive(plan)` runs after normal
physical planning. It wraps each `Partitioned` hash join whose probe side is
hash-repartitioned in an `AdaptiveHashJoinExec`. On first execution that
operator collects the build side, which a hash join must do anyway, up to a
threshold:
- If the build side fits, it runs the same join as `CollectLeft`, probing
the probe input as it is, without the repartition.
- If it does not fit, it stops collecting and runs `Partitioned` as the
static plan would. The rest of the build side streams in behind the batches
already collected.
- `dfbench tpch --adaptive`, plus `--adaptive-broadcast-rows` (default 4M),
`--adaptive-broadcast-rows-tracking` (default 128K) and
`--adaptive-broadcast-probe-ratio` (default 0.05).
The plan is otherwise unchanged:
- **Execution order** is unchanged, so a parent join's dynamic filter still
reaches the build sides below it.
- **Dynamic filter:** the wrapped join keeps the one the optimizer already
pushed into the probe scans.
- **Plan properties** are unchanged, so ancestors are not re-derived and
their TopK filters stay wired.
Points that turned out to matter:
1. **Re-planning from materialized stages** is the Spark-style approach. I
tried it first: run build sides bottom-up as stages, then re-optimize with
exact statistics. It made Q18 2x slower. DataFusion builds top-down along the
probe spine, so the semi-join's dynamic filter (with its `KeyRangeBitmap`)
prunes the build side of the join below it. A standalone stage loses that
filter and builds its hash table on all 150M orders instead of 85.7M.
2. **Joins that track matched build rows** (`Left`, `LeftSemi`, `LeftAnti`,
`LeftMark`, `Full`) share one mutex-protected bitmap across probe threads in
`CollectLeft`, which made Q22's anti join 20% slower. They now only broadcast
build sides of up to 128K rows.
3. **Waiting for the whole build side** before deciding cost 6-11% on joins
that end up partitioned anyway, so collection stops as soon as the threshold is
exceeded.
4. **Ancestors that rely on hash partitioning:** where an ancestor relies on
the join's output being hash partitioned, a broadcast join repartitions its
usually much smaller output instead of its probe input (for example, the inner
join under Q17's semi join).
5. **Dynamic filter barrier:** a `Partitioned` hash join waits for every
partition to report its build side before it publishes its dynamic filter.
Partitions a parent never polls, for example because the parent's own build
partition is empty, must still be created and dropped so that they report as
canceled. Otherwise the query hangs.
## Results
TPC-H SF100 parquet, `release-nonlto` build, same binary with and without
`--adaptive`, median of 3 runs. The machine is a Ryzen 9 7950X3D (16 cores / 32
threads, 124 GB):
| Query | main (ms) | AQE (ms) | Speedup |
|---|---:|---:|---:|
| Q1 | 2,499 | 2,458 | 1.02x |
| Q2 | 1,421 | 963 | 1.48x |
| Q3 | 3,390 | 3,334 | 1.02x |
| Q4 | 1,217 | 1,250 | 0.97x |
| Q5 | 6,252 | 5,816 | 1.07x |
| Q6 | 1,154 | 1,174 | 0.98x |
| Q7 | 10,510 | 8,289 | 1.27x |
| Q8 | 6,484 | 4,049 | 1.60x |
| Q9 | 11,952 | 8,169 | 1.46x |
| Q10 | 4,942 | 5,039 | 0.98x |
| Q11 | 1,294 | 627 | 2.07x |
| Q12 | 1,500 | 1,389 | 1.08x |
| Q13 | 3,420 | 3,425 | 1.00x |
| Q14 | 1,932 | 1,916 | 1.01x |
| Q15 | 3,938 | 3,983 | 0.99x |
| Q16 | 844 | 775 | 1.09x |
| Q17 | 11,625 | 9,853 | 1.18x |
| Q18 | 15,838 | 15,886 | 1.00x |
| Q19 | 2,147 | 2,036 | 1.05x |
| Q20 | 3,836 | 3,699 | 1.04x |
| Q21 | 11,744 | 10,721 | 1.10x |
| Q22 | 533 | 539 | 0.99x |
| **Total** | **108,472** | **95,390** | **1.14x** |
All the gains come from switching to `CollectLeft` once the real build-side
size is known. `dfbench tpch --debug --adaptive` prints each join's decision.
On an M3 Ultra (28 cores, 96 GB, median of 6) the total is 1.08x, with the same
wins. There, Q21 is 0.92x and Q10 0.93x. Q21 loses because it broadcasts the
1M-row `supplier` table against a narrow 379M-row probe: on that machine
repartitioning is cheap and a 1M-entry hash table misses L2, while the
7950X3D's 128 MB L3 holds it. With every join forced to `Partitioned`, Q21
matches main, so the adaptive machinery itself costs nothing there. A real
implementation would need a cost model that knows probe width and cache size,
not a row threshold.
For comparison, main with the static thresholds simply raised to about the
same level (4M rows / 128 MiB) reaches 1.07x on the M3 Ultra, against AQE's
1.08x. It gets the same big wins, because TPC-H's estimates for filtered
dimension tables are good enough. It also broadcasts on estimates where it
should not: Q22 0.83x, Q21 0.88x, Q13 0.94x and Q18 0.96x, against AQE's 1.02x,
0.92x, 0.98x and 0.99x. It wins Q10 (1.09x) where AQE does not.
SF1000 runs are in progress and will be added here.
## What is the testing strategy for this PR?
- Unit tests in `adaptive.rs` cover three cases:
- a small build side is collected, with results equal to the static plan;
- a build side over the threshold stays partitioned after the early stop;
- partitions a parent never polls do not hang the query. This test hangs
and fails at its timeout without the fix.
- All 22 TPC-H queries return the same results as main at SF1 under five
settings: defaults, everything partitioned, everything collected, batch size
97, and 3 partitions. They also match at SF100 on two different datasets.
## Are there any user-facing changes?
No. The only additions are the new `dfbench tpch` flags.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
