andygrove opened a new pull request, #2199:
URL: https://github.com/apache/datafusion-ballista/pull/2199
# Which issue does this PR close?
Part of #2198.
# Rationale for this change
#2188 made null-aware anti joins correct by collapsing them to a single
task, because DataFusion's null-aware hash join coordinates its visited-row
bitmap and probe-side NULL state through in-process atomics. That lowering
loses all parallelism and caps the build side (the outer table) at
`ballista.optimizer.broadcast_join_threshold_bytes`.
This PR implements option 2 from #2198: rewrite uncorrelated `NOT IN
(subquery)` filter predicates during logical optimization so that no null-aware
join is planned in the first place. For
```sql
SELECT a FROM t1 WHERE a NOT IN (SELECT b FROM t2)
```
the rewrite produces
```text
Projection: t1.a
Filter: __cnt = 0 OR (t1.a IS NOT NULL AND __cnt = __cnt_non_null)
CrossJoin
Aggregate: count(1) AS __cnt, count(b) AS __cnt_non_null (over t2)
LeftAnti Join: t1.a = t2.b
TableScan: t1
SubqueryAlias: t2
```
which is equivalent under SQL three-valued `NOT IN` semantics in a `WHERE`
context: an empty subquery passes every row, a matching probe value is dropped
by the anti join, and a NULL probe value or a NULL in the subquery drops the
remaining rows via the one-row aggregate. Every operator in the rewritten plan
is an ordinary distributable one, so the outer table keeps full parallelism and
the broadcast threshold no longer applies.
A side effect is that `NOT IN` now also returns correct results under
Ballista's default sort-merge-join configuration, where the null-aware flag was
previously dropped before scheduler lowering (the practical impact of #2193).
# What changes are included in this PR?
- New `NotInSubqueryRewrite` logical optimizer rule in `ballista-core`,
prepended to DataFusion's default rules so it runs before
`DecorrelatePredicateSubquery`. The rewrite only fires where null-aware
semantics would otherwise be needed: uncorrelated subqueries with a single
output column and a nullable key on either side. Correlated, multi-column,
volatile, and provably non-nullable predicates are left to DataFusion's own
decorrelation, and the #2188 single-task lowering remains as the fallback for
anything the rewrite does not cover.
- Registration in every Ballista session builder:
`SessionState::new_ballista_state`, `upgrade_for_ballista`, and the scheduler's
`default_session_builder`.
- New config key `ballista.optimizer.not_in_subquery_rewrite` (Boolean,
default `true`); setting it to `false` restores the previous null-aware
planning and single-task lowering.
- Unit tests for the rewrite shape and its skip conditions, and new
`default` and `rewrite_disabled` variants in the `null_aware` integration test.
The `default` variant fails without this change (wrong results on the
sort-merge path) and passes with it.
- Config table and tuning guide documentation updates.
# Are there any user-facing changes?
`NOT IN` queries with uncorrelated single-column subqueries now execute as
fully distributed anti joins under static and adaptive planning, including
under the default configuration, and are no longer subject to the broadcast
threshold cap or the single-task lowering. The new
`ballista.optimizer.not_in_subquery_rewrite` key (default `true`) disables the
rewrite if needed. No public API changes.
--
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]
