Offset and column index cache#66
Open
bharath-techie wants to merge 18 commits into
Open
Conversation
Add indexed_table::page_index_loader — a process-global, byte-bounded LRU that loads the parquet page index scoped to a query's predicate columns and is shared by both scan paths (DataFusion ListingTable + custom indexed executor). No scan-path wiring yet; this lands the cache + its unit tests in isolation. Design: - ColumnIndex scoped to predicate columns (NONE placeholders elsewhere); OffsetIndex kept for all columns / all row groups (read at scan time for every projected column — placeholders there break reads). - Cache stores the decoded (ColumnIndex, OffsetIndex) PAIR + a structural size estimate, NOT a full ParquetMetaData, and grafts the pair onto the caller's resident footer at lookup. No footer duplication / over-allocation. - Key is (path, predicate parquet-cols) ONLY — both paths build the identical all-RG artifact, so entries are shared across paths. Row-group scoping is a future axis (gated Step 2/3) and is deliberately not in the key yet. - Cache hit returns a fresh Arc (graft), so hits are asserted via the stats counter, not Arc::ptr_eq. Uses the deprecated read_columns_indexes/read_offset_indexes — the only public subset-decode API today; arrow-rs#8643/opensearch-project#8797/opensearch-project#8714 track the first-class replacement. 9 unit tests pass (cargo test --lib page_index_loader --test-threads=1).
…tting
Make the scoped cache (Step 1a) observable for A/B testing, and let its byte
budget be configured.
Rust:
- stats.rs: CacheStatsRepr grows a third CacheGroupRepr (scoped_page_index_cache),
packed from page_index_loader::scoped_cache_stats(). DfStatsBuffer 75->80 longs
(600->640 bytes); asserts + tests updated. New test asserts the packed group
reflects the live global counters/limit.
- ffm.rs: df_set_scoped_page_index_cache_limit(i64) — process-global, no manager
pointer; negative rejected, zero ignored.
Java:
- StatsLayout: cache_stats group now nests 3 sub-caches (metadata, statistics,
scoped_page_index); layout 80 longs/640 bytes; 5 new VarHandles; readCacheStats
builds the 3rd group.
- CacheStats: third CacheGroupStats threaded through ctor/StreamInput/writeTo/
XContent ("scoped_page_index_cache")/equals/hashCode + getter.
- NativeBridge.setScopedPageIndexCacheLimit -> df_set_scoped_page_index_cache_limit.
- CacheSettings datafusion.scoped_page_index.cache.size.limit (default 64mb);
CacheUtils pushes it to native at startup (NodeScope, not yet dynamic).
- Updated StatsLayout(Property)Tests offsets/FIELD_COUNT (75->80) + CacheStats(*)
tests for the 3-arg shape.
Tests: 19 Rust stats + 9 page_index_loader pass; 7 Java stats suites (57 tests)
pass under -PrustDebug -Dsandbox.enabled=true. No scan-path behavior change yet.
Populate the unified scoped cache (1a) on the DataFusion ListingTable / vanilla parquet scan path so it is exercised + observable via node-stats (1b). - shard_scoped_reader.rs: ScopedPageIndexReaderFactory — a ParquetFileReaderFactory whose get_metadata loads footer-only metadata, resolves predicate column names to per-file parquet leaf indices, and grafts an all-RG, column-scoped page index via load_scoped_page_index (falls back to footer-only on no-predicate / failure). All row groups (not RG-scoped): DataFusion owns RG selection on this path, so a placeholder on an RG it scans would panic. - scoped_page_index_optimizer.rs: ScopedPageIndexOptimizer — a PhysicalOptimizerRule that finds each parquet DataSourceExec, reads the pushed predicate, derives predicate column names, and REPLACES the reader factory (DataFusion always pre-installs its own full-index CachedParquetFileReaderFactory — that is what we swap). Provider-agnostic; no-op without a predicate. - session_context.rs: register the rule after ProjectRowIdOptimizer so it sees the final DataSourceExec, using this shard's store + the shared metadata cache. Tests (15 pass, --test-threads=1): the optimizer end_to_end_listing_scan_fills_ scoped_cache drives a real SessionContext + stock ListingTable and asserts the scoped cache fills with a miss; reader tests assert predicate-scoped ColumnIndex + all-column OffsetIndex and the no-predicate no-scope path. No metadata-cache strip yet (1e) — scoped-cache population does not depend on it.
ScopedPageIndexCacheIT provisions app_logs (parquet/composite) and exercises the listing-table scan path (numeric predicate — native, not Lucene- delegated), asserting via GET /_plugins/_analytics_backend_datafusion/stats (cache_stats.scoped_page_index_cache): - the scoped group is always exposed with a positive size_limit_bytes (64mb); - a listing query populates the cache (entry + memory_bytes > 0) at QUERY time; - re-running the IDENTICAL query is a pure cache hit: hits increase while misses, entries, and memory_bytes stay flat (no duplication / no over-allocation); - 10 repeated runs do not grow entries or bytes (bounded, predictable). Assertions are same-method twice-run deltas, not cold-cache baselines: the scoped cache is a process-global singleton with no reset endpoint and IT methods run in randomized order on a preserved cluster, so an entry may already exist from a prior method. Verified the scoped cache is populated ONLY at query time — its sole writer (load_scoped_page_index) is reachable only from the query-time scoped reader, never the refresh/warm path (custom_cache_manager only warms the level-1 metadata cache). 3/3 pass under -Dsandbox.enabled=true -PrustDebug.
…r-only) Make the level-1 metadata cache footer-only (row-group + file stats), so the heavy decoded page index is never retained there. This is the flip that makes the scoped page-index cache (1a/1c) load-bearing and kills the wide-schema heap blowup. - cache.rs: strip_page_index(entry) downcasts CachedParquetMetaData and rebuilds it with set_column_index(None)/set_offset_index(None); no-op (same Arc) if already footer-only. Called in MutexFileMetadataCache::put — the single chokepoint every put flows through (DataFusion scan opener, infer_schema, fetch_statistics, and the warm path all force-decode the full index before put; we refuse to retain it). - custom_cache_manager.rs: corrected the warm-path comment — warming now lands a footer-only entry through the same stripping put. Correctness: page pruning is an optimization; PagePruner::prune_rg early-returns (scans the whole RG) when no page index is present, never a wrong result. Both scan paths rebuild a predicate-scoped page index per query via the shared scoped cache, at query time only — ColumnIndex is never warmed at refresh. Tests: 2 new strip tests (put_strips_page_index_and_get_returns_footer_only, strip_is_noop_for_footer_only_entry); 295 indexed_table::tests_e2e pass with the strip in place (correctness preserved); cache/page_index_loader/shard_scoped_ reader/optimizer/stats all green.
…age index The decisive fix: do not merely strip the page index after the fact — never decode it for the level-1 cache in the first place. DataFusion 54 hardcodes the page-index policy in fetch_metadata: cache present => PageIndexPolicy::Optional (force-decodes the FULL ColumnIndex+OffsetIndex before caching), no cache => Skip. So passing our cache made it decode the whole page index every footer load, only for put() to throw it away. On wide schemas that decode is the dominant cost the whole effort targets. - parquet_bridge::load_parquet_metadata now manages the cache itself: cache get + is_valid_for => return cached footer (zero I/O, zero decode); miss => fetch WITHOUT a cache (PageIndexPolicy::Skip => page index never decoded), then put a footer-only entry. This is the loader both scoped-reader paths and the warm path use. - custom_cache_manager: warm path now delegates to load_parquet_metadata (footer-only, no decode) instead of DFParquetMetadata::fetch_metadata-with-cache. - cache.rs MutexFileMetadataCache::put keeps strip_page_index as a defensive backstop for the scan paths DataFusion drives directly (its own opener / infer_schema), which still construct fetch_metadata-with-cache out of our reach. Correctness preserved: PagePruner no-ops without a page index; scoped page index is rebuilt per query for predicate columns only. 49 unit + 295 indexed e2e pass.
Broaden ScopedPageIndexCacheIT from 3 to 8 tests so the page-index changes (scoped cache + footer-only level-1 cache) are verified to break nothing, all via GET /_plugins/_analytics_backend_datafusion/stats: Correctness (exact counts vs known app_logs cardinalities — 200 docs): - total=200; status>=400=103 (numeric listing predicate); log_level=ERROR=115 (keyword); answers stable across cold vs cached re-runs. Level-1 metadata cache still works after the strip: - holds footer entries and registers hits across repeated queries. Scoped cache (hits / misses / size): - same query re-run = pure hit (misses/entries/bytes flat); 10 repeats stay bounded; occupancy never exceeds the 64mb budget. No-breakage query sweep: - plain projection, grouped aggregation, multi-column native filter, full-text match() (Lucene-delegated), and mixed native+match predicate all execute. 8/8 pass under -Dsandbox.enabled=true -PrustDebug; zero panics/errors in node logs; caches configured metadata=250mb stats=100mb scoped=64mb.
After Step 1e the level-1 metadata cache is footer-only, so the indexed PagePruner (which reads column_index()/offset_index() straight off segment.metadata) would see no page index and conservatively no-op. Restore page pruning by re-augmenting each segment metadata with a predicate-scoped page index, built through the SAME shared scoped cache as the listing path. - indexed_executor.rs: after predicate columns are known, collect predicate column NAMES and, per segment, resolve_predicate_parquet_columns + load_scoped_page_index, replacing segment.metadata with the augmented copy. - Resolving NAMES -> per-file parquet leaf indices via the same resolve_predicate_parquet_columns the listing reader uses means both paths compute the identical scoped-cache key for the same (file, predicate columns), so entries are SHARED across paths (constraint #1). All row groups (no RG scoping in Step 1); on any failure the segment keeps footer-only metadata and pruning no-ops — never a wrong result. Tests: 295 indexed_table::tests_e2e pass with re-augmentation; page_index_loader / shard_scoped_reader / cache modules green.
Add testCrossPathSharingListingThenIndexed: a listing query filters `status`, then an indexed query (forced onto the indexed path by match(message,...)) also filters `status`. Asserts via the stats API that the indexed query HITS the scoped entry the listing query built — hits increase, entries and memory_bytes stay flat (no second, path-specific entry). This is constraint #1 (one unified cache shared by both paths), the central failure of the prior iteration, now verified on a real cluster. 9/9 ScopedPageIndexCacheIT pass under -Dsandbox.enabled=true -PrustDebug.
Reset the scoped cache and re-measure via stats without a cluster restart.
- page_index_loader::clear_scoped_cache() — drop entries + reset counters, keep
the byte budget (the next query repopulates lazily).
- ffm.rs df_clear_scoped_page_index_cache() FFI.
- NativeBridge.clearScopedPageIndexCache().
- RestClearScopedPageIndexCacheAction: POST
/_plugins/_analytics_backend_datafusion/cache/scoped_page_index/_clear
(local node; for the single-node benchmark cluster). Registered in
DataFusionPlugin.getRestHandlers.
Verified live on the 100M-doc clickbench index: populate -> 20 entries/8.1MB;
clear -> {"acknowledged":true} -> 0/0/0; repopulate -> identical 20/8.1MB.
Initial, test-backed draft of row-group scoping for the scoped page index, grounded in a close read of DF54 + parquet58 source (HANDOFF_step2_rg_scoping.md has the full analysis + citations). NOT wired into the scan paths yet — the live paths still call the unchanged all-RG load_scoped_page_index. Design (safe per the source analysis): - ColumnIndex (the heap hog: per-page string min/max) is RG-scopable — decode it only for row groups that pass footer RG-stats pruning; NONE elsewhere. NONE is handled gracefully by DF (statistics.rs get_data_page_statistics! _ arm) and the RGs we placeholder are exactly those DF also prunes. - OffsetIndex MUST stay all-RG: an empty OffsetIndex on any RG DataFusion scans panics (statistics.rs:1835 page_locations.last().unwrap()) or breaks reads (failed to skip rows). DF picks the scanned set itself, after + independent of our index load, so we cannot safely omit any RG OffsetIndex. - Survivor set = footer-stats prune (superset of DF scanned set, since DF further applies bloom/range/limit) — deterministic in (file, predicate), so both scan paths compute the identical set → identical key → cross-path sharing preserved. Code: - ScopedKey gains surviving_rgs (empty = Step-1 all-RG; non-empty = Step-2). - surviving_row_groups(footer, schema, predicate): PruningPredicate over a footer-stats PruningStatistics (mirrors DF RowGroupAccessPlanFilter; no page index); conservative all-RG fallback. - load_scoped_page_index_rgs + build_scoped_page_index refactor: ColumnIndex gated per-RG, OffsetIndex always all-RG; size estimate reflects the RG scope. - 4 unit tests (survivor calc, RG-scoped CI / all-RG OI, byte reduction, key identity). load_scoped_page_index unchanged → Step 1 fully backward compatible. Tests: 13 page_index_loader + 295 indexed e2e + cache/reader/optimizer all pass. Step 3 (RG-scope OffsetIndex) deferred — needs an adversarial repro; see handoff §4.
…+ col 0
Answer to "do we need the OffsetIndex for ALL columns?": no — only
predicate ∪ projection ∪ {0}. Verified against DF54/parquet58 source: the
OffsetIndex is dereferenced only by (a) reads, for PROJECTED columns
(in_memory_row_group.rs:80), (b) page pruning, for the PREDICATE column
(page_filter.rs:512), and (c) the page-skip METRIC, for column 0
(page_filter.rs:255 — affects only the counter, not the RowSelection). Including
col 0 keeps the metric identical to stock DataFusion.
DRAFT (not wired into scan paths):
- ScopedKey gains offset_cols (empty = all columns / Step 1).
- load_scoped_page_index_cols(.., offset_cols): defensively unions predicate cols
+ col 0, clamps/sorts/dedups; collapses to the all-columns key when the union
covers every column (shares the Step-1 entry).
- build_scoped_page_index decodes the OffsetIndex only for off_cols, scatters to
absolute column positions, empty placeholders elsewhere; size estimate follows.
- 4 unit tests: offset real only for the scoped set; projected non-predicate read
succeeds; cached bytes reduced; full-coverage collapse to all-columns entry.
This is orthogonal to RG-scoping (CI by row group) and is the bigger win on wide
schemas (402 cols, project a few). Wiring TODO: listing path has the projection in
the optimizer; indexed path must thread read_projection to the augmentation site
(both paths must pass the same offset_cols to keep cross-path sharing). See
HANDOFF_step2_rg_scoping.md §3b. load_scoped_page_index unchanged → Step 1 + the
RG-scoping draft remain backward compatible.
Tests: 17 page_index_loader + 295 indexed e2e + cache/reader/optimizer all pass.
Signed-off-by: G <bharath78910@gmail.com>
Signed-off-by: G <bharath78910@gmail.com>
Signed-off-by: G <bharath78910@gmail.com>
…r fixes Signed-off-by: G <bharath78910@gmail.com>
Signed-off-by: G <bharath78910@gmail.com>
Signed-off-by: G <bharath78910@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
[Describe what this change achieves]
Related Issues
Resolves #[Issue number to be closed when this PR is merged]
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.