opensearch-project · cocosz · Jul 28, 2026
@@ -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")

@@ -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 }

@@ -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

@@ -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;

@@ -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)
+}