Support null-aware (`NOT IN`) anti / mark joins in `SortMergeJoinExec`

[jayzhan211]

Is your feature request related to a problem or challenge?

x NOT IN (subquery) is planned as a null-aware join: three-valued logic makes the result depend on facts about the whole subquery side, not on any single key comparison. For the uncorrelated LeftAnti form:

subquery side	result
empty	every outer row, including rows whose x is NULL
contains a NULL key	no rows
otherwise	outer rows with a non-NULL x that matches nothing

Only HashJoinExec implements this (the null_aware flag, pinned to PartitionMode::CollectLeft). SortMergeJoinExec does not, so the planner always routes these joins to a hash join, even with prefer_hash_join = false (#22810 fixed the wrong results that came from not doing so).

The consequence is that a null-aware join is the one join shape with no memory-bounded execution path. It is pinned to CollectLeft, prefer_hash_join = false does not apply to it, and the sort-merge fallback proposed in #24768 / #25217 has to exclude it because there is no sort-merge join to fall back to. A large NOT IN fails with Resources exhausted and no setting changes that.

Reproducer — datafusion-cli -m 100M --mem-pool-type fair -f repro.sql:

set datafusion.execution.target_partitions = 4;
set datafusion.optimizer.prefer_hash_join = false;  -- the documented workaround for joins that do not fit
create view big   as select v as x from (select unnest(generate_series(1, 20000000)) as v);
create view small as select v as y from (select unnest(generate_series(1, 1000)) as v);

-- null-aware: fails
select count(*) from big where x not in (select y from small);

-- control: same data, same anti join, without NOT IN semantics: completes through SortMergeJoinExec
select count(*) from big where not exists (select 1 from small where small.y = big.x);

Resources exhausted: Additional allocation failed for HashJoinInput ...
Error: Failed to allocate additional 152.6 MB for HashJoinInput with 0.0 B already allocated for this reservation - 100.0 MB remain available for the total memory pool: fair(pool_size: 100.0 MB)
+----------+
| count(*) |
+----------+
| 19999000 |
+----------+

The plans show why (prefer_hash_join = false in both):

-- x NOT IN (select y from t2)
HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(x@0, y@0)], null_aware

-- NOT EXISTS (select 1 from t2 where t2.y = t1.x)
SortMergeJoinExec: join_type=LeftAnti, on=[(x@0, y@0)]
  SortExec: expr=[x@0 ASC], preserve_partitioning=[false]
  SortExec: expr=[y@0 ASC], preserve_partitioning=[false]

Describe the solution you'd like

Teach SortMergeJoinExec the null-aware semantics, then relax the !null_aware guard in the physical planner. A first step could be limited to the uncorrelated, single-key LeftAnti case.

One observation that may make this cheaper than it looks: both global facts are visible at the head of a sorted input. With nulls first on the subquery side, a partition knows after reading its first subquery row whether it holds any row and whether it holds a NULL key. So the cross-partition part could be a one-time barrier at stream start (each partition reports (saw_row, saw_null)), rather than deferring all output to the end as the hash join's probe-completion tracking has to. A simpler variant is to require a single partition for a null-aware sort-merge join. (An idea only, not prototyped.)

The correlated LeftMark form (value key on[0] plus scope keys on[1..], see the null_aware docs on HashJoinExec) needs the same facts per scope group and can follow separately.

Describe alternatives you've considered

• Keep the status quo: null-aware joins stay hash-only and keep failing under a memory limit.
• Spill inside the hash join instead (the hybrid design in [EPIC] Spilling Hash Join — run any join in a bounded memory budget #24768, CollectLeft phase). That fixes it without a sort-merge join, but is much further out.

Additional context

• [EPIC] Spilling Hash Join — run any join in a bounded memory budget #24768 — spilling hash join EPIC; this ticket was suggested there ([EPIC] Spilling Hash Join — run any join in a bounded memory budget #24768 (comment))
• feat: sort-merge fallback for partitioned hash joins under memory pressure #25217 — sort-merge fallback for hash joins, which excludes null-aware joins for this reason
• fix: NOT IN with NULL subquery returns wrong results under SortMergeJoin #22810 — wrong results when a null-aware join reached SortMergeJoinExec; the guard this ticket would relax
• NestedLoopJoinExec has no null-aware path either (noted in decorrelate_predicate_subquery.rs)
• An implementation should run null_aware_anti_join.slt / null_aware_mark_join.slt with prefer_hash_join = false: a mistake here is silently wrong rows, not an error.