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