OpenSearch integration: selective caching, LRU eviction, dynamic budgets by cocosz · Pull Request #505 · datafusion-contrib/liquid-cache

cocosz · 2026-06-08T10:37:28Z

Related : #500

Summary

Liquid Cache currently wraps every ParquetSource unconditionally. This causes severe regressions on string-heavy queries (cache misses force full parquet reads) and queries where decode cost is already minimal. This PR makes LC aware of when it can and cannot help, and gracefully opts out when engagement would hurt.

Changes

1. Selective column caching

(`cache/column.rs`)

Cache only numeric predicate columns. String and binary columns are never inserted into LC — they cannot be efficiently encoded and their misses force full parquet reads that negate any cache benefit.

2. LRU eviction policy with type-aware queues

(`cache/policies/cache/lru.rs`, `cache/policies/cache/mod.rs`)

New LRU eviction policy that manages Arrow, Liquid-encoded, and squeezed entries in separate queues. Allows bounded memory usage with predictable eviction behavior under concurrent load.

3. Dynamic budget resize

(`cache/budget.rs`, `cache/core.rs`)

Cache memory and disk budgets can be resized at runtime via `set_max_memory_bytes` / `set_max_disk_bytes`. Enables live tuning without requiring a restart.

4. Builder API extensions

(`datafusion-local/src/lib.rs`)

`with_max_disk_bytes` on `LiquidCacheLocalBuilder`
Configurable eviction policy (LRU or default Liquid)
Re-export `LiquidParquetSource` and `LiquidCacheParquetRef` from the local crate

5. Optimizer gating: skip uncacheable queries

(`optimizers/mod.rs`)

The `LocalModeOptimizer` now skips LC wrapping when:

Output projection is empty (COUNT(*) served from metadata)
Output has >4 columns (wide projections where per-column overhead exceeds savings)
Any output or predicate column is string/binary/dictionary-of-string

6. Selectivity-based delegation in the opener

(`reader/plantime/opener.rs`, `reader/plantime/source.rs`)

When the opener estimates that fewer than 50% of rows survive pruning AND a predicate is present, it delegates to plain parquet instead of using the LC stream. Few matching rows means decode cost is already minimal — LC overhead would dominate.

Also adds:

Caller-provided `ParquetFileReaderFactory` for metadata reuse (avoids redundant metadata I/O)
Early exit when all row groups are pruned (returns empty stream immediately)

7. Two-phase cache read with batched fallback

(`reader/runtime/liquid_cache_reader.rs`)

Previously, each column miss triggered an independent parquet read. Now:

Phase 1: Try cache for all projected columns, track hits/misses
Phase 2: If any miss, do ONE parquet read for all missed columns
Cache fill is async (spawned on tokio) — does not block the query hot path

8. Fix: cache disabled when no predicate

(`reader/runtime/liquid_stream.rs`)

When no predicate is present, `predicate_column_ids` was empty, causing `create_row_group` to treat all columns as non-predicate (uncacheable). Fixed to use `cache_column_ids` as the predicate set when no filter exists — so all projected columns become cacheable.

Benchmark Results (100M row ClickBench, 3 runs, warm cache)

Category	Count	Best speedup
Faster	10 queries	q04: 3.8x, q03: 2.3x, q02/q08: 1.2x
Neutral (within noise)	29 queries	—
Regression	0 queries	—

…on-only columns Signed-off-by: Tanvir Alam <tanvralm@amazon.com>

Signed-off-by: Tanvir Alam <tanvralm@amazon.com>

…local crate

Remove LiquidCacheParquetRef from the private use statement (line 18) since it's already made available via pub use (line 24). Having it in both causes E0252 (name defined multiple times). Also consolidate both pub use statements into a single line. Signed-off-by: Bukhtawar Khan <bukhtawa@amazon.com>

Fix re-export conflict: LiquidCacheParquetRef defined twice

Two fixes for the OpenSearch indexed-table integration path where predicate=None (filtering is handled externally by the BoolNode evaluator): 1. When no row filter is present, treat all projected columns as cacheable (predicate_column_ids = cache_column_ids). Previously, an empty predicate_column_ids meant is_predicate_column=false for all columns, causing get/insert to always bail out — the cache was effectively a no-op passthrough with pure overhead. 2. Move cache insert (Arrow→Liquid transcoding) into tokio::spawn so it runs asynchronously. The query batch is returned immediately without waiting for cache population. This ensures cache MISS has near-zero overhead vs the non-LC path. Together these ensure: repeated numeric queries get cache HITs (served from in-memory Arrow arrays), first execution has minimal overhead, and string columns are still correctly rejected by the is_string_type guard. Signed-off-by: Bukhtawar Khan <bukhtawa@amazon.com>

Fix cache disabled without predicate and make insert async

Two fixes for the OpenSearch indexed-table integration path where predicate=None (filtering is handled externally by the BoolNode evaluator): 1. When no row filter is present, treat all projected columns as cacheable (predicate_column_ids = cache_column_ids). Previously, an empty predicate_column_ids meant is_predicate_column=false for all columns, causing get/insert to always bail out — the cache was effectively a no-op passthrough with pure overhead. 2. Move cache insert (Arrow→Liquid transcoding) into tokio::spawn so it runs asynchronously. The query batch is returned immediately without waiting for cache population. This ensures cache MISS has near-zero overhead vs the non-LC path. Together these ensure: repeated numeric queries get cache HITs (served from in-memory Arrow arrays), first execution has minimal overhead, and string columns are still correctly rejected by the is_string_type guard. Signed-off-by: Bukhtawar Khan <bukhtawa@amazon.com>

Refactor read_from_cache: phase 1/phase 2 for future split-projection

Two fixes for the OpenSearch indexed-table integration path where predicate=None (filtering is handled externally by the BoolNode evaluator): 1. When no row filter is present, treat all projected columns as cacheable (predicate_column_ids = cache_column_ids). Previously, an empty predicate_column_ids meant is_predicate_column=false for all columns, causing get/insert to always bail out — the cache was effectively a no-op passthrough with pure overhead. 2. Move cache insert (Arrow→Liquid transcoding) into tokio::spawn so it runs asynchronously. The query batch is returned immediately without waiting for cache population. This ensures cache MISS has near-zero overhead vs the non-LC path. Together these ensure: repeated numeric queries get cache HITs (served from in-memory Arrow arrays), first execution has minimal overhead, and string columns are still correctly rejected by the is_string_type guard. Signed-off-by: Bukhtawar Khan <bukhtawa@amazon.com>

…conflict

Selectivity-gated LC with metadata passthrough and all-cols-cacheable

codacy-production · 2026-06-08T10:39:32Z

Up to standards ✅

🟢 Issues 6 high

Results:
6 new issues

Category Results

Security 6 high

View in Codacy

🟢 Metrics 279 complexity · 13 duplication

Metric Results

Complexity 279

Duplication 13

View in Codacy

_{NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer}
_{TIP This summary will be updated as you push new changes.}

Bukhtawar · 2026-06-08T10:58:28Z


        let metadata_size_hint = partitioned_file.metadata_size_hint;
+        let has_predicate = self.predicate.is_some();
+        log::info!(


Lets format logs and change to debug

Sure we can change the log level from info to debug

Bukhtawar · 2026-06-08T10:59:27Z

+            // Estimate selectivity from row_selection: how many rows survived
+            // RG pruning + page index pruning vs total rows in selected RGs.
+            let total_rows: usize = row_group_indexes
+                .iter()
+                .map(|&idx| rg_metadata[idx].num_rows() as usize)
+                .sum();
+            let selected_rows = row_selection.as_ref()
+                .map(|sel| sel.row_count())
+                .unwrap_or(total_rows);
+            let estimated_selectivity = if total_rows > 0 {
+                selected_rows as f64 / total_rows as f64
+            } else {
+                1.0
+            };


Lets generalize this as a generic policy that can be pushed

Done — extracted into a CacheEngagementPolicy trait. Consumers can push custom policies through
.with_engagement_policy(threshold)

…ation

codecov · 2026-06-09T15:05:00Z

Codecov Report

❌ Patch coverage is 69.07631% with 77 lines in your changes missing coverage. Please review.
✅ Project coverage is 29.43%. Comparing base (4597d1d) to head (880bdfc).
⚠️ Report is 3 commits behind head on main.

Files with missing lines	Patch %	Lines
src/datafusion/src/reader/plantime/opener.rs	64.61%	21 Missing and 2 partials ⚠️
src/datafusion/src/reader/plantime/source.rs	30.00%	14 Missing ⚠️
...atafusion/src/reader/plantime/engagement_policy.rs	57.69%	10 Missing and 1 partial ⚠️
src/core/src/cache/policies/cache/lru.rs	0.00%	10 Missing ⚠️
src/core/src/cache/budget.rs	56.25%	7 Missing ⚠️
src/datafusion-local/src/lib.rs	33.33%	4 Missing ⚠️
src/datafusion/src/cache/column.rs	80.00%	4 Missing ⚠️
...tafusion/src/reader/runtime/liquid_cache_reader.rs	91.89%	1 Missing and 2 partials ⚠️
src/datafusion/src/optimizers/mod.rs	97.29%	1 Missing ⚠️

❗ There is a different number of reports uploaded between BASE (4597d1d) and HEAD (880bdfc). Click for more details.

HEAD has 1 upload less than BASE

Flag BASE (4597d1d) HEAD (880bdfc)

4 3

Additional details and impacted files

@@             Coverage Diff             @@
##             main     #505       +/-   ##
===========================================
- Coverage   83.10%   29.43%   -53.68%     
===========================================
  Files          86       78        -8     
  Lines       19613    13309     -6304     
  Branches    19613    13309     -6304     
===========================================
- Hits        16300     3917    -12383     
- Misses       2974     9226     +6252     
+ Partials      339      166      -173

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

Adds a method that sets the predicate for row_filter (per-row filtering during decode) and RG-level pruning predicate, but deliberately skips page_pruning_predicate. This is used by the indexed path where the BoolNode's RowSelection is authoritative and page-level statistics must not override it.

Tanvir Alam and others added 14 commits May 21, 2026 16:11

Cache only numeric predicate columns, skip string/binary and projecti…

a59a555

…on-only columns Signed-off-by: Tanvir Alam <tanvralm@amazon.com>

Add LRU cache eviction policy with type-aware queues

a2138d4

Signed-off-by: Tanvir Alam <tanvralm@amazon.com>

Add with_max_disk_bytes to LiquidCacheLocalBuilder

62f9ab1

Signed-off-by: Tanvir Alam <tanvralm@amazon.com>

Make cache budget limits dynamically resizable at runtime

e7484d3

Signed-off-by: Tanvir Alam <tanvralm@amazon.com>

Export LiquidParquetSource and LiquidCacheParquetRef from datafusion-…

c941540

…local crate

Merge pull request #1 from Bukhtawar/fix-reexport-conflict

63a4caa

Fix re-export conflict: LiquidCacheParquetRef defined twice

Merge pull request #2 from Bukhtawar/fix-reexport-conflict

91b4429

Fix cache disabled without predicate and make insert async

Merge pull request #3 from Bukhtawar/fix-reexport-conflict

e223a24

Refactor read_from_cache: phase 1/phase 2 for future split-projection

Merge branch 'liquid-cache-opensearch-integration' into fix-reexport-…

f876600

…conflict

Merge pull request #4 from Bukhtawar/fix-reexport-conflict

c371e28

Selectivity-gated LC with metadata passthrough and all-cols-cacheable

This was referenced Jun 8, 2026

Selective Column Admission for Improved Cache Efficiency Under Constrained Memory Budgets #500

Open

Support cache pass-through for selective scans #485

Open

Bukhtawar reviewed Jun 8, 2026

View reviewed changes

cocosz force-pushed the liquid-cache-opensearch-integration branch 3 times, most recently from f30dc3a to fdcfca8 Compare June 8, 2026 19:35

Add CacheEngagementPolicy trait for pluggable selectivity-based deleg…

8a85077

…ation

cocosz force-pushed the liquid-cache-opensearch-integration branch from fdcfca8 to 8a85077 Compare June 8, 2026 19:36

Change LC log statements from info to debug level

1d5a668

cocosz force-pushed the liquid-cache-opensearch-integration branch from 04a6d4a to 1d5a668 Compare June 9, 2026 14:33

cocosz force-pushed the liquid-cache-opensearch-integration branch from 0160083 to 1d5a668 Compare June 9, 2026 23:04

cocosz force-pushed the liquid-cache-opensearch-integration branch from 9de5160 to 1c258b7 Compare June 11, 2026 07:32

cocosz force-pushed the liquid-cache-opensearch-integration branch from 1c258b7 to 9de5160 Compare June 11, 2026 09:09

Navigation Menu

OpenSearch integration: selective caching, LRU eviction, dynamic budgets - #505

OpenSearch integration: selective caching, LRU eviction, dynamic budgets#505
cocosz wants to merge 17 commits into
datafusion-contrib:mainfrom
cocosz:liquid-cache-opensearch-integration

cocosz commented Jun 8, 2026 •

edited

Loading

Uh oh!

codacy-production Bot commented Jun 8, 2026 •

edited

Loading

Uh oh!

Bukhtawar Jun 8, 2026

Uh oh!

cocosz Jun 8, 2026

Uh oh!

Bukhtawar Jun 8, 2026

Uh oh!

cocosz Jun 8, 2026

Uh oh!

codecov Bot commented Jun 9, 2026 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

cocosz commented Jun 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes

1. Selective column caching

2. LRU eviction policy with type-aware queues

3. Dynamic budget resize

4. Builder API extensions

5. Optimizer gating: skip uncacheable queries

6. Selectivity-based delegation in the opener

7. Two-phase cache read with batched fallback

8. Fix: cache disabled when no predicate

Benchmark Results (100M row ClickBench, 3 runs, warm cache)

Uh oh!

codacy-production Bot commented Jun 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Up to standards ✅

Uh oh!

Bukhtawar Jun 8, 2026

Choose a reason for hiding this comment

Uh oh!

cocosz Jun 8, 2026

Choose a reason for hiding this comment

Uh oh!

Bukhtawar Jun 8, 2026

Choose a reason for hiding this comment

Uh oh!

cocosz Jun 8, 2026

Choose a reason for hiding this comment

Uh oh!

codecov Bot commented Jun 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

cocosz commented Jun 8, 2026 •

edited

Loading

codacy-production Bot commented Jun 8, 2026 •

edited

Loading

codecov Bot commented Jun 9, 2026 •

edited

Loading