kosiew commented on code in PR #22038:
URL: https://github.com/apache/datafusion/pull/22038#discussion_r3924241103
##########
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 the remaining issue here is the API-health follow-up rather than
whether this field shape can be accepted.
This is still a valid `constructible_struct_adds_field` SemVer break because
downstream exhaustive `ExecutionOptions { ... }` literals will no longer
compile. I agree that the existing project precedent and advisory CI support
accepting this kind of config-field change, but that does not remove the
API-health requirements for an accepted break.
Could we add the `api-change` label and document the source migration in the
version-specific Upgrade Guide? It would be helpful for the guide to show
downstream users either how to add the new field explicitly or how to switch to
`..Default::default()`.
##########
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 depends on this in-stream async future completing. If
`release_chunk` returns `Pending` while waiting for the coordinator mutex, and
the consumer drops or cancels the emitter stream before the next poll,
`chunk_release_in_flight` is dropped without `release_chunk` completing.
Because the coordinator is owned by the plan, it can keep holding `current`
and its `Arc<JoinLeftData>`. Now that the reservation is owned by
`JoinLeftData`, the chunk can stay charged for as long as the plan remains
alive.
Could we make this cleanup cancellation-safe? I think a scheduling-sensitive
regression would also be valuable here: hold the mutex long enough for the
release future to become pending, drop the emitter stream while retaining the
plan, and verify that the chunk reservation is released.
--
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]
