Warm: Eager Loading + Tiered Cache

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

@@ -77,6 +77,7 @@ proptest = "=1.4.0"
 
 # Internal
 native-bridge-common = { path = "common" }
+opensearch-tiered-storage = { path = "../../tiered-storage/src/main/rust" }
 opensearch-repository-s3 = { path = "../../../plugins/native-repository-s3/src/main/rust" }
 opensearch-repository-gcs = { path = "../../../plugins/native-repository-gcs/src/main/rust" }
 opensearch-repository-azure = { path = "../../../plugins/native-repository-azure/src/main/rust" }

sandbox/libs/tiered-storage/src/main/rust/Cargo.toml

@@ -22,4 +22,10 @@ opensearch-block-cache = { path = "../../../../../plugins/block-cache-foyer/src/
 
 [dev-dependencies]
 tempfile = { workspace = true }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+parquet = { workspace = true }
+arrow = { workspace = true }
+arrow-array = { workspace = true }
+arrow-schema = { workspace = true }
+datafusion = { workspace = true }
+url = { workspace = true }

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

@@ -22,7 +22,7 @@ use opensearch_block_cache::traits::BlockCache;
 
 use crate::registry::TieredStorageRegistry;
 use crate::registry::FileRegistry;
-use crate::tiered_object_store::TieredObjectStore;
+use crate::tiered_object_store::{MetadataCachingStore, TieredObjectStore};
 use crate::types::FileLocation;
 
 const NULL_PTR: i64 = 0;
@@ -116,8 +116,12 @@ pub extern "C" fn ts_create_tiered_object_store(
     Ok(ptr)
 }
 
-/// Returns a `Box<Arc<dyn ObjectStore>>` pointer from an existing TieredObjectStore Arc pointer.
-/// This is the format that `df_create_reader` expects — a boxed fat pointer to the trait object.
+/// Returns a `Box<Arc<dyn MetadataCachingStore>>` pointer from an existing TieredObjectStore
+/// Arc pointer. This is the format that `df_create_reader` and warmup paths expect — a boxed
+/// fat pointer to the trait object. `MetadataCachingStore: ObjectStore`, so all `ObjectStore`
+/// methods remain available; in addition, callers can invoke `put_metadata` to promote bytes
+/// into the never-evict metadata tier.
+///
 /// Each call creates a new Box with its own Arc clone — caller must free with
 /// `ts_destroy_object_store_box_ptr`.
 #[ffm_safe]
@@ -129,22 +133,22 @@ pub extern "C" fn ts_get_object_store_box_ptr(tiered_store_ptr: i64) -> i64 {
     // Increment strong count so we don't consume the original Arc
     unsafe { Arc::increment_strong_count(tiered_store_ptr as *const TieredObjectStore) };
     let arc: Arc<TieredObjectStore> = unsafe { Arc::from_raw(tiered_store_ptr as *const TieredObjectStore) };
-    // Coerce to trait object and box it
-    let boxed: Box<Arc<dyn ObjectStore>> = Box::new(arc as Arc<dyn ObjectStore>);
+    // Coerce to MetadataCachingStore trait object and box it
+    let boxed: Box<Arc<dyn MetadataCachingStore>> = Box::new(arc as Arc<dyn MetadataCachingStore>);
     let ptr = Box::into_raw(boxed) as i64;
     native_bridge_common::log_info!("ffm: ts_get_object_store_box_ptr: ok");
     Ok(ptr)
 }
 
-/// Destroy a `Box<Arc<dyn ObjectStore>>` pointer returned by `ts_get_object_store_box_ptr`.
-/// Drops the Box and decrements the Arc strong count.
+/// Destroy a `Box<Arc<dyn MetadataCachingStore>>` pointer returned by
+/// `ts_get_object_store_box_ptr`. Drops the Box and decrements the Arc strong count.
 #[ffm_safe]
 #[no_mangle]
 pub extern "C" fn ts_destroy_object_store_box_ptr(ptr: i64) -> i64 {
     if ptr == NULL_PTR {
         return Err("ts_destroy_object_store_box_ptr: null pointer (0)".to_string());
     }
-    let _boxed = unsafe { Box::from_raw(ptr as *mut Arc<dyn ObjectStore>) };
+    let _boxed = unsafe { Box::from_raw(ptr as *mut Arc<dyn MetadataCachingStore>) };
     native_bridge_common::log_info!("ffm: ts_destroy_object_store_box_ptr: ok");
     Ok(0)
 }
@@ -244,6 +248,63 @@ pub extern "C" fn ts_remove_file(
     Ok(0)
 }
 
+/// Store metadata byte ranges into the metadata cache (never-evict tier).
+///
+/// Called by Java warmup after reading metadata bytes from any source (S3, local FS).
+/// The bytes are stored directly into the metadata_cache, bypassing data_cache.
+/// Subsequent reads via get_opts/get_ranges will find them on the first probe.
+///
+/// # Parameters
+/// - `store_ptr`: TieredObjectStore pointer.
+/// - `path_ptr`/`path_len`: UTF-8 file path.
+/// - `ranges_ptr`/`ranges_count`: array of `(start: i64, end: i64)` pairs.
+/// - `data_ptrs`/`data_lens`: arrays of byte buffers (one per range).
+///
+/// # Safety
+/// - `path_ptr` must point to `path_len` valid UTF-8 bytes.
+/// - `ranges_ptr` must point to `ranges_count * 2` consecutive i64 values.
+/// - `data_ptrs[i]` must point to `data_lens[i]` valid bytes for each range.
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn ts_put_metadata(
+    store_ptr: i64,
+    path_ptr: *const u8,
+    path_len: i64,
+    ranges_ptr: *const i64,
+    ranges_count: i32,
+    data_ptrs: *const *const u8,
+    data_lens: *const i64,
+) -> i64 {
+    let store = arc_from_ptr(store_ptr)?;
+    let path = str_from_raw(path_ptr, path_len)
+        .map_err(|e| format!("ts_put_metadata path: {}", e))?;
+    let path = path.strip_prefix('/').unwrap_or(path);
+
+    if ranges_ptr.is_null() || ranges_count <= 0 {
+        return Ok(0);
+    }
+
+    let mut ranges = Vec::with_capacity(ranges_count as usize);
+    let mut data = Vec::with_capacity(ranges_count as usize);
+    for i in 0..(ranges_count as usize) {
+        let start = *ranges_ptr.add(i * 2) as u64;
+        let end = *ranges_ptr.add(i * 2 + 1) as u64;
+        ranges.push(start..end);
+
+        let ptr = *data_ptrs.add(i);
+        let len = *data_lens.add(i) as usize;
+        let bytes = bytes::Bytes::copy_from_slice(std::slice::from_raw_parts(ptr, len));
+        data.push(bytes);
+    }
+
+    store.put_metadata(path, &ranges, &data);
+
+    native_bridge_common::log_debug!(
+        "ffm: ts_put_metadata path='{}' ranges={}", path, ranges_count
+    );
+    Ok(0)
+}
+
 // TODO (writable warm): add ts_get_file_location when LOCAL routing is needed.
 // TODO (writable warm): add ts_add_remote_store_ptr for late-binding remote store.