fix: NOT IN (subquery) with a constant value ignores NULLs in the subquery

datafusion/core/src/physical_planner.rs

@@ -1604,6 +1604,17 @@ impl DefaultPhysicalPlanner {
                     && session_state.config().repartition_joins()
                     && !*null_aware;
 
+                // Only `HashJoinExec` implements null-aware semantics, and it
+                // needs equi-join keys to do so. Without them the join would be
+                // planned as a nested loop (or piecewise merge) join, which
+                // silently ignores the flag and returns wrong results for
+                // `NOT IN` over a nullable subquery. Fail loudly instead.
+                if *null_aware && join_on.is_empty() {
+                    return plan_err!(
+                        "null_aware {join_type} join requires equi-join keys, but the join has none"
+                    );
+                }
+
                 // TODO: Allow PWMJ to deal with residual equijoin conditions
                 let join: Arc<dyn ExecutionPlan> = if join_on.is_empty() {
                     if join_filter.is_none() && *join_type == JoinType::Inner {

datafusion/optimizer/src/decorrelate_predicate_subquery.rs

@@ -471,7 +471,14 @@ fn build_join(
             replace_qualified_name(filter, &all_correlated_cols, &alias).map(Some)
         })?;
 
-    let join_filter = match (join_filter_opt, in_predicate_opt.cloned()) {
+    // The outer value expression of an `IN`/`NOT IN` predicate whose join filter
+    // is nothing but that predicate, recorded together with the subquery column
+    // it is compared against and a name for the column it can be projected as.
+    // Correlated subqueries are excluded on purpose: their correlation predicate
+    // is a second join key, and null-aware hash joins accept only a single key.
+    let mut in_value_expr = None;
+
+    let mut join_filter = match (join_filter_opt, in_predicate_opt.cloned()) {
         (
             Some(join_filter),
             Some(Expr::BinaryExpr(BinaryExpr {
@@ -493,14 +500,56 @@ fn build_join(
                 right,
             })),
         ) => {
+            let value_name = format!("{alias}_value");
             let right_col = create_col_from_scalar_expr(&right, alias)?;
+            let value = left.deref().clone();
+            in_value_expr = Some((value.clone(), right_col.clone(), value_name));
 
-            Expr::eq(left.deref().clone(), Expr::Column(right_col))
+            Expr::eq(value, Expr::Column(right_col))
         }
         (None, None) => lit(true),
         _ => return Ok(None),
     };
 
+    // `<constant> IN/NOT IN (<subquery>)`: the outer value expression holds no
+    // column reference, so `<constant> = __correlated_sq.col` is not a valid
+    // equi-join key (see `find_valid_equijoin_key_pair`) and stays in the join
+    // filter. Two things then go wrong for a null-aware join: the filter is
+    // right-only, so `push_down_filter` moves it into the subquery and drops the
+    // very NULLs that make `NOT IN` UNKNOWN, and a join without equi-join keys
+    // is planned as a nested loop join, which has no null-aware implementation.
+    // Projecting the constant as a column of the outer side turns the predicate
+    // into a real equi-join key so the null-aware hash join handles it.
+    let mut projected_left = None;
+    if let Some((value, right_col, mut value_name)) = in_value_expr
+        && value.column_refs().is_empty()
+        && matches!(join_type, JoinType::LeftAnti | JoinType::LeftMark)
+        && join_keys_may_be_null(&join_filter, left.schema(), sub_query_alias.schema())?
+    {
+        // The projected column is unqualified, so a left field that already has
+        // this name — however unlikely — would make the reference ambiguous.
+        let left_schema = left.schema();
+        while left_schema.fields().iter().any(|f| f.name() == &value_name) {
+            value_name.push('_');
+        }
+        let value_col = Column::new_unqualified(value_name);
+        let projections = left_schema
+            .columns()
+            .into_iter()
+            .map(Expr::from)
+            .chain(std::iter::once(value.alias(value_col.name())))
+            .collect::<Vec<_>>();
+        projected_left = Some(
+            LogicalPlanBuilder::from(left.clone())
+                .project(projections)?
+                .build()?,
+        );
+        // `in_value_expr` is only set when the `IN` equality is the whole join
+        // filter, so it can simply be rebuilt against the projected column.
+        join_filter = Expr::eq(Expr::Column(value_col), Expr::Column(right_col));
+    }
+    let left = projected_left.as_ref().unwrap_or(left);
+
     if matches!(join_type, JoinType::LeftMark | JoinType::RightMark) {
         let right_schema = sub_query_alias.schema();
 
@@ -1413,6 +1462,72 @@ mod tests {
         )
     }
 
+    /// A constant value expression has no column, so `Int32(3) = inner_t.id`
+    /// cannot be an equi-join key on its own. The rule projects the constant as
+    /// a column of the outer side; `ExtractEquijoinPredicate` (not run here)
+    /// then turns the filter into a real key for the null-aware hash join.
+    #[test]
+    fn constant_not_in_subquery_projects_value_as_join_key() -> 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)
+                .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"
+        Projection: outer_t.id, outer_t.grp [id:Int32;N, grp:Int32;N]
+          LeftAnti Join:  Filter: __correlated_sq_1_value = __correlated_sq_1.id 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]
+              Projection: inner_t.id [id:Int32;N]
+                TableScan: inner_t [id:Int32;N, grp:Int32;N]
+        "
+        )
+    }
+
+    /// 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")?;

datafusion/optimizer/src/push_down_filter.rs

@@ -406,7 +406,25 @@ fn push_down_all_join(
 ) -> Result<Transformed<LogicalPlan>> {
     let is_inner_join = join.join_type == JoinType::Inner;
     // Get pushable predicates from current optimizer state
-    let (left_preserved, right_preserved) = lr_is_preserved(join.join_type);
+    let (left_preserved, mut right_preserved) = lr_is_preserved(join.join_type);
+    let (on_left_preserved, mut on_right_preserved) = on_lr_is_preserved(join.join_type);
+
+    // Null-aware joins (e.g. `NOT IN` with a nullable subquery) implement SQL
+    // three-valued logic: a NULL join key on the right/subquery side makes the
+    // predicate UNKNOWN and empties the result. Anything pushed into the right
+    // input runs before the join can observe those NULLs, so a null-rejecting
+    // predicate would drop them and silently produce wrong results.
+    // `infer_join_predicates` skips null-aware joins for the same reason.
+    //
+    // `on_right_preserved` is what actually matters here: it is what lets a
+    // right-only join filter — the shape `<constant> NOT IN (<subquery>)`
+    // produces — reach the subquery. `right_preserved` is already false for
+    // every join type that can carry `null_aware` today, so clearing it is
+    // defence in depth.
+    if join.null_aware {
+        right_preserved = false;
+        on_right_preserved = false;
+    }
 
     // The predicates can be divided to three categories:
     // 1) can push through join to its children(left or right)
@@ -447,7 +465,6 @@ fn push_down_all_join(
     }
 
     let mut on_filter_join_conditions = vec![];
-    let (on_left_preserved, on_right_preserved) = on_lr_is_preserved(join.join_type);
     for on in on_filter {
         if on_left_preserved && checker.is_left_only(&on) {
             left_push.push(on)
@@ -3882,6 +3899,46 @@ mod tests {
         )
     }
 
+    /// Regression test for a null-aware LeftAnti join whose join filter only
+    /// references the subquery side, the shape produced by
+    /// `<constant> NOT IN (<subquery>)`. Pushing that filter into the right
+    /// input would drop the subquery's NULL rows before the join can observe
+    /// them, so `NOT IN` would wrongly evaluate to TRUE instead of UNKNOWN.
+    #[test]
+    fn null_aware_left_anti_join_keeps_right_only_join_filter() -> Result<()> {
+        let table_scan = test_table_scan_with_name("test1")?;
+        let left = LogicalPlanBuilder::from(table_scan)
+            .project(vec![col("a"), col("b")])?
+            .build()?;
+        let right_table_scan = test_table_scan_with_name("test2")?;
+        let right = LogicalPlanBuilder::from(right_table_scan)
+            .project(vec![col("a"), col("b")])?
+            .build()?;
+        let plan = LogicalPlanBuilder::from(left)
+            .join_detailed_with_options(
+                right,
+                JoinType::LeftAnti,
+                (Vec::<Column>::new(), Vec::<Column>::new()),
+                Some(lit(3u32).eq(col("test2.a"))),
+                datafusion_common::NullEquality::NullEqualsNothing,
+                true,
+            )?
+            .build()?;
+
+        // `UInt32(3) = test2.a` stays on the join: it must not become a
+        // `TableScan: test2, full_filters=[...]`.
+        assert_optimized_plan_equal!(
+            plan,
+            @r"
+        LeftAnti Join:  Filter: UInt32(3) = test2.a null_aware
+          Projection: test1.a, test1.b
+            TableScan: test1
+          Projection: test2.a, test2.b
+            TableScan: test2
+        "
+        )
+    }
+
     #[test]
     fn left_anti_join_with_filters() -> Result<()> {
         let table_scan = test_table_scan_with_name("test1")?;

datafusion/sqllogictest/test_files/null_aware_anti_join.slt

@@ -564,3 +564,105 @@ DROP TABLE nai_outer;
 
 statement ok
 DROP TABLE nai_inner;
+
+#############
+## Regression: constant (non-column) value expression in `NOT IN (subquery)`
+##
+## `3 NOT IN (1, NULL)` is UNKNOWN, so a `WHERE` clause must remove every row.
+## The value expression `3` contains no column, so on its own it cannot be an
+## equi-join key; decorrelation projects it as a column of the outer side so
+## the null-aware hash join gets a real key. Otherwise the predicate stays in
+## the join filter, where it is pushed into the subquery (dropping the NULL
+## rows before the join can observe them) and leaves a keyless join that no
+## null-aware operator can execute.
+#############
+
+statement ok
+CREATE TABLE naconst_t1(id INT) AS VALUES (1), (2);
+
+statement ok
+CREATE TABLE naconst_t2(id INT) AS VALUES (1), (NULL);
+
+# Q1: expected no rows (subquery yields {1, NULL} => `3 NOT IN {1, NULL}` is UNKNOWN)
+query I
+SELECT id FROM naconst_t1 WHERE 3 NOT IN (SELECT id FROM naconst_t2) ORDER BY id;
+----
+
+# Q2: `NOT (x IN (...))` is the same predicate spelled differently
+query I
+SELECT id FROM naconst_t1 WHERE NOT (3 IN (SELECT id FROM naconst_t2)) ORDER BY id;
+----
+
+# C1: the constant is present in the subquery => FALSE, no rows (already correct)
+query I
+SELECT id FROM naconst_t1 WHERE 1 NOT IN (SELECT id FROM naconst_t2) ORDER BY id;
+----
+
+# C2: no NULL in the subquery => plain TRUE, all rows survive
+query I
+SELECT id FROM naconst_t1 WHERE 3 NOT IN (SELECT id FROM naconst_t2 WHERE id IS NOT NULL) ORDER BY id;
+----
+1
+2
+
+# C3: column value expression (an equi-join key exists) => already correct
+query I
+SELECT id FROM naconst_t1 WHERE id + 2 NOT IN (SELECT id FROM naconst_t2) ORDER BY id;
+----
+
+# A constant that is absent from a NULL-free subquery still yields TRUE even
+# when other outer predicates are present.
+query I
+SELECT id FROM naconst_t1
+WHERE id > 1 AND 3 NOT IN (SELECT id FROM naconst_t2 WHERE id IS NOT NULL)
+ORDER BY id;
+----
+2
+
+# The projected value column is named after the subquery alias. A user column
+# that happens to share that name must not make the reference ambiguous.
+statement ok
+CREATE TABLE naconst_clash("__correlated_sq_1_value" INT) AS VALUES (1), (2);
+
+query I
+SELECT "__correlated_sq_1_value" FROM naconst_clash
+WHERE 3 NOT IN (SELECT id FROM naconst_t2)
+ORDER BY 1;
+----
+
+query I
+SELECT "__correlated_sq_1_value" FROM naconst_clash
+WHERE 3 NOT IN (SELECT id FROM naconst_t2 WHERE id IS NOT NULL)
+ORDER BY 1;
+----
+1
+2
+
+statement ok
+DROP TABLE naconst_clash;
+
+# A constant value expression with a non-equality correlation leaves the
+# null-aware join without any equi-join key. Only `HashJoinExec` implements
+# null-aware semantics and it needs a key, so the planner reports the gap
+# instead of falling back to a nested loop join that ignores the NULLs and
+# silently returns wrong results.
+statement ok
+CREATE TABLE naconst_corr_t1(id INT, g INT) AS VALUES (1, 1), (2, 2);
+
+statement ok
+CREATE TABLE naconst_corr_t2(id INT, g INT) AS VALUES (1, 1), (NULL, 2);
+
+query error DataFusion error: Error during planning: null_aware LeftAnti join requires equi\-join keys, but the join has none
+SELECT id FROM naconst_corr_t1 WHERE 3 NOT IN (SELECT id FROM naconst_corr_t2 WHERE naconst_corr_t2.g > naconst_corr_t1.g);
+
+statement ok
+DROP TABLE naconst_corr_t1;
+
+statement ok
+DROP TABLE naconst_corr_t2;
+
+statement ok
+DROP TABLE naconst_t1;
+
+statement ok
+DROP TABLE naconst_t2;