diff --git a/CHANGELOG.md b/CHANGELOG.md index 08bcc4a7a1..5a6d651e0a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/bin/test-polars.sh b/bin/test-polars.sh index 457ce82155..036ba407e6 100755 --- a/bin/test-polars.sh +++ b/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 diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index ec26f4b4cf..33deef4e91 100644 --- a/graphistry/compute/chain.py +++ b/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], diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index c4c17f34cd..a56aa4cc3c 100644 --- a/graphistry/compute/gfql_unified.py +++ b/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,9 +1059,20 @@ 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 @@ -1068,7 +1080,11 @@ def _execute_seeded_typed_hop_fast_path( # 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) diff --git a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py index 8ff8b24174..8f4d8ba7d5 100644 --- a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py +++ b/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]