perf(gfql): engage #1658 index for seeded gfql()/Cypher chains on ALL 4 engines (incl. typed edges)

CHANGELOG.md

@@ -13,6 +13,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Fixed
 - **GFQL polars/polars-gpu seeded traversal no longer pays an O(E) NaN re-scan per hop on float-column graphs (identity-stable `_pl_nan_to_null` + clean-cache)**: every `hop()` runs `_coerce_input_formats` → `_pl_nan_to_null`, which scanned `is_nan().any()` over every float column on the (unchanged) resident edge frame **on every call** — so repeated seeded hops on a resident graph (the seeded-Search / native-hop pattern) grew O(E) with edge count on polars while pandas stayed flat. Measured: an indexed polars `g.hop` on 8M edges with one float column was **33.8 ms and growing**; it is now **0.20 ms and FLAT** (~140× faster; O(degree)), matching pandas. `_pl_nan_to_null` now probes an eager polars frame for real NaN once, returns a clean frame UNCHANGED (restoring the #1726 identity guard that #1731 reverted — so frame-identity caches like the #1658 index keep engaging), rewrites only the columns that genuinely carry NaN (values identical to the old unconditional `fill_nan`), and caches the id of frames verified clean so later calls skip the probe (recycle-safe via `weakref.finalize`; a rebound/new frame re-probes, so NaN→null semantics are unchanged). Regression: full 4-engine `test_index.py` parity + new `TestPlNanCleanCache` cases (NaN-present cleaned, clean-frame cached same-object, distinct frames independent).
+- **GFQL seeded `gfql()`/Cypher chains now engage the resident #1658 adjacency index on ALL FOUR engines (incl. typed edges) instead of always scanning**: a seeded hop expressed as a `gfql([...])`/Cypher CHAIN (rather than a direct `g.hop(...)`) previously fell back to the O(E) scan even with a resident index, because both chain executors attach a synthetic per-edge id column to the edge frame (pandas/cuDF eager: `copy(deep=False)` + assign; polars native lazy: `with_row_index`) → a fresh edge-frame object → the index's `source_ref is df` identity guard missed. Both chain paths now re-point the resident edge adjacency indexes at the augmented frame via `GfqlIndexRegistry.rebind_edges` (provably safe: the shallow copy / `with_row_index` preserves the indexed src/dst columns by value; the structural fingerprint — row count + bound columns + engine — is unchanged by an added column). Additionally, a **simple scalar-equality `edge_match`** (typed edges, e.g. `-[:KNOWS]->` / `e_forward({"type": X})`) is now index-coverable on the wavefront path: `index_seeded_hop` applies the edge predicate to the CSR-matched rows each hop, parity-exact with the scan's `filter_edges_by_dict`. Predicate/membership-list `edge_match`, `edge_query`, and the direct-hop (non-wavefront) path deliberately stay on the scan path (no over-reach). Measured (dgx, 500k nodes / 4M edges / 4 edge types, warm median, index result == scan result): pandas typed 1-hop **2.1×** / untyped **2.7×**; cuDF typed **1.9×** / untyped **2.0×**; polars typed **1.3×** / untyped **5.4×**; polars-gpu typed engaged (1.0×) / untyped **12.4×**; 2-hop typed engaged on all four engines. 29 new differential + engagement regression tests (`test_index.py`), 4-engine, plus the existing 109-test index suite green.
 - **GFQL Cypher `CASE` with mixed-dtype branches is now engine-consistent on cuDF (no more `GFQLTypeError`)**: pandas coerces the two `CASE` branches to a common type, but cuDF's `.where` raised `TypeError: cudf does not support mixed types`, surfaced as a hard `GFQLTypeError`. This bit `CASE WHEN path IS NULL THEN -1 ELSE length(path) END` over an UNREACHABLE `shortestPath` (int `-1` branch vs an object/null hops branch) — pandas returned `-1`, cuDF errored. The row-AST `CASE` evaluator now unifies the branch dtypes and retries, so cuDF returns the same value pandas does. Regression test `test_case_mixed_dtype.py`.
 - **GFQL polars engine raises a clean typed error (not an opaque polars `InvalidOperationError`) for a string predicate on a non-string column**: `WHERE col STARTS WITH/CONTAINS/regex ...` on a Categorical/Enum/numeric column raised a clean `GFQLSchemaError` ("string predicate used on non-string column") on pandas/cuDF but leaked polars' internal `InvalidOperationError: expected String type, got: cat` on polars. `filter_by_dict_polars` now raises the same `GFQLSchemaError` (categorical treated as non-string, exactly as `filter_by_dict`), so all three engines agree. Regression test `test_polars_string_predicate_nonstring.py`; also confirms connected-join predicate pushdown is parity-correct on polars for string/numeric/eq/bool columns.
 - **GFQL string predicates (`Contains`/`Startswith`/`Endswith`/`Match`/`Fullmatch`) are now value-safe on non-string columns**: applying a string predicate to a numeric/temporal/bool column raised an opaque `AttributeError: Can only use .str accessor with string values!` on pandas and cuDF. They now follow openCypher semantics — a string op over a non-string value is null → excluded (matching the established per-cell behavior on an object column holding non-strings) — and never stringify the column (which would diverge pandas↔cuDF, e.g. wrongly matching `5 CONTAINS '5'`). String, mixed-object, and all-null columns are unchanged. Regression tests in `test_str.py` across dtypes; cuDF parity confirmed.

graphistry/compute/chain.py

@@ -964,6 +964,15 @@ def _chain_impl(
             indexed_edges_df = g._edges.copy(deep=False)
             indexed_edges_df[GFQL_EDGE_INDEX] = indexed_edges_df.index
             g = g.edges(indexed_edges_df, edge=GFQL_EDGE_INDEX)
+            # The shallow copy above only ADDS the synthetic id column; the indexed
+            # src/dst columns are preserved by value. Re-point any resident #1658
+            # adjacency index at the new edge frame so the seeded fast path still
+            # engages through gfql()/Cypher chains (else the identity guard misses and
+            # every chain hop falls back to the O(E) scan).
+            from graphistry.compute.gfql.index import get_registry, set_registry
+            _reg = get_registry(g)
+            if not _reg.is_empty():
+                g = set_registry(g, _reg.rebind_edges(indexed_edges_df))
         else:
             added_edge_index = False
 

graphistry/compute/gfql/index/__init__.py

@@ -16,7 +16,7 @@
 )
 from .api import (
     create_index, drop_index, show_indexes, gfql_index_edges, gfql_index_all,
-    get_registry, get_index_policy, maybe_index_hop, index_name, index_trace,
+    get_registry, set_registry, get_index_policy, maybe_index_hop, index_name, index_trace,
 )
 from .wire import (
     CreateIndex, DropIndex, ShowIndexes, IndexOp, apply_index_op, index_op_from_json,
@@ -33,7 +33,7 @@
     "EDGE_OUT_ADJ", "EDGE_IN_ADJ", "NODE_ID", "ADJ_KINDS", "ALL_KINDS",
     "AdjacencyIndex", "NodeIdIndex",
     "create_index", "drop_index", "show_indexes", "gfql_index_edges",
-    "gfql_index_all", "get_registry", "get_index_policy", "maybe_index_hop", "index_name",
+    "gfql_index_all", "get_registry", "set_registry", "get_index_policy", "maybe_index_hop", "index_name",
     "index_trace",
     "CreateIndex", "DropIndex", "ShowIndexes", "IndexOp", "apply_index_op",
     "index_op_from_json", "is_index_op", "is_index_op_json",

graphistry/compute/gfql/index/api.py

@@ -88,6 +88,14 @@ def _attach(g: Plottable, registry: GfqlIndexRegistry) -> Plottable:
     return res
 
 
+def set_registry(g: Plottable, registry: GfqlIndexRegistry) -> Plottable:
+    """Attach ``registry`` to a copy of ``g`` (public wrapper over ``_attach``).
+
+    Used by the chain executor to re-point the resident index at an edge frame it
+    augmented in place (see ``GfqlIndexRegistry.rebind_edges``)."""
+    return _attach(g, registry)
+
+
 def index_name(kind: IndexKind, column: Optional[str]) -> str:
     return f"{kind}:{column}" if column else kind
 
@@ -279,16 +287,29 @@ def _hop_is_index_coverable(
     edge_query: Optional[str],
     include_zero_hop_seed: bool,
     target_wave_front: Optional[DataFrameT],
+    return_as_wave_front: bool = False,
 ) -> bool:
     if nodes is None:
         return False
     if any(x is not None for x in (
         min_hops if (min_hops is not None and min_hops > 1) else None,
         output_min_hops, output_max_hops, label_node_hops, label_edge_hops,
-        edge_match, source_node_match, destination_node_match,
+        source_node_match, destination_node_match,
         source_node_query, destination_node_query, edge_query, target_wave_front,
     )):
         return False
+    # Typed-edge edge_match: coverable only as a simple scalar-equality filter on the
+    # wavefront (chain/Cypher) path, where index_seeded_hop applies it per-hop parity-
+    # exact with the scan's filter_edges_by_dict. Predicate/membership forms and the
+    # direct-hop (non-wavefront) path stay on scan.
+    if edge_match is not None:
+        from .traverse import is_simple_equality_edge_match
+        if not (
+            return_as_wave_front
+            and isinstance(edge_match, dict)
+            and is_simple_equality_edge_match(edge_match)
+        ):
+            return False
     if label_seeds or include_zero_hop_seed:
         return False
     if not to_fixed_point and (not isinstance(hops, int) or hops < 1):
@@ -408,6 +429,7 @@ def _bail(reason: str) -> Optional[Plottable]:
         edge_query=edge_query,
         include_zero_hop_seed=include_zero_hop_seed,
         target_wave_front=target_wave_front,
+        return_as_wave_front=return_as_wave_front,
     ):
         return _bail("query not index-coverable")
     assert nodes is not None
@@ -460,6 +482,7 @@ def _bail(reason: str) -> Optional[Plottable]:
         g, registry, nodes=nodes, node_col=node_col, src=src, dst=dst, engine=engine,
         hops=eff_hops, to_fixed_point=to_fixed_point, direction=direction,
         return_as_wave_front=return_as_wave_front,
+        edge_match=cast(Optional[dict], rest.get("edge_match")),
     )
     if trace:
         _record(cast(IndexTraceStep, {

graphistry/compute/gfql/index/registry.py

@@ -96,6 +96,46 @@ def without(self, kind: IndexKind) -> "GfqlIndexRegistry":
         new.pop(kind, None)
         return GfqlIndexRegistry(new)
 
+    def rebind_edges(self, new_edges: DataFrameT) -> "GfqlIndexRegistry":
+        """Re-point the EDGE adjacency indexes' identity guard at ``new_edges``.
+
+        Caller contract: ``new_edges`` was produced by a transform that preserves
+        the indexed src/dst columns by value (same rows, same order) — e.g. a shallow
+        copy that merely ADDS an unrelated column. The chain executor does exactly
+        this when it attaches its synthetic per-edge id (chain.py), which otherwise
+        breaks the ``source_ref is df`` identity guard and forces a full scan. The
+        CSR arrays stay valid (row positions unchanged); we only swap the strong-ref
+        so ``get_valid`` recognizes the live frame. NODE_ID is left untouched (node
+        materialization may legitimately change node rows).
+
+        ENFORCED (O(1), engine-portable): the swap happens only when ``new_edges``
+        still matches the index's structural fingerprint (row count + bound cols +
+        engine) and actually carries the indexed columns; on any mismatch the edge
+        index is DROPPED (safe miss -> scan path) instead of re-pointed at a frame
+        it wasn't built over. Value-level preservation remains the caller's promise —
+        checking it would be the O(E) scan this path exists to avoid."""
+        new = dict(self.indexes)
+        for kind in (EDGE_OUT_ADJ, EDGE_IN_ADJ):
+            idx = new.get(kind)
+            if idx is None:
+                continue
+            if not isinstance(idx, AdjacencyIndex):  # defensive; also narrows for mypy
+                new.pop(kind, None)
+                continue
+            ok = idx.fingerprint == frame_fingerprint(
+                new_edges, tuple(idx.fingerprint[1]), idx.engine)
+            if ok:
+                try:
+                    colnames = set(new_edges.columns)
+                    ok = idx.key_col in colnames and idx.other_col in colnames
+                except Exception:
+                    ok = False
+            if ok:
+                new[kind] = replace(idx, source_ref=new_edges)
+            else:
+                new.pop(kind, None)
+        return GfqlIndexRegistry(new)
+
     def get(self, kind: IndexKind) -> Optional[Union[AdjacencyIndex, NodeIdIndex]]:
         return self.indexes.get(kind)
 

graphistry/compute/gfql/index/traverse.py

@@ -6,13 +6,17 @@
 falls back to the scan/join path — correctness is never traded for speed).
 
 Covered (v1): seeded (nodes given), integer ``hops`` >= 1 or ``to_fixed_point``,
-direction forward/reverse/undirected, ``return_as_wave_front``. Not covered
-(returns None): edge/source/destination match or query, target_wave_front,
-min_hops>1, output_min/max_hops, labeling, missing node table.
+direction forward/reverse/undirected, ``return_as_wave_front``, and a simple
+scalar-equality ``edge_match`` (typed edges, e.g. Cypher ``-[:KNOWS]->``) applied on
+the wavefront path. Not covered (returns None): predicate/membership edge_match,
+source/destination match or query, edge_query, target_wave_front, min_hops>1,
+output_min/max_hops, labeling, missing node table.
 """
 from __future__ import annotations
 
-from typing import List, Optional, Tuple, cast
+from typing import Any, List, Optional, Tuple, cast
+
+from typing_extensions import TypeGuard
 
 from graphistry.Engine import Engine
 from graphistry.compute.typing import DataFrameT
@@ -23,7 +27,7 @@
 )
 from .lookup import lookup_edge_rows, lookup_node_rows
 from .registry import EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, AdjacencyIndex, GfqlIndexRegistry, NodeIdIndex
-from .types import ArrayLike, HopDirection
+from .types import ArrayLike, EdgeMatch, HopDirection, SimpleEqualityEdgeMatch
 
 
 def _indices_for_direction(
@@ -44,6 +48,89 @@ def _indices_for_direction(
     return [out_idx, in_idx]
 
 
+def is_simple_equality_edge_match(
+    edge_match: Optional[EdgeMatch],
+) -> TypeGuard[SimpleEqualityEdgeMatch]:
+    """True iff ``edge_match`` is a dict of plain scalar equalities.
+
+    This is the only ``edge_match`` shape the index path accelerates parity-exact:
+    it mirrors filter_by_dict's concrete scalar ``==`` branch. ASTPredicate values
+    (predicate path), membership lists/sets/tuples (isin path), and nested dicts are
+    NOT covered here — the caller keeps them on the scan path.
+    """
+    if not edge_match:
+        return False
+    from graphistry.compute.predicates.ASTPredicate import ASTPredicate
+    from graphistry.compute.filter_by_dict import _is_membership_filter_value
+    for v in edge_match.values():
+        if isinstance(v, ASTPredicate):
+            return False
+        # Membership must be decided by the SAME helper the scan path uses
+        # (filter_by_dict). A local (list, tuple, set, dict) check misses frozenset /
+        # pd.Index / pd.Series, which the scan lowers to isin — an equality mask over
+        # such a value is silently all-False (wrong answer), never an error.
+        if _is_membership_filter_value(v) or isinstance(v, dict):
+            return False
+    return True
+
+
+def _build_edge_keep_mask(
+    edges: DataFrameT, edge_match: EdgeMatch, engine: Engine, xp: "object"
+) -> Optional[ArrayLike]:
+    """Boolean array over ORIGINAL edge rows (length E, same indexing as
+    ``AdjacencyIndex.other_values`` / ``row_positions``) selecting rows that satisfy
+    a simple-equality ``edge_match``.
+
+    Built via each frame's native ``col == val`` (so cudf string columns stay on the
+    cudf layer instead of a cupy string compare). Returns ``None`` on ANY unexpected
+    shape or error, so the caller falls back to scan rather than risk a divergence.
+    """
+    try:
+        if not is_simple_equality_edge_match(edge_match):
+            return None
+        from graphistry.compute.filter_by_dict import (
+            _is_numeric_dtype_safe, _is_string_dtype_safe,
+        )
+        n_edges = int(edges.shape[0])
+        mask: Optional[ArrayLike] = None
+        for col, val in edge_match.items():
+            if col not in edges.columns:
+                return None
+            if engine in (Engine.POLARS, Engine.POLARS_GPU):
+                series = edges.get_column(col)
+            else:
+                series = edges[col]
+            # Obvious dtype mismatch (numeric col vs str val, string col vs numeric
+            # val): the scan raises GFQLSchemaError E302 where a naive == is silently
+            # all-False. Decline -> caller falls back to the scan, which raises the
+            # SAME error (parity-exact; mirrors filter_by_dict's exact two checks,
+            # skipped like the scan on empty frames).
+            if n_edges > 0:
+                dt = series.dtype
+                if _is_numeric_dtype_safe(dt) and isinstance(val, str):
+                    return None
+                if (_is_string_dtype_safe(dt)
+                        and isinstance(val, (int, float)) and not isinstance(val, bool)):
+                    return None
+            # Null-safe materialization: on null-carrying columns (pandas nullable
+            # Int64/boolean/string, polars nulls — which the NaN->null coercion makes
+            # common) a bare == yields NA cells, and to_numpy() then produces an
+            # OBJECT-dtype array that later explodes at rows[edge_keep[rows]]
+            # (IndexError: not int/bool). Null == val filters out on the scan path,
+            # so fill False is parity-exact.
+            if engine in (Engine.POLARS, Engine.POLARS_GPU):
+                col_mask = cast(ArrayLike, (series == val).fill_null(False).to_numpy())
+            elif engine == Engine.CUDF:
+                col_mask = cast(ArrayLike, (series == val).fillna(False).values)
+            else:
+                col_mask = cast(
+                    ArrayLike, (series == val).fillna(False).to_numpy(dtype=bool))
+            mask = col_mask if mask is None else cast(ArrayLike, cast(Any, mask) & cast(Any, col_mask))
+        return mask
+    except Exception:  # pragma: no cover - defensive parity guard
+        return None
+
+
 def index_seeded_hop(
     g: Plottable,
     registry: GfqlIndexRegistry,
@@ -57,6 +144,7 @@ def index_seeded_hop(
     to_fixed_point: bool,
     direction: HopDirection,
     return_as_wave_front: bool,
+    edge_match: Optional[EdgeMatch] = None,
 ) -> Optional[Plottable]:
     if nodes is None or g._edges is None or g._nodes is None:
         return None
@@ -77,6 +165,16 @@ def index_seeded_hop(
 
     xp, _backend = array_namespace(engine)
 
+    # Typed-edge (edge_match) support: a boolean mask over ORIGINAL edge rows that
+    # pass the match predicate, applied to the CSR-matched rows each hop. Gated to
+    # simple scalar equality + the wavefront path by the coverability check upstream
+    # (maybe_index_hop); an unsupported shape returns None here => scan (parity-safe).
+    edge_keep: Optional[ArrayLike] = None
+    if edge_match:
+        edge_keep = _build_edge_keep_mask(edges, edge_match, engine, xp)
+        if edge_keep is None:
+            return None
+
     # Do NOT narrow the seed to the index key dtype (a node-id int64 seed cast to
     # an int32 edge-endpoint key wraps large ids → false match). lookup promotes both
     # sides to a common dtype; numpy/cupy set ops promote on concat. So we keep ids at
@@ -100,6 +198,11 @@ def index_seeded_hop(
         neigh_parts: List[ArrayLike] = []
         for ix in indices:
             rows, matched = lookup_edge_rows(ix, frontier, xp)
+            if edge_keep is not None:
+                # Keep only CSR-matched rows whose edge passes edge_match. Wavefront-
+                # only (coverability gate), so the `matched`/first-hop `visited`
+                # bookkeeping below — which edge_match does NOT filter — is never read.
+                rows = rows[edge_keep[rows]]
             edge_rows_parts.append(rows)
             neigh_parts.append(ix.other_values[rows])
             matched_parts.append(matched)