This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new a3f0f939a4 fix: prevent incorrect results when pushing filters through
anti joins (#24045)
a3f0f939a4 is described below
commit a3f0f939a46737e4c53a74f0c39337bdcece1aa2
Author: Burak Şen <[email protected]>
AuthorDate: Tue Aug 4 07:28:24 2026 +0300
fix: prevent incorrect results when pushing filters through anti joins
(#24045)
## Which issue does this PR close?
- Closes #24002.
## Rationale for this change
Pushing filters to an anti join's non-output side can produce incorrect
results.
## What changes are included in this PR?
Restrict anti-join filter pushdown to the output side while preserving
two-sided join-key pushdown for semi joins.
## Are these changes tested?
Yes, with updated unit and SQL logic tests.
## Are there any user-facing changes?
no API changes but some downstream expected plans may change
---
.../tests/physical_optimizer/filter_pushdown.rs | 36 +++++++++++++-----
.../physical-plan/src/joins/hash_join/exec.rs | 32 +++++++---------
.../test_files/dynamic_filter_pushdown_config.slt | 43 +++++++++++++++++++---
3 files changed, 77 insertions(+), 34 deletions(-)
diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
index 7593fe3515..e6f51266c4 100644
--- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
+++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs
@@ -1506,10 +1506,8 @@ fn test_hashjoin_parent_filter_pushdown_mark_join() {
);
}
-/// Test that filters on join key columns are pushed to both sides of
semi/anti joins.
-/// For LeftSemi/LeftAnti, the output only contains left columns, but filters
on
-/// join key columns can also be pushed to the right (non-preserved) side
because
-/// the equijoin condition guarantees the key values match.
+/// Semi-join key filters can be pushed to both sides, but anti-join filters
must
+/// only rely on the output side to preserve their semantics.
#[test]
fn test_hashjoin_parent_filter_pushdown_semi_anti_join() {
use datafusion_common::JoinType;
@@ -1539,8 +1537,8 @@ fn test_hashjoin_parent_filter_pushdown_semi_anti_join() {
let join = Arc::new(
HashJoinExec::try_new(
left_scan,
- right_scan,
- on,
+ Arc::clone(&right_scan),
+ on.clone(),
None,
&JoinType::LeftSemi,
None,
@@ -1579,6 +1577,24 @@ fn test_hashjoin_parent_filter_pushdown_semi_anti_join()
{
- DataSourceExec: file_groups={1 group: [[test.parquet]]},
projection=[k, w], file_type=test, pushdown_supported=true, predicate=k@0 = x
"
);
+
+ let join = Arc::new(
+ HashJoinExec::try_new(
+ TestScanBuilder::new(Arc::clone(&left_schema)).build(),
+ right_scan,
+ on,
+ None,
+ &JoinType::LeftAnti,
+ None,
+ PartitionMode::Partitioned,
+ datafusion_common::NullEquality::NullEqualsNothing,
+ false,
+ )
+ .unwrap(),
+ );
+ let predicate = Arc::new(Literal::new(ScalarValue::Boolean(Some(false))));
+ let plan = Arc::new(FilterExec::try_new(predicate, join).unwrap());
+ assert_parent_filter_remains(plan);
}
#[test]
@@ -1817,13 +1833,13 @@ fn col_lit_predicate(
))
}
-fn assert_parent_filter_remains_above_aggregate(plan: Arc<dyn ExecutionPlan>) {
+fn assert_parent_filter_remains(plan: Arc<dyn ExecutionPlan>) {
let mut config = ConfigOptions::default();
config.execution.parquet.pushdown_filters = true;
let optimized = FilterPushdown::new().optimize(plan, &config).unwrap();
assert!(
optimized.downcast_ref::<FilterExec>().is_some(),
- "parent filter must remain above aggregate"
+ "parent filter must remain"
);
}
@@ -2152,7 +2168,7 @@ fn
test_no_pushdown_constant_false_through_global_aggregate() {
let predicate = Arc::new(Literal::new(ScalarValue::Boolean(Some(false))));
let plan = Arc::new(FilterExec::try_new(predicate, aggregate).unwrap());
- assert_parent_filter_remains_above_aggregate(plan);
+ assert_parent_filter_remains(plan);
}
#[test]
@@ -2189,7 +2205,7 @@ fn
test_no_pushdown_constant_false_through_empty_grouping_set() {
let predicate = Arc::new(Literal::new(ScalarValue::Boolean(Some(false))));
let plan = Arc::new(FilterExec::try_new(predicate, aggregate).unwrap());
- assert_parent_filter_remains_above_aggregate(plan);
+ assert_parent_filter_remains(plan);
}
#[test]
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs
b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index 9d9c867c27..9a1c0b0f63 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -1645,14 +1645,11 @@ impl ExecutionPlan for HashJoinExec {
};
});
- // For semi/anti joins, the non-preserved side's columns are not in the
- // output, but filters on join key columns can still be pushed there.
- // We find output columns that are join keys on the preserved side and
- // add their output indices to the non-preserved side's allowed set.
- // The name-based remap in FilterRemapper will then match them to the
- // corresponding column in the non-preserved child's schema.
+ // For semi joins, filters on output join keys can also be pushed to
the
+ // non-output side: every emitted row has an equal key there. This is
not
+ // true for anti joins, whose emitted rows have no match.
match self.join_type {
- JoinType::LeftSemi | JoinType::LeftAnti => {
+ JoinType::LeftSemi => {
let left_key_indices: HashSet<usize> = self
.on
.iter()
@@ -1666,7 +1663,7 @@ impl ExecutionPlan for HashJoinExec {
}
}
}
- JoinType::RightSemi | JoinType::RightAnti => {
+ JoinType::RightSemi => {
let right_key_indices: HashSet<usize> = self
.on
.iter()
@@ -1969,21 +1966,18 @@ impl HashJoinExec {
/// Determines which sides of a join are "preserved" for filter pushdown.
///
/// A preserved side means filters on that side's columns can be safely pushed
-/// below the join. This mirrors the logic in the logical optimizer's
-/// `lr_is_preserved` in `datafusion/optimizer/src/push_down_filter.rs`.
+/// below the join. This mostly mirrors the logical optimizer's
`lr_is_preserved`;
+/// semi joins additionally allow join-key filters on the non-output side.
fn lr_is_preserved(join_type: JoinType) -> (bool, bool) {
match join_type {
JoinType::Inner => (true, true),
JoinType::Left => (true, false),
JoinType::Right => (false, true),
JoinType::Full => (false, false),
- // Filters in semi/anti joins are either on the preserved side, or on
join keys,
- // as all output columns come from the preserved side. Join key
filters can be
- // safely pushed down into the other side.
- JoinType::LeftSemi | JoinType::LeftAnti => (true, true),
- JoinType::RightSemi | JoinType::RightAnti => (true, true),
- JoinType::LeftMark => (true, false),
- JoinType::RightMark => (false, true),
+ // Callers restrict the non-output side of semi joins to join-key
columns.
+ JoinType::LeftSemi | JoinType::RightSemi => (true, true),
+ JoinType::LeftAnti | JoinType::LeftMark => (true, false),
+ JoinType::RightAnti | JoinType::RightMark => (false, true),
}
}
@@ -6848,10 +6842,10 @@ mod tests {
assert_eq!(lr_is_preserved(JoinType::Right), (false, true));
assert_eq!(lr_is_preserved(JoinType::Full), (false, false));
assert_eq!(lr_is_preserved(JoinType::LeftSemi), (true, true));
- assert_eq!(lr_is_preserved(JoinType::LeftAnti), (true, true));
+ assert_eq!(lr_is_preserved(JoinType::LeftAnti), (true, false));
assert_eq!(lr_is_preserved(JoinType::LeftMark), (true, false));
assert_eq!(lr_is_preserved(JoinType::RightSemi), (true, true));
- assert_eq!(lr_is_preserved(JoinType::RightAnti), (true, true));
+ assert_eq!(lr_is_preserved(JoinType::RightAnti), (false, true));
assert_eq!(lr_is_preserved(JoinType::RightMark), (false, true));
}
diff --git
a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
index c51a127986..eec6e5ae17 100644
--- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
+++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
@@ -457,10 +457,9 @@ ORDER BY l.id LIMIT 2;
1 left1
3 left3
-# ANTI JOIN with TopK parent: TopK generates a dynamic filter on `id` (join
-# key) that pushes through the LeftAnti join to both the preserved and
-# non-preserved sides. The HashJoin pushes the self-generated filter to the
-# right hand side of the LeftAnti join.
+# ANTI JOIN with TopK parent: the TopK dynamic filter on `id` is pushed only
+# to the preserved output side. Filtering the non-output side can create
+# anti-join output.
query TT
EXPLAIN SELECT l.*
FROM left_parquet l
@@ -479,7 +478,7 @@ physical_plan
01)SortExec: TopK(fetch=2), expr=[id@0 ASC NULLS LAST],
preserve_partitioning=[false]
02)--HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(id@0, id@0)],
null_aware
03)----DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]},
projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ],
dynamic_rg_pruning=eligible
-04)----DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]},
projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ],
dynamic_rg_pruning=eligible
+04)----DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]},
projection=[id], file_type=parquet
# Correctness check
query IT
@@ -491,6 +490,40 @@ ORDER BY l.id LIMIT 2;
2 left2
4 left4
+# A parent filter must remain when only an anti join's non-output side accepts
+# pushdown; otherwise filtering that side creates incorrect anti-join rows.
+statement ok
+SET datafusion.optimizer.max_passes = 0;
+
+statement ok
+SET datafusion.optimizer.join_reordering = false;
+
+statement ok
+SET datafusion.execution.parquet.pushdown_filters = true;
+
+query I
+SELECT count(*)
+FROM join_left l LEFT ANTI JOIN right_parquet r USING (id)
+WHERE false;
+----
+0
+
+query I
+SELECT count(*)
+FROM right_parquet r RIGHT ANTI JOIN join_left l USING (id)
+WHERE false;
+----
+0
+
+statement ok
+RESET datafusion.optimizer.max_passes;
+
+statement ok
+RESET datafusion.optimizer.join_reordering;
+
+statement ok
+RESET datafusion.execution.parquet.pushdown_filters;
+
# Test 3: Test independent control
# Disable TopK, keep Join enabled
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
