feat(gfql): add aggregate fast paths and final hardening

.github/workflows/ci.yml

@@ -1193,7 +1193,7 @@ jobs:
         cat build/gfql-coverage-audit/gfql-coverage-audit.md >> "$GITHUB_STEP_SUMMARY"
 
     - name: Upload GFQL coverage audit
-      if: ${{ matrix.python-version == '3.12' }}
+      if: ${{ always() && matrix.python-version == '3.12' }}
       uses: actions/upload-artifact@v4
       with:
         name: gfql-coverage-audit-py3.12

bin/coverage_audit.py

@@ -614,8 +614,23 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
     print(f"Wrote {json_path}")
     if exit_code:
         print(f"pytest failed with exit code {exit_code}", file=sys.stderr)
-    if any(check.status == "fail" for check in baseline_checks):
+    failing_baseline_checks = [check for check in baseline_checks if check.status == "fail"]
+    if failing_baseline_checks:
         print("per-file coverage baseline failed", file=sys.stderr)
+        for check in failing_baseline_checks:
+            actual = "missing" if check.actual_percent is None else f"{check.actual_percent:.2f}%"
+            delta = "n/a" if check.delta_percent is None else f"{check.delta_percent:+.2f}%"
+            print(
+                "  {path}: actual={actual} floor={floor:.2f}% tolerance={tolerance:.2f}% delta={delta} reason={reason}".format(
+                    path=check.path,
+                    actual=actual,
+                    floor=check.min_percent,
+                    tolerance=check.tolerance_percent,
+                    delta=delta,
+                    reason=check.reason,
+                ),
+                file=sys.stderr,
+            )
         return exit_code or 3
     return exit_code
 

bin/test-polars.sh

@@ -3,7 +3,7 @@ set -ex
 
 # Run from project root
 # - Extra args are passed through to the pytest phase
-# - Set POLARS_COV=1 to collect coverage over the graphistry package; the coverage
+# - Set POLARS_COV=1 to collect coverage over graphistry/compute; the coverage
 #   data file location is taken from $COVERAGE_FILE (as the CI py3.12 lane sets it)
 # - Non-zero exit code on fail
 
@@ -17,25 +17,18 @@ POLARS_TEST_FILES=(
     graphistry/tests/compute/gfql/test_engine_polars_hop.py
     graphistry/tests/compute/gfql/test_engine_polars_chain.py
     graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py
+    graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py
     graphistry/tests/compute/gfql/test_engine_polars_cypher_conformance.py
     graphistry/tests/compute/gfql/test_engine_polars_conformance_matrix.py
     graphistry/tests/compute/gfql/test_conformance_ledger.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
-    # Engine coercion tests: the polars paths (df_to_engine, _pl_nan_to_null identity) only
-    # run where polars is installed, and this lane is the only coverage-collecting lane
-    # with polars — the core py3.14 lane has no polars extra, so those tests skip there
-    graphistry/tests/compute/test_engine_coercion.py
 )
 
-# The whole graphistry package is measured here (not just graphistry/compute) because
-# polars-only branches outside compute/ (e.g. Engine._pl_nan_to_null) are unreachable in
-# the core coverage lane (no polars extra); coverage sources must be dirs/packages, and a
-# dotted module source (graphistry.Engine) breaks numpy under pytest
 COV_ARGS=()
 if [ -n "${POLARS_COV:-}" ]; then
-    COV_ARGS=(--cov=graphistry --cov-report=)
+    COV_ARGS=(--cov=graphistry/compute --cov-report=)
 fi
 
 python -B -m pytest -vv "${COV_ARGS[@]}" "${POLARS_TEST_FILES[@]}" "$@"
@@ -44,7 +37,7 @@ python -B -m pytest -vv "${COV_ARGS[@]}" "${POLARS_TEST_FILES[@]}" "$@"
 # appended into the same coverage data file when POLARS_COV=1 (CI audit reads it)
 COV_APPEND_ARGS=()
 if [ -n "${POLARS_COV:-}" ]; then
-    COV_APPEND_ARGS=(--cov=graphistry --cov-report= --cov-append)
+    COV_APPEND_ARGS=(--cov=graphistry/compute --cov-report= --cov-append)
 fi
 python -B -m pytest -vv "${COV_APPEND_ARGS[@]}" \
     graphistry/tests/compute/gfql/cypher/test_lowering.py -k polars

graphistry/Engine.py

@@ -214,25 +214,11 @@ def _pl_nan_to_null(df):
     polars / Arrow / cuDF input carrying genuine NaN is treated as MISSING like the pandas
     oracle (which skipna/dropna's NaN). Without this, ``engine='polars'`` on a frame with a
     real NaN keeps rows a filter/aggregation should drop (silent divergence from pandas).
-    No-op when there are no float columns.
-
-    Identity-stable: returns the *same* frame object when no float column actually carries a
-    NaN. ``fill_nan(None)`` on a NaN-free column is a value-level no-op but still yields a
-    FRESH frame; that identity churn defeats frame-identity caches keyed on ``source_ref is
-    df`` (the #1658 CSR index) -- see #1726. So we probe for real NaN presence per float
-    column (``is_nan`` only applies to float dtypes, hence the schema gate) and rebuild only
-    when -- and only the columns where -- a NaN is genuinely present. Values are identical to
-    the old unconditional rewrite (``fill_nan(None)`` never touches non-NaN entries)."""
+    No-op when there are no float columns."""
     import polars as pl
     float_cols = [c for c, dt in df.schema.items() if dt in (pl.Float32, pl.Float64)]
     if not float_cols:
         return df
-    if isinstance(df, pl.DataFrame):
-        nan_cols = [c for c in float_cols if df.get_column(c).is_nan().any()]
-        if not nan_cols:
-            return df
-        return df.with_columns([pl.col(c).fill_nan(None) for c in nan_cols])
-    # LazyFrame (rare): no cheap eager NaN probe, keep the original unconditional rewrite.
     return df.with_columns([pl.col(c).fill_nan(None) for c in float_cols])
 
 

graphistry/compute/gfql/index/__init__.py

@@ -23,9 +23,7 @@
     is_index_op, is_index_op_json,
 )
 from .cypher_ddl import parse_index_ddl, looks_like_index_ddl
-from .cost import (
-    cost_gate_frac, cost_gate_min_frontier, reset_cost_gate_frac, set_cost_gate_frac,
-)
+from .cost import cost_gate_frac, reset_cost_gate_frac, set_cost_gate_frac
 from .explain import GfqlExplainReport
 
 __all__ = [
@@ -40,7 +38,6 @@
     "CreateIndex", "DropIndex", "ShowIndexes", "IndexOp", "apply_index_op",
     "index_op_from_json", "is_index_op", "is_index_op_json",
     "parse_index_ddl", "looks_like_index_ddl",
-    "cost_gate_frac", "cost_gate_min_frontier", "reset_cost_gate_frac",
-    "set_cost_gate_frac",
+    "cost_gate_frac", "reset_cost_gate_frac", "set_cost_gate_frac",
     "GfqlExplainReport",
 ]

graphistry/compute/gfql/index/api.py

@@ -21,7 +21,7 @@
 )
 from .build import build_adjacency_index, build_node_id_index
 from .traverse import index_seeded_hop
-from .cost import cost_gate_frac, cost_gate_min_frontier, seed_deg_sum, seed_id_array
+from .cost import cost_gate_frac, seed_deg_sum, seed_id_array
 from .policy import IndexPolicy, validate_index_policy
 from .types import (
     AdjacencyIndexKind, EdgeIndexDirection, HopDirection, IndexKind,
@@ -424,15 +424,10 @@ def _bail(reason: str) -> Optional[Plottable]:
 
     # Cost gate: if the frontier covers a large fraction of distinct sources, the
     # scan path is competitive — fall back (avoids index overhead on bulk-ish hops).
-    # Small-frontier floor: a frontier of <= cost_gate_min_frontier() seeds always
-    # indexes — the frac gate scales with n_keys, so on small/low-cardinality slices
-    # it would otherwise send even a single-seed hop to the O(E) scan. Pure routing
-    # heuristic either way: index and scan return identical results.
     idx0 = cast(Optional[AdjacencyIndex], registry.get_valid(
         EDGE_OUT_ADJ if direction != "reverse" else EDGE_IN_ADJ, g._edges, (src, dst), engine
     ))
     frac = cost_gate_frac(engine)
-    min_frontier = cost_gate_min_frontier()
     if trace and idx0 is not None:
         # Free fanout estimate (Σ seed degree) from the CSR offsets — the planner
         # signal the report wants EXPLAIN to surface (not just used-index yes/no).
@@ -442,19 +437,16 @@ def _bail(reason: str) -> Optional[Plottable]:
         diag["seed_deg_sum"] = deg_sum
         diag["est_result_rows"] = deg_sum
         diag["threshold_frac"] = frac
-        diag["min_frontier_floor"] = min_frontier
     if idx0 is None:
         # required direction not resident (undirected needs both); let driver decide
         pass
     elif resolved_policy != "force":
         try:
             frontier_n = int(nodes.shape[0])
-            if (frontier_n > min_frontier
-                    and idx0.n_keys > 0 and frontier_n >= frac * idx0.n_keys):
+            if idx0.n_keys > 0 and frontier_n >= frac * idx0.n_keys:
                 return _bail(
                     f"frontier {frontier_n} >= {frac}*n_keys "
-                    f"({frac * idx0.n_keys:.0f}) and > floor ({min_frontier}) "
-                    f"-> scan cheaper"
+                    f"({frac * idx0.n_keys:.0f}) -> scan cheaper"
                 )
         except (AttributeError, TypeError, ValueError):
             pass

graphistry/compute/gfql/index/cost.py

@@ -17,46 +17,6 @@
 _COST_GATE_FRAC_DEFAULT = 0.02
 _COST_GATE_FRAC_OVERRIDES: Dict[Engine, float] = {}
 
-# Absolute small-frontier floor: at or below this many seed rows the planner NEVER
-# bails to scan on the frac gate. The frac gate scales with distinct-key cardinality,
-# so on small / low-cardinality edge slices (e.g. per-edge-type homogeneous frames:
-# n_keys <= 1/0.02 = 50 at the polars/cudf frac) even a single-node seed trips it and
-# the hop scans O(E) despite a resident O(degree) index. A frontier of <= K seeds
-# bounds CSR lookup work to K searchsorted probes plus the matching-row gather,
-# avoiding that fractional-gate artifact. The actual index/scan crossover remains
-# engine- and data-dependent: K is a conservative, environment-tunable default,
-# not a measured universal threshold. Constant (not a function of n_keys) on
-# purpose: scaling with n_keys is the frac gate's job; the floor bounds absolute
-# per-hop probe work where frac*n_keys collapses below a handful of seeds. Uniform
-# across engines (pandas's 0.5 frac only overlaps the floor when n_keys <= 2*K).
-# Purely a routing heuristic: index and scan return identical results.
-_COST_GATE_MIN_FRONTIER_DEFAULT = 16
-
-
-def cost_gate_min_frontier() -> int:
-    """Return the absolute small-frontier floor for the index-vs-scan cost gate.
-
-    Frontiers of at most this many seeds always take a resident index, regardless
-    of the frac gate. ``0`` disables the floor (pure frac gating). Env-overridable
-    via ``GFQL_INDEX_COST_GATE_MIN_FRONTIER`` for benchmark/diagnostic tuning.
-    """
-    raw = os.environ.get("GFQL_INDEX_COST_GATE_MIN_FRONTIER")
-    if raw is None or raw == "":
-        return _COST_GATE_MIN_FRONTIER_DEFAULT
-    try:
-        val = int(raw)
-    except ValueError as ex:
-        raise ValueError(
-            f"Invalid GFQL_INDEX_COST_GATE_MIN_FRONTIER={raw!r}: "
-            f"expected a non-negative integer"
-        ) from ex
-    if val < 0:
-        raise ValueError(
-            f"Invalid GFQL_INDEX_COST_GATE_MIN_FRONTIER={raw!r}: "
-            f"expected a non-negative integer"
-        )
-    return val
-
 
 def _validate_cost_gate_frac(frac: float) -> float:
     if not 0.0 < frac <= 1.0:

graphistry/compute/gfql/index/types.py

@@ -30,7 +30,6 @@ class IndexTraceStep(TypedDict, total=False):
     seed_deg_sum: Optional[int]
     est_result_rows: Optional[int]
     threshold_frac: float
-    min_frontier_floor: int
 
 
 IndexTrace = List[IndexTraceStep]

graphistry/compute/gfql/lazy/engine/polars/chain.py

@@ -361,32 +361,28 @@ def _try_native_row_op(g_cur, op):
     from .search import search_any_polars
 
     fn = getattr(op, "function", None)
-    if (
-        fn == "rows"
-        and op.params.get("binding_ops") is not None
-        and op.params.get("alias_endpoints") is None
-    ):
-        # Multi-alias bindings table (#1709): native for fixed-length connected
-        # patterns. A decline must fall through to the pre-existing correlated
-        # pattern handler below (EXISTS/searchAny); returning None here would turn
-        # those already-native shapes into an NIE.
-        bindings_result = binding_rows_polars(
-            g_cur, op.params["binding_ops"], op.params.get("attach_prop_aliases")
-        )
-        if bindings_result is not None:
-            return bindings_result
+    if fn == "rows" and op.params.get("binding_ops") is not None:
+        # #1731: single-entity boundary rows (MATCH (n) / EXISTS seeds) are handled by
+        # the pattern-apply helper; try that narrow shape first.
+        if op.params.get("source") is None:
+            out = rows_binding_ops_polars(g_cur, op.params["binding_ops"])
+            if out is not None:
+                return out
+        # #1730 gate: only take the multi-alias bindings table when alias_endpoints is
+        # absent (the alias-endpoints shape is handled elsewhere and must fall through).
+        if op.params.get("alias_endpoints") is None:
+            # Multi-alias bindings table (#1709): native for fixed-length connected
+            # patterns. A decline must fall through to the pre-existing correlated
+            # pattern handler below (EXISTS/searchAny); returning None here would turn
+            # those already-native shapes into an NIE.
+            bindings_result = binding_rows_polars(
+                g_cur, op.params["binding_ops"], op.params.get("attach_prop_aliases")
+            )
+            if bindings_result is not None:
+                return bindings_result
     if _call_native_on_polars(op):
         # frame ops (rows/limit/skip/distinct/drop_cols) — engine-polymorphic
         return op.execute(g=g_cur, prev_node_wavefront=None, target_wave_front=None, engine=Engine.POLARS)
-    # correlated pattern-existence family (EXISTS { } / pattern predicates): native
-    # via chain_polars-computed key sets; unsupported shapes return None -> honest NIE.
-    if (
-        fn == "rows"
-        and op.params.get("binding_ops") is not None
-        and op.params.get("alias_endpoints") is None
-        and op.params.get("source") is None
-    ):
-        return rows_binding_ops_polars(g_cur, op.params["binding_ops"])
     if fn == "semi_apply_mark":
         # required params are safelist-validated — direct indexing (an or-default
         # here could only mask an unvalidated call); neq is the optional one.

graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py

@@ -968,6 +968,23 @@ def _names(lf: pl.LazyFrame) -> List[str]:
                 oriented = edges_f.rename({join_col: "__from__", result_col: "__to__"})
             if payload_renames:
                 oriented = oriented.rename(payload_renames)
+
+            next_op = ops[edge_idx + 1]
+            if not isinstance(next_op, ASTNode):
+                return None
+            next_nodes = filter_by_dict_polars(nodes_lf, next_op.filter_dict)
+            next_node_ids = next_nodes.select(node_id).unique()
+            if not sem.is_multihop:
+                # Filter endpoint candidates before joining from the current state.
+                # For graph-bench q5/q6/q7, pushed Interest/City predicates make
+                # this turn an all-edges scan into a small-domain edge semi-join.
+                oriented = oriented.join(
+                    next_node_ids,
+                    left_on="__to__",
+                    right_on=node_id,
+                    how="semi",
+                )
+
             # Column collision between edge payload and accumulated state → decline
             # (pandas resolves via merge suffixes; unreferenced-by-queries either way).
             overlap = (set(_names(oriented)) - {"__from__"}) & set(_names(state))
@@ -1012,12 +1029,8 @@ def _names(lf: pl.LazyFrame) -> List[str]:
                     .rename({"__to__": "__current__"})
                 )
 
-            next_op = ops[edge_idx + 1]
-            if not isinstance(next_op, ASTNode):
-                return None
-            next_nodes = filter_by_dict_polars(nodes_lf, next_op.filter_dict)
             state = state.join(
-                next_nodes.select(node_id).unique(),
+                next_node_ids,
                 left_on="__current__",
                 right_on=node_id,
                 how="semi",
@@ -1037,7 +1050,10 @@ def _names(lf: pl.LazyFrame) -> List[str]:
                 continue  # properties unreferenced — keep only the bare id column
             lookup_src = alias_frames[alias]
             lookup = lookup_src.select(
-                [pl.col(node_id), pl.col(node_id).alias(f"{alias}.{node_id}")]
+                [
+                    pl.col(node_id),
+                    pl.col(node_id).alias(f"{alias}.{node_id}"),
+                ]
                 + [
                     pl.col(col).alias(f"{alias}.{col}")
                     for col in _names(lookup_src)

graphistry/compute/gfql/row/frame_ops.py

@@ -27,6 +27,7 @@ def _gfql_binding_ops_row_table(
             self,
             binding_ops: List[Dict[str, JSONVal]],
             alias_prefilters: Optional[AliasPrefilters] = None,
+            attach_prop_aliases: Optional[List[str]] = None,
         ) -> "Plottable": ...
         def _gfql_bindings_row_table(self, alias_endpoints: Any) -> "Plottable": ...
 
@@ -148,7 +149,7 @@ def coerce_non_negative_int(value: Any, op_name: str) -> int:
 
 
 def rows(
-    ctx: Any,
+    ctx: RowPipelineCtx,
     table: Optional[str] = None,
     source: Optional[str] = None,
     alias_endpoints: Optional[Dict[str, str]] = None,

graphistry/compute/gfql/row/prefilter.py

@@ -27,24 +27,10 @@ def _is_alias_prefilter_spec(value: object) -> bool:
         return False
     kind = value.get("kind")
     if kind == "expr":
-        return (
-            set(value).issubset({"kind", "text"})
-            and isinstance(value.get("text"), str)
-        )
+        return isinstance(value.get("text"), str)
     if kind == "search_any":
-        if not set(value).issubset(
-            {"kind", "term", "case_sensitive", "regex", "columns"}
-        ):
-            return False
-        if not isinstance(value.get("term"), str):
-            return False
-        if "case_sensitive" in value and not isinstance(value["case_sensitive"], bool):
-            return False
-        if "regex" in value and not isinstance(value["regex"], bool):
-            return False
-        if "columns" in value and not _is_string_list(value["columns"]):
-            return False
-        return True
+        columns = value.get("columns")
+        return isinstance(value.get("term"), str) and (columns is None or _is_string_list(columns))
     return False