build: update to df 38 (#503)

Cargo.toml

@@ -36,7 +36,7 @@ resolver = "2"
 [workspace.dependencies]
 arrow = { version = "51.0.0" }
 async-trait = "0.1.80"
-datafusion = { version = "37", features = ["compression", "parquet"] }
+datafusion = { version = "38", features = ["compression", "parquet"] }
 futures = "0.3"
 noodles = { version = "0.71" }
 object_store = { version = "0.9" }

exon/exon-cli/Cargo.toml

@@ -10,7 +10,7 @@ version.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-datafusion-cli = { version = "37" }
+datafusion-cli = { version = "38" }
 clap = { version = "4", features = ["derive", "cargo"] }
 datafusion = { workspace = true }
 exon = { path = "../exon-core", version = "0.20.0", features = ["default"] }

exon/exon-core/src/physical_plan/object_store/hive_partition.rs

@@ -24,7 +24,7 @@ use arrow::{
 };
 
 use datafusion::{
-    common::{DFField, DFSchema},
+    common::DFSchema,
     datasource::listing::{ListingTableUrl, PartitionedFile},
     error::{DataFusionError, Result},
     execution::context::SessionState,
@@ -146,6 +146,7 @@ pub async fn pruned_partition_list<'a>(
                     partition_values: partition_values.clone(),
                     range: None,
                     extensions: None,
+                    statistics: None,
                 })
             }));
 
@@ -262,10 +263,10 @@ async fn prune_partitions(
     // let fields: Fields = partition_cols.collect();
     let schema = Arc::new(Schema::new(partition_cols.to_vec()));
 
-    let df_schema = DFSchema::new_with_metadata(
+    let df_schema = DFSchema::from_unqualifed_fields(
         partition_cols
             .iter()
-            .map(|f| DFField::new_unqualified(f.name(), f.data_type().clone(), f.is_nullable())) // TODO: use qualified name, remove clone
+            .map(|f| Field::new(f.name(), f.data_type().clone(), f.is_nullable())) // TODO: use qualified name, remove clone
             .collect(),
         Default::default(),
     )?;

exon/exon-core/src/physical_plan/planner/exon_physical_planner.rs

@@ -21,7 +21,7 @@ use datafusion::{
     error::Result,
     execution::context::SessionState,
     logical_expr::{Expr, LogicalPlan},
-    physical_plan::{insert::FileSinkExec, ExecutionPlan, PhysicalExpr},
+    physical_plan::{insert::DataSinkExec, ExecutionPlan, PhysicalExpr},
     physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner},
 };
 
@@ -48,7 +48,7 @@ impl PhysicalPlanner for ExonPhysicalPlanner {
         let runtime = session_state.runtime_env();
 
         // try to downcast plan as FileSinkExec
-        if let Some(file_sink) = plan.as_any().downcast_ref::<FileSinkExec>() {
+        if let Some(file_sink) = plan.as_any().downcast_ref::<DataSinkExec>() {
             let sink = file_sink.sink();
 
             // Try to downcast to a ParquetSink