diff --git a/CHANGELOG.md b/CHANGELOG.md index ef993819b4..bc04cee26b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - **GFQL secondary node property indexes (`create_index('node_prop', column=...)` / `g.gfql_index_node_props([...])`)**: a seed predicate on a NON-key column — `MATCH (m {id: 42})` where the graph's node id binding is some other column — previously cost a full node scan, because the registry only indexed the node-id binding and the CSR adjacencies. A property index is the same pay-as-you-go sidecar as the existing kinds: sorted distinct values over node **row positions** (CSR, so duplicate values are indexable), never reorders `.nodes`, fingerprint-validated so a `.nodes()` rebind is treated as absent (safe miss, never a wrong answer), engine-polymorphic (numpy host / cupy on-device), and policy-gated (`off`/`use`/`auto`/`force`). The seeded fixed-hop planner picks the **most selective** indexed scalar predicate in the seed filter using a free CSR-offset estimate, gathers those candidates, and applies every remaining predicate to them — so results are identical whether the index is present, absent, stale, or cost-gated out. `show_indexes()` lists property indexes; `drop_index('node_prop', column=...)` drops one. Only integer columns are indexable today (float NaN ordering, strings on cupy, and nulls all decline to the scan); widening that is additive. **Perf (dgx-spark, official LDBC SNB SF1, 3.18M nodes, warm median, value-identical 19-row result):** interactive-short IS7 `71.6 ms -> 19.5 ms` (**3.7x**), with a one-time `112 ms` build — the seed lookup itself goes from a `51.2 ms` scan to `0.096 ms`. ### Performance +- **GFQL indexed typed-edge traversals stop scanning the whole edge frame (#1658)**: the index path built the simple-equality `edge_match` mask over ALL `E` edges — one `(series == val)` over the full column — and then read that mask only at `rows[edge_keep[rows]]`, the handful of positions the CSR adjacency lookup had already returned. That put an **O(E) predicate scan inside an O(degree) traversal**, which is why an indexed seeded typed hop scaled with the graph while the underlying `g.hop()` stayed flat. The predicate is now evaluated on the gathered candidate rows instead: `_build_edge_row_filter` does the schema-level validation up front (the dtype-mismatch decline that keeps error parity with the scan is O(1) and unchanged) and `_EdgeMatchRowFilter.mask_for(rows)` applies `col == val` to just those rows, using each frame's native `==` exactly as before (so cuDF string columns stay on the cuDF layer). This is a cost tradeoff, not a free win, and it is bounded rather than assumed: candidate-row evaluation beats one whole-column compare only while the candidates stay a small fraction of the frame, and a fixed-point walk that reaches most of the graph inverts it — the out- and in-indices are filtered separately, so it can gather up to 2E elements by random access against the eager form's single sequential pass over E (measured before the guard: an undirected `to_fixed_point` typed walk gathered 1.94×E and ran 1.2–1.6× SLOWER than master). A cumulative-cost guard now switches to the whole-column mask once gathered rows reach E/8, so total predicate work is bounded at ~1.125×E in that regime while a seeded hop — which gathers ~degree — never approaches the threshold and never builds it. An evaluation failure mid-traversal abandons the indexed path entirely so the scan fallback stays parity-safe. Measured on a 3.18M-node / 14M-edge polars graph (LDBC SNB SF1-shaped), a seeded typed-edge query with alias markers goes **309.9 → 57.4 ms (5.4×)**, value-identical, with the full-column compare (previously 248 ms, 81% of the query) gone from the profile. Parity held across 650 differential comparisons — pandas + polars × 13 traversal shapes × 25 random seeds on a dense graph with null-carrying string and numeric columns — comparing columns, dtypes, row counts and full content, including the decline paths (membership `edge_match` still routes to the scan; a dtype mismatch still raises the same `GFQLSchemaError`). One honest limit on error parity: only the O(1) dtype gates are parity-exact. A *data-dependent* comparison failure (e.g. a list-valued cell in an object column) is now observed only if it lands in the gathered candidates, so the indexed path can succeed where the scan raises. That needs pathological cell values to reach, but it is a real narrowing of the previous guarantee. Pinned by two structural regression tests that assert the *shape* rather than a wall-clock number: the predicate must see only candidate rows, and growing the graph 8× at fixed degree must not grow the number of rows it examines. - **GFQL indexed fixed-hop bindings now dispatch BEFORE the canonical traversal (pandas / cuDF / native Polars)**: the resident-index fixed-hop path (`rows(binding_ops=...)`, the Cypher multi-alias `MATCH … RETURN` shape) was consulted only *inside* row materialization — after the engine had already run the full canonical traversal — so its compact path bag was computed on top of the graph-sized work it exists to avoid. Both the pandas/cuDF chain boundary and the native Polars chain now ask the same shared, structural gate first and, only when it can serve the WHOLE middle exactly, skip the canonical traversal and hand the compact state to the unchanged row materializer. Engagement stays operator/index/dtype/cost based — no query, schema, or hop-count recognition — and every unsupported, seeded, prefiltered, policy-bearing, shortest-path, or cost-gated shape still declines to canonical execution with identical results. On a standard LDBC SNB SF1 interactive-short profile this removed the redundant two-hop typed-mask and 16-merge buckets and cut the profiled call ~11.9× (1263 ms → 106 ms), exact 19-row oracle preserved. - **Indexed seed lookup now covers a unique node id PLUS extra scalar constraints**: `{node_id: v, other: w}` previously fell back to a full node-table scan even though the unique node-id index can gather at most one row; it now gathers that one row and applies the remaining constraints to it. Missing ids and non-matching extra constraints return the same empty result as before, and duplicate-id graphs (which cannot build the index) are unaffected. - **Seeded typed-hop property projection keeps its fast path through DISTINCT / ORDER BY / SKIP / LIMIT**: those trailing row ops are plain frame operations, so the fast path now delegates them to the canonical chain instead of declining the whole query. diff --git a/graphistry/compute/gfql/index/traverse.py b/graphistry/compute/gfql/index/traverse.py index 17f5b8773e..accc4e1e60 100644 --- a/graphistry/compute/gfql/index/traverse.py +++ b/graphistry/compute/gfql/index/traverse.py @@ -14,12 +14,12 @@ """ from __future__ import annotations -from typing import Any, List, Optional, Tuple, cast +from typing import Dict, List, Optional, Tuple, cast from typing_extensions import TypeGuard from graphistry.Engine import Engine -from graphistry.compute.typing import DataFrameT +from graphistry.compute.typing import DataFrameT, SeriesT from graphistry.Plottable import Plottable from .engine_arrays import ( array_namespace, col_to_array, ids_to_array, take_rows, select_by_ids, @@ -27,7 +27,33 @@ ) from .lookup import lookup_edge_rows, lookup_node_rows from .registry import EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, AdjacencyIndex, GfqlIndexRegistry, NodeIdIndex -from .types import ArrayLike, EdgeMatch, HopDirection, SimpleEqualityEdgeMatch +from .types import ( + ArrayLike, EdgeMatch, HopDirection, ScalarMatchValue, SimpleEqualityEdgeMatch, +) + +# Cost guard for candidate-row edge_match evaluation. Gathering candidate rows beats one +# whole-column compare only while the candidates stay a small fraction of the frame; a +# fixed-point walk that reaches most of the graph inverts that and gathers up to 2E by +# random access. Once cumulative gathered rows reach E/DIVISOR we build the whole-column +# mask once and reuse it, bounding total predicate work at ~(1 + 1/DIVISOR)*E. The FLOOR +# keeps small frames on the candidate-row path, where the whole-column compare is cheap +# anyway and the switch would only add a branch. +_EAGER_MASK_SWITCH_DIVISOR = 8 +_EAGER_MASK_SWITCH_FLOOR = 1024 + + +def _candidate_edge_mask_enabled() -> bool: + """Candidate-row ``edge_match`` evaluation is on by default; set + ``GFQL_INDEX_CANDIDATE_EDGE_MASK=0`` to force the whole-column mask on every hop. + + Follows the ``GFQL_LEAN_COMBINE`` precedent: the BOUNDARY is externally switchable so + the differential harness can exercise both sides of it and assert they agree, while the + numeric thresholds above stay private module constants like ``_LEAN_SHRINK_RATIO`` — + they are a cost heuristic, not an interface, and the guard already bounds the bad case. + """ + import os as _os + + return _os.environ.get("GFQL_INDEX_CANDIDATE_EDGE_MASK", "1") != "0" def _indices_for_direction( @@ -74,16 +100,123 @@ def is_simple_equality_edge_match( return True -def _build_edge_keep_mask( - edges: DataFrameT, edge_match: EdgeMatch, engine: Engine, xp: "object" -) -> Optional[ArrayLike]: - """Boolean array over ORIGINAL edge rows (length E, same indexing as - ``AdjacencyIndex.other_values`` / ``row_positions``) selecting rows that satisfy - a simple-equality ``edge_match``. +class _EdgeMatchRowFilter: + """Evaluates a simple-equality ``edge_match`` on the CSR-matched edge rows only. + + The mask is read exactly once per hop, as ``rows[keep[rows]]`` — at the handful of + positions the adjacency lookup returned. Materializing it over all E edges first + therefore put an O(E) predicate scan inside an O(degree) traversal, which is what + made the indexed path scale with the graph instead of with the answer. Evaluating + ``col == val`` on the gathered candidate rows makes the predicate proportional to + the edges the traversal actually visits, so a seeded hop examines O(edges traversed) + elements. + + A row is *mostly* returned once per index — frontiers are set-differenced against + ``visited`` — but not strictly: ``edge_match`` is only reachable with + ``return_as_wave_front=True``, and that mode skips the first-hop ``visited`` seeding + below, so seed ids can re-enter a later frontier and their rows be gathered twice. + The worst case is a fixed-point undirected walk reaching the whole graph, where the + out- and in-indices are filtered separately and the gathered total approaches 2E + against the eager form's single sequential pass over E. That regime is a genuine + REGRESSION for this form (measured: 1.94×E gathered, 1.2–1.6× slower than the eager + mask), which is why the caller keeps a cumulative-gathered counter and falls back to + ``full_mask()`` once it crosses a fraction of the frame. + + Column values are compared with each frame's native ``==`` (so cudf string columns + stay on the cudf layer rather than becoming a cupy string compare), matching the + eager form exactly. + """ + + __slots__ = ("_series", "_items", "_engine") + + # Typed slots: the per-column edge Series keyed by column name, the validated + # (column, scalar) equalities in ``edge_match`` order, and the frame engine. + _series: Dict[str, SeriesT] + _items: List[Tuple[str, ScalarMatchValue]] + _engine: Engine - Built via each frame's native ``col == val`` (so cudf string columns stay on the - cudf layer instead of a cupy string compare). Returns ``None`` on ANY unexpected - shape or error, so the caller falls back to scan rather than risk a divergence. + def __init__( + self, + series: Dict[str, SeriesT], + items: List[Tuple[str, ScalarMatchValue]], + engine: Engine, + ) -> None: + self._series = series + self._items = items + self._engine = engine + + def mask_for(self, rows: ArrayLike) -> Optional[ArrayLike]: + """Boolean array over ``rows`` (positional, same order), or ``None`` on any + unexpected shape/error so the caller falls back to the scan.""" + try: + mask: Optional[ArrayLike] = None + for col, val in self._items: + sub = _gather_series(self._series[col], rows, self._engine) + col_mask: ArrayLike + # Null-safe materialization: on null-carrying columns (pandas nullable + # Int64/boolean/string, polars nulls — which the NaN->null coercion + # makes common) a bare == yields NA cells, and to_numpy() then produces + # an OBJECT-dtype array that later explodes at rows[keep] (IndexError: + # not int/bool). Null == val filters out on the scan path, so fill + # False is parity-exact. + if self._engine in (Engine.POLARS, Engine.POLARS_GPU): + col_mask = (sub == val).fill_null(False).to_numpy() + elif self._engine == Engine.CUDF: + col_mask = (sub == val).fillna(False).values + else: + col_mask = (sub == val).fillna(False).to_numpy(dtype=bool) + mask = col_mask if mask is None else mask & col_mask + return mask + except Exception: # pragma: no cover - defensive parity guard + return None + + def full_mask(self) -> Optional[ArrayLike]: + """The eager whole-column mask, length E — the pre-candidate-row form. + + Candidate-row evaluation is a win exactly while the candidates are a small + fraction of the frame. A fixed-point walk that reaches most of the graph inverts + that: it gathers up to 2E elements (out- and in-indices filtered separately), by + random access, versus one sequential compare over E. The caller switches to this + once it has gathered enough to know it is in that regime; see + ``_EAGER_MASK_SWITCH_DIVISOR``. + """ + try: + mask: Optional[ArrayLike] = None + for col, val in self._items: + col_mask: ArrayLike + series = self._series[col] + # Same null-safe materialization as mask_for, so the two forms agree + # cell for cell — this is the parity-critical property of the switch. + if self._engine in (Engine.POLARS, Engine.POLARS_GPU): + col_mask = (series == val).fill_null(False).to_numpy() + elif self._engine == Engine.CUDF: + col_mask = (series == val).fillna(False).values + else: + col_mask = (series == val).fillna(False).to_numpy(dtype=bool) + mask = col_mask if mask is None else mask & col_mask + return mask + except Exception: # pragma: no cover - defensive parity guard + return None + + +def _gather_series(series: SeriesT, rows: ArrayLike, engine: Engine) -> SeriesT: + """Positionally gather ``rows`` out of a single column. O(len(rows)).""" + if engine in (Engine.POLARS, Engine.POLARS_GPU): + import numpy as np + + return series.gather(np.asarray(rows)) + # pandas / cudf: positional take accepts numpy (pandas) or cupy (cudf) int arrays + return series.take(rows) + + +def _build_edge_row_filter( + edges: DataFrameT, edge_match: EdgeMatch, engine: Engine +) -> Optional[_EdgeMatchRowFilter]: + """Validate a simple-equality ``edge_match`` against the edge schema and return a + per-row evaluator, or ``None`` when the shape isn't covered (caller falls back to + the scan rather than risk a divergence). + + All checks here are schema-level (O(1) in E); no predicate is evaluated yet. """ try: if not is_simple_equality_edge_match(edge_match): @@ -92,41 +225,31 @@ def _build_edge_keep_mask( _is_numeric_dtype_safe, _is_string_dtype_safe, ) n_edges = int(edges.shape[0]) - mask: Optional[ArrayLike] = None + series: Dict[str, SeriesT] = {} + items: List[Tuple[str, ScalarMatchValue]] = [] for col, val in edge_match.items(): if col not in edges.columns: return None + col_series: SeriesT if engine in (Engine.POLARS, Engine.POLARS_GPU): - series = edges.get_column(col) + col_series = edges.get_column(col) else: - series = edges[col] + col_series = edges[col] # Obvious dtype mismatch (numeric col vs str val, string col vs numeric # val): the scan raises GFQLSchemaError E302 where a naive == is silently # all-False. Decline -> caller falls back to the scan, which raises the # SAME error (parity-exact; mirrors filter_by_dict's exact two checks, # skipped like the scan on empty frames). if n_edges > 0: - dt = series.dtype + dt = col_series.dtype if _is_numeric_dtype_safe(dt) and isinstance(val, str): return None if (_is_string_dtype_safe(dt) and isinstance(val, (int, float)) and not isinstance(val, bool)): return None - # Null-safe materialization: on null-carrying columns (pandas nullable - # Int64/boolean/string, polars nulls — which the NaN->null coercion makes - # common) a bare == yields NA cells, and to_numpy() then produces an - # OBJECT-dtype array that later explodes at rows[edge_keep[rows]] - # (IndexError: not int/bool). Null == val filters out on the scan path, - # so fill False is parity-exact. - if engine in (Engine.POLARS, Engine.POLARS_GPU): - col_mask = cast(ArrayLike, (series == val).fill_null(False).to_numpy()) - elif engine == Engine.CUDF: - col_mask = cast(ArrayLike, (series == val).fillna(False).values) - else: - col_mask = cast( - ArrayLike, (series == val).fillna(False).to_numpy(dtype=bool)) - mask = col_mask if mask is None else cast(ArrayLike, cast(Any, mask) & cast(Any, col_mask)) - return mask + series[col] = col_series + items.append((col, val)) + return _EdgeMatchRowFilter(series, items, engine) except Exception: # pragma: no cover - defensive parity guard return None @@ -165,15 +288,28 @@ def index_seeded_hop( xp, _backend = array_namespace(engine) - # Typed-edge (edge_match) support: a boolean mask over ORIGINAL edge rows that - # pass the match predicate, applied to the CSR-matched rows each hop. Gated to - # simple scalar equality + the wavefront path by the coverability check upstream - # (maybe_index_hop); an unsupported shape returns None here => scan (parity-safe). - edge_keep: Optional[ArrayLike] = None + # Typed-edge (edge_match) support: the match predicate is evaluated on the + # CSR-matched rows of each hop, so it costs O(edges visited) rather than O(E). + # Gated to simple scalar equality + the wavefront path by the coverability check + # upstream (maybe_index_hop); an unsupported shape returns None here => scan + # (parity-safe). Schema validation happens now, up front, so an uncovered + # edge_match still declines before any traversal work. + edge_filter: Optional[_EdgeMatchRowFilter] = None if edge_match: - edge_keep = _build_edge_keep_mask(edges, edge_match, engine, xp) - if edge_keep is None: + edge_filter = _build_edge_row_filter(edges, edge_match, engine) + if edge_filter is None: return None + # Cost guard for the candidate-row form (see _EdgeMatchRowFilter.full_mask). Cumulative + # gathered rows; once they reach a fraction of the frame we are demonstrably NOT in the + # seeded regime, so we pay for the whole-column mask once and reuse it. Bounds total + # predicate work at ~(1 + 1/D)*E instead of the unbounded-in-hops gather, while a seeded + # hop — which gathers ~degree — never comes close to the threshold and never builds it. + gathered_rows = 0 + eager_keep: Optional[ArrayLike] = None + switch_at = ( + max(_EAGER_MASK_SWITCH_FLOOR, len(edges) // _EAGER_MASK_SWITCH_DIVISOR) + if _candidate_edge_mask_enabled() else 0 # 0 => build the whole-column mask up front + ) # Do NOT narrow the seed to the index key dtype (a node-id int64 seed cast to # an int32 edge-endpoint key wraps large ids → false match). lookup promotes both @@ -198,11 +334,29 @@ def index_seeded_hop( neigh_parts: List[ArrayLike] = [] for ix in indices: rows, matched = lookup_edge_rows(ix, frontier, xp) - if edge_keep is not None: + if edge_filter is not None: # Keep only CSR-matched rows whose edge passes edge_match. Wavefront- # only (coverability gate), so the `matched`/first-hop `visited` # bookkeeping below — which edge_match does NOT filter — is never read. - rows = rows[edge_keep[rows]] + # Decide BEFORE gathering this batch, not after: a single hop's batch can + # be arbitrarily large, so a post-hoc check would overshoot by up to one + # whole batch. Checking the projected total keeps gathered <= switch_at. + if (eager_keep is None + and gathered_rows + int(rows.shape[0]) > switch_at): + eager_keep = edge_filter.full_mask() + if eager_keep is None: + return None + if eager_keep is not None: + rows = rows[eager_keep[rows]] + else: + gathered_rows += int(rows.shape[0]) + keep = edge_filter.mask_for(rows) + if keep is None: + # Evaluation failed on this candidate batch: abandon the indexed + # path entirely so the caller re-runs the hop on the scan. Nothing + # observable has been mutated, so this stays parity-safe. + return None + rows = rows[keep] edge_rows_parts.append(rows) neigh_parts.append(ix.other_values[rows]) matched_parts.append(matched) diff --git a/graphistry/compute/typing.py b/graphistry/compute/typing.py index 3819452158..46bfe49a2b 100644 --- a/graphistry/compute/typing.py +++ b/graphistry/compute/typing.py @@ -50,6 +50,12 @@ def __gt__(self, other: Any) -> "ArrayLike": def __invert__(self) -> "ArrayLike": ... + def __and__(self, other: Any) -> "ArrayLike": + ... + + def __rand__(self, other: Any) -> "ArrayLike": + ... + def __add__(self, other: Any) -> "ArrayLike": ... diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index baab166c2f..8ac7bee5ce 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -884,3 +884,293 @@ def test_hop_dtype_mismatch_edge_match_matches_scan_error(typed_graph, engine): if engine == "pandas": from graphistry.compute.exceptions import GFQLSchemaError assert isinstance(base_err.value, GFQLSchemaError) + + +# ---- typed-edge predicate cost is proportional to the traversal, not the graph ------- +# Regression (measured 2026-07-26, LDBC SNB SF1 lane): the index path built the +# edge_match mask over ALL E edges (`(series == val)`) and then read it at only the +# handful of CSR-matched rows, putting an O(E) predicate scan inside an O(degree) +# traversal. On a 14M-edge graph that single compare was 248ms of a 308ms query, and +# it is why the indexed surface SCALED with the graph while the native hop stayed flat. +# These tests pin the SHAPE (predicate sees only candidate rows), not a wall-clock +# number, so they can't go flaky on a loaded host. + +@pytest.mark.parametrize("engine", _cpu_engines()) +def test_typed_edge_predicate_only_reads_candidate_rows(typed_graph, engine, monkeypatch): + """The edge_match predicate must be evaluated on the CSR-matched rows only. + + A one-seed hop touches ~deg edges out of 12000; if the predicate is ever handed + the whole column again this assertion fails loudly. + """ + from graphistry.Engine import Engine as _E, df_to_engine + from graphistry.compute.gfql.index import traverse as _traverse + + g = typed_graph + if engine == "polars": + g = g.edges(df_to_engine(g._edges, _E.POLARS), "src", "dst").nodes( + df_to_engine(g._nodes, _E.POLARS), "id") + n_edges = int(g._edges.shape[0]) + gi = g.gfql_index_all(engine=engine) + + seen = [] + orig = _traverse._gather_series + + def spy(series, rows, eng): + seen.append(int(len(rows))) + return orig(series, rows, eng) + + monkeypatch.setattr(_traverse, "_gather_series", spy) + seeds = g._nodes[:1] if engine == "pandas" else g._nodes.head(1) + kwargs = dict(hops=1, return_as_wave_front=True, edge_match={"etype": 1}, engine=engine) + idx = gi.hop(nodes=seeds, **kwargs) + base = g.hop(nodes=seeds, **kwargs) + + assert seen, "edge_match predicate never ran on the index path" + assert int(idx._edges.shape[0]) == int(base._edges.shape[0]) + # Proportional to the traversal: a single seed's out-degree, not the edge count. + assert max(seen) < n_edges // 10, ( + f"predicate saw {max(seen)} rows of {n_edges} — it is scanning the graph, " + "not the traversal candidates") + + +@pytest.mark.parametrize("engine", _cpu_engines()) +def test_typed_edge_predicate_cost_flat_in_graph_size(engine): + """Growing the graph 8x while holding degree fixed must NOT grow the number of + rows the edge_match predicate examines. This is the property that makes the + indexed path O(degree); it is engine- and schema-independent.""" + from graphistry.Engine import Engine as _E, df_to_engine + from graphistry.compute.gfql.index import traverse as _traverse + + def probe(n_nodes): + rng = np.random.default_rng(7) + deg = 6 + m = n_nodes * deg + edf = pd.DataFrame({ + "src": rng.integers(0, n_nodes, m), + "dst": rng.integers(0, n_nodes, m), + "etype": rng.integers(0, 3, m), + }) + ndf = pd.DataFrame({"id": np.arange(n_nodes)}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + if engine == "polars": + g = g.edges(df_to_engine(g._edges, _E.POLARS), "src", "dst").nodes( + df_to_engine(g._nodes, _E.POLARS), "id") + gi = g.gfql_index_all(engine=engine) + seen = [] + orig = _traverse._gather_series + + def spy(series, rows, eng): + seen.append(int(len(rows))) + return orig(series, rows, eng) + + _traverse._gather_series = spy + try: + seeds = g._nodes[:1] if engine == "pandas" else g._nodes.head(1) + gi.hop(nodes=seeds, hops=1, return_as_wave_front=True, + edge_match={"etype": 1}, engine=engine) + finally: + _traverse._gather_series = orig + return sum(seen) + + small, big = probe(1000), probe(8000) + assert small > 0 and big > 0 + # Degree is held fixed, so candidate rows must not scale with |E| (allow slack + # for the random degree distribution of a single seed). + assert big <= small * 4 + 20, ( + f"predicate rows grew {small} -> {big} as the graph grew 8x; " + "the edge_match filter is scaling with the graph") + + +@pytest.mark.parametrize("engine", _cpu_engines()) +def test_typed_edge_predicate_abandons_indexed_path_on_evaluation_failure( + typed_graph, engine, monkeypatch +): + """The one branch the candidate-row form adds: `mask_for` -> None -> abandon. + + It carries all of this path's parity risk and previously had no test. Force the + gather to raise mid-traversal and assert the result is byte-identical to the + no-index scan — i.e. the hop really did fall back, rather than returning a partial + answer built from the rows it had already filtered. + """ + from graphistry.Engine import Engine as _E, df_to_engine + from graphistry.compute.gfql.index import traverse as _traverse + + g = typed_graph + if engine == "polars": + g = g.edges(df_to_engine(g._edges, _E.POLARS), "src", "dst").nodes( + df_to_engine(g._nodes, _E.POLARS), "id") + gi = g.gfql_index_all(engine=engine) + seeds = g._nodes[:1] if engine == "pandas" else g._nodes.head(1) + kwargs = dict(hops=1, return_as_wave_front=True, edge_match={"etype": 1}, engine=engine) + + expected = g.hop(nodes=seeds, **kwargs) + + calls = {"n": 0} + + def boom(series, rows, eng): + calls["n"] += 1 + raise RuntimeError("synthetic gather failure") + + monkeypatch.setattr(_traverse, "_gather_series", boom) + got = gi.hop(nodes=seeds, **kwargs) + + assert calls["n"] > 0, "the failure was never triggered — test proves nothing" + assert int(got._edges.shape[0]) == int(expected._edges.shape[0]) + assert int(got._nodes.shape[0]) == int(expected._nodes.shape[0]) + def _pairs(df): + col = (lambda c: df[c].to_list()) if hasattr(df[df.columns[0]], "to_list") \ + else (lambda c: df[c].tolist()) + return sorted(zip(col("src"), col("dst"))) + + got_pairs, exp_pairs = _pairs(got._edges), _pairs(expected._edges) + assert got_pairs == exp_pairs, "fallback did not reproduce the scan result exactly" + + +@pytest.mark.parametrize("engine", _cpu_engines()) +def test_fixed_point_typed_walk_bounds_predicate_work(engine): + """A fixed-point walk reaching most of the graph must NOT gather unboundedly. + + This is the regime where candidate-row evaluation loses to one whole-column compare + (it gathers up to 2E by random access). The cumulative-cost guard must notice and + switch to the whole-column mask, so total gathered stays a bounded fraction of E — + and the answer must be unchanged either way. + """ + from graphistry.Engine import Engine as _E, df_to_engine + from graphistry.compute.gfql.index import traverse as _traverse + + rng = np.random.default_rng(11) + n_nodes, n_edges = 4000, 40000 + nodes = pd.DataFrame({"id": np.arange(n_nodes, dtype=np.int64)}) + edges = pd.DataFrame({ + "src": rng.integers(0, n_nodes, n_edges).astype(np.int64), + "dst": rng.integers(0, n_nodes, n_edges).astype(np.int64), + "etype": rng.integers(0, 2, n_edges).astype(np.int64), + }) + g = graphistry.edges(edges, "src", "dst").nodes(nodes, "id") + if engine == "polars": + g = g.edges(df_to_engine(g._edges, _E.POLARS), "src", "dst").nodes( + df_to_engine(g._nodes, _E.POLARS), "id") + gi = g.gfql_index_all(engine=engine) + seeds = g._nodes[:1] if engine == "pandas" else g._nodes.head(1) + kwargs = dict(to_fixed_point=True, direction="undirected", return_as_wave_front=True, + edge_match={"etype": 1}, engine=engine) + + gathered = [] + orig = _traverse._gather_series + + def spy(series, rows, eng): + gathered.append(int(len(rows))) + return orig(series, rows, eng) + + import unittest.mock as _mock + with _mock.patch.object(_traverse, "_gather_series", spy): + got = gi.hop(nodes=seeds, **kwargs) + expected = g.hop(nodes=seeds, **kwargs) + + assert int(got._edges.shape[0]) == int(expected._edges.shape[0]), \ + "guard changed the answer" + total = sum(gathered) + limit = max(_traverse._EAGER_MASK_SWITCH_FLOOR, + n_edges // _traverse._EAGER_MASK_SWITCH_DIVISOR) + assert total <= limit, ( + f"gathered {total} rows of {n_edges} edges — the cost guard did not engage; " + "an unbounded gather is the regression this pins" + ) + + +@pytest.mark.parametrize("engine", _cpu_engines()) +@pytest.mark.parametrize("shape", ["one_hop", "two_hop", "fixed_point", "undirected"]) +def test_both_sides_of_the_edge_mask_cost_boundary_agree(typed_graph, engine, shape, monkeypatch): + """Either side of the optimization boundary must return the same answer. + + The candidate-row form and the whole-column form are two implementations of one + predicate; the cost guard switches between them mid-traversal based on how much has + been gathered. That switch is only safe if the two forms agree cell-for-cell, so pin + it directly: force each side via GFQL_INDEX_CANDIDATE_EDGE_MASK and compare, with the + unindexed scan as a third opinion so a shared bug in both forms cannot hide. + """ + from graphistry.Engine import Engine as _E, df_to_engine + + g = typed_graph + if engine == "polars": + g = g.edges(df_to_engine(g._edges, _E.POLARS), "src", "dst").nodes( + df_to_engine(g._nodes, _E.POLARS), "id") + gi = g.gfql_index_all(engine=engine) + seeds = g._nodes[:1] if engine == "pandas" else g._nodes.head(1) + kw = dict(one_hop=dict(hops=1, direction="forward"), + two_hop=dict(hops=2, direction="forward"), + fixed_point=dict(to_fixed_point=True, direction="forward"), + undirected=dict(to_fixed_point=True, direction="undirected"))[shape] + kwargs = dict(return_as_wave_front=True, edge_match={"etype": 1}, engine=engine, **kw) + + def run(): + return gi.hop(nodes=seeds, **kwargs) + + monkeypatch.setenv("GFQL_INDEX_CANDIDATE_EDGE_MASK", "1") + candidate = run() + monkeypatch.setenv("GFQL_INDEX_CANDIDATE_EDGE_MASK", "0") + whole_column = run() + monkeypatch.delenv("GFQL_INDEX_CANDIDATE_EDGE_MASK", raising=False) + scan = g.hop(nodes=seeds, **kwargs) # no index at all — independent oracle + + def pairs(gg): + df = gg._edges + col = (lambda c: df[c].to_list()) if hasattr(df[df.columns[0]], "to_list") \ + else (lambda c: df[c].tolist()) + return sorted(zip(col("src"), col("dst"))) + + assert pairs(candidate) == pairs(whole_column), f"[{shape}] the two forms disagree" + assert pairs(candidate) == pairs(scan), f"[{shape}] both forms disagree with the scan" + + def node_ids(gg): + s = gg._nodes["id"] + return set(s.to_list() if hasattr(s, "to_list") else s.tolist()) + + # The two forms must agree with each other EXACTLY — that is the boundary property this + # test exists for, and it holds on nodes as well as edges. + assert node_ids(candidate) == node_ids(whole_column), f"[{shape}] node sets differ" + + # Against the scan we compare only the nodes the scan also produces. There is a + # PRE-EXISTING indexed-vs-scan divergence, unrelated to this PR and present identically + # on master 84be35fb: for an undirected to_fixed_point wavefront hop the indexed path + # keeps the SEED in `_nodes` while the scan drops it when the walk never returns to it + # (edges are identical). Asserting equality here would encode that bug as expected; this + # asserts the indexed result is a superset and that any excess is exactly the seed. + seed_ids = set(seeds["id"].to_list() if hasattr(seeds["id"], "to_list") else seeds["id"].tolist()) + extra = node_ids(candidate) - node_ids(scan) + assert not (node_ids(scan) - node_ids(candidate)), f"[{shape}] indexed path LOST nodes" + assert extra <= seed_ids, f"[{shape}] indexed path gained non-seed nodes: {sorted(extra)[:5]}" + + +@pytest.mark.parametrize("engine", _cpu_engines()) +def test_forcing_the_whole_column_mask_actually_changes_the_path(typed_graph, engine, monkeypatch): + """The negative side of the boundary must be reachable — otherwise the test above is + comparing the candidate-row form against itself and proves nothing.""" + from graphistry.Engine import Engine as _E, df_to_engine + from graphistry.compute.gfql.index import traverse as _traverse + + g = typed_graph + if engine == "polars": + g = g.edges(df_to_engine(g._edges, _E.POLARS), "src", "dst").nodes( + df_to_engine(g._nodes, _E.POLARS), "id") + gi = g.gfql_index_all(engine=engine) + seeds = g._nodes[:1] if engine == "pandas" else g._nodes.head(1) + kwargs = dict(hops=1, return_as_wave_front=True, edge_match={"etype": 1}, engine=engine) + + seen = [] + orig = _traverse._gather_series + + def spy(series, rows, eng): + seen.append(int(len(rows))) + return orig(series, rows, eng) + + monkeypatch.setattr(_traverse, "_gather_series", spy) + + monkeypatch.setenv("GFQL_INDEX_CANDIDATE_EDGE_MASK", "1") + gi.hop(nodes=seeds, **kwargs) + assert seen, "candidate-row path did not gather — the ON side never engaged" + + seen.clear() + monkeypatch.setenv("GFQL_INDEX_CANDIDATE_EDGE_MASK", "0") + gi.hop(nodes=seeds, **kwargs) + assert not seen, "OFF side still gathered candidate rows — the switch does nothing" diff --git a/graphistry/tests/compute/gfql/index/test_index_gpu_edge_match.py b/graphistry/tests/compute/gfql/index/test_index_gpu_edge_match.py new file mode 100644 index 0000000000..6ccc83bebe --- /dev/null +++ b/graphistry/tests/compute/gfql/index/test_index_gpu_edge_match.py @@ -0,0 +1,127 @@ +"""GPU coverage for the indexed `edge_match` candidate-row path (#1782) and the semi-join +key sides (#1784). + +Why a separate file: the regression tests for those changes are parametrized over +``_cpu_engines()`` — pandas + polars — so the ``Engine.CUDF`` and ``Engine.POLARS_GPU`` +branches had NO coverage from them. They cannot simply be switched to ``_engines()``: +that helper includes cudf whenever cudf is *importable*, so on a dev box with cudf +installed but no working GPU every such test fails. This file gates on a real runtime +probe instead, so it runs where there is a GPU and skips cleanly where there is not. + +The specific device risks these pin, all raised in review of #1782: + * ``(sub == val).fillna(False).values`` on a GATHERED cudf column — ``.values`` RAISES on a + null-bearing cudf column, so correctness rests entirely on ``fillna`` having cleared nulls + for EVERY result dtype, not just the numeric ones. + * the EMPTY candidate batch on device (a zero-length gather map). + * the cost guard's whole-column fallback firing on device. +Pandas is the oracle in every case — comparing a GPU engine only against another GPU engine +would let a shared defect pass. + +WHAT THESE DO **NOT** PIN, measured rather than assumed: deleting the ``fillna(False)`` from +the cudf branch of ``_EdgeMatchRowFilter.mask_for`` leaves all of these GREEN on cudf 26.02. +The review's concern was that ``.values`` raises on a null-bearing cudf column; on this +version ``(sub == val)`` already yields a non-null boolean column, so the fillna is defensive +rather than load-bearing *here*. It is kept because the pandas branch genuinely needs it and +because a future cudf could restore null propagation — but do not read a green run of this +file as evidence that the fillna is required. What the file DOES pin is device-vs-pandas +parity across null-bearing dtypes and the empty-batch path, both of which had no coverage at +all before: every new test the stack added is parametrized over ``_cpu_engines()``. +""" +import numpy as np +import pandas as pd +import pytest + +import graphistry +from graphistry.compute.gfql.index import create_index # noqa: F401 (registers plottable API) + +cudf = pytest.importorskip("cudf") + + +def _gpu_available() -> bool: + """Probe by running the smallest version of what these tests do. + + Cheaper probes do not discriminate on a box with cudf installed but no CUDA runtime: + `cudf.DataFrame(...)`, `.to_pandas()`, `.values` (a small cupy alloc), a comparison, and + even `groupby().sum()` all SUCCEED there, and the suite then fails with + `OSError: libnvrtc.so.12` inside the first real kernel. So the probe is an actual indexed + typed hop on a two-edge graph — if that runs, everything below can. + """ + try: + import numpy as _np + import pandas as _pd + from graphistry.Engine import Engine as _E, df_to_engine as _to + _n = _to(_pd.DataFrame({"id": _np.arange(3, dtype=_np.int64)}), _E.CUDF) + _e = _to(_pd.DataFrame({"src": [0, 1], "dst": [1, 2], "etype": [1, 2]}), _E.CUDF) + _g = graphistry.nodes(_n, "id").edges(_e, "src", "dst").gfql_index_all(engine="cudf") + _g.hop(nodes=_n[:1], hops=1, return_as_wave_front=True, + edge_match={"etype": 1}, engine="cudf") + return True + except Exception: + return False + + +pytestmark = pytest.mark.skipif(not _gpu_available(), reason="no working cudf / GPU") + +NULL_DTYPES = ["int64", "Int64", "float", "string", "boolean"] + + +def _frames(null_col_dtype: str, n_nodes: int = 400, deg: int = 6): + """Typed graph whose edge predicate column carries NULLS — the case `.values` trips on.""" + rng = np.random.default_rng(3) + m = n_nodes * deg + etype = rng.integers(0, 3, m) + edges = pd.DataFrame({"src": rng.integers(0, n_nodes, m), + "dst": rng.integers(0, n_nodes, m), + "etype": pd.Series(etype).astype(null_col_dtype) + if null_col_dtype not in ("string", "boolean") + else pd.Series([str(v) for v in etype] if null_col_dtype == "string" + else (etype % 2).astype(bool)).astype(null_col_dtype)}) + edges.loc[edges.index[::7], "etype"] = None # ~14% nulls + nodes = pd.DataFrame({"id": np.arange(n_nodes, dtype=np.int64)}) + return nodes, edges + + +def _match_value(null_col_dtype: str): + return {"string": "1", "boolean": True}.get(null_col_dtype, 1) + + +@pytest.mark.parametrize("engine", ["cudf", "polars-gpu"]) +@pytest.mark.parametrize("null_col_dtype", NULL_DTYPES) +def test_null_bearing_edge_predicate_matches_the_pandas_oracle_on_device(engine, null_col_dtype): + """`.values` on a gathered, null-bearing device column must neither raise nor diverge.""" + nodes, edges = _frames(null_col_dtype) + g_pd = graphistry.nodes(nodes, "id").edges(edges, "src", "dst") + kwargs = dict(hops=1, return_as_wave_front=True, + edge_match={"etype": _match_value(null_col_dtype)}) + oracle = g_pd.hop(nodes=nodes[:1], engine="pandas", **kwargs) + + from graphistry.Engine import Engine as _E, df_to_engine + target = _E.CUDF if engine == "cudf" else _E.POLARS + g = graphistry.nodes(df_to_engine(nodes, target), "id").edges( + df_to_engine(edges, target), "src", "dst") + got = g.gfql_index_all(engine=engine).hop( + nodes=df_to_engine(nodes[:1], target), engine=engine, **kwargs) + + def pairs(gg): + df = gg._edges + p = df.to_pandas() if hasattr(df, "to_pandas") else df + return sorted(zip(p["src"].tolist(), p["dst"].tolist())) + + assert pairs(got) == pairs(oracle), f"[{engine}/{null_col_dtype}] diverged from pandas" + + +@pytest.mark.parametrize("engine", ["cudf", "polars-gpu"]) +def test_empty_candidate_batch_on_device(engine): + """A seed with no matching typed edges yields a zero-length gather map on device.""" + nodes, edges = _frames("int64") + from graphistry.Engine import Engine as _E, df_to_engine + target = _E.CUDF if engine == "cudf" else _E.POLARS + g_pd = graphistry.nodes(nodes, "id").edges(edges, "src", "dst") + kwargs = dict(hops=1, return_as_wave_front=True, edge_match={"etype": 99}) # matches nothing + oracle = g_pd.hop(nodes=nodes[:1], engine="pandas", **kwargs) + + g = graphistry.nodes(df_to_engine(nodes, target), "id").edges( + df_to_engine(edges, target), "src", "dst") + got = g.gfql_index_all(engine=engine).hop( + nodes=df_to_engine(nodes[:1], target), engine=engine, **kwargs) + assert int(got._edges.shape[0]) == int(oracle._edges.shape[0]) == 0