perf(csr): read target columns once when a hop fetches much of a table

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

@@ -1136,8 +1136,10 @@ impl CypherEngine {
             item.set_item("edges", entry.edges)?;
             item.set_item("delta_edges", entry.delta_edges)?;
             item.set_item("tombstones", entry.tombstones)?;
+            item.set_item("unindexed_edges", entry.unindexed_edges)?;
             item.set_item("update", entry.update)?;
             item.set_item("version", entry.version)?;
+            item.set_item("base_version", entry.base_version)?;
             item.set_item("build_ms", entry.build_ms)?;
             stats.set_item(rel_type, item)?;
         }

crates/lance-csr/src/engine.rs

@@ -168,11 +168,16 @@ pub struct CsrStats {
     pub delta_edges: u64,
     /// Removed edges held outside the base CSR since its last full build.
     pub tombstones: u64,
+    /// Live edges outside the base CSR: appended since its last full build and not
+    /// since removed. Lance's `num_unindexed_rows`, counted in edges.
+    pub unindexed_edges: u64,
     /// How the last refresh produced it: `build` (first), `rebuild` (full), `delta`
     /// (changes applied) or `reused` (no edge change).
     pub update: &'static str,
     /// Relationship dataset version it reflects (`None` for Arrow input).
     pub version: Option<u64>,
+    /// Relationship dataset version of the last full build (`None` for Arrow input).
+    pub base_version: Option<u64>,
     /// Wall time of that build or delta, both directions.
     pub build_ms: f64,
 }
@@ -264,6 +269,8 @@ struct CsrEntry {
     incoming: Arc<Adjacency>,
     /// Rows of `source`, for relationship columns the adjacencies don't hold.
     edges: Arc<EdgeRows>,
+    /// Version of the snapshot the base was fully built from.
+    base_version: Option<u64>,
     /// Wall time of the last build or delta application.
     build_ms: f64,
     /// How the last refresh produced this entry: build, rebuild, delta or reused.
@@ -342,6 +349,7 @@ async fn try_apply_csr_delta(
         outgoing: Arc::new(outgoing),
         incoming: Arc::new(incoming),
         edges: source.edge_rows(),
+        base_version: old.base_version,
         build_ms: elapsed(),
         update: "delta",
     })
@@ -797,6 +805,7 @@ async fn build_csr_entries(
                 outgoing: Arc::new(outgoing),
                 incoming: Arc::new(incoming),
                 edges: source.edge_rows(),
+                base_version: source.lance_version(),
                 build_ms: started.elapsed().as_secs_f64() * 1e3,
                 update: if previous_entry.is_some() {
                     "rebuild"
@@ -1057,8 +1066,10 @@ impl CypherEngine {
                     edges: entry.outgoing.num_edges(),
                     delta_edges: entry.outgoing.delta_edges(),
                     tombstones: entry.outgoing.tombstones(),
+                    unindexed_edges: entry.outgoing.unindexed_edges(),
                     update: entry.update,
                     version: entry.source.lance_version(),
+                    base_version: entry.base_version,
                     build_ms: entry.build_ms,
                 };
                 (rel_type.clone(), stats)

docs/csr.md

@@ -234,17 +234,20 @@ One entry per relationship type that has a CSR:
 
 ```python
 print(engine.csr_stats()["KNOWS"])
-# {'edges': 4, 'delta_edges': 0, 'tombstones': 0, 'update': 'build',
-#  'version': None, 'build_ms': 0.41}  (build_ms varies)
+# {'edges': 4, 'delta_edges': 0, 'tombstones': 0, 'unindexed_edges': 0,
+#  'update': 'build', 'version': None, 'base_version': None,
+#  'build_ms': 0.41}  (build_ms varies)
 ```
 
 | Field | Meaning |
 | --- | --- |
 | `edges` | Live indexed edges. Edges with a null endpoint are skipped. |
 | `delta_edges` | Appended edges held outside the base CSR since its last full build. |
 | `tombstones` | Removed edges masked out since the last full build. |
+| `unindexed_edges` | Live edges outside the base CSR: appended since its last full build and not since removed. Lance index statistics call this `num_unindexed_rows`. |
 | `update` | How the last build or `refresh()` produced it: `build` (first), `rebuild` (full), `delta` (changes applied) or `reused` (no edge change). |
 | `version` | The relationship dataset's Lance version, or `None` for PyArrow input. |
+| `base_version` | The version of the last full build, or `None` for PyArrow input. It trails `version` while deltas apply. |
 | `build_ms` | Wall time of that build or delta, both directions. |
 
 A relationship type missing from `csr_stats()` has no CSR (its endpoints aren't integers) and always joins.
@@ -295,7 +298,8 @@ The engine doesn't watch your datasets for changes. Keeping the CSR in step with
    after all of them commit (see [snapshots are per dataset](lance.md#lazy-version-pinned-reads)). A refresh
    with nothing new reuses the existing CSR, so calling it on a timer or before each batch of queries is fine.
 3. **Check what it did with [`csr_stats()`](#csr_stats)**: `update` says whether the CSR was `reused`, updated
-   with a `delta` or fully rebuilt (`rebuild`).
+   with a `delta` or fully rebuilt (`rebuild`). `base_version` and `unindexed_edges` say how far the base
+   CSR trails the data. Answers are exact either way: delta edges are part of every lookup.
 
 === "Python"
 

python/python/lance_csr/_internal/graph.pyi

@@ -33,8 +33,10 @@ class _CsrStats(TypedDict):
     edges: int
     delta_edges: int
     tombstones: int
+    unindexed_edges: int
     update: Literal["build", "rebuild", "delta", "reused"]
     version: int | None
+    base_version: int | None
     build_ms: float
 
 @final

python/python/lance_csr/_types.py

@@ -53,8 +53,10 @@ class CsrStats(TypedDict):
     edges: int
     delta_edges: int
     tombstones: int
+    unindexed_edges: int
     update: Literal["build", "rebuild", "delta", "reused"]
     version: int | None
+    base_version: int | None
     build_ms: float
 
 

python/python/lance_csr/cypher.py

@@ -287,10 +287,14 @@ def csr_stats(self) -> dict[str, CsrStats]:
             - `edges`: live indexed edges (null endpoints are skipped).
             - `delta_edges`, `tombstones`: edges appended and removed since the last
               full build, held outside the base CSR.
+            - `unindexed_edges`: live edges outside the base CSR, appended since its
+              last full build and not since removed.
             - `update`: how the last refresh produced it: `"build"` (first), `"rebuild"`
               (full), `"delta"` (changes applied) or `"reused"` (no edge change).
             - `version`: the relationship dataset version it reflects (`None` for
               PyArrow).
+            - `base_version`: the relationship dataset version of the last full
+              build (`None` for PyArrow). It trails `version` while deltas apply.
             - `build_ms`: wall time of that build or delta, both directions.
         """
         return self._inner.csr_stats()

python/python/tests/test_engine_csr.py

@@ -112,6 +112,36 @@ def datasets(request, tmp_path):
     }
 
 
+def test_target_columns_read_once_over_many_batches(tmp_path):
+    # Fetches more target rows than one batch and an eighth of the node table, so the
+    # expansion switches from per-batch takes to one read of the target columns.
+    n = 3000
+    people = pa.table(
+        {
+            "id": list(range(n)),
+            "name": [None if i % 11 == 0 else f"p{i}" for i in range(n)],
+            "age": [i % 90 for i in range(n)],
+        }
+    )
+    knows = pa.table(
+        {
+            "src": pa.array([i % n for i in range(6 * n)], pa.int64()),
+            "dst": pa.array([(i * 7 + 3) % n for i in range(6 * n)], pa.int64()),
+        }
+    )
+    person = lance.write_dataset(people, tmp_path / "p.lance", max_rows_per_file=700)
+    person.delete("id % 13 = 0")
+    datasets = {"Person": lance.dataset(tmp_path / "p.lance"), "KNOWS": knows}
+    csr = CypherEngine(CONFIG, datasets, use_csr=True)
+    join = CypherEngine(CONFIG, datasets)
+    for query in [
+        "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.id, b.id, b.name, b.age",
+        "MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE b.age > 40 RETURN a.id, b.name",
+        "MATCH (a:Person)<-[:KNOWS]-(b:Person) RETURN count(DISTINCT b.name) AS n",
+    ]:
+        assert rows(csr.execute(query)) == rows(join.execute(query)), query
+
+
 @pytest.mark.parametrize("query", QUERIES)
 def test_csr_matches_join_path(datasets, query):
     csr = CypherEngine(CONFIG, datasets, use_csr=True)

python/python/tests/test_engine_csr_delta.py

@@ -89,6 +89,7 @@ def test_initial_build_reports_build(graph):
     engine, _ = graph()
     s = stats(engine)
     assert (s["update"], s["delta_edges"], s["tombstones"]) == ("build", 0, 0)
+    assert s["unindexed_edges"] == 0 and s["base_version"] == s["version"]
 
 
 def test_append_applied_as_delta(graph):
@@ -97,7 +98,8 @@ def test_append_applied_as_delta(graph):
     engine.refresh()
     s = stats(engine)
     assert (s["update"], s["delta_edges"], s["tombstones"]) == ("delta", 3, 0)
-    assert s["edges"] == 203
+    assert s["edges"] == 203 and s["unindexed_edges"] == 3
+    assert s["base_version"] < s["version"]
     check(engine, uri)
 
 
@@ -159,6 +161,8 @@ def test_sequential_deltas_accumulate(graph):
     engine.refresh()
     s = stats(engine)
     assert s["update"] == "delta" and s["delta_edges"] == 3 and s["tombstones"] == 2
+    # One tombstone removes an appended edge, so two of three appends are still live.
+    assert s["unindexed_edges"] == 2
     check(engine, uri)
 
 
@@ -217,6 +221,7 @@ def test_large_delta_triggers_rebuild(graph):
     engine.refresh()
     s = stats(engine)
     assert (s["update"], s["edges"], s["delta_edges"]) == ("rebuild", 300, 0)
+    assert s["unindexed_edges"] == 0 and s["base_version"] == s["version"]
     check(engine, uri)