From aa3adb529a651bc949a696db4645cb3cd0fbf39e Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 11:58:50 -0700
Subject: [PATCH 1/9] feat(gfql): polars whole-entity aggregation Cypher
 (HAS_label disambiguation + identity-key + key_prefixes)

Three pandas-parity ports unlock the LDBC IC4 shape natively on polars:
binding_rows_polars applies the HAS_<Label> candidate-domain narrowing (only
under duplicated candidate ids, mirroring pandas exactly) instead of a blanket
decline; alias.__gfql_node_id__ resolves to the bare alias id column; and
group_by expands key_prefixes like pandas. Official IC4 runs parity-exact
(harness ok @127.8ms SF0.1, was NIE). 184 suite tests + 2 new pass on dgx.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 CHANGELOG.md                                  |  1 +
 .../compute/gfql/lazy/engine/polars/chain.py  | 12 +--
 .../gfql/lazy/engine/polars/row_pipeline.py   | 42 +++++++--
 .../gfql/test_polars_rows_entity_groupby.py   | 92 +++++++++++++++++++
 4 files changed, 135 insertions(+), 12 deletions(-)
 create mode 100644 graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7f91071bb5..ef6f5e6095 100644
--- a/CHANGELOG.md
+++ b/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 whole-entity aggregation Cypher (LDBC IC4 shape): HAS_<Label> destination disambiguation + entity identity-key resolution + `key_prefixes` group-by expansion**: three previously honest-NIE polars lowerings, each an exact pandas-parity port: (1) `binding_rows_polars` now applies pandas' `_gfql_disambiguate_has_edge_destination_nodes` candidate-domain rule (an UNLABELED next-node op after a `HAS_<Label>`-typed edge narrows to that label ONLY when candidate node ids collide across labels) instead of declining every labeled-destination pattern; (2) `alias.__gfql_node_id__` (the #1650 whole-entity identity key the aggregation lowering groups by) resolves to the bare `alias` id column (the polars bindings table intentionally omits pandas' join-residue columns — the bare alias column IS the identity key); (3) `group_by(key_prefixes=...)` expands every `<prefix>*` row-table column into the key set exactly like pandas (functionally dependent on the identity key — group sizes unchanged). Result: the official LDBC IC4 (new-topics) query — comma-MATCH, `WITH DISTINCT`, CASE projections, whole-entity `sum` aggregation, HAVING-style `WHERE` — runs natively on polars, parity-exact (harness-verified vs expected rows at SF0.1: ok, 127.8 ms). Regression test `test_polars_rows_entity_groupby.py` (pandas oracle + polars parity on a discriminating fixture). Remaining polars `rows` declines are undirected bounded var-length (IC6/IC11) and shortestPath/zero-hop-unbounded (by design).
 - **GFQL connected OPTIONAL MATCH no longer crashes on polars frames (engine-polymorphic seed extraction)**: `_optional_arm_start_nodes` (gfql_unified.py) applied pandas-only ops (`.dropna()`/`.drop_duplicates()`/`.rename(columns=)`/boolean-mask indexing) to the joined binding rows, so an IS7-shaped `MATCH ... OPTIONAL MATCH ...` on `engine='polars'` crashed with `AttributeError: 'DataFrame' object has no attribute 'dropna'` before reaching the row pipeline (hit by LDBC SNB interactive-short-7 via the official query text). Now FIXED end-to-end: the seed extraction branches on frame type (`drop_nulls().unique().rename({...})` + `filter(is_in)` on polars), the optional-arm dispatch skips the pandas-only seeded-rows contract on polars (the seed restriction is a pruning hint — arm rows are left-outer-joined on the shared aliases, so unseeded is semantically identical, and `where`-arms already ran unseeded), and the arm join/alias-synthesis block gains a polars twin (`is_in` semi-join prune, `join(how='left')`, null-preserving alias synthesis). Result: **connected OPTIONAL MATCH now runs natively on polars** (previously pandas-only), oracle-exact on the IS7 shape; the simple-CASE null-literal equality (`CASE x WHEN null`, lowered as the `__cypher_case_eq__(x, null)` marker) now lowers natively to `is_null()` on polars (pandas-parity null-mask semantics; the general `CASE x WHEN v` form still declines honestly — it carries pandas' bool/numeric cross-dtype rules), so the full LDBC IS7 query INCLUDING its `CASE r WHEN null` flag projection runs natively on polars, oracle-exact. The polars arm is also PRUNED like the pandas one: since the polars bindings-row path declines `start_nodes` by contract, the same restriction is expressed as an id-membership filter injected into the arm's first (shared) node op — measured 9.2× on the LDBC SF0.1 harness (IS7 message-replies cypher 710.8→77.2 ms). pandas/cuDF paths byte-identical. Regression tests `test_optional_match_polars_frames.py` (pandas oracle, polars native end-to-end, polars no-pandas-ism contract).
 - **GFQL polars/polars-gpu seeded traversal no longer pays an O(E) NaN re-scan per hop on float-column graphs (identity-stable `_pl_nan_to_null` + clean-cache)**: every `hop()` runs `_coerce_input_formats` → `_pl_nan_to_null`, which scanned `is_nan().any()` over every float column on the (unchanged) resident edge frame **on every call** — so repeated seeded hops on a resident graph (the seeded-Search / native-hop pattern) grew O(E) with edge count on polars while pandas stayed flat. Measured: an indexed polars `g.hop` on 8M edges with one float column was **33.8 ms and growing**; it is now **0.20 ms and FLAT** (~140× faster; O(degree)), matching pandas. `_pl_nan_to_null` now probes an eager polars frame for real NaN once, returns a clean frame UNCHANGED (restoring the #1726 identity guard that #1731 reverted — so frame-identity caches like the #1658 index keep engaging), rewrites only the columns that genuinely carry NaN (values identical to the old unconditional `fill_nan`), and caches the id of frames verified clean so later calls skip the probe (recycle-safe via `weakref.finalize`; a rebound/new frame re-probes, so NaN→null semantics are unchanged). Regression: full 4-engine `test_index.py` parity + new `TestPlNanCleanCache` cases (NaN-present cleaned, clean-frame cached same-object, distinct frames independent).
 - **GFQL seeded `gfql()`/Cypher chains now engage the resident #1658 adjacency index on ALL FOUR engines (incl. typed edges) instead of always scanning**: a seeded hop expressed as a `gfql([...])`/Cypher CHAIN (rather than a direct `g.hop(...)`) previously fell back to the O(E) scan even with a resident index, because both chain executors attach a synthetic per-edge id column to the edge frame (pandas/cuDF eager: `copy(deep=False)` + assign; polars native lazy: `with_row_index`) → a fresh edge-frame object → the index's `source_ref is df` identity guard missed. Both chain paths now re-point the resident edge adjacency indexes at the augmented frame via `GfqlIndexRegistry.rebind_edges` (provably safe: the shallow copy / `with_row_index` preserves the indexed src/dst columns by value; the structural fingerprint — row count + bound columns + engine — is unchanged by an added column). Additionally, a **simple scalar-equality `edge_match`** (typed edges, e.g. `-[:KNOWS]->` / `e_forward({"type": X})`) is now index-coverable on the wavefront path: `index_seeded_hop` applies the edge predicate to the CSR-matched rows each hop, parity-exact with the scan's `filter_edges_by_dict`. Predicate/membership-list `edge_match`, `edge_query`, and the direct-hop (non-wavefront) path deliberately stay on the scan path (no over-reach). Measured (dgx, 500k nodes / 4M edges / 4 edge types, warm median, index result == scan result): pandas typed 1-hop **2.1×** / untyped **2.7×**; cuDF typed **1.9×** / untyped **2.0×**; polars typed **1.3×** / untyped **5.4×**; polars-gpu typed engaged (1.0×) / untyped **12.4×**; 2-hop typed engaged on all four engines. 29 new differential + engagement regression tests (`test_index.py`), 4-engine, plus the existing 109-test index suite green.
diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index fd6e9ee790..fbd6f99db6 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -415,12 +415,12 @@ def _try_native_row_op(g_cur, op):
     if fn == "order_by":
         return order_by_polars(g_cur, op.params.get("keys", []))
     if fn == "group_by":
-        if op.params.get("key_prefixes"):
-            # Whole-row grouping on a bindings table (alias-prefixed key expansion):
-            # silently ignoring key_prefixes would group on the wrong keys — a
-            # wrong answer. Decline until natively ported.
-            return None
-        return group_by_polars(g_cur, op.params.get("keys", []), op.params.get("aggregations", []))
+        return group_by_polars(
+            g_cur,
+            op.params.get("keys", []),
+            op.params.get("aggregations", []),
+            key_prefixes=op.params.get("key_prefixes"),
+        )
     if fn == "unwind":
         return unwind_polars(g_cur, op.params.get("expr", ""), op.params.get("as_", "value"))
     if fn == "get_degrees":
diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index e6917ab3c7..54285b10c1 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
@@ -83,6 +83,11 @@ def _resolve_property(alias: str, prop: str, columns: Sequence[str]) -> Optional
     prefixed = f"{alias}.{prop}"
     if prefixed in columns:
         return prefixed
+    if prop == "__gfql_node_id__" and alias in columns:
+        # Whole-entity identity key (#1650 lowering groups by `alias.__gfql_node_id__`).
+        # pandas' bindings table carries it as a join-residue column; the polars table
+        # deliberately doesn't — its value IS the bare alias id column.
+        return alias
     if prop in columns and alias in columns:
         return prop
     return None
@@ -776,12 +781,28 @@ def _agg_expr(func: str, expr: Optional[str], columns: Sequence[str], alias: str
     return None
 
 
-def group_by_polars(g: Plottable, keys: Sequence[Any], aggregations: Sequence[Any]) -> Optional[Plottable]:
+def group_by_polars(
+    g: Plottable,
+    keys: Sequence[Any],
+    aggregations: Sequence[Any],
+    key_prefixes: Optional[Sequence[Any]] = None,
+) -> Optional[Plottable]:
     """Native polars group-by; None if a key/agg isn't lowerable. Matches pandas dropna=False
     (null keys kept) + non-null agg semantics; output order is first-occurrence (maintain_order),
-    though the parity gate compares order-insensitively."""
+    though the parity gate compares order-insensitively. ``key_prefixes`` mirrors the pandas
+    whole-entity expansion: every ``<prefix>*`` column of the row table joins the key set (the
+    entity's columns are functionally dependent on its identity key, so this only carries them
+    through — group sizes are unchanged)."""
     table = _active_table(g)
     cols = list(table.columns)
+    keys = [str(k) for k in keys]
+    if key_prefixes:
+        seen = set(keys)
+        for prefix in key_prefixes:
+            for col in cols:
+                if isinstance(col, str) and col.startswith(str(prefix)) and col not in seen:
+                    keys.append(col)
+                    seen.add(col)
     if not keys or not all(isinstance(k, str) and k in cols for k in keys):
         return None
     aggs: List[Any] = []
@@ -928,10 +949,6 @@ def _names(lf: pl.LazyFrame) -> List[str]:
                 return None
             if bool(op.label_seeds) or bool(op.include_zero_hop_seed):
                 return None
-            # Duplicate-id + HAS_<Label> endpoint disambiguation: pandas has a
-            # bespoke candidate-domain rule here; decline rather than diverge.
-            if RowPipelineMixin._gfql_has_edge_destination_label_col(op, nodes.columns) is not None:
-                return None
 
     node_id = str(node_id)
     src = str(src)
@@ -992,6 +1009,19 @@ def _names(lf: pl.LazyFrame) -> List[str]:
             if not isinstance(next_op, ASTNode):
                 return None
             next_nodes = filter_by_dict_polars(nodes_lf, next_op.filter_dict)
+            # HAS_<Label> destination disambiguation — polars twin of pandas'
+            # _gfql_disambiguate_has_edge_destination_nodes: when an UNLABELED next
+            # node op follows a HAS_<Label>-typed edge and candidate node ids collide
+            # across labels, narrow candidates to that label. Exactly mirrors pandas:
+            # the narrowing applies ONLY under duplicated ids (unique ids keep the
+            # unnarrowed candidate domain), and a labeled next op is left alone.
+            dis_label_col = RowPipelineMixin._gfql_has_edge_destination_label_col(edge_op, nodes.columns)
+            if dis_label_col is not None and not RowPipelineMixin._gfql_node_filter_has_label(next_op.filter_dict):
+                _ids_dup = bool(
+                    next_nodes.select(pl.col(node_id).is_duplicated().any()).collect().item()
+                )
+                if _ids_dup:
+                    next_nodes = next_nodes.filter(pl.col(dis_label_col).fill_null(False))
             next_node_ids = next_nodes.select(node_id).unique()
             if not sem.is_multihop:
                 # Filter endpoint candidates before joining from the current state.
diff --git a/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py b/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
new file mode 100644
index 0000000000..4fb0f29144
--- /dev/null
+++ b/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
@@ -0,0 +1,92 @@
+"""Regression: LDBC IC4-shaped Cypher (comma-MATCH + WITH DISTINCT + CASE + whole-entity
+sum-aggregation) runs natively on polars, parity-exact with the pandas oracle.
+
+Three polars lowerings this pins (each previously an honest NIE):
+1. HAS_<Label> destination disambiguation in ``binding_rows_polars`` (pandas'
+   ``_gfql_disambiguate_has_edge_destination_nodes`` candidate-domain rule: narrow an
+   UNLABELED next-node op after a HAS_<Label> edge to that label ONLY when candidate node
+   ids collide across labels).
+2. ``alias.__gfql_node_id__`` (whole-entity identity key, #1650) resolves to the bare
+   ``alias`` id column (the polars bindings table doesn't carry pandas' join-residue
+   columns; the bare alias column IS the identity key).
+3. ``group_by(key_prefixes=...)`` whole-entity key expansion (every ``<prefix>*`` column
+   joins the key set — functionally dependent on the identity key, so group sizes are
+   unchanged).
+"""
+import pandas as pd
+import pytest
+
+import graphistry
+
+try:
+    import polars as pl
+    HAS_POLARS = True
+except ImportError:
+    HAS_POLARS = False
+
+
+IC4_SHAPED = """
+MATCH (person:Person {id: $personId })-[:KNOWS]-(friend:Person),
+      (friend)<-[:HAS_CREATOR]-(post:Post)-[:HAS_TAG]->(tag)
+WITH DISTINCT tag, post
+WITH tag,
+     CASE
+       WHEN $startDate <= post.creationDate < $endDate THEN 1
+       ELSE 0
+     END AS valid,
+     CASE
+       WHEN post.creationDate < $startDate THEN 1
+       ELSE 0
+     END AS inValid
+WITH tag, sum(valid) AS postCount, sum(inValid) AS inValidPostCount
+WHERE postCount>0 AND inValidPostCount=0
+RETURN tag.name AS tagName, postCount
+ORDER BY postCount DESC, tagName ASC
+LIMIT 10
+"""
+
+
+def _graph_frames():
+    # person 1 KNOWS 2,3; posts 100(by 2, cD 150 in-window), 101(by 3, cD 50 BEFORE window
+    # -> its tag t2 is invalidated), 102(by 2, cD 250 in-window). Tags: t1 on 100+102
+    # (postCount 2), t2 on 101+102 (invalidated by 101). Tag node id 200 COLLIDES with
+    # nothing here; node 300 is a Forum sharing no ids (label narrowing exercised via the
+    # unlabeled `tag` op after HAS_TAG).
+    nodes = pd.DataFrame({
+        "id": [1, 2, 3, 100, 101, 102, 200, 201, 300],
+        "label__Person": [True, True, True, False, False, False, False, False, False],
+        "label__Post": [False, False, False, True, True, True, False, False, False],
+        "label__Tag": [False, False, False, False, False, False, True, True, False],
+        "label__Forum": [False, False, False, False, False, False, False, False, True],
+        "creationDate": [None, None, None, 150, 50, 250, None, None, None],
+        "name": [None, None, None, None, None, None, "t1", "t2", None],
+    })
+    edges = pd.DataFrame({
+        "src": [1, 1, 100, 101, 102, 100, 102, 101, 102],
+        "dst": [2, 3, 2, 3, 2, 200, 200, 201, 201],
+        "type": ["KNOWS", "KNOWS", "HAS_CREATOR", "HAS_CREATOR", "HAS_CREATOR",
+                 "HAS_TAG", "HAS_TAG", "HAS_TAG", "HAS_TAG"],
+    })
+    return nodes, edges
+
+
+PARAMS = {"personId": 1, "startDate": 100, "endDate": 300}
+EXPECTED = [("t1", 2)]  # t2 invalidated by pre-window post 101
+
+
+def test_ic4_shaped_pandas_oracle() -> None:
+    nodes, edges = _graph_frames()
+    g = graphistry.nodes(nodes, "id").edges(edges, "src", "dst")
+    res = g.gfql(IC4_SHAPED, params=PARAMS, engine="pandas")
+    rows = res._nodes.reset_index(drop=True)
+    assert list(zip(rows["tagName"], rows["postCount"])) == EXPECTED
+
+
+@pytest.mark.skipif(not HAS_POLARS, reason="polars not installed")
+def test_ic4_shaped_polars_native_parity() -> None:
+    nodes, edges = _graph_frames()
+    g = graphistry.nodes(pl.from_pandas(nodes), "id").edges(pl.from_pandas(edges), "src", "dst")
+    res = g.gfql(IC4_SHAPED, params=PARAMS, engine="polars")
+    rows = res._nodes.to_pandas() if hasattr(res._nodes, "to_pandas") else res._nodes
+    rows = rows.reset_index(drop=True)
+    assert list(zip(rows["tagName"], rows["postCount"])) == EXPECTED

From c08076d33c4260d0cf2a5d32e17d2f9df877b0a4 Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 12:05:16 -0700
Subject: [PATCH 2/9] fix(gfql): match pandas disambiguation call-site exactly
 (reached-domain dup probe, forward single-hop gate)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Self-review of the initial port found two parity divergences vs the pandas
call site (row/pipeline.py:3670): pandas probes id duplicates on the REACHED
candidate domain (candidate_source ∩ wavefront), not the full node table, and
applies the narrowing only for forward single-hop edges. The polars twin now
runs after the hop's state semi-join with the same gates; two new
engine-parametrized tests pin both branches (global-dup-but-unique-reached
must NOT narrow; reached-collision must narrow). 186+2 tests pass on dgx.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../gfql/lazy/engine/polars/row_pipeline.py   | 43 ++++++++++-----
 .../gfql/test_polars_rows_entity_groupby.py   | 52 +++++++++++++++++++
 2 files changed, 82 insertions(+), 13 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index 54285b10c1..b3c4b6d22f 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
@@ -1009,19 +1009,6 @@ def _names(lf: pl.LazyFrame) -> List[str]:
             if not isinstance(next_op, ASTNode):
                 return None
             next_nodes = filter_by_dict_polars(nodes_lf, next_op.filter_dict)
-            # HAS_<Label> destination disambiguation — polars twin of pandas'
-            # _gfql_disambiguate_has_edge_destination_nodes: when an UNLABELED next
-            # node op follows a HAS_<Label>-typed edge and candidate node ids collide
-            # across labels, narrow candidates to that label. Exactly mirrors pandas:
-            # the narrowing applies ONLY under duplicated ids (unique ids keep the
-            # unnarrowed candidate domain), and a labeled next op is left alone.
-            dis_label_col = RowPipelineMixin._gfql_has_edge_destination_label_col(edge_op, nodes.columns)
-            if dis_label_col is not None and not RowPipelineMixin._gfql_node_filter_has_label(next_op.filter_dict):
-                _ids_dup = bool(
-                    next_nodes.select(pl.col(node_id).is_duplicated().any()).collect().item()
-                )
-                if _ids_dup:
-                    next_nodes = next_nodes.filter(pl.col(dis_label_col).fill_null(False))
             next_node_ids = next_nodes.select(node_id).unique()
             if not sem.is_multihop:
                 # Filter endpoint candidates before joining from the current state.
@@ -1084,6 +1071,36 @@ def _names(lf: pl.LazyFrame) -> List[str]:
                 right_on=node_id,
                 how="semi",
             )
+            # HAS_<Label> destination disambiguation — polars twin of pandas'
+            # _gfql_disambiguate_has_edge_destination_nodes at its exact call site:
+            # forward single-hop only, and the duplicate probe runs on the REACHED
+            # candidate domain (full node table ∩ post-hop wavefront, pre-filter_dict
+            # — pandas' candidate_source ∩ state), NOT the whole node table. Only
+            # when ids collide there does the unlabeled next op narrow to the edge's
+            # HAS_<Label> label; unique reached ids keep the unnarrowed domain.
+            dis_label_col = RowPipelineMixin._gfql_has_edge_destination_label_col(edge_op, nodes.columns)
+            if (
+                dis_label_col is not None
+                and not sem.is_multihop
+                and edge_op.direction == "forward"
+                and not RowPipelineMixin._gfql_node_filter_has_label(next_op.filter_dict)
+            ):
+                reached_ids = state.select(pl.col("__current__").alias(node_id)).unique()
+                _reached_dup = bool(
+                    nodes_lf.select(node_id)
+                    .join(reached_ids, on=node_id, how="semi")
+                    .select(pl.col(node_id).is_duplicated().any())
+                    .collect()
+                    .item()
+                )
+                if _reached_dup:
+                    next_nodes = next_nodes.filter(pl.col(dis_label_col).fill_null(False))
+                    state = state.join(
+                        next_nodes.select(node_id).unique(),
+                        left_on="__current__",
+                        right_on=node_id,
+                        how="semi",
+                    )
             next_alias = next_op._name
             if isinstance(next_alias, str):
                 state = state.with_columns(pl.col("__current__").alias(next_alias))
diff --git a/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py b/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
index 4fb0f29144..90834f1670 100644
--- a/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
+++ b/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
@@ -90,3 +90,55 @@ def test_ic4_shaped_polars_native_parity() -> None:
     rows = res._nodes.to_pandas() if hasattr(res._nodes, "to_pandas") else res._nodes
     rows = rows.reset_index(drop=True)
     assert list(zip(rows["tagName"], rows["postCount"])) == EXPECTED
+
+
+TAG_SCAN = """
+MATCH (post:Post {id: $postId })-[:HAS_TAG]->(tag)
+RETURN tag.name AS tagName
+ORDER BY tagName
+"""
+
+
+def _disambiguation_frames():
+    # Node id 500 is duplicated across Tag/Forum but UNREACHABLE from post 600's hop;
+    # reached ids {201, 300} are unique -> pandas does NOT narrow, so the unlabeled
+    # `tag` op binds the Forum node 300 too. Node id 400 IS duplicated among reached
+    # (Tag row + Forum row) from post 601 -> pandas narrows to the Tag row only.
+    nodes = pd.DataFrame({
+        "id": [600, 601, 201, 300, 400, 400, 500, 500],
+        "label__Post": [True, True, False, False, False, False, False, False],
+        "label__Tag": [False, False, True, False, True, False, True, False],
+        "label__Forum": [False, False, False, True, False, True, False, True],
+        "name": [None, None, "t1", "f-node", "t4", "f4", "t5", "f5"],
+    })
+    edges = pd.DataFrame({
+        "src": [600, 600, 601],
+        "dst": [201, 300, 400],
+        "type": ["HAS_TAG", "HAS_TAG", "HAS_TAG"],
+    })
+    return nodes, edges
+
+
+@pytest.mark.parametrize("engine", ["pandas"] + (["polars"] if HAS_POLARS else []))
+def test_has_label_narrowing_skipped_when_reached_ids_unique(engine: str) -> None:
+    """Global id collisions must NOT trigger narrowing when the REACHED ids are unique
+    (pandas probes duplicates on candidate_source ∩ wavefront, not the whole table)."""
+    nodes, edges = _disambiguation_frames()
+    if engine == "polars":
+        nodes, edges = pl.from_pandas(nodes), pl.from_pandas(edges)
+    g = graphistry.nodes(nodes, "id").edges(edges, "src", "dst")
+    res = g.gfql(TAG_SCAN, params={"postId": 600}, engine=engine)
+    rows = res._nodes.to_pandas() if hasattr(res._nodes, "to_pandas") else res._nodes
+    assert sorted(rows["tagName"]) == ["f-node", "t1"]
+
+
+@pytest.mark.parametrize("engine", ["pandas"] + (["polars"] if HAS_POLARS else []))
+def test_has_label_narrowing_applies_on_reached_collision(engine: str) -> None:
+    """A reached id colliding across labels narrows the unlabeled op to the HAS_<Label> label."""
+    nodes, edges = _disambiguation_frames()
+    if engine == "polars":
+        nodes, edges = pl.from_pandas(nodes), pl.from_pandas(edges)
+    g = graphistry.nodes(nodes, "id").edges(edges, "src", "dst")
+    res = g.gfql(TAG_SCAN, params={"postId": 601}, engine=engine)
+    rows = res._nodes.to_pandas() if hasattr(res._nodes, "to_pandas") else res._nodes
+    assert sorted(rows["tagName"]) == ["t4"]

From 6865a4894a96fd8d7f76ca3759d959054a16efa3 Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 14:07:20 -0700
Subject: [PATCH 3/9] refactor(gfql): TypedDict-narrow group_by params + update
 stale whole-row groupby test (review)

Review feedback on #1737:
- op.params is wire-format Dict[str, Any] keyed by the runtime op.function, so
  precise typing = per-function TypedDicts narrowed at dispatch; adds
  _GroupByParams for the group_by site this PR touches (full tagged-union
  ASTCall params is a larger AST refactor, out of scope here).
- test_try_native_row_op_declines_whole_row_group_prefixes asserted the
  pre-#1737 decline behavior; whole-row (prefixed-key) grouping now lowers
  natively, so the test asserts the native result against the hand-derived
  oracle instead (was the test-polars CI failure).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 23 +++++++++++++++----
 .../gfql/test_engine_polars_row_pipeline.py   |  9 +++++---
 2 files changed, 25 insertions(+), 7 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index fbd6f99db6..695ad6fa5f 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -7,7 +7,9 @@
 (no silent pandas fallback). Deferred: variable-length/multi-hop edge sub-cases, some
 undirected multi-edge combos, node query=.
 """
-from typing import Any, List, Optional, Tuple
+from typing import Any, List, Optional, Tuple, cast
+
+from typing_extensions import TypedDict
 
 from graphistry.Plottable import Plottable
 from graphistry.compute.ast import ASTObject, ASTNode, ASTEdge
@@ -348,6 +350,14 @@ def _run_calls_polars(g_cur, calls, start_nodes, base_graph, middle):
     return g_cur
 
 
+
+class _GroupByParams(TypedDict, total=False):
+    """group_by's slice of ASTCall.params (wire JSON): keys + (out_col, agg, in_col)
+    aggregation triples + optional per-key entity prefixes ("node." / edge alias dots)."""
+    keys: List[str]
+    aggregations: List[Any]
+    key_prefixes: Optional[List[str]]
+
 def _try_native_row_op(g_cur, op):
     """Run a row-pipeline call natively on polars, or return None to defer (NIE)."""
     from graphistry.Engine import Engine
@@ -415,11 +425,16 @@ def _try_native_row_op(g_cur, op):
     if fn == "order_by":
         return order_by_polars(g_cur, op.params.get("keys", []))
     if fn == "group_by":
+        # ASTCall.params is a wire-format Dict[str, Any] whose schema is keyed by the
+        # RUNTIME value of op.function, so precise typing lives in per-function
+        # TypedDicts narrowed at dispatch (full tagged-union ASTCall is a larger AST
+        # refactor). The cast is sound: validate() has already checked these shapes.
+        gb = cast(_GroupByParams, op.params)
         return group_by_polars(
             g_cur,
-            op.params.get("keys", []),
-            op.params.get("aggregations", []),
-            key_prefixes=op.params.get("key_prefixes"),
+            gb.get("keys", []),
+            gb.get("aggregations", []),
+            key_prefixes=gb.get("key_prefixes"),
         )
     if fn == "unwind":
         return unwind_polars(g_cur, op.params.get("expr", ""), op.params.get("as_", "value"))
diff --git a/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py b/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py
index cb39a46b30..450e5e05ea 100644
--- a/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py
+++ b/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py
@@ -332,13 +332,16 @@ def test_select_extend_polars_preserves_and_updates_columns():
     assert select_extend_polars(g, [("bad", "missing.property")]) is None
 
 
-def test_try_native_row_op_declines_whole_row_group_prefixes():
-    """Whole-row grouping must defer instead of silently ignoring prefixed keys."""
+def test_try_native_row_op_handles_whole_row_group_prefixes():
+    """Whole-row grouping (prefixed keys) lowers natively via group_by_polars(key_prefixes=...)
+    instead of deferring; keys expand to their entity columns and results match the oracle."""
     from graphistry.compute.ast import group_by
     from graphistry.compute.gfql.lazy.engine.polars.chain import _try_native_row_op
 
     op = group_by(["k"], [("count", "count", "v")], key_prefixes=["node."])
-    assert _try_native_row_op(_polars_graph(), op) is None
+    out = _try_native_row_op(_polars_graph(), op)
+    assert out is not None
+    assert sorted(out._nodes.select(["k", "count"]).rows()) == [("a", 2), ("b", 2)]
 
 
 def test_polars_rows_binding_ops_undirected_self_loop_multiplicity():

From c9b54a6e9ac8f0f626e83f213154ef461cbe942e Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 14:07:44 -0700
Subject: [PATCH 4/9] ci(polars): run entity-groupby tests in the polars lane
 (changed-line gate)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 bin/test-polars.sh | 1 +
 1 file changed, 1 insertion(+)

diff --git a/bin/test-polars.sh b/bin/test-polars.sh
index 08cab09be0..4b98a1fb27 100755
--- a/bin/test-polars.sh
+++ b/bin/test-polars.sh
@@ -25,6 +25,7 @@ POLARS_TEST_FILES=(
     graphistry/tests/compute/gfql/test_conformance_ledger.py
     graphistry/tests/compute/gfql/test_polars_nan_clean.py
     graphistry/tests/compute/gfql/test_optional_match_polars_frames.py
+    graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
     # 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

From d1cf98aa7fe1f82d5d395263f63a0eae114db72a Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 16:49:06 -0700
Subject: [PATCH 5/9] refactor(gfql): precise wire-format types for polars
 row-pipeline sigs (review)

Replace Sequence[Any] with SelectItem/OrderKey/AggSpec aliases across
select/order_by/group_by lowering; group_by_polars now takes
Sequence[str] keys + Sequence[AggSpec] + Optional[Sequence[str]]
key_prefixes (matching the pandas oracle sig); _GroupByParams.aggregations
narrowed to List[AggSpec]; _agg_expr schema typed as Mapping[str, pl.DataType].

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  |  7 +-
 .../gfql/lazy/engine/polars/row_pipeline.py   | 66 ++++++++++++-------
 2 files changed, 47 insertions(+), 26 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 695ad6fa5f..487c92a8de 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -7,10 +7,13 @@
 (no silent pandas fallback). Deferred: variable-length/multi-hop edge sub-cases, some
 undirected multi-edge combos, node query=.
 """
-from typing import Any, List, Optional, Tuple, cast
+from typing import TYPE_CHECKING, Any, List, Optional, Tuple, cast
 
 from typing_extensions import TypedDict
 
+if TYPE_CHECKING:
+    from .row_pipeline import AggSpec
+
 from graphistry.Plottable import Plottable
 from graphistry.compute.ast import ASTObject, ASTNode, ASTEdge
 from .hop_eager import ensure_nodes_polars
@@ -355,7 +358,7 @@ class _GroupByParams(TypedDict, total=False):
     """group_by's slice of ASTCall.params (wire JSON): keys + (out_col, agg, in_col)
     aggregation triples + optional per-key entity prefixes ("node." / edge alias dots)."""
     keys: List[str]
-    aggregations: List[Any]
+    aggregations: List["AggSpec"]
     key_prefixes: Optional[List[str]]
 
 def _try_native_row_op(g_cur, op):
diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index b3c4b6d22f..bcbd1d7d22 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
@@ -16,7 +16,7 @@
 import contextvars
 import operator
 import re
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Sequence, Tuple
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Union, cast
 
 if TYPE_CHECKING:
     import polars as pl
@@ -27,6 +27,19 @@
 from .dtypes import is_float as _dtype_is_float, is_int as _dtype_is_int, is_numeric as _dtype_is_numeric, is_stringlike as _dtype_is_stringlike
 
 
+# Wire-format payload types for row-pipeline ops (ASTCall.params). JSON lists deserialize as
+# ``List``; Python-side lowering may pass tuples. Shapes are safelist-validated
+# (gfql/call/validation.py) before reaching these helpers, so the runtime isinstance/len
+# checks below are defense-in-depth, not the contract.
+SelectItem = Union[str, Tuple[str, JSONVal], List[JSONVal]]
+"""Projection item: ``'col'`` | ``(alias, expr)`` where a str expr is expression text and any
+other JSON scalar is a constant literal (e.g. synthetic ``__cypher_group__=1``)."""
+OrderKey = Union[Tuple[str, str], List[str]]
+"""Sort key: ``(expr, 'asc'|'desc')``."""
+AggSpec = Union[Tuple[str, str], Tuple[str, str, Optional[str]], List[Optional[str]]]
+"""Aggregation spec: ``(alias, func[, expr])``; ``expr`` None/omitted only for count(*)."""
+
+
 # Active row-table schema (col -> dtype), set around lowering so lower_expr can infer FLOAT
 # operands for the NaN guard. Free to populate — schema is already on the table, no scan.
 _SCHEMA: "contextvars.ContextVar[dict]" = contextvars.ContextVar("gfql_polars_schema", default={})
@@ -564,10 +577,12 @@ def lower_expr_str(expr: str, columns: Sequence[str]) -> Optional[pl.Expr]:
     return lower_expr(node, columns)
 
 
-def lower_select_items(items: Sequence[Any], columns: Sequence[str]) -> Optional[List[Any]]:
+def lower_select_items(items: Sequence[SelectItem], columns: Sequence[str]) -> Optional[List[Any]]:
     """Lower projection items [(alias, expr) | 'col'] to polars exprs, or None."""
     out: List[Any] = []
     for item in items:
+        alias: str
+        expr: JSONVal
         if isinstance(item, str):
             alias, expr = item, item
         elif isinstance(item, (list, tuple)) and len(item) == 2:
@@ -587,7 +602,7 @@ def lower_select_items(items: Sequence[Any], columns: Sequence[str]) -> Optional
     return out
 
 
-def lower_order_by_keys(keys: Sequence[Any], columns: Sequence[str]) -> Optional[Tuple[List[Any], List[bool]]]:
+def lower_order_by_keys(keys: Sequence[OrderKey], columns: Sequence[str]) -> Optional[Tuple[List[Any], List[bool]]]:
     """Lower order_by [(expr, direction)] to (polars exprs, descending flags)."""
     exprs: List[Any] = []
     descending: List[bool] = []
@@ -641,7 +656,7 @@ def _project_preserving_height(table: Any, exprs: List[Any]) -> Any:
     return table.select(exprs)
 
 
-def _project_polars(g: Plottable, items: Sequence[Any], extend: bool) -> Optional[Plottable]:
+def _project_polars(g: Plottable, items: Sequence[SelectItem], extend: bool) -> Optional[Plottable]:
     """Shared body of ``select_polars`` / ``with_columns_polars``; None if any item isn't
     lowerable (honest NIE, no pandas bridge)."""
     table = _active_table(g)
@@ -666,12 +681,12 @@ def _select_emits_temporal_constructor_text(out: Any) -> bool:
     return False
 
 
-def select_polars(g: Plottable, items: Sequence[Any]) -> Optional[Plottable]:
+def select_polars(g: Plottable, items: Sequence[SelectItem]) -> Optional[Plottable]:
     """Native polars projection (replaces the row table)."""
     return _project_polars(g, items, extend=False)
 
 
-def with_columns_polars(g: Plottable, items: Sequence[Any]) -> Optional[Plottable]:
+def with_columns_polars(g: Plottable, items: Sequence[SelectItem]) -> Optional[Plottable]:
     """Native polars WITH extend=True: add/overwrite columns, keep the rest. Mirrors pandas
     ``with_(extend=True)`` (``table_df.assign``): ``with_columns`` matches — an existing alias
     REPLACES in place (position kept), a new alias APPENDS at the end in item order."""
@@ -719,7 +734,7 @@ def where_rows_polars(
     return _rewrap(g, table.filter(combined))
 
 
-def order_by_polars(g: Plottable, keys: Sequence[Any]) -> Optional[Plottable]:
+def order_by_polars(g: Plottable, keys: Sequence[OrderKey]) -> Optional[Plottable]:
     """Native polars sort; None if any key isn't lowerable."""
     table = _active_table(g)
     lowered = _lower_with_schema(table, lambda: lower_order_by_keys(keys, list(table.columns)))
@@ -737,7 +752,7 @@ def order_by_polars(g: Plottable, keys: Sequence[Any]) -> Optional[Plottable]:
 
 # Native aggs: count/sum/avg/min/max/count_distinct/collect/collect_distinct; stdev/percentile
 # etc. return None → caller declines (NIE).
-def _agg_expr(func: str, expr: Optional[str], columns: Sequence[str], alias: str, schema: Optional[dict] = None) -> Optional[pl.Expr]:
+def _agg_expr(func: str, expr: Optional[str], columns: Sequence[str], alias: str, schema: Optional[Mapping[str, "pl.DataType"]] = None) -> Optional[pl.Expr]:
     import polars as pl
     func = func.lower()
     if func == "count" and (expr is None or expr == "*"):
@@ -783,9 +798,9 @@ def _agg_expr(func: str, expr: Optional[str], columns: Sequence[str], alias: str
 
 def group_by_polars(
     g: Plottable,
-    keys: Sequence[Any],
-    aggregations: Sequence[Any],
-    key_prefixes: Optional[Sequence[Any]] = None,
+    keys: Sequence[str],
+    aggregations: Sequence[AggSpec],
+    key_prefixes: Optional[Sequence[str]] = None,
 ) -> Optional[Plottable]:
     """Native polars group-by; None if a key/agg isn't lowerable. Matches pandas dropna=False
     (null keys kept) + non-null agg semantics; output order is first-occurrence (maintain_order),
@@ -795,28 +810,31 @@ def group_by_polars(
     through — group sizes are unchanged)."""
     table = _active_table(g)
     cols = list(table.columns)
-    keys = [str(k) for k in keys]
+    key_cols = [str(k) for k in keys]
     if key_prefixes:
-        seen = set(keys)
+        seen = set(key_cols)
         for prefix in key_prefixes:
             for col in cols:
-                if isinstance(col, str) and col.startswith(str(prefix)) and col not in seen:
-                    keys.append(col)
+                if isinstance(col, str) and col.startswith(prefix) and col not in seen:
+                    key_cols.append(col)
                     seen.add(col)
-    if not keys or not all(isinstance(k, str) and k in cols for k in keys):
+    if not key_cols or not all(isinstance(k, str) and k in cols for k in key_cols):
         return None
-    aggs: List[Any] = []
+    aggs: List["pl.Expr"] = []
     for agg in aggregations:
         if not isinstance(agg, (list, tuple)) or len(agg) not in (2, 3):
             return None
-        alias = str(agg[0])
-        func = str(agg[1])
-        expr = agg[2] if len(agg) == 3 else None
+        # cast: the AggSpec tuple variants make agg[2] an out-of-range index to mypy;
+        # the len guard above already proved the shape.
+        spec = cast("Sequence[Optional[str]]", agg)
+        alias = str(spec[0])
+        func = str(spec[1])
+        expr = spec[2] if len(spec) == 3 else None
         lowered = _agg_expr(func, expr, cols, alias, table.schema)
         if lowered is None:
             return None
         aggs.append(lowered)
-    out = table.group_by(list(keys), maintain_order=True).agg(aggs)
+    out = table.group_by(key_cols, maintain_order=True).agg(aggs)
     return _rewrap(g, out)
 
 
@@ -846,7 +864,7 @@ def unwind_polars(g: Plottable, expr: str, as_: str = "value") -> Optional[Plott
     return _rewrap(g, table.join(rhs, how="cross"))
 
 
-def select_extend_polars(g: Plottable, items: Sequence[Any]) -> Optional[Plottable]:
+def select_extend_polars(g: Plottable, items: Sequence[SelectItem]) -> Optional[Plottable]:
     """Native polars ``with_(items, extend=True)``: add/overwrite projected columns
     while keeping the existing row table (pandas ``assign`` semantics). Emitted by
     the bindings-path aggregate lowering (pre-aggregation group keys / agg args),
@@ -1149,9 +1167,9 @@ def _names(lf: pl.LazyFrame) -> List[str]:
     return out
 
 
-def can_select_native(items: Sequence[Any], columns: Sequence[str]) -> bool:
+def can_select_native(items: Sequence[SelectItem], columns: Sequence[str]) -> bool:
     return lower_select_items(items, columns) is not None
 
 
-def can_order_by_native(keys: Sequence[Any], columns: Sequence[str]) -> bool:
+def can_order_by_native(keys: Sequence[OrderKey], columns: Sequence[str]) -> bool:
     return lower_order_by_keys(keys, columns) is not None

From 74fe4a9fcb4d28d9972bd14eabdb47cbb996464f Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 16:49:06 -0700
Subject: [PATCH 6/9] docs(gfql): nan-clean cache soundness notes + LazyFrame
 collect_schema (deferred #1735 review)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/nan_clean.py      | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py b/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py
index c704ee5e95..dfc24bbe6a 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py
@@ -22,6 +22,12 @@
 # per-hop NaN probe on a RESIDENT graph (seeded Search / native-hop hammers the same edge
 # frame every call) from O(E)-per-call into O(1) after the first check — the dominant
 # per-call cost for polars/polars-gpu seeded traversal on float-column (i.e. real) graphs.
+#
+# Soundness rests on pl.DataFrame immutability: polars exposes no in-place cell mutation,
+# so a frame that probed clean stays clean for its lifetime (a same-id DIFFERENT frame is
+# impossible while the finalize is alive). Thread-safety: set.add/discard are atomic under
+# the GIL and eviction-vs-insert races can only lose a cache entry, never fabricate one —
+# worst case is a redundant re-probe, never a stale hit.
 _PL_NAN_CLEAN_IDS: Set[int] = set()
 
 
@@ -49,7 +55,10 @@ def _pl_nan_to_null(df: "PolarsT") -> "PolarsT":
     identity guard (reverted by #1731) AND removes the per-call O(E) re-scan that made
     polars/polars-gpu seeded Search grow with edge count (see plans/gfql-benchmark-numbers)."""
     import polars as pl
-    float_cols = [c for c, dt in df.schema.items() if dt in (pl.Float32, pl.Float64)]
+    # collect_schema(): resolves the LazyFrame schema without a PerformanceWarning
+    # (LazyFrame.schema is deprecated for that); on eager DataFrames .schema is free.
+    schema = df.collect_schema() if isinstance(df, pl.LazyFrame) else df.schema
+    float_cols = [c for c, dt in schema.items() if dt in (pl.Float32, pl.Float64)]
     if not float_cols:
         return df
     if isinstance(df, pl.DataFrame):

From 2cdc9c33e60d6dc29a093e109ba409b7099242ed Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 17:43:23 -0700
Subject: [PATCH 7/9] fix(gfql): HAS_<Label> dup-id graphs decline to honest
 NIE in binding_rows_polars (review F1)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The disambiguation gate probed the chain-combine node table, which is already
deduplicated by id — the duplicate probe could never fire, and with the old
decline removed, duplicate-id graphs returned silently row-order-dependent
first-occurrence rows where pandas narrows to the HAS_<Label> label row
(reproduced end-to-end). The label row pandas narrows to may already be gone
from the deduped table, so a faithful native answer is impossible here: probe
the pre-chain base graph (_gfql_rows_base_graph) instead — unique-id graphs
stay native (pandas would not narrow either; IC4 keeps its win), duplicate-id
graphs decline to the honest NIE.

Also from review: binding-rows-path regression tests that actually reach the
gate (dup-id NIE both row orders + unique-id parity), key_prefixes genuine
expansion/overlap test, malformed-AggSpec decline test, nan_clean soundness
comment corrected (polars does have in-place mutation APIs; state the real
no-mutation contract), AggSpec import made runtime (introspectable TypedDict),
lower_select_items/lower_order_by_keys return List[pl.Expr].

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 10 +-
 .../gfql/lazy/engine/polars/nan_clean.py      | 13 ++-
 .../gfql/lazy/engine/polars/row_pipeline.py   | 57 ++++++------
 .../gfql/test_engine_polars_row_pipeline.py   | 34 +++++++
 .../gfql/test_polars_rows_entity_groupby.py   | 92 +++++++++++++++++--
 5 files changed, 165 insertions(+), 41 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 487c92a8de..86a3e7627c 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -7,12 +7,14 @@
 (no silent pandas fallback). Deferred: variable-length/multi-hop edge sub-cases, some
 undirected multi-edge combos, node query=.
 """
-from typing import TYPE_CHECKING, Any, List, Optional, Tuple, cast
+from typing import Any, List, Optional, Tuple, cast
 
 from typing_extensions import TypedDict
 
-if TYPE_CHECKING:
-    from .row_pipeline import AggSpec
+# Runtime import (not TYPE_CHECKING): AggSpec is a pure typing Union of builtins —
+# row_pipeline imports polars lazily, so this is safe without polars installed, and it
+# keeps _GroupByParams introspectable (get_type_hints) at runtime.
+from .row_pipeline import AggSpec
 
 from graphistry.Plottable import Plottable
 from graphistry.compute.ast import ASTObject, ASTNode, ASTEdge
@@ -358,7 +360,7 @@ class _GroupByParams(TypedDict, total=False):
     """group_by's slice of ASTCall.params (wire JSON): keys + (out_col, agg, in_col)
     aggregation triples + optional per-key entity prefixes ("node." / edge alias dots)."""
     keys: List[str]
-    aggregations: List["AggSpec"]
+    aggregations: List[AggSpec]
     key_prefixes: Optional[List[str]]
 
 def _try_native_row_op(g_cur, op):
diff --git a/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py b/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py
index dfc24bbe6a..3102799fa5 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py
@@ -23,11 +23,14 @@
 # frame every call) from O(E)-per-call into O(1) after the first check — the dominant
 # per-call cost for polars/polars-gpu seeded traversal on float-column (i.e. real) graphs.
 #
-# Soundness rests on pl.DataFrame immutability: polars exposes no in-place cell mutation,
-# so a frame that probed clean stays clean for its lifetime (a same-id DIFFERENT frame is
-# impossible while the finalize is alive). Thread-safety: set.add/discard are atomic under
-# the GIL and eviction-vs-insert races can only lose a cache entry, never fabricate one —
-# worst case is a redundant re-probe, never a stale hit.
+# Soundness contract: frames handed to GFQL must not be mutated in place afterwards.
+# polars DOES expose in-place mutation (``extend``, ``replace_column``, ``insert_column``,
+# ``hstack(in_place=True)``) that keeps ``id()`` stable — a caller injecting NaN through
+# those after a hop would get a stale "clean" verdict. This is the same implicit contract
+# the pandas-input paths already rely on (``pl.from_pandas(nan_to_null=True)`` snapshots
+# at entry); typical GFQL usage rebinds frames rather than mutating them. Thread-safety:
+# set.add/discard are atomic under the GIL and eviction-vs-insert races can only lose a
+# cache entry, never fabricate one — worst case a redundant re-probe, never a stale hit.
 _PL_NAN_CLEAN_IDS: Set[int] = set()
 
 
diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index bcbd1d7d22..eacaf463cf 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
@@ -577,9 +577,9 @@ def lower_expr_str(expr: str, columns: Sequence[str]) -> Optional[pl.Expr]:
     return lower_expr(node, columns)
 
 
-def lower_select_items(items: Sequence[SelectItem], columns: Sequence[str]) -> Optional[List[Any]]:
+def lower_select_items(items: Sequence[SelectItem], columns: Sequence[str]) -> Optional[List["pl.Expr"]]:
     """Lower projection items [(alias, expr) | 'col'] to polars exprs, or None."""
-    out: List[Any] = []
+    out: List["pl.Expr"] = []
     for item in items:
         alias: str
         expr: JSONVal
@@ -602,9 +602,9 @@ def lower_select_items(items: Sequence[SelectItem], columns: Sequence[str]) -> O
     return out
 
 
-def lower_order_by_keys(keys: Sequence[OrderKey], columns: Sequence[str]) -> Optional[Tuple[List[Any], List[bool]]]:
+def lower_order_by_keys(keys: Sequence[OrderKey], columns: Sequence[str]) -> Optional[Tuple[List["pl.Expr"], List[bool]]]:
     """Lower order_by [(expr, direction)] to (polars exprs, descending flags)."""
-    exprs: List[Any] = []
+    exprs: List["pl.Expr"] = []
     descending: List[bool] = []
     for key in keys:
         if not isinstance(key, (list, tuple)) or len(key) != 2:
@@ -897,8 +897,10 @@ def binding_rows_polars(
     Returns None to DECLINE (caller raises the honest NIE) for anything outside
     the supported subset: variable-length/multi-hop edges, shortestPath scalar
     bindings, node ``query=`` / edge query or endpoint-match params, hop labels,
-    HAS_-label destination disambiguation, seeded re-entry contexts, cartesian
-    (node-only) mode, and the legacy ``alias_endpoints`` variant. NO-CHEATING:
+    HAS_-label destination disambiguation on duplicate-node-id graphs (unique-id
+    graphs run native — pandas would not narrow there either), seeded re-entry
+    contexts, cartesian (node-only) mode, and the legacy ``alias_endpoints``
+    variant. NO-CHEATING:
     never bridges to pandas. Parity gate: differential tests vs the pandas oracle.
     """
     import polars as pl
@@ -1089,13 +1091,18 @@ def _names(lf: pl.LazyFrame) -> List[str]:
                 right_on=node_id,
                 how="semi",
             )
-            # HAS_<Label> destination disambiguation — polars twin of pandas'
-            # _gfql_disambiguate_has_edge_destination_nodes at its exact call site:
-            # forward single-hop only, and the duplicate probe runs on the REACHED
-            # candidate domain (full node table ∩ post-hop wavefront, pre-filter_dict
-            # — pandas' candidate_source ∩ state), NOT the whole node table. Only
-            # when ids collide there does the unlabeled next op narrow to the edge's
-            # HAS_<Label> label; unique reached ids keep the unnarrowed domain.
+            # HAS_<Label> destination disambiguation (pandas'
+            # _gfql_disambiguate_has_edge_destination_nodes): on DUPLICATE-id graphs
+            # pandas narrows the unlabeled next op to the edge's HAS_<Label> rows
+            # taken from the ORIGINAL node table (its base_nodes still carries the
+            # colliding label rows). By the time this rows op runs, ``g``'s node
+            # table is the chain-combine result — deduplicated by id, first
+            # occurrence kept — so the label row pandas narrows to may already be
+            # GONE here and a native answer would be silently row-order-dependent.
+            # Unique-id graphs need no narrowing (pandas' duplicated() probe is
+            # False) → native is parity-exact; duplicate-id graphs DECLINE (honest
+            # NIE). Probe the pre-chain base graph stashed by _run_calls_polars —
+            # if it isn't available, decline (can't prove uniqueness).
             dis_label_col = RowPipelineMixin._gfql_has_edge_destination_label_col(edge_op, nodes.columns)
             if (
                 dis_label_col is not None
@@ -1103,22 +1110,20 @@ def _names(lf: pl.LazyFrame) -> List[str]:
                 and edge_op.direction == "forward"
                 and not RowPipelineMixin._gfql_node_filter_has_label(next_op.filter_dict)
             ):
-                reached_ids = state.select(pl.col("__current__").alias(node_id)).unique()
-                _reached_dup = bool(
-                    nodes_lf.select(node_id)
-                    .join(reached_ids, on=node_id, how="semi")
-                    .select(pl.col(node_id).is_duplicated().any())
+                base_graph = getattr(g, "_gfql_rows_base_graph", None)
+                base_nodes = getattr(base_graph, "_nodes", None)
+                if base_nodes is None:
+                    return None
+                base_lf = base_nodes.lazy()
+                if node_id not in base_lf.collect_schema().names():
+                    return None
+                _base_dup = bool(
+                    base_lf.select(pl.col(node_id).is_duplicated().any())
                     .collect()
                     .item()
                 )
-                if _reached_dup:
-                    next_nodes = next_nodes.filter(pl.col(dis_label_col).fill_null(False))
-                    state = state.join(
-                        next_nodes.select(node_id).unique(),
-                        left_on="__current__",
-                        right_on=node_id,
-                        how="semi",
-                    )
+                if _base_dup:
+                    return None
             next_alias = next_op._name
             if isinstance(next_alias, str):
                 state = state.with_columns(pl.col("__current__").alias(next_alias))
diff --git a/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py b/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py
index 450e5e05ea..0a61853003 100644
--- a/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py
+++ b/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py
@@ -344,6 +344,40 @@ def test_try_native_row_op_handles_whole_row_group_prefixes():
     assert sorted(out._nodes.select(["k", "count"]).rows()) == [("a", 2), ("b", 2)]
 
 
+def test_group_by_polars_key_prefix_expands_and_dedups_overlap():
+    """key_prefixes genuinely expands `<prefix>*` columns into the key set AND skips
+    columns already present in keys (no polars DuplicateError; pandas twin:
+    test_row_pipeline_ops key-overlap case)."""
+    import polars as pl
+    from graphistry.compute.gfql.lazy.engine.polars.row_pipeline import group_by_polars
+
+    nodes = pl.DataFrame({
+        "tag.id": [1, 1, 2, 2],
+        "tag.name": ["a", "a", "b", "b"],
+        "v": [10, 20, 30, 40],
+    })
+    g = graphistry.nodes(nodes)
+    # "tag." expands to tag.id (already a key -> deduped) + tag.name (appended)
+    out = group_by_polars(g, ["tag.id"], [("total", "sum", "v")], key_prefixes=["tag."])
+    assert out is not None
+    assert sorted(out._nodes.columns) == ["tag.id", "tag.name", "total"]
+    assert sorted(out._nodes.select(["tag.id", "tag.name", "total"]).rows()) == [
+        (1, "a", 30), (2, "b", 70),
+    ]
+
+
+def test_group_by_polars_malformed_agg_declines_not_crashes():
+    """Runtime defense-in-depth: malformed AggSpec shapes and unsupported agg
+    functions decline to None (caller raises honest NIE), never crash."""
+    from graphistry.compute.gfql.lazy.engine.polars.row_pipeline import group_by_polars
+
+    g = _polars_graph()
+    assert group_by_polars(g, ["k"], ["not-a-spec"]) is None  # non-list/tuple item
+    assert group_by_polars(g, ["k"], [("alias",)]) is None  # wrong arity
+    assert group_by_polars(g, ["k"], [("alias", "stdev", "v")]) is None  # unsupported func
+    assert group_by_polars(g, ["missing"], [("count", "count", "v")]) is None  # bad key
+
+
 def test_polars_rows_binding_ops_undirected_self_loop_multiplicity():
     """Raw undirected bindings preserve both orientations, including two self-loop paths."""
     from graphistry.compute.ast import e_undirected, n, rows, serialize_binding_ops
diff --git a/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py b/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
index 90834f1670..750d097388 100644
--- a/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
+++ b/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
@@ -2,10 +2,11 @@
 sum-aggregation) runs natively on polars, parity-exact with the pandas oracle.
 
 Three polars lowerings this pins (each previously an honest NIE):
-1. HAS_<Label> destination disambiguation in ``binding_rows_polars`` (pandas'
-   ``_gfql_disambiguate_has_edge_destination_nodes`` candidate-domain rule: narrow an
-   UNLABELED next-node op after a HAS_<Label> edge to that label ONLY when candidate node
-   ids collide across labels).
+1. HAS_<Label> destination gate in ``binding_rows_polars``: on UNIQUE-node-id graphs the
+   pattern runs native (pandas' ``_gfql_disambiguate_has_edge_destination_nodes`` would
+   not narrow there either — parity-exact); on DUPLICATE-id graphs polars declines to the
+   honest NIE (the chain-combine result has already deduplicated nodes, so the label row
+   pandas narrows to may be gone — a native answer would be row-order-dependent).
 2. ``alias.__gfql_node_id__`` (whole-entity identity key, #1650) resolves to the bare
    ``alias`` id column (the polars bindings table doesn't carry pandas' join-residue
    columns; the bare alias column IS the identity key).
@@ -122,7 +123,10 @@ def _disambiguation_frames():
 @pytest.mark.parametrize("engine", ["pandas"] + (["polars"] if HAS_POLARS else []))
 def test_has_label_narrowing_skipped_when_reached_ids_unique(engine: str) -> None:
     """Global id collisions must NOT trigger narrowing when the REACHED ids are unique
-    (pandas probes duplicates on candidate_source ∩ wavefront, not the whole table)."""
+    (pandas probes duplicates on candidate_source ∩ wavefront, not the whole table).
+    NOTE: this single-MATCH shape routes through the plain chain executor, NOT
+    ``binding_rows_polars`` — it pins that path's parity; the binding-rows gate is
+    pinned by the ``test_binding_rows_dup_id_*`` tests below."""
     nodes, edges = _disambiguation_frames()
     if engine == "polars":
         nodes, edges = pl.from_pandas(nodes), pl.from_pandas(edges)
@@ -134,7 +138,8 @@ def test_has_label_narrowing_skipped_when_reached_ids_unique(engine: str) -> Non
 
 @pytest.mark.parametrize("engine", ["pandas"] + (["polars"] if HAS_POLARS else []))
 def test_has_label_narrowing_applies_on_reached_collision(engine: str) -> None:
-    """A reached id colliding across labels narrows the unlabeled op to the HAS_<Label> label."""
+    """A reached id colliding across labels narrows the unlabeled op to the HAS_<Label> label.
+    NOTE: single-MATCH shape — plain chain executor path, not ``binding_rows_polars``."""
     nodes, edges = _disambiguation_frames()
     if engine == "polars":
         nodes, edges = pl.from_pandas(nodes), pl.from_pandas(edges)
@@ -142,3 +147,78 @@ def test_has_label_narrowing_applies_on_reached_collision(engine: str) -> None:
     res = g.gfql(TAG_SCAN, params={"postId": 601}, engine=engine)
     rows = res._nodes.to_pandas() if hasattr(res._nodes, "to_pandas") else res._nodes
     assert sorted(rows["tagName"]) == ["t4"]
+
+
+# Comma-MATCH multi-alias shape: lowers to rows(binding_ops=...) and (on polars) routes
+# through binding_rows_polars — the path carrying the HAS_<Label> duplicate-id gate.
+COLLIDE_BINDINGS = """
+MATCH (person:Person {id: $personId })-[:KNOWS]-(friend:Person),
+      (friend)<-[:HAS_CREATOR]-(post:Post)-[:HAS_TAG]->(tag)
+WITH DISTINCT tag, post
+RETURN tag.name AS tagName, post.creationDate AS cd
+ORDER BY tagName, cd
+"""
+
+
+def _dup_id_frames(tag_first: bool):
+    # Node id 200 has TWO rows: a Tag row ('t1') and a Forum row ('FX'); `tag_first`
+    # controls their order so a first-occurrence-dependent answer flips between runs.
+    # pandas narrows to the Tag row in BOTH orders; a deduplicated native polars run
+    # would return whichever row happened to come first — hence the honest NIE.
+    rows = [
+        (1, True, False, False, False, None, None),
+        (2, True, False, False, False, None, None),
+        (100, False, True, False, False, 150, None),
+    ]
+    tag_row = (200, False, False, True, False, None, "t1")
+    forum_row = (200, False, False, False, True, None, "FX")
+    rows += [tag_row, forum_row] if tag_first else [forum_row, tag_row]
+    nodes = pd.DataFrame(
+        rows,
+        columns=["id", "label__Person", "label__Post", "label__Tag", "label__Forum",
+                 "creationDate", "name"],
+    )
+    edges = pd.DataFrame({
+        "src": [1, 100, 100],
+        "dst": [2, 2, 200],
+        "type": ["KNOWS", "HAS_CREATOR", "HAS_TAG"],
+    })
+    return nodes, edges
+
+
+@pytest.mark.parametrize("tag_first", [True, False])
+def test_binding_rows_dup_id_pandas_narrows_order_independent(tag_first: bool) -> None:
+    """Oracle: pandas narrows to the HAS_TAG label row regardless of node row order."""
+    nodes, edges = _dup_id_frames(tag_first)
+    g = graphistry.nodes(nodes, "id").edges(edges, "src", "dst")
+    res = g.gfql(COLLIDE_BINDINGS, params={"personId": 1}, engine="pandas")
+    assert sorted(zip(res._nodes["tagName"], res._nodes["cd"])) == [("t1", 150.0)]
+
+
+@pytest.mark.skipif(not HAS_POLARS, reason="polars not installed")
+@pytest.mark.parametrize("tag_first", [True, False])
+def test_binding_rows_dup_id_polars_declines_nie(tag_first: bool) -> None:
+    """Duplicate node ids + HAS_<Label> gate shape → honest NIE on polars, never a
+    silently row-order-dependent native answer (parity-or-error contract)."""
+    nodes, edges = _dup_id_frames(tag_first)
+    g = graphistry.nodes(pl.from_pandas(nodes), "id").edges(pl.from_pandas(edges), "src", "dst")
+    with pytest.raises(NotImplementedError):
+        g.gfql(COLLIDE_BINDINGS, params={"personId": 1}, engine="polars")
+
+
+@pytest.mark.skipif(not HAS_POLARS, reason="polars not installed")
+def test_binding_rows_unique_ids_native_parity() -> None:
+    """Same gate shape with UNIQUE node ids stays native and parity-exact (the Forum
+    row gets its own id, so pandas would not narrow either)."""
+    nodes, edges = _dup_id_frames(tag_first=True)
+    nodes = nodes.copy()
+    nodes.loc[nodes["label__Forum"].fillna(False).astype(bool), "id"] = 300
+    gp = graphistry.nodes(nodes, "id").edges(edges, "src", "dst")
+    expected = sorted(zip(
+        gp.gfql(COLLIDE_BINDINGS, params={"personId": 1}, engine="pandas")._nodes["tagName"],
+        gp.gfql(COLLIDE_BINDINGS, params={"personId": 1}, engine="pandas")._nodes["cd"],
+    ))
+    gl = graphistry.nodes(pl.from_pandas(nodes), "id").edges(pl.from_pandas(edges), "src", "dst")
+    res = gl.gfql(COLLIDE_BINDINGS, params={"personId": 1}, engine="polars")
+    out = res._nodes.to_pandas() if hasattr(res._nodes, "to_pandas") else res._nodes
+    assert sorted(zip(out["tagName"], out["cd"])) == expected == [("t1", 150.0)]

From 48e96d4db9af5f0d6fafb8ce10caa4c946159d0c Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 19:38:41 -0700
Subject: [PATCH 8/9] refactor(gfql): hoist SelectItem/OrderKey/AggSpec to
 engine-neutral call/support (review)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The wire-format payload types describe ASTCall.params for every engine, not
just polars: move them next to their runtime validators in
gfql/call/support.py (single source of truth, kept in lockstep) and adopt
them in the engine-neutral consumers — pandas RowPipelineMixin.group_by,
the ast.group_by builder, and the cypher lowering's aggregation lists
(TYPE_CHECKING imports where the call package __init__ would cycle).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 graphistry/compute/ast.py                      |  5 ++++-
 graphistry/compute/gfql/call/support.py        | 18 +++++++++++++++++-
 graphistry/compute/gfql/cypher/lowering.py     | 13 +++++++++----
 .../compute/gfql/lazy/engine/polars/chain.py   |  7 +++----
 .../gfql/lazy/engine/polars/row_pipeline.py    | 15 +++------------
 graphistry/compute/gfql/row/pipeline.py        |  3 ++-
 6 files changed, 38 insertions(+), 23 deletions(-)

diff --git a/graphistry/compute/ast.py b/graphistry/compute/ast.py
index af15ebca0a..8f68292e2c 100644
--- a/graphistry/compute/ast.py
+++ b/graphistry/compute/ast.py
@@ -7,6 +7,9 @@
 
 if TYPE_CHECKING:
     from graphistry.compute.chain import Chain
+    # TYPE_CHECKING only: graphistry.compute.gfql.call.__init__ pulls in executor/validation
+    # at import time — a module-level import here would be circular.
+    from graphistry.compute.gfql.call.support import AggSpec
 
 from graphistry.Engine import Engine, EngineAbstract
 from graphistry.compute.dataframe_utils import dbg_df
@@ -1809,7 +1812,7 @@ def count_table(
 
 def group_by(
     keys: Iterable[str],
-    aggregations: Iterable[Sequence[Any]],
+    aggregations: Iterable["AggSpec"],
     key_prefixes: Optional[Iterable[str]] = None,
 ) -> ASTCall:
     """Create grouped aggregation operation for row pipelines.
diff --git a/graphistry/compute/gfql/call/support.py b/graphistry/compute/gfql/call/support.py
index 5637a7fd2d..b0b6f20ca6 100644
--- a/graphistry/compute/gfql/call/support.py
+++ b/graphistry/compute/gfql/call/support.py
@@ -1,8 +1,24 @@
 """Shared non-parser helpers for GFQL call safelist definitions."""
 
-from typing import Any, Callable, Dict, List, Set
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
 
 from graphistry.compute.gfql.language_defs import GFQL_AGGREGATION_FUNCTIONS
+from graphistry.utils.json import JSONVal
+
+
+# Wire-format payload types for row-pipeline ASTCall.params, ENGINE-NEUTRAL (pandas, polars,
+# cuDF consumers alike). JSON deserialization turns tuples into ``List``; Python-side builders
+# (ast.py — cf. the builder-side ``ProjectionItem``) may pass tuples. Each alias mirrors the
+# runtime validator of the same shape below / in validation.py — keep them in lockstep.
+SelectItem = Union[str, Tuple[str, JSONVal], List[JSONVal]]
+"""Projection item: ``'col'`` | ``(alias, expr)`` where a str expr is expression text and any
+other JSON scalar is a constant literal (e.g. synthetic ``__cypher_group__=1``).
+Validator: ``is_projection_items``."""
+OrderKey = Union[Tuple[str, str], List[str]]
+"""Sort key: ``(expr, 'asc'|'desc')``. Validator: ``is_order_keys``."""
+AggSpec = Union[Tuple[str, str], Tuple[str, str, Optional[str]], List[Optional[str]]]
+"""Aggregation spec: ``(alias, func[, expr])``; ``expr`` None/omitted only for count(*).
+Validator: ``is_list_of_agg_specs``."""
 
 
 def is_string(v: object) -> bool:
diff --git a/graphistry/compute/gfql/cypher/lowering.py b/graphistry/compute/gfql/cypher/lowering.py
index a47aa415a5..65895bd6fa 100644
--- a/graphistry/compute/gfql/cypher/lowering.py
+++ b/graphistry/compute/gfql/cypher/lowering.py
@@ -3,10 +3,15 @@
 from dataclasses import dataclass, field, replace
 import math
 import re
-from typing import AbstractSet, Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple, Union, cast
+from typing import TYPE_CHECKING, AbstractSet, Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple, Union, cast
 from typing_extensions import Literal
 import pandas as pd
 
+if TYPE_CHECKING:
+    # TYPE_CHECKING only: the gfql.call package __init__ imports executor/validation at
+    # import time - a module-level import here would be circular.
+    from graphistry.compute.gfql.call.support import AggSpec
+
 from graphistry.compute.ast import (
     ASTCall,
     ASTEdge,
@@ -4876,7 +4881,7 @@ def _lower_match_alias_aggregate_stage(
             projected_property_outputs.setdefault(prop, output_name)
             output_to_source_property[output_name] = prop
 
-    aggregations: List[Sequence[Any]] = []
+    aggregations: List["AggSpec"] = []
     for agg_spec in aggregate_specs:
         if agg_spec.output_name in available_columns:
             raise _unsupported(
@@ -5440,7 +5445,7 @@ def _lower_row_column_aggregate_stage(
             alias_name=item.alias,
         )
 
-    aggregations: List[Sequence[Any]] = []
+    aggregations: List["AggSpec"] = []
     for agg_spec in aggregate_specs:
         if agg_spec.output_name in available_columns:
             raise _unsupported(
@@ -6998,7 +7003,7 @@ def _lower_general_row_projection(
                 alias_name=item.alias,
             )
 
-        aggregations: List[Sequence[Any]] = []
+        aggregations: List["AggSpec"] = []
         for agg_spec in aggregate_specs:
             if agg_spec.output_name in available_columns:
                 raise _unsupported(
diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 86a3e7627c..80901e8f1a 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -11,10 +11,9 @@
 
 from typing_extensions import TypedDict
 
-# Runtime import (not TYPE_CHECKING): AggSpec is a pure typing Union of builtins —
-# row_pipeline imports polars lazily, so this is safe without polars installed, and it
-# keeps _GroupByParams introspectable (get_type_hints) at runtime.
-from .row_pipeline import AggSpec
+# Runtime import (not TYPE_CHECKING): AggSpec is a pure typing Union of builtins (engine-
+# neutral wire type), and it keeps _GroupByParams introspectable (get_type_hints) at runtime.
+from graphistry.compute.gfql.call.support import AggSpec
 
 from graphistry.Plottable import Plottable
 from graphistry.compute.ast import ASTObject, ASTNode, ASTEdge
diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index eacaf463cf..6ed3a9d75c 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
@@ -24,20 +24,11 @@
 
 from graphistry.Plottable import Plottable
 from graphistry.utils.json import JSONVal
-from .dtypes import is_float as _dtype_is_float, is_int as _dtype_is_int, is_numeric as _dtype_is_numeric, is_stringlike as _dtype_is_stringlike
-
-
-# Wire-format payload types for row-pipeline ops (ASTCall.params). JSON lists deserialize as
-# ``List``; Python-side lowering may pass tuples. Shapes are safelist-validated
+# Engine-neutral wire-format payload types (ASTCall.params). Shapes are safelist-validated
 # (gfql/call/validation.py) before reaching these helpers, so the runtime isinstance/len
 # checks below are defense-in-depth, not the contract.
-SelectItem = Union[str, Tuple[str, JSONVal], List[JSONVal]]
-"""Projection item: ``'col'`` | ``(alias, expr)`` where a str expr is expression text and any
-other JSON scalar is a constant literal (e.g. synthetic ``__cypher_group__=1``)."""
-OrderKey = Union[Tuple[str, str], List[str]]
-"""Sort key: ``(expr, 'asc'|'desc')``."""
-AggSpec = Union[Tuple[str, str], Tuple[str, str, Optional[str]], List[Optional[str]]]
-"""Aggregation spec: ``(alias, func[, expr])``; ``expr`` None/omitted only for count(*)."""
+from graphistry.compute.gfql.call.support import AggSpec, OrderKey, SelectItem
+from .dtypes import is_float as _dtype_is_float, is_int as _dtype_is_int, is_numeric as _dtype_is_numeric, is_stringlike as _dtype_is_stringlike
 
 
 # Active row-table schema (col -> dtype), set around lowering so lower_expr can infer FLOAT
diff --git a/graphistry/compute/gfql/row/pipeline.py b/graphistry/compute/gfql/row/pipeline.py
index e900189336..7e9a0d9064 100644
--- a/graphistry/compute/gfql/row/pipeline.py
+++ b/graphistry/compute/gfql/row/pipeline.py
@@ -20,6 +20,7 @@
 )
 from graphistry.compute.exceptions import ErrorCode, GFQLValidationError
 from graphistry.compute.dataframe_utils import concat_frames
+from graphistry.compute.gfql.call.support import AggSpec
 from graphistry.compute.gfql.row import frame_ops as row_frame_ops
 from graphistry.compute.gfql.row.prefilter import AliasPrefilters
 from graphistry.compute.typing import DataFrameT
@@ -4993,7 +4994,7 @@ def unwind(self, expr: Any, as_: str = "value") -> "Plottable":
     def group_by(
         self,
         keys: Sequence[str],
-        aggregations: Sequence[Sequence[Any]],
+        aggregations: Sequence[AggSpec],
         key_prefixes: Optional[Sequence[str]] = None,
     ) -> "Plottable":
         """Vectorized grouped aggregations for row-table pipelines."""

From 63ba3645e3fbe0731557a835857b3eaaba82e815 Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 20:31:55 -0700
Subject: [PATCH 9/9] chore(gfql): bump cypher surface-guard baseline
 9244->9249

Pure typing growth in cypher/lowering.py from the engine-neutral
SelectItem/OrderKey/AggSpec hoist (TYPE_CHECKING import block +
List["AggSpec"] annotations at 3 call sites). No new compiled-surface
fields or properties.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 bin/ci_cypher_surface_guard_baseline.json | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/bin/ci_cypher_surface_guard_baseline.json b/bin/ci_cypher_surface_guard_baseline.json
index ef5aa58e40..615454cca5 100644
--- a/bin/ci_cypher_surface_guard_baseline.json
+++ b/bin/ci_cypher_surface_guard_baseline.json
@@ -13,5 +13,5 @@
       "max_properties": 0
     }
   },
-  "lowering_py_max_lines": 9244
+  "lowering_py_max_lines": 9249
 }