feat: allow pushdown of dynamic filters with expr having partition cols

Author: feniljain

datafusion/datasource-parquet/src/file_format.rs

@@ -463,7 +463,8 @@ impl FileFormat for ParquetFormat {
             metadata_size_hint = Some(metadata);
         }
 
-        let mut source = ParquetSource::new(self.options.clone());
+        let mut source = ParquetSource::new(self.options.clone())
+            .with_table_partition_cols(conf.table_partition_cols.clone());
 
         // Use the CachedParquetFileReaderFactory
         let metadata_cache = state.runtime_env().cache_manager.get_file_metadata_cache();

datafusion/datasource-parquet/src/source.rs

@@ -26,6 +26,8 @@ use crate::opener::ParquetOpener;
 use crate::row_filter::can_expr_be_pushed_down_with_schemas;
 use crate::DefaultParquetFileReaderFactory;
 use crate::ParquetFileReaderFactory;
+use arrow::datatypes::Field;
+use arrow::datatypes::Schema;
 use datafusion_common::config::ConfigOptions;
 #[cfg(feature = "parquet_encryption")]
 use datafusion_common::config::EncryptionFactoryOptions;
@@ -43,6 +45,7 @@ use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_physical_expr::conjunction;
 use datafusion_physical_expr_adapter::DefaultPhysicalExprAdapterFactory;
 use datafusion_physical_expr_common::physical_expr::fmt_sql;
+use datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use datafusion_physical_plan::filter_pushdown::PushedDown;
 use datafusion_physical_plan::filter_pushdown::{
@@ -287,6 +290,7 @@ pub struct ParquetSource {
     /// Optional hint for the size of the parquet metadata
     pub(crate) metadata_size_hint: Option<usize>,
     pub(crate) projected_statistics: Option<Statistics>,
+    pub(crate) table_partition_cols: Vec<Arc<Field>>,
     #[cfg(feature = "parquet_encryption")]
     pub(crate) encryption_factory: Option<Arc<dyn EncryptionFactory>>,
 }
@@ -330,6 +334,11 @@ impl ParquetSource {
         self
     }
 
+    pub fn with_table_partition_cols(mut self, partition_cols: Vec<Arc<Field>>) -> Self {
+        self.table_partition_cols = partition_cols;
+        self
+    }
+
     /// Options passed to the parquet reader for this scan
     pub fn table_parquet_options(&self) -> &TableParquetOptions {
         &self.table_parquet_options
@@ -720,7 +729,9 @@ impl FileSource for ParquetSource {
         let filters: Vec<PushedDownPredicate> = filters
             .into_iter()
             .map(|filter| {
-                if can_expr_be_pushed_down_with_schemas(&filter, &file_schema) {
+                if is_dynamic_physical_expr(&filter) {
+                    PushedDownPredicate::supported(filter)
+                } else if can_expr_be_pushed_down_with_schemas(&filter, &file_schema) {
                     PushedDownPredicate::supported(filter)
                 } else {
                     PushedDownPredicate::unsupported(filter)