[cool9850311]

Which issue does this PR close?

• Closes Null-aware NOT IN hash join in Partitioned mode returns rows when the subquery contains NULL #26496.

Rationale for this change

x NOT IN (SELECT y FROM t) must return no rows at all once the subquery
produces a NULL: under SQL three-valued logic x <> NULL is UNKNOWN, so the
NOT IN predicate can never be TRUE for any x.

A null-aware LeftAnti hash join got this wrong in
PartitionMode::Partitioned. With build c1 = 1..8, probe c2 = {1, NULL}
and four hash partitions on each side, the join returned

+----+
| c1 |
+----+
| 3  |
| 6  |
| 8  |
+----+

where PartitionMode::CollectLeft correctly returned nothing. This reproduces
the numbers in the issue.

The cause is where the "a probe partition saw a NULL NOT IN key" fact is
stored. LeftAnti is build-driven: its rows are build rows emitted in the
final stage, after the probe side is exhausted. The flag is therefore recorded
on the build data, in JoinLeftData::probe_completion, written by
JoinLeftData::record_probe_batch and read once through
report_probe_completed.

• CollectLeft collects the build side once, with probe_threads_count set to
  the number of probe partitions, so the flag covers every probe partition. One
  NULL anywhere on the probe side suppresses the whole output. Correct.
• Partitioned builds a separate build side per output partition, with
  probe_threads_count of 1. A NULL probe key is routed by hash to exactly one
  partition, so it only suppresses that partition's build rows; the other
  partitions still emit their unmatched build rows.

NullAwareMode::try_new already rejected Partitioned for LeftMark and for
the correlated LeftAnti, which depend on the per-build-row null bitmap for
the same structural reason. The uncorrelated LeftAnti was the one member of
that build-driven family still allowed through, which is the bug.

Why reject rather than make Partitioned correct

• The planner never produces this plan. physical_planner excludes null-aware
  joins from can_repartition_join; JoinSelection::partitioned_hash_join
  pins a null-aware join to CollectLeft; enforce_distribution only rewrites
  a join that is already Partitioned and never converts CollectLeft into
  it. So there is no workload that would benefit.
• Making it correct would mean hoisting ProbeCompletion out of the
  per-partition JoinLeftData into exec-level shared state, plus a
  cross-partition barrier in the final stage (today only the last partition to
  finish gets Some(summary), while in Partitioned mode every partition
  needs the global answer before emitting). That means touching the atomic
  ordering invariants in probe_completion.rs, which are covered by a loom
  model check.
• Rejecting is what the correlated forms already do, so this keeps one rule for
  the whole family instead of two.

This is a correctness hole rather than a user-visible regression: with the
planner suppressions above, the shape is only reachable from hand-built plans
and from protobuf decode, which takes the partition mode as written. It is
worth closing anyway because the doc comment on HashJoinExec::null_aware
already states the invariant ("such joins are pinned to
PartitionMode::CollectLeft") and relies on it for the positional key-ordering
convention. Before this PR that was a planner convention only; now it is
enforced at construction.

What changes are included in this PR?

Commit 1 (fix):

• NullAwareMode::try_new: widen the arm that rejects Partitioned from
  LeftMark | LeftAnti { correlated: true } to LeftMark | LeftAnti { .. },
  and rewrite its comment to name both kinds of shared state. The error
  message is the existing string, unchanged.
• Tests, see below.

Commit 2 (refactor, separable):

• Drop the now-dead if hash_join.null_aware { CollectLeft } else { Partitioned }
  from the PartitionMode::Partitioned arm of
  statistical_join_selection_subrule. Reaching that line requires the join to
  already be Partitioned, and can_swap_hash_join only lets a null-aware
  join swap when it is an uncorrelated LeftAnti with a single key and no
  filter - exactly the combination commit 1 rejects at construction. So
  null_aware is always false there and the swap keeps the mode the arm
  matched on. (Its comment named RightAnti, which never reached the line
  either: a null-aware RightAnti fails can_swap_hash_join, and RightAnti
  with Partitioned has always been rejected at construction.)
• The sibling check in partitioned_hash_join is deliberately kept. That
  function is called from the Auto and CollectLeft arms, where a null-aware
  join is legal, and it is what keeps the planner from building one in
  Partitioned mode in the first place. What this PR changes for it is the
  consequence of removing it: previously a silently wrong plan, now a loud
  plan_err.

Commit 2 is a separate commit so it can be dropped on its own if a reviewer
prefers to keep the branch.

No public API change; no behaviour change for any plan the planner can build.

What is the testing strategy for this PR?

New tests:

• test_null_aware_validation_left_anti_partitioned
  (physical-plan, joins::hash_join::exec): an uncorrelated null-aware
  LeftAnti (single key, no filter) with Partitioned must fail
  HashJoinExec::try_new. Written first and confirmed red on unfixed code -
  try_new returned Ok and the test dumped the mode: Partitioned, null_aware: true plan. It sits next to the existing
  test_null_aware_validation_partitioned, whose comment notes it uses
  LeftMark "so the single-key check is skipped and the partition-mode check
  is reached"; the LeftAnti side of that rule was simply untested.
• try_from_proto_rejects_a_null_aware_partitioned_join (same file,
  mod proto_tests): the protobuf path is one of the two ways to reach the
  shape, and decode goes through the same builder, so it is rejected there too.

Mutation check: reverting the one-line try_new change and re-running
cargo test -p datafusion-physical-plan --lib --features proto joins::hash_join::exec
turns exactly those two tests red (591 passed, 2 failed) and nothing else, so
they are the guard rather than incidental coverage. Note the second one needs
--features proto; a default cargo test -p datafusion-physical-plan does not
compile mod proto_tests at all.

Existing tests adjusted, both forced by the fix:

• proto_tests::null_aware_round_trips now sets
  partition_mode = CollectLeft. Its decodable_node() helper hard-codes
  Partitioned, and the test then set join_type = LeftAnti and
  null_aware = true, which is precisely the rejected shape; the node it built
  can no longer be decoded. This failure is the fix working.
• test_null_aware_left_anti_swaps_to_right_anti (core,
  physical_optimizer::join_selection) loses its
  #[case(PartitionMode::Partitioned)]. The case is not asserting something
  that changed - its input plan can no longer be constructed, so the rstest
  case would fail in try_new before JoinSelection ever ran. What it checked
  (a null-aware LeftAnti swaps to a RightAnti in CollectLeft with the
  small side on the left and the schema preserved) is still covered by the
  CollectLeft and Auto cases, which reach the same assertion through
  try_collect_left. A doc comment on the rstest records why Partitioned is
  absent, so it is not helpfully added back later. The construction rejection
  it would have exercised is now asserted directly, at the layer that owns it,
  by test_null_aware_validation_left_anti_partitioned.

No .slt case: SQL cannot build this plan, so there is nothing for
sqllogictest to express. The user-visible behaviour is already covered -
null_aware_anti_join.slt "Test 2: NOT IN with NULL in subquery - should
return 0 rows" runs at the harness default target_partitions = 4 and expects
an empty result, and the file also asserts mode=CollectLeft in a plan test.

Full runs (clean worktree, release of this branch):

• cargo test -p datafusion-physical-plan --lib joins:: - 1528 passed
• cargo test -p datafusion --test core_integration - 1430 passed
• cargo test -p datafusion-proto - passed
• cargo test -p datafusion-physical-plan --lib --features proto joins::hash_join::exec::proto_tests - 19 passed
• cargo test -p datafusion-physical-optimizer - 37 passed
• sqllogictest joins and subquery - pass
• extended tests (the command in CLAUDE.md) - 12,756 passed / 0 failed /
  8 ignored, sqllogictest 535/535 files. Run both with and without commit 2,
  with identical counts.
• cargo fmt --all clean, cargo clippy --all-targets --all-features -D warnings
  clean.

Are there any user-facing changes?

No SQL-level change and no public API change. A HashJoinExec built by hand,
or decoded from protobuf, with null_aware = true, JoinType::LeftAnti and
PartitionMode::Partitioned now fails at plan construction with
null_aware joins require PartitionMode::CollectLeft, got PartitionMode::Partitioned instead of returning rows that NOT IN must not
return. Anyone constructing that plan was getting wrong results, so the error
is the fix.

Out of scope on purpose: PartitionMode::Auto with a null-aware join is still
accepted at construction and rejected in execute(), as it is for every join
type (Auto must be resolved by JoinSelection before execution). And the
fallthrough message in NullAwareMode::try_new ("null_aware can only be true
for LeftAnti joins and RightAnti joins with CollectLeft PartitionMode, or
LeftMark joins, got ...") could now be reworded, since the CollectLeft
requirement is no longer specific to RightAnti; I left the wording alone to keep this PR single-purpose and can do it here or
separately if you prefer.

🤖 Generated with Claude Code

[cool9850311]

On the second commit, in case it reads as unrelated cleanup, here is why that
branch cannot be reached.

Entering it requires three things at once:

1. partition_mode == PartitionMode::Partitioned — it is that match arm;
2. can_swap_hash_join(hash_join) — the enclosing if;
3. hash_join.null_aware — the ternary's true side.

fn can_swap_hash_join(hash_join: &HashJoinExec) -> bool {
    hash_join.join_type().supports_swap()
        && (!hash_join.null_aware
            || (*hash_join.join_type() == JoinType::LeftAnti
                && hash_join.on().len() == 1
                && hash_join.filter().is_none()))
}

Given (3), !hash_join.null_aware is false, so (2) can only hold through its
right-hand side: LeftAnti, a single key, no filter. NullAwareMode::try_new
derives correlated = num_keys > 1 || has_filter, so that is exactly
LeftAnti { correlated: false }. With (1) the join is therefore an uncorrelated
null-aware LeftAnti in Partitioned mode — the combination the first commit
rejects in try_new, which HashJoinExec::try_new, the builder and protobuf
decode all go through.

So on that line null_aware is always false and the chosen mode is always
Partitioned: for every input that can reach it, the old ternary and
swap_inputs(PartitionMode::Partitioned) compute the same value. Reverting the
first commit and re-running the crate's tests confirms the construction is what
blocks it, not the optimizer: the only two failures are the new
test_null_aware_validation_left_anti_partitioned and
try_from_proto_rejects_a_null_aware_partitioned_join.

The RightAnti named in the old comment never reached the line either: a
null-aware RightAnti fails can_swap_hash_join, and
(RightAnti, Partitioned) falls through try_new's
match (join_type, partition_mode) to plan_err.

This is not needed for the fix, which is why it is a separate commit — happy to
drop it if you would rather leave the branch in place.

[jayzhan211]

Thanks @cool9850311 👍🏻