You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/23 05:35:42 UTC
(datafusion) branch gh-readonly-queue/main/pr-26586-81549fbf36aea66ffa55b125c417aec339b47d54 created (now 0ab1fd2fb9)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-26586-81549fbf36aea66ffa55b125c417aec339b47d54
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 0ab1fd2fb9 fix: reject null-aware `LeftAnti` hash join in `Partitioned` mode (#26586)
This branch includes the following new commits:
new 0ab1fd2fb9 fix: reject null-aware `LeftAnti` hash join in `Partitioned` mode (#26586)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: reject null-aware `LeftAnti` hash join in `Partitioned` mode (#26586)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26586-81549fbf36aea66ffa55b125c417aec339b47d54
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 0ab1fd2fb9b99069473fb0c964d10952e59481f6
Author: Cheng-Yi Shih <48...@users.noreply.github.com>
AuthorDate: Fri Oct 23 03:33:10 2026 +0000
fix: reject null-aware `LeftAnti` hash join in `Partitioned` mode (#26586)
## Which issue does this PR close?
- Closes #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](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5 <no...@anthropic.com>
Co-authored-by: Jay Zhan <ja...@gmail.com>
---
.../tests/physical_optimizer/join_selection.rs | 5 +-
.../physical-optimizer/src/join_selection.rs | 13 ++--
.../physical-plan/src/joins/hash_join/exec.rs | 85 ++++++++++++++++++++--
3 files changed, 89 insertions(+), 14 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs
index 1840568f1b..9f23432990 100644
--- a/datafusion/core/tests/physical_optimizer/join_selection.rs
+++ b/datafusion/core/tests/physical_optimizer/join_selection.rs
@@ -469,10 +469,13 @@ async fn test_join_with_swap_semi() {
}
}
+/// `PartitionMode::Partitioned` is deliberately not a case here: a null-aware
+/// `LeftAnti` cannot be constructed in that mode at all, so there is no input
+/// plan to optimize. See
+/// `HashJoinExec::null_aware_mode`/`NullAwareMode::try_new`.
#[rstest]
#[case(PartitionMode::CollectLeft)]
#[case(PartitionMode::Auto)]
-#[case(PartitionMode::Partitioned)]
#[tokio::test]
async fn test_null_aware_left_anti_swaps_to_right_anti(
#[case] partition_mode: PartitionMode,
diff --git a/datafusion/physical-optimizer/src/join_selection.rs b/datafusion/physical-optimizer/src/join_selection.rs
index 2df139ea5b..11975973ed 100644
--- a/datafusion/physical-optimizer/src/join_selection.rs
+++ b/datafusion/physical-optimizer/src/join_selection.rs
@@ -319,13 +319,12 @@ fn statistical_join_selection_subrule(
if can_swap_hash_join(hash_join)
&& should_swap_join_order(left, right, context)?
{
- // Null-aware RightAnti only supports CollectLeft
- let partition_mode = if hash_join.null_aware {
- PartitionMode::CollectLeft
- } else {
- PartitionMode::Partitioned
- };
- hash_join.swap_inputs(partition_mode).map(Some)?
+ // A null-aware join is never in `Partitioned` mode
+ // (`NullAwareMode::try_new` rejects it), so the swap
+ // keeps the mode this arm matched on.
+ hash_join
+ .swap_inputs(PartitionMode::Partitioned)
+ .map(Some)?
} else {
None
}
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index 5309f4c6f5..9f77d782b2 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -300,12 +300,14 @@ impl NullAwareMode {
Self::RightAnti if num_keys != 1 => plan_err!(
"null_aware {join_type} joins only support single column join key, got {num_keys} columns"
),
- // Correlated joins share the per-build-row null bitmap across all
- // probe partitions.
- Self::LeftMark { .. }
- | Self::LeftAnti {
- correlated: true, ..
- } if partition_mode == PartitionMode::Partitioned => {
+ // Build-driven joins decide in the final stage from state shared
+ // across the probe partitions: the per-build-row null bitmap for
+ // the correlated forms, and the probe-side NULL flag for the
+ // uncorrelated ones. `Partitioned` gives each output partition its
+ // own build side, so neither is global there.
+ Self::LeftMark { .. } | Self::LeftAnti { .. }
+ if partition_mode == PartitionMode::Partitioned =>
+ {
plan_err!(
"null_aware joins require PartitionMode::CollectLeft, got PartitionMode::Partitioned"
)
@@ -2739,9 +2741,37 @@ mod proto_tests {
let mut node = decodable_node();
node.join_type = protobuf::JoinType::Leftanti.into();
node.null_aware = true;
+ // `decodable_node` is `Partitioned`, which a null-aware join rejects
+ // (see `try_from_proto_rejects_a_null_aware_partitioned_join`).
+ node.partition_mode = protobuf::PartitionMode::CollectLeft.into();
assert!(as_join(&decode(node)).null_aware);
}
+ /// A null-aware join decides in the final stage from state shared across
+ /// the probe partitions, which `PartitionMode::Partitioned` does not give
+ /// it. Decoding goes through the same builder as `try_new`, so a node
+ /// written with that combination is rejected rather than silently
+ /// returning rows for a `NOT IN` whose subquery contains NULL.
+ #[test]
+ fn try_from_proto_rejects_a_null_aware_partitioned_join() {
+ let decoder = StubPlanDecoder::ok();
+ let ctx = ExecutionPlanDecodeCtx::new(&decoder);
+ let mut node = decodable_node();
+ node.join_type = protobuf::JoinType::Leftanti.into();
+ node.null_aware = true;
+ assert_eq!(
+ node.partition_mode,
+ protobuf::PartitionMode::Partitioned as i32
+ );
+
+ let err = HashJoinExec::try_from_proto(&join_node(node), &ctx).unwrap_err();
+ assert!(
+ err.to_string()
+ .contains("null_aware joins require PartitionMode::CollectLeft"),
+ "{err}"
+ );
+ }
+
/// A multi-column `NOT IN` keeps its value key count through the round
/// trip; a node written before the field existed decodes it as 0, which
/// is read as a scalar `NOT IN`.
@@ -10322,6 +10352,49 @@ mod tests {
);
}
+ /// An uncorrelated null-aware `LeftAnti` must reject
+ /// `PartitionMode::Partitioned` just like the correlated forms do.
+ ///
+ /// `LeftAnti` emits build rows in the final stage, and the "a probe
+ /// partition saw a NULL `NOT IN` key" flag lives on the build data
+ /// (`JoinLeftData::probe_completion`). In `Partitioned` mode every output
+ /// partition builds its own build side, so a NULL probe key, hashed to a
+ /// single partition, would only silence that partition's build rows while
+ /// the others still emitted their unmatched ones.
+ #[tokio::test]
+ async fn test_null_aware_validation_left_anti_partitioned() {
+ let left =
+ build_table_two_cols(("c1", &vec![Some(1)]), ("dummy", &vec![Some(10)]));
+ let right =
+ build_table_two_cols(("c2", &vec![Some(1)]), ("dummy", &vec![Some(100)]));
+
+ // A single join key and no filter, so this is the uncorrelated
+ // `LeftAnti` form, which the partition-mode check used to let through.
+ let on = vec![(
+ Arc::new(Column::new_with_schema("c1", &left.schema()).unwrap()) as _,
+ Arc::new(Column::new_with_schema("c2", &right.schema()).unwrap()) as _,
+ )];
+
+ let err = HashJoinExec::try_new(
+ left,
+ right,
+ on,
+ None,
+ &JoinType::LeftAnti,
+ None,
+ PartitionMode::Partitioned,
+ NullEquality::NullEqualsNothing,
+ true,
+ )
+ .expect_err("null_aware LeftAnti must reject PartitionMode::Partitioned");
+
+ assert!(
+ err.to_string()
+ .contains("null_aware joins require PartitionMode::CollectLeft"),
+ "{err}"
+ );
+ }
+
/// Test that null_aware validation rejects PartitionMode::Partitioned (must be CollectLeft).
#[tokio::test]
async fn test_null_aware_validation_partitioned() {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org