diff --git a/CHANGELOG.md b/CHANGELOG.md index ef993819b4..8bf135544b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Added +- **Native polars `rows(binding_ops=...)` for UNBOUNDED directed variable-length patterns (`-[*]->` / `-[*0..]->`) (#1709)**: the Cypher multi-alias bindings table already lowered natively for fixed-length and *bounded* variable-length segments, but an unbounded fixed-point segment declined with `NotImplementedError: polars engine does not yet natively support cypher row op 'rows'`. This was the last shape blocking `engine='polars'` on LDBC SNB interactive-short-6 (`MATCH (m:Message)-[:REPLY_OF*0..]->(p:Post)<-[:CONTAINER_OF]-(f:Forum)-[:HAS_MODERATOR]->(mod)`), the one interactive-short query polars could not answer. It now runs natively: a dedup-by-node frontier walk finds the exhaustion depth, then the SAME lazy bounded pair-join loop the `-[*1..k]->` arm uses materializes one row per distinct edge SEQUENCE (Cypher path multiplicity, parallel edges included) — no pandas bridge, no `to_pandas()` round trip. A cycle reachable from the seed means infinitely many paths; that raises the same E108 "require terminating variable-length segments" error pandas raises — same exception class and same `.code` on both engines — and is detected before the path expansion blows up rather than after, bounded by the REACHABLE node count so an unreachable remainder of the graph costs nothing. Still declining honestly (NIE, never a silent answer): UNDIRECTED unbounded (`-[*]-`, needs the min_hops == 1 multiplicity reconstruction plus backtrack-aware termination — pandas rejects it outright), aliased variable-length relationships (pandas rejects those too), unbounded segments WITHOUT `to_fixed_point` (pandas silently truncates at a bound this lowering cannot reconstruct), unbounded segments with `min_hops >= 2` (`-[*2..]->`: pandas' step pairs are pruned by min_hops against a dedup-by-node eccentricity, which this raw-edge reconstruction cannot reproduce — serving it would return a different count with no error), and `to_fixed_point` combined with an explicit bound (declined on master too; it is not Cypher-reachable, since the parser only sets the flag for `*` / `*k..` where there is no maximum, but through the AST surface it hits the same reconstruction gap for `min_hops >= 3`). Cross-engine parity is the gate: differential fuzz vs the pandas oracle over random DAGs and cyclic graphs with self-loops and parallel edges, plus pinned IS6/zero-hop/multiplicity/cycle tests. + - **GFQL secondary node property indexes (`create_index('node_prop', column=...)` / `g.gfql_index_node_props([...])`)**: a seed predicate on a NON-key column — `MATCH (m {id: 42})` where the graph's node id binding is some other column — previously cost a full node scan, because the registry only indexed the node-id binding and the CSR adjacencies. A property index is the same pay-as-you-go sidecar as the existing kinds: sorted distinct values over node **row positions** (CSR, so duplicate values are indexable), never reorders `.nodes`, fingerprint-validated so a `.nodes()` rebind is treated as absent (safe miss, never a wrong answer), engine-polymorphic (numpy host / cupy on-device), and policy-gated (`off`/`use`/`auto`/`force`). The seeded fixed-hop planner picks the **most selective** indexed scalar predicate in the seed filter using a free CSR-offset estimate, gathers those candidates, and applies every remaining predicate to them — so results are identical whether the index is present, absent, stale, or cost-gated out. `show_indexes()` lists property indexes; `drop_index('node_prop', column=...)` drops one. Only integer columns are indexable today (float NaN ordering, strings on cupy, and nulls all decline to the scan); widening that is additive. **Perf (dgx-spark, official LDBC SNB SF1, 3.18M nodes, warm median, value-identical 19-row result):** interactive-short IS7 `71.6 ms -> 19.5 ms` (**3.7x**), with a one-time `112 ms` build — the seed lookup itself goes from a `51.2 ms` scan to `0.096 ms`. ### Performance @@ -20,6 +22,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - **GFQL execution context is declared rather than attached at runtime**: the private `_gfql_*` per-execution fields (index policy and registry, row-pipeline base graph, carried seed nodes, edge aliases, shortest-path backend, and the indexed-bindings handoff) are now declared on `Plottable` with defaults on `PlotterBase`, instead of being set with `setattr` and read back with `getattr(..., default)`. No public API or behaviour change; internal call sites are typed, and hand-rolled `Plottable` stand-ins must now construct the full context. ### Fixed +- **Native polars variable-length bindings dropped zero-hop rows (`-[*0..k]->`)**: the polars `rows(binding_ops=...)` builder rebuilt the bindings table from the CHAIN OUTPUT, while the pandas oracle rebuilds from the pre-chain base graph. The traversal prunes to what IT considers matched, which is not the same set the bindings builder matches — a zero-hop variable-length segment binds a seed that has no matching outgoing edge, and the traversal drops that node entirely. So `MATCH (a)-[*0..2]->(b)` could silently return fewer rows than pandas (no error, just missing rows). The polars builder now sources its node/edge tables from the same pre-chain base graph pandas uses (which is also what the indexed builder was already handed). Found by differential fuzzing while adding unbounded support (#1709); pinned by a zero-hop regression test and by `-[*0..2]->` parity cases. + - **Indexed bypass could return every node for `rows()` over an unnamed pattern**: the boundary accepted a `rows()` call carrying no binding ops over a middle with no aliases. That call reads the traversal-narrowed node table, but the bypass hands it the full graph, so the query would have returned all nodes. The gate now requires that the rows call actually consumes the whole middle as binding ops — either it already carries them, or the named-middle rewrite installs exactly them. - **Seeded property-RETURN dtype divergence on cuDF**: the lean projection applied the pandas rows-pivot artifact (int → float64, bool → object) on every engine, but cuDF's canonical pivot preserves the source dtypes — so the fast path returned `float64`/`object` where cuDF's own canonical path returns `int64`/`bool`. The cast rule is now engine-aware; the dtype-class decline guard is unchanged. diff --git a/graphistry/compute/gfql/index/bindings.py b/graphistry/compute/gfql/index/bindings.py index 0cb5487b21..7466666718 100644 --- a/graphistry/compute/gfql/index/bindings.py +++ b/graphistry/compute/gfql/index/bindings.py @@ -149,7 +149,9 @@ def _filter_frame( filter_by_dict_polars, ) - return cast(DataFrameT, filter_by_dict_polars(frame, filter_dict)) # type: ignore[arg-type] + # frame is DataFrameT (engine-neutral); on this branch it IS a polars frame, + # which the helper's constrained TypeVar cannot see through the alias + return cast(DataFrameT, filter_by_dict_polars(frame, filter_dict)) # type: ignore[type-var] from graphistry.compute.filter_by_dict import filter_by_dict return filter_by_dict(frame, filter_dict, engine) # type: ignore[arg-type] diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index dd5dbf42bd..d870c0cfc4 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -401,8 +401,27 @@ def _run_calls_polars(g_cur, calls, start_nodes, base_graph, middle): # is MECHANICAL (is_row_pipeline_call), not curated. (umap/hypergraph never reach here — # the generic chain routes schema-changers straight to execute_call.) from graphistry.compute.gfql.row.pipeline import is_row_pipeline_call + from graphistry.compute.exceptions import ErrorCode, GFQLTypeError, GFQLValidationError for op in calls: - native = _try_native_row_op(g_cur, op) + try: + native = _try_native_row_op(g_cur, op) + except GFQLTypeError: + raise + except GFQLValidationError as validation_error: + # Same wrapping `execute_call` applies (gfql/call/executor.py): a kernel that + # raises a validation error surfaces as GFQLTypeError(E303) with this message + # shape. The native attempt runs BEFORE execute_call, so without this the SAME + # query carries a different class AND a different `.code` per engine — and this + # repo's control flow keys on `.code`. Scoped to GFQLValidationError, the one + # divergence actually observed (an E108 from the var-length cycle guard); + # other exception classes are left alone rather than blanket-normalized. + fn_name = getattr(op, "function", None) + raise GFQLTypeError( + ErrorCode.E303, + f"Error executing '{fn_name}': {validation_error}", + field="function", + value=fn_name, + ) from validation_error if native is not None: g_cur = native continue diff --git a/graphistry/compute/gfql/lazy/engine/polars/predicates.py b/graphistry/compute/gfql/lazy/engine/polars/predicates.py index 5c758a5df8..a19de87ec5 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/predicates.py +++ b/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 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: @@ -315,7 +321,7 @@ def filter_by_dict_polars(df: "pl.DataFrame", filter_dict: "Optional[Dict[str, A 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 diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index 277e570991..173842eef6 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -35,6 +35,13 @@ # checks below are defense-in-depth, not the contract. from graphistry.compute.gfql.call.support import AggSpec, OrderKey, SelectItem from .dtypes import is_float as _dtype_is_float, is_int as _dtype_is_int, is_numeric as _dtype_is_numeric, is_stringlike as _dtype_is_stringlike +# Same-package sibling holding the var-length specializations. Safe at module scope: +# `varlen_rows` has no runtime module-level imports of its own (polars and the pandas +# mixin are both function-local there), so it cannot cycle back through this module. +from .varlen_rows import ( + _directed_varlen_reachable_polars, + _directed_fixed_point_binding_rows_polars, +) # Active row-table schema (col -> dtype), set around lowering so lower_expr can infer FLOAT @@ -1099,7 +1106,7 @@ def binding_rows_polars( binding_ops: Sequence[Dict[str, JSONVal]], attach_prop_aliases: Optional[Sequence[str]] = None, ) -> Optional[Plottable]: - """Native polars bindings-row table for FIXED-LENGTH connected patterns (#1709). + """Native polars bindings-row table for connected alias patterns (#1709). Materializes one row per matched path for an alternating ``n/e/n/...`` pattern (the ``rows(binding_ops=...)`` op emitted by Cypher multi-alias lowering), with @@ -1109,14 +1116,19 @@ def binding_rows_polars( columns — raw ``node_id``, ``a__a_join__``, leaked ``__gfql_edge_index__`` — that no lowered query references; those are intentionally not replicated.) + Covers fixed-length hops, bounded variable-length (directed ``-[*i..k]->`` and + undirected ``-[*1..k]-``), unbounded DIRECTED fixed point (``-[*]->`` / + ``-[*0..]->``), and the node-only cartesian mode. + Returns None to DECLINE (caller raises the honest NIE) for anything outside - the supported subset: variable-length/multi-hop edges, shortestPath scalar - bindings, node ``query=`` / edge query or endpoint-match params, hop labels, - HAS_-label destination disambiguation on duplicate-node-id graphs (unique-id - graphs run native — pandas would not narrow there either), seeded re-entry - contexts, cartesian (node-only) mode, and the legacy ``alias_endpoints`` - variant. NO-CHEATING: - never bridges to pandas. Parity gate: differential tests vs the pandas oracle. + that subset: undirected variable-length outside ``min_hops == 1`` (including + undirected unbounded), aliased variable-length relationships, unbounded + segments without ``to_fixed_point``, shortestPath scalar bindings, node + ``query=`` / edge query or endpoint-match params, hop labels, HAS_-label + destination disambiguation on duplicate-node-id graphs (unique-id graphs run + native — pandas would not narrow there either), duplicate-id re-entry seeds, + and the legacy ``alias_endpoints`` variant. NO-CHEATING: never bridges to + pandas. Parity gate: differential tests vs the pandas oracle. """ import polars as pl from graphistry.compute.ast import ASTEdge, ASTNode, ASTObject, from_json as ast_from_json @@ -1129,11 +1141,23 @@ def _names(lf: pl.LazyFrame) -> List[str]: # LazyFrame column names WITHOUT collecting data (schema-only resolve). return lf.collect_schema().names() - nodes = g._nodes - edges = g._edges - node_id = g._node - src = g._source - dst = g._destination + # Build from the PRE-CHAIN base graph, exactly like the pandas oracle + # (`_gfql_binding_rows`: `base_nodes = base_graph._nodes`, every alias step + # `op.execute(g=base_graph, ...)`) and like the indexed builder below, which is + # already handed `base_graph`. Rebuilding from the chain OUTPUT instead would + # silently under-report: the traversal prunes to nodes/edges IT considers + # matched, and that is not the same set the bindings builder matches — e.g. a + # zero-hop var-length segment (`-[*0..k]->`, `-[*0..]->`) binds a seed with no + # outgoing edge, which the traversal drops entirely. Fuzz-verified: with the + # chain output as the source, `-[*0..2]->` lost those rows against pandas. + base_graph = g._gfql_rows_base_graph + if base_graph is None: + base_graph = g + nodes = base_graph._nodes + edges = base_graph._edges + node_id = base_graph._node + src = base_graph._source + dst = base_graph._destination if nodes is None or edges is None or node_id is None or src is None or dst is None: return None seed_ids_lf: Optional[Any] = None # LazyFrame; Any avoids the union-typed seed_nodes.join mismatch @@ -1175,9 +1199,6 @@ def _names(lf: pl.LazyFrame) -> List[str]: from graphistry.compute.gfql.index import bindings as indexed_bindings from graphistry.compute.gfql.lazy import active_target, ExecutionTarget from graphistry.Engine import Engine - base_graph = g._gfql_rows_base_graph - if base_graph is None: - base_graph = g engine_concrete = ( Engine.POLARS_GPU if active_target() == ExecutionTarget.GPU @@ -1229,22 +1250,62 @@ def _names(lf: pl.LazyFrame) -> List[str]: # supported via iterative pair joins. Bounded UNDIRECTED var-length # with min_hops == 1 (`-[*1..k]-`, the LDBC IC11/IC6 shape) is now # also supported via a doubled-pair join with immediate-backtrack - # avoidance (see the execution branch below). Everything else declines: - # unbounded (`[*]`, needs fixed-point + termination error), aliased - # var-length edges (pandas rejects those outright), and undirected - # var-length with min_hops != 1 (`-[*0..k]-` / `-[*2..k]-`): pandas' - # step_pairs come from the var-length `edge_op.execute` hop, whose - # backward hop-window pruning / zero-hop handling changes the edge - # multiplicity in a way this raw-edge reconstruction only reproduces - # for min_hops == 1 (every edge is trivially a length-1 path, so no - # pruning occurs) — fuzz-verified vs the pandas oracle. Decline the - # rest honestly rather than risk silent-wrong multiplicities. - if ( - bool(op.to_fixed_point) - or (op.max_hops is None and op.hops is None) - or isinstance(op._name, str) - ): + # avoidance (see the execution branch below). UNBOUNDED DIRECTED + # fixed-point (`-[*0..]->` / `-[*]->`, the LDBC IS6 REPLY_OF ancestor + # walk, #1709) runs the same pair join to exhaustion — see + # `_directed_fixed_point_binding_rows_polars` for the parity argument. + # Everything else declines: + # - aliased var-length edges (pandas rejects those outright); + # - undirected var-length with min_hops != 1 (`-[*0..k]-` / + # `-[*2..k]-`): pandas' step_pairs come from the var-length + # `edge_op.execute` hop, whose backward hop-window pruning / + # zero-hop handling changes the edge multiplicity in a way this + # raw-edge reconstruction only reproduces for min_hops == 1 (every + # edge is trivially a length-1 path, so no pruning occurs) — + # fuzz-verified vs the pandas oracle; + # - undirected UNBOUNDED (`-[*]-`): would need both the multiplicity + # reconstruction above and backtrack-aware termination; + # - unbounded WITHOUT to_fixed_point (`min_hops=2` and no max): pandas + # silently truncates at `len(step_pairs) + 1` iterations instead of + # erroring, and its step_pairs row count is not reconstructible here, + # so the truncation depth (hence the answer) is not reproducible. + # - UNBOUNDED with min_hops >= 2 (`-[*2..]->`): same multiplicity hazard as + # the undirected min_hops != 1 case above, and it is Cypher-reachable. + # Pandas' step_pairs come from the var-length `edge_op.execute` hop, which + # returns an EMPTY frame when its `max_reached_hop < min_hops` + # (compute/hop.py) and otherwise drops edges labelled below min_hops. + # `max_reached_hop` is a dedup-by-node BFS eccentricity, NOT a longest-walk + # length, so the raw-edge reconstruction below expands a DIFFERENT edge + # multiset and disagrees SILENTLY — on a 7-node acyclic graph + # `MATCH (a)-[*3..]->(b) RETURN count(*)` gives pandas 0 vs polars 30. + # `RETURN count(*)` lowers to a pure-CALL chain, so `_is_native_multihop` + # in `_chain_traversal_polars` never runs: this gate is the only gate. + # Decline the rest honestly rather than risk silent-wrong multiplicities. + if isinstance(op._name, str): + return None + _resolved_max = op.max_hops if op.max_hops is not None else op.hops + if bool(op.to_fixed_point) and _resolved_max is not None: + # - `to_fixed_point` COMBINED WITH an explicit bound. Master declined + # this outright (it declined on `bool(op.to_fixed_point)` alone), and + # serving it is silently wrong for min_hops >= 3 — the same + # reconstruction gap as the unbounded case: `MATCH (a)-[*3..5]->(b)` + # with the flag set gives pandas 0 and polars 30 on a 7-node acyclic + # graph. Cypher never emits this combination (the parser sets + # to_fixed_point False for `*k` / `*i..k` and only leaves max_hops + # None for `*` / `*k..`), so it is reachable through the AST / + # `rows(binding_ops=...)` wire surface only — but "hard to reach" is + # not "correct", and declining it merely restores what master did. return None + if _resolved_max is None: + if not bool(op.to_fixed_point) or op.direction == "undirected": + return None + # min_hops 0 and 1 are the fuzz-verified shapes (`-[*]->`, `-[*0..]->`, + # the #1709 IS6 walk); >= 2 is the divergence above. + _resolved_min_unbounded = op.min_hops if op.min_hops is not None else ( + op.hops if op.hops is not None else 1 + ) + if _resolved_min_unbounded > 1: + return None if op.direction == "undirected": _resolved_min = op.min_hops if op.min_hops is not None else ( op.hops if op.hops is not None else 1 @@ -1287,7 +1348,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 @@ -1356,12 +1421,20 @@ def _names(lf: pl.LazyFrame) -> List[str]: edge_op.hops if edge_op.hops is not None else 1 ) max_hops_value = edge_op.max_hops if edge_op.max_hops is not None else edge_op.hops - if max_hops_value is None: - return None min_hops = int(min_hops_value) - max_hops = int(max_hops_value) state_cols = _names(state) - if sem.is_undirected: + if max_hops_value is None: + # UNBOUNDED directed fixed point (`-[*0..]->`, LDBC IS6). Gated + # above to to_fixed_point=True and a directed edge. Termination is + # data-dependent, so unlike the bounded branch this one cannot stay + # fully lazy — it collects one frontier per hop. + state = _directed_fixed_point_binding_rows_polars( + state, + oriented.select(["__from__", "__to__"]), + state_cols, + min_hops=min_hops, + ) + elif sem.is_undirected: # Bounded UNDIRECTED var-length, min_hops == 1 (gated above): the # LDBC IC11/IC6 `-[*1..k]-` shape. Mirror the pandas oracle # (`_gfql_multihop_binding_rows`, avoid_immediate_backtrack=True) @@ -1376,6 +1449,7 @@ def _names(lf: pl.LazyFrame) -> List[str]: # carries the just-left node so each hop can drop immediate # backtracks (`__to__ == __prev__`), matching pandas' Kleene mask # (null prev -> kept). + max_hops = int(max_hops_value) normal = edges_f.filter(pl.col(src) != pl.col(dst)) loops = edges_f.filter(pl.col(src) == pl.col(dst)) fwd = normal.select([pl.col(src).alias("__from__"), pl.col(dst).alias("__to__")]) @@ -1409,22 +1483,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: - 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") @@ -1441,15 +1507,13 @@ def _names(lf: pl.LazyFrame) -> List[str]: # HAS_