Optimize null-aware anti join to skip CollectLeft when keys are non-nullable

When all join keys are non-nullable on both sides, we don't need null-aware
semantics because NULLs cannot exist in the data. This allows the query to use
regular Partitioned mode instead of the more expensive CollectLeft mode.

Implementation:
- Added join_keys_may_be_null() helper function that checks schema nullability
- Modified null_aware flag logic to only enable when:
  1. It's a NOT IN subquery (not NOT EXISTS)
  2. AND at least one join key column is nullable

Benefits:
- Queries with NOT NULL constraints can use Partitioned mode (better parallelism)
- Avoids unnecessary CollectLeft overhead when null-aware semantics aren't needed
- Regular anti join is cheaper than null-aware (no atomic flag synchronization)

Example: SELECT * FROM t1 WHERE id NOT IN (SELECT id FROM t2)
- If t1.id and t2.id are NOT NULL: uses regular anti join with Partitioned mode
- If either is nullable: uses null-aware anti join with CollectLeft mode

Addresses review comment on join_selection.rs L251 by detecting nullability
earlier in the optimizer rather than in the physical optimizer.

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

Co-Authored-By: Claude Sonnet 4.5 <[email protected]>

Author: viirya

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, NullEquality, Result, assert_or_internal_err, plan_err};
+use datafusion_common::{Column, DFSchemaRef, ExprSchema, 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};
@@ -310,6 +310,39 @@ 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.
+/// 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(
+    join_filter: &Expr,
+    left_schema: &DFSchemaRef,
+    right_schema: &DFSchemaRef,
+) -> Result<bool> {
+    // 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) {
+            if field.as_ref().is_nullable() {
+                return Ok(true);
+            }
+        }
+        // Check in right schema
+        if let Ok(field) = right_schema.field_from_column(&col) {
+            if field.as_ref().is_nullable() {
+                return Ok(true);
+            }
+        }
+    }
+
+    Ok(false)
+}
+
 fn build_join(
     left: &LogicalPlan,
     subquery: &LogicalPlan,
@@ -422,8 +455,12 @@ fn build_join(
     // - NOT IN: Uses three-valued logic, requires null-aware handling
     // - NOT EXISTS: Uses two-valued logic, regular anti join is correct
     // We can distinguish them: NOT IN has in_predicate_opt, NOT EXISTS does not
-    let null_aware =
-        matches!(join_type, JoinType::LeftAnti) && in_predicate_opt.is_some();
+    //
+    // Additionally, if the join keys are non-nullable on both sides, we don't need
+    // null-aware semantics because NULLs cannot exist in the data.
+    let null_aware = matches!(join_type, JoinType::LeftAnti)
+        && in_predicate_opt.is_some()
+        && join_keys_may_be_null(&join_filter, left.schema(), sub_query_alias.schema())?;
 
     // join our sub query into the main plan
     let new_plan = if null_aware {