perf(gfql): kill per-call O(E) terms in seeded Cypher/chain queries (#1755 slice 1)

CHANGELOG.md

@@ -19,6 +19,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL polars engine natively runs multi-source (node-cartesian) MATCH (#1273)**: comma-separated disconnected aliases — `g.gfql("MATCH (a {..}), (b {..}) RETURN a.id, b.id")` — previously declined on polars (the `rows(binding_ops)` `node_cartesian` branch hard-NIE'd) while pandas handled them. New `_cartesian_node_bindings_polars` mirrors the pandas oracle `_gfql_cartesian_node_bindings_row_table`: each alias is independently filtered, projected into the per-alias lookup schema (bare `alias` id, `alias.id`, `alias.<prop>`, and pandas' leaked `alias.alias=True` flag incl. property-shadowing), then left-major cross-joined for row-order parity. Scalar/aliased/`count(*)` projections over the cartesian match pandas exactly (verified end-to-end); whole-entity `RETURN a, b` stays an honest NIE (separate projection surface). Two parity-safe declines mirror shapes where pandas itself errors (proven on master, so both engines fail identically, never diverge): an anonymous node op, and ≥4 named aliases (pandas' bare-id merge residue collides). Differential fuzz vs pandas over ~10k lowered cases: 0 disagreements. Tests in `test_engine_polars_binding_rows.py`.
 - **GFQL polars chain — variable-length node aliases are hop-distance gated; the chain is now silent-wrong-free (#1741)**: a node named after a variable-length edge (`g.gfql("MATCH (a)-[*1..2]-(b) RETURN b")`) previously carried its alias regardless of hop distance, so an undirected walk that backtracked into the seed wrongly returned it (pandas correctly excludes it — trail semantics). The polars chain now auto-injects the #1741 hop-distance label (name resolved against the user's node columns via the `reserved_columns` registry, so a user column named like the internal one is never clobbered) and applies pandas' alias `[min_hop, max_hop]` window in `_apply_node_names`. The one shape the gate can't yet cover — a node alias after a **forward/reverse `min_hops>1`** edge, whose labels need pandas' layered backward walk (not ported) — now **declines with an honest `NotImplementedError` (#1748)** instead of returning nodes outside the window; the decline is precise (differential vs pandas: 30/30 named shapes NIE, 30/30 unnamed run and match, 0 over-decline). 4-engine A/B on the stacked build: pandas/cudf 144/144, polars/polars-gpu 112 agree / **0 disagree** / 32 honest-NIE — zero silent-wrong shapes remain on the polars chain. Adapts the mechanism from the retired #1742 (declined undirected varlen+alias, now natively gated). Tests: `TestVarlenAliasHopGate` (pandas-parity across directions, the `*2..3` decline + unnamed-still-runs pins, a column-collision test).
 - **GFQL native polars `label_node_hops` on the plain BFS — correct, direction-dependent hop labels (#1741 groundwork)**: the polars eager hop now emits pandas-parity node hop-distance labels instead of declining `label_node_hops`. The labeling rule is DIRECTION-DEPENDENT (the subtle part, derived by differential fuzz vs the pandas oracle, not by reading pandas alone): forward/reverse label EVERY destination of a hop first-wins (matching pandas `hop.py` `new_node_ids`), so a seed re-entered at hop 1 IS labeled; undirected labels destinations MINUS everything already visited and pre-seeds the seen-set with the seeds, so a seed re-reached by a backtracking walk stays NULL (its shortest-path distance). Two correctness fixes over the first cut: (1) the undirected seed pre-seed must NOT depend on `return_as_wave_front` — a seed filtered out of the frontier by `source_node_match` or suppressed in wave-front mode was being re-labeled (A/B vs pandas over direction × hops × `return_as_wave_front` × `source_node_match` × `destination_node_match` × 10 graphs: was 456/24, now 480/480); (2) a requested `label_node_hops` name that collides with an existing column is redirected to `<name>_1` like pandas' `resolve_label_col`, not the polars left-join auto-suffix `<name>_right`/`DuplicateError`. `label_edge_hops` and `min_hops>1` labels stay honest NIEs. Amplified `test_engine_polars_hop.py` (`TestHopLabelsDifferential` over the seed-filter / wave-front / multi-seed axes, an explicit direction-asymmetry contrast pin, and a collision test).
+- **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).

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

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

@@ -522,6 +522,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
@@ -746,7 +771,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

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

@@ -83,6 +83,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
@@ -1171,7 +1211,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 = {