kosiew commented on code in PR #22038:
URL: https://github.com/apache/datafusion/pull/22038#discussion_r3933649450
##########
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:
There is still a cancellation cleanup edge case here, but after the
follow-up discussion I do not think it needs to block this PR.
If `release_chunk` is pending on `inner.lock()` and this stream is dropped,
`chunk_release_in_flight` is dropped with it. If the caller keeps the
`NestedLoopJoinExec` alive, the coordinator can then keep `current` and its
`Arc<JoinLeftData>` alive for the lifetime of the plan. There is also a broader
mid-probe version of the same issue: if streams are dropped before anyone
reaches `ProbeEnd`, no emitter is elected and no release future exists at all.
I agree it is better to track this as a follow-up rather than change the
coordinator ordering again here. The normal completion path is fixed and
tested, and `current` is also part of the synchronization protocol for
advancing every partition to the next chunk or to termination, so clearing it
earlier is risky.
--
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]
