This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 95bb0a0dfa fix: correlated NOT IN with a non-equality correlation
returns wrong results (#25339)
95bb0a0dfa is described below
commit 95bb0a0dfa48ca20d017faa55b2fc96ee2f300fe
Author: Adrian Garcia Badaracco <[email protected]>
AuthorDate: Wed Sep 23 21:55:24 2026 +0000
fix: correlated NOT IN with a non-equality correlation returns wrong
results (#25339)
## Which issue does this PR close?
- Closes https://github.com/apache/datafusion/issues/25336.
> [!NOTE]
> **This PR now holds only the executor fix.** It used to hold the
executor fix, the optimizer fix, and their tests together. To make
review easier, I split it:
>
> 1. Tests and benchmarks:
https://github.com/apache/datafusion/pull/25558 (merged). They record
the wrong results on `main`.
> 2. Executor fix: this PR. It flips the expectations that it fixes.
> 3. Optimizer fix: https://github.com/apache/datafusion/pull/25560,
stacked on this PR.
>
> The executor code is the same as at 78b49ba214 (the last head that was
reviewed), less one dead line. The review threads on
`decorrelate_predicate_subquery.rs` ([plain
`IN`](https://github.com/apache/datafusion/pull/25339#discussion_r4046914195),
[`InSubquery`
value](https://github.com/apache/datafusion/pull/25339#discussion_r4056392706)),
the [`on[0]`
thread](https://github.com/apache/datafusion/pull/25339#discussion_r4027554354)
and the [constant-projection
comment](https://github.com/apache/datafusion/pull/25339#issuecomment-5738402844)
are fixed in #25560.
## Rationale for this change
A correlated `NOT IN` whose correlation cannot become an equi-join key
leaves a residual join filter. The null-aware hash join ignored that
filter when deciding whether a NULL on the subquery side makes `NOT IN`
UNKNOWN, so a NULL the filter excludes still poisoned every outer row:
```sql
CREATE TABLE oc(id INT, g INT) AS VALUES
(1,5),(2,5),(3,0),(4,NULL),(NULL,5),(NULL,0);
CREATE TABLE ic(id INT) AS VALUES (1),(NULL);
SELECT id, g FROM oc WHERE oc.id NOT IN (SELECT ic.id FROM ic WHERE oc.g >
0);
```
returns no rows; DuckDB and PostgreSQL return `3|0`, `4|NULL`, `NULL|0`.
`oc.g > 0` holds only for `id` 1, 2 and the NULL-id row, so only those
three see the subquery `{1, NULL}`; the rest see an empty subquery, and
`NOT IN` over an empty set is TRUE.
The plan was already right — `LeftAnti ... Filter: oc.g > Int32(0)
null_aware` — so this is purely an execution fix. No optimizer change is
involved.
## What changes are included in this PR?
A NULL now makes `NOT IN` UNKNOWN only for the build rows whose
correlation scope and residual filter keep that NULL, recorded per build
row in a null-indices bitmap. Candidates come from a scope-map lookup
when there are correlation keys and from a cross product otherwise, then
pass the filter. Cost is proportional to the number of NULLs and is zero
when the data has none; build rows already marked UNKNOWN are skipped.
Null-aware `LeftAnti` also accepts more than one join key, which the
equality-correlated shape needs. `RightAnti` still requires exactly one.
## What is the testing strategy for this PR?
The coverage landed in #25558. This PR flips the expectations it fixes —
the diff in `null_aware_anti_join.slt` and the Q05–Q07 canaries is the
behaviour change. A "pinned to today's behaviour" note is removed only
where the expectation below it changes; the notes on shapes that #25560
fixes stay. `datafusion/physical-plan/src/joins/hash_join/exec.rs` also
gains unit tests for the filter-only anti and mark paths at several
batch sizes.
## Are there any user-facing changes?
Correlated `NOT IN` with a residual filter returns correct results. Some
shapes that failed to plan now run.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-authored-by: Adrian Garcia Badaracco <[email protected]>
Co-authored-by: Claude Opus 5 <[email protected]>
---
.../null_aware_join/benchmarks/q05.benchmark | 4 +-
.../null_aware_join/benchmarks/q06.benchmark | 4 +-
.../null_aware_join/benchmarks/q07.benchmark | 4 +-
.../physical-optimizer/src/join_selection.rs | 4 +-
.../physical-plan/src/joins/hash_join/exec.rs | 340 ++++++++++++++-----
.../physical-plan/src/joins/hash_join/stream.rs | 376 +++++++++++++++------
.../test_files/null_aware_anti_join.slt | 54 ++-
7 files changed, 568 insertions(+), 218 deletions(-)
diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q05.benchmark
b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q05.benchmark
index bde1f3e565..54fdaba978 100644
--- a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q05.benchmark
+++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q05.benchmark
@@ -15,9 +15,7 @@ SELECT count(*) = (
FROM small_outer o
WHERE o.id_n1 NOT IN (SELECT i.id_n0 FROM small_inner i WHERE i.z < o.z);
----
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
-false
+true
expect_plan HashJoinExec
expect_plan null_aware
diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q06.benchmark
b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q06.benchmark
index 25929198e7..e841f25df5 100644
--- a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q06.benchmark
+++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q06.benchmark
@@ -15,9 +15,7 @@ SELECT count(*) = (
FROM small_outer o
WHERE o.id_n50 NOT IN (SELECT i.id_n0 FROM small_inner i WHERE i.z < o.z);
----
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
-false
+true
expect_plan HashJoinExec
expect_plan null_aware
diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q07.benchmark
b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q07.benchmark
index 3364be5513..7b92e60e40 100644
--- a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q07.benchmark
+++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q07.benchmark
@@ -17,9 +17,7 @@ SELECT count(*) = (
FROM small_outer o
WHERE o.id_n0 NOT IN (SELECT i.id_n50 FROM small_inner i WHERE i.z < o.z);
----
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
-false
+true
expect_plan HashJoinExec
expect_plan null_aware
diff --git a/datafusion/physical-optimizer/src/join_selection.rs
b/datafusion/physical-optimizer/src/join_selection.rs
index f11e8612de..2b342bece7 100644
--- a/datafusion/physical-optimizer/src/join_selection.rs
+++ b/datafusion/physical-optimizer/src/join_selection.rs
@@ -172,11 +172,13 @@ impl PhysicalOptimizerRule for JoinSelection {
}
}
-/// Determines whether it is possible to swap inputs of a hash join - for
null-aware joins, we can only swap `LeftAnti` with no filters
+/// Determines whether it is possible to swap inputs of a hash join - for
null-aware joins, we can only swap an uncorrelated `LeftAnti`
+/// (a single join key and no filter), because the swapped `RightAnti` has no
per-row NULL handling
fn can_swap_hash_join(hash_join: &HashJoinExec) -> bool {
hash_join.join_type().supports_swap()
&& (!hash_join.null_aware
|| (*hash_join.join_type() == JoinType::LeftAnti
+ && hash_join.on().len() == 1
&& hash_join.filter().is_none()))
}
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs
b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index 9858a4c06c..d2fcf5191d 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -192,19 +192,20 @@ fn try_create_array_map(
Ok(Some((array_map, batch, left_values)))
}
-/// Correlation-scope hash map over only the build rows whose scalar `NOT IN`
-/// value key is NULL, used by correlated null-aware `LeftMark` joins.
+/// The build rows whose scalar `NOT IN` value key is NULL, used by correlated
+/// null-aware joins (see [`NullAwareMode`]).
///
-/// Such rows produce a NULL (UNKNOWN) mark whenever *any* probe row shares
-/// their correlation scope, so every probe row must be tested against them.
-/// Restricting this map to the NULL-valued build rows keeps that lookup
+/// Such rows are UNKNOWN whenever *any* probe row in their correlation scope
+/// passes the join filter, so every probe row must be tested against them.
+/// Restricting this lookup to the NULL-valued build rows keeps it
/// proportional to the number of NULLs instead of enumerating every scope
/// match of every probe row.
-pub(super) struct NullValueScopeMap {
+pub(super) struct NullValueBuildRows {
/// Hash table keyed by the correlation scope values of the NULL-valued
/// build rows. Stored positions index into `scope_values`/`build_indices`,
- /// not the full build batch.
- pub(super) map: Box<dyn JoinHashMapType>,
+ /// not the full build batch. `None` when the join has no correlation
+ /// scope keys, so every probe row is in scope.
+ pub(super) scope_map: Option<Box<dyn JoinHashMapType>>,
/// Correlation scope key values of the NULL-valued build rows.
pub(super) scope_values: Vec<ArrayRef>,
/// Maps positions in `map`/`scope_values` back to row indices in the full
@@ -215,19 +216,23 @@ pub(super) struct NullValueScopeMap {
/// Null-aware (`NOT IN`) semantics of a hash join, derived from
/// [`HashJoinExec::null_aware`] and the join type.
///
-/// Only these three combinations are legal (see [`Self::try_new`]), so the
+/// Only these combinations are legal (see [`Self::try_new`]), so the
/// stream matches on this instead of re-checking `null_aware && join_type ==
..`.
+///
+/// A `correlated` join has correlation scope keys (`on[1..]`, see
+/// [`HashJoinExec::null_aware`]) or a join filter, or both. A NULL then makes
+/// `NOT IN` UNKNOWN only for the build rows whose scope and filter keep that
+/// NULL, so the join records the decision per build row in the null-indices
+/// bitmap instead of in shared probe-side flags.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum NullAwareMode {
- /// Uncorrelated `build.key NOT IN (probe.key)`: emits build rows, and
- /// none of them once any probe key is NULL.
- LeftAnti,
+ /// `build.key NOT IN (probe.key)`: emits build rows. When uncorrelated,
+ /// none of them are emitted once any probe key is NULL.
+ LeftAnti { correlated: bool },
/// Uncorrelated `probe.key NOT IN (build.key)`: emits probe rows, and
/// none of them once any build key is NULL.
RightAnti,
- /// `NOT IN` as a nullable mark column on the build rows. `correlated`
- /// means `on[1..]` are correlation scope keys (see
- /// [`HashJoinExec::null_aware`]).
+ /// `NOT IN` as a nullable mark column on the build rows.
LeftMark { correlated: bool },
}
@@ -239,13 +244,12 @@ impl NullAwareMode {
num_keys: usize,
has_filter: bool,
) -> Result<Self> {
+ let correlated = num_keys > 1 || has_filter;
let mode = match (join_type, partition_mode) {
- (JoinType::LeftAnti, _) => Self::LeftAnti,
+ (JoinType::LeftAnti, _) => Self::LeftAnti { correlated },
// `PartitionMode::CollectLeft` is safe because `RightAnti` is
probe-driven
(JoinType::RightAnti, PartitionMode::CollectLeft) =>
Self::RightAnti,
- (JoinType::LeftMark, _) => Self::LeftMark {
- correlated: num_keys > 1,
- },
+ (JoinType::LeftMark, _) => Self::LeftMark { correlated },
_ => {
return plan_err!(
"null_aware can only be true for LeftAnti joins and
RightAnti joins with `CollectLeft` `PartitionMode`, or LeftMark joins, got
{join_type} with {partition_mode}"
@@ -253,10 +257,14 @@ impl NullAwareMode {
}
};
match mode {
- Self::LeftAnti | Self::RightAnti if num_keys != 1 => plan_err!(
+ Self::RightAnti if num_keys != 1 => plan_err!(
"null_aware {join_type} joins only support single column join
key, got {num_keys} columns"
),
- Self::LeftMark { .. } if partition_mode ==
PartitionMode::Partitioned => {
+ // Correlated joins share the per-build-row null bitmap across all
+ // probe partitions.
+ Self::LeftMark { .. } | Self::LeftAnti { correlated: true }
+ if partition_mode == PartitionMode::Partitioned =>
+ {
plan_err!(
"null_aware joins require PartitionMode::CollectLeft, got
PartitionMode::Partitioned"
)
@@ -267,6 +275,14 @@ impl NullAwareMode {
_ => Ok(mode),
}
}
+
+ /// Whether this join decides UNKNOWN per build row (see
[`NullAwareMode`]).
+ pub(super) fn is_correlated(self) -> bool {
+ matches!(
+ self,
+ Self::LeftAnti { correlated: true } | Self::LeftMark { correlated:
true }
+ )
+ }
}
/// HashTable and input data for the left (build side) of a join
@@ -274,17 +290,16 @@ pub(super) struct JoinLeftData {
/// The hash table with indices into `batch`
/// Arc is used to allow sharing with SharedBuildAccumulator for hash map
pushdown
pub(super) map: Arc<Map>,
- /// Hash table over correlated scope keys for scalar null-aware mark joins.
+ /// Hash table over correlated scope keys for correlated null-aware joins.
///
- /// For null-aware `LeftMark`, key 0 is the scalar `NOT IN` value key and
- /// keys 1..N are correlated equality scope keys. This map covers all build
- /// rows and is probed only with NULL-valued probe rows; the complementary
- /// direction uses `null_value_scope_map`.
- null_aware_mark_scope_map: Option<Box<dyn JoinHashMapType>>,
- /// Scope map restricted to the build rows whose value key is NULL (see
- /// [`NullValueScopeMap`]). `None` when the build side has no NULL value
- /// keys.
- null_value_scope_map: Option<NullValueScopeMap>,
+ /// Key 0 is the scalar `NOT IN` value key and keys 1..N are correlated
+ /// equality scope keys. This map covers all build rows and is probed only
+ /// with NULL-valued probe rows; the complementary direction uses
+ /// `null_value_build_rows`. `None` when there are no scope keys.
+ null_aware_scope_map: Option<Box<dyn JoinHashMapType>>,
+ /// The build rows whose value key is NULL (see [`NullValueBuildRows`]).
+ /// `None` when the build side has no NULL value keys.
+ null_value_build_rows: Option<NullValueBuildRows>,
/// The input rows for the build side
batch: RecordBatch,
/// The build side on expressions values
@@ -318,12 +333,12 @@ impl JoinLeftData {
&self.map
}
- pub(super) fn null_aware_mark_scope_map(&self) -> Option<&dyn
JoinHashMapType> {
- self.null_aware_mark_scope_map.as_deref()
+ pub(super) fn null_aware_scope_map(&self) -> Option<&dyn JoinHashMapType> {
+ self.null_aware_scope_map.as_deref()
}
- pub(super) fn null_value_scope_map(&self) -> Option<&NullValueScopeMap> {
- self.null_value_scope_map.as_ref()
+ pub(super) fn null_value_build_rows(&self) -> Option<&NullValueBuildRows> {
+ self.null_value_build_rows.as_ref()
}
/// returns a reference to the build side batch
@@ -878,13 +893,16 @@ pub struct HashJoinExec {
/// Flag to indicate if this join uses null-aware equality semantics.
///
/// Set for the physical lowering of scalar `NOT IN` subqueries (producing
- /// `JoinType::LeftAnti` when uncorrelated or `JoinType::LeftMark` when
- /// correlated). When `true`, NULLs in the join keys follow SQL `NOT IN`
- /// three-valued logic rather than ordinary equi-join semantics.
+ /// `JoinType::LeftAnti` at the top level of a filter or
`JoinType::LeftMark`
+ /// inside a larger expression). When `true`, NULLs in the join keys follow
+ /// SQL `NOT IN` three-valued logic rather than ordinary equi-join
semantics.
+ /// A join filter holds the non-equality part of a correlated subquery, and
+ /// only the probe rows that pass it take part in the three-valued logic.
///
/// Key-ordering convention (relied on positionally, not enforced): for a
- /// null-aware `LeftMark` join with more than one key, `on[0]` is the
scalar
- /// `NOT IN` value key and `on[1..N]` are the correlated equality scope
keys.
+ /// null-aware `LeftAnti` or `LeftMark` join with more than one key,
`on[0]`
+ /// is the scalar `NOT IN` value key and `on[1..N]` are the correlated
+ /// equality scope keys.
/// Reordering these keys would silently produce wrong results, which is
why
/// such joins are pinned to `PartitionMode::CollectLeft` (the only key
/// reorderer acts solely on `PartitionMode::Partitioned`).
@@ -2872,11 +2890,8 @@ async fn collect_left_input(
let schema = left_stream.schema();
// The extra scope maps + null bitmap are only built for correlated
- // null-aware LeftMark joins (`on_left[1..]` are correlation scope keys).
- let with_null_aware_mark_state = matches!(
- null_aware,
- Some(NullAwareMode::LeftMark { correlated: true })
- );
+ // null-aware joins (see `NullAwareMode`).
+ let with_null_aware_row_state =
null_aware.is_some_and(NullAwareMode::is_correlated);
let is_phj_candidate = is_perfect_hash_join_candidate(&on_left, &schema)?;
@@ -3006,42 +3021,42 @@ async fn collect_left_input(
BooleanBufferBuilder::new(0)
};
- let null_indices_bitmap = if with_null_aware_mark_state {
+ let null_indices_bitmap = if with_null_aware_row_state {
allocate_bitmap()?
} else {
BooleanBufferBuilder::new(0)
};
- let (null_aware_mark_scope_map, null_value_scope_map) = if
with_null_aware_mark_state
- {
- // Null-aware `LeftMark` convention: `on_left[0]` is the value key and
- // `on_left[1..]` the scope keys, so the scope map needs more than one
key.
- debug_assert!(
- on_left.len() > 1,
- "null-aware LeftMark needs on_left[0]=value, on_left[1..]=scope,
got {} key(s)",
- on_left.len()
- );
- // Scope-only NULL marking uses a HashMap (the primary join map may use
- // ArrayMap for full-key matches, but scope keys have arbitrary shape).
- let mut scope_map = new_join_hashmap(num_rows, &mut reservation,
&metrics)?;
-
- let mut hashes_buffer = vec![0; batch.num_rows()];
- update_hash(
- &on_left[1..],
- &batch,
- &mut *scope_map,
- 0,
- &random_state,
- &mut hashes_buffer,
- 0,
- true,
- NullEquality::NullEqualsNothing,
- )?;
+ let (null_aware_scope_map, null_value_build_rows) = if
with_null_aware_row_state {
+ // Null-aware convention: `on_left[0]` is the value key and
+ // `on_left[1..]` the (possibly empty) correlation scope keys.
+ let scope_keys = &on_left[1..];
+ let scope_map = if scope_keys.is_empty() {
+ None
+ } else {
+ // Scope-only NULL marking uses a HashMap (the primary join map may
+ // use ArrayMap for full-key matches, but scope keys have arbitrary
+ // shape).
+ let mut scope_map = new_join_hashmap(num_rows, &mut reservation,
&metrics)?;
+
+ let mut hashes_buffer = vec![0; batch.num_rows()];
+ update_hash(
+ scope_keys,
+ &batch,
+ &mut *scope_map,
+ 0,
+ &random_state,
+ &mut hashes_buffer,
+ 0,
+ true,
+ NullEquality::NullEqualsNothing,
+ )?;
+ Some(scope_map)
+ };
- // Build the dedicated scope map over the NULL-valued build rows (see
- // `NullValueScopeMap`).
+ // Collect the NULL-valued build rows (see `NullValueBuildRows`).
let value_key = &left_values[0];
- let null_value_scope_map = if value_key.null_count() > 0 {
+ let null_value_build_rows = if value_key.logical_null_count() > 0 {
let null_mask = arrow::compute::is_null(value_key.as_ref())?;
let build_indices = UInt64Array::from_iter_values(
null_mask.values().set_indices().map(|i| i as u64),
@@ -3062,14 +3077,19 @@ async fn collect_left_input(
reservation.try_grow(retained_size)?;
metrics.build_mem_used.add(retained_size);
- let null_rows = build_indices.len();
- let mut map = new_join_hashmap(null_rows, &mut reservation,
&metrics)?;
- let mut hashes_buffer = vec![0; null_rows];
- create_hashes(&scope_values, &random_state, &mut hashes_buffer)?;
-
map.update_from_iter(Box::new(hashes_buffer.iter().enumerate().rev()), 0);
+ let scope_map = if scope_values.is_empty() {
+ None
+ } else {
+ let null_rows = build_indices.len();
+ let mut map = new_join_hashmap(null_rows, &mut reservation,
&metrics)?;
+ let mut hashes_buffer = vec![0; null_rows];
+ create_hashes(&scope_values, &random_state, &mut
hashes_buffer)?;
+
map.update_from_iter(Box::new(hashes_buffer.iter().enumerate().rev()), 0);
+ Some(map)
+ };
- Some(NullValueScopeMap {
- map,
+ Some(NullValueBuildRows {
+ scope_map,
scope_values,
build_indices,
})
@@ -3077,7 +3097,7 @@ async fn collect_left_input(
None
};
- (Some(scope_map), null_value_scope_map)
+ (scope_map, null_value_build_rows)
} else {
(None, None)
};
@@ -3120,8 +3140,8 @@ async fn collect_left_input(
let data = JoinLeftData {
map,
- null_aware_mark_scope_map,
- null_value_scope_map,
+ null_aware_scope_map,
+ null_value_build_rows,
batch,
values: left_values,
visited_indices_bitmap: Mutex::new(visited_indices_bitmap),
@@ -8871,13 +8891,14 @@ mod tests {
),
];
- // Try to create null-aware anti join with 2 columns (should fail)
+ // Try to create null-aware right anti join with 2 columns (should
fail).
+ // A multi-column `LeftAnti` is a correlated `NOT IN` and is accepted.
let result = HashJoinExec::try_new(
left,
right,
on,
None,
- &JoinType::LeftAnti,
+ &JoinType::RightAnti,
None,
PartitionMode::CollectLeft,
NullEquality::NullEqualsNothing,
@@ -8887,7 +8908,7 @@ mod tests {
assert!(result.is_err());
assert!(
result.unwrap_err().to_string().contains(
- "null_aware LeftAnti joins only support single column join key"
+ "null_aware RightAnti joins only support single column join
key"
)
);
}
@@ -9183,6 +9204,151 @@ mod tests {
Ok(())
}
+ /// `left.z > right.z` over the second column of two-column tables: the
+ /// non-equality correlation of
+ /// `id NOT IN (SELECT r.id FROM r WHERE r.z < l.z)`.
+ fn prepare_second_column_gt_filter() -> JoinFilter {
+ let column_indices = vec![
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Right,
+ },
+ ];
+ let intermediate_schema = Schema::new(vec![
+ Field::new("z", DataType::Int32, true),
+ Field::new("z", DataType::Int32, true),
+ ]);
+ let filter_expression = Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("z", 0)),
+ Operator::Gt,
+ Arc::new(Column::new("z", 1)),
+ )) as Arc<dyn PhysicalExpr>;
+
+ JoinFilter::new(
+ filter_expression,
+ column_indices,
+ Arc::new(intermediate_schema),
+ )
+ }
+
+ /// Build and probe sides of a null-aware join whose only correlation is
+ /// the non-equality filter from [`prepare_second_column_gt_filter`].
+ ///
+ /// For each build row, the probe rows with a smaller `z` form its
+ /// subquery result:
+ /// - `(1, 10)` and `(2, 20)`: `{1, NULL}`
+ /// - `(NULL, 30)`: `{1, NULL}`
+ /// - `(4, 40)`: `{1, 4, NULL}`
+ /// - `(NULL, 1)` and `(5, 1)`: empty
+ ///
+ /// The probe row `(NULL, 50)` never passes the filter.
+ fn build_null_aware_filter_only_inputs()
+ -> (Arc<dyn ExecutionPlan>, Arc<dyn ExecutionPlan>, JoinOn) {
+ let left = build_table_two_cols(
+ ("id", &vec![Some(1), Some(2), None, Some(4), None, Some(5)]),
+ (
+ "z",
+ &vec![Some(10), Some(20), Some(30), Some(40), Some(1),
Some(1)],
+ ),
+ );
+ let right = build_table_two_cols(
+ ("id", &vec![Some(1), None, Some(4), None]),
+ ("z", &vec![Some(5), Some(50), Some(35), Some(2)]),
+ );
+ let on = vec![(
+ Arc::new(Column::new_with_schema("id", &left.schema()).unwrap())
as _,
+ Arc::new(Column::new_with_schema("id", &right.schema()).unwrap())
as _,
+ )];
+ (left, right, on)
+ }
+
+ /// Null-aware `LeftAnti` with a join filter and no correlation scope keys.
+ ///
+ /// A NULL on either side only makes `NOT IN` UNKNOWN for the build rows
+ /// where the filter keeps the NULL, so the NULLs must not remove every
row.
+ #[apply(hash_join_exec_configs)]
+ #[tokio::test]
+ async fn test_null_aware_left_anti_filter_only(batch_size: usize) ->
Result<()> {
+ let task_ctx = prepare_task_ctx(batch_size, false);
+ let (left, right, on) = build_null_aware_filter_only_inputs();
+
+ let join = HashJoinExec::try_new(
+ left,
+ right,
+ on,
+ Some(prepare_second_column_gt_filter()),
+ &JoinType::LeftAnti,
+ None,
+ PartitionMode::CollectLeft,
+ NullEquality::NullEqualsNothing,
+ true,
+ )?;
+
+ let stream = join.execute(0, task_ctx)?;
+ let batches = common::collect(stream).await?;
+
+ // Only the rows with an empty subquery result are TRUE.
+ allow_duplicates! {
+ assert_snapshot!(batches_to_sort_string(&batches), @r"
+ +----+---+
+ | id | z |
+ +----+---+
+ | | 1 |
+ | 5 | 1 |
+ +----+---+
+ ");
+ }
+
+ Ok(())
+ }
+
+ /// Null-aware `LeftMark` with a join filter and no correlation scope keys.
+ #[apply(hash_join_exec_configs)]
+ #[tokio::test]
+ async fn test_null_aware_left_mark_filter_only(batch_size: usize) ->
Result<()> {
+ let task_ctx = prepare_task_ctx(batch_size, false);
+ let (left, right, on) = build_null_aware_filter_only_inputs();
+
+ let join = HashJoinExec::try_new(
+ left,
+ right,
+ on,
+ Some(prepare_second_column_gt_filter()),
+ &JoinType::LeftMark,
+ None,
+ PartitionMode::CollectLeft,
+ NullEquality::NullEqualsNothing,
+ true,
+ )?;
+
+ let stream = join.execute(0, task_ctx)?;
+ let batches = common::collect(stream).await?;
+
+ // `(1, 10)` and `(4, 40)` match (true); `(2, 20)` and `(NULL, 30)`
+ // keep the NULL probe row (UNKNOWN); `(NULL, 1)` and `(5, 1)` have an
+ // empty subquery result (false).
+ allow_duplicates! {
+ assert_snapshot!(batches_to_sort_string(&batches), @r"
+ +----+----+-------+
+ | id | z | mark |
+ +----+----+-------+
+ | | 1 | false |
+ | | 30 | |
+ | 1 | 10 | true |
+ | 2 | 20 | |
+ | 4 | 40 | true |
+ | 5 | 1 | false |
+ +----+----+-------+
+ ");
+ }
+
+ Ok(())
+ }
+
#[test]
fn test_lr_is_preserved() {
assert_eq!(lr_is_preserved(JoinType::Inner), (true, true));
diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs
b/datafusion/physical-plan/src/joins/hash_join/stream.rs
index 7b70b2bf3b..56c34534fb 100644
--- a/datafusion/physical-plan/src/joins/hash_join/stream.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs
@@ -37,7 +37,6 @@ use crate::stream::EmptyRecordBatchStream;
use crate::{
RecordBatchStream, SendableRecordBatchStream, handle_state,
hash_utils::create_hashes,
- joins::SharedBitmapBuilder,
joins::utils::{
BuildProbeJoinMetrics, ColumnIndex, JoinFilter, JoinHashMapType,
StatefulStreamResult, adjust_indices_by_join_type,
apply_join_filter_to_indices,
@@ -581,8 +580,8 @@ impl HashJoinStream {
hashes_buffer,
probe_indices_buffer: Vec::with_capacity(batch_size),
build_indices_buffer: Vec::with_capacity(batch_size),
- // Left unallocated: only correlated null-aware LeftMark joins ever
- // use these, and they grow them on first use.
+ // Left unallocated: only correlated null-aware joins ever use
+ // these, and they grow them on first use.
null_mark_hashes_buffer: Vec::new(),
null_mark_probe_indices_buffer: Vec::new(),
null_mark_build_indices_buffer: Vec::new(),
@@ -837,29 +836,24 @@ impl HashJoinStream {
let timer = self.join_metrics.join_time.timer();
if let Some(mode) = self.null_aware {
- if null_aware_skip_probe_batch(
- mode,
- state,
- &build_side.left_data,
- self.filter.is_some(),
- ) {
+ if null_aware_skip_probe_batch(mode, state, &build_side.left_data)
{
timer.done();
self.state = HashJoinStreamState::FetchProbeBatch;
return Ok(StatefulStreamResult::Continue);
}
- // For correlated null-aware LeftMark, record this batch's UNKNOWN
+ // For correlated null-aware joins, record this batch's UNKNOWN
// candidates once, before the first chunked lookup
// (offset == (0, None)).
//
// Must precede the empty-build-map return below: an all-NULL-key
// build side has an empty full-key map but still needs UNKNOWN
marks.
- if matches!(mode, NullAwareMode::LeftMark { correlated: true })
- && state.offset == (0, None)
- {
+ if mode.is_correlated() && state.offset == (0, None) {
mark_null_candidates_for_probe_batch(
build_side,
state,
+ self.filter.as_ref(),
+ self.join_type,
&self.random_state,
self.batch_size,
&mut self.null_mark_hashes_buffer,
@@ -1126,18 +1120,20 @@ impl HashJoinStream {
// Null-aware joins post-process the build rows under SQL three-valued
// logic; see the helpers for the rules.
let (left_side, right_side, mark_column) = match self.null_aware {
- Some(NullAwareMode::LeftAnti) => {
+ Some(NullAwareMode::LeftAnti { correlated }) => {
let (left_side, right_side) =
null_aware_left_anti_final_indices(
&build_side.left_data,
+ correlated,
probe_summary,
left_side,
right_side,
);
(left_side, right_side, None)
}
- Some(NullAwareMode::LeftMark { .. }) => {
+ Some(NullAwareMode::LeftMark { correlated }) => {
let mark_column = null_aware_left_mark_column(
&build_side.left_data,
+ correlated,
probe_summary,
&left_side,
&right_side,
@@ -1243,24 +1239,28 @@ fn null_aware_skip_probe_batch(
mode: NullAwareMode,
state: &ProcessProbeBatchState,
left_data: &JoinLeftData,
- has_filter: bool,
) -> bool {
match mode {
NullAwareMode::RightAnti => left_data.build_side_has_null,
- NullAwareMode::LeftAnti | NullAwareMode::LeftMark { .. } => {
+ // Correlated joins decide UNKNOWN per build row instead, in
+ // `mark_null_candidates_for_probe_batch`.
+ NullAwareMode::LeftAnti { correlated: true }
+ | NullAwareMode::LeftMark { correlated: true } => false,
+ NullAwareMode::LeftAnti { correlated: false }
+ | NullAwareMode::LeftMark { correlated: false } => {
// `on[0]` is the `NOT IN` value key for both modes.
let probe_key_column = &state.values[0];
- let probe_has_null = match mode {
- NullAwareMode::LeftAnti if !has_filter => {
- probe_key_column.logical_null_count() > 0
- }
- _ => probe_key_column.null_count() > 0,
+ let is_anti = matches!(mode, NullAwareMode::LeftAnti { .. });
+ let probe_has_null = if is_anti {
+ probe_key_column.logical_null_count() > 0
+ } else {
+ probe_key_column.null_count() > 0
};
// Only batches with rows count: `NULL NOT IN (empty)` is TRUE.
left_data.record_probe_batch(state.batch.num_rows() > 0,
probe_has_null);
// Best-effort early exit; the final stage re-checks the flag
// through `report_probe_completed`.
- mode == NullAwareMode::LeftAnti &&
left_data.probe_side_has_null_hint()
+ is_anti && left_data.probe_side_has_null_hint()
}
}
}
@@ -1291,7 +1291,12 @@ fn drop_null_probe_keys(
}
/// Final-stage rules of a null-aware `LeftAnti` join, evaluated by the last
-/// probe partition from what every partition together saw:
+/// probe partition.
+///
+/// A correlated join (see `NullAwareMode`) drops the unmatched build rows
+/// whose `NOT IN` was marked UNKNOWN in the null-indices bitmap.
+///
+/// Otherwise the rules use what every partition together saw:
/// - a NULL probe key seen by any partition makes `build.key NOT IN (probe)`
/// UNKNOWN for every build row, so nothing is emitted;
/// - otherwise a NULL build key means `NULL NOT IN (probe)`, which is UNKNOWN
@@ -1299,10 +1304,23 @@ fn drop_null_probe_keys(
/// kept).
fn null_aware_left_anti_final_indices(
left_data: &JoinLeftData,
+ correlated: bool,
probe_summary: ProbeSideSummary,
left_side: UInt64Array,
right_side: UInt32Array,
) -> (UInt64Array, UInt32Array) {
+ if correlated {
+ let null_indices_bitmap = left_data.null_indices_bitmap().lock();
+ let left_side = UInt64Array::from_iter_values(
+ left_side
+ .values()
+ .iter()
+ .copied()
+ .filter(|idx| !null_indices_bitmap.get_bit(*idx as usize)),
+ );
+ let right_side = UInt32Array::new_null(left_side.len());
+ return (left_side, right_side);
+ }
if probe_summary.has_null {
return (UInt64Array::new_null(0), UInt32Array::new_null(0));
}
@@ -1324,16 +1342,14 @@ fn null_aware_left_anti_final_indices(
/// final indices and what every probe partition together saw.
fn null_aware_left_mark_column(
left_data: &JoinLeftData,
+ correlated: bool,
probe_summary: ProbeSideSummary,
left_side: &UInt64Array,
right_side: &UInt32Array,
) -> ArrayRef {
let build_key_column = &left_data.values()[0];
// Correlated joins precomputed the UNKNOWN decision per build row.
- let null_indices_bitmap = left_data
- .null_aware_mark_scope_map()
- .is_some()
- .then(|| left_data.null_indices_bitmap().lock());
+ let null_indices_bitmap = correlated.then(||
left_data.null_indices_bitmap().lock());
build_null_aware_left_mark_column(
left_side,
right_side,
@@ -1344,104 +1360,192 @@ fn null_aware_left_mark_column(
)
}
-/// Records which build rows of a correlated null-aware `LeftMark` join are
-/// UNKNOWN candidates for this probe batch.
+/// Records which build rows of a correlated null-aware join are UNKNOWN
+/// candidates for this probe batch.
///
-/// Key layout: `on[0]` is the `NOT IN` value key, `on[1..]` the correlation
-/// scope keys (see `HashJoinExec::null_aware`). A build row's mark must be
-/// NULL (SQL UNKNOWN) instead of FALSE when it is unmatched and either:
-/// 1. its value key is NULL and any probe row shares its correlation scope, or
-/// 2. some probe row in its correlation scope has a NULL value key.
+/// Key layout: `on[0]` is the `NOT IN` value key, `on[1..]` the (possibly
+/// empty) correlation scope keys (see `HashJoinExec::null_aware`). An
+/// unmatched build row's `NOT IN` is UNKNOWN instead of TRUE (its mark is NULL
+/// instead of FALSE) when either:
+/// 1. its value key is NULL and any probe row in its correlation scope passes
+/// the join filter, or
+/// 2. some probe row in its correlation scope with a NULL value key passes the
+/// join filter.
///
-/// Case 1 probes the build-side NULL-value scope map with all probe rows;
-/// case 2 probes the full scope map with only the NULL-valued probe rows.
+/// Case 1 pairs the NULL-valued build rows with all probe rows; case 2 pairs
+/// all build rows with the NULL-valued probe rows. Scope keys narrow these
+/// pairs through a hash lookup; without scope keys every pair is a candidate.
+/// The join filter, if any, then decides which candidates count.
+///
+/// A build row stays UNKNOWN once it is marked, so candidates whose build row
+/// is already marked are skipped, and the join filter is not evaluated for
+/// them. Without scope keys this also ends the pairing as soon as no unmarked
+/// build row is left.
+#[expect(clippy::too_many_arguments)]
fn mark_null_candidates_for_probe_batch(
build_side: &BuildSideReadyState,
state: &ProcessProbeBatchState,
+ filter: Option<&JoinFilter>,
+ join_type: JoinType,
random_state: &RandomState,
batch_size: usize,
hashes_buffer: &mut Vec<u64>,
probe_indices_buffer: &mut Vec<u32>,
build_indices_buffer: &mut Vec<u64>,
) -> Result<()> {
- let Some(scope_map) = build_side.left_data.null_aware_mark_scope_map()
else {
+ let left_data = &build_side.left_data;
+ let null_value_build_rows = left_data.null_value_build_rows();
+ let probe_value_key = &state.values[0];
+ let probe_has_null_values = probe_value_key.logical_null_count() > 0;
+ if null_value_build_rows.is_none() && !probe_has_null_values {
return Ok(());
- };
+ }
debug_assert_eq!(
- build_side.left_data.values().len(),
+ left_data.values().len(),
state.values.len(),
"build/probe key counts must match"
);
- debug_assert!(state.values.len() > 1, "keys must be [value, scope..]");
-
- let probe_value_key = &state.values[0];
- let build_scope_values = &build_side.left_data.values()[1..];
+ let build_scope_values = &left_data.values()[1..];
let probe_scope_values = &state.values[1..];
- let null_value_scope_map = build_side.left_data.null_value_scope_map();
- let probe_has_null_values = probe_value_key.null_count() > 0;
- if null_value_scope_map.is_none() && !probe_has_null_values {
- return Ok(());
- }
+ // Keeps the candidate pairs that pass the join filter and marks their
+ // build rows as UNKNOWN.
+ let mut mark = |build_indices: UInt64Array, probe_indices: UInt32Array| {
+ let (build_indices, probe_indices) =
+ retain_unmarked(left_data, build_indices, probe_indices);
+ if build_indices.is_empty() {
+ return Ok(());
+ }
+ let build_indices = match filter {
+ Some(filter) => {
+ apply_join_filter_to_indices(
+ left_data.batch(),
+ &state.batch,
+ build_indices,
+ probe_indices,
+ filter,
+ JoinSide::Left,
+ None,
+ join_type,
+ )?
+ .0
+ }
+ None => build_indices,
+ };
+ if !build_indices.is_empty() {
+ let mut null_bitmap = left_data.null_indices_bitmap().lock();
+ for build_idx in build_indices.values() {
+ null_bitmap.set_bit(*build_idx as usize, true);
+ }
+ }
+ Ok(())
+ };
// Case 1: build rows with a NULL value key are UNKNOWN as soon as any
- // probe row shares their correlation scope.
- if let Some(null_value_scope_map) = null_value_scope_map {
- hashes_buffer.clear();
- hashes_buffer.resize(state.batch.num_rows(), 0);
- create_hashes(probe_scope_values, random_state, hashes_buffer)?;
-
- scan_scope_matches_into_bitmap(
- null_value_scope_map.map.as_ref(),
- &null_value_scope_map.scope_values,
- probe_scope_values,
- hashes_buffer,
- batch_size,
- probe_indices_buffer,
- build_indices_buffer,
- // The map indexes only the NULL-valued build rows; translate its
- // positions back to row indices in the full build batch.
- |position| null_value_scope_map.build_indices.value(position as
usize),
- build_side.left_data.null_indices_bitmap(),
- )?;
+ // probe row in their correlation scope passes the filter.
+ if let Some(null_rows) = null_value_build_rows {
+ match &null_rows.scope_map {
+ Some(scope_map) => {
+ hashes_buffer.clear();
+ hashes_buffer.resize(state.batch.num_rows(), 0);
+ create_hashes(probe_scope_values, random_state,
hashes_buffer)?;
+
+ for_each_scope_match(
+ scope_map.as_ref(),
+ &null_rows.scope_values,
+ probe_scope_values,
+ hashes_buffer,
+ batch_size,
+ probe_indices_buffer,
+ build_indices_buffer,
+ |positions, probe_indices| {
+ // The map indexes only the NULL-valued build rows;
+ // translate its positions back to build row indices.
+ let build_indices = UInt64Array::from_iter_values(
+ positions
+ .values()
+ .iter()
+ .map(|p| null_rows.build_indices.value(*p as
usize)),
+ );
+ mark(build_indices, probe_indices)
+ },
+ )?;
+ }
+ None => {
+ for_each_unmarked_cross_product(
+ left_data,
+ null_rows.build_indices.values().iter().copied(),
+ 0..state.batch.num_rows() as u32,
+ batch_size,
+ &mut mark,
+ )?;
+ }
+ }
}
// Case 2: NULL-valued probe rows make every build row in their correlation
- // scope an UNKNOWN candidate.
+ // scope that passes the filter an UNKNOWN candidate.
if probe_has_null_values {
let null_mask = arrow::compute::is_null(probe_value_key.as_ref())?;
- let probe_null_scope_values = probe_scope_values
- .iter()
- .map(|values| Ok(arrow::compute::filter(values.as_ref(),
&null_mask)?))
- .collect::<Result<Vec<_>>>()?;
-
- hashes_buffer.clear();
- hashes_buffer.resize(null_mask.true_count(), 0);
- create_hashes(&probe_null_scope_values, random_state, hashes_buffer)?;
+ let null_probe_rows = UInt32Array::from_iter_values(
+ null_mask.values().set_indices().map(|i| i as u32),
+ );
- scan_scope_matches_into_bitmap(
- scope_map,
- build_scope_values,
- &probe_null_scope_values,
- hashes_buffer,
- batch_size,
- probe_indices_buffer,
- build_indices_buffer,
- |position| position,
- build_side.left_data.null_indices_bitmap(),
- )?;
+ match left_data.null_aware_scope_map() {
+ Some(scope_map) => {
+ let probe_null_scope_values = probe_scope_values
+ .iter()
+ .map(|values| {
+ Ok(arrow::compute::filter(values.as_ref(),
&null_mask)?)
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ hashes_buffer.clear();
+ hashes_buffer.resize(null_probe_rows.len(), 0);
+ create_hashes(&probe_null_scope_values, random_state,
hashes_buffer)?;
+
+ for_each_scope_match(
+ scope_map,
+ build_scope_values,
+ &probe_null_scope_values,
+ hashes_buffer,
+ batch_size,
+ probe_indices_buffer,
+ build_indices_buffer,
+ |build_indices, positions| {
+ // The lookup ran over only the NULL-valued probe rows;
+ // translate its positions back to probe row indices.
+ let probe_indices = UInt32Array::from_iter_values(
+ positions
+ .values()
+ .iter()
+ .map(|p| null_probe_rows.value(*p as usize)),
+ );
+ mark(build_indices, probe_indices)
+ },
+ )?;
+ }
+ None => {
+ for_each_unmarked_cross_product(
+ left_data,
+ 0..left_data.batch().num_rows() as u64,
+ null_probe_rows.values().iter().copied(),
+ batch_size,
+ &mut mark,
+ )?;
+ }
+ }
}
Ok(())
}
-/// Scans all correlation-scope matches between `build_scope_values` and
-/// `probe_scope_values` and sets the bit of every matched build row in
-/// `null_bitmap`, translating matched map positions through
-/// `map_position_to_build_row`.
+/// Calls `f` with all correlation-scope matches between `build_scope_values`
+/// and `probe_scope_values`, as chunks of at most `batch_size` pairs of
+/// (position in `build_scope_values`, position in `probe_scope_values`).
#[expect(clippy::too_many_arguments)]
-fn scan_scope_matches_into_bitmap(
+fn for_each_scope_match(
scope_map: &dyn JoinHashMapType,
build_scope_values: &[ArrayRef],
probe_scope_values: &[ArrayRef],
@@ -1449,12 +1553,11 @@ fn scan_scope_matches_into_bitmap(
batch_size: usize,
probe_indices_buffer: &mut Vec<u32>,
build_indices_buffer: &mut Vec<u64>,
- map_position_to_build_row: impl Fn(u64) -> u64,
- null_bitmap: &SharedBitmapBuilder,
+ mut f: impl FnMut(UInt64Array, UInt32Array) -> Result<()>,
) -> Result<()> {
let mut offset = (0, None);
loop {
- let (build_indices, _probe_indices, next_offset) = lookup_join_hashmap(
+ let (build_indices, probe_indices, next_offset) = lookup_join_hashmap(
scope_map,
build_scope_values,
probe_scope_values,
@@ -1468,13 +1571,7 @@ fn scan_scope_matches_into_bitmap(
)?;
if !build_indices.is_empty() {
- let mut null_bitmap = null_bitmap.lock();
-
- for build_idx in build_indices.iter() {
- let build_idx = build_idx
- .expect("scope lookup should produce non-null build
indices");
- null_bitmap.set_bit(map_position_to_build_row(build_idx) as
usize, true);
- }
+ f(build_indices, probe_indices)?;
}
let Some(next_offset) = next_offset else {
@@ -1486,6 +1583,79 @@ fn scan_scope_matches_into_bitmap(
Ok(())
}
+/// Removes the candidate pairs whose build row is already marked UNKNOWN.
+fn retain_unmarked(
+ left_data: &JoinLeftData,
+ build_indices: UInt64Array,
+ probe_indices: UInt32Array,
+) -> (UInt64Array, UInt32Array) {
+ let bitmap = left_data.null_indices_bitmap().lock();
+ let is_unmarked = |build_idx: &u64| !bitmap.get_bit(*build_idx as usize);
+ if build_indices.values().iter().all(is_unmarked) {
+ return (build_indices, probe_indices);
+ }
+ let (build, probe): (Vec<u64>, Vec<u32>) = build_indices
+ .values()
+ .iter()
+ .zip(probe_indices.values().iter())
+ .filter(|(build_idx, _)| is_unmarked(build_idx))
+ .unzip();
+ (build.into(), probe.into())
+}
+
+/// Calls `f` with the pairs of `build_rows` x `probe_rows` whose build row is
+/// not marked UNKNOWN, as chunks of at most `batch_size` pairs.
+///
+/// `f` marks build rows, so the unmarked build rows are found again after each
+/// chunk. The pairing stops when no unmarked build row is left.
+fn for_each_unmarked_cross_product(
+ left_data: &JoinLeftData,
+ build_rows: impl Iterator<Item = u64>,
+ probe_rows: impl Iterator<Item = u32>,
+ batch_size: usize,
+ mut f: impl FnMut(UInt64Array, UInt32Array) -> Result<()>,
+) -> Result<()> {
+ let retain_unmarked_rows = |rows: &mut Vec<u64>| {
+ let bitmap = left_data.null_indices_bitmap().lock();
+ rows.retain(|idx| !bitmap.get_bit(*idx as usize));
+ };
+
+ let mut build_rows: Vec<u64> = build_rows.collect();
+ retain_unmarked_rows(&mut build_rows);
+
+ let mut build_chunk = Vec::with_capacity(batch_size);
+ let mut probe_chunk = Vec::with_capacity(batch_size);
+ let mut marks_since_refresh = false;
+ for probe_row in probe_rows {
+ // Refresh only after a chunk was sent, so the cost of the refresh
+ // stays proportional to the pairs already evaluated.
+ if marks_since_refresh {
+ retain_unmarked_rows(&mut build_rows);
+ marks_since_refresh = false;
+ }
+ if build_rows.is_empty() {
+ break;
+ }
+ for build_row in &build_rows {
+ build_chunk.push(*build_row);
+ probe_chunk.push(probe_row);
+ if build_chunk.len() == batch_size {
+ f(
+ std::mem::replace(&mut build_chunk,
Vec::with_capacity(batch_size))
+ .into(),
+ std::mem::replace(&mut probe_chunk,
Vec::with_capacity(batch_size))
+ .into(),
+ )?;
+ marks_since_refresh = true;
+ }
+ }
+ }
+ if !build_chunk.is_empty() {
+ f(build_chunk.into(), probe_chunk.into())?;
+ }
+ Ok(())
+}
+
impl Stream for HashJoinStream {
type Item = Result<RecordBatch>;
diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
index 4feea136d9..050f223f9b 100644
--- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
+++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
@@ -400,8 +400,10 @@ SELECT * FROM test_table WHERE (c1 NOT IN (SELECT c2 FROM
test_table)) = true;
# The correlated subquery version from issue #10583 creates a multi-column join
# (correlation + NOT IN condition). It is supported now; see the correlated
# sections below.
-query error DataFusion error: Error during planning: null_aware LeftAnti joins
only support single column join key, got 2 columns
+query II rowsort
SELECT * FROM test_table t1 WHERE c1 NOT IN (SELECT c2 FROM test_table t2
WHERE t1.c1 = t2.c1);
+----
+NULL 0
#############
## Cleanup
@@ -739,17 +741,17 @@ statement ok
CREATE TABLE nai_res_t2(id INT, z INT) AS VALUES (1,5), (NULL,50);
# The NULL row (z = 50) never passes `t2.z < t1.z`, so it does not affect the
result.
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
query I
SELECT id FROM nai_res_t1 WHERE id NOT IN (SELECT nai_res_t2.id FROM
nai_res_t2 WHERE nai_res_t2.z < nai_res_t1.z) ORDER BY id;
----
+2
+4
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
query I
SELECT id FROM nai_res_t1 WHERE NOT (id IN (SELECT nai_res_t2.id FROM
nai_res_t2 WHERE nai_res_t2.z < nai_res_t1.z)) ORDER BY id;
----
+2
+4
query I
SELECT id FROM nai_res_t1 WHERE id IN (SELECT nai_res_t2.id FROM nai_res_t2
WHERE nai_res_t2.z < nai_res_t1.z) ORDER BY id;
@@ -790,49 +792,58 @@ CREATE TABLE nai_res_inner(id INT, z INT, g INT) AS VALUES
# Per outer row, the residual decides which inner rows (NULL or not) are in
# the subquery. A NULL outer value is TRUE only when that set is empty.
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
query II rowsort
SELECT id, z FROM nai_res_outer
WHERE id NOT IN (SELECT i.id FROM nai_res_inner i WHERE i.z < nai_res_outer.z);
----
+5 1
+NULL 1
query II rowsort
SELECT id, z FROM nai_res_outer
WHERE id NOT IN (SELECT i.id FROM nai_res_inner i WHERE i.z > nai_res_outer.z);
----
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
query II rowsort
SELECT id, z FROM nai_res_outer
WHERE id NOT IN (SELECT i.id FROM nai_res_inner i WHERE i.z < nai_res_outer.z
AND i.z > 3);
----
+2 20
+5 1
+NULL 1
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
query II rowsort
SELECT id, z FROM nai_res_outer
WHERE id + 0 NOT IN (SELECT i.id FROM nai_res_inner i WHERE i.z <
nai_res_outer.z);
----
+5 1
+NULL 1
# The residual references the subquery value itself.
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
query II rowsort
SELECT id, z FROM nai_res_outer
WHERE id NOT IN (SELECT i.id FROM nai_res_inner i WHERE i.id + i.z >
nai_res_outer.z);
----
+1 10
+2 20
+4 40
+5 1
# Equality and non-equality correlation together.
-query error DataFusion error: Error during planning: null_aware LeftAnti joins
only support single column join key, got 2 columns
+query II rowsort
SELECT id, z FROM nai_res_outer
WHERE id NOT IN (SELECT i.id FROM nai_res_inner i WHERE i.g = nai_res_outer.g
AND i.z < nai_res_outer.z);
+----
+2 20
+5 1
+NULL 1
# Equality correlation only.
-query error DataFusion error: Error during planning: null_aware LeftAnti joins
only support single column join key, got 2 columns
+query II rowsort
SELECT id, z FROM nai_res_outer
WHERE id NOT IN (SELECT i.id FROM nai_res_inner i WHERE i.g = nai_res_outer.g);
+----
+5 1
# NOT EXISTS uses two-valued logic and must not change.
query II rowsort
@@ -849,16 +860,20 @@ NULL 30
statement ok
SET datafusion.execution.batch_size = 1;
-# Pinned to today's behaviour, which is incorrect. See
-# https://github.com/apache/datafusion/issues/25336 -- the fix flips this.
query II rowsort
SELECT id, z FROM nai_res_outer
WHERE id NOT IN (SELECT i.id FROM nai_res_inner i WHERE i.z < nai_res_outer.z);
----
+5 1
+NULL 1
-query error DataFusion error: Error during planning: null_aware LeftAnti joins
only support single column join key, got 2 columns
+query II rowsort
SELECT id, z FROM nai_res_outer
WHERE id NOT IN (SELECT i.id FROM nai_res_inner i WHERE i.g = nai_res_outer.g
AND i.z < nai_res_outer.z);
+----
+2 20
+5 1
+NULL 1
statement ok
RESET datafusion.execution.batch_size;
@@ -882,6 +897,9 @@ CREATE TABLE nai_res_ig(id INT) AS VALUES (1), (NULL);
query II
SELECT id, g FROM nai_res_og WHERE nai_res_og.id NOT IN (SELECT i.id FROM
nai_res_ig i WHERE nai_res_og.g > 0) ORDER BY id;
----
+3 0
+4 NULL
+NULL 0
# The shape must stay a single anti join that carries the correlation as a
# filter. The alternative lowering materializes the mark of three joins, which
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
