NestedLoopJoin coordinated fallback hangs surviving partitions when one is dropped unfinished

[viirya]

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:

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.