From 50fc7f60652465a3284915cd33f9dc840d6a0360 Mon Sep 17 00:00:00 2001
From: Zac Farrell <zac@hotdata.dev>
Date: Wed, 2 Sep 2026 21:16:19 -0700
Subject: [PATCH 1/7] fix(cache): release evicted entries eagerly, not via
 epoch GC

---
 src/core/src/cache/index.rs        | 118 +++++++++----
 src/core/tests/memory_footprint.rs | 260 +++++++++++++++++++++++++++++
 2 files changed, 346 insertions(+), 32 deletions(-)
 create mode 100644 src/core/tests/memory_footprint.rs

diff --git a/src/core/src/cache/index.rs b/src/core/src/cache/index.rs
index a25fec75..fbcc2691 100644
--- a/src/core/src/cache/index.rs
+++ b/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,56 @@ 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)
+        // A slot emptied by a concurrent remove reads as a miss, exactly as if
+        // the remove had won the race outright.
+        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 +164,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);
+    }
 }
diff --git a/src/core/tests/memory_footprint.rs b/src/core/tests/memory_footprint.rs
new file mode 100644
index 00000000..83625588
--- /dev/null
+++ b/src/core/tests/memory_footprint.rs
@@ -0,0 +1,260 @@
+//! Regression test for liquid-cache#43: the process heap must track the
+//! cache's own budget tally for a working set larger than the memory tier.
+//! Every allocation in this test binary goes through a counting allocator, so
+//! "live" below is exact live heap, not RSS.
+
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use arrow::array::{ArrayRef, StringViewArray};
+use liquid_cache::cache::{
+    AlwaysHydrate, CacheEntry, EntryID, LiquidCache, LiquidCacheBuilder, LiquidPolicy,
+    TranscodeSqueezeEvict,
+};
+
+struct Counting;
+
+static LIVE: AtomicUsize = AtomicUsize::new(0);
+static PEAK: AtomicUsize = AtomicUsize::new(0);
+
+fn on_alloc(size: usize) {
+    let live = LIVE.fetch_add(size, Ordering::Relaxed) + size;
+    PEAK.fetch_max(live, Ordering::Relaxed);
+}
+
+unsafe impl GlobalAlloc for Counting {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        let p = unsafe { System.alloc(layout) };
+        if !p.is_null() {
+            on_alloc(layout.size());
+        }
+        p
+    }
+    unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
+        let p = unsafe { System.alloc_zeroed(layout) };
+        if !p.is_null() {
+            on_alloc(layout.size());
+        }
+        p
+    }
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        unsafe { System.dealloc(ptr, layout) };
+        LIVE.fetch_sub(layout.size(), Ordering::Relaxed);
+    }
+    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
+        let p = unsafe { System.realloc(ptr, layout, new_size) };
+        if !p.is_null() {
+            if new_size >= layout.size() {
+                on_alloc(new_size - layout.size());
+            } else {
+                LIVE.fetch_sub(layout.size() - new_size, Ordering::Relaxed);
+            }
+        }
+        p
+    }
+}
+
+#[global_allocator]
+static ALLOC: Counting = Counting;
+
+fn live() -> usize {
+    LIVE.load(Ordering::Relaxed)
+}
+fn reset_peak() {
+    PEAK.store(live(), Ordering::Relaxed);
+}
+fn peak() -> usize {
+    PEAK.load(Ordering::Relaxed)
+}
+fn mib(b: usize) -> f64 {
+    b as f64 / (1024.0 * 1024.0)
+}
+
+const WORDS: &[&str] = &[
+    "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel", "india", "juliet",
+    "kilo", "lima", "mike", "november", "oscar", "papa", "quebec", "romeo", "sierra", "tango",
+    "uniform", "victor", "whiskey", "xray", "yankee", "zulu", "server", "request", "latency",
+    "status", "payload", "region", "tenant", "shard",
+];
+
+/// ~1 KiB rows of semi-compressible text, unique per row, like a log/JSON column.
+fn make_entry(seed: u64, rows: usize) -> ArrayRef {
+    let mut state = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1;
+    let mut next = move || {
+        state ^= state << 13;
+        state ^= state >> 7;
+        state ^= state << 17;
+        state
+    };
+    let values: Vec<String> = (0..rows)
+        .map(|row| {
+            let mut s = format!("row={row:08} seed={seed:04} ");
+            while s.len() < 1024 {
+                let w = WORDS[(next() % WORDS.len() as u64) as usize];
+                s.push_str(w);
+                s.push_str(&format!("={:x} ", next() & 0xffff));
+            }
+            s
+        })
+        .collect();
+    Arc::new(StringViewArray::from_iter_values(values))
+}
+
+fn indexed_bytes(cache: &LiquidCache) -> usize {
+    let mut sum = 0;
+    cache.for_each_entry(|_, e| sum += e.memory_usage_bytes());
+    sum
+}
+
+/// Sum of `disk_bytes` over every on-disk entry, read straight from the
+/// index rather than the budget, so the budget can be checked against it.
+fn indexed_disk_bytes(cache: &LiquidCache) -> usize {
+    let mut sum = 0;
+    cache.for_each_entry(|_, e| {
+        sum += match e {
+            CacheEntry::DiskLiquid { disk_bytes, .. }
+            | CacheEntry::DiskArrow { disk_bytes, .. } => *disk_bytes,
+            _ => 0,
+        };
+    });
+    sum
+}
+
+fn report(cache: &LiquidCache, label: &str, baseline: usize) {
+    let stats = cache.stats();
+    let tally = cache.budget().memory_usage_bytes();
+    eprintln!(
+        "[{label}] live={:.1} MiB peak={:.1} MiB tally={:.1} MiB indexed={:.1} MiB disk={:.1} MiB \
+         entries(arrow={} liquid={} squeezed={} disk_liquid={} disk_arrow={})",
+        mib(live().saturating_sub(baseline)),
+        mib(peak().saturating_sub(baseline)),
+        mib(tally),
+        mib(indexed_bytes(cache)),
+        mib(cache.budget().disk_usage_bytes()),
+        stats.memory_arrow_entries,
+        stats.memory_liquid_entries,
+        stats.memory_squeezed_liquid_entries,
+        stats.disk_liquid_entries,
+        stats.disk_arrow_entries,
+    );
+}
+
+#[tokio::test]
+async fn heap_footprint_tracks_budget_for_oversized_working_set() {
+    const MEMORY_TIER: usize = 32 * 1024 * 1024;
+    const ROWS: usize = 2048;
+    const ENTRIES: usize = 96; // ~2 MiB arrow each → ~6x the memory tier
+
+    let dir = tempfile::tempdir().unwrap();
+    let store = liquid_cache::store::mount(&dir.path().join("cache.t4"))
+        .await
+        .unwrap();
+    let cache = LiquidCacheBuilder::new()
+        .with_max_memory_bytes(MEMORY_TIER)
+        .with_max_disk_bytes(4 << 30)
+        .with_batch_size(8192)
+        .with_cache_policy(Box::new(LiquidPolicy::new()))
+        .with_squeeze_policy(Box::new(TranscodeSqueezeEvict))
+        .with_hydration_policy(Box::new(AlwaysHydrate::new()))
+        .with_store(store)
+        .build()
+        .await;
+
+    let baseline = live();
+    reset_peak();
+
+    // Fill: mimics read_parquet_batch_and_fill_cache inserting each decoded
+    // column batch as arrow, dropping the caller's copy right after.
+    for i in 0..ENTRIES {
+        let arr = make_entry(i as u64, ROWS);
+        cache.insert(EntryID::from(i), arr).await.unwrap();
+    }
+    report(&cache, "after fill", baseline);
+    let idle_after_fill = live() - baseline;
+    let tally_after_fill = cache.budget().memory_usage_bytes();
+    assert!(
+        idle_after_fill as f64 <= 1.15 * tally_after_fill as f64,
+        "live ({:.1} MiB) must track the budget tally ({:.1} MiB) within 1.15x: \
+         evicted entries must not outlive the index (congee deferred-drop measured 6.14x)",
+        mib(idle_after_fill),
+        mib(tally_after_fill),
+    );
+    // Reset the IO counters here so the write count measured after the read
+    // pass below reflects only the warm churn, not the fill's own writes.
+    let _ = cache.observer().runtime_snapshot();
+
+    // Warm churn: a second scan over the same working set, every batch read
+    // once per pass, the materialized array dropped immediately.
+    reset_peak();
+    for _pass in 0..2 {
+        for i in 0..ENTRIES {
+            let arr = cache.get(&EntryID::from(i)).await.unwrap();
+            assert_eq!(arr.len(), ROWS);
+            drop(arr);
+        }
+    }
+    // Read before `report` (which also drains the counters via `cache.stats()`),
+    // so this reflects only the warm churn since the reset after fill above.
+    let rt = cache.observer().runtime_snapshot();
+    report(&cache, "after reads", baseline);
+    let peak_reads = peak() - baseline;
+    assert!(
+        peak_reads <= 2 * MEMORY_TIER + MEMORY_TIER / 2,
+        "peak during reads ({:.1} MiB) must stay within 2.5x the memory tier ({:.1} MiB) \
+         (measured 6.09x before the congee fix, 1.71x after)",
+        mib(peak_reads),
+        mib(MEMORY_TIER),
+    );
+    assert!(
+        (rt.write_io_count as usize) <= ENTRIES / 2,
+        "disk writes during the read pass ({}) must stay under half of the entry count ({}): \
+         a hydrated entry must not rewrite its disk copy on re-eviction (measured 190 for 96 entries before the fix, 30 after)",
+        rt.write_io_count,
+        ENTRIES,
+    );
+
+    // Flush every remaining memory entry to a disk stub: one with no disk
+    // copy yet must write, one that was hydrated must flip for free via the
+    // `write_in_memory_batch_to_disk` shortcut. Snapshot stats and IO
+    // counters right before, so both checks below are scoped to the flush.
+    let stats_before_flush = cache.stats();
+    let memory_entries_before_flush = stats_before_flush.memory_arrow_entries
+        + stats_before_flush.memory_liquid_entries
+        + stats_before_flush.memory_squeezed_liquid_entries;
+    cache.flush_all_to_disk().await.unwrap();
+    let flush_rt = cache.observer().runtime_snapshot();
+    assert!(
+        (flush_rt.write_io_count as usize) < memory_entries_before_flush,
+        "flush wrote {} times for {} in-memory entries: a hydrated entry's re-eviction must \
+         not reserve its disk object twice (drifted to 2.7x before the fix)",
+        flush_rt.write_io_count,
+        memory_entries_before_flush,
+    );
+
+    let budget_disk_bytes = cache.budget().disk_usage_bytes();
+    let index_disk_bytes = indexed_disk_bytes(&cache);
+    eprintln!(
+        "after flush: writes={} for {memory_entries_before_flush} memory entries, \
+         disk budget={:.1} MiB indexed_disk_entries={:.1} MiB",
+        flush_rt.write_io_count,
+        mib(budget_disk_bytes),
+        mib(index_disk_bytes),
+    );
+    assert_eq!(
+        budget_disk_bytes, index_disk_bytes,
+        "disk budget must equal the indexed on-disk bytes: a hydrated entry's re-eviction \
+         must not reserve its disk object twice (drifted to 2.7x before the fix)"
+    );
+
+    // What survives once the index is emptied is held by the store, the
+    // policy, or the compressor state — not by indexed entries.
+    cache.reset();
+    report(&cache, "after reset", baseline);
+    let idle_after_reset = live() - baseline;
+    assert!(
+        idle_after_reset <= 1024 * 1024,
+        "live after reset ({:.2} MiB) must drop back under 1 MiB",
+        mib(idle_after_reset),
+    );
+}

From f6c03d68fe53135e62dbdedc75997ac24f7641d9 Mon Sep 17 00:00:00 2001
From: Zac Farrell <zac@hotdata.dev>
Date: Wed, 2 Sep 2026 21:16:29 -0700
Subject: [PATCH 2/7] fix(cache): reuse a hydrated entry's disk copy on
 eviction

---
 src/core/src/cache/core.rs                    | 154 ++++++++++++++++--
 ...ests__policies__insert_wont_fit_cache.snap |   1 -
 2 files changed, 139 insertions(+), 16 deletions(-)

diff --git a/src/core/src/cache/core.rs b/src/core/src/cache/core.rs
index f26f2877..2e940f49 100644
--- a/src/core/src/cache/core.rs
+++ b/src/core/src/cache/core.rs
@@ -24,10 +24,48 @@ use crate::liquid_array::{
     LiquidSqueezedArrayRef, SqueezeIoHandler, SqueezedBacking, SqueezedDate32Array,
     VariantStructSqueezedArray,
 };
-use crate::sync::Arc;
+use crate::sync::{Arc, Mutex};
+use std::collections::HashMap;
 
 // CacheStats and RuntimeStats moved to stats.rs
 
+/// What the disk tier holds for an entry that is currently (also) in memory.
+///
+/// Hydrating a disk entry replaces its index entry with a memory one, but the
+/// bytes stay in the store under the same key and stay counted against the
+/// disk budget. Without this record, evicting the hydrated entry serialised
+/// and wrote the same bytes again — one redundant write per read of an
+/// oversized working set, and a second disk reservation for one object, so
+/// the disk tally drifted up until the tier evicted real entries early
+/// (liquid-cache#43).
+#[derive(Debug, Clone, Copy)]
+struct DiskCopy {
+    kind: DiskKind,
+    bytes: usize,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum DiskKind {
+    Liquid,
+    Arrow,
+}
+
+impl DiskCopy {
+    fn of(entry: &CacheEntry) -> Option<Self> {
+        match entry {
+            CacheEntry::DiskLiquid { disk_bytes, .. } => Some(Self {
+                kind: DiskKind::Liquid,
+                bytes: *disk_bytes,
+            }),
+            CacheEntry::DiskArrow { disk_bytes, .. } => Some(Self {
+                kind: DiskKind::Arrow,
+                bytes: *disk_bytes,
+            }),
+            _ => None,
+        }
+    }
+}
+
 /// Cache storage for liquid cache.
 ///
 /// Example (async read):
@@ -60,6 +98,7 @@ pub struct LiquidCache {
     metadata: Arc<dyn EntryMetadata>,
     store: t4::Store,
     squeeze_victims_concurrently: bool,
+    disk_copies: Mutex<HashMap<EntryID, DiskCopy>>,
 }
 
 /// Builder returned by [`LiquidCache::insert`] for configuring cache writes.
@@ -183,6 +222,7 @@ impl LiquidCache {
     pub fn reset(&self) {
         self.index.reset();
         self.budget.reset_usage();
+        self.disk_copies.lock().unwrap().clear();
     }
 
     /// Check if a batch is cached.
@@ -243,6 +283,19 @@ impl LiquidCache {
                     }
                 }
                 CacheEntry::MemoryLiquid(liquid_array) => {
+                    let data_type = liquid_array.original_arrow_data_type();
+                    if let Some(DiskCopy {
+                        kind: DiskKind::Liquid,
+                        bytes,
+                    }) = self.disk_copy(&entry_id)
+                    {
+                        // Hydrated from disk and never modified since: the
+                        // bytes are already there, flip the index rather
+                        // than re-serialising and rewriting them.
+                        self.try_insert(entry_id, CacheEntry::disk_liquid(data_type, bytes))
+                            .expect("failed to insert disk liquid entry");
+                        continue;
+                    }
                     let liquid_bytes = liquid_array.to_bytes();
                     let disk_bytes = liquid_bytes.len();
                     match self
@@ -252,10 +305,7 @@ impl LiquidCache {
                         Ok(()) => {
                             self.try_insert(
                                 entry_id,
-                                CacheEntry::disk_liquid(
-                                    liquid_array.original_arrow_data_type(),
-                                    disk_bytes,
-                                ),
+                                CacheEntry::disk_liquid(data_type, disk_bytes),
                             )
                             .expect("failed to insert disk liquid entry");
                         }
@@ -311,14 +361,19 @@ impl LiquidCache {
                 Ok(new_batch)
             }
             CacheEntry::MemoryLiquid(liquid_array) => {
+                let data_type = liquid_array.original_arrow_data_type();
+                if let Some(DiskCopy {
+                    kind: DiskKind::Liquid,
+                    bytes,
+                }) = self.disk_copy(&entry_id)
+                {
+                    return Ok(CacheEntry::disk_liquid(data_type, bytes));
+                }
                 let liquid_bytes = Bytes::from(liquid_array.to_bytes());
                 let disk_bytes = liquid_bytes.len();
                 self.write_batch_to_disk(entry_id, &batch, liquid_bytes)
                     .await?;
-                Ok(CacheEntry::disk_liquid(
-                    liquid_array.original_arrow_data_type(),
-                    disk_bytes,
-                ))
+                Ok(CacheEntry::disk_liquid(data_type, disk_bytes))
             }
             CacheEntry::MemorySqueezedLiquid(squeezed_array) => {
                 // The full data is already on disk, so we just need to mark ourself as disk entry
@@ -398,6 +453,43 @@ impl LiquidCache {
             metadata,
             store,
             squeeze_victims_concurrently,
+            disk_copies: Mutex::new(HashMap::new()),
+        }
+    }
+
+    fn disk_copy(&self, entry_id: &EntryID) -> Option<DiskCopy> {
+        self.disk_copies.lock().unwrap().get(entry_id).copied()
+    }
+
+    /// If `outcome` demotes an entry to a disk stub whose bytes are already in
+    /// the store, drop the write and point the stub at the existing copy.
+    fn reuse_disk_copy(&self, entry_id: &EntryID, outcome: SqueezeOutcome) -> SqueezeOutcome {
+        let SqueezeOutcome::Replace {
+            entry,
+            bytes_to_write: Some(_),
+        } = &outcome
+        else {
+            return outcome;
+        };
+        let (Some(copy), Some(stub)) = (self.disk_copy(entry_id), DiskCopy::of(entry)) else {
+            return outcome;
+        };
+        if copy.kind != stub.kind {
+            return outcome;
+        }
+        let data_type = match entry {
+            CacheEntry::DiskLiquid { data_type, .. } | CacheEntry::DiskArrow { data_type, .. } => {
+                data_type.clone()
+            }
+            _ => unreachable!("DiskCopy::of only matches disk stubs"),
+        };
+        let entry = match copy.kind {
+            DiskKind::Liquid => CacheEntry::disk_liquid(data_type, copy.bytes),
+            DiskKind::Arrow => CacheEntry::disk_arrow(data_type, copy.bytes),
+        };
+        SqueezeOutcome::Replace {
+            entry,
+            bytes_to_write: None,
         }
     }
 
@@ -466,6 +558,7 @@ impl LiquidCache {
             .remove(&entry_id_to_key(&entry_id))
             .await
             .expect("disk remove failed");
+        self.disk_copies.lock().unwrap().remove(&entry_id);
         self.budget.release_disk(disk_bytes);
         self.cache_policy.notify_remove(&entry_id);
         self.trace(InternalEvent::DiskEvict {
@@ -524,12 +617,31 @@ impl LiquidCache {
         ));
 
         loop {
-            let outcome = self.squeeze_policy.squeeze(
-                to_squeeze_batch.as_ref(),
-                compressor.as_ref(),
-                squeeze_hint,
-                &squeeze_io,
-            );
+            // A liquid entry hydrated from the disk tier still has its bytes
+            // there: demote it by flipping the index entry rather than
+            // re-serialising it into a hybrid whose backing would have to be
+            // written all over again.
+            let outcome = match (to_squeeze_batch.as_ref(), self.disk_copy(&to_squeeze)) {
+                (
+                    CacheEntry::MemoryLiquid(liquid),
+                    Some(DiskCopy {
+                        kind: DiskKind::Liquid,
+                        bytes,
+                    }),
+                ) => SqueezeOutcome::Replace {
+                    entry: CacheEntry::disk_liquid(liquid.original_arrow_data_type(), bytes),
+                    bytes_to_write: None,
+                },
+                _ => {
+                    let outcome = self.squeeze_policy.squeeze(
+                        to_squeeze_batch.as_ref(),
+                        compressor.as_ref(),
+                        squeeze_hint,
+                        &squeeze_io,
+                    );
+                    self.reuse_disk_copy(&to_squeeze, outcome)
+                }
+            };
 
             match outcome {
                 SqueezeOutcome::Replace {
@@ -815,6 +927,18 @@ impl LiquidCache {
             .put(entry_id_to_key(&entry_id), bytes.to_vec())
             .await
             .expect("write failed");
+        let kind = match batch {
+            CacheEntry::DiskArrow { .. } => DiskKind::Arrow,
+            CacheEntry::MemorySqueezedLiquid(squeezed) => match squeezed.disk_backing() {
+                SqueezedBacking::Arrow(_) => DiskKind::Arrow,
+                SqueezedBacking::Liquid(_) => DiskKind::Liquid,
+            },
+            _ => DiskKind::Liquid,
+        };
+        self.disk_copies
+            .lock()
+            .unwrap()
+            .insert(entry_id, DiskCopy { kind, bytes: len });
         Ok(())
     }
 
diff --git a/src/core/src/cache/tests/snapshots/liquid_cache__cache__tests__policies__insert_wont_fit_cache.snap b/src/core/src/cache/tests/snapshots/liquid_cache__cache__tests__policies__insert_wont_fit_cache.snap
index 16dbb27b..5c06ad9c 100644
--- a/src/core/src/cache/tests/snapshots/liquid_cache__cache__tests__policies__insert_wont_fit_cache.snap
+++ b/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
 ]

From 2cf823e077960f8e3784d258bf2e885175e2a9b6 Mon Sep 17 00:00:00 2001
From: Zac Farrell <zac@hotdata.dev>
Date: Wed, 2 Sep 2026 21:16:39 -0700
Subject: [PATCH 3/7] test(liquid_array): guard decoded byte-view size
 accounting

---
 .../src/liquid_array/byte_view_array/tests.rs | 29 +++++++++++++++++++
 1 file changed, 29 insertions(+)

diff --git a/src/core/src/liquid_array/byte_view_array/tests.rs b/src/core/src/liquid_array/byte_view_array/tests.rs
index 63267e34..9921252a 100644
--- a/src/core/src/liquid_array/byte_view_array/tests.rs
+++ b/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);
+}

From 7cb9f7c16233eb1e46ccf191a28f0c0e28563226 Mon Sep 17 00:00:00 2001
From: Zac Farrell <zac@hotdata.dev>
Date: Thu, 3 Sep 2026 08:18:42 -0700
Subject: [PATCH 4/7] fix(cache): retry index lookup after a concurrent replace

`ArtIndex::insert` empties the replaced slot, so a reader that had
already loaded the old slot saw `None` for a key that is still present.
`is_cached` then reported false and `try_insert` reserved fresh memory
for the new value instead of releasing the replaced one. Look the key
up once more when the slot is empty.
---
 src/core/src/cache/index.rs | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/src/core/src/cache/index.rs b/src/core/src/cache/index.rs
index fbcc2691..cf8881ab 100644
--- a/src/core/src/cache/index.rs
+++ b/src/core/src/cache/index.rs
@@ -57,8 +57,15 @@ impl ArtIndex {
 
     pub(crate) fn get(&self, entry_id: &EntryID) -> Option<Arc<CacheEntry>> {
         let guard = self.art.pin();
-        // A slot emptied by a concurrent remove reads as a miss, exactly as if
-        // the remove had won the race outright.
+        // 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()
     }
 

From 3ae1a4e1cd346fca8d7570fbc191d72be2f0b375 Mon Sep 17 00:00:00 2001
From: Zac Farrell <zac@hotdata.dev>
Date: Thu, 3 Sep 2026 08:18:43 -0700
Subject: [PATCH 5/7] fix(cache): drop a superseded disk copy on overwrite

A caller overwriting an entry left the disk-copy record pointing at
the previous value, so a later demotion flipped the index to a stub
over stale bytes and `get` returned them. `Insert` now removes the
object and its reservation before inserting; hydration, which shares
`insert_inner`, keeps the record because the bytes are still current.

Also in `write_batch_to_disk`: a flushed arrow entry was recorded as a
liquid copy, so a hydrated-then-transcoded entry could be demoted to a
liquid stub over Arrow IPC bytes; and a put that replaces the object
under a key now releases the previous copy's reservation.

The hydrated-liquid shortcut in `squeeze_victim_inner` now applies
only without a squeeze hint (no `LiquidArray::squeeze` produces a
squeezed form unhinted); with one the policy runs, so the entry can
return to the squeezed tier, and `reuse_disk_copy` drops the write
when the squeezed form's backing is the copy already on disk.
---
 src/core/src/cache/builders.rs |   1 +
 src/core/src/cache/core.rs     | 314 ++++++++++++++++++++++++++++++---
 2 files changed, 286 insertions(+), 29 deletions(-)

diff --git a/src/core/src/cache/builders.rs b/src/core/src/cache/builders.rs
index 2607f750..af0572da 100644
--- a/src/core/src/cache/builders.rs
+++ b/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
     }
 }
diff --git a/src/core/src/cache/core.rs b/src/core/src/cache/core.rs
index 2e940f49..e519e3fc 100644
--- a/src/core/src/cache/core.rs
+++ b/src/core/src/cache/core.rs
@@ -51,7 +51,9 @@ enum DiskKind {
 }
 
 impl DiskCopy {
-    fn of(entry: &CacheEntry) -> Option<Self> {
+    /// The store object an entry refers to: a disk stub's bytes, or the
+    /// full serialisation a squeezed entry reads back through.
+    fn referenced_by(entry: &CacheEntry) -> Option<Self> {
         match entry {
             CacheEntry::DiskLiquid { disk_bytes, .. } => Some(Self {
                 kind: DiskKind::Liquid,
@@ -61,7 +63,17 @@ impl DiskCopy {
                 kind: DiskKind::Arrow,
                 bytes: *disk_bytes,
             }),
-            _ => None,
+            CacheEntry::MemorySqueezedLiquid(squeezed) => Some(match squeezed.disk_backing() {
+                SqueezedBacking::Liquid(bytes) => Self {
+                    kind: DiskKind::Liquid,
+                    bytes,
+                },
+                SqueezedBacking::Arrow(bytes) => Self {
+                    kind: DiskKind::Arrow,
+                    bytes,
+                },
+            }),
+            CacheEntry::MemoryArrow(_) | CacheEntry::MemoryLiquid(_) => None,
         }
     }
 }
@@ -461,31 +473,83 @@ impl LiquidCache {
         self.disk_copies.lock().unwrap().get(entry_id).copied()
     }
 
-    /// If `outcome` demotes an entry to a disk stub whose bytes are already in
-    /// the store, drop the write and point the stub at the existing copy.
+    /// A caller-supplied value supersedes whatever the store holds for the
+    /// entry. Hydration keeps the record, because the bytes on disk are still
+    /// the value in memory; an overwrite must not, or a later demotion would
+    /// flip the index to a stub over the previous value. Only [`Insert`] calls
+    /// this: `insert_inner` is shared with `maybe_hydrate`.
+    ///
+    /// A concurrent overwrite and squeeze of one entry is not serialised here
+    /// or anywhere else in the cache (a squeeze that read the old value can
+    /// still land its result after the new one), so this covers the
+    /// sequential case only.
+    pub(crate) async fn supersede_disk_copy(&self, entry_id: EntryID) {
+        if self.disk_copy(&entry_id).is_none() {
+            return;
+        }
+        if let Some(entry) = self.index.get(&entry_id)
+            && matches!(
+                entry.as_ref(),
+                CacheEntry::DiskLiquid { .. } | CacheEntry::DiskArrow { .. }
+            )
+        {
+            // Still a stub: the whole entry is the superseded object.
+            self.remove_disk_entry(entry_id).await;
+            return;
+        }
+        let Some(copy) = self.disk_copies.lock().unwrap().remove(&entry_id) else {
+            return;
+        };
+        self.store
+            .remove(&entry_id_to_key(&entry_id))
+            .await
+            .expect("disk remove failed");
+        self.budget.release_disk(copy.bytes);
+    }
+
+    /// If `outcome` demotes an entry to a form backed by a store object whose
+    /// bytes are already there, drop the write and point the entry at the
+    /// existing copy.
     fn reuse_disk_copy(&self, entry_id: &EntryID, outcome: SqueezeOutcome) -> SqueezeOutcome {
-        let SqueezeOutcome::Replace {
+        let (entry, bytes) = match outcome {
+            SqueezeOutcome::Replace {
+                entry,
+                bytes_to_write: Some(bytes),
+            } => (entry, bytes),
+            other => return other,
+        };
+        let keep_write = move |entry| SqueezeOutcome::Replace {
             entry,
-            bytes_to_write: Some(_),
-        } = &outcome
-        else {
-            return outcome;
+            bytes_to_write: Some(bytes),
         };
-        let (Some(copy), Some(stub)) = (self.disk_copy(entry_id), DiskCopy::of(entry)) else {
-            return outcome;
+        let (Some(copy), Some(wanted)) =
+            (self.disk_copy(entry_id), DiskCopy::referenced_by(&entry))
+        else {
+            return keep_write(entry);
         };
-        if copy.kind != stub.kind {
-            return outcome;
+        if copy.kind != wanted.kind {
+            return keep_write(entry);
         }
-        let data_type = match entry {
-            CacheEntry::DiskLiquid { data_type, .. } | CacheEntry::DiskArrow { data_type, .. } => {
-                data_type.clone()
+        let entry = match entry {
+            // A squeezed entry reads back through the full serialisation the
+            // policy handed over to be written. A copy of the same kind and
+            // length is that serialisation (the array was hydrated from it),
+            // so the entry can keep its backing as chosen.
+            CacheEntry::MemorySqueezedLiquid(_) => {
+                if wanted.bytes != copy.bytes {
+                    return keep_write(entry);
+                }
+                entry
+            }
+            CacheEntry::DiskLiquid { data_type, .. } => {
+                CacheEntry::disk_liquid(data_type, copy.bytes)
+            }
+            CacheEntry::DiskArrow { data_type, .. } => {
+                CacheEntry::disk_arrow(data_type, copy.bytes)
+            }
+            CacheEntry::MemoryArrow(_) | CacheEntry::MemoryLiquid(_) => {
+                unreachable!("referenced_by only matches entries backed by a store object")
             }
-            _ => unreachable!("DiskCopy::of only matches disk stubs"),
-        };
-        let entry = match copy.kind {
-            DiskKind::Liquid => CacheEntry::disk_liquid(data_type, copy.bytes),
-            DiskKind::Arrow => CacheEntry::disk_arrow(data_type, copy.bytes),
         };
         SqueezeOutcome::Replace {
             entry,
@@ -618,16 +682,25 @@ impl LiquidCache {
 
         loop {
             // A liquid entry hydrated from the disk tier still has its bytes
-            // there: demote it by flipping the index entry rather than
-            // re-serialising it into a hybrid whose backing would have to be
-            // written all over again.
-            let outcome = match (to_squeeze_batch.as_ref(), self.disk_copy(&to_squeeze)) {
+            // there. Without a squeeze hint every policy demotes it to a disk
+            // stub (no `LiquidArray::squeeze` produces a squeezed form
+            // unhinted), so flip the index entry rather than re-serialising
+            // the array to arrive at the same stub. With a hint the policy
+            // runs, so the entry can return to the squeezed tier, and
+            // `reuse_disk_copy` drops the write when the squeezed form's
+            // backing is the copy already on disk.
+            let outcome = match (
+                to_squeeze_batch.as_ref(),
+                self.disk_copy(&to_squeeze),
+                squeeze_hint,
+            ) {
                 (
                     CacheEntry::MemoryLiquid(liquid),
                     Some(DiskCopy {
                         kind: DiskKind::Liquid,
                         bytes,
                     }),
+                    None,
                 ) => SqueezeOutcome::Replace {
                     entry: CacheEntry::disk_liquid(liquid.original_arrow_data_type(), bytes),
                     bytes_to_write: None,
@@ -927,18 +1000,26 @@ impl LiquidCache {
             .put(entry_id_to_key(&entry_id), bytes.to_vec())
             .await
             .expect("write failed");
+        // `bytes` is whatever `batch` serialises to: Arrow IPC for an arrow
+        // entry (the flush path writes those directly), liquid otherwise.
         let kind = match batch {
-            CacheEntry::DiskArrow { .. } => DiskKind::Arrow,
+            CacheEntry::DiskArrow { .. } | CacheEntry::MemoryArrow(_) => DiskKind::Arrow,
             CacheEntry::MemorySqueezedLiquid(squeezed) => match squeezed.disk_backing() {
                 SqueezedBacking::Arrow(_) => DiskKind::Arrow,
                 SqueezedBacking::Liquid(_) => DiskKind::Liquid,
             },
-            _ => DiskKind::Liquid,
+            CacheEntry::DiskLiquid { .. } | CacheEntry::MemoryLiquid(_) => DiskKind::Liquid,
         };
-        self.disk_copies
+        let previous = self
+            .disk_copies
             .lock()
             .unwrap()
             .insert(entry_id, DiskCopy { kind, bytes: len });
+        if let Some(previous) = previous {
+            // The put replaced the object under this key, so the previous
+            // copy's reservation goes with it.
+            self.budget.release_disk(previous.bytes);
+        }
         Ok(())
     }
 
@@ -1090,7 +1171,7 @@ impl LiquidCache {
 mod tests {
     use super::*;
     use crate::cache::{
-        CacheEntry, CacheExpression, CachePolicy, LiquidCacheBuilder, LiquidPolicy,
+        AlwaysHydrate, CacheEntry, CacheExpression, CachePolicy, LiquidCacheBuilder, LiquidPolicy,
         TranscodeSqueezeEvict, transcode_liquid_inner,
         utils::{
             LiquidCompressorStates, arrow_to_bytes, create_cache_store, create_test_array,
@@ -1487,4 +1568,179 @@ mod tests {
         assert_eq!(result, Ok(()));
         assert!(!cache.is_cached(&entry_id));
     }
+
+    async fn hydrating_cache() -> Arc<LiquidCache> {
+        LiquidCacheBuilder::new()
+            .with_max_memory_bytes(1 << 20)
+            .with_max_disk_bytes(1 << 20)
+            .with_squeeze_policy(Box::new(TranscodeSqueezeEvict))
+            .with_cache_policy(Box::new(LiquidPolicy::new()))
+            .with_hydration_policy(Box::new(AlwaysHydrate::new()))
+            .build()
+            .await
+    }
+
+    /// Two squeezes take a fresh arrow entry through liquid to a disk stub.
+    async fn demote_to_disk(cache: &LiquidCache, id: EntryID) -> usize {
+        cache.squeeze_victims(vec![id]).await.unwrap();
+        cache.squeeze_victims(vec![id]).await.unwrap();
+        let entry = cache.index().get(&id).expect("entry present");
+        let CacheEntry::DiskLiquid { disk_bytes, .. } = entry.as_ref() else {
+            panic!("expected a disk stub, got {entry:?}");
+        };
+        *disk_bytes
+    }
+
+    /// Overwriting an entry that sits on disk must drop the disk copy of the
+    /// value it replaces: demoting the new value must not flip the index to a
+    /// stub over the old bytes, and the old reservation must be released.
+    #[tokio::test]
+    async fn overwrite_of_disk_stub_invalidates_disk_copy() {
+        let cache = hydrating_cache().await;
+        let id = EntryID::from(920usize);
+        let v1: ArrayRef = Arc::new(Int32Array::from_iter_values(0..16));
+        let v2: ArrayRef = Arc::new(Int32Array::from_iter_values(100..164));
+
+        cache.insert(id, v1).await.unwrap();
+        let v1_disk_bytes = demote_to_disk(&cache, id).await;
+        assert_eq!(cache.budget().disk_usage_bytes(), v1_disk_bytes);
+
+        cache.insert(id, v2.clone()).await.unwrap();
+        let disk_after_overwrite = cache.budget().disk_usage_bytes();
+
+        let v2_disk_bytes = demote_to_disk(&cache, id).await;
+        let read = cache.get(&id).await.expect("present");
+        assert_eq!(read.as_ref(), v2.as_ref(), "read back the superseded value");
+        assert_eq!(
+            disk_after_overwrite, 0,
+            "the superseded object's reservation must be released"
+        );
+        assert_eq!(cache.budget().disk_usage_bytes(), v2_disk_bytes);
+    }
+
+    /// The same, for an entry that was hydrated back into memory before the
+    /// overwrite, so the index holds a memory entry and only the disk-copy
+    /// record points at the old bytes.
+    #[tokio::test]
+    async fn overwrite_of_hydrated_entry_invalidates_disk_copy() {
+        let cache = hydrating_cache().await;
+        let id = EntryID::from(921usize);
+        let v1: ArrayRef = Arc::new(Int32Array::from_iter_values(0..16));
+        let v2: ArrayRef = Arc::new(Int32Array::from_iter_values(100..164));
+
+        cache.insert(id, v1.clone()).await.unwrap();
+        let v1_disk_bytes = demote_to_disk(&cache, id).await;
+        let read = cache.get(&id).await.expect("present");
+        assert_eq!(read.as_ref(), v1.as_ref());
+        assert!(matches!(
+            cache.index().get(&id).unwrap().as_ref(),
+            CacheEntry::MemoryLiquid(_)
+        ));
+        assert_eq!(cache.budget().disk_usage_bytes(), v1_disk_bytes);
+
+        cache.insert(id, v2.clone()).await.unwrap();
+        let disk_after_overwrite = cache.budget().disk_usage_bytes();
+
+        let v2_disk_bytes = demote_to_disk(&cache, id).await;
+        let read = cache.get(&id).await.expect("present");
+        assert_eq!(read.as_ref(), v2.as_ref(), "read back the superseded value");
+        assert_eq!(disk_after_overwrite, 0);
+        assert_eq!(cache.budget().disk_usage_bytes(), v2_disk_bytes);
+    }
+
+    /// A flush writes an arrow entry as Arrow IPC and must record the copy as
+    /// such: once hydrated and transcoded, the entry is demoted through the
+    /// policy to freshly written liquid bytes, not flipped to a liquid stub
+    /// over the arrow bytes. The replaced object's reservation is released.
+    #[tokio::test]
+    async fn flushed_arrow_copy_is_not_reused_as_liquid() {
+        let cache = hydrating_cache().await;
+        let id = EntryID::from(922usize);
+        let array: ArrayRef = Arc::new(Int32Array::from_iter_values(0..64));
+
+        cache.insert(id, array.clone()).await.unwrap();
+        cache.flush_all_to_disk().await.unwrap();
+        assert!(matches!(
+            cache.index().get(&id).unwrap().as_ref(),
+            CacheEntry::DiskArrow { .. }
+        ));
+        let read = cache.get(&id).await.expect("present");
+        assert_eq!(read.as_ref(), array.as_ref());
+        assert!(matches!(
+            cache.index().get(&id).unwrap().as_ref(),
+            CacheEntry::MemoryArrow(_)
+        ));
+
+        let disk_bytes = demote_to_disk(&cache, id).await;
+        assert_eq!(cache.budget().disk_usage_bytes(), disk_bytes);
+        let read = cache.get(&id).await.expect("present");
+        assert_eq!(read.as_ref(), array.as_ref());
+    }
+
+    /// A hinted entry rehydrated from its liquid disk copy must still reach
+    /// the squeezed tier on its next demotion, and must not rewrite the copy
+    /// its squeezed form reads back through.
+    #[tokio::test]
+    async fn rehydrated_hinted_entry_returns_to_squeezed_tier_without_rewrite() {
+        let cache = hydrating_cache().await;
+        let id = EntryID::from(923usize);
+        let dates: ArrayRef = Arc::new(Date32Array::from(vec![
+            Some(2),
+            Some(365 + 1),
+            None,
+            Some(365 + 100),
+        ]));
+        let expr = Arc::new(CacheExpression::extract_date32(Date32Field::Year));
+
+        cache
+            .insert(id, dates.clone())
+            .with_squeeze_hint(expr.clone())
+            .await
+            .unwrap();
+        for _ in 0..3 {
+            cache.squeeze_victims(vec![id]).await.unwrap();
+        }
+        let entry = cache.index().get(&id).unwrap();
+        let CacheEntry::DiskLiquid { disk_bytes, .. } = entry.as_ref() else {
+            panic!("expected a disk stub, got {entry:?}");
+        };
+        let disk_bytes = *disk_bytes;
+        assert_eq!(cache.budget().disk_usage_bytes(), disk_bytes);
+        // Drain the IO counters so the count below covers only the re-eviction.
+        let _ = cache.observer().runtime_snapshot();
+
+        let read = cache.get(&id).await.expect("present");
+        assert_eq!(read.as_ref(), dates.as_ref());
+        assert!(matches!(
+            cache.index().get(&id).unwrap().as_ref(),
+            CacheEntry::MemoryLiquid(_)
+        ));
+
+        cache.squeeze_victims(vec![id]).await.unwrap();
+        assert!(
+            matches!(
+                cache.index().get(&id).unwrap().as_ref(),
+                CacheEntry::MemorySqueezedLiquid(_)
+            ),
+            "the squeeze policy must run for a hinted entry"
+        );
+        assert_eq!(
+            cache.observer().runtime_snapshot().write_io_count,
+            0,
+            "the squeezed form's backing is already on disk"
+        );
+        assert_eq!(cache.budget().disk_usage_bytes(), disk_bytes);
+
+        let years = cache
+            .get(&id)
+            .with_expression_hint(expr)
+            .read()
+            .await
+            .expect("present");
+        let years = years.as_any().downcast_ref::<Date32Array>().unwrap();
+        assert_eq!(years.value(0), 0);
+        assert_eq!(years.value(1), 365);
+        assert!(years.is_null(2));
+        assert_eq!(years.value(3), 365);
+    }
 }

From 511f5120689d972ae84d5597464f15d286da522f Mon Sep 17 00:00:00 2001
From: Zac Farrell <zac@hotdata.dev>
Date: Thu, 3 Sep 2026 08:43:14 -0700
Subject: [PATCH 6/7] test(datafusion-local): refresh snapshots for disk-copy
 reuse

---
 ...quid_cache_datafusion_local__tests__referer_filtering.snap | 2 +-
 ...che_datafusion_local__tests__squeeze__squeeze_strings.snap | 1 -
 ...e_datafusion_local__tests__url_selection_and_ordering.snap | 4 ++--
 3 files changed, 3 insertions(+), 4 deletions(-)

diff --git a/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__referer_filtering.snap b/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__referer_filtering.snap
index df9e1613..34345c2b 100644
--- a/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__referer_filtering.snap
+++ b/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__referer_filtering.snap
@@ -41,7 +41,7 @@ entries.memory.squeezed_liquid: 0
 entries.disk.liquid: 0
 entries.disk.arrow: 0
 usage.memory_bytes: 917947
-usage.disk_bytes: 877216
+usage.disk_bytes: 729144
 RuntimeStatsSnapshot:
   get: 2
   get_with_selection: 2
diff --git a/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__squeeze__squeeze_strings.snap b/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__squeeze__squeeze_strings.snap
index b54a6066..9b54c6e5 100644
--- a/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__squeeze__squeeze_strings.snap
+++ b/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__squeeze__squeeze_strings.snap
@@ -26,7 +26,6 @@ event=squeeze_begin victims=[851969,851968,917504]
 event=squeeze_victim entry=851969
 event=insert_success entry=851969 kind=MemoryLiquid
 event=squeeze_victim entry=851968
-event=io_write entry=851968 kind=DiskLiquid bytes=139416
 event=insert_success entry=851968 kind=DiskLiquid
 event=squeeze_victim entry=917504
 event=io_write entry=917504 kind=MemorySqueezedLiquid bytes=136440
diff --git a/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__url_selection_and_ordering.snap b/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__url_selection_and_ordering.snap
index 7b6d7f6d..284a11df 100644
--- a/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__url_selection_and_ordering.snap
+++ b/src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__url_selection_and_ordering.snap
@@ -42,7 +42,7 @@ entries.memory.squeezed_liquid: 1
 entries.disk.liquid: 1
 entries.disk.arrow: 0
 usage.memory_bytes: 189553
-usage.disk_bytes: 706252
+usage.disk_bytes: 427460
 RuntimeStatsSnapshot:
   get: 3
   get_with_selection: 3
@@ -52,7 +52,7 @@ RuntimeStatsSnapshot:
   try_read_liquid_calls: 0
   hit_date32_expression_calls: 0
   read_io_count: 4
-  write_io_count: 4
+  write_io_count: 2
   disk_evictions: 0
   disk_reservation_failures: 0
   eval_predicate_on_liquid_failed: 0

From ad483adf88a90e814fa5dc1866fa9439b473ada6 Mon Sep 17 00:00:00 2001
From: Zac Farrell <zac@hotdata.dev>
Date: Thu, 3 Sep 2026 08:50:44 -0700
Subject: [PATCH 7/7] fix(cache): drop a squeezed entry and its copy on
 overwrite

A squeezed entry reads back through its store object, so
`supersede_disk_copy` cannot delete that object and leave the entry in
the index: an insert that then fails with `CacheFull` left an entry
whose next read panicked. The squeezed entry now goes the way of a
disk stub. `drop_memory_entry` likewise discards the disk copy of the
entry it drops, which the flush overflow path leaked as an orphaned
object and a permanent reservation.

The hydrated-liquid shortcut in `squeeze_victim_inner` is gone: float
arrays squeeze without a hint, so it kept them from the squeezed tier.
The policy always runs and `reuse_disk_copy` drops the rewrite.
---
 src/core/src/cache/core.rs | 189 ++++++++++++++++++++++++++++---------
 1 file changed, 145 insertions(+), 44 deletions(-)

diff --git a/src/core/src/cache/core.rs b/src/core/src/cache/core.rs
index e519e3fc..2f85a1a7 100644
--- a/src/core/src/cache/core.rs
+++ b/src/core/src/cache/core.rs
@@ -291,7 +291,7 @@ impl LiquidCache {
                             )
                             .expect("failed to insert disk arrow entry");
                         }
-                        Err(CacheFull) => self.drop_memory_entry(entry_id, &batch),
+                        Err(CacheFull) => self.drop_memory_entry(entry_id, &batch).await,
                     }
                 }
                 CacheEntry::MemoryLiquid(liquid_array) => {
@@ -321,7 +321,7 @@ impl LiquidCache {
                             )
                             .expect("failed to insert disk liquid entry");
                         }
-                        Err(CacheFull) => self.drop_memory_entry(entry_id, &batch),
+                        Err(CacheFull) => self.drop_memory_entry(entry_id, &batch).await,
                     }
                 }
                 CacheEntry::MemorySqueezedLiquid(array) => {
@@ -487,16 +487,27 @@ impl LiquidCache {
         if self.disk_copy(&entry_id).is_none() {
             return;
         }
-        if let Some(entry) = self.index.get(&entry_id)
-            && matches!(
-                entry.as_ref(),
-                CacheEntry::DiskLiquid { .. } | CacheEntry::DiskArrow { .. }
-            )
-        {
-            // Still a stub: the whole entry is the superseded object.
-            self.remove_disk_entry(entry_id).await;
-            return;
+        match self.index.get(&entry_id).as_deref() {
+            Some(CacheEntry::DiskLiquid { .. } | CacheEntry::DiskArrow { .. }) => {
+                // Still a stub: the whole entry is the superseded object.
+                self.remove_disk_entry(entry_id).await;
+            }
+            Some(squeezed @ CacheEntry::MemorySqueezedLiquid(_)) => {
+                // A squeezed entry reads back through the object too, so it
+                // cannot stay in the index over a deleted one, not even for
+                // the span of an insert that then fails with `CacheFull`.
+                self.drop_memory_entry(entry_id, squeezed).await;
+            }
+            Some(CacheEntry::MemoryArrow(_) | CacheEntry::MemoryLiquid(_)) | None => {
+                self.discard_disk_copy(entry_id).await;
+            }
         }
+    }
+
+    /// Delete the store object recorded for `entry_id`, if any, and release
+    /// its reservation. The index entry, if one remains, must not be a form
+    /// that reads through the object.
+    async fn discard_disk_copy(&self, entry_id: EntryID) {
         let Some(copy) = self.disk_copies.lock().unwrap().remove(&entry_id) else {
             return;
         };
@@ -590,7 +601,10 @@ impl LiquidCache {
         Ok(())
     }
 
-    fn drop_memory_entry(&self, entry_id: EntryID, _expected: &CacheEntry) {
+    /// Drop a memory entry from the cache altogether, including the disk copy
+    /// it may hold: with the index entry gone nothing could reach that object
+    /// again, and its reservation would shrink the disk tier for good.
+    async fn drop_memory_entry(&self, entry_id: EntryID, _expected: &CacheEntry) {
         let Some(removed) = self.index.remove(&entry_id) else {
             return;
         };
@@ -606,6 +620,7 @@ impl LiquidCache {
         self.budget
             .try_update_memory_usage(removed.memory_usage_bytes(), 0)
             .expect("memory release cannot fail");
+        self.discard_disk_copy(entry_id).await;
         self.cache_policy.notify_remove(&entry_id);
     }
 
@@ -681,40 +696,18 @@ impl LiquidCache {
         ));
 
         loop {
-            // A liquid entry hydrated from the disk tier still has its bytes
-            // there. Without a squeeze hint every policy demotes it to a disk
-            // stub (no `LiquidArray::squeeze` produces a squeezed form
-            // unhinted), so flip the index entry rather than re-serialising
-            // the array to arrive at the same stub. With a hint the policy
-            // runs, so the entry can return to the squeezed tier, and
-            // `reuse_disk_copy` drops the write when the squeezed form's
-            // backing is the copy already on disk.
-            let outcome = match (
+            // The policy always decides the next form, so an entry hydrated
+            // from the disk tier can still reach the squeezed tier (floats
+            // squeeze even without a hint). `reuse_disk_copy` then drops the
+            // write when the form's backing is the copy already on disk; the
+            // serialisation the policy produced for it is the only cost.
+            let outcome = self.squeeze_policy.squeeze(
                 to_squeeze_batch.as_ref(),
-                self.disk_copy(&to_squeeze),
+                compressor.as_ref(),
                 squeeze_hint,
-            ) {
-                (
-                    CacheEntry::MemoryLiquid(liquid),
-                    Some(DiskCopy {
-                        kind: DiskKind::Liquid,
-                        bytes,
-                    }),
-                    None,
-                ) => SqueezeOutcome::Replace {
-                    entry: CacheEntry::disk_liquid(liquid.original_arrow_data_type(), bytes),
-                    bytes_to_write: None,
-                },
-                _ => {
-                    let outcome = self.squeeze_policy.squeeze(
-                        to_squeeze_batch.as_ref(),
-                        compressor.as_ref(),
-                        squeeze_hint,
-                        &squeeze_io,
-                    );
-                    self.reuse_disk_copy(&to_squeeze, outcome)
-                }
-            };
+                &squeeze_io,
+            );
+            let outcome = self.reuse_disk_copy(&to_squeeze, outcome);
 
             match outcome {
                 SqueezeOutcome::Replace {
@@ -1743,4 +1736,112 @@ mod tests {
         assert!(years.is_null(2));
         assert_eq!(years.value(3), 365);
     }
+
+    /// A policy with no eviction advice, so an insert that does not fit
+    /// falls straight through to the disk tier.
+    #[derive(Debug)]
+    struct NoVictims;
+
+    impl CachePolicy for NoVictims {
+        fn find_memory_victim(&self, _cnt: usize) -> Vec<EntryID> {
+            Vec::new()
+        }
+    }
+
+    /// Overwriting an entry that sits in the squeezed tier must take the
+    /// entry out of the index along with the object it reads through, even
+    /// when the new value then fails to insert: a squeezed entry left over a
+    /// deleted object would panic on its next read.
+    #[tokio::test]
+    async fn overwrite_of_squeezed_entry_that_fails_to_insert_leaves_no_entry() {
+        let dates: ArrayRef = Arc::new(Date32Array::from(vec![
+            Some(2),
+            Some(365 + 1),
+            None,
+            Some(365 + 100),
+        ]));
+        let expr = Arc::new(CacheExpression::extract_date32(Date32Field::Year));
+        let squeeze_to_tier = |cache: Arc<LiquidCache>, id| {
+            let dates = dates.clone();
+            let expr = expr.clone();
+            async move {
+                cache
+                    .insert(id, dates)
+                    .with_squeeze_hint(expr)
+                    .await
+                    .unwrap();
+                cache.squeeze_victims(vec![id]).await.unwrap();
+                cache.squeeze_victims(vec![id]).await.unwrap();
+                assert!(matches!(
+                    cache.index().get(&id).unwrap().as_ref(),
+                    CacheEntry::MemorySqueezedLiquid(_)
+                ));
+            }
+        };
+        // Learn the backing size, then size the disk tier to exactly it.
+        let probe = hydrating_cache().await;
+        squeeze_to_tier(probe.clone(), EntryID::from(1usize)).await;
+        let backing_bytes = probe.budget().disk_usage_bytes();
+        assert!(backing_bytes > 0);
+
+        let cache = LiquidCacheBuilder::new()
+            .with_max_memory_bytes(64 * 1024)
+            .with_max_disk_bytes(backing_bytes)
+            .with_squeeze_policy(Box::new(TranscodeSqueezeEvict))
+            .with_cache_policy(Box::new(NoVictims))
+            .build()
+            .await;
+        let id = EntryID::from(924usize);
+        squeeze_to_tier(cache.clone(), id).await;
+        assert_eq!(cache.budget().disk_usage_bytes(), backing_bytes);
+
+        // Too big for memory, and the disk tier is full with no victims.
+        let too_big: ArrayRef = Arc::new(Int32Array::from_iter_values(0..(1 << 16)));
+        let result = cache.insert(id, too_big).await;
+        assert_eq!(result, Err(CacheFull));
+
+        assert!(!cache.is_cached(&id));
+        assert!(cache.get(&id).await.is_none());
+        assert_eq!(cache.budget().disk_usage_bytes(), 0);
+        assert_eq!(cache.budget().memory_usage_bytes(), 0);
+    }
+
+    /// A flush that cannot write an entry drops it; if that entry still held
+    /// a disk copy, the object and its reservation must go with it, or the
+    /// disk tier shrinks by that much for good.
+    #[tokio::test]
+    async fn flush_dropping_hydrated_entry_releases_its_disk_copy() {
+        let array: ArrayRef = Arc::new(Int32Array::from_iter_values(0..64));
+        let disk_bytes = arrow_to_bytes(&array).unwrap().len();
+        let cache = LiquidCacheBuilder::new()
+            .with_max_memory_bytes(1 << 20)
+            .with_max_disk_bytes(disk_bytes)
+            .with_squeeze_policy(Box::new(TranscodeSqueezeEvict))
+            .with_cache_policy(Box::new(LiquidPolicy::new()))
+            .with_hydration_policy(Box::new(AlwaysHydrate::new()))
+            .build()
+            .await;
+        let id = EntryID::from(925usize);
+
+        cache.insert(id, array.clone()).await.unwrap();
+        cache.flush_all_to_disk().await.unwrap();
+        let read = cache.get(&id).await.expect("present");
+        assert_eq!(read.as_ref(), array.as_ref());
+        assert!(matches!(
+            cache.index().get(&id).unwrap().as_ref(),
+            CacheEntry::MemoryArrow(_)
+        ));
+        assert_eq!(cache.budget().disk_usage_bytes(), disk_bytes);
+
+        // The second flush wants to write the arrow bytes again into a tier
+        // that is full with the entry's own copy, so the entry is dropped.
+        cache.flush_all_to_disk().await.unwrap();
+
+        assert!(!cache.is_cached(&id));
+        assert_eq!(
+            cache.budget().disk_usage_bytes(),
+            0,
+            "the dropped entry's disk copy must be released"
+        );
+    }
 }