From f19f2ccd02fccf28cee232d33d8fad002f30d26d Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 20 Jul 2026 15:36:39 -0700 Subject: [PATCH] feat(gfql): native polars undirected var-length bindings (-[*1..k]-) for IC11/IC6 cross-alias WHERE binding_rows_polars now materializes the bounded UNDIRECTED variable-length bindings table for min_hops==1 (the LDBC IC11/IC6 -[*1..k]- shape), unblocking the cross-alias same-path `WHERE NOT a = b` / `a <> b` clause on polars (the WHERE lowering itself was already native; the residual was the undirected `rows(binding_ops=...)` materialization declining). Exact port of the pandas oracle `_gfql_multihop_binding_rows` (avoid_immediate_backtrack=True): a `__prev__` marker (seeded null, dtype-matched to the id column) drops immediate backtracks each hop, and the step-pair set reproduces pandas' edge multiplicity (each non-loop edge each directed orientation x2; self-loops (u,u) x2 only). Scoped to min_hops==1 where the raw-edge reconstruction provably matches pandas; min_hops 0 and >=2 decline with an honest NotImplementedError (never silent-wrong). Differential fuzz vs pandas over ~2500 random graphs (self-loops, parallel + antiparallel edges, *1..2..*1..5, all WHERE/RETURN variants): 0 disagreements; out-of-scope windows decline. Tests: parity + multiplicity/backtrack/self-loop/ string-id pins + *0..2/*2..3/*2..2 decline pins. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- CHANGELOG.md | 1 + .../gfql/lazy/engine/polars/row_pipeline.py | 104 ++++++++++++++---- .../gfql/test_engine_polars_binding_rows.py | 90 ++++++++++++++- 3 files changed, 173 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 69a186f7cd..94bf276b82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Fixed - **GFQL polars engine natively runs `UNWIND` of a carried `collect()` list column**: `... WITH collect(x) AS xs UNWIND xs AS y RETURN ...` (exploding a list-valued column produced by `collect()`, the row-pipeline analogue of the IC6 unwind shape) previously declined on polars — `unwind_polars` only accepted a scalar literal list. Now explodes a `List`-dtype column reference (`with_columns` copy → `filter(list.len() > 0)` → `explode`), matching the pandas oracle exactly: empty-list/null cells → 0 rows, nulls *within* a list survive, source column retained; the pre-filter also makes it independent of the polars-2.0 `empty_as_null` default. Non-list columns, name collisions, unknown identifiers, and nested-list literals still decline (honest NIE, no pandas bridge). Differential fuzz vs pandas over ~3500 collect→UNWIND→{RETURN, grouped, filtered, ORDER BY, aggregate, nested-unwind} queries: 0 disagreements. (The `MATCH ... WITH collect(..) UNWIND .. MATCH` form — IC6 with a trailing re-traversal — is compile-rewritten into WITH→MATCH reentry, a separate path.) Tests in `test_engine_polars_row_pipeline.py`. +- **GFQL polars engine natively runs the LDBC IC11/IC6 undirected variable-length bindings table (`-[*1..k]-`), unblocking the cross-alias `WHERE NOT(person=friend)` clause**: a bounded UNDIRECTED variable-length MATCH that materializes a bindings row table (`rows(binding_ops=...)`, emitted whenever a downstream clause — e.g. a cross-alias same-path `WHERE NOT a = b` / `a <> b` — references two node aliases on the path) previously declined on polars (`binding_rows_polars` hard-NIE'd every undirected multihop), while pandas handled it. This was the single residual blocking the official LDBC IC11/IC6 queries on polars (the cross-alias WHERE lowering itself was already native — directed var-length + `WHERE NOT a = b` already matched pandas). `binding_rows_polars` now supports undirected var-length with **`min_hops == 1`** via a doubled-pair join with immediate-backtrack avoidance, an exact port of the pandas oracle `_gfql_multihop_binding_rows` (`avoid_immediate_backtrack=True`): a `__prev__` marker (seeded null, dtype-matched to the id column) drops immediate backtracks each hop (Kleene mask: null prev kept), and the step-pair set reproduces pandas' edge multiplicity exactly — each non-loop edge contributes each directed orientation TWICE (so a length-1 pair appears x2, length-2 x4) while self-loops contribute `(u,u)` x2 only (not double-counted). The multiplicity rule was derived by instrumenting pandas' `step_pairs` (which flow from the var-length `edge_op.execute` hop + `orient_edges`), NOT by reading code alone. Scoped to `min_hops == 1` because that is where the raw-edge reconstruction provably matches pandas: every edge is trivially a length-1 path so the var-length hop's backward pruning removes nothing; `min_hops == 0` (zero-hop, undoubled) and `min_hops >= 2` (backward-pruned / long-walk divergence) still **decline with an honest `NotImplementedError`** rather than risk silent-wrong multiplicities. Differential fuzz vs the pandas oracle over ~2500 random graphs (self-loops, parallel + antiparallel edges, `*1..2`…`*1..5`, all WHERE/RETURN variants): 0 disagreements; the out-of-scope windows decline (never diverge). Tests in `test_engine_polars_binding_rows.py` (parity + multiplicity/backtrack/self-loop/string-id pins + `*0..2`/`*2..3`/`*2..2` decline pins). - **GFQL polars engine natively runs multi-source (node-cartesian) MATCH (#1273)**: comma-separated disconnected aliases — `g.gfql("MATCH (a {..}), (b {..}) RETURN a.id, b.id")` — previously declined on polars (the `rows(binding_ops)` `node_cartesian` branch hard-NIE'd) while pandas handled them. New `_cartesian_node_bindings_polars` mirrors the pandas oracle `_gfql_cartesian_node_bindings_row_table`: each alias is independently filtered, projected into the per-alias lookup schema (bare `alias` id, `alias.id`, `alias.`, and pandas' leaked `alias.alias=True` flag incl. property-shadowing), then left-major cross-joined for row-order parity. Scalar/aliased/`count(*)` projections over the cartesian match pandas exactly (verified end-to-end); whole-entity `RETURN a, b` stays an honest NIE (separate projection surface). Two parity-safe declines mirror shapes where pandas itself errors (proven on master, so both engines fail identically, never diverge): an anonymous node op, and ≥4 named aliases (pandas' bare-id merge residue collides). Differential fuzz vs pandas over ~10k lowered cases: 0 disagreements. Tests in `test_engine_polars_binding_rows.py`. - **GFQL polars chain — variable-length node aliases are hop-distance gated; the chain is now silent-wrong-free (#1741)**: a node named after a variable-length edge (`g.gfql("MATCH (a)-[*1..2]-(b) RETURN b")`) previously carried its alias regardless of hop distance, so an undirected walk that backtracked into the seed wrongly returned it (pandas correctly excludes it — trail semantics). The polars chain now auto-injects the #1741 hop-distance label (name resolved against the user's node columns via the `reserved_columns` registry, so a user column named like the internal one is never clobbered) and applies pandas' alias `[min_hop, max_hop]` window in `_apply_node_names`. The one shape the gate can't yet cover — a node alias after a **forward/reverse `min_hops>1`** edge, whose labels need pandas' layered backward walk (not ported) — now **declines with an honest `NotImplementedError` (#1748)** instead of returning nodes outside the window; the decline is precise (differential vs pandas: 30/30 named shapes NIE, 30/30 unnamed run and match, 0 over-decline). 4-engine A/B on the stacked build: pandas/cudf 144/144, polars/polars-gpu 112 agree / **0 disagree** / 32 honest-NIE — zero silent-wrong shapes remain on the polars chain. Adapts the mechanism from the retired #1742 (declined undirected varlen+alias, now natively gated). Tests: `TestVarlenAliasHopGate` (pandas-parity across directions, the `*2..3` decline + unnamed-still-runs pins, a column-collision test). - **GFQL native polars `label_node_hops` on the plain BFS — correct, direction-dependent hop labels (#1741 groundwork)**: the polars eager hop now emits pandas-parity node hop-distance labels instead of declining `label_node_hops`. The labeling rule is DIRECTION-DEPENDENT (the subtle part, derived by differential fuzz vs the pandas oracle, not by reading pandas alone): forward/reverse label EVERY destination of a hop first-wins (matching pandas `hop.py` `new_node_ids`), so a seed re-entered at hop 1 IS labeled; undirected labels destinations MINUS everything already visited and pre-seeds the seen-set with the seeds, so a seed re-reached by a backtracking walk stays NULL (its shortest-path distance). Two correctness fixes over the first cut: (1) the undirected seed pre-seed must NOT depend on `return_as_wave_front` — a seed filtered out of the frontier by `source_node_match` or suppressed in wave-front mode was being re-labeled (A/B vs pandas over direction × hops × `return_as_wave_front` × `source_node_match` × `destination_node_match` × 10 graphs: was 456/24, now 480/480); (2) a requested `label_node_hops` name that collides with an existing column is redirected to `_1` like pandas' `resolve_label_col`, not the polars left-join auto-suffix `_right`/`DuplicateError`. `label_edge_hops` and `min_hops>1` labels stay honest NIEs. Amplified `test_engine_polars_hop.py` (`TestHopLabelsDifferential` over the seed-filter / wave-front / multi-seed axes, an explicit direction-asymmetry contrast pin, and a collision test). diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index 967e16267d..10e1477d9c 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -1057,17 +1057,31 @@ def _names(lf: pl.LazyFrame) -> List[str]: sem = EdgeSemantics.from_edge(op) if sem.is_multihop: # Bounded directed var-length (`-[*1..k]->`, graph-bench q3) is - # supported via iterative pair joins; everything else declines: - # unbounded (`[*]`, needs fixed-point + termination error), - # undirected multihop (immediate-backtrack avoidance not ported), - # and aliased var-length edges (pandas rejects those outright). + # 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 ( - op.direction == "undirected" - or bool(op.to_fixed_point) + bool(op.to_fixed_point) or (op.max_hops is None and op.hops is None) or isinstance(op._name, str) ): return None + if op.direction == "undirected": + _resolved_min = op.min_hops if op.min_hops is not None else ( + op.hops if op.hops is not None else 1 + ) + if _resolved_min != 1: + return None if op.direction not in ("forward", "reverse", "undirected"): return None if any( @@ -1174,23 +1188,71 @@ def _names(lf: pl.LazyFrame) -> List[str]: return None min_hops = int(min_hops_value) max_hops = int(max_hops_value) - pairs = oriented.select(["__from__", "__to__"]) state_cols = _names(state) - reachable = [state] if min_hops == 0 else [] - current = state - # Lazy: build all max_hops iterations (no eager .height early-break — - # empty intermediates lazily join to empty, so the result is - # identical; the pandas break is an optimization, not semantics). - for _hop in range(1, max_hops + 1): - current = ( - current.join(pairs, left_on="__current__", right_on="__from__", how="inner") - .drop("__current__") - .rename({"__to__": "__current__"}) - .select(state_cols) + if 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) + # EXACTLY, including its edge multiplicity: pandas' `step_pairs` + # come from the undirected var-length hop + `orient_edges`, which + # emits each NON-loop edge as (u,v)x2 AND (v,u)x2, and each + # SELF-loop as (u,u)x2 (loops are not double-counted). Reconstruct + # that here: `exec_rows` = both directions of non-loops + one row + # per self-loop; the final `pairs` doubles `exec_rows` + # (fuzz-verified vs pandas over random graphs incl. self-loops, + # parallel + antiparallel edges). A `__prev__` column (seeded null) + # carries the just-left node so each hop can drop immediate + # backtracks (`__to__ == __prev__`), matching pandas' Kleene mask + # (null prev -> kept). + 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__")]) + rev = normal.select([pl.col(dst).alias("__from__"), pl.col(src).alias("__to__")]) + loop = loops.select([pl.col(src).alias("__from__"), pl.col(dst).alias("__to__")]) + exec_rows = pl.concat([fwd, rev, loop], how="vertical") + pairs = pl.concat([exec_rows, exec_rows], how="vertical") + prev_col = "__prev__" + reachable = [state.select(state_cols)] if min_hops == 0 else [] + # Seed the backtrack marker with the SAME dtype as __current__ so a + # non-Int64 node id (e.g. string ids) compares/concats cleanly. + current = state.with_columns( + pl.lit(None).cast(state.collect_schema()["__current__"]).alias(prev_col) ) - if _hop >= min_hops: - reachable.append(current) - state = pl.concat(reachable, how="vertical") if reachable else state.limit(0) + for _hop in range(1, max_hops + 1): + joined = current.join( + pairs, left_on="__current__", right_on="__from__", how="inner" + ) + joined = joined.filter( + pl.col(prev_col).is_null() | (pl.col("__to__") != pl.col(prev_col)) + ) + # new prev = the node we are leaving (old __current__); new + # __current__ = __to__. Set prev BEFORE dropping __current__. + joined = ( + joined.with_columns(pl.col("__current__").alias(prev_col)) + .drop("__current__") + .rename({"__to__": "__current__"}) + ) + current = joined.select(state_cols + [prev_col]) + if _hop >= min_hops: + reachable.append(current.select(state_cols)) + state = pl.concat(reachable, how="vertical") if reachable else state.limit(0) + else: + pairs = oriented.select(["__from__", "__to__"]) + reachable = [state] if min_hops == 0 else [] + current = state + # Lazy: build all max_hops iterations (no eager .height early-break — + # empty intermediates lazily join to empty, so the result is + # identical; the pandas break is an optimization, not semantics). + for _hop in range(1, max_hops + 1): + current = ( + current.join(pairs, left_on="__current__", right_on="__from__", how="inner") + .drop("__current__") + .rename({"__to__": "__current__"}) + .select(state_cols) + ) + if _hop >= min_hops: + reachable.append(current) + state = pl.concat(reachable, how="vertical") if reachable else state.limit(0) else: state = ( state.join(oriented, left_on="__current__", right_on="__from__", how="inner") diff --git a/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py b/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py index 0a6b44f7e8..cb2ee0ca76 100644 --- a/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py +++ b/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py @@ -81,6 +81,15 @@ def _assert_parity(query, *, order_sensitive=False): "MATCH (a)-[*2..2]->(b) RETURN count(*) AS c", # exactly-k "MATCH (a)-[:F*1..2]->(b) RETURN count(*) AS c", # typed var-length "MATCH (a)-[*1..2]->(b)-[]->(c) RETURN count(*) AS c", # var-length + fixed hop + # bounded UNDIRECTED var-length, min_hops == 1 (LDBC IC11/IC6 `-[*1..k]-` shape): + # doubled-pair join + immediate-backtrack avoidance, exact pandas multiplicity. + "MATCH (a)-[*1..2]-(b) RETURN count(*) AS c", + "MATCH (a)-[*1..3]-(b) RETURN count(*) AS c", + "MATCH (a)-[:F*1..2]-(b) RETURN count(*) AS c", # typed undirected var-length + # the residual IC11 clause: cross-alias node inequality over an undirected + # var-length bindings table (previously NIE'd on the undirected `rows` op). + "MATCH (a)-[*1..2]-(b) WHERE NOT a = b RETURN b.id AS id ORDER BY id", + "MATCH (a)-[*1..2]-(b) WHERE a <> b RETURN a.id AS ai, b.id AS bi ORDER BY ai, bi", # node-only cartesian: disconnected multi-source MATCH (#1273). <=3 named aliases. "MATCH (a), (b) RETURN a.id AS ai, b.id AS bi ORDER BY ai, bi", "MATCH (a {kind: 'a'}), (b {kind: 'b'}) RETURN a.id AS ai, b.id AS bi", @@ -94,7 +103,13 @@ def _assert_parity(query, *, order_sensitive=False): # no silent wrong answer). DEFERRED = [ "MATCH (a)-[*]->(b) RETURN count(*) AS c", # unbounded var-length - "MATCH (a)-[*1..2]-(b) RETURN count(*) AS c", # undirected var-length + # undirected var-length is native ONLY for min_hops == 1 (see SUPPORTED). Other + # windows still DECLINE: pandas' step_pairs come from the var-length hop whose + # backward-pruning / zero-hop handling changes edge multiplicity in a way the + # raw-edge reconstruction only reproduces for min_hops == 1. + "MATCH (a)-[*0..2]-(b) RETURN count(*) AS c", # undirected, min_hops 0 + "MATCH (a)-[*2..3]-(b) RETURN count(*) AS c", # undirected, min_hops 2 + "MATCH (a)-[*2..2]-(b) RETURN count(*) AS c", # undirected, exactly-2 "MATCH (a)-[e]->(b) WHERE a.age < b.age RETURN a.id", # cross-alias same-path WHERE # cartesian outside the pandas-reliable subset (#1273): pandas itself errors or # is fragile here, so polars declines rather than diverge. @@ -240,6 +255,79 @@ def test_polars_binding_rows_undirected_self_loop(): +def _undirected_chain_graph(): + # chain 0-1-2-3 (directed edges, read undirected by the query) + nodes = pd.DataFrame({"id": [0, 1, 2, 3]}) + edges = pd.DataFrame({"s": [0, 1, 2], "d": [1, 2, 3]}) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +def test_polars_undirected_varlen_min1_backtrack_and_multiplicity_pins(): + """Undirected `-[*1..k]-` binding table (min_hops == 1) — pin the EXACT pandas + oracle semantics, not just parity, so a future drift on either engine is caught: + + * immediate-backtrack avoidance (0->1->0 excluded, so `WHERE NOT a = b` on a chain + never re-reaches the start via a single edge), + * pandas' edge multiplicity (each non-loop edge contributes each directed + orientation TWICE), so a length-1 pair appears x2 and a length-2 pair appears x4. + """ + g = _undirected_chain_graph() + # length-1 pairs appear x2, length-2 pairs appear x4 (pandas step_pairs doubling). + q = "MATCH (a)-[*1..2]-(b) WHERE NOT a = b RETURN a.id AS ai, b.id AS bi" + rpl = g.gfql(q, engine="polars")._nodes.to_pandas() + counts = rpl.groupby(["ai", "bi"]).size().to_dict() + # 1-hop neighbours (both orientations), each doubled + assert counts[(0, 1)] == 2 and counts[(1, 0)] == 2 + assert counts[(1, 2)] == 2 and counts[(2, 1)] == 2 + assert counts[(2, 3)] == 2 and counts[(3, 2)] == 2 + # 2-hop reaches (backtrack-free), each x4 + assert counts[(0, 2)] == 4 and counts[(2, 0)] == 4 + assert counts[(1, 3)] == 4 and counts[(3, 1)] == 4 + # backtrack pairs (a==b via 0->1->0) are excluded -> no (0,0)/(1,1)/... rows here + assert (0, 0) not in counts and (1, 1) not in counts + # and it exactly matches the pandas oracle + _assert_parity(q) + + +def test_polars_undirected_varlen_min1_self_loop_multiplicity(): + """Self-loops contribute (u,u) x2 only (NOT x4 like non-loops) — mirrors pandas, + which sources a single self-loop row from the var-length hop before orienting.""" + nodes = pd.DataFrame({"id": [0, 1, 2]}) + edges = pd.DataFrame({"s": [0, 1], "d": [1, 1]}) # edge 0->1 + self-loop 1->1 + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + q = "MATCH (a)-[*1..2]-(b) RETURN a.id AS ai, b.id AS bi" + rpd = g.gfql(q, engine="pandas")._nodes + rpl = g.gfql(q, engine="polars")._nodes + assert "polars" in type(rpl).__module__ + key = ["ai", "bi"] + pd.testing.assert_frame_equal( + rpd.sort_values(key).reset_index(drop=True), + rpl.to_pandas().sort_values(key).reset_index(drop=True), + check_dtype=False, + ) + + +def test_polars_undirected_varlen_min1_parity_string_ids_and_parallel(): + """String node ids + parallel/antiparallel edges: the `__prev__` backtrack marker + is dtype-matched to the id column and multiplicity stays pandas-exact.""" + nodes = pd.DataFrame({"id": ["x", "y", "z"]}) + edges = pd.DataFrame({"s": ["x", "y", "y"], "d": ["y", "x", "z"]}) # antiparallel + extra + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + for q in [ + "MATCH (a)-[*1..2]-(b) RETURN a.id AS ai, b.id AS bi", + "MATCH (a)-[*1..3]-(b) WHERE a <> b RETURN a.id AS ai, b.id AS bi", + ]: + rpd = g.gfql(q, engine="pandas")._nodes + rpl = g.gfql(q, engine="polars")._nodes + assert "polars" in type(rpl).__module__ + key = ["ai", "bi"] + pd.testing.assert_frame_equal( + rpd.sort_values(key).reset_index(drop=True), + rpl.to_pandas().sort_values(key).reset_index(drop=True), + check_dtype=False, + ) + + def test_polars_binding_rows_focused_native_coverage(): """Focused coverage for narrow native-polars helpers used by binding rows.""" from graphistry.Engine import Engine