fix: NOT IN with NULL subquery returns wrong results under SortMergeJoin (#22810)

## Problem

`NOT IN (subquery)` is a null-aware anti join: when the subquery yields
a NULL the predicate is never TRUE, so the query must return zero rows.
With `prefer_hash_join = false` and multiple partitions, the planner
routed the null-aware anti join to `SortMergeJoinExec`, which is not
null-aware, so it returned wrong results. HashJoin (the default) was
already correct.

## Proof

```sql
set datafusion.optimizer.prefer_hash_join = false;
create table t1(x int) as values (1);
create table t2(y int) as values (NULL);
select x from t1 where x not in (select y from t2);
```

Expected 0 rows (the subquery contains a NULL). Before this change it
returned `1`. With `prefer_hash_join = true` it correctly returned 0
rows. `EXPLAIN` showed the wrong config selecting `SortMergeJoinExec:
join_type=LeftAnti`.

## Solution

The planner already requires null-aware joins to use the CollectLeft
HashJoin, and the HashJoin branch guards on `!null_aware`. The
SortMergeJoin branch was missing the same guard, so this adds `&&
!*null_aware` to it. Null-aware anti joins now fall through to the
CollectLeft HashJoin regardless of `prefer_hash_join`.
`SortMergeJoinExec` has no `null_aware` parameter and cannot honor these
semantics.

Added a regression test in `subquery.slt` (under `prefer_hash_join =
false`) covering both a null-containing subquery (zero rows) and a
null-free subquery (normal anti join). All 61 SortMergeJoin unit tests
pass.

Author: nathanb9

datafusion/core/src/physical_planner.rs

@@ -1803,6 +1803,11 @@ impl DefaultPhysicalPlanner {
                 } else if session_state.config().target_partitions() > 1
                     && session_state.config().repartition_joins()
                     && !prefer_hash_join
+                    && !*null_aware
+                // Null-aware joins (e.g. `NOT IN` with a nullable subquery) must
+                // use the CollectLeft HashJoin below: SortMergeJoinExec does not
+                // implement null-aware anti-join semantics and would return wrong
+                // results when the right side contains a null join key.
                 {
                     // Use SortMergeJoin if hash join is not preferred
                     let join_on_len = join_on.len();

datafusion/sqllogictest/test_files/subquery.slt

@@ -2599,3 +2599,61 @@ DROP TABLE sq_count_customer;
 
 statement ok
 DROP TABLE sq_count_orders;
+
+# Regression test: `NOT IN` is a null-aware anti join. When the subquery yields a
+# NULL the predicate is never TRUE, so the query must return zero rows. This must
+# hold regardless of the chosen physical join operator. Previously, with
+# prefer_hash_join = false and multiple partitions, the planner routed the
+# null-aware anti join to SortMergeJoin (which is not null-aware) and returned
+# wrong results; null-aware anti joins must use the CollectLeft HashJoin.
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+statement ok
+CREATE TABLE nia_left(x INT) AS VALUES (1), (2), (3), (4);
+
+statement ok
+CREATE TABLE nia_right_with_null(y INT) AS VALUES (2), (NULL);
+
+statement ok
+CREATE TABLE nia_right_no_null(y INT) AS VALUES (2), (4);
+
+# Subquery contains a NULL -> NOT IN must return no rows.
+query I
+SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_null) ORDER BY x;
+----
+
+# The null-aware anti join must be planned as a CollectLeft HashJoinExec even with
+# prefer_hash_join = false: SortMergeJoinExec is not null-aware and must not be used.
+query TT
+EXPLAIN SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_null);
+----
+logical_plan
+01)LeftAnti Join: nia_left.x = __correlated_sq_1.y null_aware
+02)--TableScan: nia_left projection=[x]
+03)--SubqueryAlias: __correlated_sq_1
+04)----TableScan: nia_right_with_null projection=[y]
+physical_plan
+01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(x@0, y@0)], null_aware
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
+03)--DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Subquery has no NULL -> NOT IN behaves like a normal anti join.
+query I
+SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_no_null) ORDER BY x;
+----
+1
+3
+
+statement ok
+DROP TABLE nia_left;
+
+statement ok
+DROP TABLE nia_right_with_null;
+
+statement ok
+DROP TABLE nia_right_no_null;
+
+statement ok
+reset datafusion.optimizer.prefer_hash_join;