From fddb9c1f5208b505b3edbd48fa18035c2f6959b8 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 18:10:28 -0700 Subject: [PATCH 1/3] perf(gfql): seeded typed-hop fast paths use resident indexes (#1658 x #1755) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With gfql_index_all() resident, the seeded helpers replace their three full-frame scans (O(N) seed row, O(E) frontier isin, O(N) endpoint gather) with node-id searchsorted + CSR adjacency gathers. Decline-gated to the identical scan body: absent/stale index (get_valid fingerprint + identity), non-numeric id families, seed not id-filtered. Local 50k/200k covered shape: pandas 3.11->1.85ms, polars 1.74->1.10ms; win scales with graph size. Pinned: parity+engagement (pandas/polars), string-id decline, stale-index decline. Stacked on perf/gfql-index-preserve-polars-frames (#1767) — polars index residency requires it. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- CHANGELOG.md | 1 + graphistry/compute/chain_fast_paths.py | 218 ++++++++++++++---- .../gfql/test_seeded_typed_hop_fastpath.py | 85 +++++++ 3 files changed, 263 insertions(+), 41 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ed1337b5a8..425eb1ca20 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Performance - **GFQL seeded fast path covers single-alias property RETURNs (LDBC IS5 shape) (#1755)**: `MATCH (m {id})-[:T]->(p) RETURN p.a AS x, p.b` — which lowers to `rows(source=p)+select(items)` rather than a whole-row projection — now takes the seeded typed-hop fast path: the deduped destination rows are renamed/selected directly (value-identical to the rows-pivot + select pipeline; row order may differ). Dtype parity with the full pandas path is exact: non-id properties ride the rows-pivot there, which upcasts int/float to float64 and bool to object, so the lean tail applies the same casts (the id property keeps its dtype); dtype classes not verified against the pivot (datetimes, pandas extension dtypes, categoricals) decline. `res._edges` is the same empty edges frame the full path yields (never None). Conservative declines keep full-path semantics for everything else: cross-alias refs (`RETURN m.x, p.y`), mixed whole-row+property, DISTINCT/ORDER BY/LIMIT (extra lowered ops), expr items, properties absent from the node frame (the full path's null/error semantics must apply), and requested-vs-actual engine mismatches (the full path converts to the requested engine). Differential fast-vs-full sweep across all of the above on pandas + polars: exact parity incl. dtypes and edges, engagement pinned both ways. Measured on LDBC SNB SF1 (dgx, harness row-validated): `message-creator` (IS5) 116.3→38.3 ms on polars (the residue is the seed scan, targeted separately by resident-index coverage). +- **GFQL seeded typed-hop fast paths use resident indexes (#1658 x #1755)**: with `gfql_index_all()` resident, the seeded fast-path helpers (`_seeded_typed_hop_pandas_cudf`, `_seeded_typed_return_dst_pandas_cudf`, `_seeded_typed_return_dst_polars`) replace their three full-frame scans — O(N) seed-row lookup, O(E) frontier `isin`, O(N) endpoint gather — with positional index lookups (node-id searchsorted + CSR adjacency gather), so a seeded Cypher lookup stops paying graph-size costs entirely. Decline-gated to fall back to the identical scan body: absent/stale index (fingerprint+identity via `get_valid`), non-numeric id families (object/str ids keep the scan path's null semantics), seed not id-filtered. Results identical either way (row order may differ; node-id index implies unique ids). Local 50k/200k covered-shape: pandas 3.11→1.85ms, polars 1.74→1.10ms; the win scales with graph size (the residue is parse/lowering, not scans). Pinned in `TestResidentIndexSeededFastPath` (parity+engagement pandas/polars, string-id decline, stale-index decline). - **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 typed-hop fast path on polars/polars-gpu (#1755)**: extends the seeded typed-hop Cypher fast path to the polars engines via native polars filters (`_seeded_typed_return_dst_polars`), so a seeded typed 1-hop RETURN also lands in single-digit ms on polars (13.7→3.4ms, 4×) and polars-gpu (24.6→2.5ms, 10×). Dispatch is on the ACTUAL frame type (`is_polars_df`), not the requested engine, because WITH..MATCH reentry can request polars while handing pandas-materialized frames. Same decline contract as the pandas path (undirected/multi-hop/varlen/predicates fall through; value-identical for the covered shape), verified by the per-engine differential sweep and oracle tests in `test_seeded_typed_hop_fastpath.py`. - **GFQL two-star grouped-count OLAP queries collect once on polars (#1755)**: profiling showed the residual two-star path (graph-benchmark q5-q7 shapes) spent ~72% of its runtime on the fixed per-op cost of ~27 eager polars operations. When every residual translates natively (see the entry below), the whole plan — base filter_dicts, residual filters, typed-edge filters, semi-joins, group-prop lookup — now composes as ONE `pl.LazyFrame` plan collected once at the join (`filter_by_dict_polars` gains a `filter_expr_by_dict_polars` twin: same column/dtype resolution and typed-error/NIE contract, expression only). Value-identical to the eager lane it replaces (same filters/joins/aggregation), INCLUDING the empty-match boundary: the eager all-left-counts==1 shortcut's single `n=0` row (the openCypher count over no rows) is reproduced — the left-counts frame rides the same single collect. Edges are engine-converted before going lazy, so pandas frames under `engine='polars'` (the WITH..MATCH reentry shape) take the lane instead of crashing; LazyFrame inputs and untranslatable residuals decline to the eager path. Measured on dgx vs warm Kuzu, same session: q5 7.7→3.8ms (flips to a GFQL win vs 5.2), q6 9.8→6.1 (flips vs 7.3), q7 8.7→5.7 (flips vs 6.9) — moving the graph-benchmark scoreboard from 4W/2T/3L to 7W/1T/1L. Differential tests spy the extracted `_connected_join_two_star_fused_polars` helper and ASSERT lane engagement (a `count(*)` query declines the whole two-star path — pinned as a decline shape): fused vs forced-eager exact-row parity (ORDER BY pinned), the ungrouped empty-match n=0 row, pandas-frames-under-polars-engine parity, empty-result shape, pandas oracle. diff --git a/graphistry/compute/chain_fast_paths.py b/graphistry/compute/chain_fast_paths.py index fa9dfb1a54..0369f9acc6 100644 --- a/graphistry/compute/chain_fast_paths.py +++ b/graphistry/compute/chain_fast_paths.py @@ -41,6 +41,81 @@ def _seeded_scalar_filters(fd: Optional[Dict[str, Any]], df: DataFrameT) -> Opti return out +def _resident_seed_indexes( + g: Plottable, nodes_df: DataFrameT, edges_df: DataFrameT, + node: str, src: str, dst: str, direction: Direction, +) -> Optional[Tuple[Any, Any, Any, Any]]: + """(node_id_index, adjacency_index, xp, engine) when BOTH resident indexes + validly cover this directed seeded hop on these EXACT frames (fingerprint + + identity via get_valid), else None — callers keep the scan path, so a stale + or absent index can never change results, only speed.""" + from graphistry.Engine import Engine, is_polars_df + from graphistry.compute.gfql.index import get_registry + from graphistry.compute.gfql.index.registry import EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID + from graphistry.compute.gfql.index.engine_arrays import array_namespace + registry = get_registry(g) + if registry.is_empty(): + return None + mod = str(type(nodes_df).__module__) + if is_polars_df(nodes_df): + engine = Engine.POLARS + elif 'cudf' in mod: + engine = Engine.CUDF + elif mod.startswith('pandas'): + engine = Engine.PANDAS + else: + return None + kind = EDGE_OUT_ADJ if direction == "forward" else EDGE_IN_ADJ + adj = registry.get_valid(kind, edges_df, (src, dst), engine) + nid = registry.get_valid(NODE_ID, nodes_df, (node,), engine) + if adj is None or nid is None: + return None + xp, _ = array_namespace(engine) + return nid, adj, xp, engine + + +def _ids_to_key_array(vals: Any, keys: Any, xp: Any) -> Optional[Any]: + """Values (python list / Series / array) -> deduped backend array in the index + key dtype, nulls dropped (null ids never link — matching the scan path's + dropna semantics). None when the cast is not value-safe (mismatched families + like str-vs-int decline to the scan path rather than risk false matches).""" + try: + arr = xp.asarray(vals.to_numpy() if hasattr(vals, "to_numpy") else vals) + if arr.dtype.kind == "f": + arr = arr[~xp.isnan(arr)] + if arr.dtype.kind not in "iuf" or keys.dtype.kind not in "iuf": + return None # numeric id families only: object/str ids keep the scan path (null-object semantics) + if arr.dtype != keys.dtype: + common = xp.promote_types(arr.dtype, keys.dtype) + arr = arr.astype(common) + return xp.unique(arr) + except (TypeError, ValueError): + return None + + +def _index_node_rows(nid: Any, ids: Any, xp: Any, engine: Any, nodes_df: DataFrameT) -> Optional[DataFrameT]: + """Node rows whose id is in ``ids`` via the resident node-id index (positional + gather; row order is id-sorted, covered by the value-identical contract).""" + from graphistry.compute.gfql.index.lookup import lookup_node_rows + from graphistry.compute.gfql.index.engine_arrays import take_rows + arr = _ids_to_key_array(ids, nid.keys_sorted, xp) + if arr is None: + return None + return take_rows(nodes_df, lookup_node_rows(nid, arr, xp), engine) + + +def _index_edge_rows(adj: Any, ids: Any, xp: Any, engine: Any, edges_df: DataFrameT) -> Optional[DataFrameT]: + """Edge rows incident to ``ids`` on the indexed side via the CSR adjacency + (searchsorted gather; replaces the O(E) isin scan).""" + from graphistry.compute.gfql.index.lookup import lookup_edge_rows + from graphistry.compute.gfql.index.engine_arrays import take_rows + arr = _ids_to_key_array(ids, adj.keys_sorted, xp) + if arr is None: + return None + rows, _ = lookup_edge_rows(adj, arr, xp) + return take_rows(edges_df, rows, engine) + + def _seeded_typed_hop_pandas_cudf( g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge, src: str, dst: str, node: str, direction: Direction, @@ -71,26 +146,54 @@ def _seeded_typed_hop_pandas_cudf( # 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: - 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].dropna())] - else: - edges = edges_df - if ef: # typed edge (edge_match) — now on the reduced frontier - for k, v in ef.items(): - 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. 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].dropna()) | nodes_df[node].isin(edges[dst].dropna()) - ].drop_duplicates(subset=[node]) + # Resident-index acceleration (#1658 x #1755): when node-id + directional + # adjacency indexes are valid for these exact frames, the O(N) seed scan, the + # O(E) frontier isin, and the O(N) candidate gather become positional lookups. + # Any decline (no index, stale fingerprint, unsafe id cast) falls back to the + # scan body below — identical results either way, only speed differs. + ctx = _resident_seed_indexes(g, nodes_df, edges_df, node, src, dst, direction) if (n0f and node in n0f) else None + seed_nodes = edges = cand = None + if ctx is not None: + nid, adj, xp, idx_engine = ctx + seed_nodes = _index_node_rows(nid, [n0f[node]], xp, idx_engine, nodes_df) + if seed_nodes is not None: + for k, v in n0f.items(): + if k != node: + seed_nodes = seed_nodes[seed_nodes[k] == v] + edges = _index_edge_rows(adj, seed_nodes[node], xp, idx_engine, edges_df) + if edges is not None: + if ef: + for k, v in ef.items(): + edges = edges[edges[k] == v] + if 'cudf' in str(type(edges).__module__): + import cudf as _cd # type: ignore + endpoint_ids = _cd.concat([edges[src], edges[dst]]) + else: + import pandas as _pd + endpoint_ids = _pd.concat([edges[src], edges[dst]]) + cand = _index_node_rows(nid, endpoint_ids, xp, idx_engine, nodes_df) + if cand is None: + if n0f: + 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].dropna())] + else: + edges = edges_df + if ef: # typed edge (edge_match) — now on the reduced frontier + for k, v in ef.items(): + 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. 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].dropna()) | nodes_df[node].isin(edges[dst].dropna()) + ].drop_duplicates(subset=[node]) + assert edges is not None and cand is not None # both branches above assign if n2f: # destination-node filter (to-side) n2_cand = cand for k, v in n2f.items(): @@ -130,16 +233,33 @@ def _seeded_typed_return_dst_pandas_cudf( # 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].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].dropna())] + ctx = _resident_seed_indexes(g, nodes_df, edges_df, node, src, dst, direction) if node in n0f else None + seed_nodes = edges = dstn = None + if ctx is not None: + nid, adj, xp, idx_engine = ctx + seed_nodes = _index_node_rows(nid, [n0f[node]], xp, idx_engine, nodes_df) + if seed_nodes is not None: + for k, v in n0f.items(): + if k != node: + seed_nodes = seed_nodes[seed_nodes[k] == v] + edges = _index_edge_rows(adj, seed_nodes[node], xp, idx_engine, edges_df) + if edges is not None: + if ef: + for k, v in ef.items(): + edges = edges[edges[k] == v] + dstn = _index_node_rows(nid, edges[to_col], xp, idx_engine, nodes_df) + if dstn is None: + 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].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].dropna())] + assert edges is not None and dstn is not None # both branches above assign if n2f: for k, v in n2f.items(): dstn = dstn[dstn[k] == v] @@ -178,17 +298,33 @@ def _seeded_typed_return_dst_polars( # Membership sets are drop_nulls()'d (null ids/endpoints never link, matching # the full pipeline's joins) and passed via .implode() (Series-arg is_in is # deprecated in polars 1.42, see polars#22149). - seed_nodes = nodes_df - for k, v in n0f.items(): - seed_nodes = seed_nodes.filter(pl.col(k) == v) - from_ids = seed_nodes.get_column(node).drop_nulls() - if from_ids.len() == 0: - return nodes_df.clear(), edges_df.clear() - edges = edges_df.filter(pl.col(from_col).is_in(from_ids.implode())) - for k, v in ef.items(): # typed edge on the reduced frontier - edges = edges.filter(pl.col(k) == v) - dst_ids = edges.get_column(to_col).drop_nulls().unique() - dstn = nodes_df.filter(pl.col(node).is_in(dst_ids.implode())) + ctx = _resident_seed_indexes(g, nodes_df, edges_df, node, src, dst, direction) if node in n0f else None + seed_nodes = edges = dstn = None + if ctx is not None: + nid, adj, xp, idx_engine = ctx + seed_nodes = _index_node_rows(nid, [n0f[node]], xp, idx_engine, nodes_df) + if seed_nodes is not None: + for k, v in n0f.items(): + if k != node: + seed_nodes = seed_nodes.filter(pl.col(k) == v) + edges = _index_edge_rows(adj, seed_nodes.get_column(node), xp, idx_engine, edges_df) + if edges is not None: + for k, v in ef.items(): + edges = edges.filter(pl.col(k) == v) + dstn = _index_node_rows(nid, edges.get_column(to_col), xp, idx_engine, nodes_df) + if dstn is None: + seed_nodes = nodes_df + for k, v in n0f.items(): + seed_nodes = seed_nodes.filter(pl.col(k) == v) + from_ids = seed_nodes.get_column(node).drop_nulls() + if from_ids.len() == 0: + return nodes_df.clear(), edges_df.clear() + edges = edges_df.filter(pl.col(from_col).is_in(from_ids.implode())) + for k, v in ef.items(): # typed edge on the reduced frontier + edges = edges.filter(pl.col(k) == v) + dst_ids = edges.get_column(to_col).drop_nulls().unique() + dstn = nodes_df.filter(pl.col(node).is_in(dst_ids.implode())) + assert edges is not None and dstn is not None # both branches above assign for k, v in n2f.items(): # destination-node filter dstn = dstn.filter(pl.col(k) == v) # drop dangling edges + dedup destination nodes (mirror the pandas tail) 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 a62a2ce489..0b126f6586 100644 --- a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -787,3 +787,88 @@ def test_string_dtype_property_engages_and_matches(self): fast, full = self._fast_and_full(g, "pandas", q) assert str(fast._nodes["nm"].dtype) == str(full._nodes["nm"].dtype) pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) +class TestResidentIndexSeededFastPath: + """#1658 x #1755: when node-id + adjacency indexes are resident and valid, the + seeded helpers replace their O(N)/O(E) scans with positional lookups. Results + must be identical to the un-indexed scan path (which stays the fallback for + every decline: absent/stale index, non-numeric ids).""" + + def _frames(self): + rng = np.random.default_rng(0) + N, E = 5000, 30000 + ndf = pd.DataFrame({ + "id": np.arange(N), + "type": np.where(np.arange(N) % 3 == 0, "Person", "Message"), + "age": rng.integers(20, 60, N).astype(float), + }) + edf = pd.DataFrame({ + "src": rng.integers(0, N, E).astype(float), + "dst": rng.integers(0, N, E).astype(float), + "type": np.where(np.arange(E) % 2 == 0, "KNOWS", "LIKES"), + }) + edf.loc[5, "dst"] = np.nan # null endpoint must not link + edf.loc[9, "dst"] = N + 999 # dangling endpoint + return ndf, edf + + def _canon(self, res): + df = res._nodes + df = df.to_pandas() if hasattr(df, "to_pandas") else pd.DataFrame(df) + cols = sorted(map(str, df.columns)) + df.columns = list(map(str, df.columns)) + return df[cols].sort_values(cols).reset_index(drop=True) + + def _spied(self, g, q, engine, monkeypatch): + import graphistry.compute.chain_fast_paths as cfp + counts = {"n": 0} + on = cfp._index_node_rows + monkeypatch.setattr(cfp, "_index_node_rows", + lambda *a, **k: (counts.__setitem__("n", counts["n"] + 1), on(*a, **k))[1]) + res = g.gfql(q, engine=engine) + monkeypatch.setattr(cfp, "_index_node_rows", on) + return res, counts["n"] + + Q = "MATCH (m {id: 33})-[:KNOWS]->(p) RETURN p" + + @pytest.mark.parametrize("engine", ["pandas", "polars"]) + def test_indexed_parity_and_engagement(self, engine, monkeypatch): + if engine == "polars": + pytest.importorskip("polars") + ndf, edf = self._frames() + if engine == "polars": + import polars as pl + mk = lambda: graphistry.nodes(pl.from_pandas(ndf), "id").edges(pl.from_pandas(edf), "src", "dst") # noqa: E731 + else: + mk = lambda: graphistry.nodes(ndf, "id").edges(edf, "src", "dst") # noqa: E731 + plain = mk().gfql(self.Q, engine=engine) + indexed, hits = self._spied(mk().gfql_index_all(), self.Q, engine, monkeypatch) + assert hits > 0, "resident index did not serve the seeded fast path" + pd.testing.assert_frame_equal(self._canon(indexed), self._canon(plain)) + + def test_string_ids_decline_to_scan(self, monkeypatch): + # numeric-only gate: object/str ids keep the scan path (null-object semantics) + ndf, edf = self._frames() + ndf = ndf.assign(id="n" + ndf["id"].astype(str)) + edf = edf.dropna(subset=["src", "dst"]).astype({"src": int, "dst": int}) + edf = edf.assign(src="n" + edf["src"].astype(str), dst="n" + edf["dst"].astype(str)) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst").gfql_index_all() + q = "MATCH (m {id: 'n33'})-[:KNOWS]->(p) RETURN p" + plain = graphistry.nodes(ndf, "id").edges(edf, "src", "dst").gfql(q, engine="pandas") + got = g.gfql(q, engine="pandas") + pd.testing.assert_frame_equal(self._canon(got), self._canon(plain)) + + def test_stale_index_declines_to_scan(self, monkeypatch): + # rebinding edges invalidates the fingerprint: results must stay correct + import graphistry.compute.chain_fast_paths as cfp + ndf, edf = self._frames() + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst").gfql_index_all() + edf2 = pd.DataFrame(edf).copy() + edf2 = edf2[edf2["type"] == "KNOWS"] + g2 = g.edges(edf2, "src", "dst") # stale registry rides along + counts = {"n": 0} + oe = cfp._index_edge_rows + monkeypatch.setattr(cfp, "_index_edge_rows", + lambda *a, **k: (counts.__setitem__("n", counts["n"] + 1), oe(*a, **k))[1]) + got = g2.gfql(self.Q, engine="pandas") + plain = graphistry.nodes(ndf, "id").edges(edf2, "src", "dst").gfql(self.Q, engine="pandas") + assert counts["n"] == 0, "stale index must not serve" + pd.testing.assert_frame_equal(self._canon(got), self._canon(plain)) From a9293317e852fdb5d4bd7c77e298ba6e2f8c73f6 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 18:28:36 -0700 Subject: [PATCH 2/3] =?UTF-8?q?fix(gfql):=20resident-index=20gate=20?= =?UTF-8?q?=E2=80=94=20no=20float-collapsing=20id=20promotes,=20cudf=20dev?= =?UTF-8?q?ice=20path,=20polars-gpu=20tags,=20serve-asserted=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review findings (plans/review-pr-1768/review.md): B1 int64<->uint64 promotes to float64 and collapses ids >= 2^53 into false matches (repro'd: seed 2^62+1 matched node 2^62) — integral pairs whose promote is float now DECLINE to the exact-compare scan path. M2 spies count SERVES (non-None), not calls, so a gate regression that declines everything fails the suite. M1 native-chain hop indexed branch + reverse/EDGE_IN_ADJ now tested (serve-asserted, parity). M3 cudf ids via .dropna().values (device array; to_numpy raised on nulls and round-tripped host). M4 polars frames also accept polars-gpu-tagged indexes (same numpy sidecars + polars gather). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- CHANGELOG.md | 2 +- graphistry/compute/chain_fast_paths.py | 26 ++++++- .../gfql/test_seeded_typed_hop_fastpath.py | 67 ++++++++++++++++++- 3 files changed, 88 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 425eb1ca20..267a72c416 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Performance - **GFQL seeded fast path covers single-alias property RETURNs (LDBC IS5 shape) (#1755)**: `MATCH (m {id})-[:T]->(p) RETURN p.a AS x, p.b` — which lowers to `rows(source=p)+select(items)` rather than a whole-row projection — now takes the seeded typed-hop fast path: the deduped destination rows are renamed/selected directly (value-identical to the rows-pivot + select pipeline; row order may differ). Dtype parity with the full pandas path is exact: non-id properties ride the rows-pivot there, which upcasts int/float to float64 and bool to object, so the lean tail applies the same casts (the id property keeps its dtype); dtype classes not verified against the pivot (datetimes, pandas extension dtypes, categoricals) decline. `res._edges` is the same empty edges frame the full path yields (never None). Conservative declines keep full-path semantics for everything else: cross-alias refs (`RETURN m.x, p.y`), mixed whole-row+property, DISTINCT/ORDER BY/LIMIT (extra lowered ops), expr items, properties absent from the node frame (the full path's null/error semantics must apply), and requested-vs-actual engine mismatches (the full path converts to the requested engine). Differential fast-vs-full sweep across all of the above on pandas + polars: exact parity incl. dtypes and edges, engagement pinned both ways. Measured on LDBC SNB SF1 (dgx, harness row-validated): `message-creator` (IS5) 116.3→38.3 ms on polars (the residue is the seed scan, targeted separately by resident-index coverage). -- **GFQL seeded typed-hop fast paths use resident indexes (#1658 x #1755)**: with `gfql_index_all()` resident, the seeded fast-path helpers (`_seeded_typed_hop_pandas_cudf`, `_seeded_typed_return_dst_pandas_cudf`, `_seeded_typed_return_dst_polars`) replace their three full-frame scans — O(N) seed-row lookup, O(E) frontier `isin`, O(N) endpoint gather — with positional index lookups (node-id searchsorted + CSR adjacency gather), so a seeded Cypher lookup stops paying graph-size costs entirely. Decline-gated to fall back to the identical scan body: absent/stale index (fingerprint+identity via `get_valid`), non-numeric id families (object/str ids keep the scan path's null semantics), seed not id-filtered. Results identical either way (row order may differ; node-id index implies unique ids). Local 50k/200k covered-shape: pandas 3.11→1.85ms, polars 1.74→1.10ms; the win scales with graph size (the residue is parse/lowering, not scans). Pinned in `TestResidentIndexSeededFastPath` (parity+engagement pandas/polars, string-id decline, stale-index decline). +- **GFQL seeded typed-hop fast paths use resident indexes (#1658 x #1755)**: with `gfql_index_all()` resident, the seeded fast-path helpers (`_seeded_typed_hop_pandas_cudf`, `_seeded_typed_return_dst_pandas_cudf`, `_seeded_typed_return_dst_polars`) replace their three full-frame scans — O(N) seed-row lookup, O(E) frontier `isin`, O(N) endpoint gather — with positional index lookups (node-id searchsorted + CSR adjacency gather), so a seeded Cypher lookup stops paying graph-size costs entirely. Decline-gated to fall back to the identical scan body: absent/stale index (fingerprint+identity via `get_valid`), non-numeric id families (object/str ids keep the scan path's null semantics), seed not id-filtered. Results identical either way (row order may differ; node-id index implies unique ids). dgx 50k/200k covered-shape: pandas 1.71→0.86ms, polars 2.02→0.69ms (vs Kuzu's same-box 1.06ms); SF1-scale 3.2M/17M polars 47.2→2.25ms — polars numbers measured with a polars-engine index resident, which today requires `gfql_index_all(engine='polars')` on polars frames (until #1767 lands, AUTO index builds swap polars frames to pandas; pandas/cuDF flows are unaffected and independent). Pinned in `TestResidentIndexSeededFastPath` (parity+engagement pandas/polars, string-id decline, stale-index decline, uint64/int64 promote decline, reverse-direction serving). - **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 typed-hop fast path on polars/polars-gpu (#1755)**: extends the seeded typed-hop Cypher fast path to the polars engines via native polars filters (`_seeded_typed_return_dst_polars`), so a seeded typed 1-hop RETURN also lands in single-digit ms on polars (13.7→3.4ms, 4×) and polars-gpu (24.6→2.5ms, 10×). Dispatch is on the ACTUAL frame type (`is_polars_df`), not the requested engine, because WITH..MATCH reentry can request polars while handing pandas-materialized frames. Same decline contract as the pandas path (undirected/multi-hop/varlen/predicates fall through; value-identical for the covered shape), verified by the per-engine differential sweep and oracle tests in `test_seeded_typed_hop_fastpath.py`. - **GFQL two-star grouped-count OLAP queries collect once on polars (#1755)**: profiling showed the residual two-star path (graph-benchmark q5-q7 shapes) spent ~72% of its runtime on the fixed per-op cost of ~27 eager polars operations. When every residual translates natively (see the entry below), the whole plan — base filter_dicts, residual filters, typed-edge filters, semi-joins, group-prop lookup — now composes as ONE `pl.LazyFrame` plan collected once at the join (`filter_by_dict_polars` gains a `filter_expr_by_dict_polars` twin: same column/dtype resolution and typed-error/NIE contract, expression only). Value-identical to the eager lane it replaces (same filters/joins/aggregation), INCLUDING the empty-match boundary: the eager all-left-counts==1 shortcut's single `n=0` row (the openCypher count over no rows) is reproduced — the left-counts frame rides the same single collect. Edges are engine-converted before going lazy, so pandas frames under `engine='polars'` (the WITH..MATCH reentry shape) take the lane instead of crashing; LazyFrame inputs and untranslatable residuals decline to the eager path. Measured on dgx vs warm Kuzu, same session: q5 7.7→3.8ms (flips to a GFQL win vs 5.2), q6 9.8→6.1 (flips vs 7.3), q7 8.7→5.7 (flips vs 6.9) — moving the graph-benchmark scoreboard from 4W/2T/3L to 7W/1T/1L. Differential tests spy the extracted `_connected_join_two_star_fused_polars` helper and ASSERT lane engagement (a `count(*)` query declines the whole two-star path — pinned as a decline shape): fused vs forced-eager exact-row parity (ORDER BY pinned), the ungrouped empty-match n=0 row, pandas-frames-under-polars-engine parity, empty-result shape, pandas oracle. diff --git a/graphistry/compute/chain_fast_paths.py b/graphistry/compute/chain_fast_paths.py index 0369f9acc6..89aecbd417 100644 --- a/graphistry/compute/chain_fast_paths.py +++ b/graphistry/compute/chain_fast_paths.py @@ -66,8 +66,17 @@ def _resident_seed_indexes( else: return None kind = EDGE_OUT_ADJ if direction == "forward" else EDGE_IN_ADJ - adj = registry.get_valid(kind, edges_df, (src, dst), engine) - nid = registry.get_valid(NODE_ID, nodes_df, (node,), engine) + engines = [engine] + if engine == Engine.POLARS: + # an index built with explicit engine='polars-gpu' serves the same eager + # polars frames (same numpy sidecars + polars row-gather) + engines.append(Engine.POLARS_GPU) + adj = nid = None + for eng_try in engines: + adj = registry.get_valid(kind, edges_df, (src, dst), eng_try) + nid = registry.get_valid(NODE_ID, nodes_df, (node,), eng_try) + if adj is not None and nid is not None: + break if adj is None or nid is None: return None xp, _ = array_namespace(engine) @@ -80,13 +89,24 @@ def _ids_to_key_array(vals: Any, keys: Any, xp: Any) -> Optional[Any]: dropna semantics). None when the cast is not value-safe (mismatched families like str-vs-int decline to the scan path rather than risk false matches).""" try: - arr = xp.asarray(vals.to_numpy() if hasattr(vals, "to_numpy") else vals) + if 'cudf' in str(type(vals).__module__): + vals = vals.dropna() + raw = vals.values # device array; to_numpy() raises on nulls + round-trips host + elif hasattr(vals, "to_numpy"): + raw = vals.to_numpy() + else: + raw = vals + arr = xp.asarray(raw) if arr.dtype.kind == "f": arr = arr[~xp.isnan(arr)] if arr.dtype.kind not in "iuf" or keys.dtype.kind not in "iuf": return None # numeric id families only: object/str ids keep the scan path (null-object semantics) if arr.dtype != keys.dtype: common = xp.promote_types(arr.dtype, keys.dtype) + if arr.dtype.kind in "iu" and keys.dtype.kind in "iu" and common.kind == "f": + # int64<->uint64 promotes to float64, which collapses distinct ids + # >= 2^53 into false matches; the scan path compares exactly -> decline. + return None arr = arr.astype(common) return xp.unique(arr) except (TypeError, ValueError): 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 0b126f6586..c748201743 100644 --- a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -818,11 +818,17 @@ def _canon(self, res): return df[cols].sort_values(cols).reset_index(drop=True) def _spied(self, g, q, engine, monkeypatch): + # counts SERVES (non-None returns), not calls: a gate regression that + # declines everything must fail these tests, not silently pass them import graphistry.compute.chain_fast_paths as cfp counts = {"n": 0} on = cfp._index_node_rows - monkeypatch.setattr(cfp, "_index_node_rows", - lambda *a, **k: (counts.__setitem__("n", counts["n"] + 1), on(*a, **k))[1]) + + def spy(*a, **k): + out = on(*a, **k) + counts["n"] += out is not None + return out + monkeypatch.setattr(cfp, "_index_node_rows", spy) res = g.gfql(q, engine=engine) monkeypatch.setattr(cfp, "_index_node_rows", on) return res, counts["n"] @@ -840,7 +846,11 @@ def test_indexed_parity_and_engagement(self, engine, monkeypatch): else: mk = lambda: graphistry.nodes(ndf, "id").edges(edf, "src", "dst") # noqa: E731 plain = mk().gfql(self.Q, engine=engine) - indexed, hits = self._spied(mk().gfql_index_all(), self.Q, engine, monkeypatch) + # explicit engine: on master, gfql_index_all(AUTO) on polars frames still + # swaps them to pandas (resolve_engine legacy input-format policy), so the + # polars index must be requested explicitly; #1767 (separate PR) makes + # AUTO preserve polars frames, at which point bare gfql_index_all() also works. + indexed, hits = self._spied(mk().gfql_index_all(engine=engine), self.Q, engine, monkeypatch) assert hits > 0, "resident index did not serve the seeded fast path" pd.testing.assert_frame_equal(self._canon(indexed), self._canon(plain)) @@ -872,3 +882,54 @@ def test_stale_index_declines_to_scan(self, monkeypatch): plain = graphistry.nodes(ndf, "id").edges(edf2, "src", "dst").gfql(self.Q, engine="pandas") assert counts["n"] == 0, "stale index must not serve" pd.testing.assert_frame_equal(self._canon(got), self._canon(plain)) + + + def test_native_chain_hop_indexed_parity_forward_and_reverse(self, monkeypatch): + """M1 pin: the native-chain hop helper's indexed branch (only reachable via + chain ops, never Cypher) — forward (EDGE_OUT_ADJ) and reverse (EDGE_IN_ADJ), + serve-asserted, parity vs the un-indexed graph.""" + from graphistry.compute.ast import n, e_forward, e_reverse + import graphistry.compute.chain_fast_paths as cfp + ndf, edf = self._frames() + mk = lambda: graphistry.nodes(ndf, "id").edges(edf, "src", "dst") # noqa: E731 + for ops in ( + [n({"id": 33}), e_forward(edge_match={"type": "KNOWS"}), n({"type": "Person"})], + [n({"id": 33}), e_reverse(edge_match={"type": "KNOWS"}), n({"type": "Person"})], + ): + serves = {"n": 0} + oe = cfp._index_edge_rows + + def spy(*a, **k): + out = oe(*a, **k) + serves["n"] += out is not None + return out + monkeypatch.setattr(cfp, "_index_edge_rows", spy) + got = mk().gfql_index_all().gfql(ops, engine="pandas") + monkeypatch.setattr(cfp, "_index_edge_rows", oe) + assert serves["n"] > 0, f"indexed hop branch did not serve for {ops[1].direction}" + plain = mk().gfql(ops, engine="pandas") + pd.testing.assert_frame_equal(self._canon(got), self._canon(plain)) + + def test_uint64_int64_id_mix_declines_not_collapses(self, monkeypatch): + """B1 pin: int64<->uint64 promotes to float64, which collapses ids >= 2**53 + into false matches; the gate must DECLINE (scan path compares exactly).""" + import graphistry.compute.chain_fast_paths as cfp + big = np.uint64(2**62) + ndf = pd.DataFrame({"id": np.array([big, big + np.uint64(1), np.uint64(7)], dtype=np.uint64), + "type": ["Person"] * 3}) + edf = pd.DataFrame({"src": np.array([int(big) + 1], dtype=np.int64), + "dst": np.array([7], dtype=np.int64), "type": ["KNOWS"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst").gfql_index_all() + q = f"MATCH (m {{id: {int(big) + 1}}})-[:KNOWS]->(p) RETURN p" + serves = {"n": 0} + on = cfp._index_node_rows + + def spy(*a, **k): + out = on(*a, **k) + serves["n"] += out is not None + return out + monkeypatch.setattr(cfp, "_index_node_rows", spy) + got = g.gfql(q, engine="pandas") + monkeypatch.setattr(cfp, "_index_node_rows", on) + plain = graphistry.nodes(ndf, "id").edges(edf, "src", "dst").gfql(q, engine="pandas") + pd.testing.assert_frame_equal(self._canon(got), self._canon(plain)) From eb57062bb747a17fc792ed29783ff3b6bfa00274 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 21 Jul 2026 19:20:44 -0700 Subject: [PATCH 3/3] refactor(gfql): typed resident-index helpers + opt-in perf regression gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback: _ids_to_key_array & friends were Any-typed — now NodeIdIndex/AdjacencyIndex/ArrayLike/ArrayNamespace/SeriesT protocol types (isnan added to the ArrayNamespace protocol), localized ignores only at the cudf module-check narrows. New GFQL_PERF_TESTS=1 opt-in wall-clock gate (indexed must beat scan >=1.5x at 500k edges) for the benchmarking-rig protocol; CI keeps the deterministic engagement asserts. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/chain_fast_paths.py | 31 +++++++++++++------ graphistry/compute/typing.py | 3 ++ .../gfql/test_seeded_typed_hop_fastpath.py | 29 +++++++++++++++++ 3 files changed, 54 insertions(+), 9 deletions(-) diff --git a/graphistry/compute/chain_fast_paths.py b/graphistry/compute/chain_fast_paths.py index 89aecbd417..3cd1b42727 100644 --- a/graphistry/compute/chain_fast_paths.py +++ b/graphistry/compute/chain_fast_paths.py @@ -7,11 +7,15 @@ """ # ruff: noqa: E501 -from typing import Any, Dict, Optional, Tuple +from typing import Any, Dict, Optional, Sequence, Tuple, TYPE_CHECKING, Union, cast from graphistry.Plottable import Plottable from .ast import ASTNode, ASTEdge, Direction -from .typing import DataFrameT +from .typing import ArrayLike, ArrayNamespace, DataFrameT, SeriesT + +if TYPE_CHECKING: + from graphistry.Engine import Engine + from graphistry.compute.gfql.index.registry import AdjacencyIndex, NodeIdIndex def _seeded_scalar_filters(fd: Optional[Dict[str, Any]], df: DataFrameT) -> Optional[Dict[str, Any]]: @@ -44,7 +48,7 @@ def _seeded_scalar_filters(fd: Optional[Dict[str, Any]], df: DataFrameT) -> Opti def _resident_seed_indexes( g: Plottable, nodes_df: DataFrameT, edges_df: DataFrameT, node: str, src: str, dst: str, direction: Direction, -) -> Optional[Tuple[Any, Any, Any, Any]]: +) -> Optional[Tuple["NodeIdIndex", "AdjacencyIndex", ArrayNamespace, "Engine"]]: """(node_id_index, adjacency_index, xp, engine) when BOTH resident indexes validly cover this directed seeded hop on these EXACT frames (fingerprint + identity via get_valid), else None — callers keep the scan path, so a stale @@ -80,18 +84,21 @@ def _resident_seed_indexes( if adj is None or nid is None: return None xp, _ = array_namespace(engine) - return nid, adj, xp, engine + # get_valid returns the union type; kind selection above guarantees the concrete classes + return cast("NodeIdIndex", nid), cast("AdjacencyIndex", adj), xp, engine -def _ids_to_key_array(vals: Any, keys: Any, xp: Any) -> Optional[Any]: +def _ids_to_key_array( + vals: Union["SeriesT", Sequence[Any]], keys: ArrayLike, xp: ArrayNamespace, +) -> Optional[ArrayLike]: """Values (python list / Series / array) -> deduped backend array in the index key dtype, nulls dropped (null ids never link — matching the scan path's dropna semantics). None when the cast is not value-safe (mismatched families like str-vs-int decline to the scan path rather than risk false matches).""" try: if 'cudf' in str(type(vals).__module__): - vals = vals.dropna() - raw = vals.values # device array; to_numpy() raises on nulls + round-trips host + vals = vals.dropna() # type: ignore[union-attr] # cudf Series by module check + raw = vals.values # type: ignore[union-attr] # device array; to_numpy() raises on nulls + round-trips host elif hasattr(vals, "to_numpy"): raw = vals.to_numpy() else: @@ -113,7 +120,10 @@ def _ids_to_key_array(vals: Any, keys: Any, xp: Any) -> Optional[Any]: return None -def _index_node_rows(nid: Any, ids: Any, xp: Any, engine: Any, nodes_df: DataFrameT) -> Optional[DataFrameT]: +def _index_node_rows( + nid: "NodeIdIndex", ids: Union["SeriesT", Sequence[Any]], + xp: ArrayNamespace, engine: "Engine", nodes_df: DataFrameT, +) -> Optional[DataFrameT]: """Node rows whose id is in ``ids`` via the resident node-id index (positional gather; row order is id-sorted, covered by the value-identical contract).""" from graphistry.compute.gfql.index.lookup import lookup_node_rows @@ -124,7 +134,10 @@ def _index_node_rows(nid: Any, ids: Any, xp: Any, engine: Any, nodes_df: DataFra return take_rows(nodes_df, lookup_node_rows(nid, arr, xp), engine) -def _index_edge_rows(adj: Any, ids: Any, xp: Any, engine: Any, edges_df: DataFrameT) -> Optional[DataFrameT]: +def _index_edge_rows( + adj: "AdjacencyIndex", ids: Union["SeriesT", Sequence[Any]], + xp: ArrayNamespace, engine: "Engine", edges_df: DataFrameT, +) -> Optional[DataFrameT]: """Edge rows incident to ``ids`` on the indexed side via the CSR adjacency (searchsorted gather; replaces the O(E) isin scan).""" from graphistry.compute.gfql.index.lookup import lookup_edge_rows diff --git a/graphistry/compute/typing.py b/graphistry/compute/typing.py index a72c287274..3819452158 100644 --- a/graphistry/compute/typing.py +++ b/graphistry/compute/typing.py @@ -110,5 +110,8 @@ def sort(self, a: ArrayLike) -> ArrayLike: def unique(self, a: ArrayLike) -> ArrayLike: ... + def isnan(self, a: ArrayLike) -> ArrayLike: + ... + def promote_types(self, type1: Any, type2: Any) -> Any: ... 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 c748201743..75c596eb0a 100644 --- a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -933,3 +933,32 @@ def spy(*a, **k): monkeypatch.setattr(cfp, "_index_node_rows", on) plain = graphistry.nodes(ndf, "id").edges(edf, "src", "dst").gfql(q, engine="pandas") pd.testing.assert_frame_equal(self._canon(got), self._canon(plain)) + + @pytest.mark.skipif("GFQL_PERF_TESTS" not in __import__("os").environ, + reason="wall-clock perf gate: opt-in (GFQL_PERF_TESTS=1) — run on the " + "benchmarking rig, not CI runners (noisy shared hardware)") + def test_perf_indexed_beats_scan_at_scale(self): + """Perf regression gate for the resident-index path: at 500k edges the + indexed covered-shape query must beat the scan fast path by >=1.5x + (measured margin is ~2-20x engine/scale dependent; 1.5x is the alarm + threshold, not the claim). Deterministic engagement is asserted by the + functional tests; this catches a silently-slow index path.""" + import time, statistics + rng = np.random.default_rng(0) + N, E = 125_000, 500_000 + ndf = pd.DataFrame({"id": np.arange(N), "type": np.where(np.arange(N) % 3 == 0, "Person", "Message")}) + edf = pd.DataFrame({"src": rng.integers(0, N, E), "dst": rng.integers(0, N, E), "type": "KNOWS"}) + q = "MATCH (m {id: 33})-[:KNOWS]->(p) RETURN p" + + def med(g): + g.gfql(q, engine="pandas") + ts = [] + for _ in range(15): + t0 = time.perf_counter() + g.gfql(q, engine="pandas") + ts.append(time.perf_counter() - t0) + return statistics.median(ts) + + scan = med(graphistry.nodes(ndf, "id").edges(edf, "src", "dst")) + indexed = med(graphistry.nodes(ndf, "id").edges(edf, "src", "dst").gfql_index_all()) + assert indexed * 1.5 < scan, f"indexed {indexed*1e3:.2f}ms not >=1.5x faster than scan {scan*1e3:.2f}ms"