Enable Foyer disk based cache for Parquet reads

Author: Vishwas Garg

modules/parquet-data-format/src/main/resources/native/macos-aarch64/libparquet_dataformat_jni.dylib

@@ -19,6 +19,8 @@ arrow-array = "57.3.0"
 arrow-schema = "57.3.0"
 arrow-buffer = "57.3.0"
 downcast-rs = "1.2"
+foyer = { version = "=0.11.5" }
+bytes = "1.9"
 
 
 # JNI dependencies

plugins/engine-datafusion/Cargo.toml

@@ -68,6 +68,12 @@ url = { workspace = true }
 # Liquid Cache for byte-level caching
 liquid-cache-datafusion-local = { workspace = true }
 
+# Foyer hybrid in-memory+disk cache for Parquet page caching
+foyer = { workspace = true }
+
+# serde_bytes: efficient Bytes serialization needed for Foyer's StorageValue bound
+serde_bytes = "0.11"
+
 # Substrait support
 substrait = { workspace = true }
 

plugins/engine-datafusion/jni/Cargo.toml

@@ -7,11 +7,13 @@ use datafusion::execution::cache::CacheAccessor;
 use object_store::ObjectMeta;
 use vectorized_exec_spi::log_error;
 
-pub const ALL_CACHE_TYPES: &[&str] = &[CACHE_TYPE_METADATA, CACHE_TYPE_STATS];
+pub const ALL_CACHE_TYPES: &[&str] = &[CACHE_TYPE_METADATA, CACHE_TYPE_STATS, CACHE_TYPE_PAGES];
 
 // Cache type constants
 pub const CACHE_TYPE_METADATA: &str = "METADATA";
 pub const CACHE_TYPE_STATS: &str = "STATISTICS";
+/// Foyer-backed byte-level page cache for Parquet column chunk data (Cache Layer 3)
+pub const CACHE_TYPE_PAGES: &str = "PAGES";
 
 // Helper function to handle cache errors
 #[allow(dead_code)]

plugins/engine-datafusion/jni/src/cache.rs

@@ -1,14 +1,40 @@
-use jni::objects::{JClass, JObjectArray, JString};
-use jni::sys::jlong;
+use jni::objects::{JByteArray, JClass, JObjectArray, JString};
+use jni::sys::{jbyteArray, jint, jlong};
 use jni::{JNIEnv};
 use crate::custom_cache_manager::CustomCacheManager;
 use crate::util::{parse_string_arr};
 use crate::cache;
 use crate::DataFusionRuntime;
 use datafusion::execution::cache::cache_unit::DefaultFilesMetadataCache;
 use std::sync::Arc;
+use bytes::Bytes;
 use vectorized_exec_spi::{log_info, log_error, log_debug};
 
+// Default page cache budget — overridden by Java settings via createCache()
+const DEFAULT_PAGE_CACHE_DISK_BYTES: usize = 10 * 1024 * 1024 * 1024;  // 10 GB disk
+const DEFAULT_PAGE_CACHE_DIR: &str = "/tmp/foyer-page-cache";
+
+/// Parse the eviction_type string for PAGES cache type.
+/// Expected format: "<disk_capacity_bytes>|<disk_dir>"
+/// Falls back to defaults if the string is malformed (e.g. plain "LRU" from old Java code).
+fn parse_page_cache_params(eviction_str: &str) -> (usize, String) {
+    if let Some(sep) = eviction_str.find('|') {
+        let disk_bytes_str = &eviction_str[..sep];
+        let disk_dir = eviction_str[sep + 1..].to_string();
+        if let Ok(disk_bytes) = disk_bytes_str.parse::<usize>() {
+            let dir = if disk_dir.is_empty() { DEFAULT_PAGE_CACHE_DIR.to_string() } else { disk_dir };
+            return (disk_bytes, dir);
+        }
+    }
+    // Fallback: plain eviction type like "LRU" from legacy config
+    log_info!(
+        "[FOYER-PAGE-CACHE] eviction_type '{}' is not in '<disk_bytes>|<dir>' format; \
+         using defaults: disk={}B, dir={}",
+        eviction_str, DEFAULT_PAGE_CACHE_DISK_BYTES, DEFAULT_PAGE_CACHE_DIR
+    );
+    (DEFAULT_PAGE_CACHE_DISK_BYTES, DEFAULT_PAGE_CACHE_DIR.to_string())
+}
+
 /// Create a CustomCacheManager instance
 #[no_mangle]
 pub extern "system" fn Java_org_opensearch_datafusion_jni_NativeBridge_createCustomCacheManager(
@@ -89,6 +115,25 @@ pub extern "system" fn Java_org_opensearch_datafusion_jni_NativeBridge_createCac
             manager.set_statistics_cache(stats_cache);
             log_info!("[CACHE INFO] Successfully created {} cache in CustomCacheManager", cache_type_str);
         }
+        cache::CACHE_TYPE_PAGES => {
+            // Create Foyer disk-only page cache — Cache Layer 3.
+            // `size_limit` is ignored for PAGES (disk-only, no memory tier).
+            // The disk budget and disk directory are passed via eviction_type as
+            // "<disk_bytes>|<disk_dir>".
+            let (disk_bytes, disk_dir) = parse_page_cache_params(&eviction_type_str);
+            log_info!(
+                "[FOYER-PAGE-CACHE] creating disk-only page cache: disk={}B, dir={}",
+                disk_bytes, disk_dir
+            );
+            let page_cache = Arc::new(crate::tiered::foyer_cache::FoyerDiskPageCache::new(
+                disk_bytes,
+                disk_dir,
+            ));
+            manager.set_page_cache(page_cache);
+            log_info!(
+                "[FOYER-PAGE-CACHE] successfully created Foyer disk-only page cache in CustomCacheManager"
+            );
+        }
         _ => {
             let msg = format!("Invalid cache type: {}", cache_type_str);
             log_error!("[CACHE ERROR] {}", msg);
@@ -444,3 +489,128 @@ pub extern "system" fn Java_org_opensearch_datafusion_jni_NativeBridge_cacheMana
         }
     }
 }
+
+// ============================================================================
+// Foyer page cache JNI operations (Layer 3: Parquet byte range cache)
+// Called by DataFusionPlugin.FoyerCacheProvider implementation to serve
+// PassthroughCacheStrategy → FoyerParquetCacheStrategy in the tiered-storage module.
+// ============================================================================
+
+/// Look up a cached byte range for a Parquet file.
+/// Returns the cached bytes as a Java byte[], or null on cache miss.
+#[no_mangle]
+pub extern "system" fn Java_org_opensearch_datafusion_jni_NativeBridge_foyerPageCacheGet(
+    mut env: JNIEnv,
+    _class: JClass,
+    runtime_ptr: jlong,
+    path: JString,
+    start: jint,
+    end: jint,
+) -> jbyteArray {
+    if runtime_ptr == 0 {
+        return std::ptr::null_mut();
+    }
+
+    let runtime = unsafe { &*(runtime_ptr as *const DataFusionRuntime) };
+    let path_str: String = match env.get_string(&path) {
+        Ok(s) => s.into(),
+        Err(_) => return std::ptr::null_mut(),
+    };
+
+    let page_cache = match runtime.custom_cache_manager.as_ref().and_then(|m| m.get_page_cache()) {
+        Some(c) => c,
+        None => return std::ptr::null_mut(),
+    };
+
+    // FoyerDiskPageCache.get() is async (disk I/O). Use get_blocking() since JNI is synchronous.
+    match page_cache.get_blocking(&path_str, start as usize, end as usize) {
+        Some(bytes) => {
+            log_debug!(
+                "[FOYER-PAGE-CACHE] JNI get HIT: path={}, range={}..{}, size={}B",
+                path_str, start, end, bytes.len()
+            );
+            match env.byte_array_from_slice(&bytes) {
+                Ok(arr) => arr.into_raw(),
+                Err(e) => {
+                    log_debug!("[FOYER-PAGE-CACHE] JNI get: failed to create Java byte[]: {}", e);
+                    std::ptr::null_mut()
+                }
+            }
+        }
+        None => {
+            log_debug!(
+                "[FOYER-PAGE-CACHE] JNI get MISS: path={}, range={}..{}",
+                path_str, start, end
+            );
+            std::ptr::null_mut()
+        }
+    }
+}
+
+/// Store a byte range for a Parquet file in the Foyer page cache.
+#[no_mangle]
+pub extern "system" fn Java_org_opensearch_datafusion_jni_NativeBridge_foyerPageCachePut(
+    mut env: JNIEnv,
+    _class: JClass,
+    runtime_ptr: jlong,
+    path: JString,
+    start: jint,
+    end: jint,
+    data: JByteArray,
+) {
+    if runtime_ptr == 0 {
+        return;
+    }
+
+    let runtime = unsafe { &*(runtime_ptr as *const DataFusionRuntime) };
+    let path_str: String = match env.get_string(&path) {
+        Ok(s) => s.into(),
+        Err(e) => {
+            log_debug!("[FoyerCache] foyerPageCachePut: failed to convert path: {}", e);
+            return;
+        }
+    };
+
+    let page_cache = match runtime.custom_cache_manager.as_ref().and_then(|m| m.get_page_cache()) {
+        Some(c) => c,
+        None => return,
+    };
+
+    let bytes_vec: Vec<u8> = match env.convert_byte_array(data) {
+        Ok(v) => v,
+        Err(e) => {
+            log_debug!("[FoyerCache] foyerPageCachePut: failed to convert byte array: {}", e);
+            return;
+        }
+    };
+
+    page_cache.put(path_str, start as usize, end as usize, Bytes::from(bytes_vec));
+}
+
+/// Evict all cached byte ranges for a given Parquet file.
+/// Called when a file is deleted (merged/compacted/tiered out).
+#[no_mangle]
+pub extern "system" fn Java_org_opensearch_datafusion_jni_NativeBridge_foyerPageCacheEvictFile(
+    mut env: JNIEnv,
+    _class: JClass,
+    runtime_ptr: jlong,
+    path: JString,
+) {
+    if runtime_ptr == 0 {
+        return;
+    }
+
+    let runtime = unsafe { &*(runtime_ptr as *const DataFusionRuntime) };
+    let path_str: String = match env.get_string(&path) {
+        Ok(s) => s.into(),
+        Err(e) => {
+            log_debug!("[FoyerCache] foyerPageCacheEvictFile: failed to convert path: {}", e);
+            return;
+        }
+    };
+
+    if let Some(page_cache) = runtime.custom_cache_manager.as_ref().and_then(|m| m.get_page_cache()) {
+        page_cache.evict_file(&path_str);
+        log_debug!("[FoyerCache] evicted file from page cache: {}", path_str);
+    }
+}

plugins/engine-datafusion/jni/src/cache_jni.rs

@@ -10,37 +10,60 @@ use crate::util::{create_object_meta_from_file};
 use object_store::path::Path;
 use object_store::ObjectMeta;
 use datafusion::datasource::physical_plan::parquet::metadata::DFParquetMetadata;
-use vectorized_exec_spi::{log_debug, log_error};
+use vectorized_exec_spi::{log_debug, log_error, log_info};
+use crate::tiered::foyer_cache::FoyerDiskPageCache;
 
 /// Custom CacheManager that holds cache references directly
 pub struct CustomCacheManager {
-    /// Direct reference to the file metadata cache
+    /// Direct reference to the file metadata cache (Cache Layer 1: Parquet footer/schema)
     file_metadata_cache: Option<Arc<MutexFileMetadataCache>>,
-    /// Direct reference to the statistics cache
-    statistics_cache: Option<Arc<CustomStatisticsCache>>
+    /// Direct reference to the statistics cache (Cache Layer 2: row counts, min/max stats)
+    statistics_cache: Option<Arc<CustomStatisticsCache>>,
+    /// Foyer-backed hybrid (memory+disk) page cache (Cache Layer 3: Parquet column chunk byte ranges)
+    pub page_cache: Option<Arc<FoyerDiskPageCache>>,
 }
 
 impl CustomCacheManager {
     /// Create a new CustomCacheManager
     pub fn new() -> Self {
         Self {
             file_metadata_cache: None,
-            statistics_cache: None
+            statistics_cache: None,
+            page_cache: None,
         }
     }
 
-    /// Set the file metadata cache
+    /// Set the file metadata cache (Layer 1)
     pub fn set_file_metadata_cache(&mut self, cache: Arc<MutexFileMetadataCache>) {
         self.file_metadata_cache = Some(cache);
         log_debug!("[CACHE INFO] File metadata cache set in CustomCacheManager");
     }
 
-    /// Set the statistics cache
+    /// Set the statistics cache (Layer 2)
     pub fn set_statistics_cache(&mut self, cache: Arc<CustomStatisticsCache>) {
         self.statistics_cache = Some(cache);
         log_debug!("[CACHE INFO] Statistics cache set in CustomCacheManager");
     }
 
+    /// Set the Foyer page cache (Layer 3).
+    ///
+    /// Once set, the `CachingObjectStore` (wrapping this cache) will intercept all
+    /// `get_range()` calls to DataFusion's ObjectStore, returning cached bytes on HIT
+    /// and populating the cache on MISS.
+    pub fn set_page_cache(&mut self, cache: Arc<FoyerDiskPageCache>) {
+        log_info!(
+            "[FOYER-PAGE-CACHE] page cache set in CustomCacheManager: disk={}B, dir={}",
+            cache.disk_capacity_bytes(),
+            cache.disk_dir().display()
+        );
+        self.page_cache = Some(cache);
+    }
+
+    /// Get the Foyer page cache (Layer 3), if configured.
+    pub fn get_page_cache(&self) -> Option<Arc<FoyerDiskPageCache>> {
+        self.page_cache.clone()
+    }
+
     /// Get the statistics cache
     pub fn get_statistics_cache(&self) -> Option<Arc<CustomStatisticsCache>> {
         self.statistics_cache.clone()
@@ -167,6 +190,12 @@ impl CustomCacheManager {
                 }
             }
 
+            // Evict all cached page byte ranges for this file (Layer 3)
+            if let Some(page_cache) = &self.page_cache {
+                page_cache.evict_file(file_path);
+                any_removed = true; // evict_file is best-effort, count as success
+            }
+
             let removed = if !errors.is_empty() && !any_removed {
                 false
             } else {
@@ -257,29 +286,36 @@ impl CustomCacheManager {
     pub fn get_total_memory_consumed(&self) -> usize {
         let mut total = 0;
 
-        // Add metadata cache memory
+        // Layer 1: metadata cache memory
         if let Some(cache) = &self.file_metadata_cache {
             if let Ok(cache_guard) = cache.inner.lock() {
                 total += cache_guard.memory_used();
             }
         }
 
-        // Add statistics cache memory
+        // Layer 2: statistics cache memory
         if let Some(cache) = &self.statistics_cache {
             total += cache.memory_consumed();
         }
 
+        // Layer 3: page cache is disk-only — no memory contribution to track here.
+
         total
     }
 
-    /// Clear all caches
+    /// Clear all caches (Layers 1, 2, and 3)
     pub fn clear_all(&self) {
         if let Some(cache) = &self.file_metadata_cache {
             cache.clear();
         }
         if let Some(cache) = &self.statistics_cache {
             cache.clear();
         }
+        // FoyerDiskPageCache.clear() is async — use the blocking wrapper
+        if let Some(cache) = &self.page_cache {
+            log_info!("[FOYER-PAGE-CACHE] clear_all: clearing page cache (memory + disk)");
+            cache.clear_blocking();
+        }
     }
 
     /// Clear specific cache type
@@ -301,6 +337,15 @@ impl CustomCacheManager {
                     Err("No statistics cache configured".to_string())
                 }
             }
+            crate::cache::CACHE_TYPE_PAGES => {
+                if let Some(cache) = &self.page_cache {
+                    log_info!("[FOYER-PAGE-CACHE] clear_cache_type PAGES: clearing page cache (memory + disk)");
+                    cache.clear_blocking();
+                    Ok(())
+                } else {
+                    Err("No page cache configured".to_string())
+                }
+            }
             _ => Err(format!("Unknown cache type: {}", cache_type))
         }
     }
@@ -326,6 +371,10 @@ impl CustomCacheManager {
                     Err("No statistics cache configured".to_string())
                 }
             }
+            crate::cache::CACHE_TYPE_PAGES => {
+                // Disk-only cache — reports 0 heap memory consumed.
+                Ok(0)
+            }
             _ => Err(format!("Unknown cache type: {}", cache_type))
         }
     }