From 2a6f9d9b1975aed71d6a4d9e264abff53e3143c8 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sun, 5 Jul 2026 18:26:06 -0700 Subject: [PATCH 01/10] perf(gfql/index): index-accelerated get_degrees (#5 degree-cache / #3 membership) Read node degrees straight from a resident CSR adjacency index (degree = diff(group_offsets), gathered per node via searchsorted) instead of the O(E) group_by(endpoint)+join. Bulk over all nodes -> always profitable when a valid index is resident; index_policy='off' skips; try/except falls back to the scan path (never wrong). New graphistry/compute/gfql/index/degrees.py (degrees_from_index), engine-native (numpy pandas/polars, cupy cudf). Wired into ComputeMixin.get_degrees (pandas/cudf). Prototype (1M/5M pandas): exists_prune 342->2.7ms (128x), q8 541->0.8ms (664x), exact parity. Unit + end-to-end parity green (arbitrary node ids + self-loops). WIP: polars get_degrees_polars wiring + index-suite parity tests + 4-engine dgx. Co-Authored-By: Claude Opus 4.8 (1M context) --- graphistry/compute/ComputeMixin.py | 28 ++++++++++++ graphistry/compute/gfql/index/degrees.py | 54 ++++++++++++++++++++++++ 2 files changed, 82 insertions(+) create mode 100644 graphistry/compute/gfql/index/degrees.py diff --git a/graphistry/compute/ComputeMixin.py b/graphistry/compute/ComputeMixin.py index 4edda70b6a..587342d709 100644 --- a/graphistry/compute/ComputeMixin.py +++ b/graphistry/compute/ComputeMixin.py @@ -321,6 +321,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") diff --git a/graphistry/compute/gfql/index/degrees.py b/graphistry/compute/gfql/index/degrees.py new file mode 100644 index 0000000000..3128814afe --- /dev/null +++ b/graphistry/compute/gfql/index/degrees.py @@ -0,0 +1,54 @@ +"""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 From af259b35a18572746f5dc5e5dbfeb0a45cbcf931 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sun, 5 Jul 2026 18:29:06 -0700 Subject: [PATCH 02/10] perf(gfql/index): wire index-accelerated get_degrees into polars engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same resident-CSR fast path (degrees_from_index) in get_degrees_polars — eager frames only (LazyFrame would force a collect), policy 'off' skips, try/except falls back to the group_by scan. Parity green vs group_by baseline (degree / degree_in / degree_out, arbitrary node ids + self-loops). Covers polars + polars-gpu (numpy-host arrays). cudf/polars-gpu GPU parity pending dgx verify. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../gfql/lazy/engine/polars/degrees.py | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/graphistry/compute/gfql/lazy/engine/polars/degrees.py b/graphistry/compute/gfql/lazy/engine/polars/degrees.py index 37b01ff3bf..0321d4b04d 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/degrees.py +++ b/graphistry/compute/gfql/lazy/engine/polars/degrees.py @@ -43,6 +43,30 @@ 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 + from graphistry.Engine import Engine + _reg = get_registry(g) + if not _reg.is_empty(): + _d = degrees_from_index(_reg, nodes, node_col, edges, (src, dst), Engine.POLARS) + 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) From 99c7ecf6084827991768bb2facf923825578bd4e Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sun, 5 Jul 2026 19:34:46 -0700 Subject: [PATCH 03/10] test(gfql/index): get_degrees index fast-path parity (4 engines) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Parity vs group_by scan + self-loop/isolated contract + index_policy='off' fallback, engine-parametrized. Local: pandas/polars/polars-gpu pass; cudf needs GPU (cupy init) — verified on dgx. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../tests/compute/gfql/index/test_index.py | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index 2d3c17913f..77a5f58a2c 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -503,3 +503,62 @@ 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) + 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)) From 6ef7083c01ebc58baea7f5b7e606d38b703d4f3f Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sun, 5 Jul 2026 19:38:44 -0700 Subject: [PATCH 04/10] perf(gfql/index): index-accelerate EXISTS {(n)--()} prune-isolated (#3 membership) Bare EXISTS pattern (no endpoint/edge filters, no drop-self neq) = "node has an edge in this direction" = CSR adjacency membership. Route _pattern_alias_keys_polars through adjacency_membership_keys (union of edge_out/in_adj keys_sorted for undirected; per-direction otherwise) instead of the O(E) chain_polars; strict guard falls through for any richer shape. New adjacency_membership_keys in index/degrees.py. Parity + policy-off tests. Perf (200k/800k, polars): scan 1206ms -> index 16.5ms (73x); parity exact incl self-loop-only nodes. Co-Authored-By: Claude Opus 4.8 (1M context) --- graphistry/compute/gfql/index/degrees.py | 28 +++++++++++++++ .../gfql/lazy/engine/polars/pattern_apply.py | 34 +++++++++++++++++++ .../tests/compute/gfql/index/test_index.py | 26 ++++++++++++++ 3 files changed, 88 insertions(+) diff --git a/graphistry/compute/gfql/index/degrees.py b/graphistry/compute/gfql/index/degrees.py index 3128814afe..c1e2647aa6 100644 --- a/graphistry/compute/gfql/index/degrees.py +++ b/graphistry/compute/gfql/index/degrees.py @@ -52,3 +52,31 @@ def degrees_from_index( 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": + return oi.keys_sorted + if direction == "reverse": + return ii.keys_sorted + xp, _ = array_namespace(engine) + return union1d(oi.keys_sorted, ii.keys_sorted, xp) diff --git a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py b/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py index ed23aa27bb..dc3481d96c 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py +++ b/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,39 @@ 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 + _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): + _mk = adjacency_membership_keys(_reg, _mdir, base_graph._edges, (_src, _dst), _Engine.POLARS) + 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 diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index 77a5f58a2c..e6d297779f 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -562,3 +562,29 @@ def test_get_degrees_index_policy_off(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 From a2708b8edf0d08417b6fcd857bf9695f6c0aef19 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 7 Jul 2026 17:16:03 -0700 Subject: [PATCH 05/10] fix(gfql): narrow adjacency membership index typing --- graphistry/compute/gfql/index/degrees.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/graphistry/compute/gfql/index/degrees.py b/graphistry/compute/gfql/index/degrees.py index c1e2647aa6..b404386a67 100644 --- a/graphistry/compute/gfql/index/degrees.py +++ b/graphistry/compute/gfql/index/degrees.py @@ -75,8 +75,12 @@ def adjacency_membership_keys( 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) From 592677bb61755c795056d78dc108f345e17272d2 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 7 Jul 2026 17:46:23 -0700 Subject: [PATCH 06/10] fix(gfql): honor polars-gpu index fast paths --- graphistry/compute/gfql/call/executor.py | 2 +- .../gfql/lazy/engine/polars/degrees.py | 15 +++-- .../gfql/lazy/engine/polars/pattern_apply.py | 4 +- .../tests/compute/gfql/index/test_index.py | 55 ++++++++++++++++++- 4 files changed, 69 insertions(+), 7 deletions(-) diff --git a/graphistry/compute/gfql/call/executor.py b/graphistry/compute/gfql/call/executor.py index 1b0b1a25c8..5aa14dddb2 100644 --- a/graphistry/compute/gfql/call/executor.py +++ b/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( diff --git a/graphistry/compute/gfql/lazy/engine/polars/degrees.py b/graphistry/compute/gfql/lazy/engine/polars/degrees.py index 0321d4b04d..429da096c1 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/degrees.py +++ b/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).""" @@ -51,10 +59,9 @@ def get_degrees_polars( try: from graphistry.compute.gfql.index import get_registry from graphistry.compute.gfql.index.degrees import degrees_from_index - from graphistry.Engine import Engine _reg = get_registry(g) if not _reg.is_empty(): - _d = degrees_from_index(_reg, nodes, node_col, edges, (src, dst), Engine.POLARS) + _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)] @@ -127,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()" diff --git a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py b/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py index dc3481d96c..bd9bec2758 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py +++ b/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py @@ -142,6 +142,7 @@ def _pattern_alias_keys_polars( 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") @@ -153,7 +154,8 @@ def _pattern_alias_keys_polars( _mdir = "reverse" _src, _dst = base_graph._source, base_graph._destination if isinstance(_src, str) and isinstance(_dst, str): - _mk = adjacency_membership_keys(_reg, _mdir, base_graph._edges, (_src, _dst), _Engine.POLARS) + _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: diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index e6d297779f..a737282455 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -520,7 +520,7 @@ def _to_engine_frames(g, engine): 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) + return get_degrees_polars(g, engine=engine) return g.get_degrees() @@ -588,3 +588,56 @@ def ids(gg): 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): + import polars as pl + 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): + import polars as pl + 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] From 4cdec5a9308b13a6c08882d53d7989fdccb75ff6 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Tue, 7 Jul 2026 17:56:06 -0700 Subject: [PATCH 07/10] test(gfql): skip polars-gpu observability without polars --- graphistry/tests/compute/gfql/index/test_index.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index a737282455..c8a48bf216 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -591,7 +591,7 @@ def ids(gg): def test_get_degrees_polars_gpu_uses_matching_index_engine(monkeypatch): - import polars as pl + 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 @@ -615,7 +615,7 @@ def wrapped(registry, nodes_df, node_col, edges_df, cols, engine): def test_exists_prune_membership_polars_gpu_uses_matching_index_engine(monkeypatch): - import polars as pl + 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 fa4bbe7776daae65789a014f005aa4c9a1af463f Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Wed, 8 Jul 2026 19:07:21 -0700 Subject: [PATCH 08/10] fix(gfql): tighten index degree typing --- graphistry/compute/ComputeMixin.py | 5 +- graphistry/compute/gfql/index/__init__.py | 4 +- graphistry/compute/gfql/index/api.py | 88 +++++++++++-------- graphistry/compute/gfql/index/degrees.py | 39 +++++--- .../gfql/lazy/engine/polars/degrees.py | 5 +- .../gfql/lazy/engine/polars/pattern_apply.py | 16 ++-- 6 files changed, 97 insertions(+), 60 deletions(-) diff --git a/graphistry/compute/ComputeMixin.py b/graphistry/compute/ComputeMixin.py index 587342d709..4d1d533fb7 100644 --- a/graphistry/compute/ComputeMixin.py +++ b/graphistry/compute/ComputeMixin.py @@ -326,7 +326,8 @@ def get_degrees( # 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": + 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 @@ -346,7 +347,7 @@ def get_degrees( col: (_nf[degree_in] + _nf[degree_out]).astype("int32"), }) return g.nodes(_nf, node_id) - except Exception: + except (AttributeError, ImportError, NotImplementedError, TypeError, ValueError): pass in_df = _degree_agg(g._edges, g._destination, degree_in, node_id) diff --git a/graphistry/compute/gfql/index/__init__.py b/graphistry/compute/gfql/index/__init__.py index 8ab8796d4f..7ee6cbf4f8 100644 --- a/graphistry/compute/gfql/index/__init__.py +++ b/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", diff --git a/graphistry/compute/gfql/index/api.py b/graphistry/compute/gfql/index/api.py index 0151e386c2..0855a90b0a 100644 --- a/graphistry/compute/gfql/index/api.py +++ b/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: DataFrameT, hops: Optional[int], direction: HopDirection, return_as_wave_front: bool, + to_fixed_point: bool = False, policy: Optional[str] = "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,11 +360,11 @@ 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: + except (AttributeError, TypeError, ValueError): pass def _bail(reason: str) -> Optional[Plottable]: @@ -367,27 +372,41 @@ def _bail(reason: str) -> Optional[Plottable]: _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") @@ -396,8 +415,8 @@ def _bail(reason: str) -> Optional[Plottable]: 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,7 +438,7 @@ 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: @@ -427,15 +446,14 @@ def _bail(reason: str) -> Optional[Plottable]: 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, diff --git a/graphistry/compute/gfql/index/degrees.py b/graphistry/compute/gfql/index/degrees.py index b404386a67..5e4e940cbc 100644 --- a/graphistry/compute/gfql/index/degrees.py +++ b/graphistry/compute/gfql/index/degrees.py @@ -9,14 +9,27 @@ """ from __future__ import annotations -from typing import Any, Optional, Tuple +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 EDGE_OUT_ADJ, EDGE_IN_ADJ, GfqlIndexRegistry +from .registry import AdjacencyIndex, EDGE_OUT_ADJ, EDGE_IN_ADJ, GfqlIndexRegistry +from .types import AdjacencyIndexKind, HopDirection -def _degree_for_nodes(adj_index: Any, node_ids: Any, xp: Any) -> Any: +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 @@ -27,7 +40,7 @@ def _degree_for_nodes(adj_index: Any, node_ids: Any, xp: Any) -> Any: 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): + if f.dtype != keys.dtype: common = xp.promote_types(f.dtype, keys.dtype) f = f.astype(common) keys = keys.astype(common) @@ -38,13 +51,13 @@ def _degree_for_nodes(adj_index: Any, node_ids: Any, xp: Any) -> Any: def degrees_from_index( - registry: GfqlIndexRegistry, nodes_df: Any, node_col: str, - edges_df: Any, cols: Tuple[str, ...], engine: Engine, -) -> Optional[Tuple[Any, Any]]: + 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 = registry.get_valid(EDGE_OUT_ADJ, edges_df, cols, engine) - ii = registry.get_valid(EDGE_IN_ADJ, edges_df, cols, engine) + 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) @@ -55,8 +68,8 @@ def degrees_from_index( def adjacency_membership_keys( - registry: GfqlIndexRegistry, direction: str, edges_df: Any, cols: Tuple[str, ...], engine: Engine, -) -> Optional[Any]: + 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); @@ -67,11 +80,11 @@ def adjacency_membership_keys( """ from .engine_arrays import union1d if direction in ("forward", "undirected"): - oi = registry.get_valid(EDGE_OUT_ADJ, edges_df, cols, engine) + oi = _valid_adjacency(registry, 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) + ii = _valid_adjacency(registry, EDGE_IN_ADJ, edges_df, cols, engine) if ii is None: return None if direction == "forward": diff --git a/graphistry/compute/gfql/lazy/engine/polars/degrees.py b/graphistry/compute/gfql/lazy/engine/polars/degrees.py index 429da096c1..38e5bc41e8 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/degrees.py +++ b/graphistry/compute/gfql/lazy/engine/polars/degrees.py @@ -55,7 +55,8 @@ def get_degrees_polars( # 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": + from graphistry.compute.gfql.index import get_index_policy + if not is_lazy(nodes) and not is_lazy(edges) and get_index_policy(g) != "off": try: from graphistry.compute.gfql.index import get_registry from graphistry.compute.gfql.index.degrees import degrees_from_index @@ -71,7 +72,7 @@ def get_degrees_polars( 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: + except (AttributeError, ImportError, NotImplementedError, TypeError, ValueError): pass node_dt = col_dtype(nodes, node_col) diff --git a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py b/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py index bd9bec2758..6cb2d79574 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py +++ b/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py @@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Dict, List, Optional, Sequence from graphistry.utils.json import JSONVal +from graphistry.compute.gfql.index.types import HopDirection from graphistry.Plottable import Plottable @@ -129,12 +130,14 @@ def _pattern_alias_keys_polars( # 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. + from graphistry.compute.gfql.index import get_index_policy 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 get_index_policy(g) != "off" + and not n0.filter_dict and not n2.filter_dict + and not edge_op.edge_match + and edge_op.edge_query is None + and base_graph._edges is not None and not is_lazy(base_graph._edges) ): try: @@ -145,7 +148,8 @@ def _pattern_alias_keys_polars( 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") + _edir = edge_op.direction + _mdir: HopDirection if _edir == "undirected": _mdir = "undirected" elif (alias == n0._name) == (_edir == "forward"): @@ -158,7 +162,7 @@ def _pattern_alias_keys_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: + except (AttributeError, ImportError, NotImplementedError, TypeError, ValueError): pass if neq: # EXISTS { (n)--(m) WHERE m <> n } — for the single-edge shape, endpoint From 370722a8de0eb76c52793cd94192c4bede4a15fa Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Wed, 8 Jul 2026 19:40:33 -0700 Subject: [PATCH 09/10] fix(gfql): type index policy handoff --- graphistry/compute/gfql/index/api.py | 8 +++++--- graphistry/compute/gfql/lazy/engine/polars/chain.py | 3 ++- graphistry/compute/gfql/lazy/engine/polars/hop.py | 4 ++-- graphistry/compute/hop.py | 4 ++-- 4 files changed, 11 insertions(+), 8 deletions(-) diff --git a/graphistry/compute/gfql/index/api.py b/graphistry/compute/gfql/index/api.py index 0855a90b0a..bc6f751e5a 100644 --- a/graphistry/compute/gfql/index/api.py +++ b/graphistry/compute/gfql/index/api.py @@ -340,8 +340,8 @@ def _ensure_indexes( def maybe_index_hop( - g: Plottable, engine: Engine, *, nodes: DataFrameT, hops: Optional[int], direction: HopDirection, return_as_wave_front: bool, - to_fixed_point: bool = False, policy: Optional[str] = "use", **rest: object, + 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. @@ -363,7 +363,8 @@ def maybe_index_hop( "policy": resolved_policy, "engine": engine.value, } try: - diag["frontier_n"] = int(nodes.shape[0]) + if nodes is not None: + diag["frontier_n"] = int(nodes.shape[0]) except (AttributeError, TypeError, ValueError): pass @@ -409,6 +410,7 @@ def _bail(reason: str) -> Optional[Plottable]: 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 diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index 377311aa30..463da88112 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -578,7 +578,8 @@ def _plain_edge(op): # `MATCH (a {id-filter})-[e]->(b)` (forward/reverse, no destination filter) — # the canonical seeded query. This native chain fast path does its own O(E) # semi-join, so it must consult the index here too (not just compute/hop.py). - _idx_pol = getattr(self, "_gfql_index_policy", "use") + from graphistry.compute.gfql.index import get_index_policy + _idx_pol = get_index_policy(self) if (start_nodes is None and len(ops) == 3 and _fp_node(ops[0]) and _plain_edge(ops[1]) and _fp_node(ops[2]) and ops[0].filter_dict and not ops[2].filter_dict and ops[1].direction in ("forward", "reverse")): diff --git a/graphistry/compute/gfql/lazy/engine/polars/hop.py b/graphistry/compute/gfql/lazy/engine/polars/hop.py index d365b0e2c5..6c9e91f9c7 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/hop.py +++ b/graphistry/compute/gfql/lazy/engine/polars/hop.py @@ -17,8 +17,8 @@ def hop_lazy_or_eager(self: Plottable, nodes: Optional[Any] = None, hops: Option # GFQL physical index fast path (pay-as-you-go). chain_polars reaches the # polars hop here without passing through compute/hop.py, so the index hook # lives here too to cover polars gfql() chains (not just direct .hop()). - from graphistry.compute.gfql.index import get_registry, maybe_index_hop - _pol = getattr(self, "_gfql_index_policy", "use") + from graphistry.compute.gfql.index import get_index_policy, get_registry, maybe_index_hop + _pol = get_index_policy(self) if (not get_registry(self).is_empty()) or _pol in ("auto", "force"): from graphistry.Engine import Engine from graphistry.compute.gfql.lazy import active_target, ExecutionTarget diff --git a/graphistry/compute/hop.py b/graphistry/compute/hop.py index 853965c523..28e6b1317c 100644 --- a/graphistry/compute/hop.py +++ b/graphistry/compute/hop.py @@ -114,9 +114,9 @@ def hop(self: Plottable, # the O(degree) CSR gather instead of the O(E) scan below. Engine-uniform; # returns None to fall back. Coercion above is a no-op when already in-engine, # so the index fingerprint (keyed on the live edge frame) still matches. - from graphistry.compute.gfql.index import get_registry, maybe_index_hop + from graphistry.compute.gfql.index import get_index_policy, get_registry, maybe_index_hop from graphistry.compute.gfql.index.types import HopDirection - _idx_policy = getattr(self, "_gfql_index_policy", "use") + _idx_policy = get_index_policy(self) if (not get_registry(self).is_empty()) or _idx_policy in ("auto", "force"): _idx_nodes = df_to_engine(nodes, engine_concrete) if nodes is not None else None _indexed = maybe_index_hop( From 9cc408592ede6a9e529471262d00f8cc3330feb2 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Wed, 8 Jul 2026 19:49:33 -0700 Subject: [PATCH 10/10] fix(gfql): tighten polars index fast paths --- .../gfql/lazy/engine/polars/degrees.py | 54 +++++++++---------- .../gfql/lazy/engine/polars/pattern_apply.py | 45 ++++++++-------- 2 files changed, 47 insertions(+), 52 deletions(-) diff --git a/graphistry/compute/gfql/lazy/engine/polars/degrees.py b/graphistry/compute/gfql/lazy/engine/polars/degrees.py index 38e5bc41e8..2c29ac2125 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/degrees.py +++ b/graphistry/compute/gfql/lazy/engine/polars/degrees.py @@ -6,15 +6,15 @@ """ from typing import Optional -from graphistry.Engine import Engine +from graphistry.Engine import Engine, EngineAbstract, EngineAbstractType, resolve_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) +def _index_engine(engine: Optional[EngineAbstractType] = None) -> Engine: + if engine is not None and engine not in (EngineAbstract.AUTO, "auto"): + return resolve_engine(engine) from graphistry.compute.gfql.lazy import active_target, ExecutionTarget return Engine.POLARS_GPU if active_target() == ExecutionTarget.GPU else Engine.POLARS @@ -33,15 +33,16 @@ def get_degrees_polars( col: str = "degree", degree_in: str = "degree_in", degree_out: str = "degree_out", - engine: Optional[str] = None, + engine: Optional[EngineAbstractType] = None, ) -> Plottable: """Native ``get_degrees`` — parity with ComputeMixin.get_degrees. group_by-count per endpoint, left-joined onto the node table (materialized from edges when absent). Oracle contract: isolated and src-only/dst-only nodes get 0; self-loops double-count (one in + one out); all three columns Int32. Empty edges need no special case — left-join + - fill_null(0) yields all-zero, matching the pandas empty-edges branch. The ``engine`` safelist - sub-param is accepted and ignored (execution already committed to polars, parity-equal). + fill_null(0) yields all-zero, matching the pandas empty-edges branch. The ``engine`` + sub-param selects the resident index engine when an index fast path is available; + the scan path is already committed to polars execution. """ import polars as pl @@ -53,27 +54,24 @@ def get_degrees_polars( # 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). + # (LazyFrame get_column would force a collect); policy 'off' skips. Missing + # or stale indexes return None and fall through; real errors should surface. from graphistry.compute.gfql.index import get_index_policy if not is_lazy(nodes) and not is_lazy(edges) and 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, 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 (AttributeError, ImportError, NotImplementedError, TypeError, ValueError): - pass + 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) node_dt = col_dtype(nodes, node_col) in_counts = _endpoint_counts(edges, dst, node_dt, node_col, degree_in) @@ -135,14 +133,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", engine: Optional[str] = None) -> Plottable: +def get_indegrees_polars(g: Plottable, col: str = "degree_in", engine: Optional[EngineAbstractType] = 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", engine: Optional[str] = None) -> Plottable: +def get_outdegrees_polars(g: Plottable, col: str = "degree_out", engine: Optional[EngineAbstractType] = 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()" diff --git a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py b/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py index 6cb2d79574..2b2b7e36e4 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py +++ b/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py @@ -140,30 +140,27 @@ def _pattern_alias_keys_polars( and base_graph._edges is not 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 = edge_op.direction - _mdir: HopDirection - 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 (AttributeError, ImportError, NotImplementedError, TypeError, ValueError): - pass + 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 = edge_op.direction + _mdir: HopDirection + 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))}) 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