Changes to upgrade datafusion and arrow to latest versions (opensearch-project#21590)

Signed-off-by: bharath-techie <bharath78910@gmail.com>

Author: bharath-techie

sandbox/libs/dataformat-native/rust/Cargo.toml

@@ -16,62 +16,62 @@ members = [
 
 [workspace.dependencies]
 # Arrow / Parquet
-arrow = { version = "57.3.0", features = ["ffi"] }
-arrow-array = "57.3.0"
-arrow-ipc = "57.3.0"
-arrow-schema = "57.3.0"
-arrow-buffer = "57.3.0"
-parquet = "57.3.0"
+arrow = { version = "=58.2.0", features = ["ffi"] }
+arrow-array = "=58.2.0"
+arrow-ipc = "=58.2.0"
+arrow-schema = "=58.2.0"
+arrow-buffer = "=58.2.0"
+parquet = "=58.2.0"
 
 # DataFusion
-datafusion = "52.1.0"
-datafusion-expr = "52.1.0"
-datafusion-datasource = "52.1.0"
-datafusion-common = "52.1.0"
-datafusion-execution = "52.1.0"
-datafusion-physical-expr = "52.1.0"
-datafusion-substrait = "52.1.0"
+datafusion = "=53.1.0"
+datafusion-expr = "=53.1.0"
+datafusion-datasource = "=53.1.0"
+datafusion-common = "=53.1.0"
+datafusion-execution = "=53.1.0"
+datafusion-physical-expr = "=53.1.0"
+datafusion-substrait = "=53.1.0"
 
 # Async
-tokio = { version = "1.0", features = ["full"] }
-tokio-util = "0.7"
-futures = "0.3"
-tokio-stream = "0.1.17"
+tokio = { version = "=1.51.0", features = ["full"] }
+tokio-util = "=0.7.18"
+futures = "=0.3.32"
+tokio-stream = "=0.1.18"
 
 # Serialization
-prost = "0.14"
-substrait = "=0.62.0"
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0"
+prost = "=0.14.3"
+substrait = "=0.62.2"
+serde = { version = "=1.0.228", features = ["derive"] }
+serde_json = "=1.0.149"
 
 # Logging
-log = "0.4"
+log = "=0.4.29"
 
 # Allocator
 # disable_initial_exec_tls: Required because this library is loaded at runtime via dlopen/JVM FFM.
 # Without it, jemalloc uses initial-exec TLS which fails on aarch64 Linux with:
 #   "cannot allocate memory in static TLS block"
 # The feature switches to global-dynamic TLS model, compatible with runtime loading.
-tikv-jemallocator = { version = "0.6", features = ["disable_initial_exec_tls"] }
-tikv-jemalloc-ctl = { version = "0.6", features = ["stats"] }
+tikv-jemallocator = { version = "=0.6.1", features = ["disable_initial_exec_tls"] }
+tikv-jemalloc-ctl = { version = "=0.6.1", features = ["stats"] }
 
 # Misc
-dashmap = "5.5"
-num_cpus = "1.16"
-object_store = "0.12.5"
-url = "2.0"
-tempfile = "3.0"
-chrono = "0.4"
-once_cell = "1.21.3"
-crc32fast = "1.4"
-parking_lot = "0.12.5"
-lazy_static = "1.4.0"
-rayon = "1.10"
-thiserror = "1.0"
-async-trait = "0.1"
-bytes = "1"
-criterion = { version = "0.5", features = ["async_tokio"] }
-tokio-metrics = { version = "0.5", features = ["rt"] }
+dashmap = "=5.5.3"
+num_cpus = "=1.17.0"
+object_store = "=0.13.2"
+url = "=2.5.8"
+tempfile = "=3.27.0"
+chrono = "=0.4.44"
+once_cell = "=1.21.4"
+crc32fast = "=1.5.0"
+parking_lot = "=0.12.5"
+lazy_static = "=1.5.0"
+rayon = "=1.11.0"
+thiserror = "=1.0.69"
+async-trait = "=0.1.89"
+bytes = "=1.11.1"
+criterion = { version = "=0.5.1", features = ["async_tokio"] }
+tokio-metrics = { version = "=0.5.0", features = ["rt"] }
 
 # Internal
 native-bridge-common = { path = "common" }

sandbox/libs/dataformat-native/rust/macros/Cargo.toml

@@ -8,6 +8,6 @@ license = "Apache-2.0"
 proc-macro = true
 
 [dependencies]
-quote = "1"
-syn = { version = "2", features = ["full"] }
-proc-macro2 = "1"
+quote = "=1.0.45"
+syn = { version = "=2.0.117", features = ["full"] }
+proc-macro2 = "=1.0.106"

sandbox/libs/tiered-storage/src/main/rust/src/tiered_object_store.rs

@@ -19,16 +19,14 @@
 //! registry's atomics and DashMap — no locks are held during I/O.
 
 use std::fmt;
-use std::ops::Range;
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use bytes::Bytes;
 use futures::stream::BoxStream;
 use futures::StreamExt;
 use object_store::{
-    path::Path, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
-    PutMultipartOptions, PutOptions, PutPayload, PutResult, Result as OsResult,
+    path::Path, CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
+    ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result as OsResult,
 };
 
 use crate::registry::traits::FileRegistry;
@@ -220,9 +218,36 @@ impl ObjectStore for TieredObjectStore {
 
     /// Primary read path: check registry for remote routing, otherwise local.
     /// If local read fails with NotFound and file transitioned to REMOTE, retries from remote.
+    ///
+    /// Also handles head requests (options.head == true) by returning cached
+    /// size from the registry when available — avoids I/O for the common case.
     async fn get_opts(&self, location: &Path, options: GetOptions) -> OsResult<GetResult> {
         let path_str = location.as_ref();
 
+        // Fast path for head: return cached size from registry if available
+        if options.head {
+            if let Some(guard) = self.registry.get(path_str) {
+                let size = guard.size();
+                if size > 0 {
+                    let meta = ObjectMeta {
+                        location: location.clone(),
+                        last_modified: chrono::DateTime::<chrono::Utc>::default(),
+                        size,
+                        e_tag: None,
+                        version: None,
+                    };
+                    return Ok(GetResult {
+                        payload: object_store::GetResultPayload::Stream(
+                            futures::stream::empty().boxed(),
+                        ),
+                        meta,
+                        range: 0..size,
+                        attributes: Default::default(),
+                    });
+                }
+            }
+        }
+
         if let Some((rp, store)) = self.resolve_remote(path_str) {
             native_bridge_common::log_debug!(
                 "TieredObjectStore: get_opts REMOTE path='{}'",
@@ -240,76 +265,20 @@ impl ObjectStore for TieredObjectStore {
         result
     }
 
-    /// Range read with local-fail-retry-remote.
-    async fn get_range(&self, location: &Path, range: Range<u64>) -> OsResult<Bytes> {
-        let path_str = location.as_ref();
-
-        if let Some((rp, store)) = self.resolve_remote(path_str) {
-            return store.get_range(&rp, range).await;
-        }
-
-        let result = self.local.get_range(location, range.clone()).await;
-        if let Err(ref e) = result {
-            if let Some((rp, store)) = self.should_retry_remote(path_str, e) {
-                return store.get_range(&rp, range).await;
-            }
-        }
-        result
-    }
-
-    /// Multi-range read with local-fail-retry-remote.
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> OsResult<Vec<Bytes>> {
-        let path_str = location.as_ref();
-
-        if let Some((rp, store)) = self.resolve_remote(path_str) {
-            return store.get_ranges(&rp, ranges).await;
-        }
-
-        let result = self.local.get_ranges(location, ranges).await;
-        if let Err(ref e) = result {
-            if let Some((rp, store)) = self.should_retry_remote(path_str, e) {
-                return store.get_ranges(&rp, ranges).await;
-            }
-        }
-        result
-    }
-
-    /// Head: check registry first (cached size), then local, then remote.
-    /// Head: check registry for cached size first (no I/O), then route via resolve_remote.
-    async fn head(&self, location: &Path) -> OsResult<ObjectMeta> {
-        let path_str = location.as_ref();
-
-        // Check registry for cached size — return immediately if available
-        if let Some(guard) = self.registry.get(path_str) {
-            let size = guard.size();
-            if size > 0 {
-                return Ok(ObjectMeta {
-                    location: location.clone(),
-                    last_modified: chrono::DateTime::<chrono::Utc>::default(),
-                    size,
-                    e_tag: None,
-                    version: None,
-                });
+    /// Delete stream: remove each path from registry only, NO local delete.
+    /// Local file deletion is handled by the Java layer.
+    fn delete_stream(
+        &self,
+        locations: BoxStream<'static, OsResult<Path>>,
+    ) -> BoxStream<'static, OsResult<Path>> {
+        let registry = Arc::clone(&self.registry);
+        let mapped = locations.map(move |result| {
+            if let Ok(ref path) = result {
+                registry.remove(path.as_ref(), true);
             }
-        }
-
-        // Size not cached — route via resolve_remote (REMOTE → remote store)
-        if let Some((rp, store)) = self.resolve_remote(path_str) {
-            return store.head(&rp).await;
-        }
-
-        // Not remote — try local
-        self.local.head(location).await
-    }
-
-    /// Delete: remove from registry only, NO local delete.
-    /// Local file deletion is handled by the Java layer
-    /// (TieredSubdirectoryAwareDirectory.deleteFile). Eviction (local copy
-    /// removal after sync) is a writable warm concern — not implemented.
-    async fn delete(&self, location: &Path) -> OsResult<()> {
-        let path_str = location.as_ref();
-        self.registry.remove(path_str, true);
-        Ok(())
+            result
+        });
+        Box::pin(mapped)
     }
 
     /// List: local entries first, then remote-only entries from registry (deduplicated).
@@ -364,23 +333,11 @@ impl ObjectStore for TieredObjectStore {
         Ok(result)
     }
 
-    async fn copy(&self, _from: &Path, _to: &Path) -> OsResult<()> {
+    async fn copy_opts(&self, _from: &Path, _to: &Path, _options: CopyOptions) -> OsResult<()> {
         Err(object_store::Error::NotSupported {
             source: "TieredObjectStore does not support copy".into(),
         })
     }
-
-    async fn copy_if_not_exists(&self, _from: &Path, _to: &Path) -> OsResult<()> {
-        Err(object_store::Error::NotSupported {
-            source: "TieredObjectStore does not support copy_if_not_exists".into(),
-        })
-    }
-
-    async fn rename_if_not_exists(&self, _from: &Path, _to: &Path) -> OsResult<()> {
-        Err(object_store::Error::NotSupported {
-            source: "TieredObjectStore does not support rename_if_not_exists".into(),
-        })
-    }
 }
 
 // ---------------------------------------------------------------------------

sandbox/libs/tiered-storage/src/main/rust/src/tiered_object_store_tests.rs

@@ -1,7 +1,7 @@
 use super::*;
 use futures::StreamExt;
 use object_store::memory::InMemory;
-use object_store::PutPayload;
+use object_store::{CopyOptions, ObjectStoreExt, PutPayload};
 use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
 
 /// Helper: create a registry + tiered store backed by in-memory stores.
@@ -514,10 +514,6 @@ impl ObjectStore for CallCountingStore {
         self.inner.get_opts(location, options).await
     }
 
-    async fn head(&self, location: &Path) -> OsResult<ObjectMeta> {
-        self.inner.head(location).await
-    }
-
     async fn put_opts(
         &self,
         location: &Path,
@@ -535,8 +531,11 @@ impl ObjectStore for CallCountingStore {
         self.inner.put_multipart_opts(location, opts).await
     }
 
-    async fn delete(&self, location: &Path) -> OsResult<()> {
-        self.inner.delete(location).await
+    fn delete_stream(
+        &self,
+        locations: BoxStream<'static, OsResult<Path>>,
+    ) -> BoxStream<'static, OsResult<Path>> {
+        self.inner.delete_stream(locations)
     }
 
     fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, OsResult<ObjectMeta>> {
@@ -547,16 +546,8 @@ impl ObjectStore for CallCountingStore {
         self.inner.list_with_delimiter(prefix).await
     }
 
-    async fn copy(&self, from: &Path, to: &Path) -> OsResult<()> {
-        self.inner.copy(from, to).await
-    }
-
-    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> OsResult<()> {
-        self.inner.copy_if_not_exists(from, to).await
-    }
-
-    async fn rename_if_not_exists(&self, from: &Path, to: &Path) -> OsResult<()> {
-        self.inner.rename_if_not_exists(from, to).await
+    async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> OsResult<()> {
+        self.inner.copy_opts(from, to, options).await
     }
 }
 
@@ -617,13 +608,6 @@ impl ObjectStore for ErrorStore {
         })
     }
 
-    async fn head(&self, _location: &Path) -> OsResult<ObjectMeta> {
-        Err(object_store::Error::Generic {
-            store: "ErrorStore",
-            source: "simulated error".into(),
-        })
-    }
-
     async fn put_opts(
         &self,
         _location: &Path,
@@ -647,11 +631,14 @@ impl ObjectStore for ErrorStore {
         })
     }
 
-    async fn delete(&self, _location: &Path) -> OsResult<()> {
-        Err(object_store::Error::Generic {
+    fn delete_stream(
+        &self,
+        locations: BoxStream<'static, OsResult<Path>>,
+    ) -> BoxStream<'static, OsResult<Path>> {
+        Box::pin(locations.map(|_| Err(object_store::Error::Generic {
             store: "ErrorStore",
             source: "simulated error".into(),
-        })
+        })))
     }
 
     fn list(&self, _prefix: Option<&Path>) -> BoxStream<'static, OsResult<ObjectMeta>> {
@@ -665,19 +652,7 @@ impl ObjectStore for ErrorStore {
         })
     }
 
-    async fn copy(&self, _from: &Path, _to: &Path) -> OsResult<()> {
-        Err(object_store::Error::NotSupported {
-            source: "not supported".into(),
-        })
-    }
-
-    async fn copy_if_not_exists(&self, _from: &Path, _to: &Path) -> OsResult<()> {
-        Err(object_store::Error::NotSupported {
-            source: "not supported".into(),
-        })
-    }
-
-    async fn rename_if_not_exists(&self, _from: &Path, _to: &Path) -> OsResult<()> {
+    async fn copy_opts(&self, _from: &Path, _to: &Path, _options: CopyOptions) -> OsResult<()> {
         Err(object_store::Error::NotSupported {
             source: "not supported".into(),
         })