diff --git a/CHANGELOG.md b/CHANGELOG.md index bdb8034c9a..0e49d8aeb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,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 two-star grouped-count OLAP queries collect once on polars (#1755)**: profiling showed the residual two-star path (graph-benchmark q5-q7 shapes) spent ~72% of its runtime on the fixed per-op cost of ~27 eager polars operations. When every residual translates natively (see the entry below), the whole plan — base filter_dicts, residual filters, typed-edge filters, semi-joins, group-prop lookup — now composes as ONE `pl.LazyFrame` plan collected once at the join (`filter_by_dict_polars` gains a `filter_expr_by_dict_polars` twin: same column/dtype resolution and typed-error/NIE contract, expression only). Value-identical to the eager lane it replaces (same filters/joins/aggregation), INCLUDING the empty-match boundary: the eager all-left-counts==1 shortcut's single `n=0` row (the openCypher count over no rows) is reproduced — the left-counts frame rides the same single collect. Edges are engine-converted before going lazy, so pandas frames under `engine='polars'` (the WITH..MATCH reentry shape) take the lane instead of crashing; LazyFrame inputs and untranslatable residuals decline to the eager path. Measured on dgx vs warm Kuzu, same session: q5 7.7→3.8ms (flips to a GFQL win vs 5.2), q6 9.8→6.1 (flips vs 7.3), q7 8.7→5.7 (flips vs 6.9) — moving the graph-benchmark scoreboard from 4W/2T/3L to 7W/1T/1L. Differential tests spy the extracted `_connected_join_two_star_fused_polars` helper and ASSERT lane engagement (a `count(*)` query declines the whole two-star path — pinned as a decline shape): fused vs forced-eager exact-row parity (ORDER BY pinned), the ungrouped empty-match n=0 row, pandas-frames-under-polars-engine parity, empty-result shape, pandas oracle. - **GFQL connected-join simple residuals filter natively on polars (#1729/#1755)**: on the polars engines, a connected-join node residual of the #1729 scalar shapes — case-insensitive equality ``(tolower(a.col) = tolower('lit'))`` and scalar comparisons ``(a.col literal)`` for ``= >= <= > <`` — previously dispatched a full sub-``chain()`` per alias (the polars row pipeline has no native ``where_rows``), costing several ms per residual on OLAP group-aggregate queries. `_residual_polars_expr` now translates these shapes to native ``pl.Expr`` filters applied directly to the alias frame (graph-benchmark q5 10.2→6.6ms, q6 10.7→8.9ms on dgx). Null semantics match the ``where_rows`` evaluator (null comparisons drop the row); any unrecognized shape, alias mismatch, or absent column falls back to the existing chain path — and if *any* expr in a residual group fails to translate, the whole group falls back (never a partial mix). Byte-parity verified across an 8-case adversarial differential (nulls, unicode/casefold, numeric ranges, mixed residuals) plus the full suite. - **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. diff --git a/graphistry/compute/gfql/lazy/engine/polars/predicates.py b/graphistry/compute/gfql/lazy/engine/polars/predicates.py index 5bad911bd8..5c758a5df8 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/predicates.py +++ b/graphistry/compute/gfql/lazy/engine/polars/predicates.py @@ -309,10 +309,22 @@ def _mismatch(v: Any) -> bool: def filter_by_dict_polars(df: "pl.DataFrame", filter_dict: "Optional[Dict[str, Any]]") -> "pl.DataFrame": """Return rows of polars ``df`` matching all entries in ``filter_dict`` via one filter.""" + combined = filter_expr_by_dict_polars(df, filter_dict) + if combined is None: + return df + return df.filter(combined) + + +def filter_expr_by_dict_polars(df: "pl.DataFrame", filter_dict: "Optional[Dict[str, Any]]") -> "Optional[pl.Expr]": + """Build the combined boolean ``pl.Expr`` filter_by_dict_polars would apply, or None + for an empty/absent filter dict. ``df`` supplies the schema for column/dtype + resolution only — callers may apply the expr to a LazyFrame over the same schema + (the fused connected-join lane), with identical semantics incl. the same typed + error/NIE contract for unsupported shapes.""" import polars as pl if not filter_dict: - return df + return None exprs: "List[pl.Expr]" = [] for col, val in filter_dict.items(): @@ -373,8 +385,8 @@ def filter_by_dict_polars(df: "pl.DataFrame", filter_dict: "Optional[Dict[str, A exprs.append(pl.col(resolved_col) == resolved_val) if not exprs: - return df + return None combined = exprs[0] for e in exprs[1:]: combined = combined & e - return df.filter(combined) + return combined diff --git a/graphistry/compute/gfql_fast_paths.py b/graphistry/compute/gfql_fast_paths.py index 1cb27fbc62..2632a9c72a 100644 --- a/graphistry/compute/gfql_fast_paths.py +++ b/graphistry/compute/gfql_fast_paths.py @@ -767,6 +767,141 @@ def _connected_join_filter_node_frames_by_residuals( return out + +def _connected_join_two_star_fused_polars( + nodes: DataFrameT, + edges: DataFrameT, + *, + node_col: str, + src_col: str, + dst_col: str, + residual_map: Dict[str, List[Any]], + shared_alias: str, + first_end_alias: str, + second_end_alias: str, + first_start_fd: Optional[dict], + second_start_fd: Optional[dict], + first_end_fd: Optional[dict], + second_end_fd: Optional[dict], + first_edge_match: Optional[dict], + second_edge_match: Optional[dict], + group_prop_refs: List[Tuple[str, str]], + output_group_keys: List[str], + agg_alias: str, + order_keys: List[Tuple[str, bool]], + limit_value: Optional[int], + select_items: Optional[List[Tuple[str, str]]], +) -> Optional[DataFrameT]: + """FUSED lazy lane (#1755 lane-1): the whole two-star grouped-count as ONE lazy + plan, collected once at the join (the eager path pays a fixed collect cost per + op; ~27 collects/exec dominated q5-q7 profiles). Value-identical to the eager + lane -- same filters/semi-joins/aggregation, and the empty-match boundary + reproduces the eager all-left-counts==1 shortcut's single n=0 row (openCypher + count over no rows). Returns None to decline (untranslatable residual, missing + group property) so the caller falls through to the eager path. Both frames must + already be engine-converted polars frames. + """ + import polars as pl + from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_expr_by_dict_polars + + if isinstance(nodes, pl.LazyFrame) or isinstance(edges, pl.LazyFrame): + return None # eager lane owns LazyFrame inputs (schema probes on a LazyFrame warn/cost) + frame_aliases = {shared_alias, first_end_alias, second_end_alias} + node_schema = dict(nodes.schema) + residual_exprs: Dict[str, List["pl.Expr"]] = {} + for r_alias, r_exprs in residual_map.items(): + if r_alias not in frame_aliases: + continue # mirror the eager path: residuals for unbound aliases are not applied here + r_translated = [_residual_polars_expr(e, r_alias, node_schema) for e in r_exprs] + if any(t is None for t in r_translated): + return None + residual_exprs[r_alias] = [t for t in r_translated if t is not None] + for _, prop in group_prop_refs: + if prop not in nodes.columns: + return None + lf_nodes = nodes.lazy() + lf_edges = edges.lazy() + + def _alias_nodes_lf(fds: List[Optional[dict]], r_alias: str) -> "pl.LazyFrame": + lf = lf_nodes + for fd in fds: + fe = filter_expr_by_dict_polars(nodes, fd) + if fe is not None: + lf = lf.filter(fe) + for rexpr in residual_exprs.get(r_alias, []): + lf = lf.filter(rexpr) + return lf + + shared_lf = _alias_nodes_lf([first_start_fd, second_start_fd], shared_alias) + second_leaf_lf = _alias_nodes_lf([second_end_fd], second_end_alias) + shared_ids_lf = shared_lf.select(node_col).unique() + first_leaf_ids_lf = _alias_nodes_lf([first_end_fd], first_end_alias).select(node_col).unique() + second_leaf_ids_lf = second_leaf_lf.select(node_col).unique() + fe1 = filter_expr_by_dict_polars(edges, first_edge_match) + fe2 = filter_expr_by_dict_polars(edges, second_edge_match) + first_edges_lf = lf_edges.filter(fe1) if fe1 is not None else lf_edges + second_edges_lf = lf_edges.filter(fe2) if fe2 is not None else lf_edges + left_counts_lf = ( + first_edges_lf + .join(shared_ids_lf, left_on=src_col, right_on=node_col, how="semi") + .join(first_leaf_ids_lf, left_on=dst_col, right_on=node_col, how="semi") + .group_by(src_col) + .len("__left_count__") + .rename({src_col: shared_alias}) + ) + right_base_lf = ( + second_edges_lf + .join(shared_ids_lf, left_on=src_col, right_on=node_col, how="semi") + .join(second_leaf_ids_lf, left_on=dst_col, right_on=node_col, how="semi") + ) + if group_prop_refs: + fused_lookup_key = "__gfql_fast_second_leaf_id__" + lookup_lf = second_leaf_lf.select( + [pl.col(node_col).alias(fused_lookup_key)] + + [pl.col(prop).alias(out_col) for out_col, prop in group_prop_refs] + ).unique(subset=[fused_lookup_key]) + right_base_lf = right_base_lf.join(lookup_lf, left_on=dst_col, right_on=fused_lookup_key, how="inner") + right_rows_lf = right_base_lf.select( + [pl.col(src_col).alias(shared_alias)] + [pl.col(key) for key in output_group_keys] + ) + joined_lf = right_rows_lf.join(left_counts_lf, on=shared_alias, how="inner") + # HOT PATH: one collect. left_counts is collected ONLY on the empty-match + # boundary below (collect_all of both plans measured +2.5ms/query on the + # 20k graphbench q5-q7 -- CSE does not absorb the left-arm recompute). + joined = joined_lf.collect() + if len(joined) == 0: + # Eager-lane parity on the empty match: the eager all-left-counts==1 + # shortcut counts matched rows with pl.len(), emitting a single n=0 row + # when the first arm is live but nothing joins (the openCypher-correct + # count over zero rows). Every other empty shape returns the 0x0 frame, + # exactly like the eager generic branch. + left_counts_df = left_counts_lf.collect() + if ( + not output_group_keys + and len(left_counts_df) > 0 + and bool(left_counts_df.select((pl.col("__left_count__") == 1).all()).item()) + ): + out_df = pl.DataFrame({agg_alias: [0]}).with_columns(pl.col(agg_alias).cast(pl.Int64)) + else: + return cast(DataFrameT, joined.select([])) + elif output_group_keys: + out_df = joined.group_by(output_group_keys, maintain_order=True).agg( + pl.col("__left_count__").sum().cast(pl.Int64).alias(agg_alias)) + else: + out_df = joined.select(pl.col("__left_count__").sum().cast(pl.Int64).alias(agg_alias)) + if order_keys: + out_df = out_df.sort( + [key for key, _ in order_keys], + descending=[desc for _, desc in order_keys], + nulls_last=[not desc for _, desc in order_keys], + ) + if limit_value is not None: + out_df = out_df.head(limit_value) + if select_items is not None: + out_df = out_df.select([pl.col(s_col).alias(d_col) for s_col, d_col in select_items]) + return cast(DataFrameT, out_df) + + def _connected_join_two_star_fast_grouped_count( base_graph: Plottable, plan: ConnectedMatchJoinPlan, @@ -969,6 +1104,37 @@ def _connected_join_two_star_fast_grouped_count( if shared_ids is None or first_leaf_ids is None or second_leaf_ids is None: return None else: + # FUSED lazy lane (#1755 lane-1): one lazy plan, one collect at the join. + # Returns None to fall through to the eager path (untranslatable residual, + # missing group property). Edges are engine-converted HERE because the + # fused lane runs native polars ops directly on them (the eager path's + # cached edge filter does its own conversion) -- pandas edges with + # engine='polars' (incl. WITH..MATCH reentry frames) crash otherwise. + fused_out = _connected_join_two_star_fused_polars( + nodes, + df_to_engine(edges, engine), + node_col=node_col, + src_col=src_col, + dst_col=dst_col, + residual_map=residual_map, + shared_alias=shared_alias, + first_end_alias=first_end_alias, + second_end_alias=second_end_alias, + first_start_fd=cast(Optional[dict], first_start.filter_dict), + second_start_fd=cast(Optional[dict], second_start.filter_dict), + first_end_fd=cast(Optional[dict], first_end.filter_dict), + second_end_fd=cast(Optional[dict], second_end.filter_dict), + first_edge_match=cast(Optional[dict], first_edge.edge_match), + second_edge_match=cast(Optional[dict], second_edge.edge_match), + group_prop_refs=group_prop_refs, + output_group_keys=output_group_keys, + agg_alias=agg_alias, + order_keys=order_keys, + limit_value=limit_value, + select_items=select_items, + ) + if fused_out is not None: + return fused_out shared_nodes = filter_by_dict_polars(nodes, cast(Optional[dict], first_start.filter_dict)) shared_nodes = filter_by_dict_polars(shared_nodes, cast(Optional[dict], second_start.filter_dict)) first_leaf_nodes = filter_by_dict_polars(nodes, cast(Optional[dict], first_end.filter_dict)) diff --git a/graphistry/tests/compute/gfql/test_residual_polars_native.py b/graphistry/tests/compute/gfql/test_residual_polars_native.py index 3321ad2d6d..3b0c9a63ac 100644 --- a/graphistry/tests/compute/gfql/test_residual_polars_native.py +++ b/graphistry/tests/compute/gfql/test_residual_polars_native.py @@ -216,3 +216,135 @@ def test_dtype_mismatch_group_reaches_designed_error(self): with pytest.raises(NotImplementedError): fp._connected_join_apply_node_residuals( g, nodes, "a", ["(a.name >= 25)"], "node_id", engine=Engine.POLARS) + + +class TestFusedTwoStarLane: + """#1755 lane-1: the fused single-collect two-star plan must be value-identical + to the eager path (which it replaces when residuals translate natively). + Every fused-arm test ASSERTS lane engagement via a spy on the extracted + _connected_join_two_star_fused_polars helper -- the original tests silently + compared slow-path vs slow-path because count(*) lowers to a 2-tuple agg that + declines the whole two-star fast path before either lane.""" + + def _star_graph(self): + pl2 = pytest.importorskip("polars") + ndf = pl2.DataFrame({ + "node_id": list(range(1, 11)), + "node_type": ["Person"] * 4 + ["Interest"] * 3 + ["City"] * 3, + "interest": [None] * 4 + ["Fine Dining", "fine dining", "tennis"] + [None] * 3, + "city": [None] * 7 + ["London", "london", "Paris"], + "gender": ["male", "female", "male", "female"] + [None] * 6, + }) + edf = pl2.DataFrame({ + "src": [1, 1, 2, 2, 3, 4, 1, 2, 3, 4], + "dst": [5, 6, 5, 7, 6, 5, 8, 8, 9, 10], + "rel": ["HAS_INTEREST"] * 6 + ["LIVES_IN"] * 4, + }) + return graphistry.nodes(ndf, "node_id").edges(edf, "src", "dst") + + # count(p) -- count(*) lowers to a 2-tuple agg and declines the two-star fast + # path entirely (pinned below), so it can never reach the fused lane. + Q = ("MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('FINE DINING') AND p.gender = 'male' " + "RETURN c.city AS city, count(p) AS n ORDER BY n DESC, city LIMIT 5") + + def _spy_fused(self, monkeypatch): + calls = [] + orig = fp._connected_join_two_star_fused_polars + + def spy(*a, **k): + out = orig(*a, **k) + calls.append(out is not None) + return out + + monkeypatch.setattr(fp, "_connected_join_two_star_fused_polars", spy) + return calls + + @staticmethod + def _rows(res): + df = res._nodes + df = df.to_pandas() if hasattr(df, "to_pandas") else df + return df.to_dict("records") + + @requires_polars + def test_fused_matches_eager_chain_path(self, monkeypatch): + g = self._star_graph() + calls = self._spy_fused(monkeypatch) + fused = g.gfql(self.Q, engine="polars") + assert calls and calls[-1], "fused lane did not engage (vacuous comparison)" + # forcing every translation to decline disables the fused lane AND the + # residual fast lane -> full eager path + where_rows chain fallback + monkeypatch.setattr(fp, "_residual_polars_expr", lambda *a, **k: None) + eager = g.gfql(self.Q, engine="polars") + assert self._rows(fused) == self._rows(eager) + assert self._rows(fused) # non-empty: ORDER BY pinned, exact row order compared + + @requires_polars + def test_fused_empty_result(self, monkeypatch): + g = self._star_graph() + q = self.Q.replace("FINE DINING", "no such interest") + calls = self._spy_fused(monkeypatch) + fused = g.gfql(q, engine="polars") + assert calls and calls[-1], "fused lane did not engage" + monkeypatch.setattr(fp, "_residual_polars_expr", lambda *a, **k: None) + eager = g.gfql(q, engine="polars") + + def shape(res): + df = res._nodes + df = df.to_pandas() if hasattr(df, "to_pandas") else df + return (len(df), sorted(map(str, df.columns))) + assert shape(fused) == shape(eager) + + @requires_polars + def test_fused_matches_pandas_oracle(self, monkeypatch): + g = self._star_graph() + gpd = graphistry.nodes(g._nodes.to_pandas(), "node_id").edges(g._edges.to_pandas(), "src", "dst") + calls = self._spy_fused(monkeypatch) + got = g.gfql(self.Q, engine="polars")._nodes + assert calls and calls[-1], "fused lane did not engage" + got = (got.to_pandas() if hasattr(got, "to_pandas") else got).to_dict("records") + oracle = gpd.gfql(self.Q, engine="pandas")._nodes.to_dict("records") + assert got == oracle + + @requires_polars + def test_pandas_frames_polars_engine_no_crash(self, monkeypatch): + """BLOCKER-1 pin: pandas frames + engine='polars' (the WITH..MATCH reentry + shape) must run the residual two-star query, not AttributeError on + edges.lazy() -- the fused lane converts edges before going lazy.""" + g = self._star_graph() + gpd = graphistry.nodes(g._nodes.to_pandas(), "node_id").edges(g._edges.to_pandas(), "src", "dst") + res = gpd.gfql(self.Q, engine="polars") + assert self._rows(res) == self._rows(g.gfql(self.Q, engine="polars")) + + @requires_polars + def test_fused_ungrouped_empty_match_returns_zero_row(self, monkeypatch): + """BLOCKER-2 pin: ungrouped count with a live first arm but empty join must + return the single n=0 row (the eager all-left-counts==1 shortcut / the + openCypher count over no rows), not a 0x0 frame.""" + g = self._star_graph() + # tennis -> only person 2, one HAS_INTEREST edge (left counts all == 1, non-empty); + # NoSuchCity -> right arm empty -> empty join + q = ("MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('TENNIS') AND c.city = 'NoSuchCity' " + "RETURN count(p) AS n") + calls = self._spy_fused(monkeypatch) + fused = g.gfql(q, engine="polars") + assert calls and calls[-1], "fused lane did not engage" + assert self._rows(fused) == [{"n": 0}] + monkeypatch.setattr(fp, "_residual_polars_expr", lambda *a, **k: None) + eager = g.gfql(q, engine="polars") + assert self._rows(fused) == self._rows(eager) + + @requires_polars + def test_count_star_declines_two_star_fast_path(self, monkeypatch): + """Decline-shape pin: count(*) lowers to a 2-tuple agg, so the two-star fast + path (fused AND eager) declines and the general path answers -- and the + fused lane must NOT engage.""" + g = self._star_graph() + q = self.Q.replace("count(p)", "count(*)") + calls = self._spy_fused(monkeypatch) + res = g.gfql(q, engine="polars") + assert not any(calls), "count(*) unexpectedly reached the fused lane" + assert self._rows(res) # still answered (general path)