kosiew commented on code in PR #22038:
URL: https://github.com/apache/datafusion/pull/22038#discussion_r3921309513
##########
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:
I think there is still a cancellation-sensitive cleanup gap here.
Final-chunk cleanup currently depends on this in-stream async future
completing. If `release_chunk` returns `Pending` while waiting on the
coordinator mutex, and the consumer then drops or cancels this emitter stream
before it is polled again, `chunk_release_in_flight` is dropped without
`release_chunk` completing.
Because the coordinator is owned by the plan, it can continue holding
`current` and its `Arc<JoinLeftData>`. With the reservation now owned by
`JoinLeftData`, that means the chunk can remain charged for as long as the plan
remains alive.
Could we make this cleanup cancellation-safe? I think it would also be
useful to add a scheduling-sensitive regression that holds the mutex long
enough for the release future to become pending, drops the emitter stream while
retaining the plan, and verifies that the chunk reservation is released.
##########
datafusion/common/src/config.rs:
##########
@@ -1031,6 +1031,25 @@ config_namespace! {
/// Default: 128 MB
pub max_spill_file_size_bytes: ConfigNonZeroUsize, default =
non_zero_usize_default(128 * 1024 * 1024)
+ /// Enables the memory-limited fallback for `NestedLoopJoinExec` join
+ /// types that emit unmatched left rows in the final output (LEFT, LEFT
+ /// SEMI, LEFT ANTI, LEFT MARK, FULL) when the right side has multiple
+ /// partitions.
+ ///
+ /// This fallback coordinates per-chunk left state (visited bitmap and
+ /// probe-thread counter) across all right-side partitions, which
+ /// assumes every partition runs in the same process. Distributed
+ /// engines that execute each output partition as an independent task
+ /// (e.g. Ballista, datafusion-distributed) build a separate
coordinator
+ /// per task and poll only one partition, so the cross-partition
+ /// counter never reaches zero and the fallback would stall. Such
+ /// engines should set this to `false`: the coordinated fallback is
then
+ /// disabled for left-emitting multi-partition joins, which instead
fail
+ /// with a resource-exhaustion error under memory pressure rather than
+ /// deadlocking. Single-partition and non-left-emitting joins are
+ /// unaffected and always keep the fallback.
+ pub enable_nlj_coordinated_fallback: bool, default = true
Review Comment:
I think this still needs a different approach.
`enable_nlj_coordinated_fallback` adds a new public field to the exhaustively
constructible `ExecutionOptions` struct, so downstream code using
`ExecutionOptions { ... }` literals will stop compiling.
That matches the bot's `constructible_struct_adds_field` finding. I don't
think the existing precedent or the advisory nature of that CI check changes
the fact that this is a public API break.
Could we avoid adding a new field directly to this public struct, or
otherwise get explicit approval for the SemVer break, while still keeping a
distributed-safe way to configure this behavior?
--
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]
