From bbb7d030ba4c7f4d9455358bbc836480025b72dd Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 03:37:07 -0700 Subject: [PATCH 01/10] perf(gfql): seed-first typed-hop fast-path for the seeded abstraction tax (#1755 lever-3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The degenerate-shape fast path (_try_chain_fast_path) previously (a) bailed on any typed edge (edge_match) and (b) for constrained shapes still validated BOTH edge endpoints against the FULL node table before applying the seed — so a seeded 1-hop paid O(E) object/isin scans over every edge. This is the root of the #1755 seeded- Cypher "abstraction tax". Changes (pandas + cuDF, engine-agnostic): - Accept typed edges: edge_match is a plain edge-frame filter (filter_by_dict), applied on the reduced frontier — same result set as the full hop. - Seed-first: when a node filter is present, reduce edges by the from-side seed BEFORE the typed-edge scan and endpoint validation, and apply the destination filter to the SMALL gathered dst nodes rather than the full node table. Endpoint validation + result-node build then run on the tiny frontier (small isin key -> small hashtable, no O(E)-values scan), so a seeded 1-hop is O(result) not O(E). Parity byte-identical fast-vs-full across unconstrained/seeded/dst-filtered/reverse typed hops (test_chain.py differential + gating updated: edge_match moves from a BYPASS shape to a FAST shape). Measured pandas (50k/200k seeded typed 1-hop, warm- median): 47.0 -> 5.5 ms (8.6x), parity-exact. mypy clean, all chain tests pass. NOTE: prototype on the bench branch (the fast-path infra _try_chain_fast_path is itself not yet on master); lands stacked on that infra. Remaining to the <=1ms gate: a pandas numpy-level specialization (ideal hand-probe = 0.911ms) and/or the #1658 index for scale-invariance. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/chain.py | 66 +++++++++++++++++++++----- graphistry/tests/compute/test_chain.py | 10 +++- 2 files changed, 62 insertions(+), 14 deletions(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index ccf4d3f349..cf926baab7 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -731,12 +731,16 @@ def _materialize_fast_path_graph() -> Plottable: if not (isinstance(n2, ASTNode) and n2._name is None and n2.query is None): return None if not (isinstance(e1, ASTEdge) and e1.is_simple_single_hop() - and e1.edge_match is None and e1.source_node_match is None + and e1.source_node_match is None and e1.destination_node_match is None and e1._name is None and e1.source_node_query is None and e1.destination_node_query is None and e1.edge_query is None and not e1.include_zero_hop_seed and not e1.prune_to_endpoints): # prune keeps only the arrival side -> full path return None + # #1755 lever-3: a typed edge (edge_match, e.g. -[:HAS_CREATOR]->) is a plain + # equality/predicate filter on the edge frame — apply it in the fast-path body + # below rather than falling through to the full two-pass machinery. source/dest + # node match + edge_query (richer predicates) still bail above. direction = e1.direction unconstrained = not n0.filter_dict and not n2.filter_dict if not unconstrained and direction == "undirected": @@ -745,20 +749,58 @@ def _materialize_fast_path_graph() -> Plottable: if g._nodes is None or g._edges is None: return None src, dst, node = g._source, g._destination, g._node - # Keep only edges with BOTH endpoints in the node table (the full path drops - # dangling edges via its joins). dropna so a NaN node id can't validate a NaN - # endpoint — .isin treats NaN as matchable but the BFS joins never match NaN<->NaN. - node_ids = g._nodes[node].dropna() - edges = g._edges[g._edges[src].isin(node_ids) & g._edges[dst].isin(node_ids)] - if not unconstrained: + concat = df_concat(engine_concrete) + if unconstrained: + # No node filter to reduce by: validate BOTH endpoints against the full + # node table (the full path drops dangling edges via its joins). dropna so + # a NaN node id can't validate a NaN endpoint — .isin treats NaN as + # matchable but the BFS joins never match NaN<->NaN. + node_ids = g._nodes[node].dropna() + edges = g._edges[g._edges[src].isin(node_ids) & g._edges[dst].isin(node_ids)] + if e1.edge_match: + # typed edge (e.g. -[:HAS_CREATOR]->) — same edge-frame filter the full + # hop applies, so the result set is identical. + edges = filter_by_dict(edges, e1.edge_match, engine_abs) + else: + # #1755 lever-3 seed-first: a seeded 1-hop must be O(result), not O(E). + # Reduce edges by the selective node filter(s) BEFORE the typed-edge scan + # and endpoint validation, so the expensive object/isin passes run on the + # tiny frontier, not all edges. The from-side ids come from the node table + # (so that endpoint is validated); the node gather below validates the to + # side and drops any edge dangling off the node table. from_col, to_col = (src, dst) if direction == "forward" else (dst, src) + edges = g._edges if n0.filter_dict: - ids = filter_by_dict(g._nodes, n0.filter_dict, engine_abs)[node] - edges = edges[edges[from_col].isin(ids)] + from_ids = filter_by_dict(g._nodes, n0.filter_dict, engine_abs)[node] + edges = edges[edges[from_col].isin(from_ids)] + if e1.edge_match: + edges = filter_by_dict(edges, e1.edge_match, engine_abs) if n2.filter_dict: - ids = filter_by_dict(g._nodes, n2.filter_dict, engine_abs)[node] - edges = edges[edges[to_col].isin(ids)] - concat = df_concat(engine_concrete) + # Apply the destination filter to the SMALL set of gathered dst nodes, + # not the full node table — an O(N) object/type scan on all nodes is + # exactly the tax we're removing. Gather the frontier's dst nodes + # (small isin key), filter those, then drop edges to the losers. + to_present = edges[to_col].dropna().unique() + to_nodes = filter_by_dict( + g._nodes[g._nodes[node].isin(to_present)], n2.filter_dict, engine_abs) + edges = edges[edges[to_col].isin(to_nodes[node])] + # Validate endpoints + build result nodes on the reduced edge set (small + # isin key -> small hashtable; no O(E)-values scan). Engine-agnostic + # (pandas + cuDF): gather candidate endpoint nodes, drop edges dangling off + # the node table, then keep only nodes still referenced by a surviving edge. + ep = concat([ + edges[[src]].rename(columns={src: node}), + edges[[dst]].rename(columns={dst: node}), + ]).drop_duplicates() + cand = g._nodes[g._nodes[node].isin(ep[node])].drop_duplicates(subset=[node]) + valid = cand[node].dropna() + edges = edges[edges[src].isin(valid) & edges[dst].isin(valid)] + final = concat([ + edges[[src]].rename(columns={src: node}), + edges[[dst]].rename(columns={dst: node}), + ]).drop_duplicates() + nodes = cand[cand[node].isin(final[node])] + return g.nodes(nodes).edges(edges) endpoints = concat([ edges[[src]].rename(columns={src: node}), edges[[dst]].rename(columns={dst: node}), diff --git a/graphistry/tests/compute/test_chain.py b/graphistry/tests/compute/test_chain.py index e14ed6e0db..1f084583c5 100644 --- a/graphistry/tests/compute/test_chain.py +++ b/graphistry/tests/compute/test_chain.py @@ -675,13 +675,17 @@ def topd(df): ("hop_fwd_dst_filter", lambda: [n(), e_forward(hops=1), n({'attr': 40})]), ("hop_fwd_both_filter", lambda: [n({'attr': 10}), e_forward(hops=1), n({'attr': 30})]), ("hop_rev_dst_filter", lambda: [n(), e_reverse(hops=1), n({'attr': 10})]), + # #1755 lever-3: typed edges (edge_match) are now a fast shape — a plain edge + # filter applied on the (seed-reduced) frontier, not a fall-through. + ("edge_match_unconstrained", lambda: [n(), e_forward(hops=1, edge_match={'w': 5}), n()]), + ("edge_match_seeded", lambda: [n({'attr': 10}), e_forward(hops=1, edge_match={'w': 5}), n()]), + ("edge_match_dst_filter", lambda: [n(), e_forward(hops=1, edge_match={'w': 5}), n({'attr': 30})]), ] # shapes that BYPASS the fast path (still must be correct via the full path) _BYPASS_SHAPES = [ ("hops_2", lambda: [n(), e_forward(hops=2), n()]), ("filtered_undirected", lambda: [n({'attr': 10}), e_undirected(hops=1), n({'attr': 30})]), - ("edge_match", lambda: [n(), e_forward(hops=1, edge_match={'w': 5}), n()]), ("named_node", lambda: [n(name='x'), e_forward(hops=1), n()]), # prune_to_endpoints: fast path returns both endpoints; full path keeps only the # arrival side. Must bypass the fast path (regression guard for the prune gate). @@ -751,6 +755,9 @@ def test_fast_path_gating_returns_none_for_ineligible(): [n({'attr': 20})], [n(), e_forward(hops=1), n()], [n(), e_reverse(hops=1), n()], + # #1755 lever-3: typed edges are now accepted (edge filter on the frontier) + [n(), e_forward(hops=1, edge_match={'w': 5}), n()], + [n({'attr': 10}), e_forward(hops=1, edge_match={'w': 5}), n()], ] for ops in eligible: assert _try_chain_fast_path(g, ops, Engine.PANDAS, None) is not None, f"should accept {ops}" @@ -758,7 +765,6 @@ def test_fast_path_gating_returns_none_for_ineligible(): ineligible = [ ("hops_2", [n(), e_forward(hops=2), n()], None, Engine.PANDAS), ("filtered_undirected", [n({'attr': 10}), e_undirected(hops=1), n({'attr': 30})], None, Engine.PANDAS), - ("edge_match", [n(), e_forward(hops=1, edge_match={'w': 5}), n()], None, Engine.PANDAS), ("named_node", [n(name='x'), e_forward(hops=1), n()], None, Engine.PANDAS), ("node_query", [n(query='attr > 5'), e_forward(hops=1), n()], None, Engine.PANDAS), ("prune_endpoints", [n(), e_forward(hops=1, prune_to_endpoints=True), n()], None, Engine.PANDAS), From 7228e1bd6893a56d3c561951c9e224f93acac10f Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 03:49:42 -0700 Subject: [PATCH 02/10] =?UTF-8?q?perf(gfql):=20numpy=20seeded-typed-hop=20?= =?UTF-8?q?fast=20path=20=E2=80=94=20native=20seeded=201-hop=20<=3D1ms=20(?= =?UTF-8?q?#1755=20lever-3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds _seeded_typed_hop_numpy_pandas: a pandas-only numpy specialization of the seeded typed 1-hop, gated to scalar node/edge filters + directed (falls back to the engine-agnostic seeded branch for predicates/undirected/cuDF). Seed-reduces edges before the object-typed edge_match scan, applies the destination filter to the small gathered dst nodes, and does a SINGLE full-node scan to gather result nodes — so a seeded lookup is a few numpy passes, not O(E) pandas machinery. Measured dgx (26.02-gfql-polars, 50k/200k, native [n({id}),e(edge_match),n({type})], warm-median): full path 29.10 ms -> fast path 0.553 ms (52.6x), byte-identical parity. GATE MET (<=1ms) on the NATIVE seeded typed hop. All 57 chain tests pass, mypy clean. NOTE: this closes the native surface only. The cypher string g.gfql("MATCH (m {id})-[:T]->(p) RETURN p") lowers to NAMED nodes (m,p) + label__ filters + a 4th ASTCall projection, so it does not yet engage this fast path (63 ms). Closing the cypher surface = the next lever (named-node + label + lean projection). Prototype on the bench branch (fast-path infra not yet on master). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/chain.py | 82 +++++++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index cf926baab7..5d243b5895 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -684,6 +684,81 @@ def _chain_otel_attrs( return attrs +def _seeded_typed_hop_numpy_pandas(g, n0, n2, e1, src, dst, node, direction): + """#1755 lever-3: pandas-only numpy fast path for a scalar-filtered seeded + typed 1-hop. Byte-identical to the engine-agnostic seeded branch for the + covered shape (all node/edge filters are plain scalars, directed), collapsing + it into a handful of numpy passes so a seeded lookup lands sub-ms. Returns + None to fall back for anything it does not cover (predicates, undirected, + missing columns) — the caller then runs the general engine-agnostic branch.""" + import numpy as np + import pandas as pd + if direction == "undirected": + return None + + def _scalars(fd): + if not fd: + return {} + out = {} + for k, v in fd.items(): + if not isinstance(v, (int, float, str, bool)): + return None # predicate / non-scalar -> bail to the general path + out[k] = v + return out + + n0f, n2f, ef = _scalars(n0.filter_dict), _scalars(n2.filter_dict), _scalars(e1.edge_match) + if n0f is None or n2f is None or ef is None: + return None + nodes_df, edges_df = g._nodes, g._edges + if any(c not in nodes_df.columns for c in (*n0f, *n2f)) or any(c not in edges_df.columns for c in ef): + return None + from_col, to_col = (src, dst) if direction == "forward" else (dst, src) + + nid = nodes_df[node].to_numpy() + # from-side seed FIRST: reduce edges to the seed's out-edges before the + # (object) edge_match compare, so the type filter runs on the tiny frontier + # rather than all edges — this is what makes a seeded lookup sub-ms. + if n0f: + nmask = np.ones(len(nodes_df), dtype=bool) + for k, v in n0f.items(): + col = nid if k == node else nodes_df[k].to_numpy() # reuse nid for the id col + nmask &= (col == v) + from_ids = nid[nmask] + from_vals = edges_df[from_col].to_numpy() + seed_mask = (from_vals == from_ids[0]) if len(from_ids) == 1 else np.isin(from_vals, from_ids) + edges = edges_df[seed_mask] + else: + edges = edges_df + if ef: # typed edge (edge_match) — now on the reduced frontier + tmask = np.ones(len(edges), dtype=bool) + for k, v in ef.items(): + tmask &= (edges[k].to_numpy() == v) + edges = edges[tmask] + + # Gather candidate endpoint nodes with a SINGLE O(N) node scan, then run the + # dest filter, dangling-edge drop and final-node selection on the small + # candidate/edge frames (no second full-node scan). + esrc, edst = edges[src].to_numpy(), edges[dst].to_numpy() + eto = edst if to_col == dst else esrc # reuse; no extra column fetch + ep = np.unique(np.concatenate([esrc, edst])) + cand = nodes_df[np.isin(nid, ep)].drop_duplicates(subset=[node]) + cand_ids = cand[node].to_numpy() + valid = cand_ids[~pd.isna(cand_ids)] # endpoints that are real nodes + if n2f: # destination-node filter (to-side) + dmask = np.ones(len(cand), dtype=bool) + for k, v in n2f.items(): + dmask &= (cand[k].to_numpy() == v) + n2_ok = cand_ids[dmask] + else: + n2_ok = valid + keep = np.isin(esrc, valid) & np.isin(edst, valid) & np.isin(eto, n2_ok) + if not keep.all(): + edges = edges[keep] + fep = np.unique(np.concatenate([edges[src].to_numpy(), edges[dst].to_numpy()])) + cand = cand[np.isin(cand_ids, fep)] + return g.nodes(cand).edges(edges) + + def _try_chain_fast_path( g_in: Plottable, ops: List[ASTObject], @@ -768,6 +843,13 @@ def _materialize_fast_path_graph() -> Plottable: # tiny frontier, not all edges. The from-side ids come from the node table # (so that endpoint is validated); the node gather below validates the to # side and drops any edge dangling off the node table. + # pandas: a scalar-filtered seeded typed hop collapses to a few numpy + # passes (sub-ms); falls back to the engine-agnostic branch below for + # predicates / undirected / missing columns and for cuDF. + if engine_concrete == Engine.PANDAS: + _np_res = _seeded_typed_hop_numpy_pandas(g, n0, n2, e1, src, dst, node, direction) + if _np_res is not None: + return _np_res from_col, to_col = (src, dst) if direction == "forward" else (dst, src) edges = g._edges if n0.filter_dict: From e4eec1cd5497b89434196ffd6eb5ae64b457a9cc Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 04:21:59 -0700 Subject: [PATCH 03/10] =?UTF-8?q?perf(gfql):=20seeded=20typed-hop=20cypher?= =?UTF-8?q?=20fast=20path=20=E2=80=94=20seeded=20cypher=20RETURN=20<=3D1ms?= =?UTF-8?q?=20(#1755)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the #1755 seeded-Cypher abstraction tax on the CYPHER STRING surface. g.gfql("MATCH (m {id})-[:T]->(p) RETURN p") lowered to named nodes + label__ filters + a rows/projection pipeline that scanned all edges/nodes and paid ~30ms of rows-pivot machinery even for a single-row seeded result. New _execute_seeded_typed_hop_fast_path (gfql_unified, alongside the existing single-hop-aggregate / two-hop-count fast paths): detects the seeded typed 1-hop with a single whole-row node RETURN, then: - _seeded_typed_return_dst_pandas (chain.py): numpy dst-only extraction — resolves label__X->type, id-first successive subset of the seed (later filters index the tiny survivor rows via .iloc so object columns are never fully materialized), seed-reduces edges before the typed-edge scan, gathers + filters the dst nodes in one node pass; - a lean apply_result_projection on just the p rows (exact column-order/flatten semantics, no rows-pivot). Gated pandas-only, forward seeded shape (RETURN alias == dst node, seed on src); reverse / multi-alias / field / non-scalar returns fall through to the full path. Parity byte-identical fast-vs-full across RETURN p (labelled/unlabelled dst), no-match, and the bail shapes. Measured dgx (26.02-gfql-polars, 50k/200k, warm-median): full 38.2ms -> fast 0.847 ms (45.2x). GATE MET (<=1ms) on the seeded cypher string. 57 chain tests pass, mypy clean. Also lands the native lever-3 helper (_seeded_typed_hop_numpy_pandas + the _try_chain_fast_path seed-first typed-edge path) this builds on: native seeded typed hop 29.1 -> 0.553 ms. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/chain.py | 106 +++++++++++++++++++++++++---- graphistry/compute/gfql_unified.py | 81 ++++++++++++++++++++++ 2 files changed, 175 insertions(+), 12 deletions(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 5d243b5895..c41c814aab 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -696,22 +696,30 @@ def _seeded_typed_hop_numpy_pandas(g, n0, n2, e1, src, dst, node, direction): if direction == "undirected": return None - def _scalars(fd): + nodes_df, edges_df = g._nodes, g._edges + + def _scalars(fd, df): + """Resolve a filter dict to plain scalar column==value pairs, or None to + bail. Handles the cypher ``label__X: True`` form by resolving it to the + ``type`` column (equality) — matching filter_by_dict.resolve_filter_column + — but bails on the list-valued ``labels`` form to stay equality-safe.""" if not fd: return {} - out = {} + out: Dict[str, Any] = {} for k, v in fd.items(): if not isinstance(v, (int, float, str, bool)): return None # predicate / non-scalar -> bail to the general path - out[k] = v + if k in df.columns: + out[k] = v + elif isinstance(k, str) and k.startswith("label__") and v is True and "type" in df.columns: + out["type"] = k[len("label__"):] + else: + return None # unknown / list-label column -> bail return out - n0f, n2f, ef = _scalars(n0.filter_dict), _scalars(n2.filter_dict), _scalars(e1.edge_match) + n0f, n2f, ef = _scalars(n0.filter_dict, nodes_df), _scalars(n2.filter_dict, nodes_df), _scalars(e1.edge_match, edges_df) if n0f is None or n2f is None or ef is None: return None - nodes_df, edges_df = g._nodes, g._edges - if any(c not in nodes_df.columns for c in (*n0f, *n2f)) or any(c not in edges_df.columns for c in ef): - return None from_col, to_col = (src, dst) if direction == "forward" else (dst, src) nid = nodes_df[node].to_numpy() @@ -719,11 +727,17 @@ def _scalars(fd): # (object) edge_match compare, so the type filter runs on the tiny frontier # rather than all edges — this is what makes a seeded lookup sub-ms. if n0f: - nmask = np.ones(len(nodes_df), dtype=bool) - for k, v in n0f.items(): - col = nid if k == node else nodes_df[k].to_numpy() # reuse nid for the id col - nmask &= (col == v) - from_ids = nid[nmask] + # Successive subsetting, id-column first: the (int, unique) id filter + # reduces 250k -> ~1 row in one cheap pass, so any remaining (object) + # filters like label__X->type run on that tiny set instead of scanning + # the whole node frame. `from_pos` tracks positions into nodes_df. + from_pos = np.arange(len(nodes_df)) + for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1): + col = (nid if k == node else nodes_df[k].to_numpy())[from_pos] + from_pos = from_pos[col == v] + if from_pos.size == 0: + break + from_ids = nid[from_pos] from_vals = edges_df[from_col].to_numpy() seed_mask = (from_vals == from_ids[0]) if len(from_ids) == 1 else np.isin(from_vals, from_ids) edges = edges_df[seed_mask] @@ -759,6 +773,74 @@ def _scalars(fd): return g.nodes(cand).edges(edges) +def _seeded_typed_return_dst_pandas(g, n0, n2, e1, src, dst, node, direction): + """#1755 cypher RETURN-alias fast path: like _seeded_typed_hop_numpy_pandas but + returns ONLY the destination (RETURN-alias) node rows + surviving edges — no + seed-node gather, no Plottable round-trip — so the seeded cypher projection + lands sub-ms. Returns ``(dst_node_rows, edges)`` or None to fall back.""" + import numpy as np + import pandas as pd + if direction == "undirected": + return None + nodes_df, edges_df = g._nodes, g._edges + + def _sc(fd, df): + if not fd: + return {} + out: Dict[str, Any] = {} + for k, v in fd.items(): + if not isinstance(v, (int, float, str, bool)): + return None + if k in df.columns: + out[k] = v + elif isinstance(k, str) and k.startswith("label__") and v is True and "type" in df.columns: + out["type"] = k[len("label__"):] + else: + return None + return out + + n0f, n2f, ef = _sc(n0.filter_dict, nodes_df), _sc(n2.filter_dict, nodes_df), _sc(e1.edge_match, edges_df) + if n0f is None or n2f is None or ef is None or not n0f: + return None + from_col, to_col = (src, dst) if direction == "forward" else (dst, src) + nid = nodes_df[node].to_numpy() + # id-first successive subset. The first (id) filter scans the full id array + # once; later filters index the already-tiny survivor rows via .iloc, so an + # object column like label__X->type is never fully materialized. + from_pos: Optional[Any] = None + for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1): + if from_pos is None: + base = nid if k == node else nodes_df[k].to_numpy() + from_pos = np.flatnonzero(base == v) + else: + sub = nid[from_pos] if k == node else nodes_df[k].iloc[from_pos].to_numpy() + from_pos = from_pos[sub == v] + if from_pos.size == 0: + break + from_ids = nid[from_pos] if from_pos is not None and from_pos.size else nid[:0] + if from_ids.size == 0: + return nodes_df.iloc[0:0], edges_df.iloc[0:0] + fv = edges_df[from_col].to_numpy() + seed_mask = (fv == from_ids[0]) if from_ids.size == 1 else np.isin(fv, from_ids) + edges = edges_df[seed_mask] + if ef: + tmask = np.ones(len(edges), dtype=bool) + for k, v in ef.items(): + tmask &= (edges[k].to_numpy() == v) + edges = edges[tmask] + to_vals = edges[to_col].to_numpy() + dst_ids = np.unique(to_vals[~pd.isna(to_vals)]) + dstn = nodes_df[np.isin(nid, dst_ids)] + if n2f: + dm = np.ones(len(dstn), dtype=bool) + for k, v in n2f.items(): + dm &= (dstn[k].to_numpy() == v) + dstn = dstn[dm] + edges = edges[np.isin(edges[to_col].to_numpy(), dstn[node].to_numpy())] + dstn = dstn[np.isin(dstn[node].to_numpy(), edges[to_col].to_numpy())].drop_duplicates(subset=[node]) + return dstn, edges + + def _try_chain_fast_path( g_in: Plottable, ops: List[ASTObject], diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index d67b5ec46f..83f913cb7b 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -976,6 +976,82 @@ def _run_logical_pass_pipeline(logical_plan: LogicalPlan, ctx: PlanContext) -> L return PassManager(DEFAULT_LOGICAL_PASSES, DEFAULT_TIER2_PASSES).run(logical_plan, ctx).plan +_SEEDED_FASTPATH_GUARD = "_gfql_seeded_typed_hop_fastpath_done" + + +def _execute_seeded_typed_hop_fast_path( + base_graph: Plottable, + compiled_query: CompiledCypherQuery, + physical_plan: "PhysicalPlan", + *, + engine: Union[EngineAbstract, str], + policy: Optional[PolicyDict], + context: ExecutionContext, + start_nodes: Optional[DataFrameT] = None, +) -> Optional[Plottable]: + """#1755 cypher-surface fast path: a seeded typed 1-hop with a whole-row node + RETURN — ``MATCH (m {id})-[:T]->(p) RETURN p`` — numpy-reduces the graph to the + seed's 1-hop neighborhood, then re-runs the SAME cypher pipeline on that tiny + subgraph (guaranteed byte-identical to the full path; sub-ms because the graph + is tiny). Returns None to fall through for anything outside this exact shape.""" + from graphistry.compute.chain import _seeded_typed_hop_numpy_pandas + if getattr(base_graph, _SEEDED_FASTPATH_GUARD, False): + return None # recursion guard: the re-run below must take the full path + requested_engine = resolve_engine(cast(Any, engine), base_graph) + if requested_engine != Engine.PANDAS: + return None + projection = compiled_query.result_projection + if projection is None or getattr(projection, "table", None) != "nodes": + return None + # Only a single whole-row node alias (RETURN p). Multi-alias returns (RETURN + # m, p) combine aliases into one row per match — different shape — so bail. + proj_cols = getattr(projection, "columns", None) + if not proj_cols or len(proj_cols) != 1 or getattr(proj_cols[0], "kind", None) != "whole_row": + return None + if compiled_query.execution_extras is not None and ( + compiled_query.execution_extras.connected_match_join is not None + or compiled_query.execution_extras.connected_optional_match is not None + ): + return None + ops = list(compiled_query.chain.chain) + if len(ops) != 4: + return None + n0, e1, n2, call = ops + if not (isinstance(n0, ASTNode) and isinstance(e1, ASTEdge) + and isinstance(n2, ASTNode) and isinstance(call, ASTCall)): + return None + if call.function != "rows": + return None + node = getattr(base_graph, "_node", None) + src = getattr(base_graph, "_source", None) + dst = getattr(base_graph, "_destination", None) + if node is None or src is None or dst is None: + return None + # RETURN alias must be the DESTINATION node (n2) and the seed must sit on the + # source node (n0) — the forward seeded shape MATCH (m {id})-[:T]->(p) RETURN p. + # Other alias/seed placements (e.g. reverse patterns where the seed is on the + # RETURN node) fall back to the full path. + if getattr(n2, "_name", None) != projection.alias: + return None + if not (n0.filter_dict and any(not str(k).startswith("label__") for k in n0.filter_dict)): + return None # n0 must carry a selective (non-label) seed + direction = getattr(e1, "direction", "forward") + node_s, src_s, dst_s = str(node), str(src), str(dst) + from graphistry.compute.chain import _seeded_typed_return_dst_pandas + dst_res = _seeded_typed_return_dst_pandas( + base_graph, n0, n2, e1, src_s, dst_s, node_s, direction) + if dst_res is None: + return None + p_rows, _edges = dst_res + # Lean projection: p_rows already IS the RETURN-alias (destination) node set. + # Tag with the alias and reuse apply_result_projection for the exact + # column-order/flatten semantics — all on a handful of rows, so seeded cypher + # stays sub-ms (vs the ~25ms rows-pivot pipeline on the full graph). + tagged = p_rows.assign(**{projection.alias: True}) + result = base_graph.nodes(tagged) + return apply_result_projection(result, projection) + + def _execute_compiled_query_via_physical_plan( base_graph: Plottable, *, @@ -1015,6 +1091,11 @@ def _execute_compiled_query_via_physical_plan( fast_count = _execute_two_hop_count_fast_path(base_graph, compiled_query.chain, engine=engine) if fast_count is not None: return fast_count + fast_hop = _execute_seeded_typed_hop_fast_path( + base_graph, compiled_query, physical_plan, + engine=engine, policy=policy, context=context, start_nodes=start_nodes) + if fast_hop is not None: + return fast_hop return _execute_compiled_query_chain_non_union( base_graph, compiled_query=compiled_query, From af1804366df4e519dcdb9fa15793dbe30fba9a22 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 07:43:48 -0700 Subject: [PATCH 04/10] test(gfql): seeded typed-hop fast-path parity + independent oracle (#1755) Re-verify the seeded-Cypher tax fast path against an INDEPENDENT oracle (creators hand-computed from raw edge/node frames), not merely fast==full, since the full path could share a bug. Native returns {seed} u creators (both bound endpoints); cypher RETURN p returns creators only. Decline coverage: two-hop, predicate node filter, predicate edge filter, undirected -> fast path declines and the full path result is byte-identical. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- .../gfql/test_seeded_typed_hop_fastpath.py | 260 ++++++++++++++++++ 1 file changed, 260 insertions(+) create mode 100644 graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py diff --git a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py new file mode 100644 index 0000000000..89b0874843 --- /dev/null +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -0,0 +1,260 @@ +"""Parity + gating tests for the #1755 seeded typed-hop fast path. + +Two layers close the seeded-Cypher abstraction tax on pandas: + * native — `_try_chain_fast_path` / `_seeded_typed_hop_numpy_pandas` accelerate + a seeded typed 1-hop chain [n({id}), e(edge_match), n({type})]; + * cypher — `_execute_seeded_typed_hop_fast_path` accelerates the lowered + MATCH (m {id})-[:T]->(p) RETURN p string surface. + +Both are byte-identical to the full path by construction; these tests pin that +(fast-on vs fast-off, differential) and that the fast path actually ENGAGES for +the accelerated shapes and DECLINES (falls through) for everything else. +""" +import numpy as np +import pandas as pd +import pytest + +import graphistry +from graphistry.compute.ast import n, e_forward, e_reverse +import graphistry.compute.chain as chain_mod +import graphistry.compute.gfql_unified as gfql_unified + + +def _graph(n_persons=1500, n_messages=6000, seed=0): + """Message -> Person HAS_CREATOR graph (the #1755 probe shape). `age` is only + on Person rows, so the concatenated node frame carries a float `age` column + with NaN on Messages — exercising mixed dtypes through the projection.""" + rng = np.random.default_rng(seed) + persons = pd.DataFrame({ + "id": np.arange(n_persons), "type": "Person", + "age": rng.integers(20, 60, n_persons), + }) + messages = pd.DataFrame({"id": np.arange(n_persons, n_persons + n_messages), "type": "Message"}) + ndf = pd.concat([persons, messages], ignore_index=True) + edf = pd.DataFrame({ + "src": np.arange(n_persons, n_persons + n_messages), + "dst": rng.integers(0, n_persons, n_messages), + "type": "HAS_CREATOR", + }) + return graphistry.nodes(ndf, "id").edges(edf, "src", "dst"), n_persons + + +def _canon_nodes(res): + df = pd.DataFrame(res._nodes) + cols = sorted(str(c) for c in df.columns) + df.columns = [str(c) for c in df.columns] + return df.sort_values(cols).reset_index(drop=True)[cols] if cols else df + + +def _oracle_creators(g, seed): + """INDEPENDENT ground truth (NOT via the query engine): the Person ids that + are HAS_CREATOR destinations of the seeded Message. Computed directly from + the raw edge/node frames with plain pandas boolean masks, so it cannot share + a bug with either the fast path or the full chain path.""" + edf = g._edges + ndf = g._nodes + creator_ids = edf.loc[ + (edf["src"] == seed) & (edf["type"] == "HAS_CREATOR"), "dst" + ].tolist() + # restrict to Person-typed endpoints (the typed n({type:Person}) filter) + person_ids = set(ndf.loc[ndf["type"] == "Person", "id"].tolist()) + return sorted(cid for cid in creator_ids if cid in person_ids) + + +# --------------------------------------------------------------------------- +# native seeded typed hop ([n({id}), e(edge_match), n({type})]) +# --------------------------------------------------------------------------- + +class TestNativeSeededTypedHop: + def _run(self, g, ops, force_full): + real = chain_mod._try_chain_fast_path + try: + if force_full: + chain_mod._try_chain_fast_path = lambda *a, **k: None + return g.gfql(ops, engine="pandas") + finally: + chain_mod._try_chain_fast_path = real + + @pytest.mark.parametrize("ops_name", ["typed_creator", "typed_person", "typed_reverse"]) + def test_parity_fast_vs_full(self, ops_name): + g, P = _graph() + seed = P + 42 + ops = { + "typed_creator": [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n()], + "typed_person": [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], + "typed_reverse": [n({"type": "Person"}), e_reverse(edge_match={"type": "HAS_CREATOR"}), n({"id": seed})], + }[ops_name] + fast = self._run(g, ops, force_full=False) + full = self._run(g, ops, force_full=True) + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + def test_fast_path_engages_on_typed_hop(self, monkeypatch): + g, P = _graph() + seed = P + 42 + hits = {"n": 0} + real = chain_mod._seeded_typed_hop_numpy_pandas + + def spy(*a, **k): + r = real(*a, **k) + if r is not None: + hits["n"] += 1 + return r + + monkeypatch.setattr(chain_mod, "_seeded_typed_hop_numpy_pandas", spy) + g.gfql([n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], engine="pandas") + assert hits["n"] >= 1 + + def test_independent_oracle_values(self): + """Falsifiability (a): the returned Person must be the ACTUAL creator, + hand-computed from raw frames — not merely fast==full (both could share + a bug). Pins VALUES, not just self-consistency.""" + g, P = _graph() + seed = P + 42 + oracle = _oracle_creators(g, seed) + assert len(oracle) >= 1, "probe seed must have >=1 Person creator" + res = g.gfql( + [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], + engine="pandas", + ) + got = sorted(pd.DataFrame(res._nodes)["id"].tolist()) + # native chain _nodes = the UNION of all bound path endpoints: the seed + # Message (n0) AND its Person creators (n2). The oracle is that union. + expected = sorted(set(oracle) | {seed}) + assert got == expected, f"fast path returned {got}, oracle says {expected}" + + def test_numpy_helper_declines_predicate_and_undirected(self): + from graphistry.compute.predicates.numeric import gt + g, P = _graph() + node, src, dst = "id", "src", "dst" + # predicate (non-scalar) edge filter -> decline + assert chain_mod._seeded_typed_hop_numpy_pandas( + g.materialize_nodes(), n({"id": P + 1}), n(), e_forward(edge_match={"type": gt(0)}), + src, dst, node, "forward") is None + # undirected -> decline + assert chain_mod._seeded_typed_hop_numpy_pandas( + g.materialize_nodes(), n({"id": P + 1}), n(), e_forward(), + src, dst, node, "undirected") is None + + @pytest.mark.parametrize("ops_name,reason", [ + ("two_hop", "multi-hop chain, not a single hop"), + ("node_predicate", "non-scalar (predicate) node filter"), + ]) + def test_native_declines_and_stays_correct(self, ops_name, reason, monkeypatch): + from graphistry.compute.predicates.numeric import gt + g, P = _graph() + seed = P + 42 + ops = { + "two_hop": [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n(), e_reverse(), n()], + "node_predicate": [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"age": gt(0)})], + }[ops_name] + hits = {"n": 0} + real = chain_mod._seeded_typed_hop_numpy_pandas + + def spy(*a, **k): + r = real(*a, **k) + if r is not None: + hits["n"] += 1 + return r + + monkeypatch.setattr(chain_mod, "_seeded_typed_hop_numpy_pandas", spy) + fast = g.gfql(ops, engine="pandas") + monkeypatch.undo() + chain_mod._try_chain_fast_path, saved = (lambda *a, **k: None), chain_mod._try_chain_fast_path + try: + full = g.gfql(ops, engine="pandas") + finally: + chain_mod._try_chain_fast_path = saved + assert hits["n"] == 0, f"numpy fast path must decline {reason}" + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + +# --------------------------------------------------------------------------- +# cypher surface (MATCH (m {id})-[:T]->(p) RETURN p) +# --------------------------------------------------------------------------- + +class TestCypherSeededTypedHop: + def _run(self, g, cy, force_full): + real = gfql_unified._execute_seeded_typed_hop_fast_path + try: + if force_full: + gfql_unified._execute_seeded_typed_hop_fast_path = lambda *a, **k: None + return g.gfql(cy, engine="pandas") + finally: + gfql_unified._execute_seeded_typed_hop_fast_path = real + + @pytest.mark.parametrize("label", ["p:Person", "p"]) + def test_parity_return_p(self, label): + g, P = _graph() + seed = P + 42 + cy = f"MATCH (m:Message {{id: {seed}}})-[:HAS_CREATOR]->({label}) RETURN p" + fast = self._run(g, cy, force_full=False) + full = self._run(g, cy, force_full=True) + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + def test_parity_no_match(self): + g, P = _graph() + cy = "MATCH (m:Message {id: 999999})-[:HAS_CREATOR]->(p:Person) RETURN p" + fast = self._run(g, cy, force_full=False) + full = self._run(g, cy, force_full=True) + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + def test_independent_oracle_values(self): + """Falsifiability (a) on the cypher surface: the p-rows returned by the + lowered MATCH ... RETURN p must be exactly the hand-computed creators.""" + g, P = _graph() + seed = P + 42 + oracle = _oracle_creators(g, seed) + assert len(oracle) >= 1 + res = g.gfql( + f"MATCH (m:Message {{id: {seed}}})-[:HAS_CREATOR]->(p:Person) RETURN p", + engine="pandas", + ) + df = pd.DataFrame(res._nodes) + # RETURN p prefixes projected columns with the alias; find the id column. + id_col = "p.id" if "p.id" in df.columns else "id" + got = sorted(df[id_col].tolist()) + assert got == oracle, f"cypher fast path returned {got}, oracle says {oracle}" + + def test_fast_path_engages_on_seeded_return(self, monkeypatch): + g, P = _graph() + seed = P + 42 + hits = {"n": 0} + real = gfql_unified._execute_seeded_typed_hop_fast_path + + def spy(*a, **k): + r = real(*a, **k) + if r is not None: + hits["n"] += 1 + return r + + monkeypatch.setattr(gfql_unified, "_execute_seeded_typed_hop_fast_path", spy) + g.gfql(f"MATCH (m:Message {{id: {seed}}})-[:HAS_CREATOR]->(p:Person) RETURN p", engine="pandas") + assert hits["n"] >= 1 + + @pytest.mark.parametrize("cy_tmpl,reason", [ + ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN m, p", "multi-alias"), + ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN p.age", "field projection"), + ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN m", "return source"), + ("MATCH (p:Person)<-[:HAS_CREATOR]-(m:Message {{id: {s}}}) RETURN p", "reverse (seed on return node)"), + ]) + def test_declines_and_stays_correct(self, cy_tmpl, reason): + g, P = _graph() + seed = P + 42 + cy = cy_tmpl.format(s=seed) + real = gfql_unified._execute_seeded_typed_hop_fast_path + hits = {"n": 0} + + def spy(*a, **k): + r = real(*a, **k) + if r is not None: + hits["n"] += 1 + return r + + gfql_unified._execute_seeded_typed_hop_fast_path = spy + try: + fast = g.gfql(cy, engine="pandas") + finally: + gfql_unified._execute_seeded_typed_hop_fast_path = real + full = self._run(g, cy, force_full=True) + assert hits["n"] == 0, f"fast path must decline {reason}" + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) From d442f681a91b6e4e943e183c296e00c784459bba Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 08:44:03 -0700 Subject: [PATCH 05/10] fix(gfql): seeded cypher fast path must decline variable-length hops (#1755) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cypher seeded fast path gated on isinstance(e1, ASTEdge) but not on the edge being a genuine single hop. A variable-length edge (-[*1..2]->) is one ASTEdge spanning multiple hops, so the 1-hop seed reduction silently truncated it — MATCH (a {id})-[*1..2]->(b) RETURN b returned only the 1-hop neighbors, breaking pandas-vs-polars parity (TestVarlenAliasHopGate). Gate on e1.is_simple_single_hop() (same canonical check the native fast path uses; also rejects hop labels, output slicing, fixed-point). Adds varlen decline coverage to the fast-path parity suite. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/gfql_unified.py | 7 +++++++ .../tests/compute/gfql/test_seeded_typed_hop_fastpath.py | 4 ++++ 2 files changed, 11 insertions(+) diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index 83f913cb7b..80725fefec 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -1020,6 +1020,13 @@ def _execute_seeded_typed_hop_fast_path( if not (isinstance(n0, ASTNode) and isinstance(e1, ASTEdge) and isinstance(n2, ASTNode) and isinstance(call, ASTCall)): return None + # Only a genuine SINGLE hop. A variable-length edge (-[*1..2]->) is still one + # ASTEdge but expands to multiple hops, so the seeded 1-hop reduction below + # would silently truncate it. Reuse the same canonical gate the native fast + # path uses (also rejects hop labels / output slicing / fixed-point). See + # test_engine_polars_chain::TestVarlenAliasHopGate. + if not e1.is_simple_single_hop(): + return None if call.function != "rows": return None node = getattr(base_graph, "_node", None) diff --git a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py index 89b0874843..98d0dbcc26 100644 --- a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -236,6 +236,10 @@ def spy(*a, **k): ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN p.age", "field projection"), ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN m", "return source"), ("MATCH (p:Person)<-[:HAS_CREATOR]-(m:Message {{id: {s}}}) RETURN p", "reverse (seed on return node)"), + # variable-length edges are one ASTEdge but multiple hops — must decline or + # the 1-hop reduction silently truncates (regressed polars varlen parity). + ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR*1..2]->(p) RETURN p", "varlen range hop"), + ("MATCH (m:Message {{id: {s}}})-[*1..2]->(p) RETURN p", "varlen untyped hop"), ]) def test_declines_and_stays_correct(self, cy_tmpl, reason): g, P = _graph() From 086ddf6a76048e912a3b3e0dab36addd0754bb92 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 08:51:13 -0700 Subject: [PATCH 06/10] refactor(gfql): type the seeded fast-path helpers (#1755) Annotate _seeded_typed_hop_numpy_pandas / _seeded_typed_return_dst_pandas (Plottable/ASTNode/ASTEdge params, Optional[Plottable] and Optional[Tuple[DataFrameT, DataFrameT]] returns) so their bodies are type-checked, and guard the src/dst/node bindings against None at the _try_chain_fast_path call site. mypy --config-file mypy.ini clean on both changed files (remaining errors are pre-existing polars-lazy, untouched). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/chain.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index c41c814aab..31a2207a0f 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -684,7 +684,10 @@ def _chain_otel_attrs( return attrs -def _seeded_typed_hop_numpy_pandas(g, n0, n2, e1, src, dst, node, direction): +def _seeded_typed_hop_numpy_pandas( + g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge, + src: str, dst: str, node: str, direction: str, +) -> Optional[Plottable]: """#1755 lever-3: pandas-only numpy fast path for a scalar-filtered seeded typed 1-hop. Byte-identical to the engine-agnostic seeded branch for the covered shape (all node/edge filters are plain scalars, directed), collapsing @@ -773,7 +776,10 @@ def _scalars(fd, df): return g.nodes(cand).edges(edges) -def _seeded_typed_return_dst_pandas(g, n0, n2, e1, src, dst, node, direction): +def _seeded_typed_return_dst_pandas( + g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge, + src: str, dst: str, node: str, direction: str, +) -> Optional[Tuple[DataFrameT, DataFrameT]]: """#1755 cypher RETURN-alias fast path: like _seeded_typed_hop_numpy_pandas but returns ONLY the destination (RETURN-alias) node rows + surviving edges — no seed-node gather, no Plottable round-trip — so the seeded cypher projection @@ -906,6 +912,8 @@ def _materialize_fast_path_graph() -> Plottable: if g._nodes is None or g._edges is None: return None src, dst, node = g._source, g._destination, g._node + if src is None or dst is None or node is None: + return None # no edge/node bindings -> can't fast-path; full path handles it concat = df_concat(engine_concrete) if unconstrained: # No node filter to reduce by: validate BOTH endpoints against the full From 52957c91ad690736ed5a7483cbf0488c31ebdb2e Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 11:59:30 -0700 Subject: [PATCH 07/10] refactor(gfql): static typing + cuDF-generic seeded fast-path helpers (#1755) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback on the seeded typed-hop fast path: - direction: str -> Direction (the canonical Literal from ast.py) in both helpers. - Make the helpers ENGINE-GENERIC (pandas + cuDF): _seeded_typed_hop_numpy_pandas -> _seeded_typed_hop_pandas_cudf and _seeded_typed_return_dst_pandas -> _seeded_typed_return_dst_pandas_cudf, rewritten to use only the shared pandas/cuDF DataFrame API (no numpy array drops), so the same body runs on both engines. Relax the native (_try_chain_fast_path) and cypher (_execute_seeded_typed_hop_fast_path) gates to Engine.PANDAS|CUDF. - Replace dynamic getattr with static attribute access in gfql_unified (projection.table/.columns, proj_cols[0].kind, n2._name, e1.direction, base_graph._node/_source/_destination) now that types are statically known. - Drop the dead _SEEDED_FASTPATH_GUARD marker (never set; vestigial from the earlier re-run design) — removes the last getattr. Pandas parity + independent-oracle tests still byte-identical (19 pass); mypy clean on both files (only pre-existing polars-lazy errors remain). cuDF parity verified separately on GPU. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/chain.py | 179 ++++++++---------- graphistry/compute/gfql_unified.py | 37 ++-- .../gfql/test_seeded_typed_hop_fastpath.py | 14 +- 3 files changed, 100 insertions(+), 130 deletions(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 31a2207a0f..f28579d87a 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -8,7 +8,7 @@ from graphistry.Engine import safe_merge from graphistry.util import setup_logger from graphistry.utils.json import JSONVal -from .ast import ASTObject, ASTNode, ASTEdge, from_json as ASTObject_from_json, serialize_binding_ops +from .ast import ASTObject, ASTNode, ASTEdge, Direction, from_json as ASTObject_from_json, serialize_binding_ops from .typing import DataFrameT, SeriesT from .util import generate_safe_column_name from graphistry.compute.validate.validate_schema import validate_chain_schema @@ -684,24 +684,26 @@ def _chain_otel_attrs( return attrs -def _seeded_typed_hop_numpy_pandas( +def _seeded_typed_hop_pandas_cudf( g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge, - src: str, dst: str, node: str, direction: str, + src: str, dst: str, node: str, direction: Direction, ) -> Optional[Plottable]: - """#1755 lever-3: pandas-only numpy fast path for a scalar-filtered seeded - typed 1-hop. Byte-identical to the engine-agnostic seeded branch for the - covered shape (all node/edge filters are plain scalars, directed), collapsing - it into a handful of numpy passes so a seeded lookup lands sub-ms. Returns - None to fall back for anything it does not cover (predicates, undirected, - missing columns) — the caller then runs the general engine-agnostic branch.""" - import numpy as np - import pandas as pd + """#1755 lever-3: engine-generic (pandas + cuDF) fast path for a scalar-filtered + seeded typed 1-hop. Byte-identical to the general seeded branch for the covered + shape (all node/edge filters are plain scalars, directed), collapsing it into a + few DataFrame filters so a seeded lookup lands sub-ms. Uses only the shared + pandas/cuDF DataFrame API (no numpy array drops) so the same body runs on both + engines. Returns None to fall back for anything it does not cover (predicates, + undirected, missing columns) — the caller then runs the general branch.""" if direction == "undirected": return None nodes_df, edges_df = g._nodes, g._edges + if nodes_df is None or edges_df is None: + return None + node_cols, edge_cols = set(nodes_df.columns), set(edges_df.columns) - def _scalars(fd, df): + def _scalars(fd: Optional[Dict[str, Any]], cols: set) -> Optional[Dict[str, Any]]: """Resolve a filter dict to plain scalar column==value pairs, or None to bail. Handles the cypher ``label__X: True`` form by resolving it to the ``type`` column (equality) — matching filter_by_dict.resolve_filter_column @@ -712,138 +714,113 @@ def _scalars(fd, df): for k, v in fd.items(): if not isinstance(v, (int, float, str, bool)): return None # predicate / non-scalar -> bail to the general path - if k in df.columns: + if k in cols: out[k] = v - elif isinstance(k, str) and k.startswith("label__") and v is True and "type" in df.columns: + elif isinstance(k, str) and k.startswith("label__") and v is True and "type" in cols: out["type"] = k[len("label__"):] else: return None # unknown / list-label column -> bail return out - n0f, n2f, ef = _scalars(n0.filter_dict, nodes_df), _scalars(n2.filter_dict, nodes_df), _scalars(e1.edge_match, edges_df) + n0f = _scalars(n0.filter_dict, node_cols) + n2f = _scalars(n2.filter_dict, node_cols) + ef = _scalars(e1.edge_match, edge_cols) if n0f is None or n2f is None or ef is None: return None from_col, to_col = (src, dst) if direction == "forward" else (dst, src) - nid = nodes_df[node].to_numpy() # from-side seed FIRST: reduce edges to the seed's out-edges before the - # (object) edge_match compare, so the type filter runs on the tiny frontier - # rather than all edges — this is what makes a seeded lookup sub-ms. + # edge_match compare, so the type filter runs on the tiny frontier rather than + # all edges — this is what makes a seeded lookup sub-ms. The id filter goes + # first (int, unique -> ~1 row in one pass) so any remaining object filters + # (label__X->type) run on that tiny survivor frame, not the whole node table. if n0f: - # Successive subsetting, id-column first: the (int, unique) id filter - # reduces 250k -> ~1 row in one cheap pass, so any remaining (object) - # filters like label__X->type run on that tiny set instead of scanning - # the whole node frame. `from_pos` tracks positions into nodes_df. - from_pos = np.arange(len(nodes_df)) + seed_nodes = nodes_df for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1): - col = (nid if k == node else nodes_df[k].to_numpy())[from_pos] - from_pos = from_pos[col == v] - if from_pos.size == 0: - break - from_ids = nid[from_pos] - from_vals = edges_df[from_col].to_numpy() - seed_mask = (from_vals == from_ids[0]) if len(from_ids) == 1 else np.isin(from_vals, from_ids) - edges = edges_df[seed_mask] + seed_nodes = seed_nodes[seed_nodes[k] == v] + edges = edges_df[edges_df[from_col].isin(seed_nodes[node])] else: edges = edges_df if ef: # typed edge (edge_match) — now on the reduced frontier - tmask = np.ones(len(edges), dtype=bool) for k, v in ef.items(): - tmask &= (edges[k].to_numpy() == v) - edges = edges[tmask] - - # Gather candidate endpoint nodes with a SINGLE O(N) node scan, then run the - # dest filter, dangling-edge drop and final-node selection on the small - # candidate/edge frames (no second full-node scan). - esrc, edst = edges[src].to_numpy(), edges[dst].to_numpy() - eto = edst if to_col == dst else esrc # reuse; no extra column fetch - ep = np.unique(np.concatenate([esrc, edst])) - cand = nodes_df[np.isin(nid, ep)].drop_duplicates(subset=[node]) - cand_ids = cand[node].to_numpy() - valid = cand_ids[~pd.isna(cand_ids)] # endpoints that are real nodes - if n2f: # destination-node filter (to-side) - dmask = np.ones(len(cand), dtype=bool) + edges = edges[edges[k] == v] + + # Gather candidate endpoint nodes (both endpoints of surviving edges), then run + # the dest filter, dangling-edge drop and final-node selection on the small + # candidate/edge frames. Selecting from nodes_df keeps only real nodes, so the + # endpoint-in-nodes check subsumes the old NaN-endpoint guard. + cand = nodes_df[ + nodes_df[node].isin(edges[src]) | nodes_df[node].isin(edges[dst]) + ].drop_duplicates(subset=[node]) + if n2f: # destination-node filter (to-side) + n2_cand = cand for k, v in n2f.items(): - dmask &= (cand[k].to_numpy() == v) - n2_ok = cand_ids[dmask] + n2_cand = n2_cand[n2_cand[k] == v] + n2_ok = n2_cand[node] else: - n2_ok = valid - keep = np.isin(esrc, valid) & np.isin(edst, valid) & np.isin(eto, n2_ok) - if not keep.all(): - edges = edges[keep] - fep = np.unique(np.concatenate([edges[src].to_numpy(), edges[dst].to_numpy()])) - cand = cand[np.isin(cand_ids, fep)] + n2_ok = cand[node] + to_vals = edges[dst] if to_col == dst else edges[src] + keep = edges[src].isin(cand[node]) & edges[dst].isin(cand[node]) & to_vals.isin(n2_ok) + edges = edges[keep] + cand = cand[cand[node].isin(edges[src]) | cand[node].isin(edges[dst])] return g.nodes(cand).edges(edges) -def _seeded_typed_return_dst_pandas( +def _seeded_typed_return_dst_pandas_cudf( g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge, - src: str, dst: str, node: str, direction: str, + src: str, dst: str, node: str, direction: Direction, ) -> Optional[Tuple[DataFrameT, DataFrameT]]: - """#1755 cypher RETURN-alias fast path: like _seeded_typed_hop_numpy_pandas but + """#1755 cypher RETURN-alias fast path: like _seeded_typed_hop_pandas_cudf but returns ONLY the destination (RETURN-alias) node rows + surviving edges — no seed-node gather, no Plottable round-trip — so the seeded cypher projection - lands sub-ms. Returns ``(dst_node_rows, edges)`` or None to fall back.""" - import numpy as np - import pandas as pd + lands sub-ms. Engine-generic (pandas + cuDF): only the shared DataFrame API, + no numpy array drops. Returns ``(dst_node_rows, edges)`` or None to fall back.""" if direction == "undirected": return None nodes_df, edges_df = g._nodes, g._edges + if nodes_df is None or edges_df is None: + return None + node_cols, edge_cols = set(nodes_df.columns), set(edges_df.columns) - def _sc(fd, df): + def _sc(fd: Optional[Dict[str, Any]], cols: set) -> Optional[Dict[str, Any]]: if not fd: return {} out: Dict[str, Any] = {} for k, v in fd.items(): if not isinstance(v, (int, float, str, bool)): return None - if k in df.columns: + if k in cols: out[k] = v - elif isinstance(k, str) and k.startswith("label__") and v is True and "type" in df.columns: + elif isinstance(k, str) and k.startswith("label__") and v is True and "type" in cols: out["type"] = k[len("label__"):] else: return None return out - n0f, n2f, ef = _sc(n0.filter_dict, nodes_df), _sc(n2.filter_dict, nodes_df), _sc(e1.edge_match, edges_df) + n0f = _sc(n0.filter_dict, node_cols) + n2f = _sc(n2.filter_dict, node_cols) + ef = _sc(e1.edge_match, edge_cols) if n0f is None or n2f is None or ef is None or not n0f: return None from_col, to_col = (src, dst) if direction == "forward" else (dst, src) - nid = nodes_df[node].to_numpy() - # id-first successive subset. The first (id) filter scans the full id array - # once; later filters index the already-tiny survivor rows via .iloc, so an - # object column like label__X->type is never fully materialized. - from_pos: Optional[Any] = None + # id-first seed reduction: filter by the id column first (int/unique -> ~1 row) + # so any remaining object filters (label__X->type) run on the tiny survivor + # frame, never materializing an object column over the whole node table. + seed_nodes = nodes_df for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1): - if from_pos is None: - base = nid if k == node else nodes_df[k].to_numpy() - from_pos = np.flatnonzero(base == v) - else: - sub = nid[from_pos] if k == node else nodes_df[k].iloc[from_pos].to_numpy() - from_pos = from_pos[sub == v] - if from_pos.size == 0: - break - from_ids = nid[from_pos] if from_pos is not None and from_pos.size else nid[:0] - if from_ids.size == 0: - return nodes_df.iloc[0:0], edges_df.iloc[0:0] - fv = edges_df[from_col].to_numpy() - seed_mask = (fv == from_ids[0]) if from_ids.size == 1 else np.isin(fv, from_ids) - edges = edges_df[seed_mask] + seed_nodes = seed_nodes[seed_nodes[k] == v] + edges = edges_df[edges_df[from_col].isin(seed_nodes[node])] if ef: - tmask = np.ones(len(edges), dtype=bool) for k, v in ef.items(): - tmask &= (edges[k].to_numpy() == v) - edges = edges[tmask] - to_vals = edges[to_col].to_numpy() - dst_ids = np.unique(to_vals[~pd.isna(to_vals)]) - dstn = nodes_df[np.isin(nid, dst_ids)] + edges = edges[edges[k] == v] + # destination nodes = real nodes that are edge to-endpoints, then the dest + # filter, dangling-edge drop and dedup on the small dst/edge frames. + dstn = nodes_df[nodes_df[node].isin(edges[to_col])] if n2f: - dm = np.ones(len(dstn), dtype=bool) for k, v in n2f.items(): - dm &= (dstn[k].to_numpy() == v) - dstn = dstn[dm] - edges = edges[np.isin(edges[to_col].to_numpy(), dstn[node].to_numpy())] - dstn = dstn[np.isin(dstn[node].to_numpy(), edges[to_col].to_numpy())].drop_duplicates(subset=[node]) + dstn = dstn[dstn[k] == v] + edges = edges[edges[to_col].isin(dstn[node])] + dstn = dstn[dstn[node].isin(edges[to_col])].drop_duplicates(subset=[node]) return dstn, edges @@ -933,13 +910,13 @@ def _materialize_fast_path_graph() -> Plottable: # tiny frontier, not all edges. The from-side ids come from the node table # (so that endpoint is validated); the node gather below validates the to # side and drops any edge dangling off the node table. - # pandas: a scalar-filtered seeded typed hop collapses to a few numpy - # passes (sub-ms); falls back to the engine-agnostic branch below for - # predicates / undirected / missing columns and for cuDF. - if engine_concrete == Engine.PANDAS: - _np_res = _seeded_typed_hop_numpy_pandas(g, n0, n2, e1, src, dst, node, direction) - if _np_res is not None: - return _np_res + # pandas + cuDF: a scalar-filtered seeded typed hop collapses to a few + # DataFrame filters (sub-ms); falls back to the general branch below for + # predicates / undirected / missing columns (and non-pandas/cuDF engines). + if engine_concrete in (Engine.PANDAS, Engine.CUDF): + _fast_res = _seeded_typed_hop_pandas_cudf(g, n0, n2, e1, src, dst, node, direction) + if _fast_res is not None: + return _fast_res from_col, to_col = (src, dst) if direction == "forward" else (dst, src) edges = g._edges if n0.filter_dict: diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index 80725fefec..3fd18c7380 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -976,8 +976,6 @@ def _run_logical_pass_pipeline(logical_plan: LogicalPlan, ctx: PlanContext) -> L return PassManager(DEFAULT_LOGICAL_PASSES, DEFAULT_TIER2_PASSES).run(logical_plan, ctx).plan -_SEEDED_FASTPATH_GUARD = "_gfql_seeded_typed_hop_fastpath_done" - def _execute_seeded_typed_hop_fast_path( base_graph: Plottable, @@ -990,23 +988,21 @@ def _execute_seeded_typed_hop_fast_path( start_nodes: Optional[DataFrameT] = None, ) -> Optional[Plottable]: """#1755 cypher-surface fast path: a seeded typed 1-hop with a whole-row node - RETURN — ``MATCH (m {id})-[:T]->(p) RETURN p`` — numpy-reduces the graph to the - seed's 1-hop neighborhood, then re-runs the SAME cypher pipeline on that tiny - subgraph (guaranteed byte-identical to the full path; sub-ms because the graph - is tiny). Returns None to fall through for anything outside this exact shape.""" - from graphistry.compute.chain import _seeded_typed_hop_numpy_pandas - if getattr(base_graph, _SEEDED_FASTPATH_GUARD, False): - return None # recursion guard: the re-run below must take the full path + RETURN — ``MATCH (m {id})-[:T]->(p) RETURN p`` — reduces the graph to the seed's + 1-hop neighborhood with a few DataFrame filters (pandas + cuDF), then applies the + RETURN projection to just the destination rows (guaranteed byte-identical to the + full path; sub-ms because the frame is tiny). Returns None to fall through for + anything outside this exact shape.""" requested_engine = resolve_engine(cast(Any, engine), base_graph) - if requested_engine != Engine.PANDAS: + if requested_engine not in (Engine.PANDAS, Engine.CUDF): return None projection = compiled_query.result_projection - if projection is None or getattr(projection, "table", None) != "nodes": + if projection is None or projection.table != "nodes": return None # Only a single whole-row node alias (RETURN p). Multi-alias returns (RETURN # m, p) combine aliases into one row per match — different shape — so bail. - proj_cols = getattr(projection, "columns", None) - if not proj_cols or len(proj_cols) != 1 or getattr(proj_cols[0], "kind", None) != "whole_row": + proj_cols = projection.columns + if len(proj_cols) != 1 or proj_cols[0].kind != "whole_row": return None if compiled_query.execution_extras is not None and ( compiled_query.execution_extras.connected_match_join is not None @@ -1029,24 +1025,21 @@ def _execute_seeded_typed_hop_fast_path( return None if call.function != "rows": return None - node = getattr(base_graph, "_node", None) - src = getattr(base_graph, "_source", None) - dst = getattr(base_graph, "_destination", None) + node, src, dst = base_graph._node, base_graph._source, base_graph._destination if node is None or src is None or dst is None: return None # RETURN alias must be the DESTINATION node (n2) and the seed must sit on the # source node (n0) — the forward seeded shape MATCH (m {id})-[:T]->(p) RETURN p. # Other alias/seed placements (e.g. reverse patterns where the seed is on the # RETURN node) fall back to the full path. - if getattr(n2, "_name", None) != projection.alias: + if n2._name != projection.alias: return None if not (n0.filter_dict and any(not str(k).startswith("label__") for k in n0.filter_dict)): return None # n0 must carry a selective (non-label) seed - direction = getattr(e1, "direction", "forward") - node_s, src_s, dst_s = str(node), str(src), str(dst) - from graphistry.compute.chain import _seeded_typed_return_dst_pandas - dst_res = _seeded_typed_return_dst_pandas( - base_graph, n0, n2, e1, src_s, dst_s, node_s, direction) + direction = e1.direction + from graphistry.compute.chain import _seeded_typed_return_dst_pandas_cudf + dst_res = _seeded_typed_return_dst_pandas_cudf( + base_graph, n0, n2, e1, src, dst, node, direction) if dst_res is None: return None p_rows, _edges = dst_res diff --git a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py index 98d0dbcc26..681b6ec497 100644 --- a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -1,7 +1,7 @@ """Parity + gating tests for the #1755 seeded typed-hop fast path. Two layers close the seeded-Cypher abstraction tax on pandas: - * native — `_try_chain_fast_path` / `_seeded_typed_hop_numpy_pandas` accelerate + * native — `_try_chain_fast_path` / `_seeded_typed_hop_pandas_cudf` accelerate a seeded typed 1-hop chain [n({id}), e(edge_match), n({type})]; * cypher — `_execute_seeded_typed_hop_fast_path` accelerates the lowered MATCH (m {id})-[:T]->(p) RETURN p string surface. @@ -92,7 +92,7 @@ def test_fast_path_engages_on_typed_hop(self, monkeypatch): g, P = _graph() seed = P + 42 hits = {"n": 0} - real = chain_mod._seeded_typed_hop_numpy_pandas + real = chain_mod._seeded_typed_hop_pandas_cudf def spy(*a, **k): r = real(*a, **k) @@ -100,7 +100,7 @@ def spy(*a, **k): hits["n"] += 1 return r - monkeypatch.setattr(chain_mod, "_seeded_typed_hop_numpy_pandas", spy) + monkeypatch.setattr(chain_mod, "_seeded_typed_hop_pandas_cudf", spy) g.gfql([n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], engine="pandas") assert hits["n"] >= 1 @@ -127,11 +127,11 @@ def test_numpy_helper_declines_predicate_and_undirected(self): g, P = _graph() node, src, dst = "id", "src", "dst" # predicate (non-scalar) edge filter -> decline - assert chain_mod._seeded_typed_hop_numpy_pandas( + assert chain_mod._seeded_typed_hop_pandas_cudf( g.materialize_nodes(), n({"id": P + 1}), n(), e_forward(edge_match={"type": gt(0)}), src, dst, node, "forward") is None # undirected -> decline - assert chain_mod._seeded_typed_hop_numpy_pandas( + assert chain_mod._seeded_typed_hop_pandas_cudf( g.materialize_nodes(), n({"id": P + 1}), n(), e_forward(), src, dst, node, "undirected") is None @@ -148,7 +148,7 @@ def test_native_declines_and_stays_correct(self, ops_name, reason, monkeypatch): "node_predicate": [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"age": gt(0)})], }[ops_name] hits = {"n": 0} - real = chain_mod._seeded_typed_hop_numpy_pandas + real = chain_mod._seeded_typed_hop_pandas_cudf def spy(*a, **k): r = real(*a, **k) @@ -156,7 +156,7 @@ def spy(*a, **k): hits["n"] += 1 return r - monkeypatch.setattr(chain_mod, "_seeded_typed_hop_numpy_pandas", spy) + monkeypatch.setattr(chain_mod, "_seeded_typed_hop_pandas_cudf", spy) fast = g.gfql(ops, engine="pandas") monkeypatch.undo() chain_mod._try_chain_fast_path, saved = (lambda *a, **k: None), chain_mod._try_chain_fast_path From 311cb5e8bf8b8f22b6bc87d30e0492fbdbbf8cae Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 12:01:37 -0700 Subject: [PATCH 08/10] =?UTF-8?q?test(gfql):=20differential=20fast-vs-full?= =?UTF-8?q?=20sweep=20across=20shapes=20=C3=97=20engines=20(#1755)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers the coverage question: broaden ENGAGED-shape verification beyond the few hardcoded cases. Adds a 7-shape × {pandas, cuDF} matrix asserting fast-on == fast-off byte-identical (native typed/untyped/reverse/type-src + cypher labelled/unlabelled/no-match), plus an independent-oracle check per engine. cuDF cases importorskip locally, run on GPU/CI. Complements the ~1114 implicit decline-path exercises the cypher conformance suite already drives through the dispatch (real queries that fall through + still return correct values). Also makes _canon_nodes engine-aware (cuDF/polars -> pandas via to_pandas). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- .../gfql/test_seeded_typed_hop_fastpath.py | 84 ++++++++++++++++++- 1 file changed, 83 insertions(+), 1 deletion(-) diff --git a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py index 681b6ec497..6a761f852d 100644 --- a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -40,12 +40,41 @@ def _graph(n_persons=1500, n_messages=6000, seed=0): def _canon_nodes(res): - df = pd.DataFrame(res._nodes) + nodes = res._nodes + df = nodes.to_pandas() if hasattr(nodes, "to_pandas") else pd.DataFrame(nodes) cols = sorted(str(c) for c in df.columns) df.columns = [str(c) for c in df.columns] return df.sort_values(cols).reset_index(drop=True)[cols] if cols else df +def _engine_graph(engine): + """Build the probe graph in the requested engine (cuDF skips if unavailable).""" + g, P = _graph() + if engine == "pandas": + return g, P + if engine == "cudf": + cudf = pytest.importorskip("cudf") + return graphistry.nodes( + cudf.from_pandas(pd.DataFrame(g._nodes)), "id" + ).edges(cudf.from_pandas(pd.DataFrame(g._edges)), "src", "dst"), P + raise ValueError(engine) + + +def _run_diff(g, engine, query, fast): + """Run `query` with BOTH seeded fast paths (native + cypher) either live or + forced-off, so a single differential asserts fast==full for any shape/engine.""" + real_native = chain_mod._try_chain_fast_path + real_cyp = gfql_unified._execute_seeded_typed_hop_fast_path + try: + if not fast: + chain_mod._try_chain_fast_path = lambda *a, **k: None + gfql_unified._execute_seeded_typed_hop_fast_path = lambda *a, **k: None + return g.gfql(query, engine=engine) + finally: + chain_mod._try_chain_fast_path = real_native + gfql_unified._execute_seeded_typed_hop_fast_path = real_cyp + + def _oracle_creators(g, seed): """INDEPENDENT ground truth (NOT via the query engine): the Person ids that are HAS_CREATOR destinations of the seeded Message. Computed directly from @@ -262,3 +291,56 @@ def spy(*a, **k): full = self._run(g, cy, force_full=True) assert hits["n"] == 0, f"fast path must decline {reason}" pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + +# --------------------------------------------------------------------------- +# Differential fast-vs-full sweep across shapes × engines +# --------------------------------------------------------------------------- + +class TestSeededFastPathDifferentialSweep: + """Broad fast-vs-full byte-parity across a matrix of seeded shapes × engines. + + Answers the coverage question: is every ENGAGED shape verified through BOTH + the fast path and the full path, on every supported engine (pandas + cuDF)? + This complements the ~1114 implicit DECLINE-path exercises the cypher + conformance suite already drives through `_execute_seeded_typed_hop_fast_path` + (real queries that fall through and still return correct values). Here we pin + the ENGAGED shapes byte-identical fast-vs-full so the specialization can never + silently diverge from the general path on any covered engine.""" + + SHAPES = { + "native_typed": lambda s: [n({"id": s}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], + "native_untyped": lambda s: [n({"id": s}), e_forward(), n()], + "native_reverse": lambda s: [n({"type": "Person"}), e_reverse(edge_match={"type": "HAS_CREATOR"}), n({"id": s})], + "native_type_src": lambda s: [n({"type": "Message"}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], + "cypher_p_labelled": lambda s: f"MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN p", + "cypher_p_unlabelled": lambda s: f"MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p) RETURN p", + "cypher_no_match": lambda s: "MATCH (m:Message {id: 9999999})-[:HAS_CREATOR]->(p:Person) RETURN p", + } + + @pytest.mark.parametrize("engine", ["pandas", "cudf"]) + @pytest.mark.parametrize("shape", list(SHAPES)) + def test_fast_vs_full_parity(self, engine, shape): + g, P = _engine_graph(engine) + query = self.SHAPES[shape](P + 42) + fast = _run_diff(g, engine, query, fast=True) + full = _run_diff(g, engine, query, fast=False) + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + @pytest.mark.parametrize("engine", ["pandas", "cudf"]) + def test_engaged_shapes_match_independent_oracle(self, engine): + """The two headline engaged shapes must equal the hand-computed creator set + (native = {seed} ∪ creators; cypher RETURN p = creators) on each engine — + not merely fast==full, since the full path could share a bug.""" + g, P = _engine_graph(engine) + pd_g, _ = _graph() # pandas twin for the independent oracle + seed = P + 42 + oracle = _oracle_creators(pd_g, seed) + assert len(oracle) >= 1 + native = _run_diff(g, engine, self.SHAPES["native_typed"](seed), fast=True) + got_native = sorted(_canon_nodes(native)["id"].tolist()) + assert got_native == sorted(set(oracle) | {seed}) + cyp = _run_diff(g, engine, self.SHAPES["cypher_p_labelled"](seed), fast=True) + cdf = _canon_nodes(cyp) + idc = "p.id" if "p.id" in cdf.columns else "id" + assert sorted(cdf[idc].tolist()) == oracle From dec4e4ad68757f191b190ab261dbf460ae4d918d Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 13:48:29 -0700 Subject: [PATCH 09/10] docs(changelog): seeded typed-hop fast path entry (#1755) 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 fe8191113e..d9001fb396 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Performance +- **GFQL seeded typed-hop fast path (#1755)**: a seeded typed 1-hop — native chain `[n({id}), e_forward(), n({type})]` or Cypher `MATCH (m {id})-[:T]->(p) RETURN p` — previously paid the full two-pass chain machinery (~20-40ms: whole-frame combines, full-column type filters, rows-pivot projection). A new seed-first fast path recognizes exactly this shape and reduces the graph to the seed's 1-hop neighborhood before any of that work: native chain 32.6→0.93ms (35×), Cypher RETURN 39.2→1.9ms (20×) on pandas, with cuDF covered via the shared DataFrame API (30.8→4.7ms). Byte-identical to the full path by construction — the helpers return `None` and fall through for anything outside the exact shape (multi-hop, variable-length, undirected, predicate filters, reverse patterns, missing bindings all decline). Verified with an independent oracle (results checked against the creator set hand-computed from the raw frames, not merely fast-vs-slow agreement) plus a differential fast-vs-full sweep across shapes × engines in `test_seeded_typed_hop_fastpath.py`. - **GFQL seeded-chain lean combine (#1755)**: The two-pass chain executor (`_chain_impl` backward pass + `combine_steps`) reconciled wavefront boundaries with full-node-frame `safe_merge`/`pandas.merge` calls even when the seeded result was a single row, so a seeded 1-hop paid a whole-frame join. When the small side is at least `4x` smaller than the full frame (and is unique on the id with no extra columns), the byte-identical result is now obtained by an `isin` membership filter instead of the join machinery. Gated narrowly on cardinality/uniqueness/column-set and pandas-only (cuDF's GPU hash-join is already sub-ms; polars takes its own engine path); `GFQL_LEAN_COMBINE=0` restores the legacy merges. Parity is byte-identical (lean-on vs lean-off) across seeded node lookup, seeded 1-hop expand, and the native typed chain, including a null-key equivalence guard. ### Documentation From efeaccc0d3b6b099603711fbddf286f6aa985361 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 14:23:44 -0700 Subject: [PATCH 10/10] fix(gfql): seeded fast-path side-channel decline gates + exact filter semantics (#1755) Review-skill wave findings, all empirically confirmed then fixed: - decline under policy (prechain/postchain/postload hooks were silently skipped) - decline on same-path WHERE (cross-alias predicates were silently dropped) - decline on OPTIONAL MATCH empty_result_row (null row became empty frame) - decline on carried reentry seeds (start_nodes ignored -> seed silently widened) - label__X resolution now mirrors resolve_filter_column exactly (list-'labels' column takes precedence; edge frames decline) via shared _seeded_scalar_filters (also dedups the twin _scalars/_sc closures) - membership sets dropna()'d: NaN ids/endpoints never link (pandas .isin matches NaN<->NaN; the full pipeline's joins never join null keys) - 'byte-identical' claims corrected to value-identical (row order/index may differ) - 6 regression tests pinning each gate (TestFastPathSideChannelGates) Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- CHANGELOG.md | 2 +- graphistry/compute/chain.py | 104 ++++++++--------- graphistry/compute/gfql_unified.py | 27 ++++- .../gfql/test_seeded_typed_hop_fastpath.py | 108 +++++++++++++++++- 4 files changed, 180 insertions(+), 61 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d9001fb396..08bcc4a7a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Performance -- **GFQL seeded typed-hop fast path (#1755)**: a seeded typed 1-hop — native chain `[n({id}), e_forward(), n({type})]` or Cypher `MATCH (m {id})-[:T]->(p) RETURN p` — previously paid the full two-pass chain machinery (~20-40ms: whole-frame combines, full-column type filters, rows-pivot projection). A new seed-first fast path recognizes exactly this shape and reduces the graph to the seed's 1-hop neighborhood before any of that work: native chain 32.6→0.93ms (35×), Cypher RETURN 39.2→1.9ms (20×) on pandas, with cuDF covered via the shared DataFrame API (30.8→4.7ms). Byte-identical to the full path by construction — the helpers return `None` and fall through for anything outside the exact shape (multi-hop, variable-length, undirected, predicate filters, reverse patterns, missing bindings all decline). Verified with an independent oracle (results checked against the creator set hand-computed from the raw frames, not merely fast-vs-slow agreement) plus a differential fast-vs-full sweep across shapes × engines in `test_seeded_typed_hop_fastpath.py`. +- **GFQL seeded typed-hop fast path (#1755)**: a seeded typed 1-hop — native chain `[n({id}), e_forward(), n({type})]` or Cypher `MATCH (m {id})-[:T]->(p) RETURN p` — previously paid the full two-pass chain machinery (~20-40ms: whole-frame combines, full-column type filters, rows-pivot projection). A new seed-first fast path recognizes exactly this shape and reduces the graph to the seed's 1-hop neighborhood before any of that work: native chain 32.6→0.93ms (35×), Cypher RETURN 39.2→1.9ms (20×) on pandas, with cuDF covered via the shared DataFrame API (30.8→4.7ms). Value-identical to the full path by construction (same rows/columns/dtypes; row order and index may differ) — the helpers return `None` and fall through for anything outside the exact shape or carrying full-path side-channels (multi-hop, variable-length, undirected, predicate filters, reverse patterns, missing bindings, list-`labels` columns, policy hooks, same-path WHERE, OPTIONAL MATCH null rows, and WITH..MATCH carried seeds all decline). Null ids/endpoints never link (membership sets are null-dropped, matching the full pipeline's joins). Verified with an independent oracle (results checked against the creator set hand-computed from the raw frames, not merely fast-vs-slow agreement) plus a differential fast-vs-full sweep across shapes × engines in `test_seeded_typed_hop_fastpath.py`. - **GFQL seeded-chain lean combine (#1755)**: The two-pass chain executor (`_chain_impl` backward pass + `combine_steps`) reconciled wavefront boundaries with full-node-frame `safe_merge`/`pandas.merge` calls even when the seeded result was a single row, so a seeded 1-hop paid a whole-frame join. When the small side is at least `4x` smaller than the full frame (and is unique on the id with no extra columns), the byte-identical result is now obtained by an `isin` membership filter instead of the join machinery. Gated narrowly on cardinality/uniqueness/column-set and pandas-only (cuDF's GPU hash-join is already sub-ms; polars takes its own engine path); `GFQL_LEAN_COMBINE=0` restores the legacy merges. Parity is byte-identical (lean-on vs lean-off) across seeded node lookup, seeded 1-hop expand, and the native typed chain, including a null-key equivalence guard. ### Documentation diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index f28579d87a..ec26f4b4cf 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -684,13 +684,41 @@ def _chain_otel_attrs( return attrs +def _seeded_scalar_filters(fd: Optional[Dict[str, Any]], df: DataFrameT) -> Optional[Dict[str, Any]]: + """Resolve a filter dict to plain scalar column==value pairs, or None to bail + to the general path. Mirrors filter_by_dict.resolve_filter_column exactly for + the shapes it accepts: the cypher ``label__X: True`` form maps to ``type`` + equality ONLY when no list-valued ``labels`` column exists (labels-containment + is not scalar equality) and the frame is not edge-shaped — same precedence as + the live resolver. Anything else (predicates, non-scalar values, absent + columns) bails, so the full path keeps its exact semantics incl. E301.""" + from graphistry.compute.filter_by_dict import _looks_like_edge_dataframe + if not fd: + return {} + cols = set(df.columns) + out: Dict[str, Any] = {} + for k, v in fd.items(): + if not isinstance(v, (int, float, str, bool)): + return None # predicate / non-scalar -> bail to the general path + if k in cols: + out[k] = v + elif (isinstance(k, str) and k.startswith("label__") and v is True + and "labels" not in cols and "type" in cols + and not _looks_like_edge_dataframe(df)): + out["type"] = k[len("label__"):] + else: + return None # labels-list / unknown column -> bail + return out + + def _seeded_typed_hop_pandas_cudf( g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge, src: str, dst: str, node: str, direction: Direction, ) -> Optional[Plottable]: """#1755 lever-3: engine-generic (pandas + cuDF) fast path for a scalar-filtered - seeded typed 1-hop. Byte-identical to the general seeded branch for the covered - shape (all node/edge filters are plain scalars, directed), collapsing it into a + seeded typed 1-hop. Value-identical to the general seeded branch for the covered + shape (all node/edge filters are plain scalars, directed) — same rows, columns, + and dtypes; row order and RangeIndex may differ — collapsing it into a few DataFrame filters so a seeded lookup lands sub-ms. Uses only the shared pandas/cuDF DataFrame API (no numpy array drops) so the same body runs on both engines. Returns None to fall back for anything it does not cover (predicates, @@ -701,30 +729,9 @@ def _seeded_typed_hop_pandas_cudf( nodes_df, edges_df = g._nodes, g._edges if nodes_df is None or edges_df is None: return None - node_cols, edge_cols = set(nodes_df.columns), set(edges_df.columns) - - def _scalars(fd: Optional[Dict[str, Any]], cols: set) -> Optional[Dict[str, Any]]: - """Resolve a filter dict to plain scalar column==value pairs, or None to - bail. Handles the cypher ``label__X: True`` form by resolving it to the - ``type`` column (equality) — matching filter_by_dict.resolve_filter_column - — but bails on the list-valued ``labels`` form to stay equality-safe.""" - if not fd: - return {} - out: Dict[str, Any] = {} - for k, v in fd.items(): - if not isinstance(v, (int, float, str, bool)): - return None # predicate / non-scalar -> bail to the general path - if k in cols: - out[k] = v - elif isinstance(k, str) and k.startswith("label__") and v is True and "type" in cols: - out["type"] = k[len("label__"):] - else: - return None # unknown / list-label column -> bail - return out - - n0f = _scalars(n0.filter_dict, node_cols) - n2f = _scalars(n2.filter_dict, node_cols) - ef = _scalars(e1.edge_match, edge_cols) + n0f = _seeded_scalar_filters(n0.filter_dict, nodes_df) + n2f = _seeded_scalar_filters(n2.filter_dict, nodes_df) + ef = _seeded_scalar_filters(e1.edge_match, edges_df) if n0f is None or n2f is None or ef is None: return None from_col, to_col = (src, dst) if direction == "forward" else (dst, src) @@ -738,7 +745,7 @@ def _scalars(fd: Optional[Dict[str, Any]], cols: set) -> Optional[Dict[str, Any] seed_nodes = nodes_df for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1): seed_nodes = seed_nodes[seed_nodes[k] == v] - edges = edges_df[edges_df[from_col].isin(seed_nodes[node])] + edges = edges_df[edges_df[from_col].isin(seed_nodes[node].dropna())] else: edges = edges_df if ef: # typed edge (edge_match) — now on the reduced frontier @@ -748,9 +755,11 @@ def _scalars(fd: Optional[Dict[str, Any]], cols: set) -> Optional[Dict[str, Any] # Gather candidate endpoint nodes (both endpoints of surviving edges), then run # the dest filter, dangling-edge drop and final-node selection on the small # candidate/edge frames. Selecting from nodes_df keeps only real nodes, so the - # endpoint-in-nodes check subsumes the old NaN-endpoint guard. + # endpoint-in-nodes check subsumes the old NaN-endpoint guard. Membership sets + # are dropna()'d: pandas .isin matches NaN<->NaN, but the general branch's BFS + # joins never join on null keys, so a null id/endpoint must not link. cand = nodes_df[ - nodes_df[node].isin(edges[src]) | nodes_df[node].isin(edges[dst]) + nodes_df[node].isin(edges[src].dropna()) | nodes_df[node].isin(edges[dst].dropna()) ].drop_duplicates(subset=[node]) if n2f: # destination-node filter (to-side) n2_cand = cand @@ -759,8 +768,8 @@ def _scalars(fd: Optional[Dict[str, Any]], cols: set) -> Optional[Dict[str, Any] n2_ok = n2_cand[node] else: n2_ok = cand[node] - to_vals = edges[dst] if to_col == dst else edges[src] - keep = edges[src].isin(cand[node]) & edges[dst].isin(cand[node]) & to_vals.isin(n2_ok) + to_vals = edges[to_col] + keep = edges[src].isin(cand[node].dropna()) & edges[dst].isin(cand[node].dropna()) & to_vals.isin(n2_ok.dropna()) edges = edges[keep] cand = cand[cand[node].isin(edges[src]) | cand[node].isin(edges[dst])] return g.nodes(cand).edges(edges) @@ -780,47 +789,32 @@ def _seeded_typed_return_dst_pandas_cudf( nodes_df, edges_df = g._nodes, g._edges if nodes_df is None or edges_df is None: return None - node_cols, edge_cols = set(nodes_df.columns), set(edges_df.columns) - - def _sc(fd: Optional[Dict[str, Any]], cols: set) -> Optional[Dict[str, Any]]: - if not fd: - return {} - out: Dict[str, Any] = {} - for k, v in fd.items(): - if not isinstance(v, (int, float, str, bool)): - return None - if k in cols: - out[k] = v - elif isinstance(k, str) and k.startswith("label__") and v is True and "type" in cols: - out["type"] = k[len("label__"):] - else: - return None - return out - - n0f = _sc(n0.filter_dict, node_cols) - n2f = _sc(n2.filter_dict, node_cols) - ef = _sc(e1.edge_match, edge_cols) + n0f = _seeded_scalar_filters(n0.filter_dict, nodes_df) + n2f = _seeded_scalar_filters(n2.filter_dict, nodes_df) + ef = _seeded_scalar_filters(e1.edge_match, edges_df) if n0f is None or n2f is None or ef is None or not n0f: return None from_col, to_col = (src, dst) if direction == "forward" else (dst, src) # id-first seed reduction: filter by the id column first (int/unique -> ~1 row) # so any remaining object filters (label__X->type) run on the tiny survivor # frame, never materializing an object column over the whole node table. + # Membership sets are dropna()'d: pandas .isin matches NaN<->NaN, but the full + # pipeline's joins never join on null keys, so a null id/endpoint must not link. seed_nodes = nodes_df for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1): seed_nodes = seed_nodes[seed_nodes[k] == v] - edges = edges_df[edges_df[from_col].isin(seed_nodes[node])] + edges = edges_df[edges_df[from_col].isin(seed_nodes[node].dropna())] if ef: for k, v in ef.items(): edges = edges[edges[k] == v] # destination nodes = real nodes that are edge to-endpoints, then the dest # filter, dangling-edge drop and dedup on the small dst/edge frames. - dstn = nodes_df[nodes_df[node].isin(edges[to_col])] + dstn = nodes_df[nodes_df[node].isin(edges[to_col].dropna())] if n2f: for k, v in n2f.items(): dstn = dstn[dstn[k] == v] - edges = edges[edges[to_col].isin(dstn[node])] - dstn = dstn[dstn[node].isin(edges[to_col])].drop_duplicates(subset=[node]) + edges = edges[edges[to_col].isin(dstn[node].dropna())] + dstn = dstn[dstn[node].isin(edges[to_col].dropna())].drop_duplicates(subset=[node]) return dstn, edges diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index 3fd18c7380..c4c17f34cd 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -990,9 +990,30 @@ def _execute_seeded_typed_hop_fast_path( """#1755 cypher-surface fast path: a seeded typed 1-hop with a whole-row node RETURN — ``MATCH (m {id})-[:T]->(p) RETURN p`` — reduces the graph to the seed's 1-hop neighborhood with a few DataFrame filters (pandas + cuDF), then applies the - RETURN projection to just the destination rows (guaranteed byte-identical to the - full path; sub-ms because the frame is tiny). Returns None to fall through for - anything outside this exact shape.""" + RETURN projection to just the destination rows (value-identical to the full + path: same rows/columns/dtypes; row order and RangeIndex may differ; sub-ms + because the frame is tiny). Returns None to fall through for anything outside + this exact shape or carrying side-channels (policy, same-path WHERE, OPTIONAL + null-row, carried reentry seeds).""" + if start_nodes is not None: + # A carried seed set (WITH..MATCH reentry) restricts n0 to those rows; the fast + # path derives its seed from n0.filter_dict alone, so engaging here would + # silently widen the seed back to the whole graph. Decline. + return None + if policy: + # The full path fires prechain/postchain/postload policy hooks; the lean + # projection below never enters chain(), so engaging would silently skip + # them. Policy-gated queries always take the full path (native twin gates + # identically in chain._try_chain_fast_path). + return None + if compiled_query.chain.where: + # Same-path WHERE entries (e.g. WHERE p.id < m.id) are evaluated by the + # full pipeline, not by the seed-first reduction — engaging would drop them. + return None + if compiled_query.empty_result_row is not None: + # OPTIONAL MATCH null-row semantics: on no match the full path emits the + # openCypher null row; the lean projection would return an empty frame. + return None requested_engine = resolve_engine(cast(Any, engine), base_graph) if requested_engine not in (Engine.PANDAS, Engine.CUDF): return None diff --git a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py index 6a761f852d..8ff8b24174 100644 --- a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -6,9 +6,12 @@ * cypher — `_execute_seeded_typed_hop_fast_path` accelerates the lowered MATCH (m {id})-[:T]->(p) RETURN p string surface. -Both are byte-identical to the full path by construction; these tests pin that +Both are value-identical to the full path by construction (same rows/columns; +row order and index may differ — comparisons canonicalize); these tests pin that (fast-on vs fast-off, differential) and that the fast path actually ENGAGES for -the accelerated shapes and DECLINES (falls through) for everything else. +the accelerated shapes and DECLINES (falls through) for everything else, +including full-path side-channels (policy hooks, same-path WHERE, OPTIONAL +null rows, WITH..MATCH carried seeds, list-`labels` columns, null ids). """ import numpy as np import pandas as pd @@ -344,3 +347,104 @@ def test_engaged_shapes_match_independent_oracle(self, engine): cdf = _canon_nodes(cyp) idc = "p.id" if "p.id" in cdf.columns else "id" assert sorted(cdf[idc].tolist()) == oracle + + +# --------------------------------------------------------------------------- +# side-channel decline gates + semantics parity (review-skill blockers, wave 2026-07-21) +# --------------------------------------------------------------------------- + +class TestFastPathSideChannelGates: + """Each test pins a shape where ENGAGING the fast path was empirically proven + to diverge from the full path (probes in plans/review-pr-1759/): the fix is + to decline (or match semantics exactly), and these lock that in.""" + + def _canon_edges(self, res): + edges = res._edges + df = edges.to_pandas() if hasattr(edges, "to_pandas") else pd.DataFrame(edges) + cols = sorted(str(c) for c in df.columns) + df.columns = [str(c) for c in df.columns] + return df.sort_values(cols).reset_index(drop=True)[cols] if cols else df + + def test_labels_column_precedence(self): + """label__X must follow resolve_filter_column: `labels` (list containment) + wins over `type` — the fast path's type-equality shortcut must decline.""" + ndf = pd.DataFrame({ + "id": [0, 1], + "type": ["Person", "Person"], + "labels": [["Admin"], ["Person"]], # containment says only id=1 is a Person + }) + edf = pd.DataFrame({"src": [0], "dst": [1], "type": ["KNOWS"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + q = "MATCH (m {id: 0})-[{type:'KNOWS'}]->(p:Person) RETURN p" + fast = _canon_nodes(_run_diff(g, "pandas", q, fast=True)) + full = _canon_nodes(_run_diff(g, "pandas", q, fast=False)) + pd.testing.assert_frame_equal(fast, full) + + def test_null_ids_never_link(self): + """NaN node id + NaN edge endpoint: pandas .isin matches NaN<->NaN but the + full pipeline's joins never link null keys — nodes AND edges must agree.""" + ndf = pd.DataFrame({"id": [0.0, 1.0, np.nan], + "type": ["Person", "Message", "Message"]}) + edf = pd.DataFrame({"src": [1.0, np.nan], "dst": [0.0, 0.0], + "type": ["HAS_CREATOR", "HAS_CREATOR"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + ops = [n({"type": "Message"}), e_forward(edge_match={"type": "HAS_CREATOR"}), + n({"type": "Person"})] + fast_r = _run_diff(g, "pandas", ops, fast=True) + full_r = _run_diff(g, "pandas", ops, fast=False) + pd.testing.assert_frame_equal(_canon_nodes(fast_r), _canon_nodes(full_r)) + pd.testing.assert_frame_equal(self._canon_edges(fast_r), self._canon_edges(full_r)) + + def test_policy_hooks_not_skipped(self): + """A policy dict must force the full path so pre/post hooks fire.""" + g, P = _graph() + seed = P + 7 + fired = [] + + def hook(ctx): + fired.append(ctx.get("phase", "?")) + return None + q = f"MATCH (m {{id: {seed}}})-[{{type:'HAS_CREATOR'}}]->(p {{type:'Person'}}) RETURN p" + policy = {"prechain": hook, "postchain": hook} + r_pol = g.gfql(q, engine="pandas", policy=policy) + assert fired, "policy hooks must fire (fast path must decline under policy)" + r_nopol = g.gfql(q, engine="pandas") + pd.testing.assert_frame_equal(_canon_nodes(r_pol), _canon_nodes(r_nopol)) + + def test_same_path_where_not_dropped(self): + """WHERE p.id < m.id (same-path cross-alias) must not be silently dropped.""" + ndf = pd.DataFrame({"id": [0, 1, 2], "type": ["Message", "Person", "Person"], + "age": [np.nan, 31.0, 32.0]}) + edf = pd.DataFrame({"src": [0, 0], "dst": [1, 2], "type": ["T", "T"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + q = "MATCH (m {id: 0})-[{type:'T'}]->(p {type:'Person'}) WHERE p.id > m.id + 1 RETURN p" + fast = _canon_nodes(_run_diff(g, "pandas", q, fast=True)) + full = _canon_nodes(_run_diff(g, "pandas", q, fast=False)) + pd.testing.assert_frame_equal(fast, full) + + def test_optional_match_null_row(self): + """No-match OPTIONAL MATCH returns the openCypher null row, not empty.""" + ndf = pd.DataFrame({"id": [0, 1], "type": ["Message", "Person"]}) + edf = pd.DataFrame({"src": [0], "dst": [1], "type": ["OTHER"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + q = "OPTIONAL MATCH (m {id: 0})-[{type:'MISSING'}]->(p {type:'Person'}) RETURN p" + fast = _canon_nodes(_run_diff(g, "pandas", q, fast=True)) + full = _canon_nodes(_run_diff(g, "pandas", q, fast=False)) + pd.testing.assert_frame_equal(fast, full) + + def test_with_match_reentry_carried_seeds(self): + """WITH..MATCH reentry hands carried seeds via start_nodes; the fast path + deriving its seed from the filter alone would silently widen the seed set. + Node 0 is excluded by the WITH but matches the reentry filter and has a + LIKES edge -> divergence unless the fast path declines.""" + ndf = pd.DataFrame({"id": [0, 1, 2, 10, 11, 13], + "kind": ["person"] * 3 + ["post"] * 3}) + edf = pd.DataFrame({"src": [0, 0, 1, 2, 0], "dst": [1, 2, 10, 11, 13], + "type": ["KNOWS", "KNOWS", "LIKES", "LIKES", "LIKES"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + q = ("MATCH (p {id:0})-[{type:'KNOWS'}]->(a) WITH a " + "MATCH (a {kind:'person'})-[{type:'LIKES'}]->(b) RETURN b") + fast = _canon_nodes(_run_diff(g, "pandas", q, fast=True)) + full = _canon_nodes(_run_diff(g, "pandas", q, fast=False)) + pd.testing.assert_frame_equal(fast, full) + assert sorted(fast["b.id"].tolist()) == [10, 11]