jayzhan211 opened a new issue, #24768: URL: https://github.com/apache/datafusion/issues/24768
# [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](https://docs.google.com/document/d/1yF-_6neNxHZulgw5O1cFDfFPa57F54F5/edit?usp=sharing&ouid=102243744706766255778&rtpof=true&sd=true)** *(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 `SortExec`s 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 ```text 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): ```text 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](https://github.com/duckdb/duckdb/pull/4189), ["Saving Private Hash Join", VLDB '25](https://duckdb.org/library/saving-private-hash-join-vldb/)) | 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](https://facebookincubator.github.io/velox/develop/spilling.html)) | 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](https://clickhouse.com/blog/clickhouse-fully-supports-joins-hash-joins-part2)) | 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](https://issues.apache.org/jira/browse/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](https://trino.io/docs/current/admin/spill.html)) | 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** (#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 - [ ] #1599, #12952 β the asks this epic closes Β· #17267 β prior proposal (suggest converging discussion here) - [ ] #14078 spill format (orthogonal) Β· #3941 memory-error reporting Β· #15512 / #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](https://github.com/duckdb/duckdb/pull/4189) Β· ["Saving Private Hash Join" VLDB '25](https://duckdb.org/library/saving-private-hash-join-vldb/) Β· [Velox spilling](https://facebookincubator.github.io/velox/develop/spilling.html) Β· [ClickHouse grace_hash](https://clickhouse.com/blog/clickhouse-fully-supports-joins-hash-joins-part2) Β· [SPARK-32634](https://issues.apache.org/jira/browse/SPARK-32634) Β· [Trino spill](https://trino.io/docs/current/admin/spill.html) Β· Kitsuregawa '83 (GRACE); DeWitt '84 / Shapiro '86 (hybrid hash); Graefe et al. VLDB '98 (dynamic destaging, role reversal). Experiment scripts: design doc, Appendix A. -- 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]
