Add disk-bounded cache with eviction and source fallback

src/core/src/cache/budget.rs

@@ -3,14 +3,22 @@ use crate::sync::atomic::{AtomicUsize, Ordering};
 #[derive(Debug)]
 pub struct BudgetAccounting {
     max_memory_bytes: usize,
+    max_disk_bytes: usize,
+    watermark_ratio: f64,
     used_memory_bytes: AtomicUsize,
     used_disk_bytes: AtomicUsize,
 }
 
 impl BudgetAccounting {
-    pub(super) fn new(max_memory_bytes: usize) -> Self {
+    pub(super) fn new(
+        max_memory_bytes: usize,
+        max_disk_bytes: usize,
+        watermark_ratio: f64,
+    ) -> Self {
         Self {
             max_memory_bytes,
+            max_disk_bytes,
+            watermark_ratio,
             used_memory_bytes: AtomicUsize::new(0),
             used_disk_bytes: AtomicUsize::new(0),
         }
@@ -69,6 +77,20 @@ impl BudgetAccounting {
     pub fn add_used_disk_bytes(&self, bytes: usize) {
         self.used_disk_bytes.fetch_add(bytes, Ordering::Relaxed);
     }
+
+    pub fn sub_used_disk_bytes(&self, bytes: usize) {
+        self.used_disk_bytes.fetch_sub(bytes, Ordering::Relaxed);
+    }
+
+    pub fn disk_budget_exceeded(&self) -> bool {
+        self.max_disk_bytes != usize::MAX
+            && self.used_disk_bytes.load(Ordering::Relaxed)
+                > (self.max_disk_bytes as f64 * self.watermark_ratio) as usize
+    }
+
+    pub fn max_disk_bytes(&self) -> usize {
+        self.max_disk_bytes
+    }
 }
 
 #[cfg(test)]
@@ -78,7 +100,7 @@ mod tests {
 
     #[test]
     fn test_memory_reservation_and_accounting() {
-        let config = BudgetAccounting::new(1000);
+        let config = BudgetAccounting::new(1000, usize::MAX, 0.9);
 
         assert_eq!(config.memory_usage_bytes(), 0);
 
@@ -95,6 +117,34 @@ mod tests {
         assert_eq!(config.memory_usage_bytes(), 0);
     }
 
+    #[test]
+    fn test_disk_budget_tracking() {
+        let budget = BudgetAccounting::new(1000, 500, 0.9);
+
+        assert_eq!(budget.disk_usage_bytes(), 0);
+        assert!(!budget.disk_budget_exceeded());
+
+        budget.add_used_disk_bytes(400);
+        assert_eq!(budget.disk_usage_bytes(), 400);
+        assert!(!budget.disk_budget_exceeded());
+
+        budget.add_used_disk_bytes(200);
+        assert_eq!(budget.disk_usage_bytes(), 600);
+        assert!(budget.disk_budget_exceeded());
+
+        budget.sub_used_disk_bytes(300);
+        assert_eq!(budget.disk_usage_bytes(), 300);
+        assert!(!budget.disk_budget_exceeded());
+    }
+
+    #[test]
+    fn test_disk_budget_unlimited() {
+        let budget = BudgetAccounting::new(1000, usize::MAX, 0.9);
+
+        budget.add_used_disk_bytes(usize::MAX / 2);
+        assert!(!budget.disk_budget_exceeded());
+    }
+
     #[test]
     fn test_concurrent_memory_operations() {
         test_concurrent_memory_budget();
@@ -111,7 +161,7 @@ mod tests {
         let max_memory = 10000;
         let operations_per_thread = 100;
 
-        let budget = Arc::new(BudgetAccounting::new(max_memory));
+        let budget = Arc::new(BudgetAccounting::new(max_memory, usize::MAX, 0.9));
         let barrier = Arc::new(Barrier::new(num_threads));
 
         let mut thread_handles = vec![];

src/core/src/cache/builders.rs

@@ -32,6 +32,8 @@ use crate::sync::Arc;
 pub struct LiquidCacheBuilder {
     batch_size: usize,
     max_memory_bytes: usize,
+    max_disk_bytes: usize,
+    disk_watermark: f64,
     cache_policy: Box<dyn CachePolicy>,
     hydration_policy: Box<dyn HydrationPolicy>,
     squeeze_policy: Box<dyn SqueezePolicy>,
@@ -52,6 +54,8 @@ impl LiquidCacheBuilder {
         Self {
             batch_size: 8192,
             max_memory_bytes: 1024 * 1024 * 1024,
+            max_disk_bytes: usize::MAX,
+            disk_watermark: 0.9,
             cache_policy: Box::new(LiquidPolicy::new()),
             hydration_policy: Box::new(super::AlwaysHydrate::new()),
             squeeze_policy: Box::new(TranscodeSqueezeEvict),
@@ -75,6 +79,20 @@ impl LiquidCacheBuilder {
         self
     }
 
+    /// Set the max disk bytes for the cache.
+    /// Default is unlimited — the cache will use available disk space without a cap.
+    pub fn with_max_disk_bytes(mut self, max_disk_bytes: usize) -> Self {
+        self.max_disk_bytes = max_disk_bytes;
+        self
+    }
+
+    /// Set the disk watermark ratio (0.0–1.0). Default is 0.9.
+    /// Eviction triggers when disk usage exceeds this fraction of max_disk_bytes.
+    pub fn with_disk_watermark(mut self, ratio: f64) -> Self {
+        self.disk_watermark = ratio;
+        self
+    }
+
     /// Set the cache policy for the cache.
     /// Default is [LiquidPolicy].
     pub fn with_cache_policy(mut self, policy: Box<dyn CachePolicy>) -> Self {
@@ -137,6 +155,8 @@ impl LiquidCacheBuilder {
         Arc::new(LiquidCache::new(
             self.batch_size,
             self.max_memory_bytes,
+            self.max_disk_bytes,
+            self.disk_watermark,
             self.squeeze_policy,
             self.cache_policy,
             self.hydration_policy,

src/core/src/cache/core.rs

@@ -113,6 +113,7 @@ impl LiquidCache {
             memory_usage_bytes,
             disk_usage_bytes,
             max_memory_bytes: self.config.max_memory_bytes(),
+            max_disk_bytes: self.config.max_disk_bytes(),
             runtime,
         }
     }
@@ -338,17 +339,23 @@ impl LiquidCache {
     pub(crate) fn new(
         batch_size: usize,
         max_memory_bytes: usize,
+        max_disk_bytes: usize,
+        disk_watermark: f64,
         squeeze_policy: Box<dyn SqueezePolicy>,
         cache_policy: Box<dyn CachePolicy>,
         hydration_policy: Box<dyn HydrationPolicy>,
         metadata: Arc<dyn EntryMetadata>,
         store: t4::Store,
         squeeze_victims_concurrently: bool,
     ) -> Self {
-        let config = CacheConfig::new(batch_size, max_memory_bytes);
+        let config = CacheConfig::new(batch_size, max_memory_bytes, max_disk_bytes, disk_watermark);
         Self {
             index: ArtIndex::new(),
-            budget: BudgetAccounting::new(config.max_memory_bytes()),
+            budget: BudgetAccounting::new(
+                config.max_memory_bytes(),
+                config.max_disk_bytes(),
+                config.disk_watermark(),
+            ),
             config,
             cache_policy,
             hydration_policy,
@@ -693,6 +700,10 @@ impl LiquidCache {
     }
 
     async fn write_batch_to_disk(&self, entry_id: EntryID, batch: &CacheEntry, bytes: Bytes) {
+        if self.budget.disk_budget_exceeded() {
+            self.evict_disk_entries().await;
+        }
+
         self.trace(InternalEvent::IoWrite {
             entry: entry_id,
             kind: CachedBatchType::from(batch),
@@ -706,6 +717,29 @@ impl LiquidCache {
         self.budget.add_used_disk_bytes(len);
     }
 
+    async fn evict_disk_entries(&self) {
+        while self.budget.disk_budget_exceeded() {
+            let victims = self.cache_policy.find_disk_victims(8);
+            if victims.is_empty() {
+                break;
+            }
+            for victim in victims {
+                if !self.budget.disk_budget_exceeded() {
+                    break;
+                }
+                if let Some(_removed) = self.index.remove(&victim) {
+                    let key = crate::cache::io_context::entry_id_to_key(&victim);
+                    let _ = self.store.remove(&key).await;
+                    self.budget.sub_used_disk_bytes(
+                        self.budget
+                            .disk_usage_bytes()
+                            .min(self.config.batch_size() * 8),
+                    );
+                }
+            }
+        }
+    }
+
     async fn read_disk_arrow_array(&self, entry_id: &EntryID) -> ArrayRef {
         let bytes = self
             .store

src/core/src/cache/index.rs

@@ -80,6 +80,15 @@ impl ArtIndex {
     pub(crate) fn entry_count(&self) -> usize {
         self.entry_count.load(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
+    }
 }
 
 #[cfg(test)]
@@ -134,4 +143,26 @@ mod tests {
         let entry_id: EntryID = EntryID::from(1);
         assert!(!store.is_cached(&entry_id));
     }
+
+    #[test]
+    fn test_remove() {
+        let store = ArtIndex::new();
+        let entry_id = EntryID::from(1);
+        let entry_id2 = EntryID::from(2);
+        let array = create_test_array(50);
+
+        store.insert(&entry_id, array.clone());
+        store.insert(&entry_id2, array.clone());
+        assert_eq!(store.entry_count(), 2);
+
+        let removed = store.remove(&entry_id);
+        assert!(removed.is_some());
+        assert!(!store.is_cached(&entry_id));
+        assert!(store.is_cached(&entry_id2));
+        assert_eq!(store.entry_count(), 1);
+
+        let removed = store.remove(&EntryID::from(99));
+        assert!(removed.is_none());
+        assert_eq!(store.entry_count(), 1);
+    }
 }

src/core/src/cache/observer/stats.rs

@@ -146,6 +146,8 @@ pub struct CacheStats {
     pub disk_usage_bytes: usize,
     /// Maximum memory size.
     pub max_memory_bytes: usize,
+    /// Maximum disk size.
+    pub max_disk_bytes: usize,
     /// Runtime counters snapshot.
     pub runtime: RuntimeStatsSnapshot,
 }

src/core/src/cache/policies/cache/mod.rs

@@ -24,6 +24,12 @@ pub trait CachePolicy: std::fmt::Debug + Send + Sync {
     /// Give cnt amount of entries to evict when cache is full.
     fn find_victim(&self, cnt: usize) -> Vec<EntryID>;
 
+    /// Give cnt amount of disk entries to evict when disk is full.
+    /// Default returns empty — policies that track disk entries can override.
+    fn find_disk_victims(&self, _cnt: usize) -> Vec<EntryID> {
+        vec![]
+    }
+
     /// Notify the cache policy that an entry was inserted.
     fn notify_insert(&self, _entry_id: &EntryID, _batch_type: CachedBatchType) {}
 

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

@@ -185,6 +185,24 @@ impl CachePolicy for LiquidPolicy {
         victims
     }
 
+    fn find_disk_victims(&self, cnt: usize) -> Vec<EntryID> {
+        if cnt == 0 {
+            return vec![];
+        }
+
+        let mut inner = self.inner.lock().unwrap();
+        let mut victims = Vec::with_capacity(cnt);
+
+        while victims.len() < cnt {
+            match inner.pop_front(QueueKind::Disk) {
+                Some(entry) => victims.push(entry),
+                None => break,
+            }
+        }
+
+        victims
+    }
+
     fn notify_access(&self, _entry_id: &EntryID, _batch_type: CachedBatchType) {}
 }
 
@@ -289,4 +307,49 @@ mod tests {
         let victims = policy.find_victim(2);
         assert_eq!(victims, vec![entry_id]);
     }
+
+    #[test]
+    fn test_find_disk_victims_returns_disk_entries_fifo() {
+        let policy = LiquidPolicy::new();
+
+        let d1 = entry(10);
+        let d2 = entry(11);
+        let d3 = entry(12);
+
+        policy.notify_insert(&d1, CachedBatchType::DiskLiquid);
+        policy.notify_insert(&d2, CachedBatchType::DiskArrow);
+        policy.notify_insert(&d3, CachedBatchType::DiskLiquid);
+
+        let victims = policy.find_disk_victims(2);
+        assert_eq!(victims, vec![d1, d2]);
+
+        let victims = policy.find_disk_victims(5);
+        assert_eq!(victims, vec![d3]);
+
+        assert!(policy.find_disk_victims(1).is_empty());
+    }
+
+    #[test]
+    fn test_find_disk_victims_zero_returns_empty() {
+        let policy = LiquidPolicy::new();
+        policy.notify_insert(&entry(1), CachedBatchType::DiskLiquid);
+        assert!(policy.find_disk_victims(0).is_empty());
+    }
+
+    #[test]
+    fn test_find_disk_victims_does_not_affect_memory_eviction() {
+        let policy = LiquidPolicy::new();
+
+        let mem = entry(1);
+        let disk = entry(2);
+
+        policy.notify_insert(&mem, CachedBatchType::MemoryArrow);
+        policy.notify_insert(&disk, CachedBatchType::DiskLiquid);
+
+        let disk_victims = policy.find_disk_victims(5);
+        assert_eq!(disk_victims, vec![disk]);
+
+        let mem_victims = policy.find_victim(5);
+        assert_eq!(mem_victims, vec![mem]);
+    }
 }