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

A MATCH re-traversing from a preceding WITH's projected/aggregated bindings previously
declined on polars ("could not recover carried node identities"). Fixes two pandas-only
gaps: the native projector now emits the _cypher_entity_projection_meta side-channel the
bounded-reentry executor reads to re-seed; the executor id-handling + seeded binding
pipeline are engine-aware (polars is_not_null/filter/order-preserving join; semi-join to
carried ids). RETURN entity/property/count/count(*)/DISTINCT over WITH->MATCH now native,
pandas-parity. Differential fuzz ~3500 graphs: 0 disagreements, 0 crashes. Scalar-column
carry + duplicate-id / cartesian-under-seed stay honest NIEs (were crashes/silent-wrong).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL

Author: lmeyerov

CHANGELOG.md

@@ -12,6 +12,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL engine-selection docs (pandas / polars / cuDF / polars-gpu)**: New :doc:`Choosing a GFQL Engine <gfql/engines>` page — a numbers-first, persona-tested guide to the four interchangeable engines. Adds the one-keyword `engine='polars'` speedup (up to ~38× over pandas on real graphs, no GPU), a motivating warm-median comparison table on real public graphs (LiveJournal 35M / Orkut 117M), a decision matrix (workload shape × size × hardware → engine, with the measured ~10K-edge CPU crossover, GPU-work-bound rule, polars-gpu memory-pressure caveat, and GPU-or-error contract), a cuDF-vs-polars-gpu disambiguation (eager-op vs fused-lazy; cuDF is not deprecated), an honest "when *not* to use Polars" section, the differential-parity guarantee, and a methodology + reproducer-script disclosure. Rewrote the top of `gfql/performance.rst` to lead with the engine comparison (de-marketed the prose), wired the new page into the GFQL toctree + recommended paths, and added Polars/polars-gpu to the engine examples in `gfql/quick.rst` and `gfql/about.rst` (previously only pandas/cuDF were documented). Driven by 4-persona doc user-testing (pandas DS, RAPIDS user, perf engineer, skeptical evaluator).
 
 ### Fixed
+- **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

graphistry/compute/gfql/cypher/reentry/execution.py

@@ -23,6 +23,51 @@
 from graphistry.Engine import is_polars_df as _is_polars_df
 
 
+def _series_is_series_like(s: Any) -> bool:
+    """True for a pandas/cuDF Series (``.dropna``) or a polars Series (module check).
+
+    The reentry executor recovers carried node identities from a projection-meta ``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: Any) -> Any:
+    """Non-null boolean mask, engine-aware (polars ``is_not_null`` vs pandas ``notna``)."""
+    if _is_polars_df(s):
+        return s.is_not_null()
+    return s.notna()
+
+
+def _series_filter(s: Any, mask: Any) -> Any:
+    """Filter a Series by a boolean mask, engine-aware, dropping the old index (pandas)."""
+    if _is_polars_df(s):
+        return s.filter(mask)
+    return s[mask].reset_index(drop=True)
+
+
+def _frame_filter(df: Any, mask: Any) -> Any:
+    """Filter a DataFrame's rows by a boolean mask, engine-aware, dropping the old index."""
+    if _is_polars_df(df):
+        return df.filter(mask)
+    return df.loc[mask].reset_index(drop=True)
+
+
+def _ordered_left_join(left: Any, right: Any, *, on: str) -> Any:
+    """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 WITH-row ordering the reentry seed needs.
+
+    Under ``engine='polars'`` the carried ids come from the natively-projected (polars) prefix
+    while the base node table can still be pandas (the base graph is converted lazily), so align
+    ``right`` onto ``left``'s engine before the polars join."""
+    if _is_polars_df(left):
+        from graphistry.Engine import Engine, df_to_engine
+        if not _is_polars_df(right):
+            right = df_to_engine(right, Engine.POLARS)
+        return left.join(right, on=on, how="left", maintain_order="left")
+    return safe_merge(left, right, on=on, how="left")
+
+
 def _reentry_row(prefix_rows: Any, row_index: int) -> Any:
     """One prefix row as a col->scalar mapping, engine-aware (``row[col]`` works for
     both the pandas Series and the polars named-row dict)."""
@@ -274,7 +319,7 @@ def compiled_query_reentry_state(
         )
     ids = meta["ids"]
     id_column = meta["id_column"]
-    if not hasattr(ids, "dropna"):
+    if not _series_is_series_like(ids):
         raise reentry_validation_error(
             "Cypher MATCH after WITH could not recover carried node identities from the prefix stage",
             value=output_name,
@@ -306,6 +351,17 @@ def compiled_query_reentry_state(
             value=output_name,
             suggestion=REENTRY_SCALAR_SUGGESTION,
         )
+    if _is_polars_df(carried_ids) or _is_polars_df(aligned_prefix_rows) or _is_polars_df(base_nodes):
+        # Whole-row re-entry that ALSO carries scalar WITH columns (e.g. `WITH a, a.val AS av
+        # MATCH (a)-...`) threads hidden ``__cypher_reentry_*`` payload columns through the
+        # binding pipeline; that carry path is pandas-only so far. The seed-only whole-row case
+        # (no carried scalars) returned above. Decline honestly rather than run the pandas-only
+        # payload merge on polars frames (parity-or-NIE; no silent bridge).
+        raise NotImplementedError(
+            "polars engine does not yet natively support Cypher WITH -> MATCH re-entry that carries "
+            "scalar WITH columns into the trailing MATCH; use engine='pandas' for this query "
+            "(no pandas fallback; parity-or-error by design)"
+        )
     duplicate_mask = carried_ids.duplicated()
     if bool(duplicate_mask.any()) if hasattr(duplicate_mask, "any") else False:
         raise reentry_validation_error(
@@ -488,18 +544,18 @@ def aligned_reentry_rows(
             value=output_name,
             suggestion="Retry with a direct whole-row carry through WITH or inspect intermediate row-shaping before MATCH re-entry.",
         )
-    if not hasattr(ids, "notna"):
+    if not _series_is_series_like(ids):
         raise reentry_validation_error(
             "Cypher MATCH after WITH could not align carried node identities from the prefix stage",
             value=output_name,
             suggestion=REENTRY_WHOLE_ROW_SUGGESTION,
         )
 
-    non_null_mask = cast(SeriesT, ids.notna())
-    carried_ids = cast(SeriesT, ids[non_null_mask].reset_index(drop=True))
+    non_null_mask = _series_not_null_mask(ids)
+    carried_ids = cast(SeriesT, _series_filter(ids, non_null_mask))
     if prefix_rows is None:
         return carried_ids, None
-    return carried_ids, cast(DataFrameT, prefix_rows.loc[non_null_mask].reset_index(drop=True))
+    return carried_ids, cast(DataFrameT, _frame_filter(prefix_rows, non_null_mask))
 
 
 def reentry_carry_payload(
@@ -533,4 +589,4 @@ def ordered_reentry_start_nodes(
     id_column: str,
 ) -> DataFrameT:
     # MATCH re-entry must preserve the WITH row order, not the base node-table order.
-    return cast(DataFrameT, safe_merge(carried_node_ids, node_rows, on=id_column, how="left"))
+    return cast(DataFrameT, _ordered_left_join(carried_node_ids, node_rows, on=id_column))

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

@@ -132,12 +132,43 @@ def _flat_entity_exprs_polars(rows_df: pl.DataFrame, projection: ResultProjectio
     return out
 
 
+def _record_entity_meta(
+    entity_meta: dict,
+    rows_df: pl.DataFrame,
+    projection: ResultProjectionPlan,
+    source_alias: str,
+    output_name: str,
+    id_column: Optional[str],
+) -> None:
+    """Record whole-entity projection metadata for one column, mirroring the pandas projector.
+
+    ``_try_native_projection`` reaches this only in the single-entity branch (flat exprs and the
+    entity-text path both decline multi-entity prefixed columns), so ``rows_df`` is the aligned
+    source frame and ``rows_df[id_column]`` is the carried alias's id column, row-aligned with the
+    projected output. Snapshot (``.clone()``) the id column so downstream reentry recovery never
+    aliases a later-mutated working frame (see #1356)."""
+    if id_column is None or id_column not in rows_df.columns:  # pragma: no cover - defensive: node re-entry always carries the id column
+        return
+    entity_meta[output_name] = {
+        "table": projection.table,
+        "alias": source_alias,
+        "id_column": id_column,
+        "ids": rows_df.get_column(id_column).clone(),
+    }
+
+
 def _try_native_projection(result: Plottable, rows_df: pl.DataFrame, projection: ResultProjectionPlan, structured: bool) -> Optional[Plottable]:
     """Native projection for property/expr columns already in the polars row table + structured-
     flat or entity-text whole-entity returns; None → caller raises NIE."""
     import polars as pl
 
     exprs = []
+    # Whole-entity projection metadata side-channel (#1273 WITH->MATCH re-entry): mirror the
+    # pandas projector (result_postprocess._apply_result_projection_pandas), which records the
+    # carried alias's id column so the bounded-reentry executor can recover carried node
+    # identities. Without it a WITH-projected node alias feeding a trailing MATCH declines.
+    entity_meta: dict = {}
+    id_column = result._node
     for column in projection.columns:
         if column.kind == "whole_row":
             if projection.table != "nodes":
@@ -146,15 +177,16 @@ def _try_native_projection(result: Plottable, rows_df: pl.DataFrame, projection:
             if structured:
                 # #1650 default: flatten to {output}.{field} (near-free, any dtype);
                 # text fallback only for synthesized-absent rows.
-                id_column = result._node
                 flat = _flat_entity_exprs_polars(rows_df, projection, source_alias, column.output_name, id_column)
                 if flat is not None:
                     exprs.extend(flat)
+                    _record_entity_meta(entity_meta, rows_df, projection, source_alias, column.output_name, id_column)
                     continue
             ent = _native_node_entity_text_expr(rows_df, source_alias, projection.exclude_columns)
             if ent is None:
                 return None
             exprs.append(ent.alias(column.output_name))
+            _record_entity_meta(entity_meta, rows_df, projection, source_alias, column.output_name, id_column)
             continue
         src = column.source_name
         if src is None or src not in rows_df.columns:
@@ -172,6 +204,8 @@ def _try_native_projection(result: Plottable, rows_df: pl.DataFrame, projection:
         return None
     out = result.bind()
     out._nodes = rows_df.select(exprs)
+    if entity_meta:
+        setattr(out, "_cypher_entity_projection_meta", entity_meta)
     edges_df = result._edges
     if edges_df is not None:
         out._edges = edges_df.clear() if _is_polars_frame(edges_df) else edges_df[:0]

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

@@ -1008,15 +1008,37 @@ def _names(lf: pl.LazyFrame) -> List[str]:
     dst = g._destination
     if nodes is None or edges is None or node_id is None or src is None or dst is None:
         return None
-    if getattr(g, "_gfql_start_nodes", None) is not None:
-        # Bounded re-entry seeds the first alias from carried rows — pandas-only.
-        return None
+    seed_ids_lf: Optional[Any] = None  # LazyFrame; Any avoids the union-typed seed_nodes.join mismatch
+    start_nodes = getattr(g, "_gfql_start_nodes", None)
+    if start_nodes is not None:
+        # Bounded WITH->MATCH re-entry (#1273): the carried WITH rows seed the first
+        # alias. Constrain the first node alias to the carried ids via a semi-join —
+        # the native twin of the pandas wavefront seed. Support only UNIQUE carried
+        # ids: then the semi-join contributes each seed node exactly once, matching the
+        # pandas seed row-for-row; duplicate carried ids could change path multiplicity,
+        # so decline (return None -> honest NIE), never risk a silent-wrong count.
+        from graphistry.Engine import Engine as _Engine, df_to_engine as _df_to_engine, is_polars_df as _is_polars
+        sn = start_nodes.collect() if isinstance(start_nodes, pl.LazyFrame) else start_nodes
+        if not _is_polars(sn):
+            sn = _df_to_engine(sn, _Engine.POLARS)
+        if node_id not in sn.columns:
+            return None
+        seed_ids = sn.select(pl.col(node_id)).drop_nulls()
+        if seed_ids.height != seed_ids.unique().height:
+            return None
+        seed_ids_lf = seed_ids.lazy()
 
     ops: List[ASTObject] = [ast_from_json(op_json, validate=False) for op_json in binding_ops]
     # Shared validation (engine-agnostic): raises the canonical GFQLValidationError
     # for malformed op sequences / duplicate aliases — same error as pandas.
     RowPipelineMixin._gfql_validate_binding_ops(ops)
     if RowPipelineMixin._gfql_binding_ops_mode(ops) == "node_cartesian":
+        if seed_ids_lf is not None:
+            # A WITH->MATCH seed constrains the FIRST alias, but the cartesian builder
+            # below does not thread it; running it would silently ignore the seed
+            # (wrong cross-product). Decline honestly (the alternating-path seed is
+            # applied at seed_nodes; node-cartesian re-entry stays pandas-only).
+            return None
         # MATCH (a), (b), ... disconnected node aliases: native cross-product (#1273).
         return _cartesian_node_bindings_polars(g, ops, node_id)
     if RowPipelineMixin._gfql_is_shortest_path_scalar_binding_ops(ops):
@@ -1076,6 +1098,9 @@ def _names(lf: pl.LazyFrame) -> List[str]:
         if not isinstance(first_op, ASTNode):
             return None
         seed_nodes = filter_by_dict_polars(nodes_lf, first_op.filter_dict)
+        if seed_ids_lf is not None:
+            # WITH->MATCH re-entry seed: constrain the first alias to the carried ids.
+            seed_nodes = seed_nodes.join(seed_ids_lf, on=node_id, how="semi")
         state = seed_nodes.select(pl.col(node_id).alias("__current__"))
         alias_frames: Dict[str, pl.LazyFrame] = {}
         node_aliases: List[str] = []