rename

Author: jayshrivastava

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -2835,7 +2835,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_is_used() {
 
         // Verify that a dynamic filter was created
         let dynamic_filter = hash_join
-            .dynamic_filter()
+            .dynamic_filter_expr()
             .expect("Dynamic filter should be created");
 
         // Verify that is_used() returns the expected value based on probe side support.

datafusion/physical-plan/src/aggregates/mod.rs

@@ -893,6 +893,47 @@ impl AggregateExec {
         &self.filter_expr
     }
 
+    /// Returns the dynamic filter expression for this aggregate, if set.
+    pub fn dynamic_filter_expr(&self) -> Option<&Arc<DynamicFilterPhysicalExpr>> {
+        self.dynamic_filter.as_ref().map(|df| &df.filter)
+    }
+
+    /// Replace the dynamic filter expression. This method errors if the aggregate does not
+    /// support dynamic filtering or if the filter expression is incompatible with this
+    /// [`AggregateExec`].
+    pub fn with_dynamic_filter_expr(
+        mut self,
+        filter: Arc<DynamicFilterPhysicalExpr>,
+    ) -> Result<Self> {
+        // If there is no dynamic filter state initialized via `try_new`, then
+        // we can safely assume that the aggregate does not support dynamic filtering.
+        let Some(dyn_filter) = self.dynamic_filter.as_ref() else {
+            return internal_err!("Aggregate does not support dynamic filtering");
+        };
+
+        // Validate that the filter is compatible with the aggregation columns.
+        let cols = self.cols_for_dynamic_filter(&dyn_filter.supported_accumulators_info);
+        if cols.len() != filter.children().len() {
+            return internal_err!(
+                "Dynamic filter expression is incompatible with aggregate due to mismatched number of columns"
+            );
+        }
+        for (col, child) in cols.iter().zip(filter.children()) {
+            if !col.eq(child) {
+                return internal_err!(
+                    "Dynamic filter expression is incompatible with aggregate due to mismatched column references {col} != {child}"
+                );
+            }
+        }
+
+        // Overwrite our filter
+        self.dynamic_filter = Some(Arc::new(AggrDynFilter {
+            filter,
+            supported_accumulators_info: dyn_filter.supported_accumulators_info.clone(),
+        }));
+        Ok(self)
+    }
+
     /// Input plan
     pub fn input(&self) -> &Arc<dyn ExecutionPlan> {
         &self.input
@@ -1048,47 +1089,6 @@ impl AggregateExec {
         &self.input_order_mode
     }
 
-    /// Returns the dynamic filter expression for this aggregate, if set.
-    pub fn dynamic_filter(&self) -> Option<&Arc<DynamicFilterPhysicalExpr>> {
-        self.dynamic_filter.as_ref().map(|df| &df.filter)
-    }
-
-    /// Replace the dynamic filter expression. This method errors if the aggregate does not
-    /// support dynamic filtering or if the filter expression is incompatible with this
-    /// [`AggregateExec`].
-    pub fn with_dynamic_filter(
-        mut self,
-        filter: Arc<DynamicFilterPhysicalExpr>,
-    ) -> Result<Self> {
-        // If there is no dynamic filter state initialized via `try_new`, then
-        // we can safely assume that the aggregate does not support dynamic filtering.
-        let Some(dyn_filter) = self.dynamic_filter.as_ref() else {
-            return internal_err!("Aggregate does not support dynamic filtering");
-        };
-
-        // Validate that the filter is compatible with the aggregation columns.
-        let cols = self.cols_for_dynamic_filter(&dyn_filter.supported_accumulators_info);
-        if cols.len() != filter.children().len() {
-            return internal_err!(
-                "Dynamic filter expression is incompatible with aggregate due to mismatched number of columns"
-            );
-        }
-        for (col, child) in cols.iter().zip(filter.children()) {
-            if !col.eq(child) {
-                return internal_err!(
-                    "Dynamic filter expression is incompatible with aggregate due to mismatched column references {col} != {child}"
-                );
-            }
-        }
-
-        // Overwrite our filter
-        self.dynamic_filter = Some(Arc::new(AggrDynFilter {
-            filter,
-            supported_accumulators_info: dyn_filter.supported_accumulators_info.clone(),
-        }));
-        Ok(self)
-    }
-
     /// Estimates output statistics for this aggregate node.
     ///
     /// For grouped aggregations with known input row count > 1, the output row
@@ -4842,7 +4842,7 @@ mod tests {
         Ok(())
     }
 
-    /// Test that [`AggregateExec::with_dynamic_filter`] overrides the existing dynamic filter
+    /// Test that [`AggregateExec::with_dynamic_filter_expr`] overrides the existing dynamic filter
     #[test]
     fn test_with_dynamic_filter() -> Result<()> {
         let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
@@ -4869,11 +4869,11 @@ mod tests {
             vec![col("a", &schema)?],
             lit(false),
         ));
-        let agg = agg.with_dynamic_filter(Arc::clone(&new_df))?;
+        let agg = agg.with_dynamic_filter_expr(Arc::clone(&new_df))?;
 
         // The aggregate's filter should now resolve to the new inner expression.
         let swapped = agg
-            .dynamic_filter()
+            .dynamic_filter_expr()
             .expect("should still have dynamic filter")
             .current()?;
         assert_eq!(format!("{swapped}"), format!("{}", lit(false)));
@@ -4884,19 +4884,18 @@ mod tests {
             Arc::<DynamicFilterPhysicalExpr>::clone(&new_df);
         let remapped_pexpr =
             new_df_as_pexpr.with_new_children(vec![col("a", &schema)?])?;
-        let Ok(remapped_df) = (remapped_pexpr
-            as Arc<dyn std::any::Any + Send + Sync>)
+        let Ok(remapped_df) = (remapped_pexpr as Arc<dyn std::any::Any + Send + Sync>)
             .downcast::<DynamicFilterPhysicalExpr>()
         else {
             panic!("should be DynamicFilterPhysicalExpr after with_new_children");
         };
         // Hard to assert this because the filter is identical. No error means
         // the filter was accepted. That's a good enough assertion for now.
-        let _agg = agg.with_dynamic_filter(remapped_df)?;
+        let _agg = agg.with_dynamic_filter_expr(remapped_df)?;
         Ok(())
     }
 
-    /// Test that [`AggregateExec::with_dynamic_filter`] errors when the aggregate does not support dynamic filtering
+    /// Test that [`AggregateExec::with_dynamic_filter_expr`] errors when the aggregate does not support dynamic filtering
     #[test]
     fn test_with_dynamic_filter_error_unsupported() -> Result<()> {
         let schema = Arc::new(Schema::new(vec![
@@ -4919,17 +4918,17 @@ mod tests {
             child,
             Arc::clone(&schema),
         )?;
-        assert!(agg.dynamic_filter().is_none());
+        assert!(agg.dynamic_filter_expr().is_none());
 
         let df = Arc::new(DynamicFilterPhysicalExpr::new(
             vec![col("a", &schema)?],
             lit(true),
         ));
-        assert!(agg.with_dynamic_filter(df).is_err());
+        assert!(agg.with_dynamic_filter_expr(df).is_err());
         Ok(())
     }
 
-    /// Test that [`AggregateExec::with_dynamic_filter`] errors when the column is not in the schema
+    /// Test that [`AggregateExec::with_dynamic_filter_expr`] errors when the column is not in the schema
     #[test]
     fn test_with_dynamic_filter_error_column_mismatch() -> Result<()> {
         let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
@@ -4953,7 +4952,7 @@ mod tests {
             vec![Arc::new(Column::new("bad", 99)) as _],
             lit(true),
         ));
-        assert!(agg.with_dynamic_filter(df).is_err());
+        assert!(agg.with_dynamic_filter_expr(df).is_err());
         Ok(())
     }
 }

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

@@ -904,7 +904,7 @@ impl HashJoinExec {
 
     /// Get the dynamic filter expression for testing purposes.
     /// Returns the dynamic filter expression for this hash join, if set.
-    pub fn dynamic_filter(&self) -> Option<&Arc<DynamicFilterPhysicalExpr>> {
+    pub fn dynamic_filter_expr(&self) -> Option<&Arc<DynamicFilterPhysicalExpr>> {
         self.dynamic_filter.as_ref().map(|df| &df.filter)
     }
 
@@ -914,7 +914,7 @@ impl HashJoinExec {
     ///
     /// Validates that the filter's children reference valid columns in
     /// the probe (right) side's schema.
-    pub fn with_dynamic_filter(
+    pub fn with_dynamic_filter_expr(
         mut self,
         filter: Arc<DynamicFilterPhysicalExpr>,
     ) -> Result<Self> {
@@ -6345,15 +6345,17 @@ mod tests {
             NullEquality::NullEqualsNothing,
             false,
         )?;
-        assert!(join.dynamic_filter().is_none());
+        assert!(join.dynamic_filter_expr().is_none());
 
         let df = Arc::new(DynamicFilterPhysicalExpr::new(
             vec![Arc::new(Column::new("b1", 1)) as _],
             lit(true),
         ));
-        let join = join.with_dynamic_filter(Arc::clone(&df))?;
+        let join = join.with_dynamic_filter_expr(Arc::clone(&df))?;
 
-        let restored = join.dynamic_filter().expect("should have dynamic filter");
+        let restored = join
+            .dynamic_filter_expr()
+            .expect("should have dynamic filter");
         assert_eq!(
             restored
                 .expression_id()
@@ -6387,7 +6389,7 @@ mod tests {
             vec![Arc::new(Column::new("bad", 99)) as _],
             lit(true),
         ));
-        assert!(join.with_dynamic_filter(df).is_err());
+        assert!(join.with_dynamic_filter_expr(df).is_err());
         Ok(())
     }
 }

datafusion/physical-plan/src/sorts/sort.rs

@@ -979,7 +979,7 @@ impl SortExec {
     }
 
     /// Returns the dynamic filter expression for this sort (TopK), if set.
-    pub fn dynamic_filter(&self) -> Option<Arc<DynamicFilterPhysicalExpr>> {
+    pub fn dynamic_filter_expr(&self) -> Option<Arc<DynamicFilterPhysicalExpr>> {
         self.filter.as_ref().map(|f| f.read().expr())
     }
 
@@ -990,7 +990,7 @@ impl SortExec {
     ///
     /// Validates that the filter's children reference valid columns in
     /// the sort's input schema.
-    pub fn with_dynamic_filter(
+    pub fn with_dynamic_filter_expr(
         mut self,
         filter: Arc<DynamicFilterPhysicalExpr>,
     ) -> Result<Self> {
@@ -2764,7 +2764,7 @@ mod tests {
 
         // SortExec with fetch creates a dynamic filter automatically.
         let original_id = sort
-            .dynamic_filter()
+            .dynamic_filter_expr()
             .expect("should have dynamic filter with fetch")
             .expression_id()
             .expect("DynamicFilterPhysicalExpr always has an expression_id");
@@ -2777,9 +2777,9 @@ mod tests {
         let new_id = new_df
             .expression_id()
             .expect("DynamicFilterPhysicalExpr always has an expression_id");
-        let sort = sort.with_dynamic_filter(Arc::clone(&new_df))?;
+        let sort = sort.with_dynamic_filter_expr(Arc::clone(&new_df))?;
         let restored_id = sort
-            .dynamic_filter()
+            .dynamic_filter_expr()
             .expect("should still have dynamic filter")
             .expression_id()
             .expect("DynamicFilterPhysicalExpr always has an expression_id");
@@ -2808,7 +2808,7 @@ mod tests {
             vec![Arc::new(Column::new("bad", 99)) as _],
             lit(true),
         ));
-        assert!(sort.with_dynamic_filter(df).is_err());
+        assert!(sort.with_dynamic_filter_expr(df).is_err());
         Ok(())
     }
 

datafusion/proto/src/physical_plan/mod.rs

@@ -1313,7 +1313,7 @@ impl protobuf::PhysicalPlanNode {
             if let Ok(df) = (dynamic_filter_expr as Arc<dyn Any + Send + Sync>)
                 .downcast::<DynamicFilterPhysicalExpr>()
             {
-                agg.with_dynamic_filter(df)?
+                agg.with_dynamic_filter_expr(df)?
             } else {
                 agg
             }
@@ -1448,7 +1448,7 @@ impl protobuf::PhysicalPlanNode {
             if let Ok(df) = (dynamic_filter_expr as Arc<dyn Any + Send + Sync>)
                 .downcast::<DynamicFilterPhysicalExpr>()
             {
-                hash_join = hash_join.with_dynamic_filter(df)?;
+                hash_join = hash_join.with_dynamic_filter_expr(df)?;
             }
         }
 
@@ -1699,7 +1699,7 @@ impl protobuf::PhysicalPlanNode {
             if let Ok(df) = (dynamic_filter_expr as Arc<dyn Any + Send + Sync>)
                 .downcast::<DynamicFilterPhysicalExpr>()
             {
-                new_sort.with_dynamic_filter(df)?
+                new_sort.with_dynamic_filter_expr(df)?
             } else {
                 new_sort
             }
@@ -2514,7 +2514,7 @@ impl protobuf::PhysicalPlanNode {
         };
 
         let dynamic_filter = exec
-            .dynamic_filter()
+            .dynamic_filter_expr()
             .map(|df| {
                 let df_expr: Arc<dyn PhysicalExpr> =
                     Arc::clone(df) as Arc<dyn PhysicalExpr>;
@@ -2867,7 +2867,7 @@ impl protobuf::PhysicalPlanNode {
                     limit,
                     has_grouping_set: exec.group_expr().has_grouping_set(),
                     dynamic_filter: exec
-                        .dynamic_filter()
+                        .dynamic_filter_expr()
                         .map(|df| {
                             let df_expr: Arc<dyn PhysicalExpr> =
                                 Arc::clone(df) as Arc<dyn PhysicalExpr>;
@@ -3168,7 +3168,7 @@ impl protobuf::PhysicalPlanNode {
             })
             .collect::<Result<Vec<_>>>()?;
         let dynamic_filter = exec
-            .dynamic_filter()
+            .dynamic_filter_expr()
             .map(|df| {
                 let df_expr: Arc<dyn PhysicalExpr> = df as Arc<dyn PhysicalExpr>;
                 proto_converter.physical_expr_to_proto(&df_expr, codec)

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -3538,7 +3538,7 @@ fn test_hash_join_with_dynamic_filter_roundtrip() -> Result<()> {
         .downcast_ref::<HashJoinExec>()
         .expect("Should be HashJoinExec");
     let deserialized_hash_join_df = deserialized_join
-        .dynamic_filter()
+        .dynamic_filter_expr()
         .expect("HashJoinExec should have a dynamic filter after roundtrip");
 
     // Extract the dynamic filter pushed down to the probe side's ParquetSource.
@@ -3691,7 +3691,7 @@ fn test_aggregate_with_dynamic_filter_roundtrip() -> Result<()> {
         .downcast_ref::<AggregateExec>()
         .expect("Should be AggregateExec");
     let deserialized_agg_df = deserialized_agg
-        .dynamic_filter()
+        .dynamic_filter_expr()
         .expect("AggregateExec should have a dynamic filter after roundtrip");
 
     // Extract the dynamic filter pushed down to the child ParquetSource.
@@ -3759,7 +3759,7 @@ fn test_sort_topk_with_dynamic_filter_roundtrip() -> Result<()> {
         .downcast_ref::<SortExec>()
         .expect("Should be SortExec");
     let deserialized_sort_df = deserialized_sort
-        .dynamic_filter()
+        .dynamic_filter_expr()
         .expect("SortExec should have a dynamic filter after roundtrip");
 
     // Extract the dynamic filter pushed down to the child ParquetSource.