wirybeaver opened a new pull request, #2714:
URL: https://github.com/apache/datafusion-ballista/pull/2714
## Motivation
A hot shuffle bucket can serialize most of a partitioned join on one task.
This ports the skew-join AQE step: split its mapper outputs and replicate the
matching bucket so the same join runs across smaller parallel tasks.
## Modification
The opt-in rule runs after coalescing, computes each side's skew threshold
once, and attaches paired mapper ranges to both input exchanges. Readers
declare unknown partitioning, and stage completion uses the executable output
width so expanded results are retained.
Skew detection averages the middle pair for even partition counts, avoiding
missed splits caused by using the upper middle value alone.
```text
resolved shuffle statistics
-> coalesce inputs when applicable
-> detect skew and split mapper ranges
-> pair shards / replicate the matching bucket
-> aligned shuffle readers -> parallel join tasks
-> repartition or collect every output partition
```
The first version supports one partitioned join per stage. It excludes paths
that need complete groups, fixed row limits, or ordered shuffle reads; partial
aggregation before a reshuffle remains eligible. Dynamic-filter pushdown must
be disabled. Configuration and its generated reference are included.
## Benchmarking Result
Both are warm-cache experiments on one Apple M5 host (10 logical CPUs, 24
GiB RAM), with two executors (4 slots and a 4 GiB memory pool each), 32 hash
partitions, and one partition per task. In the split-only A/B comparisons, AQE
is enabled; coalescing, dynamic-filter pushdown, and broadcast are disabled.
The join algorithm is fixed within each comparison. **A disables split; B
enables it.** Latency changes use `mean(B)/mean(A)-1` and
`median(B)/median(A)-1`; negative values mean lower latency.
### Synthetic dataset
**Workload:** 16M fact rows in 32 Parquet files; 90% share one hot key. A
uniform control spreads 16M rows across 100,004 keys. The dimension has eight
rows per key. The query joins on the key and groups counts and weighted sums by
`key % 32`.
```sql
SELECT d.bucket,
COUNT(*) AS matched_rows,
SUM(f.measure * d.weight) AS weighted_sum
FROM fact f
JOIN dim d ON f.join_key = d.join_key
GROUP BY d.bucket
ORDER BY d.bucket;
```
**Parameters/process:** skew factor **5**, threshold **32 MiB**, advisory
size **16 MiB**. After one warmup per arm, run **ABBAAB**, with three queries
per visit. Every answer matches an independent row-ID oracle. Only the skewed B
cases split, increasing join tasks **32 → 47**. The sort-merge baseline's hot
sort spills; splitting avoids that spill.
#### Per-visit results
| Visit | Arm | Skew SMJ (s / M rows/s) | Skew hash (s / M rows/s) | Uniform
SMJ (s / M rows/s) |
| --- | --- | ---: | ---: | ---: |
| 1 | A | 1.726 / 9.268 | 1.049 / 15.253 | 0.502 / 31.876 |
| 2 | B | 0.426 / 37.601 | 0.333 / 48.195 | 0.509 / 31.471 |
| 3 | B | 0.426 / 37.568 | 0.325 / 49.308 | 0.521 / 30.696 |
| 4 | A | 1.754 / 9.122 | 1.038 / 15.410 | 0.506 / 31.624 |
| 5 | A | 1.756 / 9.114 | 1.034 / 15.474 | 0.517 / 30.989 |
| 6 | B | 0.446 / 35.934 | 0.318 / 50.370 | 0.515 / 31.080 |
| Workload | Mean latency A → B (s) | Mean latency change | Median latency
change |
| --- | ---: | ---: | ---: |
| Skew, sort-merge | 1.745 → 0.432 | -75.22% | -75.71% |
| Skew, hash | 1.040 → 0.325 | -68.74% | -68.73% |
| Uniform, sort-merge | 0.508 → 0.515 | +1.36% | +1.84% |
Skewed-join latency improves **4.04× for sort-merge** and **3.20× for
hash**; the uniform control has a small slowdown. These synthetic measurements
predate the even-partition median correction; recorded revisions, scripts, and
evidence are in the [synthetic benchmark
gist](https://gist.github.com/wirybeaver/17d83998b6d10df852f15b637a196c86).
### TPC-DS SF100 — custom NULL-hotkey query
**Workload:** unmodified reference-generated SF100 data (24 tables, 35.7 GB
Parquet). A custom NULL-safe join matches `store_sales` to return groups from
`store_returns`, then aggregates counts and weighted units by date bucket. The
27.7M-row build side and NULL hot key produce 10.67B logical matches. [Full SQL
and
reproduction](https://gist.github.com/wirybeaver/8a8848c997d8e7f6b2f8dafa5ac135ad#file-query-sql).
```sql
WITH return_buckets AS (
SELECT sr_customer_sk, sr_return_amt, COUNT(*) AS return_events
FROM store_returns
WHERE sr_customer_sk IS NOT NULL
OR (sr_customer_sk IS NULL AND sr_return_amt BETWEEN 0 AND 5.11)
GROUP BY sr_customer_sk, sr_return_amt
)
SELECT s.ss_sold_date_sk % 8 AS bucket,
COUNT(*) AS matched_rows,
SUM(CAST(s.ss_quantity AS BIGINT) * r.return_events) AS weighted_units
FROM store_sales s
JOIN return_buckets r
ON s.ss_customer_sk IS NOT DISTINCT FROM r.sr_customer_sk
GROUP BY s.ss_sold_date_sk % 8
ORDER BY bucket NULLS FIRST;
```
**Parameters/process:** partitioned hash join with the ordinary **64 MiB**
hash-build limit; skew factor **1.5**, threshold **8 MiB**, advisory size **8
MiB**, small-partition factor **0.2**. Freeze the SQL/parameters, warm each arm
once, then run **ABBA**. All answers exactly match an independent factored
DuckDB oracle. Split expands the join stage **32 → 62** logical tasks.
#### Per-visit results
| Visit | Arm | Query latency (s) | Join tasks | Longest join task (s) |
| --- | --- | ---: | ---: | ---: |
| 1 | A | 82.368 | 32 | 79.924 |
| 2 | B | 28.051 | 62 | 4.637 |
| 3 | B | 28.622 | 62 | 7.213 |
| 4 | A | 81.940 | 32 | 79.440 |
#### Join-strategy comparison
| Configuration | Measurement | Query latency (s) |
| --- | --- | ---: |
| Fixed partitioned hash, split disabled | Mean of 2 ABBA visits | 82.154 |
| Automatic join selection, split disabled (broadcast selected) | Single
reference run | 40.199 |
| Fixed partitioned hash, split enabled | Mean of 2 ABBA visits | **28.337**
|
Split reduces mean latency by **65.51% (2.90× speedup)** against the fixed
partitioned baseline; mean and median latency changes are both **-65.51%**. It
also beats the automatic-join reference by **29.51% (1.42× speedup)**. The
automatic-join result is a single measurement, separate from the repeated
split-only ABBA comparison. These are warm, single-host custom-query results.
Scripts and recorded results are in the [SF100 reproduction
gist](https://gist.github.com/wirybeaver/8a8848c997d8e7f6b2f8dafa5ac135ad).
`benchmarks/` remains unchanged against the PR base.
--
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]
