Skip to content

fix: null-aware anti join could emit rows after a sibling probe partition saw NULL - #25076

Merged
kosiew merged 8 commits into
apache:mainfrom
jayzhan211:fix/hash-join-null-aware-probe-race
Sep 10, 2026
Merged

kosiew merged 8 commits into
apache:mainfrom
jayzhan211:fix/hash-join-null-aware-probe-race

Conversation

@jayzhan211

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

  • Closes #.

Rationale for this change

x NOT IN (subquery) is planned as a null-aware LeftAnti hash join in
CollectLeft mode with several probe partitions. If any probe partition sees
a NULL key, the predicate is UNKNOWN for every build row and the join must
return nothing.

The final stage of HashJoinStream read the shared "probe side saw NULL"
flag before decrementing the probe-partition counter. That allowed this
interleaving:

  1. partition B finishes its probe input and reads the flag as false;
  2. partition A processes a batch with a NULL key, sets the flag, finishes and
    decrements the counter;
  3. partition B decrements the counter, becomes the last partition, and emits
    the unmatched build rows.

The result then contains rows that NOT IN must suppress. The counter was
also decremented with Ordering::Relaxed, so nothing ordered the flag stores
of a finishing partition before the reads of the last one, even when the
reads happened after the decrement.

What changes are included in this PR?

  • JoinLeftData::report_probe_completed now decrements with AcqRel and
    returns Option<ProbeSideSummary>: Some only for the last partition, with
    the shared flags read after the decrement. The two AtomicBool flags are
    private; the probe phase stores through record_probe_batch and the early
    exit reads a probe_side_has_null_hint that is documented as best-effort.
    The final stage therefore cannot read the flags before its own decrement,
    which guards the fix structurally rather than by a test that would have to
    force a few-instruction race window.
  • process_unmatched_build_batch decrements first and hands the summary to
    the null-aware helpers. The NULL-probe rule of LeftAnti moved into
    null_aware_left_anti_final_indices, so all its final-stage rules are in
    one place.
  • No change for joins that are not null-aware beyond the AcqRel ordering
    on the counter.

What is the testing strategy for this PR?

Two new tests in hash_join/exec.rs,
test_null_aware_anti_join_probe_null_in_other_partition and
test_null_aware_left_mark_probe_null_in_other_partition, run a CollectLeft
join with a two-partition probe side where only partition 0 has a NULL key,
draining the partitions in both orders. They cover the cross-partition flag
propagation that had no test. The race itself cannot be reproduced
deterministically without a test hook, which is why the fix makes the wrong
call order unrepresentable instead. Existing null-aware join tests and the
join unit test suite pass unchanged.

Are there any user-facing changes?

Wrong results from NOT IN under this interleaving are fixed. No API changes;
all touched items are pub(super).

@jayzhan211
jayzhan211 requested review from AdamGS and kosiew September 8, 2026 13:17
@github-actions github-actions Bot added the physical-plan Changes to the physical-plan crate label Sep 8, 2026
.unwrap()
}

/// Drains the probe partitions of `join` one after another in the given

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These partitions are drained sequentially, so the old implementation also passes both orders: the NULL flag is always written before the final partition examines it. The tests would also remain green if the counter ordering changed back to Relaxed.

Could we force the failing overlap with a synchronization hook where the non-NULL partition reads false, the sibling records NULL, and the non-NULL partition then becomes last or add a model-checked atomic test? This wrong-results race otherwise has no regression coverage.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I add the loom test

@codecov-commenter

codecov-commenter commented Sep 8, 2026 •

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 96.52778% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.90%. Comparing base (46ec37d) to head (9834679).
⚠️ Report is 2 commits behind head on main.

Files with missing lines Patch % Lines
...tafusion/physical-plan/src/joins/hash_join/exec.rs 95.12% 2 Missing and 2 partials ⚠️
...sical-plan/src/joins/hash_join/probe_completion.rs 97.87% 0 Missing and 1 partial ⚠️
Additional details and impacted files
@@            Coverage Diff             @@
##             main   #25076      +/-   ##
==========================================
+ Coverage   81.89%   81.90%   +0.01%     
==========================================
  Files        1131     1132       +1     
  Lines      419945   420155     +210     
  Branches   419945   420155     +210     
==========================================
+ Hits       343901   344121     +220     
+ Misses      55763    55744      -19     
- Partials    20281    20290       +9     

☔ 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.

@kosiew kosiew left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@jayzhan211, thanks for working on this. The synchronization change looks good to me. Encapsulating the probe-side state behind the final summary makes the ordering requirement much clearer, and the new two-partition tests cover the race in both completion orders. I have one non-blocking suggestion for some additional dictionary-key coverage.


/// Builds a two-partition probe side for the cross-partition null-aware
/// tests: partition 0 holds the only NULL key, partition 1 holds none.
fn build_two_partition_probe_with_null_in_partition_0() -> Arc<dyn ExecutionPlan> {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we also add a two-partition LeftAnti case where the NULL-bearing probe partition uses a dictionary key containing only a logical NULL? The existing dictionary test is single-partition, while this new helper covers physical Int32 NULLs. I think that would be useful for exercising the logical_null_count() path through the new cross-partition summary. This is non-blocking.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

added them

@kumarUjjawal kumarUjjawal left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @jayzhan211

Looks good!

@kosiew
kosiew enabled auto-merge September 10, 2026 09:41
@kosiew

kosiew commented Sep 10, 2026

Copy link
Copy Markdown
Contributor

🚀
@jayzhan211
Thank you for your contribution.

…sion

Merging main's chunked emission of final build rows split the final stage
into `prepare_unmatched_build_rows` and `emit_unmatched_build_rows`, leaving
the null-aware post-processing in the second one while `probe_summary` was
still bound in the first. `report_probe_completed` hands the summary out
exactly once, so capture it alongside the bitmap snapshot in
`EmitUnmatchedBuildRowsState` and read it back per chunk.
@jayzhan211

Copy link
Copy Markdown
Contributor Author

One of my CI failure is nested_key_spill_keeps_groups_unique which seems added in #24889
Another is sql::unparser::test_tpch_unparser_roundtrip stack overflow — most likely collateral from the same memory pressure. 🤔

`loom` was a plain dev-dependency, so it joined every `cargo test
--workspace` resolve. It is not inert there: loom pulls in
tracing-subscriber, whose env-filter turns on `regex-automata`'s
`dfa-build`/`dfa-search` features for the whole workspace build. Those
change which engine the shared `regex` selects, and the new path costs
enough extra stack to abort unrelated deep-recursion tests --
`sql::unparser::test_tpch_unparser_roundtrip` overflowed its stack in CI
on every run since the dependency was added.

Measured on `core_integration`, varying only whether loom is in the
resolve (`RUST_MIN_STACK` at which the test passes):

    without loom   480K ok   1M ok     2M ok     4M ok
    with loom      480K ok   1M FAIL   2M FAIL   4M ok

Declaring loom under `cfg(datafusion_loom)` leaves the ordinary build
byte-identical to one that never mentioned it. The cfg is namespaced
because a bare `loom` is also read by tokio, which would switch tokio
into its own loom build and drop `tokio::fs`.

Run the model checks with:

    RUSTFLAGS="--cfg datafusion_loom" \
        cargo test -p datafusion-physical-plan --lib loom_tests
@kosiew
kosiew added this pull request to the merge queue Sep 10, 2026
Merged via the queue into apache:main with commit 5747e87 Sep 10, 2026
42 checks passed
jayzhan211 added a commit to cning112/datafusion that referenced this pull request Oct 24, 2026
## Which issue does this PR close?

- None.

## Rationale for this change

`HashJoinExec` supports SQL `NOT IN` through `null_aware = true`, in
three forms: `LeftAnti`, `LeftMark` and `RightAnti`, each correlated or
not. The rules for which rows such a join emits are spread out today:

- `exec.rs`: `NullAwareMode`, four `JoinLeftData` fields and one
`JoinBuildData` field that only some forms use, and the correlated
per-row state built inside the build function.
- `stream.rs`: about 700 lines inside the regular join stream. These are
the batch-skip decision, the correlated marking pass and its helpers,
the RightAnti NULL and UNKNOWN drops, and two final-stage functions.
- `utils.rs`: a second copy of the final rule, used for the LeftMark
column.

The two copies of the final rule already drifted once: LeftMark checked
logical NULLs since apache#25693, LeftAnti physical ones until apache#26578.

This code changes often. Since July it gained four features (apache#23957,
apache#21585, apache#19857, apache#26213) and seven correctness fixes (apache#23104, apache#23173,
apache#25076, apache#25339, apache#25693, apache#26586, apache#26578), and apache#26214, apache#25679 and apache#26088
propose more. Each such change has to find the `NOT IN` rules in three
files, usually about six functions, interleaved with the regular join. A
change to the regular probe loop has to read past them too.

This PR moves the rules into `hash_join/not_in.rs`, with one copy of the
final rule, so a `NOT IN` change touches one module and a regular-join
change doesn't need to read it. The join reaches it at four points:

1. `NotIn::try_new` decides when the plan is built or executed whether
`null_aware` is legal for the join, and which form it is: `BuildRows {
scope, output: Rows | Mark }` for LeftAnti and LeftMark, or `ProbeRows {
scope }` for RightAnti, where `scope` is `Uncorrelated` or `Correlated`.
2. `NotInBuild::try_new` runs once per build. It has one variant per
form, holding only that form's state. `JoinLeftData` keeps a single
`Option<NotInBuild>`.
3. `NotInProbe::before_lookup` / `after_lookup` run around each probe
batch's lookup, and `after_empty_build` replaces `after_lookup` when the
build map is empty. They skip the batch, record what the probe saw, mark
build rows or this batch's probe rows UNKNOWN, and drop the probe rows a
RightAnti must not emit; for other joins they do nothing.
4. `NotInBuild::final_chunk` runs on the build rows emitted at the end.
It applies one rule for an unmatched row's mark, used both by LeftAnti
(keep the rows whose mark is FALSE) and by LeftMark (the mark column).

## What changes are included in this PR?

Six commits, each compiling and passing clippy on its own:

1. **Move the correlated marking helpers into `not_in.rs`.** The code is
unchanged, apart from one signature that now takes the build data and
probe batch instead of the stream's private state structs. This commit
also adds a test for correlated marking when every build value key is
NULL. The full-key map is empty then, and the marking must still run
before the empty-map shortcut; nothing covered this before.
2. **`NotIn` replaces `NullAwareMode`.** The legality checks, messages
and order are the same.
3. **`NotInBuild` holds the per-form build state, with one final rule.**
It replaces the `JoinLeftData` / `JoinBuildData` fields, the two
final-stage functions in `stream.rs`, and
`build_null_aware_left_mark_column` in `utils.rs`.
4. **The `NotInProbe` hooks replace the stream's `NOT IN` branches and
fields.** The correlated RightAnti's per-batch UNKNOWN probe rows move
from `ProcessProbeBatchState` into `NotInProbe`, which recomputes them
on each batch's first lookup.
5. **Review follow-ups.** `HashJoinExec::not_in()` and the `not_in`
parameters replace `null_aware_mode()` and `null_aware: Option<NotIn>`,
the helpers used only inside `not_in.rs` become private, and
`final_chunk` asserts that a probe side with a NULL key is not empty.
6. **Test the correlated RightAnti empty-build path.**
`test_null_aware_correlated_all_null_build_values` also runs RightAnti,
which must drop the UNKNOWN probe rows when the build map is empty
(`NotInProbe::after_empty_build`).

Types: `NullAwareMode` goes, and `NotIn`, `NotInScope`, `NotInOutput`,
`NotInBuild` and `NotInProbe` come in. Gone: the `NOT IN` fields of
`JoinLeftData`, `JoinBuildData`, `HashJoinStream` and
`ProcessProbeBatchState` and their constructor parameters, 4 stream
functions (`null_aware_skip_probe_batch`, `drop_null_probe_keys`,
`null_aware_left_anti_final_indices`, `null_aware_left_mark_column`),
and the second copy of the final rule.

What does not change:
- the planner rules, including the `JoinSelection` swap to RightAnti;
- the dynamic-filter gate for null-aware joins;
- the prefilter and prepared-build exclusions;
- the cross-partition `ProbeCompletion`.

### Behavior notes

No query results change. Two internal orders do:

- **Reservation order.** For an uncorrelated LeftAnti or LeftMark join
whose build key has newly materialized logical NULLs (a dictionary, REE
or Null key), the NULL buffer is now charged to the memory reservation
right after the visited bitmap. On main it was charged after the
membership step's optional pruning bitmap. Under a tight memory limit, a
different allocation may be the one that fails. The pruning bitmap only
exists with a dynamic filter, which a null-aware join with a nullable
build key doesn't get.
- **The form is computed twice per execution.** The stream reads it from
the build (`JoinLeftData::not_in`); `execute` derives it again
(`HashJoinExec::not_in()`) for the dynamic filter. Main derived it three
times (build, stream, accumulator). The two can only disagree if an
executed plan is rebuilt and run again without `reset_state`, which is
already unsupported. Main panicked in that case; this PR uses the
build's form.

## Performance

Benchmark bot (GKE `c4a-highmem-16`, aarch64) on `a9d4896d7e` against
its merge base `702803d022`; the later commit only adds a test:

- `null_aware_join`: geomean 0.989. Q01 and Q03 are 4% faster, Q02, Q05
and Q06 about 2% faster, Q08 2% slower (20.4 → 20.8 ms), the rest are
within 1%.
- `hj`: total 6870 → 6885 ms (+0.2%). Q22 and Q24 are unchanged. The
queries the bot flags (Q11, Q14, Q15, Q17 slower; Q16, Q18 faster) move
within their run-to-run spread.
- `tpch10`: all 22 queries unchanged, total 7078 → 7049 ms.

Earlier runs on an M-series Mac showed this refactor slowing `hj` Q24 by
up to 9% through different LLVM inlining of the probe loop. That doesn't
reproduce on the bot, so the PR leaves inlining to the compiler. Not
measured on x86_64.

## What is the testing strategy for this PR?

- New test: `test_null_aware_correlated_all_null_build_values`, covering
LeftAnti, LeftMark and RightAnti with the probe in two batches, at every
batch-size config.
- Existing tests:
  - physical-plan unit tests;
- the null-aware join fuzz test (`null_aware_join_fuzz`, with
`extended_tests`), which covers uncorrelated and correlated RightAnti
plans after the `JoinSelection` swap;
- the `null_aware_*` and `subquery*` sqllogictest files, with no changes
to expected results or `EXPLAIN` output.

## Are there any user-facing changes?

No.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

physical-plan Changes to the physical-plan crate v56.0.0

Projects

None yet

Development

Successfully merging this pull request may close these issues.

5 participants