review: share the bounded var-length expansion; type the polars path bag lazily

- extract `_directed_varlen_reachable_polars` so the unbounded fixed-point arm and
  the bounded `-[*1..k]->` arm run literally the same loop instead of two copies
  kept in sync by a comment.
- pin the generic bindings builder's path bag as a LazyFrame. It always was one;
  mypy inferred eager because `filter_by_dict_polars` declared the eager type.
- make `filter_by_dict_polars` frame-polymorphic via a constrained TypeVar
  (DataFrame | LazyFrame) rather than declaring one flavour and being called with
  both -- the eager viz lane and the lazy chain lane take the same `.filter(expr)`
  path, and the TypeVar keeps the caller's flavour on the way out. Removes the
  mypy noise this file was carrying, so the new code lands at delta 0.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NxxTFpKiqA4FQa9Pjy8yR1

Author: lmeyerov

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

@@ -10,7 +10,7 @@
 
 import operator
 import re
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, TypeVar, Union
 
 from graphistry.compute.predicates.ASTPredicate import ASTPredicate
 from graphistry.compute.predicates.str import Contains, Endswith, Fullmatch, Match, Startswith
@@ -21,6 +21,12 @@
     import datetime
     import polars as pl
 
+    # These helpers are frame-polymorphic: the eager (viz/crossfilter) lane hands them
+    # DataFrames, the lazy chain/bindings lane hands them LazyFrames, and both take the
+    # SAME `.filter(expr)` path. A constrained TypeVar says exactly that and keeps the
+    # caller's flavour on the way out (a plain union would not).
+    PolarsFrameT = TypeVar("PolarsFrameT", "pl.DataFrame", "pl.LazyFrame")
+
 # 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).
@@ -282,7 +288,7 @@ def _is_membership(value: Any) -> bool:
     return isinstance(value, (list, tuple, set, frozenset))
 
 
-def _is_cross_type_predicate(df: "pl.DataFrame", col: str, pred: ASTPredicate) -> bool:
+def _is_cross_type_predicate(df: "Union[pl.DataFrame, pl.LazyFrame]", col: str, pred: ASTPredicate) -> bool:
     """True iff the predicate compares a numeric column to a string value (or vice versa):
     polars raises `cannot compare string with numeric type` (an uncatchable Rust panic when
     nested); pandas/cypher return a value/null. Recurses into AllOf (fold of x>a AND x<b) and
@@ -307,15 +313,15 @@ def _mismatch(v: Any) -> bool:
     return _mismatch(val)
 
 
-def filter_by_dict_polars(df: "pl.DataFrame", filter_dict: "Optional[Dict[str, Any]]") -> "pl.DataFrame":
+def filter_by_dict_polars(df: "PolarsFrameT", filter_dict: "Optional[Dict[str, Any]]") -> "PolarsFrameT":
     """Return rows of polars ``df`` matching all entries in ``filter_dict`` via one filter."""
     combined = filter_expr_by_dict_polars(df, filter_dict)
     if combined is None:
         return df
     return df.filter(combined)
 
 
-def filter_expr_by_dict_polars(df: "pl.DataFrame", filter_dict: "Optional[Dict[str, Any]]") -> "Optional[pl.Expr]":
+def filter_expr_by_dict_polars(df: "Union[pl.DataFrame, pl.LazyFrame]", filter_dict: "Optional[Dict[str, Any]]") -> "Optional[pl.Expr]":
     """Build the combined boolean ``pl.Expr`` filter_by_dict_polars would apply, or None
     for an empty/absent filter dict. ``df`` supplies the schema for column/dtype
     resolution only — callers may apply the expr to a LazyFrame over the same schema

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

@@ -1094,6 +1094,41 @@ def _cartesian_node_bindings_polars(
     return _rewrap(g, out_df)
 
 
+def _directed_varlen_reachable_polars(
+    state: "pl.LazyFrame",
+    pairs: "pl.LazyFrame",
+    state_cols: List[str],
+    *,
+    min_hops: int,
+    max_hops: int,
+) -> "pl.LazyFrame":
+    """Bounded DIRECTED variable-length expansion of a bindings path bag.
+
+    One row per distinct edge SEQUENCE: ``pairs`` is NOT deduped, so parallel edges
+    multiply per hop, matching pandas' ``_gfql_multihop_binding_rows`` merge. Zero-hop
+    rows (``min_hops == 0``) keep the seed row (endpoint == start) and come first, then
+    hop 1, 2, ... — the same ``reachable`` concat order pandas builds.
+
+    Stays fully lazy: all ``max_hops`` iterations are built without an eager
+    ``.height`` early-break, because an empty intermediate lazily joins to empty and
+    yields the identical result (pandas' break is an optimization, not semantics).
+    """
+    import polars as pl
+
+    reachable: List["pl.LazyFrame"] = [state] if min_hops == 0 else []
+    current = state
+    for hop in range(1, max_hops + 1):
+        current = (
+            current.join(pairs, left_on="__current__", right_on="__from__", how="inner")
+            .drop("__current__")
+            .rename({"__to__": "__current__"})
+            .select(state_cols)
+        )
+        if hop >= min_hops:
+            reachable.append(current)
+    return pl.concat(reachable, how="vertical") if reachable else state.limit(0)
+
+
 def _directed_fixed_point_binding_rows_polars(
     state: "pl.LazyFrame",
     pairs: "pl.LazyFrame",
@@ -1169,19 +1204,10 @@ def _directed_fixed_point_binding_rows_polars(
             "Cypher multi-alias row bindings currently require terminating variable-length segments"
         )
 
-    # (b) bounded path expansion, identical to the `-[*1..k]->` arm with max_hops=depth.
-    reachable: List["pl.LazyFrame"] = [state] if min_hops == 0 else []
-    current = state
-    for hop in range(1, depth + 1):
-        current = (
-            current.join(pairs_lf, left_on="__current__", right_on="__from__", how="inner")
-            .drop("__current__")
-            .rename({"__to__": "__current__"})
-            .select(state_cols)
-        )
-        if hop >= min_hops:
-            reachable.append(current)
-    return pl.concat(reachable, how="vertical") if reachable else state.limit(0)
+    # (b) the SAME bounded expansion the `-[*1..k]->` arm runs, with max_hops = depth.
+    return _directed_varlen_reachable_polars(
+        state, pairs_lf, state_cols, min_hops=min_hops, max_hops=depth
+    )
 
 
 def binding_rows_polars(
@@ -1397,7 +1423,11 @@ def _names(lf: pl.LazyFrame) -> List[str]:
         if seed_ids_lf is not None:
             # WITH->MATCH re-entry seed: constrain the first alias to the carried ids.
             seed_nodes = seed_nodes.join(seed_ids_lf, on=node_id, how="semi")
-        state = seed_nodes.select(pl.col(node_id).alias("__current__"))
+        # The whole generic builder works in LazyFrames (`nodes_lf` / `edges_lf` above);
+        # `filter_by_dict_polars` is frame-polymorphic at runtime but declares the eager
+        # type, so pin the path bag lazy here instead of leaving every downstream lazy
+        # op to fight an eager inference.
+        state: pl.LazyFrame = seed_nodes.select(pl.col(node_id).alias("__current__"))  # type: ignore[assignment]
         alias_frames: Dict[str, pl.LazyFrame] = {}
         node_aliases: List[str] = []
         first_alias = first_op._name
@@ -1528,23 +1558,14 @@ def _names(lf: pl.LazyFrame) -> List[str]:
                             reachable.append(current.select(state_cols))
                     state = pl.concat(reachable, how="vertical") if reachable else state.limit(0)
                 else:
-                    max_hops = int(max_hops_value)
-                    pairs = oriented.select(["__from__", "__to__"])
-                    reachable = [state] if min_hops == 0 else []
-                    current = state
-                    # Lazy: build all max_hops iterations (no eager .height early-break —
-                    # empty intermediates lazily join to empty, so the result is
-                    # identical; the pandas break is an optimization, not semantics).
-                    for _hop in range(1, max_hops + 1):
-                        current = (
-                            current.join(pairs, left_on="__current__", right_on="__from__", how="inner")
-                            .drop("__current__")
-                            .rename({"__to__": "__current__"})
-                            .select(state_cols)
-                        )
-                        if _hop >= min_hops:
-                            reachable.append(current)
-                    state = pl.concat(reachable, how="vertical") if reachable else state.limit(0)
+                    # Bounded directed var-length (`-[*1..k]->`, graph-bench q3).
+                    state = _directed_varlen_reachable_polars(
+                        state,
+                        oriented.select(["__from__", "__to__"]),
+                        state_cols,
+                        min_hops=min_hops,
+                        max_hops=int(max_hops_value),
+                    )
             else:
                 state = (
                     state.join(oriented, left_on="__current__", right_on="__from__", how="inner")

graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py

@@ -630,15 +630,24 @@ def test_run_calls_polars_binding_ops_native():
 
 
 def test_run_calls_polars_binding_ops_unbounded_multihop_defers():
-    """UNBOUNDED variable-length binding patterns stay outside the native subset
-    (bounded `-[*1..k]->` is native, #1709) -> NotImplementedError (NO pandas
-    bridge, see plan.md NO-CHEATING)."""
+    """Unbounded variable-length binding patterns: DIRECTED fixed point is native
+    (#1709, LDBC IS6); the rest stay outside the subset -> NotImplementedError (NO
+    pandas bridge, see plan.md NO-CHEATING)."""
     from graphistry.compute.gfql.lazy.engine.polars.chain import _run_calls_polars
-    from graphistry.compute.ast import call, n, e_forward
+    from graphistry.compute.ast import call, n, e_forward, e_undirected
     g = _polars_graph()
-    middle = [n(name="a"), e_forward(to_fixed_point=True), n(name="b")]
-    with pytest.raises(NotImplementedError):
-        _run_calls_polars(g, [call("rows", {})], None, g, middle)
+    native = _run_calls_polars(
+        g, [call("rows", {})], None, g, [n(name="a"), e_forward(to_fixed_point=True), n(name="b")]
+    )
+    assert "polars" in type(native._nodes).__module__
+    for middle in [
+        # undirected unbounded: multiplicity + backtrack-aware termination unmodeled
+        [n(name="a"), e_undirected(to_fixed_point=True), n(name="b")],
+        # aliased variable-length relationship: pandas rejects it outright
+        [n(name="a"), e_forward(to_fixed_point=True, name="r"), n(name="b")],
+    ]:
+        with pytest.raises(NotImplementedError):
+            _run_calls_polars(g, [call("rows", {})], None, g, middle)
 
 
 def test_frame_ops_polars_rows_empty_table():