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,22 +153,21 @@ impl FileSource for TestSource {
 
     fn try_pushdown_filters(
         &self,
-        filters: &[Arc<dyn PhysicalExpr>],
+        mut filters: Vec<Arc<dyn PhysicalExpr>>,
         config: &ConfigOptions,
     ) -> Result<FilterPushdownPropagation<Arc<dyn FileSource>>> {
-        let mut all_filters = filters.iter().map(Arc::clone).collect::<Vec<_>>();
         if self.support && config.execution.parquet.pushdown_filters {
             if let Some(internal) = self.predicate.as_ref() {
-                all_filters.push(Arc::clone(internal));
+                filters.push(Arc::clone(internal));
             }
             let new_node = Arc::new(TestSource {
                 support: true,
-                predicate: Some(conjunction(all_filters.clone())),
+                predicate: Some(conjunction(filters.clone())),
                 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(filters),
+                updated_node: Some(new_node),
             })
         } else {
             Ok(FilterPushdownPropagation::unsupported(filters))
@@ -196,7 +195,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 +219,7 @@ fn test_pushdown_into_scan_with_config_options() {
     insta::assert_snapshot!(
         OptimizationTest::new(
             Arc::clone(&plan),
-            PushdownFilter {},
+            FilterPushdown {},
             false
         ),
         @r"
@@ -239,7 +238,7 @@ fn test_pushdown_into_scan_with_config_options() {
     insta::assert_snapshot!(
         OptimizationTest::new(
             plan,
-            PushdownFilter {},
+            FilterPushdown {},
             true
         ),
         @r"
@@ -264,7 +263,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 +293,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 +317,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 +346,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 +410,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,28 @@ 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 {
-            Some(new_node) => {
-                let mut new_data_source = self.clone();
-                new_data_source.file_source = new_node;
+        match result.updated_node {
+            Some(new_file_source) => {
+                let file_scan_config = FileScanConfigBuilder::new(
+                    self.object_store_url.clone(),
+                    Arc::clone(&self.file_schema),
+                    new_file_source,
+                )
+                .build();
                 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(file_scan_config) 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,
             }),
         }
     }