Merge pull request #165 from simonvandel/df-47

WIP: Upgrade to DataFusion 47

Author: JanKaul

Cargo.lock

@@ -8,33 +8,36 @@ members = [
     "datafusion-iceberg-sql",
     "datafusion_iceberg",
     "iceberg-rust",
-    "iceberg-rust-spec"
+    "iceberg-rust-spec",
 ]
 
 resolver = "2"
 
 [workspace.dependencies]
 apache-avro = "0.17.0"
-arrow = "54"
-arrow-schema = "54"
+arrow = "55"
+arrow-schema = "55"
 async-trait = "0.1"
 bytes = "1"
-chrono = { version = "0.4", default-features = false, features = ["serde", "clock"] }
-datafusion = "46"
-datafusion-common = "46"
-datafusion-execution = "46"
-datafusion-expr = "46"
-datafusion-functions = { version = "46", features = ["crypto_expressions"] }
-datafusion-functions-aggregate = "46"
-datafusion-sql = "46"
+chrono = { version = "0.4", default-features = false, features = [
+    "serde",
+    "clock",
+] }
+datafusion = "47"
+datafusion-common = "47"
+datafusion-execution = "47"
+datafusion-expr = "47"
+datafusion-functions = { version = "47", features = ["crypto_expressions"] }
+datafusion-functions-aggregate = "47"
+datafusion-sql = "47"
 derive-getters = "0.5.0"
 derive_builder = "0.20"
 futures = "0.3.31"
 getrandom = { version = "0.3.1", features = ["std"] }
 itertools = "0.14.0"
-object_store = { version = "0.11.2", features = ["aws", "gcp"] }
+object_store = { version = "0.12", features = ["aws", "gcp"] }
 once_map = "0.4"
-parquet = { version = "54", features = ["async", "object_store"] }
+parquet = { version = "55", features = ["async", "object_store"] }
 pin-project-lite = "0.2"
 serde = "^1.0"
 serde_derive = "^1.0"

Cargo.toml

@@ -37,15 +37,6 @@ impl UserDefinedLogicalNodeCore for PosDeltaNode {
         write!(f, "PosDelta")
     }
 
-    fn from_template(&self, exprs: &[Expr], inputs: &[LogicalPlan]) -> Self {
-        assert_eq!(inputs.len(), 1, "input size inconsistent");
-        assert_eq!(exprs.len(), 0, "expression size inconsistent");
-        Self {
-            input: Arc::new(inputs[0].clone()),
-            aliases: BTreeMap::new(),
-        }
-    }
-
     fn with_exprs_and_inputs(
         &self,
         exprs: Vec<Expr>,
@@ -117,15 +108,6 @@ impl UserDefinedLogicalNodeCore for NegDeltaNode {
         write!(f, "NegDelta")
     }
 
-    fn from_template(&self, exprs: &[Expr], inputs: &[LogicalPlan]) -> Self {
-        assert_eq!(inputs.len(), 1, "input size inconsistent");
-        assert_eq!(exprs.len(), 0, "expression size inconsistent");
-        Self {
-            input: Arc::new(inputs[0].clone()),
-            aliases: BTreeMap::new(),
-        }
-    }
-
     fn with_exprs_and_inputs(
         &self,
         _exprs: Vec<Expr>,

datafusion_iceberg/src/materialized_view/delta_queries/delta_node.rs

@@ -1,4 +1,3 @@
-use core::panic;
 use std::{
     fmt::{self, Debug},
     hash::Hash,
@@ -117,10 +116,6 @@ impl UserDefinedLogicalNodeCore for ForkNode {
         write!(f, "ForkNode")
     }
 
-    fn from_template(&self, _exprs: &[Expr], _inputs: &[LogicalPlan]) -> Self {
-        panic!("Creating fork node from template is not allowed");
-    }
-
     fn with_exprs_and_inputs(
         &self,
         exprs: Vec<Expr>,

datafusion_iceberg/src/materialized_view/delta_queries/fork_node.rs

@@ -2,8 +2,8 @@ use std::{fmt::Debug, hash::Hash, sync::Arc};
 
 use async_trait::async_trait;
 use datafusion_expr::{
-    ColumnarValue, CreateCatalogSchema, CreateView, DropCatalogSchema, DropTable, ScalarUDFImpl,
-    Signature, Volatility,
+    ColumnarValue, CreateCatalogSchema, CreateView, DropCatalogSchema, DropTable,
+    ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
 use itertools::Itertools;
 use regex::Regex;
@@ -796,10 +796,11 @@ impl ScalarUDFImpl for RefreshMaterializedView {
         Ok(DataType::Utf8)
     }
 
-    fn invoke(
+    fn invoke_with_args(
         &self,
-        args: &[datafusion_expr::ColumnarValue],
+        args: ScalarFunctionArgs,
     ) -> datafusion::error::Result<datafusion_expr::ColumnarValue> {
+        let args = args.args;
         let ColumnarValue::Scalar(ScalarValue::Utf8(Some(name))) = &args[0] else {
             return Err(DataFusionError::Execution(
                 "Refresh function only takes a scalar string input.".to_string(),
@@ -961,7 +962,7 @@ OPTIONS ('has_header' 'true');";
             .expect("Failed to execute query plan.");
 
         ctx.sql(
-            "INSERT INTO iceberg.public.orders (id, customer_id, product_id, order_date, quantity) VALUES 
+            "INSERT INTO iceberg.public.orders (id, customer_id, product_id, order_date, quantity) VALUES
                 (1, 1, 1, '2020-01-01', 1),
                 (2, 2, 1, '2020-01-01', 1),
                 (3, 3, 1, '2020-01-01', 3),

datafusion_iceberg/src/materialized_view/delta_queries/transform.rs

@@ -29,8 +29,8 @@ use datafusion::{
     scalar::ScalarValue,
 };
 use datafusion_expr::{
-    expr::ScalarFunction, BinaryExpr, ColumnarValue, Expr, ScalarUDF, ScalarUDFImpl, Signature,
-    TypeSignature, Volatility,
+    expr::ScalarFunction, BinaryExpr, ColumnarValue, Expr, ScalarFunctionArgs, ScalarUDF,
+    ScalarUDFImpl, Signature, TypeSignature, Volatility,
 };
 use iceberg_rust::{
     arrow::transform::transform_arrow,
@@ -387,7 +387,11 @@ impl ScalarUDFImpl for DateTransform {
         Ok(DataType::Int32)
     }
 
-    fn invoke(&self, args: &[ColumnarValue]) -> datafusion::error::Result<ColumnarValue> {
+    fn invoke_with_args(
+        &self,
+        args: ScalarFunctionArgs,
+    ) -> datafusion::error::Result<ColumnarValue> {
+        let args = args.args;
         let transform = &args[0];
         let array = &args[1];
         let ColumnarValue::Scalar(ScalarValue::Utf8(Some(transform))) = transform else {