Unify analytics cache interfaces behind one management plane

[alchemist51]

Description

Unifies the interfaces of all four DataFusion analytics-backend caches (footer metadata, statistics, column-index, offset-index) so a future eviction-policy change is a one-place plug-in — without changing any eviction policy or behavior, and without adding any new dependency (no foyer).

Previously each cache had its own eviction trait, stats shape, and per-type match arms in the cache manager and FFI layer, so introducing a new policy (e.g. S3-FIFO) meant touching four places.

1. One eviction-policy trait

EvictionPolicy<K> replaces the two previous traits (CachePolicy with hardcoded String keys for the statistics cache, and ScopedEvictionPolicy<K> private to BoundedCache). LruPolicy, LfuPolicy, and FifoPolicy all implement it; the eviction protocol is unified on next_victim() (pop one candidate), so the statistics cache's batch select_for_eviction API is gone. FifoPolicy moves from cache_store.rs into eviction_policy.rs and swaps its UnsafeCell for an uncontended parking_lot::Mutex so it is sound as a shared trait object.

Adding a new policy = one trait impl + one CacheEvictionPolicy variant.

2. New cache/unified.rs — the management plane

• CacheKind — closed enum for the four cache types. Parses the Java FFI type string once, and validate_policy() centralizes the per-cache validity rules that used to be scattered as match-arm guards in df_create_cache (statistics = LRU/LFU, CI/OI = FIFO-only, metadata = accepted but not forwarded).
• ManagedCache — the uniform lifecycle/observability surface (stats, set_limit, clear, reset_counters, remove_file, contains_file) with a shared CacheStats snapshot. All four caches implement it; the process-global CI/OI singletons via zero-sized handles.

Query-path get/put intentionally stays on each cache's typed API — the key/value shapes differ per cache and erasing them would cost hot-path performance for no management benefit.

3. Registry-based CustomCacheManager

The manager now holds a CacheKind → Arc<dyn ManagedCache> registry; clear_cache_type, get_memory_consumed_by_type, contains_file_by_type, remove_files, and the ~10 per-cache stat accessors collapse into registry dispatch. df_create_cache and df_cache_manager_update_size_limit route via CacheKind::parse; the latter now accepts all four kinds (previously METADATA/STATISTICS only). The separate CI/OI limit FFIs are unchanged.

Non-changes

• No new dependencies; no foyer.
• Java side untouched: same settings, same FFI signatures, same CacheType enum.
• Eviction behavior, sizes, and defaults are identical.
• One per-type correction: clear_cache_type("COLUMN_INDEX") now clears only CI instead of both CI+OI (the Java caller clears both via the same path, so nothing user-visible changes).

Related Issues

Interface-unification slice of #22282 (which additionally swapped the implementations to foyer + S3-FIFO); this PR does only the unification so policy changes can land separately.

Check List

• Functionality includes unit tests — cargo test --lib: 1273 passed; rustfmt + clippy clean on touched files; plugin compileJava/compileTestJava/test pass.
• Commits are signed per the DCO using --signoff.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 5af614b)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Semantics Change ColumnIndexCacheHandle::reset_counters and OffsetIndexCacheHandle::reset_counters call clear_keep_limit(), which drops all cached entries in addition to resetting counters. The trait doc for reset_counters states it should reset diagnostic counters "without touching entries". Callers invoking reset_cache_stats(CacheKind::ColumnIndex/OffsetIndex) will unexpectedly wipe the entire page-index cache. Consider adding a counter-only reset to BoundedCache or documenting this as a known limitation. fn reset_counters(&self) { // clear_keep_limit is the only counter reset the scoped caches expose; // entries and counters reset together (as before this interface). COLUMN_INDEX_CACHE.clear_keep_limit(); } fn remove_file(&self, file_path: &str) -> bool { COLUMN_INDEX_CACHE.evict_by_prefix(file_path) } fn contains_file(&self, _file_path: &str) -> bool { // Cell-keyed cache — a per-file scan is not worth a boolean probe; // approximate with "has any entries" (diagnostics only, as before). COLUMN_INDEX_CACHE.stats().entries > 0 } } impl ManagedCache for OffsetIndexCacheHandle { fn kind(&self) -> CacheKind { CacheKind::OffsetIndex } fn stats(&self) -> CacheStats { OFFSET_INDEX_CACHE.stats() } fn set_limit(&self, limit: usize) { set_offset_index_cache_limit(limit); } fn clear(&self) { OFFSET_INDEX_CACHE.clear_keep_limit(); } fn reset_counters(&self) { OFFSET_INDEX_CACHE.clear_keep_limit(); } Performance Regression LruPolicy::next_victim / LfuPolicy::next_victim perform an O(n) DashMap scan per call. When the owning cache evicts to free bytes it may call next_victim repeatedly, giving O(n·k) work per insert where the previous batched select_for_eviction sorted once (O(n log n)) and returned the whole set. Under bulk insert / limit-shrink paths on large statistics caches, this could noticeably increase latency. fn next_victim(&self) -> Option<K> { // Least recently accessed entry. O(n) scan per victim — same asymptotic // work as the previous batch-sort selection, spread across calls. let victim = self .entries .iter() .min_by_key(|entry| entry.value().last_accessed) .map(|entry| entry.key().clone())?; if let Some((_, entry)) = self.entries.remove(&victim) { self.total_size.fetch_sub(entry.size, Ordering::Relaxed); } Some(victim) } Behavior Change evict_victims uses Path::from(victim.as_str()) to look up the cache entry, but keys stored by the policy come from k.to_string() where k: &Path. If Path::to_string() isn't an exact roundtrip through Path::from(&str) for all inputs (e.g. leading slashes or encoded characters), evictions will silently fail to remove entries and byte accounting will drift. The previous implementation used parse_key_to_path which did the same thing, but this is worth verifying for the paths actually stored. fn evict_victims(&self, target_size: usize) -> usize { if target_size == 0 { return 0; } let policy = self.policy(); let mut freed_size = 0; while freed_size < target_size { let Some(victim) = policy.next_victim() else { break; }; let path = Path::from(victim.as_str()); if self.inner_cache.remove(&path).is_some() { if let Ok(mut state) = self.memory_state.lock() { if let Some(old_size) = state.tracker.remove(&victim) { state.total = state.total.saturating_sub(old_size); freed_size += old_size; } } } } freed_size Log Noise remove_files logs a debug message for every cache that doesn't contain the file. Since every removal iterates all four registered caches and typically only one (metadata or statistics) contains the file, a single removal of N files now emits up to 3N "File not found" debug entries. Consider only logging when no cache had the file, or dropping the per-cache miss log. for cache in self.caches.values() { if cache.remove_file(file_path) { any_removed = true; } else { log_debug!( "[CACHE INFO] File not found in {} cache: {}", cache.kind(), file_path ); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 5af614b
Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Avoid clearing entries in reset_counters reset_counters is documented as resetting hit/miss counters without touching entries, but here (and in the offset-index handle) it calls clear_keep_limit, which drops all cached entries. This silently wipes cache contents whenever a caller invokes reset_cache_stats(CacheKind::ColumnIndex/OffsetIndex). Either implement a counter-only reset on BoundedCache or make this a no-op with a doc note. sandbox/plugins/analytics-backend-datafusion/rust/src/cache/page_index/mod.rs [218-222] fn reset_counters(&self) { - // clear_keep_limit is the only counter reset the scoped caches expose; - // entries and counters reset together (as before this interface). - COLUMN_INDEX_CACHE.clear_keep_limit(); + // BoundedCache does not currently expose a counter-only reset; + // treat this as a no-op rather than silently dropping entries. } Suggestion importance[1-10]: 8 __ Why: Strong catch: the trait contract says reset_counters should reset counters without touching entries, but the implementation calls clear_keep_limit which drops all entries. This is a real behavioral bug that would cause data loss on stats reset.	Medium
	Handle poisoned lock during eviction accounting When inner_cache.remove succeeds but the memory_state lock is poisoned, the entry is dropped from the store but state.tracker/state.total are never updated, permanently desynchronizing accounting. Use unwrap_or_else(|e| e.into_inner()) (as elsewhere in this file) so poisoning doesn't leak leaked bytes into total. sandbox/plugins/analytics-backend-datafusion/rust/src/cache/statistics_cache.rs [235-256] fn evict_victims(&self, target_size: usize) -> usize { if target_size == 0 { return 0; } let policy = self.policy(); let mut freed_size = 0; while freed_size < target_size { let Some(victim) = policy.next_victim() else { break; }; let path = Path::from(victim.as_str()); if self.inner_cache.remove(&path).is_some() { - if let Ok(mut state) = self.memory_state.lock() { - if let Some(old_size) = state.tracker.remove(&victim) { - state.total = state.total.saturating_sub(old_size); - freed_size += old_size; - } + let mut state = self + .memory_state + .lock() + .unwrap_or_else(|e| e.into_inner()); + if let Some(old_size) = state.tracker.remove(&victim) { + state.total = state.total.saturating_sub(old_size); + freed_size += old_size; } } } freed_size } Suggestion importance[1-10]: 5 __ Why: Valid concern about lock poisoning causing accounting desync, though the original code also used if let Ok. The fix aligns with the pattern used elsewhere in the file (policy() method uses unwrap_or_else).	Low
General	Ensure iterator locks released before remove Calling min_by_key while holding DashMap iterator locks and then invoking self.entries.remove(&victim) on the same map risks a deadlock on the same shard (DashMap iterator holds read locks). Drop the iterator before removing by collecting the key first, which is already done, but ensure min_by_key fully consumes the iterator (it does) — however the same concern applies to LfuPolicy::next_victim. Verify DashMap semantics or explicitly scope the iterator to release locks before remove. sandbox/plugins/analytics-backend-datafusion/rust/src/cache/eviction_policy.rs [190-202] fn next_victim(&self) -> Option<K> { - // Least recently accessed entry. O(n) scan per victim — same asymptotic - // work as the previous batch-sort selection, spread across calls. - let victim = self - .entries - .iter() - .min_by_key(|entry| entry.value().last_accessed) - .map(|entry| entry.key().clone())?; + let victim = { + self.entries + .iter() + .min_by_key(|entry| entry.value().last_accessed) + .map(|entry| entry.key().clone())? + }; if let Some((_, entry)) = self.entries.remove(&victim) { self.total_size.fetch_sub(entry.size, Ordering::Relaxed); } Some(victim) } Suggestion importance[1-10]: 4 __ Why: The concern about DashMap iterator/remove deadlock is worth verifying, but the existing code already collects the key before removing (the iterator is consumed by min_by_key). The suggested change is essentially identical to existing code.	Low
	Preserve error reporting on file removal The previous implementation surfaced per-cache errors (e.g. metadata lock failure) and treated "cache configured but errored" differently from "not present". The new loop swallows all failures into a boolean and unconditionally reports success/failure with no error propagation, which can mask real cache-manager bugs. Consider returning Err or logging at error level when a ManagedCache operation fails, or add an error-reporting return from remove_file. sandbox/plugins/analytics-backend-datafusion/rust/src/cache/custom_cache_manager.rs [319-335] for file_path in file_paths { let mut any_removed = false; for cache in self.caches.values() { - if cache.remove_file(file_path) { - any_removed = true; - } else { - log_debug!( + match cache.remove_file(file_path) { + true => any_removed = true, + false => log_debug!( "[CACHE INFO] File not found in {} cache: {}", cache.kind(), file_path - ); + ), } } results.push((file_path.clone(), any_removed)); } Suggestion importance[1-10]: 3 __ Why: Minor observation about reduced error granularity, but the improved_code is essentially unchanged from existing_code and the ManagedCache trait's remove_file returns bool by design, so this is not actionable.	Low

---

Previous suggestions

Suggestions up to commit 5af614b

Category	Suggestion	Impact
Possible issue	Avoid clearing entries in reset_counters reset_counters is documented as resetting hit/miss diagnostic counters without touching entries, but this implementation clears the entire cache. This will cause unintended data loss when callers invoke reset_cache_stats(ColumnIndex) (e.g. metadata_cache_reset_stats-style calls). Either add a counters-only reset to BoundedCache or make this a no-op, but do not wipe entries. sandbox/plugins/analytics-backend-datafusion/rust/src/cache/page_index/mod.rs [219-222] fn reset_counters(&self) { - // clear_keep_limit is the only counter reset the scoped caches expose; - // entries and counters reset together (as before this interface). - COLUMN_INDEX_CACHE.clear_keep_limit(); + // Scoped caches do not currently expose a counters-only reset; + // avoid wiping entries here to honor the ManagedCache contract. } Suggestion importance[1-10]: 8 __ Why: Strong catch: reset_counters is documented as not touching entries, but the implementation calls clear_keep_limit() which wipes all entries. This is a real behavioral bug that could cause data loss when stats are reset.	Medium
	Drop DashMap iterator before remove Holding a DashMap::iter() reference across the subsequent entries.remove(&victim) call can deadlock, because iter() holds shard read guards and remove requires a write guard on the same shard. Ensure the iterator is fully dropped before calling remove by binding the result to a local before the remove call (the current .map(...).clone()? chain drops the guard, but making it explicit avoids future regressions and is safer under refactoring). Same concern applies to the LFU min_by variant. sandbox/plugins/analytics-backend-datafusion/rust/src/cache/eviction_policy.rs [190-202] fn next_victim(&self) -> Option<K> { - // Least recently accessed entry. O(n) scan per victim — same asymptotic - // work as the previous batch-sort selection, spread across calls. - let victim = self - .entries - .iter() - .min_by_key(|entry| entry.value().last_accessed) - .map(|entry| entry.key().clone())?; + let victim = { + let iter_min = self + .entries + .iter() + .min_by_key(|entry| entry.value().last_accessed)?; + iter_min.key().clone() + }; if let Some((_, entry)) = self.entries.remove(&victim) { self.total_size.fetch_sub(entry.size, Ordering::Relaxed); } Some(victim) } Suggestion importance[1-10]: 4 __ Why: The concern about DashMap deadlock is valid in principle, but the existing code's .map(|entry| entry.key().clone())? already drops the guard before remove. The suggestion is largely defensive/stylistic.	Low
General	Handle poisoned lock consistently in eviction next_victim pops the key from the policy's bookkeeping, so if inner_cache.remove returns None (stale victim already removed) we correctly skip. But if inner_cache.remove succeeds while memory_state.lock() fails (poisoning), we permanently lose the entry from the cache while leaving state.total inconsistent, and freed_size never advances — the loop can spin until the policy is empty. Use unwrap_or_else(|e| e.into_inner()) (matching policy()'s pattern) so accounting stays consistent with the store. sandbox/plugins/analytics-backend-datafusion/rust/src/cache/statistics_cache.rs [235-255] fn evict_victims(&self, target_size: usize) -> usize { if target_size == 0 { return 0; } let policy = self.policy(); let mut freed_size = 0; while freed_size < target_size { let Some(victim) = policy.next_victim() else { break; }; let path = Path::from(victim.as_str()); if self.inner_cache.remove(&path).is_some() { - if let Ok(mut state) = self.memory_state.lock() { - if let Some(old_size) = state.tracker.remove(&victim) { - state.total = state.total.saturating_sub(old_size); - freed_size += old_size; - } + let mut state = self + .memory_state + .lock() + .unwrap_or_else(|e| e.into_inner()); + if let Some(old_size) = state.tracker.remove(&victim) { + state.total = state.total.saturating_sub(old_size); + freed_size += old_size; } } } freed_size } Suggestion importance[1-10]: 5 __ Why: Valid edge-case: if the mutex is poisoned after removing from inner_cache, accounting becomes inconsistent and the loop may spin. Handling poisoning consistently with policy() improves robustness, though poisoning is rare.	Low
	Iterate caches in deterministic order Iterating self.caches.values() uses HashMap iteration order, which is non-deterministic. This changes observable log ordering across runs and can subtly affect test stability. Iterate over CacheKind::ALL (or a sorted order) and look up each kind for deterministic behavior. sandbox/plugins/analytics-backend-datafusion/rust/src/cache/custom_cache_manager.rs [322-332] -for cache in self.caches.values() { - if cache.remove_file(file_path) { - any_removed = true; - } else { - log_debug!( - "[CACHE INFO] File not found in {} cache: {}", - cache.kind(), - file_path - ); +for kind in CacheKind::ALL { + if let Some(cache) = self.managed(kind) { + if cache.remove_file(file_path) { + any_removed = true; + } else { + log_debug!( + "[CACHE INFO] File not found in {} cache: {}", + cache.kind(), + file_path + ); + } } } Suggestion importance[1-10]: 3 __ Why: Minor improvement for log/test determinism. Functionally the code is correct; the impact is limited to log ordering.	Low

[github-actions]

Persistent review updated to latest commit 5af614b

[github-actions]

❌ Gradle check result for 5af614b: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?