Repository navigation
Understanding Lance‐CSR
Note
Current as of release version v0.5.1, published 1 November 2026
lance-csr is an openCypher engine for graphs whose nodes and relationships are ordinary Lance datasets (or Arrow tables). It is built on Apache DataFusion, and its goal is fast multi-hop traversal over data that already lives in Lance: no import into a graph database, no second copy of the data, no server.
DataFusion is an excellent relational engine, but a graph hop expressed relationally is two hash joins, and a five-hop query is ten of them. lance-csr keeps DataFusion for everything it already does well (filters, projections, aggregates, sorts, joins, vectorized multi-threaded execution) and adds graph-aware pieces exactly where relational plans fall over. The CSR adjacency index is a means to that end: it is the data structure the custom operators read.
This page is a tour of those extensions. For user-facing documentation see the docs site, in particular How it works and The CSR index.
- Design principles
- The query pipeline
- Every DataFusion extension point we use
- The custom operators
- Logical optimizer rules
- Graph rewrites before DataFusion sees the plan
- Join statistics for graph workloads
- Indexes that live next to Lance
- Execution plumbing
- Benchmarks
- Source map
-
The data stays in Lance. Tables are read through Lance's own
LanceTableProvider. The graph indexes store Lance row addresses and row ids, so a hop fetches exactly the rows it needs withtake, never a full scan, and an index never holds a copy of node or relationship properties. - Every graph operator is a DataFusion extension node. It sits inside an ordinary DataFusion plan, so everything around it is still planned, optimized and executed by DataFusion.
- Results never depend on the fast path. Any hop the CSR can't express is planned as a hash join in the same plan. The test suite runs each supported shape with the CSR on and off and compares the answers.
- 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 queries never see a half-updated index.
flowchart TD
T["Cypher text"] --> P["Parse<br/><code>parser.rs</code>"]
P --> S["Semantic analysis<br/><code>semantic.rs</code><br/>scopes, labels, $parameters"]
S --> R["Graph rewrites<br/><code>pattern_order.rs</code><br/><code>pattern_alternatives.rs</code>"]
R --> G["Graph logical plan<br/><code>logical_plan.rs</code><br/>ScanByLabel, Expand, VariableLengthExpand,<br/>ShortestPath, PathMatch, OptionalMatch, Apply"]
G --> D["DataFusion logical plan<br/><code>datafusion_planner/</code><br/>scans + joins + <b>extension nodes</b>"]
D --> O["DataFusion optimizer<br/>+ PushCsrTargetFilters<br/>+ PruneCsrTargetColumns"]
O --> X["Physical planning<br/>CsrQueryPlanner → CsrExtensionPlanner"]
X --> E["Execution<br/>CsrExpandExec, NodeLookupExec, ShortestPathExec,<br/>PathMatchExec, ApplyExec, KeyStatisticsExec"]
E --> B["Arrow record batches"]
classDef ours fill:#e8f1ff,stroke:#3b6fd8,color:#0b2a6b;
class R,G,O,X,E ours;
The highlighted stages are where lance-csr adds its own machinery. engine.explain(query) prints the graph logical
plan, the DataFusion logical plan and the physical plan, and every custom operator shows up in it by name.
| DataFusion trait | lance-csr implementation | Role |
|---|---|---|
UserDefinedLogicalNodeCore |
CsrExpandNode, NodeLookupNode, ShortestPathNode, PathMatchNode, ApplyNode
|
Graph operators as logical plan nodes |
ExecutionPlan |
CsrExpandExec, NodeLookupExec, ShortestPathExec, PathMatchExec, ApplyExec, KeyStatisticsExec
|
Their physical operators |
ExtensionPlanner |
CsrExtensionPlanner |
Lowers each graph node to its ExecutionPlan
|
QueryPlanner |
CsrQueryPlanner |
Installs the extension planner on the session |
OptimizerRule |
PushCsrTargetFilters, PruneCsrTargetColumns
|
Filter pushdown into, and column pruning of, extension nodes |
PhysicalOptimizerRule |
DeadlineJoins (query_deadline_joins) |
Cooperative deadlines in nested-loop and cross joins |
TableProvider |
KeyStatisticsProvider |
Reports distinct counts of join keys so DataFusion sizes graph joins correctly |
ScalarUDFImpl |
cypher_list, cypher_field, Cypher math and scalar functions, vector distance UDFs |
Cypher semantics DataFusion's built-ins don't match |
The session is assembled once per snapshot:
SessionStateBuilder::new()
.with_default_features()
.with_config(session_config)
.with_query_planner(Arc::new(CsrQueryPlanner))
// Move target-only filters into CSR nodes first, so pruning can then drop
// the filter columns from the take.
.with_optimizer_rule(Arc::new(PushCsrTargetFilters))
.with_optimizer_rule(Arc::new(PruneCsrTargetColumns))
.build()Rust callers that bring their own SessionContext install CsrQueryPlanner the same way; contexts created by
CypherQuery and CypherEngine do it themselves.
The core operator. Without it, (a:Person)-[:KNOWS]->(b:Person) is a hash join of a against the whole KNOWS
table and a second hash join against the whole Person table to fetch b. With it, each input row looks up
a's neighbours in the CSR and fetches b's rows by position.
flowchart LR
subgraph join ["Relational hop: two hash joins"]
direction TB
S1["Scan Person (a)"] --> HJ1["HashJoin<br/>a.id = KNOWS.src"]
K1["Scan KNOWS"] --> HJ1
HJ1 --> HJ2["HashJoin<br/>KNOWS.dst = b.id"]
S2["Scan Person (b)"] --> HJ2
end
subgraph csr ["lance-csr hop: one operator"]
direction TB
S3["Scan Person (a)"] --> CE["CsrExpandExec<br/>neighbours(a.id) from the CSR,<br/>take b rows by position"]
end
join ~~~ csr
Here is the same two-hop query planned both ways, straight from explain() (lance-csr 0.5.1, a four-person toy graph):
MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(c:Person {name: 'Dan'}) RETURN a.nameuse_csr=False use_csr=True
───────────── ────────────
ProjectionExec ProjectionExec
HashJoinExec knows_2.src = a.id FilterExec (relationship uniqueness)
HashJoinExec b.id = knows_2.dst ProjectionExec
HashJoinExec knows_1.src = b.id CsrExpandExec: key=b__id, take=a[a__id*, a__name]
HashJoinExec c.id = knows_1.dst CsrExpandExec: key=c__id, take=b[b__id*]
FilterExec name = Dan → scan Person (c) FilterExec name = Dan → scan Person (c)
scan KNOWS
scan Person (b)
scan KNOWS
scan Person (a)
Four hash joins and five scans become two CsrExpandExecs over one filtered scan. Note that both plans start
from c (Dan), not a: that is pattern re-rooting at work.
What one CsrExpandExec can do, and how each shows up in explain():
| Capability | explain() |
What it replaces |
|---|---|---|
Neighbour lookup in one direction, or both for -[:R]-
|
key=a__id, edges=N, undirected
|
Join against the relationship table |
Fused target lookup: neighbour id → row position → take of only the columns the query reads |
take=b[b__name] |
Join against the target node table |
| Relationship properties fetched by edge row (each CSR entry records the Lance row address its edge came from) | edge=k[k__since] |
Keeping the relationship table in the join |
| Pushed target filter: reads the filter's columns first, fetches the rest only for passing neighbours | filter=age > 35 |
FilterExec after a full fetch |
| Pushed relationship filter: drops failing edges before anything else is fetched | edge_filter=since > 2020 |
FilterExec after the join |
| Closing edge: target already bound (a cycle), so keep only neighbours equal to it | match=b__id |
Expand everything, then filter |
| Iterative variable-length traversal: one cursor per hop, typed (label, key) frontier, bounded output batches | iterative hops=1..3 |
One join branch per path length, UNION ALLed |
A hop that uses most of these at once:
MATCH (a:Person {id: 1})-[k:KNOWS]->(b:Person) WHERE k.since > 2020 AND b.age > 35 RETURN b.name
ProjectionExec: expr=[b__name@3 as b.name]
CsrExpandExec: key=a__id, edges=4, take=b[b__name], edge=k[], filter=age@0 > 35, edge_filter=since@0 > 2020
RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
ProjectionExec: expr=[id@0 as a__id]
FilterExec: id@0 = 1
KeyStatisticsExec: distinct=[id=4]
DataSourceExec: partitions=1, partition_sizes=[1]
And a variable-length path, which is a single operator instead of three unrolled branches:
MATCH (a:Person {id: 1})-[:KNOWS*1..3]->(b:Person) RETURN b.name
ProjectionExec: expr=[b__name@3 as b.name]
CsrExpandExec: key=a__id, edges=4, take=b[b__name], iterative hops=1..3
...
Some details that matter for performance:
- Adaptive I/O for lazy Lance tables. Fetching scattered rows is cheap for small hops but, for a hop touching a large share of a table, each batch of scattered rows reads most of a column's pages. Once a relationship filter has expanded more than 1/256 of the relationship's edges, the operator reads the filter's columns once, sequentially, builds a per-fragment bitset, and checks edges against it from then on. Target-column fetches switch to one sequential read the same way. In-memory Arrow tables always gather directly.
- Bounded memory per partition. Each partition holds one input batch and at most one output chunk of neighbour positions (sized by DataFusion's batch size) before fetching properties, and yields mid-way through a high-degree node's neighbour list.
-
Fast hashing on the hot path. Integer-keyed maps on the per-edge lookup path use
ahash(what Arrow and DataFusion use); SipHash dominated expansion time in profiles. -
Predicate pushdown boundary. The node declares its own output columns in
prevent_predicate_push_down_columns, so DataFusion pushes filters on input columns below it and keeps filters on the columns it creates above it.
Eligibility: integer endpoint columns, exactly one relationship type per hop (alternation is split into branches first), and no inline relationship property map. Anything else stays a join.
WITH b, count(*) AS n groups by the node b, which relationally means grouping by every one of b's columns.
Rows before a grouping can outnumber nodes many times over (they come out of an expansion), so carrying b's
properties through the aggregate for each of them is expensive. When b's label has a node index (unique ids), the
planner groups by b's id alone and adds a NodeLookup above the aggregate that fetches b's other properties once
per group:
MATCH (a:Person)-[:KNOWS]->(b:Person) WITH b, count(*) AS n RETURN b.name, n
ProjectionExec: expr=[b__name@2 as b.name, __agg_0@1 as n]
NodeLookupExec: key=b__id, take=b[b__name]
AggregateExec: mode=FinalPartitioned, gby=[b__id@0 as b__id], aggr=[count(1) as __agg_0]
RepartitionExec: partitioning=Hash([b__id@0], 10), input_partitions=10
AggregateExec: mode=Partial, gby=[b__id@0 as b__id], aggr=[count(1) as __agg_0]
RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
ProjectionExec: expr=[b__id@3 as b__id]
CsrExpandExec: key=a__id, edges=4, take=b[b__id*, b__name]
...
Unlike a join back to the node table, its cost follows the number of groups, not the table's size. This applies to in-memory node tables; for a lazily read Lance table the properties were already read by an earlier step, so the planner keeps grouping by them rather than reading them from disk twice.
shortestPath and allShortestPaths run as a breadth-first search over the same CSR adjacency (or, without a CSR, over
an adjacency built once per query from a scan of the relationship's endpoint columns and shared by all partitions).
- A source with one target searches from both ends using the outgoing and incoming CSR, expanding the smaller frontier and finishing each layer before joining paths, so every shortest alternative is kept.
- Several targets for the same source share one forward search.
- Predecessors retain every shortest path, including parallel edges.
- The operator appends
<path>__lengthand<path>__nodesto each input row, once per path; rows with no path are dropped (MATCH) or kept with nulls through a left join (OPTIONAL MATCH).
ShortestPathExec: a__id -> b__id, Undirected, hops 1..4294967295, one
CrossJoinExec
... scan a ...
... scan b ...
CSR neighbours don't carry edge identity or properties, but MATCH p = ... with relationships(p) needs them, and
parallel edges must stay distinct even through collect() and UNWIND. PathMatchExec builds one immutable,
query-local edge inventory in which an edge's identity is its actual scanned occurrence, enumerates trails
resumably over it, and reuses the bidirectional BFS on a bipartite node/edge graph for shortest paths. Path values are
Arrow values with properties kept in a marked child, so user property names can't collide with identity metadata.
A read-only CALL { ... } body must run once per incoming row: its LIMIT and aggregates apply per row, not globally.
DataFusion has no operator for that, so the body is kept as an unoptimized child template. For each outer row,
ApplyExec binds the explicit imports to concrete values, plans the child, and streams its batches. It holds one outer
batch and one child stream per partition, and dropping the result stream drops both.
A transparent wrapper over a table scan that reports distinct counts for join keys (see
Join statistics). Filters, projections, limits and sort pushdown pass
through to the inner scan unchanged, like DataFusion's own CooperativeExec. It shows in every plan as
KeyStatisticsExec: distinct=[id=4].
Two DataFusion OptimizerRules exist because the stock optimizer treats extension nodes as opaque.
DataFusion can push a filter below a join, but not into an extension node. This rule splits a Filter above a
CsrExpand into conjuncts (looking through column-only projections that projection pushdown may have left between
them) and moves:
- conjuncts that read only the fused target's columns into
target_filter; - conjuncts that read only the named relationship's own columns into
edge_filter.
Mixed or volatile conjuncts stay where they were. The operator then reads just the filter's columns for each batch of candidates, evaluates the predicate, and fetches the remaining columns only for rows that pass. (Evaluating the filter once over the whole target table was tried first; it cost more than it saved whenever the expansion was already small, which re-rooting usually ensures.)
DataFusion's projection pushdown prunes an extension node's inputs but not its outputs, so without this rule a
fused lookup would fetch every target column for every row. The rule computes the set of columns actually used
anywhere in the plan (conservatively: expression references, plan outputs, each CSR node's key and closing column, and
all inputs of nodes whose column use is implicit, such as DISTINCT and UNION) and narrows each CsrExpand's
take and edge lists and each NodeLookup to that set. A NodeLookup whose columns are all unused is removed.
The two run in that order on purpose: after a filter moves into the node, the columns only the filter reads are no longer referenced above it and can be pruned from the output. Compare the logical and physical plans of the earlier example:
logical: CsrExpand: ... take=b[b__id*, b__name, b__age], edge=k[k__since] (before the rules)
physical: CsrExpandExec: ... take=b[b__name], edge=k[], filter=age@0 > 35, edge_filter=since@0 > 2020
The logical planner scans a path's first node and expands left to right, so
MATCH (f:Forum)-[:hasMember]->(p:Person) WHERE p.id = 42 would expand every forum's memberships before filtering.
The join path partly recovers because DataFusion pushes the filter below the join; a CSR expansion can't. So
pattern_order.rs rewrites each MATCH into an equivalent traversal that starts at the best-scoring node (one with a
key or property filter), walking segments backwards where needed (an outgoing edge walked from its target is
incoming). The same edges, endpoints, labels and variables are matched and the full WHERE still applies; only the
order of work changes.
flowchart LR
subgraph before ["As written"]
direction LR
F1["(f:Forum)<br/>all forums"] -->|hasMember| P1["(p:Person)<br/>WHERE p.id = 42"]
end
subgraph after ["Re-rooted"]
direction LR
P2["(p:Person {id: 42})<br/>one row"] -->|"hasMember (incoming)"| F2["(f:Forum)"]
end
before ~~~ after
-[:A|B]-> and (m:Post:Comment) are rewritten into branches with exactly one type per relationship and at most one
label per node, keeping only combinations consistent with the configured endpoint labels, and UNION ALLed. Each
branch is an ordinary pattern, so each keeps the CSR and re-rooting. A multi-type variable-length segment
(-[:hasType|isSubclassOf*1..20]->) is unrolled into the type sequences whose endpoint labels chain, unless every type
and label is indexed, in which case the typed iterative traversal handles it in one operator. Combinations the
configuration can't pin down are an error rather than a guess, because joining an edge type to the wrong node table
silently matches wrong rows whenever two labels share ids.
Not every hop runs on the CSR, and the joins that remain need good cardinality estimates. DataFusion estimates an
equi-join's output as L × R / max(NDV_l, NDV_r). Neither MemTable nor LanceTableProvider reports distinct counts,
so DataFusion substitutes the row count. For a graph join chain that turns a multi-million-row intermediate into an
estimate of a few thousand rows, under the 128k-row CollectLeft threshold, and DataFusion then builds that
intermediate as one single-threaded hash table.
KeyStatisticsProvider wraps each table and reports distinct counts for its node-id and edge-endpoint columns (taken
from the CSR when there is one, otherwise from one scan of the endpoint columns); every other statistic comes from the
inner table. The counts are cached per dataset version, so an unchanged refresh() costs nothing.
The operators above read two in-memory indexes, both built from Lance data and both pointing back into it.
Lance datasets (pinned version)
┌───────────────────────────┬───────────────────────────┐
│ KNOWS (src, dst, since) │ Person (id, name, age) │
└─────────────┬─────────────┴─────────────┬─────────────┘
scan src, dst, │ │ scan id + row id
_rowaddr ▼ ▼
┌─────────────────────────────────────┐ ┌───────────────────────────┐
│ Adjacency (per type, per direction) │ │ NodeRows (per label) │
│ offsets [0, 2, 3, 4] │ │ id -> Lance row id │
│ neighbours [2, 3, 4, 4] raw keys │ │ (unique integer ids) │
│ edge rows [row addresses] │ └─────────────┬─────────────┘
│ + delta: appended edges, │ │
│ tombstones │ │
└──────────────────┬──────────────────┘ │
│ neighbours(a.id) │ take_rows(positions, columns)
└─────────────► CsrExpandExec ◄──────┘
-
Adjacency(csr_expand.rs,csr_index.rs): outgoing and incoming compressed sparse row arrays per relationship type, built from the relationship table alone. Neighbours are stored as raw target keys, not node-table positions, so node-table writes never invalidate it. Each neighbour also records its edge's Lance row address, which is what lets a hop fetch relationship properties bytakeand lets a refresh find deleted edges without reading the old dataset version. About 32 bytes per edge stay resident for both directions. -
NodeRows: per label with unique integer keys, a map from key to Lance row id (or row index for Arrow input), built by scanning only the id column. It powers the fused target lookup andNodeLookup.
refresh() diffs two relationship-dataset versions by fragment, from the manifests: new fragments are added,
missing ones removed, rewritten ones swapped, and changed deletion files turn into tombstones. Fragments whose endpoint
data is unchanged are skipped, and nothing is read from the old version beyond its manifest, so compaction and
cleanup_old_versions() don't force a rebuild.
flowchart TD
R["engine.refresh()"] --> V{"Relationship dataset<br/>changed?"}
V -- no --> U["reused"]
V -- yes --> D{"Same endpoint<br/>columns?"}
D -- no --> RB["rebuild"]
D -- yes --> S{"Pending changes > 25%<br/>of the base CSR?"}
S -- yes --> RB
S -- no --> DL["delta: appended edges +<br/>tombstones for deleted ones"]
The delta sits on top of an immutable base shared between snapshots, exactly like Lance's indexed and unindexed rows:
| Lance index | CSR |
|---|---|
| Fragments the index was built from | The base CSR, at base_version
|
| Unindexed rows, scanned alongside the index | The delta, read by every lookup |
num_unindexed_rows |
unindexed_edges in csr_stats()
|
optimize_indices() |
optimize_csr() |
-
Partitioning in-memory tables. A
MemTableholding one large batch is one partition, andRoundRobinBatchrepartitioning moves whole batches, so an operator that doesn't hash-repartition (likeCsrExpandExec) would run on one thread. In-memory tables are split into zero-copy slices, one partition per core. Lance datasets use Lance's own parallel reader. -
Plan template cache. Each snapshot keeps a small LRU of unoptimized DataFusion plan templates, keyed by the query
text and its parameters' types, so re-running a query with new scalar
$parametersskips parsing and graph planning. A refresh starts an empty cache. -
Cooperative deadlines.
with_execution_deadlineinstalls theDeadlineJoinsphysical optimizer rule, which adds a volatile deadline check to nested-loop joins and yielding to cross joins, so a runaway Cartesian product stops withQuery timeout exceededinstead of running to completion. -
Cypher-correct UDFs.
cypher_list,cypher_field, Cypher math and scalar functions, list folds and three-valued list predicates built on DataFusion lambdas, and vector distance functions for similarity search inside Cypher. -
GIL released. From Python, several threads can query one engine at once, and
Ctrl-Cinterrupts a running query.
Independent, unaudited measurements on one machine (Apple M5, 24 GB); geometric means of per-query ratios over the queries both engines answered correctly. Full tables and reproduction steps are on the benchmarks page.
GDC SNB interactive complex reads (IC1–IC14, Neo4j's Cypher text unchanged)
| lance-csr mode | vs Neo4j | vs Kuzu | vs Ladybug |
|---|---|---|---|
| In memory | 3.9× faster | 6.1× faster | 3.2× faster |
| Lazy (Lance on disk) | 1.8× faster | 2.8× faster | 1.3× faster |
GDC SNB SF1, 30-query workload (3.18M nodes, 17.3M relationships)
| lance-csr mode | vs Neo4j | vs Kuzu | vs Ladybug | vs lance-graph |
|---|---|---|---|---|
| In memory | 3.6× faster | 4.1× faster | 2.6× faster | 4.6× faster |
| Lazy (Lance on disk) | 2.1× faster | 2.3× faster | 1.5× faster | 2.6× faster |
lance-graph, the project lance-csr was forked from, runs the same query text over the same Lance datasets and plans every hop as a join.
All paths are under crates/lance-csr/src/.
| File | What it holds |
|---|---|
csr_expand.rs |
Adjacency, NodeRows, EdgeRows, CsrExpandNode/Exec, CsrExtensionPlanner, CsrQueryPlanner, both optimizer rules, fragment diffing for refresh |
csr_expand/node_lookup.rs |
NodeLookupNode/Exec
|
csr_index.rs |
The compact CSR neighbour store |
csr_traversal.rs |
Resumable, typed variable-length walks |
shortest_path.rs |
ShortestPathNode/Exec, bidirectional BFS |
path_match.rs, path_inventory.rs, path_value.rs
|
Named paths, the edge inventory, Arrow path values |
apply.rs |
ApplyNode/Exec for CALL subqueries |
key_statistics.rs |
KeyStatisticsProvider/Exec
|
pattern_order.rs, pattern_alternatives.rs
|
Re-rooting, alternation and multi-label branching |
logical_plan.rs |
Graph logical operators |
datafusion_planner/ |
Graph plan → DataFusion plan; builder/expand_ops.rs decides CSR vs join per hop |
engine.rs |
CypherEngine: snapshots, session setup, refresh, plan template cache |
query/deadline.rs |
DeadlineJoins and with_execution_deadline
|