ranflarion opened a new pull request, #24677:
URL: https://github.com/apache/datafusion/pull/24677
## Which issue does this PR close?
- Closes #24676.
## 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)`.
--
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]
