apache / apache/datafusion-ballista
Hash-join build side OOMs on TPC-H Q18 at SF1000 and does not shrink with target_partitions
- Dominant language
- Rust
- Stars
- 2.1k
- Forks
- 320
- Avg merge
- 1d 22h
- Merged PRs (30d)
- 66
Description
## 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 --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.
Contributor guide
Research direction
Reproduce TPC-H Q18 with the benchmark command using 32 and 64 partitions, then inspect the physical plan to distinguish a CollectLeft build side from hash-partitioning skew. Review the related issues #1905, #1922, and #1643, and note the failure path at benchmarks/src/bin/tpch.rs:501. Done means the query no longer exhausts the memory pool, or oversized build sides spill or fall back as described, while the benchmark failure is handled appropriately.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- databases, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 46/100