jayzhan211 commented on issue #24768:
URL: https://github.com/apache/datafusion/issues/24768#issuecomment-5463211988
> how would you like feedback on this proposal? Would you like it in the
attached google doc or on this issue?
@alamb Let's keep the discussion here in this issue.
Thanks — this is useful, and the second half of your reading is exactly
right. Let me correct the first half, because the difference matters for how
the rest of the design reads.
**(2) "add per-partition spilling to the Partitioned mode"** — yes, that's
the core of it. Each output partition owns a disjoint slice of keys and a
private hash table, so it can spill and recover on its own; that's what makes
this tractable without the cross-thread spill coordination DuckDB and Velox
need.
**(1) "add a dynamic transition from CollectLeft → Partitioned if the single
collect-left hash table is too large"** — not quite, and I should have made
this clearer. Mode selection stays a **plan-time** decision: when
`enable_hash_join_spilling=true`, the planner steers joins to `Partitioned`
unless the build side is provably below the CollectLeft threshold (T8). There
is no runtime CollectLeft→Partitioned switch — a build side that lands in
CollectLeft anyway (statistics misestimate) keeps today's failure behavior, and
coordinated CollectLeft spilling is the first follow-up issue.
The transition that *is* dynamic is a different one, one level down —
**inside a single partition**, and it's the zero-regression mechanism:
```text
Phase A (= today's code, exactly) Phase B (only after memory
pressure)
batches.push(batch) ──try_grow──► hash + split buffered batches
into 16 buckets,
no hashing, no routing, fails destage the largest to disk,
route new batches
no files from here on
```
So "adaptive" in this proposal means *buffer → bucketed+spilling within a
partition*, never *CollectLeft → Partitioned across the plan*. With no memory
limit configured, Phase B is unreachable and the operator is unchanged. The
solution sketch in the issue body says this less directly than it should have;
take the wording below as the intended reading.
---
On the three suggestions — all fair, all my fault for writing the symptoms
section in the abstract:
**Concrete reproducer.** Added below; it's self-contained, and I've
re-verified it end to end on the 54.0.0 tag. One pass produces every row of the
table.
**The options, labelled.** Rewritten table — the middle column now says what
each row *is* (a configuration a user can actually choose today, vs. a control
that isn't a way to run the join at all), and the two ceiling rows are
collapsed into one as you suggested:
| # | role | query / setting | result |
|---|---|---|---|
| 1 | **default settings** — the planner picks `HashJoinExec` | `count(*)
FROM t_probe p JOIN t_build b ON p.k = b.k` | ❌ fails in 0.2 s — `Failed to
allocate additional 95.4 MB for HashJoinInput[3] with 38.2 MB already allocated
… fair(pool_size: 300.0 MB)` |
| 2 | **the only workaround** — `prefer_hash_join=false`, so the planner
always picks `SortMergeJoinExec`, whose sorts can spill | same query | ✅ 3.7 s
(`SortExec` spills `spill_count=16`, 155 MB; the join itself spills nothing) |
| 3 | **where the ceiling is** — same query, build input shrunk by a filter
| `JOIN (SELECT * FROM t_build WHERE k <= N) b` | ✅ N = 10M: 0.7 s · ❌ N = 12M:
fails at the hash-table build (`+57.2 MB` with `22.9 MB` held) |
| 4 | **what the workaround costs when it isn't needed** — row 3's *fitting*
join forced through SMJ | N = 10M, `prefer_hash_join=false` | ✅ 2.5 s — **3.5×
tax on a join that would have fit** |
| 5 | **control, not a way to run the join** — shows the budget itself is
workable | `count(DISTINCT payload) FROM t_build` | ✅ 11.1 s — hash
*aggregation* spills 1079 MB through the same 300 MB pool and completes |
**"Can't we get it to run by switching the inputs?"** No, and you're right
that the table should say so outright. When statistics are available the
planner already places the smaller input on the build side (the `JoinSelection`
swap), so the failure mode is precisely the case where *even the smaller side*
exceeds the budget. In this reproducer both tables are identical, which makes
that concrete: there is no better side to choose.
One thing I can now state more sharply than "the fatal allocation is the
hash-table build". Since the query is `count(*)` on `k` alone, projection
pushdown drops `payload` — the build side is one `Int64` column, ~8 B/row. The
hash map is sized by `estimate_memory_size::<(u32, u64)>` ≈ **19.4 B/row, ~2.4×
the data it indexes**, and it is requested as a *single* `try_grow` after every
build batch has already been admitted:
| | rows/partition | batches held | hash-map grow | total after grow |
|---|---|---|---|---|
| 20M / 4 partitions | 5.0M | 38.2 MB | **95.4 MB** | 133.6 MB |
| 12M / 4 partitions | 3.0M | 22.9 MB | **57.2 MB** | 80.2 MB |
Every consumer in the failure dump decomposes into those two columns (the
133.6 MB consumer is a partition that already finished its map; the ~38 MB ones
are still holding batches when one of them dies making the same grow). That's
the concrete argument for the table-headroom reservation in T9 — per-batch
backpressure alone cannot avoid this allocation.
<details>
<summary>reproducer — <code>datafusion-cli</code>, verified on the 54.0.0
tag</summary>
Generate once, with no memory flag (building the 20M-row file under a 300 MB
pool is its own fight and not the thing under test):
```sql
-- gen.sql: datafusion-cli -f gen.sql
COPY (SELECT v AS k, concat('payload-', v, '-', repeat('x', 24)) AS payload
FROM (SELECT unnest(generate_series(1, 20000000)) AS v))
TO 't.parquet';
```
`--top-memory-consumers 8` raises the failure dump above its default of 3
entries so that all four join partitions are listed — at the default one is cut
off, and for the N = 12M case the one cut off is the partition that actually
fails:
```sql
-- repro.sql: datafusion-cli -m 300M --mem-pool-type fair
--top-memory-consumers 8 -f repro.sql
SET datafusion.execution.target_partitions = 4;
CREATE EXTERNAL TABLE t_build STORED AS PARQUET LOCATION 't.parquet';
CREATE EXTERNAL TABLE t_probe STORED AS PARQUET LOCATION 't.parquet';
-- row 1: default settings
SELECT count(*) FROM t_probe p JOIN t_build b ON p.k = b.k;
-- row 2: the workaround
SET datafusion.optimizer.prefer_hash_join = false;
SELECT count(*) FROM t_probe p JOIN t_build b ON p.k = b.k;
-- row 4: the tax — the workaround on a join that fits
SELECT count(*) FROM t_probe p
JOIN (SELECT * FROM t_build WHERE k <= 10000000) b ON p.k = b.k;
-- row 3: the ceiling
SET datafusion.optimizer.prefer_hash_join = true;
SELECT count(*) FROM t_probe p
JOIN (SELECT * FROM t_build WHERE k <= 10000000) b ON p.k = b.k; -- ✅
SELECT count(*) FROM t_probe p
JOIN (SELECT * FROM t_build WHERE k <= 12000000) b ON p.k = b.k; -- ❌
-- row 5: control
SELECT count(DISTINCT payload) FROM t_build;
```
The CLI reports the two `Resources exhausted` failures inline and continues,
so a single pass produces all five rows and exits 0. The `95.4 MB` / `38.2 MB`
figures are stable run to run; the trailing `… remain available` value and the
failing partition index vary with how many sibling partitions had reached their
hash-map grow.
</details>
Restating the solution-sketch opening in the words it should have used,
since that paragraph is where the ambiguity came from:
> Mode selection is a **plan-time** decision: with
`enable_hash_join_spilling=true` the planner steers joins to `Partitioned`
unless the build side is provably below the CollectLeft threshold — so the
joins that *can* hit the limit are exactly the joins that *can* spill. The
adaptivity is **within** a partition, not across the plan. Each partition
starts on today's exact code path — append batches to a list, no hashing, no
routing, no files — and only when a `try_grow` is first refused does it split
what it has buffered into 16 buckets, destage the largest to disk, and route
subsequent batches. With no memory limit configured, that transition is
unreachable and the operator is bit-for-bit unchanged.
I'm leaving the issue body as it stands and treating this comment as the
clarification of record, so the thread reads in order — the labelled table, the
reproducer, and the paragraph above supersede the corresponding parts of the
description. The reproducer also becomes T0's benchmark, so the matrix stays
reproducible as the epic progresses.
--
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]
