Scan-resistant eviction + footprint-based cache admission gate

dev/dev-tools/src/components/cache_state_view.rs

@@ -50,7 +50,7 @@ pub fn CacheStateView(simulator: Signal<CacheSimulator>) -> Element {
 
     // Create a unified sorted list of all entry IDs (both actual and failed)
     let mut all_entry_ids: Vec<u64> = entries.iter().map(|e| e.entry_id).collect();
-    for (entry_id, _) in state.failed_inserts.iter() {
+    for entry_id in state.failed_inserts.keys() {
         if !all_entry_ids.contains(entry_id) {
             all_entry_ids.push(*entry_id);
         }

src/core/src/cache/policies/cache/three_queue.rs

@@ -1,4 +1,7 @@
-use std::{collections::HashMap, ptr::NonNull};
+use std::{
+    collections::{HashMap, HashSet},
+    ptr::NonNull,
+};
 
 use crate::{
     cache::{CachePolicy, EntryID, cached_batch::CachedBatchType},
@@ -19,6 +22,12 @@ enum QueueKind {
 struct QueueNode {
     entry_id: EntryID,
     queue: QueueKind,
+    /// Second-chance (CLOCK) reference bit. Set when the entry is accessed
+    /// (a cache hit); cleared when it is passed over during victim search.
+    /// A fresh insert starts cold (`false`), so a batch read exactly once —
+    /// e.g. every batch of a large one-pass scan — is evicted before any
+    /// batch that has been reused, keeping the working set scan-resistant.
+    referenced: bool,
 }
 
 type NodePtr = NonNull<DoublyLinkedNode<QueueNode>>;
@@ -68,6 +77,7 @@ impl LiquidQueueInternalState {
         let node = DoublyLinkedNode::new(QueueNode {
             entry_id,
             queue: target,
+            referenced: false,
         });
         let node_ptr = NonNull::from(Box::leak(node));
 
@@ -107,6 +117,46 @@ impl LiquidQueueInternalState {
         }
         Some(removed)
     }
+
+    /// Entry id and reference bit of the queue's front node, if any.
+    fn head_info(&self, queue: QueueKind) -> Option<(EntryID, bool)> {
+        let list = match queue {
+            QueueKind::Arrow => &self.arrow,
+            QueueKind::Liquid => &self.liquid,
+            QueueKind::Squeezed => &self.squeezed,
+            QueueKind::Disk => &self.disk,
+        };
+        let head_ptr = list.head()?;
+        let data = unsafe { &head_ptr.as_ref().data };
+        Some((data.entry_id, data.referenced))
+    }
+
+    /// Set the reference bit of an entry, if present.
+    fn set_referenced(&mut self, entry_id: &EntryID, value: bool) {
+        if let Some(node_ptr) = self.map.get(entry_id).copied() {
+            unsafe {
+                (*node_ptr.as_ptr()).data.referenced = value;
+            }
+        }
+    }
+
+    /// Give the queue's front node a second chance: clear its reference bit and
+    /// move it to the back. Caller has already confirmed the queue is non-empty.
+    fn second_chance_head(&mut self, queue: QueueKind) {
+        let head_ptr = match queue {
+            QueueKind::Arrow => self.arrow.head(),
+            QueueKind::Liquid => self.liquid.head(),
+            QueueKind::Squeezed => self.squeezed.head(),
+            QueueKind::Disk => self.disk.head(),
+        };
+        if let Some(head_ptr) = head_ptr {
+            unsafe {
+                (*head_ptr.as_ptr()).data.referenced = false;
+                self.detach(head_ptr);
+                self.push_back(queue, head_ptr);
+            }
+        }
+    }
 }
 
 impl Drop for LiquidQueueInternalState {
@@ -173,23 +223,29 @@ impl CachePolicy for LiquidPolicy {
         let mut inner = self.inner.lock().unwrap();
         let mut victims = Vec::with_capacity(cnt);
 
-        while victims.len() < cnt {
-            if let Some(entry) = inner.pop_front(QueueKind::Arrow) {
-                victims.push(entry);
-                continue;
-            }
-
-            if let Some(entry) = inner.pop_front(QueueKind::Liquid) {
-                victims.push(entry);
-                continue;
+        // Evict decoded (Arrow) first, then Liquid, then Squeezed — cheapest to
+        // reconstruct first. Within each queue apply CLOCK/second-chance: a
+        // referenced entry is passed over once (bit cleared, moved to the back)
+        // before it can be evicted, so a reused batch outlives a read-once one.
+        // `chanced` bounds each entry to a single second chance, so the loop
+        // always terminates (at most one pass of reprieves, then eviction).
+        for queue in [QueueKind::Arrow, QueueKind::Liquid, QueueKind::Squeezed] {
+            let mut chanced: HashSet<EntryID> = HashSet::new();
+            while victims.len() < cnt {
+                let Some((entry_id, referenced)) = inner.head_info(queue) else {
+                    break;
+                };
+                if referenced && chanced.insert(entry_id) {
+                    inner.second_chance_head(queue);
+                } else {
+                    let popped = inner.pop_front(queue);
+                    debug_assert_eq!(popped, Some(entry_id));
+                    victims.push(entry_id);
+                }
             }
-
-            if let Some(entry) = inner.pop_front(QueueKind::Squeezed) {
-                victims.push(entry);
-                continue;
+            if victims.len() >= cnt {
+                break;
             }
-
-            break;
         }
 
         victims
@@ -213,7 +269,24 @@ impl CachePolicy for LiquidPolicy {
         victims
     }
 
-    fn notify_access(&self, _entry_id: &EntryID, _batch_type: CachedBatchType) {}
+    fn notify_access(&self, entry_id: &EntryID, batch_type: CachedBatchType) {
+        // Only a memory-resident hit counts as reuse for CLOCK. A disk hit must
+        // not set the bit: the disk read path calls `notify_access` before
+        // hydration, and the subsequent Disk->Memory `notify_insert` preserves
+        // the bit — so a batch read once from disk would arrive in memory
+        // looking hot and could evict a genuinely reused memory batch during a
+        // large one-pass scan.
+        let is_memory = matches!(
+            batch_type,
+            CachedBatchType::MemoryArrow
+                | CachedBatchType::MemoryLiquid
+                | CachedBatchType::MemorySqueezedLiquid
+        );
+        if is_memory {
+            let mut inner = self.inner.lock().unwrap();
+            inner.set_referenced(entry_id, true);
+        }
+    }
 
     fn notify_remove(&self, entry_id: &EntryID) {
         let mut inner = self.inner.lock().unwrap();
@@ -249,6 +322,66 @@ mod tests {
         assert_eq!(policy.find_memory_victim(1), vec![liquid_b]);
     }
 
+    #[test]
+    fn test_accessed_entry_gets_second_chance() {
+        let policy = LiquidPolicy::new();
+        let a = entry(1);
+        let b = entry(2);
+        let c = entry(3);
+        policy.notify_insert(&a, CachedBatchType::MemoryArrow);
+        policy.notify_insert(&b, CachedBatchType::MemoryArrow);
+        policy.notify_insert(&c, CachedBatchType::MemoryArrow);
+
+        // `a` is reused, so it should be spared on the next victim search even
+        // though it was inserted first (plain FIFO would evict it).
+        policy.notify_access(&a, CachedBatchType::MemoryArrow);
+
+        assert_eq!(policy.find_memory_victim(1), vec![b]);
+        assert_eq!(policy.find_memory_victim(1), vec![c]);
+        // `a` survived the longest and is evicted last (its bit was cleared when
+        // it was passed over, so it gets no further reprieve).
+        assert_eq!(policy.find_memory_victim(1), vec![a]);
+    }
+
+    #[test]
+    fn test_disk_hit_does_not_protect_only_memory_hit_does() {
+        // A hit served from disk must NOT protect the entry — otherwise a
+        // one-pass disk-backed scan arrives hot on hydration and evicts the
+        // real working set. Only a memory hit grants a second chance.
+        let policy = LiquidPolicy::new();
+        let a = entry(1);
+        let b = entry(2);
+        let c = entry(3);
+        for id in [&a, &b, &c] {
+            policy.notify_insert(id, CachedBatchType::MemoryArrow);
+        }
+        policy.notify_access(&a, CachedBatchType::DiskArrow); // must not protect
+        policy.notify_access(&b, CachedBatchType::MemoryArrow); // protects
+
+        // `a` (disk hit) and `c` (untouched) evicted; `b` (memory hit) survives.
+        assert_eq!(policy.find_memory_victim(2), vec![a, c]);
+        assert_eq!(policy.find_memory_victim(1), vec![b]);
+    }
+
+    #[test]
+    fn test_read_once_entries_evicted_before_reused() {
+        // Simulates a large one-pass scan (read-once batches) alongside a single
+        // reused batch: the scan batches must all be evicted before the reused
+        // one, so the scan cannot displace the working set.
+        let policy = LiquidPolicy::new();
+        let ids: Vec<_> = (1..=5).map(entry).collect();
+        for id in &ids {
+            policy.notify_insert(id, CachedBatchType::MemoryArrow);
+        }
+        // ids[2] (entry 3) is reused.
+        policy.notify_access(&ids[2], CachedBatchType::MemoryArrow);
+
+        let victims = policy.find_memory_victim(4);
+        assert_eq!(victims, vec![ids[0], ids[1], ids[3], ids[4]]);
+        // The reused entry is the sole survivor.
+        assert_eq!(policy.find_memory_victim(4), vec![ids[2]]);
+    }
+
     #[test]
     fn test_queue_priority_order() {
         let policy = LiquidPolicy::new();

src/datafusion-local/src/lib.rs

@@ -69,8 +69,10 @@ pub struct LiquidCacheLocalBuilder {
     squeeze_policy: Box<dyn SqueezePolicy>,
     /// Hydration policy
     hydration_policy: Box<dyn HydrationPolicy>,
-    /// Maximum total file size for a scan to be routed through LiquidCache.
-    max_scan_bytes: Option<u64>,
+    /// Footprint-based admission gate `(expansion, safety, tolerance, strict)`.
+    /// When set, a scan is cached only if its estimated liquid footprint stays
+    /// within `budget × tolerance`; `strict` toggles fail-loud panic handling.
+    admission: Option<(f64, f64, f64, bool)>,
     span: fastrace::Span,
 }
 
@@ -86,7 +88,7 @@ impl Default for LiquidCacheLocalBuilder {
             cache_policy: Box::new(LiquidPolicy::new()),
             squeeze_policy: Box::new(TranscodeSqueezeEvict),
             hydration_policy: Box::new(AlwaysHydrate::new()),
-            max_scan_bytes: None,
+            admission: None,
             span: fastrace::Span::enter_with_local_parent("liquid_cache_datafusion_local_builder"),
         }
     }
@@ -148,11 +150,22 @@ impl LiquidCacheLocalBuilder {
         self
     }
 
-    /// Set the maximum total file size (in bytes) for a scan to be routed
-    /// through LiquidCache. Scans exceeding this threshold are read directly
-    /// from the parquet source, bypassing the cache entirely.
-    pub fn with_max_scan_bytes(mut self, max_bytes: u64) -> Self {
-        self.max_scan_bytes = Some(max_bytes);
+    /// Enable the footprint-based admission gate. A scan is cached only when its
+    /// estimated liquid footprint (raw required bytes x `expansion` x `safety`)
+    /// stays within `budget × tolerance`; larger scans are read directly from the
+    /// parquet source, bypassing the cache. `expansion`/`safety` are `>= 1.0`
+    /// (inflate the estimate); `tolerance` is `>= 1.0` (overcommit the budget,
+    /// clamped to the measured ~5x compaction crossover). `strict == true` lets a
+    /// footprint-estimation panic abort the query (fail loud); `false` catches it
+    /// and caches the scan normally.
+    pub fn with_admission_gate(
+        mut self,
+        expansion: f64,
+        safety: f64,
+        tolerance: f64,
+        strict: bool,
+    ) -> Self {
+        self.admission = Some((expansion, safety, tolerance, strict));
         self
     }
 
@@ -202,8 +215,8 @@ impl LiquidCacheLocalBuilder {
         let cache_ref = Arc::new(cache);
 
         let mut optimizer = LocalModeOptimizer::new(cache_ref.clone());
-        if let Some(max_bytes) = self.max_scan_bytes {
-            optimizer = optimizer.with_max_scan_bytes(max_bytes);
+        if let Some((expansion, safety, tolerance, strict)) = self.admission {
+            optimizer = optimizer.with_admission_gate(expansion, safety, tolerance, strict);
         }
 
         let state = datafusion::execution::SessionStateBuilder::new()

src/datafusion-local/src/tests/snapshots/liquid_cache_datafusion_local__tests__squeeze__squeeze_strings.snap

@@ -22,15 +22,15 @@ event=hydrate entry=851968 cached=DiskLiquid new=MemoryLiquid
 event=insert_success entry=851968 kind=MemoryLiquid
 event=insert_success entry=851969 kind=MemoryArrow
 event=insert_failed entry=917505 kind=MemoryArrow
-event=squeeze_begin victims=[851969,917504,851968]
+event=squeeze_begin victims=[851969,851968,917504]
 event=squeeze_victim entry=851969
 event=insert_success entry=851969 kind=MemoryLiquid
-event=squeeze_victim entry=917504
-event=io_write entry=917504 kind=MemorySqueezedLiquid bytes=136440
-event=insert_success entry=917504 kind=MemorySqueezedLiquid
 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
+event=insert_success entry=917504 kind=MemorySqueezedLiquid
 event=insert_success entry=917505 kind=MemoryArrow
 event=eval_predicate entry=917505 selection=true cached=MemoryArrow
 event=read entry=851969 selection=true expr=None cached=MemoryLiquid