feat: Add automatic null-aware anti join for NOT IN subqueries

This commit implements Phase 2 of null-aware anti join support, enabling
automatic detection and configuration of null-aware semantics for SQL
NOT IN subqueries.

DataFusion now automatically provides correct SQL NOT IN semantics with
three-valued logic. When users write NOT IN subqueries, the optimizer
automatically detects them and enables null-aware execution.

- Added `null_aware: bool` field to `Join` struct in logical plan
- Updated `Join::try_new()` and related APIs to accept null_aware parameter
- Added `LogicalPlanBuilder::join_detailed_with_options()` for explicit
  null_aware control
- Updated all Join construction sites across the codebase

- Modified `DecorrelatePredicateSubquery` optimizer to automatically set
  `null_aware: true` for LeftAnti joins (NOT IN subqueries)
- Uses new `join_detailed_with_options()` API to pass the flag
- Conservative approach: all LeftAnti joins use null-aware semantics

- Added checks in `JoinSelection` physical optimizer to prevent swapping
  null-aware anti joins
- Null-aware LeftAnti joins cannot be swapped to RightAnti because:
  - Validation only allows LeftAnti with null_aware=true
  - NULL-handling semantics are asymmetric between sides
- Added checks in 5 locations: try_collect_left, partitioned_hash_join,
  partition mode optimization, and hash_join_swap_subrule

- Added new SQL logic test file with 13 comprehensive test scenarios
- Tests cover: NULL in subquery, NULL in outer table, empty subquery,
  complex expressions, multiple NOT IN conditions, correlated subqueries
- Includes EXPLAIN tests to verify correct plan generation
- All existing optimizer and hash join tests continue to pass

- datafusion/expr/src/logical_plan/plan.rs
- datafusion/expr/src/logical_plan/builder.rs
- datafusion/expr/src/logical_plan/tree_node.rs
- datafusion/optimizer/src/decorrelate_predicate_subquery.rs
- datafusion/optimizer/src/eliminate_cross_join.rs
- datafusion/optimizer/src/eliminate_outer_join.rs
- datafusion/optimizer/src/extract_equijoin_predicate.rs
- datafusion/physical-optimizer/src/join_selection.rs
- datafusion/physical-optimizer/src/enforce_distribution.rs
- datafusion/core/src/physical_planner.rs
- datafusion/proto/src/physical_plan/mod.rs
- datafusion/sqllogictest/test_files/null_aware_anti_join.slt (new)

Before (Phase 1 - manual):
```rust
HashJoinExec::try_new(..., true /* null_aware */)
```

After (Phase 2 - automatic):
```sql
SELECT * FROM orders WHERE order_id NOT IN (SELECT order_id FROM cancelled)
```

The optimizer automatically handles null-aware semantics.

- SQL logic tests: All passed
- Optimizer tests: 568 passed
- Hash join tests: 610 passed
- Physical optimizer tests: 16 passed

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: viirya

datafusion/core/src/physical_planner.rs

@@ -1091,6 +1091,7 @@ impl DefaultPhysicalPlanner {
                 filter,
                 join_type,
                 null_equality,
+                null_aware,
                 schema: join_schema,
                 ..
             }) => {
@@ -1497,6 +1498,7 @@ impl DefaultPhysicalPlanner {
                         None,
                         PartitionMode::Auto,
                         *null_equality,
+                        *null_aware,
                     )?)
                 } else {
                     Arc::new(HashJoinExec::try_new(
@@ -1508,6 +1510,7 @@ impl DefaultPhysicalPlanner {
                         None,
                         PartitionMode::CollectLeft,
                         *null_equality,
+                        *null_aware,
                     )?)
                 };
 

datafusion/expr/src/logical_plan/builder.rs

@@ -1011,6 +1011,18 @@ impl LogicalPlanBuilder {
         join_keys: (Vec<impl Into<Column>>, Vec<impl Into<Column>>),
         filter: Option<Expr>,
         null_equality: NullEquality,
+    ) -> Result<Self> {
+        self.join_detailed_with_options(right, join_type, join_keys, filter, null_equality, false)
+    }
+
+    pub fn join_detailed_with_options(
+        self,
+        right: LogicalPlan,
+        join_type: JoinType,
+        join_keys: (Vec<impl Into<Column>>, Vec<impl Into<Column>>),
+        filter: Option<Expr>,
+        null_equality: NullEquality,
+        null_aware: bool,
     ) -> Result<Self> {
         if join_keys.0.len() != join_keys.1.len() {
             return plan_err!("left_keys and right_keys were not the same length");
@@ -1128,6 +1140,7 @@ impl LogicalPlanBuilder {
             join_constraint: JoinConstraint::On,
             schema: DFSchemaRef::new(join_schema),
             null_equality,
+            null_aware,
         })))
     }
 
@@ -1201,6 +1214,7 @@ impl LogicalPlanBuilder {
                 join_type,
                 JoinConstraint::Using,
                 NullEquality::NullEqualsNothing,
+                false, // null_aware
             )?;
 
             Ok(Self::new(LogicalPlan::Join(join)))
@@ -1217,6 +1231,7 @@ impl LogicalPlanBuilder {
             JoinType::Inner,
             JoinConstraint::On,
             NullEquality::NullEqualsNothing,
+            false, // null_aware
         )?;
 
         Ok(Self::new(LogicalPlan::Join(join)))
@@ -1471,6 +1486,7 @@ impl LogicalPlanBuilder {
             join_type,
             JoinConstraint::On,
             NullEquality::NullEqualsNothing,
+            false, // null_aware
         )?;
 
         Ok(Self::new(LogicalPlan::Join(join)))

datafusion/expr/src/logical_plan/plan.rs

@@ -661,6 +661,7 @@ impl LogicalPlan {
                 on,
                 schema: _,
                 null_equality,
+                null_aware,
             }) => {
                 let schema =
                     build_join_schema(left.schema(), right.schema(), &join_type)?;
@@ -682,6 +683,7 @@ impl LogicalPlan {
                     filter,
                     schema: DFSchemaRef::new(schema),
                     null_equality,
+                    null_aware,
                 }))
             }
             LogicalPlan::Subquery(_) => Ok(self),
@@ -942,6 +944,7 @@ impl LogicalPlan {
                     filter: filter_expr,
                     schema: DFSchemaRef::new(schema),
                     null_equality: *null_equality,
+                    null_aware: false,
                 }))
             }
             LogicalPlan::Subquery(Subquery {
@@ -3781,6 +3784,12 @@ pub struct Join {
     pub schema: DFSchemaRef,
     /// Defines the null equality for the join.
     pub null_equality: NullEquality,
+    /// Whether this is a null-aware anti join (for NOT IN semantics).
+    /// Only applies to LeftAnti joins. When true, implements SQL NOT IN semantics where:
+    /// - If the right side (subquery) contains any NULL in join keys, no rows are output
+    /// - Left side rows with NULL in join keys are not output
+    /// This is required for correct NOT IN subquery behavior with three-valued logic.
+    pub null_aware: bool,
 }
 
 impl Join {
@@ -3798,6 +3807,7 @@ impl Join {
     /// * `join_type` - Type of join (Inner, Left, Right, etc.)
     /// * `join_constraint` - Join constraint (On, Using)
     /// * `null_equality` - How to handle nulls in join comparisons
+    /// * `null_aware` - Whether this is a null-aware anti join (for NOT IN semantics)
     ///
     /// # Returns
     ///
@@ -3810,6 +3820,7 @@ impl Join {
         join_type: JoinType,
         join_constraint: JoinConstraint,
         null_equality: NullEquality,
+        null_aware: bool,
     ) -> Result<Self> {
         let join_schema = build_join_schema(left.schema(), right.schema(), &join_type)?;
 
@@ -3822,6 +3833,7 @@ impl Join {
             join_constraint,
             schema: Arc::new(join_schema),
             null_equality,
+            null_aware,
         })
     }
 
@@ -3877,6 +3889,7 @@ impl Join {
                 join_constraint: original_join.join_constraint,
                 schema: Arc::new(join_schema),
                 null_equality: original_join.null_equality,
+                null_aware: original_join.null_aware,
             },
             requalified,
         ))

datafusion/expr/src/logical_plan/tree_node.rs

@@ -133,6 +133,7 @@ impl TreeNode for LogicalPlan {
                 join_constraint,
                 schema,
                 null_equality,
+                null_aware,
             }) => (left, right).map_elements(f)?.update_data(|(left, right)| {
                 LogicalPlan::Join(Join {
                     left,
@@ -143,6 +144,7 @@ impl TreeNode for LogicalPlan {
                     join_constraint,
                     schema,
                     null_equality,
+                    null_aware,
                 })
             }),
             LogicalPlan::Limit(Limit { skip, fetch, input }) => input
@@ -564,6 +566,7 @@ impl LogicalPlan {
                 join_constraint,
                 schema,
                 null_equality,
+                null_aware,
             }) => (on, filter).map_elements(f)?.update_data(|(on, filter)| {
                 LogicalPlan::Join(Join {
                     left,
@@ -574,6 +577,7 @@ impl LogicalPlan {
                     join_constraint,
                     schema,
                     null_equality,
+                    null_aware,
                 })
             }),
             LogicalPlan::Sort(Sort { expr, input, fetch }) => expr

datafusion/optimizer/src/decorrelate_predicate_subquery.rs

@@ -27,7 +27,7 @@ use crate::{OptimizerConfig, OptimizerRule};
 
 use datafusion_common::alias::AliasGenerator;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
-use datafusion_common::{Column, Result, assert_or_internal_err, plan_err};
+use datafusion_common::{Column, NullEquality, Result, assert_or_internal_err, plan_err};
 use datafusion_expr::expr::{Exists, InSubquery};
 use datafusion_expr::expr_rewriter::create_col_from_scalar_expr;
 use datafusion_expr::logical_plan::{JoinType, Subquery};
@@ -403,6 +403,8 @@ fn build_join(
             // Degenerate case: no right columns referenced by the predicate(s)
             sub_query_alias.clone()
         };
+
+        // Mark joins don't use null-aware semantics (they use three-valued logic with mark column)
         let new_plan = LogicalPlanBuilder::from(left.clone())
             .join_on(right_projected, join_type, Some(join_filter))?
             .build()?;
@@ -415,10 +417,30 @@ fn build_join(
         return Ok(Some(new_plan));
     }
 
+    // Determine if this should be a null-aware anti join
+    // For LeftAnti joins (NOT IN), we need null-aware semantics if:
+    // 1. The join type is LeftAnti
+    // 2. The join predicate involves nullable columns (conservative: assume nullable)
+    let null_aware = matches!(join_type, JoinType::LeftAnti);
+
     // join our sub query into the main plan
-    let new_plan = LogicalPlanBuilder::from(left.clone())
-        .join_on(sub_query_alias, join_type, Some(join_filter))?
-        .build()?;
+    let new_plan = if null_aware {
+        // Use join_detailed_with_options to set null_aware flag
+        LogicalPlanBuilder::from(left.clone())
+            .join_detailed_with_options(
+                sub_query_alias,
+                join_type,
+                (Vec::<Column>::new(), Vec::<Column>::new()), // No equijoin keys, filter-based join
+                Some(join_filter),
+                NullEquality::NullEqualsNothing,
+                true, // null_aware
+            )?
+            .build()?
+    } else {
+        LogicalPlanBuilder::from(left.clone())
+            .join_on(sub_query_alias, join_type, Some(join_filter))?
+            .build()?
+    };
     debug!(
         "predicate subquery optimized:\n{}",
         new_plan.display_indent()

datafusion/optimizer/src/eliminate_cross_join.rs

@@ -341,6 +341,7 @@ fn find_inner_join(
                 filter: None,
                 schema: join_schema,
                 null_equality,
+                null_aware: false,
             }));
         }
     }
@@ -363,6 +364,7 @@ fn find_inner_join(
         join_type: JoinType::Inner,
         join_constraint: JoinConstraint::On,
         null_equality,
+        null_aware: false,
     }))
 }
 
@@ -1367,6 +1369,7 @@ mod tests {
             filter: None,
             schema: join_schema,
             null_equality: NullEquality::NullEqualsNull, // Test preservation
+            null_aware: false,
         });
 
         // Apply filter that can create join conditions

datafusion/optimizer/src/eliminate_outer_join.rs

@@ -119,6 +119,7 @@ impl OptimizerRule for EliminateOuterJoin {
                         filter: join.filter.clone(),
                         schema: Arc::clone(&join.schema),
                         null_equality: join.null_equality,
+                        null_aware: join.null_aware,
                     }));
                     Filter::try_new(filter.predicate, new_join)
                         .map(|f| Transformed::yes(LogicalPlan::Filter(f)))

datafusion/optimizer/src/extract_equijoin_predicate.rs

@@ -76,6 +76,7 @@ impl OptimizerRule for ExtractEquijoinPredicate {
                 join_constraint,
                 schema,
                 null_equality,
+                null_aware,
             }) => {
                 let left_schema = left.schema();
                 let right_schema = right.schema();
@@ -117,6 +118,7 @@ impl OptimizerRule for ExtractEquijoinPredicate {
                             // According to `is not distinct from`'s semantics, it's
                             // safe to override it
                             null_equality: NullEquality::NullEqualsNull,
+                            null_aware,
                         })));
                     }
                 }
@@ -132,6 +134,7 @@ impl OptimizerRule for ExtractEquijoinPredicate {
                         join_constraint,
                         schema,
                         null_equality,
+                        null_aware,
                     })))
                 } else {
                     Ok(Transformed::no(LogicalPlan::Join(Join {
@@ -143,6 +146,7 @@ impl OptimizerRule for ExtractEquijoinPredicate {
                         join_constraint,
                         schema,
                         null_equality,
+                        null_aware,
                     })))
                 }
             }

datafusion/physical-optimizer/src/enforce_distribution.rs

@@ -295,6 +295,7 @@ pub fn adjust_input_keys_ordering(
         projection,
         mode,
         null_equality,
+        null_aware,
         ..
     }) = plan.as_any().downcast_ref::<HashJoinExec>()
     {
@@ -314,6 +315,7 @@ pub fn adjust_input_keys_ordering(
                         projection.clone(),
                         PartitionMode::Partitioned,
                         *null_equality,
+                        *null_aware,
                     )
                     .map(|e| Arc::new(e) as _)
                 };
@@ -618,6 +620,7 @@ pub fn reorder_join_keys_to_inputs(
         projection,
         mode,
         null_equality,
+        null_aware,
         ..
     }) = plan_any.downcast_ref::<HashJoinExec>()
     {
@@ -644,6 +647,7 @@ pub fn reorder_join_keys_to_inputs(
                     projection.clone(),
                     PartitionMode::Partitioned,
                     *null_equality,
+                    *null_aware,
                 )?));
             }
         }

datafusion/physical-optimizer/src/join_selection.rs

@@ -184,7 +184,9 @@ pub(crate) fn try_collect_left(
 
     match (left_can_collect, right_can_collect) {
         (true, true) => {
+            // Don't swap null-aware anti joins as they have specific side requirements
             if hash_join.join_type().supports_swap()
+                && !hash_join.null_aware
                 && should_swap_join_order(&**left, &**right)?
             {
                 Ok(Some(hash_join.swap_inputs(PartitionMode::CollectLeft)?))
@@ -198,6 +200,7 @@ pub(crate) fn try_collect_left(
                     hash_join.projection.clone(),
                     PartitionMode::CollectLeft,
                     hash_join.null_equality(),
+                    hash_join.null_aware,
                 )?)))
             }
         }
@@ -210,9 +213,11 @@ pub(crate) fn try_collect_left(
             hash_join.projection.clone(),
             PartitionMode::CollectLeft,
             hash_join.null_equality(),
+            hash_join.null_aware,
         )?))),
         (false, true) => {
-            if hash_join.join_type().supports_swap() {
+            // Don't swap null-aware anti joins as they have specific side requirements
+            if hash_join.join_type().supports_swap() && !hash_join.null_aware {
                 hash_join.swap_inputs(PartitionMode::CollectLeft).map(Some)
             } else {
                 Ok(None)
@@ -232,7 +237,10 @@ pub(crate) fn partitioned_hash_join(
 ) -> Result<Arc<dyn ExecutionPlan>> {
     let left = hash_join.left();
     let right = hash_join.right();
-    if hash_join.join_type().supports_swap() && should_swap_join_order(&**left, &**right)?
+    // Don't swap null-aware anti joins as they have specific side requirements
+    if hash_join.join_type().supports_swap()
+        && !hash_join.null_aware
+        && should_swap_join_order(&**left, &**right)?
     {
         hash_join.swap_inputs(PartitionMode::Partitioned)
     } else {
@@ -245,6 +253,7 @@ pub(crate) fn partitioned_hash_join(
             hash_join.projection.clone(),
             PartitionMode::Partitioned,
             hash_join.null_equality(),
+            hash_join.null_aware,
         )?))
     }
 }
@@ -277,7 +286,9 @@ fn statistical_join_selection_subrule(
                 PartitionMode::Partitioned => {
                     let left = hash_join.left();
                     let right = hash_join.right();
+                    // Don't swap null-aware anti joins as they have specific side requirements
                     if hash_join.join_type().supports_swap()
+                        && !hash_join.null_aware
                         && should_swap_join_order(&**left, &**right)?
                     {
                         hash_join
@@ -484,6 +495,7 @@ pub fn hash_join_swap_subrule(
     if let Some(hash_join) = input.as_any().downcast_ref::<HashJoinExec>()
         && hash_join.left.boundedness().is_unbounded()
         && !hash_join.right.boundedness().is_unbounded()
+        && !hash_join.null_aware // Don't swap null-aware anti joins
         && matches!(
             *hash_join.join_type(),
             JoinType::Inner | JoinType::Left | JoinType::LeftSemi | JoinType::LeftAnti