Sync upstream (#524–#527): adopt upstream's file-id fix

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");

benchmark/src/lib.rs

@@ -8,7 +8,7 @@ use fastrace::Span;
 use fastrace::future::FutureExt as _;
 use liquid_cache::cache::CacheStats;
 use liquid_cache::cache::{Evict, EvictionPolicy, TranscodeEvict};
-use liquid_cache_common::rpc::ExecutionMetricsResponse;
+use liquid_cache_common::metrics::ExecutionMetricsResponse;
 use liquid_cache_datafusion_server::{ApiResponse, ExecutionStats};
 use log::info;
 use serde::Serialize;

examples/core.rs

@@ -18,10 +18,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await;
 
     let entry_id = EntryID::from(7);
-    // The identity names which source this data came from. A cache key packs
-    // its fields into fixed widths, so two sources can compute one key; the
-    // identity is what keeps a read from being served the other's data. One
-    // source here, so any constant will do.
+    // 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

src/common/Cargo.toml

@@ -7,18 +7,26 @@ readme = "README.md"
 description = { workspace = true }
 repository = { workspace = true }
 
+[features]
+# Arrow Flight RPC message types used between the client and the server.
+rpc = ["dep:arrow-flight", "dep:prost"]
+
 [dependencies]
-arrow-flight = { workspace = true }
+arrow-flight = { workspace = true, optional = true }
 async-trait = { workspace = true }
 bytes = { workspace = true }
 chrono = "0.4.45"
 futures = { workspace = true }
 object_store = { workspace = true }
-prost = { workspace = true }
+prost = { workspace = true, optional = true }
 serde = { workspace = true }
 thiserror = "2.0.18"
 tokio = { workspace = true }
 url = { workspace = true }
 
 [dev-dependencies]
 tempfile = { workspace = true }
+tokio = { workspace = true, features = ["macros"] }
+
+[package.metadata.docs.rs]
+features = ["rpc"]

src/common/README.md

@@ -3,3 +3,5 @@
 Common types and utilities shared across LiquidCache components.
 
 This crate contains shared data structures, enums, and utility functions used throughout the LiquidCache ecosystem. It defines the core abstractions for cache modes, schema transformations, and RPC message types that enable communication between client and server components.
+
+The RPC message types (`rpc` module) are behind the `rpc` feature, which is off by default. The client and server crates enable it; in-process use through `liquid-cache-datafusion-local` does not need it, and so does not depend on Arrow Flight or gRPC.

src/common/src/lib.rs

@@ -1,5 +1,7 @@
 #![doc = include_str!("../README.md")]
 
+pub mod metrics;
 pub mod mock_store;
+#[cfg(feature = "rpc")]
 pub mod rpc;
 pub mod utils;

src/common/src/metrics.rs

@@ -0,0 +1,18 @@
+//! Execution metrics reported for a query plan.
+
+#[derive(Clone, PartialEq, serde::Serialize, serde::Deserialize, Debug)]
+pub struct ExecutionMetricsResponse {
+    pub pushdown_eval_time: u64,
+    pub cache_memory_usage: u64,
+    pub liquid_cache_usage: u64,
+}
+
+impl ExecutionMetricsResponse {
+    pub fn zero() -> Self {
+        Self {
+            pushdown_eval_time: 0,
+            cache_memory_usage: 0,
+            liquid_cache_usage: 0,
+        }
+    }
+}

src/common/src/rpc.rs

@@ -246,20 +246,3 @@ impl ProstMessageExt for FetchResults {
         }
     }
 }
-
-#[derive(Clone, PartialEq, serde::Serialize, serde::Deserialize, Debug)]
-pub struct ExecutionMetricsResponse {
-    pub pushdown_eval_time: u64,
-    pub cache_memory_usage: u64,
-    pub liquid_cache_usage: u64,
-}
-
-impl ExecutionMetricsResponse {
-    pub fn zero() -> Self {
-        Self {
-            pushdown_eval_time: 0,
-            cache_memory_usage: 0,
-            liquid_cache_usage: 0,
-        }
-    }
-}

src/core/README.md

@@ -24,14 +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
-// `0` is the file identity: the unnarrowed name of the source this data
-// belongs to. Cache keys pack their fields into fixed widths, so two sources
-// can compute one key; the identity is what keeps a read from being served
-// the other's data. One source here, so any constant will do.
+// 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.is_cached(&entry_id, 0));
+assert!(storage.contains(&entry_id, 0));
 });
 ```
 

src/core/src/cache/builders.rs

@@ -175,8 +175,6 @@ pub fn default_max_memory_bytes() -> usize {
 pub struct Insert<'a> {
     pub(super) storage: &'a Arc<LiquidCache>,
     pub(super) entry_id: EntryID,
-    /// The unnarrowed name of the source this data belongs to, recorded with
-    /// the entry so a key collision cannot serve it to anyone else.
     pub(super) identity: u64,
     pub(super) batch: ArrayRef,
     pub(super) skip_gc: bool,
@@ -223,11 +221,7 @@ impl<'a> Insert<'a> {
         }
         let batch = CacheEntry::memory_arrow(batch);
         self.storage
-            .insert_inner(
-                self.entry_id,
-                crate::cache::index::WriteIdentity::Owned(self.identity),
-                batch,
-            )
+            .insert_inner(self.entry_id, self.identity, batch)
             .await
     }
 }