Fix cache metrics

quickwit/quickwit-storage/src/cache/base_cache.rs

@@ -205,8 +205,9 @@ impl<K: Hash + Eq, V: ValueLen + Clone> Lru<K, V> {
             }
             return;
         }
-        if let Some(previous_data) = self.lru_cache.pop(&key) {
-            self.drop_item(previous_data.len() as u64);
+        if let Some(item) = self.lru_cache.get_mut(&key) {
+            item.touch();
+            return;
         }
 
         let now = Instant::now();
@@ -334,6 +335,9 @@ impl<K: Hash + Eq, V: ValueLen + Clone> S3Fifo<K, V> {
             }
             return;
         }
+        if self.cache.get(&key).is_some() {
+            return;
+        }
 
         self.cache_metrics.in_cache_count.inc();
         self.cache_metrics
@@ -459,6 +463,9 @@ impl<K: Hash + Eq + Send + Sync + 'static, V: ValueLen + Clone + Send + Sync + '
             );
             return;
         }
+        if self.cache.get(&key).is_some() {
+            return;
+        }
 
         self.cache_metrics.in_cache_count.inc();
         self.cache_metrics
@@ -571,4 +578,21 @@ mod tests {
             );
         }
     }
+
+    #[test]
+    fn test_any_cache_reputting_same_key_is_not_counted_as_eviction() {
+        for policy in [CachePolicy::Lru, CachePolicy::S3Fifo, CachePolicy::TinyLfu] {
+            let cache_metrics =
+                ComponentCacheMetrics::for_component_in_tests(&format!("reput_test_{policy}"))
+                    .active_cache_metrics;
+            let mut cache: AnyCache<String, OwnedBytes> =
+                AnyCache::from_policy_and_capacity(policy, ByteSize::kb(10), cache_metrics.clone());
+            cache.put("key".to_string(), OwnedBytes::new(&b"hello"[..]));
+            // cached values are assumed immutable: re-putting the same key must be a no-op,
+            // not an eviction of the existing entry.
+            cache.put("key".to_string(), OwnedBytes::new(&b"hello"[..]));
+            assert_eq!(cache_metrics.evict_num_items.get(), 0, "policy {policy}");
+            assert_eq!(cache_metrics.in_cache_count.get(), 1, "policy {policy}");
+        }
+    }
 }

quickwit/quickwit-storage/src/cache/stored_item.rs

@@ -33,10 +33,14 @@ impl<V> StoredItem<V> {
 
 impl<V: ValueLen + Clone> StoredItem<V> {
     pub fn payload(&mut self) -> V {
-        self.last_access_time = Instant::now();
+        self.touch();
         self.payload.clone()
     }
 
+    pub fn touch(&mut self) {
+        self.last_access_time = Instant::now();
+    }
+
     pub fn len(&self) -> usize {
         self.payload.len()
     }

quickwit/quickwit-storage/src/file_descriptor_cache.rs

@@ -101,25 +101,32 @@ impl FileDescriptorCache {
 
     fn put_split_file(&self, split_id: Ulid, split_file: SplitFile) {
         let mut fd_cache_lock = self.fd_cache.lock().unwrap();
-        fd_cache_lock.push(split_id, split_file);
+        let evicted = fd_cache_lock.push(split_id, split_file);
         self.fd_cache_metrics
             .in_cache_count
             .set(fd_cache_lock.len() as i64);
+        if let Some((evicted_split_id, _)) = evicted
+            && split_id != evicted_split_id
+        {
+            self.fd_cache_metrics.evict_num_items.inc();
+        }
     }
 
     /// Evicts the given list of split ids from the file descriptor cache.
     /// This method does NOT remove the actual files.
     pub fn evict_split_files(&self, split_ids: &[Ulid]) {
         let mut fd_cache_lock = self.fd_cache.lock().unwrap();
+        let mut evicted_count = 0;
         for split_id in split_ids {
-            fd_cache_lock.pop(split_id);
+            let evicted = fd_cache_lock.pop(split_id);
+            if evicted.is_some() {
+                evicted_count += 1;
+            }
         }
         self.fd_cache_metrics
             .in_cache_count
             .set(fd_cache_lock.len() as i64);
-        self.fd_cache_metrics
-            .evict_num_items
-            .inc_by(split_ids.len() as u64);
+        self.fd_cache_metrics.evict_num_items.inc_by(evicted_count);
     }
 
     pub async fn get_or_open_split_file(
@@ -224,6 +231,7 @@ mod tests {
         assert_eq!(cache_metrics.in_cache_count.get(), 10);
         assert_eq!(cache_metrics.hits_num_items.get(), 20);
         assert_eq!(cache_metrics.misses_num_items.get(), 10);
+        assert_eq!(cache_metrics.evict_num_items.get(), 0);
     }
 
     // This mimics Quickwit's workload where the fd cache is much smaller than the number of
@@ -257,6 +265,9 @@ mod tests {
         assert_eq!(cache_metrics.in_cache_count.get(), 10);
         assert_eq!(cache_metrics.hits_num_items.get(), 100 * 9);
         assert_eq!(cache_metrics.misses_num_items.get(), 100);
+        // 100 distinct splits went through a 10-entry cache: the 90 oldest ones were pushed
+        // out by capacity, one at a time, as later splits were first opened.
+        assert_eq!(cache_metrics.evict_num_items.get(), 90);
     }
 
     #[tokio::test]