feat(gfql): native polars WITH...MATCH re-traversal reentry (IC6/IC11 core, #1709-adjacent)

CHANGELOG.md

@@ -15,6 +15,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **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 whole-entity `count(DISTINCT n)` / identity aggregation (#1709)**: `g.gfql("MATCH (a {..}) RETURN count(DISTINCT a) AS c")` and its grouped/filtered/traversal variants (`(a)-[]->(b) RETURN count(DISTINCT b)`) previously declined on polars — the Cypher aggregation lowers `count(DISTINCT b)` to the `__gfql_node_id__` identity sentinel, and the polars `lower_expr` `Identifier` branch resolved only the prefixed `alias.__gfql_node_id__` form, not the bare sentinel, so it NIE'd. Fix threads the graph node-id column through a `_NODE_ID` contextvar (published alongside the schema in `_lower_with_schema`, wired through the four callers) so the bare sentinel resolves to the id column **only if present**, else declines (never invents/mis-resolves a column). Differential fuzz vs the pandas oracle: 600 + 300 cases, 0 disagreements. Deliberately still honest-NIE (verified declines, not silent-wrong): whole-entity `collect(b)` (list-of-entities repr not ported, cf #1650) and multi-alias binding-table identity aggregation (bare id absent on connected-pattern binding tables — adjacent to #1273). Tests in `test_engine_polars_row_pipeline.py`.
+- **GFQL polars engine natively runs `WITH ... MATCH ...` re-traversal (the IC6/IC11 core)**: a MATCH that re-traverses from a preceding `WITH`'s projected/aggregated bindings — `MATCH (p {id:X})-[:KNOWS]-(friend) WITH DISTINCT friend MATCH (friend)-[:HAS_CREATOR]-(post) RETURN count(post)` — previously declined on polars (`Cypher MATCH after WITH could not recover carried node identities from the prefix stage`). Two pandas-only gaps fixed: (1) the native polars projector now emits the `_cypher_entity_projection_meta` side-channel (carried alias `id`/`ids`) that the bounded-reentry executor reads to re-seed the next MATCH; (2) the reentry executor's id-handling and the seeded binding pipeline are made engine-aware (polars `is_not_null`/`filter`/order-preserving left-join; a semi-join of the first alias to the carried ids). `RETURN entity / property / count / count(*) / DISTINCT` over WITH→MATCH now run natively with pandas parity. Differential fuzz vs pandas over ~3500 random graphs × WITH→MATCH shapes (DISTINCT/filtered/ORDER-BY prefixes × fwd/rev/undirected re-MATCH × entity/property/count/distinct RETURN): **0 disagreements, 0 crashes** (declines are honest NIEs). Still honest-NIE (clean decline, was a crash): a WITH that also carries a **scalar column** into the trailing MATCH (hidden-payload threading is still pandas-only), and duplicate carried ids / node-cartesian trailing patterns under a seed (preserve pandas path-multiplicity / avoid silent-wrong). New `test_engine_polars_with_match_reentry.py`.
 - **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).

bin/test-polars.sh

@@ -18,6 +18,7 @@ POLARS_TEST_FILES=(
     graphistry/tests/compute/gfql/test_engine_polars_chain.py
     graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py
     graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py
+    graphistry/tests/compute/gfql/test_engine_polars_with_match_reentry.py
     graphistry/tests/compute/gfql/test_engine_polars_cypher_conformance.py
     graphistry/tests/compute/gfql/test_engine_polars_conformance_matrix.py
     graphistry/tests/compute/gfql/test_polars_string_predicate_nonstring.py
@@ -29,11 +30,14 @@ POLARS_TEST_FILES=(
     # index tests exercise the seeded-index hook in the polars hop entry (hop.py) — without
     # them the hook dominates the now-thin file and trips its per-file coverage floor
     graphistry/tests/compute/gfql/index/test_index.py
+    # engine-agnostic frame/series primitives (graphistry/Engine.py) — the polars branches of
+    # these dispatch helpers are only measured when this lane covers graphistry (see cov widen below)
+    graphistry/tests/test_engine_frame_helpers.py
 )
 
 COV_ARGS=()
 if [ -n "${POLARS_COV:-}" ]; then
-    COV_ARGS=(--cov=graphistry/compute --cov-report=)
+    COV_ARGS=(--cov=graphistry --cov-report=)
 fi
 
 python -B -m pytest -vv "${COV_ARGS[@]}" "${POLARS_TEST_FILES[@]}" "$@"
@@ -42,7 +46,7 @@ python -B -m pytest -vv "${COV_ARGS[@]}" "${POLARS_TEST_FILES[@]}" "$@"
 # appended into the same coverage data file when POLARS_COV=1 (CI audit reads it)
 COV_APPEND_ARGS=()
 if [ -n "${POLARS_COV:-}" ]; then
-    COV_APPEND_ARGS=(--cov=graphistry/compute --cov-report= --cov-append)
+    COV_APPEND_ARGS=(--cov=graphistry --cov-report= --cov-append)
 fi
 python -B -m pytest -vv "${COV_APPEND_ARGS[@]}" \
     graphistry/tests/compute/gfql/cypher/test_lowering.py -k polars

graphistry/Engine.py

@@ -3,12 +3,18 @@
 import numpy as np
 import pandas as pd
 import pyarrow as pa
-from typing import Any, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Sequence, Union
 from typing_extensions import Literal
 from enum import Enum
 
 from graphistry.models.types import ValidationParam
 
+if TYPE_CHECKING:
+    # Frame aliases (pandas types for mypy; ``Any`` at runtime). Imported under TYPE_CHECKING
+    # and referenced via string annotations below so Engine.py — imported very early — never
+    # triggers ``graphistry.compute`` package init at runtime (would be circular).
+    from graphistry.compute.typing import DataFrameT, SeriesT
+
 
 class Engine(Enum):
     PANDAS = 'pandas'
@@ -850,3 +856,103 @@ def safe_merge(
         raise ValueError("Must specify either 'on' or both 'left_on' and 'right_on'")
 
     return result
+
+
+# ---------------------------------------------------------------------------
+# Engine-agnostic series / frame primitives (pandas / cuDF / polars dispatch).
+#
+# Pure per-row/per-column dispatch helpers with no domain knowledge — the polars
+# branches genuinely return polars objects mypy can't narrow to the pandas frame
+# aliases, so the localized ``# type: ignore`` lives here at the dispatch point and
+# callers get a clean ``SeriesT`` / ``DataFrameT`` contract with no ``cast()``.
+# Annotations are strings so the TYPE_CHECKING-only alias import stays runtime-free.
+# ---------------------------------------------------------------------------
+
+
+def is_series_like(s: object) -> bool:
+    """True for a pandas/cuDF Series (``.dropna``) or a polars Series (module check).
+
+    Some engine-agnostic callers accept an ``ids`` Series that is pandas under
+    ``engine='pandas'`` and polars under ``engine='polars'``; both are valid."""
+    return hasattr(s, "dropna") or is_polars_df(s)
+
+
+def series_not_null_mask(s: "SeriesT") -> "SeriesT":
+    """Non-null boolean mask, engine-aware (polars ``is_not_null`` vs pandas ``notna``)."""
+    if is_polars_df(s):
+        return s.is_not_null()  # type: ignore[attr-defined,no-any-return]
+    return s.notna()
+
+
+def series_filter(s: "SeriesT", mask: "SeriesT") -> "SeriesT":
+    """Filter a Series by a boolean mask, engine-aware, dropping the old index (pandas)."""
+    if is_polars_df(s):
+        return s.filter(mask)  # type: ignore[attr-defined,no-any-return]
+    return s[mask].reset_index(drop=True)
+
+
+def frame_filter(df: "DataFrameT", mask: "SeriesT") -> "DataFrameT":
+    """Filter a DataFrame's rows by a boolean mask, engine-aware, dropping the old index."""
+    if is_polars_df(df):
+        return df.filter(mask)  # type: ignore[attr-defined,no-any-return]
+    return df.loc[mask].reset_index(drop=True)
+
+
+def ordered_left_join(left: "DataFrameT", right: "DataFrameT", *, on: str) -> "DataFrameT":
+    """Left join preserving ``left`` row order, engine-aware. Polars ``.merge`` does not exist;
+    ``safe_merge`` (pandas/cuDF ``.merge``) cannot run on polars frames, so branch to
+    ``.join(..., maintain_order='left')`` which pins the left-row ordering the caller needs.
+
+    ``right`` may arrive on a different engine than ``left`` (e.g. natively-projected polars
+    ``left`` against a still-pandas base table), so align ``right`` onto ``left``'s engine
+    before the polars join."""
+    if is_polars_df(left):
+        if not is_polars_df(right):
+            right = df_to_engine(right, Engine.POLARS)
+        return left.join(right, on=on, how="left", maintain_order="left")  # type: ignore[call-arg,no-any-return]
+    return safe_merge(left, right, on=on, how="left")
+
+
+def row_as_mapping(rows: "DataFrameT", row_index: int) -> Mapping[str, Any]:
+    """One frame row as a col->scalar mapping, engine-aware (``row[col]`` works for
+    both the pandas Series and the polars named-row dict)."""
+    if is_polars_df(rows):
+        return rows.row(row_index, named=True)  # type: ignore[attr-defined,no-any-return]
+    return rows.iloc[row_index]
+
+
+def assign_constant_columns(df: "DataFrameT", values: Dict[str, Any]) -> "DataFrameT":
+    """Broadcast scalar ``values`` as constant columns, engine-aware."""
+    if not values:
+        return df
+    if is_polars_df(df):
+        import polars as pl
+        return df.with_columns([pl.lit(v).alias(k) for k, v in values.items()])  # type: ignore[attr-defined,no-any-return]
+    return df.assign(**values)
+
+
+def drop_columns(df: "DataFrameT", cols: Sequence[str]) -> "DataFrameT":
+    """Drop columns by name, engine-aware (polars ``drop(list)`` vs pandas ``drop(columns=)``)."""
+    if is_polars_df(df):
+        return df.drop(list(cols))  # type: ignore[no-any-return]
+    return df.drop(columns=list(cols))
+
+
+def series_to_pylist(values: "SeriesT") -> List[Any]:
+    """Series -> python list, engine-aware, with defensive arrow/pandas fallbacks."""
+    if hasattr(values, "to_arrow"):
+        try:
+            return list(values.to_arrow().to_pylist())
+        except Exception:
+            pass
+    if hasattr(values, "to_pandas"):
+        try:
+            return list(values.to_pandas().tolist())
+        except Exception:
+            pass
+    if hasattr(values, "tolist"):
+        try:
+            return list(values.tolist())
+        except Exception:
+            pass
+    return list(values)