Share per-chunk JoinLeftData across right partitions in NLJ memory-limited fallback by viirya · Pull Request #22038 · apache/datafusion

viirya · 2026-05-06T05:55:18Z

Which issue does this PR close?

Closes NestedLoopJoin memory-limited fallback lacks cross-partition left-bitmap coordination, so LEFT/FULL joins over a partitioned right side cannot spill #24764.

Rationale for this change

The memory-limited fallback path builds a per-output-partition JoinLeftData for each left chunk, with probe_threads_counter == 1. The single-pass path instead shares one JoinLeftData seeded via collect_left_input(.., probe_threads_count). With a private visited bitmap per partition, each right partition sees only its own right rows, so unmatched left rows would be emitted once per partition, and rows matched only in another partition would be emitted as unmatched.

Because those results would be wrong, main refuses the fallback for the affected join types rather than producing them:

let left_emission_multi_partition =
    need_produce_result_in_final(self.join_type) && right_partition_count > 1;
let can_spill = context.runtime_env().disk_manager.tmp_files_enabled()
    && !left_emission_multi_partition;

So the state on main today is not incorrect output but a missing capability: with target_partitions > 1, a LEFT / LEFT SEMI / LEFT ANTI / LEFT MARK / FULL nested loop join has no spilling path at all and fails with ResourcesExhausted under memory pressure. This PR makes those joins spill instead.

(When this PR was first opened the guard covered only FULL, so the wrong results were still reachable for the other join types. #21833 and #24675 have since widened it to every left-emitting type, which is why the framing above differs from the original description.)

What changes are included in this PR?

A plan-level FallbackCoordinator that:

Owns the shared left spill stream and the per-chunk MemoryReservation, so each chunk is read exactly once no matter how many partitions probe it,
Has the first partition reaching a chunk become its "leader": it loads the chunk and publishes an Arc<JoinLeftData> (with probe_threads_counter == right_partition_count) into a shared slot,
Lets every other right partition take an Arc clone of that same JoinLeftData, so the visited bitmap and probe-thread counter are shared exactly as in the single-pass collect_left_input path,
Releases the slot only after the partition that drives the counter to zero finishes emitting unmatched left rows for the chunk, then notifies waiters so the next chunk can be loaded.

The per-chunk in-flight fetch and release are driven through future fields on SpillStateActive (chunk_fetch_in_flight / chunk_release_in_flight), polled across poll_next iterations. Partitions waiting for a chunk sleep on a Notify rather than busy-looping, and every early return after a leader claim clears loader_in_flight and wakes waiters so a failed load cannot leave others parked.

The left_emission_multi_partition guard is removed, so these join types now take the coordinated path.

datafusion.execution.enable_nlj_coordinated_fallback (default true) lets distributed engines opt out. The coordination assumes all right partitions run in the same process; an engine that runs each partition as an independent task would get one coordinator per task, so the shared probe-thread counter would never reach zero and the fallback would stall. Setting the flag to false restores the current fail-fast behavior for the affected join types.

Rebase notes (2026-08-28)

Rebased onto current main. Two things changed underneath this PR:

main now shares the spilled left side across partitions itself, resolving a shared OnceAsync<LeftLoad> so the left child is executed and spilled exactly once. That was one half of this PR's original design (left_spill_fut: OnceFut<LeftSpillData>), so that half is dropped in favor of main's mechanism. What remains — and what main still lacks — is sharing the per-chunk visited bitmap and probe-thread counter.
fix: emit deferred unmatched rows when memory-limited NestedLoopJoin exhausts its left side #24746 added EmitGlobalRightUnmatched routing so the memory-limited path does not drop deferred unmatched probe-side rows when the left side is exhausted. That routing is preserved on the coordinator's exhausted-left path, and its three regression tests in datafusion/core/tests/memory_limit/nlj_spill_unmatched.rs pass.

Design discussion from the original round: #21833 (comment) and #21833 (comment).

Are these changes tested?

Yes:

Seven multi-partition memory-limited correctness tests in nested_loop_join.rs: one each for LEFT, FULL, LEFT SEMI, LEFT ANTI, LEFT MARK, plus two multi-chunk cases (LEFT, FULL) that exercise the chunk hand-off between leaders. Each compares the memory-limited result against the unrestricted one. All seven fail on main (as ResourcesExhausted) and pass here.
Multi-partition cases added to nested_loop_join_spill.slt, including an EXPLAIN ANALYZE assertion on spill_count so the tests fail if the fallback silently stops being taken.
enable_nlj_coordinated_fallback listed in information_schema.slt.

Regression-checked after the rebase: 55 NLJ unit tests, 1125 joins tests, 36 memory_limit tests (including the three #24746 tests above), 259 datafusion-proto tests, and the nested_loop_join* / information_schema sqllogictest files. cargo clippy --all-targets --all-features -- -D warnings is clean.

Are there any user-facing changes?

Yes, two:

LEFT, LEFT SEMI, LEFT ANTI, LEFT MARK, and FULL nested loop joins over a multi-partition right side now spill under memory pressure instead of failing with ResourcesExhausted.
New config option datafusion.execution.enable_nlj_coordinated_fallback (default true), documented in docs/source/user-guide/configs.md. No public API signature changes.

kosiew

@viirya
Thanks for working on this. I think the coordinator changes need one more regression test before this lands, specifically one that exercises the multi-chunk path.

kosiew · 2026-06-22T09:02:27Z

+04)------AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))], metrics=[<slt:ignore>]
+05)--------NestedLoopJoinExec: join_type=Left, filter=v1@0 + v2@1 = 101, projection=[], metrics=[output_rows=5.00 K, <slt:ignore> spill_count=2, <slt:ignore>]
+06)----------ProjectionExec: expr=[value@0 as v1], metrics=[<slt:ignore>]
+07)------------LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=5000, batch_size=8192], metrics=[<slt:ignore>]


Thanks for adding the regression coverage here. I do not think this currently exercises the multi-chunk coordinator path though.

The plan shows the left generate_series(1, 5000) being produced as one batch_size=8192 batch, and load_one_chunk accepts a single over-budget batch so it can make progress. Because of that, these SLT cases appear to validate only the one-chunk fallback path.

The main invariant added by the coordinator is per-chunk sharing, followed by releasing the current chunk and advancing to the next one. Without a test where the left input is split into at least two chunks, the carryover, release_chunk, waiter notification, and global-right accumulation across chunks are not really covered.

Could you please add a multi-partition spill regression with multiple left batches or chunks? For example, this could be a Rust test that feeds more than one left batch, or an SLT shape that produces multiple left batches. It would be good to assert the same LEFT/FULL or LEFT SEMI/ANTI counts, plus spill_count > 0.

kosiew · 2026-06-22T09:02:27Z

+                );
+            }
+
+            // Case 1: requested chunk is already loaded.


Small follow-up suggestion: while a leader is loading a chunk, waiters can see left_stream.is_none() and reopen the spill stream before they reach the loader_in_flight wait path.

The leader later overwrites that stream, so this looks like wasted I/O and state churn. It may be worth guarding lazy stream/reservation initialization with !inner.loader_in_flight, or moving that initialization into the leader-claim branch so only the loader sets up those fields.

Previously the shared left spill stream and chunk reservation were lazily initialized at the top of `FallbackCoordinator::next_chunk`, before the `loader_in_flight` guard. While a leader was loading a chunk, a waiting partition could observe `left_stream.is_none()` and reopen the spill stream, which the leader would then overwrite — wasted I/O and state churn. Move the lazy initialization into the leader-claim branch (under the `loader_in_flight` guard), so only the leader sets up the stream, schema, and reservation. Waiters fall straight through to the notify wait path. Addresses a review suggestion on apache#22038. Co-authored-by: Claude Code

The existing multi-partition spill tests feed the left side as a single batch, so the memory-limited fallback only ever loads one chunk. That leaves the coordinator's multi-chunk machinery untested: `carryover` between chunks, `release_chunk` advancing to the next chunk, waiter notification, and (for FULL) the global right-unmatched bitmap accumulated across chunks. Add `*_multi_chunk_left_join` and `*_multi_chunk_full_join`: the left input is one row per batch under a tight memory limit, so the coordinator loads each row as a separate chunk (verified to load multiple chunks with spill_count > 0). Both assert exact row sets, so a regression in per-chunk `JoinLeftData` sharing (duplicate unmatched rows) or cross-chunk right accumulation would fail. These collect the four output partitions concurrently: a chunk is not released until all partitions finish probing it, so sequential collection would deadlock — concurrent collection mirrors how partitions run under the runtime. Also drop a stray `dbg!` in `join_inner_with_filter`. Addresses review feedback on apache#22038. Co-authored-by: Claude Code

viirya · 2026-06-22T16:51:41Z

@kosiew Thanks for the careful review — both points were spot on.

Multi-chunk coverage. You're right that the existing SLT cases only exercised the single-chunk path: generate_series(1, 5000) arrives as one 8192-row batch, so load_one_chunk accepts the single over-budget batch and the coordinator never loads a second chunk. The spill_count there comes from the right side, not from left chunking. I couldn't reliably target multi-chunk from SLT (a memory limit tight enough to split the left side OOMs other operators first), so I added Rust regression tests instead: test_nlj_memory_limited_multi_partition_multi_chunk_{left,full}_join. The left input is one row per batch under a tight memory limit, which I verified loads multiple distinct chunks with spill_count > 0. They assert exact row sets, so a regression in per-chunk JoinLeftData sharing (duplicate unmatched rows) or in the cross-chunk global right-unmatched accumulation (FULL) would fail.

One thing worth noting: these tests collect the four output partitions concurrently. Sequential collection deadlocks on the multi-chunk path — a chunk isn't released until every partition finishes probing it (the probe_threads_counter reaching zero), and no partition can advance to the next chunk until the current one is released. So if only one partition is driven at a time, it blocks waiting for a release that can never happen. Concurrent collection mirrors how partitions actually run under the runtime; I documented this in the helper.

Leader-only stream init. Good catch — moved the lazy stream/schema/reservation initialization into the leader-claim branch (under the loader_in_flight guard) so waiters no longer open a throwaway spill stream that the leader overwrites.

Pushed both as separate commits. Ready for another look when you have a chance.

kosiew

@viirya

Thanks for the updates. I re-reviewed the changes and the two follow-ups I raised have been addressed:

Added multi-chunk regression coverage with concurrent output partitions, strict memory pressure, exact LEFT/FULL output assertions, and spill verification.
Moved stream initialization under the leader claim so waiting tasks no longer create throwaway spill streams.

I only have one small non-blocking suggestion.

kosiew · 2026-06-24T04:17:53Z

+                    None => {
+                        let stream = spill_data
+                            .spill_manager
+                            .read_spill_as_stream(spill_data.spill_file.clone(), None)?;


Nice improvement moving stream initialization behind the leader claim.

One small thing that caught my eye: loader_in_flight is set to true before calling read_spill_as_stream(...), and the call uses ?. Today that seems safe because read_spill_as_stream only constructs the buffered stream and returns Ok(...); any real file/open/schema errors happen later while polling the stream, where load_one_chunk already clears loader_in_flight and wakes any waiters.

This feels a little fragile though. If stream construction ever becomes genuinely fallible in the future, an early return here could leave waiters blocked indefinitely. It might be worth avoiding ? while the flag is set, or using a small guard/helper that guarantees the flag is cleared and waiters are notified if leader setup fails.

Good catch — agreed it's worth hardening even though it isn't reachable today (read_spill_as_stream only builds the buffered stream; the real I/O errors surface later while polling, where load_one_chunk already clears loader_in_flight and notifies).

Fixed: on stream-construction failure the leader now clears loader_in_flight, drops the lock, wakes waiters, and propagates the error, so another partition can claim the leader role and retry rather than blocking on a release that never comes. Pushed as a separate commit.

addressed

kosiew · 2026-06-24T04:20:33Z

Leaving this open for more review as this is a big PR.

…ails In `FallbackCoordinator::next_chunk`, the leader sets `loader_in_flight` before constructing the spill stream. If that construction returned an error via `?`, the leader would bail out while still holding `loader_in_flight = true` and without notifying waiters — leaving every other partition blocked forever on a chunk release that the failed leader can never make. `read_spill_as_stream` only builds the buffered stream today (real I/O errors surface later while polling, where the flag is already cleared), so this is not currently reachable. Handle it defensively anyway: on construction failure, clear `loader_in_flight`, drop the lock, wake waiters, and propagate the error so another partition can retry. Addresses a review suggestion on apache#22038. Co-authored-by: Claude Code

andygrove · 2026-07-02T16:12:58Z

Thanks for fixing this. A question about distributed execution.

Downstream engines like Ballista and datafusion-distributed run each output partition of a plan as an independent task, often in a separate process. Each task deserializes and instantiates the physical plan on its own and calls execute(partition) for only its one partition.

If I read FallbackCoordinator correctly, it initializes probe_threads_counter to right_partition_count and loads the next left chunk only after the partition that brings the counter to zero notifies the waiters. In a distributed setup a process builds its own FallbackCoordinator but polls a single right partition, so the counter would never reach zero and the Notify waiters would wait forever. Would that stall the memory-limited fallback whenever the right side is spread across tasks?

Could a distributed consumer opt out of this cross-partition coordination through a config flag, the way datafusion.optimizer.enable_dynamic_filter_pushdown lets them disable another single-process assumption? #22671 looks like it already takes the distributed-safe route for left-emitting joins by disabling the fallback instead of coordinating across partitions.

This review was assisted by an LLM.

@andygrove

The memory-limited NestedLoopJoin fallback coordinates per-chunk left state (visited bitmap + probe-thread counter) across all right-side partitions via `FallbackCoordinator`. That coordination assumes every right partition runs in the same process. Distributed engines (Ballista, datafusion-distributed) execute each output partition as an independent task that builds its own coordinator and polls a single partition, so the shared probe-thread counter never reaches zero, no partition is elected to release the current chunk, and the waiters block forever — the memory-limited fallback deadlocks. Add `datafusion.execution.enable_nlj_coordinated_fallback` (default true, preserving current behavior). When set to false, the coordinated fallback is disabled for the join types that would deadlock — left-emitting joins (LEFT, LEFT SEMI, LEFT ANTI, LEFT MARK, FULL) with a multi-partition right side — which then use `SpillState::Disabled` and fail with resource exhaustion under memory pressure rather than deadlocking. Single-partition and non-left-emitting joins are always safe and keep the coordinated fallback regardless of the flag. Raised by @andygrove on apache#22038. Co-authored-by: Claude Code

viirya · 2026-07-18T21:00:16Z

@andygrove Thanks for looking at this closely — your analysis is correct, and it's a real deadlock, not a stall in theory.

To confirm the mechanism: the coordinator initializes probe_threads_counter to the global right_partition_count, and a chunk is only released (and the next one loaded) once the partition that drives the counter to zero notifies the waiters. When a process builds its own FallbackCoordinator but only executes one right partition, the counter stops at N-1, nobody is elected to release the chunk, and the other next_chunk waiters block forever on the Notify.

One nuance I want to be honest about: the single-process assumption itself isn't new — the single-pass path (collect_left_input(..., right_partition_count)) already sets the counter to the global partition count, and a distributed run there produces incorrect results (missing unmatched-left rows) because non-emitter partitions just return early. What this PR changes is that the coordinator makes that assumption load-bearing for liveness, turning a wrong-results degradation into a hang — a strictly worse failure mode. So the concern is fair and specific to the coordinated fallback.

I've pushed an opt-out along the lines you suggested: a new config flag datafusion.execution.enable_nlj_coordinated_fallback (default true). When a distributed consumer sets it to false, the coordinated fallback is disabled for exactly the cases that would deadlock — left-emitting joins with a multi-partition right side — and those fall back to SpillState::Disabled (resource exhaustion under memory pressure) instead of coordinating, mirroring how enable_dynamic_filter_pushdown lets consumers opt out of another single-process assumption. Single-partition and non-left-emitting joins keep the coordinated fallback regardless.

I kept the default as true so single-process users (the common case) get the coordinated spill behavior without configuration, and distributed engines flip it off — but I don't have a strong opinion on the default and am happy to invert it if you think distributed-safety should be the default, given a hang is a harder failure than an OOM.

Previously the shared left spill stream and chunk reservation were lazily initialized at the top of `FallbackCoordinator::next_chunk`, before the `loader_in_flight` guard. While a leader was loading a chunk, a waiting partition could observe `left_stream.is_none()` and reopen the spill stream, which the leader would then overwrite — wasted I/O and state churn. Move the lazy initialization into the leader-claim branch (under the `loader_in_flight` guard), so only the leader sets up the stream, schema, and reservation. Waiters fall straight through to the notify wait path. Addresses a review suggestion on apache#22038. Co-authored-by: Claude Code

The existing multi-partition spill tests feed the left side as a single batch, so the memory-limited fallback only ever loads one chunk. That leaves the coordinator's multi-chunk machinery untested: `carryover` between chunks, `release_chunk` advancing to the next chunk, waiter notification, and (for FULL) the global right-unmatched bitmap accumulated across chunks. Add `*_multi_chunk_left_join` and `*_multi_chunk_full_join`: the left input is one row per batch under a tight memory limit, so the coordinator loads each row as a separate chunk (verified to load multiple chunks with spill_count > 0). Both assert exact row sets, so a regression in per-chunk `JoinLeftData` sharing (duplicate unmatched rows) or cross-chunk right accumulation would fail. These collect the four output partitions concurrently: a chunk is not released until all partitions finish probing it, so sequential collection would deadlock — concurrent collection mirrors how partitions run under the runtime. Also drop a stray `dbg!` in `join_inner_with_filter`. Addresses review feedback on apache#22038. Co-authored-by: Claude Code

…ails In `FallbackCoordinator::next_chunk`, the leader sets `loader_in_flight` before constructing the spill stream. If that construction returned an error via `?`, the leader would bail out while still holding `loader_in_flight = true` and without notifying waiters — leaving every other partition blocked forever on a chunk release that the failed leader can never make. `read_spill_as_stream` only builds the buffered stream today (real I/O errors surface later while polling, where the flag is already cleared), so this is not currently reachable. Handle it defensively anyway: on construction failure, clear `loader_in_flight`, drop the lock, wake waiters, and propagate the error so another partition can retry. Addresses a review suggestion on apache#22038. Co-authored-by: Claude Code

@andygrove

The memory-limited NestedLoopJoin fallback coordinates per-chunk left state (visited bitmap + probe-thread counter) across all right-side partitions via `FallbackCoordinator`. That coordination assumes every right partition runs in the same process. Distributed engines (Ballista, datafusion-distributed) execute each output partition as an independent task that builds its own coordinator and polls a single partition, so the shared probe-thread counter never reaches zero, no partition is elected to release the current chunk, and the waiters block forever — the memory-limited fallback deadlocks. Add `datafusion.execution.enable_nlj_coordinated_fallback` (default true, preserving current behavior). When set to false, the coordinated fallback is disabled for the join types that would deadlock — left-emitting joins (LEFT, LEFT SEMI, LEFT ANTI, LEFT MARK, FULL) with a multi-partition right side — which then use `SpillState::Disabled` and fail with resource exhaustion under memory pressure rather than deadlocking. Single-partition and non-left-emitting joins are always safe and keep the coordinated fallback regardless of the flag. Raised by @andygrove on apache#22038. Co-authored-by: Claude Code

github-actions · 2026-07-19T02:28:19Z

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-common v55.0.0 (current)
       Built [  35.617s] (current)
     Parsing datafusion-common v55.0.0 (current)
      Parsed [   0.062s] (current)
    Building datafusion-common v55.0.0 (baseline)
       Built [  34.814s] (baseline)
     Parsing datafusion-common v55.0.0 (baseline)
      Parsed [   0.061s] (baseline)
    Checking datafusion-common v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.737s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field ExecutionOptions.enable_nlj_coordinated_fallback in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:894

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  72.646s] datafusion-common
    Building datafusion-physical-plan v55.0.0 (current)
       Built [  39.198s] (current)
     Parsing datafusion-physical-plan v55.0.0 (current)
      Parsed [   0.151s] (current)
    Building datafusion-physical-plan v55.0.0 (baseline)
       Built [  38.838s] (baseline)
     Parsing datafusion-physical-plan v55.0.0 (baseline)
      Parsed [   0.152s] (baseline)
    Checking datafusion-physical-plan v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.747s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  80.457s] datafusion-physical-plan
    Building datafusion-sqllogictest v55.0.0 (current)
       Built [ 103.056s] (current)
     Parsing datafusion-sqllogictest v55.0.0 (current)
      Parsed [   0.023s] (current)
    Building datafusion-sqllogictest v55.0.0 (baseline)
       Built [ 102.945s] (baseline)
     Parsing datafusion-sqllogictest v55.0.0 (baseline)
      Parsed [   0.023s] (baseline)
    Checking datafusion-sqllogictest v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.092s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 208.686s] datafusion-sqllogictest

codecov-commenter · 2026-07-19T05:56:56Z

Codecov Report

❌ Patch coverage is 89.00000% with 77 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.53%. Comparing base (16cce96) to head (d237a30).
⚠️ Report is 12 commits behind head on main.

Files with missing lines	Patch %	Lines
...fusion/physical-plan/src/joins/nested_loop_join.rs	89.00%	43 Missing and 34 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #22038      +/-   ##
==========================================
+ Coverage   81.50%   81.53%   +0.03%     
==========================================
  Files        1123     1123              
  Lines      404763   406545    +1782     
  Branches   404763   406545    +1782     
==========================================
+ Hits       329898   331490    +1592     
- Misses      55553    55657     +104     
- Partials    19312    19398      +86

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

…joins with a multi-partition probe side (apache#24675) ## Which issue does this PR close? - Closes apache#22641. ## Rationale for this change The memory-limited fallback returns wrong rows for join types whose final emission reads the visited-left bitmap (LEFT, LEFT SEMI, LEFT ANTI, LEFT MARK) when the probe side has more than one partition. The fallback rebuilds a per-partition `JoinLeftData` per left chunk with `probe_threads_counter == 1` and a per-partition bitmap, so every partition believes it is the last to finish probing and emits its own unmatched set, from a bitmap that only saw its own right rows. Unmatched rows are emitted once per partition, and rows matched only in another partition are emitted as unmatched. Proof, using this file's own `build_table` / `prepare_join_filter` / `multi_partitioned_join_collect` fixtures (4 right partitions) with a 100-byte memory limit versus unbounded: ```rust for join_type in [JoinType::Left, JoinType::LeftAnti] { // unbounded memory -> correct; 100-byte pool -> memory-limited fallback let (_, correct, _) = multi_partitioned_join_collect(left, right, &join_type, Some(filter), ample_ctx).await?; let (_, wrong, _) = multi_partitioned_join_collect(left, right, &join_type, Some(filter), tight_ctx).await?; } // Left: correct=19 memory_limited=49 // LeftAnti: correct=1 memory_limited=31 ``` The existing gate covers only FULL. This extends it to every `need_produce_result_in_final` type, so an over-budget build side fails with `ResourcesExhausted` instead of returning wrong rows. Right-side emission types (RIGHT, RIGHT SEMI, RIGHT ANTI, RIGHT MARK) keep the fallback: each partition owns its right rows exclusively, so their bitmaps are complete per partition. Proper cross-partition coordination of the left bitmap, which would re-enable the fallback for these types, is apache#22038. ## What changes are included in this PR? The `full_join_multi_partition` condition in `NestedLoopJoinExec::execute` becomes `need_produce_result_in_final(self.join_type) && right_partition_count > 1`, with the comment updated to describe the failure mode. `test_overallocation` moves LEFT/LEFT SEMI/LEFT ANTI/LEFT MARK from the succeed-via-fallback group (which collected but never checked the row values) into the must-OOM group alongside FULL. ## Are these changes tested? `test_overallocation` now asserts the `ResourcesExhausted` refusal for all five gated join types with a multi-partition probe side, and still asserts fallback success for Inner and the right-emission types. All 42 `nested_loop_join` tests pass; `./dev/rust_lint.sh` is clean. ## Are there any user-facing changes? Left-family nested loop joins with a multi-partition probe side whose build side exceeds the memory budget now fail with `ResourcesExhausted` instead of returning incorrect results. No API changes.

…lback Rebase of the coordinated-fallback work onto current main. The memory-limited NestedLoopJoin fallback built a per-partition `JoinLeftData` with `probe_threads_counter == 1`. For join types whose final emission reads the visited-left bitmap (LEFT, LEFT SEMI, LEFT ANTI, LEFT MARK, FULL) each right partition therefore emitted from a bitmap that had only seen its own right rows, so unmatched rows came out once per partition and rows matched only in another partition came out as unmatched. Because those results are wrong, the fallback was refused for that combination and the query failed with ResourcesExhausted instead of spilling. `FallbackCoordinator` now loads each chunk once via a leader partition and publishes it as a shared `Arc<JoinLeftData>` whose probe-thread counter is seeded with `right_partition_count`, so the last partition to finish a chunk emits its unmatched left rows -- matching how the single-pass path coordinates through `collect_left_input(.., probe_threads_count)`. Those join types now spill instead of erroring. The coordination assumes all right partitions run in one process. Distributed engines run each partition as an independent task with its own coordinator, so the shared counter would never reach zero and the fallback would stall; `enable_nlj_coordinated_fallback = false` lets them opt out and keep the previous fail-fast behavior for the affected join types. Rebase notes: main since shares the spilled left side across partitions via `OnceAsync<LeftLoad>`, so the original `left_spill_fut`/`OnceFut` half of this change is dropped in favor of main's mechanism. The `EmitGlobalRightUnmatched` routing added in apache#24746 is preserved on the coordinator's exhausted-left path, keeping its three regression tests green. Co-authored-by: Claude Code

kosiew

@viirya,

Thanks for working on this. The cross-partition coordination approach looks promising, and the multi-partition and multi-chunk coverage is helpful.

I found one issue with the final chunk lifecycle that I think needs to be fixed before merging. The coordinator can retain the final chunk and its memory reservation after the query has completed if the physical plan remains alive. I left an inline comment with the details.

I also left one non-blocking suggestion for additional opt-out coverage.

kosiew · 2026-08-29T10:07:10Z

-                        // recomputed when `ProbeEnd` is re-entered for the next
-                        // chunk, so it does not need to be reset here.
                        self.state = NLJState::BufferingLeft;
                    } else if self.is_memory_limited()


I think the final chunk is never actually released here.

The release future is created above, but chunk_release_in_flight is only polled when we return to BufferingLeft or re-enter handle_emit_left_unmatched. That works for non-final chunks because they transition back to BufferingLeft. For the final chunk, though, a LEFT join goes directly to Done, while a FULL join goes to EmitGlobalRightUnmatched.

As a result, FallbackCoordinatorInner.current can keep its Arc<JoinLeftData> along with the coordinator reservation until the physical plan itself is dropped. If a completed query's plan stays alive, the final chunk's batch and bitmap remain accounted against the memory pool and could cause later queries to hit the memory limit.

Could we make sure the final chunk release is driven to completion before transitioning to Done or EmitGlobalRightUnmatched? It would also be good to add a regression that keeps the plan alive after collection and verifies that the coordinator reservation has been released.

Good catch, and confirmed — this was a real leak, not just a theoretical one.

I reproduced it before fixing: holding the plan alive after collecting every partition, the pool still had 1921 bytes reserved for both LEFT and FULL. Your reading of the control flow was exactly right: the release future is created, then the final chunk goes straight to Done / EmitGlobalRightUnmatched, neither of which polls chunk_release_in_flight.

Fixing it turned out to need two parts, the second of which I only found because the test still failed:

A new ReleasingFinalChunk state that polls the release future before finishing, then continues to whichever state would have followed. (My first attempt just re-entered EmitLeftUnmatched so its existing drain would run, but that re-runs process_left_unmatched() and panics with LeftData should be available, since buffered_left_data has already been cleared by then.)

release_chunk also has to resize(0) the coordinator reservation once the left side is exhausted. With only fix 1 the retained figure stayed at exactly 1921 — the per-chunk JoinLeftData carries just an empty RAII placeholder (reservation.new_empty()), so dropping the Arc frees nothing by itself; the real bytes live in the coordinator's reservation, which otherwise waits for a load_one_chunk that never comes after the final chunk.

Added test_nlj_memory_limited_releases_final_chunk_{left,full}_join, which keep the plan alive after collection and assert pool.reserved() == 0. Both fail without the change.

kosiew · 2026-08-29T10:07:10Z

+        Ok(())
+    }
+
+    /// When `enable_nlj_coordinated_fallback` is disabled, a LEFT join with


Could we also add an opt-out regression for one non-left-emitting join, such as RIGHT or INNER, under the same multi-partition memory pressure?

The documented contract says enable_nlj_coordinated_fallback = false only disables the coordinated fallback for left-emitting multi-partition joins. A small regression here would help ensure that a future change to this guard does not accidentally disable spill fallback for unaffected join types.

Agreed — added test_nlj_memory_limited_fallback_disabled_right_join_still_spills, covering both RIGHT and INNER under the same multi-partition memory pressure with enable_nlj_coordinated_fallback = false, asserting spill_count > 0. That pins the guard's scope so a future change can't quietly disable the spill fallback for join types that never needed left-bitmap coordination.

One thing worth recording from writing it: my first version drained the partitions sequentially (reusing multi_partition_memory_limited_join_collect) and hung. The coordinator seeds every chunk's probe counter with right_partition_count regardless of join type — with_visited_bitmap only controls whether a bitmap is allocated, not the counter — so with a multi-chunk left side, partition 0 waits on the other three to report before chunk 0 can be released, and collecting them one after another deadlocks. That's why all the existing multi-chunk coordinated tests use the concurrent helper; the new test does too, and I noted the reason in a comment so nobody reuses the sequential helper here.

It's a test-harness constraint rather than a product issue (a real plan always polls its partitions concurrently), but it does mean "RIGHT joins don't need coordination" is true of the bitmap and not of the chunk lockstep. Happy to make the counter join-type-aware in a follow-up if you think that lockstep is worth removing for non-left-emitting joins — it would let those partitions advance independently, but it's a behavioural change beyond this PR's scope.

Addresses review feedback on the coordinated fallback. `release_chunk` is what drops the coordinator's `Arc<JoinLeftData>` and returns the chunk's memory. Non-final chunks were released on the way back through `BufferingLeft`, but after the last chunk the stream went straight to `Done` (LEFT) or `EmitGlobalRightUnmatched` (FULL), and neither state polls `chunk_release_in_flight`. The coordinator hangs off the exec rather than the stream, so the final chunk's batch, bitmap and reservation stayed accounted against the memory pool for as long as the plan was alive -- enough to push a later query over the limit if a finished plan is still referenced. Two changes were needed. A new `ReleasingFinalChunk` state polls the release future before finishing, and `release_chunk` now also resizes the coordinator reservation to zero once the left side is exhausted: the per-chunk `JoinLeftData` carries only an empty RAII placeholder, so dropping the `Arc` frees nothing on its own, and the real bytes would otherwise wait for a `load_one_chunk` that never comes. Tests: `test_nlj_memory_limited_releases_final_chunk_{left,full}_join` keep the plan alive after collecting every partition and assert the pool is back to zero. Both fail without this change (1921 bytes retained). Also adds `test_nlj_memory_limited_fallback_disabled_right_join_still_spills`, pinning the scope of the `enable_nlj_coordinated_fallback` opt-out: RIGHT and INNER do not need left-bitmap coordination, so the opt-out must leave their spill fallback intact. It drains partitions concurrently because the coordinator seeds every chunk's probe counter with `right_partition_count` whatever the join type, so a multi-chunk left side cannot advance when the partitions are collected one after another. Co-authored-by: Claude Code

kosiew

@viirya,

Thanks for the updates here. The multi-chunk coordinator coverage, leader-only stream initialization, and opt-out coverage for RIGHT and INNER joins all look addressed.

I found two remaining issues that I think should be resolved before merging. The first is around the lifetime of the final chunk's memory reservation, and the second is the SemVer impact of adding a field to ExecutionOptions.

I left inline comments with the details. The focused NLJ tests I ran are passing.

kosiew · 2026-09-01T07:31:56Z

+            if inner.left_exhausted
+                && let Some(reservation) = inner.reservation.as_mut()
+            {
+                reservation.resize(0);


I think there is still a memory-accounting race here. release_chunk calls reservation.resize(0) as soon as the probe-counter emitter releases the final chunk, but that emitter is only the last stream to finish probing. It is not necessarily the last stream to drop its buffered_left_data reference.

For example, another partition can already be in EmitLeftUnmatched, flush a completed matched-output batch from maybe_flush_ready_batch, and return while still holding its Arc<JoinLeftData>. The emitter can then reach this cleanup and release the reservation even though that chunk's RecordBatch and bitmap are still live in the other partition. At that point the pool under-accounts the actual live memory and could allow the configured limit to be exceeded.

Could we keep the reservation charged until every partition has relinquished its reference to the chunk? One option would be a separate per-chunk release acknowledgement. Another would be to tie reservation ownership directly to the shared chunk data so its lifetime follows the data naturally.

It would also be good to add a scheduling-sensitive regression that holds a non-emitter after it flushes output while allowing the emitter to release the final chunk. That should catch this lifetime gap.

kosiew · 2026-09-01T07:31:56Z

+        /// with a resource-exhaustion error under memory pressure rather than
+        /// deadlocking. Single-partition and non-left-emitting joins are
+        /// unaffected and always keep the fallback.
+        pub enable_nlj_coordinated_fallback: bool, default = true


I think the SemVer concern raised by the bot still needs to be addressed. ExecutionOptions is public and can be constructed exhaustively, so adding this public field breaks downstream struct literals. cargo-semver-checks reports constructible_struct_adds_field for this field.

Could we use a configuration mechanism that does not add a field to the publicly constructible struct? Otherwise, if this API break is intentional, I think it needs to be explicitly approved and targeted for the appropriate major-version change.

github-actions Bot added sqllogictest SQL Logic Tests (.slt) physical-plan Changes to the physical-plan crate labels May 6, 2026

viirya force-pushed the nlj-multi-partition-unmatched-fix branch from a1a8bbe to 15e05fe Compare June 20, 2026 20:41

viirya requested a review from kosiew June 20, 2026 23:11

kosiew previously requested changes Jun 22, 2026

View reviewed changes

kosiew reviewed Jun 24, 2026

View reviewed changes

github-actions Bot added documentation Improvements or additions to documentation common Related to common crate labels Jul 18, 2026

viirya force-pushed the nlj-multi-partition-unmatched-fix branch from fdb63b4 to 8f2fd93 Compare July 19, 2026 02:19

github-actions Bot added the auto detected api change Auto detected API change label Jul 19, 2026

This was referenced Aug 25, 2026

fix: refuse memory-limited NestedLoopJoin fallback for left-emission joins with a multi-partition probe side #24675

Merged

NestedLoopJoin memory-limited fallback re-executes the build side, losing rows for non-replayable inputs #24676

Closed

viirya force-pushed the nlj-multi-partition-unmatched-fix branch from fa6902c to 03c8821 Compare August 28, 2026 23:32

viirya mentioned this pull request Aug 28, 2026

NestedLoopJoin memory-limited fallback lacks cross-partition left-bitmap coordination, so LEFT/FULL joins over a partitioned right side cannot spill #24764

Open

kosiew requested changes Aug 29, 2026

View reviewed changes

ranflarion mentioned this pull request Aug 31, 2026

NestedLoopJoin buffers the build side into a single concat_batches allocation: 2x transient peak, invisible to the memory pool #24819

Open

kosiew requested changes Sep 1, 2026

View reviewed changes

Uh oh!

Conversation

viirya commented May 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Rebase notes (2026-08-28)

Are these changes tested?

Are there any user-facing changes?

Uh oh!

kosiew left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya commented Jun 22, 2026

Uh oh!

kosiew left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya Jun 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kosiew commented Jun 24, 2026

Uh oh!

andygrove commented Jul 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

viirya commented Jul 18, 2026

Uh oh!

github-actions Bot commented Jul 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov-commenter commented Jul 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

kosiew left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kosiew left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

viirya commented May 6, 2026 •

edited

Loading

kosiew left a comment •

edited

Loading

kosiew left a comment •

edited

Loading

viirya Jun 24, 2026 •

edited

Loading

andygrove commented Jul 2, 2026 •

edited

Loading

github-actions Bot commented Jul 19, 2026 •

edited

Loading

codecov-commenter commented Jul 19, 2026 •

edited

Loading