feat(gfql): native polars whole-entity count(DISTINCT n) / identity aggregation (#1709)

CHANGELOG.md

@@ -14,6 +14,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Fixed
 - **GFQL polars engine natively runs `UNWIND` of a carried `collect()` list column**: `... WITH collect(x) AS xs UNWIND xs AS y RETURN ...` (exploding a list-valued column produced by `collect()`, the row-pipeline analogue of the IC6 unwind shape) previously declined on polars — `unwind_polars` only accepted a scalar literal list. Now explodes a `List`-dtype column reference (`with_columns` copy → `filter(list.len() > 0)` → `explode`), matching the pandas oracle exactly: empty-list/null cells → 0 rows, nulls *within* a list survive, source column retained; the pre-filter also makes it independent of the polars-2.0 `empty_as_null` default. Non-list columns, name collisions, unknown identifiers, and nested-list literals still decline (honest NIE, no pandas bridge). Differential fuzz vs pandas over ~3500 collect→UNWIND→{RETURN, grouped, filtered, ORDER BY, aggregate, nested-unwind} queries: 0 disagreements. (The `MATCH ... WITH collect(..) UNWIND .. MATCH` form — IC6 with a trailing re-traversal — is compile-rewritten into WITH→MATCH reentry, a separate path.) Tests in `test_engine_polars_row_pipeline.py`.
 - **GFQL polars engine natively runs the LDBC IC11/IC6 undirected variable-length bindings table (`-[*1..k]-`), unblocking the cross-alias `WHERE NOT(person=friend)` clause**: a bounded UNDIRECTED variable-length MATCH that materializes a bindings row table (`rows(binding_ops=...)`, emitted whenever a downstream clause — e.g. a cross-alias same-path `WHERE NOT a = b` / `a <> b` — references two node aliases on the path) previously declined on polars (`binding_rows_polars` hard-NIE'd every undirected multihop), while pandas handled it. This was the single residual blocking the official LDBC IC11/IC6 queries on polars (the cross-alias WHERE lowering itself was already native — directed var-length + `WHERE NOT a = b` already matched pandas). `binding_rows_polars` now supports undirected var-length with **`min_hops == 1`** via a doubled-pair join with immediate-backtrack avoidance, an exact port of the pandas oracle `_gfql_multihop_binding_rows` (`avoid_immediate_backtrack=True`): a `__prev__` marker (seeded null, dtype-matched to the id column) drops immediate backtracks each hop (Kleene mask: null prev kept), and the step-pair set reproduces pandas' edge multiplicity exactly — each non-loop edge contributes each directed orientation TWICE (so a length-1 pair appears x2, length-2 x4) while self-loops contribute `(u,u)` x2 only (not double-counted). The multiplicity rule was derived by instrumenting pandas' `step_pairs` (which flow from the var-length `edge_op.execute` hop + `orient_edges`), NOT by reading code alone. Scoped to `min_hops == 1` because that is where the raw-edge reconstruction provably matches pandas: every edge is trivially a length-1 path so the var-length hop's backward pruning removes nothing; `min_hops == 0` (zero-hop, undoubled) and `min_hops >= 2` (backward-pruned / long-walk divergence) still **decline with an honest `NotImplementedError`** rather than risk silent-wrong multiplicities. Differential fuzz vs the pandas oracle over ~2500 random graphs (self-loops, parallel + antiparallel edges, `*1..2`…`*1..5`, all WHERE/RETURN variants): 0 disagreements; the out-of-scope windows decline (never diverge). Tests in `test_engine_polars_binding_rows.py` (parity + multiplicity/backtrack/self-loop/string-id pins + `*0..2`/`*2..3`/`*2..2` decline pins).
+- **GFQL polars engine natively runs whole-entity `count(DISTINCT n)` / identity aggregation (#1709)**: `g.gfql("MATCH (a {..}) RETURN count(DISTINCT a) AS c")` and its grouped/filtered/traversal variants (`(a)-[]->(b) RETURN count(DISTINCT b)`) previously declined on polars — the Cypher aggregation lowers `count(DISTINCT b)` to the `__gfql_node_id__` identity sentinel, and the polars `lower_expr` `Identifier` branch resolved only the prefixed `alias.__gfql_node_id__` form, not the bare sentinel, so it NIE'd. Fix threads the graph node-id column through a `_NODE_ID` contextvar (published alongside the schema in `_lower_with_schema`, wired through the four callers) so the bare sentinel resolves to the id column **only if present**, else declines (never invents/mis-resolves a column). Differential fuzz vs the pandas oracle: 600 + 300 cases, 0 disagreements. Deliberately still honest-NIE (verified declines, not silent-wrong): whole-entity `collect(b)` (list-of-entities repr not ported, cf #1650) and multi-alias binding-table identity aggregation (bare id absent on connected-pattern binding tables — adjacent to #1273). Tests in `test_engine_polars_row_pipeline.py`.
 - **GFQL polars engine natively runs multi-source (node-cartesian) MATCH (#1273)**: comma-separated disconnected aliases — `g.gfql("MATCH (a {..}), (b {..}) RETURN a.id, b.id")` — previously declined on polars (the `rows(binding_ops)` `node_cartesian` branch hard-NIE'd) while pandas handled them. New `_cartesian_node_bindings_polars` mirrors the pandas oracle `_gfql_cartesian_node_bindings_row_table`: each alias is independently filtered, projected into the per-alias lookup schema (bare `alias` id, `alias.id`, `alias.<prop>`, and pandas' leaked `alias.alias=True` flag incl. property-shadowing), then left-major cross-joined for row-order parity. Scalar/aliased/`count(*)` projections over the cartesian match pandas exactly (verified end-to-end); whole-entity `RETURN a, b` stays an honest NIE (separate projection surface). Two parity-safe declines mirror shapes where pandas itself errors (proven on master, so both engines fail identically, never diverge): an anonymous node op, and ≥4 named aliases (pandas' bare-id merge residue collides). Differential fuzz vs pandas over ~10k lowered cases: 0 disagreements. Tests in `test_engine_polars_binding_rows.py`.
 - **GFQL polars chain — variable-length node aliases are hop-distance gated; the chain is now silent-wrong-free (#1741)**: a node named after a variable-length edge (`g.gfql("MATCH (a)-[*1..2]-(b) RETURN b")`) previously carried its alias regardless of hop distance, so an undirected walk that backtracked into the seed wrongly returned it (pandas correctly excludes it — trail semantics). The polars chain now auto-injects the #1741 hop-distance label (name resolved against the user's node columns via the `reserved_columns` registry, so a user column named like the internal one is never clobbered) and applies pandas' alias `[min_hop, max_hop]` window in `_apply_node_names`. The one shape the gate can't yet cover — a node alias after a **forward/reverse `min_hops>1`** edge, whose labels need pandas' layered backward walk (not ported) — now **declines with an honest `NotImplementedError` (#1748)** instead of returning nodes outside the window; the decline is precise (differential vs pandas: 30/30 named shapes NIE, 30/30 unnamed run and match, 0 over-decline). 4-engine A/B on the stacked build: pandas/cudf 144/144, polars/polars-gpu 112 agree / **0 disagree** / 32 honest-NIE — zero silent-wrong shapes remain on the polars chain. Adapts the mechanism from the retired #1742 (declined undirected varlen+alias, now natively gated). Tests: `TestVarlenAliasHopGate` (pandas-parity across directions, the `*2..3` decline + unnamed-still-runs pins, a column-collision test).
 - **GFQL native polars `label_node_hops` on the plain BFS — correct, direction-dependent hop labels (#1741 groundwork)**: the polars eager hop now emits pandas-parity node hop-distance labels instead of declining `label_node_hops`. The labeling rule is DIRECTION-DEPENDENT (the subtle part, derived by differential fuzz vs the pandas oracle, not by reading pandas alone): forward/reverse label EVERY destination of a hop first-wins (matching pandas `hop.py` `new_node_ids`), so a seed re-entered at hop 1 IS labeled; undirected labels destinations MINUS everything already visited and pre-seeds the seen-set with the seeds, so a seed re-reached by a backtracking walk stays NULL (its shortest-path distance). Two correctness fixes over the first cut: (1) the undirected seed pre-seed must NOT depend on `return_as_wave_front` — a seed filtered out of the frontier by `source_node_match` or suppressed in wave-front mode was being re-labeled (A/B vs pandas over direction × hops × `return_as_wave_front` × `source_node_match` × `destination_node_match` × 10 graphs: was 456/24, now 480/480); (2) a requested `label_node_hops` name that collides with an existing column is redirected to `<name>_1` like pandas' `resolve_label_col`, not the polars left-join auto-suffix `<name>_right`/`DuplicateError`. `label_edge_hops` and `min_hops>1` labels stay honest NIEs. Amplified `test_engine_polars_hop.py` (`TestHopLabelsDifferential` over the seed-filter / wave-front / multi-seed axes, an explicit direction-asymmetry contrast pin, and a collision test).

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

@@ -0,0 +1,21 @@
+"""Per-engine lowering context for the native polars GFQL engine.
+
+Contextvars threaded around a polars row-op lowering pass so the pure expr-lowering helpers
+(``lower_expr`` and friends) can read ambient state without plumbing it through every signature.
+Set + reset them as a pair around a lowering call — see ``_lower_with_schema`` in ``row_pipeline.py``.
+
+Declared together in one place per engine (the contextvar analogue of ``reserved_columns.py``) so
+the lowering's ambient state is discoverable rather than inlined across the executor. New
+polars-lowering contextvars SHOULD be added here.
+"""
+import contextvars
+from typing import Optional
+
+#: Table schema (column name -> polars dtype) of the frame being lowered — lets float-operand
+#: inference run for the NaN guard without a scan.
+SCHEMA: "contextvars.ContextVar[dict]" = contextvars.ContextVar("gfql_polars_schema", default={})
+
+#: Active graph node-id column name — lets ``lower_expr`` resolve the bare whole-entity identity
+#: sentinel (``same_path_types.NODE_IDENTITY_COLUMN``) to the real id column. None when unknown, so
+#: the identity sentinel then declines to an honest NIE rather than resolving to a wrong column.
+NODE_ID: "contextvars.ContextVar[Optional[str]]" = contextvars.ContextVar("gfql_polars_node_id", default=None)

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

@@ -13,10 +13,9 @@
 """
 from __future__ import annotations
 
-import contextvars
 import operator
 import re
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Union, cast
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, TypeVar, Union, cast
 
 if TYPE_CHECKING:
     import polars as pl
@@ -33,8 +32,11 @@
 
 
 # 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={})
+# operands for the NaN guard. Lowering contextvars live in the per-engine `lowering_context`
+# registry (aliased here to keep call sites terse); the whole-entity identity sentinel is the
+# shared cypher-lowering constant, NOT a local literal.
+from .lowering_context import SCHEMA as _SCHEMA, NODE_ID as _NODE_ID
+from graphistry.compute.gfql.same_path_types import NODE_IDENTITY_COLUMN as _NODE_ID_TOKEN
 
 # Ops needing the NaN guard: polars treats NaN as the LARGEST value (>/>=/== TRUE), but
 # IEEE/Python/pandas/Cypher compare NaN as FALSE (!= TRUE; Neo4j TCK agrees). Float operands
@@ -88,7 +90,7 @@ 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:
+    if prop == _NODE_ID_TOKEN 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.
@@ -473,7 +475,16 @@ def lower_expr(node: ExprNode, columns: Sequence[str]) -> Optional[pl.Expr]:
     if isinstance(node, ListLiteral):
         return _lower_list_literal(node.items, columns)
     if isinstance(node, Identifier):
-        return pl.col(node.name) if node.name in columns else None
+        if node.name in columns:
+            return pl.col(node.name)
+        # Bare whole-entity identity sentinel -> the graph node-id column (pandas
+        # _gfql_resolve_token bare form). Only when the id column is actually present;
+        # otherwise decline (None -> NIE) rather than invent a column.
+        if node.name == _NODE_ID_TOKEN:
+            node_id = _NODE_ID.get()
+            if node_id is not None and node_id in columns:
+                return pl.col(node_id)
+        return None
     if isinstance(node, PropertyAccessExpr):
         if isinstance(node.value, Identifier):
             src = _resolve_property(node.value.name, node.property, columns)
@@ -625,14 +636,21 @@ def _rewrap(g: Plottable, table_df: Any) -> Plottable:
     return frame_ops.row_table(_RowPipelineAdapter(g), table_df)
 
 
-def _lower_with_schema(table: Any, fn):
+_LowerT = TypeVar("_LowerT")
+
+
+def _lower_with_schema(table: "pl.DataFrame", fn: Callable[[], _LowerT],
+                       node_id: Optional[str] = None) -> _LowerT:
     """Run a lowering callable with the table schema published to ``_SCHEMA`` (float-operand
-    inference for the NaN guard)."""
-    token = _SCHEMA.set(dict(table.schema))
+    inference for the NaN guard) and the graph node-id column published to ``_NODE_ID`` (bare
+    ``__gfql_node_id__`` identity-sentinel resolution)."""
+    schema_token = _SCHEMA.set(dict(table.schema))
+    node_id_token = _NODE_ID.set(node_id)
     try:
         return fn()
     finally:
-        _SCHEMA.reset(token)
+        _SCHEMA.reset(schema_token)
+        _NODE_ID.reset(node_id_token)
 
 
 def _project_preserving_height(table: Any, exprs: List[Any]) -> Any:
@@ -652,7 +670,7 @@ def _project_polars(g: Plottable, items: Sequence[SelectItem], extend: bool) ->
     """Shared body of ``select_polars`` / ``with_columns_polars``; None if any item isn't
     lowerable (honest NIE, no pandas bridge)."""
     table = _active_table(g)
-    exprs = _lower_with_schema(table, lambda: lower_select_items(items, list(table.columns)))
+    exprs = _lower_with_schema(table, lambda: lower_select_items(items, list(table.columns)), node_id=g._node)
     if exprs is None:
         return None
     out = table.with_columns(exprs) if extend else _project_preserving_height(table, exprs)
@@ -714,7 +732,7 @@ def where_rows_polars(
     if expr is not None:
         if not isinstance(expr, str):
             return None
-        lowered = _lower_with_schema(table, lambda: lower_expr_str(expr, columns))
+        lowered = _lower_with_schema(table, lambda: lower_expr_str(expr, columns), node_id=g._node)
         if lowered is None:
             return None
         preds.append(lowered)
@@ -729,17 +747,16 @@ def where_rows_polars(
 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)))
+    lowered = _lower_with_schema(table, lambda: lower_order_by_keys(keys, list(table.columns)), node_id=g._node)
     if lowered is None:
         return None
     exprs, descending = lowered
     # openCypher orders NULL as the LARGEST value: ASC -> nulls last, DESC -> nulls FIRST.
     # (Previously hardcoded nulls_last=True, which mis-ordered DESC keys and silently returned
-    # the wrong `... DESC LIMIT k` top-k over a column containing NULLs.) Normalize `descending`
-    # to a per-key list (it may arrive as a scalar bool) so `nulls_last` mirrors it per key.
-    descending_list = descending if isinstance(descending, list) else [descending] * len(exprs)
-    nulls_last = [not d for d in descending_list]
-    return _rewrap(g, table.sort(exprs, descending=descending_list, nulls_last=nulls_last))
+    # the wrong `... DESC LIMIT k` top-k over a column containing NULLs.) `descending` is one
+    # bool per key (see lower_order_by_keys), so `nulls_last` mirrors it per key.
+    nulls_last = [not d for d in descending]
+    return _rewrap(g, table.sort(exprs, descending=descending, nulls_last=nulls_last))
 
 
 # Native aggs: count/sum/avg/min/max/count_distinct/collect/collect_distinct; stdev/percentile
@@ -887,7 +904,7 @@ def select_extend_polars(g: Plottable, items: Sequence[SelectItem]) -> Optional[
     the bindings-path aggregate lowering (pre-aggregation group keys / agg args),
     so it is required for binding-row queries (#1709). None → NIE."""
     table = _active_table(g)
-    exprs = _lower_with_schema(table, lambda: lower_select_items(items, list(table.columns)))
+    exprs = _lower_with_schema(table, lambda: lower_select_items(items, list(table.columns)), node_id=g._node)
     if exprs is None:
         return None
     out = table.with_columns(exprs)

graphistry/tests/compute/gfql/coverage_baselines/ci-polars-py3.12.json

@@ -9,6 +9,7 @@
     "graphistry/compute/gfql/lazy/engine/polars/dtypes.py": 92.0,
     "graphistry/compute/gfql/lazy/engine/polars/hop.py": 87.0,
     "graphistry/compute/gfql/lazy/engine/polars/hop_eager.py": 90.0,
+    "graphistry/compute/gfql/lazy/engine/polars/lowering_context.py": 90.0,
     "graphistry/compute/gfql/lazy/engine/polars/nan_clean.py": 90.0,
     "graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py": 67.0,
     "graphistry/compute/gfql/lazy/engine/polars/predicates.py": 83.0,