think this does it?

Signed-off-by: Andrew Duffy <[email protected]>

Author: a10y

vortex-datafusion/src/convert/exprs.rs

@@ -9,17 +9,22 @@ use datafusion_expr::Operator as DFOperator;
 use datafusion_functions::core::getfield::GetFieldFunc;
 use datafusion_physical_expr::PhysicalExpr;
 use datafusion_physical_expr::ScalarFunctionExpr;
-use datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr;
 use datafusion_physical_expr_common::physical_expr::PhysicalExprRef;
+use datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr;
 use datafusion_physical_plan::expressions as df_expr;
 use itertools::Itertools;
 use vortex::compute::LikeOptions;
-use vortex::dtype::arrow::FromArrowType;
 use vortex::dtype::DType;
 use vortex::dtype::Nullability;
+use vortex::dtype::arrow::FromArrowType;
+use vortex::error::VortexResult;
 use vortex::error::vortex_bail;
 use vortex::error::vortex_err;
-use vortex::error::VortexResult;
+use vortex::expr::Binary;
+use vortex::expr::Expression;
+use vortex::expr::Like;
+use vortex::expr::Operator;
+use vortex::expr::VTableExt;
 use vortex::expr::and;
 use vortex::expr::cast;
 use vortex::expr::get_item;
@@ -28,11 +33,6 @@ use vortex::expr::list_contains;
 use vortex::expr::lit;
 use vortex::expr::not;
 use vortex::expr::root;
-use vortex::expr::Binary;
-use vortex::expr::Expression;
-use vortex::expr::Like;
-use vortex::expr::Operator;
-use vortex::expr::VTableExt;
 use vortex::scalar::Scalar;
 
 use crate::convert::FromDataFusion;
@@ -301,8 +301,8 @@ mod tests {
     use arrow_schema::Schema;
     use arrow_schema::TimeUnit as ArrowTimeUnit;
     use datafusion::functions::core::getfield::GetFieldFunc;
-    use datafusion_common::config::ConfigOptions;
     use datafusion_common::ScalarValue;
+    use datafusion_common::config::ConfigOptions;
     use datafusion_expr::Operator as DFOperator;
     use datafusion_expr::ScalarUDF;
     use datafusion_physical_expr::PhysicalExpr;

vortex-datafusion/src/persistent/adapter.rs

@@ -24,19 +24,30 @@
 //!
 //! `datafusion/datasource/src/schema_adapter.rs` -> for can_cast_field (which is crate-private)
 
-use arrow_schema::{DataType, Field, FieldRef, Schema, SchemaRef};
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use arrow_schema::DataType;
+use arrow_schema::Field;
+use arrow_schema::FieldRef;
+use arrow_schema::Schema;
+use arrow_schema::SchemaRef;
 use datafusion_common::ScalarValue;
 use datafusion_common::arrow::compute::can_cast_types;
+use datafusion_common::exec_err;
 use datafusion_common::nested_struct::validate_struct_compatibility;
-use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
-use datafusion_common::{exec_err, plan_err};
+use datafusion_common::plan_err;
+use datafusion_common::tree_node::Transformed;
+use datafusion_common::tree_node::TransformedResult;
+use datafusion_common::tree_node::TreeNode;
 use datafusion_functions::core::getfield::GetFieldFunc;
-use datafusion_physical_expr::expressions::{CastExpr, Column};
-use datafusion_physical_expr::{ScalarFunctionExpr, expressions};
-use datafusion_physical_expr_adapter::{PhysicalExprAdapter, PhysicalExprAdapterFactory};
+use datafusion_physical_expr::ScalarFunctionExpr;
+use datafusion_physical_expr::expressions;
+use datafusion_physical_expr::expressions::CastExpr;
+use datafusion_physical_expr::expressions::Column;
+use datafusion_physical_expr_adapter::PhysicalExprAdapter;
+use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
-use std::fmt::Debug;
-use std::sync::Arc;
 
 #[derive(Debug, Clone)]
 pub struct DefaultPhysicalExprAdapterFactory;

vortex-datafusion/src/persistent/mod.rs

@@ -2,13 +2,13 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 //! Persistent implementation of a Vortex table provider.
+pub(crate) mod adapter;
 mod cache;
 mod format;
 pub mod metrics;
 mod opener;
 mod sink;
 mod source;
-pub(crate) mod adapter;
 
 pub use format::VortexFormat;
 pub use format::VortexFormatFactory;

vortex-datafusion/src/persistent/opener.rs

@@ -2,8 +2,9 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::ops::Range;
+use std::sync::Arc;
+use std::sync::LazyLock;
 use std::sync::Weak;
-use std::sync::{Arc, LazyLock};
 
 use arrow_schema::ArrowError;
 use arrow_schema::Field;
@@ -16,7 +17,8 @@ use datafusion_datasource::PartitionedFile;
 use datafusion_datasource::file_meta::FileMeta;
 use datafusion_datasource::file_stream::FileOpenFuture;
 use datafusion_datasource::file_stream::FileOpener;
-use datafusion_datasource::schema_adapter::{DefaultSchemaAdapterFactory, SchemaAdapterFactory};
+use datafusion_datasource::schema_adapter::DefaultSchemaAdapterFactory;
+use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
 use datafusion_physical_expr::PhysicalExprRef;
 use datafusion_physical_expr::simplifier::PhysicalExprSimplifier;
 use datafusion_physical_expr::split_conjunction;
@@ -180,28 +182,26 @@ impl FileOpener for VortexOpener {
             // for schema evolution and divergence between the table's schema and individual files.
             filter = filter
                 .map(|filter| {
-                    // Rewrite the filter to cast into the logical file schema field types
+                    // Rewrite the filter to properly handle values in the table schema
                     let expr = expr_adapter_factory
                         .create(table_schema.clone(), physical_file_schema.clone())
                         .with_partition_values(partition_values)
-                        .rewrite(filter)
-                        .expect("rewrite");
+                        .rewrite(filter)?;
 
                     // Expression might now reference columns that don't exist in the file, so we can give it
                     // another simplification pass.
-                    PhysicalExprSimplifier::new(table_schema.as_ref()).simplify(expr)
+                    PhysicalExprSimplifier::new(physical_file_schema.as_ref())
+                        .simplify(expr.clone())
                 })
                 .transpose()?;
 
-            let predicate_file_schema = physical_file_schema;
-
             let (schema_mapping, adapted_projections) =
-                schema_adapter.map_schema(&predicate_file_schema)?;
+                schema_adapter.map_schema(&physical_file_schema)?;
 
             // We use the field names from pushdown expression instead.
             let field_names: Vec<FieldName> = adapted_projections
                 .into_iter()
-                .map(|index| FieldName::from(predicate_file_schema.field(index).name().as_str()))
+                .map(|index| FieldName::from(physical_file_schema.field(index).name().as_str()))
                 .collect();
             let projection_expr = select(field_names, root());
 
@@ -247,7 +247,7 @@ impl FileOpener for VortexOpener {
                 .and_then(|f| {
                     let exprs = split_conjunction(&f)
                         .into_iter()
-                        .filter(|expr| can_be_pushed_down(expr, &predicate_file_schema))
+                        .filter(|expr| can_be_pushed_down(expr, &physical_file_schema))
                         .collect::<Vec<_>>();
 
                     make_vortex_predicate(&exprs).transpose()

vortex-datafusion/tests/schema_evolution.rs

@@ -142,11 +142,13 @@ async fn test_filter_with_schema_evolution() {
         record_batch(
             &table_schema,
             vec![
+                // a
                 Arc::new(StringViewArray::from(vec![
                     Some("one"),
                     Some("two"),
                     Some("three"),
                 ])) as ArrowArrayRef,
+                // b
                 Arc::new(StringViewArray::from(vec![
                     Option::<&str>::None,
                     None,