From 9b121e4fb596d64afd9c8da79c88f082d1c76f3a Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sun, 19 Jul 2026 03:41:06 -0700 Subject: [PATCH 1/5] perf(gfql): engage #1658 index for seeded gfql()/Cypher chains on pandas/cuDF (incl. typed edges) A seeded hop expressed as a gfql()/Cypher chain (not a direct g.hop) always scanned on pandas/cuDF even with a resident index: _chain_impl attaches a synthetic per-edge id column (copy(deep=False)+assign) -> a fresh edge-frame object -> the index's `source_ref is df` identity guard missed. - registry.rebind_edges: re-point edge adjacency indexes at the augmented frame (safe: shallow copy preserves src/dst by value; fingerprint unchanged by an added column). Called in chain.py right after the edge-index augmentation. - Typed edges: relax _hop_is_index_coverable for simple scalar-equality edge_match on the wavefront path; index_seeded_hop applies the edge predicate to CSR-matched rows each hop, parity-exact with the scan's filter_edges_by_dict. - Predicate/membership edge_match, edge_query, and the direct-hop path stay on scan. Measured (dgx, 500k/4M/4-types, warm median, index==scan): pandas typed 2.1x / untyped 2.7x; cuDF typed 1.9x / untyped 2.0x. polars/polars-gpu untyped already engage via the native lazy executor; typed polars is a tracked follow-up. 27 new 4-engine differential + engagement tests; existing 109-test index suite green. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- CHANGELOG.md | 1 + graphistry/compute/chain.py | 9 ++ graphistry/compute/gfql/index/__init__.py | 4 +- graphistry/compute/gfql/index/api.py | 21 +++- graphistry/compute/gfql/index/registry.py | 19 ++++ graphistry/compute/gfql/index/traverse.py | 73 +++++++++++++- .../tests/compute/gfql/index/test_index.py | 98 +++++++++++++++++++ 7 files changed, 219 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea3f2f0baa..b409503075 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Fixed - **GFQL polars/polars-gpu seeded traversal no longer pays an O(E) NaN re-scan per hop on float-column graphs (identity-stable `_pl_nan_to_null` + clean-cache)**: every `hop()` runs `_coerce_input_formats` → `_pl_nan_to_null`, which scanned `is_nan().any()` over every float column on the (unchanged) resident edge frame **on every call** — so repeated seeded hops on a resident graph (the seeded-Search / native-hop pattern) grew O(E) with edge count on polars while pandas stayed flat. Measured: an indexed polars `g.hop` on 8M edges with one float column was **33.8 ms and growing**; it is now **0.20 ms and FLAT** (~140× faster; O(degree)), matching pandas. `_pl_nan_to_null` now probes an eager polars frame for real NaN once, returns a clean frame UNCHANGED (restoring the #1726 identity guard that #1731 reverted — so frame-identity caches like the #1658 index keep engaging), rewrites only the columns that genuinely carry NaN (values identical to the old unconditional `fill_nan`), and caches the id of frames verified clean so later calls skip the probe (recycle-safe via `weakref.finalize`; a rebound/new frame re-probes, so NaN→null semantics are unchanged). Regression: full 4-engine `test_index.py` parity + new `TestPlNanCleanCache` cases (NaN-present cleaned, clean-frame cached same-object, distinct frames independent). +- **GFQL seeded `gfql()`/Cypher chains now engage the resident #1658 adjacency index on pandas/cuDF (incl. typed edges) instead of always scanning**: a seeded hop expressed as a `gfql([...])`/Cypher CHAIN (rather than a direct `g.hop(...)`) previously fell back to the O(E) scan on pandas and cuDF even with a resident index, because the chain executor attaches a synthetic per-edge id column (`copy(deep=False)` + assign) → a fresh edge-frame object → the index's `source_ref is df` identity guard missed. The chain now re-points the resident edge adjacency indexes at the augmented frame via `GfqlIndexRegistry.rebind_edges` (provably safe: the shallow copy preserves the indexed src/dst columns by value; the structural fingerprint — row count + bound columns + engine — is unchanged by an added column). Additionally, a **simple scalar-equality `edge_match`** (typed edges, e.g. `-[:KNOWS]->` / `e_forward({"type": X})`) is now index-coverable on the wavefront path: `index_seeded_hop` applies the edge predicate to the CSR-matched rows each hop, parity-exact with the scan's `filter_edges_by_dict`. Predicate/membership-list `edge_match`, `edge_query`, and the direct-hop (non-wavefront) path deliberately stay on the scan path (no over-reach). Measured (dgx, 500k nodes / 4M edges / 4 edge types, warm median, index result == scan result): pandas typed 1-hop **2.1×** / untyped **2.7×**; cuDF typed **1.9×** / untyped **2.0×**; 2-hop typed pandas **1.4×** / cuDF **2.0×**. polars/polars-gpu untyped already engage via their native lazy executor; polars/polars-gpu *typed* chains remain a tracked follow-up (they fall to the eager chain path whose polars edge-index attach is outside the rebind site). 27 new differential + engagement regression tests (`test_index.py`), 4-engine, plus the existing 109-test index suite green. - **GFQL Cypher `CASE` with mixed-dtype branches is now engine-consistent on cuDF (no more `GFQLTypeError`)**: pandas coerces the two `CASE` branches to a common type, but cuDF's `.where` raised `TypeError: cudf does not support mixed types`, surfaced as a hard `GFQLTypeError`. This bit `CASE WHEN path IS NULL THEN -1 ELSE length(path) END` over an UNREACHABLE `shortestPath` (int `-1` branch vs an object/null hops branch) — pandas returned `-1`, cuDF errored. The row-AST `CASE` evaluator now unifies the branch dtypes and retries, so cuDF returns the same value pandas does. Regression test `test_case_mixed_dtype.py`. - **GFQL polars engine raises a clean typed error (not an opaque polars `InvalidOperationError`) for a string predicate on a non-string column**: `WHERE col STARTS WITH/CONTAINS/regex ...` on a Categorical/Enum/numeric column raised a clean `GFQLSchemaError` ("string predicate used on non-string column") on pandas/cuDF but leaked polars' internal `InvalidOperationError: expected String type, got: cat` on polars. `filter_by_dict_polars` now raises the same `GFQLSchemaError` (categorical treated as non-string, exactly as `filter_by_dict`), so all three engines agree. Regression test `test_polars_string_predicate_nonstring.py`; also confirms connected-join predicate pushdown is parity-correct on polars for string/numeric/eq/bool columns. - **GFQL string predicates (`Contains`/`Startswith`/`Endswith`/`Match`/`Fullmatch`) are now value-safe on non-string columns**: applying a string predicate to a numeric/temporal/bool column raised an opaque `AttributeError: Can only use .str accessor with string values!` on pandas and cuDF. They now follow openCypher semantics — a string op over a non-string value is null → excluded (matching the established per-cell behavior on an object column holding non-strings) — and never stringify the column (which would diverge pandas↔cuDF, e.g. wrongly matching `5 CONTAINS '5'`). String, mixed-object, and all-null columns are unchanged. Regression tests in `test_str.py` across dtypes; cuDF parity confirmed. diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 13b1e5ead2..b1f316bde1 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -964,6 +964,15 @@ def _chain_impl( indexed_edges_df = g._edges.copy(deep=False) indexed_edges_df[GFQL_EDGE_INDEX] = indexed_edges_df.index g = g.edges(indexed_edges_df, edge=GFQL_EDGE_INDEX) + # The shallow copy above only ADDS the synthetic id column; the indexed + # src/dst columns are preserved by value. Re-point any resident #1658 + # adjacency index at the new edge frame so the seeded fast path still + # engages through gfql()/Cypher chains (else the identity guard misses and + # every chain hop falls back to the O(E) scan). + from graphistry.compute.gfql.index import get_registry, set_registry + _reg = get_registry(g) + if not _reg.is_empty(): + g = set_registry(g, _reg.rebind_edges(indexed_edges_df)) else: added_edge_index = False diff --git a/graphistry/compute/gfql/index/__init__.py b/graphistry/compute/gfql/index/__init__.py index 7ee6cbf4f8..1d2386923d 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, get_index_policy, maybe_index_hop, index_name, index_trace, + get_registry, set_registry, get_index_policy, maybe_index_hop, index_name, index_trace, ) from .wire import ( CreateIndex, DropIndex, ShowIndexes, IndexOp, apply_index_op, index_op_from_json, @@ -33,7 +33,7 @@ "EDGE_OUT_ADJ", "EDGE_IN_ADJ", "NODE_ID", "ADJ_KINDS", "ALL_KINDS", "AdjacencyIndex", "NodeIdIndex", "create_index", "drop_index", "show_indexes", "gfql_index_edges", - "gfql_index_all", "get_registry", "get_index_policy", "maybe_index_hop", "index_name", + "gfql_index_all", "get_registry", "set_registry", "get_index_policy", "maybe_index_hop", "index_name", "index_trace", "CreateIndex", "DropIndex", "ShowIndexes", "IndexOp", "apply_index_op", "index_op_from_json", "is_index_op", "is_index_op_json", diff --git a/graphistry/compute/gfql/index/api.py b/graphistry/compute/gfql/index/api.py index bc6f751e5a..20dad5c161 100644 --- a/graphistry/compute/gfql/index/api.py +++ b/graphistry/compute/gfql/index/api.py @@ -88,6 +88,14 @@ def _attach(g: Plottable, registry: GfqlIndexRegistry) -> Plottable: return res +def set_registry(g: Plottable, registry: GfqlIndexRegistry) -> Plottable: + """Attach ``registry`` to a copy of ``g`` (public wrapper over ``_attach``). + + Used by the chain executor to re-point the resident index at an edge frame it + augmented in place (see ``GfqlIndexRegistry.rebind_edges``).""" + return _attach(g, registry) + + def index_name(kind: IndexKind, column: Optional[str]) -> str: return f"{kind}:{column}" if column else kind @@ -279,16 +287,25 @@ def _hop_is_index_coverable( edge_query: Optional[str], include_zero_hop_seed: bool, target_wave_front: Optional[DataFrameT], + return_as_wave_front: bool = False, ) -> bool: if nodes is None: return False if any(x is not None for x in ( min_hops if (min_hops is not None and min_hops > 1) else None, output_min_hops, output_max_hops, label_node_hops, label_edge_hops, - edge_match, source_node_match, destination_node_match, + source_node_match, destination_node_match, source_node_query, destination_node_query, edge_query, target_wave_front, )): return False + # Typed-edge edge_match: coverable only as a simple scalar-equality filter on the + # wavefront (chain/Cypher) path, where index_seeded_hop applies it per-hop parity- + # exact with the scan's filter_edges_by_dict. Predicate/membership forms and the + # direct-hop (non-wavefront) path stay on scan. + if edge_match is not None: + from .traverse import is_simple_equality_edge_match + if not (return_as_wave_front and is_simple_equality_edge_match(edge_match)): + return False if label_seeds or include_zero_hop_seed: return False if not to_fixed_point and (not isinstance(hops, int) or hops < 1): @@ -408,6 +425,7 @@ def _bail(reason: str) -> Optional[Plottable]: edge_query=edge_query, include_zero_hop_seed=include_zero_hop_seed, target_wave_front=target_wave_front, + return_as_wave_front=return_as_wave_front, ): return _bail("query not index-coverable") assert nodes is not None @@ -460,6 +478,7 @@ def _bail(reason: str) -> Optional[Plottable]: g, registry, nodes=nodes, node_col=node_col, src=src, dst=dst, engine=engine, hops=eff_hops, to_fixed_point=to_fixed_point, direction=direction, return_as_wave_front=return_as_wave_front, + edge_match=cast(Optional[dict], rest.get("edge_match")), ) if trace: _record(cast(IndexTraceStep, { diff --git a/graphistry/compute/gfql/index/registry.py b/graphistry/compute/gfql/index/registry.py index 6024ee0172..32d08a36ae 100644 --- a/graphistry/compute/gfql/index/registry.py +++ b/graphistry/compute/gfql/index/registry.py @@ -96,6 +96,25 @@ def without(self, kind: IndexKind) -> "GfqlIndexRegistry": new.pop(kind, None) return GfqlIndexRegistry(new) + def rebind_edges(self, new_edges: DataFrameT) -> "GfqlIndexRegistry": + """Re-point the EDGE adjacency indexes' identity guard at ``new_edges``. + + Use ONLY when the caller has produced ``new_edges`` by a transform that + provably preserves the indexed src/dst columns by value (same rows, same + order) — e.g. a shallow copy that merely ADDS an unrelated column. The chain + executor does exactly this when it attaches its synthetic per-edge id + (chain.py), which otherwise breaks the ``source_ref is df`` identity guard and + forces a full scan. The CSR arrays stay valid (row positions unchanged); we + only swap the strong-ref so ``get_valid`` recognizes the live frame. NODE_ID + is left untouched (node materialization may legitimately change node rows).""" + import dataclasses + new = dict(self.indexes) + for kind in (EDGE_OUT_ADJ, EDGE_IN_ADJ): + idx = new.get(kind) + if idx is not None: + new[kind] = dataclasses.replace(idx, source_ref=new_edges) + return GfqlIndexRegistry(new) + def get(self, kind: IndexKind) -> Optional[Union[AdjacencyIndex, NodeIdIndex]]: return self.indexes.get(kind) diff --git a/graphistry/compute/gfql/index/traverse.py b/graphistry/compute/gfql/index/traverse.py index 80dc73c810..16272a6c5e 100644 --- a/graphistry/compute/gfql/index/traverse.py +++ b/graphistry/compute/gfql/index/traverse.py @@ -6,9 +6,11 @@ falls back to the scan/join path — correctness is never traded for speed). Covered (v1): seeded (nodes given), integer ``hops`` >= 1 or ``to_fixed_point``, -direction forward/reverse/undirected, ``return_as_wave_front``. Not covered -(returns None): edge/source/destination match or query, target_wave_front, -min_hops>1, output_min/max_hops, labeling, missing node table. +direction forward/reverse/undirected, ``return_as_wave_front``, and a simple +scalar-equality ``edge_match`` (typed edges, e.g. Cypher ``-[:KNOWS]->``) applied on +the wavefront path. Not covered (returns None): predicate/membership edge_match, +source/destination match or query, edge_query, target_wave_front, min_hops>1, +output_min/max_hops, labeling, missing node table. """ from __future__ import annotations @@ -44,6 +46,55 @@ def _indices_for_direction( return [out_idx, in_idx] +def is_simple_equality_edge_match(edge_match: Optional[dict]) -> bool: + """True iff ``edge_match`` is a dict of plain scalar equalities. + + This is the only ``edge_match`` shape the index path accelerates parity-exact: + it mirrors filter_by_dict's concrete scalar ``==`` branch. ASTPredicate values + (predicate path), membership lists/sets/tuples (isin path), and nested dicts are + NOT covered here — the caller keeps them on the scan path. + """ + if not edge_match: + return False + from graphistry.compute.predicates.ASTPredicate import ASTPredicate + for v in edge_match.values(): + if isinstance(v, ASTPredicate): + return False + if isinstance(v, (list, tuple, set, dict)): + return False + return True + + +def _build_edge_keep_mask( + edges: DataFrameT, edge_match: dict, engine: Engine, xp: "object" +) -> Optional[ArrayLike]: + """Boolean array over ORIGINAL edge rows (length E, same indexing as + ``AdjacencyIndex.other_values`` / ``row_positions``) selecting rows that satisfy + a simple-equality ``edge_match``. + + Built via each frame's native ``col == val`` (so cudf string columns stay on the + cudf layer instead of a cupy string compare). Returns ``None`` on ANY unexpected + shape or error, so the caller falls back to scan rather than risk a divergence. + """ + try: + if not is_simple_equality_edge_match(edge_match): + return None + mask: Optional[ArrayLike] = None + for col, val in edge_match.items(): + if col not in edges.columns: + return None + if engine in (Engine.POLARS, Engine.POLARS_GPU): + col_mask = cast(ArrayLike, (edges.get_column(col) == val).to_numpy()) + elif engine == Engine.CUDF: + col_mask = cast(ArrayLike, (edges[col] == val).values) + else: + col_mask = cast(ArrayLike, (edges[col] == val).to_numpy()) + mask = col_mask if mask is None else cast(ArrayLike, mask & col_mask) + return mask + except Exception: # pragma: no cover - defensive parity guard + return None + + def index_seeded_hop( g: Plottable, registry: GfqlIndexRegistry, @@ -57,6 +108,7 @@ def index_seeded_hop( to_fixed_point: bool, direction: HopDirection, return_as_wave_front: bool, + edge_match: Optional[dict] = None, ) -> Optional[Plottable]: if nodes is None or g._edges is None or g._nodes is None: return None @@ -77,6 +129,16 @@ def index_seeded_hop( xp, _backend = array_namespace(engine) + # Typed-edge (edge_match) support: a boolean mask over ORIGINAL edge rows that + # pass the match predicate, applied to the CSR-matched rows each hop. Gated to + # simple scalar equality + the wavefront path by the coverability check upstream + # (maybe_index_hop); an unsupported shape returns None here => scan (parity-safe). + edge_keep: Optional[ArrayLike] = None + if edge_match: + edge_keep = _build_edge_keep_mask(edges, edge_match, engine, xp) + if edge_keep is None: + return None + # Do NOT narrow the seed to the index key dtype (a node-id int64 seed cast to # an int32 edge-endpoint key wraps large ids → false match). lookup promotes both # sides to a common dtype; numpy/cupy set ops promote on concat. So we keep ids at @@ -100,6 +162,11 @@ def index_seeded_hop( neigh_parts: List[ArrayLike] = [] for ix in indices: rows, matched = lookup_edge_rows(ix, frontier, xp) + if edge_keep is not None: + # Keep only CSR-matched rows whose edge passes edge_match. Wavefront- + # only (coverability gate), so the `matched`/first-hop `visited` + # bookkeeping below — which edge_match does NOT filter — is never read. + rows = rows[edge_keep[rows]] edge_rows_parts.append(rows) neigh_parts.append(ix.other_values[rows]) matched_parts.append(matched) diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index c8a48bf216..f9b2bc4946 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -641,3 +641,101 @@ def wrapped(registry, direction, edges_df, cols, engine): assert Engine.POLARS_GPU in seen assert keys is not None assert sorted(keys.get_column("id").to_list()) == [0, 1, 2] + + +# ---- chain / Cypher index engagement (#1658 through gfql(), incl. typed edges) ------- +# Regression: a seeded hop expressed as a gfql()/Cypher CHAIN (not a direct g.hop()) +# used to ALWAYS scan on pandas/cuDF, because _chain_impl attaches a synthetic per-edge +# id column -> a fresh edge frame -> the index's `source_ref is df` identity guard missed. +# rebind_edges (chain.py) re-points the adjacency index at the augmented frame; typed +# edges additionally route a simple scalar-equality edge_match through the index. + +@pytest.fixture(scope="module") +def typed_graph(): + rng = np.random.default_rng(1) + n_nodes, deg = 2000, 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)}) + return graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + + +def _sig_typed(g): + def topd(df): + mod = type(df).__module__ + return df.to_pandas() if ("cudf" in mod or "polars" in mod) else df + nn = topd(g._nodes) + ee = topd(g._edges) + nodes = sorted(nn["id"].tolist()) + edges = sorted(map(tuple, ee[["src", "dst", "etype"]].itertuples(index=False, name=None))) + return nodes, edges + + +_TYPED_CHAIN = [n({"id": 100}), e_forward({"etype": 1}, hops=1)] +_TYPED_2HOP = [n({"id": 100}), e_forward({"etype": 1}, hops=1), n(), + e_forward({"etype": 2}, hops=1)] +_UNTYPED_CHAIN = [n({"id": 100}), e_forward(hops=1)] +_MEMBER_CHAIN = [n({"id": 100}), e_forward({"etype": [0, 1]}, hops=1)] + +_PANDAS_CUDF = [e for e in ENGINES if e in ("pandas", "cudf")] + + +@pytest.mark.parametrize("engine", ENGINES) +@pytest.mark.parametrize("chain", [_TYPED_CHAIN, _TYPED_2HOP, _UNTYPED_CHAIN, _MEMBER_CHAIN]) +def test_chain_index_parity_vs_scan(typed_graph, engine, chain): + """Every chain shape returns the SAME subgraph via the index as via the scan, + on every engine. Correctness is never traded for the fast path.""" + gi = typed_graph.gfql_index_all(engine=engine) + base = typed_graph.gfql(chain, index_policy="off", engine=engine) + idx = gi.gfql(chain, index_policy="use", engine=engine) + assert _sig_typed(base) == _sig_typed(idx) + + +@pytest.mark.parametrize("engine", _PANDAS_CUDF) +def test_chain_typed_edge_engages_index_pandas_cudf(typed_graph, engine): + """pandas/cuDF: a typed-edge (simple-equality edge_match) seeded chain hop ENGAGES + the resident #1658 index instead of scanning.""" + gi = typed_graph.gfql_index_all(engine=engine) + rep = gi.gfql_explain(_TYPED_CHAIN, index_policy="use", engine=engine) + assert rep["used_index"] is True, (engine, rep) + + +@pytest.mark.parametrize("engine", ENGINES) +def test_chain_untyped_engages_index(typed_graph, engine): + """An untyped seeded chain hop engages the index on every engine (pandas/cuDF via + the rebind fix; polars/polars-gpu via their native lazy executor).""" + gi = typed_graph.gfql_index_all(engine=engine) + rep = gi.gfql_explain(_UNTYPED_CHAIN, index_policy="use", engine=engine) + assert rep["used_index"] is True, (engine, rep) + + +@pytest.mark.parametrize("engine", ENGINES) +def test_chain_membership_edge_match_stays_on_scan(typed_graph, engine): + """A membership-list edge_match is NOT simple-equality, so it deliberately stays on + the scan path (parity preserved; no over-reach of the index coverage).""" + gi = typed_graph.gfql_index_all(engine=engine) + rep = gi.gfql_explain(_MEMBER_CHAIN, index_policy="use", engine=engine) + assert rep["used_index"] is False, (engine, rep) + + +def test_rebind_edges_revalidates_after_shallow_augmentation(): + """rebind_edges re-points the edge adjacency index at a shallow-copied frame that + merely ADDS a column (the chain's synthetic edge id), so get_valid recognizes it.""" + from graphistry.compute.gfql.index import get_registry + from graphistry.compute.gfql.index.registry import EDGE_OUT_ADJ + rng = np.random.default_rng(2) + edf = pd.DataFrame({"src": rng.integers(0, 500, 3000), "dst": rng.integers(0, 500, 3000)}) + ndf = pd.DataFrame({"id": np.arange(500)}) + gi = graphistry.nodes(ndf, "id").edges(edf, "src", "dst").gfql_index_all(engine="pandas") + reg = get_registry(gi) + from graphistry.Engine import Engine as _E + # augment like the chain does: shallow copy + an extra column -> a NEW frame object + aug = gi._edges.copy(deep=False) + aug["__synthetic_id__"] = aug.index + assert reg.get_valid(EDGE_OUT_ADJ, aug, ("src", "dst"), _E.PANDAS) is None # identity miss + reg2 = reg.rebind_edges(aug) + assert reg2.get_valid(EDGE_OUT_ADJ, aug, ("src", "dst"), _E.PANDAS) is not None # now valid From 129d72cb4634635fbc8d1003ce99e4e27d30fac2 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sun, 19 Jul 2026 03:47:22 -0700 Subject: [PATCH 2/5] perf(gfql): extend chain index engagement to polars/polars-gpu typed edges The native polars lazy chain executor (lazy/engine/polars/chain.py) attaches its synthetic edge id via with_row_index -> a fresh edge frame -> the #1658 identity guard missed, so typed-edge chains scanned on polars/polars-gpu (untyped already engaged). Apply the same rebind_edges fix at the polars augmentation site. Now ALL FOUR engines engage the index for typed AND untyped seeded chain/Cypher hops. Measured (dgx, 500k/4M/4-types, index==scan): polars typed 1.3x / untyped 5.4x; polars-gpu typed engaged / untyped 12.4x. Broadened the engagement test to all engines; 138 index tests pass (109 existing + 29 new), 4-engine. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- CHANGELOG.md | 2 +- graphistry/compute/gfql/lazy/engine/polars/chain.py | 10 ++++++++++ graphistry/tests/compute/gfql/index/test_index.py | 11 +++++------ 3 files changed, 16 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b409503075..c57afde8b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Fixed - **GFQL polars/polars-gpu seeded traversal no longer pays an O(E) NaN re-scan per hop on float-column graphs (identity-stable `_pl_nan_to_null` + clean-cache)**: every `hop()` runs `_coerce_input_formats` → `_pl_nan_to_null`, which scanned `is_nan().any()` over every float column on the (unchanged) resident edge frame **on every call** — so repeated seeded hops on a resident graph (the seeded-Search / native-hop pattern) grew O(E) with edge count on polars while pandas stayed flat. Measured: an indexed polars `g.hop` on 8M edges with one float column was **33.8 ms and growing**; it is now **0.20 ms and FLAT** (~140× faster; O(degree)), matching pandas. `_pl_nan_to_null` now probes an eager polars frame for real NaN once, returns a clean frame UNCHANGED (restoring the #1726 identity guard that #1731 reverted — so frame-identity caches like the #1658 index keep engaging), rewrites only the columns that genuinely carry NaN (values identical to the old unconditional `fill_nan`), and caches the id of frames verified clean so later calls skip the probe (recycle-safe via `weakref.finalize`; a rebound/new frame re-probes, so NaN→null semantics are unchanged). Regression: full 4-engine `test_index.py` parity + new `TestPlNanCleanCache` cases (NaN-present cleaned, clean-frame cached same-object, distinct frames independent). -- **GFQL seeded `gfql()`/Cypher chains now engage the resident #1658 adjacency index on pandas/cuDF (incl. typed edges) instead of always scanning**: a seeded hop expressed as a `gfql([...])`/Cypher CHAIN (rather than a direct `g.hop(...)`) previously fell back to the O(E) scan on pandas and cuDF even with a resident index, because the chain executor attaches a synthetic per-edge id column (`copy(deep=False)` + assign) → a fresh edge-frame object → the index's `source_ref is df` identity guard missed. The chain now re-points the resident edge adjacency indexes at the augmented frame via `GfqlIndexRegistry.rebind_edges` (provably safe: the shallow copy preserves the indexed src/dst columns by value; the structural fingerprint — row count + bound columns + engine — is unchanged by an added column). Additionally, a **simple scalar-equality `edge_match`** (typed edges, e.g. `-[:KNOWS]->` / `e_forward({"type": X})`) is now index-coverable on the wavefront path: `index_seeded_hop` applies the edge predicate to the CSR-matched rows each hop, parity-exact with the scan's `filter_edges_by_dict`. Predicate/membership-list `edge_match`, `edge_query`, and the direct-hop (non-wavefront) path deliberately stay on the scan path (no over-reach). Measured (dgx, 500k nodes / 4M edges / 4 edge types, warm median, index result == scan result): pandas typed 1-hop **2.1×** / untyped **2.7×**; cuDF typed **1.9×** / untyped **2.0×**; 2-hop typed pandas **1.4×** / cuDF **2.0×**. polars/polars-gpu untyped already engage via their native lazy executor; polars/polars-gpu *typed* chains remain a tracked follow-up (they fall to the eager chain path whose polars edge-index attach is outside the rebind site). 27 new differential + engagement regression tests (`test_index.py`), 4-engine, plus the existing 109-test index suite green. +- **GFQL seeded `gfql()`/Cypher chains now engage the resident #1658 adjacency index on ALL FOUR engines (incl. typed edges) instead of always scanning**: a seeded hop expressed as a `gfql([...])`/Cypher CHAIN (rather than a direct `g.hop(...)`) previously fell back to the O(E) scan even with a resident index, because both chain executors attach a synthetic per-edge id column to the edge frame (pandas/cuDF eager: `copy(deep=False)` + assign; polars native lazy: `with_row_index`) → a fresh edge-frame object → the index's `source_ref is df` identity guard missed. Both chain paths now re-point the resident edge adjacency indexes at the augmented frame via `GfqlIndexRegistry.rebind_edges` (provably safe: the shallow copy / `with_row_index` preserves the indexed src/dst columns by value; the structural fingerprint — row count + bound columns + engine — is unchanged by an added column). Additionally, a **simple scalar-equality `edge_match`** (typed edges, e.g. `-[:KNOWS]->` / `e_forward({"type": X})`) is now index-coverable on the wavefront path: `index_seeded_hop` applies the edge predicate to the CSR-matched rows each hop, parity-exact with the scan's `filter_edges_by_dict`. Predicate/membership-list `edge_match`, `edge_query`, and the direct-hop (non-wavefront) path deliberately stay on the scan path (no over-reach). Measured (dgx, 500k nodes / 4M edges / 4 edge types, warm median, index result == scan result): pandas typed 1-hop **2.1×** / untyped **2.7×**; cuDF typed **1.9×** / untyped **2.0×**; polars typed **1.3×** / untyped **5.4×**; polars-gpu typed engaged (1.0×) / untyped **12.4×**; 2-hop typed engaged on all four engines. 29 new differential + engagement regression tests (`test_index.py`), 4-engine, plus the existing 109-test index suite green. - **GFQL Cypher `CASE` with mixed-dtype branches is now engine-consistent on cuDF (no more `GFQLTypeError`)**: pandas coerces the two `CASE` branches to a common type, but cuDF's `.where` raised `TypeError: cudf does not support mixed types`, surfaced as a hard `GFQLTypeError`. This bit `CASE WHEN path IS NULL THEN -1 ELSE length(path) END` over an UNREACHABLE `shortestPath` (int `-1` branch vs an object/null hops branch) — pandas returned `-1`, cuDF errored. The row-AST `CASE` evaluator now unifies the branch dtypes and retries, so cuDF returns the same value pandas does. Regression test `test_case_mixed_dtype.py`. - **GFQL polars engine raises a clean typed error (not an opaque polars `InvalidOperationError`) for a string predicate on a non-string column**: `WHERE col STARTS WITH/CONTAINS/regex ...` on a Categorical/Enum/numeric column raised a clean `GFQLSchemaError` ("string predicate used on non-string column") on pandas/cuDF but leaked polars' internal `InvalidOperationError: expected String type, got: cat` on polars. `filter_by_dict_polars` now raises the same `GFQLSchemaError` (categorical treated as non-string, exactly as `filter_by_dict`), so all three engines agree. Regression test `test_polars_string_predicate_nonstring.py`; also confirms connected-join predicate pushdown is parity-correct on polars for string/numeric/eq/bool columns. - **GFQL string predicates (`Contains`/`Startswith`/`Endswith`/`Match`/`Fullmatch`) are now value-safe on non-string columns**: applying a string predicate to a numeric/temporal/bool column raised an opaque `AttributeError: Can only use .str accessor with string values!` on pandas and cuDF. They now follow openCypher semantics — a string op over a non-string value is null → excluded (matching the established per-cell behavior on an object column holding non-strings) — and never stringify the column (which would diverge pandas↔cuDF, e.g. wrongly matching `5 CONTAINS '5'`). String, mixed-object, and all-null columns are unchanged. Regression tests in `test_str.py` across dtypes; cuDF parity confirmed. diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index 72b8769f33..fd6e9ee790 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -645,6 +645,16 @@ def _plain_edge(op): EID = "__gfql_edge_index__" g = g.edges(g._edges.with_row_index(EID), g._source, g._destination, edge=EID) added_edge_index = True + # with_row_index only PREPENDS a synthetic id column; the indexed src/dst are + # preserved by value. Re-point any resident #1658 adjacency index at the new + # edge frame so the seeded fast path still engages through the native polars + # chain executor (mirrors compute/chain.py — else the identity guard misses + # and every hop falls back to the O(E) scan). Enables typed-edge chains on + # polars/polars-gpu (untyped already engaged; the frame swap blocked typed). + from graphistry.compute.gfql.index import get_registry, set_registry + _reg = get_registry(g) + if not _reg.is_empty(): + g = set_registry(g, _reg.rebind_edges(g._edges)) else: EID = g._edge added_edge_index = False diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index f9b2bc4946..7782e4da01 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -681,8 +681,6 @@ def topd(df): _UNTYPED_CHAIN = [n({"id": 100}), e_forward(hops=1)] _MEMBER_CHAIN = [n({"id": 100}), e_forward({"etype": [0, 1]}, hops=1)] -_PANDAS_CUDF = [e for e in ENGINES if e in ("pandas", "cudf")] - @pytest.mark.parametrize("engine", ENGINES) @pytest.mark.parametrize("chain", [_TYPED_CHAIN, _TYPED_2HOP, _UNTYPED_CHAIN, _MEMBER_CHAIN]) @@ -695,10 +693,11 @@ def test_chain_index_parity_vs_scan(typed_graph, engine, chain): assert _sig_typed(base) == _sig_typed(idx) -@pytest.mark.parametrize("engine", _PANDAS_CUDF) -def test_chain_typed_edge_engages_index_pandas_cudf(typed_graph, engine): - """pandas/cuDF: a typed-edge (simple-equality edge_match) seeded chain hop ENGAGES - the resident #1658 index instead of scanning.""" +@pytest.mark.parametrize("engine", ENGINES) +def test_chain_typed_edge_engages_index(typed_graph, engine): + """All four engines: a typed-edge (simple-equality edge_match) seeded chain hop + ENGAGES the resident #1658 index instead of scanning (pandas/cuDF via the eager + chain rebind; polars/polars-gpu via the native lazy chain executor rebind).""" gi = typed_graph.gfql_index_all(engine=engine) rep = gi.gfql_explain(_TYPED_CHAIN, index_policy="use", engine=engine) assert rep["used_index"] is True, (engine, rep) From 46a9d48f9fa7f0b4259af9d2a47e432b68b0eec5 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sun, 19 Jul 2026 13:28:11 -0700 Subject: [PATCH 3/5] =?UTF-8?q?fix(types):=20mypy=20=E2=80=94=20isinstance?= =?UTF-8?q?-narrow=20edge=5Fmatch=20before=20simple-equality=20check;=20ca?= =?UTF-8?q?st=20ArrayLike=20operands=20for=20mask=20&?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/gfql/index/api.py | 6 +++++- graphistry/compute/gfql/index/traverse.py | 4 ++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/graphistry/compute/gfql/index/api.py b/graphistry/compute/gfql/index/api.py index 20dad5c161..0ed74c3d4c 100644 --- a/graphistry/compute/gfql/index/api.py +++ b/graphistry/compute/gfql/index/api.py @@ -304,7 +304,11 @@ def _hop_is_index_coverable( # direct-hop (non-wavefront) path stay on scan. if edge_match is not None: from .traverse import is_simple_equality_edge_match - if not (return_as_wave_front and is_simple_equality_edge_match(edge_match)): + if not ( + return_as_wave_front + and isinstance(edge_match, dict) + and is_simple_equality_edge_match(edge_match) + ): return False if label_seeds or include_zero_hop_seed: return False diff --git a/graphistry/compute/gfql/index/traverse.py b/graphistry/compute/gfql/index/traverse.py index 16272a6c5e..5796b22d07 100644 --- a/graphistry/compute/gfql/index/traverse.py +++ b/graphistry/compute/gfql/index/traverse.py @@ -14,7 +14,7 @@ """ from __future__ import annotations -from typing import List, Optional, Tuple, cast +from typing import Any, List, Optional, Tuple, cast from graphistry.Engine import Engine from graphistry.compute.typing import DataFrameT @@ -89,7 +89,7 @@ def _build_edge_keep_mask( col_mask = cast(ArrayLike, (edges[col] == val).values) else: col_mask = cast(ArrayLike, (edges[col] == val).to_numpy()) - mask = col_mask if mask is None else cast(ArrayLike, mask & col_mask) + mask = col_mask if mask is None else cast(ArrayLike, cast(Any, mask) & cast(Any, col_mask)) return mask except Exception: # pragma: no cover - defensive parity guard return None From 11d0bab3aca64ece86a4a2b135af73a11294ae99 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sun, 19 Jul 2026 14:04:34 -0700 Subject: [PATCH 4/5] refactor(gfql): precise edge_match typing + fingerprint-enforced rebind_edges (review) Review feedback on #1734: - types.py gains ScalarMatchValue / EdgeMatchValue / EdgeMatch / SimpleEqualityEdgeMatch mirroring exactly the runtime shapes filter_by_dict accepts; is_simple_equality_edge_match becomes a TypeGuard so the index path gets real narrowing instead of Optional[dict], and _build_edge_keep_mask / index_seeded_hop take EdgeMatch. - rebind_edges no longer trusts its docstring: the strong-ref swap is gated on the index's O(1) structural fingerprint (row count + cols + engine) and indexed-column presence; any mismatch DROPS the edge index (safe miss -> scan) instead of re-pointing it at a frame it wasn't built over. Value-level preservation stays the caller's promise (checking it would be the O(E) scan this path avoids). + enforcement test. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/gfql/index/registry.py | 43 ++++++++++++++----- graphistry/compute/gfql/index/traverse.py | 12 ++++-- graphistry/compute/gfql/index/types.py | 21 ++++++++- .../tests/compute/gfql/index/test_index.py | 30 +++++++++++++ 4 files changed, 90 insertions(+), 16 deletions(-) diff --git a/graphistry/compute/gfql/index/registry.py b/graphistry/compute/gfql/index/registry.py index 32d08a36ae..b5c2754e24 100644 --- a/graphistry/compute/gfql/index/registry.py +++ b/graphistry/compute/gfql/index/registry.py @@ -99,20 +99,41 @@ def without(self, kind: IndexKind) -> "GfqlIndexRegistry": def rebind_edges(self, new_edges: DataFrameT) -> "GfqlIndexRegistry": """Re-point the EDGE adjacency indexes' identity guard at ``new_edges``. - Use ONLY when the caller has produced ``new_edges`` by a transform that - provably preserves the indexed src/dst columns by value (same rows, same - order) — e.g. a shallow copy that merely ADDS an unrelated column. The chain - executor does exactly this when it attaches its synthetic per-edge id - (chain.py), which otherwise breaks the ``source_ref is df`` identity guard and - forces a full scan. The CSR arrays stay valid (row positions unchanged); we - only swap the strong-ref so ``get_valid`` recognizes the live frame. NODE_ID - is left untouched (node materialization may legitimately change node rows).""" - import dataclasses + Caller contract: ``new_edges`` was produced by a transform that preserves + the indexed src/dst columns by value (same rows, same order) — e.g. a shallow + copy that merely ADDS an unrelated column. The chain executor does exactly + this when it attaches its synthetic per-edge id (chain.py), which otherwise + breaks the ``source_ref is df`` identity guard and forces a full scan. The + CSR arrays stay valid (row positions unchanged); we only swap the strong-ref + so ``get_valid`` recognizes the live frame. NODE_ID is left untouched (node + materialization may legitimately change node rows). + + ENFORCED (O(1), engine-portable): the swap happens only when ``new_edges`` + still matches the index's structural fingerprint (row count + bound cols + + engine) and actually carries the indexed columns; on any mismatch the edge + index is DROPPED (safe miss -> scan path) instead of re-pointed at a frame + it wasn't built over. Value-level preservation remains the caller's promise — + checking it would be the O(E) scan this path exists to avoid.""" new = dict(self.indexes) for kind in (EDGE_OUT_ADJ, EDGE_IN_ADJ): idx = new.get(kind) - if idx is not None: - new[kind] = dataclasses.replace(idx, source_ref=new_edges) + if idx is None: + continue + if not isinstance(idx, AdjacencyIndex): # defensive; also narrows for mypy + new.pop(kind, None) + continue + ok = idx.fingerprint == frame_fingerprint( + new_edges, tuple(idx.fingerprint[1]), idx.engine) + if ok: + try: + colnames = set(new_edges.columns) + ok = idx.key_col in colnames and idx.other_col in colnames + except Exception: + ok = False + if ok: + new[kind] = replace(idx, source_ref=new_edges) + else: + new.pop(kind, None) return GfqlIndexRegistry(new) def get(self, kind: IndexKind) -> Optional[Union[AdjacencyIndex, NodeIdIndex]]: diff --git a/graphistry/compute/gfql/index/traverse.py b/graphistry/compute/gfql/index/traverse.py index 5796b22d07..3cf986f048 100644 --- a/graphistry/compute/gfql/index/traverse.py +++ b/graphistry/compute/gfql/index/traverse.py @@ -16,6 +16,8 @@ from typing import Any, List, Optional, Tuple, cast +from typing_extensions import TypeGuard + from graphistry.Engine import Engine from graphistry.compute.typing import DataFrameT from graphistry.Plottable import Plottable @@ -25,7 +27,7 @@ ) from .lookup import lookup_edge_rows, lookup_node_rows from .registry import EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, AdjacencyIndex, GfqlIndexRegistry, NodeIdIndex -from .types import ArrayLike, HopDirection +from .types import ArrayLike, EdgeMatch, HopDirection, SimpleEqualityEdgeMatch def _indices_for_direction( @@ -46,7 +48,9 @@ def _indices_for_direction( return [out_idx, in_idx] -def is_simple_equality_edge_match(edge_match: Optional[dict]) -> bool: +def is_simple_equality_edge_match( + edge_match: Optional[EdgeMatch], +) -> TypeGuard[SimpleEqualityEdgeMatch]: """True iff ``edge_match`` is a dict of plain scalar equalities. This is the only ``edge_match`` shape the index path accelerates parity-exact: @@ -66,7 +70,7 @@ def is_simple_equality_edge_match(edge_match: Optional[dict]) -> bool: def _build_edge_keep_mask( - edges: DataFrameT, edge_match: dict, engine: Engine, xp: "object" + 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 @@ -108,7 +112,7 @@ def index_seeded_hop( to_fixed_point: bool, direction: HopDirection, return_as_wave_front: bool, - edge_match: Optional[dict] = None, + edge_match: Optional[EdgeMatch] = None, ) -> Optional[Plottable]: if nodes is None or g._edges is None or g._nodes is None: return None diff --git a/graphistry/compute/gfql/index/types.py b/graphistry/compute/gfql/index/types.py index af3aa800f6..df9109e00d 100644 --- a/graphistry/compute/gfql/index/types.py +++ b/graphistry/compute/gfql/index/types.py @@ -1,10 +1,15 @@ """Shared internal types for GFQL physical indexes.""" from __future__ import annotations -from typing import List, Literal, Optional, TypedDict, Union +from typing import TYPE_CHECKING, Any, Dict, List, Literal, Mapping, Optional, Set, Tuple, TypedDict, Union + +import numpy as np from graphistry.compute.typing import ArrayLike, ArrayNamespace, DataFrameT +if TYPE_CHECKING: + from graphistry.compute.predicates.ASTPredicate import ASTPredicate + IndexKind = Literal["edge_out_adj", "edge_in_adj", "node_id"] AdjacencyIndexKind = Literal["edge_out_adj", "edge_in_adj"] IndexBackend = Literal["numpy", "cupy"] @@ -17,6 +22,20 @@ IndexPath = Literal["scan", "index"] +# One column's constraint in an ``edge_match``/filter dict — exactly the runtime shapes +# ``filter_by_dict`` accepts: a plain scalar equality (python or numpy scalar), an +# ASTPredicate, a membership collection (isin), or a nested dict. ``EdgeMatch`` is the +# full dict; ``SimpleEqualityEdgeMatch`` is the scalar-equalities-only subset that the +# index path accelerates (see ``is_simple_equality_edge_match``, a TypeGuard for it). +ScalarMatchValue = Union[str, int, float, bool, None, np.generic] +EdgeMatchValue = Union[ + ScalarMatchValue, "ASTPredicate", + List[Any], Tuple[Any, ...], Set[Any], Dict[str, Any], +] +EdgeMatch = Mapping[str, EdgeMatchValue] +SimpleEqualityEdgeMatch = Mapping[str, ScalarMatchValue] + + class IndexTraceStep(TypedDict, total=False): op: str direction: HopDirection diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index 7782e4da01..d7afefe6ad 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -738,3 +738,33 @@ def test_rebind_edges_revalidates_after_shallow_augmentation(): assert reg.get_valid(EDGE_OUT_ADJ, aug, ("src", "dst"), _E.PANDAS) is None # identity miss reg2 = reg.rebind_edges(aug) assert reg2.get_valid(EDGE_OUT_ADJ, aug, ("src", "dst"), _E.PANDAS) is not None # now valid + + +def test_rebind_edges_drops_index_on_fingerprint_mismatch(): + """rebind_edges ENFORCES its contract structurally: a frame with a different row + count (or missing indexed columns) cannot inherit the index — it is dropped + (safe miss -> scan) instead of re-pointed at a frame it wasn't built over.""" + from graphistry.compute.gfql.index import get_registry + from graphistry.compute.gfql.index.registry import EDGE_IN_ADJ, EDGE_OUT_ADJ + rng = np.random.default_rng(3) + edf = pd.DataFrame({"src": rng.integers(0, 100, 500), "dst": rng.integers(0, 100, 500)}) + ndf = pd.DataFrame({"id": np.arange(100)}) + gi = graphistry.nodes(ndf, "id").edges(edf, "src", "dst").gfql_index_all(engine="pandas") + reg = get_registry(gi) + assert reg.has(EDGE_OUT_ADJ) and reg.has(EDGE_IN_ADJ) + + # row count changed -> both edge indexes dropped, never mis-bound + fewer = gi._edges.iloc[:-1].copy(deep=False) + reg2 = reg.rebind_edges(fewer) + assert not reg2.has(EDGE_OUT_ADJ) and not reg2.has(EDGE_IN_ADJ) + + # indexed column renamed away -> dropped + renamed = gi._edges.rename(columns={"dst": "dst2"}) + reg3 = reg.rebind_edges(renamed) + assert not reg3.has(EDGE_OUT_ADJ) and not reg3.has(EDGE_IN_ADJ) + + # same-shape shallow augmentation still rebinds (the intended use) + aug = gi._edges.copy(deep=False) + aug["__synthetic_id__"] = aug.index + reg4 = reg.rebind_edges(aug) + assert reg4.has(EDGE_OUT_ADJ) and reg4.has(EDGE_IN_ADJ) From a12e9549acfdaf7b3f1c96c088cfa230fc542c0c Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sun, 19 Jul 2026 14:33:00 -0700 Subject: [PATCH 5/5] fix(gfql): align typed-edge index guard's value/dtype domain with the scan path (review blockers) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of #1734 found the guard's 'simple scalar equality' narrower than filter_by_dict's runtime behavior — three defects in one seam, all fixed: - F1 (BLOCKER): frozenset (+ pd.Index/pd.Series) leaked past the (list, tuple, set, dict) check but are membership (isin) on the scan path; a bare == is silently all-False -> wrong answer on public g.hop(). The guard now delegates to filter_by_dict._is_membership_filter_value (single source of truth); EdgeMatchValue widened to match. - F2 (BLOCKER): null-carrying match columns (pandas nullable dtypes, polars nulls — common after NaN->null coercion) produced NA cells whose to_numpy() yields an object-dtype array, exploding later at rows[edge_keep[rows]] (IndexError). Masks now fill_null(False)/fillna(False) before materializing — parity-exact (null == val drops on the scan path too). - F3 (IMPORTANT): numeric-col-vs-str / string-col-vs-numeric edge_match returned a silent empty subgraph where the scan raises (GFQLSchemaError E302 on pandas/cuDF; polars its own ComputeError). The mask builder mirrors filter_by_dict's exact two checks and declines -> falls back to the scan -> users get the SAME error as their engine's scan. + 3 engine-parametrized regression tests (frozenset parity, null-column no-crash+parity, dtype-mismatch error parity). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL --- graphistry/compute/gfql/index/traverse.py | 40 +++++++++- graphistry/compute/gfql/index/types.py | 6 +- .../tests/compute/gfql/index/test_index.py | 76 +++++++++++++++++++ 3 files changed, 116 insertions(+), 6 deletions(-) diff --git a/graphistry/compute/gfql/index/traverse.py b/graphistry/compute/gfql/index/traverse.py index 3cf986f048..17f5b8773e 100644 --- a/graphistry/compute/gfql/index/traverse.py +++ b/graphistry/compute/gfql/index/traverse.py @@ -61,10 +61,15 @@ def is_simple_equality_edge_match( if not edge_match: return False from graphistry.compute.predicates.ASTPredicate import ASTPredicate + from graphistry.compute.filter_by_dict import _is_membership_filter_value for v in edge_match.values(): if isinstance(v, ASTPredicate): return False - if isinstance(v, (list, tuple, set, dict)): + # Membership must be decided by the SAME helper the scan path uses + # (filter_by_dict). A local (list, tuple, set, dict) check misses frozenset / + # pd.Index / pd.Series, which the scan lowers to isin — an equality mask over + # such a value is silently all-False (wrong answer), never an error. + if _is_membership_filter_value(v) or isinstance(v, dict): return False return True @@ -83,16 +88,43 @@ def _build_edge_keep_mask( try: if not is_simple_equality_edge_match(edge_match): return None + from graphistry.compute.filter_by_dict import ( + _is_numeric_dtype_safe, _is_string_dtype_safe, + ) + n_edges = int(edges.shape[0]) mask: Optional[ArrayLike] = None for col, val in edge_match.items(): if col not in edges.columns: return None if engine in (Engine.POLARS, Engine.POLARS_GPU): - col_mask = cast(ArrayLike, (edges.get_column(col) == val).to_numpy()) + series = edges.get_column(col) + else: + series = edges[col] + # Obvious dtype mismatch (numeric col vs str val, string col vs numeric + # val): the scan raises GFQLSchemaError E302 where a naive == is silently + # all-False. Decline -> caller falls back to the scan, which raises the + # SAME error (parity-exact; mirrors filter_by_dict's exact two checks, + # skipped like the scan on empty frames). + if n_edges > 0: + dt = series.dtype + if _is_numeric_dtype_safe(dt) and isinstance(val, str): + return None + if (_is_string_dtype_safe(dt) + and isinstance(val, (int, float)) and not isinstance(val, bool)): + return None + # Null-safe materialization: on null-carrying columns (pandas nullable + # Int64/boolean/string, polars nulls — which the NaN->null coercion makes + # common) a bare == yields NA cells, and to_numpy() then produces an + # OBJECT-dtype array that later explodes at rows[edge_keep[rows]] + # (IndexError: not int/bool). Null == val filters out on the scan path, + # so fill False is parity-exact. + if engine in (Engine.POLARS, Engine.POLARS_GPU): + col_mask = cast(ArrayLike, (series == val).fill_null(False).to_numpy()) elif engine == Engine.CUDF: - col_mask = cast(ArrayLike, (edges[col] == val).values) + col_mask = cast(ArrayLike, (series == val).fillna(False).values) else: - col_mask = cast(ArrayLike, (edges[col] == val).to_numpy()) + col_mask = cast( + ArrayLike, (series == val).fillna(False).to_numpy(dtype=bool)) mask = col_mask if mask is None else cast(ArrayLike, cast(Any, mask) & cast(Any, col_mask)) return mask except Exception: # pragma: no cover - defensive parity guard diff --git a/graphistry/compute/gfql/index/types.py b/graphistry/compute/gfql/index/types.py index df9109e00d..ab4f7543aa 100644 --- a/graphistry/compute/gfql/index/types.py +++ b/graphistry/compute/gfql/index/types.py @@ -1,9 +1,10 @@ """Shared internal types for GFQL physical indexes.""" from __future__ import annotations -from typing import TYPE_CHECKING, Any, Dict, List, Literal, Mapping, Optional, Set, Tuple, TypedDict, Union +from typing import TYPE_CHECKING, Any, Dict, FrozenSet, List, Literal, Mapping, Optional, Set, Tuple, TypedDict, Union import numpy as np +import pandas as pd from graphistry.compute.typing import ArrayLike, ArrayNamespace, DataFrameT @@ -30,7 +31,8 @@ ScalarMatchValue = Union[str, int, float, bool, None, np.generic] EdgeMatchValue = Union[ ScalarMatchValue, "ASTPredicate", - List[Any], Tuple[Any, ...], Set[Any], Dict[str, Any], + List[Any], Tuple[Any, ...], Set[Any], FrozenSet[Any], Dict[str, Any], + "pd.Index", "pd.Series", ] EdgeMatch = Mapping[str, EdgeMatchValue] SimpleEqualityEdgeMatch = Mapping[str, ScalarMatchValue] diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index d7afefe6ad..0303ce0495 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -768,3 +768,79 @@ def test_rebind_edges_drops_index_on_fingerprint_mismatch(): aug["__synthetic_id__"] = aug.index reg4 = reg.rebind_edges(aug) assert reg4.has(EDGE_OUT_ADJ) and reg4.has(EDGE_IN_ADJ) + + +# --- review F1/F2/F3 regressions: the guard's value/dtype domain must equal the scan's ---- + +def _cpu_engines(): + return [e for e in ENGINES if e in ("pandas", "polars")] + + +@pytest.mark.parametrize("engine", _cpu_engines()) +def test_hop_frozenset_edge_match_parity_with_scan(typed_graph, engine): + """frozenset is membership (isin) on the scan path; the index path must not treat + it as scalar equality (a bare == is silently all-False). Parity, not zero rows.""" + 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[:5] if engine == "pandas" else g._nodes.head(5) + kwargs = dict(hops=1, return_as_wave_front=True, + edge_match={"etype": frozenset({0, 1})}, engine=engine) + base = g.hop(nodes=seeds, **kwargs) + idx = gi.hop(nodes=seeds, **kwargs) + + def n_edges(h): + return int(h._edges.shape[0]) + assert n_edges(base) > 0 # membership filter genuinely selects rows + assert n_edges(idx) == n_edges(base) + + +@pytest.mark.parametrize("engine", _cpu_engines()) +def test_hop_null_carrying_match_column_no_crash_and_parity(engine): + """Null-carrying match columns (pandas nullable Int64 / polars nulls — common after + NaN->null coercion) must not blow up mask indexing; null == val drops like scan.""" + from graphistry.Engine import Engine as _E, df_to_engine + edf = pd.DataFrame({ + "src": [0, 0, 1, 1, 2, 2], + "dst": [1, 2, 2, 3, 3, 0], + "etype": pd.array([0, None, 0, 1, None, 0], dtype="Int64"), + }) + ndf = pd.DataFrame({"id": np.arange(4)}) + 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) + seeds = g._nodes[:4] if engine == "pandas" else g._nodes.head(4) + kwargs = dict(hops=1, return_as_wave_front=True, edge_match={"etype": 0}, engine=engine) + base = g.hop(nodes=seeds, **kwargs) + idx = gi.hop(nodes=seeds, **kwargs) # was: IndexError from object-dtype mask + assert int(idx._edges.shape[0]) == int(base._edges.shape[0]) == 3 + + +@pytest.mark.parametrize("engine", _cpu_engines()) +def test_hop_dtype_mismatch_edge_match_matches_scan_error(typed_graph, engine): + """Numeric column vs string value: the scan raises (GFQLSchemaError E302 on pandas; + polars raises its own ComputeError). The index path must decline (fall back to + scan) so users get the SAME error as their engine's scan — never a silent empty + subgraph.""" + 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[:5] if engine == "pandas" else g._nodes.head(5) + kwargs = dict(hops=1, return_as_wave_front=True, + edge_match={"etype": "zero"}, engine=engine) + with pytest.raises(Exception) as base_err: + g.hop(nodes=seeds, **kwargs) + with pytest.raises(Exception) as idx_err: + gi.hop(nodes=seeds, **kwargs) + assert type(idx_err.value) is type(base_err.value) + if engine == "pandas": + from graphistry.compute.exceptions import GFQLSchemaError + assert isinstance(base_err.value, GFQLSchemaError)