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/07/16 22:27:11 UTC
(datafusion) branch gh-readonly-queue/main/pr-23104-943e02932326617137d27248b959173f8f5e3aa5 created (now 8d680db14a)
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-23104-943e02932326617137d27248b959173f8f5e3aa5
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 8d680db14a fix: keep null-aware anti-join NULLs in the pushed dynamic filter (#23104)
This branch includes the following new commits:
new 8d680db14a fix: keep null-aware anti-join NULLs in the pushed dynamic filter (#23104)
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: keep null-aware anti-join NULLs in the pushed dynamic filter (#23104)
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-23104-943e02932326617137d27248b959173f8f5e3aa5
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 8d680db14a9134837bbd3d53497dd58c985c51a7
Author: Moe <md...@gmail.com>
AuthorDate: Thu Jul 16 15:26:49 2026 -0700
fix: keep null-aware anti-join NULLs in the pushed dynamic filter (#23104)
## Which issue does this close?
Closes #23103.
## Rationale for this change
A hash join pushes a build-side dynamic filter (`key IN build_keys`)
down to the probe scan. For a null-aware anti join (`NOT IN`), that
filter drops the probe's NULL rows. But `NOT IN` three-valued logic
needs a probe-side NULL to collapse the whole result to zero rows. With
the NULL filtered away at the scan, before the join's null-aware check
runs, the join returns rows that shouldn't be there.
## What changes are included in this PR?
`SharedBuildAccumulator::build_filter` now ORs `probe_key IS NULL` into
the pushed predicate when the join is `null_aware`. Non-NULL probe rows
still get filtered, so the optimization stays. `HashJoinExec`'s
`null_aware` validation already guarantees a single probe key.
## Are these changes tested?
Yes. Added a parquet-backed case to `null_aware_anti_join.slt`. The
existing cases use in-memory `VALUES`, whose scans never apply the
pushed filter, so they passed despite the bug. The new one sets
`parquet.pushdown_filters = true` so the filter runs row-level. Without
the fix it returns `1, 3`; with it, zero rows.
## Are there any user-facing changes?
A `NOT IN` over a NULL-bearing inner now returns zero rows instead of
leaking rows, when join dynamic filter pushdown and row-level scan
filtering are both on.
---------
Co-authored-by: Andrew Lamb <an...@nerdnetworks.org>
Co-authored-by: Adrian Garcia Badaracco <17...@users.noreply.github.com>
Co-authored-by: rjhallsted <rj...@gmail.com>
---
.../physical-plan/src/joins/hash_join/exec.rs | 1 +
.../src/joins/hash_join/shared_bounds.rs | 42 +++++++++++++-
.../test_files/null_aware_anti_join.slt | 65 ++++++++++++++++++++++
3 files changed, 105 insertions(+), 3 deletions(-)
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index 6f60155ea3..c622663f03 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -1405,6 +1405,7 @@ impl ExecutionPlan for HashJoinExec {
filter,
on_right,
repartition_random_state,
+ self.null_aware,
))
})))
})
diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
index 7146e8dc2e..1fa06b5c6c 100644
--- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs
@@ -37,7 +37,7 @@ use datafusion_common::{DataFusionError, Result, ScalarValue, SharedResult};
use datafusion_expr::Operator;
use datafusion_functions::core::r#struct as struct_func;
use datafusion_physical_expr::expressions::{
- BinaryExpr, CaseExpr, DynamicFilterPhysicalExpr, InListExpr, lit,
+ BinaryExpr, CaseExpr, DynamicFilterPhysicalExpr, InListExpr, IsNullExpr, lit,
};
use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef, ScalarFunctionExpr};
@@ -255,6 +255,9 @@ pub(crate) struct SharedBuildAccumulator {
repartition_random_state: SeededRandomState,
/// Schema of the probe (right) side for evaluating filter expressions
probe_schema: Arc<Schema>,
+ /// Null-aware anti join (`NOT IN`). A probe-side NULL must reach the join so its
+ /// three-valued logic can collapse the result, so the pushed filter keeps NULL rows.
+ null_aware: bool,
}
/// Strategy for filter pushdown (decided at collection time)
@@ -358,6 +361,7 @@ impl SharedBuildAccumulator {
dynamic_filter: Arc<DynamicFilterPhysicalExpr>,
on_right: Vec<PhysicalExprRef>,
repartition_random_state: SeededRandomState,
+ null_aware: bool,
) -> Self {
// Troubleshooting: If partition counts are incorrect, verify this logic matches
// the actual execution pattern in collect_build_side()
@@ -404,6 +408,7 @@ impl SharedBuildAccumulator {
on_right,
repartition_random_state,
probe_schema: right_child.schema(),
+ null_aware,
}
}
@@ -579,7 +584,8 @@ impl SharedBuildAccumulator {
if let Some(filter_expr) =
combine_membership_and_bounds(membership_expr, bounds_expr)
{
- self.dynamic_filter.update(filter_expr)?;
+ self.dynamic_filter
+ .update(self.null_aware_filter(filter_expr))?;
}
}
PartitionStatus::Pending => {
@@ -685,12 +691,40 @@ impl SharedBuildAccumulator {
)?) as Arc<dyn PhysicalExpr>
};
- self.dynamic_filter.update(filter_expr)?;
+ self.dynamic_filter
+ .update(self.null_aware_filter(filter_expr))?;
}
}
Ok(())
}
+
+ /// Wraps a pushdown filter so a null-aware anti join keeps its probe-side NULL rows.
+ ///
+ /// The build-side predicate drops probe rows whose key is NULL, but `NOT IN` three-valued
+ /// logic needs that NULL to reach the join. OR-ing `probe_key IS NULL` preserves the dynamic
+ /// filter's selectivity for non-NULL rows while letting the NULL through.
+ fn null_aware_filter(
+ &self,
+ filter_expr: Arc<dyn PhysicalExpr>,
+ ) -> Arc<dyn PhysicalExpr> {
+ if !self.null_aware {
+ return filter_expr;
+ }
+ debug_assert_eq!(
+ self.on_right.len(),
+ 1,
+ "null_aware anti join must have exactly one probe key"
+ );
+ let probe_key_is_null: Arc<dyn PhysicalExpr> =
+ Arc::new(IsNullExpr::new(Arc::clone(&self.on_right[0])));
+ // Cheap null check first short-circuits before the costlier dynamic filter.
+ Arc::new(BinaryExpr::new(
+ probe_key_is_null,
+ Operator::Or,
+ filter_expr,
+ ))
+ }
}
impl fmt::Debug for SharedBuildAccumulator {
@@ -722,6 +756,7 @@ pub(super) fn make_partitioned_accumulator_for_test(
on_right: vec![],
repartition_random_state: SeededRandomState::with_seed(1),
probe_schema,
+ null_aware: false,
}
}
@@ -778,6 +813,7 @@ mod tests {
on_right,
repartition_random_state: SeededRandomState::with_seed(1),
probe_schema: test_probe_schema(),
+ null_aware: false,
}
}
diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
index b18f3b3ae7..1d12fc33c9 100644
--- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
+++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt
@@ -451,3 +451,68 @@ DROP TABLE customers_test;
statement ok
DROP TABLE all_null_banned;
+
+#############
+## Test: dynamic filter pushdown must not drop inner (probe-side) NULLs.
+## With join dynamic filter pushdown on, the build-side filter pushed to the probe scan would drop
+## inner NULLs, but NOT IN three-valued logic needs them to collapse the result to zero rows. The
+## in-memory VALUES scans above never apply the pushed filter, so this case needs a parquet scan.
+#############
+
+statement ok
+set datafusion.optimizer.enable_join_dynamic_filter_pushdown = true;
+
+# Row-level parquet filtering, so the pushed filter actually drops matching rows instead of only
+# pruning row groups. Without this the single row group is read whole and the NULL never gets dropped.
+statement ok
+set datafusion.execution.parquet.pushdown_filters = true;
+
+statement ok
+CREATE TABLE asa_outer(id INT) AS VALUES (1), (2), (3);
+
+statement ok
+CREATE TABLE asa_inner(eid INT) AS VALUES (2), (NULL);
+
+query I
+COPY asa_outer TO 'test_files/scratch/null_aware_anti_join/asa_outer.parquet' STORED AS PARQUET;
+----
+3
+
+query I
+COPY asa_inner TO 'test_files/scratch/null_aware_anti_join/asa_inner.parquet' STORED AS PARQUET;
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE asa_outer_parquet(id INT)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/null_aware_anti_join/asa_outer.parquet';
+
+statement ok
+CREATE EXTERNAL TABLE asa_inner_parquet(eid INT)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/null_aware_anti_join/asa_inner.parquet';
+
+# Expected: zero rows. Before the fix the pushed dynamic filter dropped inner NULLs, so the join
+# wrongly returned id = 1 and id = 3.
+query I
+SELECT id FROM asa_outer_parquet WHERE id NOT IN (SELECT eid FROM asa_inner_parquet) ORDER BY id;
+----
+
+statement ok
+DROP TABLE asa_outer;
+
+statement ok
+DROP TABLE asa_inner;
+
+statement ok
+DROP TABLE asa_outer_parquet;
+
+statement ok
+DROP TABLE asa_inner_parquet;
+
+statement ok
+RESET datafusion.execution.parquet.pushdown_filters;
+
+statement ok
+RESET datafusion.optimizer.enable_join_dynamic_filter_pushdown;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org