Avoid single-task execution for null-aware anti joins (NOT IN subqueries)

[andygrove]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

#2188 fixed the correctness bug in #2187 by executing null-aware anti joins (NOT IN subqueries) in a single task. This is the only correct lowering available today, but it has two significant costs:

1. No parallelism. The join stage collapses to one task regardless of cluster size.
2. The size cap lands on the outer table. DataFusion only supports null_aware on LeftAnti joins with a single-column key (hash_join/exec.rs validation), which means the build side is always the outer table, not the subquery. So ballista.optimizer.broadcast_join_threshold_bytes (default 10 MiB) now effectively caps the size of the fact table in SELECT ... FROM big_fact WHERE key NOT IN (SELECT k FROM small_dim). Any large outer table with known statistics gets a hard planning error.

Why single-task is currently unavoidable: DataFusion's null-aware hash join coordinates three pieces of global state across probe partitions of one operator instance, using in-process shared memory (probe_side_has_null: AtomicBool, probe_side_non_empty: AtomicBool, and the visited-build-row bitmap). The last probe partition to finish emits the unmatched build rows, suppressing all output if any partition saw a NULL key. Split the probe across processes and each task gets private copies of all three, producing duplicated or incorrect rows. Hash-partitioning both sides on the key does not help either, because "probe side has a NULL" is a global property: a NULL key hashes to one partition and every other partition needs to know about it.

For comparison, Spark (since 3.1, SPARK-32290) plans single-column NOT IN with the opposite build side: it broadcasts the subquery as a special HashedRelation that carries "is empty" and "contains NULL" flags computed at build time. Each outer partition then applies purely local logic, so the outer table stays fully distributed and the broadcast constraint lands on the subquery, which is almost always the small side.

Describe the solution you'd like

Three complementary options, roughly in order of leverage:

1. Upstream: null-aware RightAnti support in DataFusion. Add null_aware support for RightAnti, building on the subquery side. RightAnti is probe-driven (already broadcast-safe in Ballista's collect_left_broadcast_safe), and the NULL logic reduces to build-side facts known at build time, exactly like Spark's design. Once upstream supports it, the existing swap optimizations produce it naturally and Ballista's normal broadcast lowering applies with full probe parallelism. This also improves single-node DataFusion: build-side memory drops from O(outer) to O(subquery).

2. Ballista logical rewrite (no upstream dependency). Rewrite negated InSubquery before decorrelation into the classic expansion: compute SELECT count(*), count(k) FROM subquery (one row), cross-join it in, and plan outer LeftAnti subquery ON key = k filtered by count = count_nonnull AND key IS NOT NULL, with the count = 0 case passing all outer rows through. Every operator in that plan is a plain, fully distributable one. Cost is a tiny aggregate stage plus a one-row broadcast. This fixes both the static and AQE paths.

3. AQE runtime branch. Under AQE the subquery is already its own stage. Run it first and report row count and NULL-key count via the runtime-stats transport added in feat(core,scheduler): transport RuntimeStatsExec reports to scheduler; log merged quantile cuts per stage #2175, then re-plan: subquery has NULLs -> replace the subtree with an empty exec (the empty-stage propagation machinery already exists); subquery empty -> drop the join entirely; otherwise -> plan an ordinary distributed anti join with a key IS NOT NULL filter on the outer side. Every branch is parallel and no new operator semantics are needed.

The single-task lowering from #2188 should remain as the fallback for any shape these do not cover.

Describe alternatives you've considered

• Partition-aligned null-aware join with a coordination phase. Hash-partition both sides and broadcast the global has-NULL / non-empty bits between phases. Requires a new two-phase distributed operator; strictly more machinery than the options above for no additional benefit.
• A separate, larger threshold for null-aware builds. Mitigates the sharp edge (the current knob conflates "cheap to broadcast N times" with "must fit one executor's memory") but keeps the single task, so it is at best a stopgap.

Additional context

• Null-aware anti joins can be swapped or repartitioned by scheduler planning #2187 (original correctness bug), fix(scheduler): execute null-aware anti joins in one task #2188 (single-task fix), Default join planning loses NOT IN null semantics in SortMergeJoinExec #2193 (the default sort-merge path drops the null-aware flag entirely, a separate correctness issue).
• Relevant DataFusion code: datafusion-physical-plan/src/joins/hash_join/exec.rs (null_aware validation restricting to LeftAnti + single column) and stream.rs (shared atomics and end-of-probe emission).