From 134af67ab6bca3fa17d8af9b9998eb63a1e16898 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sat, 25 Jul 2026 21:38:30 -0700 Subject: [PATCH 1/6] feat(gfql): native polars unbounded directed var-length binding rows (LDBC IS6, #1709) The Cypher multi-alias bindings table (`rows(binding_ops=...)`) lowered natively for fixed-length and BOUNDED variable-length segments, but an unbounded fixed-point segment (`-[*]->` / `-[*0..]->`) declined with "polars engine does not yet natively support cypher row op 'rows'". That was the last shape blocking engine='polars' on LDBC SNB interactive-short-6, the one interactive-short query polars could not answer. Unbounded DIRECTED fixed point now lowers natively. Termination is data-dependent, so instead of pandas' expand-paths-until-empty (which materializes every partial path at every hop) the lowering runs a dedup-by-node frontier walk to find the exhaustion depth D, then reuses the SAME lazy bounded pair-join loop the `-[*1..k]->` arm uses with max_hops=D. Deduping by endpoint changes no walk's EXISTENCE, only its multiplicity, which the bounded loop then reproduces in full -- so the emitted rows are pandas-identical while the exponential path expansion happens once, lazily. A cycle reachable from the seed means infinitely many paths: that raises the same E108 "require terminating variable-length segments" error pandas raises, detected by a pigeonhole bound on the distinct nodes rather than after the blow-up. Also fixes a latent silent-wrong found by the differential fuzz: the polars builder rebuilt bindings from the CHAIN OUTPUT while pandas rebuilds from the pre-chain base graph. The traversal prunes to what IT considers matched, which is not the bindings builder's match set -- a zero-hop var-length segment binds a seed with no matching outgoing edge, and the traversal drops that node. So `-[*0..k]->` could silently return fewer rows than pandas. The builder now sources its node/edge tables from the same base graph pandas uses (and that the indexed builder was already handed). Honest declines unchanged in spirit (NIE, never a silent answer): undirected unbounded, aliased var-length relationships, and unbounded segments without to_fixed_point (pandas truncates at a bound this lowering cannot reconstruct). Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01NxxTFpKiqA4FQa9Pjy8yR1 --- CHANGELOG.md | 4 + .../gfql/lazy/engine/polars/row_pipeline.py | 198 +++++++++++++---- .../gfql/test_engine_polars_binding_rows.py | 199 +++++++++++++++++- 3 files changed, 358 insertions(+), 43 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ef993819b4..6cff767e26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Added +- **Native polars `rows(binding_ops=...)` for UNBOUNDED directed variable-length patterns (`-[*]->` / `-[*0..]->`) (#1709)**: the Cypher multi-alias bindings table already lowered natively for fixed-length and *bounded* variable-length segments, but an unbounded fixed-point segment declined with `NotImplementedError: polars engine does not yet natively support cypher row op 'rows'`. This was the last shape blocking `engine='polars'` on LDBC SNB interactive-short-6 (`MATCH (m:Message)-[:REPLY_OF*0..]->(p:Post)<-[:CONTAINER_OF]-(f:Forum)-[:HAS_MODERATOR]->(mod)`), the one interactive-short query polars could not answer. It now runs natively: a dedup-by-node frontier walk finds the exhaustion depth, then the SAME lazy bounded pair-join loop the `-[*1..k]->` arm uses materializes one row per distinct edge SEQUENCE (Cypher path multiplicity, parallel edges included) — no pandas bridge, no `to_pandas()` round trip. A cycle reachable from the seed means infinitely many paths; that raises the same E108 "require terminating variable-length segments" error pandas raises, and is detected before the path expansion blows up rather than after. Still declining honestly (NIE, never a silent answer): UNDIRECTED unbounded (`-[*]-`, needs the min_hops == 1 multiplicity reconstruction plus backtrack-aware termination — pandas rejects it outright), aliased variable-length relationships (pandas rejects those too), and unbounded segments WITHOUT `to_fixed_point` (pandas silently truncates at a bound this lowering cannot reconstruct). Cross-engine parity is the gate: differential fuzz vs the pandas oracle over random DAGs and cyclic graphs with self-loops and parallel edges, plus pinned IS6/zero-hop/multiplicity/cycle tests. + - **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`. ### Performance @@ -20,6 +22,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - **GFQL execution context is declared rather than attached at runtime**: the private `_gfql_*` per-execution fields (index policy and registry, row-pipeline base graph, carried seed nodes, edge aliases, shortest-path backend, and the indexed-bindings handoff) are now declared on `Plottable` with defaults on `PlotterBase`, instead of being set with `setattr` and read back with `getattr(..., default)`. No public API or behaviour change; internal call sites are typed, and hand-rolled `Plottable` stand-ins must now construct the full context. ### Fixed +- **Native polars variable-length bindings dropped zero-hop rows (`-[*0..k]->`)**: the polars `rows(binding_ops=...)` builder rebuilt the bindings table from the CHAIN OUTPUT, while the pandas oracle rebuilds from the pre-chain base graph. The traversal prunes to what IT considers matched, which is not the same set the bindings builder matches — a zero-hop variable-length segment binds a seed that has no matching outgoing edge, and the traversal drops that node entirely. So `MATCH (a)-[*0..2]->(b)` could silently return fewer rows than pandas (no error, just missing rows). The polars builder now sources its node/edge tables from the same pre-chain base graph pandas uses (which is also what the indexed builder was already handed). Found by differential fuzzing while adding unbounded support (#1709); pinned by a zero-hop regression test and by `-[*0..2]->` parity cases. + - **Indexed bypass could return every node for `rows()` over an unnamed pattern**: the boundary accepted a `rows()` call carrying no binding ops over a middle with no aliases. That call reads the traversal-narrowed node table, but the bypass hands it the full graph, so the query would have returned all nodes. The gate now requires that the rows call actually consumes the whole middle as binding ops — either it already carries them, or the named-middle rewrite installs exactly them. - **Seeded property-RETURN dtype divergence on cuDF**: the lean projection applied the pandas rows-pivot artifact (int → float64, bool → object) on every engine, but cuDF's canonical pivot preserves the source dtypes — so the fast path returned `float64`/`object` where cuDF's own canonical path returns `int64`/`bool`. The cast rule is now engine-aware; the dtype-class decline guard is unchanged. diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index 277e570991..ac936b73ba 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -1094,6 +1094,96 @@ def _cartesian_node_bindings_polars( return _rewrap(g, out_df) +def _directed_fixed_point_binding_rows_polars( + state: "pl.LazyFrame", + pairs: "pl.LazyFrame", + state_cols: List[str], + *, + min_hops: int, +) -> "pl.LazyFrame": + """Unbounded DIRECTED variable-length binding rows (``-[*0..]->`` / ``-[*]->``), #1709. + + The native twin of the ``max_hops is None and to_fixed_point`` arm of pandas' + ``RowPipelineMixin._gfql_multihop_binding_rows``. One row per distinct edge + SEQUENCE (Cypher path multiplicity): ``pairs`` is never deduped, so parallel + edges multiply per hop exactly as the pandas merge does. ``min_hops == 0`` + contributes the zero-hop rows (endpoint == start) first, then hop 1, 2, ... — + the same ``reachable`` concat order pandas builds. + + Pandas discovers the traversal depth by expanding the PATH frontier until it is + empty, which materializes every partial path at every hop. This lowering splits + that into (a) a cheap dedup-by-node frontier walk that computes the exhaustion + depth ``D``, then (b) the SAME lazy bounded pair-join loop the ``-[*1..k]->`` arm + uses, with ``max_hops = D``. Step (a) is exact, not an approximation: the path + frontier at hop ``h`` is non-empty iff a walk of length ``h`` leaves some seed, + and deduping by endpoint node changes no walk's EXISTENCE — only its + multiplicity, which step (b) then reproduces in full. So the emitted rows are + identical to pandas' while the exponential path blow-up happens once, lazily. + + Non-termination: a cycle reachable from a seed makes the walk infinite, and + pandas raises ``E108`` ("require terminating variable-length segments"). We raise + the same error via the same helper, and we detect it strictly: with ``N`` distinct + nodes touched by ``pairs``, any walk of ``N`` edges visits ``N + 1`` nodes and so + must repeat one (pigeonhole) — a non-empty frontier at hop ``N`` is a reachable + cycle, exactly the condition under which pandas' own cap + (``max(len(step_pairs), 1) + 1``) also fails to exhaust. Conversely an acyclic + reachable subgraph has every walk shorter than ``N``, so both engines exhaust. + Same outcome on both sides of the branch, without pandas' cost of expanding paths + into the cycle before giving up. + """ + import polars as pl + from graphistry.compute.gfql.lazy import collect as _lazy_collect + from graphistry.compute.gfql.row.pipeline import RowPipelineMixin + + pairs_df = _lazy_collect(pairs) + pairs_lf = pairs_df.lazy() + node_cap = int( + pl.concat( + [ + pairs_df.select(pl.col("__from__").alias("__n__")), + pairs_df.select(pl.col("__to__").alias("__n__")), + ], + how="vertical", + ) + .select(pl.col("__n__").n_unique()) + .item() + ) + + # (a) depth probe: dedup-by-node frontier, so each hop costs O(N) not O(paths). + frontier = _lazy_collect(state.select(pl.col("__current__")).unique()) + depth = 0 + exhausted = node_cap == 0 # no matched edges at all -> no walk of length >= 1 + for hop in range(1, node_cap + 1): + frontier = _lazy_collect( + frontier.lazy() + .join(pairs_lf, left_on="__current__", right_on="__from__", how="inner") + .select(pl.col("__to__").alias("__current__")) + .unique() + ) + if frontier.height == 0: + exhausted = True + break + depth = hop + if not exhausted: + RowPipelineMixin._gfql_bindings_error( + "Cypher multi-alias row bindings currently require terminating variable-length segments" + ) + + # (b) bounded path expansion, identical to the `-[*1..k]->` arm with max_hops=depth. + reachable: List["pl.LazyFrame"] = [state] if min_hops == 0 else [] + current = state + for hop in range(1, depth + 1): + current = ( + current.join(pairs_lf, left_on="__current__", right_on="__from__", how="inner") + .drop("__current__") + .rename({"__to__": "__current__"}) + .select(state_cols) + ) + if hop >= min_hops: + reachable.append(current) + return pl.concat(reachable, how="vertical") if reachable else state.limit(0) + + def binding_rows_polars( g: Plottable, binding_ops: Sequence[Dict[str, JSONVal]], @@ -1129,11 +1219,23 @@ def _names(lf: pl.LazyFrame) -> List[str]: # LazyFrame column names WITHOUT collecting data (schema-only resolve). return lf.collect_schema().names() - nodes = g._nodes - edges = g._edges - node_id = g._node - src = g._source - dst = g._destination + # Build from the PRE-CHAIN base graph, exactly like the pandas oracle + # (`_gfql_binding_rows`: `base_nodes = base_graph._nodes`, every alias step + # `op.execute(g=base_graph, ...)`) and like the indexed builder below, which is + # already handed `base_graph`. Rebuilding from the chain OUTPUT instead would + # silently under-report: the traversal prunes to nodes/edges IT considers + # matched, and that is not the same set the bindings builder matches — e.g. a + # zero-hop var-length segment (`-[*0..k]->`, `-[*0..]->`) binds a seed with no + # outgoing edge, which the traversal drops entirely. Fuzz-verified: with the + # chain output as the source, `-[*0..2]->` lost those rows against pandas. + base_graph = g._gfql_rows_base_graph + if base_graph is None: + base_graph = g + nodes = base_graph._nodes + edges = base_graph._edges + node_id = base_graph._node + src = base_graph._source + dst = base_graph._destination if nodes is None or edges is None or node_id is None or src is None or dst is None: return None seed_ids_lf: Optional[Any] = None # LazyFrame; Any avoids the union-typed seed_nodes.join mismatch @@ -1175,9 +1277,6 @@ def _names(lf: pl.LazyFrame) -> List[str]: from graphistry.compute.gfql.index import bindings as indexed_bindings from graphistry.compute.gfql.lazy import active_target, ExecutionTarget from graphistry.Engine import Engine - base_graph = g._gfql_rows_base_graph - if base_graph is None: - base_graph = g engine_concrete = ( Engine.POLARS_GPU if active_target() == ExecutionTarget.GPU @@ -1229,20 +1328,31 @@ def _names(lf: pl.LazyFrame) -> List[str]: # supported via iterative pair joins. Bounded UNDIRECTED var-length # with min_hops == 1 (`-[*1..k]-`, the LDBC IC11/IC6 shape) is now # also supported via a doubled-pair join with immediate-backtrack - # avoidance (see the execution branch below). Everything else declines: - # unbounded (`[*]`, needs fixed-point + termination error), aliased - # var-length edges (pandas rejects those outright), and undirected - # var-length with min_hops != 1 (`-[*0..k]-` / `-[*2..k]-`): pandas' - # step_pairs come from the var-length `edge_op.execute` hop, whose - # backward hop-window pruning / zero-hop handling changes the edge - # multiplicity in a way this raw-edge reconstruction only reproduces - # for min_hops == 1 (every edge is trivially a length-1 path, so no - # pruning occurs) — fuzz-verified vs the pandas oracle. Decline the - # rest honestly rather than risk silent-wrong multiplicities. - if ( - bool(op.to_fixed_point) - or (op.max_hops is None and op.hops is None) - or isinstance(op._name, str) + # avoidance (see the execution branch below). UNBOUNDED DIRECTED + # fixed-point (`-[*0..]->` / `-[*]->`, the LDBC IS6 REPLY_OF ancestor + # walk, #1709) runs the same pair join to exhaustion — see + # `_directed_fixed_point_binding_rows_polars` for the parity argument. + # Everything else declines: + # - aliased var-length edges (pandas rejects those outright); + # - undirected var-length with min_hops != 1 (`-[*0..k]-` / + # `-[*2..k]-`): pandas' step_pairs come from the var-length + # `edge_op.execute` hop, whose backward hop-window pruning / + # zero-hop handling changes the edge multiplicity in a way this + # raw-edge reconstruction only reproduces for min_hops == 1 (every + # edge is trivially a length-1 path, so no pruning occurs) — + # fuzz-verified vs the pandas oracle; + # - undirected UNBOUNDED (`-[*]-`): would need both the multiplicity + # reconstruction above and backtrack-aware termination; + # - unbounded WITHOUT to_fixed_point (`min_hops=2` and no max): pandas + # silently truncates at `len(step_pairs) + 1` iterations instead of + # erroring, and its step_pairs row count is not reconstructible here, + # so the truncation depth (hence the answer) is not reproducible. + # Decline the rest honestly rather than risk silent-wrong multiplicities. + if isinstance(op._name, str): + return None + _resolved_max = op.max_hops if op.max_hops is not None else op.hops + if _resolved_max is None and ( + not bool(op.to_fixed_point) or op.direction == "undirected" ): return None if op.direction == "undirected": @@ -1356,12 +1466,20 @@ def _names(lf: pl.LazyFrame) -> List[str]: edge_op.hops if edge_op.hops is not None else 1 ) max_hops_value = edge_op.max_hops if edge_op.max_hops is not None else edge_op.hops - if max_hops_value is None: - return None min_hops = int(min_hops_value) - max_hops = int(max_hops_value) state_cols = _names(state) - if sem.is_undirected: + if max_hops_value is None: + # UNBOUNDED directed fixed point (`-[*0..]->`, LDBC IS6). Gated + # above to to_fixed_point=True and a directed edge. Termination is + # data-dependent, so unlike the bounded branch this one cannot stay + # fully lazy — it collects one frontier per hop. + state = _directed_fixed_point_binding_rows_polars( + state, + oriented.select(["__from__", "__to__"]), + state_cols, + min_hops=min_hops, + ) + elif sem.is_undirected: # Bounded UNDIRECTED var-length, min_hops == 1 (gated above): the # LDBC IC11/IC6 `-[*1..k]-` shape. Mirror the pandas oracle # (`_gfql_multihop_binding_rows`, avoid_immediate_backtrack=True) @@ -1376,6 +1494,7 @@ def _names(lf: pl.LazyFrame) -> List[str]: # carries the just-left node so each hop can drop immediate # backtracks (`__to__ == __prev__`), matching pandas' Kleene mask # (null prev -> kept). + max_hops = int(max_hops_value) normal = edges_f.filter(pl.col(src) != pl.col(dst)) loops = edges_f.filter(pl.col(src) == pl.col(dst)) fwd = normal.select([pl.col(src).alias("__from__"), pl.col(dst).alias("__to__")]) @@ -1409,6 +1528,7 @@ def _names(lf: pl.LazyFrame) -> List[str]: reachable.append(current.select(state_cols)) state = pl.concat(reachable, how="vertical") if reachable else state.limit(0) else: + max_hops = int(max_hops_value) pairs = oriented.select(["__from__", "__to__"]) reachable = [state] if min_hops == 0 else [] current = state @@ -1441,15 +1561,13 @@ def _names(lf: pl.LazyFrame) -> List[str]: # HAS_