|
| 1 | +"""Variable-length (``-[*i..k]->`` / unbounded) binding-row specializations. |
| 2 | +
|
| 3 | +Extracted from ``row_pipeline.py`` so the core row flow stays readable and so the |
| 4 | +other in-flight PRs that touch that file do not have to rebase across ~270 lines |
| 5 | +of specialization. Pure code move: no behaviour change. |
| 6 | +
|
| 7 | +The unbounded arm computes an exhaustion depth with a dedup-by-node frontier walk |
| 8 | +(O(N) per hop, not O(paths)) and then runs the SAME bounded pair-join loop the |
| 9 | +`-[*1..k]->` arm uses. Deduping by endpoint changes no walk's EXISTENCE, only its |
| 10 | +multiplicity, which the bounded loop reproduces in full. |
| 11 | +""" |
| 12 | +from __future__ import annotations |
| 13 | + |
| 14 | +from typing import Dict, List, Optional, Sequence, TYPE_CHECKING |
| 15 | + |
| 16 | +# NOTE: RowPipelineMixin is imported FUNCTION-LOCALLY below, exactly as in the |
| 17 | +# original site. Hoisting it to module scope reintroduces an import cycle |
| 18 | +# (row.pipeline -> lazy engine -> back), so the move keeps it where it was. |
| 19 | + |
| 20 | +if TYPE_CHECKING: |
| 21 | + import polars as pl |
| 22 | + |
| 23 | +def _directed_varlen_reachable_polars( |
| 24 | + state: "pl.LazyFrame", |
| 25 | + pairs: "pl.LazyFrame", |
| 26 | + state_cols: List[str], |
| 27 | + *, |
| 28 | + min_hops: int, |
| 29 | + max_hops: int, |
| 30 | +) -> "pl.LazyFrame": |
| 31 | + """Bounded DIRECTED variable-length expansion of a bindings path bag. |
| 32 | +
|
| 33 | + One row per distinct edge SEQUENCE: ``pairs`` is NOT deduped, so parallel edges |
| 34 | + multiply per hop, matching pandas' ``_gfql_multihop_binding_rows`` merge. Zero-hop |
| 35 | + rows (``min_hops == 0``) keep the seed row (endpoint == start) and come first, then |
| 36 | + hop 1, 2, ... — the same ``reachable`` concat order pandas builds. |
| 37 | +
|
| 38 | + Stays fully lazy: all ``max_hops`` iterations are built without an eager |
| 39 | + ``.height`` early-break, because an empty intermediate lazily joins to empty and |
| 40 | + yields the identical result (pandas' break is an optimization, not semantics). |
| 41 | + """ |
| 42 | + import polars as pl |
| 43 | + |
| 44 | + reachable: List["pl.LazyFrame"] = [state] if min_hops == 0 else [] |
| 45 | + current = state |
| 46 | + for hop in range(1, max_hops + 1): |
| 47 | + current = ( |
| 48 | + current.join(pairs, left_on="__current__", right_on="__from__", how="inner") |
| 49 | + .drop("__current__") |
| 50 | + .rename({"__to__": "__current__"}) |
| 51 | + .select(state_cols) |
| 52 | + ) |
| 53 | + if hop >= min_hops: |
| 54 | + reachable.append(current) |
| 55 | + return pl.concat(reachable, how="vertical") if reachable else state.limit(0) |
| 56 | + |
| 57 | + |
| 58 | +def _directed_fixed_point_binding_rows_polars( |
| 59 | + state: "pl.LazyFrame", |
| 60 | + pairs: "pl.LazyFrame", |
| 61 | + state_cols: List[str], |
| 62 | + *, |
| 63 | + min_hops: int, |
| 64 | +) -> "pl.LazyFrame": |
| 65 | + """Unbounded DIRECTED variable-length binding rows (``-[*0..]->`` / ``-[*]->``), #1709. |
| 66 | +
|
| 67 | + The native twin of the ``max_hops is None and to_fixed_point`` arm of pandas' |
| 68 | + ``RowPipelineMixin._gfql_multihop_binding_rows``. One row per distinct edge |
| 69 | + SEQUENCE (Cypher path multiplicity): ``pairs`` is never deduped, so parallel |
| 70 | + edges multiply per hop exactly as the pandas merge does. ``min_hops == 0`` |
| 71 | + contributes the zero-hop rows (endpoint == start) first, then hop 1, 2, ... — |
| 72 | + the same ``reachable`` concat order pandas builds. |
| 73 | +
|
| 74 | + Pandas discovers the traversal depth by expanding the PATH frontier until it is |
| 75 | + empty, which materializes every partial path at every hop. This lowering splits |
| 76 | + that into (a) a cheap dedup-by-node frontier walk that computes the exhaustion |
| 77 | + depth ``D``, then (b) the SAME lazy bounded pair-join loop the ``-[*1..k]->`` arm |
| 78 | + uses, with ``max_hops = D``. Step (a) is exact, not an approximation: the path |
| 79 | + frontier at hop ``h`` is non-empty iff a walk of length ``h`` leaves some seed, |
| 80 | + and deduping by endpoint node changes no walk's EXISTENCE — only its |
| 81 | + multiplicity, which step (b) then reproduces in full. So the emitted rows are |
| 82 | + identical to pandas' while the exponential path blow-up happens once, lazily. |
| 83 | +
|
| 84 | + Non-termination: a cycle reachable from a seed makes the walk infinite, and |
| 85 | + pandas raises ``E108`` ("require terminating variable-length segments"). We raise |
| 86 | + the same error via the same helper, and we detect it strictly: with ``N`` distinct |
| 87 | + nodes touched by ``pairs``, any walk of ``N`` edges visits ``N + 1`` nodes and so |
| 88 | + must repeat one (pigeonhole) — a non-empty frontier at hop ``N`` is a reachable |
| 89 | + cycle, exactly the condition under which pandas' own cap |
| 90 | + (``max(len(step_pairs), 1) + 1``) also fails to exhaust. Conversely an acyclic |
| 91 | + reachable subgraph has every walk shorter than ``N``, so both engines exhaust. |
| 92 | + Same outcome on both sides of the branch, without pandas' cost of expanding paths |
| 93 | + into the cycle before giving up. |
| 94 | + """ |
| 95 | + import polars as pl |
| 96 | + from graphistry.compute.gfql.lazy import collect as _lazy_collect |
| 97 | + from graphistry.compute.gfql.row.pipeline import RowPipelineMixin |
| 98 | + |
| 99 | + pairs_df = _lazy_collect(pairs) |
| 100 | + pairs_lf = pairs_df.lazy() |
| 101 | + node_cap = int( |
| 102 | + pl.concat( |
| 103 | + [ |
| 104 | + pairs_df.select(pl.col("__from__").alias("__n__")), |
| 105 | + pairs_df.select(pl.col("__to__").alias("__n__")), |
| 106 | + ], |
| 107 | + how="vertical", |
| 108 | + ) |
| 109 | + .select(pl.col("__n__").n_unique()) |
| 110 | + .item() |
| 111 | + ) |
| 112 | + |
| 113 | + # (a) depth probe: dedup-by-node frontier, so each hop costs O(N) not O(paths). |
| 114 | + frontier_lf = state.select(pl.col("__current__")).unique() |
| 115 | + depth = 0 |
| 116 | + exhausted = node_cap == 0 # no matched edges at all -> no walk of length >= 1 |
| 117 | + for hop in range(1, node_cap + 1): |
| 118 | + frontier = _lazy_collect( |
| 119 | + frontier_lf.join(pairs_lf, left_on="__current__", right_on="__from__", how="inner") |
| 120 | + .select(pl.col("__to__").alias("__current__")) |
| 121 | + .unique() |
| 122 | + ) |
| 123 | + if frontier.height == 0: |
| 124 | + exhausted = True |
| 125 | + break |
| 126 | + frontier_lf = frontier.lazy() |
| 127 | + depth = hop |
| 128 | + if not exhausted: |
| 129 | + RowPipelineMixin._gfql_bindings_error( |
| 130 | + "Cypher multi-alias row bindings currently require terminating variable-length segments" |
| 131 | + ) |
| 132 | + |
| 133 | + # (b) the SAME bounded expansion the `-[*1..k]->` arm runs, with max_hops = depth. |
| 134 | + return _directed_varlen_reachable_polars( |
| 135 | + state, pairs_lf, state_cols, min_hops=min_hops, max_hops=depth |
| 136 | + ) |
0 commit comments