Upgrade deltalake and df

Author: matthewmturner

Cargo.lock

@@ -14,14 +14,14 @@ version = "0.2.2"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-arrow-flight = { version = "54.1.0", features = [
+arrow-flight = { version = "54", features = [
   "flight-sql-experimental",
 ], optional = true }
 axum = { version = "0.7.9", features = ["macros"], optional = true }
 clap = { version = "4.5.27", features = ["derive"] }
 color-eyre = "0.6.3"
 crossterm = { version = "0.28.1", features = ["event-stream"] }
-datafusion = { version = "45.0.0" }
+datafusion = { version = "46" }
 datafusion-app = { version = "0.1.0", path = "crates/datafusion-app" }
 directories = "5.0.1"
 env_logger = "0.11.5"
@@ -34,7 +34,7 @@ lazy_static = "1.4.0"
 log = "0.4.22"
 metrics = { version = "0.24.0", optional = true }
 metrics-exporter-prometheus = { version = "0.16.0", optional = true }
-parquet = "54.1.0"
+parquet = "54"
 pin-project-lite = { version = "0.2.14" }
 prost = "0.13.1"
 ratatui = "0.28.0"

Cargo.toml

@@ -6,21 +6,19 @@ version = "0.1.0"
 # Leaving Hudi and Iceberg around in case traction on those picks up.  It would be better if I just remove them from main and create PRs with the diff - maybe ill get around to that.
 
 [dependencies]
-arrow-flight = { version = "54.1.0", features = [
+arrow-flight = { version = "54", features = [
   "flight-sql-experimental",
 ], optional = true }
 async-trait = "0.1.80"
 base64 = { version = "0.22.1", optional = true }
 color-eyre = "0.6.3"
-datafusion = "45.0.0"
-datafusion-functions-json = { version = "0.45", optional = true }
-datafusion-functions-parquet = { version = "0.1.0", path = "..//datafusion-functions-parquet", optional = true }
+datafusion = "46"
+datafusion-functions-json = { version = "0.46", optional = true }
+datafusion-functions-parquet = { version = "0.1.0", path = "../datafusion-functions-parquet", optional = true }
 datafusion-udfs-wasm = { version = "0.1.0", path = "../datafusion-udfs-wasm", features = [
   "serde",
 ], optional = true }
-deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "d8995e64c305dc795f62d33ce2a4614b335944ff", features = [
-  "datafusion",
-], optional = true }
+deltalake = { version = "0.25", features = ["datafusion"], optional = true }
 directories = "5.0.1"
 futures = "0.3.30"
 indexmap = { version = "2.8.0", features = ["serde"] }

crates/datafusion-app/Cargo.toml

@@ -16,7 +16,7 @@
 // under the License.
 
 use datafusion::{
-    datasource::physical_plan::ParquetExec,
+    datasource::{physical_plan::ParquetSource, source::DataSourceExec},
     physical_plan::{
         aggregates::AggregateExec,
         filter::FilterExec,
@@ -354,11 +354,19 @@ impl PlanIOVisitor {
             self.time_opening = metrics.sum_by_name("time_elapsed_opening");
             self.time_scanning = metrics.sum_by_name("time_elapsed_scanning_total");
 
-            if plan.as_any().downcast_ref::<ParquetExec>().is_some() {
-                self.parquet_output_rows = metrics.output_rows();
-                self.parquet_rg_pruned_stats = metrics.sum_by_name("row_groups_pruned_statistics");
-                self.parquet_rg_matched_stats =
-                    metrics.sum_by_name("row_groups_matched_statistics");
+            if let Some(data_source_exec) = plan.as_any().downcast_ref::<DataSourceExec>() {
+                if data_source_exec
+                    .data_source()
+                    .as_any()
+                    .downcast_ref::<ParquetSource>()
+                    .is_some()
+                {
+                    self.parquet_output_rows = metrics.output_rows();
+                    self.parquet_rg_pruned_stats =
+                        metrics.sum_by_name("row_groups_pruned_statistics");
+                    self.parquet_rg_matched_stats =
+                        metrics.sum_by_name("row_groups_matched_statistics");
+                }
             }
         }
     }

crates/datafusion-app/src/stats.rs

@@ -11,7 +11,7 @@ repository = "https://github.com/datafusion-contrib/datafusion-dft/tree/main/cra
 version = "0.1.0"
 
 [dependencies]
-arrow = { version = "54.1.0" }
+arrow = { version = "54" }
 async-trait = "0.1.41"
-datafusion = { version = "45.0.0" }
-parquet = { version = "54.1.0", default-features = false }
+datafusion = { version = "46" }
+parquet = { version = "54", default-features = false }

crates/datafusion-functions-parquet/Cargo.toml

@@ -23,10 +23,10 @@ use async_trait::async_trait;
 use datafusion::catalog::Session;
 use datafusion::catalog::TableFunctionImpl;
 use datafusion::common::{plan_err, Column};
+use datafusion::datasource::memory::MemorySourceConfig;
 use datafusion::datasource::TableProvider;
 use datafusion::error::Result;
 use datafusion::logical_expr::Expr;
-use datafusion::physical_plan::memory::MemoryExec;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::scalar::ScalarValue;
 use parquet::basic::ConvertedType;
@@ -51,7 +51,7 @@ impl TableProvider for ParquetMetadataTable {
         self
     }
 
-    fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef {
+    fn schema(&self) -> arrow::datatypes::SchemaRef {
         self.schema.clone()
     }
 
@@ -66,11 +66,11 @@ impl TableProvider for ParquetMetadataTable {
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        Ok(Arc::new(MemoryExec::try_new(
+        Ok(MemorySourceConfig::try_new_exec(
             &[vec![self.batch.clone()]],
             TableProvider::schema(self),
             projection.cloned(),
-        )?))
+        )?)
     }
 }
 
@@ -183,7 +183,7 @@ impl TableFunctionImpl for ParquetMetadataFunc {
             Field::new("total_uncompressed_size", DataType::Int64, true),
         ]));
 
-        // construct recordbatch from metadata
+        // construct record batch from metadata
         let mut filename_arr = vec![];
         let mut row_group_id_arr = vec![];
         let mut row_group_num_rows_arr = vec![];

crates/datafusion-functions-parquet/src/lib.rs

@@ -4,7 +4,7 @@ name = "datafusion-udfs-wasm"
 version = "0.1.0"
 
 [dependencies]
-datafusion = { version = "45.0.0", default-features = false }
+datafusion = { version = "46", default-features = false }
 log = "0.4.25"
 serde = { version = "1.0.217", optional = true }
 wasi-common = "29.0.1"

crates/datafusion-udfs-wasm/Cargo.toml

@@ -7,4 +7,4 @@ version = "0.1.0"
 crate-type = ["cdylib"]
 
 [dependencies]
-arrow = "53"
+arrow = "54"