diff --git a/graphistry/compute/gfql/index/api.py b/graphistry/compute/gfql/index/api.py index ab19c8a58f..8fa76166bd 100644 --- a/graphistry/compute/gfql/index/api.py +++ b/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 diff --git a/graphistry/compute/gfql/index/build.py b/graphistry/compute/gfql/index/build.py index 472fb94709..cd27c2413e 100644 --- a/graphistry/compute/gfql/index/build.py +++ b/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])) diff --git a/graphistry/compute/gfql/index/traverse.py b/graphistry/compute/gfql/index/traverse.py index fcfaa7dae6..3a997d3c8b 100644 --- a/graphistry/compute/gfql/index/traverse.py +++ b/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 diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain_specializations/admission.py b/graphistry/compute/gfql/lazy/engine/polars/chain_specializations/admission.py index 6b259ee2b1..2e00682dd3 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain_specializations/admission.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain_specializations/admission.py @@ -24,12 +24,14 @@ 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 @@ -37,8 +39,10 @@ def polars_plain_single_hop_admits(ops: Sequence[ASTObject], start_nodes: Option 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 diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain_specializations/hotpaths.py b/graphistry/compute/gfql/lazy/engine/polars/chain_specializations/hotpaths.py index cd54a06727..22f4ac4214 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain_specializations/hotpaths.py +++ b/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, ) diff --git a/graphistry/tests/compute/gfql/index/test_node_type_segments.py b/graphistry/tests/compute/gfql/index/test_node_type_segments.py index a4d45446ce..511bf8d044 100644 --- a/graphistry/tests/compute/gfql/index/test_node_type_segments.py +++ b/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.""" diff --git a/graphistry/tests/compute/gfql/lazy/engine/polars/chain_specializations/test_polars_admission.py b/graphistry/tests/compute/gfql/lazy/engine/polars/chain_specializations/test_polars_admission.py index a215038b6d..d04854f219 100644 --- a/graphistry/tests/compute/gfql/lazy/engine/polars/chain_specializations/test_polars_admission.py +++ b/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,