review suggestions

Author: berkaysynnada

datafusion/core/tests/physical_optimizer/push_down_filter.rs

@@ -44,10 +44,10 @@ use datafusion_physical_expr::{
     aggregate::AggregateExprBuilder, conjunction, Partitioning,
 };
 use datafusion_physical_expr_common::physical_expr::fmt_sql;
-use datafusion_physical_optimizer::push_down_filter::PushdownFilter;
+use datafusion_physical_optimizer::filter_pushdown::FilterPushdown;
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
 use datafusion_physical_plan::filter_pushdown::{
-    FilterPushdownPropagation, FilterPushdowns,
+    FilterPushdownPropagation, PredicateSupports,
 };
 use datafusion_physical_plan::{
     aggregates::{AggregateExec, AggregateMode, PhysicalGroupBy},
@@ -153,7 +153,7 @@ impl FileSource for TestSource {
 
     fn try_pushdown_filters(
         &self,
-        filters: &[Arc<dyn PhysicalExpr>],
+        filters: Vec<Arc<dyn PhysicalExpr>>,
         config: &ConfigOptions,
     ) -> Result<FilterPushdownPropagation<Arc<dyn FileSource>>> {
         let mut all_filters = filters.iter().map(Arc::clone).collect::<Vec<_>>();
@@ -167,8 +167,8 @@ impl FileSource for TestSource {
                 statistics: self.statistics.clone(), // should be updated in reality
             });
             Ok(FilterPushdownPropagation {
-                parent_filter_result: FilterPushdowns::all_supported(&all_filters),
-                new_node: Some(new_node),
+                filters: PredicateSupports::all_supported(all_filters),
+                updated_node: Some(new_node),
             })
         } else {
             Ok(FilterPushdownPropagation::unsupported(filters))
@@ -196,7 +196,7 @@ fn test_pushdown_into_scan() {
 
     // expect the predicate to be pushed down into the DataSource
     insta::assert_snapshot!(
-        OptimizationTest::new(plan, PushdownFilter{}, true),
+        OptimizationTest::new(plan, FilterPushdown{}, true),
         @r"
     OptimizationTest:
       input:
@@ -220,7 +220,7 @@ fn test_pushdown_into_scan_with_config_options() {
     insta::assert_snapshot!(
         OptimizationTest::new(
             Arc::clone(&plan),
-            PushdownFilter {},
+            FilterPushdown {},
             false
         ),
         @r"
@@ -239,7 +239,7 @@ fn test_pushdown_into_scan_with_config_options() {
     insta::assert_snapshot!(
         OptimizationTest::new(
             plan,
-            PushdownFilter {},
+            FilterPushdown {},
             true
         ),
         @r"
@@ -264,7 +264,7 @@ fn test_filter_collapse() {
     let plan = Arc::new(FilterExec::try_new(predicate2, filter1).unwrap());
 
     insta::assert_snapshot!(
-        OptimizationTest::new(plan, PushdownFilter{}, true),
+        OptimizationTest::new(plan, FilterPushdown{}, true),
         @r"
     OptimizationTest:
       input:
@@ -294,7 +294,7 @@ fn test_filter_with_projection() {
 
     // expect the predicate to be pushed down into the DataSource but the FilterExec to be converted to ProjectionExec
     insta::assert_snapshot!(
-        OptimizationTest::new(plan, PushdownFilter{}, true),
+        OptimizationTest::new(plan, FilterPushdown{}, true),
         @r"
     OptimizationTest:
       input:
@@ -318,7 +318,7 @@ fn test_filter_with_projection() {
             .unwrap(),
     );
     insta::assert_snapshot!(
-        OptimizationTest::new(plan, PushdownFilter{},true),
+        OptimizationTest::new(plan, FilterPushdown{},true),
         @r"
     OptimizationTest:
       input:
@@ -347,7 +347,7 @@ fn test_push_down_through_transparent_nodes() {
 
     // expect the predicate to be pushed down into the DataSource
     insta::assert_snapshot!(
-        OptimizationTest::new(plan, PushdownFilter{},true),
+        OptimizationTest::new(plan, FilterPushdown{},true),
         @r"
     OptimizationTest:
       input:
@@ -411,7 +411,7 @@ fn test_no_pushdown_through_aggregates() {
 
     // expect the predicate to be pushed down into the DataSource
     insta::assert_snapshot!(
-        OptimizationTest::new(plan, PushdownFilter{}, true),
+        OptimizationTest::new(plan, FilterPushdown{}, true),
         @r"
     OptimizationTest:
       input:

datafusion/datasource/src/file.rs

@@ -102,7 +102,7 @@ pub trait FileSource: Send + Sync {
     /// [`ExecutionPlan::handle_child_pushdown_result`]: datafusion_physical_plan::ExecutionPlan::handle_child_pushdown_result
     fn try_pushdown_filters(
         &self,
-        filters: &[Arc<dyn PhysicalExpr>],
+        filters: Vec<Arc<dyn PhysicalExpr>>,
         _config: &ConfigOptions,
     ) -> Result<FilterPushdownPropagation<Arc<dyn FileSource>>> {
         Ok(FilterPushdownPropagation::unsupported(filters))

datafusion/datasource/src/file_scan_config.rs

@@ -594,24 +594,24 @@ impl DataSource for FileScanConfig {
 
     fn try_pushdown_filters(
         &self,
-        filters: &[Arc<dyn PhysicalExpr>],
+        filters: Vec<Arc<dyn PhysicalExpr>>,
         config: &ConfigOptions,
     ) -> Result<FilterPushdownPropagation<Arc<dyn DataSource>>> {
         let result = self.file_source.try_pushdown_filters(filters, config)?;
-        match result.new_node {
+        match result.updated_node {
             Some(new_node) => {
                 let mut new_data_source = self.clone();
                 new_data_source.file_source = new_node;
                 Ok(FilterPushdownPropagation {
-                    parent_filter_result: result.parent_filter_result,
-                    new_node: Some(Arc::new(new_data_source) as _),
+                    filters: result.filters,
+                    updated_node: Some(Arc::new(new_data_source) as _),
                 })
             }
             None => {
                 // If the file source does not support filter pushdown, return the original config
                 Ok(FilterPushdownPropagation {
-                    parent_filter_result: result.parent_filter_result,
-                    new_node: None,
+                    filters: result.filters,
+                    updated_node: None,
                 })
             }
         }

datafusion/datasource/src/source.rs

@@ -88,7 +88,7 @@ pub trait DataSource: Send + Sync + Debug {
     /// [`ExecutionPlan::handle_child_pushdown_result`]: datafusion_physical_plan::ExecutionPlan::handle_child_pushdown_result
     fn try_pushdown_filters(
         &self,
-        filters: &[Arc<dyn PhysicalExpr>],
+        filters: Vec<Arc<dyn PhysicalExpr>>,
         _config: &ConfigOptions,
     ) -> Result<FilterPushdownPropagation<Arc<dyn DataSource>>> {
         Ok(FilterPushdownPropagation::unsupported(filters))
@@ -214,23 +214,23 @@ impl ExecutionPlan for DataSourceExec {
     ) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
         // Push any remaining filters into our data source
         let res = self.data_source.try_pushdown_filters(
-            &child_pushdown_result.parent_filters.into_inner_filters(),
+            child_pushdown_result.parent_filters.collect_all(),
             config,
         )?;
-        match res.new_node {
+        match res.updated_node {
             Some(data_source) => {
                 let mut new_node = self.clone();
                 new_node.data_source = data_source;
                 new_node.cache =
                     Self::compute_properties(Arc::clone(&new_node.data_source));
                 Ok(FilterPushdownPropagation {
-                    parent_filter_result: res.parent_filter_result,
-                    new_node: Some(Arc::new(new_node)),
+                    filters: res.filters,
+                    updated_node: Some(Arc::new(new_node)),
                 })
             }
             None => Ok(FilterPushdownPropagation {
-                parent_filter_result: res.parent_filter_result,
-                new_node: None,
+                filters: res.filters,
+                updated_node: None,
             }),
         }
     }