Hash join spill - #45
Merged
Merged
Conversation
Groundwork for the spillable hash join (no behavior change): - Extract build_left_data() from collect_left_input() — the hash-table/ ArrayMap/concat/bitmap/membership construction — so the same code will serve both the classic in-memory build and per-partition builds when the join spills. - collect_left_input() now returns BuildPhaseOutput (single InMemory variant for now; Spilled arrives with the scatter phase). - Register the build-side reservation with_can_spill(true) when the new execution.enable_hash_join_spill flag is on. - New config knobs (default off/auto): enable_hash_join_spill, hash_join_spill_partition_count, hash_join_spill_headroom_bytes, hash_join_spill_max_recursion_depth; config docs + information_schema expectations regenerated. Verified: full physical-plan suite (1268 passed; spill_compression zstd feature quirk pre-exists on clean base), joins suite 926 passed, full sqllogictest suite green modulo the pre-existing local encrypted_parquet quirk, clippy -D warnings clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
When enable_hash_join_spill is on and the build-side reservation first fails, the build switches into scatter mode instead of erroring: buffered batches and the rest of the build input are hash-partitioned into K disk partitions (new spill.rs), the probe side is scattered with the same seeds, and the join runs partition-pair by pair, each pair building an ordinary in-memory hash table under its own reservation. The in-memory fast path is unchanged. - spill.rs: SideScatter / PartitionSpillWriter (coalesced IPC writes, file rotation, StringView compaction), chained spill readers, SpillJoinDriver state machine, auto partition-count sizing. - exec.rs: engage-on-overflow inside the build fold with exact reservation shrinks; BuildPhaseOutput::Spilled; per-partition spill context in execute(); spill disabled when it would break the advertised probe-side output ordering. - stream.rs: SpillJoin stream state; bounds-only dynamic-filter reporting for spilled partitions (prevents both the bounds-report deadlock and wrong pruning); output flows through the existing coalescer/fetch. - shared_bounds.rs: PushdownStrategy::BoundsOnly (unlike Empty, a spilled partition's probe rows must not be pruned by the CASE filter). M2 scope: PartitionMode::Partitioned + JoinType::Inner only (gate in execute()); oversized partitions surface a clean resources-exhausted error until recursion lands (M4). Tests (8 new): scatter round-trip, seed independence from repartition hash, file rotation + temp cleanup, forced-spill inner join equals in-memory results with spill metrics recorded, fast path untouched with ample memory, pool fully released after completion and after mid-stream drop, disk-cap exhaustion and flag-off both keep today's clean errors. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Remove the inner-join-only gate: disk partitions are key-disjoint, so each pair's in-memory join has complete knowledge for its keys — per-pair visited bitmaps (unmatched build emission) and probe-side null padding work unchanged. The driver's pair-skip logic is now a per-type matrix: pairs are skipped only when the join type provably emits nothing for an empty side; pairs that emit unmatched rows from the non-empty side still run (the inner stream handles the empty other side natively). The ordering guard from M2 (no spilling when the operator's advertised probe-side order maintenance is live) now carries the weight for Inner/Right/RightSemi/RightAnti/RightMark. Null-aware anti joins remain excluded. Tests: forced-spill matrix over all 10 join types vs in-memory reference (multiset-fingerprint comparison), and an Inner/Left/Full x NullEqualsNothing/NullEqualsNull matrix with NULL join keys (all NULLs scatter to one partition, so NullEqualsNull matching stays complete; NULL-heavy skew is deferred to the chunked fallback in M4). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
A pair whose build partition exceeds the budget is now handled in three tiers instead of erroring: 1. Recursive repartition (up to hash_join_spill_max_recursion_depth passes): already-loaded batches and the rest of the pair's build stream re-scatter into 8 children at the next seed level, the pair's probe partition follows with the same seed, and the children are queued front-first so their disk space is reclaimed promptly. A no-shrink shortcut (largest child > 3/4 of the parent) marks children as duplicate-key mass and sends them straight to the fallback. 2. Chunked build fallback for build-side-emission types (Inner, Left, LeftSemi, LeftAnti, LeftMark): build rows load in budget-sized chunks, each joined against a full re-read of the pair's probe partition — correct because each build row lives in exactly one chunk. 3. Probe-side-emission types (Right family, Full) in this corner surface a descriptive resources-exhausted error (their outputs depend on matches against the whole build side, which chunks cannot track) — never wrong results. Two real bugs found by the new tests and fixed: - Seed derivation: per-level scatter seeds differed only by a small additive constant in one ahash seed word, and ahash folds that word in additively — hashes at consecutive levels differed by a constant, so under % K a whole parent partition landed in ONE child and recursion could never split anything. Seeds are now derived through a SplitMix64 mixer (regression test: recursive_scatter_levels_are_decorrelated). - Memory accounting: loaders reserved only batch bytes, so a chunk/pair could fill the budget with data and then fail on the hash table's estimate (~2x the data for narrow rows). Spill-mode loaders (build fold, pair load, chunk load) now pre-reserve the incremental table estimate per batch — overflow surfaces where the fallback logic can engage — and build_left_data takes prereserved_table_bytes so the accounting stays exact (fast path passes 0: unchanged). New metrics: join_spill_repartition_passes, join_spill_fallback_chunks. Tests (15 total now): extreme duplicate-key skew via chunked fallback (Inner + LeftAnti) vs in-memory reference, unsupported-type skew clean error, moderate-skew recursion with zero fallback chunks, per-level scatter decorrelation. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
CollectLeft joins (single shared build, N probe partitions) can now spill: the build scatters once into K disk partitions, each output partition scatters ITS OWN probe stream with the same seed and walks k = 0..K, and every k must run in every partition (skipping would leave the shared completion counter high and suppress unmatched emission). Design decision recorded by the tests: per-k hash tables are NOT shared across probe partitions. Sharing them (first attempt, via per-k OnceAsync) meant a retained-table WINDOW spanning the progress gap between the fastest and slowest partition — up to K tables resident, observed directly as three concurrent multi-MB tables blowing a pool sized for two. Instead each partition builds its own copy of k's table from the shared files (bounded: at most one resident table per partition, matching the auto-K sizing, which now also divides its per-partition target by the consumer count). What IS shared per k is exactly what correctness needs: the visited bitmap and the probe-completion counter, injected into each copy's JoinLeftData — JoinLeftData's bitmap/counter are Arc'd now — so matches from all partitions mark one bitmap and the existing last-finisher logic emits unmatched build rows exactly once. Row order across copies is deterministic (same files, same order), so bitmap indices agree. Oversized shared partitions surface a clean descriptive error (no recursion for the shared build: a bitmap cannot be split across repartition levels). Null-aware anti joins remain excluded. Test: forced CollectLeft spill over Inner/Left/LeftAnti/Full with two concurrently-polled output partitions — fingerprint-identical to the in-memory join (exactly-once unmatched emission), spill engaged, pool drained after the plan drops (the exec node's left_fut cache retains the scatter headroom until then, as it retains the whole build today). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… pass Two more cancel/error-safety tests: dropping the output stream deep inside the chunked skew fallback releases every reservation (chunk data, table estimate, headroom, scratch), and a probe-side error arriving mid-scatter propagates verbatim with the pool drained. Regression status at this point: 18 spill tests, full physical-plan suite 1286 passed (spill_compression zstd feature quirk pre-exists on clean base), full sqllogictest suite green except the pre-existing local encrypted_parquet environment quirk, clippy -D warnings and fmt clean. Performance parity gates (flag-on vs flag-off at SF1/SF10) run with the rustice integration, where a release build and the benchmark harness exist. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Operational visibility for long spilled joins on remote deployments: one line when a build side switches to scatter mode (with buffered bytes/rows), one per recursive repartition pass, one per chunked- fallback chunk. Mirrors SortExec's spill logging. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…tually available with_can_spill() was driven by the raw config flag, but under FairSpillPool a spillable consumer is capped at pool/n_spillable — a join excluded from the spill fallback (null-aware anti join, order-preserving probe side, disabled DiskManager) got the tiny fair share with no fallback to honor it, failing queries that passed before the feature existed. Found by TPC-H Q16 at SF100: its NOT IN subquery plans as a null-aware anti join, which hit a ~155 MB share of a 16 GB pool and errored. The reservation is now marked spillable exactly when the spill context exists (the full gate), so non-degradable joins keep unspillable semantics. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
github-actions
Bot
added
documentation
osipovartem
pushed a commit
that referenced
this pull request
Aug 30, 2026
* Hash join spill M1: neutral build-phase refactor + config plumbing Groundwork for the spillable hash join (no behavior change): - Extract build_left_data() from collect_left_input() — the hash-table/ ArrayMap/concat/bitmap/membership construction — so the same code will serve both the classic in-memory build and per-partition builds when the join spills. - collect_left_input() now returns BuildPhaseOutput (single InMemory variant for now; Spilled arrives with the scatter phase). - Register the build-side reservation with_can_spill(true) when the new execution.enable_hash_join_spill flag is on. - New config knobs (default off/auto): enable_hash_join_spill, hash_join_spill_partition_count, hash_join_spill_headroom_bytes, hash_join_spill_max_recursion_depth; config docs + information_schema expectations regenerated. Verified: full physical-plan suite (1268 passed; spill_compression zstd feature quirk pre-exists on clean base), joins suite 926 passed, full sqllogictest suite green modulo the pre-existing local encrypted_parquet quirk, clippy -D warnings clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Hash join spill M2: Partitioned-mode inner-join grace spill path When enable_hash_join_spill is on and the build-side reservation first fails, the build switches into scatter mode instead of erroring: buffered batches and the rest of the build input are hash-partitioned into K disk partitions (new spill.rs), the probe side is scattered with the same seeds, and the join runs partition-pair by pair, each pair building an ordinary in-memory hash table under its own reservation. The in-memory fast path is unchanged. - spill.rs: SideScatter / PartitionSpillWriter (coalesced IPC writes, file rotation, StringView compaction), chained spill readers, SpillJoinDriver state machine, auto partition-count sizing. - exec.rs: engage-on-overflow inside the build fold with exact reservation shrinks; BuildPhaseOutput::Spilled; per-partition spill context in execute(); spill disabled when it would break the advertised probe-side output ordering. - stream.rs: SpillJoin stream state; bounds-only dynamic-filter reporting for spilled partitions (prevents both the bounds-report deadlock and wrong pruning); output flows through the existing coalescer/fetch. - shared_bounds.rs: PushdownStrategy::BoundsOnly (unlike Empty, a spilled partition's probe rows must not be pruned by the CASE filter). M2 scope: PartitionMode::Partitioned + JoinType::Inner only (gate in execute()); oversized partitions surface a clean resources-exhausted error until recursion lands (M4). Tests (8 new): scatter round-trip, seed independence from repartition hash, file rotation + temp cleanup, forced-spill inner join equals in-memory results with spill metrics recorded, fast path untouched with ample memory, pool fully released after completion and after mid-stream drop, disk-cap exhaustion and flag-off both keep today's clean errors. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Hash join spill M3: all join types under the Partitioned spill path Remove the inner-join-only gate: disk partitions are key-disjoint, so each pair's in-memory join has complete knowledge for its keys — per-pair visited bitmaps (unmatched build emission) and probe-side null padding work unchanged. The driver's pair-skip logic is now a per-type matrix: pairs are skipped only when the join type provably emits nothing for an empty side; pairs that emit unmatched rows from the non-empty side still run (the inner stream handles the empty other side natively). The ordering guard from M2 (no spilling when the operator's advertised probe-side order maintenance is live) now carries the weight for Inner/Right/RightSemi/RightAnti/RightMark. Null-aware anti joins remain excluded. Tests: forced-spill matrix over all 10 join types vs in-memory reference (multiset-fingerprint comparison), and an Inner/Left/Full x NullEqualsNothing/NullEqualsNull matrix with NULL join keys (all NULLs scatter to one partition, so NullEqualsNull matching stays complete; NULL-heavy skew is deferred to the chunked fallback in M4). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Hash join spill M4: recursive repartitioning + chunked skew fallback A pair whose build partition exceeds the budget is now handled in three tiers instead of erroring: 1. Recursive repartition (up to hash_join_spill_max_recursion_depth passes): already-loaded batches and the rest of the pair's build stream re-scatter into 8 children at the next seed level, the pair's probe partition follows with the same seed, and the children are queued front-first so their disk space is reclaimed promptly. A no-shrink shortcut (largest child > 3/4 of the parent) marks children as duplicate-key mass and sends them straight to the fallback. 2. Chunked build fallback for build-side-emission types (Inner, Left, LeftSemi, LeftAnti, LeftMark): build rows load in budget-sized chunks, each joined against a full re-read of the pair's probe partition — correct because each build row lives in exactly one chunk. 3. Probe-side-emission types (Right family, Full) in this corner surface a descriptive resources-exhausted error (their outputs depend on matches against the whole build side, which chunks cannot track) — never wrong results. Two real bugs found by the new tests and fixed: - Seed derivation: per-level scatter seeds differed only by a small additive constant in one ahash seed word, and ahash folds that word in additively — hashes at consecutive levels differed by a constant, so under % K a whole parent partition landed in ONE child and recursion could never split anything. Seeds are now derived through a SplitMix64 mixer (regression test: recursive_scatter_levels_are_decorrelated). - Memory accounting: loaders reserved only batch bytes, so a chunk/pair could fill the budget with data and then fail on the hash table's estimate (~2x the data for narrow rows). Spill-mode loaders (build fold, pair load, chunk load) now pre-reserve the incremental table estimate per batch — overflow surfaces where the fallback logic can engage — and build_left_data takes prereserved_table_bytes so the accounting stays exact (fast path passes 0: unchanged). New metrics: join_spill_repartition_passes, join_spill_fallback_chunks. Tests (15 total now): extreme duplicate-key skew via chunked fallback (Inner + LeftAnti) vs in-memory reference, unsupported-type skew clean error, moderate-skew recursion with zero fallback chunks, per-level scatter decorrelation. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Hash join spill M5: CollectLeft shared-build spill CollectLeft joins (single shared build, N probe partitions) can now spill: the build scatters once into K disk partitions, each output partition scatters ITS OWN probe stream with the same seed and walks k = 0..K, and every k must run in every partition (skipping would leave the shared completion counter high and suppress unmatched emission). Design decision recorded by the tests: per-k hash tables are NOT shared across probe partitions. Sharing them (first attempt, via per-k OnceAsync) meant a retained-table WINDOW spanning the progress gap between the fastest and slowest partition — up to K tables resident, observed directly as three concurrent multi-MB tables blowing a pool sized for two. Instead each partition builds its own copy of k's table from the shared files (bounded: at most one resident table per partition, matching the auto-K sizing, which now also divides its per-partition target by the consumer count). What IS shared per k is exactly what correctness needs: the visited bitmap and the probe-completion counter, injected into each copy's JoinLeftData — JoinLeftData's bitmap/counter are Arc'd now — so matches from all partitions mark one bitmap and the existing last-finisher logic emits unmatched build rows exactly once. Row order across copies is deterministic (same files, same order), so bitmap indices agree. Oversized shared partitions surface a clean descriptive error (no recursion for the shared build: a bitmap cannot be split across repartition levels). Null-aware anti joins remain excluded. Test: forced CollectLeft spill over Inner/Left/LeftAnti/Full with two concurrently-polled output partitions — fingerprint-identical to the in-memory join (exactly-once unmatched emission), spill engaged, pool drained after the plan drops (the exec node's left_fut cache retains the scatter headroom until then, as it retains the whole build today). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Hash join spill M6: lifecycle hardening tests + full-suite regression pass Two more cancel/error-safety tests: dropping the output stream deep inside the chunked skew fallback releases every reservation (chunk data, table estimate, headroom, scratch), and a probe-side error arriving mid-scatter propagates verbatim with the pool drained. Regression status at this point: 18 spill tests, full physical-plan suite 1286 passed (spill_compression zstd feature quirk pre-exists on clean base), full sqllogictest suite green except the pre-existing local encrypted_parquet environment quirk, clippy -D warnings and fmt clean. Performance parity gates (flag-on vs flag-off at SF1/SF10) run with the rustice integration, where a release build and the benchmark harness exist. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Hash join spill: log engage/repartition/chunk progress at info Operational visibility for long spilled joins on remote deployments: one line when a build side switches to scatter mode (with buffered bytes/rows), one per recursive repartition pass, one per chunked- fallback chunk. Mirrors SortExec's spill logging. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Hash join spill: mark reservations spillable only when spilling is actually available with_can_spill() was driven by the raw config flag, but under FairSpillPool a spillable consumer is capped at pool/n_spillable — a join excluded from the spill fallback (null-aware anti join, order-preserving probe side, disabled DiskManager) got the tiny fair share with no fallback to honor it, failing queries that passed before the feature existed. Found by TPC-H Q16 at SF100: its NOT IN subquery plans as a null-aware anti join, which hit a ~155 MB share of a 16 GB pool and errored. The reservation is now marked spillable exactly when the spill context exists (the full gate), so non-degradable joins keep unspillable semantics. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
What changes are included in this PR?
Are these changes tested?
Are there any user-facing changes?