diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d105a6d42..2e70779cb5 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 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 +- **Native polars chain combine is proportional to the traversal result, not to the graph**: two graph-sized terms sat inside a combine whose answer is a handful of rows, and both are gone. (1) `_combine_edges` ran the prev/next endpoint gates for EVERY step, including the node steps whose edge frame is `g._edges.clear()` — zero rows. The eager combine skipped those, but the collect-once rewrite lazified the step frames and `.lazy()` erases the height, so the skip silently went dead. The cost lands on the side that is NOT empty: for the first step the gate's key side is the whole node table, and polars builds the hash table on that side before discovering the probe side has no rows (isolated: 6.99 ms for one such join at N=2M, and a chain pays one per node step). The pre-lazy row count is now recorded when the step frame is still eager and an empty step is dropped from the id union — it can contribute no ids, so the result is unchanged by construction. The skip keys on KNOWN-empty only; a frame that arrives already lazy reports no height and is planned normally. (2) The output node rows were materialized in TWO passes over the node table — one for the ids the steps kept, one more for the surviving edges' endpoints the first pass missed — then concatenated. The output node set is the UNION of those two id sides, so the ids are unioned first and the node table is read ONCE. The row-level `unique(subset=[node])` is preserved verbatim: those rows feed `how="left"` alias joins where a node table carrying the same id twice would multiply rows. Measured on synthetic LDBC-IS5-shaped graphs (one-row answer, polars-engine resident indexes), varying one dimension at a time: **at fixed E=2M, N=250k → 4M went 10.17 → 45.48 ms before and 7.60 → 17.17 ms after (2.65× at 4M, and the node-count slope is 3.7× flatter)**; the edge-count slope is unchanged, as expected for a node-side fix, with the constant ~6 ms lower. Parity: identical full frames (all columns, row order included) across 280 shape × graph combinations and 400 duplicate-node-id combinations, plus the 1003-case polars chain differential suite. Pinned by tests that assert the boundary rather than a wall clock: the node universe must not appear in the edge plan at all, and the node table must be read at most once per query. - **GFQL polars chain stops deduplicating semi-join key sides**: the native polars executor applied `.unique()` to every frame it fed into a `how="semi"` join. A semi-join emits a left row iff at least one matching right row exists, so duplicate keys can neither change which rows come back nor multiply them the way an inner join would — the deduplication was a full hash pass over the key column bought for no observable effect. On an unfiltered hop the key side **is** the node table, so this put **O(N) work inside a query whose answer is O(degree)**: the seeded single-hop plan built two such key frames per hop, each costing ~53 ms at 3.18M nodes — more than the rest of the query combined. The `.unique()` is now dropped everywhere the frame is provably a semi key side only (the `_semi` helper, the alias hop-window and next-edge endpoint gates, the two-hop fast path's endpoint gate, the `start_nodes` gate, the single-hop planner's id frames, and the index layer's `select_by_ids` polars branch — where the cuDF and pandas branches already used `isin` with no dedup, so the three engines now agree). It is deliberately **kept** on the alias frame that feeds a `how="left"` join, where duplicates genuinely would multiply rows, and the eager multi-hop loop is untouched (its frames also flow into concat/anti-join bookkeeping, a separate argument). Measured on a 3.18M-node / 14M-edge polars graph (LDBC SNB SF1-shaped), a seeded typed hop goes **127.4 → 57.1 ms (2.23×)** with identical row counts; at 1.75M edges **102.8 → 31.4 ms (3.28×)**, confirming the removed cost scales with node count, not edge count. Parity held across 380 differential comparisons covering duplicate node keys, null ids, dangling edges, duplicate `start_nodes`, and 11 traversal shapes; the gfql and chain suites show identical failure sets before and after. Pinned by tests that assert the boundary rather than the speed: duplicates reaching a semi key side must not change results or multiply rows, a dangling endpoint must still be excluded (the gate is load-bearing, not vacuous), and duplicate `start_nodes` must be inert. - **Seeded chain combine stops joining against the full frame when the intermediate is empty**: `_lean_prefilter_right` shrinks the big side of the combine's `how='left'` merge to the keys actually present on the left, but it declined to do so in the one case where shrinking is both maximally profitable and trivially correct — an **empty** left. A left merge keeps only the right rows that match, so a zero-row left yields a zero-row result whatever the right side holds; the merge was nonetheless materializing the whole graph-sized frame. It now hands back a zero-row slice of `right` (same columns and dtypes, so the merge still produces an identical schema). Measured: a single-node query whose 0-row intermediate was joined against 14M edges went **112.64 → 17.29 ms**. The shrink is used at exactly one call site, which is `how='left'`; a merge that retains unmatched right rows (`right`/`outer`) would NOT be safe to shrink this way, and the tests pin both directions — the empty-left case must return an empty result, and the non-empty cases must be byte-identical to the unshrunk merge. - **Native polars chain reuses the synthetic edge id as its stable row order**: when the executor had already added a synthetic edge-index column it also added a second, separate row-index column purely to restore input order at the end — two full `with_row_index` passes over the edge frame, and a redundant column carried through every intermediate. The existing synthetic id is already a contiguous, order-preserving row index, so it is now reused as the sort key and dropped once at the end. Only the pre-existing column is reused; when no synthetic id was added the separate order column is still created, so graphs that bring their own edge id are unaffected. Value-identical including row order (148.22 → 135.43 ms on a 3.18M-node / 14M-edge polars graph). diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index e2ff7aed93..750a728630 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -19,7 +19,9 @@ from graphistry.compute.ast import ASTObject, ASTNode, ASTEdge if TYPE_CHECKING: + import polars as pl from graphistry.compute.gfql.index.bindings import IndexedBindingsState + from .dtypes import PolarsFrame from .hop_eager import ensure_nodes_polars from .dtypes import is_lazy, colnames, endpoint_ids from .degrees import get_degrees_polars, get_indegrees_polars, get_outdegrees_polars @@ -219,22 +221,57 @@ def _is_native_multihop(op: ASTObject) -> bool: class _LazyShim: """Track B collect-once shim: carries _nodes/_edges as LazyFrames (+ col names) so the eager - combine helpers run lazily over already-materialized hop frames without Plottable rebinds.""" - __slots__ = ("_nodes", "_edges", "_node", "_source", "_destination", "_edge") - - def __init__(self, nodes_lf, edges_lf, node, source, destination, edge): + combine helpers run lazily over already-materialized hop frames without Plottable rebinds. + + ``edges_empty`` records whether the step's edge frame was empty while it was still eager + (tri-state: True/False, or None when unknown). ``.lazy()`` throws that fact away — a + LazyFrame has no height without collecting — and the combine's cardinality shortcuts then go + dead, which is a graph-sized mistake: an empty relation annihilates a join, but polars cannot + know the relation is empty until it has already built the hash table over the OTHER side. + Capturing it here costs nothing (the frames are materialized at construction) and this is the + only place in the lazy combine where the count is still available.""" + __slots__ = ("_nodes", "_edges", "_node", "_source", "_destination", "_edge", "edges_empty") + + # Bare annotations only — a class-level VALUE would collide with __slots__ at class + # creation. These make the slots statically typed rather than inferred from __init__. + # LazyFrame, not the PolarsFrame union: every construction site (`step` and the Track-B + # entry) calls `.lazy()` first, which is the whole point of the shim, so the union would + # be both less true and unusable — `pl.concat`'s TypeVar rejects a DataFrame|LazyFrame. + _nodes: "Optional[pl.LazyFrame]" + _edges: "Optional[pl.LazyFrame]" + _node: Optional[str] + _source: Optional[str] + _destination: Optional[str] + _edge: Optional[str] + edges_empty: Optional[bool] + + def __init__(self, nodes_lf: "Optional[pl.LazyFrame]", edges_lf: "Optional[pl.LazyFrame]", + node: Optional[str], source: Optional[str], destination: Optional[str], + edge: Optional[str], edges_empty: Optional[bool] = None) -> None: self._nodes = nodes_lf self._edges = edges_lf self._node = node self._source = source self._destination = destination self._edge = edge + self.edges_empty = edges_empty @staticmethod - def step(p): + def step(p: Plottable) -> "_LazyShim": nd = p._nodes.lazy() if p._nodes is not None else None ed = p._edges.lazy() if p._edges is not None else None - return _LazyShim(nd, ed, None, None, None, None) + return _LazyShim(nd, ed, None, None, None, None, edges_empty=_known_empty(p._edges)) + + +def _known_empty(frame: "Optional[PolarsFrame]") -> Optional[bool]: + """Tri-state emptiness of an already-materialized frame: True/False, or None when unknown + (frame absent, or already lazy so the height is not available without collecting).""" + if frame is None or is_lazy(frame): + return None + # `is_lazy` is a plain bool predicate, so the else-branch narrowing has to be asserted + # here rather than inferred. Widening `is_lazy` to a TypeIs would narrow this for every + # caller in the engine, but that is a dtypes.py-wide change, not this PR's. + return cast("pl.DataFrame", frame).height == 0 def _combine_edges(g, steps, label_steps, has_multihop=False): @@ -247,7 +284,16 @@ def _combine_edges(g, steps, label_steps, has_multihop=False): edges_df = g_step._edges if edges_df is None: continue - if not is_lazy(edges_df) and edges_df.height == 0: + # A step with no edges contributes no ids to the union below, so drop it BEFORE the + # endpoint gates rather than semi-joining an empty frame against the graph. The gates + # are the expensive part and their cost is on the side we do NOT need: polars builds + # the hash table on the RIGHT (the node universe / a neighbouring step's node frame) + # and only then probes with the empty left, so an unfiltered `prev_nodes = g._nodes` + # costs a full O(N) hash build to produce the zero rows we already knew about + # (measured: 6.99 ms for one such join at N=2M, and a chain hits one per node step). + # Height is read from the pre-lazy fact recorded by _LazyShim.step because `.lazy()` + # erases it; `not is_lazy(...)` keeps the direct-eager-frame case working. + if g_step.edges_empty is True or (not is_lazy(edges_df) and edges_df.height == 0): continue if has_multihop or (isinstance(op, ASTEdge) and not op.is_simple_single_hop()): # has_multihop: every edge step was already recomputed path-valid (forward re-exec over @@ -287,20 +333,63 @@ def _combine_edges(g, steps, label_steps, has_multihop=False): return out -def _combine_nodes(g, steps): +def _combine_node_ids(g: "_LazyShim", + steps: List[Tuple[ASTObject, "_LazyShim"]]) -> "pl.LazyFrame": + """One-column frame of the node ids the traversal kept, unioned over the pruned steps. + + IDS ONLY, not the node rows: the caller still has to fold in the surviving edges' endpoints, + and materializing the node rows before that fold means scanning the node table TWICE (once + here, once for the endpoints the first scan missed). The union is over per-step id columns, + so it is proportional to the traversal result, not to the graph. + + Not deduplicated: the single consumer is a ``how="semi"`` key side, where duplicate keys can + neither change which rows come back nor multiply them (see the module note on semi-join key + frames). The caller's own ``.unique()`` on the materialized node rows is a DIFFERENT dedup + (by node id, over rows) and is still required.""" import polars as pl node_col = g._node assert node_col is not None + all_nodes = g._nodes + assert all_nodes is not None frames = [ g_step._nodes.select(pl.col(node_col)) for _, g_step in steps if g_step._nodes is not None and node_col in colnames(g_step._nodes) ] - if frames: - ids = pl.concat(frames, how="vertical_relaxed").unique(subset=[node_col]) - else: - ids = g._nodes.select(pl.col(node_col)).limit(0) - return g._nodes.join(ids, on=node_col, how="semi") + if not frames: + return all_nodes.select(pl.col(node_col)).limit(0) + if len(frames) == 1: + return frames[0] + return pl.concat(frames, how="vertical_relaxed") + + +def _materialize_node_rows(all_nodes: "pl.LazyFrame", step_ids: "pl.LazyFrame", + endpoint_ids_frame: "pl.LazyFrame", node_col: str) -> "pl.LazyFrame": + """The output node ROWS: every node the steps kept, plus every endpoint of a surviving edge. + + Union the two ID sides FIRST, then read the node table ONCE. Materializing the step rows and + then fetching the endpoint rows the first pass missed reads the whole node frame TWICE for + the same answer — two pure O(N) passes for a result that is usually a handful of rows + (measured: 0.90 + 0.86 ms at N=2M). Row identity is unchanged: semi-joining the UNION of two + key sets selects exactly the rows the two semi-joins selected between them. + + Neither id side is deduplicated — both feed a ``how="semi"`` key side, where duplicates + cannot change or multiply the rows that come back. The trailing ``unique`` is a DIFFERENT + dedup and is REQUIRED: it is over the node ROWS, and these rows go on to feed ``how="left"`` + alias joins where a node table carrying the same id twice would multiply every matching row. + + Row ORDER out of here is arbitrary — a polars semi-join does not preserve left-frame order — + and the caller restores input-frame order with an explicit sort. ``maintain_order`` is kept + verbatim from the pre-refactor call so that WHICH duplicate row survives is decided the same + way it was before: A/B over 400 duplicate-id combos gives identical full frames **under the + default in-memory collect**. Scoped deliberately — under streaming collect the survivor DOES + differ from the pre-refactor call (measured), so it is not a guaranteed property of this + helper, only a stable one on the default engine. Anything needing a specific survivor must + order explicitly rather than rely on this.""" + import polars as pl + ids = pl.concat([step_ids, endpoint_ids_frame], how="vertical_relaxed") + return all_nodes.join(ids, on=node_col, how="semi").unique( + subset=[node_col], maintain_order=True) def _apply_node_names(out, g, steps, auto_hop_col: str = _AUTO_NODE_HOP): @@ -337,7 +426,16 @@ def _apply_node_names(out, g, steps, auto_hop_col: str = _AUTO_NODE_HOP): on=node_col, how="semi") if idx + 1 < len(step_list): next_op, next_step = step_list[idx + 1] - if isinstance(next_op, ASTEdge) and next_step._edges is not None and (is_lazy(next_step._edges) or next_step._edges.height > 0): + # Cardinality guard, restated against a fact that SURVIVES lazification. The old + # spelling was `is_lazy(df) or df.height > 0`, and `_apply_node_names` is always + # called with lazified steps — so `is_lazy` short-circuited True and the height + # test was unreachable. That is the identical silent death this commit fixes one + # function above; leaving a second copy of it here is how the bug recurs. + # Unlike the edges combine this one is SEMANTIC, not a cost guard: an empty next + # edge step must not empty `named` via the gate below. + next_edges_empty = getattr(next_step, "edges_empty", None) + if (isinstance(next_op, ASTEdge) and next_step._edges is not None + and next_edges_empty is not True): e = next_step._edges if next_op.direction == "forward": part = e.select(pl.col(src).alias(node_col)) @@ -953,14 +1051,12 @@ def _plain_edge(op): edge_steps_lz = [(op, _LazyShim.step(p)) for op, p in edge_steps] label_lz = [(op, _LazyShim.step(p)) for op, p in label_steps] - final_nodes = _combine_nodes(g_lz, steps_lz) + node_ids = _combine_node_ids(g_lz, steps_lz) final_edges = _combine_edges(g_lz, edge_steps_lz, label_lz, has_multihop) - # Endpoint (lazy: always compute; maintain_order keeps the semi-join order). - endpoints = endpoint_ids(final_edges, src, dst, node_col).unique(subset=[node_col]) - missing = endpoints.join(final_nodes.select(pl.col(node_col)), on=node_col, how="anti") - extra = g_lz._nodes.join(missing, on=node_col, how="semi") - final_nodes = pl.concat([final_nodes, extra], how="diagonal_relaxed").unique( - subset=[node_col], maintain_order=True) + all_nodes_lz = g_lz._nodes + assert all_nodes_lz is not None # constructed from g._nodes two statements above + final_nodes = _materialize_node_rows( + all_nodes_lz, node_ids, endpoint_ids(final_edges, src, dst, node_col), node_col) final_nodes = _apply_node_names(final_nodes, g_lz, steps_lz, auto_hop_col=auto_hop_col) final_nodes = final_nodes.sort(NORD).drop(NORD) @@ -968,6 +1064,9 @@ def _plain_edge(op): # single drop above removes it. There is no `added_edge_index and EID != EORD` case left # to handle: the only branch that sets added_edge_index also sets EORD = EID. final_edges = final_edges.sort(EORD).drop(EORD) - final_edges, final_nodes = collect_all([final_edges, final_nodes]) - final_edges = _restore_edge_dtypes(final_edges, src, dst, _endpoint_restore) - return self.nodes(final_nodes, node_col).edges(final_edges, src, dst) + # Distinct names on the eager side: `final_nodes` is statically a LazyFrame all the way + # down this block, and `collect_all` hands back DataFrames — rebinding would be a type + # error, and silencing it would cost the lazy/eager distinction the shim exists to keep. + final_edges_eager, final_nodes_eager = collect_all([final_edges, final_nodes]) + final_edges_eager = _restore_edge_dtypes(final_edges_eager, src, dst, _endpoint_restore) + return self.nodes(final_nodes_eager, node_col).edges(final_edges_eager, src, dst) diff --git a/graphistry/tests/compute/gfql/test_engine_polars_narrow_combine.py b/graphistry/tests/compute/gfql/test_engine_polars_narrow_combine.py new file mode 100644 index 0000000000..1cc2ba5f1b --- /dev/null +++ b/graphistry/tests/compute/gfql/test_engine_polars_narrow_combine.py @@ -0,0 +1,397 @@ +"""The polars chain combine must be proportional to the TRAVERSAL RESULT, not to the graph. + +Two graph-sized terms used to sit inside a combine that answers with a handful of rows: + +1. ``_combine_edges`` ran the prev/next endpoint gates for EVERY step, including the node steps + whose edge frame is ``g._edges.clear()`` — zero rows. The eager code skipped those; Track B + lazified the step frames and ``.lazy()`` erased the height, so the skip silently went dead. + The cost lands on the side that is NOT empty: for the first step ``prev_nodes`` is the whole + node table, and polars builds the hash table on that side before discovering the probe side + has no rows (measured: 6.99 ms per such join at N=2M). +2. The node rows were materialized in TWO passes over the node table — once for the ids the + steps kept, once more for the edge endpoints the first pass missed — then concatenated. + +These tests pin the BOUNDARY, not a wall clock (which goes flaky on a shared host): + * the node universe must not enter the EDGE plan at all when the chain starts with a node step + (test 1 exercises ``_combine_edges`` directly with a marker column no step frame carries), and + * the node universe must be read AT MOST ONCE by the whole node plan (test 2). +Both are backed by semantics: pandas is the oracle for a differential matrix over multi-step, +undirected, multi-hop, fixed-point, aliased and ``rows(...)``-projected shapes, plus explicit +row-ORDER, endpoint-materialization and duplicate-node-id pins that the refactor could break +without changing any id set. +""" +import pandas as pd +import pytest + +import graphistry +from graphistry.compute.ast import n, e_forward, e_reverse, e_undirected, rows + +pl = pytest.importorskip("polars") + +from graphistry.compute.gfql.lazy.engine.polars import chain as chain_mod # noqa: E402 +from graphistry.tests.compute.gfql.polars_test_utils import ( # noqa: E402 + graph_sig, to_pandas_any) + + +# --------------------------------------------------------------------------- fixtures + +def _clean_frames(): + nodes = pd.DataFrame({"key": [0, 1, 2, 3, 4, 5], + "id": ["a", "b", "c", "d", "e", "f"], + "grp": [1, 2, 2, 1, 2, 1]}) + edges = pd.DataFrame({"s": [0, 0, 1, 2, 3, 4], + "d": [1, 2, 3, 3, 4, 5], + "type": ["K", "K", "L", "K", "K", "K"]}) + return nodes, edges + + +def _dup_key_frames(): + """The same node key twice — the case the combine's trailing unique() exists for.""" + nodes, edges = _clean_frames() + nodes = pd.concat([nodes, nodes[nodes["key"].isin([1, 3])]], ignore_index=True) + return nodes, edges + + +def _dangling_frames(): + """An edge whose destination is absent from the node table: the endpoint gate is NOT vacuous.""" + nodes, edges = _clean_frames() + edges = pd.concat([edges, pd.DataFrame({"s": [0], "d": [999], "type": ["K"]})], + ignore_index=True) + return nodes, edges + + +def _null_id_frames(): + nodes, edges = _clean_frames() + nodes.loc[2, "id"] = None + return nodes, edges + + +def _unsorted_frames(): + """Node/edge frames whose row order is NOT id order — the order the output must restore.""" + nodes, edges = _clean_frames() + return (nodes.iloc[[4, 0, 5, 2, 1, 3]].reset_index(drop=True), + edges.iloc[[3, 5, 0, 4, 1, 2]].reset_index(drop=True)) + + +BUILDERS = { + "clean": _clean_frames, + "dup_keys": _dup_key_frames, + "dangling": _dangling_frames, + "null_ids": _null_id_frames, + "unsorted": _unsorted_frames, +} + +SHAPES = { + "node_only": [n({"id": "a"}, name="m")], + "fwd_typed": [n({"id": "a"}, name="m"), e_forward({"type": "K"}, name="r"), n(name="p")], + "rev_typed": [n({"id": "d"}, name="m"), e_reverse({"type": "K"}, name="r"), n(name="p")], + "undirected": [n({"id": "a"}, name="m"), e_undirected({"type": "K"}, name="r"), n(name="p")], + "multi_step": [n({"id": "a"}, name="m"), e_forward(name="r1"), n(name="mid"), + e_forward(name="r2"), n(name="p")], + "multi_hop": [n({"id": "a"}, name="m"), e_forward(hops=2, name="r"), n(name="p")], + "fixed_point": [n({"id": "a"}, name="m"), e_forward(to_fixed_point=True, name="r"), + n(name="p")], + "mixed_single_multi": [n({"id": "a"}), e_forward(hops=2), n(), e_forward(), n(name="p")], + "no_match": [n({"id": "zzz"}, name="m"), e_forward(name="r"), n(name="p")], + "trailing_node_filter": [n({"grp": 1}, name="m"), e_forward(name="r"), n({"grp": 2}, name="p")], +} + +ROWS_SHAPES = { + "rows_nodes": [n({"id": "a"}, name="m"), e_forward({"type": "K"}, name="r"), n(name="p"), + rows(table="nodes", source="p")], + "rows_edges": [n({"id": "a"}, name="m"), e_forward({"type": "K"}, name="r"), n(name="p"), + rows(table="edges", source="r")], +} + + +def _pair(nodes_pd, edges_pd): + g_pd = graphistry.edges(edges_pd, "s", "d").nodes(nodes_pd, "key") + g_pl = graphistry.edges(pl.from_pandas(edges_pd), "s", "d").nodes( + pl.from_pandas(nodes_pd), "key") + return g_pd, g_pl + + +# --------------------------------------------------------------------------- structure + +def _shim_step(nodes_df, edges_df): + """A combine step built the way the chain builds one (through the eager->lazy shim, which is + where the row count is captured), without standing up a whole Plottable.""" + g = graphistry.edges(edges_df, "s", "d", edge="eid").nodes(nodes_df, "key") + return chain_mod._LazyShim.step(g) + + +def test_empty_edge_step_never_reaches_the_node_universe(): + """A step with zero edges contributes zero ids, so its endpoint gates must not be planned. + + The universe frame carries a marker column NO step frame has, so its presence anywhere in + the optimized edge plan means the empty step's ``prev_nodes = g._nodes`` gate was built — + the graph-sized hash build this change removes. + """ + universe_nodes = pl.DataFrame({"key": [0, 1, 2, 3], "probe_universe_marker": [9, 9, 9, 9]}) + universe_edges = pl.DataFrame({"eid": [0, 1, 2], "s": [0, 1, 2], "d": [1, 2, 3]}) + step_nodes = pl.DataFrame({"key": [0, 1]}) + + g_lz = chain_mod._LazyShim(universe_nodes.lazy(), universe_edges.lazy(), + "key", "s", "d", "eid") + node_step = _shim_step(step_nodes, universe_edges.clear()) # ASTNode -> cleared edges + edge_step = _shim_step(step_nodes, universe_edges.head(1)) # ASTEdge -> one real edge + steps = [(n(), node_step), (e_forward(), edge_step), (n(), node_step)] + + out = chain_mod._combine_edges(g_lz, steps, steps) + plan = out.explain(optimized=True) + assert "probe_universe_marker" not in plan, ( + "the empty node step's endpoint gate was planned against the whole node table:\n" + plan) + # ...and the combine still returns the right edge, so the skip is not a silent drop. + assert out.collect().sort("eid")["eid"].to_list() == [0] + + +def test_a_step_of_unknown_height_is_planned_not_dropped(): + """The skip must key on KNOWN-empty, never on 'not known to be non-empty'. + + ``_LazyShim.step`` can only record the height when the step frame is still eager; a frame + that arrives already lazy records nothing. Treating that as empty would silently drop real + edges from the result — the failure mode a cardinality shortcut has to be safe against. + """ + edges = pl.DataFrame({"eid": [0, 1, 2], "s": [0, 1, 2], "d": [1, 2, 3]}) + assert chain_mod._known_empty(edges) is False + assert chain_mod._known_empty(edges.clear()) is True + assert chain_mod._known_empty(edges.lazy()) is None, "a lazy frame cannot report a height" + assert chain_mod._known_empty(None) is None + + g_lz = chain_mod._LazyShim(pl.DataFrame({"key": [0, 1, 2, 3]}).lazy(), edges.lazy(), + "key", "s", "d", "eid") + step_nodes = pl.DataFrame({"key": [0, 1]}) + unknown = chain_mod._LazyShim(step_nodes.lazy(), edges.head(1).lazy(), + None, None, None, None, edges_empty=None) + steps = [(n(), _shim_step(step_nodes, edges.clear())), (e_forward(), unknown)] + out = chain_mod._combine_edges(g_lz, steps, steps) + assert out.collect()["eid"].to_list() == [0], \ + "a step with an unrecorded height was dropped from the edge union" + + +def _node_universe_scans(plan: str) -> int: + """How many times the plan reads the chain's node-universe frame. + + The universe is the ONLY frame carrying the synthetic node-order column (the chain attaches + it with ``with_row_index`` to restore input order), so counting the ``DF [...]`` headers that + mention it counts reads of the node table specifically — step frames are separate materialized + frames and never carry it. + """ + return sum(1 for line in plan.splitlines() + if line.strip().startswith("DF [") and "norder" in line) + + +def test_node_rows_are_materialized_in_one_pass_over_the_node_table(): + """The output node set is (step ids) UNION (surviving edge endpoints). Unioning the two ID + sides first means one scan of the node table; materializing the step rows and then fetching + the endpoint rows the first pass missed means two.""" + _, g_pl = _pair(*_clean_frames()) + captured = [] + import graphistry.compute.gfql.lazy as lazy_mod + orig = lazy_mod.collect_all + + def spy(frames, *a, **k): + captured.append(list(frames)) + return orig(frames, *a, **k) + + lazy_mod.collect_all = spy + try: + g_pl.chain([n({"id": "a"}, name="m"), e_forward(name="r"), n(name="p")], engine="polars") + finally: + lazy_mod.collect_all = orig + + assert captured, "the chain did not go through the collect-once combine" + plans = [f.explain(optimized=True) for f in captured[-1]] + worst = max(_node_universe_scans(p) for p in plans) + assert worst <= 1, ( + f"the node table is read {worst}x for one traversal result; " + "the combine should union the id sides and scan it once") + + +# --------------------------------------------------------------------------- semantics + +@pytest.mark.parametrize("frames", sorted(BUILDERS)) +@pytest.mark.parametrize("shape", sorted(SHAPES)) +def test_parity_with_pandas_oracle(frames, shape): + g_pd, g_pl = _pair(*BUILDERS[frames]()) + chain = list(SHAPES[shape]) + try: + expected = graph_sig(g_pd.chain(chain, engine="pandas")) + except Exception as ex: # a broken oracle is not a polars failure + pytest.skip(f"pandas oracle raised {type(ex).__name__}") + try: + got = graph_sig(g_pl.chain(chain, engine="polars")) + except NotImplementedError: + pytest.skip("shape declined by the native polars chain") + assert expected == got, f"polars diverged from the pandas oracle [{frames}/{shape}]" + + +@pytest.mark.parametrize("frames", sorted(BUILDERS)) +@pytest.mark.parametrize("shape", sorted(ROWS_SHAPES)) +def test_parity_with_pandas_oracle_for_rows_projections(frames, shape): + """``rows(...)`` reads the combine's OUTPUT ORDER (its slicing is positional), so a combine + that returns the right rows in the wrong order shows up here and nowhere else.""" + g_pd, g_pl = _pair(*BUILDERS[frames]()) + query = list(ROWS_SHAPES[shape]) + try: + expected = g_pd.gfql(query, engine="pandas") + except Exception as ex: + pytest.skip(f"pandas oracle raised {type(ex).__name__}") + try: + got = g_pl.gfql(query, engine="polars") + except NotImplementedError: + pytest.skip("shape declined by the native polars chain") + # Column ORDER, and whether the engine's own synthetic `__gfql_*` bookkeeping column + # survives into the projection, already differ between the two row pipelines (pre-existing, + # and not something the combine controls). Compare on a canonical column order over the + # USER columns — ROW order is what this test is for and it is compared as-is. + def _user_cols(df): + df = to_pandas_any(df).reset_index(drop=True) + return df[[c for c in sorted(df.columns) if not str(c).startswith("__gfql_")]] + + exp, act = _user_cols(expected._nodes), _user_cols(got._nodes) + assert list(exp.columns) == list(act.columns), f"[{frames}/{shape}] column set diverged" + pd.testing.assert_frame_equal(exp, act, check_dtype=False) + + +@pytest.mark.parametrize("frames", sorted(BUILDERS)) +@pytest.mark.parametrize("shape", sorted(SHAPES)) +def test_output_row_order_follows_the_input_frame_order(frames, shape): + """The combine explicitly sorts back to input-frame order. Row SETS matching is not enough: + ``graph_sig`` sorts rows, so an order regression would pass every parity test above.""" + _, g_pl = _pair(*BUILDERS[frames]()) + try: + out = g_pl.chain(list(SHAPES[shape]), engine="polars") + except NotImplementedError: + pytest.skip("shape declined by the native polars chain") + + def _first_positions(keys): + """key -> FIRST position in the input frame (duplicate node ids collapse to their first + row, so first-occurrence is the position the output should be sorted by).""" + pos: dict = {} + for i, k in enumerate(keys): + pos.setdefault(k, i) + return pos + + node_pos_of = _first_positions(g_pl._nodes["key"].to_list()) + node_pos = [node_pos_of[k] for k in out._nodes["key"].to_list()] + assert node_pos == sorted(node_pos), f"[{frames}/{shape}] node rows came back out of order" + # edges have no user id column here; compare (s, d) positions against the input edge frame + edge_pos_of = _first_positions(list(zip(g_pl._edges["s"].to_list(), + g_pl._edges["d"].to_list()))) + edge_pos = [edge_pos_of[e] for e in zip(out._edges["s"].to_list(), + out._edges["d"].to_list())] + assert edge_pos == sorted(edge_pos), f"[{frames}/{shape}] edge rows came back out of order" + + +def test_endpoint_only_nodes_are_materialized_with_their_attributes(): + """A node reached ONLY as an endpoint of a surviving edge must still come back, with its real + columns — the safety net the second node-table pass used to provide, now folded into the id + union. Driven at the helper, deliberately: across 6300 generated chain executions I could not + build a graph/shape where the step ids MISS an endpoint, so an end-to-end test of this would + pass with the endpoint side removed entirely (verified — it does). The helper takes the two id + sides as arguments, so here the case can be constructed. + """ + all_nodes = pl.DataFrame({"key": [0, 1, 2], "grp": [7, 8, 9]}).lazy() + step_ids = pl.DataFrame({"key": [0]}).lazy() # the steps kept node 0 only + endpoints = pl.DataFrame({"key": [0, 1]}).lazy() # a surviving edge 0 -> 1 + got = chain_mod._materialize_node_rows(all_nodes, step_ids, endpoints, "key").collect() + # sorted: a semi-join does NOT preserve left-frame order, which is why the chain restores + # it with an explicit sort afterwards. Row identity is what this test is about. + assert got.sort("key").rows() == [(0, 7), (1, 8)], \ + "the endpoint-only node was dropped or lost its attributes" + + +def test_materialize_node_rows_dedups_rows_but_not_key_sides(): + """The two dedups in the helper are different and only one is needed. + + Duplicates on either ID side are inert (both are ``how="semi"`` key sides), while a node + table carrying the same id twice must still collapse to ONE row — those rows go on to feed + ``how="left"`` alias joins, where a duplicate key multiplies every matching row. + """ + all_nodes = pl.DataFrame({"key": [0, 1, 1, 2], "tag": ["x", "one", "one", "z"]}).lazy() + step_ids = pl.DataFrame({"key": [1, 1, 1]}).lazy() + endpoints = pl.DataFrame({"key": [1, 2, 2]}).lazy() + got = chain_mod._materialize_node_rows(all_nodes, step_ids, endpoints, "key").collect() + assert got.sort("key").rows() == [(1, "one"), (2, "z")], \ + "duplicate keys multiplied rows, or the wrong rows came back" + + +def test_duplicate_node_ids_still_collapse_to_one_row(): + """The combine's trailing ``unique(subset=[node])`` is load-bearing: the node rows feed + ``how="left"`` alias joins downstream, where a duplicated key multiplies rows. + + WHICH duplicate survives is left to the pandas oracle rather than asserted directly — the + semi-join feeding the dedup does not preserve node-frame order, so 'the first row' is not a + property this engine guarantees on its own (it is stable in practice, and unchanged by this + change: 400 dup-key combos A/B, identical full frames). + """ + nodes = pd.DataFrame({"key": [0, 1, 1, 2], "id": ["a", "b", "b", "c"], + "tag": ["x", "dup", "dup", "z"]}) + edges = pd.DataFrame({"s": [0], "d": [1], "type": ["K"]}) + g_pd, g_pl = _pair(nodes, edges) + chain = [n({"id": "a"}, name="m"), e_forward(name="r"), n(name="p")] + out = g_pl.chain(chain, engine="polars") + got = to_pandas_any(out._nodes) + assert (got["key"] == 1).sum() == 1, "duplicate node ids multiplied the output" + assert graph_sig(g_pd.chain(chain, engine="pandas")) == graph_sig(out) + + +@pytest.mark.parametrize("streaming", [False, True], ids=["in-memory", "streaming"]) +def test_output_row_order_survives_a_frame_big_enough_to_parallelize(streaming): + """Order at fixture scale proves little: polars' joins happen to come back in left order on a + handful of rows and only reorder once the hash join actually runs in parallel. Use a frame + large enough to reorder, shuffled so input order is not id order, and pin that the combine's + explicit sorts put both output frames back into input-frame order. + + Parametrized over the collect engine because the IN-MEMORY engine hides the edge sort: with + `final_edges.sort(EORD)` deleted, in-memory still returns EORD-ordered rows at every size + probed, so an in-memory-only test would call that sort dead code. Under STREAMING it does + not, and a trailing rows(limit=)/skip would then slice the wrong rows. Both engines here, + so neither sort can be removed on the strength of the other's silence.""" + from graphistry.compute.gfql.lazy import set_cpu_streaming + size = 60_000 + rng = list(range(size)) + order = rng[1::2] + rng[0::2][::-1] # deterministic shuffle + nodes = pd.DataFrame({"key": order, "grp": [k % 3 for k in order]}) + edges = pd.DataFrame({"s": [(k * 7) % size for k in order], + "d": [(k * 13 + 1) % size for k in order], + "type": ["K" if k % 2 else "L" for k in order]}) + _, g_pl = _pair(nodes, edges) + set_cpu_streaming(streaming) + try: + out = g_pl.chain([n({"grp": 1}, name="m"), e_forward({"type": "K"}, name="r"), n(name="p")], + engine="polars") + finally: + set_cpu_streaming(None) + assert out._nodes.height > 1000 and out._edges.height > 1000, "fixture stopped being big" + + node_rank = {k: i for i, k in enumerate(nodes["key"].tolist())} + node_pos = [node_rank[k] for k in out._nodes["key"].to_list()] + assert node_pos == sorted(node_pos), "node rows came back out of input-frame order" + + edge_rank: dict = {} + for i, e in enumerate(zip(edges["s"].tolist(), edges["d"].tolist())): + edge_rank.setdefault(e, i) + edge_pos = [edge_rank[e] for e in zip(out._edges["s"].to_list(), out._edges["d"].to_list())] + assert edge_pos == sorted(edge_pos), "edge rows came back out of input-frame order" + + +def test_endpoint_gate_still_excludes_a_dangling_edge(): + """Non-vacuity: the gates that survive must still do their job. Skipping the EMPTY steps + must not be mistaken for skipping the gates.""" + g_pd, g_pl = _pair(*_dangling_frames()) + chain = [n({"id": "a"}, name="m"), e_forward({"type": "K"}, name="r"), n(name="p")] + out = g_pl.chain(chain, engine="polars") + assert 999 not in out._nodes["key"].to_list(), "dangling endpoint leaked into the nodes" + assert graph_sig(g_pd.chain(chain, engine="pandas")) == graph_sig(out) + + +def test_chain_with_no_edge_steps_returns_no_edges(): + """Every step empty -> the union has no frames at all. That branch must still produce an + empty, correctly-typed edge frame rather than the whole edge table.""" + g_pd, g_pl = _pair(*_clean_frames()) + chain = [n({"grp": 1}, name="m")] + out = g_pl.chain(chain, engine="polars") + assert out._edges.height == 0 + assert graph_sig(g_pd.chain(chain, engine="pandas")) == graph_sig(out)