From f031ee7fb94b3753857d13c28c684c58c5d79efd Mon Sep 17 00:00:00 2001 From: naman Date: Wed, 23 Sep 2026 09:04:58 +0530 Subject: [PATCH 1/4] fix: Decide NOT IN null-aware joins from expression nullability, not just column nullability closes: #25473 closes: #25474 `join_keys_may_be_null` only checked the schema nullability of the columns a join filter references. A NULL constant or an expression that is nullable even though every column it reads is NOT NULL (CASE without ELSE, NULLIF, arithmetic with a NULL operand) was therefore never treated as needing null-aware semantics, so a NOT IN predicate that should be UNKNOWN silently kept the outer row instead. Replace the column walk with a per-equality-conjunct nullability check that evaluates each operand as an expression against whichever schema it resolves to, falling back to nullable when that can't be determined. --- .../src/decorrelate_predicate_subquery.rs | 166 ++++++++++++++++-- .../test_files/null_aware_anti_join.slt | 137 +++++++++++++++ 2 files changed, 285 insertions(+), 18 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 135d1fc7f519d..40a948838b17f 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -34,10 +34,12 @@ use datafusion_common::{ use datafusion_expr::expr::{Exists, InSubquery}; use datafusion_expr::expr_rewriter::create_col_from_scalar_expr; use datafusion_expr::logical_plan::{JoinType, Subquery}; -use datafusion_expr::utils::{conjunction, expr_to_columns, split_conjunction_owned}; +use datafusion_expr::utils::{ + conjunction, expr_to_columns, split_conjunction, split_conjunction_owned, +}; use datafusion_expr::{ - BinaryExpr, Expr, Filter, LogicalPlan, LogicalPlanBuilder, Operator, exists, - in_subquery, lit, not, not_exists, not_in_subquery, when, + BinaryExpr, Expr, ExprSchemable, Filter, LogicalPlan, LogicalPlanBuilder, Operator, + exists, in_subquery, lit, not, not_exists, not_in_subquery, when, }; use log::debug; @@ -479,7 +481,10 @@ fn mark_join( /// Check if join keys in the join filter may contain NULL values /// -/// Returns true if any join key column is nullable on either side. +/// Returns true if either side of an equality conjunct can itself be NULL, as +/// an expression rather than just the columns it references. A `CASE` without +/// `ELSE`, `NULLIF`, or a NULL constant is nullable even when every column it +/// reads is declared `NOT NULL`. /// This is used to optimize null-aware anti joins: if all join keys are non-nullable, /// we can use a regular anti join instead of the more expensive null-aware variant. fn join_keys_may_be_null( @@ -487,21 +492,20 @@ fn join_keys_may_be_null( left_schema: &DFSchemaRef, right_schema: &DFSchemaRef, ) -> Result { - // Extract columns from the join filter - let mut columns = std::collections::HashSet::new(); - expr_to_columns(join_filter, &mut columns)?; - - // Check if any column is nullable - for col in columns { - // Check in left schema - if let Ok(field) = left_schema.field_from_column(&col) - && field.as_ref().is_nullable() - { + for conjunct in split_conjunction(join_filter) { + let Expr::BinaryExpr(BinaryExpr { + left, + op: Operator::Eq, + right, + }) = conjunct + else { + // Only equality conjuncts become hash-join keys; a residual + // conjunct's nullability isn't decided here, so stay conservative. return Ok(true); - } - // Check in right schema - if let Ok(field) = right_schema.field_from_column(&col) - && field.as_ref().is_nullable() + }; + + if operand_may_be_null(left, left_schema, right_schema)? + || operand_may_be_null(right, left_schema, right_schema)? { return Ok(true); } @@ -522,6 +526,35 @@ fn in_predicate_first(in_predicate: Expr, correlation: Option) -> Expr { } } +/// Whether `expr` can evaluate to NULL, resolving it against whichever side's +/// schema its column references belong to. Falls back to `true`, the +/// conservative answer, when that side can't be determined uniquely: treating +/// a key as nullable costs a null-aware join, treating it as non-nullable +/// risks wrong results. +fn operand_may_be_null( + expr: &Expr, + left_schema: &DFSchemaRef, + right_schema: &DFSchemaRef, +) -> Result { + let mut columns = std::collections::HashSet::new(); + expr_to_columns(expr, &mut columns)?; + + let references_left = columns + .iter() + .any(|col| left_schema.field_from_column(col).is_ok()); + let references_right = columns + .iter() + .any(|col| right_schema.field_from_column(col).is_ok()); + + match (references_left, references_right) { + (true, true) => Ok(true), + (true, false) => expr.nullable(left_schema.as_ref()), + // A constant references neither schema; either one gives the same + // answer, since a literal's nullability doesn't depend on the schema. + (false, true) | (false, false) => expr.nullable(right_schema.as_ref()), + } +} + fn build_join( left: &LogicalPlan, subquery: &LogicalPlan, @@ -805,6 +838,16 @@ mod tests { table_scan(Some(name), &schema, None)?.build() } + fn has_null_aware_left_anti_join(plan: &LogicalPlan) -> bool { + if let LogicalPlan::Join(join) = plan + && join.join_type == JoinType::LeftAnti + { + return join.null_aware; + } + + plan.inputs().into_iter().any(has_null_aware_left_anti_join) + } + fn has_null_aware_left_mark_join(plan: &LogicalPlan) -> bool { if let LogicalPlan::Join(join) = plan && join.join_type == JoinType::LeftMark @@ -1542,6 +1585,93 @@ mod tests { ) } + /// A NULL constant against a `NOT NULL` subquery column must still + /// trigger the null-aware rewrite: `join_keys_may_be_null` has to see the + /// constant's own nullability, not just the nullability of the columns + /// the join filter references. + /// + #[test] + fn constant_null_not_in_subquery_is_null_aware_over_not_null_column() -> Result<()> { + let outer_scan = nullable_scalar_mark_scan("outer_t")?; + let subquery = test_subquery_with_name("inner_t")?; + + let plan = LogicalPlanBuilder::from(outer_scan) + .filter(not_in_subquery( + Expr::Literal(ScalarValue::Int32(None), None), + subquery, + ))? + .build()?; + + let optimized = optimize_with_decorrelate(plan)?; + assert!( + has_null_aware_left_anti_join(&optimized), + "{}", + optimized.display_indent_schema() + ); + + Ok(()) + } + + /// An expression that can be NULL even though every column it reads is + /// `NOT NULL` (`CASE` without `ELSE`, here) must also trigger the + /// null-aware rewrite. `join_keys_may_be_null` has to check the + /// expression's own nullability rather than only the schema nullability + /// of the columns it references. + /// + #[test] + fn nullable_expression_not_in_subquery_is_null_aware_over_not_null_columns() + -> Result<()> { + let outer_scan = test_table_scan_with_name("outer_t")?; + let subquery = test_subquery_with_name("inner_t")?; + + let value = when(col("outer_t.a").gt(lit(100u32)), col("outer_t.a")).end()?; + let plan = LogicalPlanBuilder::from(outer_scan) + .filter(not_in_subquery(value, subquery))? + .build()?; + + let optimized = optimize_with_decorrelate(plan)?; + assert!( + has_null_aware_left_anti_join(&optimized), + "{}", + optimized.display_indent_schema() + ); + + Ok(()) + } + + /// The same rewrite must not fire for a correlated subquery: the + /// correlation predicate is a second equi-join key, and null-aware hash + /// joins accept only one. + #[test] + fn constant_not_in_correlated_subquery_is_not_rewritten() -> Result<()> { + let outer_scan = nullable_scalar_mark_scan("outer_t")?; + let inner_scan = nullable_scalar_mark_scan("inner_t")?; + + let subquery = Arc::new( + LogicalPlanBuilder::from(inner_scan) + .filter( + out_ref_col(DataType::Int32, "outer_t.grp").eq(col("inner_t.grp")), + )? + .project(vec![col("inner_t.id")])? + .build()?, + ); + + let plan = LogicalPlanBuilder::from(outer_scan) + .filter(not_in_subquery(lit(3i32), subquery))? + .build()?; + + assert_optimized_plan_equal!( + plan, + @r" + LeftAnti Join: Filter: Int32(3) = __correlated_sq_1.id AND outer_t.grp = __correlated_sq_1.grp null_aware [id:Int32;N, grp:Int32;N] + TableScan: outer_t [id:Int32;N, grp:Int32;N] + SubqueryAlias: __correlated_sq_1 [id:Int32;N, grp:Int32;N] + Projection: inner_t.id, inner_t.grp [id:Int32;N, grp:Int32;N] + TableScan: inner_t [id:Int32;N, grp:Int32;N] + " + ) + } + #[test] fn correlated_not_in_mark_join_is_null_aware_for_hashable_filter() -> Result<()> { let outer_scan = nullable_scalar_mark_scan("outer_t")?; diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt index 33e6099900203..506fdb2882c44 100644 --- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt +++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt @@ -942,3 +942,140 @@ DROP TABLE nai_res_outer; statement ok DROP TABLE nai_res_inner; + +############# +## Regression: `join_keys_may_be_null` must see expression nullability, not +## just the schema nullability of the columns an expression references. +## https://github.com/apache/datafusion/issues/25473 +## https://github.com/apache/datafusion/issues/25474 +## +## A NULL constant, or an expression that is nullable even though every +## column it reads is `NOT NULL` (`CASE` without `ELSE`, `NULLIF`), must still +## make `NOT IN` null-aware. Otherwise the NULL outer value is never +## recognized: the predicate is pushed into the subquery and the row is kept +## instead of becoming UNKNOWN. +############# + +statement ok +CREATE TABLE nainull_o(x INT NOT NULL) AS VALUES (1), (2); + +statement ok +CREATE TABLE nainull_i(id INT NOT NULL) AS VALUES (1), (3); + +statement ok +CREATE TABLE nainull_ie(id INT NOT NULL) AS SELECT * FROM (VALUES (1)) t(id) WHERE false; + +# A typed NULL constant against a NOT NULL subquery column. +query I +SELECT x FROM nainull_o WHERE CAST(NULL AS INT) NOT IN (SELECT id FROM nainull_i) ORDER BY x; +---- + +query I +SELECT x FROM nainull_o WHERE NULL NOT IN (SELECT id FROM nainull_i) ORDER BY x; +---- + +query I +SELECT x FROM nainull_o WHERE NULLIF(1, 1) NOT IN (SELECT id FROM nainull_i) ORDER BY x; +---- + +# The mark form behind an OR: the NULL constant's UNKNOWN must survive. +query I +SELECT x FROM nainull_o WHERE CAST(NULL AS INT) NOT IN (SELECT id FROM nainull_i) OR x = 99 ORDER BY x; +---- + +# Double negation flips the outer row set: false -> true. +query I +SELECT x FROM nainull_o WHERE NOT (CAST(NULL AS INT) NOT IN (SELECT id FROM nainull_i)) ORDER BY x; +---- + +# Control: `NOT IN ()` is TRUE even for a NULL outer value. +query I +SELECT x FROM nainull_o WHERE CAST(NULL AS INT) NOT IN (SELECT id FROM nainull_ie) ORDER BY x; +---- +1 +2 + +query TT +EXPLAIN SELECT x FROM nainull_o WHERE CAST(NULL AS INT) NOT IN (SELECT id FROM nainull_i); +---- +logical_plan +01)Projection: nainull_o.x +02)--LeftAnti Join: __correlated_sq_1_value = __correlated_sq_1.id null_aware +03)----Projection: nainull_o.x, Int32(NULL) AS __correlated_sq_1_value +04)------TableScan: nainull_o projection=[x] +05)----SubqueryAlias: __correlated_sq_1 +06)------TableScan: nainull_i projection=[id] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(__correlated_sq_1_value@1, id@0)], projection=[x@0], null_aware +02)--ProjectionExec: expr=[x@0 as x, NULL as __correlated_sq_1_value] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)--DataSourceExec: partitions=1, partition_sizes=[1] + +statement ok +DROP TABLE nainull_o; + +statement ok +DROP TABLE nainull_i; + +statement ok +DROP TABLE nainull_ie; + +# An outer expression that is nullable even though the only column it reads +# (`x`) is `NOT NULL`. `on2` is the same data with a nullable `x`, as a +# control that this was already correct via column nullability. +statement ok +CREATE TABLE nainull_e_o(x INT NOT NULL) AS VALUES (1), (2); + +statement ok +CREATE TABLE nainull_e_o2(x INT) AS VALUES (1), (2); + +statement ok +CREATE TABLE nainull_e_i(id INT NOT NULL) AS VALUES (1), (3); + +# `CASE` without `ELSE` is NULL for every row here. +query I +SELECT x FROM nainull_e_o WHERE (CASE WHEN x > 100 THEN 1 END) NOT IN (SELECT id FROM nainull_e_i) ORDER BY x; +---- + +query I +SELECT x FROM nainull_e_o WHERE NULLIF(x, x) NOT IN (SELECT id FROM nainull_e_i) ORDER BY x; +---- + +query I +SELECT x FROM nainull_e_o WHERE x + CAST(NULL AS INT) NOT IN (SELECT id FROM nainull_e_i) ORDER BY x; +---- + +query I +SELECT x FROM nainull_e_o WHERE (CASE WHEN x > 100 THEN 1 END) NOT IN (SELECT id FROM nainull_e_i) OR x = 99 ORDER BY x; +---- + +# Same expression, but the column it reads is nullable: must give the same +# (correct) result as the `NOT NULL` cases above. +query I +SELECT x FROM nainull_e_o2 WHERE (CASE WHEN x > 100 THEN 1 END) NOT IN (SELECT id FROM nainull_e_i) ORDER BY x; +---- + +query TT +EXPLAIN SELECT x FROM nainull_e_o WHERE (CASE WHEN x > 100 THEN 1 END) NOT IN (SELECT id FROM nainull_e_i); +---- +logical_plan +01)LeftAnti Join: CASE WHEN nainull_e_o.x > Int32(100) THEN Int64(1) END = __correlated_sq_1.nainull_e_i.id null_aware +02)--TableScan: nainull_e_o projection=[x] +03)--SubqueryAlias: __correlated_sq_1 +04)----Projection: CAST(nainull_e_i.id AS Int64) +05)------TableScan: nainull_e_i projection=[id] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(nainull_e_i.id@0, CASE WHEN nainull_e_o.x > Int32(100) THEN Int64(1) END@1)], projection=[x@0], null_aware +02)--ProjectionExec: expr=[CAST(id@0 AS Int64) as nainull_e_i.id] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)--ProjectionExec: expr=[x@0 as x, CASE WHEN x@0 > 100 THEN 1 END as CASE WHEN nainull_e_o.x > Int32(100) THEN Int64(1) END] +05)----DataSourceExec: partitions=1, partition_sizes=[1] + +statement ok +DROP TABLE nainull_e_o; + +statement ok +DROP TABLE nainull_e_o2; + +statement ok +DROP TABLE nainull_e_i; From b0220d87e5254157d2ab344ef1406c023660cdbc Mon Sep 17 00:00:00 2001 From: naman Date: Wed, 23 Sep 2026 10:11:19 +0530 Subject: [PATCH 2/4] Correct a misleading comment in the NOT IN null-aware regression test The double-negation query was labeled as demonstrating a flip from false to true. NOT UNKNOWN is UNKNOWN, not TRUE; the query was always empty and is a control from the issue's own table, not evidence of the fix. --- datafusion/sqllogictest/test_files/null_aware_anti_join.slt | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt index 506fdb2882c44..b59137044d69b 100644 --- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt +++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt @@ -983,7 +983,9 @@ query I SELECT x FROM nainull_o WHERE CAST(NULL AS INT) NOT IN (SELECT id FROM nainull_i) OR x = 99 ORDER BY x; ---- -# Double negation flips the outer row set: false -> true. +# Control from the issue's own table: `NOT (NULL NOT IN (...))` is UNKNOWN, +# same as the bare predicate, so this was already correct on `main` and must +# stay that way. query I SELECT x FROM nainull_o WHERE NOT (CAST(NULL AS INT) NOT IN (SELECT id FROM nainull_i)) ORDER BY x; ---- From 5509608bb2803170cf4144a7bd1cd0ff49ba4718 Mon Sep 17 00:00:00 2001 From: naman Date: Wed, 23 Sep 2026 11:34:22 +0530 Subject: [PATCH 3/4] Add unit tests for branches codecov flagged as uncovered - A correlated NOT IN whose value equality is NOT NULL but whose correlation equality is nullable, checked second, exercises the loop continuing past the first conjunct. - An operand referencing columns from both schemas exercises the conservative fallback in operand_may_be_null directly. Verified each fails when the corresponding code path is broken. --- .../src/decorrelate_predicate_subquery.rs | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 40a948838b17f..55328a522a824 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -838,6 +838,17 @@ mod tests { table_scan(Some(name), &schema, None)?.build() } + /// A scan with one `NOT NULL` column (`a`) and one nullable column + /// (`grp`), for tests that need the two conjuncts of a correlated + /// `NOT IN` join filter to disagree on nullability. + fn mixed_nullability_scan(name: &str) -> Result { + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("grp", DataType::Int32, true), + ]); + table_scan(Some(name), &schema, None)?.build() + } + fn has_null_aware_left_anti_join(plan: &LogicalPlan) -> bool { if let LogicalPlan::Join(join) = plan && join.join_type == JoinType::LeftAnti @@ -1639,6 +1650,54 @@ mod tests { Ok(()) } + /// `join_keys_may_be_null` walks every equality conjunct rather than + /// stopping at the first: a correlated `NOT IN` whose value equality is + /// `NOT NULL` must still come out nullable when the correlation equality, + /// checked second, is not. + #[test] + fn correlated_not_in_is_null_aware_when_only_the_correlation_key_is_nullable() + -> Result<()> { + let outer_scan = mixed_nullability_scan("outer_t")?; + let inner_scan = mixed_nullability_scan("inner_t")?; + + let subquery = Arc::new( + LogicalPlanBuilder::from(inner_scan) + .filter( + out_ref_col(DataType::Int32, "outer_t.grp").eq(col("inner_t.grp")), + )? + .project(vec![col("inner_t.a")])? + .build()?, + ); + + let plan = LogicalPlanBuilder::from(outer_scan) + .filter(not_in_subquery(col("outer_t.a"), subquery))? + .build()?; + + let optimized = optimize_with_decorrelate(plan)?; + assert!( + has_null_aware_left_anti_join(&optimized), + "{}", + optimized.display_indent_schema() + ); + + Ok(()) + } + + /// An operand that mixes columns from both sides of the join can't be + /// attributed to either schema, so `operand_may_be_null` must fall back + /// to the conservative answer rather than guessing or erroring. + #[test] + fn operand_referencing_both_sides_is_treated_as_nullable() -> Result<()> { + let left_schema = Arc::clone(test_table_scan_with_name("left_t")?.schema()); + let right_schema = Arc::clone(test_table_scan_with_name("right_t")?.schema()); + + let mixed = col("left_t.a").add(col("right_t.a")); + + assert!(operand_may_be_null(&mixed, &left_schema, &right_schema)?); + + Ok(()) + } + /// The same rewrite must not fire for a correlated subquery: the /// correlation predicate is a second equi-join key, and null-aware hash /// joins accept only one. From e85696b758c031588ade1fe41095fe1e22bbf3f3 Mon Sep 17 00:00:00 2001 From: naman Date: Fri, 25 Sep 2026 02:53:23 +0530 Subject: [PATCH 4/4] Update a stale snapshot after rebasing onto main #25560, merged after this branch, changed build_join to project a constant IN/NOT IN value as a real column so it takes the on[0] key position instead of the correlation predicate. That now also fires for this test's correlated + null-aware case, which is exactly the scenario #25560's own comment on build_join describes. The plan is still null-aware with both equi-join conditions; only the constant's representation changed. --- .../src/decorrelate_predicate_subquery.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 55328a522a824..057d840ecdbe1 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -1721,12 +1721,14 @@ mod tests { assert_optimized_plan_equal!( plan, - @r" - LeftAnti Join: Filter: Int32(3) = __correlated_sq_1.id AND outer_t.grp = __correlated_sq_1.grp null_aware [id:Int32;N, grp:Int32;N] - TableScan: outer_t [id:Int32;N, grp:Int32;N] - SubqueryAlias: __correlated_sq_1 [id:Int32;N, grp:Int32;N] - Projection: inner_t.id, inner_t.grp [id:Int32;N, grp:Int32;N] - TableScan: inner_t [id:Int32;N, grp:Int32;N] + @" + Projection: outer_t.id, outer_t.grp [id:Int32;N, grp:Int32;N] + LeftAnti Join: Filter: __correlated_sq_1_value = __correlated_sq_1.id AND outer_t.grp = __correlated_sq_1.grp null_aware [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32] + Projection: outer_t.id, outer_t.grp, Int32(3) AS __correlated_sq_1_value [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32] + TableScan: outer_t [id:Int32;N, grp:Int32;N] + SubqueryAlias: __correlated_sq_1 [id:Int32;N, grp:Int32;N] + Projection: inner_t.id, inner_t.grp [id:Int32;N, grp:Int32;N] + TableScan: inner_t [id:Int32;N, grp:Int32;N] " ) }