fix(cache): serve cached entries only to the file they came from

README.md

@@ -40,18 +40,21 @@ use std::sync::Arc;
 tokio_test::block_on(async {
     let cache = LiquidCacheBuilder::new().build().await;
     let entry_id = EntryID::from(1);
+    // The identity names the source the data belongs to; reads under any
+    // other identity miss. One source here, so any constant will do.
+    let identity = 0;
     let values = Arc::new(UInt64Array::from(vec![10, 11, 12, 13, 14, 15]));
 
     // 1) insert
-    cache.insert(entry_id, values.clone()).await;
+    cache.insert(entry_id, identity, values.clone()).await;
 
     // 2) get
-    let all_rows = cache.get(&entry_id).await.expect("entry should exist");
+    let all_rows = cache.get(&entry_id, identity).await.expect("entry should exist");
 
     // 3) get filtered (selection pushdown): keep rows 0, 2, 4
     let selection = BooleanBuffer::from(vec![true, false, true, false, true, false]);
     let selected_rows = cache
-        .get(&entry_id)
+        .get(&entry_id, identity)
         .with_selection(&selection)
         .await
         .expect("entry should exist");
@@ -63,7 +66,7 @@ tokio_test::block_on(async {
         Arc::new(Literal::new(ScalarValue::UInt64(Some(12)))),
     ));
     let predicate_mask = cache
-        .eval_predicate(&entry_id, &predicate)
+        .eval_predicate(&entry_id, identity, &predicate)
         .await
         .expect("entry should exist")
         .expect("predicate should be evaluated in cache");

examples/core.rs

@@ -18,13 +18,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await;
 
     let entry_id = EntryID::from(7);
+    // The identity names the source this data belongs to. One source here,
+    // so any constant will do.
+    let identity = 0;
     let arrow_array = Arc::new(UInt64Array::from_iter_values(0..16));
-    storage.insert(entry_id, arrow_array.clone()).await.unwrap();
+    storage
+        .insert(entry_id, identity, arrow_array.clone())
+        .await
+        .unwrap();
 
     // Move data to disk so the read demonstrates async I/O
     storage.flush_all_to_disk().await.unwrap();
 
-    let retrieved = storage.get(&entry_id).await.unwrap();
+    let retrieved = storage.get(&entry_id, identity).await.unwrap();
     assert_eq!(retrieved.as_ref(), arrow_array.as_ref());
 
     Ok(())

src/core/README.md

@@ -24,10 +24,12 @@ let storage = LiquidCacheBuilder::new().build().await;
 let entry_id = EntryID::from(42);
 let arrow_array = Arc::new(UInt64Array::from_iter_values(0..1000));
 
-// Insert once; replacement/placement is handled by the cache policy
-storage.insert(entry_id, arrow_array.clone()).await;
+// Insert once; replacement/placement is handled by the cache policy.
+// `0` is the identity: the source this data belongs to. Reads under any
+// other identity miss. One source here, so any constant will do.
+storage.insert(entry_id, 0, arrow_array.clone()).await;
 
-assert!(storage.contains(&entry_id));
+assert!(storage.contains(&entry_id, 0));
 });
 ```
 
@@ -43,13 +45,13 @@ let storage = LiquidCacheBuilder::new().build().await;
 
 let entry_id = EntryID::from(7);
 let arrow_array = Arc::new(UInt64Array::from_iter_values(0..16));
-storage.insert(entry_id, arrow_array.clone()).await;
+storage.insert(entry_id, 0, arrow_array.clone()).await;
 
 // Move data to disk so the read will demonstrate async I/O
 storage.flush_all_to_disk().await;
 
 // Read asynchronously
-let retrieved = storage.get(&entry_id).await.unwrap();
+let retrieved = storage.get(&entry_id, 0).await.unwrap();
 assert_eq!(retrieved.as_ref(), arrow_array.as_ref());
 });
 ```
@@ -74,7 +76,7 @@ let entry_id = EntryID::from(8);
 let data = Arc::new(StringArray::from(vec![
     Some("apple"), Some("banana"), None, Some("apple"), Some("cherry"),
 ]));
-storage.insert(entry_id, data.clone()).await;
+storage.insert(entry_id, 0, data.clone()).await;
 
 // Move data to disk so the read will demonstrate async I/O
 storage.flush_all_to_disk().await;
@@ -93,7 +95,7 @@ let liquid_expr = liquid_cache::cache::LiquidExpr::try_new(
 
 // Read with predicate pushdown
 let mask = storage
-    .eval_predicate(&entry_id, &liquid_expr)
+    .eval_predicate(&entry_id, 0, &liquid_expr)
     .with_selection(&selection)
     .await
     .unwrap();

src/core/src/cache/builders.rs

@@ -175,16 +175,23 @@ pub fn default_max_memory_bytes() -> usize {
 pub struct Insert<'a> {
     pub(super) storage: &'a Arc<LiquidCache>,
     pub(super) entry_id: EntryID,
+    pub(super) identity: u64,
     pub(super) batch: ArrayRef,
     pub(super) skip_gc: bool,
     pub(super) lineage: Option<Arc<CacheExpression>>,
 }
 
 impl<'a> Insert<'a> {
-    pub(super) fn new(storage: &'a Arc<LiquidCache>, entry_id: EntryID, batch: ArrayRef) -> Self {
+    pub(super) fn new(
+        storage: &'a Arc<LiquidCache>,
+        entry_id: EntryID,
+        identity: u64,
+        batch: ArrayRef,
+    ) -> Self {
         Self {
             storage,
             entry_id,
+            identity,
             batch,
             skip_gc: false,
             lineage: None,
@@ -213,7 +220,9 @@ impl<'a> Insert<'a> {
             self.storage.add_lineage(&self.entry_id, lineage);
         }
         let batch = CacheEntry::memory_arrow(batch);
-        self.storage.insert_inner(self.entry_id, batch).await
+        self.storage
+            .insert_inner(self.entry_id, self.identity, batch)
+            .await
     }
 }
 
@@ -231,15 +240,17 @@ impl<'a> IntoFuture for Insert<'a> {
 pub struct Get<'a> {
     pub(super) storage: &'a LiquidCache,
     pub(super) entry_id: &'a EntryID,
+    pub(super) identity: u64,
     pub(super) selection: Option<&'a BooleanBuffer>,
     pub(super) expression_hint: Option<Arc<CacheExpression>>,
 }
 
 impl<'a> Get<'a> {
-    pub(super) fn new(storage: &'a LiquidCache, entry_id: &'a EntryID) -> Self {
+    pub(super) fn new(storage: &'a LiquidCache, entry_id: &'a EntryID, identity: u64) -> Self {
         Self {
             storage,
             entry_id,
+            identity,
             selection: None,
             expression_hint: None,
         }
@@ -271,6 +282,7 @@ impl<'a> Get<'a> {
         self.storage
             .read_arrow_array(
                 self.entry_id,
+                self.identity,
                 self.selection,
                 self.expression_hint.as_deref(),
             )
@@ -326,6 +338,7 @@ fn maybe_gc_view_arrays(array: &ArrayRef) -> Option<ArrayRef> {
 pub struct EvaluatePredicate<'a> {
     pub(super) storage: &'a LiquidCache,
     pub(super) entry_id: &'a EntryID,
+    pub(super) identity: u64,
     pub(super) predicate: &'a LiquidExpr,
     pub(super) selection: Option<&'a BooleanBuffer>,
 }
@@ -334,11 +347,13 @@ impl<'a> EvaluatePredicate<'a> {
     pub(super) fn new(
         storage: &'a LiquidCache,
         entry_id: &'a EntryID,
+        identity: u64,
         predicate: &'a LiquidExpr,
     ) -> Self {
         Self {
             storage,
             entry_id,
+            identity,
             predicate,
             selection: None,
         }
@@ -353,7 +368,7 @@ impl<'a> EvaluatePredicate<'a> {
     /// Evaluate the predicate against the cached data.
     pub async fn read(self) -> Option<BooleanArray> {
         self.storage
-            .eval_predicate_internal(self.entry_id, self.selection, self.predicate)
+            .eval_predicate_internal(self.entry_id, self.identity, self.selection, self.predicate)
             .await
     }
 }
@@ -451,9 +466,9 @@ mod tests {
 
         let cache = LiquidCacheBuilder::new().build().await;
         let entry_id = EntryID::from(123usize);
-        cache.insert(entry_id, root.clone()).await.unwrap();
+        cache.insert(entry_id, 0, root.clone()).await.unwrap();
 
-        let stored = cache.get(&entry_id).await.expect("array present");
+        let stored = cache.get(&entry_id, 0).await.expect("array present");
         let post_size = stored.get_array_memory_size();
 
         // GC should have compacted the view arrays, reducing memory footprint.