Repository navigation
Support null-aware (NOT IN) anti / mark joins in SortMergeJoinExec #25546
Description
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 NestedLoopJoinExechas no null-aware path either (noted indecorrelate_predicate_subquery.rs)- An implementation should run
null_aware_anti_join.slt/null_aware_mark_join.sltwithprefer_hash_join = false: a mistake here is silently wrong rows, not an error.
Activity
TinyMurky commentedon Sep 23, 2026
take
test(smj): Cover null-aware NOT IN semantics
test(smj): Cover null-aware NOT IN semantics
test(smj): Cover null-aware executor boundaries
test(smj): Cover null-aware NOT IN semantics
test(smj): Cover null-aware executor boundaries
test(smj): Cover null-aware NOT IN semantics
test(smj): Cover null-aware executor boundaries
TinyMurky commentedon Oct 25, 2026
Hello @jayzhan211!
Sorry for the delay. The initial implementation is ready for review. Could you take a look when you have a chance?
This PR introduces two public API changes:
null_awareparameter toSortMergeJoinExec::try_new, following the interface ofHashJoinExec::try_new.datafusion/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs
Lines 211 to 227 in 7798c94
null_awarefield to the protobuf message and the generated RustSortMergeJoinExecNodestruct.datafusion/datafusion/proto-models/proto/datafusion.proto
Lines 1829 to 1839 in 7798c94
However, the api change mention above trigger the following warning from cargo-semver-checks, the full warning message can be found in here
Could you advise on how to handle the SemVer check for these intentional API changes?
Thank you!
test(smj): Cover null-aware NOT IN semantics
test(smj): Cover null-aware executor boundaries
test(smj): Cover null-aware NOT IN semantics
test(smj): Cover null-aware executor boundaries
test(smj): Cover null-aware NOT IN semantics
test(smj): Cover null-aware executor boundaries
test(smj): Cover null-aware NOT IN semantics
test(smj): Cover null-aware executor boundaries