From fefed1067d22f2434823bbea9d582fbb5593e0d5 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 20 Jul 2026 10:37:05 -0700 Subject: [PATCH 1/5] feat(gfql): native polars whole-entity count(DISTINCT n) via __gfql_node_id__ (#1709) Resolve the bare `__gfql_node_id__` identity sentinel (emitted by whole-entity aggregation lowering, e.g. RETURN count(DISTINCT b)) to the graph node-id column in the polars row-expression lowering, mirroring pandas _gfql_resolve_token. The prefixed `alias.__gfql_node_id__` form was already handled by _resolve_property; only the single-source bare form declined (NIE on 'with_'). Published via a _NODE_ID contextvar alongside _SCHEMA; declines (NIE) when the id column is unknown or absent so a multi-alias binding table never resolves to a wrong/absent column. Differential fuzz vs pandas oracle: 600/600 value-identical. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- .../gfql/lazy/engine/polars/row_pipeline.py | 42 ++++-- .../gfql/test_engine_polars_row_pipeline.py | 132 ++++++++++++++++++ 2 files changed, 165 insertions(+), 9 deletions(-) diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index 10e1477d9c..d7782dc944 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -36,6 +36,18 @@ # 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={}) +# Whole-entity identity sentinel token (mirrors same_path_types.NODE_IDENTITY_COLUMN). The +# aggregation lowering emits a BARE ``__gfql_node_id__`` as the agg arg for whole-entity +# reductions (e.g. ``RETURN count(DISTINCT b)``); it must resolve to the graph's node-id +# column, exactly like pandas ``_gfql_resolve_token``. (The ``alias.__gfql_node_id__`` prefixed +# form is already handled by ``_resolve_property``.) +_NODE_ID_TOKEN = "__gfql_node_id__" + +# Active graph node-id column name, published around lowering so ``lower_expr`` can resolve the +# bare identity sentinel above to the real id column. None when unknown (identity sentinel then +# declines -> honest NIE, never a wrong column). +_NODE_ID: "contextvars.ContextVar[Optional[str]]" = contextvars.ContextVar("gfql_polars_node_id", default=None) + # 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 # get masked to the IEEE answer; ``is_nan()`` is float-only, hence the dtype inference. @@ -473,7 +485,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 +646,17 @@ def _rewrap(g: Plottable, table_df: Any) -> Plottable: return frame_ops.row_table(_RowPipelineAdapter(g), table_df) -def _lower_with_schema(table: Any, fn): +def _lower_with_schema(table: Any, fn, node_id: Optional[str] = None): """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 +676,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 +738,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,7 +753,7 @@ 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 @@ -887,7 +911,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) 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 ee4ea10a32..6163773e05 100644 --- a/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py +++ b/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py @@ -108,6 +108,15 @@ def _assert_parity(query, *, order_sensitive=True): "MATCH (n) UNWIND [1, 2, 3] AS x RETURN x", # multi-entity property projection via native rows(binding_ops) (#1709) "MATCH (n)-[e]->(m) RETURN n.val, m.val", + # whole-entity identity aggregation (#1709): count(DISTINCT n) lowers the agg arg to the + # bare ``__gfql_node_id__`` identity sentinel, now resolved to the node-id column natively. + "MATCH (n) RETURN count(DISTINCT n) AS c", + "MATCH (n) WHERE n.val > 25 RETURN count(DISTINCT n) AS c", + "MATCH (n) RETURN n.kind, count(DISTINCT n) AS c", + "MATCH (n) RETURN n.kind, count(DISTINCT n) AS c ORDER BY c DESC", + "MATCH (a)-[e]->(b) RETURN count(DISTINCT b) AS c", + # property-arg distinct (already native; guard against regression) + "MATCH (n) RETURN n.kind, count(DISTINCT n.val) AS c", ] # NO-CHEATING (see plan.md): no native impl yet -> NotImplementedError, never a @@ -117,6 +126,13 @@ def _assert_parity(query, *, order_sensitive=True): DEFERRED = [ "MATCH (n)-[e]->(m) WHERE n.val < m.val RETURN n, m", # cross-entity WHERE "MATCH (n)-[e]->(m) RETURN n, m", # whole-row multi-entity render + # whole-entity collect: agg arg is the __node_entity__(n) whole-entity token (not the bare + # identity sentinel), whose native list-of-entities representation isn't ported yet -> NIE. + "MATCH (n) RETURN collect(n) AS xs", + # multi-alias binding-table identity aggregation: the bare ``__gfql_node_id__`` sentinel has + # no bare node-id column on the connected-pattern binding table (identity lives in the bare + # ALIAS column), so it declines honestly rather than resolve to a wrong/absent column. + "MATCH (a)-[e]->(b) RETURN b.kind, count(DISTINCT b) AS c, count(b) AS t", ] @@ -261,6 +277,30 @@ def test_row_expr_lowering_unit(): assert lower_order_by_keys(["bad-shape"], cols) is None +def test_node_identity_sentinel_resolution(): + """Bare ``__gfql_node_id__`` (whole-entity identity, #1709) resolves ONLY when the graph + node-id column is published AND present in the frame; else declines (None -> NIE), never a + wrong column. Mirrors pandas ``_gfql_resolve_token`` bare form.""" + from graphistry.compute.gfql.lazy.engine.polars import row_pipeline as rp + cols = ["id", "val", "kind"] + # no node-id published -> decline + assert rp.lower_expr_str("__gfql_node_id__", cols) is None + # published + present -> resolves to that column + tok = rp._NODE_ID.set("id") + try: + expr = rp.lower_expr_str("__gfql_node_id__", cols) + assert expr is not None + assert expr.meta.root_names() == ["id"] + finally: + rp._NODE_ID.reset(tok) + # published but absent from frame -> decline (never invent a column) + tok = rp._NODE_ID.set("node_id_missing") + try: + assert rp.lower_expr_str("__gfql_node_id__", cols) is None + finally: + rp._NODE_ID.reset(tok) + + def test_polars_frame_op_limit_matches_slice(): """limit/skip operate on a polars active table without index artifacts.""" g = BASE.gfql("MATCH (n) RETURN n LIMIT 4", engine="polars") @@ -313,6 +353,98 @@ def test_polars_empty_result_shape(): assert list(g._nodes.columns) == list(g_pd._nodes.columns) +@pytest.mark.parametrize("query,expected", [ + ("MATCH (n) RETURN count(DISTINCT n) AS c", 6), # 6 distinct ids + ("MATCH (n) WHERE n.val > 25 RETURN count(DISTINCT n) AS c", 4), # ids with val 30/40/50/60 +]) +def test_polars_count_distinct_entity_absolute(query, expected): + """Explicit pin: whole-entity count(DISTINCT n) returns the TRUE distinct-identity count + on the native polars engine (pandas oracle can't mask a wrong constant).""" + rpl = BASE.gfql(query, engine="polars")._nodes + assert "polars" in type(rpl).__module__ + assert rpl.height == 1 + assert rpl["c"].to_list() == [expected] + + +def test_polars_count_distinct_entity_grouped_values(): + """Grouped whole-entity count(DISTINCT n) matches pandas value-for-value and equals the + per-group node count (ids unique). BASE kinds: a=3, b=2, c=1.""" + q = "MATCH (n) RETURN n.kind, count(DISTINCT n) AS c" + rpd = _to_pandas(BASE.gfql(q, engine="pandas")._nodes) + rpl = _to_pandas(BASE.gfql(q, engine="polars")._nodes) + a = rpd.sort_values("n.kind").reset_index(drop=True) + b = rpl.sort_values("n.kind").reset_index(drop=True) + pd.testing.assert_frame_equal(a, b, check_dtype=False) + assert dict(zip(b["n.kind"], b["c"])) == {"a": 3, "b": 2, "c": 1} + + +@pytest.mark.parametrize("seed", list(range(60))) +def test_polars_identity_aggregation_fuzz_matches_pandas(seed): + """Bounded differential fuzz (#1709): random small graphs x whole-entity identity + aggregation shapes; native polars MUST equal the pandas oracle value-for-value, or both + decline. A silent divergence is the worst outcome, so any polars-ok/pandas-nie or value + mismatch fails. Larger sweep lives in scratch (600/600 clean); this pins a deterministic + subset into the coverage lane.""" + import random + from .polars_test_utils import graph_sig + import numpy as np + + def _norm(sig): + def cell(v): + if isinstance(v, np.ndarray): + return tuple(v.tolist()) + if isinstance(v, list): + return tuple(v) + return v + + def frame(f): + if f is None: + return None + c, rows = f + return (c, tuple(tuple(cell(v) for v in r) for r in rows)) + return tuple(frame(f) for f in sig) + + rng = random.Random(seed) + n = rng.randint(1, 7) + ids = [f"n{i}" for i in range(n)] + nodes = pd.DataFrame({ + "id": pd.Series(ids, dtype="object"), + "kind": pd.Series([rng.choice(["A", "B", "C"]) for _ in ids], dtype="object"), + "v": pd.Series([rng.choice([1, 2, 3, None]) for _ in ids], dtype="float64"), + "grp": pd.Series([rng.choice(["x", "y", None]) for _ in ids], dtype="object"), + }) + ne = rng.randint(0, 8) + s = [rng.choice(ids) for _ in range(ne)] + d = [rng.choice(ids) for _ in range(ne)] + g = graphistry.nodes(nodes, "id").edges( + pd.DataFrame({"s": pd.Series(s, dtype="object"), "d": pd.Series(d, dtype="object")}), + "s", "d", + ) + queries = [ + "MATCH (b) RETURN count(DISTINCT b) AS c", + "MATCH (b {kind:'A'}) RETURN count(DISTINCT b) AS c", + "MATCH (b) RETURN b.kind, count(DISTINCT b) AS c", + "MATCH (b) RETURN b.grp, count(DISTINCT b) AS c, count(b) AS t", + "MATCH (a)-[]->(b) RETURN count(DISTINCT b) AS c", + "MATCH (b) RETURN b.kind, count(DISTINCT b.v) AS c", + ] + for q in queries: + try: + base = ("ok", _norm(graph_sig(g.gfql(q, engine="pandas")))) + except NotImplementedError: + base = ("nie",) + try: + got = ("ok", _norm(graph_sig(g.gfql(q, engine="polars")))) + except NotImplementedError: + got = ("nie",) + # polars may decline where pandas succeeds (honest NIE); never the reverse, never a + # value mismatch. + if base[0] == "ok" and got[0] == "ok": + assert base[1] == got[1], f"value divergence {q!r} seed={seed}: {base} vs {got}" + elif got[0] == "ok": + raise AssertionError(f"polars ok but pandas nie for {q!r} seed={seed}") + + # Direct frame-op coverage: each native polars branch on a real polars-framed graph, independent # of which cypher shapes compile to which ops — pins the engine-polymorphic frame_ops layer. def _polars_graph(): From 0d8584247d9e4d0042883f36985e0ae046437e58 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 20 Jul 2026 11:44:37 -0700 Subject: [PATCH 2/5] docs(gfql): CHANGELOG [Development] entry for #1709 (native count(DISTINCT n)) Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 94bf276b82..36e599a827 100644 --- a/CHANGELOG.md +++ b/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.`, 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 `_1` like pandas' `resolve_label_col`, not the polars left-join auto-suffix `_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). From 5ebeeaaa5d2e89fdf21635e6b1c879a9c8b6916f Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 20 Jul 2026 14:39:52 -0700 Subject: [PATCH 3/5] refactor(gfql): dedupe node-id sentinel + move polars lowering contextvars to a registry (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback on #1750: _NODE_ID_TOKEN was a bare literal duplicating the canonical same_path_types.NODE_IDENTITY_COLUMN — now imports it. The polars-lowering contextvars (_SCHEMA/_NODE_ID) move to a new per-engine registry lowering_context.py (the contextvar analogue of reserved_columns.py), answering "per-engine or overall?" -> per-engine, since they thread polars-specific lowering state. Registered lowering_context.py in the polars per-file coverage baseline; dropped the now-unused `import contextvars`. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- .../lazy/engine/polars/lowering_context.py | 21 ++++++++++++++++++ .../gfql/lazy/engine/polars/row_pipeline.py | 22 +++++-------------- .../coverage_baselines/ci-polars-py3.12.json | 1 + 3 files changed, 28 insertions(+), 16 deletions(-) create mode 100644 graphistry/compute/gfql/lazy/engine/polars/lowering_context.py diff --git a/graphistry/compute/gfql/lazy/engine/polars/lowering_context.py b/graphistry/compute/gfql/lazy/engine/polars/lowering_context.py new file mode 100644 index 0000000000..660af2918f --- /dev/null +++ b/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) diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index d7782dc944..131c1ecec0 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -13,7 +13,6 @@ """ 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 @@ -33,20 +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={}) - -# Whole-entity identity sentinel token (mirrors same_path_types.NODE_IDENTITY_COLUMN). The -# aggregation lowering emits a BARE ``__gfql_node_id__`` as the agg arg for whole-entity -# reductions (e.g. ``RETURN count(DISTINCT b)``); it must resolve to the graph's node-id -# column, exactly like pandas ``_gfql_resolve_token``. (The ``alias.__gfql_node_id__`` prefixed -# form is already handled by ``_resolve_property``.) -_NODE_ID_TOKEN = "__gfql_node_id__" - -# Active graph node-id column name, published around lowering so ``lower_expr`` can resolve the -# bare identity sentinel above to the real id column. None when unknown (identity sentinel then -# declines -> honest NIE, never a wrong column). -_NODE_ID: "contextvars.ContextVar[Optional[str]]" = contextvars.ContextVar("gfql_polars_node_id", default=None) +# 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 @@ -100,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. diff --git a/graphistry/tests/compute/gfql/coverage_baselines/ci-polars-py3.12.json b/graphistry/tests/compute/gfql/coverage_baselines/ci-polars-py3.12.json index 08ee6d8c7e..652835ad4f 100644 --- a/graphistry/tests/compute/gfql/coverage_baselines/ci-polars-py3.12.json +++ b/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, From 4968f45a3c39c1feb9358b15acf857c0dadff61c Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 20 Jul 2026 16:36:04 -0700 Subject: [PATCH 4/5] refactor(gfql): type _lower_with_schema (table: pl.DataFrame, fn generic) (review) Review feedback on #1750: _lower_with_schema(table: Any, fn) -> untyped. Now table: pl.DataFrame, fn: Callable[[], _LowerT] -> _LowerT (the lowering result flows through). mypy: 0 new errors; ruff clean; 175 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- .../compute/gfql/lazy/engine/polars/row_pipeline.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index 131c1ecec0..8eef63e118 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -15,7 +15,7 @@ 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 @@ -636,7 +636,11 @@ def _rewrap(g: Plottable, table_df: Any) -> Plottable: return frame_ops.row_table(_RowPipelineAdapter(g), table_df) -def _lower_with_schema(table: Any, fn, node_id: Optional[str] = None): +_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) and the graph node-id column published to ``_NODE_ID`` (bare ``__gfql_node_id__`` identity-sentinel resolution).""" From bdd0b7abad1106519edf637b10c09bd44db9d1c8 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 20 Jul 2026 17:07:37 -0700 Subject: [PATCH 5/5] fix(gfql): drop dead scalar-normalization branch in order_by_polars lower_order_by_keys always returns List[bool] for descending flags, so the isinstance(descending, list) ternary was statically dead and tripped mypy [list-item] under py3.8 (List[bool] where bool expected). Use descending directly for per-key nulls_last; behavior unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- .../compute/gfql/lazy/engine/polars/row_pipeline.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index 8eef63e118..0a59a5c91e 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -753,11 +753,10 @@ def order_by_polars(g: Plottable, keys: Sequence[OrderKey]) -> Optional[Plottabl 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