feat(csr)!: store the csr inside each lance dataset and pack edge rows

crates/lance-csr-python/src/graph.rs

@@ -26,7 +26,7 @@ use datafusion::execution::context::SessionContext;
 use futures::StreamExt;
 use lance_csr::{
     ast::{DistanceMetric as RustDistanceMetric, GraphPattern, ReadingClause},
-    engine::{CypherEngine as RustCypherEngine, EngineSource, MAX_TARGET_PARTITIONS},
+    engine::{CypherEngine as RustCypherEngine, EngineOptions, EngineSource, MAX_TARGET_PARTITIONS},
     CypherQuery as RustCypherQuery, ExecutionStrategy as RustExecutionStrategy,
     GraphConfig as RustGraphConfig, GraphError as RustGraphError, SqlDialect as RustSqlDialect,
     SqlQuery as RustSqlQuery, VectorSearch as RustVectorSearch,
@@ -1015,14 +1015,18 @@ pub struct CypherEngine {
 #[pymethods]
 impl CypherEngine {
     #[new]
-    #[pyo3(signature = (config, datasets, use_csr = false, *, target_partitions = None))]
+    #[pyo3(signature = (config, datasets, use_csr = false, *, target_partitions = None, persist_csr = false))]
     fn new(
         py: Python,
         config: &GraphConfig,
         datasets: &Bound<'_, PyDict>,
         use_csr: bool,
         target_partitions: Option<&Bound<'_, PyAny>>,
+        persist_csr: bool,
     ) -> PyResult<Self> {
+        if persist_csr && !use_csr {
+            return Err(PyValueError::new_err("persist_csr requires use_csr=True"));
+        }
         let target_partitions = target_partitions
             .map(|value| {
                 if value.is_instance_of::<PyBool>() || !value.is_instance_of::<PyInt>() {
@@ -1051,11 +1055,14 @@ impl CypherEngine {
         let inner = RT
             .block_on(
                 Some(py),
-                RustCypherEngine::with_target_partitions(
+                RustCypherEngine::with_options(
                     config.inner.clone(),
                     sources,
-                    use_csr,
-                    target_partitions,
+                    EngineOptions {
+                        use_csr,
+                        target_partitions,
+                        persist_csr,
+                    },
                 ),
             )?
             .map_err(graph_error_to_pyerr)?;
@@ -1112,16 +1119,26 @@ impl CypherEngine {
             .map_err(graph_error_to_pyerr)
     }
 
-    fn refresh(&self, py: Python) -> PyResult<()> {
-        RT.block_on(Some(py), self.inner.refresh())?
+    fn checkout_latest(&self, py: Python) -> PyResult<()> {
+        RT.block_on(Some(py), self.inner.checkout_latest())?
             .map_err(graph_error_to_pyerr)
     }
 
+    /// Deprecated alias; the Python wrapper warns before calling it.
+    fn refresh(&self, py: Python) -> PyResult<()> {
+        self.checkout_latest(py)
+    }
+
     fn optimize_csr(&self, py: Python) -> PyResult<()> {
         RT.block_on(Some(py), self.inner.optimize_csr())?
             .map_err(graph_error_to_pyerr)
     }
 
+    fn delete_stored_csr(&self, py: Python) -> PyResult<usize> {
+        RT.block_on(Some(py), self.inner.delete_stored_csr())?
+            .map_err(graph_error_to_pyerr)
+    }
+
     fn node_index_stats(&self, py: Python) -> PyResult<Py<PyAny>> {
         let stats = PyDict::new(py);
         for (label, entry) in self.inner.node_index_stats() {

crates/lance-csr/Cargo.toml

@@ -13,7 +13,7 @@ categories = ["database", "data-structures", "science"]
 
 [dependencies]
 ahash = "0.8"
-arrow = { version = "58", features = ["prettyprint"] }
+arrow = { version = "58", features = ["prettyprint", "ipc_compression"] }
 arrow-array = "58"
 arrow-schema = "58"
 async-trait = "0.1"
@@ -38,7 +38,9 @@ lance = "13.0.0"
 lance-linalg = "13.0.0"
 lance-namespace = "13.0.0"
 lance-table = "13.0.0"
+log = "0.4"
 nom = "7.1"
+object_store = "0.14"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 snafu = "0.8"

crates/lance-csr/README.md

@@ -15,7 +15,7 @@ crate.
 - Semantic validation with typed `GraphError` diagnostics
 - Pluggable execution strategies (DataFusion planner by default, Lance Native placeholder)
 - `CypherEngine`: register datasets once, optionally build the CSR, and serve many queries
-- Lance datasets read lazily and pinned to a version until `refresh()`
+- Lance datasets read lazily and pinned to a version until `checkout_latest()`
 - Async query execution that returns Arrow `RecordBatch` results or a stream
 - JSON-serializable parameter binding for reusable query templates
 - Logical plan debugging via `CypherQuery::explain`
@@ -41,7 +41,7 @@ let batch = engine
 ```
 
 `examples/embedded_query.rs` is a complete program that writes two Lance datasets, queries them, and picks up a
-new write with `refresh()`:
+new write with `checkout_latest()`:
 
 ```bash
 cargo run -p lance-csr --example embedded_query
@@ -154,7 +154,7 @@ clauses, patterns, expressions and functions the engine runs.
 - `logical_plan` – Builders for graph logical plans.
 - `datafusion_planner` – DataFusion-based execution planning.
 - `config` – Graph configuration types and builders.
-- `engine` – `CypherEngine`: cached snapshot, CSR adjacencies, `refresh()`.
+- `engine` – `CypherEngine`: cached snapshot, CSR adjacencies, `checkout_latest()`.
 - `query` – High level `CypherQuery` API and runtime.
 - `csr_index`, `csr_expand` – The CSR adjacency index and the traversal operators that use it.
 - `error` – `GraphError` and result helpers.

crates/lance-csr/examples/embedded_query.rs

@@ -2,7 +2,7 @@
 //!
 //! Writes a small social graph as two Lance datasets in a temporary directory, opens
 //! them in a CSR-backed engine, runs a two-hop query, appends an edge and picks it up
-//! with `refresh()`, runs a query under a deadline, and repeats the in-memory Arrow
+//! with `checkout_latest()`, runs a query under a deadline, and repeats the in-memory Arrow
 //! examples of the home page and the getting started guide. The documentation includes
 //! this file section by section (the `--8<--` markers), so keep it compiling and its printed
 //! output in step.
@@ -80,7 +80,7 @@ async fn main() -> lance_csr::Result<()> {
         .node_table("Person", "id")
         .rel_table("KNOWS", "Person", "Person", "src_id", "dst_id")
         .build()?;
-    // Each dataset is pinned to the version opened here until `engine.refresh()`.
+    // Each dataset is pinned to the version opened here until `engine.checkout_latest()`.
     let sources = HashMap::from([
         (
             "Person".to_string(),
@@ -108,10 +108,10 @@ async fn main() -> lance_csr::Result<()> {
     // +------------------+
     // --8<-- [end:query]
 
-    // --8<-- [start:refresh]
-    // Bob now knows Alice too. The engine doesn't see the write until refresh().
+    // --8<-- [start:checkout_latest]
+    // Bob now knows Alice too. The engine doesn't see the write until checkout_latest().
     write_lance(knows(vec![2], vec![1])?, &knows_uri, WriteMode::Append).await?;
-    engine.refresh().await?;
+    engine.checkout_latest().await?;
     let result = engine.execute(query, params()).await?;
     print_batches(&[result])?; // Alice, Carol, David
 
@@ -121,7 +121,7 @@ async fn main() -> lance_csr::Result<()> {
         stats.edges, stats.update, stats.version
     );
     // 5 edges, update=delta, version=Some(2)
-    // --8<-- [end:refresh]
+    // --8<-- [end:checkout_latest]
 
     // --8<-- [start:timeout]
     // One deadline for both: checks inside nested-loop joins, and a timer that stops

crates/lance-csr/src/csr_expand.rs

@@ -100,9 +100,10 @@ use crate::error::{GraphError, Result};
 /// sorted offsets into each source's base-plus-appended neighbours. Offsets let
 /// readers skip deleted occurrences without copying per-source deletion state.
 /// Identical edges remain distinct occurrences. Each neighbour carries the row its
-/// edge came from (a Lance row address, or a row index for in-memory input), so the
-/// base holds 16 bytes per edge: relationship columns can be fetched by row, and
-/// deleted edges are found by row without reading the old dataset version.
+/// edge came from (a Lance row address, or a row index for in-memory input), so
+/// relationship columns can be fetched by row, and deleted edges are found by row
+/// without reading the old dataset version. Edge rows are packed (see
+/// [`crate::csr_index::PackedU64`]), so the base holds about 10 bytes per edge, not 16.
 ///
 /// # Design note for future changes (read before "optimizing" the delta)
 ///
@@ -220,14 +221,16 @@ impl Adjacency {
                 .map(|(k, v)| (k, Some(v)))
                 .chain(added_only.map(|&k| (k, None)))
             {
-                let base_rows = vertex.map_or(&[][..], |v| self.base.csr.edge_rows(v));
+                let base_range = vertex.map_or(0..0, |v| self.base.csr.range(v));
+                let base_len = base_range.len();
                 let added_rows = self.added.get(&key).map_or(&[][..], Vec::as_slice);
                 let prior = self.removed.get(&key).map_or(&[][..], Vec::as_slice);
                 let mut prior = prior.iter().copied().peekable();
                 let mut positions = Vec::new();
-                let rows = base_rows
-                    .iter()
-                    .copied()
+                let rows = self
+                    .base
+                    .csr
+                    .edge_rows_at(base_range)
                     .chain(added_rows.iter().map(|&(_, r)| r));
                 for (position, row) in rows.enumerate() {
                     if prior.peek() == Some(&position) {
@@ -239,7 +242,7 @@ impl Adjacency {
                 if !positions.is_empty() {
                     next.removed_count += positions.len() as u64;
                     next.removed_added_count +=
-                        positions.iter().filter(|&&p| p >= base_rows.len()).count() as u64;
+                        positions.iter().filter(|&&p| p >= base_len).count() as u64;
                     let tombstones = next.removed.entry(key).or_default();
                     tombstones.extend(positions);
                     tombstones.sort_unstable();
@@ -252,17 +255,47 @@ impl Adjacency {
         next
     }
 
+    /// An adjacency with no delta over a stored base: `keys` are the source keys in
+    /// CSR position order, as [`Self::base_parts`] returns them. `None` if a key
+    /// repeats.
+    pub(crate) fn from_base(keys: &[i64], csr: CsrIndex) -> Option<Self> {
+        let mut dense = FastMap::with_capacity_and_hasher(keys.len(), Default::default());
+        dense.extend(keys.iter().enumerate().map(|(i, &k)| (k, i as u64)));
+        if dense.len() != keys.len() {
+            return None;
+        }
+        Some(Self {
+            base: Arc::new(CsrBase { dense, csr }),
+            added: FastMap::default(),
+            removed: FastMap::default(),
+            added_count: 0,
+            removed_count: 0,
+            removed_added_count: 0,
+        })
+    }
+
+    /// The base CSR: its source keys in CSR position order, and the index over them.
+    /// The delta is not included.
+    pub(crate) fn base_parts(&self) -> (Vec<i64>, &CsrIndex) {
+        let mut keys = vec![0; self.base.dense.len()];
+        for (&key, &position) in &self.base.dense {
+            keys[position as usize] = key;
+        }
+        (keys, &self.base.csr)
+    }
+
     /// Call `f` for each target endpoint value reachable from `key`, duplicates included.
     pub fn for_each_neighbor(&self, key: i64, f: impl FnMut(u64)) {
         self.neighbors_from(key, &mut 0).for_each(f);
     }
 
-    /// Neighbours and their edge rows in the base.
+    /// Positions of `key`'s neighbours in the base.
     #[inline(always)]
-    fn base_slots(&self, key: i64) -> (&[u64], &[u64]) {
-        self.base.dense.get(&key).map_or((&[], &[]), |&v| {
-            (self.base.csr.neighbors(v), self.base.csr.edge_rows(v))
-        })
+    fn base_range(&self, key: i64) -> std::ops::Range<usize> {
+        self.base
+            .dense
+            .get(&key)
+            .map_or(0..0, |&v| self.base.csr.range(v))
     }
 
     /// Resume at an occurrence offset, borrowing all neighbour and deletion storage.
@@ -286,17 +319,27 @@ impl Adjacency {
         key: i64,
         offset: &'a mut usize,
     ) -> impl Iterator<Item = Option<(u64, u64)>> + 'a {
-        let (base, base_rows) = self.base_slots(key);
+        let base = self.base_range(key);
         let added = self.added.get(&key).map_or(&[][..], Vec::as_slice);
         let removed = self.removed.get(&key).map_or(&[][..], Vec::as_slice);
         let mut removed = &removed[removed.partition_point(|&position| position < *offset)..];
         let added_offset = offset.saturating_sub(base.len());
-        let base_start = (*offset).min(base.len());
-        let base_rows = &base_rows[base_start..];
-        base[base_start..]
+        let rest = base.start + (*offset).min(base.len())..base.end;
+        let csr = &self.base.csr;
+        // Never advanced without `ROWS`, so plain hops don't unpack edge rows.
+        let mut base_rows = csr.edge_rows_at(if ROWS { rest.clone() } else { 0..0 });
+        csr.neighbors_at(rest)
             .iter()
-            .enumerate()
-            .map(move |(i, &n)| (n, if ROWS { base_rows[i] } else { 0 }))
+            .map(move |&n| {
+                (
+                    n,
+                    if ROWS {
+                        base_rows.next().unwrap_or(0)
+                    } else {
+                        0
+                    },
+                )
+            })
             .chain(
                 added
                     .get(added_offset..)
@@ -461,15 +504,42 @@ pub async fn edge_changes(
     src: &str,
     dst: &str,
 ) -> Result<Option<EdgeChanges>> {
-    let endpoint_ids = |ds: &Dataset| -> Option<[(i32, DataType); 2]> {
-        let field = |name: &str| ds.schema().field(name).map(|f| (f.id, f.data_type()));
-        Some([field(src)?, field(dst)?])
+    let Some(old_fields) = endpoint_fields(old, src, dst) else {
+        return Ok(None);
+    };
+    let old_frags: Vec<Fragment> = old
+        .get_fragments()
+        .into_iter()
+        .map(|f| f.metadata().clone())
+        .collect();
+    edge_changes_since(&old_frags, &old_fields, new, src, dst).await
+}
+
+/// Lance field id and type of a relationship table's two endpoint columns.
+pub(crate) type EndpointFields = [(i32, String); 2];
+
+pub(crate) fn endpoint_fields(ds: &Dataset, src: &str, dst: &str) -> Option<EndpointFields> {
+    let field = |name: &str| {
+        ds.schema()
+            .field(name)
+            .map(|f| (f.id, f.data_type().to_string()))
     };
-    let old_ids = endpoint_ids(old);
-    if old_ids.is_none() || old_ids != endpoint_ids(new) {
+    Some([field(src)?, field(dst)?])
+}
+
+/// [`edge_changes`] from a version known only by its fragments and endpoint fields,
+/// as a stored CSR records them.
+pub(crate) async fn edge_changes_since(
+    old_fragments: &[Fragment],
+    old_fields: &EndpointFields,
+    new: &Dataset,
+    src: &str,
+    dst: &str,
+) -> Result<Option<EdgeChanges>> {
+    if endpoint_fields(new, src, dst).as_ref() != Some(old_fields) {
         return Ok(None);
     }
-    let fields: Vec<i32> = old_ids.unwrap().iter().map(|(id, _)| *id).collect();
+    let fields: Vec<i32> = old_fields.iter().map(|(id, _)| *id).collect();
     let touches = |f: &[i32]| f.iter().any(|id| fields.contains(id));
     // The parts of a fragment that determine its endpoint values.
     let endpoint_data = |frag: &Fragment| {
@@ -482,11 +552,8 @@ pub async fn edge_changes(
         format!("{:?}{:?}", files, overlays)
     };
 
-    let old_frags: HashMap<u64, Fragment> = old
-        .get_fragments()
-        .into_iter()
-        .map(|f| (f.id() as u64, f.metadata().clone()))
-        .collect();
+    let old_frags: HashMap<u64, Fragment> =
+        old_fragments.iter().map(|f| (f.id, f.clone())).collect();
     let new_frags: HashMap<u64, Fragment> = new
         .get_fragments()
         .into_iter()