Custom operators¶

Every graph operator in lance-csr is a DataFusion extension: a logical node that implements UserDefinedLogicalNodeCore, lowered by CsrExtensionPlanner to its own ExecutionPlan. This page covers each one, with the plans engine.explain(query) prints for it.

All of these operators were added in lance-csr. lance-graph, which lance-csr was forked from, has no DataFusion extension nodes: it plans every hop as hash joins, which is the use_csr=False plan below and the fallback lance-csr keeps for hops the CSR can't take.

The plans below come from a four-person graph: a Person table (id, name, age) and a KNOWS table (src, dst, since) with four edges, in memory, with use_csr=True. Plans of real graphs differ in their numbers and partitioning, not in their operators.

CsrExpand: a hop as an array lookup¶

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.

Here is the same two-hop query planned both ways:

MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(c:Person {name: 'Dan'}) RETURN a.name

ProjectionExec: expr=[a__name@0 as a.name]
  FilterExec: b__id@1 != a__id@2 OR c__id@0 != b__id@1, projection=[a__name@3]
    ProjectionExec: expr=[c__id@0 as c__id, b__id@3 as b__id, a__id@6 as a__id, a__name@7 as a__name]
      CsrExpandExec: key=b__id, edges=4, take=a[a__id*, a__name]
        CsrExpandExec: key=c__id, edges=4, take=b[b__id*]
          RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
            ProjectionExec: expr=[id@0 as c__id]
              FilterExec: name@1 = Dan, projection=[id@0]
                KeyStatisticsExec: distinct=[id=4]
                  DataSourceExec: partitions=1, partition_sizes=[1]

ProjectionExec: expr=[a__name@0 as a.name]
  HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(knows_2__src@2, a__id@0)], ...
    HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(b__id@1, knows_2__dst@1)], ...
      HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(knows_1__src@1, b__id@0)], ...
        HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(c__id@0, knows_1__dst@1)], ...
          ProjectionExec: expr=[id@0 as c__id]
            FilterExec: name@1 = Dan, projection=[id@0]
              KeyStatisticsExec: distinct=[id=4]
                DataSourceExec: partitions=1, partition_sizes=[1]
          ProjectionExec: expr=[src@0 as knows_1__src, dst@1 as knows_1__dst]
            KeyStatisticsExec: distinct=[dst=3, src=3]
              DataSourceExec: partitions=1, partition_sizes=[1]
        ProjectionExec: expr=[id@0 as b__id]
          KeyStatisticsExec: distinct=[id=4]
            DataSourceExec: partitions=1, partition_sizes=[1]
      ProjectionExec: expr=[src@0 as knows_2__src, dst@1 as knows_2__dst]
        KeyStatisticsExec: distinct=[dst=3, src=3]
          DataSourceExec: partitions=1, partition_sizes=[1]
    ProjectionExec: expr=[id@0 as a__id, name@1 as a__name]
      KeyStatisticsExec: distinct=[id=4]
        DataSourceExec: partitions=1, partition_sizes=[1]

Four hash joins over five scans become two CsrExpandExecs over one filtered scan. The FilterExec above them keeps the two hops from reusing one edge. Both plans start from c (Dan), not a: that is re-rooting.

What one expansion can do¶

A hop that uses several 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]

Both WHERE conditions moved into the expansion, and the columns only they read (age, since) are gone from its output. Optimizer rules explains how.

A variable-length path is one operator instead of a branch per length:

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

Performance details¶

• Adaptive reads of lazy Lance tables. A batch of scattered rows from a large Lance table touches most of a column's pages. Once a relationship filter has seen more than 1/256 of the relationship's edges, the operator reads the filter's columns once, in order, builds a bitset per fragment, 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 it fetches properties. It yields partway 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, as Arrow and DataFusion do; with the standard library's SipHash, hashing dominated expansion time in profiles.
• A predicate pushdown boundary. The node lists 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.

A hop is eligible when its endpoint columns are integers, it has exactly one relationship type (alternation is split into branches first) and it has no inline relationship property map. Which hops use it has the full list. Any other hop stays a join.

NodeLookup: group by id, then look up¶

WITH b, count(*) AS n groups by the node b, which relationally means grouping by every one of b's columns, and RETURN b.id, b.name, count(*) groups by two of them. Rows before a grouping can outnumber nodes many times over, because 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) and the grouping includes b's id, whole or as b.id, the planner groups by the id alone and adds a NodeLookup above the aggregate that fetches b's other grouped 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*]
                ...

The expansion below takes only b's id: the lookup's b__name is a column of its own, so PruneCsrTargetColumns doesn't count it as a use of the expansion's. Grouping by b.name without b.id keeps the plain grouping, since two nodes may share a name. Unlike a join back to the node table, the lookup's cost follows the number of groups, not the size of the table. A lazily read Lance node table gets the lookup only when a CsrExpand in the aggregate's input fetches the grouped columns and no other node in that input reads them (a projection passing them through doesn't count; a Filter or Distinct does). Pruning then drops them from the expansion, so the lookup replaces that per-edge fetch instead of reading the columns a second time; see grouping by a node. From Lance, once the looked-up rows pass ⅛ of the table, the lookup reads its columns in one sequential pass and gathers from memory, with the same threshold and shared state as CsrExpand's target columns. A list comprehension over a list of nodes uses it too, to bind each element to its node.

PathExists: pattern predicates walked in place¶

A pattern predicate such as WHERE NOT (b)-[:KNOWS]->(a) asks, for each row, whether a path exists. Relationally, the planner answers it by planning everything before the predicate a second time, matching the pattern from that copy, and left-joining a mark back onto the rows. That doubles the plan beneath every predicate, and DataFusion plans and runs both copies. When the pattern qualifies, PathExistsExec instead walks the CSR from each row's start node, stops at the first path it finds, and appends the same mark (true, or null for no path), so the filter or CASE above it is unchanged:

MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE NOT (b)-[:KNOWS]->(a) RETURN a.name, b.name

ProjectionExec: expr=[a__name@0 as a.name, b__name@1 as b.name]
  FilterExec: __exists_1@2 IS NULL, projection=[a__name@0, b__name@1]
    ProjectionExec: expr=[a__name@1 as a__name, b__name@5 as b__name, __exists_1@6 as __exists_1]
      PathExistsExec: start=b__id, hops=1, match=a__id
        RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
          CsrExpandExec: key=a__id, edges=4, take=b[b__id*, b__name]
            ProjectionExec: expr=[id@0 as a__id, name@1 as a__name]
              KeyStatisticsExec: distinct=[id=4]
                DataSourceExec: partitions=1, partition_sizes=[1]

ProjectionExec: expr=[a__name@0 as a.name, b__name@1 as b.name]
  FilterExec: __exists_1@2 IS NULL, projection=[a__name@0, b__name@1]
    HashJoinExec: mode=CollectLeft, join_type=Left, on=[(b__id@2, __optional__b__id@0), (a__id@0, __optional__a__id@1)], ...
      HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(knows_1__dst@2, b__id@0)], ...
        ...                         (the MATCH, planned once for the rows)
      ProjectionExec: expr=[b__id@0 as __optional__b__id, a__id@3 as __optional__a__id, __exists_1@6 as __exists_1]
        AggregateExec: mode=FinalPartitioned, gby=[b__id@0 as b__id, ..., __exists_1@6 as __exists_1], aggr=[]
          ...
            HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(b__id@0, knows_3__src@0), (a__id@3, knows_3__dst@1)], ...
              ...
                HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(knows_2__dst@3, b__id@0)], ...
                  ...               (the MATCH again, as the seed of the predicate)

A predicate qualifies when it is one path that:

• starts at a node bound earlier in the query;
• passes through anonymous nodes, with no properties and only the label the relationship implies;
• uses anonymous relationships of one type each, all different, so no two hops can bind the same relationship, and none variable-length;
• ends at an anonymous node or at another bound node (match= in the plan);
• has no WHERE of its own.

Every node the walk reaches must be in its label's node index, as the relational plan joins each node to its table, so an edge to a missing node matches nothing. A predicate that doesn't qualify, a hop or label without a CSR, or an endpoint that isn't an integer id keeps the relational plan, which the logical operator carries as its fallback. Both give the same rows. The walk keeps no set of visited nodes, so a long chain of high-degree hops ending at a bound node costs the product of their degrees per row.

ShortestPath: breadth-first search over the adjacency¶

shortestPath and allShortestPaths run as a breadth-first search over the CSR adjacency. Without a CSR, the node gets a second input, a scan of the relationship's two endpoint columns, and builds an adjacency from it once per query, shared by all partitions.

• A source with one target searches from both ends using the outgoing and incoming adjacency. It expands the smaller frontier and finishes each layer before joining paths, so it keeps every shortest alternative.
• Several targets for the same source share one forward search.
• Predecessors retain every shortest path, including parallel edges.
• The operator appends <path>__length and <path>__nodes to each input row, once per path. Rows with no path are dropped for MATCH, and kept with nulls through a left join for OPTIONAL MATCH.

MATCH (a:Person {id: 1}), (b:Person {id: 4}) MATCH p = shortestPath((a)-[:KNOWS*]-(b)) RETURN length(p)

ProjectionExec: expr=[p__length@6 as length(p)]
  ShortestPathExec: a__id -> b__id, Undirected, hops 1..4294967295, one
    CrossJoinExec
      ...

PathMatch: named paths with real edge identity¶

CSR neighbours don't carry edge identity or properties, but MATCH p = ... with relationships(p) needs both, 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. It enumerates trails over the inventory, resuming between output batches, and reuses the bidirectional breadth-first search on a bipartite node/edge graph for shortest paths. Path values keep properties in a marked child, so user property names can't collide with identity metadata. See path values for what users see.

Apply: correlated CALL subqueries¶

A read-only CALL { ... } body runs once per incoming row: its LIMIT and aggregates apply to each row, not to the whole input. 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.

KeyStatistics: statistics on the scan¶

KeyStatisticsExec is 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, as with DataFusion's own CooperativeExec. It shows in plans as KeyStatisticsExec: distinct=[id=4].