chore(deps): upgrade datafusion to 52.0.0

Cargo.toml

@@ -38,11 +38,12 @@ arrow-select = { version = "57" }
 object_store = { version = "0.12.1" }
 parquet = { version = "57" }
 
-# datafusion
-datafusion = "51.0"
-datafusion-datasource = "51.0"
-datafusion-ffi = "51.0"
-datafusion-proto = "51.0"
+# datafusion 52.1
+datafusion = { version = "52.1.0" }
+datafusion-datasource = { version = "52.1.0" }
+datafusion-ffi = { version = "52.1.0" }
+datafusion-proto = { version = "52.1.0" }
+datafusion-physical-expr-adapter = { version = "52.1.0" }
 
 # serde
 serde = { version = "1.0.194", features = ["derive"] }

crates/core/Cargo.toml

@@ -38,6 +38,7 @@ object_store = { workspace = true }
 datafusion = { workspace = true, optional = true }
 datafusion-datasource = { workspace = true, optional = true }
 datafusion-proto = { workspace = true, optional = true }
+datafusion-physical-expr-adapter = { workspace = true, optional = true }
 
 # serde
 serde = { workspace = true, features = ["derive"] }
@@ -105,7 +106,12 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [features]
 default = ["rustls"]
-datafusion = ["dep:datafusion", "datafusion-datasource", "datafusion-proto"]
+datafusion = [
+    "dep:datafusion",
+    "datafusion-datasource",
+    "datafusion-proto",
+    "datafusion-physical-expr-adapter",
+]
 datafusion-ext = ["datafusion"]
 json = ["parquet/json"]
 python = ["arrow/pyarrow"]

crates/core/src/delta_datafusion/mod.rs

@@ -92,7 +92,6 @@ mod find_files;
 pub mod logical;
 pub mod physical;
 pub mod planner;
-mod schema_adapter;
 mod session;
 mod table_provider;
 pub(crate) mod utils;

crates/core/src/delta_datafusion/session.rs

@@ -48,7 +48,16 @@ impl Default for DeltaSessionConfig {
     fn default() -> Self {
         DeltaSessionConfig {
             inner: SessionConfig::default()
-                .set_bool("datafusion.sql_parser.enable_ident_normalization", false),
+                .set_bool("datafusion.sql_parser.enable_ident_normalization", false)
+                .set_bool("datafusion.execution.parquet.schema_force_view_types", true)
+                // Workaround: hash-join dynamic filtering (IN-list pushdown) can panic when join
+                // keys include dictionary arrays (still reproducible with DF 52.1.x crates).
+                // Disable IN-list pushdown and fall back to hash lookups.
+                .set_usize("datafusion.optimizer.hash_join_inlist_pushdown_max_size", 0)
+                .set_usize(
+                    "datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values",
+                    0,
+                ),
         }
     }
 }

crates/core/src/delta_datafusion/table_provider.rs

@@ -12,14 +12,15 @@ use datafusion::catalog::TableProvider;
 use datafusion::catalog::memory::DataSourceExec;
 use datafusion::common::pruning::PruningStatistics;
 use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion};
-use datafusion::common::{Column, DFSchemaRef, Result, Statistics, ToDFSchema};
+use datafusion::common::{Column, ColumnStatistics, DFSchemaRef, Result, Statistics, ToDFSchema};
 use datafusion::config::{ConfigOptions, TableParquetOptions};
 use datafusion::datasource::TableType;
 use datafusion::datasource::physical_plan::{
-    FileGroup, FileSource, wrap_partition_type_in_dict, wrap_partition_value_in_dict,
+    FileGroup, wrap_partition_type_in_dict, wrap_partition_value_in_dict,
 };
 use datafusion::datasource::physical_plan::{FileScanConfigBuilder, ParquetSource};
 use datafusion::datasource::sink::DataSinkExec;
+use datafusion::datasource::table_schema::TableSchema;
 use datafusion::error::DataFusionError;
 use datafusion::execution::{SendableRecordBatchStream, TaskContext};
 use datafusion::logical_expr::dml::InsertOp;
@@ -50,7 +51,7 @@ use url::Url;
 use uuid::Uuid;
 
 use crate::delta_datafusion::engine::AsObjectStoreUrl;
-use crate::delta_datafusion::schema_adapter::DeltaSchemaAdapterFactory;
+
 use crate::delta_datafusion::table_provider::next::SnapshotWrapper;
 use crate::delta_datafusion::{
     DataFusionMixins as _, FindFilesExprProperties, LogDataHandler, get_null_of_arrow_type,
@@ -151,7 +152,7 @@ impl DeltaScanConfigBuilder {
                 Some(name) => {
                     if column_names.contains(name) {
                         return Err(DeltaTableError::Generic(format!(
-                            "Unable to add file path column since column with name {name} exits"
+                            "Unable to add file path column since column with name {name} exists"
                         )));
                     }
 
@@ -179,7 +180,7 @@ impl DeltaScanConfigBuilder {
             wrap_partition_values: self.wrap_partition_values.unwrap_or(true),
             enable_parquet_pushdown: self.enable_parquet_pushdown,
             schema: self.schema.clone(),
-            schema_force_view_types: false,
+            schema_force_view_types: true,
         })
     }
 }
@@ -538,12 +539,82 @@ impl<'a> DeltaScanBuilder<'a> {
 
         let stats = stats.unwrap_or(Statistics::new_unknown(&schema));
 
+        // DF52's TableSchema outputs columns as: file_schema + partition_columns
+        // Source stats are indexed by TableConfiguration.schema() field order, which may differ
+        // from the scan schema order. We need name-based remapping, not index-based.
+        let partition_col_names = self.snapshot.metadata().partition_columns();
+
+        // Build name -> ColumnStatistics map from source stats (keyed by TableConfiguration schema order)
+        let source_schema = self.snapshot.schema();
+        let stats_by_name: HashMap<String, ColumnStatistics> = source_schema
+            .fields()
+            .enumerate()
+            .filter_map(|(idx, field)| {
+                stats
+                    .column_statistics
+                    .get(idx)
+                    .map(|s| (field.name().to_string(), s.clone()))
+            })
+            .collect();
+
+        // Build stats in DF52 order: file_schema columns first, then partition_columns
+        // file_schema columns are in file_schema field order (non-partition from logical_schema)
+        let file_col_stats: Vec<ColumnStatistics> = file_schema
+            .fields()
+            .iter()
+            .map(|f| {
+                stats_by_name
+                    .get(f.name())
+                    .cloned()
+                    .unwrap_or_else(ColumnStatistics::new_unknown)
+            })
+            .collect();
+
+        // Partition columns must be in metadata.partition_columns() order (not schema encounter order)
+        let partition_col_stats: Vec<ColumnStatistics> = partition_col_names
+            .iter()
+            .map(|name| {
+                stats_by_name
+                    .get(name)
+                    .cloned()
+                    .unwrap_or_else(ColumnStatistics::new_unknown)
+            })
+            .collect();
+
+        // Combine: file columns first, then partition columns
+        let mut reordered_stats = file_col_stats;
+        reordered_stats.extend(partition_col_stats);
+
+        let stats = Statistics {
+            num_rows: stats.num_rows,
+            total_byte_size: stats.total_byte_size,
+            column_statistics: reordered_stats,
+        };
+
+        // Add unknown stats for file_column if present (it's added as partition field but not in original schema)
+        let stats = if config.file_column_name.is_some() {
+            let mut col_stats = stats.column_statistics;
+            col_stats.push(ColumnStatistics::new_unknown());
+            Statistics {
+                num_rows: stats.num_rows,
+                total_byte_size: stats.total_byte_size,
+                column_statistics: col_stats,
+            }
+        } else {
+            stats
+        };
+
         let parquet_options = TableParquetOptions {
             global: self.session.config().options().execution.parquet.clone(),
             ..Default::default()
         };
 
-        let mut file_source = ParquetSource::new(parquet_options);
+        let partition_fields: Vec<Arc<Field>> =
+            table_partition_cols.into_iter().map(Arc::new).collect();
+        let table_schema = TableSchema::new(file_schema, partition_fields);
+
+        let mut file_source =
+            ParquetSource::new(table_schema).with_table_parquet_options(parquet_options);
 
         // 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
@@ -553,11 +624,9 @@ impl<'a> DeltaScanBuilder<'a> {
         {
             file_source = file_source.with_predicate(predicate);
         };
-        let file_source =
-            file_source.with_schema_adapter_factory(Arc::new(DeltaSchemaAdapterFactory {}))?;
 
         let file_scan_config =
-            FileScanConfigBuilder::new(self.log_store.object_store_url(), file_schema, file_source)
+            FileScanConfigBuilder::new(self.log_store.object_store_url(), Arc::new(file_source))
                 .with_file_groups(
                     // If all files were filtered out, we still need to emit at least one partition to
                     // pass datafusion sanity checks.
@@ -570,9 +639,8 @@ impl<'a> DeltaScanBuilder<'a> {
                     },
                 )
                 .with_statistics(stats)
-                .with_projection_indices(self.projection.cloned())
+                .with_projection_indices(self.projection.cloned())?
                 .with_limit(self.limit)
-                .with_table_partition_cols(table_partition_cols)
                 .build();
 
         let metrics = ExecutionPlanMetricsSet::new();