[Analytics] Vendor the liquid-cache library crates

sandbox/libs/dataformat-native/build.gradle

@@ -77,6 +77,11 @@ task buildRustLibrary(type: Exec) {
 
     def cargoArgs = [cargoExecutable, 'build', '-p', 'opensearch-native-lib']
     if (buildType == 'release') { cargoArgs.add('--release') }
+    // Liquid Cache: -PliquidCache (or LIQUID_CACHE=1) turns on the liquid_cache cargo feature (off by default).
+    if (project.hasProperty('liquidCache') || System.getenv('LIQUID_CACHE') == '1') {
+        logger.lifecycle('Liquid Cache: feature enabled')
+        cargoArgs.addAll(['--features', 'opensearch-datafusion/liquid_cache'])
+    }
     commandLine cargoArgs
 
     inputs.files fileTree("${rustWorkspaceDir}/common/src")

sandbox/plugins/analytics-backend-datafusion/rust/Cargo.toml

@@ -95,6 +95,13 @@ crc32fast = { workspace = true }
 # rewrite. STANDARD alphabet matches the OpenSearch `binary` field wire contract.
 base64 = "0.22"
 
+# Liquid Cache (off by default): uncomment the dep + the dep: entry below and build with -PliquidCache to use liquid cache.
+# opensearch-liquid-cache = { path = "../../liquid-cache/src/main/rust", optional = true }
+
+[features]
+# Off by default; when on, the hook in src/liquid_cache.rs engages the process-global cache.
+liquid_cache = [] # ["dep:opensearch-liquid-cache"]
+
 [dev-dependencies]
 criterion = { workspace = true }
 tempfile = { workspace = true }

sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/parquet_bridge.rs

@@ -250,9 +250,11 @@ fn create_stream_with_access_plan(
         }
     }
 
-    let mut config_builder =
-        FileScanConfigBuilder::new(config.store_url.clone(), Arc::new(parquet_source))
-            .with_file(partitioned_file);
+    // Liquid Cache: wrap the ParquetSource with LiquidParquetSource for eligible scans when the feature is on; a no-op otherwise.
+    let file_source: Arc<dyn datafusion::datasource::physical_plan::FileSource> =
+        crate::liquid_cache::maybe_wrap_parquet_source(parquet_source, config);
+    let mut config_builder = FileScanConfigBuilder::new(config.store_url.clone(), file_source)
+        .with_file(partitioned_file);
 
     if let Some(ref proj) = config.projection {
         // Empty projection (e.g. COUNT(*)) is honoured as "read no

sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs

@@ -31,6 +31,7 @@ pub mod ffm;
 pub mod helper;
 pub mod indexed_executor;
 pub mod indexed_table;
+pub mod liquid_cache;
 pub mod local_executor;
 pub mod memory;
 pub mod memory_guard;

sandbox/plugins/analytics-backend-datafusion/rust/src/liquid_cache.rs

@@ -0,0 +1,49 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+//! Liquid Cache hook for the indexed scan path; gated behind the `liquid_cache` feature (off by default = inlined no-op). Policy and cache live in the `opensearch-liquid-cache` crate.
+
+use std::sync::Arc;
+
+use datafusion::datasource::physical_plan::{FileSource, ParquetSource};
+
+use crate::indexed_table::parquet_bridge::RowGroupStreamConfig;
+
+/// Wrap the source with LiquidParquetSource when the `liquid_cache` feature is
+/// compiled in and the cache is enabled for an eligible scan; otherwise (and
+/// always in the default build) return the plain source unchanged.
+pub(crate) fn maybe_wrap_parquet_source(
+    parquet_source: ParquetSource,
+    config: &RowGroupStreamConfig,
+) -> Arc<dyn FileSource> {
+    #[cfg(feature = "liquid_cache")]
+    {
+        use opensearch_liquid_cache::{
+            indexed_scan_eligible, LiquidOnlyRuntime, LiquidParquetSource,
+        };
+
+        if LiquidOnlyRuntime::is_enabled_globally() {
+            if let Some(cache_ref) = LiquidOnlyRuntime::cache_ref_globally() {
+                if indexed_scan_eligible(
+                    &config.full_schema,
+                    config.projection.as_deref(),
+                    config.predicate.as_ref(),
+                ) {
+                    return Arc::new(LiquidParquetSource::from_parquet_source(
+                        parquet_source,
+                        cache_ref,
+                    ));
+                }
+            }
+        }
+    }
+    #[cfg(not(feature = "liquid_cache"))]
+    let _ = config;
+
+    Arc::new(parquet_source)
+}

sandbox/plugins/liquid-cache/.gitignore

@@ -0,0 +1,2 @@
+src/main/rust/target/
+src/main/rust/Cargo.lock

sandbox/plugins/liquid-cache/src/main/rust/Cargo.toml

@@ -0,0 +1,40 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# Liquid Cache native crates — a self-contained Cargo workspace holding the
+# in-memory decoded-batch cache (./core) and its DataFusion parquet reader
+# (./datafusion). These crates are compiled into the analytics engine's native
+# library only when it is built with the `liquid_cache` cargo feature; on their
+# own they are just a buildable, self-testing library.
+[workspace]
+resolver = "2"
+members = ["core", "datafusion"]
+
+[workspace.dependencies]
+arrow = { version = "=58.3.0", features = ["ffi"] }
+arrow-schema = "=58.3.0"
+parquet = "=58.3.0"
+datafusion = "=54.0.0"
+datafusion-common = "=54.0.0"
+datafusion-expr-common = "=54.0.0"
+datafusion-physical-expr = "=54.0.0"
+datafusion-datasource = "=54.0.0"
+tokio = { version = "=1.52.3", features = ["full"] }
+futures = "=0.3.32"
+object_store = "=0.13.2"
+bytes = "=1.11.1"
+log = "=0.4.29"
+serde = { version = "=1.0.228", features = ["derive"] }
+ahash = "=0.8.12"
+congee = "=0.4.1"
+fastlanes = "=0.5.2"
+num-traits = "=0.2.19"
+tempfile = "=3.27.0"
+opensearch-liquid-cache-core = { path = "core" }
+opensearch-liquid-cache-datafusion = { path = "datafusion" }
+
+[profile.release]
+lto = true
+codegen-units = 1
+
+[profile.dev]
+opt-level = 1

sandbox/plugins/liquid-cache/src/main/rust/core/Cargo.toml

@@ -0,0 +1,31 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# Vendored in-memory subset of liquid-cache core.
+# Provenance: https://github.com/cocosz/liquid-cache branch lc-opensearch-df54-v2
+# commit 8311bccc258756127adbebee3e54140b60fc09ba. See ../README.md.
+
+[package]
+name = "opensearch-liquid-cache-core"
+version = "0.1.0"
+edition = "2024"
+license = "Apache-2.0"
+publish = false
+
+[lib]
+# Keep the upstream crate name so vendored code and the integration keep
+# their `use liquid_cache::...` paths, making future upstream re-syncs diffable.
+name = "liquid_cache"
+path = "src/lib.rs"
+
+[dependencies]
+ahash = { workspace = true }
+arrow = { workspace = true }
+arrow-schema = { workspace = true }
+congee = { workspace = true }
+datafusion-common = { workspace = true }
+datafusion-expr-common = { workspace = true }
+datafusion-physical-expr = { workspace = true }
+fastlanes = { workspace = true }
+log = { workspace = true }
+num-traits = { workspace = true }
+serde = { workspace = true }

sandbox/plugins/liquid-cache/src/main/rust/core/src/cache/budget.rs

@@ -0,0 +1,164 @@
+use crate::sync::atomic::{AtomicUsize, Ordering};
+
+#[derive(Debug)]
+pub struct BudgetAccounting {
+    max_memory_bytes: AtomicUsize,
+    used_memory_bytes: AtomicUsize,
+}
+
+impl BudgetAccounting {
+    pub(super) fn new(max_memory_bytes: usize) -> Self {
+        Self {
+            max_memory_bytes: AtomicUsize::new(max_memory_bytes),
+            used_memory_bytes: AtomicUsize::new(0),
+        }
+    }
+
+    pub(super) fn reset_usage(&self) {
+        self.used_memory_bytes.store(0, Ordering::Relaxed);
+    }
+
+    /// Dynamically update the max memory limit. Takes effect for new reservations.
+    pub fn set_max_memory_bytes(&self, new_limit: usize) {
+        self.max_memory_bytes.store(new_limit, Ordering::Relaxed);
+    }
+
+    pub fn max_memory_bytes(&self) -> usize {
+        self.max_memory_bytes.load(Ordering::Relaxed)
+    }
+
+    /// Try to reserve memory in the cache.
+    /// Returns ok if the memory was reserved, err if the memory budget is full.
+    pub(super) fn try_reserve_memory(&self, request_bytes: usize) -> Result<(), ()> {
+        let used = self.used_memory_bytes.load(Ordering::Relaxed);
+        if used + request_bytes > self.max_memory_bytes.load(Ordering::Relaxed) {
+            return Err(());
+        }
+
+        match self.used_memory_bytes.compare_exchange(
+            used,
+            used + request_bytes,
+            Ordering::Relaxed,
+            Ordering::Relaxed,
+        ) {
+            Ok(_) => Ok(()),
+            Err(_) => self.try_reserve_memory(request_bytes),
+        }
+    }
+
+    /// Adjust memory usage after transcoding.
+    /// Returns ok if the usage was adjusted, err if the memory budget is full when new_size is larger than old_size.
+    pub(super) fn try_update_memory_usage(
+        &self,
+        old_size: usize,
+        new_size: usize,
+    ) -> Result<(), ()> {
+        if old_size < new_size {
+            let diff = new_size - old_size;
+            self.try_reserve_memory(diff)?;
+            Ok(())
+        } else {
+            self.used_memory_bytes
+                .fetch_sub(old_size - new_size, Ordering::Relaxed);
+            Ok(())
+        }
+    }
+
+    pub fn memory_usage_bytes(&self) -> usize {
+        self.used_memory_bytes.load(Ordering::Relaxed)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::sync::{Arc, Barrier, thread};
+
+    fn test_budget(max_memory_bytes: usize) -> BudgetAccounting {
+        BudgetAccounting::new(max_memory_bytes)
+    }
+
+    #[test]
+    fn test_memory_reservation_and_accounting() {
+        let config = test_budget(1000);
+
+        assert_eq!(config.memory_usage_bytes(), 0);
+
+        assert!(config.try_reserve_memory(500).is_ok());
+        assert_eq!(config.memory_usage_bytes(), 500);
+
+        assert!(config.try_reserve_memory(300).is_ok());
+        assert_eq!(config.memory_usage_bytes(), 800);
+
+        assert!(config.try_reserve_memory(300).is_err());
+        assert_eq!(config.memory_usage_bytes(), 800);
+
+        config.reset_usage();
+        assert_eq!(config.memory_usage_bytes(), 0);
+    }
+
+    #[test]
+    fn test_concurrent_memory_operations() {
+        test_concurrent_memory_budget();
+    }
+
+    fn test_concurrent_memory_budget() {
+        let num_threads = 3;
+        let max_memory = 10000;
+        let operations_per_thread = 100;
+
+        let budget = Arc::new(test_budget(max_memory));
+        let barrier = Arc::new(Barrier::new(num_threads));
+
+        let mut thread_handles = vec![];
+
+        for _ in 0..num_threads {
+            let budget_clone = budget.clone();
+            let barrier_clone = barrier.clone();
+
+            let handle = thread::spawn(move || {
+                let mut successful_reservations = Vec::new();
+
+                barrier_clone.wait();
+
+                for i in 0..operations_per_thread {
+                    let reserve_size = 10 + (i % 20) * 5; // 10 to 105 bytes
+                    if budget_clone.try_reserve_memory(reserve_size).is_ok() {
+                        successful_reservations.push(reserve_size);
+                    }
+
+                    if i.is_multiple_of(5) && !successful_reservations.is_empty() {
+                        let idx = i % successful_reservations.len();
+                        let old_size = successful_reservations[idx];
+                        let new_size = if i.is_multiple_of(2) {
+                            old_size + 5 // Grow
+                        } else {
+                            old_size.saturating_sub(5) // Shrink
+                        };
+
+                        if budget_clone
+                            .try_update_memory_usage(old_size, new_size)
+                            .is_ok()
+                        {
+                            successful_reservations[idx] = new_size;
+                        }
+                    }
+                }
+                successful_reservations
+            });
+
+            thread_handles.push(handle);
+        }
+
+        let mut expected_memory_usage = 0;
+        for handle in thread_handles {
+            let reservations = handle.join().unwrap();
+            for size in reservations {
+                expected_memory_usage += size;
+            }
+        }
+
+        assert_eq!(budget.memory_usage_bytes(), expected_memory_usage);
+        assert!(budget.memory_usage_bytes() <= max_memory);
+    }
+}