perf(gfql): lower simple destination node filters into indexed hops

graphistry/compute/gfql/index/api.py

@@ -1165,6 +1165,10 @@ def _hop_is_index_coverable(
         return_as_wave_front
         and is_simple_equality_node_match(source_node_match)
         and is_simple_equality_node_match(destination_node_match)
+    ) and not (
+        not return_as_wave_front
+        and source_node_match is None
+        and is_simple_equality_node_match(destination_node_match)
     ):
         return False
     # Typed-edge edge_match: coverable only as a simple scalar-equality filter on the

graphistry/compute/gfql/index/build.py

@@ -219,7 +219,7 @@ def build_node_type_segment(
         col_to_array(matching, position_column, engine), dtype=xp.int64,
     )
     if engine == Engine.CUDF and isinstance(type_value, str):
-        # cupy cannot encode text scalars; segments are looked up by registry key, so retain string values natively.
+        # cupy cannot encode text; registry keys drive lookup, not keys_sorted.
         import cudf  # type: ignore[import]
 
         keys = cast(ArrayLike, cudf.Series([type_value]))

graphistry/compute/gfql/index/traverse.py

@@ -466,9 +466,17 @@ def index_seeded_hop(
 
         neighbors = neigh_parts[0] if len(neigh_parts) == 1 else xp.concatenate(neigh_parts)
         if first and not return_as_wave_front:
-            matched_all = (
-                matched_parts[0] if len(matched_parts) == 1 else xp.concatenate(matched_parts)
-            )
+            if edge_filter is None and dest_allowed is None:
+                matched_all = (
+                    matched_parts[0] if len(matched_parts) == 1 else xp.concatenate(matched_parts)
+                )
+            else:
+                # A filter can reject a source's only edge: first-hop visited = surviving rows' sources.
+                surviving = (
+                    edge_rows_parts[0] if len(edge_rows_parts) == 1
+                    else xp.concatenate(edge_rows_parts)
+                )
+                matched_all = col_to_array(edges, src, engine)[surviving]
             visited = sorted_unique(xp, matched_all)
         first = False
 

graphistry/compute/gfql/lazy/engine/polars/chain_specializations/admission.py

@@ -24,21 +24,25 @@ def _plain_edge(op: ASTObject) -> bool:
 
 
 def polars_plain_single_hop_admits(ops: Sequence[ASTObject], start_nodes: Optional[DataFrameT]) -> Optional[PolarsPlainSingleHopShape]:
-    """The polars chain's plain single-hop branch for ``ops``: ``"seeded-index"`` when the
-    resident-index hop is consulted first (seed filter, no destination filter, directed),
-    ``"skip-combine"`` when the one-hop endpoint filter serves it without the
-    forward/backward/combine passes, None when the full chain runs. The dispatcher calls
-    this; unnamed, unqueried nodes and an unnamed, unmatched simple edge are the shape.
-    A filtered undirected hop is the one plain shape that still takes the full chain."""
+    """The polars chain's plain single-hop branch for ``ops``.
+
+    ``"seeded-index"`` consults the resident-index hop first (seed filter, directed).
+    A simple destination node filter is also an index candidate: the indexed hop can
+    apply it as a node segment or canonical residual predicate. ``"skip-combine"``
+    serves the remaining plain single hops without forward/backward/combine; ``None``
+    means the full chain runs. A filtered undirected hop still takes the full chain.
+    """
     if start_nodes is not None or len(ops) != 3:
         return None
     n0, e1, n2 = ops
     if not (_plain_node(n0) and _plain_edge(e1) and _plain_node(n2)):
         return None
     assert isinstance(n0, ASTNode) and isinstance(e1, ASTEdge) and isinstance(n2, ASTNode)
     directed = e1.direction in ("forward", "reverse")
-    if n0.filter_dict and not n2.filter_dict and directed:
-        return "seeded-index"
+    if n0.filter_dict and directed:
+        from graphistry.compute.gfql.index.traverse import is_simple_equality_node_match
+        if not n2.filter_dict or is_simple_equality_node_match(n2.filter_dict):
+            return "seeded-index"
     unconstrained = not n0.filter_dict and not n2.filter_dict
     return "skip-combine" if (unconstrained or directed) else None
 

graphistry/compute/gfql/lazy/engine/polars/chain_specializations/hotpaths.py

@@ -54,8 +54,8 @@ def _plain_seeded_index_hop_polars(g: Plottable, ops: Sequence[ASTObject]) -> Op
     from graphistry.Engine import Engine
     from graphistry.compute.gfql.index import get_index_policy, get_registry, maybe_index_hop
     from graphistry.compute.gfql.lazy import active_target, ExecutionTarget
-    n0, e1 = ops[0], ops[1]
-    assert isinstance(n0, ASTNode) and isinstance(e1, ASTEdge)  # the predicate admitted this shape
+    n0, e1, n2 = ops
+    assert isinstance(n0, ASTNode) and isinstance(e1, ASTEdge) and isinstance(n2, ASTNode)  # the predicate admitted this shape
     policy = get_index_policy(g)
     if get_registry(g).is_empty() and policy not in ("auto", "force"):
         return None
@@ -65,6 +65,7 @@ def _plain_seeded_index_hop_polars(g: Plottable, ops: Sequence[ASTObject]) -> Op
     return maybe_index_hop(
         gf0, engine, nodes=seed0, hops=1, direction=e1.direction,
         return_as_wave_front=False, to_fixed_point=False, policy=policy,
+        destination_node_match=n2.filter_dict or None,
     )
 
 

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

@@ -105,6 +105,76 @@ def test_node_segment_matches_scan_single_hop(engine: str, source_match, dest_ma
 ]
 
 
+def _partial_reject_graph(engine: str):
+    """Seed 2's only reverse candidate (4->2) fails a destination-Q filter.
+
+    Seed 1 also has a reverse Q candidate (3->1), while 4->2 exercises
+    rejection of the candidate itself rather than absence of an incoming edge.
+    """
+    nodes = pd.DataFrame({"id": [1, 2, 3, 4], "type": ["P", "P", "Q", "P"]})
+    edges = pd.DataFrame({"src": [1, 2, 3, 4], "dst": [3, 4, 1, 2]})
+    if engine == "polars":
+        pl = pytest.importorskip("polars")
+        nodes, edges = pl.from_pandas(nodes), pl.from_pandas(edges)
+    elif engine == "cudf":
+        cudf = pytest.importorskip("cudf")
+        nodes, edges = cudf.from_pandas(nodes), cudf.from_pandas(edges)
+    return graphistry.bind(node="id", source="src", destination="dst").nodes(nodes).edges(edges)
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars", "cudf"])
+@pytest.mark.parametrize("direction", ["forward", "reverse", "undirected"])
+def test_direct_hop_partial_seed_reject_matches_scan(engine: str, direction: str) -> None:
+    """First-hop visited must drop seeds whose only edge a destination filter rejects."""
+    graph = _partial_reject_graph(engine)
+    indexed = graph.gfql_index_all(engine=engine).gfql_index_node_types(
+        type_column="type", type_values=["Q"], engine=engine)
+    kw = dict(
+        hops=1, direction=direction, return_as_wave_front=False,
+        destination_node_match={"type": "Q"},
+    )
+    seeds = graph._nodes[graph._nodes["id"].isin([1, 2])] if engine == "pandas" else graph._nodes.head(2)
+    if direction == "reverse":
+        prefilter = graph.hop(
+            nodes=seeds, engine=engine, hops=1, direction=direction,
+            return_as_wave_front=False,
+        )
+        assert _hop_sig(prefilter) == (
+            [1, 2, 3, 4], [(3, 1), (4, 2)],
+        )
+    expected = graph.hop(nodes=seeds, engine=engine, **kw)
+    for policy in ["off", "force"]:
+        actual = with_index_policy(indexed, policy).hop(nodes=seeds, engine=engine, **kw)
+        assert actual is not None
+        assert _hop_sig(actual) == _hop_sig(expected), policy
+    # Node 4 is non-Q, so seed 2's only reverse candidate is rejected.
+    expected_edges = {
+        "forward": [(1, 3)],
+        "reverse": [(3, 1)],
+        "undirected": [(1, 3), (3, 1)],
+    }[direction]
+    assert _hop_sig(expected) == ([1, 3], expected_edges)
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars", "cudf"])
+def test_direct_hop_all_destinations_rejected_matches_scan(engine: str) -> None:
+    """A destination filter matching no node must yield the same empty result."""
+    graph = _partial_reject_graph(engine)
+    indexed = graph.gfql_index_all(engine=engine).gfql_index_node_types(
+        type_column="type", type_values=["Z"], engine=engine)
+    kw = dict(
+        hops=1, direction="forward", return_as_wave_front=False,
+        destination_node_match={"type": "Z"},
+    )
+    seeds = graph._nodes[graph._nodes["id"].isin([1, 2])] if engine == "pandas" else graph._nodes.head(2)
+    expected = graph.hop(nodes=seeds, engine=engine, **kw)
+    assert len(_pandas(expected._nodes)) == 0
+    for policy in ["off", "force"]:
+        actual = with_index_policy(indexed, policy).hop(nodes=seeds, engine=engine, **kw)
+        assert actual is not None
+        _assert_same_graph(actual, expected)
+
+
 @pytest.mark.parametrize("engine", ["pandas", "polars", "cudf"])
 @pytest.mark.parametrize("direction", ["forward", "reverse", "undirected"])
 @pytest.mark.parametrize("source_match,dest_match", DIRECT_HOP_CASES)
@@ -153,6 +223,30 @@ def test_node_segment_direct_hop_engagement(
     } for step in steps)
 
 
+@pytest.mark.route_engaged("index-hop", "polars-plain")
+def test_plain_polars_chain_consumes_destination_node_segment() -> None:
+    """A destination node-type filter is lowered into the indexed single hop.
+
+    The chain seed has already applied the source filter, so this shape admits a
+    destination-only node match without asking the segment path to re-filter seeds.
+    """
+    engine = "polars"
+    graph = _graph(engine)
+    indexed = _indexed(graph, engine)
+    query = [
+        n({"id": 1, "type": "P"}), e_forward(), n({"type": "Q"}), rows(),
+    ]
+    expected = graph.gfql(query, engine=engine, index_policy="off")
+    with index_trace() as steps:
+        actual = indexed.gfql(query, engine=engine, index_policy="force")
+    _assert_same_graph(actual, expected)
+    assert any(
+        step.get("path") == "index"
+        and step.get("node_segments", {}).get("destination") == "resident"
+        for step in steps
+    ), steps
+
+
 @pytest.mark.parametrize("engine", ["pandas", "polars", "cudf"])
 def test_node_segment_two_hop_chain_matches_scan(engine: str) -> None:
     """The IC4-realistic shape: labeled nodes on every pattern of a two-hop chain."""

graphistry/tests/compute/gfql/lazy/engine/polars/chain_specializations/test_polars_admission.py

@@ -1,9 +1,10 @@
 """``polars_plain_single_hop_admits`` is the polars chain's own gate for its plain single-hop
 branches.
 
-Pins: the decision table over the shared corpus (``"seeded-index"`` for a directed seeded hop
-without a destination filter, ``"skip-combine"`` for the other plain single hops except the
-filtered undirected one, None otherwise), a seeded chain (``start_nodes``) declines, and every
+Pins: the decision table over the shared corpus (``"seeded-index"`` for a directed seeded hop,
+including a simple destination node filter, ``"skip-combine"`` for the other plain single hops
+except the filtered undirected one, None otherwise), a seeded chain (``start_nodes``) declines,
+and every
 admitted shape stays value-identical to the pandas full path.
 """
 import pandas as pd
@@ -22,7 +23,7 @@
     "plain single hop, unseeded": "skip-combine",
     "plain single hop, seeded": "seeded-index",
     "plain single hop, seeded, reverse": "seeded-index",
-    "plain single hop, seeded, destination filter": "skip-combine",
+    "plain single hop, seeded, destination filter": "seeded-index",
     "plain single hop, undirected, unconstrained": "skip-combine",
     "plain single hop, undirected, seeded": None,
     "single hop, prune to endpoints": None,