Support null-aware RightAnti hash join (build on the subquery side) for NOT IN

[andygrove]

Is your feature request related to a problem or challenge?

DataFusion plans NOT IN (subquery) as a null-aware anti join, but HashJoinExec only supports null_aware = true for LeftAnti with a single join key (validated in datafusion-physical-plan/src/joins/hash_join/exec.rs). Since HashJoinExec always builds on the left input, the build side of a null-aware anti join is the outer table, not the subquery.

This has two costs:

1. Memory scales with the wrong side. For SELECT ... FROM big_fact WHERE key NOT IN (SELECT k FROM small_dim), the hash table is built over the entire fact table. Memory is O(outer) when it could be O(subquery).

2. The operator cannot be distributed. The null-aware logic coordinates three pieces of global state across probe partitions through in-process shared memory: probe_side_has_null: AtomicBool, probe_side_non_empty: AtomicBool, and the visited-build-row bitmap, with the last probe partition to finish emitting the unmatched build rows (hash_join/stream.rs). This is correct and cheap in one process, but engines that split probe partitions across processes get independent copies of all three and produce duplicated or incorrect rows. Ballista hit exactly this (Null-aware anti joins can be swapped or repartitioned by scheduler planning datafusion-ballista#2187) and currently has to force the join into a single task (fix(scheduler): execute null-aware anti joins in one task datafusion-ballista#2188), losing all parallelism.

For comparison, Spark's null-aware anti join (SPARK-32290) builds on the subquery side: the broadcast HashedRelation carries "is empty" and "contains NULL" flags computed at build time, and each streamed partition of the outer table applies purely local logic. The outer table stays fully distributed and the memory requirement lands on the (typically small) subquery.

Describe the solution you'd like

Support null_aware = true for RightAnti joins, where the build (left) input is the subquery and the probe (right) input is the outer table. The semantics become build-side-local:

• While building the hash table, record build_is_empty and build_has_null for the single join key.
• For each probe row: if the build is empty, emit the row. Otherwise, if the probe key is NULL, or build_has_null, or the key matches, drop it.

No shared probe-side state, no visited bitmap, and no end-of-probe emission phase are needed. The natural partition mode is CollectLeft, which is safe here because RightAnti is probe-driven (Partitioned mode would need a globally computed build_has_null, since a NULL key hashes to a single partition, so it could be left unsupported initially, mirroring Spark which only supports the broadcast form).

With this in place, the existing swap optimizations in JoinSelection can rewrite a null-aware LeftAnti into a null-aware RightAnti when the statistics favor building on the subquery, which is nearly always for NOT IN. Single-process DataFusion gets the memory win, and distributed engines built on DataFusion get an operator whose build side can be broadcast while the outer table keeps full parallelism.

Describe alternatives you've considered

• Plan-level rewrite. Ballista now rewrites uncorrelated NOT IN filters into a plain anti join plus a one-row count(*)/count(k) aggregate cross join (feat(core): rewrite NOT IN subqueries into distributable anti join plans datafusion-ballista#2199). This works and distributes, but it scans the subquery twice and lives outside the operator, so every downstream engine has to reimplement it. An operator-level fix benefits everyone, including single-process DataFusion.
• Keep the current design. Correct, but the O(outer) build memory and the single-process coordination remain.

Additional context

• Null-Aware Anti Join Support #4211 is the original null-aware anti join request that led to the current LeftAnti implementation.
• Current restriction: null_aware can only be true for LeftAnti joins and single-column keys, hash_join/exec.rs.
• Shared probe-side state: probe_side_has_null / probe_side_non_empty atomics and the end-of-probe emission in hash_join/stream.rs.
• Spark reference: SPARK-32290 (single-column null-aware anti join, broadcast-only).
• Downstream discussion of the distribution problem and workarounds: Avoid single-task execution for null-aware anti joins (NOT IN subqueries) datafusion-ballista#2198.

[saadtajwar]

Hey @andygrove , happy to take a stab at this one! :)

[saadtajwar]

take

[saadtajwar]

@andygrove - first of two PRs up at #23957 ! 🥳 🎉