Add in-memory liquid cache for DataFusion parquet scans

sandbox/libs/dataformat-native/rust/Cargo.toml

@@ -4,6 +4,8 @@ members = [
     "common",
     "macros",
     "lib",
+    "liquid-cache/core",
+    "liquid-cache/datafusion",
     "../../../plugins/analytics-backend-datafusion/rust",
     "../../../plugins/parquet-data-format/src/main/rust",
     "../../../plugins/native-repository-s3/src/main/rust",
@@ -30,6 +32,8 @@ datafusion-datasource = "=54.0.0"
 datafusion-common = "=54.0.0"
 datafusion-execution = "=54.0.0"
 datafusion-physical-expr = "=54.0.0"
+datafusion-physical-expr-common = "=54.0.0"
+datafusion-expr-common = "=54.0.0"
 datafusion-substrait = "=54.0.0"
 
 # Async
@@ -56,6 +60,12 @@ tikv-jemallocator = { version = "=0.6.1", features = ["disable_initial_exec_tls"
 tikv-jemalloc-ctl = { version = "=0.6.1", features = ["stats"] }
 tikv-jemalloc-sys = { version = "=0.6.1", features = ["stats"] }
 
+# Liquid cache (vendored subset — see liquid-cache/README.md)
+ahash = "=0.8.12"
+congee = "=0.4.1"
+fastlanes = "=0.5.2"
+num-traits = "=0.2.19"
+
 # Misc
 dashmap = { version = "=5.5.3", features = ["raw-api", "serde"] }
 num_cpus = "=1.17.0"
@@ -77,6 +87,8 @@ proptest = "=1.4.0"
 
 # Internal
 native-bridge-common = { path = "common" }
+opensearch-liquid-cache-core = { path = "liquid-cache/core" }
+opensearch-liquid-cache-datafusion = { path = "liquid-cache/datafusion" }
 opensearch-tiered-storage = { path = "../../tiered-storage/src/main/rust" }
 opensearch-repository-s3 = { path = "../../../plugins/native-repository-s3/src/main/rust" }
 opensearch-repository-gcs = { path = "../../../plugins/native-repository-gcs/src/main/rust" }

sandbox/libs/dataformat-native/rust/common/Cargo.toml

@@ -13,6 +13,7 @@ native-bridge-macros = { path = "../macros" }
 tikv-jemalloc-ctl = { workspace = true }
 tikv-jemalloc-sys = { workspace = true }
 tokio = { workspace = true }
+log = { workspace = true }
 
 [dev-dependencies]
 tikv-jemallocator = { workspace = true }

sandbox/libs/dataformat-native/rust/common/src/logger.rs

@@ -49,10 +49,42 @@ pub extern "C" fn native_logger_set_level(level: i32) {
     MAX_LEVEL.store(clamped, Ordering::Relaxed);
 }
 
+/// A `log` crate backend that forwards to our native bridge callback.
+/// This makes `log::info!()` from any crate (including liquid-cache)
+/// appear in OpenSearch's log file via the Java RustLoggerBridge.
+struct NativeBridgeLogger;
+
+impl ::log::Log for NativeBridgeLogger {
+    fn enabled(&self, metadata: &::log::Metadata) -> bool {
+        metadata.level() <= ::log::Level::Info
+    }
+
+    fn log(&self, record: &::log::Record) {
+        if !self.enabled(record.metadata()) {
+            return;
+        }
+        let level = match record.level() {
+            ::log::Level::Error => LogLevel::Error,
+            ::log::Level::Warn | ::log::Level::Info => LogLevel::Info,
+            _ => LogLevel::Debug,
+        };
+        let msg = format!("{}", record.args());
+        log(level, &msg);
+    }
+
+    fn flush(&self) {}
+}
+
+static NATIVE_BRIDGE_LOGGER: NativeBridgeLogger = NativeBridgeLogger;
+
 /// Called by Java at startup to register the log callback.
 #[no_mangle]
 pub unsafe extern "C" fn native_logger_init(callback: LogCallback) {
     LOG_CALLBACK.store(callback as *mut (), Ordering::Release);
+    // Initialize the `log` crate facade so that log::info!() from any
+    // dependency (e.g. liquid-cache) routes through our native bridge.
+    let _ = ::log::set_logger(&NATIVE_BRIDGE_LOGGER);
+    ::log::set_max_level(::log::LevelFilter::Info);
     log(LogLevel::Info, "Native logger initialized successfully");
 }
 

sandbox/libs/dataformat-native/rust/liquid-cache/README.md

@@ -0,0 +1,43 @@
+# opensearch-liquid-cache (vendored subset)
+
+In-memory-only subset of [liquid-cache](https://github.com/XiangpengHao/liquid-cache),
+vendored so the OpenSearch sandbox takes **no Cargo dependency** on the upstream
+package.
+
+## Provenance
+
+- Vendored from: https://github.com/cocosz/liquid-cache
+- Branch: `lc-opensearch-df54-v2`
+- Commit: `8311bccc258756127adbebee3e54140b60fc09ba`
+- License: Apache-2.0 (same as upstream)
+
+## What was removed relative to upstream
+
+Everything needed only for the disk tier, client/server mode, or data types the
+OpenSearch integration gate never caches:
+
+- **Disk tier**: `t4` store (io-uring), `DiskLiquid`/`DiskArrow` cache entries,
+  `SqueezeIoHandler`/`DefaultSqueezeIo`, squeeze-to-disk policies, `ipc.rs`
+  serialization and per-array `to_bytes`/`from_bytes`. The cache is memory-only;
+  under pressure entries are transcoded (Arrow → Liquid) and then evicted.
+- **Client/server**: the whole `liquid-cache-common` crate (arrow-flight/tonic/axum),
+  `datafusion-client`, `datafusion-server`.
+- **String/binary caching**: `byte_view_array` + FSST (`fsst-rs`). The integration
+  gate only engages LC for numeric/date/timestamp/boolean projections.
+- **Variant/JSON shredding**: `variant_array`, `parquet-variant-*` deps, variant UDFs.
+- **Date32 squeeze**: `squeezed_date32_array` (a squeeze-to-disk optimization).
+- **LineageOptimizer**: excluded by the POC already (planning overhead).
+- **Tracing**: `fastrace`, `sysinfo`.
+
+## Crates
+
+- `core/` — package `opensearch-liquid-cache-core`, **lib name `liquid_cache`**:
+  cache storage (index, budget, eviction policies, transcode) + numeric
+  LiquidArray encodings.
+- `datafusion/` — package `opensearch-liquid-cache-datafusion`, **lib name
+  `liquid_cache_datafusion`**: `LiquidParquetSource` (the ParquetSource
+  replacement), reader/stream machinery, `LocalModeOptimizer`.
+
+The lib names intentionally match upstream so vendored code and the
+`analytics-backend-datafusion` integration compile with unchanged `use` paths,
+which also keeps future re-syncs against upstream reviewable.

sandbox/libs/dataformat-native/rust/liquid-cache/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 }