Support large input batches in null-aware sort-merge joins under memory limits #27054

Is your feature request related to a problem or challenge?

This issue originated from PR #26090 and Issue #25546.

PR #26090 adds support for uncorrelated, single-key null-aware LeftAnti sort-merge joins, allowing supported NOT IN queries to use SortMergeJoinExec when datafusion.optimizer.prefer_hash_join = false.

The initial implementation requires single-partition inputs. However the original reproduce from #25546 was reported to still fail under a 100 MB memory limit because unnest(generate_series(...)) produces a single 20-million-row batch (in left partitions). This batch is not split before reaching the sort.

Run:

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

Contents of repro.sql:

set datafusion.execution.target_partitions = 4;
set datafusion.optimizer.prefer_hash_join = false;

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);

select count(*) from big where x not in (select y from small);

Reported error on the PR #26090:

Failed to allocate additional 152.7 MB for ExternalSorter[0]

With normally batched input using the table-function form of
generate_series, the reviewer reported that the query succeeds and returns
19999000.

Describe the solution you'd like

Allow this null-aware sort-merge join query to complete under the same
memory limit when its input arrives as a single giant batch.

(Or allow this null-aware sort-merge join query to complete under a 100 MB memory limit even when the input arrives as a single batch containing 20 million rows.)

One possible approach, mentioned in the TODO in PR #26090 SortMergeJoinExec::input_distribution_requirements (file: datafusion/physical-plan/src/joins/sort_merge_join/exec.rs), is to allow multiple left partitions while sharing the right-side summary needed for null-aware semantics. The partitioning and right-input handling would need to preserve
correct join results.

The implementation must preserve the NOT IN semetics:

• An empty right input retains every left row, including NULL keys.
• A right input containing a NULL key produces no output.
• Otherwise, only unmatched, non-NULL left keys are retained.

Add regression coverage for giant input batches under a constrained memory pool, alongside coverage of these null-aware semantics.

Describe alternatives you've considered

• Split oversized batches before sorting, if this can be done within the memory limit.

Additional context

This is a follow-up to PR #26090 and the original request in Issue #25546.

Review discussion:
#26090 (comment)

The failure above was reported during review of PR #26090 ; it has not been independently reverified for this issue.