fixup pushdown filter

Author: aditanase

crates/core/src/delta_datafusion/mod.rs

@@ -41,15 +41,14 @@ use datafusion::catalog::{Session, TableProviderFactory};
 use datafusion::config::TableParquetOptions;
 use datafusion::datasource::memory::DataSourceExec;
 use datafusion::datasource::physical_plan::{
-    wrap_partition_type_in_dict, wrap_partition_value_in_dict, FileScanConfig,
-    ParquetSource,
+    wrap_partition_type_in_dict, wrap_partition_value_in_dict, FileScanConfig, ParquetSource,
 };
 use datafusion::datasource::{listing::PartitionedFile, MemTable, TableProvider, TableType};
 use datafusion::execution::context::{SessionConfig, SessionContext, SessionState, TaskContext};
 use datafusion::execution::runtime_env::RuntimeEnv;
 use datafusion::execution::FunctionRegistry;
 use datafusion::optimizer::simplify_expressions::ExprSimplifier;
-use datafusion::physical_optimizer::pruning::PruningPredicate;
+use datafusion::physical_optimizer::pruning::{PruningPredicate, PruningStatistics};
 use datafusion_common::scalar::ScalarValue;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion, TreeNodeVisitor};
 use datafusion_common::{
@@ -60,7 +59,9 @@ 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::{col, Expr, Extension, LogicalPlan, TableProviderFilterPushDown, Volatility};
+use datafusion_expr::{
+    col, BinaryExpr, Expr, Extension, LogicalPlan, TableProviderFilterPushDown, Volatility,
+};
 use datafusion_physical_expr::{create_physical_expr, PhysicalExpr};
 use datafusion_physical_plan::filter::FilterExec;
 use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
@@ -571,18 +572,17 @@ impl<'a> DeltaScanBuilder<'a> {
         };
 
         let context = SessionContext::new();
-        let df_schema = logical_schema.clone().to_dfschema()?;
 
         let logical_filter = self.filter.map(|expr| {
             // Simplify the expression first
             let props = ExecutionProps::new();
             let simplify_context =
-                SimplifyContext::new(&props).with_schema(df_schema.clone().into());
+                SimplifyContext::new(&props).with_schema(filter_df_schema.clone().into());
             let simplifier = ExprSimplifier::new(simplify_context).with_max_cycles(10);
             let simplified = simplifier.simplify(expr).unwrap();
 
             context
-                .create_physical_expr(simplified, &df_schema)
+                .create_physical_expr(simplified, &filter_df_schema)
                 .unwrap()
         });