Skip to content

fix: spill the NestedLoopJoin build side from the pass that consumed it - #24677

Open
ranflarion wants to merge 1 commit into
apache:mainfrom
ranflarion:nlj-spill-from-load
Open

fix: spill the NestedLoopJoin build side from the pass that consumed it#24677
ranflarion wants to merge 1 commit into
apache:mainfrom
ranflarion:nlj-spill-from-load

Conversation

@ranflarion

Copy link
Copy Markdown

Which issue does this PR close?

Rationale for this change

The memory-limited fallback re-executes the build-side child: when the in-memory load fails with ResourcesExhausted, initiate_fallback calls self.left.execute(0, ctx) a second time and spills that stream — but the first execution already consumed part of the input. For a child that cannot be executed twice, the consumed batches are gone.

Proof on current main, wrapping this file's own build_left_table() in an ExecutionPlan whose second execute returns an empty stream (the way an exhausted external reader behaves), under the 50-byte limit the existing memory-limited tests use:

executions=2  result=Err(Internal("Left side produced no data to spill"))

For a child whose stream resumes instead of restarting, the same shape loses the consumed prefix silently: embedding DataFusion 54.1.0 over a JVM-fed source we measured a 200k-row build side returning 134,464 rows with no error. Replayable inputs are also executed twice for no benefit.

This PR moves the spill into the load: when a reservation fails inside collect_left_input, the batches buffered so far plus the remainder of the same stream are written to one spill file, and the shared OnceAsync now resolves to a LeftLoad enum (InMemory(JoinLeftData) or Spilled(LeftSpillData)) that every partition consumes. The build side is executed exactly once whether or not it fits, SpillState::Pending no longer carries the child plan, and an empty build side stays on the in-memory path instead of erroring as "no data to spill".

What changes are included in this PR?

collect_left_input takes an optional SpillManager (built up front, since by the time the limit is hit the stream is already partly consumed) and returns LeftLoad; a new spill_left_input writes buffered batches plus the stream remainder; initiate_fallback becomes enter_memory_limited_mode(Arc<LeftSpillData>); the per-partition chunk reader opens the shared spill file directly instead of awaiting a spill future; the load reservation is created with_can_spill(can_spill) since the consumer now genuinely spills.

Are these changes tested?

New test memory_limited_left_side_reads_the_child_once: a one-shot child under the tight limit must produce the full result with spill_count > 0 and exactly one execute of the build side — on unpatched main this scenario fails as shown above. All 43 nested_loop_join tests pass (the existing memory-limited tests cover every join type through the new path), 1783 crate tests pass, ./dev/rust_lint.sh is clean.

Are there any user-facing changes?

Memory-limited nested loop joins over non-replayable inputs now return complete results instead of failing (or silently truncating), and the build side is no longer executed twice. No API changes; LeftLoad/LeftSpillData are pub(crate).

@github-actions github-actions Bot added the physical-plan Changes to the physical-plan crate label Aug 25, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

physical-plan Changes to the physical-plan crate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

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

1 participant