fix: NOT IN with NULL subquery returns wrong results under SortMergeJoin

datafusion/core/src/physical_planner.rs

@@ -1741,6 +1741,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

@@ -2437,3 +2437,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;