From 5d3b1883230eefdfd7699b71ebc4ae8fbf1b911e Mon Sep 17 00:00:00 2001 From: Burak Sen Date: Sat, 1 Aug 2026 18:29:56 +0300 Subject: [PATCH 1/2] dont delete filter in filterpushdown antijoin --- .../physical_optimizer/filter_pushdown.rs | 36 +++++++++++++------ .../physical-plan/src/joins/hash_join/exec.rs | 32 +++++++---------- .../dynamic_filter_pushdown_config.slt | 9 +++-- 3 files changed, 43 insertions(+), 34 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 7593fe351548e..e6f51266c4611 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) { +fn assert_parent_filter_remains(plan: Arc) { let mut config = ConfigOptions::default(); config.execution.parquet.pushdown_filters = true; let optimized = FilterPushdown::new().optimize(plan, &config).unwrap(); assert!( optimized.downcast_ref::().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 9d9c867c2724b..9a1c0b0f63545 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 = self .on .iter() @@ -1666,7 +1663,7 @@ impl ExecutionPlan for HashJoinExec { } } } - JoinType::RightSemi | JoinType::RightAnti => { + JoinType::RightSemi => { let right_key_indices: HashSet = 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 c51a127986421..4d241e1fca8f9 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 From f682806f48c513e2c8ce9568f44c173b90191218 Mon Sep 17 00:00:00 2001 From: Burak Sen Date: Sun, 2 Aug 2026 12:43:57 +0300 Subject: [PATCH 2/2] add slt test for failure case --- .../dynamic_filter_pushdown_config.slt | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 4d241e1fca8f9..eec6e5ae179bc 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -490,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