[adriangb]

Which issue does this PR close?

• Closes Null-aware join: an equality correlation makes a correlated NOT IN slower, not faster #25438.

Rationale for this change

An equality correlation made a correlated NOT IN slower, not faster. The cost grew with the square of the scope size. This is the reproduction from the issue, on the null_aware_join benchmark tables at 100,000 rows each:

-- (a) non-equality correlation only
SELECT count(*) FROM small_outer o
WHERE o.id_n50 NOT IN (SELECT i.id_n0 FROM small_inner i WHERE i.z < o.z);

-- (b) the same, plus an equality correlation on k
SELECT count(*) FROM small_outer o
WHERE o.z > 900
   OR o.id_n50 NOT IN (SELECT i.id_n0 FROM small_inner i WHERE i.k = o.k AND i.z < o.z);

Release datafusion-cli, median of 10 runs:

Query	main	This PR
(a)	7 ms	7 ms
(b)	538 ms	7 ms

Query (b) is now as fast as query (a). Both queries return the same results as before.

What changes are included in this PR?

A correlated null-aware join marks a row UNKNOWN when a candidate pair passes the join filter. LeftAnti and LeftMark mark build rows, and RightAnti marks probe rows. A marked row stays marked. With a scope key, the candidate pairs come from a hash lookup on the scope key. Before this PR, each probe row was paired with every row of its scope, marked or not.

The change is in UnmarkedPairs::scope_matches in hash_join/not_in.rs:

1. The lookup drops the matches whose build row is marked, before it makes pairs.
2. That alone still walks each scope once per probe row, and a RightAnti probe row is still paired with its whole scope. So the join counts the matches that the lookup per probe row wastes. When they cost more than a sort, it estimates from a sample of the remaining probe rows whether groups of equal hashes save work:
   • with marks on build rows, when a group holds more than one probe row, which then share one walk of the hash chain;
   • with marks on probe rows, when the pairs of a group fill more than a chunk, so that the later chunks can skip the marked probe rows.
3. If the groups save work, the join sorts the remaining probe rows by hash, looks up each hash once, and pairs each group like a cross product that drops the rows as they are marked. A group of probe rows stops walking its hash chain when all of them are marked. Otherwise the lookup per probe row goes on, so scopes of a few rows never pay for a sort.
4. With marks on build rows, a check of the marks that drops nothing makes the next checks less frequent. So a filter that never passes adds almost no cost.

The PR has three commits, for review one at a time:

1. refactor: no behavior change. The cross product of the path without scope keys and the scope-key lookup share one struct, UnmarkedPairs. The scope-key lookup yields full row indices, and the join checks the scope-key equality per chunk. So NullValueBuildRows drops its copy of the scope values.
2. test: a randomized differential test, two tests that pin how many hash-chain matches the lookups walk today, and null_aware_scope_groups.slt.
3. perf: the change. The test diff shows that the walk no longer grows with the product of the rows.

Cost on other shapes. Release datafusion-cli, median of 3 interleaved runs, in ms, main → this PR. The outer side is all NULL, except the unique-key shape (50% NULL). "None" has no residual filter, "never" a filter that no pair passes, "always" one that every pair passes. One partition unless noted. The 64-row shape comes from a separate run, because its first run was noisy.

Shape	RightAnti none	RightAnti never	RightAnti always	LeftMark none	LeftMark never	LeftMark always
One scope, 8,192 × 8,192	370 → 1	299 → 232	419 → 1	316 → 1	352 → 293	308 → 1
Unique scope keys, 2M × 2M	55 → 55	54 → 53	57 → 54	65 → 66	65 → 68	66 → 67
12 inner rows per scope, 300k × 1M	79 → 66	59 → 53	88 → 75	83 → 61	89 → 88	80 → 61
16 inner rows per scope, 400k × 1M	92 → 87	77 → 71	107 → 101	103 → 71	112 → 105	99 → 71
16 inner rows per scope, random scopes, 400k × 1M	200 → 189	186 → 180	228 → 212	333 → 237	388 → 381	333 → 225
64 inner rows per scope, 1.6M × 2M	892 → 871	814 → 758	1045 → 1012	898 → 541	1003 → 938	839 → 526
16 inner rows per scope, 400k × 1M, all partitions	20 → 19	18 → 16	22 → 21	43 → 39	42 → 36	45 → 39
64 inner rows per scope, 1.6M × 2M, all partitions	191 → 179	179 → 166	220 → 210	254 → 223	247 → 217	258 → 218

No shape is more than 5% slower, and the remaining gaps are inside the run-to-run noise. Shapes where rows are marked are up to several hundred times faster.

Not in this PR. Each probe batch walks a fully marked scope again at least once. The path without scope keys has the same cost. Keeping each scope's unmarked rows across batches is a possible follow-up.

What is the testing strategy for this PR?

• No result changes. The null_aware_*.slt files, subquery.slt, the full sqllogictest suite and the hash_join unit tests pass.
• unmarked_pairs_match_brute_force runs UnmarkedPairs on 3,000 random inputs and compares the marks with those of every pair. The inputs have hash collisions between scopes, rows marked before the pairing or during it by other partitions, filters that pass none to all of the pairs, marks on build or probe rows, and batch sizes from 1 to 8192. More than 100 inputs of each mark kind reach the grouped pairing.
• scope_lookups_stay_linear_in_rows and marked_probe_groups_stop_walking bound the hash-chain matches that the lookups walk. Emitted pairs alone do not show a repeated walk.
• The cost decisions (estimate_groups, groups_save_work, next_unchecked_lookups) are small pure functions with exact unit tests. small_scopes_are_not_grouped and lookups_skip_checks_while_nothing_is_marked check them through the pairing.
• null_aware_scope_groups.slt compares LeftMark and RightAnti NOT IN queries with a reference built from an outer join and an aggregate. The scopes are large enough for the grouped pairing, and a residual filter leaves some rows unmarked. It runs with configMatrix for batch sizes 1, 7 and 8192 and 1 and 4 partitions. With an instrumented build, I confirmed that both mark kinds reach the grouped pairing.
• Mutation testing with cargo-mutants on the changed lines of not_in.rs catches every mutant except one, which moves the threshold for considering groups by one wasted match.

Are there any user-facing changes?

No. Correlated NOT IN queries with an equality correlation are faster.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 90.30303% with 32 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.49%. Comparing base (95bb0a0) to head (0bbdfa9).

Files with missing lines	Patch %	Lines
...fusion/physical-plan/src/joins/hash_join/stream.rs	91.60%	10 Missing and 11 partials ⚠️
...on/optimizer/src/decorrelate_predicate_subquery.rs	86.07%	3 Missing and 8 partials ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #25679    +/-   ##
========================================
  Coverage   82.48%   82.49%            
========================================
  Files        1140     1140            
  Lines      437614   437727   +113     
  Branches   437614   437727   +113     
========================================
+ Hits       360986   361092   +106     
- Misses      54834    54849    +15     
+ Partials    21794    21786     -8     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[viirya]

Thanks for this — the grouping idea fixes the pathology nicely. I rebased the last commit onto current main to check it against the repro sunchao raised on #26214 (#26214 (comment)). Conflicts are small: the multi-column mark closure from #19857, MapOffset → ProbeOffset (#26203), and equal_rows_arr now needs the comparator slot (the build scope values are normalized, so equal_rows_arr_with_normalized_left fits). With 8192 × 8192 rows and a constant scope key, a NOT IN (SELECT x FROM inner_t i WHERE i.y = o.b) goes from ~206 ms to ~1 ms, including the variant where one outer row sits in a scope that is never marked. The issue's Q08 at 100k goes from ~367 ms to ~10 ms, and the unique-scope 2M × 2M shapes cost +2–6% wall, in line with your numbers.

Now that #25560 has landed, do you plan to take this out of draft? #26214 makes NOT NULL tuple elements of a multi-column NOT IN into scope keys, which routes more queries into this path, so I'd like to stack it on top of this. If you don't plan to continue with it, we'd be happy to open a PR that builds on your commit.

[viirya]

Following up on my earlier comment: #26214 is on hold until the scope-key marking stops re-pairing already-marked rows, which is what this PR does. Would you have a chance to pick this up again soon? If you're busy with other things, I'd be glad to help move it forward: I could open a PR that builds on your commit and rebases it onto current main (the null-aware join code has since moved to hash_join/not_in.rs), with you credited as co-author. Maybe sometime next week, if that works for you? Happy to hand it back anytime if you'd rather finish it yourself.

[adriangb]

Sorry for the delay @viirya. Marked as ready now!

[Copilot]

🟡 Changes recommended

Regression coverage does not validate grouped pairing with remaining unmarked rows or detect loss of the hash-chain traversal optimization.

1 open issue

• Add selective grouped-pairing regression and bound hash-chain traversal · New

What changed in this PR

This PR addresses #25438 by reducing repeated work in DataFusion’s scope-key null-aware joins, improving correlated NOT IN performance without intended result changes.

Changes:

• Skip marked rows during scope lookup and adaptively group remaining probe rows by hash.
• Share chunked pairing logic through UnmarkedPairs and remove retained copies of NULL-build scope values.
• Add a regression test for emitted-pair growth.

File	Description
datafusion/​physical-plan/​src/​joins/​hash_join/​not_in.rs	Optimizes scope matching, consolidates pairing logic, reduces retained state, and adds regression coverage.

_{Review effort: Balanced}

[viirya]

Thanks for picking this back up! I re-ran the repros on the current head (440b285). The build-side path is fixed: with 8192 × 8192 rows and a constant scope key (outer_t: a = NULL, b = 1; inner_t: x = 2, y = 1), the LeftMark

SELECT count(r) FROM (SELECT a NOT IN (SELECT x FROM inner_t i WHERE i.y = o.b) AS r FROM outer_t o);

goes from ~210 ms to ~1 ms.

RightAnti still pairs every row of a scope bucket, and is now slower. On the same data, the anti-join form

SELECT count(*) FROM outer_t o WHERE a NOT IN (SELECT x FROM inner_t i WHERE i.y = o.b);

plans as a null-aware RightAnti (probe = outer_t) and goes from ~300 ms on main to ~410 ms with this PR. The comment in scope_matches says a probe row is looked up once, so grouping never pays off for marks on probe rows. But that single lookup returns the probe row's whole bucket (up to batch_size matches), and they are all pushed before the first chunk is sent. lookup_unmarked only drops matches of a probe row that a chunk already sent has marked, so it never skips a probe row's own matches. Each NULL-valued probe row is still paired with every build row in its scope, O(probe × bucket) as before, with the bit check and push added per match. With a medium bucket (64 scopes, 128 rows each) the difference is small (3 → 4 ms), so it is mostly the large-bucket shape.

Would it work to group the probe rows by scope hash for UnknownMarks::Probe too, and pair each group with its chain's build rows the way cross_product already does for Probe marks (the group's probe rows as decided_rows in pair_unmarked, the build rows as the outer loop), stopping the chain walk once the whole group is marked? A constant-key bucket would then cost about one chunk per group instead of one bucket per probe row. The grouped phase's lookup_unmarked maps positions to group indices, so it can't be reused as is for Probe marks.

This matters for #26214: after #26213, its multi-column repro (a, b) NOT IN (SELECT x, y FROM inner_t) with b/y NOT NULL also plans as RightAnti and goes through the same path. I'm happy to do the probe-side part as a follow-up if you'd rather keep this PR to the build side. Which would you prefer?

Fully marked buckets are walked again in every probe batch. Nothing carries over between probe batches, so each batch walks the whole marked chain at least once per group. The LeftMark query above at N × N rows with a constant key takes ~260 ms at 1M, ~960 ms at 2M and ~3.7 s at 4M. The no-scope cross product grows the same way ((a, b) NOT IN without a scope key: 84 ms at 1M, 274 ms at 2M), so this isn't specific to this PR and is fine as a follow-up. I'm only flagging it in case you had considered keeping each scope's unmarked build rows across batches.

Coverage of the grouped path. +1 to Copilot's point. Grouping starts only after more than 8 skipped matches per probe row, and only unmarked_pairs_do_not_grow_with_probe_rows reaches it, with a closure that marks every pair. The real mark closure (multi-column value tuples, join filter, LeftMark) is never run together with the grouping: a group split across two lookup chunks, the MidChain row paired again, the per-group refresh with rows left unmarked. The null-aware fuzz tables are too small to get there. Could you add an slt case that does, for example a LeftMark with a constant scope key, more than 8 NULL-valued build rows in that scope, a residual filter that leaves some of them unmarked, and batch_size 1 and 7? It would pass before and after, but it would protect the grouped path from later changes. For reference, comparing NOT IN with NOT EXISTS (... (o.a = i.x) IS NOT FALSE) over low-cardinality scope keys (NULL and -0.0 scope values, dictionary scope keys, filters, 1–3 value keys, batch sizes 1/7/64/8192, 1 and 4 partitions) showed no differences.

[viirya]

Thanks for the update, this covers everything I raised. On ac9173e, with 8192 × 8192 rows and a constant scope key, the RightAnti form a NOT IN (SELECT x FROM inner_t i WHERE i.y = o.b) goes from ~295 ms on main to ~1 ms. With #26214 applied on top, its multi-column repros ((a, b) NOT IN (SELECT x, y FROM inner_t) with b/y NOT NULL, also planned as RightAnti) go from ~296 ms to ~1 ms as well. The medium-bucket shape that got slower last time is back to main (64 scopes: 3 ms → 3 ms; unique scope: 31 ms → 31 ms), and the RightAnti walk now grows linearly with the rows (13 / 25 / 51 ms at 1M / 2M / 4M). Agreed that the per-batch re-walk of fully marked scopes is fine as a follow-up, since the path without scope keys has the same cost.

The new slt and the brute-force test against every pair cover the grouped path well, including rows left unmarked by the filter and marks from other partitions.

Two small inline comments: the license header check failing on the new slt, and two doc comments that still say marked pairs never reach the join filter.

[viirya]

nit: since mark no longer calls retain_unmarked, this (and the UnmarkedPairs doc at L991–993, "drops the pairs that decide a marked row before it sends them") is now a bit stronger than what happens. A pair of a marked build row can reach the join filter when lookup_unmarked skips its check, and the per-probe-row lookup with marks on probe rows doesn't check at all. Both are fine since a pair only sets a mark, as the comment in lookup_unmarked says. Maybe something like "...so the pairing drops most pairs that decide a row already known to be UNKNOWN before the join filter is evaluated (see UnmarkedPairs::lookup_unmarked for the checks it skips)"?

[adriangb]

b8f1a0c