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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL

Author: lmeyerov

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.<prop>`, 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 `<name>_1` like pandas' `resolve_label_col`, not the polars left-join auto-suffix `<name>_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).

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")