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

bin/test-polars.sh

@@ -122,6 +122,7 @@ POLARS_TEST_FILES=(
     graphistry/tests/compute/gfql/row/test_alias_prefilter_alignment_2020.py
     graphistry/tests/compute/gfql/lazy/engine/polars/test_chain_alias_column_collision_2039.py
     graphistry/tests/compute/test_chain_alias_column_collision.py
+    graphistry/tests/compute/gfql/routes/test_route_harness.py
     graphistry/tests/compute/gfql/test_engine_polars_semi_key_dedup.py
     graphistry/tests/compute/gfql/test_engine_polars_call_modality.py
     graphistry/tests/compute/gfql/test_engine_polars_gpu.py

graphistry/tests/compute/gfql/cypher/test_lowering.py

@@ -18626,6 +18626,7 @@ def test_t6_col_stats_decisions_are_visible_in_the_trace() -> None:
         outcomes={"nodes.id": "served", "edges.s": "served"})
 
 
+@pytest.mark.route_engaged("cypher-fast")
 def test_t6_assert_col_stats_helper_fails_loudly() -> None:
     """The helper must FAIL when the optimization did not fire -- an engagement
     pin that cannot fail is worse than none, which is the whole failure mode
@@ -19225,6 +19226,7 @@ def _mk_h3_case_data(fixture: str) -> Tuple[pd.DataFrame, pd.DataFrame]:
     raise AssertionError(f"unknown fixture {fixture}")
 
 
+@pytest.mark.route_engaged("cypher-fast")
 @pytest.mark.parametrize("engine", ["polars", "polars-gpu"])
 @pytest.mark.parametrize("label,fixture,query", _H3_DIFFERENTIAL_CASES, ids=[c[0] for c in _H3_DIFFERENTIAL_CASES])
 def test_h3_fused_two_hop_count_matches_eager_twin_and_pandas(
@@ -19249,6 +19251,7 @@ def test_h3_fused_two_hop_count_matches_eager_twin_and_pandas(
     assert fused == oracle, f"{label}: fused lane diverged from the pandas oracle"
 
 
+@pytest.mark.route_engaged("cypher-fast")
 @pytest.mark.parametrize("engine", ["polars", "polars-gpu"])
 def test_h3_fused_two_hop_count_empty_match_counts_zero(engine: str, monkeypatch: pytest.MonkeyPatch) -> None:
     """openCypher counts over no rows as 0 -- not an empty frame."""
@@ -19359,6 +19362,7 @@ def test_h3_two_hop_count_fast_path_has_no_order_by_or_limit_surface(suffix: str
     assert _two_hop_count_alias(compiled.chain) == expect_alias
 
 
+@pytest.mark.route_engaged("cypher-fast")
 @pytest.mark.parametrize("engine", ["polars", "polars-gpu"])
 def test_h3_fused_two_hop_count_handles_degenerate_bindings(engine: str, monkeypatch: pytest.MonkeyPatch) -> None:
     """The node key may share a name with an endpoint column, and source/destination may be bound

graphistry/tests/compute/gfql/index/test_degree_consult.py

@@ -102,6 +102,7 @@ def test_identity_anchors_to_the_bound_frame_not_the_partition() -> None:
     assert fact.source_ref is g._edges
 
 
+@pytest.mark.route_engaged("cypher-fast")
 @pytest.mark.parametrize("engine", ENGINES)
 @pytest.mark.parametrize("n_p,n_c", [(3, 3), (5, 1), (2, 8), (7, 2)])
 def test_slice_is_exact_across_domain_shapes(n_p: int, n_c: int, engine: str) -> None:
@@ -117,6 +118,7 @@ def test_slice_is_exact_across_domain_shapes(n_p: int, n_c: int, engine: str) ->
     assert value == oracle
 
 
+@pytest.mark.route_engaged("cypher-fast")
 def test_gapped_node_space_builds_facts_and_stays_exact() -> None:
     """Density is NOT required for the degree arrays: ids absent from the span
     contribute ZERO to the dot, so a gapped node space builds valid facts. (The
@@ -137,6 +139,7 @@ def test_gapped_node_space_builds_facts_and_stays_exact() -> None:
     assert used, "P-domain [0,2] is dense, so the kernel must consult the fact"
 
 
+@pytest.mark.route_engaged("cypher-fast")
 @pytest.mark.parametrize("seed", range(6))
 def test_differential_vs_the_scan_on_random_typed_graphs(seed: int) -> None:
     """Values must be identical with and without the fact, on arbitrary degree

graphistry/tests/compute/gfql/index/test_index.py

@@ -1465,6 +1465,7 @@ def _polars_indexed_graph():
     return g.gfql_index_all(engine="polars")
 
 
+@pytest.mark.route_engaged("index-hop")
 def test_auto_engine_gfql_serves_polars_index_1767_cliff():
     """#1767 cliff pin: polars frames + explicit polars index + gfql with NO engine
     argument must serve path=index on engine=polars (AUTO routes native, so the
@@ -1701,6 +1702,7 @@ def test_col_stats_auto_narrows_lazy_frames(self):
         gi = gl.gfql_index_col_stats()  # AUTO on lazy frames must not crash
         assert gi is not None
 
+    @pytest.mark.route_engaged("index-hop")
     def test_inversion_auto_index_auto_gfql_serves_polars_index(self):
         """THE INVERSION PIN. The exact scenario the retracted #1767 regressed
         to the scan floor: ``gfql_index_all()`` with NO engine + ``g.gfql(<index-

graphistry/tests/compute/gfql/index/test_index_gpu_edge_match.py

@@ -85,6 +85,7 @@ def _match_value(null_col_dtype: str):
     return {"string": "1", "boolean": True}.get(null_col_dtype, 1)
 
 
+@pytest.mark.route_engaged("index-hop")
 @pytest.mark.parametrize("engine", ["cudf", "polars-gpu"])
 @pytest.mark.parametrize("null_col_dtype", NULL_DTYPES)
 def test_null_bearing_edge_predicate_matches_the_pandas_oracle_on_device(engine, null_col_dtype):
@@ -110,6 +111,7 @@ def pairs(gg):
     assert pairs(got) == pairs(oracle), f"[{engine}/{null_col_dtype}] diverged from pandas"
 
 
+@pytest.mark.route_engaged("index-hop")
 @pytest.mark.parametrize("engine", ["cudf", "polars-gpu"])
 def test_empty_candidate_batch_on_device(engine):
     """A seed with no matching typed edges yields a zero-length gather map on device."""

graphistry/tests/compute/gfql/index/test_indexed_bindings.py

@@ -405,6 +405,7 @@ def test_connected_path_bag_trace_and_lifecycle(
 ]
 
 
+@pytest.mark.route_engaged("indexed-kernel")
 @pytest.mark.parametrize("engine", ENGINES)
 @pytest.mark.parametrize("query", STANDARD_DERIVED_POSITIVES)
 def test_standard_derived_connected_parity(
@@ -454,6 +455,7 @@ def unexpected_traversal(*args: Any, **kwargs: Any) -> Any:
     _assert_decision(decisions[0], seam="connected_bindings", served=True)
 
 
+@pytest.mark.route_engaged("cypher-fast")
 @pytest.mark.parametrize("engine", ENGINES)
 def test_destination_property_projection_dtype_parity(
     engine: str,
@@ -726,6 +728,7 @@ def test_node_property_index_prefers_the_most_selective_column(
         pytest.param({"grp": 0}, "grp", False, id="unselective-keeps-scan"),
     ],
 )
+@pytest.mark.route_engaged("index-hop", "indexed-kernel")
 def test_node_property_index_cost_gate_under_policy_use(
     seed: Dict[str, Any],
     indexed_column: str,

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)