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

graphistry/compute/ComputeMixin.py

@@ -315,6 +315,34 @@ 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).
+        if getattr(g, "_gfql_index_policy", "use") != "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 Exception:
+                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/degrees.py

@@ -0,0 +1,86 @@
+"""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 Any, Optional, Tuple
+
+from graphistry.Engine import Engine
+from .engine_arrays import array_namespace, col_to_array
+from .registry import EDGE_OUT_ADJ, EDGE_IN_ADJ, GfqlIndexRegistry
+
+
+def _degree_for_nodes(adj_index: Any, node_ids: Any, xp: Any) -> Any:
+    """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 getattr(f, "dtype", None) != getattr(keys, "dtype", None):
+        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: Any, node_col: str,
+    edges_df: Any, cols: Tuple[str, ...], engine: Engine,
+) -> Optional[Tuple[Any, Any]]:
+    """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 = registry.get_valid(EDGE_OUT_ADJ, edges_df, cols, engine)
+    ii = registry.get_valid(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: str, edges_df: Any, cols: Tuple[str, ...], engine: Engine,
+) -> Optional[Any]:
+    """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 = registry.get_valid(EDGE_OUT_ADJ, edges_df, cols, engine)
+        if oi is None:
+            return None
+    if direction in ("reverse", "undirected"):
+        ii = registry.get_valid(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)

graphistry/compute/gfql/lazy/engine/polars/degrees.py

@@ -6,11 +6,19 @@
 """
 from typing import Optional
 
+from graphistry.Engine import Engine
 from graphistry.Plottable import Plottable
 from .dtypes import is_lazy, colnames, col_dtype
 from .hop_eager import ensure_nodes_polars
 
 
+def _index_engine(engine: Optional[str] = None) -> Engine:
+    if engine is not None:
+        return Engine(engine)
+    from graphistry.compute.gfql.lazy import active_target, ExecutionTarget
+    return Engine.POLARS_GPU if active_target() == ExecutionTarget.GPU else Engine.POLARS
+
+
 def _endpoint_counts(edges, key_col: str, node_dt, node_col: str, alias: str):
     """group_by-count on ONE edge endpoint, key cast to node-id dtype so left-join keys match
     when node/endpoint dtypes diverge (polars won't auto-cast int<->float keys; pandas merges fine)."""
@@ -43,6 +51,29 @@ def get_degrees_polars(
     assert g._nodes is not None and g._edges is not None
     nodes, edges = g._nodes, g._edges
 
+    # GFQL #1658 index fast path (#5 degree-cache / #3 membership): degrees from a
+    # resident CSR index (O(N) gather) instead of the group_by below. Eager-only
+    # (LazyFrame get_column would force a collect); policy 'off' skips; try/except
+    # falls through to the scan path (never wrong).
+    if not is_lazy(nodes) and not is_lazy(edges) and getattr(g, "_gfql_index_policy", "use") != "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, nodes, node_col, edges, (src, dst), _index_engine(engine))
+                if _d is not None:
+                    _in, _out = _d
+                    drop0 = [c for c in colnames(nodes) if c in (degree_in, degree_out, col)]
+                    base0 = nodes.drop(drop0) if drop0 else nodes
+                    out0 = base0.with_columns(
+                        pl.Series(degree_in, _in).cast(pl.Int32),
+                        pl.Series(degree_out, _out).cast(pl.Int32),
+                    ).with_columns((pl.col(degree_in) + pl.col(degree_out)).cast(pl.Int32).alias(col))
+                    return g.nodes(out0, node_col)
+        except Exception:
+            pass
+
     node_dt = col_dtype(nodes, node_col)
     in_counts = _endpoint_counts(edges, dst, node_dt, node_col, degree_in)
     out_counts = _endpoint_counts(edges, src, node_dt, node_col, degree_out)
@@ -103,14 +134,14 @@ def _single_direction_degree_polars(g: Plottable, key_col: str, col: str) -> Plo
     return g.nodes(out, node_col)
 
 
-def get_indegrees_polars(g: Plottable, col: str = "degree_in") -> Plottable:
+def get_indegrees_polars(g: Plottable, col: str = "degree_in", engine: Optional[str] = None) -> Plottable:
     """Native ``get_indegrees`` (parity with ComputeMixin.get_indegrees): in-degree = count of
     edges whose DESTINATION endpoint is the node."""
     assert g._destination is not None, "Missing destination binding; set via .bind() or .edges()"
     return _single_direction_degree_polars(g, g._destination, col)
 
 
-def get_outdegrees_polars(g: Plottable, col: str = "degree_out") -> Plottable:
+def get_outdegrees_polars(g: Plottable, col: str = "degree_out", engine: Optional[str] = None) -> Plottable:
     """Native ``get_outdegrees`` (parity with ComputeMixin.get_outdegrees): out-degree = count of
     edges whose SOURCE endpoint is the node."""
     assert g._source is not None, "Missing source binding; set via .bind() or .edges()"

graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py

@@ -17,6 +17,7 @@
     import polars as pl
     from graphistry.compute.ast import ASTObject
 
+from .dtypes import is_lazy
 from .row_pipeline import _active_table, _rewrap
 
 
@@ -124,6 +125,41 @@ def _pattern_alias_keys_polars(
         return None
     base_graph = _rows_base_graph(g)
     node_id = base_graph._node or "id"  # Plottable._node: Optional[str]
+    # GFQL #1658 index fast path (#3 membership): bare EXISTS pattern (no endpoint/
+    # edge filters, no drop-self neq) -> participating nodes == "has an edge in this
+    # direction" = CSR adjacency membership. Skips the O(E) chain_polars below.
+    # Strict guard; anything richer (filters/neq/multi-hop) falls through unchanged.
+    if (
+        neq is None
+        and getattr(g, "_gfql_index_policy", "use") != "off"
+        and not getattr(n0, "filter_dict", None) and not getattr(n2, "filter_dict", None)
+        and not getattr(edge_op, "edge_match", None)
+        and getattr(edge_op, "edge_query", None) is None
+        and not is_lazy(base_graph._edges)
+    ):
+        try:
+            import numpy as _np
+            from graphistry.compute.gfql.index import get_registry
+            from graphistry.compute.gfql.index.degrees import adjacency_membership_keys
+            from graphistry.Engine import Engine as _Engine
+            from graphistry.compute.gfql.lazy import active_target as _active_target, ExecutionTarget as _ExecutionTarget
+            _reg = get_registry(g)
+            if not _reg.is_empty():
+                _edir = getattr(edge_op, "direction", "forward")
+                if _edir == "undirected":
+                    _mdir = "undirected"
+                elif (alias == n0._name) == (_edir == "forward"):
+                    _mdir = "forward"
+                else:
+                    _mdir = "reverse"
+                _src, _dst = base_graph._source, base_graph._destination
+                if isinstance(_src, str) and isinstance(_dst, str):
+                    _eng = _Engine.POLARS_GPU if _active_target() == _ExecutionTarget.GPU else _Engine.POLARS
+                    _mk = adjacency_membership_keys(_reg, _mdir, base_graph._edges, (_src, _dst), _eng)
+                    if _mk is not None:
+                        return pl.DataFrame({node_id: pl.Series(node_id, _np.asarray(_mk))})
+        except Exception:
+            pass
     if neq:
         # EXISTS { (n)--(m) WHERE m <> n } — for the single-edge shape, endpoint
         # inequality == "witnessed by a NON-self-loop edge": pre-drop self loops and

graphistry/tests/compute/gfql/index/test_index.py

@@ -416,3 +416,141 @@ def test_drop_index_if_exists_semantics():
     # and a resident index still drops through the plain form
     gi = g.create_index("edge_out_adj")
     assert gi.gfql("DROP GFQL INDEX FOR edge_out_adj").show_indexes().shape[0] == 0
+
+
+# --- get_degrees index fast path (#5 degree-cache / #3 membership) ---
+
+def _to_engine_frames(g, engine):
+    if engine == "cudf":
+        import cudf
+        return g.nodes(cudf.from_pandas(g._nodes), g._node).edges(cudf.from_pandas(g._edges), g._source, g._destination)
+    if engine in ("polars", "polars-gpu"):
+        import polars as pl
+        return g.nodes(pl.from_pandas(g._nodes), g._node).edges(pl.from_pandas(g._edges), g._source, g._destination)
+    return g
+
+
+def _degrees(g, engine):
+    if engine in ("polars", "polars-gpu"):
+        from graphistry.compute.gfql.lazy.engine.polars.degrees import get_degrees_polars
+        return get_degrees_polars(g, engine=engine)
+    return g.get_degrees()
+
+
+def _degcols(g_res):
+    nn = g_res._nodes
+    nn = nn.to_pandas() if hasattr(nn, "to_pandas") else nn
+    nn = nn.sort_values("id").reset_index(drop=True)
+    return {c: nn[c].tolist() for c in ("degree", "degree_in", "degree_out")}
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_get_degrees_index_parity(graph, engine):
+    g = _to_engine_frames(graph, engine)
+    base = _degcols(_degrees(g, engine))                       # group_by scan path
+    gi = g.gfql_index_all(engine=engine)
+    idx = _degcols(_degrees(gi, engine))                       # index fast path
+    assert base == idx
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_get_degrees_self_loops_and_isolated(graph, engine):
+    # graph with self-loops + isolated nodes (0..N-1; some have no edge)
+    import numpy as _np
+    ndf = pd.DataFrame({"id": _np.arange(300)})
+    edf = pd.DataFrame({"src": [0, 1, 2, 5, 5], "dst": [0, 2, 1, 5, 6]})  # self-loops at 0,5
+    g0 = graphistry.nodes(ndf, "id").edges(edf, "src", "dst")
+    g = _to_engine_frames(g0, engine)
+    base = _degcols(_degrees(g, engine))
+    gi = g.gfql_index_all(engine=engine)
+    idx = _degcols(_degrees(gi, engine))
+    assert base == idx
+    # self-loop double-count contract: node 0 (self-loop only) degree==2
+    assert idx["degree"][0] == 2
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_get_degrees_index_policy_off(graph, engine):
+    g = _to_engine_frames(graph, engine)
+    gi = g.gfql_index_all(engine=engine)
+    setattr(gi, "_gfql_index_policy", "off")
+    assert _degcols(_degrees(gi, engine)) == _degcols(_degrees(g, engine))
+
+
+# --- EXISTS { (n)--() } prune-isolated via #3 membership (polars fast path) ---
+
+@pytest.mark.parametrize("engine", [e for e in ENGINES if e in ("polars", "polars-gpu")])
+def test_exists_prune_membership_parity(engine):
+    import polars as pl
+    rng = np.random.default_rng(4)
+    N, E = 5000, 20000
+    src = rng.integers(0, N, E)
+    dst = rng.integers(0, N, E)
+    dst[:50] = src[:50]  # self-loops (membership must still include them)
+    g = graphistry.nodes(pl.DataFrame({"id": np.arange(N)}), "id").edges(
+        pl.DataFrame({"src": src, "dst": dst}), "src", "dst")
+    q = "MATCH (n) WHERE EXISTS { (n)--() } RETURN n.id AS id"
+    def ids(gg):
+        nn = gg.gfql(q, engine=engine)._nodes
+        nn = nn.to_pandas() if hasattr(nn, "to_pandas") else nn
+        return sorted(nn["id"].tolist())
+    base = ids(g)
+    gi = g.gfql_index_all(engine=engine)
+    assert ids(gi) == base
+    # policy='off' still correct
+    gi2 = g.gfql_index_all(engine=engine)
+    setattr(gi2, "_gfql_index_policy", "off")
+    assert ids(gi2) == base
+
+
+def test_get_degrees_polars_gpu_uses_matching_index_engine(monkeypatch):
+    pl = pytest.importorskip("polars")
+    from graphistry.Engine import Engine
+    import graphistry.compute.gfql.index.degrees as index_degrees
+    from graphistry.compute.gfql.lazy.engine.polars.degrees import get_degrees_polars
+
+    g = graphistry.nodes(pl.DataFrame({"id": [0, 1, 2]}), "id").edges(
+        pl.DataFrame({"src": [0, 1], "dst": [1, 2]}), "src", "dst")
+    gi = g.gfql_index_all(engine="polars-gpu")
+
+    seen = []
+    orig = index_degrees.degrees_from_index
+
+    def wrapped(registry, nodes_df, node_col, edges_df, cols, engine):
+        seen.append(engine)
+        return orig(registry, nodes_df, node_col, edges_df, cols, engine)
+
+    monkeypatch.setattr(index_degrees, "degrees_from_index", wrapped)
+    out = get_degrees_polars(gi, engine="polars-gpu")
+
+    assert Engine.POLARS_GPU in seen
+    assert "degree" in out._nodes.columns
+
+
+def test_exists_prune_membership_polars_gpu_uses_matching_index_engine(monkeypatch):
+    pl = pytest.importorskip("polars")
+    from graphistry.Engine import Engine
+    from graphistry.compute.ast import n, e_undirected, serialize_binding_ops
+    import graphistry.compute.gfql.index.degrees as index_degrees
+    from graphistry.compute.gfql.lazy import ExecutionTarget, target_mode
+    from graphistry.compute.gfql.lazy.engine.polars.pattern_apply import _pattern_alias_keys_polars
+
+    g = graphistry.nodes(pl.DataFrame({"id": [0, 1, 2]}), "id").edges(
+        pl.DataFrame({"src": [0, 1], "dst": [1, 2]}), "src", "dst")
+    gi = g.gfql_index_all(engine="polars-gpu")
+    binding_ops = serialize_binding_ops([n(name="n"), e_undirected(), n(name="m")])
+
+    seen = []
+    orig = index_degrees.adjacency_membership_keys
+
+    def wrapped(registry, direction, edges_df, cols, engine):
+        seen.append(engine)
+        return orig(registry, direction, edges_df, cols, engine)
+
+    monkeypatch.setattr(index_degrees, "adjacency_membership_keys", wrapped)
+    with target_mode(ExecutionTarget.GPU):
+        keys = _pattern_alias_keys_polars(gi, binding_ops, "n")
+
+    assert Engine.POLARS_GPU in seen
+    assert keys is not None
+    assert sorted(keys.get_column("id").to_list()) == [0, 1, 2]