feat(gfql): polars whole-entity aggregation Cypher — HAS_label disambiguation + identity-key + key_prefixes (LDBC IC4 native)

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.

bin/ci_cypher_surface_guard_baseline.json

@@ -13,5 +13,5 @@
       "max_properties": 0
     }
   },
-  "lowering_py_max_lines": 9244
+  "lowering_py_max_lines": 9249
 }

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

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.

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:

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(

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

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