You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/06/08 00:45:56 UTC
(datafusion) branch gh-readonly-queue/main/pr-22791-8036b94051daf3d3af93e8744b005bd4977dcd41 created (now d0d993d8bf)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-22791-8036b94051daf3d3af93e8744b005bd4977dcd41
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at d0d993d8bf fix: NestedLoopJoinExec emits spurious unmatched-left rows with multiple probe partitions (#22791)
This branch includes the following new commits:
new d0d993d8bf fix: NestedLoopJoinExec emits spurious unmatched-left rows with multiple probe partitions (#22791)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: NestedLoopJoinExec emits spurious unmatched-left rows with multiple probe partitions (#22791)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-22791-8036b94051daf3d3af93e8744b005bd4977dcd41
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit d0d993d8bfec447d8d809d30aaf862be12c4a55f
Author: Nathan <56...@users.noreply.github.com>
AuthorDate: Sun Jun 7 20:45:34 2026 -0400
fix: NestedLoopJoinExec emits spurious unmatched-left rows with multiple probe partitions (#22791)
## Problem
`NestedLoopJoinExec` can return incorrect and non deterministic results
for LEFT, RIGHT and FULL joins when the probe side has more than one
partition (the common case for any join whose condition is not a pure
equijoin). Some unmatched left rows are emitted twice: once correctly
and once again as an extra NULL padded row.
## Proof
Correct answer is 5 (every t1 row is unmatched). Verify with
`target_partitions = 1`. With multiple partitions an unpatched build
intermittently returns 6, 7 or 8. Because it is a scheduling race, run
it a few times to observe the divergence.
```sql
set datafusion.execution.target_partitions = 4;
set datafusion.execution.batch_size = 2;
create table t1(v bigint) as values (4),(72),(41),(98),(91);
create table t2(s varchar, w bigint) as values
('aaaaaaaaaaaaaaaa', 1),('bbbbbbbbbbbbbbbb', 2),('cccccccccccccccc', 3);
select count(*) as left_join_rows
from t1 left join t2 on (t1.v < t2.w and t2.s <= 'A');
```
## Solution
The probe streams share a `probe_threads_counter`; the stream that
drives it to 0 is the one that emits unmatched left rows, after all
partitions finish probing. In `handle_emit_left_unmatched`, after
`process_left_unmatched` returns `Ok(false)`,
`maybe_flush_ready_batch()` can return early with a ready batch before
the state advances to `Done`. The stream stays in `EmitLeftUnmatched`,
so the next poll re enters `process_left_unmatched` with `left_emit_idx
== 0` and decrements the counter a second time. The counter then reaches
0 before every partition has finished probing, so a partition emits
unmatched left rows early.
The fix decrements the counter at most once per probe stream using a new
`probe_completed_reported` flag (reset per chunk in the memory limited
path). All 42 existing `nested_loop_join` tests pass.
This is separate from #22641, which covers the memory limited spill
fallback path; this bug reproduces with unbounded memory and with
spilling disabled.
---
.../physical-plan/src/joins/nested_loop_join.rs | 44 +++++++++++++++++++---
datafusion/sqllogictest/test_files/joins.slt | 44 ++++++++++++++++++++++
2 files changed, 83 insertions(+), 5 deletions(-)
diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs
index a18ec0cbe4..0bd053a9db 100644
--- a/datafusion/physical-plan/src/joins/nested_loop_join.rs
+++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs
@@ -1064,6 +1064,17 @@ pub(crate) struct NestedLoopJoinStream {
/// Memory-limited spill fallback state. See [`SpillState`] for details.
spill_state: SpillState,
+
+ /// Whether this stream has already reported probe completion for the current
+ /// left chunk via [`JoinLeftData::report_probe_completed`]. The shared
+ /// probe-threads counter must be decremented exactly once per probe stream;
+ /// without this guard a stream that yields a ready batch while finishing the
+ /// `EmitLeftUnmatched` state (and is then re-polled with `left_emit_idx`
+ /// still 0) would decrement the counter twice, driving it to zero
+ /// prematurely and causing a sibling partition to emit unmatched-left rows
+ /// before all partitions finished probing (spurious NULL-padded rows).
+ /// Reset to `false` when starting a new left chunk in memory-limited mode.
+ probe_completed_reported: bool,
}
pub(crate) struct NestedLoopJoinMetrics {
@@ -1337,6 +1348,7 @@ impl NestedLoopJoinStream {
handled_empty_output: false,
should_track_unmatched_right: need_produce_right_in_final(join_type),
spill_state,
+ probe_completed_reported: false,
}
}
@@ -1863,6 +1875,10 @@ impl NestedLoopJoinStream {
self.buffered_left_data = None;
self.left_probe_idx = 0;
self.left_emit_idx = 0;
+ // Each memory-limited chunk gets a fresh per-chunk
+ // `JoinLeftData`/counter, so allow this stream to report
+ // completion again for the next chunk.
+ self.probe_completed_reported = false;
self.state = NLJState::BufferingLeft;
} else if self.is_memory_limited()
&& self.should_track_unmatched_right
@@ -2341,7 +2357,9 @@ impl NestedLoopJoinStream {
/// true -> continue in the same EmitLeftUnmatched state
/// false -> next state (Done)
fn process_left_unmatched(&mut self) -> Result<bool> {
- let left_data = self.get_left_data()?;
+ // Clone the shared `Arc<JoinLeftData>` so the immutable borrow of `self`
+ // ends here and we can update `self.probe_completed_reported` below.
+ let left_data = Arc::clone(self.get_left_data()?);
let left_batch = left_data.batch();
// ========
@@ -2350,9 +2368,25 @@ 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
- let handled_by_other_partition =
- self.left_emit_idx == 0 && !left_data.report_probe_completed();
+ // 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();
@@ -2368,7 +2402,7 @@ impl NestedLoopJoinStream {
let end_idx = std::cmp::min(start_idx + self.batch_size, left_batch.num_rows());
if let Some(batch) =
- self.process_left_unmatched_range(left_data, start_idx, end_idx)?
+ self.process_left_unmatched_range(&left_data, start_idx, end_idx)?
{
self.output_buffer.push_batch(batch)?;
}
diff --git a/datafusion/sqllogictest/test_files/joins.slt b/datafusion/sqllogictest/test_files/joins.slt
index e0be63fe71..b037aef3c2 100644
--- a/datafusion/sqllogictest/test_files/joins.slt
+++ b/datafusion/sqllogictest/test_files/joins.slt
@@ -5527,3 +5527,47 @@ DROP TABLE t1;
statement ok
DROP TABLE t2;
+
+# Regression test for a LEFT JOIN with a non-equijoin predicate (forces
+# NestedLoopJoinExec) and a multi-partition probe side. Previously the unmatched
+# left rows could be emitted before all partitions finished probing, adding
+# spurious NULL-padded rows. The result must include every left row exactly once.
+statement ok
+set datafusion.execution.target_partitions = 4;
+
+statement ok
+set datafusion.execution.batch_size = 2;
+
+statement ok
+CREATE TABLE nlj_left(id INT, v INT) AS VALUES (1, 4), (2, 72), (3, 41), (4, 98), (5, 91);
+
+statement ok
+CREATE TABLE nlj_right(w INT) AS VALUES (49), (58), (83), (3), (76);
+
+query III
+SELECT id, v, w FROM nlj_left LEFT JOIN nlj_right ON nlj_left.v < nlj_right.w ORDER BY id, w;
+----
+1 4 49
+1 4 58
+1 4 76
+1 4 83
+2 72 76
+2 72 83
+3 41 49
+3 41 58
+3 41 76
+3 41 83
+4 98 NULL
+5 91 NULL
+
+statement ok
+DROP TABLE nlj_left;
+
+statement ok
+DROP TABLE nlj_right;
+
+statement ok
+set datafusion.execution.target_partitions = 4;
+
+statement ok
+reset datafusion.execution.batch_size;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org