From c2138b0d35826baa43b248553bff51e7c652032c Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Thu, 16 Jul 2026 05:51:43 -0700
Subject: [PATCH] feat(gfql): add aggregate fast paths and final hardening

---
 .github/workflows/ci.yml                      |   2 +-
 CHANGELOG.md                                  |   4 +-
 bin/ci_cypher_surface_guard_baseline.json     |   2 +-
 bin/coverage_audit.py                         |  17 +-
 bin/test-polars.sh                            |  15 +-
 graphistry/Engine.py                          |  16 +-
 graphistry/compute/gfql/cypher/lowering.py    |  28 +-
 graphistry/compute/gfql/index/__init__.py     |   7 +-
 graphistry/compute/gfql/index/api.py          |  14 +-
 graphistry/compute/gfql/index/cost.py         |  40 --
 graphistry/compute/gfql/index/types.py        |   1 -
 .../compute/gfql/lazy/engine/polars/chain.py  |  17 +-
 .../gfql/lazy/engine/polars/row_pipeline.py   |  84 ++-
 graphistry/compute/gfql/row/frame_ops.py      |   3 +-
 graphistry/compute/gfql/row/prefilter.py      |  20 +-
 graphistry/compute/gfql_unified.py            | 619 +++++++++++++++++-
 .../gfql/cypher/test_compiler_policy.py       |  72 +-
 .../compute/gfql/cypher/test_lowering.py      | 456 ++++++++++++-
 .../compute/gfql/cypher/test_row_pushdown.py  |  34 +-
 .../tests/compute/gfql/index/test_index.py    |  91 +--
 .../gfql/test_engine_polars_binding_rows.py   |  75 +++
 .../tests/compute/test_engine_coercion.py     |  42 --
 22 files changed, 1318 insertions(+), 341 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 4b95768bc2..43a5e955b9 100644
--- a/.github/workflows/ci.yml
+++ b/.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
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8a7d6c1e3c..cc55e0e5a3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,7 +12,6 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL engine-selection docs (pandas / polars / cuDF / polars-gpu)**: New :doc:`Choosing a GFQL Engine <gfql/engines>` page — a numbers-first, persona-tested guide to the four interchangeable engines. Adds the one-keyword `engine='polars'` speedup (up to ~38× over pandas on real graphs, no GPU), a motivating warm-median comparison table on real public graphs (LiveJournal 35M / Orkut 117M), a decision matrix (workload shape × size × hardware → engine, with the measured ~10K-edge CPU crossover, GPU-work-bound rule, polars-gpu memory-pressure caveat, and GPU-or-error contract), a cuDF-vs-polars-gpu disambiguation (eager-op vs fused-lazy; cuDF is not deprecated), an honest "when *not* to use Polars" section, the differential-parity guarantee, and a methodology + reproducer-script disclosure. Rewrote the top of `gfql/performance.rst` to lead with the engine comparison (de-marketed the prose), wired the new page into the GFQL toctree + recommended paths, and added Polars/polars-gpu to the engine examples in `gfql/quick.rst` and `gfql/about.rst` (previously only pandas/cuDF were documented). Driven by 4-persona doc user-testing (pandas DS, RAPIDS user, perf engineer, skeptical evaluator).
 
 ### Added
-- **GFQL Cypher row-pipeline single-alias predicate pushdown**: Planner-generated `rows(alias_prefilters=...)` hints pre-filter eligible node and single-hop edge alias frames before binding joins on pandas/cuDF while retaining the original post-join filters for exact semantics and engine-safe fallback.
 - **GFQL viz-filter-pipeline acceptance suite + regression benchmark (viz-filter L3)**: `test_viz_pipeline_conformance.py` — curated full-panel pipelines (node+edge filters, exclusion-dominates composition, `(pred OR IS NULL)` keep-null leaves, both EXISTS prune-isolated flavors, searchAny composition, deterministic paging), graph-state prune shapes with exact node+edge pins, a 40-seed panel-state fuzzer with an independent plain-pandas second oracle, and a case/regex/unicode trick matrix (ß/İ full-case-mapping pins, metachar literal-vs-regex, null cells, Categorical) — all parity-or-NIE across pandas/cuDF/polars/polars-gpu. `benchmarks/gfql/viz_filter_pipeline.py` — six streamgl-viz panel scenarios (filters, keep-self GRAPH prune, EXISTS prune, node/edge search, combined) at 100K/1M/10M with native-frame-per-engine fairness, an NIE-tolerant matrix, and JSON receipts (first receipt: 100k × 4 engines, everything within the ~350ms interactive reference except pandas combined). Documented findings from the first runs: the same-path WHERE route dedupes parallel edges (diverges from the panel algebra's edge multiplicity — pinned + tracked), and edge-alias searchAny declines on polars (tracked).
 - **GFQL Cypher `searchAny(entity, term[, opts])` cross-column search predicate + `g.search_nodes()`/`g.search_edges()` (viz-filter L2, native on all four engines)**: True where ANY of the entity's columns matches the term — the streamgl-viz inspector's table-search semantics as a composable WHERE predicate: OR across columns; case-insensitive substring by DEFAULT (case-folded, never regex — the common call avoids every engine regex limit); regex opt-in obeying the same per-engine decline rules as `=~`; dtype gate AS SEMANTICS (string columns always; integer columns iff the term is a numeric literal, per the inspector's `/^[0-9.-]+$/` gate; floats/dates/booleans reachable via the explicit `columns:` list). Options map `{caseSensitive, regex, columns}` is strict-validated (unknown keys error listing the valid ones); unbound aliases and missing explicit columns error clearly; null cells never match. Lowered like the pattern-predicate markers (a `search_any` row op + fresh marker column), so it composes through AND/OR/NOT and different node/edge terms coexist in one pipeline. Per-column matching reuses the parity-hardened `Contains` predicate on pandas/cuDF and a lowercase-fold/any_horizontal lowering on polars; oracle-pinned + 4-engine parity-or-NIE conformance cases. Python twins `g.search_nodes(term, columns=, case_sensitive=, regex=)` / `g.search_edges(...)` filter their own table and return a Plottable (polars-frame twins decline honestly for now — use the cypher op). Honest declines (NIE, use engine='pandas'): edge-alias searchAny on polars, and explicit columns beyond string/int/bool dtypes on polars AND cuDF (incl. floats: repr diverges across engines — dgx-probed).
 - **GFQL Cypher `EXISTS { <pattern> }` pattern-existence subqueries (openCypher-standard), native on all four engines**: `WHERE EXISTS { (n)-[:R]->() }` and `WHERE NOT EXISTS { (n)--() }` now parse and run — the declarative prune-isolated building blocks for the streamgl-viz filter pipeline. An `EXISTS` body reuses the existing pattern-predicate lowering wholesale (`semi_apply_mark` / `anti_semi_apply` row ops), so pandas/cuDF worked immediately; the polars engine gains NATIVE lowerings for the semi-apply family (correlated key sets computed by the polars chain executor's named-flag columns; order-preserving `is_in` joins) plus `rows(binding_ops=...)` for the single-entity row table — previously all honest-NIE. Aliases introduced inside the braces are existentially scoped (`EXISTS { (n)--(m) }` allowed with `m` unbound outside — bare pattern predicates keep the conservative guard), inline property maps work, and the one supported inner `WHERE` form is endpoint inequality — `EXISTS { (n)--(m) WHERE m <> n }`, the drop-self-loop prune-isolated flavor (pandas/cuDF filter the correlated bindings; polars excludes self-loop edges, which is exactly the `m <> n` witness). Both prune flavors are oracle-pinned in the conformance matrix on a self-loop discriminator graph, 4-engine parity-or-NIE. Honest declines with clear errors: `EXISTS` in RETURN/WITH projections, general inner `WHERE`, multi-pattern bodies, full `MATCH..RETURN` subquery bodies, multi-alias correlation on polars.
@@ -41,9 +40,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL native Polars bindings-row tables (`rows(binding_ops)`) — traversal Cypher on polars (#1709)**: the Cypher multi-alias lowering's `rows(binding_ops=...)` op (one row per matched path) now runs natively on `engine='polars'` for **fixed-length connected patterns** — unblocking traversal-shaped Cypher that previously NIE'd: multi-alias property projections (`MATCH (a)-[e]->(b) RETURN a.x, e.w, b.y`), top-k in-degree (`RETURN b.id, count(a) ... ORDER BY ... LIMIT`, graph-benchmark q1/q2), and fixed multi-hop counts (`MATCH (a)-->(b)-->(c) RETURN count(*)`, q8/q9), with forward/reverse/undirected edges, node/edge filters, and edge-alias payload columns — **plus bounded directed variable-length segments** (`-[*i..k]->`, typed `-[:TYPE*i..k]->`, exactly-k; iterative pair joins with Cypher path multiplicity and zero-hop rows), covering the q3 shape (`MATCH (a:Person)-[:FOLLOWS*1..2]->(b:Person) ... RETURN avg(b.age)`). With this, **all graph-benchmark q1–q9 shapes run as real GFQL Cypher on pandas, cuDF, and polars**. Also adds native `with_(extend=True)` (emitted by the bindings-path aggregate lowering) and an honest decline for `group_by(key_prefixes=...)` (whole-row bindings grouping — was a latent silent-wrong-key trap). **Honestly deferred** (`NotImplementedError`, no pandas bridge): unbounded `[*]`, undirected/aliased variable-length, shortestPath scalar bindings, node/edge `query=`/endpoint-match params, cartesian (`MATCH (a),(b)`) mode, seeded re-entry contexts. Differential parity vs the pandas oracle (+20 tests incl. path multiplicity and undirected self-loops; conformance corpus extended). **Perf @500k nodes/2M edges (CPU):** q1 top-k in-degree **17×** vs pandas (2559→150ms), seeded 2-hop count 7.3×, seeded 1-hop projection 13×. polars-gpu runs the same code path (eager joins; GPU verification queued on the shared GPU box).
 
 ### Fixed
-- **GFQL seeded indexes now engage for small Polars/cuDF frontiers**: NaN-free native Polars frames retain object identity during normalization, preserving resident CSR validity, while a tunable absolute floor prevents the fractional cost gate from routing tiny seeded hops over low-cardinality slices to a full scan. Scan/index result parity is unchanged; the default floor is a routing heuristic whose engine/data crossover remains subject to measurement.
 - **GFQL Cypher `GRAPH { }` residual predicates now fail safely or apply as graph masks**: `GRAPH { MATCH ... WHERE ... }` no longer silently drops predicates that the graph-state path cannot apply. Safe one-node/one-edge residual filters, including disjunctions and `searchAny(...)`, are applied before graph-state matching; unsupported pattern-predicate, multi-alias, and Polars graph-residual cases now raise clear validation errors instead of returning an over-broad subgraph.
-- **GFQL Cypher `=~` / scalar-fn cross-engine hardening (review wave, dgx-verified)**: (1) composed `=~` (`WHERE … =~ … OR …`, `RETURN`-expression position) now works on `engine='cudf'` — the series evaluator used raw `.str.fullmatch`, which cuDF lacks, and the resulting `AttributeError` was masked as "unsupported predicate op" (it now routes through the `Fullmatch` predicate's engine workarounds, and honest `NotImplementedError` declines pass through instead of being re-labeled); (2) the cuDF fullmatch emulation anchors alternations as a whole (`^(ab|cd)$` — bare `^ab|cd$` silently matched `'abXXX'`); (3) the cuDF case-insensitive `(?i)` lowercase-folding workaround now declines the fold-unsound shapes — uppercase escape classes (`.lower()` turns `\D` into `\d`, silently inverting the predicate), case-crossing character ranges (`(?i)[A-z]` silently narrowed; `[X-b]` folded to an invalid range), and non-ASCII patterns — while lowercase escapes (`\d`, `\.`) keep folding as before; lookaround, backreferences, and named-group refs decline up front (libcudf rejects them at kernel-compile time); (4) polars `Match`/`Fullmatch` lowering applies the same Rust-regex guard as `Contains` (lookaround/backrefs decline instead of a non-NIE `ComputeError` at collect); (5) `toLower`/`toUpper` on a non-string column decline like neo4j's type error instead of broadcasting the stringified Series repr (pandas/cuDF) or raising a non-NIE `SchemaError` (polars-gpu); (6) polars `floor`/`ceil`/`round` cast to `Float64` so integer columns return Float like neo4j and the pandas engine; (7) invalid regex patterns on the composed path raise a clear "invalid regex pattern" error instead of "unsupported predicate op".
+- **GFQL Cypher `=~` / scalar-fn cross-engine hardening (review wave, dgx-verified)**: (1) composed `=~` (`WHERE … =~ … OR …`, `RETURN`-expression position) now works on `engine='cudf'` — the series evaluator used raw `.str.fullmatch`, which cuDF lacks, and the resulting `AttributeError` was masked as "unsupported predicate op" (it now routes through the `Fullmatch` predicate's engine workarounds, and honest `NotImplementedError` declines pass through instead of being re-labeled); (2) the cuDF fullmatch emulation anchors alternations as a whole (`^(ab|cd)$` — bare `^ab|cd$` silently matched `'abXXX'`); (3) the cuDF case-insensitive `(?i)` lowercase-folding workaround now declines the fold-unsound shapes — uppercase escape classes (`.lower()` turns `\D` into `\d`, silently inverting the predicate), case-crossing character ranges (`(?i)[A-z]` silently narrowed; `[X-b]` folded to an invalid range), and non-ASCII patterns — while lowercase escapes (`\d`, `\.`) keep folding as before; lookaround, backreferences, and named-group refs decline up front (libcudf rejects them at kernel-compile time); (4) polars `Match`/`Fullmatch` lowering applies the same Rust-regex guard as `Contains` (lookaround/backrefs decline instead of a non-NIE `ComputeError` at collect); (5) `toLower`/`toUpper` on a non-string column decline with a Cypher-standard type error instead of broadcasting the stringified Series repr (pandas/cuDF) or raising a non-NIE `SchemaError` (polars-gpu); (6) polars `floor`/`ceil`/`round` cast to `Float64` so integer columns return Float per Cypher semantics, matching the pandas engine; (7) invalid regex patterns on the composed path raise a clear "invalid regex pattern" error instead of "unsupported predicate op".
 - **GFQL Cypher `round()` hardening (review wave, dgx-verified)**: (1) the `polars` extra's floor is now `polars>=1.29` — `Expr.round(mode=)` shipped in py-1.29.0 (pola-rs/polars#22248), not 1.5 as previously pinned, so 1.5–1.28 installs crashed with a raw `TypeError` on `round(x, p>0)` under `engine='polars'`; (2) `round(x, p>308)` is the identity on both engines (a float64 has no digits there) instead of pandas raising through an unclear decline while polars returned identity — parity restored, `10.0**p` overflow guarded; (3) polars `round(x, p>0)` normalizes `-0.0` to `+0.0` like the pandas kernel (`round(-0.04, 1)` was `0.0` on pandas vs `-0.0` on polars — invisible to value equality, pinned by a sign-bit test); (4) documented the precision>0 decimal-string deviation vs neo4j (`round(2.675, 2)` = `2.67` binary-double here vs `2.68` BigDecimal there) and added deterministic tie/hazard matrix cases so ties actually reach cuDF/polars-gpu (the fixture's normal-distribution floats never tied).
 - **GFQL `materialize_nodes()` on an edges-only graph under `engine='polars'`**: crashed with `AttributeError: 'Series' object has no attribute 'drop_duplicates'` — the node-id derivation used pandas-only `.drop_duplicates()` / `.reset_index()` / `.rename(columns=)` on a polars Series/DataFrame. Now polars-aware (`.unique(maintain_order=True)`, `.select().rename({...})`), matching the pandas oracle. Surfaced by any `let()` DAG on an edges-only graph under Polars (the DAG pre-materializes nodes). pandas/cuDF paths unchanged.
 - **CI: `compute_networkx` HITS test was scipy-version flaky**: `test_compute_networkx_hits_outputs_hubs_and_authorities` used a pure directed 3-cycle (`a→b→c→a`), whose adjacency is a permutation matrix (all singular values equal). networkx HITS computes scores via scipy `svds(k=1)`, which then returns an arbitrary vector from that fully-degenerate singular space; when its components sum to ~0, HITS's `h /= h.sum()` normalization blows up to `inf`/`nan`. Which vector `svds` returns is scipy/LAPACK-version dependent, so the test passed on the pinned scipy but failed on the `nx-upper-scipy` CI profile's newer scipy — a latent flake, not a code regression. Switched to a graph with a well-defined hub/authority structure (`a→b`, `a→c`, `b→c`), whose dominant singular vector is unique (Perron-Frobenius) and therefore version-stable. Test-only change.- **GFQL `materialize_nodes()` on an edges-only graph under `engine='polars'`**: crashed with `AttributeError: 'Series' object has no attribute 'drop_duplicates'` — the node-id derivation used pandas-only `.drop_duplicates()` / `.reset_index()` / `.rename(columns=)` on a polars Series/DataFrame. Now polars-aware (`.unique(maintain_order=True)`, `.select().rename({...})`), matching the pandas oracle. Surfaced by any `let()` DAG on an edges-only graph under Polars (the DAG pre-materializes nodes). pandas/cuDF paths unchanged.- **GFQL `engine='polars-gpu'` LazyFrame-input coercion now collects on the GPU executor**: `Engine.df_to_engine(lazyframe, POLARS_GPU)` materialized a `polars.LazyFrame` input with a bare `.collect()` on the CPU default executor — ignoring `gpu_executor()` and not distinguishing `POLARS` from `POLARS_GPU`. It now routes through the target-aware lazy collect, so a LazyFrame coerced under `polars-gpu` collects on the cudf-polars GPU executor (and `polars` honors `cpu_streaming()`). No change for already-materialized frame inputs (the common path).
diff --git a/bin/ci_cypher_surface_guard_baseline.json b/bin/ci_cypher_surface_guard_baseline.json
index bbf8dbb481..393b035ad3 100644
--- a/bin/ci_cypher_surface_guard_baseline.json
+++ b/bin/ci_cypher_surface_guard_baseline.json
@@ -13,5 +13,5 @@
       "max_properties": 0
     }
   },
-  "lowering_py_max_lines": 8511
+  "lowering_py_max_lines": 9125
 }
diff --git a/bin/coverage_audit.py b/bin/coverage_audit.py
index 980e8fdd6f..8a5db836d2 100755
--- a/bin/coverage_audit.py
+++ b/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
 
diff --git a/bin/test-polars.sh b/bin/test-polars.sh
index 6d7965cee6..a6a6f5b936 100755
--- a/bin/test-polars.sh
+++ b/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
diff --git a/graphistry/Engine.py b/graphistry/Engine.py
index a19c8f1449..a3d93c2939 100644
--- a/graphistry/Engine.py
+++ b/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])
 
 
diff --git a/graphistry/compute/gfql/cypher/lowering.py b/graphistry/compute/gfql/cypher/lowering.py
index e42a16db65..d7a67ba019 100644
--- a/graphistry/compute/gfql/cypher/lowering.py
+++ b/graphistry/compute/gfql/cypher/lowering.py
@@ -2265,7 +2265,7 @@ def _projection_ref_from_expr_safe(
 
 
 def _clause_has_mixed_aggregate_item(
-    query: Any,
+    query: CypherQuery,
     *,
     alias_targets: Mapping[str, ASTObject],
     params: Optional[Mapping[str, Any]],
@@ -2275,12 +2275,12 @@ def _clause_has_mixed_aggregate_item(
     cross-source items have ambiguous multiplicity and must keep the conservative
     fail-fast; a clean split (``c.city`` and ``avg(p.age)`` as separate items) does
     not trip this (#1273 / rejects-unsound-multi-source-overlap contract)."""
-    return_clause = getattr(query, "return_", None)
+    return_clause = query.return_
     exprs: List[Tuple[str, int, int]] = []
     if return_clause is not None:
         for item in return_clause.items:
             exprs.append((item.expression.text, item.span.line, item.span.column))
-    order_by = getattr(query, "order_by", None)  # top-level ORDER BY (not on ReturnClause)
+    order_by = query.order_by  # top-level ORDER BY (not on ReturnClause)
     if order_by is not None:
         for order_item in order_by.items:
             exprs.append((order_item.expression.text, order_item.span.line, order_item.span.column))
@@ -2324,7 +2324,7 @@ def _expr_has_aggregate(node: ExprNode) -> bool:
 
 
 def _binding_prop_alias_set(
-    query: Any,
+    query: CypherQuery,
     *,
     alias_targets: Mapping[str, ASTObject],
     params: Optional[Mapping[str, Any]],
@@ -2342,18 +2342,18 @@ def _binding_prop_alias_set(
     itself is EXACT: ``_expr_match_alias_usage`` non-aggregate refs are precisely the
     property / whole-entity uses; aggregate-only refs (``count(a)``) are excluded.
     """
-    if getattr(query, "with_stages", None):
+    if query.with_stages:
         return None
-    if getattr(query, "where", None) is not None:
+    if query.where is not None:
         return None
-    matches = cast(Sequence[MatchClause], getattr(query, "matches", ()) or ())
+    matches = query.matches or ()
     if len(matches) != 1:
         return None  # multi-MATCH / cartesian — conservative
     match_clause = matches[0]
-    if getattr(match_clause, "where", None) is not None or getattr(match_clause, "optional", False):
+    if match_clause.where is not None or match_clause.optional:
         return None
-    return_clause = getattr(query, "return_", None)
-    if return_clause is None or getattr(return_clause, "where", None) is not None:
+    return_clause = query.return_
+    if return_clause is None:
         return None
 
     # A repeated node alias (e.g. `MATCH (n)-[:LOOP]->(n)`) enforces n==n via hidden
@@ -2365,7 +2365,7 @@ def _binding_prop_alias_set(
             for el in _match_pattern_elements(match_clause)
             if isinstance(el, NodePattern) and el.variable is not None
         ]
-    except Exception:
+    except (GFQLValidationError, RuntimeError):
         return None
     if len(node_vars) != len(set(node_vars)):
         return None
@@ -2374,9 +2374,9 @@ def _binding_prop_alias_set(
         agg_specs = _collect_aggregate_specs_for_clause(
             return_clause, params=params, alias_targets=alias_targets
         )
-    except Exception:
+    except (GFQLValidationError, RuntimeError):
         return None
-    if any(getattr(spec, "func", None) == "collect" for spec in agg_specs):
+    if any(spec.func == "collect" for spec in agg_specs):
         return None
 
     node_aliases = {a for a, t in alias_targets.items() if isinstance(t, ASTNode)}
@@ -2386,7 +2386,7 @@ def _binding_prop_alias_set(
     exprs: List[Tuple[str, int, int]] = []
     for item in return_clause.items:
         exprs.append((item.expression.text, item.span.line, item.span.column))
-    order_by = getattr(query, "order_by", None)  # top-level ORDER BY (not on ReturnClause)
+    order_by = query.order_by  # top-level ORDER BY (not on ReturnClause)
     if order_by is not None:
         for order_item in order_by.items:
             exprs.append((order_item.expression.text, order_item.span.line, order_item.span.column))
diff --git a/graphistry/compute/gfql/index/__init__.py b/graphistry/compute/gfql/index/__init__.py
index f58588bb79..7ee6cbf4f8 100644
--- a/graphistry/compute/gfql/index/__init__.py
+++ b/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",
 ]
diff --git a/graphistry/compute/gfql/index/api.py b/graphistry/compute/gfql/index/api.py
index f5296bd911..bc6f751e5a 100644
--- a/graphistry/compute/gfql/index/api.py
+++ b/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
diff --git a/graphistry/compute/gfql/index/cost.py b/graphistry/compute/gfql/index/cost.py
index bd8ca53862..754db74482 100644
--- a/graphistry/compute/gfql/index/cost.py
+++ b/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:
diff --git a/graphistry/compute/gfql/index/types.py b/graphistry/compute/gfql/index/types.py
index 47a088b73c..af3aa800f6 100644
--- a/graphistry/compute/gfql/index/types.py
+++ b/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]
diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 012a76e5ee..a546ef849a 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -362,6 +362,14 @@ def _try_native_row_op(g_cur, op):
 
     fn = getattr(op, "function", None)
     if fn == "rows" and op.params.get("binding_ops") is not None:
+        # Single-entity boundary rows emitted by MATCH (n) / EXISTS seeds are
+        # handled by the pattern-apply helper. Try that narrow shape before the
+        # connected multi-alias bindings table path, which intentionally declines
+        # node-only binding_ops.
+        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
         # Multi-alias bindings table (#1709): native for fixed-length connected
         # patterns; binding_rows_polars declines (None → NIE) outside that subset.
         if op.params.get("alias_endpoints") is not None:
@@ -372,15 +380,6 @@ def _try_native_row_op(g_cur, op):
     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.
diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index 88853cb45a..17b071446b 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
@@ -23,6 +23,7 @@
     from graphistry.compute.gfql.expr_parser import ExprNode, FunctionCall
 
 from graphistry.Plottable import Plottable
+from graphistry.utils.json import JSONVal
 from .dtypes import is_float as _dtype_is_float, is_int as _dtype_is_int, is_numeric as _dtype_is_numeric, is_stringlike as _dtype_is_stringlike
 
 
@@ -822,8 +823,8 @@ def select_extend_polars(g: Plottable, items: Sequence[Any]) -> Optional[Plottab
 
 def binding_rows_polars(
     g: Plottable,
-    binding_ops: Sequence[Any],
-    attach_prop_aliases: Optional[List[str]] = None,
+    binding_ops: Sequence[Dict[str, JSONVal]],
+    attach_prop_aliases: Optional[Sequence[str]] = None,
 ) -> Optional[Plottable]:
     """Native polars bindings-row table for FIXED-LENGTH connected patterns (#1709).
 
@@ -843,13 +844,13 @@ def binding_rows_polars(
     never bridges to pandas. Parity gate: differential tests vs the pandas oracle.
     """
     import polars as pl
-    from graphistry.compute.ast import ASTEdge, ASTNode, from_json as ast_from_json
+    from graphistry.compute.ast import ASTEdge, ASTNode, ASTObject, from_json as ast_from_json
     from graphistry.compute.gfql.lazy import collect as _lazy_collect
     from graphistry.compute.gfql.row.pipeline import RowPipelineMixin
     from graphistry.compute.gfql.same_path.edge_semantics import EdgeSemantics
     from .predicates import filter_by_dict_polars
 
-    def _names(lf: Any) -> List[str]:
+    def _names(lf: pl.LazyFrame) -> List[str]:
         # LazyFrame column names WITHOUT collecting data (schema-only resolve).
         return lf.collect_schema().names()
 
@@ -864,7 +865,7 @@ def _names(lf: Any) -> List[str]:
         # Bounded re-entry seeds the first alias from carried rows — pandas-only.
         return None
 
-    ops = [ast_from_json(op_json, validate=False) for op_json in binding_ops]
+    ops: List[ASTObject] = [ast_from_json(op_json, validate=False) for op_json in binding_ops]
     # Shared validation (engine-agnostic): raises the canonical GFQLValidationError
     # for malformed op sequences / duplicate aliases — same error as pandas.
     RowPipelineMixin._gfql_validate_binding_ops(ops)
@@ -875,7 +876,7 @@ def _names(lf: Any) -> List[str]:
 
     for idx, op in enumerate(ops):
         if idx % 2 == 0:
-            if not isinstance(op, ASTNode) or getattr(op, "query", None) is not None:
+            if not isinstance(op, ASTNode) or op.query is not None:
                 return None
         else:
             if not isinstance(op, ASTEdge):
@@ -891,22 +892,22 @@ def _names(lf: Any) -> List[str]:
                     op.direction == "undirected"
                     or bool(op.to_fixed_point)
                     or (op.max_hops is None and op.hops is None)
-                    or isinstance(getattr(op, "_name", None), str)
+                    or isinstance(op._name, str)
                 ):
                     return None
             if op.direction not in ("forward", "reverse", "undirected"):
                 return None
             if any(
-                getattr(op, attr, None) is not None
-                for attr in (
-                    "edge_query", "source_node_match", "destination_node_match",
-                    "source_node_query", "destination_node_query",
-                    "label_node_hops", "label_edge_hops",
-                    "output_min_hops", "output_max_hops",
+                value is not None
+                for value in (
+                    op.edge_query, op.source_node_match, op.destination_node_match,
+                    op.source_node_query, op.destination_node_query,
+                    op.label_node_hops, op.label_edge_hops,
+                    op.output_min_hops, op.output_max_hops,
                 )
             ):
                 return None
-            if bool(getattr(op, "label_seeds", False)) or bool(getattr(op, "include_zero_hop_seed", False)):
+            if bool(op.label_seeds) or bool(op.include_zero_hop_seed):
                 return None
             # Duplicate-id + HAS_<Label> endpoint disambiguation: pandas has a
             # bespoke candidate-domain rule here; decline rather than diverge.
@@ -927,11 +928,14 @@ def _names(lf: Any) -> List[str]:
         # NIE → use engine='pandas'/'polars'), never a silent CPU fallback.
         nodes_lf = nodes.lazy()
         edges_lf = edges.lazy()
-        seed_nodes = filter_by_dict_polars(nodes_lf, getattr(ops[0], "filter_dict", None))
+        first_op = ops[0]
+        if not isinstance(first_op, ASTNode):
+            return None
+        seed_nodes = filter_by_dict_polars(nodes_lf, first_op.filter_dict)
         state = seed_nodes.select(pl.col(node_id).alias("__current__"))
-        alias_frames: dict = {}
+        alias_frames: Dict[str, pl.LazyFrame] = {}
         node_aliases: List[str] = []
-        first_alias = getattr(ops[0], "_name", None)
+        first_alias = first_op._name
         if isinstance(first_alias, str):
             state = state.with_columns(pl.col("__current__").alias(first_alias))
             alias_frames[first_alias] = seed_nodes
@@ -939,9 +943,11 @@ def _names(lf: Any) -> List[str]:
 
         for edge_idx in range(1, len(ops), 2):
             edge_op = ops[edge_idx]
+            if not isinstance(edge_op, ASTEdge):
+                return None
             sem = EdgeSemantics.from_edge(edge_op)
-            edges_f = filter_by_dict_polars(edges_lf, getattr(edge_op, "edge_match", None))
-            edge_alias = getattr(edge_op, "_name", None)
+            edges_f = filter_by_dict_polars(edges_lf, edge_op.edge_match)
+            edge_alias = edge_op._name
             if isinstance(edge_alias, str):
                 payload_renames = {
                     col: f"{edge_alias}.{col}"
@@ -962,6 +968,23 @@ def _names(lf: Any) -> 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))
@@ -974,10 +997,14 @@ def _names(lf: Any) -> List[str]:
                 # `_gfql_multihop_binding_rows`). Zero-hop rows (min 0) keep the
                 # seed row (endpoint == start), also matching pandas.
                 # Same defaults as the pandas builder: bare hops=k means exactly-k.
-                min_hops = edge_op.min_hops if edge_op.min_hops is not None else (
+                min_hops_value = edge_op.min_hops if edge_op.min_hops is not None else (
                     edge_op.hops if edge_op.hops is not None else 1
                 )
-                max_hops = edge_op.max_hops if edge_op.max_hops is not None else edge_op.hops
+                max_hops_value = edge_op.max_hops if edge_op.max_hops is not None else edge_op.hops
+                if max_hops_value is None:
+                    return None
+                min_hops = int(min_hops_value)
+                max_hops = int(max_hops_value)
                 pairs = oriented.select(["__from__", "__to__"])
                 state_cols = _names(state)
                 reachable = [state] if min_hops == 0 else []
@@ -985,7 +1012,7 @@ def _names(lf: Any) -> List[str]:
                 # Lazy: build all max_hops iterations (no eager .height early-break —
                 # empty intermediates lazily join to empty, so the result is
                 # identical; the pandas break is an optimization, not semantics).
-                for _hop in range(1, int(max_hops) + 1):
+                for _hop in range(1, max_hops + 1):
                     current = (
                         current.join(pairs, left_on="__current__", right_on="__from__", how="inner")
                         .drop("__current__")
@@ -1002,15 +1029,13 @@ def _names(lf: Any) -> List[str]:
                     .rename({"__to__": "__current__"})
                 )
 
-            next_op = ops[edge_idx + 1]
-            next_nodes = filter_by_dict_polars(nodes_lf, getattr(next_op, "filter_dict", None))
             state = state.join(
-                next_nodes.select(node_id).unique(),
+                next_node_ids,
                 left_on="__current__",
                 right_on=node_id,
                 how="semi",
             )
-            next_alias = getattr(next_op, "_name", None)
+            next_alias = next_op._name
             if isinstance(next_alias, str):
                 state = state.with_columns(pl.col("__current__").alias(next_alias))
                 alias_frames[next_alias] = next_nodes
@@ -1025,7 +1050,10 @@ def _names(lf: Any) -> 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)
@@ -1048,7 +1076,7 @@ def _names(lf: Any) -> List[str]:
     edge_aliases = {
         alias
         for op in ops[1::2]
-        for alias in [getattr(op, "_name", None)]
+        for alias in [op._name]
         if isinstance(alias, str)
     }
     setattr(out, "_gfql_rows_edge_aliases", edge_aliases)
diff --git a/graphistry/compute/gfql/row/frame_ops.py b/graphistry/compute/gfql/row/frame_ops.py
index deed488a97..e734793096 100644
--- a/graphistry/compute/gfql/row/frame_ops.py
+++ b/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,
diff --git a/graphistry/compute/gfql/row/prefilter.py b/graphistry/compute/gfql/row/prefilter.py
index b92a3ee0ae..9c859e1750 100644
--- a/graphistry/compute/gfql/row/prefilter.py
+++ b/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
 
 
diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py
index 602e57221d..9852f2cd90 100644
--- a/graphistry/compute/gfql_unified.py
+++ b/graphistry/compute/gfql_unified.py
@@ -605,7 +605,7 @@ def _connected_join_cached_node_filter(
     return cast(DataFrameT, filtered)
 
 
-def _connected_join_cached_node_ids(
+def _connected_join_cached_node_ids(  # pragma: no cover - polars-only, covered by polars lane
     base_graph: Plottable,
     nodes_obj: DataFrameT,
     node_matches: Sequence[Optional[dict]],
@@ -713,7 +713,7 @@ def _connected_join_cached_singleton_dst_source_counts(
     if cache is not None and full_key in cache:
         return cast(DataFrameT, cache[full_key])
 
-    if engine in POLARS_ENGINES:
+    if engine in POLARS_ENGINES:  # pragma: no cover - polars-only, covered by polars lane
         import polars as pl
         counts = cast(
             DataFrameT,
@@ -731,7 +731,7 @@ def _connected_join_cached_singleton_dst_source_counts(
     return counts
 
 
-def _connected_join_cached_first_arm_shared_counts(
+def _connected_join_cached_first_arm_shared_counts(  # pragma: no cover - polars-only, covered by polars lane
     base_graph: Plottable,
     nodes_obj: DataFrameT,
     edges_obj: DataFrameT,
@@ -795,7 +795,7 @@ def _connected_join_cached_first_arm_shared_counts(
     return cast(DataFrameT, counts)
 
 
-def _connected_join_cached_second_arm_group_rows(
+def _connected_join_cached_second_arm_group_rows(  # pragma: no cover - polars-only, covered by polars lane
     base_graph: Plottable,
     nodes_obj: DataFrameT,
     edges_obj: DataFrameT,
@@ -878,13 +878,59 @@ def _connected_join_cached_second_arm_group_rows(
     return cast(DataFrameT, right_rows)
 
 
-def _property_ref(expr: Any, valid_aliases: Sequence[str]) -> Optional[Tuple[str, str]]:
-    if not isinstance(expr, str) or "." not in expr:
-        return None
-    alias, prop = expr.split(".", 1)
-    if alias not in valid_aliases or not prop:
+def _two_hop_cached_equal_domain_degree_counts(
+    base_graph: Plottable,
+    nodes_obj: DataFrameT,
+    edges_obj: DataFrameT,
+    domain_nodes: DataFrameT,
+    edge_domain: DataFrameT,
+    *,
+    node_match: Optional[dict],
+    edge_match: Optional[dict],
+    node_col: str,
+    src_col: str,
+    dst_col: str,
+    engine: Engine,
+) -> Optional[Tuple[DataFrameT, DataFrameT]]:
+    node_key = _connected_join_simple_filter_cache_key(node_match)
+    edge_key = _connected_join_simple_filter_cache_key(edge_match)
+    if node_key is None or edge_key is None:
         return None
-    return alias, prop
+
+    cache_attr = "_gfql_two_hop_equal_domain_degree_counts_cache"
+    cache = getattr(base_graph, cache_attr, None)
+    if not isinstance(cache, dict):
+        cache = {}
+        try:
+            setattr(base_graph, cache_attr, cache)
+        except Exception:
+            cache = None
+    full_key = (id(nodes_obj), id(edges_obj), engine.value, node_col, src_col, dst_col, node_key, edge_key)
+    if cache is not None and full_key in cache:
+        return cast(Tuple[DataFrameT, DataFrameT], cache[full_key])
+
+    if engine in POLARS_ENGINES:
+        domain_ids = domain_nodes.select(node_col).unique()
+        filtered_edges = (
+            edge_domain
+            .join(domain_ids, left_on=src_col, right_on=node_col, how="semi")
+            .join(domain_ids, left_on=dst_col, right_on=node_col, how="semi")
+        )
+        counts = (
+            cast(DataFrameT, filtered_edges.group_by(dst_col).len("__in_count__")),
+            cast(DataFrameT, filtered_edges.group_by(src_col).len("__out_count__")),
+        )
+    else:
+        domain_ids = domain_nodes[node_col].drop_duplicates()
+        filtered_edges = edge_domain[edge_domain[src_col].isin(domain_ids) & edge_domain[dst_col].isin(domain_ids)]
+        counts = (
+            cast(DataFrameT, filtered_edges.groupby(dst_col, sort=False).size().reset_index(name="__in_count__")),
+            cast(DataFrameT, filtered_edges.groupby(src_col, sort=False).size().reset_index(name="__out_count__")),
+        )
+
+    if cache is not None:
+        cache[full_key] = counts
+    return counts
 
 
 def _connected_join_post_property_columns(plan: ConnectedMatchJoinPlan, alias: str) -> List[str]:
@@ -1057,7 +1103,7 @@ def _connected_join_two_star_fast_grouped_count(
     nodes = df_to_engine(nodes_source, engine)
     edges = cast(DataFrameT, edges_obj)
 
-    if engine in POLARS_ENGINES:
+    if engine in POLARS_ENGINES:  # pragma: no cover - polars-only, covered by polars lane
         import polars as pl
         from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_by_dict_polars
 
@@ -1616,6 +1662,458 @@ def _apply_graph_residual_filters(
             graph = graph.edges(cast(DataFrameT, edges_df.loc[edge_mask]))
     return graph
 
+def _filter_nodes_for_fast_count(nodes: DataFrameT, filter_dict: Optional[dict], *, engine: Engine) -> DataFrameT:
+    if engine in POLARS_ENGINES:
+        from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_by_dict_polars
+        return cast(DataFrameT, filter_by_dict_polars(nodes, filter_dict))
+    return filter_by_dict(nodes, filter_dict, engine=EngineAbstract(engine.value))
+
+
+def _two_hop_count_alias(chain: Chain) -> Optional[str]:
+    ops = list(chain.chain)
+    if len(ops) != 4 or not all(isinstance(op, ASTCall) for op in ops):
+        return None
+    rows_call, with_call, group_call, select_call = cast(Tuple[ASTCall, ASTCall, ASTCall, ASTCall], tuple(ops))
+    if rows_call.function != "rows" or with_call.function != "with_" or group_call.function != "group_by" or select_call.function != "select":
+        return None
+    if rows_call.params.get("table") != "nodes":
+        return None
+    if with_call.params.get("items") != [("__cypher_group__", 1)]:
+        return None
+    aggs = group_call.params.get("aggregations")
+    if group_call.params.get("keys") != ["__cypher_group__"] or not isinstance(aggs, list) or len(aggs) != 1:
+        return None
+    agg = aggs[0]
+    if not isinstance(agg, (tuple, list)) or len(agg) != 2 or str(agg[1]).lower() != "count":
+        return None
+    alias = str(agg[0])
+    if select_call.params.get("items") != [(alias, alias)]:
+        return None
+    return alias
+
+
+def _two_hop_count_binding_ops(chain: Chain) -> Optional[Tuple[ASTNode, ASTEdge, ASTNode, ASTEdge, ASTNode]]:
+    if not chain.chain or not isinstance(chain.chain[0], ASTCall):
+        return None
+    rows_call = cast(ASTCall, chain.chain[0])
+    binding_ops = rows_call.params.get("binding_ops")
+    if not isinstance(binding_ops, list) or len(binding_ops) != 5:
+        return None
+    from graphistry.compute.ast import from_json as ast_from_json
+    ops = [ast_from_json(op_json, validate=False) for op_json in binding_ops]
+    n0, e0, n1, e1, n2 = ops
+    if not isinstance(n0, ASTNode) or not isinstance(e0, ASTEdge) or not isinstance(n1, ASTNode) or not isinstance(e1, ASTEdge) or not isinstance(n2, ASTNode):
+        return None
+    if not _is_connected_fast_single_hop(e0) or not _is_connected_fast_single_hop(e1):
+        return None
+    return n0, e0, n1, e1, n2
+
+
+def _property_ref(expr: Any, valid_aliases: Sequence[str]) -> Optional[Tuple[str, str]]:
+    if not isinstance(expr, str) or "." not in expr:
+        return None
+    alias, prop = expr.split(".", 1)
+    if alias not in valid_aliases or not prop:
+        return None
+    return alias, prop
+
+
+def _execute_single_hop_grouped_aggregate_fast_path(
+    base_graph: Plottable,
+    chain: Chain,
+    *,
+    engine: Union[EngineAbstract, str],
+) -> Optional[Plottable]:
+    ops = list(chain.chain)
+    if len(ops) not in (3, 4, 5) or not all(isinstance(op, ASTCall) for op in ops):
+        return None
+    rows_call = cast(ASTCall, ops[0])
+    with_call = cast(ASTCall, ops[1])
+    group_call = cast(ASTCall, ops[2])
+    suffix = [cast(ASTCall, op) for op in ops[3:]]
+    if rows_call.function != "rows" or with_call.function != "with_" or group_call.function != "group_by":
+        return None
+    if rows_call.params.get("table") != "nodes" or with_call.params.get("extend") is not True:
+        return None
+
+    order_call: Optional[ASTCall] = None
+    limit_call: Optional[ASTCall] = None
+    for op in suffix:
+        if op.function == "order_by" and order_call is None and limit_call is None:
+            order_call = op
+        elif op.function == "limit" and limit_call is None:
+            limit_call = op
+        else:
+            return None
+
+    binding_ops = rows_call.params.get("binding_ops")
+    if not isinstance(binding_ops, list) or len(binding_ops) != 3:
+        return None
+    from graphistry.compute.ast import from_json as ast_from_json
+    parsed_ops = [ast_from_json(op_json, validate=False) for op_json in binding_ops]
+    start_op, edge_op, end_op = parsed_ops
+    if not isinstance(start_op, ASTNode) or not isinstance(edge_op, ASTEdge) or not isinstance(end_op, ASTNode):
+        return None
+    if not _is_connected_fast_single_hop(edge_op):
+        return None
+    start_alias = getattr(start_op, "_name", None)
+    end_alias = getattr(end_op, "_name", None)
+    if not isinstance(start_alias, str) or not isinstance(end_alias, str) or start_alias == end_alias:
+        return None
+    aliases = (start_alias, end_alias)
+
+    with_items_raw = with_call.params.get("items")
+    if not isinstance(with_items_raw, list):
+        return None
+    with_items: Dict[str, Tuple[str, Optional[str]]] = {}
+    for item in with_items_raw:
+        if not isinstance(item, (tuple, list)) or len(item) != 2 or not isinstance(item[0], str):
+            return None
+        prop_ref = _property_ref(item[1], aliases)
+        ref: Tuple[str, Optional[str]]
+        if prop_ref is None:
+            if item[1] not in aliases:
+                return None
+            ref = (cast(str, item[1]), None)
+        else:
+            ref = prop_ref
+        with_items[item[0]] = ref
+
+    group_keys_raw = group_call.params.get("keys")
+    if not isinstance(group_keys_raw, list) or not group_keys_raw or not all(isinstance(key, str) for key in group_keys_raw):
+        return None
+    group_keys = cast(List[str], group_keys_raw)
+    if not all(key in with_items and with_items[key][1] is not None for key in group_keys):
+        return None
+
+    aggs_raw = group_call.params.get("aggregations")
+    if not isinstance(aggs_raw, list) or not aggs_raw:
+        return None
+    aggregations: List[Tuple[str, str, Optional[str]]] = []
+    for agg in aggs_raw:
+        if not isinstance(agg, (tuple, list)) or len(agg) not in (2, 3):
+            return None
+        alias = str(agg[0])
+        func = str(agg[1]).lower()
+        expr_alias = agg[2] if len(agg) == 3 else None
+        if func not in {"count", "avg", "sum", "min", "max"}:
+            return None
+        if expr_alias is not None and not isinstance(expr_alias, str):
+            return None
+        if expr_alias is not None and expr_alias not in with_items:
+            return None
+        if func != "count" and expr_alias is None:
+            return None
+        if expr_alias is not None and with_items[expr_alias][1] is None and func != "count":
+            return None
+        aggregations.append((alias, func, cast(Optional[str], expr_alias)))
+
+    order_keys: List[Tuple[str, bool]] = []
+    if order_call is not None:
+        raw_order = order_call.params.get("keys")
+        if not isinstance(raw_order, list):
+            return None
+        available = set(group_keys) | {alias for alias, _, _ in aggregations}
+        for item in raw_order:
+            if not isinstance(item, (tuple, list)) or len(item) != 2 or not isinstance(item[0], str):
+                return None
+            if item[0] not in available:
+                return None
+            direction = str(item[1]).lower()
+            if direction not in {"asc", "ascending", "desc", "descending"}:
+                return None
+            order_keys.append((item[0], direction in {"desc", "descending"}))
+
+    limit_value: Optional[int] = None
+    if limit_call is not None:
+        raw_limit = limit_call.params.get("value")
+        if not isinstance(raw_limit, int) or raw_limit < 0:
+            return None
+        limit_value = raw_limit
+
+    requested_engine = resolve_engine(cast(Any, engine), base_graph)
+    nodes_obj = getattr(base_graph, "_nodes", None)
+    edges_obj = getattr(base_graph, "_edges", None)
+    node_col = getattr(base_graph, "_node", None)
+    src_col = getattr(base_graph, "_source", None)
+    dst_col = getattr(base_graph, "_destination", None)
+    if nodes_obj is None or edges_obj is None or node_col is None or src_col is None or dst_col is None:
+        return None
+    node_col = str(node_col)
+    src_col = str(src_col)
+    dst_col = str(dst_col)
+    if node_col not in nodes_obj.columns or src_col not in edges_obj.columns or dst_col not in edges_obj.columns:
+        return None
+
+    nodes = cast(DataFrameT, nodes_obj)
+    start_nodes = _connected_join_cached_node_filter(base_graph, nodes, cast(Optional[dict], start_op.filter_dict), engine=requested_engine)
+    end_nodes = _connected_join_cached_node_filter(base_graph, nodes, cast(Optional[dict], end_op.filter_dict), engine=requested_engine)
+    edges = _connected_join_cached_edge_filter(base_graph, cast(DataFrameT, edges_obj), cast(Optional[dict], edge_op.edge_match), engine=requested_engine)
+
+    needed_by_alias: Dict[str, List[Tuple[str, str]]] = {start_alias: [], end_alias: []}
+    for out_col, ref in with_items.items():
+        alias, prop = ref
+        if prop is not None:
+            needed_by_alias[alias].append((out_col, prop))
+
+    if requested_engine in POLARS_ENGINES:  # pragma: no cover - polars-only, covered by polars lane
+        import polars as pl
+        start_ids = start_nodes.select(node_col).unique()
+        end_ids = end_nodes.select(node_col).unique()
+        work = (
+            edges
+            .join(start_ids, left_on=src_col, right_on=node_col, how="semi")
+            .join(end_ids, left_on=dst_col, right_on=node_col, how="semi")
+            .select([src_col, dst_col])
+        )
+
+        def join_props_polars(work_df: Any, alias: str, node_df: Any, edge_col: str) -> Any:
+            props = needed_by_alias.get(alias, [])
+            if not props:
+                return work_df
+            lookup_key = f"__gfql_t3_{alias}_id__"
+            exprs = [pl.col(node_col).alias(lookup_key)]
+            for out_col, prop in props:
+                if prop not in node_df.columns:
+                    return None
+                exprs.append(pl.col(prop).alias(out_col))
+            lookup = node_df.select(exprs)
+            return work_df.join(lookup, left_on=edge_col, right_on=lookup_key, how="inner")
+
+        work = join_props_polars(work, start_alias, start_nodes, src_col)
+        if work is None:
+            return None
+        work = join_props_polars(work, end_alias, end_nodes, dst_col)
+        if work is None:
+            return None
+        agg_exprs = []
+        for alias, func, expr_alias in aggregations:
+            if func == "count" and (expr_alias is None or with_items[expr_alias][1] is None):
+                agg_exprs.append(pl.len().alias(alias))
+            elif func == "count" and expr_alias is not None:
+                agg_exprs.append(pl.col(expr_alias).count().alias(alias))
+            elif func == "avg" and expr_alias is not None:
+                agg_exprs.append(pl.col(expr_alias).mean().alias(alias))
+            elif func == "sum" and expr_alias is not None:
+                agg_exprs.append(pl.col(expr_alias).sum().alias(alias))
+            elif func == "min" and expr_alias is not None:
+                agg_exprs.append(pl.col(expr_alias).min().alias(alias))
+            elif func == "max" and expr_alias is not None:
+                agg_exprs.append(pl.col(expr_alias).max().alias(alias))
+            else:
+                return None
+        out_nodes = work.group_by(group_keys, maintain_order=True).agg(agg_exprs)
+        if order_keys:
+            out_nodes = out_nodes.sort([key for key, _ in order_keys], descending=[desc for _, desc in order_keys])
+        if limit_value is not None:
+            out_nodes = out_nodes.head(limit_value)
+        out_df = cast(DataFrameT, out_nodes)
+    else:
+        start_ids = start_nodes[node_col].drop_duplicates()
+        end_ids = end_nodes[node_col].drop_duplicates()
+        work = edges[edges[src_col].isin(start_ids) & edges[dst_col].isin(end_ids)][[src_col, dst_col]]
+
+        def join_props_df(work_df: DataFrameT, alias: str, node_df: DataFrameT, edge_col: str) -> Optional[DataFrameT]:
+            props = needed_by_alias.get(alias, [])
+            if not props:
+                return work_df
+            prop_cols = []
+            for _, prop in props:
+                if prop not in node_df.columns:
+                    return None
+                if prop != node_col and prop not in prop_cols:
+                    prop_cols.append(prop)
+            lookup_key = f"__gfql_t3_{alias}_id__"
+            lookup = node_df[[node_col] + prop_cols].drop_duplicates(subset=[node_col]).copy()
+            for out_col, prop in props:
+                lookup[out_col] = lookup[prop]
+            lookup = lookup.rename(columns={node_col: lookup_key})
+            return cast(DataFrameT, work_df.merge(lookup, left_on=edge_col, right_on=lookup_key, how="inner"))
+
+        work = cast(DataFrameT, work)
+        joined_start = join_props_df(work, start_alias, start_nodes, src_col)
+        if joined_start is None:
+            return None
+        joined_end = join_props_df(joined_start, end_alias, end_nodes, dst_col)
+        if joined_end is None:
+            return None
+        work = joined_end
+        try:
+            grouped = work.groupby(group_keys, sort=False, dropna=False)
+        except TypeError:
+            grouped = work.groupby(group_keys, sort=False)
+        out_df = grouped.size().reset_index(name="__gfql_group_size__")[group_keys]
+        for alias, func, expr_alias in aggregations:
+            if func == "count" and (expr_alias is None or with_items[expr_alias][1] is None):
+                agg_df = grouped.size().reset_index(name=alias)
+            elif func == "count" and expr_alias is not None:
+                agg_df = grouped[expr_alias].count().reset_index(name=alias)
+            elif func == "avg" and expr_alias is not None:
+                agg_df = grouped[expr_alias].mean().reset_index(name=alias)
+            elif func == "sum" and expr_alias is not None:
+                agg_df = grouped[expr_alias].sum().reset_index(name=alias)
+            elif func == "min" and expr_alias is not None:
+                agg_df = grouped[expr_alias].min().reset_index(name=alias)
+            elif func == "max" and expr_alias is not None:
+                agg_df = grouped[expr_alias].max().reset_index(name=alias)
+            else:
+                return None
+            out_df = cast(DataFrameT, out_df.merge(agg_df, on=group_keys, how="left", sort=False))
+        if order_keys:
+            out_df = cast(DataFrameT, out_df.sort_values(by=[key for key, _ in order_keys], ascending=[not desc for _, desc in order_keys]))
+        if limit_value is not None:
+            out_df = cast(DataFrameT, out_df.head(limit_value))
+        out_df = df_to_engine(out_df.reset_index(drop=True), requested_engine)
+
+    out = base_graph.bind()
+    out._nodes = out_df
+    out._edges = df_cons(requested_engine)()
+    return out
+
+
+def _execute_two_hop_count_fast_path(
+    base_graph: Plottable,
+    chain: Chain,
+    *,
+    engine: Union[EngineAbstract, str],
+) -> Optional[Plottable]:
+    alias = _two_hop_count_alias(chain)
+    if alias is None:
+        return None
+    ops = _two_hop_count_binding_ops(chain)
+    if ops is None:
+        return None
+    start_op, first_edge, middle_op, second_edge, end_op = ops
+
+    requested_engine = resolve_engine(cast(Any, engine), base_graph)
+    nodes_obj = getattr(base_graph, "_nodes", None)
+    edges_obj = getattr(base_graph, "_edges", None)
+    node_col = getattr(base_graph, "_node", None)
+    src_col = getattr(base_graph, "_source", None)
+    dst_col = getattr(base_graph, "_destination", None)
+    if nodes_obj is None or edges_obj is None or node_col is None or src_col is None or dst_col is None:
+        return None
+    node_col = str(node_col)
+    src_col = str(src_col)
+    dst_col = str(dst_col)
+    if node_col not in nodes_obj.columns or src_col not in edges_obj.columns or dst_col not in edges_obj.columns:
+        return None
+
+    nodes = cast(DataFrameT, nodes_obj)
+    start_nodes = _connected_join_cached_node_filter(base_graph, nodes, cast(Optional[dict], start_op.filter_dict), engine=requested_engine)
+    middle_nodes = (
+        start_nodes
+        if middle_op.filter_dict == start_op.filter_dict
+        else _connected_join_cached_node_filter(base_graph, nodes, cast(Optional[dict], middle_op.filter_dict), engine=requested_engine)
+    )
+    end_nodes = (
+        middle_nodes
+        if end_op.filter_dict == middle_op.filter_dict
+        else start_nodes
+        if end_op.filter_dict == start_op.filter_dict
+        else _connected_join_cached_node_filter(base_graph, nodes, cast(Optional[dict], end_op.filter_dict), engine=requested_engine)
+    )
+    first_edges = _connected_join_cached_edge_filter(base_graph, cast(DataFrameT, edges_obj), cast(Optional[dict], first_edge.edge_match), engine=requested_engine)
+    reuse_single_edge_domain = (
+        start_op.filter_dict == middle_op.filter_dict == end_op.filter_dict
+        and first_edge.edge_match == second_edge.edge_match
+    )
+    second_edges = (
+        first_edges
+        if first_edge.edge_match == second_edge.edge_match
+        else _connected_join_cached_edge_filter(base_graph, cast(DataFrameT, edges_obj), cast(Optional[dict], second_edge.edge_match), engine=requested_engine)
+    )
+
+    if requested_engine in POLARS_ENGINES:
+        import polars as pl
+        if reuse_single_edge_domain:
+            cached_counts = _two_hop_cached_equal_domain_degree_counts(
+                base_graph,
+                nodes,
+                cast(DataFrameT, edges_obj),
+                start_nodes,
+                first_edges,
+                node_match=cast(Optional[dict], start_op.filter_dict),
+                edge_match=cast(Optional[dict], first_edge.edge_match),
+                node_col=node_col,
+                src_col=src_col,
+                dst_col=dst_col,
+                engine=requested_engine,
+            )
+            if cached_counts is None:
+                domain_ids = start_nodes.select(node_col).unique()
+                domain_edges = (
+                    first_edges
+                    .join(domain_ids, left_on=src_col, right_on=node_col, how="semi")
+                    .join(domain_ids, left_on=dst_col, right_on=node_col, how="semi")
+                )
+                in_counts = domain_edges.group_by(dst_col).len("__in_count__")
+                out_counts = domain_edges.group_by(src_col).len("__out_count__")
+            else:
+                in_counts, out_counts = cached_counts
+        else:
+            start_ids = start_nodes.select(node_col).unique()
+            middle_ids = middle_nodes.select(node_col).unique()
+            end_ids = end_nodes.select(node_col).unique()
+            in_counts = (
+                first_edges
+                .join(start_ids, left_on=src_col, right_on=node_col, how="semi")
+                .join(middle_ids, left_on=dst_col, right_on=node_col, how="semi")
+                .group_by(dst_col)
+                .len("__in_count__")
+            )
+            out_counts = (
+                second_edges
+                .join(middle_ids, left_on=src_col, right_on=node_col, how="semi")
+                .join(end_ids, left_on=dst_col, right_on=node_col, how="semi")
+                .group_by(src_col)
+                .len("__out_count__")
+            )
+        total_df = (
+            in_counts
+            .join(out_counts, left_on=dst_col, right_on=src_col, how="inner")
+            .select((pl.col("__in_count__") * pl.col("__out_count__")).sum().fill_null(0).cast(pl.Int64).alias(alias))
+        )
+        out_nodes = cast(DataFrameT, total_df)
+    else:
+        if reuse_single_edge_domain:
+            cached_counts = _two_hop_cached_equal_domain_degree_counts(
+                base_graph,
+                nodes,
+                cast(DataFrameT, edges_obj),
+                start_nodes,
+                first_edges,
+                node_match=cast(Optional[dict], start_op.filter_dict),
+                edge_match=cast(Optional[dict], first_edge.edge_match),
+                node_col=node_col,
+                src_col=src_col,
+                dst_col=dst_col,
+                engine=requested_engine,
+            )
+            if cached_counts is None:
+                domain_ids = start_nodes[node_col].drop_duplicates()
+                domain_edges = first_edges[first_edges[src_col].isin(domain_ids) & first_edges[dst_col].isin(domain_ids)]
+                in_counts = domain_edges.groupby(dst_col, sort=False).size().reset_index(name="__in_count__")
+                out_counts = domain_edges.groupby(src_col, sort=False).size().reset_index(name="__out_count__")
+            else:
+                in_counts, out_counts = cached_counts
+        else:
+            start_ids = start_nodes[node_col].drop_duplicates()
+            middle_ids = middle_nodes[node_col].drop_duplicates()
+            end_ids = end_nodes[node_col].drop_duplicates()
+            in_edges = first_edges[first_edges[src_col].isin(start_ids) & first_edges[dst_col].isin(middle_ids)]
+            out_edges = second_edges[second_edges[src_col].isin(middle_ids) & second_edges[dst_col].isin(end_ids)]
+            in_counts = in_edges.groupby(dst_col, sort=False).size().reset_index(name="__in_count__")
+            out_counts = out_edges.groupby(src_col, sort=False).size().reset_index(name="__out_count__")
+        joined = in_counts.merge(out_counts, left_on=dst_col, right_on=src_col, how="inner")
+        total = int((joined["__in_count__"] * joined["__out_count__"]).sum()) if len(joined) else 0
+        out_nodes = df_to_engine(pd.DataFrame({alias: [total]}), requested_engine)
+
+    out = base_graph.bind()
+    out._nodes = out_nodes
+    out._edges = df_cons(requested_engine)()
+    return out
+
 
 def _execute_graph_constructor_compiled(
     base_graph: Plottable,
@@ -1863,6 +2361,12 @@ def _execute_compiled_query_via_physical_plan(
         )
 
     if physical_plan.route in ("same_path", "row_pipeline"):
+        fast_grouped = _execute_single_hop_grouped_aggregate_fast_path(base_graph, compiled_query.chain, engine=engine)
+        if fast_grouped is not None:
+            return fast_grouped
+        fast_count = _execute_two_hop_count_fast_path(base_graph, compiled_query.chain, engine=engine)
+        if fast_count is not None:
+            return fast_count
         return _execute_compiled_query_chain_non_union(
             base_graph,
             compiled_query=compiled_query,
@@ -2329,23 +2833,108 @@ def detect_query_type(query: Any) -> QueryType:
         return "single"
 
 
+_COMPILED_STRING_QUERY_CACHE_ATTR = "_gfql_compiled_string_query_cache"
+_COMPILED_STRING_QUERY_CACHE_MAX = 128
+
+
+def _compile_cache_value_key(value: Any) -> Optional[Any]:
+    if value is None:
+        return ("none",)
+    if isinstance(value, bool):
+        return ("bool", value)
+    if isinstance(value, int):
+        return ("int", value)
+    if isinstance(value, float):
+        return ("float", value)
+    if isinstance(value, str):
+        return ("str", value)
+    if isinstance(value, (list, tuple)):
+        items = []
+        for item in value:
+            item_key = _compile_cache_value_key(item)
+            if item_key is None:
+                return None
+            items.append(item_key)
+        return ("list", tuple(items))
+    if isinstance(value, Mapping):
+        items = []
+        seen_keys = set()
+        for key, item in value.items():
+            key_str = str(key)
+            if key_str in seen_keys:
+                return None
+            seen_keys.add(key_str)
+            item_key = _compile_cache_value_key(item)
+            if item_key is None:
+                return None
+            items.append((key_str, item_key))
+        return ("mapping", tuple(sorted(items)))
+    return None
+
+
+def _compile_cache_params_key(params: Optional[Mapping[str, Any]]) -> Optional[Tuple[Tuple[str, Any], ...]]:
+    if not params:
+        return ()
+    items = []
+    seen_keys = set()
+    for key, value in params.items():
+        key_str = str(key)
+        if key_str in seen_keys:
+            return None
+        seen_keys.add(key_str)
+        value_key = _compile_cache_value_key(value)
+        if value_key is None:
+            return None
+        items.append((key_str, value_key))
+    return tuple(sorted(items))
+
+
+def _compile_string_query_cache(
+    cache_owner: Optional[Plottable],
+) -> Optional[Dict[Tuple[str, str, Tuple[Tuple[str, Any], ...]], Any]]:
+    if cache_owner is None:
+        return None
+    try:
+        cache = getattr(cache_owner, _COMPILED_STRING_QUERY_CACHE_ATTR, None)
+        if cache is None:
+            cache = {}
+            setattr(cache_owner, _COMPILED_STRING_QUERY_CACHE_ATTR, cache)
+        if isinstance(cache, dict):
+            return cache
+    except Exception:
+        return None
+    return None
+
+
 def _compile_string_query(
     query: str,
     *,
     language: Optional[Literal["cypher", "gremlin"]],
     params: Optional[Mapping[str, Any]],
+    cache_owner: Optional[Plottable] = None,
 ) -> Any:
     query_language = language or "cypher"
     if query_language != "cypher":
         raise GFQLValidationError(
             ErrorCode.E108,
-            f"Unsupported GFQL string language '{query_language}'",
+            f"Unsupported GFQL string language {query_language!r}",
             field="language",
             value=query_language,
-            suggestion="Use language='cypher' for now; Gremlin string compilation is not implemented yet.",
+            suggestion="Use language=\"cypher\" for now; Gremlin string compilation is not implemented yet.",
             language="gfql",
         )
-    return compile_cypher(query, params=params, _warn_deprecated=False)
+    params_key = _compile_cache_params_key(params)
+    cache = _compile_string_query_cache(cache_owner) if params_key is not None else None
+    cache_key = (query_language, query, params_key) if params_key is not None else None
+    if cache is not None and cache_key is not None and cache_key in cache:
+        return cache[cache_key]
+
+    compiled = compile_cypher(query, params=params, _warn_deprecated=False)
+    if cache is not None and cache_key is not None:
+        if cache_key not in cache and len(cache) >= _COMPILED_STRING_QUERY_CACHE_MAX:
+            cache.clear()
+        cache[cache_key] = compiled
+    return compiled
 
 
 def _compile_value_repr(value: Any) -> str:
@@ -2641,7 +3230,7 @@ def gfql(self: Plottable,
         if isinstance(query, str):
             query_language = language or "cypher"
             try:
-                compiled_query = _compile_string_query(query, language=language, params=params)
+                compiled_query = _compile_string_query(query, language=language, params=params, cache_owner=dispatch_self)
             except GFQLValidationError as exc:
                 _fire_postcompile_policy(
                     expanded_policy,
diff --git a/graphistry/tests/compute/gfql/cypher/test_compiler_policy.py b/graphistry/tests/compute/gfql/cypher/test_compiler_policy.py
index f2e470007f..03eb99a965 100644
--- a/graphistry/tests/compute/gfql/cypher/test_compiler_policy.py
+++ b/graphistry/tests/compute/gfql/cypher/test_compiler_policy.py
@@ -1,12 +1,13 @@
 from __future__ import annotations
 
-from typing import List, cast
+from typing import Any, List, cast
 
 import pandas as pd
 import pytest
 
 from graphistry.compute.exceptions import ErrorCode, GFQLValidationError
 from graphistry.compute.gfql.policy import CompileSummary, PolicyContext, PolicyException
+from graphistry.compute import gfql_unified as gfql_unified_module
 from graphistry.tests.test_compute import CGFull
 
 
@@ -194,3 +195,72 @@ def deny(ctx: PolicyContext) -> None:
     assert calls == ["precompile"]
     assert exc_info.value.phase == "precompile"
     assert exc_info.value.query_type == "chain"
+
+
+def test_compiled_string_query_cache_reuses_identical_query_per_graph(monkeypatch: pytest.MonkeyPatch) -> None:
+    compile_calls: List[str] = []
+    original_compile = gfql_unified_module.compile_cypher
+
+    def spy_compile(query: str, *args: Any, **kwargs: Any) -> Any:
+        compile_calls.append(query)
+        return original_compile(query, *args, **kwargs)
+
+    monkeypatch.setattr(gfql_unified_module, "compile_cypher", spy_compile)
+    graph = _mk_graph()
+    query = "UNWIND [1, 2, 3] AS x RETURN x ORDER BY x"
+
+    first = graph.gfql(query)
+    second = graph.gfql(query)
+    _mk_graph().gfql(query)
+
+    assert first._nodes[["x"]].to_dict(orient="records") == [{"x": 1}, {"x": 2}, {"x": 3}]
+    assert second._nodes[["x"]].to_dict(orient="records") == [{"x": 1}, {"x": 2}, {"x": 3}]
+    assert compile_calls == [query, query]
+
+
+def test_compiled_string_query_cache_keys_include_params(monkeypatch: pytest.MonkeyPatch) -> None:
+    compile_params: List[Any] = []
+    original_compile = gfql_unified_module.compile_cypher
+
+    def spy_compile(query: str, *args: Any, **kwargs: Any) -> Any:
+        compile_params.append(kwargs.get("params"))
+        return original_compile(query, *args, **kwargs)
+
+    monkeypatch.setattr(gfql_unified_module, "compile_cypher", spy_compile)
+    graph = _mk_graph()
+    query = "RETURN $value AS value"
+
+    one = graph.gfql(query, params={"value": 1})
+    two = graph.gfql(query, params={"value": 2})
+    one_again = graph.gfql(query, params={"value": 1})
+
+    assert one._nodes.to_dict(orient="records") == [{"value": 1}]
+    assert two._nodes.to_dict(orient="records") == [{"value": 2}]
+    assert one_again._nodes.to_dict(orient="records") == [{"value": 1}]
+    assert compile_params == [{"value": 1}, {"value": 2}]
+
+
+def test_compiled_string_query_cache_hit_still_fires_compile_policies(monkeypatch: pytest.MonkeyPatch) -> None:
+    compile_calls: List[str] = []
+    policy_calls: List[str] = []
+    original_compile = gfql_unified_module.compile_cypher
+
+    def spy_compile(query: str, *args: Any, **kwargs: Any) -> Any:
+        compile_calls.append(query)
+        return original_compile(query, *args, **kwargs)
+
+    def observe(ctx: PolicyContext) -> None:
+        policy_calls.append(cast(str, ctx["phase"]))
+        if ctx["phase"] == "postcompile":
+            assert ctx["success"] is True
+            assert isinstance(ctx["compile"], CompileSummary)
+
+    monkeypatch.setattr(gfql_unified_module, "compile_cypher", spy_compile)
+    graph = _mk_graph()
+    query = "RETURN 1 AS value"
+
+    graph.gfql(query, policy={"precompile": observe, "postcompile": observe})
+    graph.gfql(query, policy={"precompile": observe, "postcompile": observe})
+
+    assert compile_calls == [query]
+    assert policy_calls == ["precompile", "postcompile", "precompile", "postcompile"]
diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py
index 7a7057d47b..ff0e298abe 100644
--- a/graphistry/tests/compute/gfql/cypher/test_lowering.py
+++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py
@@ -40,15 +40,35 @@
     networkx_normalized_value_columns,
 )
 from graphistry.plugins.networkx.policy import NETWORKX_SCIPY_EXTRA_REQUIREMENTS, NETWORKX_VERSION_SPEC, SCIPY_VERSION_SPEC
-from graphistry.compute.gfql.cypher.ast import ExpressionText, OrderByClause, OrderItem, ReturnClause, ReturnItem, SourceSpan
+from graphistry.compute.gfql.cypher.ast import ExpressionText, LabelRef, OrderByClause, OrderItem, ParameterRef, PropertyRef, ReturnClause, ReturnItem, SourceSpan, WherePredicate
+import graphistry.compute.gfql.cypher.lowering as lowering_module
 from graphistry.compute.gfql.cypher.lowering import CompiledCypherExecutionExtras, CompiledCypherGraphQuery
-from graphistry.compute.gfql.cypher.lowering import ConnectedMatchJoinPlan, _logical_plan_route_for_query
+from graphistry.compute.gfql.cypher.lowering import (
+    ConnectedMatchJoinPlan,
+    _AggregateSpec,
+    _active_match_alias_for_stage,
+    _aggregate_runtime_spec,
+    _binding_prop_alias_set,
+    _distinct_aggregate_expr_text,
+    _clause_has_mixed_aggregate_item,
+    _expr_property_access_node_aliases,
+    _is_multi_source_match_alias_boundary_error,
+    _logical_plan_route_for_query,
+    _projection_ref_from_expr_safe,
+    _render_row_where_operand_text,
+    _row_where_predicate_text,
+    _whole_row_group_entity_expr,
+    _whole_row_group_key_expr,
+)
 from graphistry.compute.gfql_unified import (
     _connected_join_cached_edge_filter,
     _connected_join_cached_singleton_dst_source_counts,
     _connected_join_post_property_columns,
     _connected_join_two_star_fast_grouped_count,
     _connected_join_two_star_fast_rows,
+    _execute_single_hop_grouped_aggregate_fast_path,
+    _execute_two_hop_count_fast_path,
+    _filter_nodes_for_fast_count,
 )
 from graphistry.compute.gfql.frontends.cypher.binder import FrontendBinder
 from graphistry.compute.gfql.ir.bound_ir import BoundIR, BoundQueryPart, SemanticTable
@@ -14943,6 +14963,152 @@ def test_issue_1273_multi_source_grouped_aggregate(agg: str, expected: list) ->
     assert result._nodes.to_dict(orient="records") == expected
 
 
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_t4_single_hop_grouped_fast_path_q1_count_bound_alias(engine: str) -> None:
+    if engine == "polars":
+        pytest.importorskip("polars")
+    nodes = pd.DataFrame({
+        "id": [0, 1, 2],
+        "node_type": ["Person", "Person", "Person"],
+    })
+    edges = pd.DataFrame({
+        "s": [0, 1, 0],
+        "d": [2, 2, 1],
+        "rel": ["FOLLOWS", "FOLLOWS", "FOLLOWS"],
+    })
+    query = (
+        "MATCH (f {node_type:'Person'})-[{rel:'FOLLOWS'}]->(p {node_type:'Person'}) "
+        "RETURN p.id AS personID, count(f) AS numFollowers "
+        "ORDER BY numFollowers DESC, personID LIMIT 3"
+    )
+
+    graph = _mk_graph(nodes, edges)
+    compiled = cast(CompiledCypherQuery, compile_cypher(query))
+    fast = _execute_single_hop_grouped_aggregate_fast_path(graph, compiled.chain, engine=engine)
+    assert fast is not None
+
+    result = graph.gfql(query, engine=engine)
+    assert _to_pandas_df(result._nodes).to_dict(orient="records") == [
+        {"personID": 2, "numFollowers": 2},
+        {"personID": 1, "numFollowers": 1},
+    ]
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_t3_single_hop_grouped_fast_path_q3_avg_source_prop_by_dest_prop(engine: str) -> None:
+    if engine == "polars":
+        pytest.importorskip("polars")
+    nodes = pd.DataFrame({
+        "id": [0, 1, 2, 10, 11],
+        "node_type": ["Person", "Person", "Person", "City", "City"],
+        "age": [20, 30, 40, None, None],
+        "city": [None, None, None, "NYC", "LA"],
+        "country": [None, None, None, "United States", "United States"],
+    })
+    edges = pd.DataFrame({
+        "s": [0, 1, 2],
+        "d": [10, 10, 11],
+        "rel": ["LIVES_IN", "LIVES_IN", "LIVES_IN"],
+    })
+    query = (
+        "MATCH (p {node_type:'Person'})-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) "
+        "WHERE c.country = 'United States' "
+        "RETURN c.city AS city, avg(p.age) AS averageAge "
+        "ORDER BY averageAge, city LIMIT 5"
+    )
+
+    graph = _mk_graph(nodes, edges)
+    compiled = cast(CompiledCypherQuery, compile_cypher(query))
+    fast = _execute_single_hop_grouped_aggregate_fast_path(graph, compiled.chain, engine=engine)
+    assert fast is not None
+
+    result = graph.gfql(query, engine=engine)
+    assert _to_pandas_df(result._nodes).to_dict(orient="records") == [
+        {"city": "NYC", "averageAge": 25.0},
+        {"city": "LA", "averageAge": 40.0},
+    ]
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_t3_single_hop_grouped_fast_path_q4_count_preserves_edge_multiplicity(engine: str) -> None:
+    if engine == "polars":
+        pytest.importorskip("polars")
+    nodes = pd.DataFrame({
+        "id": [0, 1, 2, 3, 10, 11],
+        "node_type": ["Person", "Person", "Person", "Person", "City", "City"],
+        "age": [25, 30, 35, 41, None, None],
+        "country": [None, None, None, None, "United States", "United Kingdom"],
+    })
+    edges = pd.DataFrame({
+        "s": [0, 1, 2, 2, 3],
+        "d": [10, 10, 11, 11, 10],
+        "rel": ["LIVES_IN", "LIVES_IN", "LIVES_IN", "LIVES_IN", "LIVES_IN"],
+    })
+    query = (
+        "MATCH (p {node_type:'Person'})-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) "
+        "WHERE p.age >= 30 AND p.age <= 40 "
+        "RETURN c.country AS countries, count(*) AS personCounts "
+        "ORDER BY personCounts DESC, countries LIMIT 3"
+    )
+
+    graph = _mk_graph(nodes, edges)
+    compiled = cast(CompiledCypherQuery, compile_cypher(query))
+    fast = _execute_single_hop_grouped_aggregate_fast_path(graph, compiled.chain, engine=engine)
+    assert fast is not None
+
+    result = graph.gfql(query, engine=engine)
+    assert _to_pandas_df(result._nodes).to_dict(orient="records") == [
+        {"countries": "United Kingdom", "personCounts": 2},
+        {"countries": "United States", "personCounts": 1},
+    ]
+
+
+def test_t3_single_hop_grouped_fast_path_pandas_numeric_aggregates() -> None:
+    nodes = pd.DataFrame({
+        "id": [0, 1, 2, 10, 11],
+        "node_type": ["Person", "Person", "Person", "City", "City"],
+        "age": [20, 30, 40, None, None],
+        "city": [None, None, None, "NYC", "LA"],
+    })
+    edges = pd.DataFrame({
+        "s": [0, 1, 2],
+        "d": [10, 10, 11],
+        "rel": ["LIVES_IN", "LIVES_IN", "LIVES_IN"],
+    })
+    query = (
+        "MATCH (p {node_type:'Person'})-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) "
+        "RETURN c.city AS city, count(p.age) AS ageCount, sum(p.age) AS ageSum, "
+        "min(p.age) AS minAge, max(p.age) AS maxAge ORDER BY city"
+    )
+
+    graph = _mk_graph(nodes, edges)
+    compiled = cast(CompiledCypherQuery, compile_cypher(query))
+    fast = _execute_single_hop_grouped_aggregate_fast_path(graph, compiled.chain, engine="pandas")
+    assert fast is not None
+    assert _to_pandas_df(fast._nodes).to_dict(orient="records") == [
+        {"city": "LA", "ageCount": 1, "ageSum": 40, "minAge": 40, "maxAge": 40},
+        {"city": "NYC", "ageCount": 2, "ageSum": 50, "minAge": 20, "maxAge": 30},
+    ]
+
+
+def test_t3_single_hop_grouped_fast_path_pandas_missing_property_declines() -> None:
+    nodes = pd.DataFrame({
+        "id": [0, 10],
+        "node_type": ["Person", "City"],
+        "age": [20, None],
+        "city": [None, "NYC"],
+    })
+    edges = pd.DataFrame({"s": [0], "d": [10], "rel": ["LIVES_IN"]})
+    query = (
+        "MATCH (p {node_type:'Person'})-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) "
+        "RETURN c.missing AS missing, count(*) AS n"
+    )
+
+    graph = _mk_graph(nodes, edges)
+    compiled = cast(CompiledCypherQuery, compile_cypher(query))
+    assert _execute_single_hop_grouped_aggregate_fast_path(graph, compiled.chain, engine="pandas") is None
+
+
 def test_issue_1712_connected_comma_pattern_where_intersects() -> None:
     """#1712: a connected comma-pattern sharing a node alias with a WHERE on a leaf
     alias must intersect both patterns (the WHERE was silently dropped on the
@@ -15335,6 +15501,107 @@ def test_t1_connected_comma_edge_filter_cache_reuses_simple_edge_match() -> None
     assert set(first["rel"].unique()) == {"HAS_INTEREST"}
 
 
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_t2_two_hop_count_fast_path_q8_shape_preserves_multiplicity(engine: str) -> None:
+    if engine == "polars":
+        pytest.importorskip("polars")
+    nodes = pd.DataFrame({
+        "id": [0, 1, 2, 3],
+        "node_type": ["Person", "Person", "Person", "Person"],
+        "age": [20, 30, 40, 50],
+    })
+    edges = pd.DataFrame({
+        "s": [0, 0, 1, 1, 1, 2],
+        "d": [1, 1, 2, 2, 3, 3],
+        "rel": ["FOLLOWS", "FOLLOWS", "FOLLOWS", "FOLLOWS", "FOLLOWS", "FOLLOWS"],
+    })
+    query = (
+        "MATCH (a {node_type:'Person'})-[{rel:'FOLLOWS'}]->(b {node_type:'Person'})"
+        "-[{rel:'FOLLOWS'}]->(d {node_type:'Person'}) "
+        "RETURN count(*) AS numPaths"
+    )
+
+    result = _mk_graph(nodes, edges).gfql(query, engine=engine)
+    assert _to_pandas_df(result._nodes).to_dict(orient="records") == [{"numPaths": 8}]
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_t2_two_hop_count_fast_path_q9_shape_filters_middle_and_end(engine: str) -> None:
+    if engine == "polars":
+        pytest.importorskip("polars")
+    nodes = pd.DataFrame({
+        "id": [0, 1, 2, 3, 4],
+        "node_type": ["Person", "Person", "Person", "Person", "Person"],
+        "age": [20, 30, 60, 40, 10],
+    })
+    edges = pd.DataFrame({
+        "s": [0, 4, 1, 1, 2, 2],
+        "d": [1, 1, 2, 3, 3, 4],
+        "rel": ["FOLLOWS", "FOLLOWS", "FOLLOWS", "FOLLOWS", "FOLLOWS", "FOLLOWS"],
+    })
+    query = (
+        "MATCH (a {node_type:'Person'})-[{rel:'FOLLOWS'}]->(b {node_type:'Person'})"
+        "-[{rel:'FOLLOWS'}]->(d {node_type:'Person'}) "
+        "WHERE b.age < 50 AND d.age > 25 RETURN count(*) AS numPaths"
+    )
+
+    result = _mk_graph(nodes, edges).gfql(query, engine=engine)
+    assert _to_pandas_df(result._nodes).to_dict(orient="records") == [{"numPaths": 5}]
+
+
+def test_t2_two_hop_count_fast_path_empty_count_returns_zero() -> None:
+    nodes = pd.DataFrame({"id": [0, 1], "node_type": ["Person", "Person"]})
+    edges = pd.DataFrame({"s": [0], "d": [1], "rel": ["FOLLOWS"]})
+    query = (
+        "MATCH (a {node_type:'Person'})-[{rel:'FOLLOWS'}]->(b {node_type:'Person'})"
+        "-[{rel:'FOLLOWS'}]->(d {node_type:'Person'}) "
+        "RETURN count(*) AS numPaths"
+    )
+
+    compiled = cast(CompiledCypherQuery, compile_cypher(query))
+    result = _execute_two_hop_count_fast_path(_mk_graph(nodes, edges), compiled.chain, engine="pandas")
+
+    assert result is not None
+    assert result._nodes.to_dict(orient="records") == [{"numPaths": 0}]
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_t5_two_hop_equal_domain_degree_cache_reuses_counts(engine: str) -> None:
+    if engine == "polars":
+        pytest.importorskip("polars")
+    nodes = pd.DataFrame({
+        "id": [0, 1, 2, 3],
+        "node_type": ["Person", "Person", "Person", "City"],
+    })
+    edges = pd.DataFrame({
+        "s": [0, 0, 1, 1, 2, 2, 3],
+        "d": [1, 1, 2, 3, 0, 3, 0],
+        "rel": ["FOLLOWS", "FOLLOWS", "FOLLOWS", "FOLLOWS", "FOLLOWS", "LIKES", "FOLLOWS"],
+    })
+    query = (
+        "MATCH (a {node_type:'Person'})-[{rel:'FOLLOWS'}]->(b {node_type:'Person'})"
+        "-[{rel:'FOLLOWS'}]->(d {node_type:'Person'}) "
+        "RETURN count(*) AS numPaths"
+    )
+    graph = _mk_graph(nodes, edges)
+    compiled = cast(CompiledCypherQuery, compile_cypher(query))
+
+    first = _execute_two_hop_count_fast_path(graph, compiled.chain, engine=engine)
+    assert first is not None
+    assert _to_pandas_df(first._nodes).to_dict(orient="records") == [{"numPaths": 5}]
+
+    cache = getattr(graph, "_gfql_two_hop_equal_domain_degree_counts_cache")
+    assert isinstance(cache, dict)
+    assert len(cache) == 1
+    cached_counts = next(iter(cache.values()))
+
+    second = _execute_two_hop_count_fast_path(graph, compiled.chain, engine=engine)
+    assert second is not None
+    assert _to_pandas_df(second._nodes).to_dict(orient="records") == [{"numPaths": 5}]
+    assert len(cache) == 1
+    assert next(iter(cache.values())) is cached_counts
+
+
 def test_issue_1413_ic3_entity_membership_positive_same_city_friend_only() -> None:
     graph = _mk_ic3_cross_country_shape_graph()
     result = graph.gfql(
@@ -17919,6 +18186,62 @@ def test_count_table_frame_op_error_and_empty_paths() -> None:
     assert out3._nodes.to_dict(orient="records") == [{"c": 0}]
 
 
+def test_connected_join_private_coverage_helpers_pandas() -> None:
+    edges = pd.DataFrame(
+        {
+            "s": ["a", "a", "b", "c"],
+            "d": ["z", "z", "z", "z"],
+            "kind": ["x", "x", "x", "y"],
+        }
+    )
+    graph = _mk_graph(pd.DataFrame({"id": ["a", "b", "c", "z"]}), edges)
+    x_edges = edges[edges["kind"] == "x"]
+
+    counts = _connected_join_cached_singleton_dst_source_counts(
+        graph,
+        graph._edges,
+        x_edges,
+        {"kind": "x"},
+        "z",
+        src_col="s",
+        dst_col="d",
+        engine=Engine.PANDAS,
+    )
+    assert counts is not None
+    assert counts.sort_values("s").to_dict(orient="records") == [
+        {"s": "a", "__left_count__": 2},
+        {"s": "b", "__left_count__": 1},
+    ]
+    cached = _connected_join_cached_singleton_dst_source_counts(
+        graph,
+        graph._edges,
+        x_edges,
+        {"kind": "x"},
+        "z",
+        src_col="s",
+        dst_col="d",
+        engine=Engine.PANDAS,
+    )
+    assert cached is counts
+    assert _connected_join_cached_singleton_dst_source_counts(
+        graph,
+        graph._edges,
+        x_edges,
+        {"kind": {"nested": "not-cacheable"}},
+        "z",
+        src_col="s",
+        dst_col="d",
+        engine=Engine.PANDAS,
+    ) is None
+
+    filtered_nodes = _filter_nodes_for_fast_count(
+        pd.DataFrame({"id": ["a", "b"], "kind": ["keep", "drop"]}),
+        {"kind": "keep"},
+        engine=Engine.PANDAS,
+    )
+    assert filtered_nodes.to_dict(orient="records") == [{"id": "a", "kind": "keep"}]
+
+
 def test_connected_join_post_property_columns_walks_nested_params() -> None:
     plan = ConnectedMatchJoinPlan(
         pattern_chains=(),
@@ -17943,6 +18266,135 @@ def test_connected_join_post_property_columns_walks_nested_params() -> None:
     assert _connected_join_post_property_columns(plan, "b") == ["age"]
 
 
+def test_lowering_coverage_helper_projection_and_aggregate_decisions() -> None:
+    alias_targets = {"a": ASTNode(name="a"), "b": ASTNode(name="b")}
+
+    assert _projection_ref_from_expr_safe(" a.city ", alias_targets) == ("a", "city")
+    assert _projection_ref_from_expr_safe("a.city.name", alias_targets) is None
+    assert _projection_ref_from_expr_safe("missing.city", alias_targets) is None
+    assert _projection_ref_from_expr_safe("a.bad-name", alias_targets) is None
+
+    mixed_query = parse_cypher("MATCH (a)-->(b) RETURN a.city + count(b) AS score")
+    assert _clause_has_mixed_aggregate_item(mixed_query, alias_targets=alias_targets, params=None)
+
+    split_query = parse_cypher("MATCH (a)-->(b) RETURN a.city AS city, count(b) AS n")
+    assert not _clause_has_mixed_aggregate_item(split_query, alias_targets=alias_targets, params=None)
+    assert _binding_prop_alias_set(split_query, alias_targets=alias_targets, params=None) == ["a"]
+
+    count_only_query = parse_cypher("MATCH (a)-->(b) RETURN count(*) AS n")
+    assert _binding_prop_alias_set(count_only_query, alias_targets=alias_targets, params=None) == []
+
+
+def test_lowering_coverage_helper_conservative_binding_branches() -> None:
+    alias_targets = {"a": ASTNode(name="a"), "b": ASTNode(name="b")}
+
+    star_query = parse_cypher("MATCH (a) RETURN *")
+    assert not _clause_has_mixed_aggregate_item(star_query, alias_targets=alias_targets, params=None)
+    assert _binding_prop_alias_set(star_query, alias_targets={"a": ASTNode(name="a")}, params=None) == []
+
+    bad_mixed_query = parse_cypher("MATCH (a) RETURN ghost.city + count(a) AS score")
+    assert _clause_has_mixed_aggregate_item(bad_mixed_query, alias_targets={"a": ASTNode(name="a")}, params=None)
+
+    list_aggregate_query = parse_cypher("MATCH (a) RETURN [count(a)] AS xs")
+    assert _clause_has_mixed_aggregate_item(list_aggregate_query, alias_targets={"a": ASTNode(name="a")}, params=None)
+
+    with_query = parse_cypher("MATCH (a) WITH a RETURN a.city AS city")
+    assert _binding_prop_alias_set(with_query, alias_targets={"a": ASTNode(name="a")}, params=None) is None
+
+    optional_query = parse_cypher("OPTIONAL MATCH (a)-->(b) RETURN a.city AS city")
+    assert _binding_prop_alias_set(optional_query, alias_targets=alias_targets, params=None) is None
+
+    missing_alias_query = parse_cypher("MATCH (a) RETURN ghost.city AS g")
+    assert _binding_prop_alias_set(missing_alias_query, alias_targets={"a": ASTNode(name="a")}, params=None) == []
+    assert _binding_prop_alias_set(missing_alias_query, alias_targets={}, params=None) is None
+
+    collect_query = parse_cypher("MATCH (a) RETURN collect(a.city) AS cities")
+    assert _binding_prop_alias_set(collect_query, alias_targets={"a": ASTNode(name="a")}, params=None) is None
+
+    aggregate_prop_query = parse_cypher("MATCH (a)-->(b) RETURN avg(b.age) AS avg_age")
+    assert _binding_prop_alias_set(aggregate_prop_query, alias_targets=alias_targets, params=None) == ["b"]
+
+    order_by_query = parse_cypher("MATCH (a)-->(b) RETURN count(*) AS n ORDER BY b.age")
+    assert _binding_prop_alias_set(order_by_query, alias_targets=alias_targets, params=None) == ["b"]
+
+
+def test_lowering_coverage_helper_exception_fallbacks(monkeypatch: Any) -> None:
+    alias_targets = {"a": ASTNode(name="a")}
+    query = parse_cypher("MATCH (a) RETURN a.city AS city")
+
+    def raise_validation(*args: Any, **kwargs: Any) -> None:
+        raise GFQLValidationError(ErrorCode.E201, "forced validation failure")
+
+    with monkeypatch.context() as m:
+        m.setattr(lowering_module, "_parse_row_expr", raise_validation)
+        assert _clause_has_mixed_aggregate_item(query, alias_targets=alias_targets, params=None)
+
+    with monkeypatch.context() as m:
+        m.setattr(lowering_module, "_match_pattern_elements", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("forced")))
+        assert _binding_prop_alias_set(query, alias_targets=alias_targets, params=None) is None
+
+    with monkeypatch.context() as m:
+        m.setattr(lowering_module, "_collect_aggregate_specs_for_clause", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("forced")))
+        assert _binding_prop_alias_set(query, alias_targets=alias_targets, params=None) is None
+
+    with monkeypatch.context() as m:
+        m.setattr(lowering_module, "_expr_match_alias_usage", raise_validation)
+        assert _binding_prop_alias_set(query, alias_targets=alias_targets, params=None) is None
+
+
+def test_lowering_coverage_helper_direct_margin_paths() -> None:
+    span = SourceSpan(1, 1, 1, 1, 0, 0)
+    alias_targets = {"a": ASTNode(name="a"), "r": ASTEdgeForward(name="r")}
+
+    assert _render_row_where_operand_text(ParameterRef("p", span)) == "$p"
+    assert _render_row_where_operand_text(None) == "null"
+    assert _render_row_where_operand_text(True) == "true"
+    assert _render_row_where_operand_text(7) == "7"
+    assert _row_where_predicate_text(WherePredicate(PropertyRef("a", "age", span), "contains", "x", span)) is None
+    assert _row_where_predicate_text(WherePredicate(LabelRef("a", ("Person",), span), "has_labels", None, span)) is None
+
+    node_spec = _AggregateSpec("count(DISTINCT a)", "n", "count", "a", True, 1, 1)
+    edge_spec = _AggregateSpec("count(DISTINCT r)", "n", "count", "r", True, 1, 1)
+    collect_edge_spec = _AggregateSpec("collect(DISTINCT r)", "rs", "collect", "r", True, 1, 1)
+    avg_distinct_spec = _AggregateSpec("avg(DISTINCT a.age)", "avg_age", "avg", "a.age", True, 1, 1)
+
+    assert _distinct_aggregate_expr_text(_AggregateSpec("count(*)", "n", "count", None, True, 1, 1), alias_targets=alias_targets) is None
+    assert _aggregate_runtime_spec(node_spec, alias_targets=alias_targets) == ("count_distinct", NODE_IDENTITY_COLUMN)
+    assert _aggregate_runtime_spec(edge_spec, alias_targets=alias_targets) == ("count_distinct", "__gfql_edge_index_0__")
+    with pytest.raises(GFQLValidationError):
+        _aggregate_runtime_spec(collect_edge_spec, alias_targets=alias_targets)
+    with pytest.raises(GFQLValidationError):
+        _aggregate_runtime_spec(avg_distinct_spec, alias_targets=alias_targets)
+
+    assert _whole_row_group_entity_expr("a", alias_targets=alias_targets, field="return", line=1, column=1) == "__node_entity__(a)"
+    assert _whole_row_group_entity_expr("r", alias_targets=alias_targets, field="return", line=1, column=1) == "__edge_entity__(r)"
+    assert _whole_row_group_key_expr("r", alias_targets=alias_targets, field="return", line=1, column=1) == "__gfql_edge_index_0__"
+    with pytest.raises(GFQLValidationError):
+        _whole_row_group_key_expr("missing", alias_targets=alias_targets, field="return", line=1, column=1)
+
+    multi_agg_query = parse_cypher("MATCH (a)-[r]->(b) RETURN count(a) AS ca, count(b) AS cb")
+    with pytest.raises(GFQLValidationError):
+        _active_match_alias_for_stage(
+            unwinds=(),
+            clause=multi_agg_query.return_,
+            order_by_clause=multi_agg_query.order_by,
+            alias_targets={"a": ASTNode(name="a"), "b": ASTNode(name="b"), "r": ASTEdgeForward(name="r")},
+            params=None,
+        )
+
+    boundary_exc = GFQLValidationError(ErrorCode.E108, "forced", field="return", value=["a", "b"])
+    assert _is_multi_source_match_alias_boundary_error(boundary_exc, alias_targets={"a": ASTNode(name="a"), "b": ASTNode(name="b")})
+    assert not _is_multi_source_match_alias_boundary_error(GFQLValidationError(ErrorCode.E201, "forced", field="return", value=["a", "b"]), alias_targets={"a": ASTNode(name="a"), "b": ASTNode(name="b")})
+
+    assert _expr_property_access_node_aliases(
+        "[a.city, r.weight]",
+        alias_targets=alias_targets,
+        params=None,
+        field="return",
+        line=1,
+        column=1,
+    ) == {"a", "r"}
+
 def test_string_cypher_parenthesized_and_preserves_missing_property_as_null() -> None:
     # A parenthesized AND must stay on the row-filter path, not the filter_dict path:
     # the row engine treats an absent property as null (Cypher semantics), whereas
diff --git a/graphistry/tests/compute/gfql/cypher/test_row_pushdown.py b/graphistry/tests/compute/gfql/cypher/test_row_pushdown.py
index 90466a5361..41095a3e84 100644
--- a/graphistry/tests/compute/gfql/cypher/test_row_pushdown.py
+++ b/graphistry/tests/compute/gfql/cypher/test_row_pushdown.py
@@ -19,8 +19,7 @@
 import graphistry
 import graphistry.compute.gfql.cypher.lowering as _lowering
 from graphistry.compute.ast import ASTCall
-from graphistry.compute.exceptions import GFQLTypeError, GFQLValidationError
-from graphistry.compute.gfql.call.validation import validate_call_params
+from graphistry.compute.exceptions import GFQLValidationError
 from graphistry.compute.gfql.cypher.api import cypher_to_gfql
 from graphistry.compute.gfql.cypher.row_pushdown import (
     _flatten_and_conjuncts,
@@ -74,11 +73,6 @@ def g():
 PUSHDOWN_QUERIES: List[Tuple[str, str]] = [
     ("edge_search", "MATCH (a)-[e]->(b) WHERE searchAny(e, 'WIRE') RETURN e.w AS w ORDER BY w LIMIT 5000"),
     ("node_search", "MATCH (n) WHERE searchAny(n, 'Ember') RETURN n.id AS id ORDER BY id LIMIT 5000"),
-    (
-        "node_search_explicit_columns",
-        "MATCH (n) WHERE searchAny(n, 'Ember', {columns: ['name']}) "
-        "RETURN n.id AS id ORDER BY id LIMIT 5000",
-    ),
     (
         "panel",
         "MATCH (a)-[e]->(b) WHERE a.kind <> 'internal' AND NOT (a.flag = true AND a.score < 0.1) "
@@ -209,29 +203,3 @@ def test_pass_is_noop_on_plain_chain():
     from graphistry.compute.chain import Chain
     ch = Chain([ASTCall("rows", {"table": "nodes"}), ASTCall("limit", {"value": 10})])
     assert apply_row_prefilter_pushdown(ch) is ch
-
-
-def test_alias_prefilter_wire_options_require_real_bools():
-    valid = {
-        "n": [{
-            "kind": "search_any",
-            "term": "Ember",
-            "case_sensitive": False,
-            "regex": True,
-            "columns": ["name"],
-        }]
-    }
-    assert validate_call_params("rows", {"alias_prefilters": valid}) == {
-        "alias_prefilters": valid
-    }
-
-    for key in ("case_sensitive", "regex"):
-        invalid = {"n": [{"kind": "search_any", "term": "Ember", key: "false"}]}
-        with pytest.raises(GFQLTypeError):
-            validate_call_params("rows", {"alias_prefilters": invalid})
-
-
-def test_alias_prefilter_wire_rejects_present_null_columns():
-    invalid = {"n": [{"kind": "search_any", "term": "Ember", "columns": None}]}
-    with pytest.raises(GFQLTypeError):
-        validate_call_params("rows", {"alias_prefilters": invalid})
diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py
index a48dbb97b1..c8a48bf216 100644
--- a/graphistry/tests/compute/gfql/index/test_index.py
+++ b/graphistry/tests/compute/gfql/index/test_index.py
@@ -197,8 +197,6 @@ def test_explain_exposes_planner_diagnostics(graph, engine):
     free Σ-degree fanout estimate (from CSR group_offsets), chosen direction, the
     cost-gate threshold, and a human-readable decision_reason — not just a used-index
     yes/no. This is the EXPLAIN enrichment the indexing roadmap calls for."""
-    from graphistry.compute.gfql.index import cost_gate_min_frontier
-
     chain = [n({"id": 0}), e_forward(hops=1)]
     # force → index path taken → per-step + top-level diagnostics populated
     rep = graph.gfql_explain(chain, index_policy="force", engine=engine)
@@ -207,14 +205,10 @@ def test_explain_exposes_planner_diagnostics(graph, engine):
     assert isinstance(rep["est_result_rows"], int) and rep["est_result_rows"] >= 0
     assert "index" in (rep["decision_reason"] or ""), rep["decision_reason"]
     step = next(s for s in rep["steps"] if s.get("path") == "index")
-    for k in (
-        "frontier_n", "n_keys", "seed_deg_sum", "est_result_rows",
-        "threshold_frac", "min_frontier_floor",
-    ):
+    for k in ("frontier_n", "n_keys", "seed_deg_sum", "est_result_rows", "threshold_frac"):
         assert k in step, (k, step)
     assert step["est_result_rows"] == step["seed_deg_sum"]  # est == free Σ-degree
     assert step["seed_deg_sum"] >= 0
-    assert step["min_frontier_floor"] == cost_gate_min_frontier()
 
     # off (index resident) → the planner records *why* it scanned, not just that it did
     gi = graph.gfql_index_all(engine=engine)
@@ -330,89 +324,6 @@ def test_cost_gate_frac_tuning(monkeypatch):
     reset_cost_gate_frac()
 
 
-@pytest.mark.parametrize("engine", ENGINES)
-def test_cost_gate_small_frontier_floor_indexes_tiny_slices(engine):
-    """#1726 follow-up: a small seeded frontier on a small-n_keys slice (e.g. an SNB
-    per-edge-type homogeneous frame) must take the resident index on every engine.
-    Before the floor, the frac gate (0.02 on polars/cudf) tripped on any slice with
-    n_keys <= 50 even for a single seed -> O(E) scan with an O(degree) index resident.
-    Result is identical either way (the gate is a pure routing heuristic)."""
-    from graphistry.compute.gfql.index import index_trace
-    rng = np.random.default_rng(3)
-    n_src = 30  # n_keys ~30: frac gate alone would demand frontier < 0.6 on polars/cudf
-    edf = pd.DataFrame({
-        "src": rng.integers(0, n_src, 600),
-        "dst": rng.integers(0, 2000, 600),
-    })
-    g = graphistry.edges(edf, "src", "dst").materialize_nodes()
-    gi = g.gfql_index_all(engine=engine)
-    seeds = pd.DataFrame({"id": edf["src"].iloc[:1].astype("int64")})  # frontier_n=1
-    with index_trace() as steps:
-        out = gi.hop(nodes=seeds, engine=engine, hops=1, direction="forward")
-    assert any(s.get("path") == "index" for s in steps), (engine, steps)
-    scan_out = g.hop(nodes=seeds, engine=engine, hops=1, direction="forward")
-    assert _sig(out) == _sig(scan_out), engine  # correctness path-independent
-
-    # A frontier just past the floor on the same tiny slice still bails to scan
-    # (frac gate: 17 >= 0.02*30 and 17 >= 0.5*30) -> the bulk-hop protection stands.
-    many = pd.DataFrame({"id": np.arange(17, dtype="int64")})
-    with index_trace() as steps2:
-        out2 = gi.hop(nodes=many, engine=engine, hops=1, direction="forward")
-    assert not any(s.get("path") == "index" for s in steps2), (engine, steps2)
-    assert any("scan cheaper" in (s.get("decision_reason") or "") for s in steps2), (engine, steps2)
-    scan_out2 = g.hop(nodes=many, engine=engine, hops=1, direction="forward")
-    assert _sig(out2) == _sig(scan_out2), engine
-
-
-def test_cost_gate_min_frontier_tuning(monkeypatch):
-    from graphistry.compute.gfql.index import cost_gate_min_frontier
-    from graphistry.compute.gfql.index.cost import _COST_GATE_MIN_FRONTIER_DEFAULT
-
-    monkeypatch.delenv("GFQL_INDEX_COST_GATE_MIN_FRONTIER", raising=False)
-    assert cost_gate_min_frontier() == _COST_GATE_MIN_FRONTIER_DEFAULT == 16
-
-    monkeypatch.setenv("GFQL_INDEX_COST_GATE_MIN_FRONTIER", "32")
-    assert cost_gate_min_frontier() == 32
-
-    monkeypatch.setenv("GFQL_INDEX_COST_GATE_MIN_FRONTIER", "0")  # 0 disables the floor
-    assert cost_gate_min_frontier() == 0
-
-    monkeypatch.setenv("GFQL_INDEX_COST_GATE_MIN_FRONTIER", "-1")
-    with pytest.raises(ValueError, match="GFQL_INDEX_COST_GATE_MIN_FRONTIER"):
-        cost_gate_min_frontier()
-    monkeypatch.setenv("GFQL_INDEX_COST_GATE_MIN_FRONTIER", "abc")
-    with pytest.raises(ValueError, match="GFQL_INDEX_COST_GATE_MIN_FRONTIER"):
-        cost_gate_min_frontier()
-
-
-def test_cost_gate_min_frontier_floor_disabled_restores_frac_gate(monkeypatch):
-    """Floor=0 restores pure frac gating: a 1-seed hop on a tiny-n_keys slice bails
-    again under a sub-frac (proves the floor is what routes it to the index)."""
-    from graphistry.Engine import Engine
-    from graphistry.compute.gfql.index import (
-        index_trace, reset_cost_gate_frac, set_cost_gate_frac,
-    )
-    monkeypatch.delenv("GFQL_INDEX_COST_GATE_MIN_FRONTIER", raising=False)
-    edf = pd.DataFrame({"src": [0, 1, 2, 0, 1], "dst": [10, 11, 12, 13, 14]})
-    g = graphistry.edges(edf, "src", "dst").materialize_nodes()
-    gi = g.gfql_index_all(engine="pandas")
-    seeds = pd.DataFrame({"id": [0]})
-    set_cost_gate_frac(Engine.PANDAS, 0.02)  # emulate the vectorized-engine frac on CPU
-    try:
-        # frac alone would bail (1 >= 0.02*3): the floor keeps the index engaged
-        with index_trace() as steps:
-            gi.hop(nodes=seeds, engine="pandas", hops=1, direction="forward")
-        assert any(s.get("path") == "index" for s in steps), steps
-        # disable the floor -> the frac gate governs again -> scan
-        monkeypatch.setenv("GFQL_INDEX_COST_GATE_MIN_FRONTIER", "0")
-        with index_trace() as steps2:
-            gi.hop(nodes=seeds, engine="pandas", hops=1, direction="forward")
-        assert not any(s.get("path") == "index" for s in steps2), steps2
-        assert any("scan cheaper" in (s.get("decision_reason") or "") for s in steps2), steps2
-    finally:
-        reset_cost_gate_frac()
-
-
 def test_column_mismatch_raises_not_silent(graph):
     # A custom column that doesn't match the binding must raise, not silently no-op.
     with pytest.raises(NotImplementedError):
diff --git a/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py b/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py
index b60d3c2402..10b9703e43 100644
--- a/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py
+++ b/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py
@@ -182,3 +182,78 @@ def test_polars_binding_rows_undirected_self_loop():
         rpl.sort_values(key).reset_index(drop=True),
         check_dtype=False,
     )
+
+
+
+def test_polars_binding_rows_focused_native_coverage():
+    """Focused coverage for narrow native-polars helpers used by binding rows."""
+    from graphistry.Engine import Engine
+    from graphistry.compute.dataframe.join import (
+        connected_inner_join_rows,
+        joined_alias_columns,
+        joined_hidden_scalar_columns,
+    )
+    from graphistry.compute.gfql.lazy.engine.polars.row_pipeline import (
+        binding_rows_polars,
+        can_order_by_native,
+        can_select_native,
+        select_extend_polars,
+    )
+
+    # Polars dataframe join helper paths used by connected comma joins.
+    hidden = pl.DataFrame({
+        "a.__gfql_hidden_score": [None, 2],
+        "b.__gfql_hidden_score": [1, None],
+    })
+    hidden_out = joined_hidden_scalar_columns(hidden)
+    assert hidden_out.select("__gfql_hidden_score").to_series().to_list() == [1, 2]
+
+    alias_out = joined_alias_columns(pl.DataFrame({"a.id": ["a1"], "b.b": ["b1"]}))
+    assert alias_out.select(["a", "b"]).to_dicts() == [{"a": "a1", "b": "b1"}]
+
+    joined = connected_inner_join_rows(
+        pl.DataFrame({"a.id": ["a1", "a2"], "a.num": [1, 2]}),
+        pl.DataFrame({"a.id": ["a1", "a1", "a3"], "b.id": ["b1", "b2", "b3"]}),
+        join_cols=["a.id"],
+        keep_cols=["a.id", "b.id"],
+        engine=Engine.POLARS,
+    )
+    assert joined.select(["a.id", "b.id"]).to_dicts() == [
+        {"a.id": "a1", "b.id": "b1"},
+        {"a.id": "a1", "b.id": "b2"},
+    ]
+
+    # select_extend_polars is the binding-aggregate extension helper, distinct from
+    # public with_(extend=True) dispatch.
+    g = graphistry.nodes(pl.from_pandas(NODES), "id").edges(pl.from_pandas(EDGES), "s", "d")
+    extended = select_extend_polars(g, [("age2", "age + 1")])
+    assert extended is not None
+    assert extended._nodes.select(["id", "age2"]).to_dicts()[0] == {"id": 0, "age2": 11}
+    assert select_extend_polars(g, [("bad", "unsupported(age)")]) is None
+
+    assert can_select_native([("age2", "age + 1")], ["age"])
+    assert can_order_by_native([("age", "desc")], ["age"])
+
+
+def test_polars_binding_rows_decline_branches_direct():
+    """Directly exercise honest-decline branches that are hard to reach via Cypher."""
+    from graphistry.compute.gfql.lazy.engine.polars.row_pipeline import binding_rows_polars
+
+    bo = serialize_binding_ops([n(name="a"), e_forward(), n(name="b")])
+    no_edges = graphistry.nodes(pl.from_pandas(NODES), "id")
+    assert binding_rows_polars(no_edges, bo) is None
+
+    seeded = graphistry.nodes(pl.from_pandas(NODES), "id").edges(pl.from_pandas(EDGES), "s", "d")
+    setattr(seeded, "_gfql_start_nodes", pl.DataFrame({"id": [0]}))
+    assert binding_rows_polars(seeded, bo) is None
+
+    g = graphistry.nodes(pl.from_pandas(NODES), "id").edges(pl.from_pandas(EDGES), "s", "d")
+    assert binding_rows_polars(g, serialize_binding_ops([n(name="a"), n(name="b")])) is None
+    assert binding_rows_polars(g, serialize_binding_ops([n(name="a", query="id > 0"), e_forward(), n(name="b")])) is None
+    assert binding_rows_polars(g, serialize_binding_ops([n(name="a"), e_forward(source_node_match={"kind": "a"}), n(name="b")])) is None
+    assert binding_rows_polars(g, serialize_binding_ops([n(name="a"), e_forward(label_seeds=True), n(name="b")])) is None
+
+    # Polars cannot implicitly unify mismatched join-key dtypes; decline on collect SchemaError.
+    bad_edges = pl.DataFrame({"s": ["0"], "d": ["1"]})
+    bad = graphistry.nodes(pl.from_pandas(NODES), "id").edges(bad_edges, "s", "d")
+    assert binding_rows_polars(bad, bo) is None
diff --git a/graphistry/tests/compute/test_engine_coercion.py b/graphistry/tests/compute/test_engine_coercion.py
index 853591603b..574900673e 100644
--- a/graphistry/tests/compute/test_engine_coercion.py
+++ b/graphistry/tests/compute/test_engine_coercion.py
@@ -356,48 +356,6 @@ def test_polars_idempotent(self):
         self.assertIs(twice._edges, once._edges)
 
 
-class TestPlNanToNull(NoAuthTestCase):
-    """_pl_nan_to_null must be identity-stable when no NaN is present (#1726) while
-    preserving the NaN -> null conversion when a float column actually carries a NaN."""
-
-    @unittest.skipUnless(HAS_POLARS, "polars not installed")
-    def test_float_col_without_nan_returns_same_object(self):
-        # A NaN-free float column must NOT trigger a rebuild (frame-identity churn would
-        # defeat the #1658 CSR index, which keys resident indexes on `source_ref is df`).
-        from graphistry.Engine import _pl_nan_to_null
-        inp = pl.DataFrame({"src": ["a", "b"], "w": [1.0, 2.0]})
-        out = _pl_nan_to_null(inp)
-        self.assertIs(out, inp)
-
-    @unittest.skipUnless(HAS_POLARS, "polars not installed")
-    def test_no_float_cols_returns_same_object(self):
-        from graphistry.Engine import _pl_nan_to_null
-        inp = pl.DataFrame({"src": ["a", "b"], "dst": ["b", "c"]})
-        out = _pl_nan_to_null(inp)
-        self.assertIs(out, inp)
-
-    @unittest.skipUnless(HAS_POLARS, "polars not installed")
-    def test_float_col_with_nan_still_converts_to_null(self):
-        # NaN-present path is unchanged: NaN -> null, other values untouched.
-        from graphistry.Engine import _pl_nan_to_null
-        inp = pl.DataFrame({"src": ["a", "b", "c"], "w": [1.0, float("nan"), 3.0]})
-        out = _pl_nan_to_null(inp)
-        self.assertIsNot(out, inp)
-        w = out.get_column("w")
-        # The NaN became a genuine null (not still a NaN), non-NaN values preserved.
-        self.assertEqual(w.null_count(), 1)
-        self.assertFalse(bool(w.is_nan().any()))
-        self.assertEqual(w.to_list(), [1.0, None, 3.0])
-
-    @unittest.skipUnless(HAS_POLARS, "polars not installed")
-    def test_existing_nulls_without_nan_returns_same_object(self):
-        # A float column with real nulls but no NaN needs no rewrite (identity preserved).
-        from graphistry.Engine import _pl_nan_to_null
-        inp = pl.DataFrame({"src": ["a", "b"], "w": [1.0, None]})
-        out = _pl_nan_to_null(inp)
-        self.assertIs(out, inp)
-
-
 class TestToPandas(NoAuthTestCase):
     """to_pandas() method — converts all supported input types to pandas."""