Skip to content

Commit ebfa720

Browse files
lmeyerovclaude
andcommitted
perf(gfql/polars): make the chain combine proportional to the traversal result
Two graph-sized terms sat inside a combine whose answer is a handful of rows. 1. `_combine_edges` ran the prev/next endpoint gates for EVERY step, including the node steps whose edge frame is `g._edges.clear()` -- zero rows. The eager combine skipped those; the collect-once (Track B) rewrite lazified the step frames and `.lazy()` erases the height, so the skip silently went dead. The cost lands on the side that is NOT empty: for the first step the gate's key side is the whole node table, and polars builds the hash table on that side before discovering the probe side has no rows. Isolated in raw polars: 6.99 ms for one such join at N=2M vs 0.22 ms against a one-row key side -- and a chain pays one per node step. `_LazyShim.step` now records the row count while the frame is still eager and an empty step is dropped from the id union. It can contribute no ids, so the result is unchanged by construction. The skip keys on KNOWN-empty only: a frame that arrives already lazy reports no height and is planned normally. 2. The output node rows were materialized in TWO passes over the node table -- one for the ids the steps kept, one more for the surviving edges' endpoints the first pass missed -- then concatenated. The output node set is the UNION of those two id sides, so `_materialize_node_rows` unions the ids first and reads the node table ONCE. The row-level `unique(subset=[node])` is preserved verbatim: those rows go on to feed `how="left"` alias joins where a node table carrying the same id twice multiplies rows. The id sides themselves are NOT deduplicated -- they are semi key sides. Note for the record: the recorded description of this residual ("UNIQUE over a UNION of the per-step FULL frames") named the wrong operator. The union inputs are 1-row step frames; the `unique` costs 0.008 ms. It was the union BRANCHES -- empty-probe joins against graph-sized build sides -- exactly as the earlier semi-dedup finding. MEASURED, one dimension at a time, synthetic LDBC-IS5-shaped graphs (one-row answer, polars-engine resident indexes, local CPU, min of 9): vary NODES at E=2M 250k -> 4M: 10.17 -> 45.48 ms before 7.60 -> 17.17 ms after (2.65x at 4M) node-count slope 3.7x flatter (+35.3 ms -> +9.6 ms over 16x N) vary EDGES at N=1M 500k -> 8M: 13.47 -> 26.02 ms before 7.08 -> 17.57 ms after edge slope essentially unchanged, as expected for a node-side fix; constant ~6 ms lower same 2-3x holds for n-e-n, n-e-n-e-n and undirected shapes PARITY: 280 shape x graph combinations (clean / self-loop / dangling / duplicate keys / multi-edge / back-edge / isolated) and 400 duplicate-node-id combinations x 5 repeats compared as FULL frames including row order -- 0 diffs vs the pre-change tree, and stable across repeated runs. `graphistry/tests/compute/` failure set is byte-identical before and after (119 pre-existing failures, 0 new, +119 passed = the new file). TESTS pin the boundary, not a wall clock. Mutation-checked; each row is a deliberate reintroduction of a bug, run against the new file: drop the empty-step skip -> 2 fail (both structural tests) restore the two-pass materialization -> 1 fail (node table read 2x) drop the row-level node dedup -> 18 fail (parity + dedup + order) drop the endpoint fold -> 2 fail drop the node order restore -> 34 fail skip steps whose height is UNKNOWN -> 1 fail NOT caught, and reported as such: dropping the edge order restore, and dropping `maintain_order` on the node dedup. Neither is observable -- the edge semi-join returns EORD order naturally at 5k/60k/300k across 5 shapes, and the node semi-join feeding the dedup is unordered either way. Both are pre-existing defensive code, untouched here. Also honest about coverage: the endpoint fold could not be shown load-bearing end-to-end (6300 generated chain executions found no graph/shape where the step ids miss an edge endpoint), so it is tested at the helper, where the case can be constructed -- an end-to-end test of it would have passed with the endpoint side removed entirely. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VARwnAqmvczhz5EmP7TYyZ
1 parent 7d508ab commit ebfa720

3 files changed

Lines changed: 463 additions & 19 deletions

File tree

CHANGELOG.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
1212
- **GFQL secondary node property indexes (`create_index('node_prop', column=...)` / `g.gfql_index_node_props([...])`)**: a seed predicate on a NON-key column — `MATCH (m {id: 42})` where the graph's node id binding is some other column — previously cost a full node scan, because the registry only indexed the node-id binding and the CSR adjacencies. A property index is the same pay-as-you-go sidecar as the existing kinds: sorted distinct values over node **row positions** (CSR, so duplicate values are indexable), never reorders `.nodes`, fingerprint-validated so a `.nodes()` rebind is treated as absent (safe miss, never a wrong answer), engine-polymorphic (numpy host / cupy on-device), and policy-gated (`off`/`use`/`auto`/`force`). The seeded fixed-hop planner picks the **most selective** indexed scalar predicate in the seed filter using a free CSR-offset estimate, gathers those candidates, and applies every remaining predicate to them — so results are identical whether the index is present, absent, stale, or cost-gated out. `show_indexes()` lists property indexes; `drop_index('node_prop', column=...)` drops one. Only integer columns are indexable today (float NaN ordering, strings on cupy, and nulls all decline to the scan); widening that is additive. **Perf (dgx-spark, official LDBC SNB SF1, 3.18M nodes, warm median, value-identical 19-row result):** interactive-short IS7 `71.6 ms -> 19.5 ms` (**3.7x**), with a one-time `112 ms` build — the seed lookup itself goes from a `51.2 ms` scan to `0.096 ms`.
1313

1414
### Performance
15+
- **Native polars chain combine is proportional to the traversal result, not to the graph**: two graph-sized terms sat inside a combine whose answer is a handful of rows, and both are gone. (1) `_combine_edges` ran the prev/next endpoint gates for EVERY step, including the node steps whose edge frame is `g._edges.clear()` — zero rows. The eager combine skipped those, but the collect-once rewrite lazified the step frames and `.lazy()` erases the height, so the skip silently went dead. The cost lands on the side that is NOT empty: for the first step the gate's key side is the whole node table, and polars builds the hash table on that side before discovering the probe side has no rows (isolated: 6.99 ms for one such join at N=2M, and a chain pays one per node step). The pre-lazy row count is now recorded when the step frame is still eager and an empty step is dropped from the id union — it can contribute no ids, so the result is unchanged by construction. The skip keys on KNOWN-empty only; a frame that arrives already lazy reports no height and is planned normally. (2) The output node rows were materialized in TWO passes over the node table — one for the ids the steps kept, one more for the surviving edges' endpoints the first pass missed — then concatenated. The output node set is the UNION of those two id sides, so the ids are unioned first and the node table is read ONCE. The row-level `unique(subset=[node])` is preserved verbatim: those rows feed `how="left"` alias joins where a node table carrying the same id twice would multiply rows. Measured on synthetic LDBC-IS5-shaped graphs (one-row answer, polars-engine resident indexes), varying one dimension at a time: **at fixed E=2M, N=250k → 4M went 10.17 → 45.48 ms before and 7.60 → 17.17 ms after (2.65× at 4M, and the node-count slope is 3.7× flatter)**; the edge-count slope is unchanged, as expected for a node-side fix, with the constant ~6 ms lower. Parity: identical full frames (all columns, row order included) across 280 shape × graph combinations and 400 duplicate-node-id combinations, plus the 1003-case polars chain differential suite. Pinned by tests that assert the boundary rather than a wall clock: the node universe must not appear in the edge plan at all, and the node table must be read at most once per query.
1516
- **GFQL polars chain stops deduplicating semi-join key sides**: the native polars executor applied `.unique()` to every frame it fed into a `how="semi"` join. A semi-join emits a left row iff at least one matching right row exists, so duplicate keys can neither change which rows come back nor multiply them the way an inner join would — the deduplication was a full hash pass over the key column bought for no observable effect. On an unfiltered hop the key side **is** the node table, so this put **O(N) work inside a query whose answer is O(degree)**: the seeded single-hop plan built two such key frames per hop, each costing ~53 ms at 3.18M nodes — more than the rest of the query combined. The `.unique()` is now dropped everywhere the frame is provably a semi key side only (the `_semi` helper, the alias hop-window and next-edge endpoint gates, the two-hop fast path's endpoint gate, the `start_nodes` gate, and the single-hop planner's id frames). It is deliberately **kept** on the alias frame that feeds a `how="left"` join, where duplicates genuinely would multiply rows, and the eager multi-hop loop is untouched (its frames also flow into concat/anti-join bookkeeping, a separate argument). Measured on a 3.18M-node / 14M-edge polars graph (LDBC SNB SF1-shaped), a seeded typed hop goes **127.4 → 57.1 ms (2.23×)** with identical row counts; at 1.75M edges **102.8 → 31.4 ms (3.28×)**, confirming the removed cost scales with node count, not edge count. Parity held across 380 differential comparisons covering duplicate node keys, null ids, dangling edges, duplicate `start_nodes`, and 11 traversal shapes; the gfql and chain suites show identical failure sets before and after. Pinned by tests that assert the boundary rather than the speed: duplicates reaching a semi key side must not change results or multiply rows, a dangling endpoint must still be excluded (the gate is load-bearing, not vacuous), and duplicate `start_nodes` must be inert.
1617
- **Seeded chain combine stops joining against the full frame when the intermediate is empty**: `_lean_prefilter_right` shrinks the big side of the combine's `how='left'` merge to the keys actually present on the left, but it declined to do so in the one case where shrinking is both maximally profitable and trivially correct — an **empty** left. A left merge keeps only the right rows that match, so a zero-row left yields a zero-row result whatever the right side holds; the merge was nonetheless materializing the whole graph-sized frame. It now hands back a zero-row slice of `right` (same columns and dtypes, so the merge still produces an identical schema). Measured: a single-node query whose 0-row intermediate was joined against 14M edges went **112.64 → 17.29 ms**. The shrink is used at exactly one call site, which is `how='left'`; a merge that retains unmatched right rows (`right`/`outer`) would NOT be safe to shrink this way, and the tests pin both directions — the empty-left case must return an empty result, and the non-empty cases must be byte-identical to the unshrunk merge.
1718
- **Native polars chain reuses the synthetic edge id as its stable row order**: when the executor had already added a synthetic edge-index column it also added a second, separate row-index column purely to restore input order at the end — two full `with_row_index` passes over the edge frame, and a redundant column carried through every intermediate. The existing synthetic id is already a contiguous, order-preserving row index, so it is now reused as the sort key and dropped once at the end. Only the pre-existing column is reused; when no synthetic id was added the separate order column is still created, so graphs that bring their own edge id are unaffected. Value-identical including row order (148.22 → 135.43 ms on a 3.18M-node / 14M-edge polars graph).

graphistry/compute/gfql/lazy/engine/polars/chain.py

Lines changed: 77 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -219,22 +219,40 @@ def _is_native_multihop(op: ASTObject) -> bool:
219219

220220
class _LazyShim:
221221
"""Track B collect-once shim: carries _nodes/_edges as LazyFrames (+ col names) so the eager
222-
combine helpers run lazily over already-materialized hop frames without Plottable rebinds."""
223-
__slots__ = ("_nodes", "_edges", "_node", "_source", "_destination", "_edge")
224-
225-
def __init__(self, nodes_lf, edges_lf, node, source, destination, edge):
222+
combine helpers run lazily over already-materialized hop frames without Plottable rebinds.
223+
224+
``edges_empty`` records whether the step's edge frame was empty while it was still eager
225+
(tri-state: True/False, or None when unknown). ``.lazy()`` throws that fact away — a
226+
LazyFrame has no height without collecting — and the combine's cardinality shortcuts then go
227+
dead, which is a graph-sized mistake: an empty relation annihilates a join, but polars cannot
228+
know the relation is empty until it has already built the hash table over the OTHER side.
229+
Capturing it here costs nothing (the frames are materialized at construction) and this is the
230+
only place in the lazy combine where the count is still available."""
231+
__slots__ = ("_nodes", "_edges", "_node", "_source", "_destination", "_edge", "edges_empty")
232+
233+
def __init__(self, nodes_lf, edges_lf, node, source, destination, edge,
234+
edges_empty: Optional[bool] = None):
226235
self._nodes = nodes_lf
227236
self._edges = edges_lf
228237
self._node = node
229238
self._source = source
230239
self._destination = destination
231240
self._edge = edge
241+
self.edges_empty = edges_empty
232242

233243
@staticmethod
234244
def step(p):
235245
nd = p._nodes.lazy() if p._nodes is not None else None
236246
ed = p._edges.lazy() if p._edges is not None else None
237-
return _LazyShim(nd, ed, None, None, None, None)
247+
return _LazyShim(nd, ed, None, None, None, None, edges_empty=_known_empty(p._edges))
248+
249+
250+
def _known_empty(frame) -> Optional[bool]:
251+
"""Tri-state emptiness of an already-materialized frame: True/False, or None when unknown
252+
(frame absent, or already lazy so the height is not available without collecting)."""
253+
if frame is None or is_lazy(frame):
254+
return None
255+
return frame.height == 0
238256

239257

240258
def _combine_edges(g, steps, label_steps, has_multihop=False):
@@ -247,7 +265,16 @@ def _combine_edges(g, steps, label_steps, has_multihop=False):
247265
edges_df = g_step._edges
248266
if edges_df is None:
249267
continue
250-
if not is_lazy(edges_df) and edges_df.height == 0:
268+
# A step with no edges contributes no ids to the union below, so drop it BEFORE the
269+
# endpoint gates rather than semi-joining an empty frame against the graph. The gates
270+
# are the expensive part and their cost is on the side we do NOT need: polars builds
271+
# the hash table on the RIGHT (the node universe / a neighbouring step's node frame)
272+
# and only then probes with the empty left, so an unfiltered `prev_nodes = g._nodes`
273+
# costs a full O(N) hash build to produce the zero rows we already knew about
274+
# (measured: 6.99 ms for one such join at N=2M, and a chain hits one per node step).
275+
# Height is read from the pre-lazy fact recorded by _LazyShim.step because `.lazy()`
276+
# erases it; `not is_lazy(...)` keeps the direct-eager-frame case working.
277+
if g_step.edges_empty is True or (not is_lazy(edges_df) and edges_df.height == 0):
251278
continue
252279
if has_multihop or (isinstance(op, ASTEdge) and not op.is_simple_single_hop()):
253280
# has_multihop: every edge step was already recomputed path-valid (forward re-exec over
@@ -287,7 +314,18 @@ def _combine_edges(g, steps, label_steps, has_multihop=False):
287314
return out
288315

289316

290-
def _combine_nodes(g, steps):
317+
def _combine_node_ids(g, steps):
318+
"""One-column frame of the node ids the traversal kept, unioned over the pruned steps.
319+
320+
IDS ONLY, not the node rows: the caller still has to fold in the surviving edges' endpoints,
321+
and materializing the node rows before that fold means scanning the node table TWICE (once
322+
here, once for the endpoints the first scan missed). The union is over per-step id columns,
323+
so it is proportional to the traversal result, not to the graph.
324+
325+
Not deduplicated: the single consumer is a ``how="semi"`` key side, where duplicate keys can
326+
neither change which rows come back nor multiply them (see the module note on semi-join key
327+
frames). The caller's own ``.unique()`` on the materialized node rows is a DIFFERENT dedup
328+
(by node id, over rows) and is still required."""
291329
import polars as pl
292330
node_col = g._node
293331
assert node_col is not None
@@ -296,11 +334,35 @@ def _combine_nodes(g, steps):
296334
for _, g_step in steps
297335
if g_step._nodes is not None and node_col in colnames(g_step._nodes)
298336
]
299-
if frames:
300-
ids = pl.concat(frames, how="vertical_relaxed").unique(subset=[node_col])
301-
else:
302-
ids = g._nodes.select(pl.col(node_col)).limit(0)
303-
return g._nodes.join(ids, on=node_col, how="semi")
337+
if not frames:
338+
return g._nodes.select(pl.col(node_col)).limit(0)
339+
if len(frames) == 1:
340+
return frames[0]
341+
return pl.concat(frames, how="vertical_relaxed")
342+
343+
344+
def _materialize_node_rows(all_nodes, step_ids, endpoint_ids_frame, node_col):
345+
"""The output node ROWS: every node the steps kept, plus every endpoint of a surviving edge.
346+
347+
Union the two ID sides FIRST, then read the node table ONCE. Materializing the step rows and
348+
then fetching the endpoint rows the first pass missed reads the whole node frame TWICE for
349+
the same answer — two pure O(N) passes for a result that is usually a handful of rows
350+
(measured: 0.90 + 0.86 ms at N=2M). Row identity is unchanged: semi-joining the UNION of two
351+
key sets selects exactly the rows the two semi-joins selected between them.
352+
353+
Neither id side is deduplicated — both feed a ``how="semi"`` key side, where duplicates
354+
cannot change or multiply the rows that come back. The trailing ``unique`` is a DIFFERENT
355+
dedup and is REQUIRED: it is over the node ROWS, and these rows go on to feed ``how="left"``
356+
alias joins where a node table carrying the same id twice would multiply every matching row.
357+
358+
Row ORDER out of here is arbitrary — a polars semi-join does not preserve left-frame order —
359+
and the caller restores input-frame order with an explicit sort. ``maintain_order`` is kept
360+
verbatim from the pre-refactor call so that WHICH duplicate row survives is decided the same
361+
way it was before (A/B over 400 duplicate-id combos: identical full frames)."""
362+
import polars as pl
363+
ids = pl.concat([step_ids, endpoint_ids_frame], how="vertical_relaxed")
364+
return all_nodes.join(ids, on=node_col, how="semi").unique(
365+
subset=[node_col], maintain_order=True)
304366

305367

306368
def _apply_node_names(out, g, steps, auto_hop_col: str = _AUTO_NODE_HOP):
@@ -953,14 +1015,10 @@ def _plain_edge(op):
9531015
edge_steps_lz = [(op, _LazyShim.step(p)) for op, p in edge_steps]
9541016
label_lz = [(op, _LazyShim.step(p)) for op, p in label_steps]
9551017

956-
final_nodes = _combine_nodes(g_lz, steps_lz)
1018+
node_ids = _combine_node_ids(g_lz, steps_lz)
9571019
final_edges = _combine_edges(g_lz, edge_steps_lz, label_lz, has_multihop)
958-
# Endpoint (lazy: always compute; maintain_order keeps the semi-join order).
959-
endpoints = endpoint_ids(final_edges, src, dst, node_col).unique(subset=[node_col])
960-
missing = endpoints.join(final_nodes.select(pl.col(node_col)), on=node_col, how="anti")
961-
extra = g_lz._nodes.join(missing, on=node_col, how="semi")
962-
final_nodes = pl.concat([final_nodes, extra], how="diagonal_relaxed").unique(
963-
subset=[node_col], maintain_order=True)
1020+
final_nodes = _materialize_node_rows(
1021+
g_lz._nodes, node_ids, endpoint_ids(final_edges, src, dst, node_col), node_col)
9641022
final_nodes = _apply_node_names(final_nodes, g_lz, steps_lz, auto_hop_col=auto_hop_col)
9651023

9661024
final_nodes = final_nodes.sort(NORD).drop(NORD)

0 commit comments

Comments
 (0)