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/13 11:50:04 UTC
(datafusion) branch gh-readonly-queue/main/pr-23104-65f9726f753bea6a13ca7f7356eede0127aa4b87 created (now 173440e35f)
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-65f9726f753bea6a13ca7f7356eede0127aa4b87
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 173440e35f fix: keep null-aware anti-join NULLs in the pushed dynamic filter (#23104)
This branch includes the following new commits:
new 173440e35f 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-65f9726f753bea6a13ca7f7356eede0127aa4b87
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 173440e35f83cdd3be181fa4b4f1decc887da171
Author: Moe <md...@gmail.com>
AuthorDate: Mon Jul 13 04:49:40 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.
---
.../physical-plan/src/joins/hash_join/exec.rs | 1 +
.../src/joins/hash_join/shared_bounds.rs | 41 +++++++++++++-
.../test_files/null_aware_anti_join.slt | 65 ++++++++++++++++++++++
3 files changed, 104 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 56e7132dc4..545ff47def 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..26f5a55072 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,
}
}
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