Merge pull request JanKaul#392 from splitgraph/datafusion-55-upgrade

chore: DataFusion 55 upgrade

Author: JanKaul

Cargo.lock

@@ -15,21 +15,21 @@ resolver = "2"
 
 [workspace.dependencies]
 apache-avro = "0.21"
-arrow = "58"
-arrow-schema = "58"
+arrow = "59"
+arrow-schema = "59"
 async-trait = "0.1"
 bytes = "1"
 chrono = { version = "0.4", default-features = false, features = [
     "serde",
     "clock",
 ] }
-datafusion = "54"
-datafusion-common = "54"
-datafusion-execution = "54"
-datafusion-expr = "54"
-datafusion-functions = { version = "54", features = ["crypto_expressions"] }
-datafusion-functions-aggregate = "54"
-datafusion-sql = "54"
+datafusion = "55"
+datafusion-common = "55"
+datafusion-execution = "55"
+datafusion-expr = "55"
+datafusion-functions = { version = "55", features = ["crypto_expressions"] }
+datafusion-functions-aggregate = "55"
+datafusion-sql = "55"
 derive-getters = "0.5.0"
 derive_builder = "0.20"
 futures = "0.3.31"
@@ -39,7 +39,7 @@ lazy_static = "1.5.0"
 lru = "0.18.0"
 object_store = { version = "0.13", features = ["aws", "gcp", "azure"] }
 murmur3 = { version = "0.5.2" }
-parquet = { version = "58", features = ["async", "object_store"] }
+parquet = { version = "59", features = ["async", "object_store"] }
 pin-project-lite = "0.2"
 regex = "1.11.1"
 rust_decimal = "1.42.0"

Cargo.toml

@@ -830,7 +830,10 @@ impl CatalogList for FileCatalogList {
 pub mod tests {
     use datafusion::{
         arrow::array::{Float64Array, Int64Array},
-        common::tree_node::{TransformedResult, TreeNode},
+        common::{
+            config::ConfigNonZeroUsize,
+            tree_node::{TransformedResult, TreeNode},
+        },
         execution::SessionStateBuilder,
         prelude::{SessionConfig, SessionContext},
     };
@@ -915,7 +918,8 @@ pub mod tests {
 
         let mut config = SessionConfig::default();
 
-        config.options_mut().execution.minimum_parallel_output_files = 1;
+        config.options_mut().execution.minimum_parallel_output_files =
+            ConfigNonZeroUsize::try_new(1).unwrap();
         config
             .options_mut()
             .execution

catalogs/iceberg-file-catalog/src/lib.rs

@@ -13,11 +13,14 @@ use std::{
 use async_trait::async_trait;
 use datafusion::{
     arrow::{array::RecordBatch, datatypes::SchemaRef},
-    common::DFSchemaRef,
+    catalog::Session,
+    common::{tree_node::TreeNodeRecursion, DFSchemaRef},
     error::DataFusionError,
-    execution::{RecordBatchStream, SendableRecordBatchStream, SessionState},
+    execution::{RecordBatchStream, SendableRecordBatchStream},
+    logical_expr::physical_planning_context::PhysicalPlanningContext,
     physical_plan::{
-        stream::RecordBatchStreamAdapter, DisplayAs, ExecutionPlan, Partitioning, PlanProperties,
+        stream::RecordBatchStreamAdapter, DisplayAs, ExecutionPlan, Partitioning, PhysicalExpr,
+        PlanProperties,
     },
     physical_planner::{ExtensionPlanner, PhysicalPlanner},
 };
@@ -179,6 +182,13 @@ impl ExecutionPlan for PhysicalForkNode {
         vec![&self.input]
     }
 
+    fn apply_expressions(
+        &self,
+        _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion, DataFusionError>,
+    ) -> Result<TreeNodeRecursion, DataFusionError> {
+        Ok(TreeNodeRecursion::Continue)
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         mut children: Vec<Arc<dyn ExecutionPlan>>,
@@ -296,7 +306,8 @@ impl ExtensionPlanner for ForkNodePlanner {
         node: &dyn UserDefinedLogicalNode,
         logical_inputs: &[&LogicalPlan],
         physical_inputs: &[Arc<dyn ExecutionPlan>],
-        _session_state: &SessionState,
+        _session: &dyn Session,
+        _planning_ctx: &PhysicalPlanningContext,
     ) -> Result<Option<Arc<dyn ExecutionPlan>>, DataFusionError> {
         if let Some(fork_node) = node.as_any().downcast_ref::<ForkNode>() {
             assert_eq!(physical_inputs.len(), 1);

datafusion_iceberg/src/materialized_view/delta_queries/fork_node.rs

@@ -7,14 +7,13 @@ use std::{
 use datafusion::common::NullEquality;
 use datafusion::{
     catalog::TableProvider,
-    common::{tree_node::Transformed, Column, DFSchema},
+    common::{tree_node::Transformed, Column, DFSchema, TableReference},
     datasource::{empty::EmptyTable, DefaultTableSource},
     error::DataFusionError,
-    sql::TableReference,
 };
 use datafusion_expr::{
     build_join_schema, expr::Alias, Aggregate, Expr, Filter, Join, JoinConstraint, JoinType,
-    LogicalPlan, Projection, SubqueryAlias, TableScan, Union,
+    LogicalPlan, Projection, SubqueryAlias, TableScanBuilder, Union,
 };
 use iceberg_rust::error::Error;
 
@@ -38,13 +37,13 @@ pub(crate) fn delta_transform_down(
 ) -> Result<Transformed<LogicalPlan>, DataFusionError> {
     let storage_table_reference = storage_table.0;
 
-    let storage_table_scan = Arc::new(LogicalPlan::TableScan(TableScan::try_new(
-        storage_table_reference.clone(),
-        Arc::new(DefaultTableSource::new(storage_table.1)),
-        None,
-        Vec::new(),
-        None,
-    )?));
+    let storage_table_scan = Arc::new(LogicalPlan::TableScan(
+        TableScanBuilder::new(
+            storage_table_reference.clone(),
+            Arc::new(DefaultTableSource::new(storage_table.1)),
+        )
+        .build()?,
+    ));
 
     let storage_table_schema = storage_table_scan.schema().clone();
     match &plan {
@@ -549,9 +548,9 @@ mod tests {
     use std::{ops::Deref, sync::Arc};
 
     use datafusion::common::tree_node::TreeNode;
+    use datafusion::common::TableReference;
     use datafusion::datasource::empty::EmptyTable;
     use datafusion::prelude::SessionContext;
-    use datafusion::sql::TableReference;
     use datafusion_expr::LogicalPlan;
     use iceberg_rust::catalog::Catalog;
     use iceberg_rust::object_store::ObjectStoreBuilder;

datafusion_iceberg/src/materialized_view/delta_queries/transform.rs

@@ -1,8 +1,11 @@
 use std::{collections::HashMap, sync::Arc};
 
 use datafusion::{
-    arrow::error::ArrowError, catalog::TableProvider, common::tree_node::TreeNode,
-    execution::SessionStateBuilder, prelude::SessionContext, sql::TableReference,
+    arrow::error::ArrowError,
+    catalog::TableProvider,
+    common::{tree_node::TreeNode, TableReference},
+    execution::SessionStateBuilder,
+    prelude::SessionContext,
 };
 use datafusion_expr::LogicalPlan;
 use delta_queries::{

datafusion_iceberg/src/materialized_view/mod.rs

@@ -14,18 +14,17 @@ use crate::{
 };
 use datafusion::{
     arrow::datatypes::{DataType, Schema as ArrowSchema},
-    catalog::CatalogProvider,
-    common::{tree_node::Transformed, SchemaReference},
+    catalog::{CatalogProvider, Session},
+    common::{tree_node::Transformed, SchemaReference, TableReference},
     error::DataFusionError,
-    execution::context::{QueryPlanner, SessionState},
+    execution::context::QueryPlanner,
     logical_expr::{
-        CreateExternalTable, DdlStatement, Extension, InvariantLevel, LogicalPlan,
-        UserDefinedLogicalNode,
+        physical_planning_context::PhysicalPlanningContext, CreateExternalTable, DdlStatement,
+        Extension, InvariantLevel, LogicalPlan, UserDefinedLogicalNode,
     },
     physical_plan::{empty::EmptyExec, ExecutionPlan},
     physical_planner::{DefaultPhysicalPlanner, ExtensionPlanner, PhysicalPlanner},
     scalar::ScalarValue,
-    sql::TableReference,
 };
 use iceberg_rust::{
     catalog::{tabular::Tabular, CatalogList},
@@ -72,11 +71,9 @@ impl QueryPlanner for IcebergQueryPlanner {
     async fn create_physical_plan(
         &self,
         logical_plan: &LogicalPlan,
-        session_state: &SessionState,
+        session: &dyn Session,
     ) -> Result<Arc<dyn ExecutionPlan>, DataFusionError> {
-        self.0
-            .create_physical_plan(logical_plan, session_state)
-            .await
+        self.0.create_physical_plan(logical_plan, session).await
     }
 }
 
@@ -85,7 +82,7 @@ pub fn iceberg_transform(node: LogicalPlan) -> Result<Transformed<LogicalPlan>,
         LogicalPlan::Ddl(DdlStatement::CreateExternalTable(table)) => {
             if table.file_type.to_lowercase() == "iceberg" {
                 Ok(Transformed::yes(LogicalPlan::Extension(Extension {
-                    node: Arc::new(CreateIcebergTable(table)),
+                    node: Arc::new(CreateIcebergTable(*table)),
                 })))
             } else {
                 Ok(Transformed::no(LogicalPlan::Ddl(
@@ -127,18 +124,19 @@ impl ExtensionPlanner for IcebergExtensionPlanner {
         node: &dyn UserDefinedLogicalNode,
         _logical_inputs: &[&LogicalPlan],
         _physical_inputs: &[Arc<dyn ExecutionPlan>],
-        session_state: &SessionState,
+        session: &dyn Session,
+        _planning_ctx: &PhysicalPlanningContext,
     ) -> Result<Option<Arc<dyn ExecutionPlan>>, DataFusionError> {
         if let Some(node) = node.as_any().downcast_ref::<CreateIcebergTable>() {
-            plan_create_table(node, session_state).await
+            plan_create_table(node, session).await
         } else if let Some(node) = node.as_any().downcast_ref::<CreateIcebergView>() {
-            plan_create_view(node, session_state).await
+            plan_create_view(node, session).await
         } else if let Some(node) = node.as_any().downcast_ref::<CreateIcebergNamespace>() {
-            plan_create_namespace(node, session_state).await
+            plan_create_namespace(node, session).await
         } else if let Some(node) = node.as_any().downcast_ref::<DropIcebergTable>() {
-            plan_drop_table(node, session_state).await
+            plan_drop_table(node, session).await
         } else if let Some(node) = node.as_any().downcast_ref::<DropIcebergNamespace>() {
-            plan_drop_namespace(node, session_state).await
+            plan_drop_namespace(node, session).await
         } else {
             return Ok(None);
         }
@@ -147,13 +145,13 @@ impl ExtensionPlanner for IcebergExtensionPlanner {
 
 async fn plan_create_table(
     node: &CreateIcebergTable,
-    session_state: &SessionState,
+    session: &dyn Session,
 ) -> Result<Option<Arc<dyn ExecutionPlan>>, DataFusionError> {
     let table_ref = &node.0.name.to_string();
 
     let identifier = TableReference::parse_str(table_ref).resolve("datafusion", "public");
 
-    let catalog_list = session_state.catalog_list();
+    let catalog_list = session.catalog_list();
     let catalog_name = &identifier.catalog;
     let namespace_name = &identifier.schema;
     let table_name: &str = &identifier.table;
@@ -206,9 +204,15 @@ async fn plan_create_table(
         .build()
         .map_err(|err| DataFusionError::External(Box::new(err)))?;
 
+    let [location] = node.0.locations.as_slice() else {
+        return Err(DataFusionError::Plan(
+            "Iceberg tables must be created with exactly one LOCATION.".to_owned(),
+        ));
+    };
+
     Table::builder()
         .with_name(table_name)
-        .with_location(&node.0.location)
+        .with_location(location)
         .with_schema(Schema::from_struct_type(schema, DEFAULT_SCHEMA_ID, None))
         .with_partition_spec(partition_spec)
         .with_properties(node.0.options.clone())
@@ -223,13 +227,13 @@ async fn plan_create_table(
 
 async fn plan_create_view(
     node: &CreateIcebergView,
-    session_state: &SessionState,
+    session: &dyn Session,
 ) -> Result<Option<Arc<dyn ExecutionPlan>>, DataFusionError> {
     let table_ref = &node.0.name.to_string();
 
     let identifier = TableReference::parse_str(table_ref).resolve("datafusion", "public");
 
-    let catalog_list = session_state.catalog_list();
+    let catalog_list = session.catalog_list();
     let catalog_name = &identifier.catalog;
     let namespace_name = &identifier.schema;
     let table_name: &str = &identifier.table;
@@ -347,7 +351,7 @@ async fn plan_create_view(
 
 async fn plan_create_namespace(
     node: &CreateIcebergNamespace,
-    session_state: &SessionState,
+    session: &dyn Session,
 ) -> Result<Option<Arc<dyn ExecutionPlan>>, DataFusionError> {
     let (catalog_name, namespace_name) =
         node.0
@@ -359,7 +363,7 @@ async fn plan_create_namespace(
                 &node.0.schema_name
             )))?;
 
-    let catalog_list = session_state.catalog_list();
+    let catalog_list = session.catalog_list();
     let datafusion_catalog = catalog_list
         .catalog(catalog_name)
         .ok_or(DataFusionError::Plan(format!(
@@ -393,13 +397,13 @@ async fn plan_create_namespace(
 
 async fn plan_drop_table(
     node: &DropIcebergTable,
-    session_state: &SessionState,
+    session: &dyn Session,
 ) -> Result<Option<Arc<dyn ExecutionPlan>>, DataFusionError> {
     let table_ref = &node.0.name.to_string();
 
     let identifier = TableReference::parse_str(table_ref).resolve("datafusion", "public");
 
-    let catalog_list = session_state.catalog_list();
+    let catalog_list = session.catalog_list();
     let catalog_name = &identifier.catalog;
     let namespace_name = &identifier.schema;
     let table_name: &str = &identifier.table;
@@ -431,14 +435,14 @@ async fn plan_drop_table(
 
 async fn plan_drop_namespace(
     node: &DropIcebergNamespace,
-    session_state: &SessionState,
+    session: &dyn Session,
 ) -> Result<Option<Arc<dyn ExecutionPlan>>, DataFusionError> {
     let (catalog_name, namespace_name) = match &node.0.name {
         SchemaReference::Bare { schema } => ("datafusion".to_owned(), schema.to_string()),
         SchemaReference::Full { schema, catalog } => (catalog.to_string(), schema.to_string()),
     };
 
-    let catalog_list = session_state.catalog_list();
+    let catalog_list = session.catalog_list();
     let datafusion_catalog = catalog_list
         .catalog(&catalog_name)
         .ok_or(DataFusionError::Plan(format!(