wip

Author: adriangb

datafusion/datasource/src/file_scan_config.rs

@@ -41,10 +41,7 @@ use datafusion_physical_expr::{
     PhysicalSortExpr,
 };
 use datafusion_physical_plan::{
-    display::{display_orderings, ProjectSchemaDisplay},
-    metrics::ExecutionPlanMetricsSet,
-    projection::{all_alias_free_columns, new_projections_for_columns, ProjectionExec},
-    DisplayAs, DisplayFormatType, ExecutionPlan,
+    display::{display_orderings, ProjectSchemaDisplay}, dynamic_filters::DynamicFilterSource, metrics::ExecutionPlanMetricsSet, projection::{all_alias_free_columns, new_projections_for_columns, ProjectionExec}, DisplayAs, DisplayFormatType, ExecutionPlan
 };
 use log::{debug, warn};
 
@@ -316,9 +313,9 @@ impl DataSource for FileScanConfig {
 
     fn with_dynamic_filter(
         &self,
-        dynamic_filters: Vec<Arc<dyn datafusion_physical_expr::PhysicalExpr>>,
+        dynamic_filter: Arc<dyn DynamicFilterSource>,
     ) -> Arc<dyn DataSource> {
-        file_source = self.file_source.with_dynamic_filter(dynamic_filters);
+        let file_source = self.file_source.with_dynamic_filter(dynamic_filter);
         Arc::new(Self {
             file_source,
             ..self.clone()

datafusion/datasource/src/source.rs

@@ -22,6 +22,7 @@ use std::fmt;
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
+use datafusion_physical_plan::dynamic_filters::DynamicFilterSource;
 use datafusion_physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
 use datafusion_physical_plan::projection::ProjectionExec;
@@ -81,7 +82,7 @@ pub trait DataSource: Send + Sync + Debug {
     ) -> datafusion_common::Result<Option<Arc<dyn ExecutionPlan>>>;
     fn with_dynamic_filter(
         &self,
-        _dynamic_filters: Vec<Arc<dyn datafusion_physical_expr::PhysicalExpr>>,
+        _dynamic_filter: Arc<dyn DynamicFilterSource>,
     ) -> Arc<dyn DataSource>;
 }
 
@@ -229,9 +230,9 @@ impl DataSourceExec {
 
     pub fn with_dynamic_filters(
         mut self,
-        dynamic_filters: Vec<Arc<dyn datafusion_physical_expr::PhysicalExpr>>,
+        dynamic_filter: Arc<dyn DynamicFilterSource>,
     ) -> Self {
-        self.data_source = self.data_source.with_dynamic_filters(dynamic_filters);
+        self.data_source = self.data_source.with_dynamic_filter(dynamic_filter);
         self
     }