diff --git a/CHANGELOG.md b/CHANGELOG.md index fe8191113e..08bcc4a7a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Performance +- **GFQL seeded typed-hop fast path (#1755)**: a seeded typed 1-hop — native chain `[n({id}), e_forward(), n({type})]` or Cypher `MATCH (m {id})-[:T]->(p) RETURN p` — previously paid the full two-pass chain machinery (~20-40ms: whole-frame combines, full-column type filters, rows-pivot projection). A new seed-first fast path recognizes exactly this shape and reduces the graph to the seed's 1-hop neighborhood before any of that work: native chain 32.6→0.93ms (35×), Cypher RETURN 39.2→1.9ms (20×) on pandas, with cuDF covered via the shared DataFrame API (30.8→4.7ms). Value-identical to the full path by construction (same rows/columns/dtypes; row order and index may differ) — the helpers return `None` and fall through for anything outside the exact shape or carrying full-path side-channels (multi-hop, variable-length, undirected, predicate filters, reverse patterns, missing bindings, list-`labels` columns, policy hooks, same-path WHERE, OPTIONAL MATCH null rows, and WITH..MATCH carried seeds all decline). Null ids/endpoints never link (membership sets are null-dropped, matching the full pipeline's joins). Verified with an independent oracle (results checked against the creator set hand-computed from the raw frames, not merely fast-vs-slow agreement) plus a differential fast-vs-full sweep across shapes × engines in `test_seeded_typed_hop_fastpath.py`. - **GFQL seeded-chain lean combine (#1755)**: The two-pass chain executor (`_chain_impl` backward pass + `combine_steps`) reconciled wavefront boundaries with full-node-frame `safe_merge`/`pandas.merge` calls even when the seeded result was a single row, so a seeded 1-hop paid a whole-frame join. When the small side is at least `4x` smaller than the full frame (and is unique on the id with no extra columns), the byte-identical result is now obtained by an `isin` membership filter instead of the join machinery. Gated narrowly on cardinality/uniqueness/column-set and pandas-only (cuDF's GPU hash-join is already sub-ms; polars takes its own engine path); `GFQL_LEAN_COMBINE=0` restores the legacy merges. Parity is byte-identical (lean-on vs lean-off) across seeded node lookup, seeded 1-hop expand, and the native typed chain, including a null-key equivalence guard. ### Documentation diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index ccf4d3f349..ec26f4b4cf 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -8,7 +8,7 @@ from graphistry.Engine import safe_merge from graphistry.util import setup_logger from graphistry.utils.json import JSONVal -from .ast import ASTObject, ASTNode, ASTEdge, from_json as ASTObject_from_json, serialize_binding_ops +from .ast import ASTObject, ASTNode, ASTEdge, Direction, from_json as ASTObject_from_json, serialize_binding_ops from .typing import DataFrameT, SeriesT from .util import generate_safe_column_name from graphistry.compute.validate.validate_schema import validate_chain_schema @@ -684,6 +684,140 @@ def _chain_otel_attrs( return attrs +def _seeded_scalar_filters(fd: Optional[Dict[str, Any]], df: DataFrameT) -> Optional[Dict[str, Any]]: + """Resolve a filter dict to plain scalar column==value pairs, or None to bail + to the general path. Mirrors filter_by_dict.resolve_filter_column exactly for + the shapes it accepts: the cypher ``label__X: True`` form maps to ``type`` + equality ONLY when no list-valued ``labels`` column exists (labels-containment + is not scalar equality) and the frame is not edge-shaped — same precedence as + the live resolver. Anything else (predicates, non-scalar values, absent + columns) bails, so the full path keeps its exact semantics incl. E301.""" + from graphistry.compute.filter_by_dict import _looks_like_edge_dataframe + if not fd: + return {} + cols = set(df.columns) + out: Dict[str, Any] = {} + for k, v in fd.items(): + if not isinstance(v, (int, float, str, bool)): + return None # predicate / non-scalar -> bail to the general path + if k in cols: + out[k] = v + elif (isinstance(k, str) and k.startswith("label__") and v is True + and "labels" not in cols and "type" in cols + and not _looks_like_edge_dataframe(df)): + out["type"] = k[len("label__"):] + else: + return None # labels-list / unknown column -> bail + return out + + +def _seeded_typed_hop_pandas_cudf( + g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge, + src: str, dst: str, node: str, direction: Direction, +) -> Optional[Plottable]: + """#1755 lever-3: engine-generic (pandas + cuDF) fast path for a scalar-filtered + seeded typed 1-hop. Value-identical to the general seeded branch for the covered + shape (all node/edge filters are plain scalars, directed) — same rows, columns, + and dtypes; row order and RangeIndex may differ — collapsing it into a + few DataFrame filters so a seeded lookup lands sub-ms. Uses only the shared + pandas/cuDF DataFrame API (no numpy array drops) so the same body runs on both + engines. Returns None to fall back for anything it does not cover (predicates, + undirected, missing columns) — the caller then runs the general branch.""" + if direction == "undirected": + return None + + nodes_df, edges_df = g._nodes, g._edges + if nodes_df is None or edges_df is None: + return None + n0f = _seeded_scalar_filters(n0.filter_dict, nodes_df) + n2f = _seeded_scalar_filters(n2.filter_dict, nodes_df) + ef = _seeded_scalar_filters(e1.edge_match, edges_df) + if n0f is None or n2f is None or ef is None: + return None + from_col, to_col = (src, dst) if direction == "forward" else (dst, src) + + # from-side seed FIRST: reduce edges to the seed's out-edges before the + # edge_match compare, so the type filter runs on the tiny frontier rather than + # all edges — this is what makes a seeded lookup sub-ms. The id filter goes + # first (int, unique -> ~1 row in one pass) so any remaining object filters + # (label__X->type) run on that tiny survivor frame, not the whole node table. + if n0f: + seed_nodes = nodes_df + for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1): + seed_nodes = seed_nodes[seed_nodes[k] == v] + edges = edges_df[edges_df[from_col].isin(seed_nodes[node].dropna())] + else: + edges = edges_df + if ef: # typed edge (edge_match) — now on the reduced frontier + for k, v in ef.items(): + edges = edges[edges[k] == v] + + # Gather candidate endpoint nodes (both endpoints of surviving edges), then run + # the dest filter, dangling-edge drop and final-node selection on the small + # candidate/edge frames. Selecting from nodes_df keeps only real nodes, so the + # endpoint-in-nodes check subsumes the old NaN-endpoint guard. Membership sets + # are dropna()'d: pandas .isin matches NaN<->NaN, but the general branch's BFS + # joins never join on null keys, so a null id/endpoint must not link. + cand = nodes_df[ + nodes_df[node].isin(edges[src].dropna()) | nodes_df[node].isin(edges[dst].dropna()) + ].drop_duplicates(subset=[node]) + if n2f: # destination-node filter (to-side) + n2_cand = cand + for k, v in n2f.items(): + n2_cand = n2_cand[n2_cand[k] == v] + n2_ok = n2_cand[node] + else: + n2_ok = cand[node] + to_vals = edges[to_col] + keep = edges[src].isin(cand[node].dropna()) & edges[dst].isin(cand[node].dropna()) & to_vals.isin(n2_ok.dropna()) + edges = edges[keep] + cand = cand[cand[node].isin(edges[src]) | cand[node].isin(edges[dst])] + return g.nodes(cand).edges(edges) + + +def _seeded_typed_return_dst_pandas_cudf( + g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge, + src: str, dst: str, node: str, direction: Direction, +) -> Optional[Tuple[DataFrameT, DataFrameT]]: + """#1755 cypher RETURN-alias fast path: like _seeded_typed_hop_pandas_cudf but + returns ONLY the destination (RETURN-alias) node rows + surviving edges — no + seed-node gather, no Plottable round-trip — so the seeded cypher projection + lands sub-ms. Engine-generic (pandas + cuDF): only the shared DataFrame API, + no numpy array drops. Returns ``(dst_node_rows, edges)`` or None to fall back.""" + if direction == "undirected": + return None + nodes_df, edges_df = g._nodes, g._edges + if nodes_df is None or edges_df is None: + return None + n0f = _seeded_scalar_filters(n0.filter_dict, nodes_df) + n2f = _seeded_scalar_filters(n2.filter_dict, nodes_df) + ef = _seeded_scalar_filters(e1.edge_match, edges_df) + if n0f is None or n2f is None or ef is None or not n0f: + return None + from_col, to_col = (src, dst) if direction == "forward" else (dst, src) + # id-first seed reduction: filter by the id column first (int/unique -> ~1 row) + # so any remaining object filters (label__X->type) run on the tiny survivor + # frame, never materializing an object column over the whole node table. + # Membership sets are dropna()'d: pandas .isin matches NaN<->NaN, but the full + # pipeline's joins never join on null keys, so a null id/endpoint must not link. + seed_nodes = nodes_df + for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1): + seed_nodes = seed_nodes[seed_nodes[k] == v] + edges = edges_df[edges_df[from_col].isin(seed_nodes[node].dropna())] + if ef: + for k, v in ef.items(): + edges = edges[edges[k] == v] + # destination nodes = real nodes that are edge to-endpoints, then the dest + # filter, dangling-edge drop and dedup on the small dst/edge frames. + dstn = nodes_df[nodes_df[node].isin(edges[to_col].dropna())] + if n2f: + for k, v in n2f.items(): + dstn = dstn[dstn[k] == v] + edges = edges[edges[to_col].isin(dstn[node].dropna())] + dstn = dstn[dstn[node].isin(edges[to_col].dropna())].drop_duplicates(subset=[node]) + return dstn, edges + + def _try_chain_fast_path( g_in: Plottable, ops: List[ASTObject], @@ -731,12 +865,16 @@ def _materialize_fast_path_graph() -> Plottable: if not (isinstance(n2, ASTNode) and n2._name is None and n2.query is None): return None if not (isinstance(e1, ASTEdge) and e1.is_simple_single_hop() - and e1.edge_match is None and e1.source_node_match is None + and e1.source_node_match is None and e1.destination_node_match is None and e1._name is None and e1.source_node_query is None and e1.destination_node_query is None and e1.edge_query is None and not e1.include_zero_hop_seed and not e1.prune_to_endpoints): # prune keeps only the arrival side -> full path return None + # #1755 lever-3: a typed edge (edge_match, e.g. -[:HAS_CREATOR]->) is a plain + # equality/predicate filter on the edge frame — apply it in the fast-path body + # below rather than falling through to the full two-pass machinery. source/dest + # node match + edge_query (richer predicates) still bail above. direction = e1.direction unconstrained = not n0.filter_dict and not n2.filter_dict if not unconstrained and direction == "undirected": @@ -745,20 +883,67 @@ def _materialize_fast_path_graph() -> Plottable: if g._nodes is None or g._edges is None: return None src, dst, node = g._source, g._destination, g._node - # Keep only edges with BOTH endpoints in the node table (the full path drops - # dangling edges via its joins). dropna so a NaN node id can't validate a NaN - # endpoint — .isin treats NaN as matchable but the BFS joins never match NaN<->NaN. - node_ids = g._nodes[node].dropna() - edges = g._edges[g._edges[src].isin(node_ids) & g._edges[dst].isin(node_ids)] - if not unconstrained: + if src is None or dst is None or node is None: + return None # no edge/node bindings -> can't fast-path; full path handles it + concat = df_concat(engine_concrete) + if unconstrained: + # No node filter to reduce by: validate BOTH endpoints against the full + # node table (the full path drops dangling edges via its joins). dropna so + # a NaN node id can't validate a NaN endpoint — .isin treats NaN as + # matchable but the BFS joins never match NaN<->NaN. + node_ids = g._nodes[node].dropna() + edges = g._edges[g._edges[src].isin(node_ids) & g._edges[dst].isin(node_ids)] + if e1.edge_match: + # typed edge (e.g. -[:HAS_CREATOR]->) — same edge-frame filter the full + # hop applies, so the result set is identical. + edges = filter_by_dict(edges, e1.edge_match, engine_abs) + else: + # #1755 lever-3 seed-first: a seeded 1-hop must be O(result), not O(E). + # Reduce edges by the selective node filter(s) BEFORE the typed-edge scan + # and endpoint validation, so the expensive object/isin passes run on the + # tiny frontier, not all edges. The from-side ids come from the node table + # (so that endpoint is validated); the node gather below validates the to + # side and drops any edge dangling off the node table. + # pandas + cuDF: a scalar-filtered seeded typed hop collapses to a few + # DataFrame filters (sub-ms); falls back to the general branch below for + # predicates / undirected / missing columns (and non-pandas/cuDF engines). + if engine_concrete in (Engine.PANDAS, Engine.CUDF): + _fast_res = _seeded_typed_hop_pandas_cudf(g, n0, n2, e1, src, dst, node, direction) + if _fast_res is not None: + return _fast_res from_col, to_col = (src, dst) if direction == "forward" else (dst, src) + edges = g._edges if n0.filter_dict: - ids = filter_by_dict(g._nodes, n0.filter_dict, engine_abs)[node] - edges = edges[edges[from_col].isin(ids)] + from_ids = filter_by_dict(g._nodes, n0.filter_dict, engine_abs)[node] + edges = edges[edges[from_col].isin(from_ids)] + if e1.edge_match: + edges = filter_by_dict(edges, e1.edge_match, engine_abs) if n2.filter_dict: - ids = filter_by_dict(g._nodes, n2.filter_dict, engine_abs)[node] - edges = edges[edges[to_col].isin(ids)] - concat = df_concat(engine_concrete) + # Apply the destination filter to the SMALL set of gathered dst nodes, + # not the full node table — an O(N) object/type scan on all nodes is + # exactly the tax we're removing. Gather the frontier's dst nodes + # (small isin key), filter those, then drop edges to the losers. + to_present = edges[to_col].dropna().unique() + to_nodes = filter_by_dict( + g._nodes[g._nodes[node].isin(to_present)], n2.filter_dict, engine_abs) + edges = edges[edges[to_col].isin(to_nodes[node])] + # Validate endpoints + build result nodes on the reduced edge set (small + # isin key -> small hashtable; no O(E)-values scan). Engine-agnostic + # (pandas + cuDF): gather candidate endpoint nodes, drop edges dangling off + # the node table, then keep only nodes still referenced by a surviving edge. + ep = concat([ + edges[[src]].rename(columns={src: node}), + edges[[dst]].rename(columns={dst: node}), + ]).drop_duplicates() + cand = g._nodes[g._nodes[node].isin(ep[node])].drop_duplicates(subset=[node]) + valid = cand[node].dropna() + edges = edges[edges[src].isin(valid) & edges[dst].isin(valid)] + final = concat([ + edges[[src]].rename(columns={src: node}), + edges[[dst]].rename(columns={dst: node}), + ]).drop_duplicates() + nodes = cand[cand[node].isin(final[node])] + return g.nodes(nodes).edges(edges) endpoints = concat([ edges[[src]].rename(columns={src: node}), edges[[dst]].rename(columns={dst: node}), diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index d67b5ec46f..c4c17f34cd 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -976,6 +976,103 @@ def _run_logical_pass_pipeline(logical_plan: LogicalPlan, ctx: PlanContext) -> L return PassManager(DEFAULT_LOGICAL_PASSES, DEFAULT_TIER2_PASSES).run(logical_plan, ctx).plan + +def _execute_seeded_typed_hop_fast_path( + base_graph: Plottable, + compiled_query: CompiledCypherQuery, + physical_plan: "PhysicalPlan", + *, + engine: Union[EngineAbstract, str], + policy: Optional[PolicyDict], + context: ExecutionContext, + start_nodes: Optional[DataFrameT] = None, +) -> Optional[Plottable]: + """#1755 cypher-surface fast path: a seeded typed 1-hop with a whole-row node + RETURN — ``MATCH (m {id})-[:T]->(p) RETURN p`` — reduces the graph to the seed's + 1-hop neighborhood with a few DataFrame filters (pandas + cuDF), then applies the + RETURN projection to just the destination rows (value-identical to the full + path: same rows/columns/dtypes; row order and RangeIndex may differ; sub-ms + because the frame is tiny). Returns None to fall through for anything outside + this exact shape or carrying side-channels (policy, same-path WHERE, OPTIONAL + null-row, carried reentry seeds).""" + if start_nodes is not None: + # A carried seed set (WITH..MATCH reentry) restricts n0 to those rows; the fast + # path derives its seed from n0.filter_dict alone, so engaging here would + # silently widen the seed back to the whole graph. Decline. + return None + if policy: + # The full path fires prechain/postchain/postload policy hooks; the lean + # projection below never enters chain(), so engaging would silently skip + # them. Policy-gated queries always take the full path (native twin gates + # identically in chain._try_chain_fast_path). + return None + if compiled_query.chain.where: + # Same-path WHERE entries (e.g. WHERE p.id < m.id) are evaluated by the + # full pipeline, not by the seed-first reduction — engaging would drop them. + return None + if compiled_query.empty_result_row is not None: + # OPTIONAL MATCH null-row semantics: on no match the full path emits the + # openCypher null row; the lean projection would return an empty frame. + return None + requested_engine = resolve_engine(cast(Any, engine), base_graph) + if requested_engine not in (Engine.PANDAS, Engine.CUDF): + return None + projection = compiled_query.result_projection + if projection is None or projection.table != "nodes": + return None + # Only a single whole-row node alias (RETURN p). Multi-alias returns (RETURN + # m, p) combine aliases into one row per match — different shape — so bail. + proj_cols = projection.columns + if len(proj_cols) != 1 or proj_cols[0].kind != "whole_row": + return None + if compiled_query.execution_extras is not None and ( + compiled_query.execution_extras.connected_match_join is not None + or compiled_query.execution_extras.connected_optional_match is not None + ): + return None + ops = list(compiled_query.chain.chain) + if len(ops) != 4: + return None + n0, e1, n2, call = ops + if not (isinstance(n0, ASTNode) and isinstance(e1, ASTEdge) + and isinstance(n2, ASTNode) and isinstance(call, ASTCall)): + return None + # Only a genuine SINGLE hop. A variable-length edge (-[*1..2]->) is still one + # ASTEdge but expands to multiple hops, so the seeded 1-hop reduction below + # would silently truncate it. Reuse the same canonical gate the native fast + # path uses (also rejects hop labels / output slicing / fixed-point). See + # test_engine_polars_chain::TestVarlenAliasHopGate. + if not e1.is_simple_single_hop(): + return None + if call.function != "rows": + return None + node, src, dst = base_graph._node, base_graph._source, base_graph._destination + if node is None or src is None or dst is None: + return None + # RETURN alias must be the DESTINATION node (n2) and the seed must sit on the + # source node (n0) — the forward seeded shape MATCH (m {id})-[:T]->(p) RETURN p. + # Other alias/seed placements (e.g. reverse patterns where the seed is on the + # RETURN node) fall back to the full path. + if n2._name != projection.alias: + return None + if not (n0.filter_dict and any(not str(k).startswith("label__") for k in n0.filter_dict)): + return None # n0 must carry a selective (non-label) seed + direction = e1.direction + from graphistry.compute.chain import _seeded_typed_return_dst_pandas_cudf + dst_res = _seeded_typed_return_dst_pandas_cudf( + base_graph, n0, n2, e1, src, dst, node, direction) + if dst_res is None: + return None + p_rows, _edges = dst_res + # Lean projection: p_rows already IS the RETURN-alias (destination) node set. + # Tag with the alias and reuse apply_result_projection for the exact + # column-order/flatten semantics — all on a handful of rows, so seeded cypher + # stays sub-ms (vs the ~25ms rows-pivot pipeline on the full graph). + tagged = p_rows.assign(**{projection.alias: True}) + result = base_graph.nodes(tagged) + return apply_result_projection(result, projection) + + def _execute_compiled_query_via_physical_plan( base_graph: Plottable, *, @@ -1015,6 +1112,11 @@ def _execute_compiled_query_via_physical_plan( fast_count = _execute_two_hop_count_fast_path(base_graph, compiled_query.chain, engine=engine) if fast_count is not None: return fast_count + fast_hop = _execute_seeded_typed_hop_fast_path( + base_graph, compiled_query, physical_plan, + engine=engine, policy=policy, context=context, start_nodes=start_nodes) + if fast_hop is not None: + return fast_hop return _execute_compiled_query_chain_non_union( base_graph, compiled_query=compiled_query, diff --git a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py new file mode 100644 index 0000000000..8ff8b24174 --- /dev/null +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -0,0 +1,450 @@ +"""Parity + gating tests for the #1755 seeded typed-hop fast path. + +Two layers close the seeded-Cypher abstraction tax on pandas: + * native — `_try_chain_fast_path` / `_seeded_typed_hop_pandas_cudf` accelerate + a seeded typed 1-hop chain [n({id}), e(edge_match), n({type})]; + * cypher — `_execute_seeded_typed_hop_fast_path` accelerates the lowered + MATCH (m {id})-[:T]->(p) RETURN p string surface. + +Both are value-identical to the full path by construction (same rows/columns; +row order and index may differ — comparisons canonicalize); these tests pin that +(fast-on vs fast-off, differential) and that the fast path actually ENGAGES for +the accelerated shapes and DECLINES (falls through) for everything else, +including full-path side-channels (policy hooks, same-path WHERE, OPTIONAL +null rows, WITH..MATCH carried seeds, list-`labels` columns, null ids). +""" +import numpy as np +import pandas as pd +import pytest + +import graphistry +from graphistry.compute.ast import n, e_forward, e_reverse +import graphistry.compute.chain as chain_mod +import graphistry.compute.gfql_unified as gfql_unified + + +def _graph(n_persons=1500, n_messages=6000, seed=0): + """Message -> Person HAS_CREATOR graph (the #1755 probe shape). `age` is only + on Person rows, so the concatenated node frame carries a float `age` column + with NaN on Messages — exercising mixed dtypes through the projection.""" + rng = np.random.default_rng(seed) + persons = pd.DataFrame({ + "id": np.arange(n_persons), "type": "Person", + "age": rng.integers(20, 60, n_persons), + }) + messages = pd.DataFrame({"id": np.arange(n_persons, n_persons + n_messages), "type": "Message"}) + ndf = pd.concat([persons, messages], ignore_index=True) + edf = pd.DataFrame({ + "src": np.arange(n_persons, n_persons + n_messages), + "dst": rng.integers(0, n_persons, n_messages), + "type": "HAS_CREATOR", + }) + return graphistry.nodes(ndf, "id").edges(edf, "src", "dst"), n_persons + + +def _canon_nodes(res): + nodes = res._nodes + df = nodes.to_pandas() if hasattr(nodes, "to_pandas") else pd.DataFrame(nodes) + cols = sorted(str(c) for c in df.columns) + df.columns = [str(c) for c in df.columns] + return df.sort_values(cols).reset_index(drop=True)[cols] if cols else df + + +def _engine_graph(engine): + """Build the probe graph in the requested engine (cuDF skips if unavailable).""" + g, P = _graph() + if engine == "pandas": + return g, P + if engine == "cudf": + cudf = pytest.importorskip("cudf") + return graphistry.nodes( + cudf.from_pandas(pd.DataFrame(g._nodes)), "id" + ).edges(cudf.from_pandas(pd.DataFrame(g._edges)), "src", "dst"), P + raise ValueError(engine) + + +def _run_diff(g, engine, query, fast): + """Run `query` with BOTH seeded fast paths (native + cypher) either live or + forced-off, so a single differential asserts fast==full for any shape/engine.""" + real_native = chain_mod._try_chain_fast_path + real_cyp = gfql_unified._execute_seeded_typed_hop_fast_path + try: + if not fast: + chain_mod._try_chain_fast_path = lambda *a, **k: None + gfql_unified._execute_seeded_typed_hop_fast_path = lambda *a, **k: None + return g.gfql(query, engine=engine) + finally: + chain_mod._try_chain_fast_path = real_native + gfql_unified._execute_seeded_typed_hop_fast_path = real_cyp + + +def _oracle_creators(g, seed): + """INDEPENDENT ground truth (NOT via the query engine): the Person ids that + are HAS_CREATOR destinations of the seeded Message. Computed directly from + the raw edge/node frames with plain pandas boolean masks, so it cannot share + a bug with either the fast path or the full chain path.""" + edf = g._edges + ndf = g._nodes + creator_ids = edf.loc[ + (edf["src"] == seed) & (edf["type"] == "HAS_CREATOR"), "dst" + ].tolist() + # restrict to Person-typed endpoints (the typed n({type:Person}) filter) + person_ids = set(ndf.loc[ndf["type"] == "Person", "id"].tolist()) + return sorted(cid for cid in creator_ids if cid in person_ids) + + +# --------------------------------------------------------------------------- +# native seeded typed hop ([n({id}), e(edge_match), n({type})]) +# --------------------------------------------------------------------------- + +class TestNativeSeededTypedHop: + def _run(self, g, ops, force_full): + real = chain_mod._try_chain_fast_path + try: + if force_full: + chain_mod._try_chain_fast_path = lambda *a, **k: None + return g.gfql(ops, engine="pandas") + finally: + chain_mod._try_chain_fast_path = real + + @pytest.mark.parametrize("ops_name", ["typed_creator", "typed_person", "typed_reverse"]) + def test_parity_fast_vs_full(self, ops_name): + g, P = _graph() + seed = P + 42 + ops = { + "typed_creator": [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n()], + "typed_person": [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], + "typed_reverse": [n({"type": "Person"}), e_reverse(edge_match={"type": "HAS_CREATOR"}), n({"id": seed})], + }[ops_name] + fast = self._run(g, ops, force_full=False) + full = self._run(g, ops, force_full=True) + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + def test_fast_path_engages_on_typed_hop(self, monkeypatch): + g, P = _graph() + seed = P + 42 + hits = {"n": 0} + real = chain_mod._seeded_typed_hop_pandas_cudf + + def spy(*a, **k): + r = real(*a, **k) + if r is not None: + hits["n"] += 1 + return r + + monkeypatch.setattr(chain_mod, "_seeded_typed_hop_pandas_cudf", spy) + g.gfql([n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], engine="pandas") + assert hits["n"] >= 1 + + def test_independent_oracle_values(self): + """Falsifiability (a): the returned Person must be the ACTUAL creator, + hand-computed from raw frames — not merely fast==full (both could share + a bug). Pins VALUES, not just self-consistency.""" + g, P = _graph() + seed = P + 42 + oracle = _oracle_creators(g, seed) + assert len(oracle) >= 1, "probe seed must have >=1 Person creator" + res = g.gfql( + [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], + engine="pandas", + ) + got = sorted(pd.DataFrame(res._nodes)["id"].tolist()) + # native chain _nodes = the UNION of all bound path endpoints: the seed + # Message (n0) AND its Person creators (n2). The oracle is that union. + expected = sorted(set(oracle) | {seed}) + assert got == expected, f"fast path returned {got}, oracle says {expected}" + + def test_numpy_helper_declines_predicate_and_undirected(self): + from graphistry.compute.predicates.numeric import gt + g, P = _graph() + node, src, dst = "id", "src", "dst" + # predicate (non-scalar) edge filter -> decline + assert chain_mod._seeded_typed_hop_pandas_cudf( + g.materialize_nodes(), n({"id": P + 1}), n(), e_forward(edge_match={"type": gt(0)}), + src, dst, node, "forward") is None + # undirected -> decline + assert chain_mod._seeded_typed_hop_pandas_cudf( + g.materialize_nodes(), n({"id": P + 1}), n(), e_forward(), + src, dst, node, "undirected") is None + + @pytest.mark.parametrize("ops_name,reason", [ + ("two_hop", "multi-hop chain, not a single hop"), + ("node_predicate", "non-scalar (predicate) node filter"), + ]) + def test_native_declines_and_stays_correct(self, ops_name, reason, monkeypatch): + from graphistry.compute.predicates.numeric import gt + g, P = _graph() + seed = P + 42 + ops = { + "two_hop": [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n(), e_reverse(), n()], + "node_predicate": [n({"id": seed}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"age": gt(0)})], + }[ops_name] + hits = {"n": 0} + real = chain_mod._seeded_typed_hop_pandas_cudf + + def spy(*a, **k): + r = real(*a, **k) + if r is not None: + hits["n"] += 1 + return r + + monkeypatch.setattr(chain_mod, "_seeded_typed_hop_pandas_cudf", spy) + fast = g.gfql(ops, engine="pandas") + monkeypatch.undo() + chain_mod._try_chain_fast_path, saved = (lambda *a, **k: None), chain_mod._try_chain_fast_path + try: + full = g.gfql(ops, engine="pandas") + finally: + chain_mod._try_chain_fast_path = saved + assert hits["n"] == 0, f"numpy fast path must decline {reason}" + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + +# --------------------------------------------------------------------------- +# cypher surface (MATCH (m {id})-[:T]->(p) RETURN p) +# --------------------------------------------------------------------------- + +class TestCypherSeededTypedHop: + def _run(self, g, cy, force_full): + real = gfql_unified._execute_seeded_typed_hop_fast_path + try: + if force_full: + gfql_unified._execute_seeded_typed_hop_fast_path = lambda *a, **k: None + return g.gfql(cy, engine="pandas") + finally: + gfql_unified._execute_seeded_typed_hop_fast_path = real + + @pytest.mark.parametrize("label", ["p:Person", "p"]) + def test_parity_return_p(self, label): + g, P = _graph() + seed = P + 42 + cy = f"MATCH (m:Message {{id: {seed}}})-[:HAS_CREATOR]->({label}) RETURN p" + fast = self._run(g, cy, force_full=False) + full = self._run(g, cy, force_full=True) + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + def test_parity_no_match(self): + g, P = _graph() + cy = "MATCH (m:Message {id: 999999})-[:HAS_CREATOR]->(p:Person) RETURN p" + fast = self._run(g, cy, force_full=False) + full = self._run(g, cy, force_full=True) + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + def test_independent_oracle_values(self): + """Falsifiability (a) on the cypher surface: the p-rows returned by the + lowered MATCH ... RETURN p must be exactly the hand-computed creators.""" + g, P = _graph() + seed = P + 42 + oracle = _oracle_creators(g, seed) + assert len(oracle) >= 1 + res = g.gfql( + f"MATCH (m:Message {{id: {seed}}})-[:HAS_CREATOR]->(p:Person) RETURN p", + engine="pandas", + ) + df = pd.DataFrame(res._nodes) + # RETURN p prefixes projected columns with the alias; find the id column. + id_col = "p.id" if "p.id" in df.columns else "id" + got = sorted(df[id_col].tolist()) + assert got == oracle, f"cypher fast path returned {got}, oracle says {oracle}" + + def test_fast_path_engages_on_seeded_return(self, monkeypatch): + g, P = _graph() + seed = P + 42 + hits = {"n": 0} + real = gfql_unified._execute_seeded_typed_hop_fast_path + + def spy(*a, **k): + r = real(*a, **k) + if r is not None: + hits["n"] += 1 + return r + + monkeypatch.setattr(gfql_unified, "_execute_seeded_typed_hop_fast_path", spy) + g.gfql(f"MATCH (m:Message {{id: {seed}}})-[:HAS_CREATOR]->(p:Person) RETURN p", engine="pandas") + assert hits["n"] >= 1 + + @pytest.mark.parametrize("cy_tmpl,reason", [ + ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN m, p", "multi-alias"), + ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN p.age", "field projection"), + ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN m", "return source"), + ("MATCH (p:Person)<-[:HAS_CREATOR]-(m:Message {{id: {s}}}) RETURN p", "reverse (seed on return node)"), + # variable-length edges are one ASTEdge but multiple hops — must decline or + # the 1-hop reduction silently truncates (regressed polars varlen parity). + ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR*1..2]->(p) RETURN p", "varlen range hop"), + ("MATCH (m:Message {{id: {s}}})-[*1..2]->(p) RETURN p", "varlen untyped hop"), + ]) + def test_declines_and_stays_correct(self, cy_tmpl, reason): + g, P = _graph() + seed = P + 42 + cy = cy_tmpl.format(s=seed) + real = gfql_unified._execute_seeded_typed_hop_fast_path + hits = {"n": 0} + + def spy(*a, **k): + r = real(*a, **k) + if r is not None: + hits["n"] += 1 + return r + + gfql_unified._execute_seeded_typed_hop_fast_path = spy + try: + fast = g.gfql(cy, engine="pandas") + finally: + gfql_unified._execute_seeded_typed_hop_fast_path = real + full = self._run(g, cy, force_full=True) + assert hits["n"] == 0, f"fast path must decline {reason}" + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + +# --------------------------------------------------------------------------- +# Differential fast-vs-full sweep across shapes × engines +# --------------------------------------------------------------------------- + +class TestSeededFastPathDifferentialSweep: + """Broad fast-vs-full byte-parity across a matrix of seeded shapes × engines. + + Answers the coverage question: is every ENGAGED shape verified through BOTH + the fast path and the full path, on every supported engine (pandas + cuDF)? + This complements the ~1114 implicit DECLINE-path exercises the cypher + conformance suite already drives through `_execute_seeded_typed_hop_fast_path` + (real queries that fall through and still return correct values). Here we pin + the ENGAGED shapes byte-identical fast-vs-full so the specialization can never + silently diverge from the general path on any covered engine.""" + + SHAPES = { + "native_typed": lambda s: [n({"id": s}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], + "native_untyped": lambda s: [n({"id": s}), e_forward(), n()], + "native_reverse": lambda s: [n({"type": "Person"}), e_reverse(edge_match={"type": "HAS_CREATOR"}), n({"id": s})], + "native_type_src": lambda s: [n({"type": "Message"}), e_forward(edge_match={"type": "HAS_CREATOR"}), n({"type": "Person"})], + "cypher_p_labelled": lambda s: f"MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p:Person) RETURN p", + "cypher_p_unlabelled": lambda s: f"MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR]->(p) RETURN p", + "cypher_no_match": lambda s: "MATCH (m:Message {id: 9999999})-[:HAS_CREATOR]->(p:Person) RETURN p", + } + + @pytest.mark.parametrize("engine", ["pandas", "cudf"]) + @pytest.mark.parametrize("shape", list(SHAPES)) + def test_fast_vs_full_parity(self, engine, shape): + g, P = _engine_graph(engine) + query = self.SHAPES[shape](P + 42) + fast = _run_diff(g, engine, query, fast=True) + full = _run_diff(g, engine, query, fast=False) + pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full)) + + @pytest.mark.parametrize("engine", ["pandas", "cudf"]) + def test_engaged_shapes_match_independent_oracle(self, engine): + """The two headline engaged shapes must equal the hand-computed creator set + (native = {seed} ∪ creators; cypher RETURN p = creators) on each engine — + not merely fast==full, since the full path could share a bug.""" + g, P = _engine_graph(engine) + pd_g, _ = _graph() # pandas twin for the independent oracle + seed = P + 42 + oracle = _oracle_creators(pd_g, seed) + assert len(oracle) >= 1 + native = _run_diff(g, engine, self.SHAPES["native_typed"](seed), fast=True) + got_native = sorted(_canon_nodes(native)["id"].tolist()) + assert got_native == sorted(set(oracle) | {seed}) + cyp = _run_diff(g, engine, self.SHAPES["cypher_p_labelled"](seed), fast=True) + cdf = _canon_nodes(cyp) + idc = "p.id" if "p.id" in cdf.columns else "id" + assert sorted(cdf[idc].tolist()) == oracle + + +# --------------------------------------------------------------------------- +# side-channel decline gates + semantics parity (review-skill blockers, wave 2026-07-21) +# --------------------------------------------------------------------------- + +class TestFastPathSideChannelGates: + """Each test pins a shape where ENGAGING the fast path was empirically proven + to diverge from the full path (probes in plans/review-pr-1759/): the fix is + to decline (or match semantics exactly), and these lock that in.""" + + def _canon_edges(self, res): + edges = res._edges + df = edges.to_pandas() if hasattr(edges, "to_pandas") else pd.DataFrame(edges) + cols = sorted(str(c) for c in df.columns) + df.columns = [str(c) for c in df.columns] + return df.sort_values(cols).reset_index(drop=True)[cols] if cols else df + + def test_labels_column_precedence(self): + """label__X must follow resolve_filter_column: `labels` (list containment) + wins over `type` — the fast path's type-equality shortcut must decline.""" + ndf = pd.DataFrame({ + "id": [0, 1], + "type": ["Person", "Person"], + "labels": [["Admin"], ["Person"]], # containment says only id=1 is a Person + }) + edf = pd.DataFrame({"src": [0], "dst": [1], "type": ["KNOWS"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + q = "MATCH (m {id: 0})-[{type:'KNOWS'}]->(p:Person) RETURN p" + fast = _canon_nodes(_run_diff(g, "pandas", q, fast=True)) + full = _canon_nodes(_run_diff(g, "pandas", q, fast=False)) + pd.testing.assert_frame_equal(fast, full) + + def test_null_ids_never_link(self): + """NaN node id + NaN edge endpoint: pandas .isin matches NaN<->NaN but the + full pipeline's joins never link null keys — nodes AND edges must agree.""" + ndf = pd.DataFrame({"id": [0.0, 1.0, np.nan], + "type": ["Person", "Message", "Message"]}) + edf = pd.DataFrame({"src": [1.0, np.nan], "dst": [0.0, 0.0], + "type": ["HAS_CREATOR", "HAS_CREATOR"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + ops = [n({"type": "Message"}), e_forward(edge_match={"type": "HAS_CREATOR"}), + n({"type": "Person"})] + fast_r = _run_diff(g, "pandas", ops, fast=True) + full_r = _run_diff(g, "pandas", ops, fast=False) + pd.testing.assert_frame_equal(_canon_nodes(fast_r), _canon_nodes(full_r)) + pd.testing.assert_frame_equal(self._canon_edges(fast_r), self._canon_edges(full_r)) + + def test_policy_hooks_not_skipped(self): + """A policy dict must force the full path so pre/post hooks fire.""" + g, P = _graph() + seed = P + 7 + fired = [] + + def hook(ctx): + fired.append(ctx.get("phase", "?")) + return None + q = f"MATCH (m {{id: {seed}}})-[{{type:'HAS_CREATOR'}}]->(p {{type:'Person'}}) RETURN p" + policy = {"prechain": hook, "postchain": hook} + r_pol = g.gfql(q, engine="pandas", policy=policy) + assert fired, "policy hooks must fire (fast path must decline under policy)" + r_nopol = g.gfql(q, engine="pandas") + pd.testing.assert_frame_equal(_canon_nodes(r_pol), _canon_nodes(r_nopol)) + + def test_same_path_where_not_dropped(self): + """WHERE p.id < m.id (same-path cross-alias) must not be silently dropped.""" + ndf = pd.DataFrame({"id": [0, 1, 2], "type": ["Message", "Person", "Person"], + "age": [np.nan, 31.0, 32.0]}) + edf = pd.DataFrame({"src": [0, 0], "dst": [1, 2], "type": ["T", "T"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + q = "MATCH (m {id: 0})-[{type:'T'}]->(p {type:'Person'}) WHERE p.id > m.id + 1 RETURN p" + fast = _canon_nodes(_run_diff(g, "pandas", q, fast=True)) + full = _canon_nodes(_run_diff(g, "pandas", q, fast=False)) + pd.testing.assert_frame_equal(fast, full) + + def test_optional_match_null_row(self): + """No-match OPTIONAL MATCH returns the openCypher null row, not empty.""" + ndf = pd.DataFrame({"id": [0, 1], "type": ["Message", "Person"]}) + edf = pd.DataFrame({"src": [0], "dst": [1], "type": ["OTHER"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + q = "OPTIONAL MATCH (m {id: 0})-[{type:'MISSING'}]->(p {type:'Person'}) RETURN p" + fast = _canon_nodes(_run_diff(g, "pandas", q, fast=True)) + full = _canon_nodes(_run_diff(g, "pandas", q, fast=False)) + pd.testing.assert_frame_equal(fast, full) + + def test_with_match_reentry_carried_seeds(self): + """WITH..MATCH reentry hands carried seeds via start_nodes; the fast path + deriving its seed from the filter alone would silently widen the seed set. + Node 0 is excluded by the WITH but matches the reentry filter and has a + LIKES edge -> divergence unless the fast path declines.""" + ndf = pd.DataFrame({"id": [0, 1, 2, 10, 11, 13], + "kind": ["person"] * 3 + ["post"] * 3}) + edf = pd.DataFrame({"src": [0, 0, 1, 2, 0], "dst": [1, 2, 10, 11, 13], + "type": ["KNOWS", "KNOWS", "LIKES", "LIKES", "LIKES"]}) + g = graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + q = ("MATCH (p {id:0})-[{type:'KNOWS'}]->(a) WITH a " + "MATCH (a {kind:'person'})-[{type:'LIKES'}]->(b) RETURN b") + fast = _canon_nodes(_run_diff(g, "pandas", q, fast=True)) + full = _canon_nodes(_run_diff(g, "pandas", q, fast=False)) + pd.testing.assert_frame_equal(fast, full) + assert sorted(fast["b.id"].tolist()) == [10, 11] diff --git a/graphistry/tests/compute/test_chain.py b/graphistry/tests/compute/test_chain.py index e14ed6e0db..1f084583c5 100644 --- a/graphistry/tests/compute/test_chain.py +++ b/graphistry/tests/compute/test_chain.py @@ -675,13 +675,17 @@ def topd(df): ("hop_fwd_dst_filter", lambda: [n(), e_forward(hops=1), n({'attr': 40})]), ("hop_fwd_both_filter", lambda: [n({'attr': 10}), e_forward(hops=1), n({'attr': 30})]), ("hop_rev_dst_filter", lambda: [n(), e_reverse(hops=1), n({'attr': 10})]), + # #1755 lever-3: typed edges (edge_match) are now a fast shape — a plain edge + # filter applied on the (seed-reduced) frontier, not a fall-through. + ("edge_match_unconstrained", lambda: [n(), e_forward(hops=1, edge_match={'w': 5}), n()]), + ("edge_match_seeded", lambda: [n({'attr': 10}), e_forward(hops=1, edge_match={'w': 5}), n()]), + ("edge_match_dst_filter", lambda: [n(), e_forward(hops=1, edge_match={'w': 5}), n({'attr': 30})]), ] # shapes that BYPASS the fast path (still must be correct via the full path) _BYPASS_SHAPES = [ ("hops_2", lambda: [n(), e_forward(hops=2), n()]), ("filtered_undirected", lambda: [n({'attr': 10}), e_undirected(hops=1), n({'attr': 30})]), - ("edge_match", lambda: [n(), e_forward(hops=1, edge_match={'w': 5}), n()]), ("named_node", lambda: [n(name='x'), e_forward(hops=1), n()]), # prune_to_endpoints: fast path returns both endpoints; full path keeps only the # arrival side. Must bypass the fast path (regression guard for the prune gate). @@ -751,6 +755,9 @@ def test_fast_path_gating_returns_none_for_ineligible(): [n({'attr': 20})], [n(), e_forward(hops=1), n()], [n(), e_reverse(hops=1), n()], + # #1755 lever-3: typed edges are now accepted (edge filter on the frontier) + [n(), e_forward(hops=1, edge_match={'w': 5}), n()], + [n({'attr': 10}), e_forward(hops=1, edge_match={'w': 5}), n()], ] for ops in eligible: assert _try_chain_fast_path(g, ops, Engine.PANDAS, None) is not None, f"should accept {ops}" @@ -758,7 +765,6 @@ def test_fast_path_gating_returns_none_for_ineligible(): ineligible = [ ("hops_2", [n(), e_forward(hops=2), n()], None, Engine.PANDAS), ("filtered_undirected", [n({'attr': 10}), e_undirected(hops=1), n({'attr': 30})], None, Engine.PANDAS), - ("edge_match", [n(), e_forward(hops=1, edge_match={'w': 5}), n()], None, Engine.PANDAS), ("named_node", [n(name='x'), e_forward(hops=1), n()], None, Engine.PANDAS), ("node_query", [n(query='attr > 5'), e_forward(hops=1), n()], None, Engine.PANDAS), ("prune_endpoints", [n(), e_forward(hops=1, prune_to_endpoints=True), n()], None, Engine.PANDAS),