fix(cache): stop evicted entries outliving the budget

src/core/src/cache/builders.rs

@@ -213,6 +213,7 @@ impl<'a> Insert<'a> {
             self.storage.add_squeeze_hint(&self.entry_id, squeeze_hint);
         }
         let batch = CacheEntry::memory_arrow(batch);
+        self.storage.supersede_disk_copy(self.entry_id).await;
         self.storage.insert_inner(self.entry_id, batch).await
     }
 }

src/core/src/cache/index.rs

@@ -1,16 +1,43 @@
 use congee::CongeeArc;
 use std::{
     fmt::{Debug, Formatter},
-    sync::{
-        Arc,
-        atomic::{AtomicUsize, Ordering},
-    },
+    sync::atomic::{AtomicUsize, Ordering},
 };
 
 use crate::cache::{cached_batch::CacheEntry, utils::EntryID};
+use crate::sync::{Arc, RwLock};
+
+/// The value stored in the ART.
+///
+/// `CongeeArc` frees a removed or replaced value through crossbeam-epoch's
+/// deferred destruction: it clones the `Arc` and drops the clone only when a
+/// later pin collects that epoch's garbage — up to 64 objects per thread wait
+/// in a thread-local bag, and the global queue drains 8 bags per 128 pins.
+/// With multi-megabyte arrays as values, that kept every evicted entry alive
+/// for an unbounded, budget-invisible stretch (liquid-cache#43: a tier
+/// reporting at its limit while the process held several times that).
+///
+/// So the tree stores a small slot and the payload is taken out of it the
+/// moment the index gives the entry up. The deferred drop then reclaims only
+/// an empty shell, and the array dies with the last caller-held reference.
+struct Slot(RwLock<Option<Arc<CacheEntry>>>);
+
+impl Slot {
+    fn new(entry: CacheEntry) -> Arc<Self> {
+        Arc::new(Self(RwLock::new(Some(Arc::new(entry)))))
+    }
+
+    fn load(&self) -> Option<Arc<CacheEntry>> {
+        self.0.read().unwrap().clone()
+    }
+
+    fn take(&self) -> Option<Arc<CacheEntry>> {
+        self.0.write().unwrap().take()
+    }
+}
 
 pub(crate) struct ArtIndex {
-    art: CongeeArc<EntryID, CacheEntry>,
+    art: CongeeArc<EntryID, Slot>,
     entry_count: AtomicUsize,
 }
 
@@ -22,62 +49,63 @@ impl Debug for ArtIndex {
 
 impl ArtIndex {
     pub(crate) fn new() -> Self {
-        let art: CongeeArc<EntryID, CacheEntry> = CongeeArc::new();
         Self {
-            art,
+            art: CongeeArc::new(),
             entry_count: AtomicUsize::new(0),
         }
     }
 
     pub(crate) fn get(&self, entry_id: &EntryID) -> Option<Arc<CacheEntry>> {
         let guard = self.art.pin();
-        let batch = self.art.get(*entry_id, &guard)?;
-        Some(batch)
+        // An empty slot means the entry was removed or replaced between the
+        // tree lookup and the load. A remove reading as a miss is exactly as
+        // if it had won the race outright, but after a replace the key is
+        // still present with a new slot, so look the key up once more rather
+        // than report a cached entry as absent.
+        let slot = self.art.get(*entry_id, &guard)?;
+        if let Some(entry) = slot.load() {
+            return Some(entry);
+        }
+        self.art.get(*entry_id, &guard)?.load()
     }
 
     pub(crate) fn is_cached(&self, entry_id: &EntryID) -> bool {
-        let guard = self.art.pin();
-        self.art.get(*entry_id, &guard).is_some()
+        self.get(entry_id).is_some()
     }
 
     pub(crate) fn insert(&self, entry_id: &EntryID, batch: CacheEntry) {
         let guard = self.art.pin();
         let existing = self
             .art
-            .insert(*entry_id, Arc::new(batch), &guard)
+            .insert(*entry_id, Slot::new(batch), &guard)
             .expect("Insertion failed");
-        if existing.is_none() {
-            self.entry_count.fetch_add(1, Ordering::Relaxed);
+        match existing {
+            Some(replaced) => drop(replaced.take()),
+            None => {
+                self.entry_count.fetch_add(1, Ordering::Relaxed);
+            }
         }
     }
 
     pub(crate) fn remove(&self, entry_id: &EntryID) -> Option<Arc<CacheEntry>> {
         let guard = self.art.pin();
-        let removed = self.art.remove(*entry_id, &guard);
-        if removed.is_some() {
-            self.entry_count.fetch_sub(1, Ordering::Relaxed);
-        }
-        removed
+        let removed = self.art.remove(*entry_id, &guard)?;
+        self.entry_count.fetch_sub(1, Ordering::Relaxed);
+        removed.take()
     }
 
     pub(crate) fn reset(&self) {
-        let guard = self.art.pin();
-        self.art.keys().into_iter().for_each(|k| {
-            _ = self.art.remove(k, &guard).unwrap();
-        });
+        for k in self.art.keys() {
+            self.remove(&k);
+        }
         self.entry_count.store(0, Ordering::Relaxed);
     }
 
     pub(crate) fn for_each(&self, mut f: impl FnMut(&EntryID, &CacheEntry)) {
-        let guard = self.art.pin();
-        for id in self.art.keys().into_iter() {
-            f(
-                &id,
-                &self
-                    .art
-                    .get(id, &guard)
-                    .expect("Failed to get value from ART"),
-            );
+        for id in self.art.keys() {
+            if let Some(entry) = self.get(&id) {
+                f(&id, &entry);
+            }
         }
     }
 
@@ -143,4 +171,37 @@ mod tests {
         let entry_id: EntryID = EntryID::from(1);
         assert!(!store.is_cached(&entry_id));
     }
+
+    /// The array behind a removed or replaced entry must die with the last
+    /// caller-held reference, not wait for epoch garbage collection.
+    #[test]
+    fn removed_and_replaced_entries_are_released_immediately() {
+        let store = ArtIndex::new();
+        let id = EntryID::from(1);
+
+        let first = create_test_array(100);
+        let CacheEntry::MemoryArrow(first_array) = &first else {
+            unreachable!()
+        };
+        let weak_first = Arc::downgrade(first_array);
+        store.insert(&id, first);
+        store.insert(&id, create_test_array(200));
+        assert!(
+            weak_first.upgrade().is_none(),
+            "replaced entry still alive: held by the index's deferred drop"
+        );
+
+        let second = store.get(&id).unwrap();
+        let removed = store.remove(&id).unwrap();
+        let CacheEntry::MemoryArrow(second_array) = removed.as_ref() else {
+            unreachable!()
+        };
+        let weak_second = Arc::downgrade(second_array);
+        drop((second, removed));
+        assert!(
+            weak_second.upgrade().is_none(),
+            "removed entry still alive: held by the index's deferred drop"
+        );
+        assert_eq!(store.entry_count(), 0);
+    }
 }

src/core/src/cache/tests/snapshots/liquid_cache__cache__tests__policies__insert_wont_fit_cache.snap

@@ -21,6 +21,5 @@ event=read entry=1 selection=false expr=None cached=DiskLiquid
 event=io_read_liquid entry=1 bytes=34600
 event=hydrate entry=1 cached=DiskLiquid new=MemoryLiquid
 event=insert_failed entry=1 kind=MemoryLiquid
-event=io_write entry=1 kind=MemoryLiquid bytes=34600
 event=insert_success entry=1 kind=DiskLiquid
 ]

src/core/src/liquid_array/byte_view_array/tests.rs

@@ -1028,3 +1028,32 @@ fn test_offset_stress() {
         assert!(offsets[i] >= offsets[i - 1], "Offsets should be monotonic");
     }
 }
+
+/// A decoded array must report the memory it actually holds: within 5% of
+/// the serialized byte count, and within 5% of the freshly transcoded array.
+#[test]
+fn decoded_memory_usage_matches_transcoded_and_bytes() {
+    let mut rng = rand::rngs::StdRng::seed_from_u64(43);
+    let mut builder = arrow::array::StringViewBuilder::new();
+    for _ in 0..2048 {
+        let s: String = (0..1024)
+            .map(|_| rng.random_range(b'a'..=b'z') as char)
+            .collect();
+        builder.append_value(&s);
+    }
+    let input = builder.finish();
+
+    let compressor = LiquidByteViewArray::<FsstArray>::train_compressor(input.iter());
+    let transcoded = LiquidByteViewArray::<FsstArray>::from_string_view_array(&input, compressor);
+    let bytes = transcoded.to_bytes();
+    let decoded = LiquidByteViewArray::<FsstArray>::from_bytes(
+        bytes.clone().into(),
+        transcoded.fsst_buffer.compressor_arc(),
+    );
+
+    let decoded_size = decoded.get_detailed_memory_usage().total() as f64;
+    let transcoded_size = transcoded.get_detailed_memory_usage().total() as f64;
+    let bytes_len = bytes.len() as f64;
+    assert!((decoded_size - bytes_len).abs() <= 0.05 * bytes_len);
+    assert!((decoded_size - transcoded_size).abs() <= 0.05 * transcoded_size);
+}