feat(gfql): count(*) short-circuit — O(1) count_table op, no full-frame materialize

A lone `RETURN count(*)` over a single node/edge pattern used to materialize
the whole matched frame and run a constant-key group_by just to count rows
(the ~770x-vs-Ladybug loss in the fair Cypher benchmark: count = 2543-8206ms
while the df-path count is ~0.01ms). The Cypher lowering now emits a new
`count_table` row op that reads the scanned table's height directly (or sums
the boolean alias-mask column when the pattern filters) in a single reduction
— no frame copy, no group_by.

Guarded to the provably-equivalent shapes only: exactly one non-DISTINCT
count(*), no group keys / post-agg exprs / row-level WHERE / UNWIND / paging /
multi-relationship binding, and either a pure node scan (relationship_count==0)
or a single relationship counted on its edge alias (== the cases
_reject_unsound_relationship_multiplicity_aggregates permits). Every other
shape falls through to the general aggregate path unchanged. The op replaces
the rows()+with_+group_by prefix and keeps the identical trailing projection,
so downstream/result is byte-identical.

count_table is a native frame op (like rows/limit) routed via
_POLARS_NATIVE_ROW_PIPELINE_CALLS -> op.execute -> frame_ops.count_table, so
one implementation covers pandas / cuDF / polars / polars-gpu.

Validation: CPU parity (pandas+polars node/edge count == oracle, fast path
confirmed taken); full test_lowering suite 1394 passed; coverage ledger green
(count_table accounted in _rowop_exercised + _ROW_OP_CASES + CALL_KNOWN_UNCOVERED);
dgx 4-engine conformance (count_all_nodes/edges + count_table rowop, chain+dag)
55 passed on pandas/cuDF/polars/polars-gpu; ruff + mypy clean. The 5M/20M
Ladybug head-to-head benchmark is the next step (needs the ladybug bench harness
rebased onto this).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: lmeyerov

CHANGELOG.md

@@ -18,6 +18,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Added
 - **GFQL native Polars engine — more cypher row coverage (`toFloat`, `collect`/`collect(DISTINCT)`, `WHERE … IN`)**: three surfaces that previously raised `NotImplementedError` on `engine='polars'` now run natively, parity-validated vs the pandas oracle across all four engines (and honest-NIE where pandas can't be matched). **`toFloat(x)`** lowers int/uint/bool/float → `Float64` (NaN preserved — float64 has no separate null sentinel, unlike `toInteger`); a non-numeric String declines (NIE) because pandas `astype(float)` *raises* rather than null-on-failure. **`collect(x)` / `collect(DISTINCT x)`** aggregations complete the native `group_by` surface (every other agg was already native): drop nulls, preserve within-group first-occurrence order (`collect` keeps dups; `DISTINCT` dedups keep-first), all-null group → `[]`. **`where_rows`/`WHERE … IN [list]`** membership lowers to `is_in` (a null cell is excluded per openCypher 3VL). No change to any already-native path.
 - **GFQL Polars-CPU streaming collect (opt-in, large traversals)**: `GFQL_POLARS_CPU_STREAMING=1` runs the polars-CPU lazy collects (`hop`/`chain`) on the polars **streaming** executor instead of the default in-memory collect. Benchmarked ~1.04–1.11× faster on big multi-hop traversals (10M nodes / 80M edges: 20.0→18.0 s) and parity-identical, but ~0.86× (slower) on small/interactive sizes (streaming overhead) — so it is **opt-in, default off** (no change to default behavior). Use for large batch traversals where CPU is the target.
+- **GFQL Cypher `count(*)` short-circuit — O(1) instead of O(N) materialize**: a lone `RETURN count(*)` over a single node or edge pattern (`MATCH (n) RETURN count(*)`, `MATCH ()-[r]->() RETURN count(*)`) previously materialized the entire matched frame and ran a constant-key `group_by` just to count its rows. The lowering now emits a new `count_table` row op that reads the scanned table's height directly (or, when the pattern applies a filter, sums the boolean alias-mask column) in a single reduction — no full-frame copy, no group_by. Applies only to the provably-equivalent shapes: exactly one non-DISTINCT `count(*)`, no group keys / post-aggregate exprs / row-level `WHERE` / `UNWIND` / paging / multi-relationship binding, and either a pure node scan (`relationship_count == 0`) or a single relationship counted on its edge alias — every other shape (node-alias-over-relationship, multi-hop paths, `count(DISTINCT …)`, grouped counts) falls through to the general aggregate path unchanged. Engine-polymorphic across pandas/cuDF/polars/polars-gpu; differential parity verified on all four engines (`count_all_nodes`/`count_all_edges` cypher conformance cases + a `count_table` row-op subject case, chain and DAG surfaces). No change to any result value — only the execution path.
 
 ### Fixed
 - **GFQL `ne()` / `<>` on NULL now follows openCypher/SQL 3-valued logic (pandas)**: `n({"col": ne(x)})` and cypher `WHERE n.col <> x` over a NULL/NA cell used to KEEP the null row on the pandas engine (`NaN != x` → True), diverging from cuDF and the polars engine (both drop it) — and even from pandas' own `WHERE NOT n.col = x` path. Per openCypher/SQL three-valued logic, `null <> x` is `null` (an unknown value cannot be proven unequal to `x`), so a null cell is **not** a match and the row is excluded — consistent with `eq`/`gt`/`lt`/`IN` (which already dropped nulls). Fixed the `NE` predicate to mask out nulls; this corrects both the `filter_dict` predicate path and the single-entity cypher `<>` WHERE path on pandas. cuDF/polars/polars-gpu were already conformant. Verified across all four engines (`ne`, `<>`, `NOT =`, `NOT IN` all drop the null). Note: this is a behavior change for `ne()` on nullable columns under the default pandas engine. (Broader openCypher null-semantics alignment + docs tracked in #1664.)

graphistry/compute/ast.py

@@ -1751,6 +1751,28 @@ def drop_cols(cols: Iterable[str]) -> ASTCall:
     return ASTCall("drop_cols", {"cols": list(cols)})
 
 
+def count_table(
+    table: str = "nodes",
+    source: Optional[str] = None,
+    alias: str = "count(*)",
+) -> ASTCall:
+    """Count matched rows without materializing them (fast path for a lone ``count(*)``).
+
+    Emitted by the Cypher lowering when a RETURN is exactly ``count(*)`` over a
+    single node/edge pattern (no DISTINCT, GROUP BY, row-level WHERE, UNWIND,
+    paging, or multi-relationship binding). Produces a one-row table
+    ``{alias: n}`` where ``n`` is the height of the active ``table`` (or, when a
+    ``source`` alias-mask column is present, the count of its truthy rows). This
+    avoids the full-frame materialize + constant-key ``group_by`` the general
+    aggregate path performs — the win that turns count(*) from O(N) into a single
+    reduction. See plans/gfql-engine-followups (BEAT LADYBUG).
+    """
+    params: Dict[str, Any] = {"table": table, "alias": alias}
+    if source is not None:
+        params["source"] = source
+    return ASTCall("count_table", params)
+
+
 def group_by(
     keys: Iterable[str],
     aggregations: Iterable[Sequence[Any]],

graphistry/compute/gfql/call/validation.py

@@ -491,6 +491,19 @@ def _semi_apply_mark_added_node_cols(params: Dict[str, object]) -> Set[str]:
         description='Drop duplicate rows from active row table',
     ),
 
+    'count_table': _safelist_entry(
+        {'table', 'source', 'alias'},
+        param_validators={
+            'table': lambda v: v in ['nodes', 'edges'],
+            'source': is_string_or_none,
+            'alias': is_non_empty_string,
+        },
+        description='Count matched rows (node/edge table height or source-alias mask) into a one-row table, without materializing the frame — fast path for a lone count(*)',
+        schema_effects=_schema_effects(
+            adds_node_cols=lambda p: [p.get('alias', 'count(*)')],
+        ),
+    ),
+
     'get_degrees': _safelist_entry(
         {'col', 'degree_in', 'degree_out', 'engine'},
         param_validators={

graphistry/compute/gfql/cypher/lowering.py

@@ -13,6 +13,7 @@
     ASTObject,
     ASTNode,
     anti_semi_apply,
+    count_table,
     distinct,
     drop_cols,
     e_forward,
@@ -2308,6 +2309,60 @@ def _match_relationship_count(clause: MatchClause) -> int:
     return sum(1 for element in _match_pattern_elements(clause) if isinstance(element, RelationshipPattern))
 
 
+def _is_pure_count_star_shortcircuit(
+    *,
+    aggregate_specs: Sequence[_AggregateSpec],
+    pre_items: Sequence[Tuple[str, Any]],
+    row_steps: Sequence[ASTObject],
+    query: CypherQuery,
+    binding_row_aliases: AbstractSet[str],
+    relationship_count: int,
+    active_match_alias: Optional[str],
+    alias_targets: Mapping[str, ASTObject],
+) -> bool:
+    """True when a RETURN is exactly ``count(*)`` over a single node/edge scan.
+
+    Guards the count_table fast path (skip the full-frame materialize + constant-
+    key group_by): the count then equals the height (or source-mask sum) of the
+    active table. Requires a lone non-DISTINCT ``count(*)`` with no group keys,
+    post-aggregate exprs, row-level WHERE, UNWIND, or multi-relationship binding,
+    and a plain ``rows(table=nodes|edges[, source])`` as the only prior step.
+    Sound only for a pure node scan (``relationship_count == 0``) or a single
+    relationship counted on its edge alias (``relationship_count == 1`` with an
+    ``ASTEdge`` active alias) — exactly the cases
+    ``_reject_unsound_relationship_multiplicity_aggregates`` permits; any other
+    shape (node-alias-over-relationship, multi-hop paths) falls through to the
+    general aggregate path (which counts bindings or rejects as unsound).
+    """
+    if len(aggregate_specs) != 1:
+        return False
+    agg = aggregate_specs[0]
+    if agg.func != "count" or agg.expr_text is not None or agg.distinct:
+        return False
+    if pre_items or binding_row_aliases or query.unwinds:
+        return False
+    # Exactly the initial rows() step: any row-level WHERE or UNWIND would have
+    # appended further steps, so len == 1 proves the count is over the raw scan.
+    if len(row_steps) != 1:
+        return False
+    base = row_steps[0]
+    if not (isinstance(base, ASTCall) and base.function == "rows"):
+        return False
+    if base.params.get("table") not in ("nodes", "edges"):
+        return False
+    if base.params.get("binding_ops") is not None or base.params.get("alias_endpoints") is not None:
+        return False
+    if relationship_count == 0:
+        return True
+    if (
+        relationship_count == 1
+        and active_match_alias is not None
+        and isinstance(alias_targets.get(active_match_alias), ASTEdge)
+    ):
+        return True
+    return False
+
+
 def _reject_unsound_relationship_multiplicity_aggregates_common(
     *,
     aggregate_specs: Sequence[_AggregateSpec],
@@ -6573,6 +6628,35 @@ def _lower_general_row_projection(
             if len(pre_items) > 0:
                 row_steps.append(with_(pre_items, extend=bindings_row_path))
             row_steps.append(group_by(key_names, aggregations, key_prefixes=alias_key_prefixes))
+        elif _is_pure_count_star_shortcircuit(
+            aggregate_specs=aggregate_specs,
+            pre_items=pre_items,
+            row_steps=row_steps,
+            query=query,
+            binding_row_aliases=binding_row_aliases,
+            relationship_count=relationship_count,
+            active_match_alias=active_match_alias,
+            alias_targets=alias_targets,
+        ):
+            # Fast path: count_table reads the scanned table's height (or the
+            # source-alias mask sum) with one reduction — no full-frame
+            # materialize + constant-key group_by. It replaces the sole rows()
+            # step and produces the same ``count(*)`` column the group_by would,
+            # so the trailing identity projection (elif not key_names, below) is
+            # unchanged. empty_result_row is belt-and-suspenders here (count_table
+            # always emits a 1-row result), kept for parity with the group_by path.
+            base_rows = cast(ASTCall, row_steps[0])
+            count_alias = aggregate_specs[0].output_name
+            row_steps = [
+                count_table(
+                    table=cast(str, base_rows.params.get("table", "nodes")),
+                    source=cast(Optional[str], base_rows.params.get("source")),
+                    alias=count_alias,
+                )
+            ]
+            available_columns = {count_alias}
+            empty_aggregate_row = _empty_aggregate_row(aggregate_specs)
+            empty_result_row = empty_aggregate_row
         else:
             global_key = _fresh_temp_name(temp_names, "__cypher_group__")
             row_steps.append(with_([(global_key, 1)] + pre_items))

graphistry/compute/gfql/row/frame_ops.py

@@ -167,6 +167,64 @@ def rows(
     return row_table(ctx, table_df)
 
 
+def count_table(
+    ctx: Any,
+    table: str = "nodes",
+    source: Optional[str] = None,
+    alias: str = "count(*)",
+) -> "Plottable":
+    """Count matched rows and set a one-row ``{alias: n}`` result table.
+
+    Fast path for a lone ``count(*)``: reads the height of the active node/edge
+    table (or the truthy count of the ``source`` alias-mask column) with a single
+    reduction, never materializing/copying the whole frame the way ``rows`` +
+    ``group_by`` would. Engine-polymorphic across pandas/cuDF/polars (eager or
+    lazy). See ``graphistry.compute.ast.count_table`` and the Cypher lowering
+    short-circuit.
+    """
+    if table not in {"nodes", "edges"}:
+        raise ValueError(
+            f"count_table(table=...) must be one of 'nodes' or 'edges', got {table!r}"
+        )
+    table_df = ctx._nodes if table == "nodes" else ctx._edges
+
+    if table_df is None:
+        return row_table(ctx, pd.DataFrame({alias: [0]}))
+
+    if _is_polars(table_df):
+        import polars as pl
+        if source is not None:
+            # LazyFrame lacks .columns without a resolve; collect_schema is lazy-safe.
+            cols = table_df.collect_schema().names()
+            if source not in cols:
+                raise ValueError(
+                    f"count_table(source=...) alias column not found: {source!r}"
+                )
+            count_expr = pl.col(source).fill_null(False).cast(pl.Boolean).sum()
+        else:
+            count_expr = pl.len()
+        res = table_df.select(count_expr.alias(alias))
+        # eager DataFrame.select -> DataFrame (no collect); LazyFrame.select -> LazyFrame.
+        if hasattr(res, "collect"):
+            res = res.collect()
+        n = int(res.item())
+        return row_table(ctx, pl.DataFrame({alias: [n]}))
+
+    # pandas / cuDF (API-compatible)
+    if source is not None:
+        if source not in table_df.columns:
+            raise ValueError(
+                f"count_table(source=...) alias column not found: {source!r}"
+            )
+        mask = table_df[source]
+        if hasattr(mask, "fillna"):
+            mask = mask.fillna(False)
+        n = int(mask.astype(bool).sum())
+    else:
+        n = int(len(table_df))
+    return row_table(ctx, template_df_cons(table_df, {alias: [n]}))
+
+
 def drop_cols(ctx: Any, cols: Sequence[str]) -> "Plottable":
     """Drop named columns from the active row table, ignoring any that don't exist."""
     table_df = get_active_table(ctx)

graphistry/compute/gfql/row/pipeline.py

@@ -180,6 +180,7 @@ def _gfql_cudf_list_sort_series_requires_host_bridge(series: Any) -> bool:
         "unwind",
         "group_by",
         "drop_cols",
+        "count_table",
     }
 )
 
@@ -4403,6 +4404,7 @@ def order_by(self, keys: List[Any]) -> "Plottable":
     skip = row_frame_ops.skip
     limit = row_frame_ops.limit
     distinct = row_frame_ops.distinct
+    count_table = row_frame_ops.count_table
 
     def unwind(self, expr: Any, as_: str = "value") -> "Plottable":
         """Vectorized UNWIND for column or literal list expressions."""
@@ -4653,14 +4655,15 @@ def bind(self) -> "Plottable":
     "unwind": RowPipelineMixin.unwind,
     "group_by": RowPipelineMixin.group_by,
     "drop_cols": RowPipelineMixin.drop_cols,
+    "count_table": RowPipelineMixin.count_table,
 }
 
 
 # Row-pipeline ops with native polars implementations (frame-level only — no
 # cypher expression engine). Everything else falls back through the guard below
 # until lowered natively. See plans/gfql-polars-engine (Phase 2).
 _POLARS_NATIVE_ROW_PIPELINE_CALLS = frozenset(
-    {"rows", "skip", "limit", "distinct", "drop_cols"}
+    {"rows", "skip", "limit", "distinct", "drop_cols", "count_table"}
 )
 
 

graphistry/tests/compute/gfql/test_conformance_ledger.py

@@ -271,6 +271,7 @@ def test_known_uncovered_function_reasons_are_nonempty():
     "order_by": "row sort; native on polars chain, NIE via call()/DAG executor; not asserted via call(). TODO.",
     "unwind": "list explode; native on polars chain, NIE via call()/DAG executor; not asserted via call(). TODO.",
     "group_by": "grouped aggregation; native on polars chain, NIE via call()/DAG executor; not asserted via call(). TODO.",
+    "count_table": "count(*) short-circuit fast path (table height / source-mask sum); native frame op emitted by the cypher lowering, exercised as a labeled subject via _ROW_OP_CASES + the count_all_nodes/edges cypher cases, not via a direct call() consistency label. TODO.",
     "semi_apply_mark": "correlated EXISTS-mark; row-pipeline op honest-NIE under polars; not asserted. TODO.",
     "anti_semi_apply": "anti-semi correlated filter; row-pipeline op honest-NIE under polars; not asserted. TODO.",
     "join_apply": "correlated row join; row-pipeline op honest-NIE under polars; not asserted. TODO.",

graphistry/tests/compute/gfql/test_engine_polars_conformance_matrix.py

@@ -262,6 +262,9 @@ def _cypher_expression_queries():
         ("count_distinct_grouped", "MATCH (n) RETURN n.flag AS k, count(DISTINCT n.num) AS cd"),
         ("count_distinct_all", "MATCH (n) RETURN count(DISTINCT n.flag) AS cd"),
         ("count_grouped", "MATCH (n) RETURN n.flag AS k, count(n.num) AS c"),
+        # count(*) short-circuit (count_table fast path): whole-graph node / edge counts.
+        ("count_all_nodes", "MATCH (n) RETURN count(*) AS c"),
+        ("count_all_edges", "MATCH ()-[r]->() RETURN count(*) AS c"),
         ("size_str", "MATCH (n) RETURN n.id AS id, size(n.name) AS sz"),
         ("substring3", "MATCH (n) RETURN n.id AS id, substring(n.name, 0, 4) AS sub"),
         ("substring2", "MATCH (n) RETURN n.id AS id, substring(n.name, 2) AS sub"),
@@ -506,6 +509,7 @@ def _rowop_exercised():
         "with_", "unwind",
         "rows", "skip", "limit", "distinct", "drop_cols",
         "order_by", "select", "return_", "where_rows", "group_by",
+        "count_table",
     }
 
 
@@ -1086,6 +1090,7 @@ def test_conformance_unwind_chain_vs_cypher_consistent():
     ("where_rows", [n(), rows(), call("where_rows", {"expr": "num > 50"})]),
     ("group_by",   [n(), rows(), call("group_by", {"keys": ["flag"],
                     "aggregations": [("c", "count"), ("s", "sum", "num")]})]),
+    ("count_table", [n(), rows(), call("count_table", {"table": "nodes", "alias": "cnt"})]),
 ]