Using lance-csr from Rust¶

lance-csr is a Rust library first. The Python package is a thin wrapper around the same engine, so everything on the other pages (the graph schema, the supported Cypher, the CSR index and version pinning) applies to Rust too. This page covers what's different: installing the crate, the async API, and the Rust types.

Every Rust snippet on this page comes from examples/embedded_query.rs, which CI compiles and runs. Run it yourself from a checkout:

cargo run -p lance-csr --example embedded_query

Install¶

The engine is published on crates.io as lance-csr:

cargo add lance-csr

Building it needs protoc, the Protocol Buffers compiler, for Lance's dependencies (see Contributing). The engine is async, so your program also needs a runtime such as Tokio.

Build an engine¶

CypherEngine is the same engine as in Python. It registers your datasets once, optionally builds the CSR, and serves many queries:

let config = GraphConfig::builder()
    .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.checkout_latest()`.
let sources = HashMap::from([
    (
        "Person".to_string(),
        EngineSource::open_lance(person_uri, None, HashMap::new()).await?,
    ),
    (
        "KNOWS".to_string(),
        EngineSource::open_lance(knows_uri.clone(), None, HashMap::new()).await?,
    ),
]);
let engine = CypherEngine::new(config, sources, true).await?;

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 checkout_latest() can reopen it with the same credentials. See Lance datasets.

EngineSource::Arrow(batch)

An in-memory Arrow RecordBatch. It never changes; build a new engine to replace it.

The third argument of CypherEngine::new is use_csr. With true, relationship types with integer endpoints get CSR adjacencies and node labels with unique integer ids get a node index, exactly as with use_csr=True in Python. CypherEngine::with_target_partitions(config, sources, use_csr, Some(n)) also sets DataFusion's query partition target (1 to 1024), like target_partitions= in Python; see Query parallelism.

Run queries¶

execute returns the whole result as one RecordBatch. Parameters are a HashMap of serde_json::Values, one per $name in the query; pass HashMap::new() when there are none:

let query = "MATCH (a:Person {name: $name})-[:KNOWS]->()-[:KNOWS]->(c:Person)
             RETURN c.name AS friend_of_friend ORDER BY friend_of_friend";
let params = || HashMap::from([("name".to_string(), serde_json::json!("Alice"))]);
let result = engine.execute(query, params()).await?;
print_batches(&[result])?;
// +------------------+
// | friend_of_friend |
// +------------------+
// | Carol            |
// | David            |
// +------------------+

The other query methods take the same arguments:

Each query runs on the snapshot that was current when it started.

Pick up new writes¶

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):

// 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.checkout_latest().await?;
let result = engine.execute(query, params()).await?;
print_batches(&[result])?; // Alice, Carol, David

let stats = &engine.csr_stats()["KNOWS"];
println!(
    "{} edges, update={}, version={:?}",
    stats.edges, stats.update, stats.version
);
// 5 edges, update=delta, version=Some(2)

csr_stats() and node_index_stats() return the same fields as in Python, as CsrStats and NodeIndexStats structs keyed by relationship type and node label. dataset_versions() gives the pinned version of each Lance dataset.

Errors¶

Every method returns lance_csr::Result<T>, whose error is GraphError. The Python package raises a different exception depending on the variant:

Async runtimes and cancellation¶

• 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.

Timeouts¶

lance_csr::query::with_execution_deadline runs a query with a cooperative deadline, the same mechanism behind Python's timeout=. Inside it, nested-loop joins (joins with non-equality conditions) check the deadline while evaluating candidates and fail with a Query timeout exceeded error, even when no rows match, and cross joins yield to the runtime between batches. Yielding only lets a timer stop the work, and other operators aren't checked, so also race the query against the same deadline with tokio::time::timeout_at:

// One deadline for both: checks inside nested-loop joins, and a timer that stops
// everything else (cross joins yield to it between batches).
let deadline = Instant::now() + Duration::from_secs(5);
let pairs = engine.execute(
    "MATCH (a:Person), (b:Person) RETURN count(*) AS pairs",
    HashMap::new(),
);
match timeout_at(deadline, with_execution_deadline(Some(deadline), pairs)).await {
    Ok(result) => print_batches(&[result?])?,
    Err(_) => println!("query timed out"),
}
// +-------+
// | pairs |
// +-------+
// | 16    |
// +-------+

The deadline applies only to work inside the call; it isn't stored in the engine. A timed-out query surfaces as Err(Elapsed) from timeout_at, or as an execution error from a join checkpoint; treat both as a timeout.

Graph mappings and pinned snapshots¶

An application that keeps a graph as plain Lance tables, one per node label and relationship type, has to record which table plays which role and which version of each table it read. The mapping module holds both, in a JSON format that applications can store next to their tables or in a publication manifest:

• GraphMapping is the graph descriptor: per label, its table and key column, and per relationship type, its source and target columns and the node labels they point to. validate() checks what needs no table data: version 1, at least one node label, at most 64 labels, labels that are ASCII identifiers and unique ignoring case, and one graph role per table.
• GraphSnapshot pins each table, by name, to a path and Lance version, and carries the mapping. It ignores fields it doesn't know, so it reads a manifest that also lists row counts or schemas.

GraphSnapshot::open opens each mapped table at its pinned version and returns the GraphConfig and sources that CypherEngine::new takes. Building the configuration also checks the tables themselves: the key, display and endpoint columns exist, keys are scalar, and each endpoint column has the type of the node key it refers to.

use std::collections::HashMap;
use lance_csr::{CypherEngine, GraphSnapshot};

let snapshot: GraphSnapshot = serde_json::from_str(&manifest)?;
let (config, sources) = snapshot.open(&HashMap::new()).await?;
let engine = CypherEngine::new(config, sources, true).await?;

A relative path in a manifest is relative to wherever the application keeps its data, so resolve it before calling open. To open one table at a pinned version without a mapping, use lance_csr::open_lance_dataset(uri, Some(version), &storage_options).

CypherQuery::execute_with_namespace accepts any Lance namespace, such as Lance's own DirectoryNamespace as well as DirNamespace. It opens each table at the version and with the storage options that the namespace's describe_table returns, and at the latest version when it returns none, which is always the case for DirNamespace.

Python and Rust side by side¶

CypherQuery (one query over in-memory batches), SqlEngine (with SqlEngine::from_engine(Arc<CypherEngine>)), SqlQuery and DirNamespace also exist in Rust. The API documentation on docs.rs lists every type and method.

Performance¶

The engine code is the same whichever language calls it. The benchmarks were all measured through the Python package; there are no separate Rust benchmark results.