From bedcd9bf1ff2b5d227efc715ddbb32a8b26fe7bd Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Sun, 11 Oct 2026 10:12:27 -0700 Subject: [PATCH 1/2] perf: read NOT NULL tuple elements of a multi-column NOT IN as scope keys A tuple element that cannot be NULL in scope on either side never makes a multi-column `NOT IN` comparison UNKNOWN, so it only selects which subquery rows a NULL in another element is compared with. Decorrelation now moves such elements after the nullable ones, so `null_aware_value_keys` counts only the nullable elements and the others become correlation scope keys. The null-aware hash join then pairs NULL-valued rows with the rows in scope through a hash lookup instead of a cross product. Closes #26088. Co-authored-by: Isaac --- .../null_aware_join/benchmarks/q11.benchmark | 7 +- .../null_aware_join/benchmarks/q12.benchmark | 8 +- .../tests/fuzz_cases/null_aware_join_fuzz.rs | 44 ++++- datafusion/expr/src/logical_plan/plan.rs | 4 +- .../src/decorrelate_predicate_subquery.rs | 76 ++++++-- .../physical-plan/src/joins/hash_join/exec.rs | 18 +- .../test_files/null_aware_anti_join.slt | 129 +++++++++++++ .../test_files/null_aware_multi_column.slt | 173 ++++++++++++++++++ .../library-user-guide/upgrading/56.0.0.md | 6 +- 9 files changed, 418 insertions(+), 47 deletions(-) diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q11.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q11.benchmark index 067616db4953c..cf4b5a30d2c2c 100644 --- a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q11.benchmark +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q11.benchmark @@ -25,9 +25,10 @@ expect_plan HashJoinExec run -- Q11: uncorrelated multi-column NOT IN with 1% NULL in one subquery element. --- Each NULL-valued subquery row is paired with every outer row and the pair is --- kept as UNKNOWN only when the other element matches, so the cost grows with --- (outer rows) x (NULL-valued subquery rows). Compare with Q10 and Q12. +-- `z` is NOT NULL on both sides, so the join reads it as a scope key: each +-- NULL-valued subquery row is paired only with the outer rows with the same +-- `z`, through a hash lookup, so the cost grows with (NULL-valued subquery +-- rows) x (outer rows per `z` value). Compare with Q10 and Q12. SELECT count(*) FROM small_outer o WHERE (o.id_n0, o.z) NOT IN (SELECT i.id_n1, i.z FROM small_inner i); diff --git a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q12.benchmark b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q12.benchmark index c126c9c16163c..a4740047da72a 100644 --- a/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q12.benchmark +++ b/benchmarks/sql_benchmarks/null_aware_join/benchmarks/q12.benchmark @@ -26,10 +26,10 @@ expect_plan HashJoinExec run -- Q12: uncorrelated multi-column NOT IN with 1% NULL in one outer element, --- the mirror of Q11. The outer side is the build side, so each NULL-valued --- build row is paired with every subquery row and kept as UNKNOWN only when --- the other element matches: the cost grows with (NULL-valued outer rows) x --- (subquery rows). Compare with Q10 and Q11. +-- the mirror of Q11. The outer side is the build side, and `z`, NOT NULL on +-- both sides, is a scope key: each NULL-valued build row is paired only with +-- the subquery rows with the same `z`, so the cost grows with (NULL-valued +-- outer rows) x (subquery rows per `z` value). Compare with Q10 and Q11. SELECT count(*) FROM small_outer o WHERE (o.id_n1, o.z) NOT IN (SELECT i.id_n0, i.z FROM small_inner i); diff --git a/datafusion/core/tests/fuzz_cases/null_aware_join_fuzz.rs b/datafusion/core/tests/fuzz_cases/null_aware_join_fuzz.rs index 4bf7171771570..755ffa88900f0 100644 --- a/datafusion/core/tests/fuzz_cases/null_aware_join_fuzz.rs +++ b/datafusion/core/tests/fuzz_cases/null_aware_join_fuzz.rs @@ -21,6 +21,8 @@ //! null-aware hash joins, are compared with `NOT EXISTS` / `EXISTS` //! formulations of the same three-valued logic, which do not. Random tables //! with NULLs on both sides are swept across batch sizes and partition counts. +//! Some columns are NOT NULL, so that tuples mix elements that can be NULL with +//! elements that cannot, which the null-aware join reads as scope keys. use std::sync::Arc; @@ -36,8 +38,8 @@ use rand::{Rng, SeedableRng}; async fn null_aware_join_matches_reference() { for seed in 0..20 { let mut rng = StdRng::seed_from_u64(seed); - let outer = random_table(&mut rng, &["id", "k", "a", "b", "c"]); - let inner = random_table(&mut rng, &["k", "x", "y", "z"]); + let outer = random_table(&mut rng, &["id", "k", "a", "b", "c"], &["n", "m"]); + let inner = random_table(&mut rng, &["k", "x", "y", "z"], &["v", "w"]); for target_partitions in [1, 4] { for batch_size in [1, 3, 8192] { let config = SessionConfig::new() @@ -93,6 +95,22 @@ fn cases() -> Vec<(String, String)> { "o.a = i.x AND o.b = i.y AND o.c = i.z", ), ("(o.a, 1)", "i.x, i.y", "o.a = i.x AND 1 = i.y"), + // `o.n`, `o.m`, `i.v` and `i.w` are NOT NULL. + ("(o.a, o.n)", "i.x, i.v", "o.a = i.x AND o.n = i.v"), + ("(o.n, o.a)", "i.v, i.x", "o.n = i.v AND o.a = i.x"), + ( + "(o.n, o.a, o.b)", + "i.v, i.x, i.y", + "o.n = i.v AND o.a = i.x AND o.b = i.y", + ), + ( + "(o.n, o.a, o.m)", + "i.v, i.x, i.w", + "o.n = i.v AND o.a = i.x AND o.m = i.w", + ), + ("(o.n, o.a)", "i.y, i.x", "o.n = i.y AND o.a = i.x"), + ("(1, o.a)", "i.v, i.x", "1 = i.v AND o.a = i.x"), + ("(o.n, o.m)", "i.v, i.w", "o.n = i.v AND o.m = i.w"), ] { let subquery = format!("SELECT {output} FROM i{correlation}"); let matched = format!( @@ -130,13 +148,23 @@ fn cases() -> Vec<(String, String)> { cases } -/// A table of nullable `Int32` columns with values in `0..3`, split into up -/// to three partitions. A column named `id` holds unique row ids instead. -fn random_table(rng: &mut StdRng, columns: &[&str]) -> Arc { +/// A table of `Int32` columns with values in `0..3`, split into up to three +/// partitions: the `nullable` columns, then the `non_nullable` ones. A column +/// named `id` holds unique row ids instead. +fn random_table( + rng: &mut StdRng, + nullable: &[&str], + non_nullable: &[&str], +) -> Arc { + let columns = nullable + .iter() + .map(|name| (*name, true)) + .chain(non_nullable.iter().map(|name| (*name, false))) + .collect::>(); let schema = Arc::new(Schema::new( columns .iter() - .map(|name| Field::new(*name, DataType::Int32, true)) + .map(|(name, nullable)| Field::new(*name, DataType::Int32, *nullable)) .collect::>(), )); let null_fraction = [0.0, 0.1, 0.3][rng.random_range(0..3)]; @@ -146,13 +174,13 @@ fn random_table(rng: &mut StdRng, columns: &[&str]) -> Arc { let rows = rng.random_range(0..6); let arrays = columns .iter() - .map(|name| { + .map(|(name, nullable)| { let values: Int32Array = if *name == "id" { (next_id..next_id + rows).map(Some).collect() } else { (0..rows) .map(|_| { - (!rng.random_bool(null_fraction)) + (!(*nullable && rng.random_bool(null_fraction))) .then(|| rng.random_range(0..3)) }) .collect() diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index 9d360d052f835..66fef1bd637da 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -4513,7 +4513,9 @@ pub struct Join { /// [`Self::null_aware`] is set. /// /// A scalar `x NOT IN (SELECT y ...)` has one value key, a multi-column - /// `(a, b) NOT IN (SELECT x, y ...)` one per tuple element. The value keys + /// `(a, b) NOT IN (SELECT x, y ...)` one per tuple element that can be + /// NULL in scope. A tuple element that cannot be NULL on either side never + /// makes the comparison UNKNOWN, so it is a scope key instead. The value keys /// are the leading equi-join keys (`on[..n]`, or the leading equality /// conjuncts of `filter` before they are extracted into `on`); any keys /// after them are correlation scope keys of a correlated subquery. A NULL diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index bfd5109761b70..848bba902dd23 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -569,7 +569,7 @@ struct MarkJoin { /// Are the `IN` equalities, in order, the first equalities of `join_filter` /// that the hash join can use as keys? A null-aware hash join reads its first /// `V` keys as the `IN` values (one per tuple element of a multi-column -/// `IN`) and the other keys as the correlation. +/// `IN` that can be NULL in scope) and the other keys as the correlation. fn values_are_first_keys( in_values: &[InValue], join_filter: &Expr, @@ -722,9 +722,11 @@ fn strip_casts(expr: &Expr) -> &Expr { /// Combines the `IN`/`NOT IN` equalities, one per value, with the correlation /// predicates, if any. /// -/// The equalities stay the leading conjuncts, in order, so that they become -/// `on[..V]`, the key positions a null-aware hash join reads as the `NOT IN` -/// value keys (see `HashJoinExec::null_aware`). +/// The equalities stay the leading conjuncts, in order, so that the first `V` +/// of them become `on[..V]`, the key positions a null-aware hash join reads as +/// the `NOT IN` value keys (see `HashJoinExec::null_aware`). The equalities +/// after them, if any, are on tuple elements that cannot be NULL, which the +/// join reads as scope keys like the correlation. fn in_join_filter(in_values: &[InValue], correlation: Option) -> Result { conjunction( in_values @@ -845,16 +847,19 @@ fn build_join( None => vec![], }; - // True if an `IN` value or the subquery column it is compared with can be - // NULL inside the scope of an outer row. - let mut value_may_be_null = false; - for in_value in &in_values { - value_may_be_null |= in_value.may_be_null_in_scope( - left.schema(), - sub_query_alias.schema(), - &pull_up.correlated_filters, - )?; - } + // For each `IN` value, true if it or the subquery column it is compared + // with can be NULL inside the scope of an outer row. + let elements_may_be_null = in_values + .iter() + .map(|in_value| { + in_value.may_be_null_in_scope( + left.schema(), + sub_query_alias.schema(), + &pull_up.correlated_filters, + ) + }) + .collect::>>()?; + let value_may_be_null = elements_may_be_null.iter().any(|may_be_null| *may_be_null); // Whether this join needs null-aware (`NOT IN` three-valued) semantics. // Decided once: the constant projection below, the mark branch and the anti @@ -929,6 +934,38 @@ fn build_join( } let left = projected_left.as_ref().unwrap_or(left); + // The first `V` keys of a null-aware join are its `NOT IN` value keys. + // + // A tuple element that cannot be NULL in scope on either side never makes + // the tuple comparison UNKNOWN: compared with a definite value it is either + // equal, which leaves the other elements to decide, or a definite mismatch, + // which makes the comparison FALSE whatever the other elements are. It + // therefore only selects which subquery rows a NULL in another element is + // compared with, which is what a correlation scope key does. Moving such + // elements after the ones that can be NULL turns them into scope keys, so + // the hash join finds the rows a NULL-valued row is compared with through a + // hash lookup instead of pairing it with every row of the other side + // (). + // + // A null-aware tuple has at least one element that can be NULL, so `V` is + // at least one. With exactly one, the join is a scalar `NOT IN` on that + // element, correlated by the other elements. + let null_aware_value_keys = if null_aware && in_values.len() > 1 { + let (nullable, non_nullable): (Vec<_>, Vec<_>) = in_values + .into_iter() + .zip(elements_may_be_null) + .partition(|(_, may_be_null)| *may_be_null); + let null_aware_value_keys = nullable.len(); + in_values = nullable + .into_iter() + .chain(non_nullable) + .map(|(in_value, _)| in_value) + .collect(); + null_aware_value_keys + } else { + in_values.len().max(1) + }; + let join_filter = match (in_values.is_empty(), join_filter_opt) { (false, correlation_opt) => in_join_filter(&in_values, correlation_opt)?, (true, Some(correlation)) => correlation, @@ -938,23 +975,22 @@ fn build_join( // Only a null-aware join reads the key order. let value_is_first_key = null_aware && values_are_first_keys( - &in_values, + &in_values[..null_aware_value_keys], &join_filter, left.schema(), sub_query_alias.schema(), )?; // A multi-column `IN` has no fallback that materializes the UNKNOWN rows - // (see below), so a tuple element that does not become its key, for - // example because its type is not hashable, is rejected instead of being - // read in the wrong key position. + // (see below), so a tuple element that can be NULL and does not become its + // key, for example because its type is not hashable, is rejected instead + // of being read in the wrong key position. The other elements need not be + // keys: like any correlation, they may stay in the join filter. if null_aware && !value_is_first_key && in_values.len() > 1 { return not_impl_err!( "Multi-column NOT IN subquery is not supported when a tuple element is not a hashable equi-join key: {join_filter}" ); } - // The first `V` keys of a null-aware join are its `NOT IN` value keys. - let null_aware_value_keys = in_values.len().max(1); // A `NOT IN` in a filter builds a `LeftAnti` join. The null-aware hash // join reads its first key as the `NOT IN` value and the other keys as the diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index de39c1638608a..4c50f95257d17 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1027,19 +1027,21 @@ pub struct HashJoinExec { /// [`Self::null_aware`] is set. /// /// A scalar `x NOT IN (SELECT y ...)` has one value key; a multi-column - /// `(a, b) NOT IN (SELECT x, y ...)` has one per tuple element. The tuple - /// comparison is UNKNOWN rather than FALSE when no element pair is a - /// definite mismatch and at least one of them involves a NULL, so a NULL in - /// one element does not make every comparison UNKNOWN the way a scalar - /// NULL does. + /// `(a, b) NOT IN (SELECT x, y ...)` has one per tuple element that can be + /// NULL. The tuple comparison is UNKNOWN rather than FALSE when no element + /// pair is a definite mismatch and at least one of them involves a NULL, + /// so a NULL in one element does not make every comparison UNKNOWN the + /// way a scalar NULL does. A tuple element that cannot be NULL on either + /// side never makes the comparison UNKNOWN; the planner passes it as a + /// scope key after the value keys, so it narrows the rows a NULL-valued + /// row is compared with. /// /// Such a join always decides UNKNOWN per build row, like a correlated /// scalar `NOT IN`, which has two costs the uncorrelated scalar join does /// not: the outer side stays the build side, since only a single-key join - /// is swapped, and without correlation keys every NULL-valued row is paired + /// is swapped, and without scope keys every NULL-valued row is paired /// with every row on the other side to check its other elements. See - /// and - /// for the follow-ups. + /// for a follow-up. pub null_aware_value_keys: usize, /// Cache holding plan properties like equivalences, output partitioning etc. cache: Arc, diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt index 916819fbb9d89..1c4219079942d 100644 --- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt +++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt @@ -479,6 +479,135 @@ physical_plan 04)------DataSourceExec: partitions=1, partition_sizes=[1] 05)------DataSourceExec: partitions=1, partition_sizes=[1] +# A tuple element that cannot be NULL on either side never makes the tuple +# comparison UNKNOWN. It follows the elements that can be NULL as a scope key, +# so `value_keys` counts only the nullable elements. +statement ok +CREATE TABLE multi_col_outer_nn(k INT NOT NULL, a INT, b INT NOT NULL) AS VALUES +(1, 1, 2), +(2, NULL, 8); + +statement ok +CREATE TABLE multi_col_inner_nn(k INT NOT NULL, x INT, y INT NOT NULL) AS VALUES +(1, 1, 2), +(2, NULL, 4); + +# `b = y` moves after `a = x`. +query TT +EXPLAIN SELECT * FROM multi_col_outer_nn +WHERE (b, a) NOT IN (SELECT y, x FROM multi_col_inner_nn); +---- +logical_plan +01)LeftAnti Join: multi_col_outer_nn.a = __correlated_sq_1.x, multi_col_outer_nn.b = __correlated_sq_1.y null_aware +02)--TableScan: multi_col_outer_nn projection=[k, a, b] +03)--SubqueryAlias: __correlated_sq_1 +04)----Projection: multi_col_inner_nn.y, multi_col_inner_nn.x +05)------TableScan: multi_col_inner_nn projection=[x, y] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(a@1, x@1), (b@2, y@0)], null_aware +02)--DataSourceExec: partitions=1, partition_sizes=[1] +03)--DataSourceExec: partitions=1, partition_sizes=[1] + +# The element scope key comes before the correlation keys. +query TT +EXPLAIN SELECT * FROM multi_col_outer_nn o +WHERE (b, a) NOT IN (SELECT y, x FROM multi_col_inner_nn i WHERE i.k = o.k); +---- +logical_plan +01)LeftAnti Join: o.a = __correlated_sq_1.x, o.b = __correlated_sq_1.y, o.k = __correlated_sq_1.k null_aware +02)--SubqueryAlias: o +03)----TableScan: multi_col_outer_nn projection=[k, a, b] +04)--SubqueryAlias: __correlated_sq_1 +05)----Projection: i.y, i.x, i.k +06)------SubqueryAlias: i +07)--------TableScan: multi_col_inner_nn projection=[k, x, y] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(a@1, x@1), (b@2, y@0), (k@0, k@2)], null_aware +02)--DataSourceExec: partitions=1, partition_sizes=[1] +03)--DataSourceExec: partitions=1, partition_sizes=[1] + +# `NULLIF` can give a NULL, so `k` stays a value key, after `a` as in the +# tuple, and only `b` moves. +query TT +EXPLAIN SELECT * FROM multi_col_outer_nn +WHERE (b, a, k) NOT IN (SELECT y, x, NULLIF(k, 0) FROM multi_col_inner_nn); +---- +logical_plan +01)LeftAnti Join: multi_col_outer_nn.a = __correlated_sq_1.x, CAST(multi_col_outer_nn.k AS Int64) = __correlated_sq_1.nullif(multi_col_inner_nn.k,Int64(0)), multi_col_outer_nn.b = __correlated_sq_1.y null_aware(value_keys=2) +02)--TableScan: multi_col_outer_nn projection=[k, a, b] +03)--SubqueryAlias: __correlated_sq_1 +04)----Projection: multi_col_inner_nn.y, multi_col_inner_nn.x, nullif(CAST(multi_col_inner_nn.k AS Int64), Int64(0)) +05)------TableScan: multi_col_inner_nn projection=[k, x, y] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(a@1, x@1), (CAST(multi_col_outer_nn.k AS Int64)@3, nullif(multi_col_inner_nn.k,Int64(0))@2), (b@2, y@0)], projection=[k@0, a@1, b@2], null_aware(value_keys=2) +02)--ProjectionExec: expr=[k@0 as k, a@1 as a, b@2 as b, CAST(k@0 AS Int64) as CAST(multi_col_outer_nn.k AS Int64)] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)--ProjectionExec: expr=[y@2 as y, x@1 as x, nullif(CAST(k@0 AS Int64), 0) as nullif(multi_col_inner_nn.k,Int64(0))] +05)----DataSourceExec: partitions=1, partition_sizes=[1] + +# A constant element is NOT NULL, so against a NOT NULL column it is a scope +# key. +query TT +EXPLAIN SELECT * FROM multi_col_outer_nn +WHERE (2, a) NOT IN (SELECT y, x FROM multi_col_inner_nn); +---- +logical_plan +01)Projection: multi_col_outer_nn.k, multi_col_outer_nn.a, multi_col_outer_nn.b +02)--LeftAnti Join: multi_col_outer_nn.a = __correlated_sq_1.x, __correlated_sq_1_value_0 = __correlated_sq_1.multi_col_inner_nn.y null_aware +03)----Projection: multi_col_outer_nn.k, multi_col_outer_nn.a, multi_col_outer_nn.b, Int64(2) AS __correlated_sq_1_value_0 +04)------TableScan: multi_col_outer_nn projection=[k, a, b] +05)----SubqueryAlias: __correlated_sq_1 +06)------Projection: CAST(multi_col_inner_nn.y AS Int64), multi_col_inner_nn.x +07)--------TableScan: multi_col_inner_nn projection=[x, y] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(a@1, x@1), (__correlated_sq_1_value_0@3, multi_col_inner_nn.y@0)], projection=[k@0, a@1, b@2], null_aware +02)--ProjectionExec: expr=[k@0 as k, a@1 as a, b@2 as b, 2 as __correlated_sq_1_value_0] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)--ProjectionExec: expr=[CAST(y@1 AS Int64) as multi_col_inner_nn.y, x@0 as x] +05)----DataSourceExec: partitions=1, partition_sizes=[1] + +# NOT NULL on the outer side only: the subquery can still give a NULL `y`, so +# `b` stays a value key. +query TT +EXPLAIN SELECT * FROM multi_col_outer_nn +WHERE (b, a) NOT IN (SELECT y, x FROM multi_col_inner); +---- +logical_plan +01)LeftAnti Join: multi_col_outer_nn.b = __correlated_sq_1.y, multi_col_outer_nn.a = __correlated_sq_1.x null_aware(value_keys=2) +02)--TableScan: multi_col_outer_nn projection=[k, a, b] +03)--SubqueryAlias: __correlated_sq_1 +04)----Projection: multi_col_inner.y, multi_col_inner.x +05)------TableScan: multi_col_inner projection=[x, y] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(b@2, y@0), (a@1, x@1)], null_aware(value_keys=2) +02)--DataSourceExec: partitions=1, partition_sizes=[1] +03)--DataSourceExec: partitions=1, partition_sizes=[1] + +# The mark join reorders the same way. +query TT +EXPLAIN SELECT b, a, (b, a) NOT IN (SELECT y, x FROM multi_col_inner_nn) AS r +FROM multi_col_outer_nn; +---- +logical_plan +01)Projection: multi_col_outer_nn.b, multi_col_outer_nn.a, NOT __correlated_sq_1.mark AS r +02)--LeftMark Join: multi_col_outer_nn.a = __correlated_sq_1.x, multi_col_outer_nn.b = __correlated_sq_1.y null_aware +03)----TableScan: multi_col_outer_nn projection=[a, b] +04)----SubqueryAlias: __correlated_sq_1 +05)------Projection: multi_col_inner_nn.y, multi_col_inner_nn.x +06)--------TableScan: multi_col_inner_nn projection=[x, y] +physical_plan +01)ProjectionExec: expr=[b@1 as b, a@0 as a, NOT mark@2 as r] +02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(a@0, x@1), (b@1, y@0)], null_aware +04)------DataSourceExec: partitions=1, partition_sizes=[1] +05)------DataSourceExec: partitions=1, partition_sizes=[1] + +statement ok +DROP TABLE multi_col_outer_nn; + +statement ok +DROP TABLE multi_col_inner_nn; + statement ok DROP TABLE multi_col_outer; diff --git a/datafusion/sqllogictest/test_files/null_aware_multi_column.slt b/datafusion/sqllogictest/test_files/null_aware_multi_column.slt index 917d1caec972f..beb2ffef6c86f 100644 --- a/datafusion/sqllogictest/test_files/null_aware_multi_column.slt +++ b/datafusion/sqllogictest/test_files/null_aware_multi_column.slt @@ -527,6 +527,167 @@ z statement ok DEALLOCATE mc_tuple_qualified; +############# +## NOT NULL tuple elements +############# + +# `b` and `y` cannot be NULL, so they never make the tuple comparison UNKNOWN +# and the join reads them as a scope key: a NULL `a` is compared only with the +# subquery rows whose `y` equals its `b`. +statement ok +CREATE TABLE mc_nn_outer(k INT NOT NULL, a INT, b INT NOT NULL, value TEXT) AS VALUES +(1, 1, 2, 'p'), +(1, NULL, 2, 'q'), +(1, NULL, 3, 'r'), +(1, 5, 2, 's'), +(2, NULL, 2, 't'), +(2, 7, 9, 'u'), +(1, 6, 4, 'v'); + +statement ok +CREATE TABLE mc_nn_inner(k INT NOT NULL, x INT, y INT NOT NULL) AS VALUES +(1, 1, 2), +(1, NULL, 4), +(2, 8, 9); + +# p matches (1, 2). q and t vs (1, 2) and v vs (NULL, 4) are UNKNOWN. r, s and +# u mismatch every row. +query T rowsort +SELECT value FROM mc_nn_outer +WHERE (a, b) NOT IN (SELECT x, y FROM mc_nn_inner); +---- +r +s +u + +# The NOT NULL element first gives the same rows. +query T rowsort +SELECT value FROM mc_nn_outer +WHERE (b, a) NOT IN (SELECT y, x FROM mc_nn_inner); +---- +r +s +u + +# With one nullable element the tuple is a scalar NOT IN on it, correlated by +# the other elements. +query T rowsort +SELECT value FROM mc_nn_outer +WHERE a NOT IN (SELECT x FROM mc_nn_inner WHERE y = b); +---- +r +s +u + +# With a correlation, t's scope is k = 2, {(8, 9)}, which it mismatches. +query T rowsort +SELECT value FROM mc_nn_outer o +WHERE (a, b) NOT IN (SELECT x, y FROM mc_nn_inner i WHERE i.k = o.k); +---- +r +s +t +u + +# The correlation as a NOT NULL tuple element gives the same rows. +query T rowsort +SELECT value FROM mc_nn_outer +WHERE (a, k, b) NOT IN (SELECT x, k, y FROM mc_nn_inner); +---- +r +s +t +u + +# Two nullable elements and a scope key. The subquery gives (1, 2, 1), +# (NULL, 4, 1) and (8, 9, NULL): t's k = 2 and u's a = 7 are definite +# mismatches. +query T rowsort +SELECT value FROM mc_nn_outer +WHERE (a, b, k) NOT IN (SELECT x, y, NULLIF(k, 2) FROM mc_nn_inner); +---- +r +s +t +u + +# A constant NOT NULL element: only (1, 2) is in scope. +query T rowsort +SELECT value FROM mc_nn_outer +WHERE (a, 2) NOT IN (SELECT x, y FROM mc_nn_inner); +---- +s +u +v + +# An empty subquery keeps every row, NULLs included. +query T rowsort +SELECT value FROM mc_nn_outer +WHERE (a, b) NOT IN (SELECT x, y FROM mc_nn_inner WHERE x > 100); +---- +p +q +r +s +t +u +v + +# A RunEndEncoded element is not hashable, so it stays in the join filter. That +# is rejected for an element that can be NULL (see above), but an element that +# cannot is read like any correlation. +statement ok +CREATE TABLE mc_nn_ree_outer AS +SELECT k, a, arrow_cast(b, 'RunEndEncoded("run_ends": non-null Int32, "values": Int32)') AS b, value +FROM mc_nn_outer; + +statement ok +CREATE TABLE mc_nn_ree_inner AS +SELECT k, x, arrow_cast(y, 'RunEndEncoded("run_ends": non-null Int32, "values": Int32)') AS y +FROM mc_nn_inner; + +query T rowsort +SELECT value FROM mc_nn_ree_outer +WHERE (b, a) NOT IN (SELECT y, x FROM mc_nn_ree_inner); +---- +r +s +u + +# As a nullable value. +query TB rowsort +SELECT value, (a, b) NOT IN (SELECT x, y FROM mc_nn_inner) AS r +FROM mc_nn_outer; +---- +p false +q NULL +r true +s true +t NULL +u true +v NULL + +query TB rowsort +SELECT value, (a, b) NOT IN (SELECT x, y FROM mc_nn_inner i WHERE i.k = o.k) AS r +FROM mc_nn_outer o; +---- +p false +q NULL +r true +s true +t true +u true +v NULL + +query T rowsort +SELECT value FROM mc_nn_outer +WHERE (a, b) NOT IN (SELECT x, y FROM mc_nn_inner) OR value = 'q'; +---- +q +r +s +u + ############# ## Cleanup ############# @@ -593,3 +754,15 @@ DROP TABLE mc_p1; statement ok DROP TABLE mc_p2; + +statement ok +DROP TABLE mc_nn_outer; + +statement ok +DROP TABLE mc_nn_inner; + +statement ok +DROP TABLE mc_nn_ree_outer; + +statement ok +DROP TABLE mc_nn_ree_inner; diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 4bc75d462dd65..b73dc30de445e 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -833,9 +833,9 @@ Wire compatibility is directional: ### Null-aware joins record their `NOT IN` value key count Multi-column `(a, b) NOT IN (SELECT x, y ...)` is now planned as a null-aware -join. Its first equi-join keys are the tuple elements, followed by any -correlation keys, so a null-aware join now records how many of its leading keys -are `NOT IN` value keys: +join. Its first equi-join keys are the tuple elements that can be NULL, +followed by the elements that cannot and any correlation keys, so a null-aware +join now records how many of its leading keys are `NOT IN` value keys: - `datafusion_expr::logical_plan::Join` has a new `null_aware_value_keys: usize` field. From 06c0885af7ac6297c3a77b431453c51c60dad41a Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Sun, 11 Oct 2026 12:54:13 -0700 Subject: [PATCH 2/2] test: expect a scope key for NOT NULL tuple elements in the proto round trip Co-authored-by: Isaac --- .../proto/tests/cases/roundtrip_logical_plan.rs | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index 4a3e6669321c4..065c6cdb9bb4a 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -3879,7 +3879,7 @@ async fn roundtrip_join_null_aware() -> Result<()> { ( " SELECT k - FROM (VALUES (1, 1, 2), (2, 3, 4)) AS t1(k, a, b) + FROM (VALUES (1, 1, 2), (2, 3, CAST(NULL AS INT))) AS t1(k, a, b) WHERE (a, b) NOT IN ( SELECT x, y FROM (VALUES (1, CAST(NULL AS INT), 2)) AS t2(k, x, y) @@ -3888,6 +3888,19 @@ async fn roundtrip_join_null_aware() -> Result<()> { ", (3, 2), ), + // `b` and `y` are NOT NULL, so `b = y` is a scope key, not a value key. + ( + " + SELECT k + FROM (VALUES (1, 1, 2), (2, 3, 4)) AS t1(k, a, b) + WHERE (a, b) NOT IN ( + SELECT x, y + FROM (VALUES (1, CAST(NULL AS INT), 2)) AS t2(k, x, y) + WHERE t2.k = t1.k + ) + ", + (3, 1), + ), ]; for (sql, expected_keys) in cases {