andygrove opened a new issue, #2025:
URL: https://github.com/apache/datafusion-ballista/issues/2025
## Describe the bug
With `datafusion.optimizer.prefer_hash_join=true`, TPC-H Q18 at SF1000 fails
with `ResourcesExhausted` in a hash-join build side, and **the per-task memory
requirement does not shrink when `target_partitions` is increased**. That
invariance is the interesting part: it means the memory is not sitting in a
key-partitioned build side, and repartitioning cannot fix it.
This matters because hash joins are otherwise a large win at this scale (see
below), and this OOM is the one thing blocking them from completing the suite.
## Evidence that it is invariant to `target_partitions`
Same cluster, same query, identical apart from `--partitions`. Per-task pool
is `--memory-pool-size / concurrent_tasks` = 28GB / 8 = **3.3 GB**:
| `target_partitions` | Failure |
|---|---|
| 32 | `Failed to allocate additional 464.0 KB for HashJoinInput[12] with
3.3 GB already allocated ... pool_size: 3.3 GB` |
| 64 | `Failed to allocate additional 544.0 MB for HashJoinInput[0] with 3.3
GB already allocated ... pool_size: 3.3 GB` |
Doubling the partition count did **not** halve the per-task build side — it
still saturates the same 3.3 GB, and in fact asks for *more* headroom. A build
side that was hash-partitioned on the join key should have roughly halved.
Raising memory instead also fails to fix it. At `--memory-pool-size=48GB` (→
5.6 GB per task slot):
```
Failed to allocate additional 1088.0 MB for HashJoinInput[4] with 4.8 GB
already
allocated for this reservation - 792.2 MB remain available for the total
memory
pool: fair(pool_size: 5.6 GB)
```
So the build side wants ~6 GB **per task**, and that figure is independent
of how finely the input is partitioned. This is the signature of a
`CollectLeft` (collected/replicated) build side — which every task materializes
in full — rather than a partitioned one. Severe skew on the join key is the
other candidate.
Sort-merge joins are unaffected: the same query with
`prefer_hash_join=false` completes in **587.9s**, because SMJ streams and
spills instead of materializing a build side in memory.
## Why this is worth fixing
Same cluster (2 executors × 8 cores / 56Gi), SF1000, single iteration, Q1–17
(all that hash joins survive before this OOM):
| Config | Q1–17 |
|---|---:|
| Ballista, `prefer_hash_join=true` | **2683.6s** |
| Spark 3.5 (AQE on) | 3441.4s |
| Ballista, `prefer_hash_join=false` (default) | 4039.1s |
Hash joins make Ballista **1.28× faster than Spark**; the SMJ default makes
it **1.17× slower**. The join strategy swings Ballista by ~1.5× at SF1000, and
this OOM is what stands between the fast config and a completed suite.
## To Reproduce
Cluster: 2 executors, 8 CPU / 32Gi each, `--memory-pool-size=28GB` (→ 3.3 GB
per task slot), 1 scheduler. TPC-H SF1000 parquet on node-local disk. Ballista
`54.0.0-rc2`.
```sh
cargo run --release --bin tpch -- benchmark ballista \
--host <scheduler> --port 50050 \
--query 18 --path /mnt/bigdata/tpch/sf1000 --format parquet \
--partitions 32 --iterations 1 \
-c datafusion.optimizer.prefer_hash_join=true \
-c datafusion.optimizer.enable_dynamic_filter_pushdown=false
```
Repeat with `--partitions 64` to observe that the per-task requirement is
unchanged.
## Expected behavior
Either the hash-join build side is partitioned so that raising
`target_partitions` reduces per-task memory, or an oversized `CollectLeft`
build side spills / falls back to a partitioned join rather than exhausting the
pool.
## Additional context
- The benchmark client `unwrap()`s a failed query into a panic
(`benchmarks/src/bin/tpch.rs:501`), so a single failing query aborts the whole
suite rather than being recorded and skipped. That is why this initially
presented as "the suite dies at Q18". Arguably worth fixing independently.
- Related: #1905 (broadcast lowering still hash-repartitions build and probe
sides of a `CollectLeft` join), #1922, #1643 (skew from hash partitioning on
low-cardinality keys).
- I have not dumped the physical plan to confirm `CollectLeft` — that is the
obvious next step and would settle CollectLeft vs skew.
--
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]
