ExecutionPlan: add APIs for filter pushdown & optimizer rule to apply them

datafusion/core/tests/physical_optimizer/mod.rs

@@ -21,6 +21,7 @@ mod aggregate_statistics;
 mod combine_partial_final_agg;
 mod enforce_distribution;
 mod enforce_sorting;
+mod filter_pushdown;
 mod join_selection;
 mod limit_pushdown;
 mod limited_distinct_aggregation;

datafusion/datasource/src/file.rs

@@ -26,8 +26,10 @@ 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_physical_expr::LexOrdering;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::{Result, Statistics};
+use datafusion_physical_expr::{LexOrdering, PhysicalExprRef};
+use datafusion_physical_plan::filter_pushdown::FilterPushdownResult;
 use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion_physical_plan::DisplayFormatType;
 
@@ -57,7 +59,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
@@ -75,7 +77,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);
         }
@@ -93,4 +95,17 @@ pub trait FileSource: Send + Sync {
         }
         Ok(None)
     }
+    /// Try to push down filters into this FileSource.
+    /// See [`ExecutionPlan::try_pushdown_filters`] for more details.
+    ///
+    /// [`ExecutionPlan::try_pushdown_filters`]: datafusion_physical_plan::ExecutionPlan::try_pushdown_filters
+    fn try_pushdown_filters(
+        &self,
+        _filters: &[PhysicalExprRef],
+        _config: &ConfigOptions,
+    ) -> Result<FileSourceFilterPushdownResult> {
+        Ok(FileSourceFilterPushdownResult::NotPushed)
+    }
 }
+
+pub type FileSourceFilterPushdownResult = FilterPushdownResult<Arc<dyn FileSource>>;

datafusion/datasource/src/file_scan_config.rs

@@ -31,14 +31,16 @@ use arrow::{
     buffer::Buffer,
     datatypes::{ArrowNativeType, DataType, Field, Schema, SchemaRef, UInt16Type},
 };
-use datafusion_common::{exec_err, ColumnStatistics, Constraints, Result, Statistics};
+use datafusion_common::{
+    config::ConfigOptions, exec_err, ColumnStatistics, Constraints, Result, Statistics,
+};
 use datafusion_common::{DataFusionError, ScalarValue};
 use datafusion_execution::{
     object_store::ObjectStoreUrl, SendableRecordBatchStream, TaskContext,
 };
 use datafusion_physical_expr::{
     expressions::Column, EquivalenceProperties, LexOrdering, Partitioning,
-    PhysicalSortExpr,
+    PhysicalExprRef, PhysicalSortExpr,
 };
 use datafusion_physical_plan::{
     display::{display_orderings, ProjectSchemaDisplay},
@@ -48,16 +50,16 @@ use datafusion_physical_plan::{
 };
 use log::{debug, warn};
 
-use crate::file_groups::FileGroup;
 use crate::{
     display::FileGroupsDisplay,
-    file::FileSource,
+    file::{FileSource, FileSourceFilterPushdownResult},
     file_compression_type::FileCompressionType,
     file_stream::FileStream,
     source::{DataSource, DataSourceExec},
     statistics::MinMaxStatistics,
     PartitionedFile,
 };
+use crate::{file_groups::FileGroup, source::DataSourceFilterPushdownResult};
 
 /// The base configurations for a [`DataSourceExec`], the a physical plan for
 /// any given file format.
@@ -584,6 +586,29 @@ impl DataSource for FileScanConfig {
             ) as _
         }))
     }
+
+    fn try_pushdown_filters(
+        &self,
+        filters: &[PhysicalExprRef],
+        config: &ConfigOptions,
+    ) -> Result<DataSourceFilterPushdownResult> {
+        match self.file_source.try_pushdown_filters(filters, config)? {
+            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,
+                })
+            }
+        }
+    }
 }
 
 impl FileScanConfig {

datafusion/datasource/src/source.rs

@@ -26,14 +26,15 @@ 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;
 use datafusion_common::config::ConfigOptions;
-use datafusion_common::{Constraints, Statistics};
+use datafusion_common::{Constraints, Result, Statistics};
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
-use datafusion_physical_expr::{EquivalenceProperties, Partitioning};
+use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExprRef};
 use datafusion_physical_expr_common::sort_expr::LexOrdering;
 
 /// Common behaviors in Data Sources for both from Files and Memory.
@@ -51,7 +52,7 @@ pub trait DataSource: Send + Sync + Debug {
         &self,
         partition: usize,
         context: Arc<TaskContext>,
-    ) -> datafusion_common::Result<SendableRecordBatchStream>;
+    ) -> Result<SendableRecordBatchStream>;
     fn as_any(&self) -> &dyn Any;
     /// Format this source for display in explain plans
     fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> fmt::Result;
@@ -62,13 +63,13 @@ pub trait DataSource: Send + Sync + Debug {
         _target_partitions: usize,
         _repartition_file_min_size: usize,
         _output_ordering: Option<LexOrdering>,
-    ) -> datafusion_common::Result<Option<Arc<dyn DataSource>>> {
+    ) -> Result<Option<Arc<dyn DataSource>>> {
         Ok(None)
     }
 
     fn output_partitioning(&self) -> Partitioning;
     fn eq_properties(&self) -> EquivalenceProperties;
-    fn statistics(&self) -> datafusion_common::Result<Statistics>;
+    fn statistics(&self) -> Result<Statistics>;
     /// Return a copy of this DataSource with a new fetch limit
     fn with_fetch(&self, _limit: Option<usize>) -> Option<Arc<dyn DataSource>>;
     fn fetch(&self) -> Option<usize>;
@@ -78,9 +79,20 @@ pub trait DataSource: Send + Sync + Debug {
     fn try_swapping_with_projection(
         &self,
         _projection: &ProjectionExec,
-    ) -> datafusion_common::Result<Option<Arc<dyn ExecutionPlan>>>;
+    ) -> Result<Option<Arc<dyn ExecutionPlan>>>;
+    /// Try to push down filters into this DataSource.
+    /// See [`ExecutionPlan::try_pushdown_filters`] for more details.
+    fn try_pushdown_filters(
+        &self,
+        _filters: &[PhysicalExprRef],
+        _config: &ConfigOptions,
+    ) -> Result<DataSourceFilterPushdownResult> {
+        Ok(DataSourceFilterPushdownResult::NotPushed)
+    }
 }
 
+pub type DataSourceFilterPushdownResult = FilterPushdownResult<Arc<dyn DataSource>>;
+
 /// [`ExecutionPlan`] handles different file formats like JSON, CSV, AVRO, ARROW, PARQUET
 ///
 /// `DataSourceExec` implements common functionality such as applying projections,
@@ -131,15 +143,15 @@ impl ExecutionPlan for DataSourceExec {
     fn with_new_children(
         self: Arc<Self>,
         _: Vec<Arc<dyn ExecutionPlan>>,
-    ) -> datafusion_common::Result<Arc<dyn ExecutionPlan>> {
+    ) -> Result<Arc<dyn ExecutionPlan>> {
         Ok(self)
     }
 
     fn repartitioned(
         &self,
         target_partitions: usize,
         config: &ConfigOptions,
-    ) -> datafusion_common::Result<Option<Arc<dyn ExecutionPlan>>> {
+    ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
         let data_source = self.data_source.repartitioned(
             target_partitions,
             config.optimizer.repartition_file_min_size,
@@ -163,15 +175,15 @@ impl ExecutionPlan for DataSourceExec {
         &self,
         partition: usize,
         context: Arc<TaskContext>,
-    ) -> datafusion_common::Result<SendableRecordBatchStream> {
+    ) -> Result<SendableRecordBatchStream> {
         self.data_source.open(partition, context)
     }
 
     fn metrics(&self) -> Option<MetricsSet> {
         Some(self.data_source.metrics().clone_inner())
     }
 
-    fn statistics(&self) -> datafusion_common::Result<Statistics> {
+    fn statistics(&self) -> Result<Statistics> {
         self.data_source.statistics()
     }
 
@@ -189,9 +201,32 @@ impl ExecutionPlan for DataSourceExec {
     fn try_swapping_with_projection(
         &self,
         projection: &ProjectionExec,
-    ) -> datafusion_common::Result<Option<Arc<dyn ExecutionPlan>>> {
+    ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
         self.data_source.try_swapping_with_projection(projection)
     }
+
+    fn try_pushdown_filters(
+        &self,
+        _plan: &Arc<dyn ExecutionPlan>,
+        parent_filters: &[PhysicalExprRef],
+        config: &ConfigOptions,
+    ) -> Result<ExecutionPlanFilterPushdownResult> {
+        match self
+            .data_source
+            .try_pushdown_filters(parent_filters, config)?
+        {
+            DataSourceFilterPushdownResult::NotPushed => {
+                Ok(ExecutionPlanFilterPushdownResult::NotPushed)
+            }
+            DataSourceFilterPushdownResult::Pushed { inner, support } => {
+                let new_self = Arc::new(DataSourceExec::new(inner));
+                Ok(ExecutionPlanFilterPushdownResult::Pushed {
+                    inner: new_self,
+                    support,
+                })
+            }
+        }
+    }
 }
 
 impl DataSourceExec {
@@ -254,3 +289,13 @@ impl DataSourceExec {
             })
     }
 }
+
+/// Create a new `DataSourceExec` from a `DataSource`
+impl<S> From<S> for DataSourceExec
+where
+    S: DataSource + 'static,
+{
+    fn from(source: S) -> Self {
+        Self::new(Arc::new(source))
+    }
+}

datafusion/physical-expr/src/lib.rs

@@ -68,7 +68,7 @@ pub use planner::{create_physical_expr, create_physical_exprs};
 pub use scalar_function::ScalarFunctionExpr;
 
 pub use datafusion_physical_expr_common::utils::reverse_order_bys;
-pub use utils::split_conjunction;
+pub use utils::{conjunction, conjunction_opt, split_conjunction};
 
 // For backwards compatibility
 pub mod tree_node {

datafusion/physical-expr/src/utils/mod.rs

@@ -47,6 +47,31 @@ pub fn split_conjunction(
     split_impl(Operator::And, predicate, vec![])
 }
 
+/// Create a conjunction of the given predicates.
+/// If the input is empty, return a literal true.
+/// If the input contains a single predicate, return the predicate.
+/// Otherwise, return a conjunction of the predicates (e.g. `a AND b AND c`).
+pub fn conjunction(
+    predicates: impl IntoIterator<Item = Arc<dyn PhysicalExpr>>,
+) -> Arc<dyn PhysicalExpr> {
+    conjunction_opt(predicates).unwrap_or_else(|| crate::expressions::lit(true))
+}
+
+/// Create a conjunction of the given predicates.
+/// If the input is empty or the return None.
+/// If the input contains a single predicate, return Some(predicate).
+/// Otherwise, return a Some(..) of a conjunction of the predicates (e.g. `Some(a AND b AND c)`).
+pub fn conjunction_opt(
+    predicates: impl IntoIterator<Item = Arc<dyn PhysicalExpr>>,
+) -> Option<Arc<dyn PhysicalExpr>> {
+    predicates
+        .into_iter()
+        .fold(None, |acc, predicate| match acc {
+            None => Some(predicate),
+            Some(acc) => Some(Arc::new(BinaryExpr::new(acc, Operator::And, predicate))),
+        })
+}
+
 /// Assume the predicate is in the form of DNF, split the predicate to a Vec of PhysicalExprs.
 ///
 /// For example, split "a1 = a2 OR b1 <= b2 OR c1 != c2" into ["a1 = a2", "b1 <= b2", "c1 != c2"]

datafusion/physical-optimizer/src/filter_pushdown.rs

@@ -0,0 +1,72 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use datafusion_common::{config::ConfigOptions, DataFusionError, Result};
+use datafusion_physical_plan::{
+    execution_plan::ExecutionPlanFilterPushdownResult, ExecutionPlan,
+};
+
+use crate::PhysicalOptimizerRule;
+
+/// A physical optimizer rule that pushes down filters in the execution plan.
+/// See [`ExecutionPlan::try_pushdown_filters`] for a detailed description of the algorithm.
+#[derive(Debug)]
+pub struct PushdownFilter {}
+
+impl Default for PushdownFilter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl PushdownFilter {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl PhysicalOptimizerRule for PushdownFilter {
+    fn optimize(
+        &self,
+        plan: Arc<dyn ExecutionPlan>,
+        config: &ConfigOptions,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        match plan.try_pushdown_filters(&plan, &Vec::new(), config)? {
+            ExecutionPlanFilterPushdownResult::NotPushed => Ok(plan),
+            ExecutionPlanFilterPushdownResult::Pushed { inner, support } => {
+                if !support.is_empty() {
+                    return Err(
+                        DataFusionError::Plan(
+                            format!("PushdownFilter: plan returned support length does not match filters length: {} != 0", support.len()
+                        ))
+                    );
+                }
+                Ok(inner)
+            }
+        }
+    }
+
+    fn name(&self) -> &str {
+        "PushdownFilter"
+    }
+
+    fn schema_check(&self) -> bool {
+        true // Filter pushdown does not change the schema of the plan
+    }
+}

datafusion/physical-optimizer/src/lib.rs

@@ -29,6 +29,7 @@ pub mod coalesce_batches;
 pub mod combine_partial_final_agg;
 pub mod enforce_distribution;
 pub mod enforce_sorting;
+pub mod filter_pushdown;
 pub mod join_selection;
 pub mod limit_pushdown;
 pub mod limited_distinct_aggregation;