[EPIC] Spilling Hash Join — run any join in a bounded memory budget

[jayzhan211]

[EPIC] Spilling Hash Join — run any join in a bounded memory budget

TL;DR. When the build side of a hash join exceeds the memory budget, DataFusion fails the query. This epic makes HashJoinExec spill to disk instead — behind a default-off flag (enable_hash_join_spilling), with zero change to joins that fit in memory and zero change for anyone running without a memory limit.

📄 Full design doc (step-by-step mechanics, join-type × mechanism matrix, figures): Google Doc
(Consolidates #1599, #12952, and the design direction of #17267 into one epic with current status and a PR-sized plan.)

Two terms used throughout. A hash join loads one input entirely into an in-memory hash table (the build side — DataFusion uses the left input) and streams the other input past it (the probe side). Also: DataFusion already splits query execution into target_partitions output partitions; this design further splits each join's build data into 16 buckets — "partition" below always means the former, "bucket" always the latter.

What is going on (symptoms)

On DataFusion 54.0.0, two 20M-row Parquet tables (~1.1 GB raw each), 300 MB FairSpillPool, target_partitions=4:

query (same data, same 300 MB budget)	result
hash join t_probe ⋈ t_build ON k	❌ fails in 0.2 s — Resources exhausted: Failed to allocate additional 95.4 MB for HashJoinInput[3] with 38.2 MB already allocated for this reservation - 51.9 MB remain available for the total memory pool: fair(pool_size: 300.0 MB)
same join via SMJ (prefer_hash_join=false)	✅ 3.7 s — the SortExecs spill (spill_count=16, 155 MB)
hash join, build side shrunk to 10M rows (fits)	✅ 0.7 s
that same fitting join forced through SMJ	✅ 2.5 s — the workaround's tax: 3.5× on a join that fits
count(DISTINCT payload) hash aggregate	✅ 11.1 s (spills)
hash join, build side 12M rows	❌ the fatal allocation is the hash-table build (+57.2 MB with 22.9 MB held)

So today users choose between two bad options: keep prefer_hash_join=true and lose big joins outright, or set it to false and pay ~3.5× on every join that would have fit. Sort, aggregation, sort-merge join, repartition, and (since DF 54) nested-loop join all spill; hash join — the default join — is the last major operator that fails instead. It also makes memory-constrained CI impossible: TPC-H/TPC-DS can't run under a stepped budget because plans die at the first big join.

What is causing the problem

collect_left_input() (hash_join/exec.rs)
    batches: Vec<RecordBatch>      ◄── every build batch appended; reservation.try_grow(batch)
    ...                                 per batch — on failure the query dies (no reaction path)
    concat_batches(...)            ◄── then ONE contiguous batch
    JoinHashMapU32/U64             ◄── then ONE hash table sized for all rows
                                        (the +57 MB allocation that actually fails in practice)

The build phase has no reaction to memory pressure, and the consumer isn't registered with_can_spill(true), so FairSpillPool can't even treat it fairly. Meanwhile the surrounding infrastructure is ready: SpillManager (IPC spill files, compression, disk quotas, metrics), SpillPool, ReplayableStreamSource, and an in-tree precedent of a join reacting to ResourcesExhausted by switching into a spill mode (DF 54's NLJ). Only the hash-join algorithm on top is missing.

The high-level solution sketch

First, the two modes DataFusion runs hash joins in, because the design leans on one of them. CollectLeft (small builds): build the table once, share it read-only across all probe streams. Partitioned (large builds — the mode that hits memory limits): both inputs are hash-redistributed across the target_partitions output partitions, and each output partition builds and probes its own private table over its own disjoint slice of keys, in a single thread. That privacy is the key insight: each output partition can spill and recover independently. The cross-thread coordination that DuckDB and Velox spend most of their spill machinery on — and the main blocker named in #17267 — does not arise, because DataFusion's plan already did the dividing. Fairness between partitions is the memory pool's existing job.

Within one output partition, the join becomes a hybrid hash join — "hybrid" meaning it keeps as much of the build side in memory as fits and spills only the remainder (vs. Grace-style, which partitions everything to disk up front):

BUILD  route rows into 16 buckets by high bits of the join hash
         ┌────┬────┬────┬────┬───┐    on memory pressure: destage the largest
         │ B0 │ B1 │ B2 │ B3 │ … │    bucket (move its buffered rows to disk)
         └────┴────┴────┴────┴───┘    + build a Bloom filter of that bucket's hashes
          mem   mem  disk  mem
       end of input: ONE JoinHashMap over the resident buckets

PROBE  hash once, route: resident rows → probe the map now (unchanged code path)
       rows for a spilled bucket → Bloom test:
           negative → resolved immediately (provably matches nothing:
                      drop, or emit null-padded for Right/Full)
           positive → append to probe.Bi.spill (only rows that might match)

CLEANUP  for each spilled pair (build.Bi, probe.Bi):
           restore build.Bi → fits? build its map, stream probe.Bi through
                              the normal probe path
                            → too big? re-split with the next window of hash
                              bits (≤ 4 levels; capacity ≈ share × 16^level)
                            → won't shrink (all-equal keys)? chunked build +
                              probe replay (the DF 54 NLJ pattern)
           role reversal: if probe.Bi ≪ build.Bi, build the map from the
           probe file instead and stream the build file through it

What runs before the first spill — exactly today's code. This is the zero-regression guarantee, and it is structural, not aspirational: while memory suffices, batches append to a single list — no routing, no hashing, no extra copies, no files. The first time try_grow fails (or an optional watermark trips), the stream computes the join hashes for the batches buffered so far, splits them into 16 buckets, destages the largest, and only from that moment routes incoming batches. With no memory limit configured (the default), that transition is unreachable and the operator is bit-for-bit unchanged.

Disk footprint. Only spilled buckets create files — one build + one probe file each — so a worst-case level is 32 files per output partition (128 at target_partitions=4; a few thousand at most on a 32-core machine with everything spilling). A pair's files are deleted as soon as it is processed, recursion drains a parent before creating children, and total bytes are already bounded by DiskManager's max_temp_directory_size.

Scope decision: CollectLeft mode

The title says any join; phase 1 implements spilling for Partitioned mode only — handled in the open rather than buried in non-goals: when enable_hash_join_spilling=true, the physical planner steers joins to Partitioned unless the build side is provably below the CollectLeft threshold (part of T8). With the flag on, the joins that can hit the limit are exactly the joins that can spill. A statistics misestimate can still land an oversized build in CollectLeft; that keeps today's failure behavior, is called out in the config docs, and coordinated CollectLeft spilling is the first follow-up issue. Relatedly, this epic makes no change to prefer_hash_join — but once spilling is default-on, prefer_hash_join=false stops being the memory-safety workaround (the symptoms table shows why you don't want it to be).

How other engines solve it

Engine	Approach	What this proposal borrows
DuckDB v1.2 (PR #4189, "Saving Private Hash Join", VLDB '25)	Adaptive radix-partitioned hash join: keep as many partitions resident as fit; repartition loop for the rest	Adaptivity (pay nothing when memory suffices); destage-largest; verify by re-running the whole suite with spilling forced (force_external)
Velox (spilling doc)	Builders coordinate to spill the same partition set; recursion advances a hash-bit window	Bit-window recursion — one hash, never re-hashed across levels; explicit depth cap
ClickHouse (grace_hash)	Grace hash join: bucket 0 in memory, all other buckets spill both sides up front	Sequential bucket processing — plus two warnings: it is not adaptive, and it shipped INNER/LEFT-only for years (join-type completeness must be day-one)
Spark (SPARK-32634)	Shuffled-hash join falls back to sort-based processing under pressure	Confirms the demand; we reject sort-fallback so a hash join stays a hash join
Trino (spill docs)	Revocable memory; a subset of build partitions spills with matching probe rows	The hybrid "spill some, keep the rest hot" discipline

No engine that solved this well made spilling a plan-time decision, and none used sort-fallback as the primary design — runtime-adaptive partitioned hybrid hash join is the consensus.

Alternatives considered

Alternative	Why not
Sort-based fallback (Spark-style)	Gives up O(1) probes; SMJ already exists as the manual fallback — and costs 3.5× on fitting joins (measured above)
Grace up-front (ClickHouse-style)	Pays partitioning + spill even when memory would have sufficed — violates zero-regression
Whole-build spill + probe replay (NLJ-style) as primary	Replay cost multiplies with the build/memory ratio; kept only as the skew fallback where re-splitting provably cannot help
Symmetric / early probing (start probing before build completes)	Must retain probe rows for late-arriving matches ⇒ buffers both sides — strictly worse memory; also forfeits build-side dynamic-filter pushdown
Probe re-scan instead of probe spill	The probe child is an arbitrary subtree — re-running it once per restore group is unbounded, and unsound if non-deterministic; kept as an opt-in follow-up for re-scannable leaf scans

Deeper treatment of the three closest calls (sparse-match economics via Bloom/role-reversal, why probe starts after build, spill vs. re-scan) is in the design doc §6.11–6.13.

What it costs (rule of thumb)

• Nothing spilled → nothing paid. Same code path as today; and with no memory limit configured, the spill machinery is structurally unreachable.
• Spilling: pay only for what didn't fit. Every spilled build byte is written once and read back once (≈ 2× the spilled bytes); probe-side spill is only the rows that might match a spilled bucket — with the Bloom filters, close to the true match volume. CPU stays linear; peak memory stays under the budget by construction. (Spilled rows carry their 8-byte join hash, ~10–15% extra spill volume, so restore/routing/Bloom never re-hash.)
• Capacity: ×16 per recursion level. A 75 MB share covers a multi-TB build side within 3–4 levels; all-equal-key skew falls back to chunked build + probe replay — slower but unbounded (an equal-key join's output is inherently quadratic anyway).
• Versus today's workaround: SMJ sorts and spills both complete sides no matter how close the build was to fitting; hybrid hash pays proportionally to the miss — ~3 GB of sequential spill traffic in the experiment above instead of query failure, and zero when it fits.

Why this is hard to fix (and how each part is handled)

• Join-type completeness is where the bodies are buried. Unmatched-build emission needs per-bucket visited bitmaps; Right/Full need Bloom-negative early emission; null-aware anti needs build-side null knowledge over all rows including spilled ones. ClickHouse shipping INNER/LEFT-only for years is the cautionary tale. The full join-type × mechanism table is in the design doc (§6.6); the test matrix is this epic's definition of done, not a fast-follow.
• The fatal allocation is the hash table, not the batches (measured above) — so the design reserves explicit table-construction headroom (hash_join_spill_reservation_bytes, the sort_spill_reservation_bytes lesson).
• Ordering property. HashJoinExec currently advertises probe-side order preservation for Inner/Right joins; cleanup emission breaks that, so with the flag on the operator stops advertising it — which can make the planner insert an explicit sort downstream even for queries that never spill. That is why the property change is tied to the flag (off ⇒ zero plan changes), T8 includes a TPC-H plan-diff audit with the flag on, and the config docs call it out.
• Dynamic-filter interplay, precisely. Dynamic filters are installed at plan time but populated once, at build completion — which is also exactly when we know what spilled, so the decision is made at a single well-defined moment, never mid-scan. Min/max bounds and InList filters are unaffected (accumulated over all rows, spilled included); only the full-map membership variant is unavailable when something spilled, and a per-bucket Bloom membership filter is its natural follow-up (T7 builds the Blooms anyway). Worst case: queries that previously failed outright lose one pushdown variant.
• Batch-splitting cost (PROPOSAL Hash Join Spilling Proposal #17267's "dual take") — only paid after the first destage, and measured by a dedicated benchmark PR (T2) before the core lands.

Past attempts and prior work

When	What	Status	Takeaway
2022	#1599 — memory-limited / externalized joins	open	earliest ask; joins named the last unspillable operators
2024	#12952 — add spilling support for HashJoin	open	recurring demand, no design attached
2024–25	#9359 — sort-merge join spilling	✅ landed	join spilling viable in DF; today's escape hatch
2025	#17267 — hybrid hash join proposal	open, unimplemented	right direction; named the hard parts — adopted here, coordination concern resolved by partition-locality
2025–26	Spill infra: SpillManager compression, SpillPool rotation, ReplayableStreamSource, disk quotas	✅ landed	the disk layer is ready
2026 / DF 54	Spilling nested-loop join	✅ landed	in-tree pattern: OOM fallback + replayable input; reused as the skew fallback

Proposed next steps (one reviewable PR each)

• T0 (S) — memory-limited join benchmark + budgeted-RuntimeEnv test harness; encode today's failure matrix (incl. the 3.5× SMJ-tax row) as the baseline
• T1 (M) — mechanical refactor: collect_left_input → BuildSideBuffer (no behavior change; bench-verified ≤ noise)
• T2 (S/M) — bucket-splitting kernel (bucket_indices(hashes, bit_window) + take) with criterion benchmarks
• T3 (M) — BucketedBuildBuffer: 16 buckets, Phase-A→Phase-B transition, destage-largest, SpillManager files, with_can_spill(true) (flag-gated)
• T4 (L) — core: INNER equi-join end-to-end, single level. Acceptance: the failing join from the symptoms table completes under 300 MB in ≤ 2× the SMJ-workaround time (≤ ~7.5 s), and no-spill join benchmarks stay within 2% of main
• T5 (L) — full join-type matrix (per-bucket bitmaps; Right/Full early emission; null-aware anti) + spill-forcing mode for the join fuzzer
• T6 (M) — recursion (bit windows, depth cap) + skew detection + chunked-build/replay fallback
• T7 (M) — Bloom filters: negative-row early resolution in pass 1 + role reversal for Inner/Semi
• T8 (S/M) — planner + property integration: CollectLeft-steering rule when the flag is on; drop probe-side maintains_input_order under the flag; membership→bounds/InList downgrade; TPC-H plan-diff audit with the flag on
• T9 (M) — memory-model polish: table-headroom reservation, victim policy, cancellation-safe temp-file cleanup, prefetch of spilled pair i+1
• T10 (S) — metrics (SpillMetrics + spilled_buckets, spill_recursion_depth) in EXPLAIN ANALYZE; user-guide memory section; degradation-curve write-up
• T11 (M) — CI: TPC-H SF1 under stepped budgets (4 GB → 256 MB). Path to default-on, stated up front: flip enable_hash_join_spilling when (a) the force-spill suite has been green for two consecutive releases, (b) the fuzzer is clean, (c) spilled hybrid ≤ the SMJ workaround on the benchmark matrix, and (d) no-spill regression ≤ 2% — proposed and decided on this issue

Dependency order: T0–T2 parallel; T3 → T4 → {T5, T6, T7} → T8–T11.

Related issues

• Memory Limited Joins (Externalized / Spill) #1599, Add spilling support for HashJoin #12952 — the asks this epic closes · PROPOSAL Hash Join Spilling Proposal #17267 — prior proposal (suggest converging discussion here)
• Optimized spill file format #14078 spill format (orthogonal) · [Epic] Generate runtime errors if the memory budget is exceeded #3941 memory-error reporting · [Epic] Dynamic filtering related items #15512 / Pushing down HashJoinExec build side dynamic filters makes tpch queries slower #19858 dynamic filters (downgrade + the re-scan follow-up build on these)

Open questions

1. Bucket count: 16 per level, or larger fan-out? (Routing cost and file count vs. recursion depth — T2/T4 benchmarks decide.)
2. When to start spilling: only when an allocation is actually refused (try_grow fails), or slightly earlier — e.g. once the join has used ~80% of the memory it can expect to get — so there is still headroom for the split work and the hash-table build? (The early trigger is only meaningful under FairSpillPool, where each consumer has a defined share; under GreedyMemoryPool the failure reaction is the trigger either way.)
3. Probe spill files: plain per-bucket InProgressSpillFile (phase 1) vs. SpillPool rotation for very large probes.

References

DuckDB PR #4189 · "Saving Private Hash Join" VLDB '25 · Velox spilling · ClickHouse grace_hash · SPARK-32634 · Trino spill · Kitsuregawa '83 (GRACE); DeWitt '84 / Shapiro '86 (hybrid hash); Graefe et al. VLDB '98 (dynamic destaging, role reversal). Experiment scripts: design doc, Appendix A.

[peterxcli]

take

[peterxcli]

let me do some initial research and then post brief design here

[alamb]

@jayzhan211 how would you like feedback on this proposal? Would you like it in the attached google doc or on this issue?

[alamb]

@peterxcli I think @jayzhan211 's writeup is a design proposal. My reading of it is:

1. Add a dynamic transition from CollectLeft --> Partitioned mode if the single collect left hash table is too large
2. Add per-partition spilling to the Partitioned mode

I may have mis read it, and I have some suggestions on how to make this proposal clearer:

Suggestions

1. "What is going on (symptoms)" heading:

2. Add a link or the exact query that is being run explicitly (the text decribes the scenario in the abstract but is not a concrete reprodicer)

3. Describe the different options available to run the query today a bit more ('default settings', "Always use Merge Join", "count(DISTINCT payload) hash aggregate") -- it was not entirely clear to me what those options

Also, what is the difference between "hash join, build side 12M rows" and "hash join t_probe ⋈ t_build ON k" ? Is the idea that we can't get the join to run by switching the inputs? If so I suggest explicitly saying so and using the same format (maybe you could collapse the table entries into a single row)

[jayzhan211]

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:

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.

reproducer — datafusion-cli, verified on the 54.0.0 tag

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):

-- 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:

-- 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.

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.