From dd3ae794a5200e4569e8d611c8c50baecd8216f9 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sat, 11 Jul 2026 10:28:27 -0700 Subject: [PATCH 1/2] bench(gfql): add typed Pokec Polars lanes --- CHANGELOG.md | 5 + benchmarks/gfql/graph_benchmark_pokec.py | 367 ++++++++++++++++-- .../gfql/test_graph_benchmark_pokec.py | 227 +++++++++++ 3 files changed, 568 insertions(+), 31 deletions(-) create mode 100644 graphistry/tests/benchmarks/gfql/test_graph_benchmark_pokec.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 652151f64b..443581ccc1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,11 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - **GFQL engine-selection docs (pandas / polars / cuDF / polars-gpu)**: New :doc:`Choosing a GFQL Engine ` page — a numbers-first, persona-tested guide to the four interchangeable engines. Adds the one-keyword `engine='polars'` speedup (up to ~38× over pandas on real graphs, no GPU), a motivating warm-median comparison table on real public graphs (LiveJournal 35M / Orkut 117M), a decision matrix (workload shape × size × hardware → engine, with the measured ~10K-edge CPU crossover, GPU-work-bound rule, polars-gpu memory-pressure caveat, and GPU-or-error contract), a cuDF-vs-polars-gpu disambiguation (eager-op vs fused-lazy; cuDF is not deprecated), an honest "when *not* to use Polars" section, the differential-parity guarantee, and a methodology + reproducer-script disclosure. Rewrote the top of `gfql/performance.rst` to lead with the engine comparison (de-marketed the prose), wired the new page into the GFQL toctree + recommended paths, and added Polars/polars-gpu to the engine examples in `gfql/quick.rst` and `gfql/about.rst` (previously only pandas/cuDF were documented). Driven by 4-persona doc user-testing (pandas DS, RAPIDS user, perf engineer, skeptical evaluator). ### Added +- **GFQL Pokec benchmark four-engine lanes**: add native Polars and Polars-GPU + selection with one-time frame conversion outside measured queries; `both` keeps + legacy pandas+cuDF behavior and `all` runs all four engines. Receipts retain + measured samples and distinguish unsupported, error, and fallback states; + indexed Polars-GPU traversal is labeled as CPU-index fallback, not a GPU result. - **GFQL viz-filter-pipeline acceptance suite + regression benchmark (viz-filter L3)**: `test_viz_pipeline_conformance.py` — curated full-panel pipelines (node+edge filters, exclusion-dominates composition, `(pred OR IS NULL)` keep-null leaves, both EXISTS prune-isolated flavors, searchAny composition, deterministic paging), graph-state prune shapes with exact node+edge pins, a 40-seed panel-state fuzzer with an independent plain-pandas second oracle, and a case/regex/unicode trick matrix (ß/İ full-case-mapping pins, metachar literal-vs-regex, null cells, Categorical) — all parity-or-NIE across pandas/cuDF/polars/polars-gpu. `benchmarks/gfql/viz_filter_pipeline.py` — six streamgl-viz panel scenarios (filters, keep-self GRAPH prune, EXISTS prune, node/edge search, combined) at 100K/1M/10M with native-frame-per-engine fairness, an NIE-tolerant matrix, and JSON receipts (first receipt: 100k × 4 engines, everything within the ~350ms interactive reference except pandas combined). Documented findings from the first runs: the same-path WHERE route dedupes parallel edges (diverges from the panel algebra's edge multiplicity — pinned + tracked), and edge-alias searchAny declines on polars (tracked). - **GFQL Cypher `searchAny(entity, term[, opts])` cross-column search predicate + `g.search_nodes()`/`g.search_edges()` (viz-filter L2, native on all four engines)**: True where ANY of the entity's columns matches the term — the streamgl-viz inspector's table-search semantics as a composable WHERE predicate: OR across columns; case-insensitive substring by DEFAULT (case-folded, never regex — the common call avoids every engine regex limit); regex opt-in obeying the same per-engine decline rules as `=~`; dtype gate AS SEMANTICS (string columns always; integer columns iff the term is a numeric literal, per the inspector's `/^[0-9.-]+$/` gate; floats/dates/booleans reachable via the explicit `columns:` list). Options map `{caseSensitive, regex, columns}` is strict-validated (unknown keys error listing the valid ones); unbound aliases and missing explicit columns error clearly; null cells never match. Lowered like the pattern-predicate markers (a `search_any` row op + fresh marker column), so it composes through AND/OR/NOT and different node/edge terms coexist in one pipeline. Per-column matching reuses the parity-hardened `Contains` predicate on pandas/cuDF and a lowercase-fold/any_horizontal lowering on polars; oracle-pinned + 4-engine parity-or-NIE conformance cases. Python twins `g.search_nodes(term, columns=, case_sensitive=, regex=)` / `g.search_edges(...)` filter their own table and return a Plottable (polars-frame twins decline honestly for now — use the cypher op). Honest declines (NIE, use engine='pandas'): edge-alias searchAny on polars, and explicit columns beyond string/int/bool dtypes on polars AND cuDF (incl. floats: repr diverges across engines — dgx-probed). - **GFQL Cypher `EXISTS { }` pattern-existence subqueries (openCypher-standard), native on all four engines**: `WHERE EXISTS { (n)-[:R]->() }` and `WHERE NOT EXISTS { (n)--() }` now parse and run — the declarative prune-isolated building blocks for the streamgl-viz filter pipeline. An `EXISTS` body reuses the existing pattern-predicate lowering wholesale (`semi_apply_mark` / `anti_semi_apply` row ops), so pandas/cuDF worked immediately; the polars engine gains NATIVE lowerings for the semi-apply family (correlated key sets computed by the polars chain executor's named-flag columns; order-preserving `is_in` joins) plus `rows(binding_ops=...)` for the single-entity row table — previously all honest-NIE. Aliases introduced inside the braces are existentially scoped (`EXISTS { (n)--(m) }` allowed with `m` unbound outside — bare pattern predicates keep the conservative guard), inline property maps work, and the one supported inner `WHERE` form is endpoint inequality — `EXISTS { (n)--(m) WHERE m <> n }`, the drop-self-loop prune-isolated flavor (pandas/cuDF filter the correlated bindings; polars excludes self-loop edges, which is exactly the `m <> n` witness). Both prune flavors are oracle-pinned in the conformance matrix on a self-loop discriminator graph, 4-engine parity-or-NIE. Honest declines with clear errors: `EXISTS` in RETURN/WITH projections, general inner `WHERE`, multi-pattern bodies, full `MATCH..RETURN` subquery bodies, multi-alias correlation on polars. diff --git a/benchmarks/gfql/graph_benchmark_pokec.py b/benchmarks/gfql/graph_benchmark_pokec.py index 8c26c6595d..6b3537d88d 100644 --- a/benchmarks/gfql/graph_benchmark_pokec.py +++ b/benchmarks/gfql/graph_benchmark_pokec.py @@ -6,7 +6,7 @@ parses the public mgBench Pokec import (``CREATE (:User {...})`` nodes and ``MATCH ... CREATE (n)-[:Friend]->(m)`` edges) into node/edge dataframes, then runs the exact mgBench read/aggregate/expansion/neighbours/pattern queries as -the *same Cypher* on GFQL (pandas/cuDF) and, optionally, Memgraph over Bolt, +the *same Cypher* on GFQL (pandas/cuDF/Polars/Polars-GPU) and, optionally, Memgraph over Bolt, using a shared fixed seed set for the seeded (``$id``) queries. Dataset: https://s3.eu-west-1.amazonaws.com/deps.memgraph.io/dataset/pokec/benchmark/ @@ -21,7 +21,20 @@ import sys import time from pathlib import Path -from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Dict, + List, + Literal, + Optional, + Sequence, + Set, + Tuple, + TypedDict, + TypeVar, +) _HERE = Path(__file__).resolve() sys.path.insert(0, str(_HERE.parent)) # benchmarks/gfql for sibling imports @@ -29,13 +42,72 @@ import pandas as pd -from graph_benchmark_q1_q9 import _maybe_to_cudf, _median +from graph_benchmark_q1_q9 import _median + +if TYPE_CHECKING: + from graphistry.Engine import Engine + from graphistry.compute.ComputeMixin import ComputeMixin + from graphistry.compute.gfql.index.types import IndexPath + from graphistry.compute.typing import DataFrameT + from graphistry.Plottable import Plottable _NODE_RE = re.compile( r'CREATE \(:User \{id: (\d+), completion_percentage: (\d+), gender: "([^"]*)", age: (\d+)\}\)' ) _EDGE_RE = re.compile(r'MATCH \(n:User \{id: (\d+)\}\), \(m:User \{id: (\d+)\}\)') +GFQLEngine = Literal["pandas", "cudf", "polars", "polars-gpu"] +QueryKind = Literal["global", "seed"] +ExecutionTarget = Literal["cpu", "gpu", "mixed"] +FailureStatus = Literal["unsupported", "error", "oom"] +ResultT = TypeVar("ResultT") + + +class QuerySpec(TypedDict): + name: str + kind: QueryKind + gfql: str + memgraph: str + + +class TimingResult(TypedDict): + median_ms: float + samples_ms: List[float] + + +class SeedTiming(TypedDict): + seed: int + median_ms: float + samples_ms: List[float] + rowcount: int + execution_paths: List["IndexPath"] + + +class ExecutionMetadata(TypedDict, total=False): + status: Literal["ok", "fallback"] + execution_target: ExecutionTarget + synchronization: Literal["none", "cuda-device"] + fallback_reason: str + + +class FailureResult(TypedDict): + status: FailureStatus + unsupported: bool + error_type: str + error: str + kind: QueryKind + + +GFQL_ENGINES: Tuple[GFQLEngine, ...] = ("pandas", "cudf", "polars", "polars-gpu") +ENGINE_SELECTIONS: Dict[str, Tuple[GFQLEngine, ...]] = { + "pandas": ("pandas",), + "cudf": ("cudf",), + "polars": ("polars",), + "polars-gpu": ("polars-gpu",), + "both": ("pandas", "cudf"), + "all": GFQL_ENGINES, +} + def parse_pokec_import(path: Path) -> Tuple[pd.DataFrame, pd.DataFrame]: ids: List[int] = [] @@ -69,7 +141,7 @@ def parse_pokec_import(path: Path) -> Tuple[pd.DataFrame, pd.DataFrame]: # "pair" runs over shared (from,to) pairs. # GFQL Cypher uses the label__User convention and untyped edges; Memgraph uses # the exact mgBench query text. -QUERIES: Tuple[Dict[str, Any], ...] = ( +QUERIES: Tuple[QuerySpec, ...] = ( {"name": "aggregate", "kind": "global", "gfql": "MATCH (n:User) RETURN n.age, COUNT(*)", "memgraph": "MATCH (n:User) RETURN n.age, COUNT(*)"}, @@ -124,19 +196,44 @@ def _rowcount(result: Any) -> int: return -1 -def _time(fn: Callable[[], Any], runs: int, warmup: int) -> Tuple[Any, float]: +def _time( + fn: Callable[[], ResultT], + runs: int, + warmup: int, + synchronize: Optional[Callable[[], None]] = None, +) -> Tuple[ResultT, TimingResult]: + if runs < 1: + raise ValueError(f"runs must be >= 1, got {runs}") + if warmup < 0: + raise ValueError(f"warmup must be >= 0, got {warmup}") for _ in range(warmup): fn() - times: List[float] = [] - result: Any = None - for _ in range(runs): + if synchronize is not None: + synchronize() + + if synchronize is not None: + synchronize() + start = time.perf_counter() + result = fn() + if synchronize is not None: + synchronize() + times = [(time.perf_counter() - start) * 1000.0] + for _ in range(1, runs): + if synchronize is not None: + synchronize() start = time.perf_counter() result = fn() + if synchronize is not None: + synchronize() times.append((time.perf_counter() - start) * 1000.0) - return result, _median(times) + return result, {"median_ms": _median(times), "samples_ms": times} -def _hop_dsts(g: Any, frontier_ids: Sequence[int], engine: str) -> List[int]: +def _hop_dsts( + g: "ComputeMixin", + frontier_ids: Sequence[int], + engine: GFQLEngine, +) -> List[int]: """One indexed forward 1-hop from a frontier; return distinct destination ids. hop(nodes=, hops=1) is routed through the #1658 CSR index on every engine @@ -154,8 +251,13 @@ def _hop_dsts(g: Any, frontier_ids: Sequence[int], engine: str) -> List[int]: NATIVE_SEEDED_NAMES = ("expansion_", "neighbours_", "pattern_short") -def _native_seeded(g: Any, spec: Dict[str, Any], seed: int, engine: str, - gfql_kwargs: Dict[str, Any], adult_ids: Optional[set] = None) -> Any: +def _native_seeded( + g: "ComputeMixin", + spec: QuerySpec, + seed: int, + engine: GFQLEngine, + adult_ids: Optional[Set[int]] = None, +) -> List[int]: """Express the seeded Pokec traversals as a loop of indexed 1-hops. expansion_k = final frontier of exactly k hops (distinct); neighbours_k = @@ -187,12 +289,137 @@ def _native_seeded(g: Any, spec: Dict[str, Any], seed: int, engine: str, return list(endpoints) -def run_gfql(nodes: pd.DataFrame, edges: pd.DataFrame, engine: str, seeds: Sequence[int], +def _frame_engine(engine: GFQLEngine) -> "Engine": + from graphistry.Engine import Engine + + if engine == "pandas": + return Engine.PANDAS + if engine == "cudf": + return Engine.CUDF + if engine in ("polars", "polars-gpu"): + # Polars-GPU is an execution target; its resident frames are ordinary Polars. + return Engine.POLARS + raise ValueError(f"unsupported GFQL benchmark engine: {engine!r}") + + +def _to_native_frames( + nodes: pd.DataFrame, edges: pd.DataFrame, engine: GFQLEngine +) -> Tuple["DataFrameT", "DataFrameT"]: + from graphistry.Engine import Engine, df_to_engine + + frame_engine = _frame_engine(engine) + if frame_engine == Engine.PANDAS: + return nodes, edges + return df_to_engine(nodes, frame_engine), df_to_engine(edges, frame_engine) + + +def _selected_engines(selection: str) -> Tuple[GFQLEngine, ...]: + try: + return ENGINE_SELECTIONS[selection] + except KeyError as exc: + raise ValueError(f"unsupported GFQL benchmark engine selection: {selection!r}") from exc + + +def _require_compute_graph(g: "Plottable") -> "ComputeMixin": + from graphistry.compute.ComputeMixin import ComputeMixin + + if not isinstance(g, ComputeMixin): + raise TypeError(f"GFQL benchmark requires ComputeMixin, got {type(g)!r}") + return g + + +def _trace_native_seeded( + g: "ComputeMixin", + spec: QuerySpec, + seed: int, + engine: GFQLEngine, + adult_ids: Optional[Set[int]], +) -> Tuple[List[int], List["IndexPath"]]: + from graphistry.compute.gfql.index import index_trace + + with index_trace() as steps: + result = _native_seeded(g, spec, seed, engine, adult_ids) + paths = [step["path"] for step in steps if "path" in step] + if not paths: + raise RuntimeError( + f"Native indexed traversal produced no planner trace for {spec['name']}" + ) + return result, paths + + +def _execution_metadata( + engine: GFQLEngine, + *, + uses_native_index: bool, + execution_paths: Sequence["IndexPath"] = (), +) -> ExecutionMetadata: + if engine == "polars-gpu" and uses_native_index: + paths = set(execution_paths) + if not paths: + raise ValueError("Polars-GPU native traversal requires execution paths") + if paths == {"scan"}: + return { + "status": "ok", + "execution_target": "gpu", + "synchronization": "cuda-device", + } + if paths == {"index"}: + target: ExecutionTarget = "cpu" + detail = "all measured seeds used the NumPy host adjacency index" + else: + target = "mixed" + detail = "planner decisions mixed NumPy host indexes with GPU scans" + return { + "status": "fallback", + "execution_target": target, + "synchronization": "cuda-device", + "fallback_reason": ( + f"{detail}; this row is not a GPU-only result" + ), + } + return { + "status": "ok", + "execution_target": "gpu" if engine in ("cudf", "polars-gpu") else "cpu", + "synchronization": ( + "cuda-device" if engine in ("cudf", "polars-gpu") else "none" + ), + } + + +def _failure_result(exc: Exception, kind: QueryKind) -> FailureResult: + unsupported = isinstance(exc, NotImplementedError) + type_name = type(exc).__name__ + normalized_type = type_name.lower().replace("_", "") + oom = isinstance(exc, MemoryError) or "outofmemory" in normalized_type + status: FailureStatus = ( + "unsupported" if unsupported else "oom" if oom else "error" + ) + return { + "status": status, + "unsupported": unsupported, + "error_type": type_name, + "error": str(exc)[:500], + "kind": kind, + } + + +def _synchronizer(engine: GFQLEngine) -> Optional[Callable[[], None]]: + if engine not in ("cudf", "polars-gpu"): + return None + import cupy as cp + + def synchronize() -> None: + cp.cuda.runtime.deviceSynchronize() + + return synchronize + + +def run_gfql(nodes: pd.DataFrame, edges: pd.DataFrame, engine: GFQLEngine, seeds: Sequence[int], runs: int, warmup: int, index_policy: str = "off", traversal: str = "cypher") -> Dict[str, Any]: import graphistry - n = _maybe_to_cudf(engine, nodes) if engine == "cudf" else nodes - e = _maybe_to_cudf(engine, edges) if engine == "cudf" else edges - g = graphistry.nodes(n, "id").edges(e, "src", "dst") + synchronize = _synchronizer(engine) + n, e = _to_native_frames(nodes, edges, engine) + g = _require_compute_graph(graphistry.nodes(n, "id").edges(e, "src", "dst")) # #1658 seeded adjacency index (opt-in). Build once as untimed setup (like a # DB's load-time index), then seeded gfql() runs with index_policy set. index_build_ms: Optional[float] = None @@ -207,7 +434,7 @@ def run_gfql(nodes: pd.DataFrame, edges: pd.DataFrame, engine: str, seeds: Seque except Exception: pass # Precompute the age>=18 endpoint set once (untimed setup) for *_with_filter. - adult_ids: Optional[set] = None + adult_ids: Optional[Set[int]] = None if traversal == "native": _nn = nodes # host pandas frame (before engine coercion) — fine for a setup lookup adult_ids = set(_nn[_nn["age"] >= 18]["id"]) @@ -217,31 +444,85 @@ def run_gfql(nodes: pd.DataFrame, edges: pd.DataFrame, engine: str, seeds: Seque try: if spec["kind"] == "global": q = spec["gfql"] - _, med = _time(lambda: g.gfql(q, **gfql_kwargs), runs, warmup) - results[name] = {"median_ms": med, "kind": "global"} + result, timing = _time( + lambda: g.gfql(q, **gfql_kwargs), + runs, + warmup, + synchronize, + ) + results[name] = { + **timing, + **_execution_metadata(engine, uses_native_index=False), + "kind": "global", + "rowcount": _rowcount(result), + "unit": "ms", + } else: use_native = traversal == "native" and name.startswith(NATIVE_SEEDED_NAMES) per_seed: List[float] = [] rowcounts: List[int] = [] + seed_timings: List[SeedTiming] = [] + execution_paths: List["IndexPath"] = [] for sid in seeds: if use_native: - res, med = _time(lambda: _native_seeded(g, spec, sid, engine, gfql_kwargs, adult_ids), runs, warmup) + res, timing = _time( + lambda: _native_seeded(g, spec, sid, engine, adult_ids), + runs, + warmup, + synchronize, + ) rc = int(len(res)) + traced_res, seed_paths = _trace_native_seeded( + g, + spec, + sid, + engine, + adult_ids, + ) + if set(traced_res) != set(res): + raise RuntimeError( + f"Trace probe changed result for {name} seed {sid}" + ) + execution_paths.extend(seed_paths) else: q = spec["gfql"].format(id=sid) - res, med = _time(lambda: g.gfql(q, **gfql_kwargs), runs, warmup) + res, timing = _time( + lambda: g.gfql(q, **gfql_kwargs), + runs, + warmup, + synchronize, + ) rc = _rowcount(res) - per_seed.append(med) + per_seed.append(timing["median_ms"]) rowcounts.append(rc) + seed_timings.append( + { + "seed": int(sid), + "median_ms": timing["median_ms"], + "samples_ms": timing["samples_ms"], + "rowcount": rc, + "execution_paths": seed_paths if use_native else [], + } + ) results[name] = { "median_ms": _median(per_seed), + "min_seed_median_ms": min(per_seed), + "max_seed_median_ms": max(per_seed), + "seed_timings": seed_timings, + "execution_paths": execution_paths, + **_execution_metadata( + engine, + uses_native_index=use_native, + execution_paths=execution_paths, + ), "kind": "seed", + "unit": "ms", "seeds": len(seeds), "traversal": "native" if use_native else "cypher", "median_rowcount": int(_median([float(r) for r in rowcounts])), } except Exception as exc: - results[name] = {"unsupported": True, "error": f"{type(exc).__name__}: {str(exc)[:160]}", "kind": spec["kind"]} + results[name] = _failure_result(exc, spec["kind"]) return results @@ -253,23 +534,47 @@ def run_memgraph(driver: Any, seeds: Sequence[int], runs: int, warmup: int) -> D q = spec["memgraph"] try: if spec["kind"] == "global": - _, med = _time(lambda: execute(driver, q), runs, warmup) - results[name] = {"median_ms": med, "kind": "global"} + result, timing = _time(lambda: execute(driver, q), runs, warmup) + results[name] = { + **timing, + "kind": "global", + "rowcount": len(result), + "unit": "ms", + } else: per_seed: List[float] = [] rowcounts: List[int] = [] + seed_timings: List[SeedTiming] = [] for sid in seeds: - res, med = _time(lambda: execute(driver, q, id=int(sid)), runs, warmup) - per_seed.append(med) - rowcounts.append(len(res)) + res, timing = _time( + lambda: execute(driver, q, id=int(sid)), + runs, + warmup, + ) + rowcount = len(res) + per_seed.append(timing["median_ms"]) + rowcounts.append(rowcount) + seed_timings.append( + { + "seed": int(sid), + "median_ms": timing["median_ms"], + "samples_ms": timing["samples_ms"], + "rowcount": rowcount, + "execution_paths": [], + } + ) results[name] = { "median_ms": _median(per_seed), + "min_seed_median_ms": min(per_seed), + "max_seed_median_ms": max(per_seed), + "seed_timings": seed_timings, "kind": "seed", + "unit": "ms", "seeds": len(seeds), "median_rowcount": int(_median([float(r) for r in rowcounts])), } except Exception as exc: - results[name] = {"unsupported": True, "error": f"{type(exc).__name__}: {str(exc)[:160]}", "kind": spec["kind"]} + results[name] = _failure_result(exc, spec["kind"]) return results @@ -299,7 +604,7 @@ def load_memgraph(driver: Any, nodes: pd.DataFrame, edges: pd.DataFrame, batch_s def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--import-cypher", type=Path, required=True, help="mgBench Pokec *_import.cypher file") - parser.add_argument("--engine", choices=["pandas", "cudf", "both"], default="both") + parser.add_argument("--engine", choices=[*GFQL_ENGINES, "both", "all"], default="both") parser.add_argument("--seeds", type=int, default=30, help="Number of shared seed vertices for seeded queries") parser.add_argument("--seed-rng", type=int, default=42) parser.add_argument("--runs", type=int, default=5) @@ -333,7 +638,7 @@ def main() -> None: "gfql": {}, } - engines = ["pandas", "cudf"] if args.engine == "both" else [args.engine] + engines = _selected_engines(args.engine) for engine in engines: output["gfql"][engine] = run_gfql(nodes, edges, engine, seeds, args.runs, args.warmup, args.gfql_index, args.gfql_traversal) diff --git a/graphistry/tests/benchmarks/gfql/test_graph_benchmark_pokec.py b/graphistry/tests/benchmarks/gfql/test_graph_benchmark_pokec.py new file mode 100644 index 0000000000..fb73d83aa1 --- /dev/null +++ b/graphistry/tests/benchmarks/gfql/test_graph_benchmark_pokec.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path +import sys +from typing import ( + TYPE_CHECKING, + Callable, + List, + Literal, + Optional, + Protocol, + runtime_checkable, + Tuple, + TypedDict, + TypeVar, +) +import unittest + +import pandas as pd + +if TYPE_CHECKING: + from graphistry.Engine import Engine + + +IndexPath = Literal["scan", "index"] +QueryKind = Literal["global", "seed"] +ExecutionTarget = Literal["cpu", "gpu", "mixed"] +FailureStatus = Literal["unsupported", "error", "oom"] +ResultT = TypeVar("ResultT") + + +class TimingReceipt(TypedDict): + median_ms: float + samples_ms: List[float] + + +class ExecutionReceipt(TypedDict, total=False): + status: Literal["ok", "fallback"] + execution_target: ExecutionTarget + synchronization: Literal["none", "cuda-device"] + fallback_reason: str + + +class FailureReceipt(TypedDict): + status: FailureStatus + unsupported: bool + error_type: str + error: str + kind: QueryKind + + +@runtime_checkable +class PokecBenchmarkModule(Protocol): + def _selected_engines(self, selection: str) -> Tuple[str, ...]: + ... + + def _frame_engine(self, engine: str) -> "Engine": + ... + + def _to_native_frames( + self, + nodes: pd.DataFrame, + edges: pd.DataFrame, + engine: str, + ) -> Tuple[pd.DataFrame, pd.DataFrame]: + ... + + def _execution_metadata( + self, + engine: str, + *, + uses_native_index: bool, + execution_paths: Tuple[IndexPath, ...] = (), + ) -> ExecutionReceipt: + ... + + def _time( + self, + fn: Callable[[], ResultT], + runs: int, + warmup: int, + synchronize: Optional[Callable[[], None]] = None, + ) -> Tuple[ResultT, TimingReceipt]: + ... + + def _failure_result( + self, + exc: Exception, + kind: QueryKind, + ) -> FailureReceipt: + ... + + +def _load_pokec_module() -> PokecBenchmarkModule: + repo_root = Path(__file__).resolve().parents[4] + module_path = repo_root / "benchmarks" / "gfql" / "graph_benchmark_pokec.py" + spec = importlib.util.spec_from_file_location("graph_benchmark_pokec", module_path) + if spec is None or spec.loader is None: + raise RuntimeError(f"Failed loading module at {module_path}") + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + if not isinstance(module, PokecBenchmarkModule): + raise RuntimeError(f"Benchmark module at {module_path} lacks its helper API") + return module + + +_POKEC = _load_pokec_module() + + +class TestPokecBenchmarkHelpers(unittest.TestCase): + def test_selected_engines(self) -> None: + cases = [ + ("pandas", ("pandas",)), + ("cudf", ("cudf",)), + ("polars", ("polars",)), + ("polars-gpu", ("polars-gpu",)), + ("both", ("pandas", "cudf")), + ("all", ("pandas", "cudf", "polars", "polars-gpu")), + ] + for selection, expected in cases: + with self.subTest(selection=selection): + self.assertEqual(_POKEC._selected_engines(selection), expected) + + def test_selected_engines_rejects_unknown(self) -> None: + with self.assertRaisesRegex( + ValueError, + "unsupported GFQL benchmark engine selection", + ): + _POKEC._selected_engines("gpu") + + def test_polars_gpu_uses_polars_resident_frames(self) -> None: + from graphistry.Engine import Engine + + self.assertEqual(_POKEC._frame_engine("polars"), Engine.POLARS) + self.assertEqual(_POKEC._frame_engine("polars-gpu"), Engine.POLARS) + + def test_pandas_frame_conversion_is_identity(self) -> None: + nodes = pd.DataFrame({"id": [1, 2]}) + edges = pd.DataFrame({"src": [1], "dst": [2]}) + + converted_nodes, converted_edges = _POKEC._to_native_frames( + nodes, + edges, + "pandas", + ) + + self.assertIs(converted_nodes, nodes) + self.assertIs(converted_edges, edges) + + def test_polars_gpu_index_execution_is_explicit_cpu_fallback(self) -> None: + metadata = _POKEC._execution_metadata( + "polars-gpu", + uses_native_index=True, + execution_paths=("index",), + ) + + self.assertEqual(metadata["status"], "fallback") + self.assertEqual(metadata["execution_target"], "cpu") + self.assertIn("not a GPU-only result", str(metadata["fallback_reason"])) + + def test_polars_gpu_cypher_execution_stays_gpu(self) -> None: + metadata = _POKEC._execution_metadata( + "polars-gpu", + uses_native_index=False, + ) + + self.assertEqual(metadata["status"], "ok") + self.assertEqual(metadata["execution_target"], "gpu") + self.assertNotIn("fallback_reason", metadata) + + def test_polars_gpu_scan_path_is_a_gpu_result(self) -> None: + metadata = _POKEC._execution_metadata( + "polars-gpu", + uses_native_index=True, + execution_paths=("scan",), + ) + + self.assertEqual(metadata["status"], "ok") + self.assertEqual(metadata["execution_target"], "gpu") + + def test_polars_gpu_mixed_path_is_explicit_fallback(self) -> None: + metadata = _POKEC._execution_metadata( + "polars-gpu", + uses_native_index=True, + execution_paths=("index", "scan"), + ) + + self.assertEqual(metadata["status"], "fallback") + self.assertEqual(metadata["execution_target"], "mixed") + self.assertIn("mixed", str(metadata["fallback_reason"])) + + def test_time_retains_all_measured_samples(self) -> None: + calls = 0 + + def measured() -> int: + nonlocal calls + calls += 1 + return calls + + result, timing = _POKEC._time(measured, runs=3, warmup=2) + + self.assertEqual(result, 5) + self.assertEqual(calls, 5) + self.assertEqual(len(timing["samples_ms"]), 3) + self.assertGreaterEqual(timing["median_ms"], 0.0) + + def test_time_synchronizes_gpu_boundaries(self) -> None: + sync_calls = 0 + + def synchronize() -> None: + nonlocal sync_calls + sync_calls += 1 + + _POKEC._time(lambda: 1, runs=2, warmup=1, synchronize=synchronize) + + self.assertEqual(sync_calls, 5) + + def test_failure_statuses_are_distinct(self) -> None: + unsupported = _POKEC._failure_result(NotImplementedError("no"), "seed") + error = _POKEC._failure_result(ValueError("bad"), "seed") + oom = _POKEC._failure_result(MemoryError("full"), "seed") + + self.assertEqual(unsupported["status"], "unsupported") + self.assertEqual(error["status"], "error") + self.assertEqual(oom["status"], "oom") From 52a22dd93888208eed0dcd22d2cb693ee09f3a55 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sat, 11 Jul 2026 10:32:51 -0700 Subject: [PATCH 2/2] fix(gfql): classify unsupported benchmark rows --- benchmarks/gfql/graph_benchmark_pokec.py | 6 +++++- .../tests/benchmarks/gfql/test_graph_benchmark_pokec.py | 8 ++++++++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/benchmarks/gfql/graph_benchmark_pokec.py b/benchmarks/gfql/graph_benchmark_pokec.py index 6b3537d88d..1118054058 100644 --- a/benchmarks/gfql/graph_benchmark_pokec.py +++ b/benchmarks/gfql/graph_benchmark_pokec.py @@ -387,7 +387,11 @@ def _execution_metadata( def _failure_result(exc: Exception, kind: QueryKind) -> FailureResult: - unsupported = isinstance(exc, NotImplementedError) + from graphistry.compute.exceptions import ErrorCode, GFQLValidationError + + unsupported = isinstance(exc, NotImplementedError) or ( + isinstance(exc, GFQLValidationError) and exc.code == ErrorCode.E108 + ) type_name = type(exc).__name__ normalized_type = type_name.lower().replace("_", "") oom = isinstance(exc, MemoryError) or "outofmemory" in normalized_type diff --git a/graphistry/tests/benchmarks/gfql/test_graph_benchmark_pokec.py b/graphistry/tests/benchmarks/gfql/test_graph_benchmark_pokec.py index fb73d83aa1..d01a53a284 100644 --- a/graphistry/tests/benchmarks/gfql/test_graph_benchmark_pokec.py +++ b/graphistry/tests/benchmarks/gfql/test_graph_benchmark_pokec.py @@ -218,10 +218,18 @@ def synchronize() -> None: self.assertEqual(sync_calls, 5) def test_failure_statuses_are_distinct(self) -> None: + from graphistry.compute.exceptions import ErrorCode, GFQLValidationError + unsupported = _POKEC._failure_result(NotImplementedError("no"), "seed") + structured_unsupported = _POKEC._failure_result( + GFQLValidationError(ErrorCode.E108, "not implemented"), + "seed", + ) error = _POKEC._failure_result(ValueError("bad"), "seed") oom = _POKEC._failure_result(MemoryError("full"), "seed") self.assertEqual(unsupported["status"], "unsupported") + self.assertEqual(structured_unsupported["status"], "unsupported") + self.assertTrue(structured_unsupported["unsupported"]) self.assertEqual(error["status"], "error") self.assertEqual(oom["status"], "oom")