From ee510f78f2dce696c3bf113b70e4fb6e80cfa39b Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 12:26:52 -0700
Subject: [PATCH 01/16] perf(gfql): kill per-call O(E) terms in seeded chains
 (keep-mask + augmentation caches)

IS1 profiling found two linear-with-E costs per seeded Cypher call on a
resident graph: the index path's simple-equality keep-mask rebuilt via a full
column compare (eq_str), and the polars chain's per-call with_row_index O(E)
copy whose fresh identity also defeated downstream id-keyed caches. Both now
recycle-safe caches keyed on the resident frame (weakref.finalize eviction,
the _pl_nan_to_null pattern). Profiled: both terms eliminated; 274 tests pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 CHANGELOG.md                                  |  1 +
 graphistry/compute/gfql/index/traverse.py     | 29 ++++++++++--
 .../compute/gfql/lazy/engine/polars/chain.py  | 27 ++++++++++-
 .../gfql/lazy/engine/polars/row_pipeline.py   | 46 ++++++++++++++++++-
 4 files changed, 98 insertions(+), 5 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index ef6f5e6095..5199bc6d67 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,7 @@ 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).
 
 ### Fixed
+- **GFQL seeded Cypher/chain queries no longer pay two per-call O(E) costs on a resident graph (typed keep-mask + edge-frame augmentation caches)**: profiling the official LDBC IS1 point query on polars found the linear-with-edge-count terms that made Cypher-surface latency GROW with scale: (1) the #1658 index path rebuilt its simple-equality edge keep-mask via a full column compare every seeded hop (`_build_edge_keep_mask` — the profiled `eq_str` O(E) scan), and (2) the polars chain executor rebuilt `with_row_index` on the edge frame every call — an O(E) copy whose fresh object identity ALSO defeated every downstream id-keyed cache. Both are now recycle-safe caches keyed on the resident frame (weakref.finalize eviction, the proven `_pl_nan_to_null` clean-cache pattern): one augmented frame per resident edge frame (stable identity for all downstream caches) and one keep-mask per (frame, engine, scalar edge_match). Profiled: both O(E) terms eliminated from the warm path (`eq_str`/`with_row_index` gone). Parity: 274 tests (index 4-engine + polars conformance) unchanged.
 - **GFQL polars engine natively runs whole-entity aggregation Cypher (LDBC IC4 shape): HAS_<Label> destination disambiguation + entity identity-key resolution + `key_prefixes` group-by expansion**: three previously honest-NIE polars lowerings, each an exact pandas-parity port: (1) `binding_rows_polars` now applies pandas' `_gfql_disambiguate_has_edge_destination_nodes` candidate-domain rule (an UNLABELED next-node op after a `HAS_<Label>`-typed edge narrows to that label ONLY when candidate node ids collide across labels) instead of declining every labeled-destination pattern; (2) `alias.__gfql_node_id__` (the #1650 whole-entity identity key the aggregation lowering groups by) resolves to the bare `alias` id column (the polars bindings table intentionally omits pandas' join-residue columns — the bare alias column IS the identity key); (3) `group_by(key_prefixes=...)` expands every `<prefix>*` row-table column into the key set exactly like pandas (functionally dependent on the identity key — group sizes unchanged). Result: the official LDBC IC4 (new-topics) query — comma-MATCH, `WITH DISTINCT`, CASE projections, whole-entity `sum` aggregation, HAVING-style `WHERE` — runs natively on polars, parity-exact (harness-verified vs expected rows at SF0.1: ok, 127.8 ms). Regression test `test_polars_rows_entity_groupby.py` (pandas oracle + polars parity on a discriminating fixture). Remaining polars `rows` declines are undirected bounded var-length (IC6/IC11) and shortestPath/zero-hop-unbounded (by design).
 - **GFQL connected OPTIONAL MATCH no longer crashes on polars frames (engine-polymorphic seed extraction)**: `_optional_arm_start_nodes` (gfql_unified.py) applied pandas-only ops (`.dropna()`/`.drop_duplicates()`/`.rename(columns=)`/boolean-mask indexing) to the joined binding rows, so an IS7-shaped `MATCH ... OPTIONAL MATCH ...` on `engine='polars'` crashed with `AttributeError: 'DataFrame' object has no attribute 'dropna'` before reaching the row pipeline (hit by LDBC SNB interactive-short-7 via the official query text). Now FIXED end-to-end: the seed extraction branches on frame type (`drop_nulls().unique().rename({...})` + `filter(is_in)` on polars), the optional-arm dispatch skips the pandas-only seeded-rows contract on polars (the seed restriction is a pruning hint — arm rows are left-outer-joined on the shared aliases, so unseeded is semantically identical, and `where`-arms already ran unseeded), and the arm join/alias-synthesis block gains a polars twin (`is_in` semi-join prune, `join(how='left')`, null-preserving alias synthesis). Result: **connected OPTIONAL MATCH now runs natively on polars** (previously pandas-only), oracle-exact on the IS7 shape; the simple-CASE null-literal equality (`CASE x WHEN null`, lowered as the `__cypher_case_eq__(x, null)` marker) now lowers natively to `is_null()` on polars (pandas-parity null-mask semantics; the general `CASE x WHEN v` form still declines honestly — it carries pandas' bool/numeric cross-dtype rules), so the full LDBC IS7 query INCLUDING its `CASE r WHEN null` flag projection runs natively on polars, oracle-exact. The polars arm is also PRUNED like the pandas one: since the polars bindings-row path declines `start_nodes` by contract, the same restriction is expressed as an id-membership filter injected into the arm's first (shared) node op — measured 9.2× on the LDBC SF0.1 harness (IS7 message-replies cypher 710.8→77.2 ms). pandas/cuDF paths byte-identical. Regression tests `test_optional_match_polars_frames.py` (pandas oracle, polars native end-to-end, polars no-pandas-ism contract).
 - **GFQL polars/polars-gpu seeded traversal no longer pays an O(E) NaN re-scan per hop on float-column graphs (identity-stable `_pl_nan_to_null` + clean-cache)**: every `hop()` runs `_coerce_input_formats` → `_pl_nan_to_null`, which scanned `is_nan().any()` over every float column on the (unchanged) resident edge frame **on every call** — so repeated seeded hops on a resident graph (the seeded-Search / native-hop pattern) grew O(E) with edge count on polars while pandas stayed flat. Measured: an indexed polars `g.hop` on 8M edges with one float column was **33.8 ms and growing**; it is now **0.20 ms and FLAT** (~140× faster; O(degree)), matching pandas. `_pl_nan_to_null` now probes an eager polars frame for real NaN once, returns a clean frame UNCHANGED (restoring the #1726 identity guard that #1731 reverted — so frame-identity caches like the #1658 index keep engaging), rewrites only the columns that genuinely carry NaN (values identical to the old unconditional `fill_nan`), and caches the id of frames verified clean so later calls skip the probe (recycle-safe via `weakref.finalize`; a rebound/new frame re-probes, so NaN→null semantics are unchanged). Regression: full 4-engine `test_index.py` parity + new `TestPlNanCleanCache` cases (NaN-present cleaned, clean-frame cached same-object, distinct frames independent).
diff --git a/graphistry/compute/gfql/index/traverse.py b/graphistry/compute/gfql/index/traverse.py
index 17f5b8773e..dabcdc7261 100644
--- a/graphistry/compute/gfql/index/traverse.py
+++ b/graphistry/compute/gfql/index/traverse.py
@@ -14,7 +14,7 @@
 """
 from __future__ import annotations
 
-from typing import Any, List, Optional, Tuple, cast
+from typing import Any, Dict, List, Optional, Tuple, cast
 
 from typing_extensions import TypeGuard
 
@@ -74,6 +74,16 @@ def is_simple_equality_edge_match(
     return True
 
 
+# Recycle-safe cache of edge keep-masks: (id(edge frame), engine, scalar items) -> mask.
+# The mask is a pure function of the resident edge frame + a scalar edge_match, but was
+# rebuilt via a full O(E) column compare on EVERY seeded hop (profiled: the eq_str scan
+# is the linear term that makes typed Cypher point queries grow with edge count — ~2.8ms
+# per call at 1M rows). Same lifetime pattern as Engine._pl_nan_to_null's clean-cache:
+# weakref.finalize on the SOURCE frame evicts, so a recycled id can never serve a stale
+# mask while the original frame is alive.
+_EDGE_KEEP_MASK_CACHE: Dict[Tuple[int, str, tuple], "ArrayLike"] = {}
+
+
 def _build_edge_keep_mask(
     edges: DataFrameT, edge_match: EdgeMatch, engine: Engine, xp: "object"
 ) -> Optional[ArrayLike]:
@@ -82,12 +92,18 @@ def _build_edge_keep_mask(
     a simple-equality ``edge_match``.
 
     Built via each frame's native ``col == val`` (so cudf string columns stay on the
-    cudf layer instead of a cupy string compare). Returns ``None`` on ANY unexpected
-    shape or error, so the caller falls back to scan rather than risk a divergence.
+    cudf layer instead of a cupy string compare), then cached per resident frame —
+    repeated seeded hops on the same typed edges hit O(1). Returns ``None`` on ANY
+    unexpected shape or error, so the caller falls back to scan rather than risk a
+    divergence.
     """
     try:
         if not is_simple_equality_edge_match(edge_match):
             return None
+        cache_key = (id(edges), engine.value, tuple(sorted(edge_match.items())))
+        cached = _EDGE_KEEP_MASK_CACHE.get(cache_key)
+        if cached is not None:
+            return cached
         from graphistry.compute.filter_by_dict import (
             _is_numeric_dtype_safe, _is_string_dtype_safe,
         )
@@ -126,6 +142,13 @@ def _build_edge_keep_mask(
                 col_mask = cast(
                     ArrayLike, (series == val).fillna(False).to_numpy(dtype=bool))
             mask = col_mask if mask is None else cast(ArrayLike, cast(Any, mask) & cast(Any, col_mask))
+        if mask is not None:
+            try:
+                import weakref
+                weakref.finalize(edges, _EDGE_KEEP_MASK_CACHE.pop, cache_key, None)
+                _EDGE_KEEP_MASK_CACHE[cache_key] = mask
+            except TypeError:  # pragma: no cover - frame not weakref-able -> don't cache
+                pass
         return mask
     except Exception:  # pragma: no cover - defensive parity guard
         return None
diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 80901e8f1a..931bb24c4e 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -456,6 +456,31 @@ def _try_native_row_op(g_cur, op):
     return None
 
 
+# Recycle-safe cache: resident edge frame -> its row-indexed augmentation. The chain
+# rebuilt `with_row_index` on EVERY call — an O(E) copy AND, worse, a fresh frame object
+# per call, which defeated every id-keyed cache downstream (#1658 rebind still worked via
+# rebind_edges, but per-frame caches like the typed keep-mask could never hit). Reusing
+# ONE augmented frame per resident frame makes repeated seeded queries O(1) here and
+# gives downstream id-keyed caches a stable identity. weakref.finalize on the SOURCE
+# frame evicts, so a recycled id can never serve a stale augmentation.
+_AUGMENTED_EDGES_CACHE: dict = {}
+
+
+def _augmented_edges_cached(edges, eid_col: str):
+    key = (id(edges), eid_col)
+    hit = _AUGMENTED_EDGES_CACHE.get(key)
+    if hit is not None:
+        return hit
+    out = edges.with_row_index(eid_col)
+    try:
+        import weakref
+        weakref.finalize(edges, _AUGMENTED_EDGES_CACHE.pop, key, None)
+        _AUGMENTED_EDGES_CACHE[key] = out
+    except TypeError:  # pragma: no cover - frame not weakref-able -> don't cache
+        pass
+    return out
+
+
 def chain_polars(self: Plottable, ops, start_nodes: Optional[Any] = None) -> Plottable:
     from graphistry.compute.ast import ASTCall
     from graphistry.compute.chain import Chain, _get_boundary_calls
@@ -662,7 +687,7 @@ def _plain_edge(op):
     g, _endpoint_restore = _align_edge_endpoints(g, g._node, g._source, g._destination)
     if g._edge is None:
         EID = "__gfql_edge_index__"
-        g = g.edges(g._edges.with_row_index(EID), g._source, g._destination, edge=EID)
+        g = g.edges(_augmented_edges_cached(g._edges, EID), g._source, g._destination, edge=EID)
         added_edge_index = True
         # with_row_index only PREPENDS a synthetic id column; the indexed src/dst are
         # preserved by value. Re-point any resident #1658 adjacency index at the new
diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index 6ed3a9d75c..c2c0875c50 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
@@ -80,6 +80,46 @@ def _apply_binop(op: str, left: pl.Expr, right: pl.Expr) -> Optional[pl.Expr]:
     return None
 
 
+
+# Recycle-safe cache of typed-edge slices: (id(edge frame), scalar edge_match items) ->
+# eager filtered frame. A seeded Cypher point query re-filtered the FULL edge frame's
+# string type column O(E) on EVERY call (~2.8ms/1M rows — the linear term that made LDBC
+# SF1 latency grow); a resident graph re-queries the same typed slice constantly. Same
+# lifetime pattern as Engine._pl_nan_to_null's clean-cache: weakref.finalize on the
+# SOURCE frame evicts, so a recycled id can never serve a stale slice.
+_TYPED_EDGE_SLICE_CACHE: Dict[Tuple[int, Any], Any] = {}
+
+
+def _typed_edge_slice(edges: Any, edge_match: Any) -> Optional[Any]:
+    """Eager typed-edge slice via cache; None when edge_match isn't scalar-only or the
+    frame isn't cacheable (caller falls back to the lazy per-call filter)."""
+    import polars as pl
+    if not isinstance(edges, pl.DataFrame) or not isinstance(edge_match, dict) or not edge_match:
+        return None
+    items = []
+    for k, v in edge_match.items():
+        if not isinstance(k, str) or isinstance(v, (dict, list, tuple, set)) or v is None:
+            return None  # predicate/membership/null forms keep the general path
+        items.append((k, v))
+    key = (id(edges), tuple(sorted(items)))
+    hit = _TYPED_EDGE_SLICE_CACHE.get(key)
+    if hit is not None:
+        return hit
+    from .predicates import filter_by_dict_polars
+    sliced_lf = filter_by_dict_polars(edges.lazy(), edge_match)
+    if sliced_lf is None:
+        return None
+    sliced = sliced_lf.collect()
+    _TYPED_EDGE_SLICE_CACHE[key] = sliced
+    try:
+        import weakref
+        weakref.finalize(edges, _TYPED_EDGE_SLICE_CACHE.pop, key, None)
+    except TypeError:  # pragma: no cover - pl.DataFrame is weakref-able; guard anyway
+        _TYPED_EDGE_SLICE_CACHE.pop(key, None)  # can't track lifetime -> don't cache
+        return sliced
+    return sliced
+
+
 def _resolve_property(alias: str, prop: str, columns: Sequence[str]) -> Optional[str]:
     """Resolve ``alias.prop`` to a row-table column (None if ambiguous/absent). Prefer the
     multi-entity prefixed form (``n.val``) over single-entity bare ``val`` + ``alias`` marker
@@ -993,7 +1033,11 @@ def _names(lf: pl.LazyFrame) -> List[str]:
             if not isinstance(edge_op, ASTEdge):
                 return None
             sem = EdgeSemantics.from_edge(edge_op)
-            edges_f = filter_by_dict_polars(edges_lf, edge_op.edge_match)
+            _cached_slice = _typed_edge_slice(edges, edge_op.edge_match)
+            if _cached_slice is not None:
+                edges_f = _cached_slice.lazy()
+            else:
+                edges_f = filter_by_dict_polars(edges_lf, edge_op.edge_match)
             edge_alias = edge_op._name
             if isinstance(edge_alias, str):
                 payload_renames = {

From e4cc21b971d00c0a91ab48225216cfadafb5c39b Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 12:31:42 -0700
Subject: [PATCH 02/16] perf(gfql): defer combine order-column stamping into
 the lazy plan (slice 2)

The _LazyShim construction eagerly with_row_index-copied both frames
O(N)+O(E) per call; lazy with_row_index numbers rows identically inside the
single fused collect. 283 tests pass (index + conformance + new regressions).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 CHANGELOG.md                                        | 2 +-
 graphistry/compute/gfql/lazy/engine/polars/chain.py | 6 +++++-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5199bc6d67..dde2163648 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,7 +12,7 @@ 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).
 
 ### Fixed
-- **GFQL seeded Cypher/chain queries no longer pay two per-call O(E) costs on a resident graph (typed keep-mask + edge-frame augmentation caches)**: profiling the official LDBC IS1 point query on polars found the linear-with-edge-count terms that made Cypher-surface latency GROW with scale: (1) the #1658 index path rebuilt its simple-equality edge keep-mask via a full column compare every seeded hop (`_build_edge_keep_mask` — the profiled `eq_str` O(E) scan), and (2) the polars chain executor rebuilt `with_row_index` on the edge frame every call — an O(E) copy whose fresh object identity ALSO defeated every downstream id-keyed cache. Both are now recycle-safe caches keyed on the resident frame (weakref.finalize eviction, the proven `_pl_nan_to_null` clean-cache pattern): one augmented frame per resident edge frame (stable identity for all downstream caches) and one keep-mask per (frame, engine, scalar edge_match). Profiled: both O(E) terms eliminated from the warm path (`eq_str`/`with_row_index` gone). Parity: 274 tests (index 4-engine + polars conformance) unchanged.
+- **GFQL seeded Cypher/chain queries no longer pay two per-call O(E) costs on a resident graph (typed keep-mask + edge-frame augmentation caches)**: profiling the official LDBC IS1 point query on polars found the linear-with-edge-count terms that made Cypher-surface latency GROW with scale: (1) the #1658 index path rebuilt its simple-equality edge keep-mask via a full column compare every seeded hop (`_build_edge_keep_mask` — the profiled `eq_str` O(E) scan), and (2) the polars chain executor rebuilt `with_row_index` on the edge frame every call — an O(E) copy whose fresh object identity ALSO defeated every downstream id-keyed cache. Both are now recycle-safe caches keyed on the resident frame (weakref.finalize eviction, the proven `_pl_nan_to_null` clean-cache pattern): one augmented frame per resident edge frame (stable identity for all downstream caches) and one keep-mask per (frame, engine, scalar edge_match). Profiled: both O(E) terms eliminated from the warm path (`eq_str`/`with_row_index` gone). Parity: 274 tests (index 4-engine + polars conformance) unchanged. Slice 2: the combine path's order-column stamping (`_LazyShim`) eagerly `with_row_index`-copied BOTH node and edge frames O(N)+O(E) on every call before entering the fused plan — now deferred into the lazy plan itself (identical row numbering, single collect); 283 tests.
 - **GFQL polars engine natively runs whole-entity aggregation Cypher (LDBC IC4 shape): HAS_<Label> destination disambiguation + entity identity-key resolution + `key_prefixes` group-by expansion**: three previously honest-NIE polars lowerings, each an exact pandas-parity port: (1) `binding_rows_polars` now applies pandas' `_gfql_disambiguate_has_edge_destination_nodes` candidate-domain rule (an UNLABELED next-node op after a `HAS_<Label>`-typed edge narrows to that label ONLY when candidate node ids collide across labels) instead of declining every labeled-destination pattern; (2) `alias.__gfql_node_id__` (the #1650 whole-entity identity key the aggregation lowering groups by) resolves to the bare `alias` id column (the polars bindings table intentionally omits pandas' join-residue columns — the bare alias column IS the identity key); (3) `group_by(key_prefixes=...)` expands every `<prefix>*` row-table column into the key set exactly like pandas (functionally dependent on the identity key — group sizes unchanged). Result: the official LDBC IC4 (new-topics) query — comma-MATCH, `WITH DISTINCT`, CASE projections, whole-entity `sum` aggregation, HAVING-style `WHERE` — runs natively on polars, parity-exact (harness-verified vs expected rows at SF0.1: ok, 127.8 ms). Regression test `test_polars_rows_entity_groupby.py` (pandas oracle + polars parity on a discriminating fixture). Remaining polars `rows` declines are undirected bounded var-length (IC6/IC11) and shortestPath/zero-hop-unbounded (by design).
 - **GFQL connected OPTIONAL MATCH no longer crashes on polars frames (engine-polymorphic seed extraction)**: `_optional_arm_start_nodes` (gfql_unified.py) applied pandas-only ops (`.dropna()`/`.drop_duplicates()`/`.rename(columns=)`/boolean-mask indexing) to the joined binding rows, so an IS7-shaped `MATCH ... OPTIONAL MATCH ...` on `engine='polars'` crashed with `AttributeError: 'DataFrame' object has no attribute 'dropna'` before reaching the row pipeline (hit by LDBC SNB interactive-short-7 via the official query text). Now FIXED end-to-end: the seed extraction branches on frame type (`drop_nulls().unique().rename({...})` + `filter(is_in)` on polars), the optional-arm dispatch skips the pandas-only seeded-rows contract on polars (the seed restriction is a pruning hint — arm rows are left-outer-joined on the shared aliases, so unseeded is semantically identical, and `where`-arms already ran unseeded), and the arm join/alias-synthesis block gains a polars twin (`is_in` semi-join prune, `join(how='left')`, null-preserving alias synthesis). Result: **connected OPTIONAL MATCH now runs natively on polars** (previously pandas-only), oracle-exact on the IS7 shape; the simple-CASE null-literal equality (`CASE x WHEN null`, lowered as the `__cypher_case_eq__(x, null)` marker) now lowers natively to `is_null()` on polars (pandas-parity null-mask semantics; the general `CASE x WHEN v` form still declines honestly — it carries pandas' bool/numeric cross-dtype rules), so the full LDBC IS7 query INCLUDING its `CASE r WHEN null` flag projection runs natively on polars, oracle-exact. The polars arm is also PRUNED like the pandas one: since the polars bindings-row path declines `start_nodes` by contract, the same restriction is expressed as an id-membership filter injected into the arm's first (shared) node op — measured 9.2× on the LDBC SF0.1 harness (IS7 message-replies cypher 710.8→77.2 ms). pandas/cuDF paths byte-identical. Regression tests `test_optional_match_polars_frames.py` (pandas oracle, polars native end-to-end, polars no-pandas-ism contract).
 - **GFQL polars/polars-gpu seeded traversal no longer pays an O(E) NaN re-scan per hop on float-column graphs (identity-stable `_pl_nan_to_null` + clean-cache)**: every `hop()` runs `_coerce_input_formats` → `_pl_nan_to_null`, which scanned `is_nan().any()` over every float column on the (unchanged) resident edge frame **on every call** — so repeated seeded hops on a resident graph (the seeded-Search / native-hop pattern) grew O(E) with edge count on polars while pandas stayed flat. Measured: an indexed polars `g.hop` on 8M edges with one float column was **33.8 ms and growing**; it is now **0.20 ms and FLAT** (~140× faster; O(degree)), matching pandas. `_pl_nan_to_null` now probes an eager polars frame for real NaN once, returns a clean frame UNCHANGED (restoring the #1726 identity guard that #1731 reverted — so frame-identity caches like the #1658 index keep engaging), rewrites only the columns that genuinely carry NaN (values identical to the old unconditional `fill_nan`), and caches the id of frames verified clean so later calls skip the probe (recycle-safe via `weakref.finalize`; a rebound/new frame re-probes, so NaN→null semantics are unchanged). Regression: full 4-engine `test_index.py` parity + new `TestPlNanCleanCache` cases (NaN-present cleaned, clean-frame cached same-object, distinct frames independent).
diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 931bb24c4e..e06b16ea81 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -787,7 +787,11 @@ def _plain_edge(op):
     from graphistry.compute.gfql.lazy import collect_all
     NORD = generate_safe_column_name("__gfql_norder__", g._nodes, prefix="__gfql_", suffix="__")
     EORD = generate_safe_column_name("__gfql_eorder__", g._edges, prefix="__gfql_", suffix="__")
-    g_lz = _LazyShim(g._nodes.with_row_index(NORD).lazy(), g._edges.with_row_index(EORD).lazy(),
+    # Defer the order-column stamping into the lazy plan: the eager form copied BOTH
+    # frames O(N)+O(E) on EVERY call before entering the fused plan; lazy
+    # with_row_index produces identical row numbering (source row order) inside the
+    # single collect.
+    g_lz = _LazyShim(g._nodes.lazy().with_row_index(NORD), g._edges.lazy().with_row_index(EORD),
                      node_col, src, dst, g._edge)
     steps_lz = [(op, _LazyShim.step(p)) for op, p in steps]
     edge_steps_lz = [(op, _LazyShim.step(p)) for op, p in edge_steps]

From 1bd2ab0f34897c6d99b726a0292fda4019fe0d4e Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 22:29:09 -0700
Subject: [PATCH 03/16] perf(gfql): lean gather-combine for seeded single-hop
 polars chains (slice 3)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The Track-B combine semi-joins the FULL node/edge frames and re-sorts on
in-plan row indices EVERY call — 55ms of the ~72ms SF1 seed-lookup wall
(probe /tmp/probe_collects.py). When the chain added __gfql_edge_index__
itself, surviving edge ids ARE row positions: a sorted position gather
rebuilds the exact edge frame in original row order for O(result), and
node rows come the same way via a recycle-safe cached id->row-position
mapping (4th cache of the weakref.finalize family; frames with null/dup
ids cache a None verdict and stay on Track B). Per-step endpoint
semi-joins + alias flags run only on the small step/gathered frames,
mirroring _combine_edges/_combine_nodes semantics exactly.

Guards -> Track B fallback: multihop, pre-bound edge ids, non-node first
step, unseeded chains, >100k total step rows, unsuitable node frame.

Parity: 1000/1000 polars chain tests incl. new lean-vs-TrackB
differential suite (byte-identical frames + engagement assertions +
fallback pins for dup-id and pre-bound-edge-id graphs).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 150 ++++++++++++++++++
 .../compute/gfql/test_engine_polars_chain.py  | 102 ++++++++++++
 2 files changed, 252 insertions(+)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index e06b16ea81..be39cabedb 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -481,6 +481,146 @@ def _augmented_edges_cached(edges, eid_col: str):
     return out
 
 
+# Recycle-safe cache #4: resident node frame -> sorted (id, row-position) lookup table for the
+# lean combine's position-gather. O(N log N) once per resident frame; weakref.finalize on the
+# source frame evicts so a recycled id can never serve stale positions. Frames with null or
+# duplicated ids cache a None verdict: position-gather keeps one row per id while the semi-join
+# combine keeps all matches, so those frames must stay on the Track-B path.
+_NODE_POS_CACHE: dict = {}
+_POS_MISS = object()
+
+
+def _node_pos_frame_cached(nodes, node_col: str):
+    key = (id(nodes), node_col)
+    hit = _NODE_POS_CACHE.get(key, _POS_MISS)
+    if hit is not _POS_MISS:
+        return hit
+    import polars as pl
+    from graphistry.compute.util import generate_safe_column_name
+    ids = nodes.get_column(node_col)
+    if ids.null_count() > 0 or ids.n_unique() != nodes.height:
+        mapping = None
+    else:
+        pos_col = generate_safe_column_name("__gfql_pos__", nodes, prefix="__gfql_", suffix="__")
+        mapping = (nodes.select(pl.col(node_col)).with_row_index(pos_col).sort(node_col), pos_col)
+    try:
+        import weakref
+        weakref.finalize(nodes, _NODE_POS_CACHE.pop, key, None)
+        _NODE_POS_CACHE[key] = mapping
+    except TypeError:  # pragma: no cover - frame not weakref-able -> don't cache
+        pass
+    return mapping
+
+
+def _gather_nodes_by_id(nodes, mapping_df, pos_col: str, node_col: str, ids):
+    """Rows of ``nodes`` whose id is in ``ids``, in original row order — semantically the
+    combine's ``nodes semi-join ids`` + NORD sort, but O(result·log N) via binary search on the
+    cached sorted mapping + position gather instead of an O(N) join/sort per call. Ids absent
+    from ``nodes`` (dangling endpoints) drop out via the equality mask, matching the semi-join."""
+    sorted_ids = mapping_df.get_column(node_col)
+    ids = ids.unique().drop_nulls().cast(sorted_ids.dtype)
+    if len(ids) == 0 or len(sorted_ids) == 0:
+        return nodes.clear()
+    idx = sorted_ids.search_sorted(ids, side="left")
+    in_range = idx < len(sorted_ids)
+    idx, ids = idx.filter(in_range), ids.filter(in_range)
+    hit = sorted_ids.gather(idx) == ids
+    pos = mapping_df.get_column(pos_col).gather(idx.filter(hit))
+    return nodes.gather(pos.sort())
+
+
+def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
+    """Slice-3 lean combine for seeded single-hop chains: O(result) instead of O(N+E) per call.
+
+    The Track-B combine rebuilds final frames by semi-joining the FULL node/edge frames and
+    re-sorting on in-plan row indices every call (~55ms of the ~72ms SF1 seed-lookup wall).
+    But when the chain added ``__gfql_edge_index__`` itself (``added_edge_index``), the surviving
+    edge ids ARE row positions in ``g._edges`` — so a sorted position gather rebuilds the exact
+    frame in original row order for free, and node rows come from the cached id→position mapping
+    the same way. Per-step endpoint semi-joins and alias flags run on the SMALL step/gathered
+    frames only, mirroring ``_combine_edges``/``_combine_nodes`` semantics exactly.
+
+    Returns ``(final_nodes, final_edges)`` (final_edges still carries ``eid_col``) or None to
+    fall back to Track B: non-node first step (its edge combine would semi-join the full node
+    frame), unsuitable node frame (null/dup ids), or step frames above the small-result guard.
+    """
+    import polars as pl
+    node_col, src, dst = g._node, g._source, g._destination
+    if not isinstance(ops[0], ASTNode):
+        return None
+    mapping = _node_pos_frame_cached(g._nodes, node_col)
+    if mapping is None:
+        return None
+    mapping_df, pos_col = mapping
+
+    # Materialize step frames in one batched collect, then bound total rows: the lean gathers
+    # and per-step joins are only a win (and only guaranteed small) on small wavefronts. On
+    # guard failure Track B proceeds with the original frames.
+    from graphistry.compute.gfql.lazy import collect_all
+    LEAN_MAX_ROWS = 100_000
+    pairs = list(steps) + list(label_steps)
+    frames: List[Any] = []
+    for _, p in pairs:
+        frames.extend((p._nodes, p._edges))
+    lazy_idx = [i for i, f in enumerate(frames) if f is not None and is_lazy(f)]
+    if lazy_idx:
+        for i, df in zip(lazy_idx, collect_all([frames[i] for i in lazy_idx])):
+            frames[i] = df
+    if sum(f.height for f in frames if f is not None) > LEAN_MAX_ROWS:
+        return None
+    shims = [
+        (op, _LazyShim(frames[2 * i], frames[2 * i + 1], node_col, src, dst, g._edge))
+        for i, (op, _) in enumerate(pairs)
+    ]
+    steps_m, label_m = shims[:len(steps)], shims[len(steps):]
+
+    # Edges: _combine_edges' per-step prev/next endpoint semi-joins, on small frames only
+    # (idx>0 always holds for edge steps — ops[0] is a node), then ONE position gather.
+    id_frames = []
+    for idx, (op, p) in enumerate(steps_m):
+        edges_df = p._edges
+        if edges_df is None or edges_df.height == 0:
+            continue
+        prev_nodes = label_m[idx - 1][1]._nodes if idx > 0 else None
+        next_nodes = label_m[idx + 1][1]._nodes if idx + 1 < len(label_m) else None
+        direction = op.direction if isinstance(op, ASTEdge) else "forward"
+        if direction == "undirected" and prev_nodes is not None and next_nodes is not None:
+            fwd = _semi(_semi(edges_df, prev_nodes, src, node_col), next_nodes, dst, node_col)
+            rev = _semi(_semi(edges_df, prev_nodes, dst, node_col), next_nodes, src, node_col)
+            edges_df = pl.concat([fwd, rev], how="vertical_relaxed").unique(subset=[eid_col])
+        else:
+            prev_col, next_col = (dst, src) if direction == "reverse" else (src, dst)
+            if prev_nodes is not None:
+                edges_df = _semi(edges_df, prev_nodes, prev_col, node_col)
+            if next_nodes is not None:
+                edges_df = _semi(edges_df, next_nodes, next_col, node_col)
+        id_frames.append(edges_df.select(pl.col(eid_col)))
+
+    if id_frames:
+        eids = pl.concat(id_frames, how="vertical_relaxed").get_column(eid_col).unique().sort()
+        out_edges = g._edges.gather(eids)  # sorted positions == original row order (EORD)
+    else:
+        out_edges = g._edges.clear()
+    for op, p in label_m:
+        if op._name is not None and isinstance(op, ASTEdge) and p._edges is not None and op._name in colnames(p._edges):
+            named = p._edges.filter(pl.col(op._name)).select(pl.col(eid_col)).with_columns(pl.lit(True).alias(op._name))
+            out_edges = out_edges.join(named, on=eid_col, how="left").with_columns(pl.col(op._name).fill_null(False))
+
+    # Nodes: step ids ∪ surviving-edge endpoints (endpoint repair), position-gathered in
+    # original row order (NORD). Dangling endpoints drop in the lookup, matching the semi-join.
+    parts = [
+        p._nodes.select(pl.col(node_col))
+        for _, p in steps_m
+        if p._nodes is not None and node_col in colnames(p._nodes)
+    ] + [
+        out_edges.select(pl.col(c).alias(node_col)) for c in (src, dst)
+    ]
+    ids = pl.concat(parts, how="vertical_relaxed").get_column(node_col)
+    final_nodes = _gather_nodes_by_id(g._nodes, mapping_df, pos_col, node_col, ids)
+    final_nodes = _apply_node_names(final_nodes, g, steps_m)
+    return final_nodes, out_edges
+
+
 def chain_polars(self: Plottable, ops, start_nodes: Optional[Any] = None) -> Plottable:
     from graphistry.compute.ast import ASTCall
     from graphistry.compute.chain import Chain, _get_boundary_calls
@@ -777,6 +917,16 @@ def _plain_edge(op):
             else:
                 edge_steps.append((op, g_step))
 
+    # Slice-3 lean combine: seeded single-hop chains skip the full-frame Track-B plan entirely
+    # (position gathers over the resident frames, O(result) per call). Guards + fallback inside.
+    if not has_multihop and added_edge_index and (
+            start_nodes is not None or (isinstance(ops[0], ASTNode) and ops[0].filter_dict)):
+        lean = _try_combine_lean(g, ops, steps, label_steps, EID)
+        if lean is not None:
+            final_nodes, final_edges = lean
+            final_edges = _restore_edge_dtypes(final_edges.drop(EID), src, dst, _endpoint_restore)
+            return self.nodes(final_nodes, node_col).edges(final_edges, src, dst)
+
     # Track B: build the WHOLE combine (combine_nodes/edges + endpoint + names) as ONE deferred
     # plan over already-materialized hop frames and collect ONCE — collapsing the ~dozen eager
     # combine ops (each internally lazy().op().collect()) into a single fused pass on the active
diff --git a/graphistry/tests/compute/gfql/test_engine_polars_chain.py b/graphistry/tests/compute/gfql/test_engine_polars_chain.py
index 58568da2da..612fee0ab3 100644
--- a/graphistry/tests/compute/gfql/test_engine_polars_chain.py
+++ b/graphistry/tests/compute/gfql/test_engine_polars_chain.py
@@ -704,3 +704,105 @@ def ids(gg):
     # temporal val -> _cmp_expr declines (None) so upstream raises honest NIE; numeric still lowers.
     assert _cmp_expr(None, operator.gt, datetime.date(2020, 1, 1)) is None
     assert _cmp_expr(pl.col("x"), operator.gt, 5) is not None
+
+
+# ---------------------------------------------------------------------------
+# Slice-3 lean gather-combine: differential vs Track B (byte-identical frames)
+# ---------------------------------------------------------------------------
+
+_LEAN_NODES = pl.DataFrame({
+    "id": ["a", "b", "c", "d", "e", "f", "g"],
+    "kind": ["x", "y", "y", "z", "x", "z", "y"],
+    "score": [10, 20, 30, 40, 50, 60, 70],
+})
+_LEAN_EDGES = pl.DataFrame({
+    "s": ["a", "a", "b", "c", "d", "e", "g", "f"],
+    "d": ["b", "c", "d", "d", "e", "b", "a", "zz_dangling"],
+    "etype": ["k", "k", "l", "l", "k", "l", "k", "k"],
+})
+
+_LEAN_CHAINS = [
+    ("named-fwd", [n({"id": "a"}, name="src"), e_forward(name="E"), n(name="dst")], True),
+    ("named-two-hop", [n({"id": "a"}), e_forward(), n(name="mid"), e_forward(name="E2"), n()], True),
+    ("undirected-filtered", [n({"id": "c"}), e_undirected(), n({"kind": "y"})], True),
+    ("reverse-named", [n({"id": "d"}, name="seed"), e_reverse(), n()], True),
+    ("edge-match", [n({"id": "a"}), e_forward(edge_match={"etype": "k"}, name="E"), n()], True),
+    ("empty-seed", [n({"id": "no_such"}), e_forward(name="E"), n()], True),
+    ("dangling-endpoint", [n({"id": "f"}, name="seed"), e_forward(), n()], True),
+]
+
+
+class TestLeanCombineDifferential:
+    """The lean position-gather combine must be byte-identical to Track B (frames, dtypes,
+    row order, alias flags) on every seeded shape it engages for, and must actually engage."""
+
+    def _graph(self):
+        return graphistry.nodes(_LEAN_NODES, "id").edges(_LEAN_EDGES, "s", "d")
+
+    @pytest.mark.parametrize("label,ch,expect_lean", [(la, c, x) for la, c, x in _LEAN_CHAINS],
+                             ids=[la for la, _, _ in _LEAN_CHAINS])
+    def test_lean_vs_trackb_byte_identical(self, label, ch, expect_lean, monkeypatch):
+        from polars.testing import assert_frame_equal
+        from graphistry.compute.gfql.lazy.engine.polars import chain as chain_mod
+
+        g = self._graph()
+        engaged = []
+        orig = chain_mod._try_combine_lean
+
+        def spy(*a, **k):
+            r = orig(*a, **k)
+            engaged.append(r is not None)
+            return r
+
+        monkeypatch.setattr(chain_mod, "_try_combine_lean", spy)
+        g_lean = g.chain(ch, engine="polars")
+        assert engaged and engaged[-1] == expect_lean, f"[{label}] lean engagement mismatch: {engaged}"
+
+        monkeypatch.setattr(chain_mod, "_try_combine_lean", lambda *a, **k: None)
+        g_trackb = g.chain(ch, engine="polars")
+
+        assert_frame_equal(g_lean._nodes, g_trackb._nodes)
+        assert_frame_equal(g_lean._edges, g_trackb._edges)
+        # and both agree with the pandas oracle on sets + aliases
+        _assert_chain_parity(g, ch, f"lean-{label}", native=True, edge_count=True, aliases=True)
+
+    def test_lean_falls_back_on_dup_node_ids(self, monkeypatch):
+        """Duplicate node ids break position-gather semantics (semi-join keeps all matches) —
+        the mapping cache must return None and Track B must serve the query."""
+        from polars.testing import assert_frame_equal
+        from graphistry.compute.gfql.lazy.engine.polars import chain as chain_mod
+
+        nodes = pl.concat([_LEAN_NODES, _LEAN_NODES.filter(pl.col("id") == "b")])
+        g = graphistry.nodes(nodes, "id").edges(_LEAN_EDGES, "s", "d")
+        ch = [n({"id": "a"}, name="src"), e_forward(name="E"), n()]
+
+        engaged = []
+        orig = chain_mod._try_combine_lean
+
+        def spy(*a, **k):
+            r = orig(*a, **k)
+            engaged.append(r is not None)
+            return r
+
+        monkeypatch.setattr(chain_mod, "_try_combine_lean", spy)
+        g_lean = g.chain(ch, engine="polars")
+        assert engaged == [False], f"dup-id graph must fall back to Track B: {engaged}"
+
+        monkeypatch.setattr(chain_mod, "_try_combine_lean", lambda *a, **k: None)
+        g_trackb = g.chain(ch, engine="polars")
+        assert_frame_equal(g_lean._nodes, g_trackb._nodes)
+        assert_frame_equal(g_lean._edges, g_trackb._edges)
+
+    def test_lean_falls_back_on_prebound_edge_id(self, monkeypatch):
+        """User-bound edge ids are values, not row positions — lean must not engage."""
+        from graphistry.compute.gfql.lazy.engine.polars import chain as chain_mod
+
+        edges = _LEAN_EDGES.with_row_index("eid").with_columns((pl.col("eid") + 100).alias("eid"))
+        g = graphistry.nodes(_LEAN_NODES, "id").edges(edges, "s", "d").bind(edge="eid")
+
+        called = []
+        orig = chain_mod._try_combine_lean
+        monkeypatch.setattr(chain_mod, "_try_combine_lean",
+                            lambda *a, **k: called.append(True) or orig(*a, **k))
+        g.chain([n({"id": "a"}, name="src"), e_forward(name="E"), n()], engine="polars")
+        assert not called, "lean guard must skip pre-bound edge ids (added_edge_index=False)"

From 821a0dbfd0023447ddad9c104303470838f89a6b Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 22:41:27 -0700
Subject: [PATCH 04/16] fix(gfql): version-safe row selection in lean combine
 (polars <1.4x lacks DataFrame.gather)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The SF1 re-measure on the 26.02 RAPIDS image (polars 1.35.2) failed every
lean-engaged query with AttributeError — DataFrame.gather is a 1.4x
addition. __getitem__ row selection is equivalent and supported on both.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 graphistry/compute/gfql/lazy/engine/polars/chain.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index be39cabedb..cbdd621cfc 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -526,7 +526,8 @@ def _gather_nodes_by_id(nodes, mapping_df, pos_col: str, node_col: str, ids):
     idx, ids = idx.filter(in_range), ids.filter(in_range)
     hit = sorted_ids.gather(idx) == ids
     pos = mapping_df.get_column(pos_col).gather(idx.filter(hit))
-    return nodes.gather(pos.sort())
+    # __getitem__ row selection, not DataFrame.gather — absent before polars 1.4x
+    return nodes[pos.sort()]
 
 
 def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
@@ -598,7 +599,7 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
 
     if id_frames:
         eids = pl.concat(id_frames, how="vertical_relaxed").get_column(eid_col).unique().sort()
-        out_edges = g._edges.gather(eids)  # sorted positions == original row order (EORD)
+        out_edges = g._edges[eids]  # sorted positions == original row order (EORD)
     else:
         out_edges = g._edges.clear()
     for op, p in label_m:

From 9efce7aea3296fee879214bb861b48db87dc8ed1 Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 22:50:12 -0700
Subject: [PATCH 05/16] perf(gfql): reuse lean-guard materialized steps in
 Track B fallback
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The lean attempt batch-collects step frames to bound result size; on
guard failure it previously returned None and Track B re-executed the
whole traversal inside its fused collect — +34-43% on big-wavefront SF1
queries (new-topics, job-referral). Return the materialized steps
instead and thread them into Track B, so the traversal runs once either
way. Protocol: ('lean', nodes, edges) | ('fallback', steps, labels) |
None (pre-collect declines).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 25 ++++++++++++-------
 .../compute/gfql/test_engine_polars_chain.py  |  4 +--
 2 files changed, 18 insertions(+), 11 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index cbdd621cfc..a54d63729e 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -541,9 +541,11 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
     the same way. Per-step endpoint semi-joins and alias flags run on the SMALL step/gathered
     frames only, mirroring ``_combine_edges``/``_combine_nodes`` semantics exactly.
 
-    Returns ``(final_nodes, final_edges)`` (final_edges still carries ``eid_col``) or None to
-    fall back to Track B: non-node first step (its edge combine would semi-join the full node
-    frame), unsuitable node frame (null/dup ids), or step frames above the small-result guard.
+    Returns ``("lean", final_nodes, final_edges)`` (final_edges still carries ``eid_col``);
+    ``("fallback", steps_m, label_m)`` when step frames exceed the small-result guard — the
+    MATERIALIZED steps, so Track B reuses them instead of re-executing the traversal; or None
+    (pre-collect declines: non-node first step — its edge combine would semi-join the full node
+    frame — or unsuitable node frame with null/dup ids).
     """
     import polars as pl
     node_col, src, dst = g._node, g._source, g._destination
@@ -556,7 +558,9 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
 
     # Materialize step frames in one batched collect, then bound total rows: the lean gathers
     # and per-step joins are only a win (and only guaranteed small) on small wavefronts. On
-    # guard failure Track B proceeds with the original frames.
+    # guard failure the MATERIALIZED steps go back to Track B (never the original lazy frames —
+    # that re-executed the whole traversal inside Track B's collect: +34-43% on the big-wavefront
+    # SF1 queries).
     from graphistry.compute.gfql.lazy import collect_all
     LEAN_MAX_ROWS = 100_000
     pairs = list(steps) + list(label_steps)
@@ -567,13 +571,13 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
     if lazy_idx:
         for i, df in zip(lazy_idx, collect_all([frames[i] for i in lazy_idx])):
             frames[i] = df
-    if sum(f.height for f in frames if f is not None) > LEAN_MAX_ROWS:
-        return None
     shims = [
         (op, _LazyShim(frames[2 * i], frames[2 * i + 1], node_col, src, dst, g._edge))
         for i, (op, _) in enumerate(pairs)
     ]
     steps_m, label_m = shims[:len(steps)], shims[len(steps):]
+    if sum(f.height for f in frames if f is not None) > LEAN_MAX_ROWS:
+        return ("fallback", steps_m, label_m)
 
     # Edges: _combine_edges' per-step prev/next endpoint semi-joins, on small frames only
     # (idx>0 always holds for edge steps — ops[0] is a node), then ONE position gather.
@@ -619,7 +623,7 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
     ids = pl.concat(parts, how="vertical_relaxed").get_column(node_col)
     final_nodes = _gather_nodes_by_id(g._nodes, mapping_df, pos_col, node_col, ids)
     final_nodes = _apply_node_names(final_nodes, g, steps_m)
-    return final_nodes, out_edges
+    return ("lean", final_nodes, out_edges)
 
 
 def chain_polars(self: Plottable, ops, start_nodes: Optional[Any] = None) -> Plottable:
@@ -923,10 +927,13 @@ def _plain_edge(op):
     if not has_multihop and added_edge_index and (
             start_nodes is not None or (isinstance(ops[0], ASTNode) and ops[0].filter_dict)):
         lean = _try_combine_lean(g, ops, steps, label_steps, EID)
-        if lean is not None:
-            final_nodes, final_edges = lean
+        if lean is not None and lean[0] == "lean":
+            _, final_nodes, final_edges = lean
             final_edges = _restore_edge_dtypes(final_edges.drop(EID), src, dst, _endpoint_restore)
             return self.nodes(final_nodes, node_col).edges(final_edges, src, dst)
+        if lean is not None:  # guard fallback: reuse the materialized steps in Track B
+            _, steps, label_steps = lean
+            edge_steps = steps
 
     # Track B: build the WHOLE combine (combine_nodes/edges + endpoint + names) as ONE deferred
     # plan over already-materialized hop frames and collect ONCE — collapsing the ~dozen eager
diff --git a/graphistry/tests/compute/gfql/test_engine_polars_chain.py b/graphistry/tests/compute/gfql/test_engine_polars_chain.py
index 612fee0ab3..1604450772 100644
--- a/graphistry/tests/compute/gfql/test_engine_polars_chain.py
+++ b/graphistry/tests/compute/gfql/test_engine_polars_chain.py
@@ -751,7 +751,7 @@ def test_lean_vs_trackb_byte_identical(self, label, ch, expect_lean, monkeypatch
 
         def spy(*a, **k):
             r = orig(*a, **k)
-            engaged.append(r is not None)
+            engaged.append(r is not None and r[0] == "lean")
             return r
 
         monkeypatch.setattr(chain_mod, "_try_combine_lean", spy)
@@ -781,7 +781,7 @@ def test_lean_falls_back_on_dup_node_ids(self, monkeypatch):
 
         def spy(*a, **k):
             r = orig(*a, **k)
-            engaged.append(r is not None)
+            engaged.append(r is not None and r[0] == "lean")
             return r
 
         monkeypatch.setattr(chain_mod, "_try_combine_lean", spy)

From 1d4bfe9a6268a636408cea006b5b84fd35bdef09 Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 22:56:56 -0700
Subject: [PATCH 06/16] perf(gfql): lean combine consumes label steps as id
 projections only
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Full forward label frames (hub fan-out) were eagerly collected by the
lean attempt while Track B only ever touches them through fused lazy
semi-joins with pushdown — new-topics cypher +33% at SF1. The lean
combine needs just prev/next node-id sets and named-edge eid sets, so
collect those projections in the same batched collect_all; on guard
fallback label steps return to Track B untouched (lazy, fusible).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 44 +++++++++++++------
 1 file changed, 31 insertions(+), 13 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index a54d63729e..13dc06be03 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -556,28 +556,46 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
         return None
     mapping_df, pos_col = mapping
 
-    # Materialize step frames in one batched collect, then bound total rows: the lean gathers
-    # and per-step joins are only a win (and only guaranteed small) on small wavefronts. On
-    # guard failure the MATERIALIZED steps go back to Track B (never the original lazy frames —
-    # that re-executed the whole traversal inside Track B's collect: +34-43% on the big-wavefront
-    # SF1 queries).
+    # Materialize the backward-pruned step frames plus ID PROJECTIONS of the forward label
+    # frames in ONE batched collect, then bound total rows: the lean gathers and per-step joins
+    # are only a win (and only guaranteed small) on small wavefronts. Label steps enter the lean
+    # combine solely as semi-join id sets (prev/next endpoints) and named-edge id sets — never
+    # materialize them in full: forward hub fan-out frames are exactly what Track B's fused
+    # lazy semi-joins avoid materializing (full-frame label collect cost new-topics +33% at SF1).
+    # On guard failure the MATERIALIZED steps go back to Track B (never the original lazy frames
+    # — that re-executed the whole traversal inside Track B's collect: +34-43% on big-wavefront
+    # SF1 queries); label steps stay as-is for Track B's fusion.
     from graphistry.compute.gfql.lazy import collect_all
     LEAN_MAX_ROWS = 100_000
-    pairs = list(steps) + list(label_steps)
     frames: List[Any] = []
-    for _, p in pairs:
+    for _, p in steps:
         frames.extend((p._nodes, p._edges))
+    n_step_frames = len(frames)
+    # label projections: nodes -> unique id column; edges -> named-flag eid column (or None)
+    for op, p in label_steps:
+        nproj = None
+        if p._nodes is not None and node_col in colnames(p._nodes):
+            nproj = p._nodes.select(pl.col(node_col)).unique()
+        eproj = None
+        if (op._name is not None and isinstance(op, ASTEdge) and p._edges is not None
+                and op._name in colnames(p._edges)):
+            eproj = p._edges.filter(pl.col(op._name)).select(pl.col(eid_col))
+        frames.extend((nproj, eproj))
     lazy_idx = [i for i, f in enumerate(frames) if f is not None and is_lazy(f)]
     if lazy_idx:
         for i, df in zip(lazy_idx, collect_all([frames[i] for i in lazy_idx])):
             frames[i] = df
-    shims = [
+    steps_m = [
         (op, _LazyShim(frames[2 * i], frames[2 * i + 1], node_col, src, dst, g._edge))
-        for i, (op, _) in enumerate(pairs)
+        for i, (op, _) in enumerate(steps)
+    ]
+    label_m = [
+        (op, _LazyShim(frames[n_step_frames + 2 * i], frames[n_step_frames + 2 * i + 1],
+                       node_col, src, dst, g._edge))
+        for i, (op, _) in enumerate(label_steps)
     ]
-    steps_m, label_m = shims[:len(steps)], shims[len(steps):]
     if sum(f.height for f in frames if f is not None) > LEAN_MAX_ROWS:
-        return ("fallback", steps_m, label_m)
+        return ("fallback", steps_m, list(label_steps))
 
     # Edges: _combine_edges' per-step prev/next endpoint semi-joins, on small frames only
     # (idx>0 always holds for edge steps — ops[0] is a node), then ONE position gather.
@@ -607,8 +625,8 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
     else:
         out_edges = g._edges.clear()
     for op, p in label_m:
-        if op._name is not None and isinstance(op, ASTEdge) and p._edges is not None and op._name in colnames(p._edges):
-            named = p._edges.filter(pl.col(op._name)).select(pl.col(eid_col)).with_columns(pl.lit(True).alias(op._name))
+        if p._edges is not None:  # pre-filtered named-edge eid projection (built above)
+            named = p._edges.with_columns(pl.lit(True).alias(op._name))
             out_edges = out_edges.join(named, on=eid_col, how="left").with_columns(pl.col(op._name).fill_null(False))
 
     # Nodes: step ids ∪ surviving-edge endpoints (endpoint repair), position-gathered in

From 7bd3dc6a6e35bd1ced363cfa8a7e23aee02d1a7b Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 23:01:50 -0700
Subject: [PATCH 07/16] perf(gfql): defer lean-combine mapping build until
 after the row guard

Pipelines that rebind node frames per call (whole-entity aggregation)
never hit the mapping cache and paid O(N log N) sort+unique per call
just to land in the guard fallback (new-topics cypher +35% at SF1).
Build the mapping only once the small-result guard passes; unsuitable
node frames now also return the materialized-steps fallback.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 20 +++++++++++--------
 1 file changed, 12 insertions(+), 8 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 13dc06be03..a78f477111 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -542,19 +542,15 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
     frames only, mirroring ``_combine_edges``/``_combine_nodes`` semantics exactly.
 
     Returns ``("lean", final_nodes, final_edges)`` (final_edges still carries ``eid_col``);
-    ``("fallback", steps_m, label_m)`` when step frames exceed the small-result guard — the
-    MATERIALIZED steps, so Track B reuses them instead of re-executing the traversal; or None
-    (pre-collect declines: non-node first step — its edge combine would semi-join the full node
-    frame — or unsuitable node frame with null/dup ids).
+    ``("fallback", steps_m, label_steps)`` when step frames exceed the small-result guard or the
+    node frame is unsuitable (null/dup ids) — with the MATERIALIZED steps, so Track B reuses
+    them instead of re-executing the traversal; or None (pre-collect decline: non-node first
+    step — its edge combine would semi-join the full node frame).
     """
     import polars as pl
     node_col, src, dst = g._node, g._source, g._destination
     if not isinstance(ops[0], ASTNode):
         return None
-    mapping = _node_pos_frame_cached(g._nodes, node_col)
-    if mapping is None:
-        return None
-    mapping_df, pos_col = mapping
 
     # Materialize the backward-pruned step frames plus ID PROJECTIONS of the forward label
     # frames in ONE batched collect, then bound total rows: the lean gathers and per-step joins
@@ -597,6 +593,14 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
     if sum(f.height for f in frames if f is not None) > LEAN_MAX_ROWS:
         return ("fallback", steps_m, list(label_steps))
 
+    # Mapping build LAST (after the row guard): it costs O(N log N) per resident node frame,
+    # and pipelines that rebind node frames per call (fresh identity -> cache miss) must not
+    # pay it just to land in the fallback anyway (new-topics cypher +35% at SF1).
+    mapping = _node_pos_frame_cached(g._nodes, node_col)
+    if mapping is None:
+        return ("fallback", steps_m, list(label_steps))
+    mapping_df, pos_col = mapping
+
     # Edges: _combine_edges' per-step prev/next endpoint semi-joins, on small frames only
     # (idx>0 always holds for edge steps — ops[0] is a node), then ONE position gather.
     id_frames = []

From a864c8d05056253db2d65c51e8f225565ccb4952 Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 23:07:35 -0700
Subject: [PATCH 08/16] perf(gfql): project step frames to combine-read columns
 in lean collect
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The eager step collect materialized full wavefront columns; both the
lean gathers and Track B rebuild output rows from the RESIDENT frames
and read only ids/endpoints/alias flags from steps — Track B's fused
plan got this for free via column pushdown, the explicit collect must
project. 1136 chain+conformance tests green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 29 +++++++++++++++++--
 1 file changed, 27 insertions(+), 2 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index a78f477111..aeabe50607 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -563,9 +563,34 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
     # SF1 queries); label steps stay as-is for Track B's fusion.
     from graphistry.compute.gfql.lazy import collect_all
     LEAN_MAX_ROWS = 100_000
+
+    def _node_cols(op, f):
+        cols = [node_col] if node_col in colnames(f) else []
+        if op._name is not None and isinstance(op, ASTNode) and op._name in colnames(f):
+            cols.append(op._name)
+        return cols
+
+    def _edge_cols(op, f):
+        cols = [c for c in (eid_col, src, dst) if c in colnames(f)]
+        if op._name is not None and isinstance(op, ASTEdge) and op._name in colnames(f):
+            cols.append(op._name)
+        return cols
+
     frames: List[Any] = []
-    for _, p in steps:
-        frames.extend((p._nodes, p._edges))
+    for op, p in steps:
+        # Project to the columns the combine actually reads (ids, endpoints, alias flags):
+        # both the lean gathers AND Track B rebuild output rows from the RESIDENT frames, so
+        # materializing full wavefront columns is pure waste (column pushdown does this inside
+        # Track B's fused plan; the eager collect must do it explicitly).
+        nf = p._nodes
+        if nf is not None:
+            cols = _node_cols(op, nf)
+            nf = nf.select([pl.col(c) for c in cols]) if cols else nf
+        ef = p._edges
+        if ef is not None:
+            cols = _edge_cols(op, ef)
+            ef = ef.select([pl.col(c) for c in cols]) if cols else ef
+        frames.extend((nf, ef))
     n_step_frames = len(frames)
     # label projections: nodes -> unique id column; edges -> named-flag eid column (or None)
     for op, p in label_steps:

From 7764096e21cefaa146945071624c4f40316077e8 Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 23:15:58 -0700
Subject: [PATCH 09/16] perf(gfql): two-sighting node-pos mapping + is_in
 first-call node gather
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The mapping build (O(N log N) sort + O(N) dup probe) ran per chain call
for pipelines that rebind node frames (fresh identity, zero cache reuse)
— the whole-entity aggregation path paid it every call (new-topics
cypher +32% at SF1). First sighting of a node frame now marks the cache
and serves nodes via the O(N) order-preserving is_in filter (same rows
and order as the semi-join); only a recurring frame builds the mapping.
Dup-id results (Track B dedupes with a first-occurrence pick) divert to
the materialized-steps fallback.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 51 ++++++++++++-------
 .../compute/gfql/test_engine_polars_chain.py  |  4 +-
 2 files changed, 35 insertions(+), 20 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index aeabe50607..ead538a53b 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -482,18 +482,31 @@ def _augmented_edges_cached(edges, eid_col: str):
 
 
 # Recycle-safe cache #4: resident node frame -> sorted (id, row-position) lookup table for the
-# lean combine's position-gather. O(N log N) once per resident frame; weakref.finalize on the
-# source frame evicts so a recycled id can never serve stale positions. Frames with null or
-# duplicated ids cache a None verdict: position-gather keeps one row per id while the semi-join
-# combine keeps all matches, so those frames must stay on the Track-B path.
+# lean combine's position-gather. weakref.finalize on the source frame evicts so a recycled id
+# can never serve stale positions. TWO-SIGHTING protocol: the O(N log N) build (+ O(N) null/dup
+# suitability probe) only pays off when the frame recurs, and pipelines that rebind node frames
+# per call (whole-entity aggregation) would otherwise pay it every call for zero reuse
+# (new-topics cypher +32% at SF1) — so the first sighting is only MARKED (caller uses the O(N)
+# order-preserving is_in filter instead) and the mapping is built on the second sighting.
+# Frames with null or duplicated ids cache a None verdict: position-gather keeps one row per id
+# while the semi-join combine keeps all matches, so those frames stay on the is_in path.
 _NODE_POS_CACHE: dict = {}
 _POS_MISS = object()
+_POS_SEEN_ONCE = object()
 
 
 def _node_pos_frame_cached(nodes, node_col: str):
     key = (id(nodes), node_col)
     hit = _NODE_POS_CACHE.get(key, _POS_MISS)
-    if hit is not _POS_MISS:
+    if hit is _POS_MISS:
+        try:
+            import weakref
+            weakref.finalize(nodes, _NODE_POS_CACHE.pop, key, None)
+            _NODE_POS_CACHE[key] = _POS_SEEN_ONCE
+        except TypeError:  # pragma: no cover - frame not weakref-able -> don't cache
+            pass
+        return None
+    if hit is not _POS_SEEN_ONCE:
         return hit
     import polars as pl
     from graphistry.compute.util import generate_safe_column_name
@@ -503,12 +516,7 @@ def _node_pos_frame_cached(nodes, node_col: str):
     else:
         pos_col = generate_safe_column_name("__gfql_pos__", nodes, prefix="__gfql_", suffix="__")
         mapping = (nodes.select(pl.col(node_col)).with_row_index(pos_col).sort(node_col), pos_col)
-    try:
-        import weakref
-        weakref.finalize(nodes, _NODE_POS_CACHE.pop, key, None)
-        _NODE_POS_CACHE[key] = mapping
-    except TypeError:  # pragma: no cover - frame not weakref-able -> don't cache
-        pass
+    _NODE_POS_CACHE[key] = mapping
     return mapping
 
 
@@ -618,13 +626,10 @@ def _edge_cols(op, f):
     if sum(f.height for f in frames if f is not None) > LEAN_MAX_ROWS:
         return ("fallback", steps_m, list(label_steps))
 
-    # Mapping build LAST (after the row guard): it costs O(N log N) per resident node frame,
-    # and pipelines that rebind node frames per call (fresh identity -> cache miss) must not
-    # pay it just to land in the fallback anyway (new-topics cypher +35% at SF1).
+    # Mapping lookup LAST (after the row guard): None means first sighting of this node frame
+    # (or null/dup ids) -> the node gather below uses the O(N) order-preserving is_in filter
+    # instead of the O(k log N) position gather; the mapping only builds for recurring frames.
     mapping = _node_pos_frame_cached(g._nodes, node_col)
-    if mapping is None:
-        return ("fallback", steps_m, list(label_steps))
-    mapping_df, pos_col = mapping
 
     # Edges: _combine_edges' per-step prev/next endpoint semi-joins, on small frames only
     # (idx>0 always holds for edge steps — ops[0] is a node), then ONE position gather.
@@ -668,7 +673,17 @@ def _edge_cols(op, f):
         out_edges.select(pl.col(c).alias(node_col)) for c in (src, dst)
     ]
     ids = pl.concat(parts, how="vertical_relaxed").get_column(node_col)
-    final_nodes = _gather_nodes_by_id(g._nodes, mapping_df, pos_col, node_col, ids)
+    if mapping is not None:
+        mapping_df, pos_col = mapping
+        final_nodes = _gather_nodes_by_id(g._nodes, mapping_df, pos_col, node_col, ids)
+    else:
+        # First sighting / unsuitable frame: order-preserving hash-probe filter — same row set
+        # as the semi-join, original order for free. Track B DEDUPES final nodes by id
+        # (unique(subset, maintain_order)); a dup-id result would need Track B's exact
+        # first-occurrence pick, so send that rare shape to the fallback.
+        final_nodes = g._nodes.filter(pl.col(node_col).is_in(ids.drop_nulls()))
+        if final_nodes.get_column(node_col).n_unique() != final_nodes.height:
+            return ("fallback", steps_m, list(label_steps))
     final_nodes = _apply_node_names(final_nodes, g, steps_m)
     return ("lean", final_nodes, out_edges)
 
diff --git a/graphistry/tests/compute/gfql/test_engine_polars_chain.py b/graphistry/tests/compute/gfql/test_engine_polars_chain.py
index 1604450772..2535794503 100644
--- a/graphistry/tests/compute/gfql/test_engine_polars_chain.py
+++ b/graphistry/tests/compute/gfql/test_engine_polars_chain.py
@@ -767,8 +767,8 @@ def spy(*a, **k):
         _assert_chain_parity(g, ch, f"lean-{label}", native=True, edge_count=True, aliases=True)
 
     def test_lean_falls_back_on_dup_node_ids(self, monkeypatch):
-        """Duplicate node ids break position-gather semantics (semi-join keeps all matches) —
-        the mapping cache must return None and Track B must serve the query."""
+        """Duplicate node ids break lean node-gather semantics (Track B dedupes by id with a
+        first-occurrence pick) — the lean attempt must detect the dup result and fall back."""
         from polars.testing import assert_frame_equal
         from graphistry.compute.gfql.lazy.engine.polars import chain as chain_mod
 

From 1f3cec645a72dfe6a2f54019641b12a9523411fb Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 23:22:40 -0700
Subject: [PATCH 10/16] perf(gfql): cap the lean-guard probe collect;
 truncation falls back to fused Track B
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Full collection of projected step frames still executed the unpruned
traversal for big-wavefront queries (new-topics cypher +34% at SF1) —
Track B's single fused plan pushes the combine's semi-join predicates
INTO the traversal, which a split collect cannot. Collect each lazy step
frame limit-capped at guard+1: a frame at <= cap is complete and
reusable; any truncated frame aborts the lean attempt so Track B runs on
the ORIGINAL lazy steps with full fusion, probe cost bounded by the cap.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 graphistry/compute/gfql/lazy/engine/polars/chain.py | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index ead538a53b..88317d12a6 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -553,7 +553,8 @@ def _try_combine_lean(g, ops, steps, label_steps, eid_col: str):
     ``("fallback", steps_m, label_steps)`` when step frames exceed the small-result guard or the
     node frame is unsuitable (null/dup ids) — with the MATERIALIZED steps, so Track B reuses
     them instead of re-executing the traversal; or None (pre-collect decline: non-node first
-    step — its edge combine would semi-join the full node frame).
+    step — its edge combine would semi-join the full node frame — or a capped-collect probe
+    truncated, meaning big wavefronts where Track B's fused lazy plan wins).
     """
     import polars as pl
     node_col, src, dst = g._node, g._source, g._destination
@@ -610,10 +611,17 @@ def _edge_cols(op, f):
                 and op._name in colnames(p._edges)):
             eproj = p._edges.filter(pl.col(op._name)).select(pl.col(eid_col))
         frames.extend((nproj, eproj))
+    # Collect lazily-capped at LEAN_MAX_ROWS+1: a frame back at <= cap is COMPLETE (the limit
+    # never truncated) and reusable; a truncated frame means big wavefronts -> return None so
+    # Track B runs on the ORIGINAL lazy steps with full traversal/combine fusion (predicate
+    # pushdown into the traversal is worth more there than reusing a partial probe).
     lazy_idx = [i for i, f in enumerate(frames) if f is not None and is_lazy(f)]
     if lazy_idx:
-        for i, df in zip(lazy_idx, collect_all([frames[i] for i in lazy_idx])):
+        for i, df in zip(lazy_idx,
+                         collect_all([frames[i].limit(LEAN_MAX_ROWS + 1) for i in lazy_idx])):
             frames[i] = df
+    if any(f is not None and f.height > LEAN_MAX_ROWS for f in frames):
+        return None
     steps_m = [
         (op, _LazyShim(frames[2 * i], frames[2 * i + 1], node_col, src, dst, g._edge))
         for i, (op, _) in enumerate(steps)
@@ -624,6 +632,7 @@ def _edge_cols(op, f):
         for i, (op, _) in enumerate(label_steps)
     ]
     if sum(f.height for f in frames if f is not None) > LEAN_MAX_ROWS:
+        # every frame complete, just jointly too big -> safe to reuse eagerly in Track B
         return ("fallback", steps_m, list(label_steps))
 
     # Mapping lookup LAST (after the row guard): None means first sighting of this node frame

From f6e42ca7f12534e72dad2c64d9430cf5f890e10f Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 23:27:46 -0700
Subject: [PATCH 11/16] perf(gfql): memoize truncated lean probes per (frame,
 compiled ops)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

In-memory limit pushdown cannot early-stop joins, so a truncated probe
means the traversal ran once for the probe and again in Track B —
new-topics cypher paid 2x traversal (+36%) on every call. Compiled-plan
caching reuses the ops objects across calls: memoize the truncation per
(edges frame id, ops ids), recycle-safe via weakref.finalize, so warm
calls go straight to the fused Track B plan. A recycled key can only
skip an eligible lean attempt — conservative, never incorrect.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 22 +++++++++++++++++--
 1 file changed, 20 insertions(+), 2 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 88317d12a6..7084551bba 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -494,6 +494,12 @@ def _augmented_edges_cached(edges, eid_col: str):
 _POS_MISS = object()
 _POS_SEEN_ONCE = object()
 
+# (id(edges frame), ids of compiled ops) whose lean probe truncated (big wavefronts): skip the
+# lean attempt entirely on warm calls — the probe would re-execute the traversal for nothing.
+# Recycle-safe: weakref.finalize on the edges frame evicts. A recycled key can only skip an
+# eligible lean attempt (conservative, never incorrect).
+_LEAN_SKIP_MEMO: dict = {}
+
 
 def _node_pos_frame_cached(nodes, node_col: str):
     key = (id(nodes), node_col)
@@ -995,8 +1001,13 @@ def _plain_edge(op):
 
     # Slice-3 lean combine: seeded single-hop chains skip the full-frame Track-B plan entirely
     # (position gathers over the resident frames, O(result) per call). Guards + fallback inside.
-    if not has_multihop and added_edge_index and (
-            start_nodes is not None or (isinstance(ops[0], ASTNode) and ops[0].filter_dict)):
+    # A truncated probe (None) means big wavefronts: the probe collect executed the traversal
+    # once and Track B will execute it again (in-memory limits don't early-stop joins), so
+    # memoize that outcome per (edges frame, compiled ops) — compiled-plan caching reuses the
+    # ops objects, so warm calls skip straight to the fused Track B plan with no probe at all.
+    _lean_key = (id(g._edges), tuple(id(op) for op in ops))
+    if (not has_multihop and added_edge_index and _lean_key not in _LEAN_SKIP_MEMO and (
+            start_nodes is not None or (isinstance(ops[0], ASTNode) and ops[0].filter_dict))):
         lean = _try_combine_lean(g, ops, steps, label_steps, EID)
         if lean is not None and lean[0] == "lean":
             _, final_nodes, final_edges = lean
@@ -1005,6 +1016,13 @@ def _plain_edge(op):
         if lean is not None:  # guard fallback: reuse the materialized steps in Track B
             _, steps, label_steps = lean
             edge_steps = steps
+        else:
+            try:
+                import weakref
+                weakref.finalize(g._edges, _LEAN_SKIP_MEMO.pop, _lean_key, None)
+                _LEAN_SKIP_MEMO[_lean_key] = True
+            except TypeError:  # pragma: no cover - frame not weakref-able -> don't memoize
+                pass
 
     # Track B: build the WHOLE combine (combine_nodes/edges + endpoint + names) as ONE deferred
     # plan over already-materialized hop frames and collect ONCE — collapsing the ~dozen eager

From 4f75f0be8e72fd4d38df2e7771c401361814614f Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 23:32:09 -0700
Subject: [PATCH 12/16] perf(gfql): gate the lean combine on a recurring edges
 frame
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The skip-memo keyed on frame identity never fires for pipelines that
rebind frames per call — every call is a fresh id, so new-topics cypher
kept paying the probe's full traversal (+35%). The augmented-edges cache
already distinguishes the cases: a HIT means this edges frame recurred
(resident graph), a MISS means per-call frames. Require the hit before
attempting the lean probe; resident graphs warm into the lean path on
their second call, rebound-frame pipelines go straight to fused Track B.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 graphistry/compute/gfql/lazy/engine/polars/chain.py      | 9 ++++++++-
 .../tests/compute/gfql/test_engine_polars_chain.py       | 2 ++
 2 files changed, 10 insertions(+), 1 deletion(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 7084551bba..82f13c2cbb 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -909,6 +909,11 @@ def _plain_edge(op):
     g, _endpoint_restore = _align_edge_endpoints(g, g._node, g._source, g._destination)
     if g._edge is None:
         EID = "__gfql_edge_index__"
+        # A cache HIT below means this edges frame recurred across calls (resident graph) —
+        # the signal the lean combine's probe gates on: per-call rebound frames (whole-entity
+        # aggregation pipelines) never recur, and probing them re-executes the traversal for
+        # nothing since in-memory limits can't early-stop joins (new-topics cypher +35% at SF1).
+        _edges_recurred = (id(g._edges), EID) in _AUGMENTED_EDGES_CACHE
         g = g.edges(_augmented_edges_cached(g._edges, EID), g._source, g._destination, edge=EID)
         added_edge_index = True
         # with_row_index only PREPENDS a synthetic id column; the indexed src/dst are
@@ -924,6 +929,7 @@ def _plain_edge(op):
     else:
         EID = g._edge
         added_edge_index = False
+        _edges_recurred = False
 
     node_col, src, dst = g._node, g._source, g._destination
     assert node_col is not None and src is not None and dst is not None
@@ -1006,7 +1012,8 @@ def _plain_edge(op):
     # memoize that outcome per (edges frame, compiled ops) — compiled-plan caching reuses the
     # ops objects, so warm calls skip straight to the fused Track B plan with no probe at all.
     _lean_key = (id(g._edges), tuple(id(op) for op in ops))
-    if (not has_multihop and added_edge_index and _lean_key not in _LEAN_SKIP_MEMO and (
+    if (not has_multihop and added_edge_index and _edges_recurred
+            and _lean_key not in _LEAN_SKIP_MEMO and (
             start_nodes is not None or (isinstance(ops[0], ASTNode) and ops[0].filter_dict))):
         lean = _try_combine_lean(g, ops, steps, label_steps, EID)
         if lean is not None and lean[0] == "lean":
diff --git a/graphistry/tests/compute/gfql/test_engine_polars_chain.py b/graphistry/tests/compute/gfql/test_engine_polars_chain.py
index 2535794503..18083bc397 100644
--- a/graphistry/tests/compute/gfql/test_engine_polars_chain.py
+++ b/graphistry/tests/compute/gfql/test_engine_polars_chain.py
@@ -746,6 +746,7 @@ def test_lean_vs_trackb_byte_identical(self, label, ch, expect_lean, monkeypatch
         from graphistry.compute.gfql.lazy.engine.polars import chain as chain_mod
 
         g = self._graph()
+        g.chain(ch, engine="polars")  # warm: lean gates on a recurring (resident) edges frame
         engaged = []
         orig = chain_mod._try_combine_lean
 
@@ -775,6 +776,7 @@ def test_lean_falls_back_on_dup_node_ids(self, monkeypatch):
         nodes = pl.concat([_LEAN_NODES, _LEAN_NODES.filter(pl.col("id") == "b")])
         g = graphistry.nodes(nodes, "id").edges(_LEAN_EDGES, "s", "d")
         ch = [n({"id": "a"}, name="src"), e_forward(name="E"), n()]
+        g.chain(ch, engine="polars")  # warm: lean gates on a recurring (resident) edges frame
 
         engaged = []
         orig = chain_mod._try_combine_lean

From c3a6255b3f9987fed4b67fd5ddd72d83fdf3e56e Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 23:38:17 -0700
Subject: [PATCH 13/16] perf(gfql): structural key for the lean skip-memo
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The trace on the real new-topics run showed the same edges frame
recurring but FRESH op objects per call — the identity-keyed memo could
never hit, so the truncating probe re-ran the traversal every call
(+40%). Key the memo on the structural chain shape instead (op kinds,
directions, hops, aliases, filter KEYS — seed values vary while fan-out
shape doesn't); collisions can only skip an eligible attempt.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../compute/gfql/lazy/engine/polars/chain.py  | 27 ++++++++++++++-----
 1 file changed, 20 insertions(+), 7 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 82f13c2cbb..2e58acbcc5 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -494,13 +494,26 @@ def _augmented_edges_cached(edges, eid_col: str):
 _POS_MISS = object()
 _POS_SEEN_ONCE = object()
 
-# (id(edges frame), ids of compiled ops) whose lean probe truncated (big wavefronts): skip the
-# lean attempt entirely on warm calls — the probe would re-execute the traversal for nothing.
-# Recycle-safe: weakref.finalize on the edges frame evicts. A recycled key can only skip an
-# eligible lean attempt (conservative, never incorrect).
+# (id(edges frame), structural chain shape) whose lean probe truncated (big wavefronts): skip
+# the lean attempt entirely on warm calls — the probe would re-execute the traversal for
+# nothing. The key is STRUCTURAL (op kinds/directions/hops/aliases/filter KEYS — not values,
+# and not object ids: lowering builds fresh op objects per call, and seed values vary while
+# fan-out shape doesn't). Recycle-safe: weakref.finalize on the edges frame evicts. A recycled
+# or colliding key can only skip an eligible lean attempt (conservative, never incorrect).
 _LEAN_SKIP_MEMO: dict = {}
 
 
+def _lean_ops_key(ops) -> tuple:
+    parts = []
+    for op in ops:
+        if isinstance(op, ASTEdge):
+            parts.append(("e", op.direction, op.hops, op.min_hops, op.max_hops, op._name,
+                          tuple(sorted(op.edge_match or {}))))
+        else:
+            parts.append(("n", op._name, tuple(sorted(getattr(op, "filter_dict", None) or {}))))
+    return tuple(parts)
+
+
 def _node_pos_frame_cached(nodes, node_col: str):
     key = (id(nodes), node_col)
     hit = _NODE_POS_CACHE.get(key, _POS_MISS)
@@ -1009,9 +1022,9 @@ def _plain_edge(op):
     # (position gathers over the resident frames, O(result) per call). Guards + fallback inside.
     # A truncated probe (None) means big wavefronts: the probe collect executed the traversal
     # once and Track B will execute it again (in-memory limits don't early-stop joins), so
-    # memoize that outcome per (edges frame, compiled ops) — compiled-plan caching reuses the
-    # ops objects, so warm calls skip straight to the fused Track B plan with no probe at all.
-    _lean_key = (id(g._edges), tuple(id(op) for op in ops))
+    # memoize that outcome per (edges frame, structural chain shape) — warm calls of the same
+    # query shape skip straight to the fused Track B plan with no probe at all.
+    _lean_key = (id(g._edges), _lean_ops_key(ops))
     if (not has_multihop and added_edge_index and _edges_recurred
             and _lean_key not in _LEAN_SKIP_MEMO and (
             start_nodes is not None or (isinstance(ops[0], ASTNode) and ops[0].filter_dict))):

From 27d0ed0b4ecedbb81898ecaec2dcb8aba4d9f47f Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Sun, 19 Jul 2026 23:50:14 -0700
Subject: [PATCH 14/16] chore(gfql): type the lean ops-key accumulator (mypy
 strict)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 graphistry/compute/gfql/lazy/engine/polars/chain.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py
index 2e58acbcc5..b30a80149c 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/chain.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py
@@ -504,7 +504,7 @@ def _augmented_edges_cached(edges, eid_col: str):
 
 
 def _lean_ops_key(ops) -> tuple:
-    parts = []
+    parts: List[Tuple[Any, ...]] = []
     for op in ops:
         if isinstance(op, ASTEdge):
             parts.append(("e", op.direction, op.hops, op.min_hops, op.max_hops, op._name,

From e1a8d0e1f0832c91fa9b64c8d00227e3e8f54854 Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Mon, 20 Jul 2026 00:32:26 -0700
Subject: [PATCH 15/16] perf(gfql): stop re-resolving lazy schemas per
 predicate (slice 4a)

LazyFrame.schema / .columns resolve the schema AND emit a
PerformanceWarning on every access; both ran per predicate per call on
the seeded fast path (the SF1 profile showed the two warning sites
firing on every query). Route polars schema reads through collect_schema
via a local _schema_of, and give filter_by_dict a lazy-safe column
probe. Zero PerformanceWarnings now; 200k/1M IS1 warm 12.1 -> 11.4ms.
Deliberately avoids row_pipeline.py so this does not conflict with the
pending #1714 OLAP rebase.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 graphistry/compute/filter_by_dict.py          | 19 +++++++++++++++----
 .../gfql/lazy/engine/polars/predicates.py     | 18 +++++++++++++-----
 2 files changed, 28 insertions(+), 9 deletions(-)

diff --git a/graphistry/compute/filter_by_dict.py b/graphistry/compute/filter_by_dict.py
index 837d4f9096..e330e249cd 100644
--- a/graphistry/compute/filter_by_dict.py
+++ b/graphistry/compute/filter_by_dict.py
@@ -82,15 +82,26 @@ def _label_series_contains(series: Any, label: str) -> Any:
     return series.apply(lambda value: label in _normalize_labels_cell(value))
 
 
+def _column_names(df: DataFrameT) -> Any:
+    """Column names without resolving a polars LazyFrame schema via ``.columns``, which warns
+    (PerformanceWarning) and re-resolves on every access — this runs per predicate per call on
+    the seeded fast path. ``collect_schema().names()`` is the sanctioned lazy form."""
+    collect_schema = getattr(df, "collect_schema", None)
+    if collect_schema is not None and type(df).__name__ == "LazyFrame":
+        return collect_schema().names()
+    return df.columns
+
+
 def resolve_filter_column(df: DataFrameT, col: str, val: Any) -> Tuple[str, Any]:
-    if col in df.columns:
+    names = _column_names(df)
+    if col in names:
         return col, val
 
     if col.startswith("label__") and val is True:
         label = col[len("label__") :]
-        if "labels" in df.columns:
+        if "labels" in names:
             return "labels", label
-        if "type" in df.columns and not _looks_like_edge_dataframe(df):
+        if "type" in names and not _looks_like_edge_dataframe(df):
             return "type", label
 
     from graphistry.compute.exceptions import ErrorCode, GFQLSchemaError
@@ -100,7 +111,7 @@ def resolve_filter_column(df: DataFrameT, col: str, val: Any) -> Tuple[str, Any]
         f'Column "{col}" does not exist in dataframe',
         field=col,
         value=val,
-        suggestion=f'Available columns: {", ".join(df.columns[:10])}{"..." if len(df.columns) > 10 else ""}'
+        suggestion=f'Available columns: {", ".join(list(names)[:10])}{"..." if len(names) > 10 else ""}'
     )
 
 
diff --git a/graphistry/compute/gfql/lazy/engine/polars/predicates.py b/graphistry/compute/gfql/lazy/engine/polars/predicates.py
index 5bad911bd8..ee639078ce 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/predicates.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/predicates.py
@@ -15,12 +15,20 @@
 from graphistry.compute.predicates.ASTPredicate import ASTPredicate
 from graphistry.compute.predicates.str import Contains, Endswith, Fullmatch, Match, Startswith
 from graphistry.compute.filter_by_dict import resolve_filter_column
-from .dtypes import is_numeric as _dtype_numeric, is_stringlike as _dtype_stringlike
+from .dtypes import is_lazy, is_numeric as _dtype_numeric, is_stringlike as _dtype_stringlike
+
 
 if TYPE_CHECKING:
     import datetime
     import polars as pl
 
+
+def _schema_of(df: Any) -> Any:
+    """Schema for an eager or lazy frame. ``LazyFrame.schema`` resolves the schema AND warns
+    (PerformanceWarning) on every access; ``collect_schema()`` is the sanctioned form. Called
+    per predicate per call on the seeded fast path, so the repeated resolution is not free."""
+    return df.collect_schema() if is_lazy(df) else df.schema
+
 # Comparison-predicate RHS: genuinely dynamic (Cypher properties are dynamically typed) — a
 # python scalar or a GFQL/py temporal matched structurally by type(val).__name__
 # (DateValue/TemporalValue/…, never imported here).
@@ -290,7 +298,7 @@ def _is_cross_type_predicate(df: "pl.DataFrame", col: str, pred: ASTPredicate) -
     name = type(pred).__name__
     if name == "AllOf" and hasattr(pred, "predicates"):
         return any(_is_cross_type_predicate(df, col, p) for p in pred.predicates)
-    dtype = df.schema.get(col)
+    dtype = _schema_of(df).get(col)
     col_num = _dtype_numeric(dtype)
     col_str = _dtype_stringlike(dtype)
 
@@ -331,7 +339,7 @@ def filter_by_dict_polars(df: "pl.DataFrame", filter_dict: "Optional[Dict[str, A
             # on a Categorical/Enum/numeric column. Raise the SAME clean, typed error so all three
             # engines agree (categorical is treated as non-string here, exactly as filter_by_dict).
             if isinstance(resolved_val, (Contains, Startswith, Endswith, Match)):
-                _col_dtype = df.schema.get(resolved_col)
+                _col_dtype = _schema_of(df).get(resolved_col)
                 if _col_dtype is not None and _col_dtype != pl.String:
                     from graphistry.compute.exceptions import ErrorCode, GFQLSchemaError
                     raise GFQLSchemaError(
@@ -342,7 +350,7 @@ def filter_by_dict_polars(df: "pl.DataFrame", filter_dict: "Optional[Dict[str, A
                         column_type=str(_col_dtype),
                         suggestion='Use numeric predicates like gt() or lt() for numeric columns',
                     )
-            expr = predicate_to_expr(resolved_col, resolved_val, df.schema.get(resolved_col))
+            expr = predicate_to_expr(resolved_col, resolved_val, _schema_of(df).get(resolved_col))
             if expr is None:
                 # decline (NIE): no native lowering for this predicate; no pandas bridge.
                 raise NotImplementedError(
@@ -354,7 +362,7 @@ def filter_by_dict_polars(df: "pl.DataFrame", filter_dict: "Optional[Dict[str, A
             exprs.append(expr)
         elif _is_membership(resolved_val):
             exprs.append(pl.col(resolved_col).is_in(list(resolved_val)))
-        elif isinstance(df.schema.get(resolved_col), pl.List):
+        elif isinstance(_schema_of(df).get(resolved_col), pl.List):
             if resolved_col == "labels":
                 # MATCH (n:Label) = scalar match on the RESERVED `labels` List column ->
                 # list.contains (empty for a non-existent label, matching pandas). A plain ==

From 639ba8becd838cc42e54215485beb45d3e86821f Mon Sep 17 00:00:00 2001
From: Leo Meyerovich <leo@graphistry.com>
Date: Mon, 20 Jul 2026 00:55:42 -0700
Subject: [PATCH 16/16] perf(gfql): fuse the per-column temporal-constructor
 probe (slice 4b)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

_select_emits_temporal_constructor_text called the per-column
_has_temporal_constructor_text helper in a loop, each doing its own
select().item() collect — 6 collects per call on LDBC IS1. Scan every
String column in ONE fused select instead (same ^-anchored pattern, same
any() semantics, same swallow-failures-as-False parity), and read the
schema lazily-safely. 71 -> 66 collects/call on the 200k/1M probe.
1233 chain+row-pipeline+conformance tests green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL
---
 .../gfql/lazy/engine/polars/row_pipeline.py   | 24 +++++++++++++++----
 1 file changed, 19 insertions(+), 5 deletions(-)

diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index c2c0875c50..5112ea015d 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
@@ -705,11 +705,25 @@ def _project_polars(g: Plottable, items: Sequence[SelectItem], extend: bool) ->
 
 def _select_emits_temporal_constructor_text(out: Any) -> bool:
     import polars as pl
-    from graphistry.compute.gfql.lazy.engine.polars.projection import _has_temporal_constructor_text
-    for name, dtype in out.schema.items():
-        if dtype == pl.String and _has_temporal_constructor_text(out, name):
-            return True
-    return False
+    from graphistry.compute.gfql.temporal.constructors import TEMPORAL_CALL_EXPR_RE
+    from .dtypes import is_lazy
+    schema = out.collect_schema() if is_lazy(out) else out.schema
+    string_cols = [name for name, dtype in schema.items() if dtype == pl.String]
+    if not string_cols:
+        return False
+    # ONE fused scan for every String column, not one collect per column (the per-column
+    # _has_temporal_constructor_text probe cost a collect each — 6 per call on LDBC IS1).
+    # Same ^-anchored pattern and same any() semantics as the per-column helper.
+    pattern = r"^\s*" + TEMPORAL_CALL_EXPR_RE.pattern
+    try:
+        probe = out.select([
+            pl.col(name).str.contains(pattern).any().alias(name) for name in string_cols
+        ])
+        row = (probe.collect() if is_lazy(probe) else probe).row(0)
+    except Exception:
+        # Parity with the per-column helper, which swallowed its own failures as False.
+        return False
+    return any(bool(v) for v in row)
 
 
 def select_polars(g: Plottable, items: Sequence[SelectItem]) -> Optional[Plottable]: