[Sandbox] feat(datafusion): expose parquet metadata and statistics cache stats on the analytics backend stats endpoint

[HarishNarasimhanK]

Description

Adds cache_stats and search_stats to GET _plugins/analytics_backend_datafusion/stats.

"cache_stats": {
  "metadata_cache": {
    "hit_count": 0,
    "miss_count": 0,
    "hit_rate": 0.0,
    "entry_count": 0,
    "memory_bytes": 0,
    "size_limit_bytes": 262144000
  },
  "statistics_cache": {
    "hit_count": 0,
    "miss_count": 0,
    "hit_rate": 0.0,
    "entry_count": 0,
    "memory_bytes": 0,
    "size_limit_bytes": 104857600
  }
},
"search_stats": {
  "elapsed_compute_ms": 0,
  "delegation_calls": 0,
  "rg_processed": 0,
  "rg_skipped": 0,
  "prefetch_wait_time_ms": 0,
  "prefetch_wait_count": 0,
  "listing_table_scan": 0,
  "single_collector_scan": 0,
  "bitmap_tree_scan": 0
}

size_limit_bytes == 0 means the cache is disabled. All search_stats values are cumulative since node start.

Field	Description
elapsed_compute_ms	Total wall-clock time inside indexed stream poll_next
delegation_calls	Total Rust→Java FFM calls to the Lucene collector
rg_processed	Total row groups scanned
rg_skipped	Total row groups skipped (zero matching docs)
prefetch_wait_time_ms	Total time the poll thread blocked waiting for Lucene prefetch
prefetch_wait_count	Total times the poll thread parked for prefetch
listing_table_scan	Queries routed through the parquet-only scan path
single_collector_scan	Queries routed through the indexed single-collector path
bitmap_tree_scan	Queries routed through the indexed bitmap-tree path

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

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 1031399)

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 Race Condition The get method increments hit_count or miss_count after reading from the cache, but the cache read and counter increment are not atomic. If the cache is cleared between the read and the counter increment, the counters may become inconsistent with actual cache state. This can lead to incorrect hit rate calculations under concurrent cache operations. fn get(&self, k: &Path) -> Option<CachedFileMetadataEntry> { match self.inner.lock() { Ok(cache) => { let result = cache.get(k); if result.is_some() { self.hit_count.fetch_add(1, Ordering::Relaxed); } else { self.miss_count.fetch_add(1, Ordering::Relaxed); } result } Err(e) => { Race Condition Similar to the metadata cache, the get method increments counters after the cache lookup without atomicity. If eviction or clearing occurs between the lookup and counter update, the statistics become unreliable. This is particularly problematic since the cache supports concurrent eviction operations. impl CacheAccessor<Path, CachedFileMetadata> for CustomStatisticsCache { fn get(&self, k: &Path) -> Option<CachedFileMetadata> { let result = self.inner_cache.get(k); if result.is_some() { self.hit_count.fetch_add(1, Ordering::Relaxed); let key = k.to_string(); let memory_size = { let state = self.memory_state.lock(); state.map(|s| s.tracker.get(&key).copied().unwrap_or(0)).unwrap_or(0) }; if let Ok(mut policy_guard) = self.policy.lock() { policy_guard.on_access(&key, memory_size); } } else { self.miss_count.fetch_add(1, Ordering::Relaxed); } result.map(|s| s.value().clone()) } Possible Issue The AccumulatingStream wrapper always accumulates metrics on drop, even if the stream was never polled or failed during initialization. If stream creation fails after wrapping but before returning, or if the stream is dropped without being polled, metrics will be accumulated with zero or incomplete values, skewing the statistics. impl Drop for AccumulatingStream { fn drop(&mut self) { crate::search_stats::accumulate(&self.stream_metrics); } }

[github-actions]

Persistent review updated to latest commit 849f3c9

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 1031399

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Use stronger memory ordering for stats reset The reset_stats method uses Relaxed ordering which may not guarantee visibility of the reset across threads. Consider using Release ordering for the stores to ensure proper synchronization when resetting statistics that may be read by other threads. sandbox/plugins/analytics-backend-datafusion/rust/src/cache.rs [53-56] pub fn reset_stats(&self) { - self.hit_count.store(0, Ordering::Relaxed); - self.miss_count.store(0, Ordering::Relaxed); + self.hit_count.store(0, Ordering::Release); + self.miss_count.store(0, Ordering::Release); } Suggestion importance[1-10]: 4 __ Why: While using Release ordering could provide stronger guarantees, Relaxed ordering is acceptable for simple counter resets where strict synchronization isn't critical. The suggestion is valid but offers marginal improvement for non-critical statistics tracking.	Low
	Prevent potential double-counting of metrics The Drop implementation unconditionally accumulates metrics, which could lead to double-counting if the stream is cloned or if metrics are accumulated elsewhere. Consider adding a flag to track whether metrics have been accumulated to prevent duplicate accumulation. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/table_provider.rs [503-507] +struct AccumulatingStream { + inner: SendableRecordBatchStream, + stream_metrics: StreamMetrics, + metrics_accumulated: std::sync::atomic::AtomicBool, +} + impl Drop for AccumulatingStream { fn drop(&mut self) { - crate::search_stats::accumulate(&self.stream_metrics); + if !self.metrics_accumulated.swap(true, std::sync::atomic::Ordering::Relaxed) { + crate::search_stats::accumulate(&self.stream_metrics); + } } } Suggestion importance[1-10]: 3 __ Why: The concern about double-counting is theoretical. SendableRecordBatchStream is not Clone, and the AccumulatingStream wrapper is created once per partition execution. The added complexity of tracking accumulation state is unnecessary without evidence of actual double-counting issues.	Low

---

Previous suggestions

Suggestions up to commit ca91da8

Category	Suggestion	Impact
General	Protect Drop implementation from panics Performing potentially expensive operations in Drop can cause panics during unwinding to abort the process. Consider wrapping the accumulation in a panic guard or ensuring accumulate is panic-safe, especially since it involves atomic operations that could fail. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/table_provider.rs [503-506] impl Drop for AccumulatingStream { fn drop(&mut self) { - crate::search_stats::accumulate(&self.stream_metrics); + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + crate::search_stats::accumulate(&self.stream_metrics); + })); } } Suggestion importance[1-10]: 6 __ Why: The suggestion addresses a real concern about panics in Drop implementations. However, accumulate performs simple atomic operations that are unlikely to panic. The added complexity of catch_unwind may not be justified unless there's evidence of panic risk.	Low
	Use stronger memory ordering for stats reset The reset_stats method uses Relaxed ordering which doesn't guarantee visibility across threads. Use Release ordering when storing to ensure other threads see the reset values, or consider using SeqCst for stronger guarantees in a statistics context where consistency matters. sandbox/plugins/analytics-backend-datafusion/rust/src/cache.rs [53-56] pub fn reset_stats(&self) { - self.hit_count.store(0, Ordering::Relaxed); - self.miss_count.store(0, Ordering::Relaxed); + self.hit_count.store(0, Ordering::Release); + self.miss_count.store(0, Ordering::Release); } Suggestion importance[1-10]: 4 __ Why: While Release ordering provides better visibility guarantees, Relaxed is acceptable for statistics counters where eventual consistency is sufficient. The suggestion is valid but the impact is minimal since stats resets are infrequent operations.	Low
	Use Release ordering for stats reset The reset_stats method uses Relaxed ordering which provides no synchronization guarantees. For statistics that may be read concurrently by monitoring threads, use Release ordering to ensure the reset is visible to subsequent reads with Acquire ordering. sandbox/plugins/analytics-backend-datafusion/rust/src/statistics_cache.rs [191-194] pub fn reset_stats(&self) { - self.hit_count.store(0, Ordering::Relaxed); - self.miss_count.store(0, Ordering::Relaxed); + self.hit_count.store(0, Ordering::Release); + self.miss_count.store(0, Ordering::Release); } Suggestion importance[1-10]: 4 __ Why: Similar to suggestion 1, this suggests using Release ordering for better visibility. While technically correct, Relaxed is acceptable for statistics counters where strict synchronization isn't critical. The impact is minimal for infrequent reset operations.	Low

Suggestions up to commit fb9dfaf

Category	Suggestion	Impact
General	Use stronger memory ordering for reset The reset_stats method should use Ordering::Release instead of Ordering::Relaxed to ensure that the reset operation is visible to other threads that subsequently read these counters. This prevents potential race conditions where a thread might read stale counter values after a reset. sandbox/plugins/analytics-backend-datafusion/rust/src/cache.rs [53-56] pub fn reset_stats(&self) { - self.hit_count.store(0, Ordering::Relaxed); - self.miss_count.store(0, Ordering::Relaxed); + self.hit_count.store(0, Ordering::Release); + self.miss_count.store(0, Ordering::Release); } Suggestion importance[1-10]: 4 __ Why: The suggestion to use Ordering::Release instead of Ordering::Relaxed for the reset_stats method is technically more correct for ensuring visibility across threads. However, given that these are simple counter resets and the counters are read with Relaxed ordering elsewhere, the practical impact is minimal. The suggestion is valid but represents a minor improvement in thread-safety guarantees.	Low
	Potential double-counting of partial metrics The Drop implementation for AccumulatingStream accumulates metrics unconditionally, which could lead to double-counting if the stream is dropped due to an error before completion. Consider tracking whether the stream completed successfully and only accumulating metrics in that case, or document that partial execution metrics are intentionally included. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/table_provider.rs [503-506] impl Drop for AccumulatingStream { fn drop(&mut self) { + // Accumulate metrics even on early drop to capture partial execution stats crate::search_stats::accumulate(&self.stream_metrics); } } Suggestion importance[1-10]: 2 __ Why: The suggestion raises a concern about potential double-counting of metrics when a stream is dropped early. However, the 'improved_code' is identical to the 'existing_code' except for adding a comment. The suggestion doesn't actually propose a solution, only documents the current behavior. Additionally, accumulating partial execution stats on drop is likely intentional for observability purposes, making this a low-priority observation rather than a bug.	Low

Suggestions up to commit c5e9c66

Category	Suggestion	Impact
General	Protect Drop from panic propagation Performing potentially expensive operations in Drop can cause issues if the drop occurs during panic unwinding or in critical sections. Consider whether accumulation should happen earlier or be made panic-safe to avoid double-panic scenarios. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/table_provider.rs [503-506] impl Drop for AccumulatingStream { fn drop(&mut self) { - crate::search_stats::accumulate(&self.stream_metrics); + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + crate::search_stats::accumulate(&self.stream_metrics); + })); } } Suggestion importance[1-10]: 6 __ Why: The suggestion to wrap the accumulate call in catch_unwind is a valid defensive programming practice to prevent double-panic scenarios during drop. While the current code may work in most cases, protecting against panic propagation in Drop implementations is a good practice, especially when the operation involves external state updates. The score reflects moderate importance as it addresses a potential runtime safety issue.	Low
	Use stronger memory ordering The reset_stats method uses Relaxed ordering for atomic stores, which may not guarantee visibility across threads. Consider using Release ordering to ensure that the reset is visible to other threads that subsequently read these counters. sandbox/plugins/analytics-backend-datafusion/rust/src/cache.rs [53-56] pub fn reset_stats(&self) { - self.hit_count.store(0, Ordering::Relaxed); - self.miss_count.store(0, Ordering::Relaxed); + self.hit_count.store(0, Ordering::Release); + self.miss_count.store(0, Ordering::Release); } Suggestion importance[1-10]: 4 __ Why: The suggestion to use Release ordering instead of Relaxed for atomic stores in reset_stats is technically valid for ensuring visibility guarantees. However, the impact is minor since these are simple counter resets, and the existing Relaxed ordering is commonly acceptable for non-critical statistics. The suggestion addresses a potential edge case but doesn't fix a critical bug.	Low

Suggestions up to commit 01fe728

Category	Suggestion	Impact
General	Prevent overflow in memory size cast The as i64 cast can silently truncate or produce negative values if memory consumption exceeds i64::MAX. This could cause incorrect statistics reporting. Consider using try_into() with proper error handling or clamping to i64::MAX. sandbox/plugins/analytics-backend-datafusion/rust/src/stats.rs [243-248] pub fn pack_cache_stats(mgr: &CustomCacheManager) -> CacheStatsRepr { let metadata_memory = mgr .get_memory_consumed_by_type(crate::cache::CACHE_TYPE_METADATA) - .unwrap_or(0) as i64; + .unwrap_or(0) + .min(i64::MAX as usize) as i64; let statistics_memory = mgr .get_memory_consumed_by_type(crate::cache::CACHE_TYPE_STATS) - .unwrap_or(0) as i64; + .unwrap_or(0) + .min(i64::MAX as usize) as i64; Suggestion importance[1-10]: 7 __ Why: The suggestion addresses a real issue where casting usize to i64 can produce negative values on overflow. Using min(i64::MAX as usize) prevents this and ensures statistics remain meaningful. This is a good defensive programming practice for metrics reporting.	Medium
	Avoid accumulating metrics on panic The Drop implementation unconditionally accumulates metrics even if the stream was dropped due to an error or cancellation. This can lead to inflated or misleading statistics. Consider tracking whether the stream completed successfully before accumulating metrics. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/table_provider.rs [503-506] impl Drop for AccumulatingStream { fn drop(&mut self) { - crate::search_stats::accumulate(&self.stream_metrics); + // Only accumulate if stream completed normally + if !std::thread::panicking() { + crate::search_stats::accumulate(&self.stream_metrics); + } } } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that metrics are accumulated even on panic/error, which could inflate statistics. However, the proposed check !std::thread::panicking() only handles panics, not normal cancellations or errors. A more comprehensive solution would be needed for full correctness.	Low
	Use stronger memory ordering for stats reset The reset_stats method is public but doesn't provide any synchronization guarantee with concurrent get operations. This can lead to inconsistent statistics if a reset occurs while cache operations are in progress. Consider documenting this race condition or using SeqCst ordering for the reset operation. sandbox/plugins/analytics-backend-datafusion/rust/src/cache.rs [53-56] pub fn reset_stats(&self) { - self.hit_count.store(0, Ordering::Relaxed); - self.miss_count.store(0, Ordering::Relaxed); + self.hit_count.store(0, Ordering::SeqCst); + self.miss_count.store(0, Ordering::SeqCst); } Suggestion importance[1-10]: 4 __ Why: The suggestion to use SeqCst ordering is valid for ensuring consistency during concurrent operations, but the impact is relatively minor since Relaxed ordering is typically sufficient for simple counter resets. The race condition mentioned is inherent to concurrent statistics and doesn't cause correctness issues in typical usage.	Low

Suggestions up to commit bbb683b

Category	Suggestion	Impact
General	Move counter increment after validation The inc_listing_table_scan() call is placed before error handling, causing the counter to increment even when the function fails early. Move the counter increment after successful validation of input parameters to ensure accurate statistics. sandbox/plugins/analytics-backend-datafusion/rust/src/ffm.rs [714] pub unsafe extern "C" fn df_create_session_context( shard_view_ptr: i64, runtime_ptr: i64, table_name_ptr: *const u8, table_name_len: i64, context_id: i64, query_config_ptr: i64, plan_ptr: *const u8, plan_len: i64, ) -> i64 { - crate::search_stats::inc_listing_table_scan(); let table_name = str_from_raw(table_name_ptr, table_name_len) .map_err(|e| format!("df_create_session_context: {}", e))?; + crate::search_stats::inc_listing_table_scan(); Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that inc_listing_table_scan() is called before input validation. Moving it after validation ensures the counter only increments for valid requests, improving metric accuracy.	Medium
	Track stream completion before accumulating The Drop implementation unconditionally accumulates metrics even when the stream is dropped due to errors or cancellation. This can lead to inflated or misleading statistics. Consider tracking stream completion status and only accumulating metrics for successfully completed streams. sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_table/table_provider.rs [503-506] +struct AccumulatingStream { + inner: SendableRecordBatchStream, + stream_metrics: StreamMetrics, + completed: bool, +} + impl Drop for AccumulatingStream { fn drop(&mut self) { - crate::search_stats::accumulate(&self.stream_metrics); + if self.completed { + crate::search_stats::accumulate(&self.stream_metrics); + } } } Suggestion importance[1-10]: 5 __ Why: The suggestion raises a valid concern about accumulating metrics for failed/cancelled streams. However, the proposed solution requires tracking completion status, which adds complexity. The current implementation may be intentional to capture all stream activity.	Low
	Prevent overflow in hit rate calculation The hitRate() calculation can overflow when hitCount + missCount exceeds Long.MAX_VALUE, resulting in a negative total and incorrect hit rate. Use Math.addExact() or check for overflow before performing the addition to prevent silent arithmetic errors. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/stats/CacheGroupStats.java [97-100] public double hitRate() { - long total = hitCount + missCount; - return total == 0 ? 0.0 : (double) hitCount / (double) total; + try { + long total = Math.addExact(hitCount, missCount); + return total == 0 ? 0.0 : (double) hitCount / (double) total; + } catch (ArithmeticException e) { + return hitCount > 0 ? 1.0 : 0.0; + } } Suggestion importance[1-10]: 3 __ Why: While overflow protection is theoretically good practice, the likelihood of hitCount + missCount exceeding Long.MAX_VALUE in a cache statistics context is extremely low. The added complexity and exception handling may not be justified for this edge case.	Low

[github-actions]

❌ Gradle check result for 849f3c9: 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?

[github-actions]

Persistent review updated to latest commit 1412929

[github-actions]

✅ Gradle check result for 1412929: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.46%. Comparing base (dad63c0) to head (1031399).
⚠️ Report is 5 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #21854      +/-   ##
============================================
- Coverage     73.51%   73.46%   -0.06%     
+ Complexity    75582    75557      -25     
============================================
  Files          6034     6034              
  Lines        342661   342661              
  Branches      49294    49294              
============================================
- Hits         251918   251734     -184     
- Misses        70712    70946     +234     
+ Partials      20031    19981      -50     

☔ View full report in Codecov by Sentry.
📢 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.

[github-actions]

Persistent review updated to latest commit 594236b

[github-actions]

❌ Gradle check result for 594236b: null

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?

[github-actions]

Persistent review updated to latest commit a87901e

[github-actions]

❕ Gradle check result for a87901e: UNSTABLE

Please review all flaky tests that succeeded after retry and create an issue if one does not already exist to track the flaky failure.

[github-actions]

Persistent review updated to latest commit 7588444

[github-actions]

✅ Gradle check result for 7588444: SUCCESS

[github-actions]

Persistent review updated to latest commit c053876

[github-actions]

✅ Gradle check result for c053876: SUCCESS

[github-actions]

Persistent review updated to latest commit 0009f43

[github-actions]

✅ Gradle check result for 0009f43: SUCCESS

[github-actions]

Persistent review updated to latest commit c33a288

[github-actions]

❌ Gradle check result for c33a288: 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?

[github-actions]

Persistent review updated to latest commit cdf51dc

[github-actions]

Persistent review updated to latest commit 28a767b

[github-actions]

Persistent review updated to latest commit 2619389

[github-actions]

✅ Gradle check result for 2619389: SUCCESS

[github-actions]

Persistent review updated to latest commit 3b49a1f

[github-actions]

❌ Gradle check result for 3b49a1f:

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?

[github-actions]

Persistent review updated to latest commit 837cf2c

[github-actions]

✅ Gradle check result for 837cf2c: SUCCESS

[github-actions]

Persistent review updated to latest commit bbb683b

[github-actions]

❌ Gradle check result for bbb683b: 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?

[github-actions]

Persistent review updated to latest commit 01fe728

[github-actions]

❌ Gradle check result for 01fe728: 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?

[github-actions]

Persistent review updated to latest commit c5e9c66

[github-actions]

❌ Gradle check result for c5e9c66: null

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?

[github-actions]

Persistent review updated to latest commit fb9dfaf

[github-actions]

✅ Gradle check result for fb9dfaf: SUCCESS

[github-actions]

Persistent review updated to latest commit ca91da8

[github-actions]

Persistent review updated to latest commit 1031399

[github-actions]

✅ Gradle check result for 1031399: SUCCESS