You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "nathanb9 (via GitHub)" <gi...@apache.org> on 2026/06/10 01:08:57 UTC
[PR] refactor: introduce ProbeEnd state in NestedLoopJoinExec [datafusion]
nathanb9 opened a new pull request, #22865:
URL: https://github.com/apache/datafusion/pull/22865
## Which issue does this PR close?
- Closes #22808.
## Rationale for this change
Follow-up to #22791, as suggested in review by @2010YOUY01.
That PR fixed a double-decrement bug where `EmitLeftUnmatched` did two jobs at once — deciding whether a partition emits unmatched-left rows (which decrements the shared `probe_threads_counter`) and performing the emit. Because the state is re-enterable (a ready batch can be flushed before the state advances to `Done`), the counter could be decremented twice, driving it to zero before all partitions finished probing and emitting spurious NULL-padded rows. #22791 patched this with a `probe_completed_reported` guard flag.
This refactor makes "decrement exactly once per probe stream" a structural property of the state graph rather than a runtime guard, so the inner logic is easier to follow and the bug is harder to reintroduce.
## What changes are included in this PR?
Restructures the state machine from `FetchingRight → EmitLeftUnmatched` to `FetchingRight → ProbeEnd → EmitLeftUnmatched`:
- Adds a dedicated `ProbeEnd` state, entered exactly once per left chunk when the right side is exhausted. It owns the single `report_probe_completed()` call and records whether this stream is the unmatched-left emitter.
- Replaces the `probe_completed_reported` guard flag with an `is_unmatched_left_emitter` field that `EmitLeftUnmatched` only reads.
- Removes the per-chunk flag reset in the memory-limited path (the decision is recomputed in `ProbeEnd` for each chunk) and reverts the `Arc::clone` workaround #22791 needed in `process_left_unmatched`.
- Updates the state-transition doc graph and arm comments.
No behavior change is expected.
## Are these changes tested?
Yes — covered by existing tests:
- All 42 `nested_loop_join` unit tests and the full `datafusion-physical-plan` suite pass.
- `joins.slt` sqllogictests pass (including the multi-partition LEFT JOIN regression test added in #22791).
- 41 `join_fuzz` tests (`cargo test --features extended_tests`) comparing `NestedLoopJoinExec` against `HashJoinExec` across every join type, filtered and unfiltered, with a multi-partition probe side — the exact scenario class of the original bug — pass.
- `cargo fmt` and `cargo clippy --all-targets --all-features -- -D warnings` are clean.
## Are there any user-facing changes?
No.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] refactor: introduce ProbeEnd state in NestedLoopJoinExec [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #22865:
URL: https://github.com/apache/datafusion/pull/22865#discussion_r3392962858
##########
datafusion/physical-plan/src/joins/nested_loop_join.rs:
##########
@@ -2368,29 +2434,11 @@ impl NestedLoopJoinStream {
// Early return if join type can't have unmatched rows
let join_type_no_produce_left = !need_produce_result_in_final(self.join_type);
- // Early return if another thread is already processing unmatched rows.
- //
- // The shared probe-threads counter must be decremented exactly once per
- // probe stream. This function can be re-entered with `left_emit_idx`
- // still 0 (e.g. when a ready batch was flushed via an early return in
- // `handle_emit_left_unmatched` before the state advanced), so guard the
- // decrement with `probe_completed_reported` instead of relying solely on
- // `left_emit_idx == 0`. Decrementing twice would drive the counter to
- // zero prematurely and let a partition emit unmatched-left rows before
- // all partitions finished probing, producing spurious NULL-padded rows.
- let handled_by_other_partition = if self.probe_completed_reported {
- // Already counted this stream's completion; if we're the designated
- // emitter we have `left_emit_idx > 0` (or are mid-emit) and continue,
- // otherwise another partition is handling emission.
- self.left_emit_idx == 0
- } else {
- self.probe_completed_reported = true;
- self.left_emit_idx == 0 && !left_data.report_probe_completed()
- };
// Stop processing unmatched rows, the caller will go to the next state
let finished = self.left_emit_idx >= left_batch.num_rows();
- if join_type_no_produce_left || handled_by_other_partition || finished {
+ // `ProbeEnd` already recorded whether this stream emits unmatched-left rows.
+ if join_type_no_produce_left || !self.is_unmatched_left_emitter || finished {
Review Comment:
Here `self.is_unmatched_left_emitter` seems like always true, if a partition has entered the current `emit left` state? If it's true, we could instead `debug_assert(self.is_unmatched_left_emitter)` here
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] refactor: introduce ProbeEnd state in NestedLoopJoinExec [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew merged PR #22865:
URL: https://github.com/apache/datafusion/pull/22865
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] refactor: introduce ProbeEnd state in NestedLoopJoinExec [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22865:
URL: https://github.com/apache/datafusion/pull/22865#discussion_r3386024891
##########
datafusion/physical-plan/src/joins/nested_loop_join.rs:
##########
@@ -2368,25 +2434,20 @@ impl NestedLoopJoinStream {
// Early return if join type can't have unmatched rows
let join_type_no_produce_left = !need_produce_result_in_final(self.join_type);
- // Early return if another thread is already processing unmatched rows.
+ // Early return if another partition is the designated emitter.
Review Comment:
Nice simplification. Since the invariant is already captured by `ProbeEnd` and documented on the state/helper, I wonder if this comment and the negated temporary could be trimmed down a bit.
Something along these lines seems a little easier to scan:
```rust
// `ProbeEnd` already recorded whether this stream emits unmatched-left rows.
let finished = self.left_emit_idx >= left_batch.num_rows();
if join_type_no_produce_left || !self.is_unmatched_left_emitter || finished {
return Ok(false);
}
```
That keeps the emit state focused on the local decision while preserving the existing 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] refactor: introduce ProbeEnd state in NestedLoopJoinExec [datafusion]
Posted by "nathanb9 (via GitHub)" <gi...@apache.org>.
nathanb9 commented on code in PR #22865:
URL: https://github.com/apache/datafusion/pull/22865#discussion_r3400280199
##########
datafusion/physical-plan/src/joins/nested_loop_join.rs:
##########
@@ -2368,29 +2434,11 @@ impl NestedLoopJoinStream {
// Early return if join type can't have unmatched rows
let join_type_no_produce_left = !need_produce_result_in_final(self.join_type);
- // Early return if another thread is already processing unmatched rows.
- //
- // The shared probe-threads counter must be decremented exactly once per
- // probe stream. This function can be re-entered with `left_emit_idx`
- // still 0 (e.g. when a ready batch was flushed via an early return in
- // `handle_emit_left_unmatched` before the state advanced), so guard the
- // decrement with `probe_completed_reported` instead of relying solely on
- // `left_emit_idx == 0`. Decrementing twice would drive the counter to
- // zero prematurely and let a partition emit unmatched-left rows before
- // all partitions finished probing, producing spurious NULL-padded rows.
- let handled_by_other_partition = if self.probe_completed_reported {
- // Already counted this stream's completion; if we're the designated
- // emitter we have `left_emit_idx > 0` (or are mid-emit) and continue,
- // otherwise another partition is handling emission.
- self.left_emit_idx == 0
- } else {
- self.probe_completed_reported = true;
- self.left_emit_idx == 0 && !left_data.report_probe_completed()
- };
// Stop processing unmatched rows, the caller will go to the next state
let finished = self.left_emit_idx >= left_batch.num_rows();
- if join_type_no_produce_left || handled_by_other_partition || finished {
+ // `ProbeEnd` already recorded whether this stream emits unmatched-left rows.
+ if join_type_no_produce_left || !self.is_unmatched_left_emitter || finished {
Review Comment:
not always true bc every partition passes through this state but only the last to finish is the emitter, so the flag is false for the others.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org