From 23b0e7bf78a6f27916f00a63de0bb90d273e716e Mon Sep 17 00:00:00 2001 From: prrao87 Date: Mon, 2 Nov 2026 12:33:20 -0500 Subject: [PATCH 1/2] refactor(engine): rename refresh() to checkout_latest() to match lance Lance moves a dataset to its newest version with checkout_latest() and has no refresh(). CypherEngine.refresh() does the same for every dataset the engine reads, so it is now checkout_latest(), in Rust and Python, which also pairs it with optimize_csr() the way Lance pairs checkout_latest() with optimize_indices(). refresh() stays as a deprecated alias: Python warns with DeprecationWarning, and Rust marks it #[deprecated]. Docs, examples, tests and benchmarks use the new name, and the lance_snapshot_refresh.py example is now lance_snapshots.py. --- README.md | 2 +- crates/lance-csr-python/src/graph.rs | 9 +- crates/lance-csr/README.md | 6 +- crates/lance-csr/examples/embedded_query.rs | 12 +-- crates/lance-csr/src/engine.rs | 78 ++++++++------- crates/lance-csr/tests/test_engine.rs | 8 +- docs/AGENTS.md | 2 +- docs/benchmarks.md | 6 +- docs/csr.md | 54 +++++------ docs/cypher.md | 10 +- docs/execution.md | 8 +- docs/getting-started.md | 2 +- docs/index.md | 2 +- docs/internals/index.md | 2 +- docs/internals/indexes.md | 6 +- docs/internals/optimizer.md | 2 +- docs/lance.md | 48 ++++++---- docs/rust.md | 10 +- docs/sql.md | 2 +- examples/README.md | 6 +- ...snapshot_refresh.py => lance_snapshots.py} | 14 +-- python/README.md | 12 +-- python/benchmarks/csr_delta.py | 6 +- python/benchmarks/csr_delta_accumulated.py | 4 +- python/benchmarks/csr_expand.py | 4 +- python/benchmarks/engine_cache.py | 10 +- python/python/lance_csr/_internal/graph.pyi | 1 + python/python/lance_csr/cypher.py | 96 +++++++++++-------- python/python/tests/test_call_subqueries.py | 2 +- python/python/tests/test_engine_csr.py | 10 +- python/python/tests/test_engine_csr_delta.py | 38 ++++---- python/python/tests/test_engine_lance.py | 33 ++++--- python/python/tests/test_path_values.py | 4 +- python/python/tests/test_target_partitions.py | 2 +- python/python/tests/test_typing.py | 2 +- python/python/tests/typing/public_api.py | 2 +- 36 files changed, 282 insertions(+), 233 deletions(-) rename examples/{lance_snapshot_refresh.py => lance_snapshots.py} (84%) diff --git a/README.md b/README.md index 99db039..bd64b03 100644 --- a/README.md +++ b/README.md @@ -45,7 +45,7 @@ case needs distributed graph queries or graph algorithms, that's the kind of eng - **CSR traversal** (`use_csr=True`): relationship hops and variable-length paths become array lookups instead of joins, updated incrementally as the Lance datasets change. See [the CSR index](https://lance-csr.hdclabs.ai/csr/). - **A Lance-native engine** that reads each table lazily from a pinned Lance snapshot or holds it in memory, and - `refresh()`es to new versions. + `checkout_latest()`es to new versions. - **Much wider Cypher**, enough to run the Graph Data Council's Interactive complex queries: `OPTIONAL MATCH`, pattern predicates, chained `WITH`, `UNWIND`, list and pattern comprehensions, `shortestPath`, `$parameters` and more. diff --git a/crates/lance-csr-python/src/graph.rs b/crates/lance-csr-python/src/graph.rs index 0725dd3..d57be91 100644 --- a/crates/lance-csr-python/src/graph.rs +++ b/crates/lance-csr-python/src/graph.rs @@ -1119,11 +1119,16 @@ 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) diff --git a/crates/lance-csr/README.md b/crates/lance-csr/README.md index 1b4a9c4..167ddd8 100644 --- a/crates/lance-csr/README.md +++ b/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. diff --git a/crates/lance-csr/examples/embedded_query.rs b/crates/lance-csr/examples/embedded_query.rs index 71cf107..5e7ecff 100644 --- a/crates/lance-csr/examples/embedded_query.rs +++ b/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 diff --git a/crates/lance-csr/src/engine.rs b/crates/lance-csr/src/engine.rs index 5450bee..de01402 100644 --- a/crates/lance-csr/src/engine.rs +++ b/crates/lance-csr/src/engine.rs @@ -5,7 +5,7 @@ //! //! [`CypherEngine`] holds a snapshot of every input: the DataFusion context, the //! catalog, join-key statistics and, with `use_csr`, the CSR adjacencies and node -//! indexes. Queries share the snapshot read-only; [`CypherEngine::refresh`] builds a +//! indexes. Queries share the snapshot read-only; [`CypherEngine::checkout_latest`] builds a //! complete replacement and swaps it in. //! //! ```no_run @@ -63,8 +63,8 @@ use crate::{GraphSourceCatalog, InMemoryCatalog}; pub enum EngineSource { /// On-disk Lance dataset, read lazily through `LanceTableProvider`. /// - /// `dataset` is pinned to one version; only `refresh()` replaces it. `uri` and - /// `storage_options` are kept so refresh can reopen the same location with the + /// `dataset` is pinned to one version; only `checkout_latest()` replaces it. `uri` and + /// `storage_options` are kept so `checkout_latest()` can reopen the same location with the /// same credentials. Lance { uri: String, @@ -79,7 +79,7 @@ pub enum EngineSource { impl EngineSource { /// Open a Lance dataset at `uri`, pinned to `version` (`None` means the latest). /// - /// `storage_options` are kept so [`CypherEngine::refresh`] can reopen the same + /// `storage_options` are kept so [`CypherEngine::checkout_latest`] can reopen the same /// location with the same credentials. pub async fn open_lance( uri: impl Into, @@ -172,7 +172,7 @@ pub struct CsrStats { /// 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 or `optimize_csr()` produced it: `build` (first), `rebuild` + /// How the last build, `checkout_latest()` or `optimize_csr()` produced it: `build` (first), `rebuild` /// (full), `loaded` (from a stored file, with any changes since applied as a /// delta), `delta` (changes applied) or `reused` (no edge change). pub update: &'static str, @@ -197,11 +197,11 @@ pub struct NodeIndexStats { /// Everything a query needs, built once and shared read-only by concurrent queries. /// -/// `refresh()` builds a complete replacement and swaps it in; queries already running +/// `checkout_latest()` builds a complete replacement and swaps it in; queries already running /// keep the `Arc` they started with, so they finish on their original snapshot. struct EngineState { sources: HashMap, - /// Unoptimized plan templates for this snapshot; a refresh starts an empty cache. + /// Unoptimized plan templates for this snapshot; `checkout_latest()` starts an empty cache. plans: Mutex>, catalog: Arc, context: SessionContext, @@ -674,7 +674,7 @@ fn memtable_partitions(batch: &RecordBatch) -> Vec> { /// Node ids count as unique (the row count). Edge endpoints use the CSR's distinct /// source and target counts when the CSR is built, and otherwise one scan of the two /// endpoint columns; only integer endpoints are counted. Reused while a dataset is -/// the same snapshot, so an unchanged `refresh()` costs nothing. +/// the same snapshot, so an unchanged `checkout_latest()` costs nothing. // shortcut: the fallback scan reads every edge endpoint at construction and on each changed refresh (about one CSR build's scan); switch to an approximate counter (HyperLogLog) or Lance column statistics if that shows up in startup time. async fn build_key_statistics( config: &GraphConfig, @@ -900,12 +900,12 @@ async fn build_csr_entries( /// /// Lance datasets are read lazily from storage, pinned to the version they were /// opened at: writes made later (by this or another process) are not visible until -/// [`refresh`](Self::refresh). Each dataset is pinned separately; versions across +/// [`checkout_latest`](Self::checkout_latest). Each dataset is pinned separately; versions across /// tables are not a transactional multi-table snapshot. Arrow batches are held in /// memory and never change; build a new engine to replace them. /// /// Building the CSR (`use_csr`) is CPU-bound work inside an async function: in an -/// application with its own async runtime, construct and refresh the engine on a +/// application with its own async runtime, construct the engine and call `checkout_latest()` on a /// blocking thread (for example `tokio::task::spawn_blocking` plus `block_on`). /// Options for [`CypherEngine::with_options`]. #[derive(Debug, Clone, Default, PartialEq)] @@ -917,7 +917,7 @@ pub struct EngineOptions { /// Store each fully built CSR inside its Lance relationship dataset, at /// `_lance_csr/.arrow`, and load it instead of building when an engine /// starts. Changes since the stored version apply as a delta, under - /// the same 25% threshold as `refresh()`. Ignored without `use_csr` and for Arrow + /// the same 25% threshold as `checkout_latest()`. Ignored without `use_csr` and for Arrow /// input. A failed write is logged and doesn't fail the engine. pub persist_csr: bool, } @@ -926,8 +926,9 @@ pub struct CypherEngine { config: GraphConfig, state: RwLock>, options: EngineOptions, - /// Held for a whole refresh, so a slower refresh can't swap in older data over a newer one. - refresh_lock: futures::lock::Mutex<()>, + /// Held for a whole update of the snapshot, so a slower `checkout_latest()` can't swap in older + /// data over a newer one. + update_lock: futures::lock::Mutex<()>, } impl CypherEngine { @@ -953,7 +954,7 @@ impl CypherEngine { /// Like [`new`](Self::new), with a target for DataFusion query partitioning. /// /// `Some(n)` sets DataFusion's `target_partitions` for every query, kept across - /// [`refresh`](Self::refresh); `None` keeps DataFusion's default (the number of + /// [`checkout_latest`](Self::checkout_latest); `None` keeps DataFusion's default (the number of /// cores). It is not a thread or memory limit. In-memory Arrow inputs are still /// sliced into one partition per core. /// @@ -998,11 +999,11 @@ impl CypherEngine { config, state: RwLock::new(Arc::new(state)), options, - refresh_lock: Default::default(), + update_lock: Default::default(), }) } - /// The current snapshot. Cloning the `Arc` lets a query outlive a concurrent refresh. + /// The current snapshot. Cloning the `Arc` lets a query outlive a concurrent `checkout_latest()`. fn snapshot(&self) -> Arc { self.state .read() @@ -1087,7 +1088,7 @@ impl CypherEngine { /// /// Batches are computed as the stream is polled. Scans, filters and projections /// stream with bounded memory; joins, sorts and aggregations may still buffer. - /// The stream keeps what it reads alive, so a concurrent `refresh()` does not + /// The stream keeps what it reads alive, so a concurrent `checkout_latest()` does not /// affect it. Planning errors return here; execution errors arrive in the stream. /// Dropping the stream cancels the remaining work. pub async fn execute_stream( @@ -1123,15 +1124,16 @@ impl CypherEngine { vector_search.search(&candidates).await } - /// Move every Lance dataset to its latest version. + /// Move every Lance dataset to its latest version: the engine's counterpart of + /// Lance's `Dataset::checkout_latest()`. /// /// Datasets whose version and commit time are unchanged keep their open handle. /// The replacement is validated in full before it is swapped in; on error nothing /// changes and the previous snapshot stays usable. Queries already running finish /// on the snapshot they started with. Arrow inputs are unchanged. Concurrent calls /// run one at a time. - pub async fn refresh(&self) -> Result<()> { - let _refreshing = self.refresh_lock.lock().await; + pub async fn checkout_latest(&self) -> Result<()> { + let _updating = self.update_lock.lock().await; let current = self.snapshot(); let mut sources = current.sources.clone(); for source in sources.values_mut() { @@ -1153,16 +1155,22 @@ impl CypherEngine { Ok(()) } + /// Former name of [`checkout_latest`](Self::checkout_latest). + #[deprecated(note = "renamed to `checkout_latest`, after Lance's `Dataset::checkout_latest`")] + pub async fn refresh(&self) -> Result<()> { + self.checkout_latest().await + } + /// Rebuild each CSR that carries a delta, at the snapshot the engine is pinned to: the /// CSR's counterpart of Lance's `optimize_indices()`. /// - /// `refresh()` applies changes as a delta over the base CSR and rebuilds on its own + /// `checkout_latest()` applies changes as a delta over the base CSR and rebuilds on its own /// only once they pass 25% of the base. This rebuilds every CSR that carries a delta /// now, so lookups stop carrying it; one without a delta, and everything else, is /// reused. Results never change. The replacement is validated in full before it is - /// swapped in, as for `refresh()`, and the two run one at a time. + /// swapped in, as for `checkout_latest()`, and the two run one at a time. pub async fn optimize_csr(&self) -> Result<()> { - let _refreshing = self.refresh_lock.lock().await; + let _updating = self.update_lock.lock().await; let current = self.snapshot(); let state = build_engine_state( &self.config, @@ -1181,10 +1189,10 @@ impl CypherEngine { /// /// The engine keeps the CSR it has in memory, so queries are unaffected. An /// engine with `persist_csr` stores a new file at its next full build; create - /// engines without it to stay in memory only. Runs one at a time with `refresh()` + /// engines without it to stay in memory only. Runs one at a time with `checkout_latest()` /// and `optimize_csr()`. pub async fn delete_stored_csr(&self) -> Result { - let _refreshing = self.refresh_lock.lock().await; + let _updating = self.update_lock.lock().await; let current = self.snapshot(); let mut deleted = 0; for (_, mapping) in self.config.iter_relationship_mappings() { @@ -1201,7 +1209,7 @@ impl CypherEngine { /// CSR adjacency statistics per relationship type (experimental, `use_csr`). /// - /// An identical map after `refresh()` means the snapshot itself was unchanged. + /// An identical map after `checkout_latest()` means the snapshot itself was unchanged. pub fn csr_stats(&self) -> HashMap { self.snapshot() .csr @@ -1280,7 +1288,7 @@ mod tests { } #[tokio::test] - async fn target_partitions_survive_refresh_without_changing_other_defaults() { + async fn target_partitions_survive_checkout_latest_without_changing_other_defaults() { let defaults = SessionConfig::new(); for target_partitions in [None, Some(1), Some(2)] { let engine = CypherEngine::with_target_partitions( @@ -1292,7 +1300,7 @@ mod tests { .await .unwrap(); let before = engine.snapshot(); - engine.refresh().await.unwrap(); + engine.checkout_latest().await.unwrap(); for snapshot in [before, engine.snapshot()] { let state = snapshot.context.state(); assert_eq!( @@ -1318,7 +1326,7 @@ mod tests { } #[tokio::test] - async fn plan_cache_reuses_bounds_and_resets_on_refresh() { + async fn plan_cache_reuses_bounds_and_resets_on_checkout_latest() { let engine = CypherEngine::new(person_config(), people_engine_sources(), false) .await .unwrap(); @@ -1385,7 +1393,7 @@ mod tests { assert_eq!(first.num_rows(), 1); // A refresh publishes a new snapshot with an empty cache; the old one is untouched. - engine.refresh().await.unwrap(); + engine.checkout_latest().await.unwrap(); assert!(engine.snapshot().plans.lock().unwrap().is_empty()); assert_eq!(cached(), PLAN_CACHE_ENTRIES); } @@ -1434,10 +1442,10 @@ mod tests { .unwrap(); } - /// A refresh that starts while another is in flight waits for it, then publishes + /// A `checkout_latest()` that starts while another is in flight waits for it, then publishes /// the newest version: an older snapshot can never land after a newer one. #[tokio::test] - async fn refresh_waits_for_an_in_flight_refresh() { + async fn checkout_latest_waits_for_an_in_flight_checkout() { let dir = tempfile::tempdir().unwrap(); let uri = dir.path().join("person.lance").display().to_string(); write_people(&uri, vec![1], WriteMode::Create).await; @@ -1452,13 +1460,13 @@ mod tests { let engine = Arc::new(CypherEngine::new(config, sources, true).await.unwrap()); // Stand in for refresh A, still building its replacement. - let in_flight = engine.refresh_lock.lock().await; + let in_flight = engine.update_lock.lock().await; let b = tokio::spawn({ let engine = engine.clone(); - async move { engine.refresh().await } + async move { engine.checkout_latest().await } }); tokio::time::sleep(std::time::Duration::from_millis(50)).await; - assert!(!b.is_finished(), "refresh B ran while A was in flight"); + assert!(!b.is_finished(), "checkout B ran while A was in flight"); // A newer version lands before A finishes; B must publish it, not anything older. write_people(&uri, vec![2], WriteMode::Append).await; diff --git a/crates/lance-csr/tests/test_engine.rs b/crates/lance-csr/tests/test_engine.rs index fe1b18d..5e5adcc 100644 --- a/crates/lance-csr/tests/test_engine.rs +++ b/crates/lance-csr/tests/test_engine.rs @@ -116,7 +116,7 @@ async fn parameters_and_streaming() { } #[tokio::test] -async fn lance_input_is_pinned_until_refresh() { +async fn lance_input_is_pinned_until_checkout_latest() { let dir = tempfile::tempdir().unwrap(); let person_uri = dir.path().join("person.lance"); let knows_uri = dir.path().join("knows.lance"); @@ -144,14 +144,14 @@ async fn lance_input_is_pinned_until_refresh() { ["Carol"] ); - // A write after construction stays invisible until refresh(). + // A write after construction stays invisible until checkout_latest(). write(knows(&[(2, 4)]), knows_uri, WriteMode::Append).await; assert_eq!( names(&engine.execute(TWO_HOPS, HashMap::new()).await.unwrap()), ["Carol"] ); - engine.refresh().await.unwrap(); + engine.checkout_latest().await.unwrap(); assert_eq!( names(&engine.execute(TWO_HOPS, HashMap::new()).await.unwrap()), ["Carol", "David"] @@ -179,7 +179,7 @@ async fn optimize_csr_folds_the_delta_into_the_base() { let engine = CypherEngine::new(config(), sources, true).await.unwrap(); write(knows(&[(2, 4)]), knows_uri, WriteMode::Append).await; - engine.refresh().await.unwrap(); + engine.checkout_latest().await.unwrap(); let csr = engine.csr_stats()["KNOWS"].clone(); assert_eq!( (csr.update, csr.delta_edges, csr.unindexed_edges), diff --git a/docs/AGENTS.md b/docs/AGENTS.md index b295d0e..cd1e3df 100644 --- a/docs/AGENTS.md +++ b/docs/AGENTS.md @@ -83,7 +83,7 @@ them. caveats and known issues; `???` makes a collapsible one for long detail; - content tabs (`=== "Tab"`) for alternatives, such as install methods or input types; - code annotations (`# (1)!` plus a numbered list after the block) to explain specific lines; - - Mermaid diagrams (```` ```mermaid ````) for flows: query pipelines, refresh decisions, snapshots; + - Mermaid diagrams (```` ```mermaid ````) for flows: query pipelines, `checkout_latest()` decisions, snapshots; - tables for options and comparisons, definition lists for method references; - grid cards on the home page only. - Plain, direct sentences. Use "you". Avoid filler and marketing adjectives. Spell out a term the first time diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 6b3a416..ebc825d 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -163,10 +163,10 @@ uv run --no-sync python benchmarks/csr_expand.py --data /tmp/csr-bench --edges 1 | Script | Measures | | --- | --- | -| `csr_expand.py` | CSR build cost, single-hop latency against the join path, refresh cost and peak RSS, per edge count | -| `csr_delta.py` | `refresh()` with a delta against a full rebuild, after one append or delete commit (uses `csr_expand.py`'s data) | +| `csr_expand.py` | CSR build cost, single-hop latency against the join path, `checkout_latest()` cost and peak RSS, per edge count | +| `csr_delta.py` | `checkout_latest()` with a delta against a full rebuild, after one append or delete commit (uses `csr_expand.py`'s data) | | `csr_delta_accumulated.py` | Refresh cost as deltas accumulate over many small commits (the basis of the 25% rebuild threshold) | -| `engine_cache.py` | Engine construction, first-query, warm-query and refresh timings; can compare against upstream `lance_graph` | +| `engine_cache.py` | Engine construction, first-query, warm-query and `checkout_latest()` timings; can compare against upstream `lance_graph` | | `bounded_memory.py` | Peak memory of streamed scans over Lance data larger than a memory cap (`generate`, then `run`) | Each script's docstring lists its options. When you change performance-sensitive code, include before and diff --git a/docs/csr.md b/docs/csr.md index 74aa3dc..838373f 100644 --- a/docs/csr.md +++ b/docs/csr.md @@ -99,7 +99,7 @@ Node index During query execution, each CSR expansion partition keeps one input batch and collects at most one output chunk of neighbour positions before fetching properties. It yields that chunk before visiting the remaining neighbours, even for a single high-degree node. Deleted edge positions are read from the - shared snapshot rather than copied for each query; refresh resolves those positions by scanning the + shared snapshot rather than copied for each query; `checkout_latest()` resolves those positions by scanning the affected sources' neighbour lists. Cancellation checkpoints count source rows and frontiers with no surviving edges, @@ -273,7 +273,7 @@ print(engine.csr_stats()["KNOWS"]) | `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, `refresh()` or [`optimize_csr()`](#rebuilding-the-csr) produced it: `build` (first), `rebuild` (full), `loaded` (from a [stored file](#storing-the-csr), with any changes since applied as a delta), `delta` (changes applied) or `reused` (no edge change). | +| `update` | How the last build, `checkout_latest()` or [`optimize_csr()`](#rebuilding-the-csr) produced it: `build` (first), `rebuild` (full), `loaded` (from a [stored file](#storing-the-csr), with any changes since applied as a delta), `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. | @@ -298,7 +298,7 @@ each other, but it also means the CSR falls behind as soon as anyone writes to t Two calls bring it back up to date, and they do different jobs: -1. **[`refresh()`](#refresh)** moves the engine to the newest version of each Lance dataset. Instead of rebuilding +1. **[`checkout_latest()`](#checkout-latest)** moves the engine to the newest version of each Lance dataset. Instead of rebuilding the CSR, it records the edges added or removed since the last build as a *delta*, a small set of changes kept next to the CSR. Call it whenever your queries should see new writes. 2. **[`optimize_csr()`](#rebuilding-the-csr)** rebuilds a CSR from scratch, so it no longer needs its delta. Call @@ -307,21 +307,21 @@ Two calls bring it back up to date, and they do different jobs: If you gave the engine in-memory PyArrow tables instead of Lance datasets, neither call applies: those tables can't change underneath the engine, so their CSR is never out of date. To query new in-memory data, create a new engine. -### `refresh()`: pick up new writes { #refresh } +### `checkout_latest()`: pick up new writes { #checkout-latest } Because each Lance dataset can evolve independently as you write to it, the engine has to be told that a newer -version is available. `refresh()` does that: it checks every Lance dataset for a newer version and moves the engine to +version is available. `checkout_latest()` does that: it checks every Lance dataset for a newer version and moves the engine to it. Call it after the writes your queries should see, whether they came from `pylance`, LanceDB or another process. -If one writer updates several tables, wait until all of them have committed before you refresh, since each Lance +If one writer updates several tables, wait until all of them have committed before you call it, since each Lance dataset is pinned on its own (see [snapshots are per Lance dataset](lance.md#lazy-version-pinned-reads)). When nothing -has changed, a refresh keeps what the engine already has and returns quickly, so calling it on a timer or before +has changed, `checkout_latest()` keeps what the engine already has and returns quickly, so calling it on a timer or before each batch of queries is fine. -Rebuilding the whole CSR on every refresh would mean scanning every edge again, even when only a handful changed. -So `refresh()` reads only the new parts of each relationship dataset and applies the changes on top of the CSR from +Rebuilding the whole CSR on every call would mean scanning every edge again, even when only a handful changed. +So `checkout_latest()` reads only the new parts of each relationship dataset and applies the changes on top of the CSR from the last full build, which the rest of this page calls the *base*. Appended edges go into the delta, and deleted edges are marked so that lookups skip them; `csr_stats()` counts those marks as `tombstones`. Answers are exactly -what a full rebuild would give. To see what a refresh did, check `update` in [`csr_stats()`](#csr_stats): `reused` +what a full rebuild would give. To see what a call did, check `update` in [`csr_stats()`](#csr_stats): `reused` means nothing changed, `delta` means the changes were applied, and `rebuild` means the CSR was built again from scratch, which happens in the cases below. @@ -338,9 +338,9 @@ scratch, which happens in the cases below. use_csr=True, ) - # David now knows Alice. The engine doesn't see the write until refresh(). + # David now knows Alice. The engine doesn't see the write until checkout_latest(). lance.write_dataset(pa.table({"src": [4], "dst": [1]}), "knows.lance", mode="append") - engine.refresh() + engine.checkout_latest() stats = engine.csr_stats()["KNOWS"] print(stats["edges"], stats["delta_edges"], stats["update"]) # 5 1 delta @@ -351,19 +351,19 @@ scratch, which happens in the cases below. This continues the on-disk example from [Using from Rust](rust.md), where Bob gains an edge to Alice. ```rust - --8<-- "crates/lance-csr/examples/embedded_query.rs:refresh" + --8<-- "crates/lance-csr/examples/embedded_query.rs:checkout_latest" ``` -??? info "How `refresh()` decides" +??? info "How `checkout_latest()` decides" ```mermaid flowchart LR - R["refresh()"] -- "no edge changes" --> U["reused"] + R["checkout_latest()"] -- "no edge changes" --> U["reused"] R -- "changes under 25% of the base" --> DL["delta"] R -- "25% or more, or new endpoint columns" --> RB["rebuild"] ``` - - A refresh rebuilds a CSR instead of adding to its delta once the changes since the last full build would reach + - `checkout_latest()` rebuilds a CSR instead of adding to its delta once the changes since the last full build would reach 25% of the base, or when the relationship dataset was overwritten or its endpoint columns changed. The threshold is fixed: every lookup reads the delta as well as the base, so a large delta slows lookups down, and a rebuild clears it. `python/benchmarks/csr_delta_accumulated.py` measures that trade-off. @@ -376,23 +376,23 @@ scratch, which happens in the cases below. ### `optimize_csr()`: rebuild to clear the delta { #rebuilding-the-csr } -The delta keeps refreshes cheap, but every lookup has to check it as well as the base. A refresh only rebuilds once +The delta keeps `checkout_latest()` cheap, but every lookup has to check it as well as the base. It only rebuilds once the delta reaches 25% of the base, so a smaller delta can stay around indefinitely and slow lookups a little the whole time. `optimize_csr()` lets you clear it at a moment you choose. It rebuilds every CSR that has a delta, from a scan of all its edges, at the version the engine is already pinned to, and leaves CSRs without a delta alone. That costs about as much as building the CSR the first time, so it's best kept for after a bulk load, before a run of -latency-sensitive queries, or a quiet period, rather than called after every refresh. Query results are the same +latency-sensitive queries, or a quiet period, rather than called after every `checkout_latest()`. Query results are the same before and after. ```python -engine.refresh() # applies the new edges as a delta +engine.checkout_latest() # applies the new edges as a delta engine.optimize_csr() # rebuilds the CSR, now including those edges stats = engine.csr_stats()["KNOWS"] print(stats["update"], stats["unindexed_edges"]) # rebuild 0 ``` Afterwards, `csr_stats()` shows `delta_edges`, `tombstones` and `unindexed_edges` at 0, and `base_version` matches -`version`. `optimize_csr()` doesn't look for newer data, so call `refresh()` first if the rebuild should include the +`version`. `optimize_csr()` doesn't look for newer data, so call `checkout_latest()` first if the rebuild should include the latest writes. ??? info "How this maps onto Lance's indexes" @@ -413,18 +413,18 @@ latest writes. ### Which one to call -!!! tip "`refresh()` or `optimize_csr()`?" +!!! tip "`checkout_latest()` or `optimize_csr()`?" - In short, `refresh()` keeps the engine up to date at little cost, and `optimize_csr()` now and then clears the - delta that refreshes leave behind. + In short, `checkout_latest()` keeps the engine up to date at little cost, and `optimize_csr()` now and then clears the + delta that `checkout_latest()` leaves behind. - | | `refresh()` | `optimize_csr()` | + | | `checkout_latest()` | `optimize_csr()` | | --- | --- | --- | | Data | Moves to the latest version of each Lance dataset | Stays at the version the engine is pinned to | | Usual effect on a CSR | Applies the new writes as a delta, reading only new fragments | Rebuilds it from a scan of all its edges | | Rebuilds a CSR when | Its delta would pass 25% of the base | It carries any delta at all | | Cost | In proportion to the new writes | Like the first build | - | When to call it | Whenever queries should see new writes | Now and then, after `refresh()`, once a delta has built up | + | When to call it | Whenever queries should see new writes | Now and then, after `checkout_latest()`, once a delta has built up | Whenever either call rebuilds a CSR, it runs the same full build and ends up with the same CSR. Both build the replacement completely before the engine switches to it, and they never run at the same time. @@ -458,10 +458,10 @@ The file belongs to the table: dropping or copying the dataset drops or copies i files it recognises. - **Changes since the stored version** apply as a delta, read from the new fragments only, exactly as - [`refresh()`](#keeping-the-csr-current) applies them. `update` says `loaded` and `base_version` names the + [`checkout_latest()`](#keeping-the-csr-current) applies them. `update` says `loaded` and `base_version` names the stored version. That holds after compaction and `cleanup_old_versions()` too. - **Past the 25% threshold**, the engine rebuilds the CSR and stores the new version. Every full build is stored, - including those from `refresh()` and [`optimize_csr()`](#rebuilding-the-csr), and older files are deleted + including those from `checkout_latest()` and [`optimize_csr()`](#rebuilding-the-csr), and older files are deleted once a newer one is written. - **A file that can't be used** (corrupt, written for other endpoint columns or by another format) is ignored and rebuilt, so a bad file costs a build, never a wrong answer. diff --git a/docs/cypher.md b/docs/cypher.md index d47a1c5..f58bfc9 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -6,7 +6,7 @@ There are two ways to run Cypher, and most code wants the first: | --- | --- | --- | | Use it for | Many queries over the same data | One query, once | | Setup cost | Paid once, in the constructor | Paid on every `execute()` | -| Lance datasets | Read lazily, pinned to a version, `refresh()` | Read into memory | +| Lance datasets | Read lazily, pinned to a version, `checkout_latest()` | Read into memory | | CSR index | `use_csr=True` | Not available | | Streaming results | `execute_batches()` | No | @@ -126,7 +126,7 @@ print(rows) # 4 - CSR expansion visits neighbours and fetches their properties a chunk at a time, including when one node has many neighbours. See [CSR memory cost](csr.md#turning-it-on) for what it keeps in memory. - Planning errors (a syntax error, an unknown label) raise immediately. Execution errors raise while reading. -- The reader keeps its snapshot alive, so a concurrent [`refresh()`](lance.md#refresh) doesn't affect it. +- The reader keeps its snapshot alive, so a concurrent [`checkout_latest()`](lance.md#checkout-latest) doesn't affect it. - Dropping the reader cancels the remaining work. ### Query deadlines @@ -163,7 +163,7 @@ print("CsrExpandExec" in plan) # True | Method | Returns | | --- | --- | | `config()` | The engine's `GraphConfig`. | -| `refresh()` | `None`. Moves every Lance dataset to its latest version. See [Lance datasets](lance.md#refresh). | +| `checkout_latest()` | `None`. Moves every Lance dataset to its latest version. See [Lance datasets](lance.md#checkout-latest). | | `dataset_versions()` | `dict[str, int]`: the pinned version of each Lance dataset. | | `csr_stats()`, `node_index_stats()` | CSR statistics. See [The CSR index](csr.md#inspecting-the-csr). | | `execute_with_vector_rerank(query, vector_search)` | A table reranked by vector similarity. See [Vector search](vector-search.md). | @@ -372,7 +372,7 @@ in `__lance_csr_identity`. Relationship records also carry the node records `__l `__lance_csr_end`. These metadata fields are distinct from user properties, even if a property has the same name. `keys(entity)` lists only non-null user properties; `keys(map)` includes null-valued keys. Prefer the Cypher property and endpoint functions when writing queries. Relationship identities identify actual occurrences within this query's inventory, not persistent -IDs that can be compared across queries or refreshes. Rich paths support the same signed integer and +IDs that can be compared across queries or snapshots. Rich paths support the same signed integer and `uint8`/`uint16`/`uint32` keys as the CSR. Mixed labels may share compatible properties; conflicting Arrow property types fail explicitly. Non-null node IDs must be unique within each participating label so relationship endpoints identify one node; duplicate IDs raise an error. Expanding a graph pattern from @@ -382,7 +382,7 @@ A path value needs a query-local inventory of the participating node and relatio properties. It therefore materializes those tables, even with `use_csr=True`. Traversal output is emitted in batches and named ordinary paths keep a cursor per hop, but inventory memory grows with the graph. Rich path queries bypass the query-template cache: each execution binds its current parameters and -creates an inventory from its own snapshot. A reader opened before `refresh()` retains the old snapshot. +creates an inventory from its own snapshot. A reader opened before `checkout_latest()` retains the old snapshot. Queries asking only for shortest-path length, node keys or path null checks, including homogeneous node-property comprehensions and paths carried through `WITH` aliases, retain the usual CSR execution route when the relationship joins the same node label. Cross-label shortest paths use typed identities even for diff --git a/docs/execution.md b/docs/execution.md index c4df547..4f8ca27 100644 --- a/docs/execution.md +++ b/docs/execution.md @@ -71,7 +71,7 @@ make it build a multi-million-row intermediate as one single-threaded hash table lance-csr computes distinct counts for every node key and edge endpoint column (from the CSR when there is one, otherwise with one scan of the endpoint columns) and reports them to DataFusion's join planner. The -counts are cached per dataset version, so an unchanged `refresh()` costs nothing. +counts are cached per dataset version, so an unchanged `checkout_latest()` costs nothing. ## Parallelism @@ -93,14 +93,14 @@ sequenceDiagram participant E as CypherEngine participant Q2 as Query 2 Q1->>E: execute() takes snapshot v1 - Note over E: refresh() builds snapshot v2
(reusing unchanged parts), then swaps it in + Note over E: checkout_latest() builds snapshot v2
(reusing unchanged parts), then swaps it in Q2->>E: execute() takes snapshot v2 E-->>Q1: finishes on v1 E-->>Q2: finishes on v2 ``` -`refresh()` builds a complete replacement next to the current snapshot, reusing everything whose source -dataset hasn't changed, and swaps it in atomically. A failed refresh leaves the old snapshot in place. +`checkout_latest()` builds a complete replacement next to the current snapshot, reusing everything whose source +dataset hasn't changed, and swaps it in atomically. A failed call leaves the old snapshot in place. ## Single-machine scope diff --git a/docs/getting-started.md b/docs/getting-started.md index d6028ba..3234d79 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -214,7 +214,7 @@ than one query against the same data. - [Graph schema](graph-schema.md): every option of `GraphConfig`. - [Querying with Cypher](cypher.md): parameters, streaming, `explain()` and the supported Cypher. -- [Lance datasets](lance.md): query data on disk without loading it, and pick up new writes with `refresh()`. +- [Lance datasets](lance.md): query data on disk without loading it, and pick up new writes with `checkout_latest()`. - [Using from Rust](rust.md): the same engine as a Rust library. - The [`examples/`](https://github.com/hyperdimensionalcomputingai/lance-csr/tree/main/examples) directory has runnable scripts. diff --git a/docs/index.md b/docs/index.md index a27133b..ae01b96 100644 --- a/docs/index.md +++ b/docs/index.md @@ -78,7 +78,7 @@ each relationship hop from a hash join into an array lookup, which is what makes --- - Lazy, version-pinned reads, `refresh()`, streaming results and LanceDB tables. + Lazy, version-pinned reads, `checkout_latest()`, streaming results and LanceDB tables. [:octicons-arrow-right-24: Lance datasets](lance.md) diff --git a/docs/internals/index.md b/docs/internals/index.md index 2532ed4..abb3059 100644 --- a/docs/internals/index.md +++ b/docs/internals/index.md @@ -53,7 +53,7 @@ matters. - **Bounded, streaming operators.** The extension operators yield bounded output batches and resume where they left off, even inside a single high-degree node's neighbour list, and they check for cancellation as they go. - **Immutable snapshots.** An engine holds a snapshot of pinned dataset versions plus the indexes built for them. - `refresh()` builds a replacement next to it, reusing whatever didn't change, and swaps it in atomically, so running + `checkout_latest()` builds a replacement next to it, reusing whatever didn't change, and swaps it in atomically, so running queries never see a half-updated index. ## Where the extensions plug in diff --git a/docs/internals/indexes.md b/docs/internals/indexes.md index 6d23cc8..43aa2d9 100644 --- a/docs/internals/indexes.md +++ b/docs/internals/indexes.md @@ -2,7 +2,7 @@ The [custom operators](operators.md) read two in-memory indexes. Both are built from the same underlying Lance dataset and point back into it, so neither holds a copy of node or relationship properties. [The CSR index](../csr.md) covers what users see: -how to turn it on, `csr_stats()`, and when to call `refresh()` and `optimize_csr()`. This page covers how the indexes +how to turn it on, `csr_stats()`, and when to call `checkout_latest()` and `optimize_csr()`. This page covers how the indexes are laid out and maintained. The compact store of offsets and neighbours came from [lance-graph](https://github.com/lance-format/lance-graph), @@ -67,7 +67,7 @@ table, so edge commits update only the adjacency and node commits rebuild only t ## How a refresh diffs Lance versions -`refresh()` compares the pinned and latest versions of each relationship dataset fragment by fragment, from their +`checkout_latest()` compares the pinned and latest versions of each relationship dataset fragment by fragment, from their manifests: | Fragment | Result | @@ -133,7 +133,7 @@ names, their Lance field ids and types, and the fragments the base was built fro When an engine starts, it reads the newest file at or before the pinned version. It then diffs that file's fragments against the pinned version, as a refresh diffs two versions (see [How a refresh diffs Lance versions](#how-a-refresh-diffs-lance-versions)), so the file's own version doesn't need to exist any more. An empty diff uses the file as it is; otherwise the changes become the delta, under the same 25% -threshold. Rebuilds after start-up (from `refresh()` or `optimize_csr()`) don't look for a file, since the stored base +threshold. Rebuilds after start-up (from `checkout_latest()` or `optimize_csr()`) don't look for a file, since the stored base is the one the engine already holds, but they do store the new one. Every batch is compressed with zstd, Arrow IPC's own buffer compression, on top of the packing. A load decompresses diff --git a/docs/internals/optimizer.md b/docs/internals/optimizer.md index 1bec696..5fe00af 100644 --- a/docs/internals/optimizer.md +++ b/docs/internals/optimizer.md @@ -114,4 +114,4 @@ graph join chain, that turns a multi-million-row intermediate into an estimate o `KeyStatisticsProvider` wraps each table and reports distinct counts for its node-id and edge-endpoint columns, taken from the CSR when there is one and otherwise from one scan of the endpoint columns. Every other statistic comes from -the inner table. The counts are cached per dataset version, so a `refresh()` with no changes costs nothing. +the inner table. The counts are cached per dataset version, so a `checkout_latest()` with no changes costs nothing. diff --git a/docs/lance.md b/docs/lance.md index e24527f..b9b3ca8 100644 --- a/docs/lance.md +++ b/docs/lance.md @@ -36,21 +36,22 @@ print(engine.dataset_versions()) # {'Person': 1, 'KNOWS': 1} ``` Each dataset is **pinned to the version you passed in**. Writes made later, by your process or another one, -stay invisible until you call `refresh()`. That gives every query a stable snapshot. +stay invisible until you call `checkout_latest()`. That gives every query a stable snapshot. !!! note "Snapshots are per dataset" Each dataset is pinned separately. The versions across tables are not a transactional multi-table - snapshot; if a writer updates `Person` and `KNOWS` one after the other, refresh after both commits. + snapshot; if a writer updates `Person` and `KNOWS` one after the other, call `checkout_latest()` after both commits. !!! warning "Use pylance 13.x to write" The engine is built against Lance 13. Datasets written by an older or newer major version of `pylance` may use a file format it can't read. Installing `lance-csr` pins a compatible `pylance` for you. -## `refresh()` { #refresh } +## `checkout_latest()` { #checkout-latest } -`refresh()` moves every Lance dataset to its latest version: +`checkout_latest()` moves every Lance dataset to its latest version. It's named after Lance's own +`LanceDataset.checkout_latest()`, which does the same for a single dataset: ```python lance.write_dataset( @@ -59,27 +60,32 @@ lance.write_dataset( lance.write_dataset(pa.table({"src": [3], "dst": [4]}), "knows.lance", mode="append") print(engine.execute("MATCH (p:Person) RETURN count(*) AS n").to_pylist()) # [{'n': 3}] -engine.refresh() +engine.checkout_latest() print(engine.execute("MATCH (p:Person) RETURN count(*) AS n").to_pylist()) # [{'n': 4}] print(engine.dataset_versions()) # {'Person': 2, 'KNOWS': 2} ``` -What `refresh()` guarantees: +!!! note "Formerly `refresh()`" + + This method was called `refresh()` before. `refresh()` still works, and does exactly the same, but it + warns with `DeprecationWarning` and will be removed in a later release. + +What `checkout_latest()` guarantees: - **Unchanged datasets keep their handle.** A dataset whose version and commit time are unchanged is reused, - along with its CSR, node index and statistics, so a no-op refresh is cheap. -- **Schema changes are picked up.** New columns are queryable after the refresh. + along with its CSR, node index and statistics, so calling it when nothing changed is cheap. +- **Schema changes are picked up.** New columns are queryable afterwards. - **All or nothing.** The replacement is validated in full before it's swapped in. On error, nothing changes and the previous snapshot keeps working. -- **Concurrent refreshes cannot overwrite a newer published snapshot with an older candidate.** If another - refresh publishes first, the overlapping refresh retries from that snapshot and reopens the latest datasets. - The retry can add work when refreshes overlap; it does not block queries while the replacement is built. -- **Running queries are unaffected.** Queries and `execute_batches()` readers that started before the refresh +- **Concurrent calls cannot overwrite a newer published snapshot with an older candidate.** If another call + publishes first, the overlapping one retries from that snapshot and reopens the latest datasets. The retry + can add work when calls overlap; it does not block queries while the replacement is built. +- **Running queries are unaffected.** Queries and `execute_batches()` readers that started before the call finish on their original snapshot. - **The CSR updates incrementally**, as described in [Keeping the CSR current](csr.md#keeping-the-csr-current). - **PyArrow inputs don't change.** To replace in-memory data, build a new engine. -A refresh reopens each dataset by its URI with the same `storage_options` it was opened with, so object-store +`checkout_latest()` reopens each dataset by its URI with the same `storage_options` it was opened with, so object-store credentials carry over. ## Repeated queries @@ -91,8 +97,8 @@ rows and results are never cached. Query text and parameter names/types identify separate from non-null values. Least recently used templates are evicted at the limit. Lists, maps and parameterized `SKIP`/`LIMIT` counts use ordinary planning because their shape or value can -change the plan. Queries whose operators cannot produce a reusable template fall back to ordinary planning. `explain()` plans the current values directly. A successful `refresh()` starts an empty cache for the -new snapshot, including a refresh whose datasets did not change; active readers retain their existing plans. +change the plan. Queries whose operators cannot produce a reusable template fall back to ordinary planning. `explain()` plans the current values directly. A successful `checkout_latest()` starts an empty cache for the +new snapshot, even when no dataset changed; active readers retain their existing plans. These rules apply with and without CSR. They do not promise a speedup for a particular workload. ## Query parallelism @@ -108,13 +114,13 @@ serial_engine = CypherEngine( target_partitions=1, ) print(serial_engine.execute("MATCH (p:Person) RETURN count(*) AS n").to_pylist()) # [{'n': 4}] -serial_engine.refresh() # retains target_partitions=1 +serial_engine.checkout_latest() # retains target_partitions=1 ``` The target is a planning preference, not a thread or memory limit; operators may use a different number of partitions. A lower target can help short queries while making larger joins and scans slower. Compare your full workload in the storage mode you use before choosing a value. The default remains unchanged, -and `refresh()` preserves your choice. Booleans and non-integers other than `None` raise `TypeError`; +and `checkout_latest()` preserves your choice. Booleans and non-integers other than `None` raise `TypeError`; zero, negative and out-of-range integers raise `ValueError`. ## Streaming large results @@ -142,7 +148,7 @@ memory. Joins, sorts and aggregations may still buffer their inputs. | Join-key statistics | In memory, a few numbers per key column | | Query results | In memory with `execute()`; streamed with `execute_batches()` | -The CSR and node index are built from the Lance data once and updated on `refresh()`; queries don't rescan +The CSR and node index are built from the Lance data once and updated on `checkout_latest()`; queries don't rescan the edge table to traverse it. ## Tables created with LanceDB @@ -166,12 +172,12 @@ config = ( engine = CypherEngine(config, {"Person": people.to_lance(), "KNOWS": knows.to_lance()}) knows.add(new_edges) # keep writing with the LanceDB API -engine.refresh() # the engine sees the new version +engine.checkout_latest() # the engine sees the new version ``` - Reads are lazy and version-pinned exactly as with `lance.dataset(...)`. - Passing the LanceDB `Table` object itself raises a `TypeError`. `table.to_arrow()` works, but loads the - whole table into memory and never refreshes. + whole table into memory and never sees later versions. - Tested with LanceDB 0.40.0 on local tables (Lance file format 2.2). A LanceDB release bundling a newer Lance major may write files this engine can't read. @@ -205,4 +211,4 @@ print(namespace.base_uri) # graph ``` `CypherQuery` (with a namespace or a dict) reads the data for each `execute()`; use `CypherEngine` for lazy -reads, refresh and the CSR. +reads, `checkout_latest()` and the CSR. diff --git a/docs/rust.md b/docs/rust.md index 32bcd14..a39aa19 100644 --- a/docs/rust.md +++ b/docs/rust.md @@ -58,7 +58,7 @@ Each input is an `EngineSource`: `EngineSource::open_lance(uri, version, storage_options)` : Opens a Lance dataset, pinned to `version` (`None` means the latest). The dataset is read lazily as - queries need it, and `storage_options` are kept so `refresh()` can reopen it with the same credentials. + queries need it, and `storage_options` are kept so `checkout_latest()` can reopen it with the same credentials. See [Lance datasets](lance.md). `EngineSource::Arrow(batch)` @@ -92,11 +92,11 @@ Each query runs on the snapshot that was current when it started. ## Pick up new writes -`refresh()` moves every Lance dataset to its latest version and updates the CSR, applying the new edges as a +`checkout_latest()` moves every Lance dataset to its latest version and updates the CSR, applying the new edges as a delta when it can (see [Keeping the CSR current](csr.md#keeping-the-csr-current)): ```rust ---8<-- "crates/lance-csr/examples/embedded_query.rs:refresh" +--8<-- "crates/lance-csr/examples/embedded_query.rs:checkout_latest" ``` `csr_stats()` and `node_index_stats()` return the same fields as in Python, as `CsrStats` and `NodeIndexStats` @@ -118,7 +118,7 @@ different exception depending on the variant: ## Async runtimes and cancellation -- Building the CSR in `new()` and `refresh()` is CPU-heavy work inside an async function, so it holds a +- Building the CSR in `new()` and `checkout_latest()` is CPU-heavy work inside an async function, so it holds a runtime worker thread until it finishes. In an application with its own runtime (a desktop or web backend), run those two calls with `tokio::task::spawn_blocking` so other tasks keep running. - Dropping a query's future, or the stream from `execute_stream`, cancels the rest of that query. @@ -147,7 +147,7 @@ The deadline applies only to work inside the call; it isn't stored in the engine | a PyArrow table in `datasets` | `EngineSource::Arrow(batch)` | | `engine.execute(query, params)` → `pyarrow.Table` | `engine.execute(query, params).await` → `RecordBatch` | | `engine.execute_batches(...)` → `RecordBatchReader` | `engine.execute_stream(...).await` → stream | -| `engine.refresh()` | `engine.refresh().await` | +| `engine.checkout_latest()` | `engine.checkout_latest().await` | | `engine.optimize_csr()` | `engine.optimize_csr().await` | | `engine.execute(query, timeout=2.0)` | `timeout_at(deadline, with_execution_deadline(Some(deadline), engine.execute(...)))` | | `CypherEngine(..., target_partitions=4)` | `CypherEngine::with_target_partitions(config, sources, use_csr, Some(4)).await` | diff --git a/docs/sql.md b/docs/sql.md index fd5bf8d..483453c 100644 --- a/docs/sql.md +++ b/docs/sql.md @@ -55,7 +55,7 @@ print(friends.to_pylist()) The SQL dialect is DataFusion's; see the [DataFusion SQL reference](https://datafusion.apache.org/user-guide/sql/index.html) for what it supports. -`SqlEngine` has no `refresh()` or CSR: those are `CypherEngine` features. +`SqlEngine` has no `checkout_latest()` or CSR: those are `CypherEngine` features. ## From Cypher to SQL diff --git a/examples/README.md b/examples/README.md index de2e6d9..6d61a34 100644 --- a/examples/README.md +++ b/examples/README.md @@ -8,7 +8,7 @@ dependencies): ```bash uv run --with lance-csr python examples/basic_cypher.py uv run --with lance-csr python examples/kg_traversal.py -uv run --with lance-csr python examples/lance_snapshot_refresh.py +uv run --with lance-csr python examples/lance_snapshots.py ``` With a local build, so the examples use your changes, set up `python/.venv` as in @@ -35,8 +35,8 @@ uv run --with lance-csr python examples/basic_cypher.py uv run --with lance-csr python examples/kg_traversal.py ``` -- `lance_snapshot_refresh.py`: lazy Lance datasets, snapshots and `refresh()`. +- `lance_snapshots.py`: lazy Lance datasets, snapshots and `checkout_latest()`. ```bash -uv run --with lance-csr python examples/lance_snapshot_refresh.py +uv run --with lance-csr python examples/lance_snapshots.py ``` diff --git a/examples/lance_snapshot_refresh.py b/examples/lance_snapshots.py similarity index 84% rename from examples/lance_snapshot_refresh.py rename to examples/lance_snapshots.py index efae887..333afa0 100644 --- a/examples/lance_snapshot_refresh.py +++ b/examples/lance_snapshots.py @@ -1,11 +1,11 @@ """ -Disk-backed CypherEngine: pinned snapshots, refresh, and streamed results. +Disk-backed CypherEngine: pinned snapshots, checkout_latest(), and streamed results. Requirements: an installed lance-csr wheel (or `maturin develop` in python/), which pulls in a compatible pylance 13.x. Run: - python examples/lance_snapshot_refresh.py + python examples/lance_snapshots.py """ from __future__ import annotations @@ -46,7 +46,7 @@ def main() -> None: "Person": lance.dataset(person_uri), "KNOWS": lance.dataset(root / "knows.lance"), }, - # Optional: choose this target after measuring your own queries; refresh keeps it. + # Optional: choose this target after measuring your own queries; checkout_latest() keeps it. target_partitions=2, ) count = "MATCH (p:Person) RETURN count(*) AS n" @@ -59,13 +59,13 @@ def main() -> None: ).to_pylist(), ) - # Writes are not visible until refresh(): the engine is pinned to version 1. + # Writes are not visible until checkout_latest(): the engine is pinned to version 1. lance.write_dataset(people(ROWS, ROWS + 10), person_uri, mode="append") lance.dataset(person_uri).add_columns({"age": "id % 100"}) # schema change - print("after writes, before refresh:", engine.execute(count).column(0)[0].as_py()) - engine.refresh() + print("after writes, before checkout_latest():", engine.execute(count).column(0)[0].as_py()) + engine.checkout_latest() print( - "after refresh:", + "after checkout_latest():", engine.execute(count).column(0)[0].as_py(), engine.dataset_versions(), ) diff --git a/python/README.md b/python/README.md index 39a051d..471d200 100644 --- a/python/README.md +++ b/python/README.md @@ -110,11 +110,11 @@ lets relationship-type alternation (`[:likes_post|likes_comment]`) pick the righ `with_node_label` and `with_relationship`, the names in lance-csr 0.1, were removed in 0.5.0; use `node_table` and `rel_table` instead. -### 3. Lance datasets: snapshots, refresh and streaming +### 3. Lance datasets: snapshots, checking out new versions and streaming `CypherEngine` reads Lance datasets lazily from storage (projections and filters are pushed into the scan). Each dataset is pinned to the version you pass in; later writes are visible only after -`refresh()`. Use `execute_batches()` to stream large results instead of materializing a table. +`checkout_latest()`. Use `execute_batches()` to stream large results instead of materializing a table. ```python import lance @@ -124,14 +124,14 @@ config = GraphConfig.builder().node_table("Person", "id").build() engine = CypherEngine(config, {"Person": lance.dataset("people.lance")}) engine.execute("MATCH (p:Person) RETURN count(*)") # sees the pinned version -engine.refresh() # move to the latest version +engine.checkout_latest() # move to the latest version for batch in engine.execute_batches("MATCH (p:Person) RETURN p.id"): ... # pyarrow.RecordBatch, one at a time ``` Experimental: `CypherEngine(config, datasets, use_csr=True)` answers single-hop patterns over integer ids with an in-memory CSR adjacency, and fetches target rows by position instead of joining. The CSR is -rebuilt on `refresh()` when the relationship table changes, and the per-label id index when its node +rebuilt on `checkout_latest()` when the relationship table changes, and the per-label id index when its node table changes. Results are identical; peak memory grows by about 95 bytes per edge while the CSR is built (both directions). Write Lance data with `pylance` 13.x: the engine is built against Lance 13 and older or newer @@ -158,11 +158,11 @@ config = ( engine = CypherEngine(config, {"Person": people.to_lance(), "KNOWS": knows.to_lance()}) knows.add(new_edges) # keep writing with the LanceDB API -engine.refresh() # the engine sees the new version +engine.checkout_latest() # the engine sees the new version ``` - Reads are lazy and pinned to the version you pass in, exactly as with `lance.dataset(...)`. -- `table.to_arrow()` also works, but loads the whole table into memory and never refreshes. +- `table.to_arrow()` also works, but loads the whole table into memory and never sees later versions. - Tested with LanceDB 0.40.0 on local tables, which write Lance file format 2.2. A LanceDB release that bundles a newer Lance major may write files this engine cannot read. - Not supported: LanceDB Cloud (remote tables have no `to_lance()`), and tables opened through a diff --git a/python/benchmarks/csr_delta.py b/python/benchmarks/csr_delta.py index e74ccbe..427bab7 100644 --- a/python/benchmarks/csr_delta.py +++ b/python/benchmarks/csr_delta.py @@ -1,14 +1,14 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright The Lance Authors -"""Incremental CSR maintenance: refresh-with-delta cost against a full rebuild. +"""Incremental CSR maintenance: ``checkout_latest()``-with-delta cost against a full rebuild. Uses the graphs written by ``csr_expand.py`` (``--data`` must point at the same directory). For each scenario, on a fresh copy of the edge table (1M-row fragments, plus a ``seq`` row-number column for targeted deletes): - apply one commit (append or delete); -- time ``refresh()`` on a CSR engine built before the commit (delta path); +- time ``checkout_latest()`` on a CSR engine built before the commit (delta path); - time a full CSR build on the new version (what a rebuild costs); - compare warm query latency with the delta in place against the clean build, checking both return identical results. @@ -116,7 +116,7 @@ def main() -> None: engine = CypherEngine(CONFIG, datasets, use_csr=True) change(uri, nodes) - refresh_ms, _ = timed(engine.refresh) + refresh_ms, _ = timed(engine.checkout_latest) stats = engine.csr_stats()["KNOWS"] latest = {"Person": person, "KNOWS": lance.dataset(uri)} diff --git a/python/benchmarks/csr_delta_accumulated.py b/python/benchmarks/csr_delta_accumulated.py index e2dcf73..5dbfac2 100644 --- a/python/benchmarks/csr_delta_accumulated.py +++ b/python/benchmarks/csr_delta_accumulated.py @@ -4,7 +4,7 @@ """Refresh cost as CSR deltas accumulate over many small commits. Uses a graph written by ``csr_expand.py``. On a fresh copy of the edge table, -appends ``--step`` of the edges ``--commits`` times, calling ``refresh()`` after +appends ``--step`` of the edges ``--commits`` times, calling ``checkout_latest()`` after each, and prints each refresh's time with the delta size so far. With deltas that are copied on every refresh, the time grows with the accumulated delta; with shared delta layers it should stay roughly proportional to the step. @@ -59,7 +59,7 @@ def main() -> None: ) lance.write_dataset(new, uri, mode="append") start = time.perf_counter() - engine.refresh() + engine.checkout_latest() refresh_ms = (time.perf_counter() - start) * 1e3 stats = engine.csr_stats()["KNOWS"] print( diff --git a/python/benchmarks/csr_expand.py b/python/benchmarks/csr_expand.py index 6ef4e92..5134b51 100644 --- a/python/benchmarks/csr_expand.py +++ b/python/benchmarks/csr_expand.py @@ -124,7 +124,7 @@ def open_engine(): out[name] = {**check, **summary(samples)} t = time.perf_counter() - engine.refresh() + engine.checkout_latest() out["refresh_unchanged_ms"] = (time.perf_counter() - t) * 1e3 # Recorded before the copy below, which materializes the edge table. out["peak_rss_mib"] = peak_rss_mib() @@ -149,7 +149,7 @@ def open_engine(): ) lance.write_dataset(new_edges, knows, mode="append") t = time.perf_counter() - engine.refresh() + engine.checkout_latest() out["refresh_after_append_ms"] = (time.perf_counter() - t) * 1e3 if mode == "csr": out["csr_rebuild_ms"] = engine.csr_stats()["KNOWS"]["build_ms"] diff --git a/python/benchmarks/engine_cache.py b/python/benchmarks/engine_cache.py index 05a1ff2..ab32a06 100644 --- a/python/benchmarks/engine_cache.py +++ b/python/benchmarks/engine_cache.py @@ -1,7 +1,7 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright The Lance Authors -"""Engine construction, first-query, warm-query and refresh timings. +"""Engine construction, first-query, warm-query and checkout_latest() timings. Runs against either the upstream build (``--module lance_graph``) or this fork (``--module lance_csr``) on the same on-disk Lance data, so the two can be compared @@ -112,11 +112,13 @@ def emit(case, **values): warm.append(t) emit("warm_query", **summary(warm)) - # Upstream has no refresh(); its equivalent is building a new engine. + # Upstream has no checkout_latest() (refresh() in older lance-csr); its equivalent + # is building a new engine. refresh = [] + checkout = getattr(engine, "checkout_latest", None) or getattr(engine, "refresh", None) for _ in range(min(args.repeats, 10)): - if hasattr(engine, "refresh"): - t, _ = timed(engine.refresh) + if checkout is not None: + t, _ = timed(checkout) else: t, engine = timed(lambda: mod.CypherEngine(config, open_datasets())) refresh.append(t) diff --git a/python/python/lance_csr/_internal/graph.pyi b/python/python/lance_csr/_internal/graph.pyi index b636099..64b74f0 100644 --- a/python/python/lance_csr/_internal/graph.pyi +++ b/python/python/lance_csr/_internal/graph.pyi @@ -133,6 +133,7 @@ class CypherEngine: def explain( self, query: str, params: dict[str, _ParameterT] | None = None ) -> str: ... + def checkout_latest(self) -> None: ... def refresh(self) -> None: ... def optimize_csr(self) -> None: ... def delete_stored_csr(self) -> int: ... diff --git a/python/python/lance_csr/cypher.py b/python/python/lance_csr/cypher.py index c8ce1ab..a2b795a 100644 --- a/python/python/lance_csr/cypher.py +++ b/python/python/lance_csr/cypher.py @@ -2,6 +2,7 @@ from __future__ import annotations +import warnings from enum import IntEnum from typing import TYPE_CHECKING, Any, Self, final @@ -47,9 +48,9 @@ class CypherEngine: Lance datasets are read lazily, pinned to the version passed in: later writes, by this or another process, aren't visible until - [`refresh`][lance_csr.CypherEngine.refresh]. Each dataset is pinned separately, so - versions across tables are not a transactional snapshot. PyArrow inputs are held in - memory and never change; create a new engine to replace them. + [`checkout_latest`][lance_csr.CypherEngine.checkout_latest]. Each dataset is pinned + separately, so versions across tables are not a transactional snapshot. PyArrow + inputs are held in memory and never change; create a new engine to replace them. [`execute`][lance_csr.CypherEngine.execute] and [`execute_batches`][lance_csr.CypherEngine.execute_batches] reuse up to 128 @@ -58,7 +59,8 @@ class CypherEngine: call. Container parameters, rich path values, `CALL` subqueries, parameterized `SKIP`/`LIMIT` and unsupported template shapes use ordinary planning. [`explain`][lance_csr.CypherEngine.explain] always plans directly, and a successful - `refresh` clears the cache. See [Repeated queries](lance.md#repeated-queries). + `checkout_latest` clears the cache. See [Repeated + queries](lance.md#repeated-queries). Parameters ---------- @@ -73,17 +75,17 @@ class CypherEngine: endpoints, and node indexes for labels with unique integer keys. target_partitions : int or None Keyword-only. A target from 1 to 1024 for DataFusion query partitioning; `None` - keeps DataFusion's default. `refresh` preserves it. It is not a thread or memory - limit. See [Query parallelism](lance.md#query-parallelism). + keeps DataFusion's default. `checkout_latest` preserves it. It is not a thread + or memory limit. See [Query parallelism](lance.md#query-parallelism). persist_csr : bool Keyword-only, experimental, needs `use_csr=True`. Store each relationship dataset's CSR inside the dataset's own directory, at `_lance_csr/.arrow`, and load it instead of building when an engine starts. Changes since the stored version apply as a delta under the same 25% - threshold as `refresh`; past it, the CSR is rebuilt and stored again. Lance's - `cleanup_old_versions` leaves the folder alone. PyArrow inputs aren't stored, - and a dataset that can't be written (read-only storage or credentials) is - built as without it. See [Storing the CSR](csr.md#storing-the-csr). + threshold as `checkout_latest`; past it, the CSR is rebuilt and stored again. + Lance's `cleanup_old_versions` leaves the folder alone. PyArrow inputs aren't + stored, and a dataset that can't be written (read-only storage or credentials) + is built as without it. See [Storing the CSR](csr.md#storing-the-csr). Raises ------ @@ -200,8 +202,8 @@ def execute_batches[ParameterT: Parameter]( Batches are computed as the reader is consumed. Scans, filters and projections stream with bounded memory; joins, sorts and aggregations may still buffer. The reader keeps its snapshot alive, so a concurrent - [`refresh`][lance_csr.CypherEngine.refresh] doesn't affect it. Planning errors - raise immediately; execution errors raise while reading. + [`checkout_latest`][lance_csr.CypherEngine.checkout_latest] doesn't affect it. + Planning errors raise immediately; execution errors raise while reading. Parameters ---------- @@ -257,36 +259,52 @@ def explain[ParameterT: Parameter]( """ return self._inner.explain(query, params) - def refresh(self) -> None: + def checkout_latest(self) -> None: """Move every Lance dataset to its latest version. - Updates the CSR, node indexes and statistics to match. Datasets whose version - and commit time are unchanged keep their open handle. The replacement is - validated in full before it's swapped in: on error nothing changes and the - previous snapshot stays usable. Queries already running finish on the snapshot - they started with. PyArrow inputs are unchanged. Concurrent calls run one at a - time, and `target_partitions` is preserved. See - [Lance datasets](lance.md#refresh). + The engine's counterpart of Lance's `LanceDataset.checkout_latest()`, for every + dataset at once. Updates the CSR, node indexes and statistics to match. Datasets + whose version and commit time are unchanged keep their open handle. The + replacement is validated in full before it's swapped in: on error nothing + changes and the previous snapshot stays usable. Queries already running finish + on the snapshot they started with. PyArrow inputs are unchanged. Concurrent + calls run one at a time, and `target_partitions` is preserved. See [Lance + datasets](lance.md#checkout-latest). + """ + self._inner.checkout_latest() + + def refresh(self) -> None: + """Deprecated alias of `checkout_latest`. + + Behaves exactly like + [`checkout_latest`][lance_csr.CypherEngine.checkout_latest], and warns with + `DeprecationWarning`. """ - self._inner.refresh() + warnings.warn( + "CypherEngine.refresh() is deprecated; use checkout_latest(), named after " + "Lance's LanceDataset.checkout_latest()", + DeprecationWarning, + stacklevel=2, + ) + self.checkout_latest() def optimize_csr(self) -> None: """Rebuild each CSR that carries a delta (experimental; needs `use_csr=True`). The CSR's counterpart of Lance's `optimize_indices()`. - [`refresh`][lance_csr.CypherEngine.refresh] applies Lance writes as a delta - over the base CSR, and rebuilds on its own only once pending changes pass 25% - of the base. This rebuilds every CSR that carries a delta now, at the version - the engine is pinned to, so lookups stop carrying it. It scans every edge - again, so call it after a bulk load or at a quiet time, not after every - refresh. It doesn't move to - newer versions: call `refresh` first for that. CSRs without a delta, and - PyArrow inputs, are left as they are, and query results never change. - Afterwards [`csr_stats`][lance_csr.CypherEngine.csr_stats] reports `update` - `"rebuild"`, `unindexed_edges` 0 and `base_version` equal to `version`. As for - `refresh`, the replacement is validated in full before it's swapped in, and the - two run one at a time. See - [Rebuilding the CSR](csr.md#rebuilding-the-csr). + [`checkout_latest`][lance_csr.CypherEngine.checkout_latest] applies Lance writes + as a delta over the base CSR, and rebuilds on its own only once pending changes + pass 25% of the base. This rebuilds every CSR that carries a delta now, at the + version the engine is pinned to, so lookups stop carrying it. It scans every + edge again, so call it after a bulk load or at a quiet time, not after every + `checkout_latest`. It doesn't move to newer versions: call `checkout_latest` + first for that. CSRs without a delta, and PyArrow inputs, are left as they are, + and query results never change. Afterwards + [`csr_stats`][lance_csr.CypherEngine.csr_stats] reports `update` `"rebuild"`, + `unindexed_edges` 0 and `base_version` equal to `version`. As for + `checkout_latest`, the replacement is validated in full before it's swapped in, + and the two run one at a time. See [Rebuilding the + CSR](csr.md#rebuilding-the-csr). """ self._inner.optimize_csr() @@ -298,7 +316,7 @@ def delete_stored_csr(self) -> int: `persist_csr`. The engine keeps the CSR it already holds in memory, so queries aren't affected. To go back to in-memory builds only, create later engines without `persist_csr`: an engine with it stores a new file at its next full - build. Runs one at a time with `refresh` and `optimize_csr`. See + build. Runs one at a time with `checkout_latest` and `optimize_csr`. See [Storing the CSR](csr.md#storing-the-csr). Returns @@ -325,9 +343,9 @@ def node_index_stats(self) -> dict[str, NodeIndexStats]: def csr_stats(self) -> dict[str, CsrStats]: """Statistics for each CSR adjacency index (experimental; needs `use_csr=True`). - An identical result after [`refresh`][lance_csr.CypherEngine.refresh] means the - snapshot itself didn't change. See - [Inspecting the CSR](csr.md#inspecting-the-csr). + An identical result after + [`checkout_latest`][lance_csr.CypherEngine.checkout_latest] means the snapshot + itself didn't change. See [Inspecting the CSR](csr.md#inspecting-the-csr). Returns ------- @@ -339,7 +357,7 @@ def csr_stats(self) -> dict[str, CsrStats]: 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 or + - `update`: how the last build, `checkout_latest` or [`optimize_csr`][lance_csr.CypherEngine.optimize_csr] produced it: `"build"` (first), `"rebuild"` (full), `"loaded"` (from a file stored with `persist_csr=True`, with any changes since applied as a delta), `"delta"` diff --git a/python/python/tests/test_call_subqueries.py b/python/python/tests/test_call_subqueries.py index 9d4a498..454dfe8 100644 --- a/python/python/tests/test_call_subqueries.py +++ b/python/python/tests/test_call_subqueries.py @@ -413,7 +413,7 @@ def test_rich_path_reader_retains_snapshot_after_refresh(graph, tmp_path): lance.write_dataset( pa.table({"src": [20], "dst": [30]}), paths["KNOWS"], mode="append" ) - engine.refresh() + engine.checkout_latest() assert engine.execute(query, {"start": 20}).column(0).to_pylist() == [ [20, 30], [20, 40], diff --git a/python/python/tests/test_engine_csr.py b/python/python/tests/test_engine_csr.py index 28aa28e..edb3c59 100644 --- a/python/python/tests/test_engine_csr.py +++ b/python/python/tests/test_engine_csr.py @@ -285,7 +285,7 @@ def test_csr_tracks_edge_changes_but_not_node_changes(tmp_path): tmp_path / "person.lance", mode="append", ) - engine.refresh() + engine.checkout_latest() assert engine.csr_stats() == before new_edge = pa.table( @@ -297,7 +297,7 @@ def test_csr_tracks_edge_changes_but_not_node_changes(tmp_path): ) lance.write_dataset(new_edge, tmp_path / "knows.lance", mode="append") assert engine.execute(query).num_rows == 0 # pinned snapshot - engine.refresh() + engine.checkout_latest() after = engine.csr_stats() assert after["KNOWS"]["version"] == 2 and after["KNOWS"]["edges"] == 8 assert engine.execute(query).column(0).to_pylist() == [1] @@ -337,7 +337,7 @@ def test_high_degree_stream_keeps_its_snapshot_across_deletions(tmp_path): assert 0 < first.num_rows < n # A live reader retains the old adjacency, including both copies of 0->1. knows.delete("dst = 1") - engine.refresh() + engine.checkout_latest() assert engine.csr_stats()["KNOWS"]["tombstones"] == 2 counts = {} for batch in chain([first], reader): @@ -429,7 +429,7 @@ def test_node_index_rebuilt_only_when_node_version_changes(tmp_path): } ) lance.write_dataset(edge, tmp_path / "knows.lance", mode="append") - engine.refresh() + engine.checkout_latest() assert engine.node_index_stats() == before query = "MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.id = 6 RETURN b.name" assert engine.execute(query).num_rows == 0 # node 7 does not exist yet @@ -439,7 +439,7 @@ def test_node_index_rebuilt_only_when_node_version_changes(tmp_path): tmp_path / "person.lance", mode="append", ) - engine.refresh() + engine.checkout_latest() after = engine.node_index_stats() assert after["Person"]["version"] == 2 and after["Person"]["rows"] == 7 assert engine.execute(query).column(0).to_pylist() == ["g"] diff --git a/python/python/tests/test_engine_csr_delta.py b/python/python/tests/test_engine_csr_delta.py index 6c928eb..345b222 100644 --- a/python/python/tests/test_engine_csr_delta.py +++ b/python/python/tests/test_engine_csr_delta.py @@ -3,7 +3,7 @@ """Incremental CSR maintenance: edge commits applied as deltas and tombstones. -After each change and ``refresh()``, the CSR engine must return exactly what a +After each change and ``checkout_latest()``, the CSR engine must return exactly what a fresh join-path engine and a freshly built CSR engine return on the new data. """ @@ -68,7 +68,7 @@ def make(**write_kwargs): def check(engine, knows_uri): - """CSR engine (after refresh) == fresh join engine == fresh CSR engine.""" + """CSR engine (after checkout_latest) == fresh join engine == fresh CSR engine.""" datasets = { "Person": lance.dataset(knows_uri.parent / "person.lance"), "KNOWS": lance.dataset(knows_uri), @@ -95,7 +95,7 @@ def test_initial_build_reports_build(graph): def test_append_applied_as_delta(graph): engine, uri = graph() lance.write_dataset(edges([(1, 2), (39, 0), (5, 5)], 1000), uri, mode="append") - engine.refresh() + engine.checkout_latest() s = stats(engine) assert (s["update"], s["delta_edges"], s["tombstones"]) == ("delta", 3, 0) assert s["edges"] == 203 and s["unindexed_edges"] == 3 @@ -106,7 +106,7 @@ def test_append_applied_as_delta(graph): def test_delete_applied_as_tombstones(graph): engine, uri = graph() lance.dataset(uri).delete("since % 9 = 0") - engine.refresh() + engine.checkout_latest() s = stats(engine) assert s["update"] == "delta" and s["tombstones"] == 23 and s["delta_edges"] == 0 check(engine, uri) @@ -116,7 +116,7 @@ def test_delete_one_of_duplicate_edges(graph): engine, uri = graph() # since 190..199 are the duplicated i -> i+1 edges; delete one copy of 2 -> 3. lance.dataset(uri).delete("since = 192") - engine.refresh() + engine.checkout_latest() query = "MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.id = 2 RETURN b.id" expected = sorted(d for s, d in BASE if s == 2) expected.remove(3) @@ -127,7 +127,7 @@ def test_delete_one_of_duplicate_edges(graph): def test_delete_whole_fragment(graph): engine, uri = graph() lance.dataset(uri).delete("since >= 40 AND since < 80") - engine.refresh() + engine.checkout_latest() assert stats(engine)["tombstones"] == 40 check(engine, uri) @@ -135,7 +135,7 @@ def test_delete_whole_fragment(graph): def test_update_endpoint_values(graph): engine, uri = graph() lance.dataset(uri).update({"dst": "0"}, where="since < 15") - engine.refresh() + engine.checkout_latest() assert stats(engine)["update"] == "delta" check(engine, uri) @@ -144,7 +144,7 @@ def test_non_endpoint_column_change_reuses_csr(graph): engine, uri = graph() before = stats(engine) lance.dataset(uri).add_columns({"weight": "since * 2"}) - engine.refresh() + engine.checkout_latest() after = stats(engine) assert after["update"] == "reused" and after["edges"] == before["edges"] assert after["version"] > before["version"] @@ -154,11 +154,11 @@ def test_non_endpoint_column_change_reuses_csr(graph): def test_sequential_deltas_accumulate(graph): engine, uri = graph() lance.write_dataset(edges([(7, 8), (7, 9)], 2000), uri, mode="append") - engine.refresh() + engine.checkout_latest() lance.dataset(uri).delete("since = 2000 OR since = 0") - engine.refresh() + engine.checkout_latest() lance.write_dataset(edges([(7, 8)], 3000), uri, mode="append") - engine.refresh() + engine.checkout_latest() 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. @@ -170,7 +170,7 @@ def test_optimize_folds_delta_into_base(graph): engine, uri = graph() lance.write_dataset(edges([(7, 8), (7, 9)], 2000), uri, mode="append") lance.dataset(uri).delete("since = 2000 OR since = 0") - engine.refresh() + engine.checkout_latest() before = stats(engine) assert before["update"] == "delta" and before["unindexed_edges"] == 1 @@ -210,7 +210,7 @@ def test_compaction_keeps_results(graph): ds = lance.dataset(uri) ds.delete("since < 30") ds.optimize.compact_files(target_rows_per_fragment=1000) - engine.refresh() + engine.checkout_latest() check(engine, uri) @@ -219,14 +219,14 @@ def test_deltas_survive_cleanup_of_old_versions(graph): engine, uri = graph() lance.write_dataset(edges([(3, 4), (4, 3)], 4000), uri, mode="append") lance.write_dataset(edges([(5, 6), (6, 5)], 4002), uri, mode="append") - engine.refresh() + engine.checkout_latest() # Compaction rewrites the two small fragments; cleanup deletes their files. lance.dataset(uri).optimize.compact_files(target_rows_per_fragment=40) lance.dataset(uri).delete("since = 0 OR since = 4003") lance.dataset(uri).cleanup_old_versions( older_than=timedelta(0), delete_unverified=True ) - engine.refresh() + engine.checkout_latest() s = stats(engine) assert (s["update"], s["delta_edges"], s["tombstones"]) == ("delta", 7, 5) check(engine, uri) @@ -235,7 +235,7 @@ def test_deltas_survive_cleanup_of_old_versions(graph): def test_non_endpoint_column_change_is_visible_through_relationship(graph): engine, uri = graph() lance.dataset(uri).add_columns({"weight": "since * 2"}) - engine.refresh() + engine.checkout_latest() assert stats(engine)["update"] == "reused" got = engine.execute( "MATCH (a:Person)-[r:KNOWS]->(b:Person) RETURN r.since, r.weight" @@ -247,7 +247,7 @@ def test_non_endpoint_column_change_is_visible_through_relationship(graph): def test_overwrite_triggers_rebuild(graph): engine, uri = graph() lance.write_dataset(edges([(1, 2), (2, 3)]), uri, mode="overwrite") - engine.refresh() + engine.checkout_latest() s = stats(engine) assert (s["update"], s["edges"], s["delta_edges"]) == ("rebuild", 2, 0) check(engine, uri) @@ -257,7 +257,7 @@ def test_large_delta_triggers_rebuild(graph): engine, uri = graph() many = [(i % N, (i + 3) % N) for i in range(100)] # 50% of the base lance.write_dataset(edges(many, 5000), uri, mode="append") - engine.refresh() + engine.checkout_latest() 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"] @@ -269,6 +269,6 @@ def test_stable_row_ids_with_deletes_and_appends(graph): ds = lance.dataset(uri) ds.delete("since % 5 = 1") lance.write_dataset(edges([(0, 39)], 9000), uri, mode="append") - engine.refresh() + engine.checkout_latest() assert stats(engine)["update"] == "delta" check(engine, uri) diff --git a/python/python/tests/test_engine_lance.py b/python/python/tests/test_engine_lance.py index 0c81356..7c14962 100644 --- a/python/python/tests/test_engine_lance.py +++ b/python/python/tests/test_engine_lance.py @@ -1,10 +1,10 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright The Lance Authors -"""CypherEngine over on-disk Lance datasets: snapshots, refresh and streaming. +"""CypherEngine over on-disk Lance datasets: snapshots, checkout_latest() and streaming. The engine pins the dataset version it was given; writes become visible only -after ``refresh()``. Expected values are computed independently of the engine. +after ``checkout_latest()``. Expected values are computed independently of the engine. """ import re @@ -109,16 +109,25 @@ def test_snapshot_until_refresh_after_append(graph): lance.write_dataset(people(N, N + 5), path / "person.lance", mode="append") assert count_people(engine) == N - engine.refresh() + engine.checkout_latest() assert count_people(engine) == N + 5 assert engine.dataset_versions()["Person"] == 2 +def test_refresh_is_a_deprecated_alias_of_checkout_latest(graph): + path, person, knows = graph + engine = CypherEngine(CONFIG, {"Person": person, "KNOWS": knows}) + lance.write_dataset(people(N, N + 5), path / "person.lance", mode="append") + with pytest.warns(DeprecationWarning, match="checkout_latest"): + engine.refresh() + assert count_people(engine) == N + 5 + + def test_refresh_without_changes_keeps_results(graph): _, person, knows = graph engine = CypherEngine(CONFIG, {"Person": person, "KNOWS": knows}) before = engine.execute("MATCH (p:Person) WHERE p.id < 3 RETURN p.name") - engine.refresh() + engine.checkout_latest() after = engine.execute("MATCH (p:Person) WHERE p.id < 3 RETURN p.name") assert before.equals(after) assert engine.dataset_versions() == {"Person": 1, "KNOWS": 1} @@ -136,7 +145,7 @@ def test_refresh_sees_overwrite_with_added_and_retyped_columns(graph): lance.write_dataset(replacement, path / "person.lance", mode="overwrite") assert count_people(engine) == N # still pinned - engine.refresh() + engine.checkout_latest() out = engine.execute("MATCH (p:Person) RETURN p.name, p.age") assert out.column(0).type == pa.int64() assert sorted(out.column(1).to_pylist()) == [30, 40] @@ -149,7 +158,7 @@ def test_refresh_sees_table_replaced_at_same_location(graph): # Delete and recreate: the new dataset is version 1 again, like the old one. shutil.rmtree(path / "person.lance") lance.write_dataset(people(0, 3), path / "person.lance") - engine.refresh() + engine.checkout_latest() assert count_people(engine) == 3 @@ -163,7 +172,7 @@ def test_failed_refresh_keeps_previous_snapshot(graph, use_csr): pa.table({"name": ["x"]}), path / "person.lance", mode="overwrite" ) with pytest.raises(ValueError, match="id"): - engine.refresh() + engine.checkout_latest() assert count_people(engine) == N assert engine.dataset_versions()["Person"] == 1 @@ -185,8 +194,8 @@ def test_historical_version_is_pinned_and_engines_independent(graph): assert count_people(old) == N assert count_people(new) == N + 5 - # refresh() moves a historical snapshot to the latest version. - old.refresh() + # checkout_latest() moves a historical snapshot to the latest version. + old.checkout_latest() assert count_people(old) == N + 5 @@ -201,7 +210,7 @@ class Wrapped(lance.LanceDataset): # Treated as Lance (pinned, refreshable), not materialized as Arrow. assert engine.dataset_versions()["Person"] == 1 lance.write_dataset(people(N, N + 1), path / "person.lance", mode="append") - engine.refresh() + engine.checkout_latest() assert count_people(engine) == N + 1 @@ -224,7 +233,7 @@ def reader(): lance.write_dataset( people(N + 5 * (i - 1), N + 5 * i), path / "person.lance", mode="append" ) - engine.refresh() + engine.checkout_latest() for t in threads: t.join() @@ -264,7 +273,7 @@ def test_execute_batches_keeps_snapshot_across_refresh(graph, use_csr): first = reader.read_next_batch() lance.write_dataset(people(N, N + 5), path / "person.lance", mode="append") - engine.refresh() + engine.checkout_latest() remaining = reader.read_all() assert first.num_rows + remaining.num_rows == N diff --git a/python/python/tests/test_path_values.py b/python/python/tests/test_path_values.py index 6a4baf4..c21e391 100644 --- a/python/python/tests/test_path_values.py +++ b/python/python/tests/test_path_values.py @@ -274,7 +274,7 @@ def test_path_inventory_keeps_lance_snapshot(tmp_path): lance.write_dataset( pa.table({"src": [1], "dst": [2], "weight": [7]}), edges_path, mode="overwrite" ) - engine.refresh() + engine.checkout_latest() assert reader.read_all().to_pylist() == [ {"names": ["old", "target"], "weights": [5]} ] @@ -599,7 +599,7 @@ def test_cross_label_shortest_inventory_isolates_refresh(cross_label_graph, tmp_ lance.write_dataset( pa.table({"src": [1], "dst": [2]}), paths["LIVES"], mode="overwrite" ) - engine.refresh() + engine.checkout_latest() assert reader.read_all().to_pylist() == [{"ids": [1, 1]}] for start in [1, 2, 1]: assert engine.execute(query, {"start": start}).to_pylist() == ( diff --git a/python/python/tests/test_target_partitions.py b/python/python/tests/test_target_partitions.py index 2bc0f1e..676ad61 100644 --- a/python/python/tests/test_target_partitions.py +++ b/python/python/tests/test_target_partitions.py @@ -58,7 +58,7 @@ def test_partition_targets_preserve_queries_and_refresh( lance.write_dataset( pa.table({"id": [5]}), tmp_path / "Person.lance", mode="append" ) - engine.refresh() + engine.checkout_latest() assert reader.read_all().to_pylist() == [{"id": value} for value in range(1, 5)] count = 5 if storage == "lance" else 4 assert engine.execute(scan).to_pylist() == [ diff --git a/python/python/tests/test_typing.py b/python/python/tests/test_typing.py index 87b21cc..21378ea 100644 --- a/python/python/tests/test_typing.py +++ b/python/python/tests/test_typing.py @@ -190,7 +190,7 @@ def test_stub_check_detects_missing_method(): node.body = [ item for item in node.body - if not isinstance(item, ast.FunctionDef) or item.name != "refresh" + if not isinstance(item, ast.FunctionDef) or item.name != "checkout_latest" ] with pytest.raises(AssertionError, match="public members differ"): check_class(node, _internal.graph.CypherEngine) diff --git a/python/python/tests/typing/public_api.py b/python/python/tests/typing/public_api.py index 539614f..b2b3c36 100644 --- a/python/python/tests/typing/public_api.py +++ b/python/python/tests/typing/public_api.py @@ -54,7 +54,7 @@ def public_api( assert_type(engine.execute(query, params), Table) assert_type(engine.execute_batches(query, params), RecordBatchReader) assert_type(engine.explain(query, params), str) - assert_type(engine.refresh(), None) + assert_type(engine.checkout_latest(), None) assert_type(engine.config(), GraphConfig) assert_type(engine.dataset_versions(), dict[str, int]) assert_type(engine.node_index_stats()["person"]["rows"], int) From 4500e39f0d38fbfc475c081d53e013299b07a4ff Mon Sep 17 00:00:00 2001 From: prrao87 Date: Mon, 2 Nov 2026 12:34:30 -0500 Subject: [PATCH 2/2] docs: say checkout_latest() follows lance and optimize_csr() rebuilds the csr --- docs/csr.md | 7 +++++++ docs/lance.md | 5 ++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/docs/csr.md b/docs/csr.md index 838373f..d6e0c80 100644 --- a/docs/csr.md +++ b/docs/csr.md @@ -304,6 +304,13 @@ Two calls bring it back up to date, and they do different jobs: 2. **[`optimize_csr()`](#rebuilding-the-csr)** rebuilds a CSR from scratch, so it no longer needs its delta. Call it occasionally, once enough changes have built up that you'd rather lookups stopped working through them. +The two calls aren't a matched pair, though. `checkout_latest()` isn't really a CSR operation: it's named after, and +works like, Lance's own `LanceDataset.checkout_latest()`, and it moves the whole engine to the newest version of every +Lance dataset it reads. Node tables, node indexes and the statistics used to plan joins all move with it, and keeping +the CSR in step is just one part of that, done as cheaply as possible with a delta. `optimize_csr()` is the call that's +about the CSR itself: it rebuilds a CSR in a running engine when you decide, rather than waiting for the delta to get +too large. It plays the part Lance's `optimize_indices()` plays for Lance's own indexes. + If you gave the engine in-memory PyArrow tables instead of Lance datasets, neither call applies: those tables can't change underneath the engine, so their CSR is never out of date. To query new in-memory data, create a new engine. diff --git a/docs/lance.md b/docs/lance.md index b9b3ca8..9c1c698 100644 --- a/docs/lance.md +++ b/docs/lance.md @@ -51,7 +51,10 @@ stay invisible until you call `checkout_latest()`. That gives every query a stab ## `checkout_latest()` { #checkout-latest } `checkout_latest()` moves every Lance dataset to its latest version. It's named after Lance's own -`LanceDataset.checkout_latest()`, which does the same for a single dataset: +`LanceDataset.checkout_latest()`, which does the same for a single dataset, and it updates everything the engine +built from that data along with it: node indexes, join statistics and, with `use_csr=True`, the CSR. It isn't a +CSR-specific call; to rebuild a CSR, use [`optimize_csr()`](csr.md#rebuilding-the-csr) (see +[Keeping the CSR current](csr.md#keeping-the-csr-current)): ```python lance.write_dataset(