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-25339-ca0a6a42d799b856f3894ca14ef4f186be0c5011 in repository https://gitbox.apache.org/repos/asf/datafusion.git
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]
