andygrove opened a new issue, #23931: URL: https://github.com/apache/datafusion/issues/23931
### 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 (apache/datafusion-ballista#2187) and currently has to force the join into a single task (apache/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 (apache/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 - #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: apache/datafusion-ballista#2198. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
