From a09b3382708ba99aba528e928bc46c2999aaf938 Mon Sep 17 00:00:00 2001 From: Tanvir Alam Date: Wed, 29 Jul 2026 02:29:35 +0530 Subject: [PATCH] Add optional liquid cache for indexed parquet scans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an opt-in, in-memory decoded-batch cache (liquid cache) for the DataFusion analytics backend, wired behind a cargo feature so it is absent from the default build. When the `liquid_cache` feature is off (the default), the integration is an inlined no-op: eligible scans use the plain ParquetSource unchanged. When on, an eligible indexed row-group scan — all projected columns numeric/date/timestamp/boolean, within a column budget, and no string or binary predicate column — is wrapped with LiquidParquetSource so decoded batches are served from and populated into the process-global cache. Enabling is a single-file edit plus a build flag: the liquid cache dependency is a commented placeholder in the analytics-backend-datafusion crate (uncomment it, set the git repo/branch, and uncomment the matching `dep:` feature entries), then build with -PliquidCache (or LIQUID_CACHE=1). The default build declares no liquid cache source, so it resolves and compiles exactly as before. Signed-off-by: Tanvir Alam --- sandbox/libs/dataformat-native/build.gradle | 5 ++ .../rust/Cargo.toml | 7 +++ .../rust/src/indexed_table/parquet_bridge.rs | 8 +-- .../rust/src/lib.rs | 1 + .../rust/src/liquid_cache.rs | 49 +++++++++++++++++++ 5 files changed, 67 insertions(+), 3 deletions(-) create mode 100644 sandbox/plugins/analytics-backend-datafusion/rust/src/liquid_cache.rs diff --git a/sandbox/libs/dataformat-native/build.gradle b/sandbox/libs/dataformat-native/build.gradle index 529ea63f2376b..d1414840572ad 100644 --- a/sandbox/libs/dataformat-native/build.gradle +++ b/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") diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/Cargo.toml b/sandbox/plugins/analytics-backend-datafusion/rust/Cargo.toml index c12a4cb13c5c2..f2ca54e663860 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/Cargo.toml +++ b/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 } diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/parquet_bridge.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/parquet_bridge.rs index 957b0d15fa0dc..3cffbc0ffddf7 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/parquet_bridge.rs +++ b/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 = + 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 diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs index 8708627abe3a4..7f2e9eda7e53a 100644 --- a/sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs +++ b/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; diff --git a/sandbox/plugins/analytics-backend-datafusion/rust/src/liquid_cache.rs b/sandbox/plugins/analytics-backend-datafusion/rust/src/liquid_cache.rs new file mode 100644 index 0000000000000..a1ea38b15eeef --- /dev/null +++ b/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 { + #[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) +}