FIXUP - only push down partial filters into parquet exec

Signed-off-by: Adrian Tanase <[email protected]>

Author: aditanase

crates/core/src/delta_datafusion/mod.rs

@@ -60,7 +60,7 @@ use datafusion_common::{
 use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_expr::logical_plan::CreateExternalTable;
 use datafusion_expr::simplify::SimplifyContext;
-use datafusion_expr::utils::conjunction;
+use datafusion_expr::utils::{conjunction, split_conjunction};
 use datafusion_expr::{
     col, BinaryExpr, Expr, Extension, LogicalPlan, Operator, TableProviderFilterPushDown,
     Volatility,
@@ -565,6 +565,8 @@ impl<'a> DeltaScanBuilder<'a> {
         let context = SessionContext::new();
         let df_schema = logical_schema.clone().to_dfschema()?;
 
+        // TODO: extract to method and untangle pushdown filter from the full logical filter
+        let mut pushdown_filter = None;
         let logical_filter = self.filter.map(|expr| {
             // Simplify the expression first
             let props = ExecutionProps::new();
@@ -573,6 +575,27 @@ impl<'a> DeltaScanBuilder<'a> {
             let simplifier = ExprSimplifier::new(simplify_context).with_max_cycles(10);
             let simplified = simplifier.simplify(expr).unwrap();
 
+            // FIXME: we probably want to do this before simplification
+            let predicates = split_conjunction(&simplified);
+            let pushdown_filters =
+                get_pushdown_filters(&predicates, self.snapshot.metadata().partition_columns.clone());
+
+            let filtered_predicates = predicates
+                .into_iter()
+                .zip(pushdown_filters.into_iter())
+                .filter_map(|(filter, pushdown)| {
+                    if pushdown == TableProviderFilterPushDown::Inexact {
+                        Some(filter.clone())
+                    } else {
+                        None
+                    }
+                });
+            let pushdown_filter = conjunction(filtered_predicates).map(|e| {
+                context
+                    .create_physical_expr(e, &df_schema)
+                    .unwrap()
+            });
+
             context
                 .create_physical_expr(simplified, &df_schema)
                 .unwrap()
@@ -728,7 +751,7 @@ impl<'a> DeltaScanBuilder<'a> {
         // Sometimes (i.e Merge) we want to prune files that don't make the
         // filter and read the entire contents for files that do match the
         // filter
-        if let Some(predicate) = logical_filter {
+        if let Some(predicate) = pushdown_filter {
             if config.enable_parquet_pushdown {
                 file_source = file_source.with_predicate(Arc::clone(&file_schema), predicate);
             }