deps: [DO NOT MERGE] test DataFusion 55.0, Arrow/Parqet 59.1

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.git", rev = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] }
+datafusion-datasource = { git = "https://github.com/apache/datafusion.git", rev = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41" }
+datafusion-physical-expr-adapter = { git = "https://github.com/apache/datafusion.git", rev = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41" }
+datafusion-spark = { git = "https://github.com/apache/datafusion.git", rev = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41", features = ["core"] }
 datafusion-comet-spark-expr = { path = "spark-expr" }
 datafusion-comet-common = { path = "common" }
 datafusion-comet-jni-bridge = { path = "jni-bridge" }
@@ -58,8 +58,8 @@ object_store = { version = "0.13.2", features = ["gcp", "azure", "aws", "http"]
 url = "2.2"
 aws-config = "1.8.18"
 aws-credential-types = "1.2.13"
-iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/mbutrovich/iceberg-rust", rev = "07de8fc8d650f7439b251d57a55613f7de693eb8" }
+iceberg-storage-opendal = { git = "https://github.com/mbutrovich/iceberg-rust", rev = "07de8fc8d650f7439b251d57a55613f7de693eb8", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
 reqsign-core = "3"
 
 [profile.release]

native/core/Cargo.toml

@@ -69,9 +69,9 @@ aws-config = { workspace = true }
 aws-credential-types = { workspace = true }
 parking_lot = "0.12.5"
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots", "http2"] }
-object_store_opendal = { version = "0.57.0", optional = true }
+object_store_opendal = { version = "0.58.0", optional = true }
 hdfs-sys = {version = "0.3", optional = true, features = ["hdfs_3_3"]}
-opendal = { version = "0.57.0", optional = true, features = ["services-hdfs"] }
+opendal = { version = "0.58.1", optional = true, features = ["services-hdfs"] }
 iceberg = { workspace = true }
 iceberg-storage-opendal = { workspace = true }
 reqsign-core = { workspace = true }
@@ -91,7 +91,7 @@ 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.git", rev = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41" }
 
 [features]
 backtrace = ["datafusion/backtrace"]

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

@@ -210,20 +210,30 @@ impl GroupsAccumulator for MergeAsPartialGroupsAccumulator {
         opt_filter: Option<&BooleanArray>,
         total_num_groups: usize,
     ) -> Result<()> {
-        // Redirect update to merge — this is the key trick.
+        // Redirect update to merge — this is the key trick. Spark's PartialMerge mode
+        // never applies a filter (filters apply once, at the Partial stage), so there's
+        // nothing to forward into merge_batch, which no longer accepts one.
+        debug_assert!(opt_filter.is_none());
         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 convert_to_state(
+        &self,
+        _values: &[ArrayRef],
+        _opt_filter: Option<&BooleanArray>,
+    ) -> Result<Vec<ArrayRef>> {
+        datafusion::common::not_impl_err!("Input batch conversion to state not implemented")
     }
 
     fn evaluate(&mut self, emit_to: EmitTo) -> Result<ArrayRef> {

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("extend failed due to offset 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("extend failed due to offset overflow");
 
                 let copied_dict = ref_copied_dict.with_values(make_array(mutable.freeze()));
                 Arc::new(copied_dict)

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

@@ -651,6 +651,8 @@ mod tests {
             start: 0,
             length: 0,
             record_count: None,
+            first_row_id: None,
+            data_sequence_number: None,
             data_file_path: "data.parquet".to_string(),
             data_file_format: DataFileFormat::Parquet,
             schema: Arc::new(Schema::builder().build().unwrap()),

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

@@ -92,7 +92,7 @@ enum ParquetWriter {
     /// The opendal::Writer is created lazily on first write
     #[cfg(feature = "hdfs-opendal")]
     Remote(
-        ArrowWriter<Cursor<Vec<u8>>>,
+        Box<ArrowWriter<Cursor<Vec<u8>>>>,
         Option<opendal::Writer>,
         Operator,
         String,
@@ -338,7 +338,7 @@ impl ParquetWriterExec {
                 // HDFS writer will be created lazily on first write
                 // Use the path from prepare_object_store_with_configs
                 Ok(ParquetWriter::Remote(
-                    arrow_parquet_buffer_writer,
+                    Box::new(arrow_parquet_buffer_writer),
                     None,
                     op,
                     object_store_path.to_string(),
@@ -656,11 +656,9 @@ mod tests {
 
         // Create OpenDAL HDFS operator
         let builder = Hdfs::default().name_node(namenode);
-        let op = Operator::new(builder)
-            .map_err(|e| {
-                DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e))
-            })?
-            .finish();
+        let op = Operator::new(builder).map_err(|e| {
+            DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e))
+        })?;
 
         let mut hdfs_writer = op.writer(output_path).await.map_err(|e| {
             DataFusionError::Execution(format!("Failed to create HDFS writer: {}", e))
@@ -707,11 +705,9 @@ mod tests {
 
         // Create OpenDAL HDFS operator
         let builder = Hdfs::default().name_node(namenode);
-        let op = Operator::new(builder)
-            .map_err(|e| {
-                DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e))
-            })?
-            .finish();
+        let op = Operator::new(builder).map_err(|e| {
+            DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e))
+        })?;
 
         // Create a single HDFS writer for the entire file
         let mut hdfs_writer = op.writer(output_path).await.map_err(|e| {

native/core/src/execution/planner.rs

@@ -110,7 +110,7 @@ use arrow::array::{
 use arrow::buffer::{BooleanBuffer, NullBuffer, OffsetBuffer};
 use arrow::row::{OwnedRow, RowConverter, SortField};
 use datafusion::common::utils::SingleRowListArrayBuilder;
-use datafusion::common::UnnestOptions;
+use datafusion::common::{NullHandling, UnnestOptions};
 use datafusion::physical_plan::filter::FilterExec;
 use datafusion::physical_plan::joins::NestedLoopJoinExec;
 use datafusion::physical_plan::limit::GlobalLimitExec;
@@ -1995,7 +1995,11 @@ impl PhysicalPlanner {
                 });
 
                 let unnest_options = UnnestOptions {
-                    preserve_nulls: explode.outer,
+                    null_handling: if explode.outer {
+                        NullHandling::Preserve
+                    } else {
+                        NullHandling::Drop
+                    },
                     recursions: vec![],
                 };
 
@@ -4065,6 +4069,8 @@ fn parse_file_scan_tasks_from_common(
                 start: proto_task.start,
                 length: proto_task.length,
                 record_count: proto_task.record_count,
+                first_row_id: None,
+                data_sequence_number: None,
                 data_file_format,
                 schema: schema_ref,
                 project_field_ids,

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)

native/core/src/parquet/parquet_support.rs

@@ -445,12 +445,10 @@ pub(crate) fn create_hdfs_operator(url: &Url) -> Result<opendal::Operator, objec
     let name_node = get_name_node_uri(url)?;
     let builder = opendal::services::Hdfs::default().name_node(&name_node);
 
-    opendal::Operator::new(builder)
-        .map_err(|error| object_store::Error::Generic {
-            store: "hdfs-opendal",
-            source: error.into(),
-        })
-        .map(|op| op.finish())
+    opendal::Operator::new(builder).map_err(|error| object_store::Error::Generic {
+        store: "hdfs-opendal",
+        source: error.into(),
+    })
 }
 
 // Creates an HDFS object store from a URL using OpenDAL

native/shuffle/src/partitioners/traits.rs

@@ -19,7 +19,7 @@ use arrow::record_batch::RecordBatch;
 use datafusion::common::Result;
 
 #[async_trait::async_trait]
-pub(crate) trait ShufflePartitioner: Send + Sync {
+pub(crate) trait ShufflePartitioner: Send {
     /// Insert a batch into the partitioner
     async fn insert_batch(&mut self, batch: RecordBatch) -> Result<()>;
     /// Write shuffle data and shuffle index file to disk

native/shuffle/src/writers/local/spill.rs

@@ -19,21 +19,21 @@ use crate::metrics::ShufflePartitionerMetrics;
 use crate::writers::BufBatchWriter;
 use crate::ShuffleBlockWriter;
 use arrow::record_batch::RecordBatch;
-use datafusion::common::DataFusionError;
-use datafusion::execution::disk_manager::RefCountedTempFile;
 use datafusion::execution::runtime_env::RuntimeEnv;
-use std::fs::{File, OpenOptions};
+use datafusion::execution::SpillFile as DfSpillFile;
+use datafusion::execution::SpillWriter as DfSpillWriter;
+use std::sync::Arc;
 
-struct SpillFile {
-    temp_file: RefCountedTempFile,
-    file: File,
+struct ActiveSpillFile {
+    temp_file: Arc<dyn DfSpillFile>,
+    writer: Box<dyn DfSpillWriter>,
 }
 
 pub(crate) struct SpillWriter {
     shuffle_block_writer: ShuffleBlockWriter,
     write_buffer_size: usize,
     batch_size: usize,
-    spill_file: Option<SpillFile>,
+    spill_file: Option<ActiveSpillFile>,
 }
 
 impl SpillWriter {
@@ -62,7 +62,7 @@ impl SpillWriter {
             let total_bytes_written = {
                 let mut buf_batch_writer = BufBatchWriter::new(
                     &mut self.shuffle_block_writer,
-                    &mut self.spill_file.as_mut().unwrap().file,
+                    &mut self.spill_file.as_mut().unwrap().writer,
                     self.write_buffer_size,
                     self.batch_size,
                 );
@@ -90,29 +90,19 @@ impl SpillWriter {
     ) -> datafusion::common::Result<()> {
         if self.spill_file.is_none() {
             // Spill file is not yet created, create it
-            let spill_file = runtime
+            let temp_file = runtime
                 .disk_manager
                 .create_tmp_file("shuffle writer spill")?;
-            let spill_data = OpenOptions::new()
-                .write(true)
-                .create(true)
-                .truncate(true)
-                .open(spill_file.path())
-                .map_err(|e| {
-                    DataFusionError::Execution(format!("Error occurred while spilling {e}"))
-                })?;
-            self.spill_file = Some(SpillFile {
-                temp_file: spill_file,
-                file: spill_data,
-            });
+            let writer = temp_file.open_writer()?;
+            self.spill_file = Some(ActiveSpillFile { temp_file, writer });
         }
         Ok(())
     }
 
     pub(crate) fn path(&self) -> Option<&std::path::Path> {
         self.spill_file
             .as_ref()
-            .map(|spill_file| spill_file.temp_file.path())
+            .and_then(|spill_file| spill_file.temp_file.path())
     }
 
     #[cfg(test)]

native/shuffle/src/writers/partition_writer.rs

@@ -32,7 +32,7 @@ use arrow::record_batch::RecordBatch;
 /// ascending id order, then a single [`finish_all`](PartitionWriter::finish_all).
 ///
 /// [`LocalPartitionWriter`]: crate::writers::local::local_partition_writer::LocalPartitionWriter
-pub(crate) trait PartitionWriter: Send + Sync {
+pub(crate) trait PartitionWriter: Send {
     /// Stages the batches from `iter` for partition `pid` without finalizing it.
     ///
     /// Used to stream single-partition output and to stage multi-partition

native/shuffle/src/writers/shuffle_block_writer.rs

@@ -18,7 +18,7 @@
 use arrow::array::RecordBatch;
 use arrow::datatypes::{DataType, Schema, SchemaRef};
 use arrow::ipc::writer::{
-    write_message, CompressionContext, DictionaryTracker, IpcDataGenerator, IpcWriteOptions,
+    write_message, DictionaryTracker, IpcDataGenerator, IpcWriteContext, IpcWriteOptions,
     StreamWriter,
 };
 use arrow::ipc::MetadataVersion;
@@ -159,7 +159,7 @@ impl ShuffleBlockWriter {
         // Fast path: reuse the pre-encoded schema message and write the record batch manually.
         let data_gen = IpcDataGenerator::default();
         let mut dictionary_tracker = DictionaryTracker::new(true);
-        let mut compression_context = CompressionContext::default();
+        let mut compression_context = IpcWriteContext::default();
         let (encoded_dictionaries, encoded_batch) = data_gen.encode(
             batch,
             &mut dictionary_tracker,