perf(gfql): fuse residual two-star grouped-count into one lazy collect — q5/q6/q7 flip to wins (#1755)

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 <op> 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.
 

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

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))