fix: datafusion predicate pushdown and dependencies

python/Cargo.toml

@@ -17,7 +17,7 @@ crate-type = ["cdylib"]
 name = "deltalake._internal"
 
 [dependencies]
-arrow-schema = { version = "28", features = ["serde"] }
+arrow-schema = { version = "29", features = ["serde"] }
 chrono = "0"
 env_logger = "0"
 futures = "0.3"

rust/Cargo.toml

@@ -10,7 +10,7 @@ description = "Native Delta Lake implementation in Rust"
 edition = "2021"
 
 [dependencies]
-arrow = { version = "28", optional = true }
+arrow = { version = "29", optional = true }
 async-trait = "0.1"
 bytes = "1"
 chrono = { version = "0.4.22", default-features = false, features = ["clock"] }
@@ -26,7 +26,7 @@ num-traits = "0.2.15"
 object_store = "0.5.3"
 once_cell = "1.16.0"
 parking_lot = "0.12"
-parquet = { version = "28", features = ["async"], optional = true }
+parquet = { version = "29", features = ["async"], optional = true }
 parquet2 = { version = "0.17", optional = true }
 percent-encoding = "2"
 serde = { version = "1", features = ["derive"] }
@@ -47,10 +47,10 @@ rusoto_dynamodb = { version = "0.48", default-features = false, optional = true
 rusoto_glue = { version = "0.48", default-features = false, optional = true }
 
 # Datafusion
-datafusion = { version = "15", optional = true }
-datafusion-expr = { version = "15", optional = true }
-datafusion-common = { version = "15", optional = true }
-datafusion-proto = { version = "15", optional = true }
+datafusion = { version = "16", optional = true }
+datafusion-expr = { version = "16", optional = true }
+datafusion-common = { version = "16", optional = true }
+datafusion-proto = { version = "16", optional = true }
 
 # NOTE dependencies only for integration tests
 fs_extra = { version = "1.2.0", optional = true }

rust/src/delta_arrow.rs

@@ -459,7 +459,7 @@ pub(crate) fn delta_log_schema_for_table(
             .for_each(|f| max_min_schema_for_fields(&mut max_min_vec, f));
 
         stats_parsed_fields.extend(
-            ["minValues", "maxValues"].iter().map(|name| {
+            ["minValues", "maxValues"].into_iter().map(|name| {
                 ArrowField::new(name, ArrowDataType::Struct(max_min_vec.clone()), true)
             }),
         );

rust/src/delta_datafusion.rs

@@ -50,7 +50,7 @@ use datafusion::physical_plan::{
 use datafusion_common::scalar::ScalarValue;
 use datafusion_common::{Column, DataFusionError, Result as DataFusionResult};
 use datafusion_expr::logical_plan::CreateExternalTable;
-use datafusion_expr::{Expr, Extension, LogicalPlan};
+use datafusion_expr::{Expr, Extension, LogicalPlan, TableProviderFilterPushDown};
 use datafusion_proto::logical_plan::LogicalExtensionCodec;
 use datafusion_proto::physical_plan::PhysicalExtensionCodec;
 use object_store::{path::Path, ObjectMeta};
@@ -321,6 +321,10 @@ fn register_store(table: &DeltaTable, env: Arc<RuntimeEnv>) {
 
 #[async_trait]
 impl TableProvider for DeltaTable {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
     fn schema(&self) -> Arc<ArrowSchema> {
         Arc::new(
             <ArrowSchema as TryFrom<&schema::Schema>>::try_from(DeltaTable::schema(self).unwrap())
@@ -332,6 +336,14 @@ impl TableProvider for DeltaTable {
         TableType::Base
     }
 
+    fn get_table_definition(&self) -> Option<&str> {
+        None
+    }
+
+    fn get_logical_plan(&self) -> Option<&LogicalPlan> {
+        None
+    }
+
     async fn scan(
         &self,
         session: &SessionState,
@@ -343,7 +355,7 @@ impl TableProvider for DeltaTable {
             DeltaTable::schema(self).unwrap(),
         )?);
 
-        register_store(self, session.runtime_env.clone());
+        register_store(self, session.runtime_env().clone());
 
         // TODO we group files together by their partition values. If the table is partitioned
         // and partitions are somewhat evenly distributed, probably not the worst choice ...
@@ -358,8 +370,8 @@ impl TableProvider for DeltaTable {
                 .files()
                 .iter()
                 .zip(files_to_prune.into_iter())
-                .for_each(|(action, prune_file)| {
-                    if !prune_file {
+                .for_each(|(action, keep_file)| {
+                    if keep_file {
                         let part = partitioned_file_from_action(action, &schema);
                         file_groups
                             .entry(part.partition_values.clone())
@@ -387,8 +399,9 @@ impl TableProvider for DeltaTable {
                 .collect(),
         ));
 
-        let parquet_scan = ParquetFormat::new(session.config_options())
+        let parquet_scan = ParquetFormat::new()
             .create_physical_plan(
+                session,
                 FileScanConfig {
                     object_store_url: self.storage.object_store_url(),
                     file_schema,
@@ -406,7 +419,7 @@ impl TableProvider for DeltaTable {
                         })
                         .collect::<Result<Vec<_>, ArrowError>>()?,
                     output_ordering: None,
-                    config_options: Default::default(),
+                    infinite_source: false,
                 },
                 filters,
             )
@@ -418,8 +431,15 @@ impl TableProvider for DeltaTable {
         }))
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
+    fn supports_filter_pushdown(
+        &self,
+        _filter: &Expr,
+    ) -> DataFusionResult<TableProviderFilterPushDown> {
+        Ok(TableProviderFilterPushDown::Inexact)
+    }
+
+    fn statistics(&self) -> Option<Statistics> {
+        Some(self.datafusion_table_statistics())
     }
 }
 

rust/src/operations/load.rs

@@ -85,7 +85,7 @@ impl std::future::IntoFuture for LoadBuilder {
 
             let ctx = SessionContext::new();
             ctx.state()
-                .runtime_env
+                .runtime_env()
                 .register_object_store(url.scheme(), "", store);
             let scan_plan = table.scan(&ctx.state(), None, &[], None).await?;
             let plan = CoalescePartitionsExec::new(scan_plan);

rust/src/table_state_arrow.rs

@@ -402,7 +402,7 @@ impl DeltaTableState {
                     .flat_map(|sub_field| {
                         if let Some(values) = getter(sub_field) {
                             let field = Field::new(
-                                sub_field
+                                *sub_field
                                     .path
                                     .last()
                                     .expect("paths must have at least one element"),