viirya opened a new issue, #25003:
URL: https://github.com/apache/datafusion/issues/25003
### Describe the bug
In the coordinated memory-limited `NestedLoopJoinExec` fallback, dropping an
unfinished output partition can strand the shared chunk and leave the other
partitions waiting indefinitely.
Chunk advancement requires every probing partition to report: the last one
to call `JoinLeftData::report_probe_completed` becomes the emitter and releases
the coordinator slot. A partition that goes away before finishing never
reports, so no emitter is elected, nothing releases the slot, and a survivor
asking for the next chunk finds
- Case 1 does not match (the slot still holds the previous chunk),
- Case 2 requires `current.is_none()`, so it does not fire,
- Case 3 requires the same, so it does not fire,
and falls through to the `notified()` wait in
`FallbackCoordinator::next_chunk`, where nothing will ever wake it.
There is a second effect on the same cause. The coordinator is owned by
`NestedLoopJoinExec` rather than by the streams, so if the plan outlives them
the slot keeps its `Arc<JoinLeftData>` — and since the reservation now lives
inside `JoinLeftData`, the chunk stays charged against the memory pool for as
long as the plan is alive.
Two shapes reach this:
1. **Release interrupted.** The emitter's release runs as a future stored in
`chunk_release_in_flight` and polled across `poll_next` calls. If it is pending
on the coordinator mutex when the stream is dropped, the future goes with it
and the slot is never cleared.
2. **No emitter at all.** If the streams are cancelled mid-probe, nobody
drives the probe counter to zero, so `release_chunk` is never called in the
first place. This is why making the release future cancellation-safe cannot fix
the problem on its own.
### To Reproduce
Driving the coordinator directly, with two nominal probe partitions so a
single holder never drives the counter to zero:
```rust
let coordinator = Arc::new(FallbackCoordinator::new(2, true));
let spill = spill_left_for_test(build_left_table(), Arc::clone(&ctx)).await?;
let (chunk, _) = Arc::clone(&coordinator)
.next_chunk(0, Arc::clone(&spill), Arc::clone(&ctx), Time::new())
.await?
.expect("chunk 0");
drop(chunk); // partition cancelled
mid-probe
// A survivor asking for the next chunk never returns.
tokio::time::timeout(
Duration::from_secs(3),
Arc::clone(&coordinator).next_chunk(1, spill, ctx, Time::new()),
).await // Err(Elapsed)
```
The retention half is observable as 505 bytes still reserved after every
stream-side reference is gone.
### Expected behavior
A partition disappearing before it finishes should not leave its peers
waiting forever. The coordinated execution has lost a partition and can no
longer produce a complete result, so the remaining partitions should fail
rather than hang — or report success from partial input.
### Additional context
Introduced with the coordinated fallback in #22038. It was found during that
review and deliberately deferred: at the time only the memory retention was
understood, which needs a caller that retains an `Arc<dyn ExecutionPlan>`
across execution — something `DataFrame::collect` and `execute_stream` never
do, since they build the plan as a local and drop it. The hang does not need
that condition, which is why this is worth fixing rather than documenting.
Note that the existing
`datafusion.execution.enable_nlj_coordinated_fallback` opt-out does not avoid
this in general: it disables the coordinated fallback only for multi-partition
joins that require final left-side emission, so single-partition and other join
types still take the coordinated path.
--
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]
