simplify according to pr feedback

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -25,8 +25,8 @@ use datafusion::{
     },
     scalar::ScalarValue,
 };
-use datafusion_common::internal_err;
 use datafusion_common::{config::ConfigOptions, Statistics};
+use datafusion_common::{internal_err, Result};
 use datafusion_datasource::file_scan_config::FileScanConfigBuilder;
 use datafusion_datasource::source::DataSourceExec;
 use datafusion_datasource::{
@@ -40,8 +40,8 @@ use datafusion_physical_optimizer::filter_pushdown::PushdownFilter;
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
 use datafusion_physical_plan::filter::FilterExec;
 use datafusion_physical_plan::{
-    displayable, execution_plan::FilterSupport, metrics::ExecutionPlanMetricsSet,
-    DisplayFormatType, ExecutionPlan,
+    displayable, filter_pushdown::FilterPushdownSupport,
+    metrics::ExecutionPlanMetricsSet, DisplayFormatType, ExecutionPlan,
 };
 use object_store::ObjectStore;
 use std::sync::{Arc, OnceLock};
@@ -53,13 +53,13 @@ use std::{
 /// A placeholder data source that accepts filter pushdown
 #[derive(Clone)]
 struct TestSource {
-    support: FilterSupport,
+    support: FilterPushdownSupport,
     predicate: Option<PhysicalExprRef>,
     statistics: Option<Statistics>,
 }
 
 impl TestSource {
-    fn new(support: FilterSupport) -> Self {
+    fn new(support: FilterPushdownSupport) -> Self {
         Self {
             support,
             predicate: None,
@@ -105,7 +105,7 @@ impl FileSource for TestSource {
         todo!("should not be called")
     }
 
-    fn statistics(&self) -> datafusion_common::Result<Statistics> {
+    fn statistics(&self) -> Result<Statistics> {
         Ok(self
             .statistics
             .as_ref()
@@ -137,23 +137,23 @@ impl FileSource for TestSource {
         }
     }
 
-    fn push_down_filters(
+    fn try_pushdown_filters(
         &self,
         filters: &[PhysicalExprRef],
-    ) -> datafusion_common::Result<Option<FileSourceFilterPushdownResult>> {
+    ) -> Result<FileSourceFilterPushdownResult> {
         let new = Arc::new(TestSource {
             support: self.support,
             predicate: Some(conjunction(filters.iter().map(Arc::clone))),
             statistics: self.statistics.clone(),
         });
-        Ok(Some(FileSourceFilterPushdownResult::new(
+        Ok(FileSourceFilterPushdownResult::new(
             new,
             vec![self.support; filters.len()],
-        )))
+        ))
     }
 }
 
-fn test_scan(support: FilterSupport) -> Arc<dyn ExecutionPlan> {
+fn test_scan(support: FilterPushdownSupport) -> Arc<dyn ExecutionPlan> {
     let schema = schema();
     let source = Arc::new(TestSource::new(support));
     let base_config = FileScanConfigBuilder::new(
@@ -167,7 +167,7 @@ fn test_scan(support: FilterSupport) -> Arc<dyn ExecutionPlan> {
 
 #[test]
 fn test_pushdown_into_scan() {
-    let scan = test_scan(FilterSupport::HandledExact);
+    let scan = test_scan(FilterPushdownSupport::Exact);
     let predicate = col_lit_predicate("a", "foo", schema());
     let plan = Arc::new(FilterExec::try_new(predicate, scan).unwrap());
 
@@ -187,9 +187,9 @@ fn test_pushdown_into_scan() {
 }
 
 #[test]
-test_filter_collapse() {
+fn test_filter_collapse() {
     // filter should be pushed down into the parquet scan with two filters
-    let scan = test_scan(FilterSupport::HandledExact);
+    let scan = test_scan(FilterPushdownSupport::Exact);
     let predicate1 = col_lit_predicate("a", "foo", schema());
     let filter1 = Arc::new(FilterExec::try_new(predicate1, scan).unwrap());
     let predicate2 = col_lit_predicate("b", "bar", schema());
@@ -205,7 +205,7 @@ test_filter_collapse() {
         -     DataSourceExec: file_groups={0 groups: []}, projection=[a, b, c], file_type=test
       output:
         Ok:
-          - DataSourceExec: file_groups={0 groups: []}, projection=[a, b, c], file_type=test, predicate=a@0 = foo AND b@1 = bar
+          - DataSourceExec: file_groups={0 groups: []}, projection=[a, b, c], file_type=test, predicate=b@1 = bar AND a@0 = foo
     "
     );
 }

datafusion/datasource/src/file.rs

@@ -26,9 +26,9 @@ use crate::file_groups::FileGroupPartitioner;
 use crate::file_scan_config::FileScanConfig;
 use crate::file_stream::FileOpener;
 use arrow::datatypes::SchemaRef;
-use datafusion_common::Statistics;
+use datafusion_common::{Result, Statistics};
 use datafusion_physical_expr::{LexOrdering, PhysicalExprRef};
-use datafusion_physical_plan::execution_plan::FilterPushdownResult;
+use datafusion_physical_plan::filter_pushdown::FilterPushdownResult;
 use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion_physical_plan::DisplayFormatType;
 
@@ -58,7 +58,7 @@ pub trait FileSource: Send + Sync {
     /// Return execution plan metrics
     fn metrics(&self) -> &ExecutionPlanMetricsSet;
     /// Return projected statistics
-    fn statistics(&self) -> datafusion_common::Result<Statistics>;
+    fn statistics(&self) -> Result<Statistics>;
     /// String representation of file source such as "csv", "json", "parquet"
     fn file_type(&self) -> &str;
     /// Format FileType specific information
@@ -76,7 +76,7 @@ pub trait FileSource: Send + Sync {
         repartition_file_min_size: usize,
         output_ordering: Option<LexOrdering>,
         config: &FileScanConfig,
-    ) -> datafusion_common::Result<Option<FileScanConfig>> {
+    ) -> Result<Option<FileScanConfig>> {
         if config.file_compression_type.is_compressed() || config.new_lines_in_values {
             return Ok(None);
         }
@@ -95,17 +95,11 @@ pub trait FileSource: Send + Sync {
         Ok(None)
     }
 
-    /// Push down filters to the file source if supported.
-    ///
-    /// Returns `Ok(None)` by default. See [`ExecutionPlan::with_filter_pushdown_result`]
-    /// for more details.
-    ///
-    /// [`ExecutionPlan::with_filter_pushdown_result`]: datafusion_physical_plan::execution_plan::ExecutionPlan::with_filter_pushdown_result
-    fn push_down_filters(
+    fn try_pushdown_filters(
         &self,
         _filters: &[PhysicalExprRef],
-    ) -> datafusion_common::Result<Option<FileSourceFilterPushdownResult>> {
-        Ok(None)
+    ) -> Result<FileSourceFilterPushdownResult> {
+        Ok(FileSourceFilterPushdownResult::NotPushed)
     }
 }
 

datafusion/datasource/src/file_scan_config.rs

@@ -50,7 +50,7 @@ use log::{debug, warn};
 
 use crate::{
     display::FileGroupsDisplay,
-    file::FileSource,
+    file::{FileSource, FileSourceFilterPushdownResult},
     file_compression_type::FileCompressionType,
     file_stream::FileStream,
     source::{DataSource, DataSourceExec},
@@ -585,19 +585,25 @@ impl DataSource for FileScanConfig {
         }))
     }
 
-    fn push_down_filters(
+    fn try_pushdown_filters(
         &self,
         filters: &[PhysicalExprRef],
-    ) -> Result<Option<DataSourceFilterPushdownResult>> {
-        if let Some(file_source_result) = self.file_source.push_down_filters(filters)? {
-            let mut new_self = self.clone();
-            new_self.file_source = file_source_result.inner;
-            Ok(Some(DataSourceFilterPushdownResult {
-                inner: Arc::new(new_self) as Arc<dyn DataSource>,
-                support: file_source_result.support,
-            }))
-        } else {
-            Ok(None)
+    ) -> Result<DataSourceFilterPushdownResult> {
+        match self.file_source.try_pushdown_filters(filters)? {
+            FileSourceFilterPushdownResult::NotPushed => {
+                Ok(DataSourceFilterPushdownResult::NotPushed)
+            }
+            FileSourceFilterPushdownResult::Pushed { inner, support } => {
+                let new_self = Arc::new(
+                    FileScanConfigBuilder::from(self.clone())
+                        .with_source(inner)
+                        .build(),
+                );
+                Ok(DataSourceFilterPushdownResult::Pushed {
+                    inner: new_self,
+                    support,
+                })
+            }
         }
     }
 }

datafusion/datasource/src/source.rs

@@ -22,14 +22,12 @@ use std::fmt;
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
-use datafusion_physical_plan::execution_plan::{
-    Boundedness, EmissionType, ExecutionPlanFilterPushdownResult, FilterPushdownResult,
-    FilterSupport,
-};
+use datafusion_physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
 use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::{
-    DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
+    DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanFilterPushdownResult,
+    FilterPushdownResult, PlanProperties,
 };
 
 use crate::file_scan_config::FileScanConfig;
@@ -83,24 +81,11 @@ pub trait DataSource: Send + Sync + Debug {
         _projection: &ProjectionExec,
     ) -> Result<Option<Arc<dyn ExecutionPlan>>>;
 
-    /// Push down filters into this `DataSource`.
-    ///
-    /// Returns `Ok(None)` if the filters cannot be evaluated within the
-    /// `DataSource`.
-    ///
-    /// If the filters can be evaluated by the `DataSource`,
-    /// return a [`FilterPushdownResult`] containing an updated
-    /// `DataSource` and the support level for each filter (exact or inexact).
-    ///
-    /// Default implementation returns `Ok(None)`. See [`ExecutionPlan::with_filter_pushdown_result`]
-    /// for more details.
-    ///
-    /// [`ExecutionPlan::push_down_filters`]: datafusion_physical_plan::execution_plan::ExecutionPlan::with_filter_pushdown_result
-    fn push_down_filters(
+    fn try_pushdown_filters(
         &self,
         _filters: &[PhysicalExprRef],
-    ) -> Result<Option<DataSourceFilterPushdownResult>>> {
-        Ok(None)
+    ) -> Result<DataSourceFilterPushdownResult> {
+        Ok(DataSourceFilterPushdownResult::NotPushed)
     }
 }
 
@@ -218,25 +203,22 @@ impl ExecutionPlan for DataSourceExec {
         self.data_source.try_swapping_with_projection(projection)
     }
 
-    fn with_filter_pushdown_result(
-        self: Arc<Self>,
-        own_filters_result: &[FilterSupport],
-        parent_filters_remaining: &[PhysicalExprRef],
-    ) -> Result<Option<ExecutionPlanFilterPushdownResult>> {
-        // We didn't give out any filters, this should be empty!
-        assert!(own_filters_result.is_empty());
-        // Forward filter pushdown to our data source.
-        if let Some(pushdown_result) = self
-            .data_source
-            .push_down_filters(parent_filters_remaining)?
-        {
-            let new_self = Arc::new(DataSourceExec::new(pushdown_result.inner));
-            Ok(Some(ExecutionPlanFilterPushdownResult::new(
-                new_self,
-                pushdown_result.support,
-            )))
-        } else {
-            Ok(None)
+    fn try_pushdown_filters(
+        &self,
+        _plan: &Arc<dyn ExecutionPlan>,
+        parent_filters: &[PhysicalExprRef],
+    ) -> Result<ExecutionPlanFilterPushdownResult> {
+        match self.data_source.try_pushdown_filters(parent_filters)? {
+            DataSourceFilterPushdownResult::NotPushed => {
+                Ok(ExecutionPlanFilterPushdownResult::NotPushed)
+            }
+            DataSourceFilterPushdownResult::Pushed { inner, support } => {
+                let new_self = Arc::new(DataSourceExec::new(inner));
+                Ok(ExecutionPlanFilterPushdownResult::Pushed {
+                    inner: new_self,
+                    support,
+                })
+            }
         }
     }
 }