refactor(gfql): route flow productized: chain_specializations layout, routes-off replay, shape registry + route harness, CI ledger

.github/workflows/ci.yml

@@ -242,11 +242,19 @@ jobs:
           '^graphistry/compute/gfql/' \
           '^graphistry/compute/gfql_validate\.py$' \
           '^graphistry/compute/gfql_unified\.py$' \
+          '^graphistry/compute/gfql_fast_paths\.py$' \
+          '^graphistry/compute/chain.*\.py$' \
+          '^graphistry/compute/hop\.py$' \
+          '^graphistry/compute/filter_by_dict\.py$' \
+          '^graphistry/compute/ast\.py$' \
+          '^graphistry/compute/predicates/' \
           '^graphistry/models/gfql/' \
           '^graphistry/Plottable\.py$' \
           '^graphistry/tests/benchmarks/gfql/' \
           '^graphistry/tests/compute/gfql/' \
           '^graphistry/tests/compute/test_gfql.*\.py$' \
+          '^graphistry/tests/compute/test_chain.*\.py$' \
+          '^graphistry/tests/compute/test_hop.*\.py$' \
           '^graphistry/tests/test_gfql_.*\.py$' \
           '^tests/gfql/'
 
@@ -1589,6 +1597,63 @@ jobs:
         source pygraphistry/bin/activate
         ./bin/test-graphviz.sh
 
+  gfql-routes-off:
+    # Non-blocking ledger: replays the GFQL suites with one hot path declined per cell
+    # (GFQL_ROUTES_OFF); engagement pins are skipped via the route_engaged marker, so the
+    # uploaded lists hold route-vs-general result divergences only.
+    needs: [changes, test-gfql-core, generate-lockfiles]
+    if: ${{ needs.changes.outputs.gfql == 'true' || github.event_name == 'workflow_dispatch' || github.event_name == 'schedule' }}
+    runs-on: ubuntu-latest
+    continue-on-error: true
+    timeout-minutes: 45
+
+    strategy:
+      fail-fast: false
+      matrix:
+        mode: [native-fast, polars-seeded, polars-plain, index-hop, indexed-kernel, cypher-fast, all-off]
+
+    steps:
+
+    - name: Checkout repo
+      uses: actions/checkout@v4
+      with:
+        lfs: true
+        persist-credentials: false
+
+    - name: Set up Python 3.12
+      uses: actions/setup-python@v5
+      with:
+        python-version: 3.12
+
+    - name: Download lockfiles
+      uses: actions/download-artifact@v4
+      with:
+        name: lockfiles
+        path: requirements
+
+    - name: Install Python dependencies
+      run: |
+        python -m venv pygraphistry
+        source pygraphistry/bin/activate
+        python -m pip install --upgrade pip uv
+        uv pip install --require-hashes -r requirements/test-polars-py3.12.lock
+        uv pip install -e . --no-deps
+
+    - name: Routes-off replay (${{ matrix.mode }})
+      run: |
+        source pygraphistry/bin/activate
+        MODES=${{ matrix.mode }} ./bin/test-routes-off.sh
+        echo "## routes-off: ${{ matrix.mode }}" >> "$GITHUB_STEP_SUMMARY"
+        sed 's/^/- /' build/routes-off/${{ matrix.mode }}.divergences >> "$GITHUB_STEP_SUMMARY"
+
+    - name: Upload ledger
+      if: always()
+      uses: actions/upload-artifact@v4
+      with:
+        name: routes-off-${{ matrix.mode }}
+        path: build/routes-off/
+        if-no-files-found: ignore
+
   test-polars:
     needs: [changes, test-minimal-python, test-gfql-core, generate-lockfiles]
     if: ${{ ((needs.changes.outputs.python == 'true' && needs.changes.outputs.narrow_python_only != 'true') || needs.changes.outputs.gfql == 'true' || needs.changes.outputs.pandas_compat == 'true' || needs.changes.outputs.core == 'true' || needs.changes.outputs.infra == 'true' || github.event_name == 'workflow_dispatch' || github.event_name == 'schedule') && !(needs.changes.outputs.docs_only_latest == 'true' && (github.event_name == 'push' || github.event_name == 'pull_request')) }}

CHANGELOG.md

@@ -11,6 +11,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Infrastructure
 
+- **CI: the gfql change filter now includes the chain engine** (`compute/chain*.py`, `hop.py`, `gfql_fast_paths.py`, `filter_by_dict.py`, `ast.py`, `predicates/`, and the chain/hop test files), so tck-gfql, the Cypher-frontend gates and the gfql benchmark lane run on a change to the chain engine; they were skipped on #2055.
 - **CI: `test-docs` runs on docs-only pull requests (#2018)**: the job needed `python-lint-types`, which a docs-only change skips, and GitHub skips a job whose prerequisite was skipped. The gate now accepts skipped prerequisites and refuses only failed or cancelled ones, so documentation changes are built and tested before merge.
 ### Tests
 
@@ -40,6 +41,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Changed
 
+* GFQL: the chain specializations move into `graphistry/compute/chain_specializations/{admission,hotpaths}.py` (pandas/cuDF single-node lane, seeded typed single hop, seeded typed RETURN-destination) and `graphistry/compute/gfql/lazy/engine/polars/chain_specializations/{admission,hotpaths}.py` (polars plain single-hop branches, seeded lane, RETURN-destination), each lane next to the admission predicate the dispatcher calls (`native_fast_path_admits`, `polars_plain_single_hop_admits`, `polars_seeded_lane_admits`); `chain.py` and the polars chain only dispatch, `chain_fast_paths.py` keeps the shared seed/index helpers. No route admits or declines anything it did not before. Tests mirror the new paths and filter one shared shape corpus per route with the route's own gate; `GFQL_ROUTES_OFF=<route,...>` (test conftest) makes named hot paths decline so every existing test replays through the other routes, and `bin/test-routes-off.sh` reports the per-route divergences.
 * GFQL: the wavefront seed-rediscovery rule moved out of `hop.py` into `graphistry/compute/gfql/seed_rediscovery.py` (pandas/cuDF) and `graphistry/compute/gfql/lazy/engine/polars/seed_rediscovery.py` (polars); `undirected_rediscovered_seed_ids` (an internal helper) is gone.
 
 ## [0.59.0 - 2026-08-31]

bin/test-polars.sh

@@ -87,6 +87,8 @@ POLARS_TEST_FILES=(
     graphistry/tests/compute/gfql/test_seeded_node_lookup_fastpath.py
     graphistry/tests/compute/gfql/test_native_seed_resolution_2027.py
     graphistry/tests/compute/gfql/test_polars_native_seed_resolution.py
+    graphistry/tests/compute/gfql/lazy/engine/polars/chain_specializations/test_polars_admission.py
+    graphistry/tests/compute/chain_specializations/test_native_admission.py
     graphistry/tests/compute/gfql/test_undirected_pairs_2026.py
     graphistry/tests/compute/gfql/test_native_seed_lane_explain.py
     # #1882/#1913-f4/#1879 crash-family pins: the polars params (filter helpers on polars
@@ -123,6 +125,7 @@ POLARS_TEST_FILES=(
     graphistry/tests/compute/gfql/cypher/test_variable_column_collision.py
     graphistry/tests/compute/test_chain_alias_column_collision.py
     graphistry/tests/compute/test_gfql_op_list_hides_internal_columns.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

bin/test-routes-off.sh

@@ -0,0 +1,17 @@
+#!/usr/bin/env bash
+# Replay the GFQL suites with each hot path declined (GFQL_ROUTES_OFF, see graphistry/tests/conftest.py).
+# Engagement pins carry the route_engaged marker and are skipped, so every remaining failure is a
+# route-vs-general result divergence. Non-blocking ledger: always exits 0; per-mode logs + id lists in $OUT.
+set -uo pipefail
+cd "$(dirname "$0")/.."
+MODES=${MODES:-native-fast polars-seeded polars-plain index-hop indexed-kernel cypher-fast all-off}
+SUITES=${SUITES:-graphistry/tests/compute/test_chain.py graphistry/tests/compute/test_hop.py graphistry/tests/compute/test_gfql.py graphistry/tests/compute/gfql}
+OUT=${OUT:-build/routes-off}
+mkdir -p "$OUT"
+for mode in $MODES; do
+  if [ "$mode" = all-off ]; then routes=native-fast,polars-seeded,polars-plain,index-hop,indexed-kernel,cypher-fast; else routes=$mode; fi
+  GFQL_ROUTES_OFF=$routes python -m pytest $SUITES -q -p no:cacheprovider -o addopts="" -rfE > "$OUT/$mode.log" 2>&1
+  grep -E "^(FAILED|ERROR) " "$OUT/$mode.log" | sed 's/ - .*//' | sort -u > "$OUT/$mode.divergences"
+  echo "$mode: $(wc -l < "$OUT/$mode.divergences") divergence id(s); $(tail -1 "$OUT/$mode.log")"
+done
+exit 0

graphistry/compute/chain.py

@@ -11,11 +11,7 @@
 from .ast import ASTObject, ASTNode, ASTEdge, ASTCall, Direction, from_json as ASTObject_from_json, serialize_binding_ops
 from .typing import DataFrameT, SeriesT
 from .util import generate_safe_column_name
-from .chain_fast_paths import (
-    _seeded_typed_hop_pandas_cudf,
-    _single_node_rows_via_index_or_filter,
-    _tag_fast_path_aliases,
-)
+from .chain_specializations.hotpaths import _try_chain_fast_path
 from graphistry.compute.validate.validate_schema import validate_chain_schema, validate_graph_shape
 from graphistry.compute.gfql.strictness import StrictInput
 from graphistry.compute.gfql.same_path_types import (
@@ -840,193 +836,6 @@ def _step_with_source_edge_columns(g: Plottable, g_step: Plottable, op: ASTObjec
     return g_step.edges(edges[edges[edge_id].isin(step_edges[edge_id])])
 
 
-def _try_chain_fast_path(
-    g_in: Plottable,
-    ops: List[ASTObject],
-    engine_concrete: Engine,
-    start_nodes: Optional[DataFrameT] = None,
-) -> Optional[Plottable]:
-    """Degenerate-shape fast path (pandas/cuDF): node-only ``MATCH (n)`` or a plain
-    single-hop ``MATCH (a)-[e]->(b)`` skip the forward/backward/combine BFS machinery.
-    Returns the result Plottable, or ``None`` to fall through to the full path.
-
-    Same node/edge sets + VALUES as the full machinery (trackA_golden + hop/chain
-    suites); the 1-hop additionally preserves int node dtypes (the full path upcasts
-    int→float via merge — the merge is the artifact, int is the Cypher-conformant type).
-    Gated to unqueried nodes + a plain single-hop edge; NAMED ops are served (the alias
-    flags are reconstructed by `_tag_fast_path_aliases`) except when undirected or when
-    the same alias is reused. filtered-undirected and seeded chains fall through.
-    polars/dask/spark also fall through (own fast path / lazy semantics)."""
-    from graphistry.compute.filter_by_dict import filter_by_dict
-
-    if engine_concrete not in (Engine.PANDAS, Engine.CUDF):
-        return None
-    if start_nodes is not None:
-        return None  # seeded chains use the full path (fast path has no seed)
-    engine_abs = EngineAbstract(engine_concrete.value)
-
-    def _materialize_fast_path_graph() -> Plottable:
-        from graphistry.compute.ComputeMixin import _coerce_input_formats  # lazy — avoids circular import
-        g = g_in.materialize_nodes(engine=EngineAbstract(engine_concrete.value))
-        return _coerce_input_formats(g, engine_concrete)
-
-    if len(ops) == 1:
-        n0 = ops[0]
-        if not (isinstance(n0, ASTNode) and n0.query is None):
-            return None
-        g = _materialize_fast_path_graph()
-        if g._nodes is None:
-            return None
-        nodes = _single_node_rows_via_index_or_filter(g, n0, engine_abs)
-        if n0._name is not None:
-            alias_was_column = n0._name in nodes.columns
-            if alias_was_column:
-                nodes = nodes.drop(columns=[n0._name])
-            nodes = nodes.assign(**{n0._name: True})
-            other_columns = [c for c in nodes.columns if c != n0._name]
-            if g._node in other_columns:
-                other_columns = [g._node, *[c for c in other_columns if c != g._node]]
-            if alias_was_column:
-                nodes = nodes[[*other_columns, n0._name]]
-            else:
-                nodes = nodes[[*other_columns[:1], n0._name, *other_columns[1:]]]
-            nodes = nodes.reset_index(drop=True)
-        edges = g._edges.iloc[0:0] if g._edges is not None else None
-        return g.nodes(nodes).edges(edges) if edges is not None else g.nodes(nodes)
-
-    if len(ops) != 3:
-        return None
-    n0, e1, n2 = ops
-    # Aliases are a PROJECTION concern, not a traversal one: capture them, serve the
-    # traversal on the fast path, and tag the result (_tag_fast_path_aliases). Rejecting
-    # them here sent a NAMED `g.gfql([n(name=..), e(..), n(name=..)])` to the full
-    # two-pass BFS purely because the ops carried names — measured ~25.2 ms before vs
-    # ~2.3 ms after (medians of 5 paired runs), on a 200-node graph where data work is ~0.
-    # SCOPE, measured: this does NOT reach the Cypher `MATCH ... RETURN` surface for the
-    # benchmark shapes. Those are served earlier by `gfql_fast_paths.py` and never consult
-    # this function at all, so do not attribute a Cypher-surface win to this gate.
-    alias_n0, alias_e1, alias_n2 = n0._name, e1._name, n2._name
-    _named = [a for a in (alias_n0, alias_e1, alias_n2) if a is not None]
-    if len(_named) != len(set(_named)):
-        # Duplicate alias reuse is an E201 error, and `combine_steps` is what raises it.
-        # Serving these here would BYPASS that check and silently succeed — decline so the
-        # full path still errors. (Caught by test_polars_duplicate_alias_declines_like_pandas.)
-        return None
-    if not (isinstance(n0, ASTNode) and n0.query is None):
-        return None
-    if not (isinstance(n2, ASTNode) and n2.query is None):
-        return None
-    if not (isinstance(e1, ASTEdge) and e1.is_simple_single_hop()
-            and e1.source_node_match is None
-            and e1.destination_node_match 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
-    if direction == "undirected" and (alias_n0 is not None or alias_n2 is not None):
-        # An undirected edge makes a node reachable as EITHER endpoint, so "which alias
-        # does this node carry" is not derivable from the endpoint columns the way it is
-        # for a directed hop. Decline to the full path rather than guess.
-        return None
-    unconstrained = not n0.filter_dict and not n2.filter_dict
-    if not unconstrained and direction == "undirected":
-        return None  # filtered-undirected (OR of both directions) -> full path
-    g = _materialize_fast_path_graph()
-    if g._nodes is None or g._edges is None:
-        return None
-    src, dst, node = g._source, g._destination, g._node
-    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
-    if alias_n0 == node or alias_n2 == node:
-        # A node alias EQUAL TO THE NODE-ID BINDING would make `_tag_fast_path_aliases`
-        # overwrite the id column with the bool flag (destroying the ids) while the full
-        # path raises on the same query ("The column label '<node>' is not unique") —
-        # a wrong-serve found by adversarial parity testing. Decline; never serve.
-        return None
-    if alias_e1 is not None and direction in ("forward", "reverse") \
-            and alias_e1 == (src if direction == "forward" else dst):
-        # An edge alias EQUAL TO THE HOP'S FROM-SIDE BINDING (forward+src / reverse+dst)
-        # made the two lanes return DIFFERENT node sets: the full path's flag overwrite
-        # corrupts its own node reduction, the fast path tags after reducing. TO-side
-        # collisions keep parity (pinned in tests) and stay served. Decline; never serve.
-        return None
-    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 _tag_fast_path_aliases(
-                    _fast_res, alias_n0, alias_e1, alias_n2, src, dst, node, direction)
-        from_col, to_col = (src, dst) if direction == "forward" else (dst, src)
-        edges = g._edges
-        if n0.filter_dict:
-            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:
-            # 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 _tag_fast_path_aliases(
-            g.nodes(nodes).edges(edges), alias_n0, alias_e1, alias_n2, src, dst, node, direction)
-    endpoints = concat([
-        edges[[src]].rename(columns={src: node}),
-        edges[[dst]].rename(columns={dst: node}),
-    ]).drop_duplicates()
-    nodes = g._nodes[g._nodes[node].isin(endpoints[node])]
-    # match the full path's merge, which collapses duplicate node-id rows
-    nodes = nodes.drop_duplicates(subset=[node])
-    return _tag_fast_path_aliases(
-        g.nodes(nodes).edges(edges), alias_n0, alias_e1, alias_n2, src, dst, node, direction)
-
-
 def reject_alias_named_like_binding(
     g: Plottable, chain_obj: "Chain", *, include_edge_endpoint_aliases: bool = False
 ) -> None: