test(gfql): shape registry fed by the specialization tables + route harness

Each specialization's test module registers the shape table it already owns
(routes corpus, the six test_chain tables, the alias-collision matrix) with
its frames and defect-class tags. The harness tries every registered shape
against every chain route whose admission predicate admits it and pins that
the lane serves, that the answer matches the same engine's general path on
values, and that node/edge sets match the pandas general path. A lane that
declines an admitted shape is recorded as an expected failure (the
attenuation ledger); filed divergences are strict expected failures keyed by
tag. The pandas bypass table's prune shapes surface #2053 on the polars route.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QztW7jYsDd66e8rb8pJNQA

Author: lmeyerov

graphistry/tests/compute/gfql/routes/corpus.py

@@ -11,6 +11,7 @@
 
 from graphistry.compute.ast import ASTObject, e_forward, e_reverse, e_undirected, n
 from graphistry.compute.predicates.numeric import GT
+from graphistry.tests.compute.gfql.routes.registry import Frames, register
 
 
 class Entry(NamedTuple):
@@ -54,3 +55,6 @@ def tagged(tag: str) -> List[Entry]:
 
 def by_name() -> Dict[str, Entry]:
     return {e.name: e for e in CORPUS}
+
+
+register("routes.corpus", [(e.name, e.ops, e.tags) for e in CORPUS], Frames(NODES, EDGES, "key", "s", "d", "eid"))

graphistry/tests/compute/gfql/routes/registry.py

@@ -0,0 +1,72 @@
+"""Shape registry for the route harness.
+
+Each specialization's own test module registers the shape table it was written against
+(``register(...)`` returns the table unchanged, so the module keeps using it), with the
+frames those shapes address and the defect classes they exercise. The harness in
+``test_route_harness.py`` then tries every registered shape against every route whose
+admission predicate admits it, so one input is exercised by several hot paths, not one.
+"""
+from typing import Callable, Dict, Iterable, List, NamedTuple, Optional, Sequence, Tuple, Union
+
+import pandas as pd
+import pytest
+
+from graphistry.Plottable import Plottable
+from graphistry.compute.ast import ASTObject
+
+Build = Callable[[], List[ASTObject]]
+Row = Union[Tuple[str, Build], Tuple[str, Build, Tuple[str, ...]]]
+
+
+class Frames(NamedTuple):
+    nodes: pd.DataFrame
+    edges: pd.DataFrame
+    node: str
+    src: str
+    dst: str
+    edge: Optional[str] = None
+
+
+class Shape(NamedTuple):
+    table: str
+    label: str
+    build: Build
+    frames: Frames
+    tags: Tuple[str, ...]
+
+    @property
+    def name(self) -> str:
+        return f"{self.table}/{self.label}"
+
+
+REGISTRY: Dict[str, Shape] = {}
+
+
+def register(table: str, rows: Sequence[Row], frames: Frames, tags: Iterable[str] = (),
+             row_tags: Optional[Dict[str, Tuple[str, ...]]] = None) -> Sequence[Row]:
+    """Register ``rows`` ((label, build[, tags]) ...) under ``table``; returns ``rows``."""
+    base = tuple(tags)
+    for row in rows:
+        label, build = row[0], row[1]
+        extra = tuple(row[2]) if len(row) > 2 else ()
+        extra += (row_tags or {}).get(label, ())
+        shape = Shape(table, label, build, frames, base + extra)
+        REGISTRY.setdefault(shape.name, shape)
+    return rows
+
+
+def to_engine(df: pd.DataFrame, engine: str):
+    if engine == "pandas":
+        return df
+    if engine == "cudf":
+        return pytest.importorskip("cudf").from_pandas(df)
+    if engine == "polars":
+        return pytest.importorskip("polars").from_pandas(df)
+    raise ValueError(engine)
+
+
+def graph_for(shape: Shape, engine: str, indexed: bool = False) -> Plottable:
+    import graphistry
+    f = shape.frames
+    g = graphistry.nodes(to_engine(f.nodes, engine), f.node).edges(to_engine(f.edges, engine), f.src, f.dst, f.edge)
+    return g.gfql_index_all(engine=engine) if indexed else g

graphistry/tests/compute/gfql/routes/switch.py

@@ -0,0 +1,51 @@
+"""Route switch for test amplification: make named GFQL hot paths decline for a scope."""
+from contextlib import contextmanager
+from typing import Iterable, Iterator, List, Tuple
+
+ROUTES = ("native-fast", "polars-seeded", "polars-plain", "index-hop", "indexed-kernel", "cypher-fast")
+
+
+def _none(*a, **k):
+    return None
+
+
+def _targets(routes: Iterable[str]) -> List[Tuple[object, str]]:
+    import graphistry.compute.chain as chain_mod
+    import graphistry.compute.gfql_unified as unified
+    import graphistry.compute.gfql.index as index_pkg
+    import graphistry.compute.gfql.index.api as index_api
+    import graphistry.compute.gfql.index.bindings as bindings
+    import graphistry.compute.gfql.lazy.engine.polars.chain as pchain
+    routes = set(routes)
+    unknown = routes - set(ROUTES)
+    assert not unknown, f"unknown route(s) {sorted(unknown)}; known: {ROUTES}"
+    out: List[Tuple[object, str]] = []
+    if "native-fast" in routes:
+        out.append((chain_mod, "_try_chain_fast_path"))
+    if "polars-seeded" in routes:
+        out.append((pchain, "_try_seeded_chain_polars"))
+    if "polars-plain" in routes:
+        out.append((pchain, "polars_plain_single_hop_admits"))
+    if "index-hop" in routes:
+        out += [(index_pkg, "maybe_index_hop"), (index_api, "maybe_index_hop")]
+    if "indexed-kernel" in routes:
+        out.append((bindings, "_try_indexed_connected_bindings_state"))
+    if "cypher-fast" in routes:
+        out += [(unified, name) for name in (
+            "_execute_seeded_node_lookup_fast_path", "_execute_seeded_typed_hop_fast_path",
+            "_execute_single_hop_grouped_aggregate_fast_path", "_execute_two_hop_count_fast_path")]
+    return out
+
+
+@contextmanager
+def routes_off(routes: Iterable[str]) -> Iterator[None]:
+    """Within the block the named routes decline, so the general path answers."""
+    saved = []
+    for mod, name in _targets(routes):
+        saved.append((mod, name, getattr(mod, name)))
+        setattr(mod, name, _none)
+    try:
+        yield
+    finally:
+        for mod, name, value in reversed(saved):
+            setattr(mod, name, value)

graphistry/tests/compute/gfql/routes/test_route_harness.py

@@ -0,0 +1,180 @@
+"""Route harness: every registered shape is tried against every chain route whose admission
+predicate admits it. Three pins per cell: the route SERVES (its lane answers; a lane that
+declines an admitted shape is recorded as an expected failure, the attenuation ledger), the
+answer matches the same engine's general path (all routes off) on node/edge values, and its
+node/edge sets match the pandas general path (the cross-engine oracle). Filed divergences are
+strict expected failures keyed by their tag, so they flip when fixed.
+"""
+import math
+import os
+from typing import Callable, Dict, List, NamedTuple, Tuple
+
+import pandas as pd
+import pytest
+
+import graphistry.compute.chain as chain_mod
+import graphistry.compute.gfql.lazy.engine.polars.chain as pchain
+from graphistry.Engine import Engine
+from graphistry.compute.ast import ASTObject
+from graphistry.compute.chain_specializations.admission import native_fast_path_admits
+from graphistry.compute.gfql.lazy.engine.polars.chain_specializations.admission import (
+    polars_plain_single_hop_admits, polars_seeded_lane_admits,
+)
+from graphistry.tests.compute.gfql.routes.registry import REGISTRY, Shape, graph_for
+from graphistry.tests.compute.gfql.routes.switch import ROUTES as ALL_ROUTES, routes_off
+
+import graphistry.tests.compute.gfql.routes.corpus  # noqa: F401  registers routes.corpus
+import graphistry.tests.compute.test_chain  # noqa: F401  registers test_chain.*
+import graphistry.tests.compute.test_chain_alias_column_collision  # noqa: F401  registers collision.*
+
+
+class Route(NamedTuple):
+    name: str
+    engines: Tuple[str, ...]
+    admits: Callable[[List[ASTObject], str], bool]
+    lane: Tuple[object, str]
+    indexed: bool
+
+
+ROUTES = [
+    Route("native-fast", ("pandas", "cudf"),
+          lambda ops, engine: native_fast_path_admits(ops, Engine(engine), None) is not None,
+          (chain_mod, "_try_chain_fast_path"), False),
+    Route("polars-plain", ("polars",),
+          lambda ops, engine: polars_plain_single_hop_admits(ops, None) is not None,
+          (pchain, "_plain_single_hop_polars"), False),
+    Route("polars-seeded", ("polars",),
+          lambda ops, engine: polars_seeded_lane_admits(ops),
+          (pchain, "_try_seeded_chain_polars"), True),
+]
+
+KNOWN: Dict[Tuple[str, str], str] = {  # (route, tag) -> issue: strict xfail until it lands
+    ("polars-plain", "#2053"): "graphistry/pygraphistry#2053",
+}
+
+
+class Case(NamedTuple):
+    route: Route
+    engine: str
+    shape: Shape
+
+    @property
+    def id(self) -> str:
+        return f"{self.route.name}/{self.engine}/{self.shape.name}"
+
+
+def _cases() -> List[Case]:
+    out = []
+    for shape in REGISTRY.values():
+        for route in ROUTES:
+            for engine in route.engines:
+                try:
+                    admitted = route.admits(shape.build(), engine)
+                except Exception:
+                    admitted = False
+                if admitted:
+                    out.append(Case(route, engine, shape))
+    return out
+
+
+CASES = _cases()
+
+
+def _topd(df):
+    if df is None:
+        return None
+    if hasattr(df, "to_pandas"):
+        return df.to_pandas()
+    return df
+
+
+def _canon(df) -> Tuple[Tuple[str, ...], List[Tuple]]:
+    df = _topd(df)
+    if df is None:
+        return ((), [])
+    cols = tuple(sorted(df.columns))
+    rows = []
+    for row in df[list(cols)].itertuples(index=False, name=None):
+        rows.append(tuple(None if (isinstance(v, float) and math.isnan(v)) or v is pd.NA or v is pd.NaT else v for v in row))
+    rows.sort(key=repr)
+    return cols, rows
+
+
+def _sig(res, frames) -> Tuple[List, List]:
+    nn, ee = _topd(res._nodes), _topd(res._edges)
+    nodes = sorted(nn[frames.node].tolist()) if nn is not None else []
+    edges = sorted(map(tuple, ee[[frames.src, frames.dst]].values.tolist())) if ee is not None and len(ee) else []
+    return nodes, edges
+
+
+def _served(case: Case, monkeypatch):
+    mod, name = case.route.lane
+    real = getattr(mod, name)
+    calls = {"served": 0}
+
+    def spy(*a, **k):
+        out = real(*a, **k)
+        calls["served"] += out is not None
+        return out
+    monkeypatch.setattr(mod, name, spy)
+    return calls
+
+
+def _skip_unavailable(engine: str) -> None:
+    if engine == "cudf":
+        if os.environ.get("TEST_CUDF") != "1":
+            pytest.skip("cuDF lane runs with TEST_CUDF=1")
+        pytest.importorskip("cudf")
+    if engine == "polars":
+        pytest.importorskip("polars")
+
+
+@pytest.mark.parametrize("case", CASES, ids=[c.id for c in CASES])
+def test_admitted_shape_is_served_and_matches_the_general_path(case: Case, request, monkeypatch):
+    _skip_unavailable(case.engine)
+    for tag in case.shape.tags:
+        if (case.route.name, tag) in KNOWN:
+            request.applymarker(pytest.mark.xfail(strict=True, reason=KNOWN[(case.route.name, tag)]))
+    g = graph_for(case.shape, case.engine, indexed=case.route.indexed)
+    calls = _served(case, monkeypatch)
+    try:
+        served = g.gfql(case.shape.build(), engine=case.engine)
+    except Exception as served_exc:
+        with routes_off(ALL_ROUTES):
+            with pytest.raises(type(served_exc)):
+                g.gfql(case.shape.build(), engine=case.engine)
+        return
+    with routes_off(ALL_ROUTES):
+        general = g.gfql(case.shape.build(), engine=case.engine)
+        oracle = _sig(graph_for(case.shape, "pandas").gfql(case.shape.build(), engine="pandas"), case.shape.frames)
+    assert _canon(served._nodes) == _canon(general._nodes), f"{case.id}: node rows differ from the general path"
+    assert _canon(served._edges) == _canon(general._edges), f"{case.id}: edge rows differ from the general path"
+    assert _sig(served, case.shape.frames) == oracle, f"{case.id}: node/edge sets differ from the pandas general path"
+    if calls["served"] == 0:
+        pytest.xfail(f"{case.id}: admitted by the predicate, declined by the lane body (attenuation ledger)")
+
+
+def test_every_route_serves_most_of_what_it_admits(monkeypatch):
+    """A lane that declines most admitted shapes has a predicate that no longer describes it."""
+    per_route: Dict[str, List[int]] = {}
+    for case in CASES:
+        if case.engine != ("polars" if case.route.name.startswith("polars") else "pandas"):
+            continue
+        if case.engine == "polars":
+            pytest.importorskip("polars")
+        g = graph_for(case.shape, case.engine, indexed=case.route.indexed)
+        calls = _served(case, monkeypatch)
+        try:
+            g.gfql(case.shape.build(), engine=case.engine)
+        except Exception:
+            continue
+        per_route.setdefault(case.route.name, []).append(calls["served"] > 0)
+    for route, served in per_route.items():
+        assert sum(served) * 2 >= len(served), f"{route}: served {sum(served)} of {len(served)} admitted shapes"
+
+
+def test_every_route_has_admitted_shapes():
+    covered = {(c.route.name, c.engine) for c in CASES}
+    for route in ROUTES:
+        for engine in route.engines:
+            assert (route.name, engine) in covered, f"{route.name}/{engine} admits no registered shape"