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/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
 }
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
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 fd6e9ee790..80901e8f1a 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -7,7 +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
+from typing import Any, List, Optional, Tuple, cast
+
+from typing_extensions import TypedDict
+
+# 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
@@ -348,6 +354,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[AggSpec]
+    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,12 +429,17 @@ 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", []))
+        # 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,
+            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"))
     if fn == "get_degrees":
diff --git a/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py b/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py
index c704ee5e95..3102799fa5 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/nan_clean.py
@@ -22,6 +22,15 @@
 # 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 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()
 
 
@@ -49,7 +58,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):
diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index e6917ab3c7..6ed3a9d75c 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
@@ -24,6 +24,10 @@
 
 from graphistry.Plottable import Plottable
 from graphistry.utils.json import JSONVal
+# 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.
+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
 
 
@@ -83,6 +87,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
@@ -559,10 +568,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["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
         if isinstance(item, str):
             alias, expr = item, item
         elif isinstance(item, (list, tuple)) and len(item) == 2:
@@ -582,9 +593,9 @@ 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["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:
@@ -636,7 +647,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)
@@ -661,12 +672,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."""
@@ -714,7 +725,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)))
@@ -732,7 +743,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 == "*"):
@@ -776,26 +787,45 @@ 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[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),
-    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)
-    if not keys or not all(isinstance(k, str) and k in cols for k in keys):
+    key_cols = [str(k) for k in keys]
+    if key_prefixes:
+        seen = set(key_cols)
+        for prefix in key_prefixes:
+            for col in cols:
+                if isinstance(col, str) and col.startswith(prefix) and col not in seen:
+                    key_cols.append(col)
+                    seen.add(col)
+    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)
 
 
@@ -825,7 +855,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),
@@ -858,8 +888,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
@@ -928,10 +960,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)
@@ -1054,6 +1082,39 @@ def _names(lf: pl.LazyFrame) -> List[str]:
                 right_on=node_id,
                 how="semi",
             )
+            # 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
+                and not sem.is_multihop
+                and edge_op.direction == "forward"
+                and not RowPipelineMixin._gfql_node_filter_has_label(next_op.filter_dict)
+            ):
+                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 _base_dup:
+                    return None
             next_alias = next_op._name
             if isinstance(next_alias, str):
                 state = state.with_columns(pl.col("__current__").alias(next_alias))
@@ -1102,9 +1163,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
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."""
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..0a61853003 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,50 @@ 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_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():
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..750d097388
--- /dev/null
+++ b/graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
@@ -0,0 +1,224 @@
+"""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 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).
+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
+
+
+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).
+    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)
+    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.
+    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)
+    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"]
+
+
+# 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)]