Refactor dynamic filter side logic in join types and update related tests

Author: kosiew

datafusion/common/src/join_type.rs

@@ -145,39 +145,6 @@ impl JoinType {
     pub fn preserves_right(self) -> bool {
         self.preserves(JoinSide::Right)
     }
-
-    /// Returns the input side eligible for dynamic filter pushdown.
-    ///
-    /// The side returned here can have a `DynamicFilterPhysicalExpr` pushed
-    /// into it, allowing values read from the opposite input to prune rows
-    /// before the join executes. When both inputs must be preserved,
-    /// dynamic filter pushdown is not supported and [`JoinSide::None`] is
-    /// returned.
-    ///
-    /// If neither input is preserving (for example with [`JoinType::Inner`],
-    /// [`JoinType::LeftSemi`], [`JoinType::RightSemi`],
-    /// [`JoinType::LeftAnti`], or [`JoinType::RightAnti`]), either side could
-    /// in principle receive the pushed filter. DataFusion selects the probe
-    /// side: for [`JoinType::LeftSemi`] and [`JoinType::LeftAnti`] this is the
-    /// left input, for [`JoinType::RightSemi`] and [`JoinType::RightAnti`] it
-    /// is the right input, and for other joins the right input is used by
-    /// default as joins typically treat the right as the probe side.
-    pub fn dynamic_filter_side(self) -> JoinSide {
-        use JoinSide::*;
-        let preserves_left = self.preserves_left();
-        let preserves_right = self.preserves_right();
-
-        match (preserves_left, preserves_right) {
-            (true, true) => None,
-            (true, false) => Right,
-            (false, true) => Left,
-            (false, false) => match self {
-                JoinType::LeftSemi | JoinType::LeftAnti => Left,
-                JoinType::RightSemi | JoinType::RightAnti => Right,
-                _ => Right,
-            },
-        }
-    }
 }
 
 impl Display for JoinType {
@@ -314,42 +281,4 @@ mod tests {
         assert!(!JoinType::RightAnti.preserves(Left));
         assert!(!JoinType::RightAnti.preserves(Right));
     }
-
-    #[test]
-    fn test_dynamic_filter_side() {
-        use JoinSide::*;
-
-        assert_eq!(JoinType::Inner.dynamic_filter_side(), Right);
-        assert_eq!(JoinType::Left.dynamic_filter_side(), Right);
-        assert_eq!(JoinType::Right.dynamic_filter_side(), Left);
-        assert_eq!(JoinType::Full.dynamic_filter_side(), None);
-        assert_eq!(JoinType::LeftSemi.dynamic_filter_side(), Left);
-        assert_eq!(JoinType::RightSemi.dynamic_filter_side(), Right);
-        assert_eq!(JoinType::LeftAnti.dynamic_filter_side(), Left);
-        assert_eq!(JoinType::RightAnti.dynamic_filter_side(), Right);
-        assert_eq!(JoinType::LeftMark.dynamic_filter_side(), Right);
-        assert_eq!(JoinType::RightMark.dynamic_filter_side(), Left);
-    }
-
-    #[test]
-    fn test_dynamic_filter_side_preservation_logic() {
-        use JoinSide::*;
-
-        for jt in [JoinType::Left, JoinType::LeftMark] {
-            assert!(jt.preserves_left());
-            assert!(!jt.preserves_right());
-            assert_eq!(jt.dynamic_filter_side(), Right);
-        }
-
-        for jt in [JoinType::Right, JoinType::RightMark] {
-            assert!(!jt.preserves_left());
-            assert!(jt.preserves_right());
-            assert_eq!(jt.dynamic_filter_side(), Left);
-        }
-
-        let jt = JoinType::Full;
-        assert!(jt.preserves_left());
-        assert!(jt.preserves_right());
-        assert_eq!(jt.dynamic_filter_side(), None);
-    }
 }

datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs

@@ -1606,7 +1606,7 @@ fn test_hashjoin_handle_child_pushdown_result_dynamic_filter_left() {
     )
     .unwrap();
 
-    assert_eq!(join.join_type.dynamic_filter_side(), JoinSide::Left);
+    assert_eq!(join.dynamic_filter_side(), JoinSide::Left);
 
     let keys: Vec<_> = on.iter().map(|(l, _)| l.clone()).collect();
     let df_expr = Arc::new(DynamicFilterPhysicalExpr::new(keys, lit(true)));

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -86,6 +86,24 @@ use parking_lot::Mutex;
 const HASH_JOIN_SEED: RandomState =
     RandomState::with_seeds('J' as u64, 'O' as u64, 'I' as u64, 'N' as u64);
 
+fn dynamic_filter_pushdown_side(join_type: JoinType) -> JoinSide {
+    use JoinSide::*;
+
+    let preserves_left = join_type.preserves_left();
+    let preserves_right = join_type.preserves_right();
+
+    match (preserves_left, preserves_right) {
+        (true, true) => None,
+        (true, false) => Right,
+        (false, true) => Left,
+        (false, false) => match join_type {
+            JoinType::LeftSemi | JoinType::LeftAnti => Left,
+            JoinType::RightSemi | JoinType::RightAnti => Right,
+            _ => Right,
+        },
+    }
+}
+
 /// HashTable and input data for the left (build side) of a join
 pub(super) struct JoinLeftData {
     /// The hash table with indices into `batch`
@@ -496,6 +514,11 @@ impl HashJoinExec {
         &self.right
     }
 
+    /// Preferred input side for installing a dynamic filter.
+    pub fn dynamic_filter_side(&self) -> JoinSide {
+        dynamic_filter_pushdown_side(self.join_type)
+    }
+
     /// Set of common columns used to join on
     pub fn on(&self) -> &[(PhysicalExprRef, PhysicalExprRef)] {
         &self.on
@@ -927,7 +950,7 @@ impl ExecutionPlan for HashJoinExec {
         }
 
         let enable_dynamic_filter_pushdown = self.dynamic_filter.is_some();
-        let df_side = self.join_type.dynamic_filter_side();
+        let df_side = self.dynamic_filter_side();
 
         let join_metrics = BuildProbeJoinMetrics::new(partition, &self.metrics);
         let left_fut = match self.mode {
@@ -1042,6 +1065,7 @@ impl ExecutionPlan for HashJoinExec {
             on_right,
             self.filter.clone(),
             self.join_type,
+            df_side,
             right_stream,
             self.random_state.clone(),
             join_metrics,
@@ -1131,7 +1155,7 @@ impl ExecutionPlan for HashJoinExec {
         parent_filters: Vec<Arc<dyn PhysicalExpr>>,
         config: &ConfigOptions,
     ) -> Result<FilterDescription> {
-        let df_side = self.join_type.dynamic_filter_side();
+        let df_side = self.dynamic_filter_side();
         self.gather_filters_for_pushdown_with_side(phase, parent_filters, config, df_side)
     }
 
@@ -1141,7 +1165,7 @@ impl ExecutionPlan for HashJoinExec {
         child_pushdown_result: ChildPushdownResult,
         config: &ConfigOptions,
     ) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
-        let df_side = self.join_type.dynamic_filter_side();
+        let df_side = self.dynamic_filter_side();
         self.handle_child_pushdown_result_with_side(
             phase,
             child_pushdown_result,
@@ -1680,7 +1704,7 @@ mod tests {
         #[case] join_type: JoinType,
         #[case] expected_side: JoinSide,
     ) {
-        assert_eq!(join_type.dynamic_filter_side(), expected_side);
+        assert_eq!(dynamic_filter_pushdown_side(join_type), expected_side);
     }
 
     #[test]

datafusion/physical-plan/src/joins/hash_join/stream.rs

@@ -238,6 +238,8 @@ pub(super) struct HashJoinStream {
     filter: Option<JoinFilter>,
     /// type of the join (left, right, semi, etc)
     join_type: JoinType,
+    /// Preferred input side for dynamic filter installation
+    dynamic_filter_side: JoinSide,
     /// right (probe) input
     right: SendableRecordBatchStream,
     /// Random state used for hashing initialization
@@ -360,6 +362,7 @@ impl HashJoinStream {
         on_right: Vec<PhysicalExprRef>,
         filter: Option<JoinFilter>,
         join_type: JoinType,
+        dynamic_filter_side: JoinSide,
         right: SendableRecordBatchStream,
         random_state: RandomState,
         join_metrics: BuildProbeJoinMetrics,
@@ -380,6 +383,7 @@ impl HashJoinStream {
             on_right,
             filter,
             join_type,
+            dynamic_filter_side,
             right,
             random_state,
             join_metrics,
@@ -469,7 +473,7 @@ impl HashJoinStream {
         // Dynamic filter coordination between partitions:
         // Report bounds to the accumulator which will handle synchronization and filter updates
         if let Some(ref bounds_accumulator) = self.bounds_accumulator {
-            if self.join_type.dynamic_filter_side() == JoinSide::Right {
+            if self.dynamic_filter_side == JoinSide::Right {
                 let bounds_accumulator = Arc::clone(bounds_accumulator);
 
                 let left_side_partition_id = match self.mode {
@@ -508,7 +512,7 @@ impl HashJoinStream {
         match ready!(self.right.poll_next_unpin(cx)) {
             None => {
                 if let Some(ref bounds_accumulator) = self.bounds_accumulator {
-                    if self.join_type.dynamic_filter_side() == JoinSide::Left {
+                    if self.dynamic_filter_side == JoinSide::Left {
                         if let Some(accs) = self.probe_bounds_accumulators.take() {
                             let right_bounds = if self.probe_side_row_count > 0 {
                                 Some(