chore: upgrade to DataFusion main + arrow/parquet 59.1.0 [WIP]

native/Cargo.toml

@@ -34,14 +34,14 @@ edition = "2021"
 rust-version = "1.88"
 
 [workspace.dependencies]
-arrow = { version = "58.4.0", features = ["prettyprint", "ffi", "chrono-tz"] }
+arrow = { version = "59.1.0", features = ["prettyprint", "ffi", "chrono-tz"] }
 async-trait = { version = "0.1" }
 bytes = { version = "1.11.1" }
-parquet = { version = "58.4.0", default-features = false, features = ["experimental"] }
-datafusion = { version = "54.1.0", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] }
-datafusion-datasource = { version = "54.1.0" }
-datafusion-physical-expr-adapter = { version = "54.1.0" }
-datafusion-spark = { version = "54.1.0", features = ["core"] }
+parquet = { version = "59.1.0", default-features = false, features = ["experimental"] }
+datafusion = { git = "https://github.com/apache/datafusion", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] }
+datafusion-datasource = { git = "https://github.com/apache/datafusion", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904" }
+datafusion-physical-expr-adapter = { git = "https://github.com/apache/datafusion", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904" }
+datafusion-spark = { git = "https://github.com/apache/datafusion", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904", features = ["core"] }
 datafusion-comet-spark-expr = { path = "spark-expr" }
 datafusion-comet-common = { path = "common" }
 datafusion-comet-jni-bridge = { path = "jni-bridge" }

native/core/Cargo.toml

@@ -72,8 +72,8 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-
 object_store_opendal = { version = "0.57.0", optional = true }
 hdfs-sys = {version = "0.3", optional = true, features = ["hdfs_3_3"]}
 opendal = { version = "0.57.0", optional = true, features = ["services-hdfs"] }
-iceberg = { workspace = true }
-iceberg-storage-opendal = { workspace = true }
+iceberg = { workspace = true, optional = true }
+iceberg-storage-opendal = { workspace = true, optional = true }
 reqsign-core = { workspace = true }
 serde_json = "1.0"
 uuid = "1.23.3"
@@ -91,12 +91,17 @@ jni = { version = "0.22.4", features = ["invocation"] }
 lazy_static = "1.4"
 assertables = "10"
 hex = "0.4.3"
-datafusion-functions-nested = { version = "54.1.0" }
+datafusion-functions-nested = { git = "https://github.com/apache/datafusion", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904" }
 
 [features]
 backtrace = ["datafusion/backtrace"]
 default = ["hdfs-opendal"]
 hdfs-opendal = ["opendal", "object_store_opendal", "hdfs-sys"]
+# The native Iceberg scan. Off by default while this branch tracks DataFusion main:
+# iceberg-rust is still on arrow 58, so its `RecordBatch` does not unify with the
+# arrow 59 that DataFusion main requires. Re-enable (and make this a default
+# feature again) once iceberg-rust upgrades to arrow 59.
+iceberg-scan = ["iceberg", "iceberg-storage-opendal"]
 jemalloc = ["tikv-jemallocator", "tikv-jemalloc-ctl"]
 
 # exclude optional packages from cargo machete verifications

native/core/src/cloud/s3/credential_bridge.rs

@@ -22,31 +22,39 @@
 use crate::execution::operators::ExecutionError;
 use crate::jvm_bridge::{jni_new_global_ref, jni_static_call, JVMClasses};
 use async_trait::async_trait;
+#[cfg(feature = "iceberg-scan")]
 use iceberg_storage_opendal::AwsCredential as IcebergAwsCredential;
 use jni::objects::{Global, JFieldID, JObject, JString, JValue};
 use jni::signature::{Primitive, ReturnType};
 use jni::strings::JNIString;
 use jni::sys::jint;
+#[cfg(feature = "iceberg-scan")]
 use log::warn;
 use object_store::aws::AwsCredential;
 use object_store::CredentialProvider;
+#[cfg(feature = "iceberg-scan")]
 use once_cell::sync::OnceCell;
+#[cfg(feature = "iceberg-scan")]
 use reqsign_core::time::Timestamp;
+#[cfg(feature = "iceberg-scan")]
 use reqsign_core::{
     Context, Error as ReqsignError, ErrorKind as ReqsignErrorKind,
     ProvideCredential as IcebergProvideCredential,
 };
 use std::collections::HashMap;
 use std::fmt;
 use std::sync::Arc;
+#[cfg(feature = "iceberg-scan")]
 use std::time::Duration;
 
 /// Cap on opendal's credential cache when the provider does not report an expiry. Prevents the
 /// executor from holding a stale credential for the entire job lifetime.
+#[cfg(feature = "iceberg-scan")]
 const DEFAULT_EXPIRY_WHEN_UNKNOWN: Duration = Duration::from_secs(300);
 
 /// Once-per-process latch for the "missing expiry" warning. Bridges are per-scan, so a per-bridge
 /// latch would re-log on every scan.
+#[cfg(feature = "iceberg-scan")]
 static WARNED_MISSING_EXPIRY: OnceCell<()> = OnceCell::new();
 
 /// Access intent forwarded to the Java SPI. Ordinal must match the JVM `CometS3AccessMode` enum.
@@ -269,7 +277,9 @@ struct RawCredentials {
     access_key_id: String,
     secret_access_key: String,
     session_token: Option<String>,
-    /// Absolute expiry. `0` means the provider did not report one.
+    /// Absolute expiry. `0` means the provider did not report one. Only consumed by the
+    /// Iceberg credential path.
+    #[cfg_attr(not(feature = "iceberg-scan"), allow(dead_code))]
     expiration_epoch_millis: i64,
 }
 
@@ -290,6 +300,7 @@ impl CredentialProvider for CometS3CredentialBridge {
     }
 }
 
+#[cfg(feature = "iceberg-scan")]
 impl IcebergProvideCredential for CometS3CredentialBridge {
     type Credential = IcebergAwsCredential;
 

native/core/src/execution/columnar_to_row.rs

@@ -2499,11 +2499,14 @@ mod tests {
         let schema = vec![DataType::FixedSizeBinary(3)];
         let mut ctx = ColumnarToRowContext::new(schema, 100);
 
-        let array: ArrayRef = Arc::new(FixedSizeBinaryArray::from(vec![
-            Some(&[1u8, 2, 3][..]),
-            Some(&[4u8, 5, 6][..]),
-            None, // Test null handling
-        ]));
+        let array: ArrayRef = Arc::new(
+            FixedSizeBinaryArray::try_from(vec![
+                Some(&[1u8, 2, 3][..]),
+                Some(&[4u8, 5, 6][..]),
+                None, // Test null handling
+            ])
+            .unwrap(),
+        );
         let arrays = vec![array];
 
         let (ptr, offsets, lengths) = ctx.convert(&arrays, 3).unwrap();

native/core/src/execution/merge_as_partial.rs

@@ -207,23 +207,22 @@ impl GroupsAccumulator for MergeAsPartialGroupsAccumulator {
         &mut self,
         values: &[ArrayRef],
         group_indices: &[usize],
-        opt_filter: Option<&BooleanArray>,
+        _opt_filter: Option<&BooleanArray>,
         total_num_groups: usize,
     ) -> Result<()> {
         // Redirect update to merge — this is the key trick.
         self.inner
-            .merge_batch(values, group_indices, opt_filter, total_num_groups)
+            .merge_batch(values, group_indices, total_num_groups)
     }
 
     fn merge_batch(
         &mut self,
         values: &[ArrayRef],
         group_indices: &[usize],
-        opt_filter: Option<&BooleanArray>,
         total_num_groups: usize,
     ) -> Result<()> {
         self.inner
-            .merge_batch(values, group_indices, opt_filter, total_num_groups)
+            .merge_batch(values, group_indices, total_num_groups)
     }
 
     fn evaluate(&mut self, emit_to: EmitTo) -> Result<ArrayRef> {
@@ -234,6 +233,18 @@ impl GroupsAccumulator for MergeAsPartialGroupsAccumulator {
         self.inner.state(emit_to)
     }
 
+    fn convert_to_state(
+        &self,
+        values: &[ArrayRef],
+        _opt_filter: Option<&BooleanArray>,
+    ) -> Result<Vec<ArrayRef>> {
+        // The input to this accumulator is already the inner accumulator's intermediate
+        // state (that is the point of redirecting update to merge), so the state for a
+        // group of one row is that row itself. The filter is ignored here for the same
+        // reason it is ignored in `update_batch`.
+        Ok(values.to_vec())
+    }
+
     fn size(&self) -> usize {
         self.inner.size()
     }

native/core/src/execution/operators/copy.rs

@@ -37,7 +37,9 @@ pub(crate) fn copy_array(array: &dyn Array) -> ArrayRef {
 
     let mut mutable = MutableArrayData::new(vec![&data], false, capacity);
 
-    mutable.extend(0, 0, capacity);
+    mutable
+        .try_extend(0, 0, capacity)
+        .expect("copy_array: extend within existing array cannot overflow");
 
     if matches!(array.data_type(), DataType::Dictionary(_, _)) {
         let copied_dict = make_array(mutable.freeze());
@@ -50,7 +52,9 @@ pub(crate) fn copy_array(array: &dyn Array) -> ArrayRef {
                 let data = values.to_data();
 
                 let mut mutable = MutableArrayData::new(vec![&data], false, values.len());
-                mutable.extend(0, 0, values.len());
+                mutable
+                    .try_extend(0, 0, values.len())
+                    .expect("copy_array: extend within existing array cannot overflow");
 
                 let copied_dict = ref_copied_dict.with_values(make_array(mutable.freeze()));
                 Arc::new(copied_dict)

native/core/src/execution/operators/mod.rs

@@ -21,13 +21,15 @@ pub use crate::errors::ExecutionError;
 
 pub use aligned_stream_reader::*;
 pub use copy::*;
+#[cfg(feature = "iceberg-scan")]
 pub use iceberg_scan::*;
 pub use scan::*;
 
 mod aligned_stream_reader;
 mod copy;
 mod expand;
 pub use expand::ExpandExec;
+#[cfg(feature = "iceberg-scan")]
 mod iceberg_scan;
 mod parquet_writer;
 pub use parquet_writer::{ParquetCompression, ParquetWriterExec};

native/core/src/execution/planner.rs

@@ -23,6 +23,7 @@ pub mod operator_registry;
 
 use crate::execution::operators::init_csv_datasource_exec;
 use crate::execution::operators::AlignedArrowStreamReader;
+#[cfg(feature = "iceberg-scan")]
 use crate::execution::operators::IcebergScanExec;
 use crate::execution::{
     expressions::list_empty_to_null::ListEmptyToNullExpr,
@@ -80,6 +81,7 @@ use datafusion_comet_spark_expr::{
     SparkBloomFilterVersion, SparkPercentile, SumInteger, ToCsv,
 };
 use datafusion_spark::function::aggregate::collect::{SparkCollectList, SparkCollectSet};
+#[cfg(feature = "iceberg-scan")]
 use iceberg::expr::Bind;
 
 use crate::execution::operators::ExecutionError::GeneralError;
@@ -1700,6 +1702,14 @@ impl PhysicalPlanner {
                     Arc::new(SparkPlan::new(spark_plan.plan_id, Arc::new(scan), vec![])),
                 ))
             }
+            #[cfg(not(feature = "iceberg-scan"))]
+            OpStruct::IcebergScan(_) => Err(GeneralError(
+                "Native Iceberg scan is not available in this build: the `iceberg-scan` \
+                 feature is disabled because iceberg-rust does not yet support arrow 59. \
+                 Disable spark.comet.scan.impl=native_iceberg_compat to fall back to Spark."
+                    .into(),
+            )),
+            #[cfg(feature = "iceberg-scan")]
             OpStruct::IcebergScan(scan) => {
                 // Extract common data and single partition's file tasks
                 // Per-partition injection happens in Scala before sending to native
@@ -2050,10 +2060,7 @@ impl PhysicalPlanner {
                     depth: 1,
                 });
 
-                let unnest_options = UnnestOptions {
-                    preserve_nulls: explode.outer,
-                    recursions: vec![],
-                };
+                let unnest_options = UnnestOptions::new().with_preserve_nulls(explode.outer);
 
                 let unnest_exec = Arc::new(UnnestExec::new(
                     project_exec,
@@ -3742,6 +3749,7 @@ fn align_shuffle_writer_input(
         .map_err(|e| ExecutionError::DataFusionError(e.to_string()))
 }
 
+#[cfg(feature = "iceberg-scan")]
 /// Converts a protobuf PartitionValue to an iceberg Literal.
 ///
 fn partition_value_to_literal(
@@ -3795,6 +3803,7 @@ fn partition_value_to_literal(
     Ok(Some(literal))
 }
 
+#[cfg(feature = "iceberg-scan")]
 /// Decodes an unscaled decimal (two's-complement big-endian) into i128.
 fn decimal_bytes_to_i128(bytes: &[u8]) -> Result<i128, ExecutionError> {
     if bytes.len() > 16 {
@@ -3815,6 +3824,7 @@ fn decimal_bytes_to_i128(bytes: &[u8]) -> Result<i128, ExecutionError> {
     Ok(i128::from_be_bytes(buf))
 }
 
+#[cfg(feature = "iceberg-scan")]
 /// Converts a protobuf PartitionData to an iceberg Struct.
 ///
 /// Uses the existing Struct::from_iter() API from iceberg-rust to construct the struct
@@ -3832,6 +3842,7 @@ fn partition_data_to_struct(
     Ok(iceberg::spec::Struct::from_iter(literals))
 }
 
+#[cfg(feature = "iceberg-scan")]
 /// Converts protobuf FileScanTasks from Scala into iceberg-rust FileScanTask objects.
 ///
 /// Each task contains a residual predicate that is used for row-group level filtering
@@ -4391,6 +4402,7 @@ fn literal_to_array_ref(
 // Iceberg Residual Predicate Conversion
 // ============================================================================
 
+#[cfg(feature = "iceberg-scan")]
 /// Converts a serialized Iceberg residual predicate into an iceberg-rust `Predicate` for row-group
 /// pruning. This is only a pruning hint -- the post-scan CometFilter enforces correctness -- so any
 /// node or literal that cannot be represented degrades to `None` (no pushdown) rather than an
@@ -4462,6 +4474,7 @@ fn iceberg_predicate_to_predicate(
     }
 }
 
+#[cfg(feature = "iceberg-scan")]
 /// Combines the two children of a logical residual node (And/Or), returning `None` unless both
 /// converted. A missing child is not expected: the Scala serde emits a logical node only when both
 /// children convert, and Rust decodes every node it emits, so both sides always convert for a
@@ -4488,6 +4501,7 @@ fn combine_logical(
     }
 }
 
+#[cfg(feature = "iceberg-scan")]
 /// Converts a serialized `IcebergLiteral` into an iceberg-rust `Datum` for predicate pushdown.
 /// Returns `None` for null and for byte-array-backed types (decimal/uuid/fixed/binary), which
 /// iceberg-rust cannot use in the page index yet; the driver does not emit those for predicates,
@@ -4560,6 +4574,7 @@ mod tests {
 
     use crate::execution::operators::ExecutionError;
     use crate::execution::planner::literal_to_array_ref;
+    #[cfg(feature = "iceberg-scan")]
     use crate::execution::planner::parse_file_scan_tasks_from_common;
     use crate::parquet::parquet_support::SparkParquetOptions;
     use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory;
@@ -5645,6 +5660,7 @@ mod tests {
         });
     }
 
+    #[cfg(feature = "iceberg-scan")]
     #[test]
     fn test_metadata_field_id_constants_match_iceberg_rust() {
         // These constants are duplicated in Scala (CometIcebergNativeScan.MetadataFieldIds)
@@ -5662,6 +5678,7 @@ mod tests {
         );
     }
 
+    #[cfg(feature = "iceberg-scan")]
     #[test]
     fn test_unified_partition_type_merges_specs_by_descending_spec_id() {
         use iceberg::spec::{NestedField, PartitionSpec, PrimitiveType, Type};
@@ -5768,6 +5785,7 @@ mod tests {
         assert_eq!(fields[1].name, "category");
     }
 
+    #[cfg(feature = "iceberg-scan")]
     #[test]
     fn test_unified_partition_type_tolerates_unparseable_spec() {
         // Regression for TestForwardCompatibility.testSparkCanReadUnknownTransform: a spec that

native/core/src/parquet/eager_page_index_reader_factory.rs

@@ -68,11 +68,11 @@ use std::sync::Arc;
 #[derive(Debug)]
 pub struct EagerPageIndexReaderFactory {
     store: Arc<dyn ObjectStore>,
-    metadata_cache: Arc<dyn FileMetadataCache>,
+    metadata_cache: Arc<FileMetadataCache>,
 }
 
 impl EagerPageIndexReaderFactory {
-    pub fn new(store: Arc<dyn ObjectStore>, metadata_cache: Arc<dyn FileMetadataCache>) -> Self {
+    pub fn new(store: Arc<dyn ObjectStore>, metadata_cache: Arc<FileMetadataCache>) -> Self {
         Self {
             store,
             metadata_cache,
@@ -118,7 +118,7 @@ struct EagerPageIndexReader {
     store: Arc<dyn ObjectStore>,
     inner: ParquetObjectReader,
     partitioned_file: PartitionedFile,
-    metadata_cache: Arc<dyn FileMetadataCache>,
+    metadata_cache: Arc<FileMetadataCache>,
     metadata_size_hint: Option<usize>,
 }
 

native/core/src/parquet/parquet_exec.rs

@@ -20,7 +20,7 @@ use crate::parquet::eager_page_index_reader_factory::EagerPageIndexReaderFactory
 use crate::parquet::encryption_support::{CometEncryptionConfig, ENCRYPTION_FACTORY_ID};
 use crate::parquet::parquet_support::SparkParquetOptions;
 use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory;
-use arrow::datatypes::{Field, SchemaRef};
+use arrow::datatypes::{Field, FieldRef, SchemaRef};
 use datafusion::config::{ParquetOptions, TableParquetOptions};
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::datasource::physical_plan::{
@@ -128,13 +128,14 @@ pub(crate) fn init_datasource_exec(
         }
         _ => (Arc::clone(&required_schema), None),
     };
-    let partition_fields: Vec<_> = partition_schema
+    let partition_fields: Vec<FieldRef> = partition_schema
         .iter()
         .flat_map(|s| s.fields().iter())
-        .map(|f| Arc::new(Field::new(f.name(), f.data_type().clone(), f.is_nullable())) as _)
+        .map(|f| Arc::new(Field::new(f.name(), f.data_type().clone(), f.is_nullable())))
         .collect();
-    let table_schema =
-        TableSchema::from_file_schema(base_schema).with_table_partition_cols(partition_fields);
+    let table_schema = TableSchema::builder(base_schema)
+        .with_table_partition_cols(partition_fields)
+        .build();
 
     let mut parquet_source = ParquetSource::new(table_schema)
         .with_table_parquet_options(table_parquet_options)