From 4f826b1a16328464333b311c90f6c5e350dd41ec Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sat, 8 Aug 2026 12:28:43 -0500 Subject: [PATCH 1/3] feat: add CacheStatistics to the Cache trait Adds a `CacheStatistics` struct holding aggregate, lifetime counters for a cache (hits, misses, evictions, bytes evicted, and inserts rejected for being larger than the whole limit) plus a `Cache::statistics()` accessor. The accessor has a default implementation returning `None` so that external implementations of the (public) `Cache` trait keep compiling. `None` means "not instrumented", which a reporting tool can distinguish from an instrumented cache whose counters are genuinely all zero. No implementation yet; `DefaultCache` is wired up in the next commit. Co-Authored-By: Claude Opus 5 --- datafusion/execution/src/cache/mod.rs | 137 ++++++++++++++++++++++++++ 1 file changed, 137 insertions(+) diff --git a/datafusion/execution/src/cache/mod.rs b/datafusion/execution/src/cache/mod.rs index f47a3f3ca49f3..0fe1b06a7344b 100644 --- a/datafusion/execution/src/cache/mod.rs +++ b/datafusion/execution/src/cache/mod.rs @@ -91,6 +91,66 @@ pub trait Cache: Send + Sync { /// Snapshot of all current entries with per-entry metadata (size, hits, /// expiration) for diagnostics and observability. fn list_entries(&self) -> HashMap>; + + /// Aggregate counters describing how this cache has behaved so far, or + /// `None` if the implementation does not track them. + /// + /// Unlike [`Cache::list_entries`], which only describes entries that are + /// currently resident, these counters cover the whole lifetime of the cache + /// and therefore survive eviction. They are what makes a thrashing cache + /// (0% hit rate because the working set does not fit the byte budget) + /// diagnosable. + /// + /// The default implementation returns `None` so that existing external + /// implementations of this trait keep compiling; `None` means + /// "not instrumented", which a reporting tool can distinguish from an + /// instrumented cache that is genuinely all zeros. + fn statistics(&self) -> Option { + None + } +} + +/// Aggregate, lifetime counters for a [`Cache`]. +/// +/// Obtained via [`Cache::statistics`]. All counters are monotonic for the +/// lifetime of the cache: they are not reset by [`Cache::clear`]. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct CacheStatistics { + /// Lookups ([`Cache::get`]) that returned a value. + pub hits: u64, + /// Lookups ([`Cache::get`]) that did not return a value, either because the + /// key was absent or because the entry had expired. + pub misses: u64, + /// Entries dropped to stay within the memory budget. + /// + /// This counts only capacity-driven eviction (including eviction triggered + /// by lowering the limit via [`Cache::update_cache_limit`]). Explicit + /// invalidation through [`Cache::remove`], [`Cache::clear`] or + /// [`Cache::drop_table_entries`], and replacing an entry with a new value + /// for the same key, are not evictions. + pub evictions: u64, + /// Total size, in bytes, of the entries counted by `evictions`. + pub bytes_evicted: u64, + /// Inserts rejected because the entry on its own was larger than the whole + /// cache limit. + /// + /// A non-zero value means the cache can never hold that entry, no matter + /// how much of it is free — a distinct failure mode from ordinary eviction + /// pressure, and one that is worth surfacing on its own. + pub inserts_rejected_too_large: u64, +} + +impl CacheStatistics { + /// Total number of lookups, i.e. `hits + misses`. + pub fn lookups(&self) -> u64 { + self.hits.saturating_add(self.misses) + } + + /// Fraction of lookups that were hits, or `None` if there were no lookups. + pub fn hit_rate(&self) -> Option { + let lookups = self.lookups(); + (lookups > 0).then(|| self.hits as f64 / lookups as f64) + } } /// Key type for entries stored in a [`Cache`]. @@ -225,6 +285,83 @@ impl DFHeapSize for SchemaFingerprint { } } +#[cfg(test)] +mod cache_statistics_tests { + use super::*; + + #[derive(Clone)] + struct Unit; + + impl CacheValue for Unit { + fn size(&self) -> usize { + 0 + } + } + + /// A [`Cache`] implementation that does not override [`Cache::statistics`], + /// standing in for an external implementation written before the counters + /// existed. It must keep compiling, and report that it is uninstrumented. + struct UninstrumentedCache; + + impl Cache for UninstrumentedCache { + fn get(&self, _key: &Path) -> Option { + None + } + fn put(&self, _key: &Path, _value: Unit) -> Option { + None + } + fn remove(&self, _k: &Path) -> Option { + None + } + fn contains_key(&self, _k: &Path) -> bool { + false + } + fn len(&self) -> usize { + 0 + } + fn clear(&self) {} + fn name(&self) -> String { + "UninstrumentedCache".to_string() + } + fn cache_limit(&self) -> usize { + 0 + } + fn update_cache_limit(&self, _limit: usize) {} + fn cache_ttl(&self) -> Option { + None + } + fn update_cache_ttl(&self, _ttl: Option) {} + fn drop_table_entries( + &self, + _table_ref: &TableReference, + ) -> datafusion_common::Result<()> { + Ok(()) + } + fn list_entries(&self) -> HashMap> { + HashMap::new() + } + } + + #[test] + fn uninstrumented_cache_reports_no_statistics() { + assert_eq!(UninstrumentedCache.statistics(), None); + } + + #[test] + fn hit_rate_and_lookups() { + assert_eq!(CacheStatistics::default().hit_rate(), None); + assert_eq!(CacheStatistics::default().lookups(), 0); + + let stats = CacheStatistics { + hits: 1, + misses: 3, + ..Default::default() + }; + assert_eq!(stats.lookups(), 4); + assert_eq!(stats.hit_rate(), Some(0.25)); + } +} + #[cfg(test)] mod schema_fingerprint_tests { use super::*; From a77aa3a6e95a5ee77e00391cbec677e5aa35887b Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sat, 8 Aug 2026 12:28:55 -0500 Subject: [PATCH 2/3] feat: track cache statistics in DefaultCache Counts hits, misses, evictions, bytes evicted and inserts rejected for exceeding the whole cache limit, and exposes them via `Cache::statistics()` and the inherent `DefaultCache::cache_statistics()`. The counters live in `DefaultCacheState`, so they are updated under the mutex that every `get`/`put` already takes rather than adding a second point of synchronization on the hot path. Semantics, covered by the new tests: - a lookup of an absent *or expired* key is a miss; `contains_key` is a probe and is not counted at all - only capacity-driven drops are evictions, including those caused by lowering the limit via `update_cache_limit`. `remove`, `clear`, `drop_table_entries` and replacing an entry are not evictions - an insert whose entry is larger than the entire limit is counted separately: no amount of eviction can ever make room for it, so it is a distinct failure mode worth surfacing on its own - counters are monotonic for the lifetime of the cache and are not reset by `clear` Co-Authored-By: Claude Opus 5 --- .../execution/src/cache/default_cache.rs | 202 +++++++++++++++++- 1 file changed, 197 insertions(+), 5 deletions(-) diff --git a/datafusion/execution/src/cache/default_cache.rs b/datafusion/execution/src/cache/default_cache.rs index bfe326f3a47e1..f5c5b80573576 100644 --- a/datafusion/execution/src/cache/default_cache.rs +++ b/datafusion/execution/src/cache/default_cache.rs @@ -23,7 +23,7 @@ use datafusion_common::instant::Instant; use datafusion_common::{HashMap, Result}; use crate::cache::lru_queue::LruQueue; -use crate::cache::{Cache, CacheEntryInfo, CacheKey, CacheValue}; +use crate::cache::{Cache, CacheEntryInfo, CacheKey, CacheStatistics, CacheValue}; /// Source of the current time used by a [`DefaultCache`] when applying TTLs. pub trait TimeProvider: Send + Sync { @@ -55,6 +55,10 @@ struct DefaultCacheState { memory_limit: usize, memory_used: usize, ttl: Option, + /// Aggregate, lifetime counters. Kept inside the state so that they are + /// updated under the mutex already held by every `get`/`put`, rather than + /// adding a second point of synchronization on the hot path. + statistics: CacheStatistics, } impl DefaultCacheState { @@ -65,19 +69,25 @@ impl DefaultCacheState { memory_limit, memory_used: 0, ttl, + statistics: CacheStatistics::default(), } } fn get(&mut self, key: &K, now: Instant) -> Option { - let entry = self.lru_queue.get(key)?; + let Some(entry) = self.lru_queue.get(key) else { + self.statistics.misses += 1; + return None; + }; if let Some(exp) = entry.expires && now > exp { self.remove(key); + self.statistics.misses += 1; return None; } let value = entry.value.clone(); *self.hits.entry(key.clone()).or_insert(0) += 1; + self.statistics.hits += 1; Some(value) } @@ -105,6 +115,8 @@ impl DefaultCacheState { let total_size = key_size + value_size; if total_size > self.memory_limit { + // The entry can never fit, no matter how much of the cache is free. + self.statistics.inserts_rejected_too_large += 1; // Remove potential stale entry return self.remove(key); } @@ -145,9 +157,11 @@ impl DefaultCacheState { self.memory_used = 0; return; }; - self.memory_used -= evicted_key.size(); - self.memory_used -= evicted.value.size(); + let evicted_size = evicted_key.size() + evicted.value.size(); + self.memory_used -= evicted_size; self.hits.remove(&evicted_key); + self.statistics.evictions += 1; + self.statistics.bytes_evicted += evicted_size as u64; } } @@ -204,6 +218,14 @@ impl DefaultCache { pub fn memory_used(&self) -> usize { self.state.lock().unwrap().memory_used } + + /// Aggregate, lifetime counters for this cache. + /// + /// See [`Cache::statistics`] and [`CacheStatistics`] for the exact meaning + /// of each counter. + pub fn cache_statistics(&self) -> CacheStatistics { + self.state.lock().unwrap().statistics + } } impl Cache for DefaultCache { @@ -293,6 +315,10 @@ impl Cache for DefaultCache { }) .collect() } + + fn statistics(&self) -> Option { + Some(self.cache_statistics()) + } } #[cfg(test)] @@ -308,7 +334,7 @@ mod tests { use crate::cache::cache_manager::{CachedFileMetadataEntry, FileMetadata}; use crate::cache::default_cache::DefaultCache; use crate::cache::default_cache::TimeProvider; - use crate::cache::{Cache, CacheEntryInfo}; + use crate::cache::{Cache, CacheEntryInfo, CacheStatistics}; use crate::cache::{CacheKey, CacheValue}; use crate::cache::{SchemaFingerprint, TableScopedPath}; use arrow::array::{Int32Array, ListArray, RecordBatch}; @@ -2012,4 +2038,170 @@ mod tests { assert!(!cache.contains_key(&key2)); assert!(cache.contains_key(&key3)); } + + #[test] + fn test_statistics_hits_and_misses() { + let cache = DefaultCache::new(DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT); + let table_ref = Some(TableReference::from("table")); + let (key1, value1) = + create_test_list_files_entry("path1", 1, 100, table_ref.clone()); + let (key2, _value2) = create_test_list_files_entry("path2", 1, 100, table_ref); + + assert_eq!(cache.cache_statistics(), CacheStatistics::default()); + assert_eq!(cache.cache_statistics().hit_rate(), None); + + // Lookup on an empty cache is a miss + assert!(cache.get(&key1).is_none()); + assert_eq!(cache.cache_statistics().misses, 1); + assert_eq!(cache.cache_statistics().hits, 0); + + cache.put(&key1, value1); + + // Two hits on the entry we just inserted + assert!(cache.get(&key1).is_some()); + assert!(cache.get(&key1).is_some()); + assert_eq!(cache.cache_statistics().hits, 2); + assert_eq!(cache.cache_statistics().misses, 1); + + // A lookup of an absent key is a miss + assert!(cache.get(&key2).is_none()); + assert_eq!(cache.cache_statistics().misses, 2); + + // `contains_key` is a probe, not a lookup, and is not counted + assert!(cache.contains_key(&key1)); + assert!(!cache.contains_key(&key2)); + assert_eq!(cache.cache_statistics().hits, 2); + assert_eq!(cache.cache_statistics().misses, 2); + + assert_eq!(cache.cache_statistics().lookups(), 4); + assert_eq!(cache.cache_statistics().hit_rate(), Some(0.5)); + + // Statistics are exposed through the `Cache` trait too + assert_eq!(cache.statistics(), Some(cache.cache_statistics())); + + // `clear` neither resets the counters nor counts as an eviction + cache.clear(); + let stats = cache.cache_statistics(); + assert_eq!(stats.hits, 2); + assert_eq!(stats.misses, 2); + assert_eq!(stats.evictions, 0); + assert_eq!(stats.bytes_evicted, 0); + + // ... and a lookup after the clear is a miss + assert!(cache.get(&key1).is_none()); + assert_eq!(cache.cache_statistics().misses, 3); + } + + #[test] + fn test_statistics_expired_entry_counts_as_miss() { + let ttl = Duration::from_millis(100); + let mock_time = Arc::new(MockTimeProvider::new()); + let cache = DefaultCache::new_with_ttl(10000, Some(ttl)) + .with_time_provider(Arc::clone(&mock_time) as Arc); + + let table_ref = Some(TableReference::from("table")); + let (key, value) = create_test_list_files_entry("path1", 1, 50, table_ref); + cache.put(&key, value); + + assert!(cache.get(&key).is_some()); + assert_eq!(cache.cache_statistics().hits, 1); + + mock_time.inc(Duration::from_millis(150)); + + // Expiration is not eviction: the entry is dropped lazily on access and + // the access itself is a miss. + assert!(cache.get(&key).is_none()); + let stats = cache.cache_statistics(); + assert_eq!(stats.hits, 1); + assert_eq!(stats.misses, 1); + assert_eq!(stats.evictions, 0); + } + + #[test] + fn test_statistics_evictions() { + let table_ref = Some(TableReference::from("table")); + let (key1, value1) = + create_test_list_files_entry("path1", 1, 100, table_ref.clone()); + let (key2, value2) = + create_test_list_files_entry("path2", 1, 100, table_ref.clone()); + let (key3, value3) = + create_test_list_files_entry("path3", 1, 100, table_ref.clone()); + let (key4, value4) = + create_test_list_files_entry("path4", 1, 100, table_ref.clone()); + + let entry_size = key1.size() + value1.size(); + + // A cache that fits exactly two entries + let cache = DefaultCache::new(entry_size * 2); + cache.put(&key1, value1); + cache.put(&key2, value2); + assert_eq!(cache.cache_statistics().evictions, 0); + + // The third insert evicts the LRU entry + cache.put(&key3, value3); + let stats = cache.cache_statistics(); + assert_eq!(stats.evictions, 1); + assert_eq!(stats.bytes_evicted, entry_size as u64); + + // Replacing an existing entry with an equally sized one is not an eviction + let (_, value3_again) = + create_test_list_files_entry("path3", 1, 100, table_ref.clone()); + cache.put(&key3, value3_again); + assert_eq!(cache.cache_statistics().evictions, 1); + + // Explicitly removing an entry is not an eviction either + cache.remove(&key2); + assert_eq!(cache.cache_statistics().evictions, 1); + + // Neither is dropping a whole table's entries + cache.put(&key4, value4); + cache + .drop_table_entries(&TableReference::from("table")) + .unwrap(); + assert_eq!(cache.cache_statistics().evictions, 1); + assert_eq!(cache.len(), 0); + + // Shrinking the limit evicts, and that is counted + let (key5, value5) = create_test_list_files_entry("path5", 1, 100, table_ref); + cache.put(&key5, value5); + cache.update_cache_limit(1); + let stats = cache.cache_statistics(); + assert_eq!(stats.evictions, 2); + assert_eq!(stats.bytes_evicted, (entry_size * 2) as u64); + } + + #[test] + fn test_statistics_inserts_rejected_too_large() { + let table_ref = Some(TableReference::from("table")); + let (key1, value1) = + create_test_list_files_entry("path1", 1, 100, table_ref.clone()); + let entry_size = key1.size() + value1.size(); + + let cache = DefaultCache::new(entry_size); + cache.put(&key1, value1); + assert_eq!(cache.cache_statistics().inserts_rejected_too_large, 0); + + // An entry larger than the whole cache can never fit and is rejected + // rather than evicting its way to space that will never be enough. + let (key_large, value_large) = + create_test_list_files_entry("large", 1, 1000, table_ref.clone()); + cache.put(&key_large, value_large); + let stats = cache.cache_statistics(); + assert_eq!(stats.inserts_rejected_too_large, 1); + assert_eq!(stats.evictions, 0); + assert!(!cache.contains_key(&key_large)); + + // A rejected insert also drops any stale entry stored under the same + // key. That removal is still a rejection, not an eviction. + assert!(cache.contains_key(&key1)); + let (_, value1_too_large) = + create_test_list_files_entry("path1", 1, 1000, table_ref); + cache.put(&key1, value1_too_large); + let stats = cache.cache_statistics(); + assert_eq!(stats.inserts_rejected_too_large, 2); + assert_eq!(stats.evictions, 0); + assert!(!cache.contains_key(&key1)); + assert_eq!(cache.len(), 0); + assert_eq!(cache.memory_used(), 0); + } } From 22de14f122ebbff75ba5e730bd1f3eeeb7e4a098 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sat, 8 Aug 2026 12:28:55 -0500 Subject: [PATCH 3/3] feat: expose cache statistics from CacheManager Adds `get_file_statistic_cache_statistics`, `get_file_metadata_cache_statistics` and `get_list_files_cache_statistics`, so the counters are reachable from the place users already go for cache limits. Each returns `None` when the cache is disabled or when the configured implementation does not track statistics. Co-Authored-By: Claude Opus 5 --- .../execution/src/cache/cache_manager.rs | 72 ++++++++++++++++++- 1 file changed, 71 insertions(+), 1 deletion(-) diff --git a/datafusion/execution/src/cache/cache_manager.rs b/datafusion/execution/src/cache/cache_manager.rs index 83dcf70975e2b..07f8840be4779 100644 --- a/datafusion/execution/src/cache/cache_manager.rs +++ b/datafusion/execution/src/cache/cache_manager.rs @@ -16,7 +16,9 @@ // under the License. use crate::cache::default_cache::DefaultCache; -pub use crate::cache::{Cache, CacheValue, SchemaFingerprint, TableScopedPath}; +pub use crate::cache::{ + Cache, CacheStatistics, CacheValue, SchemaFingerprint, TableScopedPath, +}; use datafusion_common::HashMap; use datafusion_common::heap_size::{DFHeapSize, DFHeapSizeCtx}; use datafusion_common::{Result, Statistics}; @@ -407,6 +409,30 @@ impl CacheManager { pub fn get_metadata_cache_limit(&self) -> usize { self.file_metadata_cache.cache_limit() } + + /// Aggregate counters for the file statistics cache. + /// + /// Returns `None` if the cache is disabled or if the configured + /// implementation does not track statistics. See [`CacheStatistics`]. + pub fn get_file_statistic_cache_statistics(&self) -> Option { + self.file_statistic_cache.as_ref()?.statistics() + } + + /// Aggregate counters for the list files cache. + /// + /// Returns `None` if the cache is disabled or if the configured + /// implementation does not track statistics. See [`CacheStatistics`]. + pub fn get_list_files_cache_statistics(&self) -> Option { + self.list_files_cache.as_ref()?.statistics() + } + + /// Aggregate counters for the file embedded metadata cache. + /// + /// Returns `None` if the configured implementation does not track + /// statistics. See [`CacheStatistics`]. + pub fn get_file_metadata_cache_statistics(&self) -> Option { + self.file_metadata_cache.statistics() + } } #[derive(Clone)] @@ -576,4 +602,48 @@ mod tests { "TTL should be overridden to 60 seconds when set in config" ); } + + /// The counters must be reachable from the `CacheManager` for the caches + /// whose byte budgets are the ones users actually hit. + #[test] + fn test_cache_statistics_reachable_from_cache_manager() { + let cache_manager = + CacheManager::try_new(&CacheManagerConfig::default()).unwrap(); + + // The default caches are all instrumented and start at zero. + assert_eq!( + cache_manager.get_file_metadata_cache_statistics(), + Some(CacheStatistics::default()) + ); + assert_eq!( + cache_manager.get_file_statistic_cache_statistics(), + Some(CacheStatistics::default()) + ); + assert_eq!( + cache_manager.get_list_files_cache_statistics(), + Some(CacheStatistics::default()) + ); + + // A miss on the metadata cache is visible through the manager. + assert!( + cache_manager + .get_file_metadata_cache() + .get(&Path::from("missing.parquet")) + .is_none() + ); + let stats = cache_manager + .get_file_metadata_cache_statistics() + .expect("default metadata cache is instrumented"); + assert_eq!(stats.misses, 1); + assert_eq!(stats.hits, 0); + assert_eq!(stats.hit_rate(), Some(0.0)); + + // A disabled cache has no statistics to report. + let config = CacheManagerConfig::default() + .with_file_statistics_cache_limit(0) + .with_list_files_cache_limit(0); + let cache_manager = CacheManager::try_new(&config).unwrap(); + assert_eq!(cache_manager.get_file_statistic_cache_statistics(), None); + assert_eq!(cache_manager.get_list_files_cache_statistics(), None); + } }