perf(gfql): seeded typed-hop fast paths use resident indexes (#1658 x #1755)

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL

Author: lmeyerov

CHANGELOG.md

@@ -9,6 +9,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 <!-- Do Not Erase This Section - Used for tracking unreleased changes -->
 
 ### Performance
+- **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 connected-join simple residuals filter natively on polars (#1729/#1755)**: on the polars engines, a connected-join node residual of the #1729 scalar shapes — case-insensitive equality ``(tolower(a.col) = tolower('lit'))`` and scalar comparisons ``(a.col <op> literal)`` for ``= >= <= > <`` — previously dispatched a full sub-``chain()`` per alias (the polars row pipeline has no native ``where_rows``), costing several ms per residual on OLAP group-aggregate queries. `_residual_polars_expr` now translates these shapes to native ``pl.Expr`` filters applied directly to the alias frame (graph-benchmark q5 10.2→6.6ms, q6 10.7→8.9ms on dgx). Null semantics match the ``where_rows`` evaluator (null comparisons drop the row); any unrecognized shape, alias mismatch, or absent column falls back to the existing chain path — and if *any* expr in a residual group fails to translate, the whole group falls back (never a partial mix). Byte-parity verified across an 8-case adversarial differential (nulls, unicode/casefold, numeric ranges, mixed residuals) plus the full suite.

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)