From 5b8756846a7193b03e592512305bb49b564123a1 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 6 Jul 2026 18:51:47 -0700 Subject: [PATCH 01/24] =?UTF-8?q?fix(gfql):=20real-GFQL=20graph-bench=20co?= =?UTF-8?q?unt=20queries=20=E2=80=94=20polars=20count(*)=20broadcast=20(#1?= =?UTF-8?q?707)=20+=20count(non-active=20node=20alias)=20routing=20(#1708)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two correctness fixes that let graph-benchmark q1-q9 run as real GFQL Cypher (no dataframe shortcuts) on pandas/cuDF, and remove a polars silent-wrong answer. `RETURN count(*)` (and any all-scalar-literal projection) returned constant 1 instead of the true count. Cypher WITH/RETURN preserves row cardinality, but polars DataFrame.select of an all-scalar projection collapses to 1 row, so the synthetic __cypher_group__=1 for keyless count(*) made the downstream group_by count see 1 row. select_polars now broadcasts an all-scalar projection to the frame height via with_columns(...).select(names). +5 parity + 2 value-regression tests. `MATCH (a)-[e]->(b) RETURN b.id, count(a)` (graph-bench q1 in-degree) failed with "one MATCH source alias at a time" — the aggregate referenced the other endpoint alias, routing to the single-alias table instead of the bindings-row table. _lower_general_row_projection now forces the bindings source for count() (non-distinct) over a non-active pattern node alias — sound on the bindings table (count of matched paths per group = in-degree). Narrow: property/compound cross-source aggregates keep the conservative fail-fast. +3 tests (incl. cuDF). Regression sweep (cypher/ + polars engine + row-pipeline): 2177 passed, 11 skipped, 15 xfailed. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_012rRBmezaBEi2CJgDGkMNgR --- graphistry/compute/gfql/cypher/lowering.py | 19 ++++++++ .../gfql/lazy/engine/polars/row_pipeline.py | 15 +++++- .../compute/gfql/cypher/test_lowering.py | 47 +++++++++++++++++++ .../gfql/test_engine_polars_row_pipeline.py | 30 ++++++++++++ 4 files changed, 110 insertions(+), 1 deletion(-) diff --git a/graphistry/compute/gfql/cypher/lowering.py b/graphistry/compute/gfql/cypher/lowering.py index 9d03a88e8e..0539d56cde 100644 --- a/graphistry/compute/gfql/cypher/lowering.py +++ b/graphistry/compute/gfql/cypher/lowering.py @@ -6551,6 +6551,25 @@ def _lower_general_row_projection( break continue if len(refs) > 1 or (len(refs) == 1 and base_active_alias not in refs): + # `count()` over a pattern node alias other than + # the projection's active alias (e.g. `RETURN b.id, count(a)` — + # graph-benchmark q1 "top-k by in-degree"). Counting a bare node + # alias is exactly "how many matched paths bind this node per + # group" — sound on the bindings-row table regardless of + # directedness. Force the bindings path rather than fail fast with + # the misleading "one MATCH" error. Kept narrow: only a bare-alias + # `count(...)`; property/compound aggregates (`count(x.p)`, + # `x.p + count(...)`) whose cross-source soundness isn't + # established keep the conservative fail-fast (#1708). + agg_arg = (agg_spec.expr_text or "").strip() + if ( + agg_spec.func == "count" + and not agg_spec.distinct + and agg_arg in alias_targets + and isinstance(alias_targets.get(agg_arg), ASTNode) + and refs <= set(alias_targets.keys()) + ): + continue can_force_bindings = False break if can_force_bindings: diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index 34e6dfa72c..f984acea42 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -607,6 +607,19 @@ def _lower_with_schema(table: Any, fn): _SCHEMA.reset(token) +def _project_preserving_height(table: Any, exprs: List[Any]) -> Any: + """Project ``exprs`` while preserving the frame's row cardinality. + + Cypher ``WITH``/``RETURN`` projection is a map, not a reduce. Polars + ``DataFrame.select`` collapses to one row when every projected expression is + scalar, so broadcast all-scalar projections through ``with_columns`` first. + """ + if exprs and all(len(e.meta.root_names()) == 0 for e in exprs): + names = [e.meta.output_name() for e in exprs] + return table.with_columns(exprs).select(names) + return table.select(exprs) + + def _project_polars(g: Plottable, items: Sequence[Any], extend: bool) -> Optional[Plottable]: """Shared body of ``select_polars`` / ``with_columns_polars``; None if any item isn't lowerable (honest NIE, no pandas bridge).""" @@ -614,7 +627,7 @@ def _project_polars(g: Plottable, items: Sequence[Any], extend: bool) -> Optiona exprs = _lower_with_schema(table, lambda: lower_select_items(items, list(table.columns))) if exprs is None: return None - out = table.with_columns(exprs) if extend else table.select(exprs) + out = table.with_columns(exprs) if extend else _project_preserving_height(table, exprs) if _select_emits_temporal_constructor_text(out): # decline (NIE): projected String column holds temporal-constructor text (date({...}) # etc.) that pandas normalizes to ISO, not yet native — don't leak the raw text. diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index 1517c5ff53..0e6348dd2a 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -1775,6 +1775,53 @@ def test_string_cypher_supports_cartesian_node_only_grouped_count() -> None: ] +def test_string_cypher_count_non_active_node_alias_in_degree() -> None: + """#1708: `MATCH (a)-[e]->(b) RETURN b.id, count(a)` (graph-benchmark q1 + "top-k by in-degree") counts a bare NODE alias other than the projection's + active alias. It must route to the bindings-row source and return the true + in-degree per `b`, not fail with the misleading "one MATCH" error.""" + nodes = pd.DataFrame({"id": [0, 1, 2, 3, 4, 7, 8, 9]}) + # in-degree: node 9 <- {0,1,2,4}=4 ; node 8 <- {3,0}=2 ; node 7 <- {1}=1 + edges = pd.DataFrame({"s": [0, 1, 2, 3, 0, 1, 4], "d": [9, 9, 9, 8, 8, 7, 9]}) + graph = _mk_graph(nodes, edges) + result = graph.gfql( + "MATCH (a)-[e]->(b) RETURN b.id AS id, count(a) AS c ORDER BY c DESC LIMIT 3" + ) + assert result._nodes.to_dict(orient="records") == [ + {"id": 9, "c": 4}, + {"id": 8, "c": 2}, + {"id": 7, "c": 1}, + ] + + +def test_string_cypher_count_non_active_node_alias_matches_count_star() -> None: + """#1708 corollary: `count()` equals `count(*)` and + `count()` for the same q1 shape (each row binds exactly one of + each endpoint), so all three routes agree.""" + nodes = pd.DataFrame({"id": [0, 1, 2, 3, 8, 9]}) + edges = pd.DataFrame({"s": [0, 1, 2, 3], "d": [9, 9, 9, 8]}) + graph = _mk_graph(nodes, edges) + tmpl = "MATCH (a)-[e]->(b) RETURN b.id AS id, count({arg}) AS c ORDER BY c DESC LIMIT 3" + expected = [{"id": 9, "c": 3}, {"id": 8, "c": 1}] + for arg in ("a", "b", "*"): + got = graph.gfql(tmpl.format(arg=arg))._nodes.to_dict(orient="records") + assert got == expected, f"count({arg}) disagreed: {got}" + + +def test_string_cypher_count_non_active_node_alias_cudf() -> None: + """#1708: bindings-route count of a non-active node alias also runs on cudf.""" + _require_cudf_runtime() + nodes = pd.DataFrame({"id": [0, 1, 2, 3, 8, 9]}) + edges = pd.DataFrame({"s": [0, 1, 2, 3], "d": [9, 9, 9, 8]}) + result = _mk_cudf_graph(nodes, edges).gfql( + "MATCH (a)-[e]->(b) RETURN b.id AS id, count(a) AS c ORDER BY c DESC LIMIT 3" + ) + assert result._nodes.to_pandas().to_dict(orient="records") == [ + {"id": 9, "c": 3}, + {"id": 8, "c": 1}, + ] + + def test_string_cypher_supports_cartesian_node_only_non_simple_scalar_expression() -> None: result = _mk_cartesian_node_graph().gfql( "MATCH (n), (m) " 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 3261c4b0d5..99f1f5fcfe 100644 --- a/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py +++ b/graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py @@ -95,6 +95,12 @@ def _assert_parity(query, *, order_sensitive=True): "MATCH (n) WHERE NOT n.kind = 'beta' RETURN n.kind", "MATCH (n) RETURN n.kind, count(n) AS c", "MATCH (n) RETURN count(n) AS c", + # count(*) / keyless + literal projection (#1707: must NOT collapse to 1) + "MATCH (n) RETURN count(*) AS c", + "MATCH (n) WHERE n.val > 25 RETURN count(*) AS c", + "MATCH (n) RETURN n.kind, count(*) AS c", + "MATCH (n) RETURN n.kind, count(*) AS c ORDER BY c DESC", + "MATCH (n) RETURN 1 AS one", "MATCH (n) RETURN n.kind, sum(n.val) AS s, avg(n.val) AS a", "MATCH (n) RETURN n.kind, min(n.val) AS mn, max(n.val) AS mx", "MATCH (n) RETURN n.kind, count(n) AS c ORDER BY c DESC", @@ -176,6 +182,30 @@ def test_polars_distinct_preserves_first_order(): ) +@pytest.mark.parametrize("query,expected", [ + ("MATCH (n) RETURN count(*) AS c", 6), # keyless count over all rows + ("MATCH (n) WHERE n.val > 25 RETURN count(*) AS c", 4), # filtered count +]) +def test_polars_count_star_broadcasts_not_collapses(query, expected): + """#1707 regression: keyless ``count(*)`` lowers to a synthetic constant group + column that must broadcast to the full row count. A collapse to 1 row made the + downstream count return a constant ``1`` (silent wrong answer). Assert the + absolute value (pandas oracle can't mask it) on the native polars engine.""" + rpl = BASE.gfql(query, engine="polars")._nodes + assert "polars" in type(rpl).__module__ + assert rpl.height == 1 # a keyless aggregate is a single row + assert rpl["c"].to_list() == [expected] # ...holding the TRUE count, not 1 + + +def test_polars_literal_projection_preserves_cardinality(): + """#1707 corollary: a bare-literal projection (``RETURN 1``) is a map, not a + reduce — it must keep every row, not collapse to one.""" + rpl = BASE.gfql("MATCH (n) RETURN 1 AS one", engine="polars")._nodes + assert "polars" in type(rpl).__module__ + assert rpl.height == 6 # one row per matched node + assert rpl["one"].to_list() == [1] * 6 + + def test_polars_empty_result_shape(): """A LIMIT 0 / over-skip empties to 0 rows but keeps the projected schema. Whole-entity RETURN n flattens to a.* columns (#1650), matching pandas.""" From 032b01dde06fcecb73a28dbf42de15ab0aaf2299 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 6 Jul 2026 19:34:37 -0700 Subject: [PATCH 02/24] perf(gfql): chain edge-index attach without O(E) deep copy (#1670, #887) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every chain()/Cypher call on a graph without an edge-id binding attached a synthetic per-edge index via reset_index + rename, which deep-copied AND block-consolidated the entire edge frame (~70ms @2M edges) — even for node-only queries, and repeatedly per boundary-call sub-chain. Now: shallow copy + assign the frame index as the id column — identical id values (any index type), no O(E) data copy. The column is internal-only and dropped on every exit path, so only uniqueness semantics matter (unchanged). Measured @500k nodes / 2M edges (pandas): 1-row point RETURN 95.7 -> 13.5 ms (7x) full-scan RETURN a 134 -> 55 ms (2.4x) seeded 1-hop 192 -> ~170 ms polars unchanged (own chain path) Scaling (1-row RETURN): 15.6@50k/100@500k -> 6.8@50k/13.5@500k/42@1M. Residual O(N/E) tail (backward-pass hydration merges, endpoints reconciliation) remains tracked in #1670/#887; a hydration-skip variant was evaluated and rejected (row-order semantics risk for ~1.5ms). Suites: compute 4301 passed + gfql 3014 passed (9 pre-existing umap/dask env failures, verified identical pre-fix via stash). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_012rRBmezaBEi2CJgDGkMNgR --- graphistry/compute/chain.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 96044f5c64..13b1e5ead2 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -953,10 +953,16 @@ def _chain_impl( GFQL_EDGE_INDEX = generate_safe_column_name('edge_index', g._edges, prefix='__gfql_', suffix='__') added_edge_index = True - indexed_edges_df = g._edges.reset_index(drop=False) - original_cols = set(g._edges.columns) - index_col_name = next(col for col in indexed_edges_df.columns if col not in original_cols) - indexed_edges_df = indexed_edges_df.rename(columns={index_col_name: GFQL_EDGE_INDEX}) + # Attach the synthetic per-edge id WITHOUT copying edge data (#1670): + # the previous reset_index(drop=False) + rename deep-copied AND + # block-consolidated the whole edge frame (~70ms @2M edges) on every + # chain call — even node-only queries. A shallow copy + assigning the + # index as a column yields the identical id values (the frame's index) + # with no O(E) data copy. The column is internal-only — dropped on + # every exit path (see added_edge_index consumers below) — so only + # uniqueness matters. + indexed_edges_df = g._edges.copy(deep=False) + indexed_edges_df[GFQL_EDGE_INDEX] = indexed_edges_df.index g = g.edges(indexed_edges_df, edge=GFQL_EDGE_INDEX) else: added_edge_index = False From 6145c2f5cb58fb837d50e63ffedbc33c0b11bc15 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 6 Jul 2026 19:45:11 -0700 Subject: [PATCH 03/24] perf(gfql): hop edge-index attach without O(E) deep copy (#1670, #887) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same fix as the chain.py edge-index attach (8765c0bd), applied to hop.py — the traversal hot path. reset_index(drop=False) + rename deep-copied + block-consolidated the whole post-filter edge frame (~80ms @2M edges) on every hop, even seeded ones. Now a shallow copy + assigning the frame index as the id column: identical id values (dedup/join key only, never used positionally — verified no downstream .iloc/.loc/.index dependence), no O(E) copy. Consistent ~10-25ms lower per raw hop/chain traversal @500k/2M pandas; the Cypher RETURN traversal is still dominated by binding materialization (tracked #887). No user-frame contamination; hop + chain + gfql suites green (152 + 3014 passed; 1 pre-existing umap env failure unrelated). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_012rRBmezaBEi2CJgDGkMNgR --- graphistry/compute/hop.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/graphistry/compute/hop.py b/graphistry/compute/hop.py index 28e6b1317c..79ee79e9ba 100644 --- a/graphistry/compute/hop.py +++ b/graphistry/compute/hop.py @@ -269,10 +269,14 @@ def _domain_union(left: Optional[DomainT], right: Optional[DomainT]) -> Optional GFQL_EDGE_INDEX = generate_safe_column_name('edge_index', pre_indexed_edges, prefix='__gfql_', suffix='__') - edges_indexed = pre_indexed_edges.reset_index(drop=False) - pre_indexed_cols = set(pre_indexed_edges.columns) - index_col_name = next(col for col in edges_indexed.columns if col not in pre_indexed_cols) - edges_indexed = edges_indexed.rename(columns={index_col_name: GFQL_EDGE_INDEX}) + # Attach the synthetic per-edge id WITHOUT copying edge data (#1670): + # reset_index(drop=False) + rename deep-copied + block-consolidated the + # whole (post-filter) edge frame (~80ms @2M edges) on every hop — the + # traversal hot path. A shallow copy + assigning the index as a column + # gives identical id values (used only as a dedup/join key, never + # positionally) with no O(E) copy. Mirrors the chain.py edge-index attach. + edges_indexed = pre_indexed_edges.copy(deep=False) + edges_indexed[GFQL_EDGE_INDEX] = edges_indexed.index EDGE_ID = GFQL_EDGE_INDEX else: edges_indexed = query_if_not_none(edge_query, g2.filter_edges_by_dict(edge_match)._edges) From 6f2419caef171a1b082de71ceb51d834a5e30eac Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Mon, 6 Jul 2026 20:41:10 -0700 Subject: [PATCH 04/24] feat(gfql/polars): native rows(binding_ops) for fixed-length connected patterns (#1709) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Native polars bindings-row tables — the rows(binding_ops=...) op emitted by Cypher multi-alias lowering. Unblocks traversal-shaped Cypher on polars: graph-benchmark q1/q2 (top-k in-degree), q8/q9 (fixed 2-hop counts), multi-alias property projections, cross-alias grouped aggregates. - binding_rows_polars: seed filter -> per-edge orient (fwd/rev/undirected, no-dedupe concat) + inner join -> node-filter semi join -> per-alias left-join assembly (alias.{col} node props, edge_alias.{col} payload, bare alias id cols). Same meaningful schema as pandas; pandas' internal join-residue columns intentionally not replicated. - select_extend_polars: native with_(extend=True) (bindings-path aggregate lowering emits it; previously NIE'd). - group_by key_prefixes guard: whole-row bindings grouping would have been silently dropped by the native dispatch once rows() stopped NIE-ing -> now declines honestly (latent wrong-answer trap). - DECLINES (None -> honest NIE, NO-CHEATING): var-length/multihop edges, shortestPath scalar bindings, node query=/edge query/endpoint-match params, hop labels, HAS_-label disambiguation, seeded re-entry contexts, node-cartesian mode, alias_endpoints variant, join-key dtype divergence. Conformance harness: _round_floats now renders non-bool numeric (incl clean-coercible object) columns as float64 on both sides — pandas' hop internals upcast int64->float64 on the 2nd alias (engine artifact; polars faithfully keeps Int64), so the astype(str) gate failed on '20.0' vs '20' for equal values. Same non-semantic noise class as the existing summation-order rounding and null-repr normalization; genuine value differences still fail. Tests: +20 (parity incl q1/q8/q9 shapes, NIE assertions, raw-table schema, path multiplicity, undirected self-loop); DEFERRED->supported moves in the row-pipeline + conformance corpora. gfql suite 3036 passed; compute 4323 passed (9 pre-existing umap/dask env failures). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_012rRBmezaBEi2CJgDGkMNgR --- .../compute/gfql/lazy/engine/polars/chain.py | 13 +- .../gfql/lazy/engine/polars/row_pipeline.py | 190 ++++++++++++++++++ .../gfql/test_engine_polars_binding_rows.py | 144 +++++++++++++ .../test_engine_polars_cypher_conformance.py | 26 ++- .../gfql/test_engine_polars_row_pipeline.py | 28 ++- 5 files changed, 390 insertions(+), 11 deletions(-) create mode 100644 graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index 463da88112..13a1cefa7d 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -353,7 +353,7 @@ def _try_native_row_op(g_cur, op): from graphistry.Engine import Engine from .row_pipeline import ( select_polars, with_columns_polars, order_by_polars, group_by_polars, - unwind_polars, where_rows_polars, + unwind_polars, where_rows_polars, binding_rows_polars, ) from .pattern_apply import ( rows_binding_ops_polars, semi_apply_mark_polars, anti_semi_apply_polars, @@ -361,6 +361,12 @@ def _try_native_row_op(g_cur, op): from .search import search_any_polars fn = getattr(op, "function", None) + if fn == "rows" and op.params.get("binding_ops") is not None: + # Multi-alias bindings table (#1709): native for fixed-length connected + # patterns; binding_rows_polars declines (None → NIE) outside that subset. + if op.params.get("alias_endpoints") is not None: + return None + return binding_rows_polars(g_cur, op.params["binding_ops"]) if _call_native_on_polars(op): # frame ops (rows/limit/skip/distinct/drop_cols) — engine-polymorphic return op.execute(g=g_cur, prev_node_wavefront=None, target_wave_front=None, engine=Engine.POLARS) @@ -405,6 +411,11 @@ 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", [])) if fn == "unwind": return unwind_polars(g_cur, op.params.get("expr", ""), op.params.get("as_", "value")) diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index f984acea42..240441eea5 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -804,3 +804,193 @@ def unwind_polars(g: Plottable, expr: str, as_: str = "value") -> Optional[Plott rhs = pl.DataFrame({as_: values}) return _rewrap(g, table.join(rhs, how="cross")) + +def select_extend_polars(g: Plottable, items: Sequence[Any]) -> 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), + 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))) + if exprs is None: + return None + out = table.with_columns(exprs) + if _select_emits_temporal_constructor_text(out): + return None + return _rewrap(g, out) + + +def binding_rows_polars(g: Plottable, binding_ops: Sequence[Any]) -> Optional[Plottable]: + """Native polars bindings-row table for FIXED-LENGTH connected patterns (#1709). + + Materializes one row per matched path for an alternating ``n/e/n/...`` pattern + (the ``rows(binding_ops=...)`` op emitted by Cypher multi-alias lowering), with + the same meaningful schema as the pandas engine: bare ``alias`` id columns, + ``edge_alias.col`` edge-payload columns, and ``alias.{col}`` node-property + columns per node alias. (The pandas frame additionally carries join-residue + columns — raw ``node_id``, ``a__a_join__``, leaked ``__gfql_edge_index__`` — + that no lowered query references; those are intentionally not replicated.) + + 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: + never bridges to pandas. Parity gate: differential tests vs the pandas oracle. + """ + import polars as pl + from graphistry.compute.ast import ASTEdge, ASTNode, from_json as ast_from_json + from graphistry.compute.gfql.row.pipeline import RowPipelineMixin + from graphistry.compute.gfql.same_path.edge_semantics import EdgeSemantics + from .predicates import filter_by_dict_polars + + nodes = g._nodes + edges = g._edges + node_id = g._node + src = g._source + dst = g._destination + if nodes is None or edges is None or node_id is None or src is None or dst is None: + return None + if getattr(g, "_gfql_start_nodes", None) is not None: + # Bounded re-entry seeds the first alias from carried rows — pandas-only. + return None + + ops = [ast_from_json(op_json, validate=False) for op_json in binding_ops] + # Shared validation (engine-agnostic): raises the canonical GFQLValidationError + # for malformed op sequences / duplicate aliases — same error as pandas. + RowPipelineMixin._gfql_validate_binding_ops(ops) + if RowPipelineMixin._gfql_binding_ops_mode(ops) == "node_cartesian": + return None # MATCH (a), (b) cross joins: deferred (rare; own schema study) + if RowPipelineMixin._gfql_is_shortest_path_scalar_binding_ops(ops): + return None # shortestPath scalar contract: BFS/native backends, pandas-only + + for idx, op in enumerate(ops): + if idx % 2 == 0: + if not isinstance(op, ASTNode) or getattr(op, "query", None) is not None: + return None + else: + if not isinstance(op, ASTEdge): + return None + sem = EdgeSemantics.from_edge(op) + if sem.is_multihop: + return None + if op.direction not in ("forward", "reverse", "undirected"): + return None + if any( + getattr(op, attr, None) is not None + for attr in ( + "edge_query", "source_node_match", "destination_node_match", + "source_node_query", "destination_node_query", + "label_node_hops", "label_edge_hops", + "output_min_hops", "output_max_hops", + ) + ): + return None + if bool(getattr(op, "label_seeds", False)) or bool(getattr(op, "include_zero_hop_seed", False)): + return None + # Duplicate-id + HAS_