andygrove opened a new issue, #2198: URL: https://github.com/apache/datafusion-ballista/issues/2198
**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](https://issues.apache.org/jira/browse/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 #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** - #2187 (original correctness bug), #2188 (single-task fix), #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). -- 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]
