Make use_csr=True at least match the join path on full-relationship queries #112

Problem

On graph-benchmark (100K Person nodes, 2.4M FOLLOWS edges, tables in memory), use_csr=True is slower than the default join path on every query that reads the whole FOLLOWS relationship. From the 0.6.0 benchmark run (prrao87/graph-benchmark#23), mean ms:

q3 to q7 are even or slightly faster with the CSR. Raw runs: default, use_csr=True. Queries: lance_csr/query.py.

Some of this is expected: every CsrExpandExec in these plans has edges=2417738, the whole relationship, so the CSR has no work to skip, and an in-memory hash join with a 100K-row build side is already cheap. But most of the gap comes from plan choices that can be fixed. The goal is for use_csr=True to be at least even with the join path on these queries, so the benchmark (and users) can run with the CSR on by default. All of the items below are planner/operator changes that fit in one minor release (0.7.0).

Findings

Measured on lance-csr 0.6.0 (PyPI wheel), Apple M5, 24 GB, macOS 26.6.2, median of 25 runs after 5 warm-ups, every answer checked equal between the two engines.

1. Unused relationship variables make CSR hops read edge identity

q8 and q9 name r1 and r2 but never use them. try_build_two_hop_count (crates/lance-csr/src/datafusion_planner/builder/count_ops.rs) copies each hop's relationship_variable into the hops it rebuilds, so the CSR plan binds the edge (edge=r1[]) and decodes edge rows it never reads:

Fix: in the two-hop count rewrite, drop a hop's relationship variable unless one of its pushed-down filters reads it. The distinct-edges predicate is already turned into the __s1/__s2 self-loop correction, so the rewrite does not need edge identity. Keep binding outside the rewrite, where it enforces Cypher's no-repeated-edge rule.

2. The two-hop count still emits one row per edge on the CSR

Since #96 (#97), q8/q9 count Σ n1(b) × n2(b) − s1(b) × s2(b) per middle node. On the CSR, each side is CsrExpandExec emitting 2.4M rows, then a hash aggregate by b. But a node's out-degree is offsets[b+1] − offsets[b], and the CSR's reverse direction gives in-degree the same way. On its own, the out-degree side is only about even with the join path (9.0 vs 10.5 ms), because it still walks every edge.

Fix: a CSR degree operator that returns one row per node with its degree and its self-loop count (needed for __s1/__s2), used by the rewrite when the CSR covers the hop:

• A hop with no filter on its far end (both hops of q8, (a)-->(b) in q9) reads only offsets: 100K rows instead of 2.4M.
• A hop with a far-end filter (c.age > $age_2 in q9) still checks each neighbour, but counts in place instead of emitting rows for a hash aggregate.
• Filters on the middle node (b.age < $age_1 in q9) just restrict which nodes are emitted.

Self-loop counts could be computed during the CSR build, or per node in the operator; either is fine as long as answers match the join path.

3. Grouping by a node's properties carries them through every edge row

q1 groups by person.id, person.name in RETURN. The late lookup (fetch_late_node in crates/lance-csr/src/datafusion_planner/builder/aggregate_ops.rs, NodeLookupExec in the plan) only fires when a whole node is projected (WITH person), so the CSR hop fetches person.name for all 2.4M rows and the aggregate hashes it:

Fix: when the group keys include a node's unique integer id (a node index exists and the table is in memory, the same conditions as today's late lookup), group by the id plus any non-node keys, and fetch that node's other grouped properties after the aggregate with NodeLookupExec. Results are identical because the id is unique. This extends #99 from whole nodes to their properties.

4. Projected-away properties are still fetched by the CSR hop

In the WITH person form of q2, the plan already groups by person__id and fetches person__name afterwards with NodeLookupExec, but the hop below still reads it for every edge:

NodeLookupExec: key=person__id, take=person[person__name]
AggregateExec: mode=FinalPartitioned, gby=[person__id@0 as person__id], aggr=[count(follower__id) as __agg_0]
...
CsrExpandExec: key=follower__id, edges=2417738, take=person[person__id*, person__name]

Fix: once the late lookup takes over a column, drop it from the take= of the CsrExpandExec under the aggregate. The same pruning should fix the q1 rewrite in finding 3.

5. q2 as written: the CSR plan is 3x slower and the cause isn't pinned down

q2 groups by person.name and the city's columns, without person.id, so finding 3 does not apply (grouping by the id would change the answer when two people share a name):

The CSR plan gathers person__name per edge, then city__city, city__state and city__country per row of the second hop, and repartitions on all four string keys. The join path groups by the same keys in 30 ms, so the extra 60 ms is somewhere in the string gathers or the repartition. Profile it and fix what turns up; findings 3 and 4 may cover part of it.

Done when

• On graph-benchmark with use_csr=True, q1, q2, q8 and q9 are at least as fast as the join path, measured with its harness (uv run pytest benchmark_query.py --benchmark-min-rounds=5 --benchmark-warmup-iterations=5 --benchmark-disable-gc --benchmark-sort=fullname in lance_csr/), and q3 to q7 don't regress.
• The suites in CONTRIBUTING.md don't regress in either mode, and every answer check passes.
• Tests cover: the two-hop count with named and unnamed relationships, with and without self-loops, with filters on each end and on the middle node; grouping by n.id plus other properties of n (and by properties without the id, which must keep today's plan).
• The CSR docs say when use_csr=True helps and how to spot a full-relationship hop in explain() (edges= equal to the relationship's edge count).

Reproduce

git clone https://github.com/prrao87/graph-benchmark && cd graph-benchmark
git checkout 7ca5d17a47a7c17747012377468936809a6a0a5c
uv sync --locked
bash generate_data.sh 100000
cd lance_csr && uv run build_graph.py

Then, from lance_csr/ (swap in a local build of lance-csr to test changes):

import statistics, time
from lance_csr import CypherEngine
from query import GRAPH_ROOT, build_config, load_datasets

QUERIES = {
    "q1": ("MATCH (follower:Person)-[:FOLLOWS]->(person:Person) RETURN person.id AS personid, person.name AS name, count(follower.id) AS numfollowers ORDER BY numfollowers DESC LIMIT 3", None),
    "q2": ("MATCH (follower:Person)-[:FOLLOWS]->(person:Person)-[:LIVES_IN]->(city:City) RETURN person.name AS name, count(follower.id) as numfollowers, city.city AS city, city.state AS state, city.country AS country ORDER BY numfollowers DESC LIMIT 1", None),
    "q8": ("MATCH (a:Person)-[r1:FOLLOWS]->(b:Person)-[r2:FOLLOWS]->(c:Person) RETURN count(*) AS numpaths", None),
    "q9": ("MATCH (a:Person)-[r1:FOLLOWS]->(b:Person)-[r2:FOLLOWS]->(c:Person) WHERE b.age < $age_1 AND c.age > $age_2 RETURN count(*) as numpaths", {"age_1": 50, "age_2": 25}),
}

datasets = load_datasets(GRAPH_ROOT)
engines = {csr: CypherEngine(build_config(), datasets, use_csr=csr) for csr in (False, True)}

def median_ms(engine, query, params):
    for _ in range(5):
        engine.execute(query, params)
    times = []
    for _ in range(25):
        start = time.perf_counter()
        engine.execute(query, params)
        times.append((time.perf_counter() - start) * 1e3)
    return statistics.median(times)

for name, (query, params) in QUERIES.items():
    assert engines[False].execute(query, params).to_pylist() == engines[True].execute(query, params).to_pylist()
    print(f"{name}: join {median_ms(engines[False], query, params):.1f} ms, csr {median_ms(engines[True], query, params):.1f} ms")
    print(engines[True].explain(query, params))