viirya commented on code in PR #22038:
URL: https://github.com/apache/datafusion/pull/22038#discussion_r3927713215
##########
datafusion/physical-plan/src/joins/nested_loop_join.rs:
##########
@@ -2252,29 +2668,60 @@ impl NestedLoopJoinStream {
return ControlFlow::Break(poll);
}
- if !self.left_exhausted && self.is_memory_limited() {
- // More left data to process — free current chunk and
- // go back to BufferingLeft for the next chunk
- if let SpillState::Active(ref active) =
self.spill_state {
- active.reservation.resize(0);
+ // Drop our reference to the current chunk's
+ // `JoinLeftData` before releasing the slot. Once the
+ // last partition does this, the `Arc` reaches zero
+ // refcount and the per-chunk reservation is freed.
+ self.buffered_left_data = None;
+
+ if self.is_memory_limited() {
+ let is_emitter = self.is_unmatched_left_emitter;
+ if let SpillState::Active(active) = &mut
self.spill_state {
+ // The last partition for this chunk (the
+ // unmatched-left emitter elected in `ProbeEnd`)
+ // releases the coordinator slot so the next
+ // leader can load the following chunk.
+ if is_emitter {
+ let coordinator =
Arc::clone(&active.coordinator);
+ let released_index = active.next_chunk_index;
+ active.chunk_release_in_flight = Some(
Review Comment:
Thanks — you're right that the release future can be dropped while pending,
and that the coordinator would then keep `current` and its `Arc<JoinLeftData>`.
Looking at what it takes to actually reach that, though, I think it needs
something to keep the physical plan alive after the streams are gone, and the
normal execution paths don't do that. `DataFrame::collect` and `execute_stream`
build the plan as a local, hand it to `plan.execute(..)`, and drop it — the
returned `SendableRecordBatchStream` doesn't hold the plan. So when a query is
cancelled the plan goes away with it, `fallback_coordinator` reaches zero
refcount, and the chunk it was holding is released along with it. No retention.
That leaves a caller holding an `Arc<dyn ExecutionPlan>` across execution
and cancelling midway. Two things make that hard to reach: the `DataFrame` API
never hands the physical plan to users, and re-executing this operator isn't
really supported anyway — `fallback_coordinator` is created in
`try_new`/`replace_children` and shared across `execute()` calls, so its
`left_exhausted` / `next_chunk_index` / `carryover` persist, much like the
existing `build_side_data: OnceAsync<LeftLoad>` beside it ("shared across all
output streams"). A second execution would not get any chunks regardless of
this.
There's also a shape that making the release cancellation-safe would not
cover: if the streams are cancelled mid-probe, no partition drives the probe
counter to zero, so no emitter is elected and `release_chunk` is never called
at all. Covering both really means changing how the coordinator tracks the
current slot, not just hardening the release path.
So my read is that the retention is real in principle but not reachable
through the supported paths, and that the proper fix is a bigger change than
this PR should carry. I'd prefer to handle it separately rather than grow this
further. Does that seem reasonable, or do you see a path where this is
reachable in normal use that I'm missing?
--
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]
