perf(gfql): generalize seeded typed-hop fast path to polars/polars-gpu (#1755)

CHANGELOG.md

@@ -10,6 +10,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Performance
 - **GFQL seeded typed-hop fast path (#1755)**: a seeded typed 1-hop — native chain `[n({id}), e_forward(), n({type})]` or Cypher `MATCH (m {id})-[:T]->(p) RETURN p` — previously paid the full two-pass chain machinery (~20-40ms: whole-frame combines, full-column type filters, rows-pivot projection). A new seed-first fast path recognizes exactly this shape and reduces the graph to the seed's 1-hop neighborhood before any of that work: native chain 32.6→0.93ms (35×), Cypher RETURN 39.2→1.9ms (20×) on pandas, with cuDF covered via the shared DataFrame API (30.8→4.7ms). Value-identical to the full path by construction (same rows/columns/dtypes; row order and index may differ) — the helpers return `None` and fall through for anything outside the exact shape or carrying full-path side-channels (multi-hop, variable-length, undirected, predicate filters, reverse patterns, missing bindings, list-`labels` columns, policy hooks, same-path WHERE, OPTIONAL MATCH null rows, and WITH..MATCH carried seeds all decline). Null ids/endpoints never link (membership sets are null-dropped, matching the full pipeline's joins). Verified with an independent oracle (results checked against the creator set hand-computed from the raw frames, not merely fast-vs-slow agreement) plus a differential fast-vs-full sweep across shapes × engines in `test_seeded_typed_hop_fastpath.py`.
+- **GFQL seeded typed-hop fast path on polars/polars-gpu (#1755)**: extends the seeded typed-hop Cypher fast path to the polars engines via native polars filters (`_seeded_typed_return_dst_polars`), so a seeded typed 1-hop RETURN also lands in single-digit ms on polars (13.7→3.4ms, 4×) and polars-gpu (24.6→2.5ms, 10×). Dispatch is on the ACTUAL frame type (`is_polars_df`), not the requested engine, because WITH..MATCH reentry can request polars while handing pandas-materialized frames. Same decline contract as the pandas path (undirected/multi-hop/varlen/predicates fall through; value-identical for the covered shape), verified by the per-engine differential sweep and oracle tests in `test_seeded_typed_hop_fastpath.py`.
 - **GFQL seeded-chain lean combine (#1755)**: The two-pass chain executor (`_chain_impl` backward pass + `combine_steps`) reconciled wavefront boundaries with full-node-frame `safe_merge`/`pandas.merge` calls even when the seeded result was a single row, so a seeded 1-hop paid a whole-frame join. When the small side is at least `4x` smaller than the full frame (and is unique on the id with no extra columns), the byte-identical result is now obtained by an `isin` membership filter instead of the join machinery. Gated narrowly on cardinality/uniqueness/column-set and pandas-only (cuDF's GPU hash-join is already sub-ms; polars takes its own engine path); `GFQL_LEAN_COMBINE=0` restores the legacy merges. Parity is byte-identical (lean-on vs lean-off) across seeded node lookup, seeded 1-hop expand, and the native typed chain, including a null-key equivalence guard.
 
 ### Documentation

bin/test-polars.sh

@@ -27,6 +27,7 @@ POLARS_TEST_FILES=(
     graphistry/tests/compute/gfql/test_polars_nan_clean.py
     graphistry/tests/compute/gfql/test_optional_match_polars_frames.py
     graphistry/tests/compute/gfql/test_polars_rows_entity_groupby.py
+    graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py
     # index tests exercise the seeded-index hook in the polars hop entry (hop.py) — without
     # them the hook dominates the now-thin file and trips its per-file coverage floor
     graphistry/tests/compute/gfql/index/test_index.py

graphistry/compute/chain.py

@@ -818,6 +818,56 @@ def _seeded_typed_return_dst_pandas_cudf(
     return dstn, edges
 
 
+def _seeded_typed_return_dst_polars(
+    g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge,
+    src: str, dst: str, node: str, direction: Direction,
+) -> Optional[Tuple[DataFrameT, DataFrameT]]:
+    """#1755 polars analog of _seeded_typed_return_dst_pandas_cudf: same seed-first
+    reduction (seed out-edges -> typed-edge filter -> destination nodes) expressed
+    with polars filters, so a seeded cypher RETURN on polars/polars-gpu also lands
+    sub-ms. Returns ``(dst_node_rows, edges)`` (polars frames) or None to fall back
+    to the full lazy pipeline. Value-identical node set to the full path for the
+    covered shape (scalar filters, directed, single hop); row order may differ."""
+    import polars as pl
+    if direction == "undirected":
+        return None
+    nodes_df, edges_df = g._nodes, g._edges
+    # Eager polars frames only: LazyFrame has no get_column, and mixed-engine
+    # node/edge frames must take the full path — decline rather than crash.
+    if not isinstance(nodes_df, pl.DataFrame) or not isinstance(edges_df, pl.DataFrame):
+        return None
+
+    n0f = _seeded_scalar_filters(n0.filter_dict, nodes_df)
+    n2f = _seeded_scalar_filters(n2.filter_dict, nodes_df)
+    ef = _seeded_scalar_filters(e1.edge_match, edges_df)
+    if n0f is None or n2f is None or ef is None or not n0f:
+        return None
+    from_col, to_col = (src, dst) if direction == "forward" else (dst, src)
+
+    # from-side seed: reduce the node frame to the seed rows, take their ids.
+    # Membership sets are drop_nulls()'d (null ids/endpoints never link, matching
+    # the full pipeline's joins) and passed via .implode() (Series-arg is_in is
+    # deprecated in polars 1.42, see polars#22149).
+    seed_nodes = nodes_df
+    for k, v in n0f.items():
+        seed_nodes = seed_nodes.filter(pl.col(k) == v)
+    from_ids = seed_nodes.get_column(node).drop_nulls()
+    if from_ids.len() == 0:
+        return nodes_df.clear(), edges_df.clear()
+    edges = edges_df.filter(pl.col(from_col).is_in(from_ids.implode()))
+    for k, v in ef.items():  # typed edge on the reduced frontier
+        edges = edges.filter(pl.col(k) == v)
+    dst_ids = edges.get_column(to_col).drop_nulls().unique()
+    dstn = nodes_df.filter(pl.col(node).is_in(dst_ids.implode()))
+    for k, v in n2f.items():  # destination-node filter
+        dstn = dstn.filter(pl.col(k) == v)
+    # drop dangling edges + dedup destination nodes (mirror the pandas tail)
+    keep_ids = dstn.get_column(node).drop_nulls()
+    edges = edges.filter(pl.col(to_col).is_in(keep_ids.implode()))
+    dstn = dstn.filter(pl.col(node).is_in(edges.get_column(to_col).implode())).unique(subset=[node], maintain_order=True)
+    return dstn, edges
+
+
 def _try_chain_fast_path(
     g_in: Plottable,
     ops: List[ASTObject],

graphistry/compute/gfql_unified.py

@@ -989,12 +989,13 @@ def _execute_seeded_typed_hop_fast_path(
 ) -> Optional[Plottable]:
     """#1755 cypher-surface fast path: a seeded typed 1-hop with a whole-row node
     RETURN — ``MATCH (m {id})-[:T]->(p) RETURN p`` — reduces the graph to the seed's
-    1-hop neighborhood with a few DataFrame filters (pandas + cuDF), then applies the
-    RETURN projection to just the destination rows (value-identical to the full
-    path: same rows/columns/dtypes; row order and RangeIndex may differ; sub-ms
-    because the frame is tiny). Returns None to fall through for anything outside
-    this exact shape or carrying side-channels (policy, same-path WHERE, OPTIONAL
-    null-row, carried reentry seeds)."""
+    1-hop neighborhood with a few DataFrame filters (pandas/cuDF via the shared
+    DataFrame API, or polars via polars filters), then applies the RETURN projection
+    to just the destination rows (value-identical to the full path: same rows/
+    columns/dtypes; row order and RangeIndex may differ; sub-ms because the frame
+    is tiny). Returns None to fall through for anything outside this exact shape
+    or carrying side-channels (policy, same-path WHERE, OPTIONAL null-row,
+    carried reentry seeds)."""
     if start_nodes is not None:
         # A carried seed set (WITH..MATCH reentry) restricts n0 to those rows; the fast
         # path derives its seed from n0.filter_dict alone, so engaging here would
@@ -1015,7 +1016,7 @@ def _execute_seeded_typed_hop_fast_path(
         # openCypher null row; the lean projection would return an empty frame.
         return None
     requested_engine = resolve_engine(cast(Any, engine), base_graph)
-    if requested_engine not in (Engine.PANDAS, Engine.CUDF):
+    if requested_engine not in (Engine.PANDAS, Engine.CUDF, Engine.POLARS, Engine.POLARS_GPU):
         return None
     projection = compiled_query.result_projection
     if projection is None or projection.table != "nodes":
@@ -1058,17 +1059,32 @@ def _execute_seeded_typed_hop_fast_path(
     if not (n0.filter_dict and any(not str(k).startswith("label__") for k in n0.filter_dict)):
         return None  # n0 must carry a selective (non-label) seed
     direction = e1.direction
-    from graphistry.compute.chain import _seeded_typed_return_dst_pandas_cudf
-    dst_res = _seeded_typed_return_dst_pandas_cudf(
-        base_graph, n0, n2, e1, src, dst, node, direction)
+    # Dispatch on the ACTUAL frame type, not the requested engine: the WITH..MATCH
+    # reentry path can request engine=polars while handing us a pandas-materialized
+    # intermediate graph, so trusting requested_engine would run polars ops on a
+    # pandas frame (and vice versa). The pandas branch also covers cuDF (shared API).
+    from graphistry.Engine import is_polars_df
+    from graphistry.compute.chain import (
+        _seeded_typed_return_dst_pandas_cudf, _seeded_typed_return_dst_polars,
+    )
+    nodes_frame = base_graph._nodes
+    is_polars = is_polars_df(nodes_frame)
+    if is_polars != is_polars_df(base_graph._edges):
+        return None  # mixed-engine node/edge frames: decline, full path decides
+    helper = _seeded_typed_return_dst_polars if is_polars else _seeded_typed_return_dst_pandas_cudf
+    dst_res = helper(base_graph, n0, n2, e1, src, dst, node, direction)
     if dst_res is None:
         return None
     p_rows, _edges = dst_res
     # Lean projection: p_rows already IS the RETURN-alias (destination) node set.
     # Tag with the alias and reuse apply_result_projection for the exact
     # column-order/flatten semantics — all on a handful of rows, so seeded cypher
     # stays sub-ms (vs the ~25ms rows-pivot pipeline on the full graph).
-    tagged = p_rows.assign(**{projection.alias: True})
+    if is_polars:
+        import polars as pl
+        tagged = p_rows.with_columns(pl.lit(True).alias(projection.alias))
+    else:
+        tagged = p_rows.assign(**{projection.alias: True})
     result = base_graph.nodes(tagged)
     return apply_result_projection(result, projection)
 

graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py

@@ -448,3 +448,169 @@ def test_with_match_reentry_carried_seeds(self):
         full = _canon_nodes(_run_diff(g, "pandas", q, fast=False))
         pd.testing.assert_frame_equal(fast, full)
         assert sorted(fast["b.id"].tolist()) == [10, 11]
+
+
+# ---------------------------------------------------------------------------
+# polars cypher surface  (#1755 generalization: MATCH (m {id})-[:T]->(p) RETURN p)
+# ---------------------------------------------------------------------------
+
+class TestCypherSeededTypedHopPolars:
+    """The seeded cypher fast path also covers polars via _seeded_typed_return_dst_polars
+    (dispatched from _execute_seeded_typed_hop_fast_path for Engine.POLARS/POLARS_GPU).
+    Same differential-parity + independent-oracle + decline bar as the pandas class."""
+
+    def _pl_graph(self):
+        pl = pytest.importorskip("polars")
+        g, P = _graph()
+        gp = graphistry.nodes(pl.from_pandas(pd.DataFrame(g._nodes)), "id").edges(
+            pl.from_pandas(pd.DataFrame(g._edges)), "src", "dst")
+        return gp, P, g  # g (pandas) reused for the independent oracle
+
+    def _run(self, g, cy, force_full):
+        real = gfql_unified._execute_seeded_typed_hop_fast_path
+        try:
+            if force_full:
+                gfql_unified._execute_seeded_typed_hop_fast_path = lambda *a, **k: None
+            return g.gfql(cy, engine="polars")
+        finally:
+            gfql_unified._execute_seeded_typed_hop_fast_path = real
+
+    @pytest.mark.parametrize("label", ["p:Person", "p"])
+    def test_parity_return_p(self, label):
+        gp, P, _ = self._pl_graph()
+        seed = P + 42
+        cy = f"MATCH (m:Message {{id: {seed}}})-[:HAS_CREATOR]->({label}) RETURN p"
+        fast = self._run(gp, cy, force_full=False)
+        full = self._run(gp, cy, force_full=True)
+        pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full))
+
+    def test_independent_oracle_values(self):
+        gp, P, g_pd = self._pl_graph()
+        seed = P + 42
+        oracle = _oracle_creators(g_pd, seed)
+        assert len(oracle) >= 1
+        res = self._run(gp, f"MATCH (m:Message {{id: {seed}}})-[:HAS_CREATOR]->(p:Person) RETURN p", force_full=False)
+        df = _canon_nodes(res)
+        id_col = "p.id" if "p.id" in df.columns else "id"
+        assert sorted(df[id_col].tolist()) == oracle
+
+    def test_fast_path_engages(self, monkeypatch):
+        gp, P, _ = self._pl_graph()
+        seed = P + 42
+        hits = {"n": 0}
+        real = gfql_unified._execute_seeded_typed_hop_fast_path
+
+        def spy(*a, **k):
+            r = real(*a, **k)
+            if r is not None:
+                hits["n"] += 1
+            return r
+
+        monkeypatch.setattr(gfql_unified, "_execute_seeded_typed_hop_fast_path", spy)
+        gp.gfql(f"MATCH (m:Message {{id: {seed}}})-[:HAS_CREATOR]->(p:Person) RETURN p", engine="polars")
+        assert hits["n"] >= 1
+
+    # Only shapes the full polars path can itself render (polars declines
+    # multi-entity/field whole-row RETURN by design — those aren't a parity
+    # target). Variable-length is the shape the fast path must decline (the bug
+    # this generalization inherited from the pandas path).
+    @pytest.mark.parametrize("cy_tmpl,reason", [
+        ("MATCH (m:Message {{id: {s}}})-[:HAS_CREATOR*1..2]->(p) RETURN p", "varlen range hop"),
+        ("MATCH (m:Message {{id: {s}}})-[*1..2]->(p) RETURN p", "varlen untyped hop"),
+    ])
+    def test_declines_and_stays_correct(self, cy_tmpl, reason):
+        gp, P, _ = self._pl_graph()
+        seed = P + 42
+        cy = cy_tmpl.format(s=seed)
+        real = gfql_unified._execute_seeded_typed_hop_fast_path
+        hits = {"n": 0}
+
+        def spy(*a, **k):
+            r = real(*a, **k)
+            if r is not None:
+                hits["n"] += 1
+            return r
+
+        gfql_unified._execute_seeded_typed_hop_fast_path = spy
+        try:
+            fast = gp.gfql(cy, engine="polars")
+        finally:
+            gfql_unified._execute_seeded_typed_hop_fast_path = real
+        full = self._run(gp, cy, force_full=True)
+        assert hits["n"] == 0, f"polars fast path must decline {reason}"
+        pd.testing.assert_frame_equal(_canon_nodes(fast), _canon_nodes(full))
+
+
+# ---------------------------------------------------------------------------
+# polars-specific decline gates (review-skill wave 2026-07-21, #1760)
+# ---------------------------------------------------------------------------
+
+class TestPolarsFastPathGates:
+    def _q(self, seed):
+        return f"MATCH (m {{id: {seed}}})-[{{type:'HAS_CREATOR'}}]->(p {{type:'Person'}}) RETURN p"
+
+    def test_lazyframe_declines_not_crashes(self):
+        """A LazyFrame-backed graph must decline (full path), not AttributeError."""
+        pl = pytest.importorskip("polars")
+        from graphistry.compute.chain import _seeded_typed_return_dst_polars
+        from graphistry.compute.ast import ASTNode, ASTEdge
+        g, P = _graph()
+        lazy_g = graphistry.nodes(
+            pl.from_pandas(pd.DataFrame(g._nodes)).lazy(), "id"
+        ).edges(pl.from_pandas(pd.DataFrame(g._edges)).lazy(), "src", "dst")
+        r = _seeded_typed_return_dst_polars(
+            lazy_g, n({"id": P + 1}), n({"type": "Person"}),
+            e_forward(edge_match={"type": "HAS_CREATOR"}),
+            "src", "dst", "id", "forward")
+        assert r is None
+
+    def test_mixed_engine_frames_decline(self):
+        """polars nodes + pandas edges (or vice versa) must decline at dispatch."""
+        pl = pytest.importorskip("polars")
+        g, P = _graph()
+        mixed = graphistry.nodes(
+            pl.from_pandas(pd.DataFrame(g._nodes)), "id"
+        ).edges(pd.DataFrame(g._edges), "src", "dst")
+        hits = {"n": 0}
+        real = gfql_unified._execute_seeded_typed_hop_fast_path
+
+        def spy(*a, **k):
+            r = real(*a, **k)
+            hits["n"] += 1 if r is not None else 0
+            return r
+        gfql_unified._execute_seeded_typed_hop_fast_path = spy
+        try:
+            mixed.gfql(self._q(P + 1), engine="polars")
+        except Exception:
+            pass  # full path may or may not support mixed frames; no crash IN the fast path
+        finally:
+            gfql_unified._execute_seeded_typed_hop_fast_path = real
+        assert hits["n"] == 0, "fast path must not engage on mixed-engine frames"
+
+    def test_polars_labels_column_precedence(self):
+        """label__X on a polars graph with BOTH labels+type columns: decline to the
+        full path (labels list-containment wins over type equality)."""
+        pl = pytest.importorskip("polars")
+        ndf = pd.DataFrame({
+            "id": [0, 1], "type": ["Person", "Person"],
+            "labels": [["Admin"], ["Person"]],
+        })
+        edf = pd.DataFrame({"src": [0], "dst": [1], "type": ["KNOWS"]})
+        gp = graphistry.nodes(pl.from_pandas(ndf), "id").edges(pl.from_pandas(edf), "src", "dst")
+        gpd = graphistry.nodes(ndf, "id").edges(edf, "src", "dst")
+        q = "MATCH (m {id: 0})-[{type:'KNOWS'}]->(p:Person) RETURN p"
+        got = _canon_nodes(_run_diff(gp, "polars", q, fast=True))
+        oracle = _canon_nodes(_run_diff(gpd, "pandas", q, fast=False))
+        assert got["p.id"].tolist() == oracle["p.id"].tolist()
+
+    def test_polars_null_ids_never_link(self):
+        """Null seed/node ids must not link via is_in membership on polars."""
+        pl = pytest.importorskip("polars")
+        ndf = pl.DataFrame({"id": [0, 1, None], "kind": ["person", "post", "post"]})
+        edf = pl.DataFrame({"src": [0, None], "dst": [1, 1], "type": ["T", "T"]})
+        gp = graphistry.nodes(ndf, "id").edges(edf, "src", "dst")
+        q = "MATCH (m {kind:'person'})-[{type:'T'}]->(p {kind:'post'}) RETURN p"
+        fast = _canon_nodes(_run_diff(gp, "polars", q, fast=True))
+        full = _canon_nodes(_run_diff(gp, "polars", q, fast=False))
+        pd.testing.assert_frame_equal(fast, full)
+        assert fast["p.id"].tolist() == [1]