[Analytics] Wire liquid cache into the analytics engine (plugin + integration)

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,17 @@ crc32fast = { workspace = true }
 # rewrite. STANDARD alphabet matches the OpenSearch `binary` field wire contract.
 base64 = "0.22"
 
+# Liquid Cache (optional): in-memory decoded-batch cache for parquet scans.
+# Off by default and only compiled in when the `liquid_cache` feature is enabled
+# (Gradle -PliquidCache). The crate re-exports everything needed, including its
+# datafusion pieces, so this single optional dependency is enough.
+opensearch-liquid-cache = { path = "../../liquid-cache/src/main/rust", optional = true }
+
+[features]
+# Off by default. When on, liquid cache is compiled in and 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/analytics-backend-datafusion/rust/src/session_context.rs

@@ -264,6 +264,16 @@ pub async unsafe fn create_session_context(
             )));
     }
 
+    // Liquid Cache (optional): engage on the listing-table path via the cache's
+    // own physical optimizer when the feature is compiled in and enabled. The
+    // session is rebuilt per query, so the enabled check tracks dynamic settings.
+    #[cfg(feature = "liquid_cache")]
+    if opensearch_liquid_cache::LiquidOnlyRuntime::is_enabled_globally() {
+        if let Some(rule) = opensearch_liquid_cache::LiquidOnlyRuntime::optimizer_globally() {
+            state_builder = state_builder.with_physical_optimizer_rule(rule);
+        }
+    }
+
     let state = state_builder.build();
 
     let ctx = SessionContext::new_with_state(state);

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionPlugin.java

@@ -60,6 +60,7 @@
 import org.opensearch.plugins.ActionPlugin;
 import org.opensearch.plugins.CircuitBreakerPlugin;
 import org.opensearch.plugins.DocumentLookupProvider;
+import org.opensearch.plugins.ExtensiblePlugin;
 import org.opensearch.plugins.NativeStoreHandle;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.plugins.SearchBackEndPlugin;
@@ -101,7 +102,8 @@ public class DataFusionPlugin extends Plugin
         AnalyticsSearchBackendPlugin,
         ActionPlugin,
         CircuitBreakerPlugin,
-        DocumentLookupProvider {
+        DocumentLookupProvider,
+        ExtensiblePlugin {
 
     private static final Logger logger = LogManager.getLogger(DataFusionPlugin.class);
 

sandbox/plugins/liquid-cache/.gitignore

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

sandbox/plugins/liquid-cache/README.md

@@ -0,0 +1,135 @@
+# Liquid Cache
+
+Liquid Cache is an in-memory, decoded-batch Parquet cache for the analytics
+(DataFusion) query path. When a query scans Parquet-backed composite indices,
+Liquid Cache keeps already-decoded column batches in memory so repeat scans skip
+the decode/IO work.
+
+It is an **experimental** feature and is gated behind a feature flag.
+
+## How it works
+
+Liquid Cache is compiled **into** the analytics engine's shared native library
+when that library is built with the `liquid_cache` cargo feature (Gradle
+`-PliquidCache`). When the feature is off (the default), the engine has no
+liquid cache dependency and the integration is an inlined no-op.
+
+When compiled in, the process-global cache is created at node startup and
+engages on both scan paths:
+
+- the **listing-table** path, via the cache's DataFusion optimizer added to the
+  query session;
+- the **indexed** path, via a hook in the per-row-group parquet scan.
+
+Cached decoded batches are served in place of re-decoding Parquet.
+
+```
+PPL/SQL query
+   -> analytics engine (DataFusion, built with -PliquidCache)
+      -> liquid cache (process-global, in-memory decoded batches)
+```
+
+This plugin owns the `datafusion.liquid_cache.*` settings, initializes the cache
+at startup, and exposes the stats/clear REST surface. It binds the cache's
+`lc_*` control symbols directly from the engine library.
+
+## Requirements
+
+- The `analytics-backend-datafusion` plugin must be installed (Liquid Cache
+  extends it for classloader access to the shared native library).
+- The engine native library must be **built with the `liquid_cache` feature**
+  (`-PliquidCache`). Without it, the `lc_*` symbols are absent and the plugin
+  loads inert.
+- The Liquid Cache experimental feature flag must be enabled.
+
+## Enabling
+
+1. Build the analytics engine native library with the feature enabled
+   (`-PliquidCache` / `LIQUID_CACHE=1`), and install the plugin alongside its
+   dependency.
+
+2. Enable the feature flag on every node (e.g. in `config/jvm.options` or
+   `OPENSEARCH_JAVA_OPTS`):
+
+   ```
+   -Dopensearch.experimental.feature.liquid_cache.enabled=true
+   ```
+
+3. Start the node. On startup you should see:
+
+   ```
+   Liquid Cache bound from the engine native library (liquid_cache feature enabled)
+   LiquidCachePlugin: liquid cache initialized and configured
+   ```
+
+If the flag is off, or the engine was built without the feature, the plugin
+loads but stays inert (a warning is logged) and queries run on the normal engine
+path.
+
+## Settings
+
+All settings live under `datafusion.liquid_cache.*`. Dynamic settings can be
+changed at runtime via the cluster settings API.
+
+| Setting | Default | Dynamic | Description |
+| --- | --- | --- | --- |
+| `datafusion.liquid_cache.enabled` | `true` | yes | Turn caching on/off. |
+| `datafusion.liquid_cache.size_bytes` | `1073741824` (1 GB) | yes | Cache memory budget in bytes. |
+| `datafusion.liquid_cache.eviction_policy` | `lru` | no (set at startup) | Eviction policy: `lru` or `liquid`. |
+| `datafusion.liquid_cache.indexed_query.max_columns` | `10` | yes | Max output columns for which the cache engages on the indexed-query path. |
+| `datafusion.liquid_cache.listing_table.max_columns` | `4` | yes | Max output columns for which the cache engages on the listing-table path. |
+
+Example:
+
+```
+PUT _cluster/settings
+{
+  "transient": {
+    "datafusion.liquid_cache.enabled": true,
+    "datafusion.liquid_cache.size_bytes": 2147483648
+  }
+}
+```
+
+## REST endpoints
+
+| Method | Path | Description |
+| --- | --- | --- |
+| GET | `_plugins/liquid_cache/stats` | Node-local cache counters. |
+| POST | `_plugins/liquid_cache/clear` | Clear all in-memory cache entries. |
+
+`stats` returns, per node:
+
+```json
+{
+  "cache_hit": 0, "cache_miss": 0, "predicate_evals": 0,
+  "memory_evictions": 0, "transcodes": 0, "total_entries": 0,
+  "memory_usage_bytes": 0, "max_memory_bytes": 1073741824
+}
+```
+
+A non-zero `max_memory_bytes` indicates the cache was initialized on that node
+(the engine was built with the feature and the flag is on).
+
+## Development
+
+Run a local node with the analytics stack and Liquid Cache compiled in
+(`-PliquidCache` builds the cache into the engine library):
+
+```
+PROTOC=/opt/homebrew/bin/protoc PATH="/opt/homebrew/bin:$PATH" \
+./gradlew run -Dsandbox.enabled=true -PrustDebug -PliquidCache \
+  -PinstalledPlugins="['arrow-base','arrow-flight-rpc','composite-engine','analytics-engine','parquet-data-format','analytics-backend-datafusion','liquid-cache','analytics-backend-lucene','dsl-query-executor']" \
+  -Dtests.jvm.argline="-Dopensearch.experimental.feature.pluggable.dataformat.enabled=true -Dopensearch.experimental.feature.transport.stream.enabled=true -Dopensearch.experimental.feature.liquid_cache.enabled=true"
+```
+
+The cache crates live in the standalone Cargo workspace under `src/main/rust/`
+(toolchain pinned via `rust-toolchain.toml`) and are compiled into the engine
+library as path dependencies when `-PliquidCache` is set. `protoc` must be on
+`PATH` for the Rust build.
+
+Run the Rust unit tests for the cache crates:
+
+```
+./gradlew :sandbox:plugins:liquid-cache:cargoTest
+```

sandbox/plugins/liquid-cache/build.gradle

@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+
+opensearchplugin {
+  description = 'Liquid Cache: in-memory decoded-batch Parquet cache exposed as a pluggable DataFusion session optimizer.'
+  classname = 'org.opensearch.liquidcache.LiquidCachePlugin'
+  // Extends the DataFusion back-end so the shared native library (which contains
+  // liquid cache when built with -PliquidCache) and the native-bridge SPI are
+  // visible on this plugin's classloader chain at runtime.
+  extendedPlugins = ['analytics-backend-datafusion']
+}
+
+java {
+  sourceCompatibility = JavaVersion.toVersion(25)
+  targetCompatibility = JavaVersion.toVersion(25)
+}
+
+dependencies {
+  // Shared native bridge lib (SymbolLookup, NativeCall). Provided at runtime by the
+  // parent analytics-backend-datafusion plugin (which bundles dataformat-native) via
+  // the extendedPlugins classloader chain — compileOnly here so we do NOT re-bundle
+  // it and cause jar hell when both plugins are installed on the same node.
+  compileOnly project(':sandbox:libs:dataformat-native')
+
+  compileOnly "org.apache.logging.log4j:log4j-api:${versions.log4j}"
+
+  testImplementation project(':test:framework')
+  testImplementation project(':sandbox:libs:dataformat-native')
+}
+
+test {
+  jvmArgs += ["--add-opens", "java.base/java.nio=ALL-UNNAMED"]
+  jvmArgs += ["--enable-native-access=ALL-UNNAMED"]
+  systemProperty 'native.lib.path', project(':sandbox:libs:dataformat-native').ext.nativeLibPath.absolutePath
+  dependsOn ':sandbox:libs:dataformat-native:buildRustLibrary'
+}
+
+// ═══════════════════════════════════════════════════════════════════
+// Rust unit tests for the liquid-cache crates (core + datafusion + glue).
+// Run as part of `check` so `./gradlew check` exercises them.
+// ═══════════════════════════════════════════════════════════════════
+// Liquid cache is its own standalone Cargo workspace (not part of the shared
+// dataformat-native workspace). The toolchain is pinned via rust-toolchain.toml
+// in the crate root.
+def rustWorkspaceDir = file("${projectDir}/src/main/rust")
+
+task cargoTest(type: Exec) {
+  description = 'Run Rust unit tests for the liquid-cache crates'
+  group = 'verification'
+  workingDir rustWorkspaceDir
+
+  def cargoExecutable = 'cargo'
+  def possibleCargoPaths = [
+    System.getenv('HOME') + '/.cargo/bin/cargo',
+    '/usr/local/bin/cargo',
+    'cargo'
+  ]
+  for (String path : possibleCargoPaths) {
+    if (new File(path).exists()) { cargoExecutable = path; break }
+  }
+
+  // datafusion-substrait's build script needs protoc; ensure Homebrew bin is on PATH.
+  def protoc = ['/opt/homebrew/bin/protoc', '/usr/local/bin/protoc'].find { new File(it).exists() }
+  if (protoc != null) {
+    environment 'PROTOC', protoc
+    environment 'PATH', new File(protoc).parent + File.pathSeparator + System.getenv('PATH')
+  }
+
+  commandLine cargoExecutable, 'test',
+    '-p', 'opensearch-liquid-cache-core',
+    '-p', 'opensearch-liquid-cache-datafusion',
+    '-p', 'opensearch-liquid-cache',
+    '--lib'
+
+  inputs.files fileTree("${projectDir}/src/main/rust/src")
+  inputs.files fileTree("${projectDir}/src/main/rust/core/src")
+  inputs.files fileTree("${projectDir}/src/main/rust/datafusion/src")
+  outputs.file "${projectDir}/src/main/rust/target/gradle-cargoTest.stamp"
+  doLast {
+    file("${projectDir}/src/main/rust/target").mkdirs()
+    file("${projectDir}/src/main/rust/target/gradle-cargoTest.stamp").text = new Date().toString()
+  }
+}
+
+check.dependsOn cargoTest
+
+// Liquid cache is compiled INTO the shared analytics engine native library when
+// that library is built with -PliquidCache (see dataformat-native/build.gradle).
+// There is no separate provider cdylib to build or bundle here; this plugin only
+// binds the lc_* symbols from the engine library at runtime.
+
+// missingJavadoc hardcodes --release 21 which hides FFM types (stable since JDK 22).
+tasks.matching { it.name == 'missingJavadoc' }.configureEach {
+  enabled = false
+}
+
+testingConventions.enabled = false
+
+tasks.named('forbiddenPatterns').configure {
+  exclude '**/*.parquet'
+  exclude '**/*.dylib'
+  exclude '**/*.so'
+  exclude '**/*.dll'
+}