fix: prevent incorrect results when pushing filters through anti joins

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]

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));
     }
 

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