diff --git a/crates/lance-csr-python/src/graph.rs b/crates/lance-csr-python/src/graph.rs index 70fe26b..d5b4a8a 100644 --- a/crates/lance-csr-python/src/graph.rs +++ b/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)?; } diff --git a/crates/lance-csr/src/csr_expand.rs b/crates/lance-csr/src/csr_expand.rs index 85b2cf0..500995e 100644 --- a/crates/lance-csr/src/csr_expand.rs +++ b/crates/lance-csr/src/csr_expand.rs @@ -129,6 +129,8 @@ pub struct Adjacency { removed: FastMap>, added_count: u64, removed_count: u64, + /// Tombstones that remove appended edges rather than base ones. + removed_added_count: u64, } struct CsrBase { @@ -181,6 +183,7 @@ impl Adjacency { removed: FastMap::default(), added_count: 0, removed_count: 0, + removed_added_count: 0, }) } @@ -199,6 +202,7 @@ impl Adjacency { removed: self.removed.clone(), added_count: self.added_count + added.len() as u64, removed_count: self.removed_count, + removed_added_count: self.removed_added_count, }; if !removed.is_empty() { // A row doesn't say which source holds it, so every live slot is checked. @@ -229,6 +233,8 @@ 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; let tombstones = next.removed.entry(key).or_default(); tombstones.extend(positions); tombstones.sort_unstable(); @@ -328,6 +334,11 @@ impl Adjacency { pub fn tombstones(&self) -> u64 { self.removed_count } + + /// Live edges outside the base: appended and not since removed. + pub fn unindexed_edges(&self) -> u64 { + self.added_count - self.removed_added_count + } } // Summary only: DataFusion formats extension nodes with `{:?}` during physical @@ -2513,9 +2524,11 @@ mod tests { (next.num_edges(), next.delta_edges(), next.tombstones()), (4, 2, 1) ); + assert_eq!(next.unindexed_edges(), 2); // Removing an appended edge cancels it; the original is untouched. let later = next.with_changes(&[], &rows([4])); assert!(neighbors(&later, 9).is_empty()); + assert_eq!((later.tombstones(), later.unindexed_edges()), (2, 1)); assert_eq!(neighbors(&base, 1), vec![2, 2, 3]); assert!(Arc::ptr_eq(&base.base, &later.base)); } @@ -2532,6 +2545,8 @@ mod tests { let next = base .with_changes(&[(1, 12, 2), (1, 12, 3), (1, 13, 4)], &rows([])) .with_changes(&[], &rows([1, 2])); + // Row 1 is a base edge, row 2 an appended one. + assert_eq!((next.tombstones(), next.unindexed_edges()), (2, 2)); for (mut offset, expected) in [ (0, vec![10, 12, 13]), (1, vec![12, 13]), diff --git a/crates/lance-csr/src/engine.rs b/crates/lance-csr/src/engine.rs index 4fb8c20..c3c9ec8 100644 --- a/crates/lance-csr/src/engine.rs +++ b/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, + /// Relationship dataset version of the last full build (`None` for Arrow input). + pub base_version: Option, /// Wall time of that build or delta, both directions. pub build_ms: f64, } @@ -264,6 +269,8 @@ struct CsrEntry { incoming: Arc, /// Rows of `source`, for relationship columns the adjacencies don't hold. edges: Arc, + /// Version of the snapshot the base was fully built from. + base_version: Option, /// 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) diff --git a/docs/csr.md b/docs/csr.md index 9dc6b6b..4784792 100644 --- a/docs/csr.md +++ b/docs/csr.md @@ -234,8 +234,9 @@ 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 | @@ -243,8 +244,10 @@ print(engine.csr_stats()["KNOWS"]) | `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" diff --git a/python/python/lance_csr/_internal/graph.pyi b/python/python/lance_csr/_internal/graph.pyi index 5244151..2fe3f1d 100644 --- a/python/python/lance_csr/_internal/graph.pyi +++ b/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 diff --git a/python/python/lance_csr/_types.py b/python/python/lance_csr/_types.py index ded83d9..34a9c5a 100644 --- a/python/python/lance_csr/_types.py +++ b/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 diff --git a/python/python/lance_csr/cypher.py b/python/python/lance_csr/cypher.py index 6a8c0bf..0d512fc 100644 --- a/python/python/lance_csr/cypher.py +++ b/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() diff --git a/python/python/tests/test_engine_csr_delta.py b/python/python/tests/test_engine_csr_delta.py index cd5b54f..2962622 100644 --- a/python/python/tests/test_engine_csr_delta.py +++ b/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)