kosiew commented on code in PR #22038:
URL: https://github.com/apache/datafusion/pull/22038#discussion_r3921273597
##########
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 is still a public API compatibility issue.
`enable_nlj_coordinated_fallback` adds a new public field to the exhaustively
constructible `ExecutionOptions` struct, so downstream code using
`ExecutionOptions { ... }` literals will stop compiling.
This matches the bot's `constructible_struct_adds_field` finding. I don't
think the precedent or advisory-CI argument changes the compatibility impact
here.
Could we avoid adding a new field directly to this public struct, or
otherwise get explicit approval for the breaking change, while still keeping a
distributed-safe way to configure the fallback?
##########
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 hole here.
Final-chunk cleanup depends on this async release future being polled to
completion.
If `release_chunk` returns `Pending` while waiting on the coordinator mutex,
and the consumer then drops the emitter stream before the next poll,
`chunk_release_in_flight` is dropped without `release_chunk` completing. Since
the coordinator is plan-owned, it can keep `current` and its
`Arc<JoinLeftData>` alive. With the reservation now owned by `JoinLeftData`,
that means the chunk can remain charged for as long as the plan stays alive.
Could we make this cleanup cancellation-safe? It would also be useful to add
a scheduling-sensitive regression that deliberately holds the coordinator 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 still
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]
