fix(cache): serve entries only to their own file

.github/workflows/ci.yml

@@ -139,6 +139,13 @@ jobs:
               run: |
                   cd src/core
                   cargo test --features "shuttle" --release -- --test-threads=1 shuttle
+            # The parquet cache keeps its own concurrent state — the file id
+            # pool — in this package, and it has its own `shuttle` feature.
+            # Without this step the model checker never sees it.
+            - name: Run shuttle test (datafusion)
+              run: |
+                  cd src/datafusion
+                  cargo test --features "shuttle" --release -- --test-threads=1 shuttle
 
     address_san:
         name: Address Sanitizer

benchmark/src/inprocess_runner.rs

@@ -657,7 +657,7 @@ impl InProcessBenchmarkRunner {
                 && let Some(cache) = &cache
             {
                 unsafe {
-                    cache.reset();
+                    cache.reset().await;
                 }
             }
 

examples/core.rs

@@ -18,13 +18,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await;
 
     let entry_id = EntryID::from(7);
+    // Names whose data this is. Entry ids are packed and can alias between
+    // sources; the cache compares this and treats a mismatch as a miss, so a
+    // caller only ever reads back what it put in.
+    let identity = 1;
     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

@@ -22,12 +22,16 @@ tokio_test::block_on(async {
 let storage = LiquidCacheBuilder::new().build().await;
 
 let entry_id = EntryID::from(42);
+// Names whose data this is. Entry ids are packed and can alias between
+// sources, so the cache compares this on every read and treats a mismatch as
+// a miss — a caller only ever reads back what it put in.
+let identity = 1;
 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;
+storage.insert(entry_id, identity, arrow_array.clone()).await;
 
-assert!(storage.is_cached(&entry_id));
+assert!(storage.is_cached(&entry_id, identity));
 });
 ```
 
@@ -42,14 +46,15 @@ tokio_test::block_on(async {
 let storage = LiquidCacheBuilder::new().build().await;
 
 let entry_id = EntryID::from(7);
+let identity = 1;
 let arrow_array = Arc::new(UInt64Array::from_iter_values(0..16));
-storage.insert(entry_id, arrow_array.clone()).await;
+storage.insert(entry_id, identity, 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, identity).await.unwrap();
 assert_eq!(retrieved.as_ref(), arrow_array.as_ref());
 });
 ```
@@ -71,10 +76,11 @@ tokio_test::block_on(async {
 let storage = LiquidCacheBuilder::new().build().await;
 
 let entry_id = EntryID::from(8);
+let identity = 1;
 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, identity, data.clone()).await;
 
 // Move data to disk so the read will demonstrate async I/O
 storage.flush_all_to_disk().await;
@@ -94,7 +100,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, identity, &liquid_expr)
     .with_selection(&selection)
     .await
     .unwrap();

src/core/src/cache/builders.rs

@@ -11,6 +11,7 @@ use super::core::LiquidCache;
 use super::io_context::{DefaultCacheMetadata, EntryMetadata};
 use super::policies::{CachePolicy, HydrationPolicy, SqueezePolicy, TranscodeSqueezeEvict};
 use super::{CacheExpression, CacheFull, EntryID, LiquidExpr, LiquidPolicy};
+use crate::cache::index::WriteIdentity;
 use crate::sync::Arc;
 
 /// Builder for [LiquidCache].
@@ -175,16 +176,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) squeeze_hint: 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,
             squeeze_hint: None,
@@ -214,7 +222,9 @@ impl<'a> Insert<'a> {
         }
         let batch = CacheEntry::memory_arrow(batch);
         self.storage.supersede_disk_copy(self.entry_id).await;
-        self.storage.insert_inner(self.entry_id, batch).await
+        self.storage
+            .insert_inner(self.entry_id, WriteIdentity::Owned(self.identity), batch)
+            .await
     }
 }
 
@@ -232,15 +242,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,
         }
@@ -273,6 +285,7 @@ impl<'a> Get<'a> {
         self.storage
             .read_arrow_array(
                 self.entry_id,
+                self.identity,
                 self.selection,
                 self.expression_hint.as_deref(),
             )
@@ -328,6 +341,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>,
 }
@@ -336,11 +350,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,
         }
@@ -355,7 +371,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
     }
 }
@@ -453,9 +469,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.