perf(gfql/index): index-accelerated get_degrees + EXISTS-prune (#5 degree-cache / #3 membership)

graphistry/compute/ComputeMixin.py

@@ -321,6 +321,35 @@ def get_degrees(
             nodes_df = g_nodes._nodes.assign(**{c: 0 for c in cols}).astype({c: "int32" for c in cols})
             return g.nodes(nodes_df, node_id)
 
+        # GFQL #1658 index fast path (#5 degree-cache / #3 membership): read degrees
+        # straight from a resident CSR adjacency index (O(N) searchsorted gather)
+        # instead of the O(E) group_by below. Bulk over all nodes, so always
+        # profitable when a valid index is resident; `index_policy='off'` skips.
+        # try/except -> any index issue falls through to the scan path (never wrong).
+        from graphistry.compute.gfql.index import get_index_policy
+        if get_index_policy(g) != "off":
+            try:
+                from graphistry.compute.gfql.index import get_registry
+                from graphistry.compute.gfql.index.degrees import degrees_from_index
+                _reg = get_registry(g)
+                if not _reg.is_empty():
+                    _d = degrees_from_index(
+                        _reg, g_nodes._nodes, node_id, g._edges,
+                        (g._source, g._destination), engine_concrete,
+                    )
+                    if _d is not None:
+                        _in, _out = _d
+                        _keep = [c for c in g_nodes._nodes.columns if c not in (degree_in, degree_out, col)]
+                        _nf = g_nodes._nodes[_keep].assign(**{degree_in: _in, degree_out: _out})
+                        _nf = _nf.assign(**{
+                            degree_in: _nf[degree_in].astype("int32"),
+                            degree_out: _nf[degree_out].astype("int32"),
+                            col: (_nf[degree_in] + _nf[degree_out]).astype("int32"),
+                        })
+                        return g.nodes(_nf, node_id)
+            except (AttributeError, ImportError, NotImplementedError, TypeError, ValueError):
+                pass
+
         in_df = _degree_agg(g._edges, g._destination, degree_in, node_id)
         out_df = _degree_agg(g._edges, g._source, degree_out, node_id)
         deg = safe_merge(in_df, out_df, on=node_id, how="outer")

graphistry/compute/gfql/call/executor.py

@@ -141,7 +141,7 @@ def _execute_validated_call(g: Plottable, function: str, validated_params: Dict[
     # declined by that guard.
     if _active_frames_are_polars(g) and function in ("get_degrees", "get_indegrees", "get_outdegrees"):
         from graphistry.compute.gfql.lazy.engine.polars import degrees as _pl_degrees
-        return getattr(_pl_degrees, function + "_polars")(g, **validated_params)
+        return getattr(_pl_degrees, function + "_polars")(g, engine=engine.value, **validated_params)
 
     if not hasattr(g, function):
         raise AttributeError(

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, maybe_index_hop, index_name, index_trace,
+    get_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", "maybe_index_hop", "index_name",
+    "gfql_index_all", "get_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

@@ -8,11 +8,11 @@
 from __future__ import annotations
 
 import copy
-from typing import Any, Dict, List, Literal, Optional, Union, cast
+from typing import Dict, List, Literal, Optional, cast
 
 import pandas as pd
 
-from graphistry.Engine import EngineAbstract, Engine, resolve_engine
+from graphistry.Engine import EngineAbstract, Engine, EngineAbstractType, resolve_engine
 from graphistry.compute.typing import DataFrameT
 from graphistry.Plottable import Plottable
 from .registry import (
@@ -28,7 +28,8 @@
     IndexTrace, IndexTraceStep,
 )
 
-# Private Plottable attachment key. Keep access behind get_registry()/show_indexes().
+# Private Plottable attachment keys. Keep access behind helpers.
+POLICY_ATTR = "_gfql_index_policy"
 REGISTRY_ATTR = "_gfql_index_registry"
 
 # --- lightweight, thread-local index decision trace (for gfql_explain) -------
@@ -44,7 +45,7 @@ def __enter__(self) -> IndexTrace:
         _set_trace_steps(self.steps)
         return self.steps
 
-    def __exit__(self, *exc: Any) -> Literal[False]:
+    def __exit__(self, *exc: object) -> Literal[False]:
         _set_trace_steps(self.prev)
         return False
 
@@ -77,6 +78,10 @@ def get_registry(g: Plottable) -> GfqlIndexRegistry:
     return cast(GfqlIndexRegistry, getattr(g, REGISTRY_ATTR, EMPTY_REGISTRY))
 
 
+def get_index_policy(g: Plottable) -> IndexPolicy:
+    return cast(IndexPolicy, getattr(g, POLICY_ATTR, "use"))
+
+
 def _attach(g: Plottable, registry: GfqlIndexRegistry) -> Plottable:
     res = copy.copy(g)
     setattr(res, REGISTRY_ATTR, registry)
@@ -101,10 +106,10 @@ def _check_column(column: Optional[str], expected: str, kind: IndexKind) -> None
 def _is_resident_index_valid(
     g: Plottable,
     kind: IndexKind,
-    engine: Union[EngineAbstract, str] = EngineAbstract.AUTO,
+    engine: EngineAbstractType = EngineAbstract.AUTO,
 ) -> bool:
     """True when a resident index still matches the current graph frames."""
-    eng = resolve_engine(cast(Any, engine), g)
+    eng = resolve_engine(engine, g)
     registry = get_registry(g)
     if kind in ADJ_KINDS:
         src, dst = g._source, g._destination
@@ -125,7 +130,7 @@ def create_index(
     *,
     column: Optional[str] = None,
     name: Optional[str] = None,
-    engine: Union[EngineAbstract, str] = EngineAbstract.AUTO,
+    engine: EngineAbstractType = EngineAbstract.AUTO,
 ) -> Plottable:
     """Eagerly build a GFQL physical index and return a new Plottable carrying it.
 
@@ -135,7 +140,7 @@ def create_index(
     O(E log E) once, amortized over later seeded queries.
     """
     from dataclasses import replace
-    eng = resolve_engine(cast(Any, engine), g)
+    eng = resolve_engine(engine, g)
     # Build over frames already in the target engine so the index arrays land on
     # the right backend (cupy for cudf, numpy otherwise). No-op when already in-engine.
     from graphistry.compute.ComputeMixin import _coerce_input_formats
@@ -225,7 +230,7 @@ def show_indexes(g: Plottable) -> pd.DataFrame:
 
 
 def gfql_index_edges(g: Plottable, direction: EdgeIndexDirection = "both",
-                     engine: Union[EngineAbstract, str] = EngineAbstract.AUTO) -> Plottable:
+                     engine: EngineAbstractType = EngineAbstract.AUTO) -> Plottable:
     """Convenience: build edge adjacency index(es). direction: 'forward'|'reverse'|'both'."""
     if direction in ("forward", "both"):
         g = create_index(g, EDGE_OUT_ADJ, engine=engine)
@@ -235,7 +240,7 @@ def gfql_index_edges(g: Plottable, direction: EdgeIndexDirection = "both",
 
 
 def gfql_index_all(g: Plottable,
-                   engine: Union[EngineAbstract, str] = EngineAbstract.AUTO) -> Plottable:
+                   engine: EngineAbstractType = EngineAbstract.AUTO) -> Plottable:
     """Convenience: build out+in adjacency + (when ids are unique) node_id indexes.
 
     The node_id index is an optional materialization accelerator; if node ids aren't
@@ -318,7 +323,7 @@ def _ensure_indexes(
             F = int(nodes.shape[0])
             if not (F <= max(1024, 0.001 * E)):
                 return registry  # not selective enough to amortize a build
-        except Exception:
+        except (AttributeError, TypeError, ValueError):
             return registry
     for kind in needed:
         if registry.get_valid(kind, g._edges, (src, dst), engine) is None:
@@ -335,8 +340,8 @@ def _ensure_indexes(
 
 
 def maybe_index_hop(
-    g: Plottable, engine: Engine, *, nodes: Any, hops: Optional[int], direction: HopDirection, return_as_wave_front: bool,
-    to_fixed_point: bool = False, policy: str = "use", **rest: Any,
+    g: Plottable, engine: Engine, *, nodes: Optional[DataFrameT], hops: Optional[int], direction: HopDirection, return_as_wave_front: bool,
+    to_fixed_point: bool = False, policy: Optional[IndexPolicy] = "use", **rest: object,
 ) -> Optional[Plottable]:
     """Planner entry called from hop(). Returns an index-built subgraph, or None to
     fall back to the scan/join path.
@@ -345,7 +350,7 @@ def maybe_index_hop(
     (or buildable under auto/force), (b) the query is covered, (c) the frontier is
     not so large that a full scan is cheaper. Correctness is identical either way.
     """
-    policy = validate_index_policy(policy) or "use"
+    resolved_policy: IndexPolicy = validate_index_policy(policy) or "use"
 
     # Diagnostic trace (LP1) is populated only inside an explain context — build the
     # base record + a `_bail` helper that logs *why* we fell back to scan. All of this
@@ -355,49 +360,65 @@ def maybe_index_hop(
     if trace:
         diag = {
             "op": "hop", "direction": direction, "hops": hops,
-            "policy": policy, "engine": engine.value,
+            "policy": resolved_policy, "engine": engine.value,
         }
         try:
-            diag["frontier_n"] = int(nodes.shape[0])
-        except Exception:
+            if nodes is not None:
+                diag["frontier_n"] = int(nodes.shape[0])
+        except (AttributeError, TypeError, ValueError):
             pass
 
     def _bail(reason: str) -> Optional[Plottable]:
         if trace:
             _record(cast(IndexTraceStep, {**diag, "path": "scan", "decision_reason": reason}))
         return None
 
-    if policy == "off":
+    if resolved_policy == "off":
         return _bail("policy=off")
     registry = get_registry(g)
-    if registry.is_empty() and policy not in ("auto", "force"):
+    if registry.is_empty() and resolved_policy not in ("auto", "force"):
         return _bail("no resident index (policy=use)")
+
+    min_hops = cast(Optional[int], rest.get("min_hops"))
+    max_hops = cast(Optional[int], rest.get("max_hops"))
+    output_min_hops = cast(Optional[int], rest.get("output_min_hops"))
+    output_max_hops = cast(Optional[int], rest.get("output_max_hops"))
+    label_node_hops = cast(Optional[str], rest.get("label_node_hops"))
+    label_edge_hops = cast(Optional[str], rest.get("label_edge_hops"))
+    label_seeds = cast(bool, rest.get("label_seeds", False))
+    source_node_query = cast(Optional[str], rest.get("source_node_query"))
+    destination_node_query = cast(Optional[str], rest.get("destination_node_query"))
+    edge_query = cast(Optional[str], rest.get("edge_query"))
+    include_zero_hop_seed = cast(bool, rest.get("include_zero_hop_seed", False))
+    target_wave_front = cast(Optional[DataFrameT], rest.get("target_wave_front"))
+
     if not _hop_is_index_coverable(
         nodes=nodes, to_fixed_point=to_fixed_point, hops=hops,
-        min_hops=rest.get("min_hops"), max_hops=rest.get("max_hops"),
-        output_min_hops=rest.get("output_min_hops"),
-        output_max_hops=rest.get("output_max_hops"),
-        label_node_hops=rest.get("label_node_hops"),
-        label_edge_hops=rest.get("label_edge_hops"),
-        label_seeds=rest.get("label_seeds", False),
+        min_hops=min_hops, max_hops=max_hops,
+        output_min_hops=output_min_hops,
+        output_max_hops=output_max_hops,
+        label_node_hops=label_node_hops,
+        label_edge_hops=label_edge_hops,
+        label_seeds=label_seeds,
         edge_match=rest.get("edge_match"),
         source_node_match=rest.get("source_node_match"),
         destination_node_match=rest.get("destination_node_match"),
-        source_node_query=rest.get("source_node_query"),
-        destination_node_query=rest.get("destination_node_query"),
-        edge_query=rest.get("edge_query"),
-        include_zero_hop_seed=rest.get("include_zero_hop_seed", False),
-        target_wave_front=rest.get("target_wave_front"),
+        source_node_query=source_node_query,
+        destination_node_query=destination_node_query,
+        edge_query=edge_query,
+        include_zero_hop_seed=include_zero_hop_seed,
+        target_wave_front=target_wave_front,
     ):
         return _bail("query not index-coverable")
+    assert nodes is not None
 
     node_col = g._node
     src, dst = g._source, g._destination
     if node_col is None or src is None or dst is None or g._edges is None or g._nodes is None:
         return _bail("graph missing node/edge columns")
 
-    if policy in ("auto", "force"):
-        registry = _ensure_indexes(g, registry, direction, engine, policy, nodes, src, dst, node_col)
+    if resolved_policy in ("auto", "force"):
+        registry = _ensure_indexes(g, registry, direction, engine, resolved_policy, nodes, src, dst, node_col)
     if registry.is_empty():
         return _bail("no index available (build declined)")
 
@@ -419,23 +440,22 @@ def _bail(reason: str) -> Optional[Plottable]:
     if idx0 is None:
         # required direction not resident (undirected needs both); let driver decide
         pass
-    elif policy != "force":
+    elif resolved_policy != "force":
         try:
             frontier_n = int(nodes.shape[0])
             if idx0.n_keys > 0 and frontier_n >= frac * idx0.n_keys:
                 return _bail(
                     f"frontier {frontier_n} >= {frac}*n_keys "
                     f"({frac * idx0.n_keys:.0f}) -> scan cheaper"
                 )
-        except Exception:
+        except (AttributeError, TypeError, ValueError):
             pass
 
     # Honor max_hops: the scan resolves the hop count as ``max_hops or hops``
     # (compute/hop.py); the index must run the SAME number of accumulating hops.
     # (regression: max_hops was passed through *rest and silently ignored — the index ran
     # `hops` (default 1) while the scan ran max_hops → wrong answer.)
-    _max_hops = rest.get("max_hops")
-    eff_hops = _max_hops if _max_hops is not None else hops
+    eff_hops = max_hops if max_hops is not None else hops
     result = index_seeded_hop(
         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,

graphistry/compute/gfql/index/degrees.py

@@ -0,0 +1,99 @@
+"""Index-accelerated node degrees (#5 degree-cache / #3 membership).
+
+Resident CSR adjacency indexes already carry per-node degree for free:
+degree(key) = ``group_offsets[i+1] - group_offsets[i]``. This turns
+``get_degrees`` from an O(E) ``group_by(endpoint)`` + join into an O(N)
+``searchsorted`` gather. Bulk-over-all-nodes, so it always beats the scan when a
+valid index is resident (no selectivity gate). Engine-polymorphic (numpy host for
+pandas/polars, cupy device for cudf), fingerprint-validated like the hop fast path.
+"""
+from __future__ import annotations
+
+from typing import Optional, Tuple
+
+from graphistry.Engine import Engine
+from graphistry.compute.typing import ArrayLike, ArrayNamespace, DataFrameT
+from .engine_arrays import array_namespace, col_to_array
+from .registry import AdjacencyIndex, EDGE_OUT_ADJ, EDGE_IN_ADJ, GfqlIndexRegistry
+from .types import AdjacencyIndexKind, HopDirection
+
+
+def _valid_adjacency(
+    registry: GfqlIndexRegistry,
+    kind: AdjacencyIndexKind,
+    edges_df: DataFrameT,
+    cols: Tuple[str, ...],
+    engine: Engine,
+) -> Optional[AdjacencyIndex]:
+    idx = registry.get_valid(kind, edges_df, cols, engine)
+    return idx if isinstance(idx, AdjacencyIndex) else None
+
+
+def _degree_for_nodes(adj_index: AdjacencyIndex, node_ids: ArrayLike, xp: ArrayNamespace) -> ArrayLike:
+    """Per-node degree via searchsorted into the CSR keys (0 for isolated nodes)."""
+    keys = adj_index.keys_sorted
+    offs = adj_index.group_offsets
+    U = int(keys.shape[0])
+    n = int(node_ids.shape[0])
+    if U == 0:
+        return xp.zeros(n, dtype=xp.int64)
+    key_deg = (offs[1:] - offs[:-1]).astype(xp.int64)         # degree per unique key
+    # dtype-safe searchsorted (promote, never narrow — mirrors lookup.py)
+    f = node_ids
+    if f.dtype != keys.dtype:
+        common = xp.promote_types(f.dtype, keys.dtype)
+        f = f.astype(common)
+        keys = keys.astype(common)
+    pos = xp.searchsorted(keys, f)
+    pos_c = xp.where(pos < U, pos, U - 1)
+    hit = keys[pos_c] == f
+    return xp.where(hit, key_deg[pos_c], xp.zeros(n, dtype=xp.int64))
+
+
+def degrees_from_index(
+    registry: GfqlIndexRegistry, nodes_df: DataFrameT, node_col: str,
+    edges_df: DataFrameT, cols: Tuple[str, ...], engine: Engine,
+) -> Optional[Tuple[ArrayLike, ArrayLike]]:
+    """Return (in_degree, out_degree) arrays aligned to ``nodes_df`` rows, or None
+    to fall back to the group_by path (no valid resident adjacency index)."""
+    oi = _valid_adjacency(registry, EDGE_OUT_ADJ, edges_df, cols, engine)
+    ii = _valid_adjacency(registry, EDGE_IN_ADJ, edges_df, cols, engine)
+    if oi is None or ii is None:
+        return None
+    xp, _ = array_namespace(engine)
+    node_ids = col_to_array(nodes_df, node_col, engine)
+    out_deg = _degree_for_nodes(oi, node_ids, xp)   # out = src-keyed adjacency
+    in_deg = _degree_for_nodes(ii, node_ids, xp)    # in  = dst-keyed adjacency
+    return in_deg, out_deg
+
+
+def adjacency_membership_keys(
+    registry: GfqlIndexRegistry, direction: HopDirection, edges_df: DataFrameT, cols: Tuple[str, ...], engine: Engine,
+) -> Optional[ArrayLike]:
+    """Backend array of node-ids with >=1 edge in ``direction``
+    ('forward' = has out-edge, 'reverse' = has in-edge, 'undirected' = either).
+    None if the needed valid index isn't resident. #3 membership (= degree>=1);
+    powers EXISTS {(n)--()} prune-isolated without an O(E) traversal.
+
+    NOTE: keys include nodes whose ONLY edge is a self-loop — correct for the bare
+    pattern; the drop-self (neq) flavor must NOT use this path.
+    """
+    from .engine_arrays import union1d
+    if direction in ("forward", "undirected"):
+        oi = _valid_adjacency(registry, EDGE_OUT_ADJ, edges_df, cols, engine)
+        if oi is None:
+            return None
+    if direction in ("reverse", "undirected"):
+        ii = _valid_adjacency(registry, EDGE_IN_ADJ, edges_df, cols, engine)
+        if ii is None:
+            return None
+    if direction == "forward":
+        assert oi is not None
+        return oi.keys_sorted
+    if direction == "reverse":
+        assert ii is not None
+        return ii.keys_sorted
+    xp, _ = array_namespace(engine)
+    assert oi is not None
+    assert ii is not None
+    return union1d(oi.keys_sorted, ii.keys_sorted, xp)