feat: update to DataFusion 50, pyo3 24, pyo3-arrow 0.11 (#3749)

# Description
The description of the main changes of your pull request

# Related Issue(s)
- apache/datafusion#16799

# Documentation
As part of preparing for the DataFusion 50 release I am making a PR to
upgrade delta.rs

Signed-off-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: alamb

.github/workflows/build.yml

@@ -39,7 +39,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: default
-          toolchain: "1.85"
+          toolchain: "1.86"
           override: true
 
       - uses: Swatinem/rust-cache@v2
@@ -91,7 +91,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: default
-          toolchain: "1.85"
+          toolchain: "1.86"
           override: true
 
       - uses: Swatinem/rust-cache@v2
@@ -132,7 +132,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: default
-          toolchain: "1.85"
+          toolchain: "1.86"
           override: true
 
       - uses: Swatinem/rust-cache@v2
@@ -194,7 +194,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: default
-          toolchain: "1.85"
+          toolchain: "1.86"
           override: true
 
       - uses: Swatinem/rust-cache@v2
@@ -241,7 +241,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: default
-          toolchain: "1.85"
+          toolchain: "1.86"
           override: true
 
       - uses: Swatinem/rust-cache@v2
@@ -300,7 +300,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: default
-          toolchain: "1.85"
+          toolchain: "1.86"
           override: true
 
       - uses: Swatinem/rust-cache@v2

Cargo.toml

@@ -5,7 +5,7 @@ resolver = "2"
 
 [workspace.package]
 authors = ["Qingping Hou <dave2008713@gmail.com>"]
-rust-version = "1.85"
+rust-version = "1.86"
 keywords = ["deltalake", "delta", "datalake"]
 readme = "README.md"
 edition = "2021"
@@ -26,32 +26,32 @@ debug = true
 debug = "line-tables-only"
 
 [workspace.dependencies]
-delta_kernel = { version = "0.15.1", features = [
-    "arrow-55",
+delta_kernel = { version = "0.15.2", features = [
+    "arrow-56",
     "default-engine-rustls",
     "internal-api",
 ] }
 
 
 # arrow
-arrow = { version = "55.2.0" }
-arrow-arith = { version = "55.2.0" }
-arrow-array = { version = "55.2.0", features = ["chrono-tz"] }
-arrow-buffer = { version = "55.2.0" }
-arrow-cast = { version = "55.2.0" }
-arrow-ipc = { version = "55.2.0" }
-arrow-json = { version = "55.2.0" }
-arrow-ord = { version = "55.2.0" }
-arrow-row = { version = "55.2.0" }
-arrow-schema = { version = "55.2.0" }
-arrow-select = { version = "55.2.0" }
+arrow = { version = "56.0.0" }
+arrow-arith = { version = "56.0.0" }
+arrow-array = { version = "56.0.0", features = ["chrono-tz"] }
+arrow-buffer = { version = "56.0.0" }
+arrow-cast = { version = "56.0.0" }
+arrow-ipc = { version = "56.0.0" }
+arrow-json = { version = "56.0.0" }
+arrow-ord = { version = "56.0.0" }
+arrow-row = { version = "56.0.0" }
+arrow-schema = { version = "56.0.0" }
+arrow-select = { version = "56.0.0" }
 object_store = { version = "0.12.1" }
-parquet = { version = "55.2.0" }
+parquet = { version = "56.0.0" }
 
 # datafusion
-datafusion = "49.0.0"
-datafusion-ffi = "49.0.0"
-datafusion-proto = "49.0.0"
+datafusion = "50.0.0"
+datafusion-ffi = "50.0.0"
+datafusion-proto = "50.0.0"
 
 # serde
 serde = { version = "1.0.194", features = ["derive"] }
@@ -93,4 +93,4 @@ arro3 = "arro3"
 Arro3 = "Arro3"
 AKS = "AKS"
 # to avoid using 'type' as a field name.
-tpe = "tpe"
+tpe = "tpe"

crates/core/src/delta_datafusion/expr.rs

@@ -52,7 +52,7 @@ use crate::{DeltaResult, DeltaTableError};
 
 /// This struct is like Datafusion's MakeArray but ensures that `element` is used rather than `item
 /// as the field name within the list.
-#[derive(Debug)]
+#[derive(Debug, Hash, PartialEq, Eq)]
 struct MakeParquetArray {
     /// The actual upstream UDF, which we're just totally cheating and using
     actual: MakeArray,

crates/core/src/delta_datafusion/mod.rs

@@ -389,6 +389,8 @@ pub(crate) fn get_null_of_arrow_type(t: &ArrowDataType) -> DeltaResult<ScalarVal
         )),
         //Unsupported types...
         ArrowDataType::Float16
+        | ArrowDataType::Decimal32(_, _)
+        | ArrowDataType::Decimal64(_, _)
         | ArrowDataType::Decimal256(_, _)
         | ArrowDataType::Union(_, _)
         | ArrowDataType::LargeList(_)
@@ -2185,10 +2187,10 @@ mod tests {
         assert_eq!("a", small.iter().next().unwrap().unwrap());
 
         let expected = vec![
-            ObjectStoreOperation::GetRange(LocationType::Data, 4952..4960),
-            ObjectStoreOperation::GetRange(LocationType::Data, 2399..4952),
+            ObjectStoreOperation::GetRange(LocationType::Data, 957..965),
+            ObjectStoreOperation::GetRange(LocationType::Data, 326..957),
             #[expect(clippy::single_range_in_vec_init)]
-            ObjectStoreOperation::GetRanges(LocationType::Data, vec![4..58]),
+            ObjectStoreOperation::GetRanges(LocationType::Data, vec![4..46]),
         ];
         let mut actual = Vec::new();
         operations.recv_many(&mut actual, 3).await;

crates/core/src/kernel/scalars.rs

@@ -250,6 +250,8 @@ impl ScalarExt for Scalar {
                 ))
             }
             Float16
+            | Decimal32(_, _)
+            | Decimal64(_, _)
             | Decimal256(_, _)
             | List(_)
             | LargeList(_)

crates/core/src/operations/optimize.rs

@@ -1111,7 +1111,7 @@ pub(super) mod zorder {
         }
 
         // DataFusion UDF impl for zorder_key
-        #[derive(Debug)]
+        #[derive(Debug, Hash, PartialEq, Eq)]
         pub struct ZOrderUDF;
 
         impl ScalarUDFImpl for ZOrderUDF {

python/Cargo.toml

@@ -20,7 +20,7 @@ doc = false
 [dependencies]
 delta_kernel.workspace = true
 
-pyo3-arrow = { version = "0.9.0", default-features = false }
+pyo3-arrow = { version = "0.11.0", default-features = false }
 
 # arrow
 arrow-schema = { workspace = true, features = ["serde"] }
@@ -63,7 +63,7 @@ jemallocator = { version = "0.5", features = [
 jemallocator = { version = "0.5", features = ["disable_initial_exec_tls"] }
 
 [dependencies.pyo3]
-version = "0.24.0"
+version = "0.25.1"
 features = ["extension-module", "abi3", "abi3-py39"]
 
 [dependencies.deltalake]

python/src/datafusion.rs

@@ -175,6 +175,10 @@ mod tests {
     }
 
     impl LazyBatchGenerator for TestBatchGenerator {
+        fn as_any(&self) -> &dyn Any {
+            self
+        }
+
         fn generate_next_batch(&mut self) -> DataFusionResult<Option<RecordBatch>> {
             if self.current_index < self.data.len() {
                 let batch = self.data[self.current_index].clone();

python/src/writer.rs

@@ -1,4 +1,6 @@
 //! This module contains helper functions to create a LazyTableProvider from an ArrowArrayStreamReader
+
+use std::any::Any;
 use std::fmt::{self};
 use std::sync::{Arc, Mutex};
 
@@ -66,6 +68,10 @@ impl ArrowStreamBatchGenerator {
 }
 
 impl LazyBatchGenerator for ArrowStreamBatchGenerator {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
     fn generate_next_batch(
         &mut self,
     ) -> deltalake::datafusion::error::Result<Option<deltalake::arrow::array::RecordBatch>> {