diff --git a/CHANGELOG.md b/CHANGELOG.md index ad13c533e4..c69f7ee272 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,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 +- **A single-key pure `count(*)` with provably LOW group cardinality skips polars' partitioned group-by**: inside the fused single-hop grouped-aggregate lane, `group_by(maintain_order=True).agg(pl.len())` carries a FLAT ~2 ms coordination cost that exists only at low group cardinality — measured on dgx-spark (polars 1.35.2, 20 threads, interleaved, 90 samples/arm/cell, 214 cells, ZERO value mismatches), int keys at 20,000 rows go 32 groups `2.054 ms` → 48 groups `0.411` → 64 groups `0.291`. `value_counts` has no such cost, so for a pure `count(*)` it is the same value for a fraction of the time. It is NOT a drop-in: `value_counts` scales WORSE with input rows and at 1,000,000 rows loses even at 2 groups (`4.137 → 8.591 ms`), and applied ungated the identical formulation makes the matched graph-benchmark **q1** cell (~20,000 groups over ~200,000 rows) **2.7 ms slower at 20k and 8.6 ms slower at 100k** — q1 is a cell that currently wins, so an ungated swap trades one cell's loss for another's regression. The formulation is therefore chosen only behind two STATIC, O(1), **upper** bounds: group cardinality ≤ the HEIGHT of the alias node frame supplying the group key (every group value is a property value of some row of that one frame, so distinct values cannot exceed its height), and aggregate input rows ≤ the height of the already-filtered EDGE frame (the semi-joins only remove rows; the property inner-join can multiply them, so the row bound is only claimed once exactly one alias carries properties and its node ids are unique — a check that runs on a frame already known to be ≤ 32 rows). Both bounds over-estimate, and over-estimating is the safe direction: a loose bound can only DECLINE a shape the fast formulation would have served, never route a high-cardinality aggregate into it. The thresholds — **32 groups and 100,000 rows** — were fixed from the crossover curve BEFORE the formulation was validated on any query, because a threshold chosen after seeing the verdicts is unfalsifiable; 48 groups already fails at 0.96× and 150,000 rows at 0.80× on string keys. Strictly additive: every decline falls through to the untouched `group_by`, so the blast radius is a decline away from zero. It DECLINES a non-single-key or non-pure-`count(*)` aggregate (including `count()`, which counts non-null values rather than rows), a group key not supplied by exactly one alias, a second alias also contributing property columns, a group-key alias frame that is too tall / missing its node-id column / carrying duplicate node ids, and an edge frame over the row bound. **Measured** on dgx-spark under the exclusive perf lock, matched graph-benchmark q1–q9 lane on the canonical query text (`gb_queries.py`, md5 `6e7ae268a5a41742587fcb87854b6e27`), 24 position-balanced slots per scale (12 per arm), Kuzu 0.11.3 re-run in-session, per-slot medians: **q4 at 20k `4.96 → 3.73 ms` (−1.23 ms, −24.7%), the two arms' slot ranges NOT overlapping**, which takes the board's last 20k loss from `1.65×` to `1.25×` of same-session Kuzu. Under the board's overlap rule that scores a TIE rather than a loss, but the overlap is **0.015 ms** and the median still favours Kuzu, so it is reported as *a loss narrowed to near-parity, not parity*. Engagement is exactly one cell: on the real board data the gate is consulted for q1/q2/q3/q4 and **admits only q4 at 20k** — q4 at 100k declines because its 7,117-row City frame carries only 3 distinct countries and an O(1) height bound cannot see the 3, and q1/q2 decline on BOTH bounds at both scales. q1, q2, q3 and q8 are arm-vs-arm ties with overlapping ranges, and the pandas arm — which never enters this polars-only lane — ties on all nine cells at both scales, a built-in null control. Value identity is the gate on the number: one canonical value per query across every slot, both arms, both engines and both scales, matching Kuzu on every cell. - **The polars single-hop GROUPED AGGREGATE builds ONE lazy plan instead of ~7 eager collects**: `MATCH (a {..})-[{..}]->(b {..}) [WHERE ..] RETURN . AS k, AS v ORDER BY .. [LIMIT n]` lowers to a fast path that semi-joins the edge frame against both node domains, inner-joins the projected properties on, groups, sorts and limits. Every one of those ops was issued EAGERLY — each its own `lazy().collect(_eager=True)` — so each intermediate materialized in full, the `select([src, dst])` projection could not be pushed into the semi-joins, and the `head()` could not reach back into the plan at all. That path serves three of the nine matched graph-benchmark cells (q1, q3, q4), and 74–96% of each of those queries' wall time sat inside those collects. The same op sequence is now expressed as a single lazy plan collected once — the algebra is character-identical (same `.unique()` id frames, same semi-joins, same un-deduplicated property lookups, same `group_by(maintain_order=True).agg(..)`, same per-key `nulls_last` sort, same `head`), so the value is identical, row ORDER included. The lane is strictly additive: the eager code is untouched and is the fallback on every decline. It DECLINES — never answering differently, only forgoing the speedup — for a non-eager-polars input frame, a property column missing from its alias' node frame (the eager twin discovers that MID-CHAIN and declines the whole fast path, so the guard is hoisted ahead of plan construction rather than left to be discovered after a plan already exists), source and destination bound to the same edge column, a projected column colliding with an endpoint column or the internal lookup key, an untranslatable aggregate, and — the correctness crux — **a result row order that ORDER BY does not fully determine**. Without every group key in the sort, the eager twin's order falls back to `maintain_order=True` group first-appearance order over an EAGER join output, which a lazy plan is free to change by re-ordering or re-siding joins; measured with an ungated variant of the same plan over 4 graph sizes × 4 seeds × 4 order-undetermined shapes, 47 of 64 comparisons diverged from the eager twin, and under `LIMIT` the divergence is a different ROW SET rather than a different row order. Not a GPU change: like the eager code and the fused two-star lane, it collects on CPU polars for both `polars` and `polars-gpu`. **Measured** on dgx-spark, matched graph-benchmark q1–q9 lane, one perf lock per experiment, master tree vs PR tree position-balanced `M P P M P M M P`, per-slot medians (never best-of), rows and canonical values compared on every cell, and replicated end to end in a second independent run: **`engine='polars'` q1 `13.31 → 8.96 ms` (−32.7%), q3 `8.54 → 5.53 ms` (−35.3%), q4 `6.99 → 5.05 ms` (−27.8%) at 20k**, per-slot ranges non-overlapping on all three in both runs; at 100k q1 `42.59 → 31.45 ms` (−26.2%) and q4 `12.19 → 10.43 ms` (−14.4%), with q3 `−9.1%…−13.0%` (non-overlapping in one run, overlapping in the other). Against same-session embedded Kuzu at 20k that widens q1 from a 1.12× win to **1.66×**, moves q3 from a **1.37× LOSS to a TIE** (Kuzu 6.29 ms; the two slot ranges overlap, so it is a tie and not a win), and narrows q4 from a 2.10× loss to **1.51× — still a loss**. The cells this lane is not called for are unchanged: q5 and q8 are ties at both scales in both runs, and q8 stays a win over Kuzu. The pandas arm — untouched by this change — reproduces the reference board to +0.8%…+9.4% and same-session Kuzu reproduces it to −2.2%…+1.5% on the cells at issue, which is what shows the harness matches it. Value identity is the gate throughout: a differential over 19 shapes × 10 graphs, compared ROW-ORDER and COLUMN-ORDER sensitively against both the eager code and the pandas oracle, found zero divergences from the eager code; pinned tests cover multiplicity on BOTH arms of the hop (duplicate node rows, parallel edges, self-loops), null placement on group keys and on aggregate values, empty matches, dangling endpoints, non-numeric ids, degenerate column bindings, and every decline. One PRE-EXISTING divergence is disclosed rather than quietly changed: the polars property lookup is not deduplicated by node id while the pandas one is, so a node table carrying the same id twice multiplies matched rows on polars only — the fused lane reproduces the eager polars answer exactly, and a test pins both sides. - **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. diff --git a/bin/test-polars.sh b/bin/test-polars.sh index 643d72e842..9f524cad59 100755 --- a/bin/test-polars.sh +++ b/bin/test-polars.sh @@ -38,6 +38,7 @@ POLARS_TEST_FILES=( graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py graphistry/tests/compute/gfql/test_residual_polars_native.py graphistry/tests/compute/gfql/cypher/test_grouped_aggregate_fused_polars.py + graphistry/tests/compute/gfql/cypher/test_grouped_aggregate_lowcard_count.py # module-level `importorskip("polars")` files that previously ran in no lane at all graphistry/tests/compute/gfql/test_engine_polars_narrow_combine.py graphistry/tests/compute/gfql/test_engine_polars_semi_key_dedup.py diff --git a/graphistry/compute/gfql_fast_paths.py b/graphistry/compute/gfql_fast_paths.py index ce0fb04a5a..24bd83f628 100644 --- a/graphistry/compute/gfql_fast_paths.py +++ b/graphistry/compute/gfql_fast_paths.py @@ -1574,6 +1574,136 @@ def _property_ref(expr: Any, valid_aliases: Sequence[str]) -> Optional[Tuple[str _GROUPED_AGG_LOOKUP_KEY_FMT = "__gfql_t3_{alias}_id__" +# Thresholds for the low-cardinality pure-count(*) formulation below. Both were fixed +# from an interleaved crossover sweep (polars 1.35.2, 20 threads, 90 samples/arm/cell) +# BEFORE the formulation was validated on any query, because a threshold chosen after +# seeing the verdicts is unfalsifiable. +# +# * ``group_by(maintain_order=True).agg(pl.len())`` carries a FLAT ~2 ms coordination +# cost that exists only at LOW group cardinality and vanishes between 32 and 64 +# groups (int keys, 20,000 rows: 32 groups 2.054 ms -> 48 groups 0.411 -> 64 groups +# 0.291). +# * ``value_counts`` has no such cost but scales WORSE with input rows: at 1,000,000 +# rows it loses even at 2 groups (4.137 ms -> 8.591 ms). +# +# So neither bound alone is sound; the gate needs both. 32 is the largest cardinality +# whose worst case (group_by p25 vs value_counts p75) still favours value_counts in every +# measured dtype x row-count cell -- 48 groups already fails at 0.96x on string keys. +# 100,000 is the largest measured input-row count where that holds for every cardinality +# <= 32 in both key dtypes -- 150,000 fails at 0.82x on string keys. +_LOWCARD_COUNT_MAX_GROUPS = 32 +_LOWCARD_COUNT_MAX_INPUT_ROWS = 100_000 + + +def _low_cardinality_pure_count_key( + group_keys: Sequence[str], + agg_specs: Sequence[Tuple[str, str, Optional[str]]], +) -> Optional[Tuple[str, str]]: + """``(group_key, out_alias)`` iff this is a single-key, pure ``count(*)`` aggregate. + + Pure ``count(*)`` is the ONLY aggregate the alternative formulation below can express, + and the only one measured value-identical to ``pl.len()``. Anything else -- a second + group key, a second aggregate, ``avg``/``sum``/``min``/``max``, or a ``count`` over a + named property (which counts NON-NULL values, not rows) -- declines here. + + ``out_alias == group_key`` also declines: both formulations raise polars + ``DuplicateError`` on it, and declining keeps the twin's error rather than minting a + second one. + """ + if len(group_keys) != 1 or len(agg_specs) != 1: + return None + out_alias, func, expr_col = agg_specs[0] + if func != "count" or expr_col is not None: + return None + group_key = group_keys[0] + if out_alias == group_key: + return None + return group_key, out_alias + + +def _low_cardinality_pure_count_plan( + work_lf: "pl.LazyFrame", + *, + node_col: str, + group_keys: Sequence[str], + agg_specs: Sequence[Tuple[str, str, Optional[str]]], + needed_by_alias: Mapping[str, Sequence[Tuple[str, str]]], + frames_by_alias: Mapping[str, DataFrameT], + edge_rows: int, +) -> Optional["pl.LazyFrame"]: + """STRICTLY ADDITIVE alternative for a single-key pure ``count(*)``: emit + ``value_counts`` instead of ``group_by(..).agg(pl.len())``. Returns ``None`` on every + decline and the caller keeps the existing ``group_by`` formulation, so the blast radius + is a decline away from zero. + + The two formulations are VALUE-IDENTICAL wherever this admits -- same key rows, same + counts, same ``UInt32`` count dtype, same treatment of null / NaN / empty-input keys -- + and the caller's gate has already made the following ``sort`` TOTAL over the output + rows, so neither formulation's internal row order can reach the answer. They differ + only in COST, which is why this is a routing decision and not a semantic one. + + THE BOUNDS ARE STATIC, O(1) IN THE DATA, AND THEY ARE UPPER BOUNDS: + + * **group cardinality <= height of the alias node frame supplying the group key.** + The group key column is produced by an inner join that reads ``prop`` out of that one + frame, so every group value in the aggregate input is a ``prop`` value of some row of + it; distinct values cannot exceed its height. ``height`` is metadata. + * **aggregate input rows <= height of the (already filtered) edge frame.** The two + semi-joins only remove edge rows. The property inner-join can MULTIPLY them when a + node id repeats in the node frame -- the lane deliberately does not dedup, because + the eager twin does not -- so the bound only holds once the sole property join is + known to be non-multiplying. Hence the two structural conditions below: exactly one + alias may carry properties, and its node ids must be unique. That uniqueness check + runs on a frame already known to be <= ``_LOWCARD_COUNT_MAX_GROUPS`` rows, so it is + bounded work regardless of graph size. + + BOTH BOUNDS ARE LOOSE IN THE SAFE DIRECTION. A loose bound over-estimates, so it can + only make this DECLINE a shape the alternative would have served -- it can never route + a high-cardinality or high-row aggregate into the wrong formulation, which is the + failure that would matter (that formulation is 2.7 ms slower on a ~20,000-group + aggregate). The cost of looseness is a forgone speedup: a 7,117-row City frame carrying + only 3 distinct countries declines here, because an O(1) height bound cannot see the 3. + + DECLINES: a non-single-key or non-pure-``count(*)`` aggregate; a group key not supplied + by exactly one alias; a second alias also contributing property columns (the row bound + stops holding); a group-key alias frame taller than ``_LOWCARD_COUNT_MAX_GROUPS``, + missing its node-id column, or carrying duplicate node ids; an edge frame taller than + ``_LOWCARD_COUNT_MAX_INPUT_ROWS``. + """ + import polars as pl + + keyed = _low_cardinality_pure_count_key(group_keys, agg_specs) + if keyed is None: + return None + group_key, out_alias = keyed + + owners = [ + alias + for alias, props in needed_by_alias.items() + if any(out_col == group_key for out_col, _ in props) + ] + if len(owners) != 1: + return None + owner = owners[0] + if any(alias != owner and props for alias, props in needed_by_alias.items()): + return None + + owner_frame = frames_by_alias.get(owner) + if owner_frame is None or not isinstance(owner_frame, pl.DataFrame): + return None + if owner_frame.height > _LOWCARD_COUNT_MAX_GROUPS: + return None + if node_col not in owner_frame.columns: + return None + if owner_frame.get_column(node_col).n_unique() != owner_frame.height: + return None + if edge_rows > _LOWCARD_COUNT_MAX_INPUT_ROWS: + return None + + # ``name=`` (polars >= 1.0, and the declared floor is 1.29) keeps the count column out + # of a rename, so a group key literally named ``count`` is served rather than crashing. + return work_lf.select(pl.col(group_key).value_counts(name=out_alias)).unnest(group_key) + def _single_hop_grouped_aggregate_fused_polars( start_nodes: DataFrameT, @@ -1608,6 +1738,11 @@ def _single_hop_grouped_aggregate_fused_polars( ``head``), so the value -- including row ORDER and openCypher's null-largest ordering -- is identical. + ONE aggregate shape has a second, value-identical polars formulation: see + :func:`_low_cardinality_pure_count_plan`. It is chosen only when static O(1) bounds + prove both the group cardinality and the aggregate input rows are low, and it declines + to this ``group_by`` everywhere else. + NOT a GPU change: like the eager twin and the fused two-star lane, it collects on CPU polars for both POLARS and POLARS_GPU. @@ -1699,7 +1834,21 @@ def _single_hop_grouped_aggregate_fused_polars( ) work_lf = work_lf.join(lookup_lf, left_on=edge_col, right_on=lookup_key, how="inner") - out_lf = work_lf.group_by(list(group_keys), maintain_order=True).agg(agg_exprs) + # STRICTLY ADDITIVE: a single-key pure count(*) whose group cardinality and input rows + # are both statically bounded low takes the value_counts formulation, which skips + # polars' partitioned group-by coordination. Every other shape -- and every shape whose + # bounds are not provably low -- keeps the group_by below, unchanged. + out_lf = _low_cardinality_pure_count_plan( + work_lf, + node_col=node_col, + group_keys=group_keys, + agg_specs=agg_specs, + needed_by_alias=needed_by_alias, + frames_by_alias={start_alias: start_nodes, end_alias: end_nodes}, + edge_rows=edges.height, + ) + if out_lf is None: + out_lf = work_lf.group_by(list(group_keys), maintain_order=True).agg(agg_exprs) # openCypher orders NULL as the largest value (ASC -> nulls last, DESC -> nulls # first); polars defaults nulls-first, so pin nulls_last per key exactly as the twin # does. The gate above guarantees this sort is TOTAL over the output rows. diff --git a/graphistry/tests/compute/gfql/cypher/test_grouped_aggregate_lowcard_count.py b/graphistry/tests/compute/gfql/cypher/test_grouped_aggregate_lowcard_count.py new file mode 100644 index 0000000000..adbd472bef --- /dev/null +++ b/graphistry/tests/compute/gfql/cypher/test_grouped_aggregate_lowcard_count.py @@ -0,0 +1,792 @@ +"""LOW-CARDINALITY PURE-``count(*)`` FORMULATION GATE -- +``gfql_fast_paths._low_cardinality_pure_count_plan``. + +Inside the fused single-hop grouped-aggregate lane, a single-key pure ``count(*)`` has a +second, value-identical polars formulation: ``value_counts`` instead of +``group_by(maintain_order=True).agg(pl.len())``. It is not a drop-in. Measured on +dgx-spark (polars 1.35.2, 20 threads, interleaved, 90 samples/arm/cell): + +* ``group_by`` carries a FLAT ~2 ms coordination cost that exists only at LOW group + cardinality and vanishes between 32 and 64 groups; +* ``value_counts`` has no such cost but scales WORSE with input rows -- at 1,000,000 rows + it loses even at 2 groups (4.137 ms -> 8.591 ms). + +So the choice is a ROUTING decision under two static bounds, and BOTH are needed: applied +ungated, the same formulation makes the graph-benchmark q1 cell (~20,000 groups over +~200,000 rows) 2.7 ms SLOWER at 20k and 8.6 ms slower at 100k. + +WHAT THESE TESTS PIN: + +* **The bounds are UPPER bounds.** ``test_admitted_shapes_respect_the_measured_bounds`` + reaches into the lane's own work frame on every admission across the whole graph x shape + corpus and asserts the REALIZED group cardinality and input rows are inside the + thresholds. That is the soundness claim itself, checked against data rather than argued: + an under-estimating bound would route a big aggregate into the slow formulation. +* **Which shapes are admitted and which are DECLINED**, enumerated, including the ones + that decline for a REASON THAT IS NOT CARDINALITY (a second property-bearing alias + breaks the row bound's derivation; duplicate node ids break it too). +* **Value identity against the unmodified product.** Every comparison runs the same query + twice -- once with the gate live, once with it forced to decline, which IS the + pre-change code -- row-order and column-order sensitively, plus dtypes, plus a pandas + oracle. +* **Engine reach**: pandas and cudf must never enter the lane at all; polars and + polars-gpu must. +""" +from __future__ import annotations + +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple + +import numpy as np +import pandas as pd +import pytest + +import graphistry +from graphistry.Plottable import Plottable +import graphistry.compute.gfql_fast_paths as gfql_fast_paths_module + + +MAX_GROUPS = gfql_fast_paths_module._LOWCARD_COUNT_MAX_GROUPS +MAX_INPUT_ROWS = gfql_fast_paths_module._LOWCARD_COUNT_MAX_INPUT_ROWS + + +# --------------------------------------------------------------------------- fixtures + +def _base_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + """Four persons, four cities, one non-matching edge label. The CITY frame is the + group-key side and has 4 rows, comfortably inside ``MAX_GROUPS``.""" + nodes = pd.DataFrame({ + "id": [1, 2, 3, 4, 5, 6, 7, 8], + "kind": ["P", "P", "P", "P", "C", "C", "C", "C"], + "age": [20, 30, 40, 50, None, None, None, None], + "city": [None, None, None, None, "LA", "NY", "SF", "LA"], + "country": [None, None, None, None, "US", "US", "US", "MX"], + }) + edges = pd.DataFrame({ + "s": [1, 2, 3, 4, 1, 2, 3, 4, 1], + "d": [5, 5, 6, 7, 8, 6, 8, 8, 5], + "rel": ["L", "L", "L", "L", "L", "L", "L", "L", "X"], + }) + return nodes, edges + + +def _null_group_key_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + nodes, edges = _base_data() + nodes = nodes.copy() + nodes.loc[nodes["id"].isin([6, 7]), "city"] = None + return nodes, edges + + +def _all_null_group_key_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + nodes, edges = _base_data() + nodes = nodes.copy() + nodes["city"] = None + return nodes, edges + + +def _empty_edges_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + nodes, edges = _base_data() + return nodes, edges.iloc[:0].copy() + + +def _no_matching_nodes_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + nodes, edges = _base_data() + return nodes.assign(kind="Z"), edges + + +def _string_ids_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + nodes, edges = _base_data() + return ( + nodes.assign(id=[f"n{i}" for i in nodes["id"]]), + edges.assign(s=[f"n{i}" for i in edges["s"]], d=[f"n{i}" for i in edges["d"]]), + ) + + +def _self_loops_parallel_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + nodes, edges = _base_data() + extra = pd.DataFrame({"s": [5, 1, 1, 6], "d": [5, 5, 5, 6], "rel": ["L", "L", "L", "L"]}) + return nodes, pd.concat([edges, extra], ignore_index=True) + + +def _dangling_endpoints_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + nodes, edges = _base_data() + extra = pd.DataFrame({"s": [99, 1], "d": [5, 98], "rel": ["L", "L"]}) + return nodes, pd.concat([edges, extra], ignore_index=True) + + +def _dup_end_node_rows_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + """The CITY frame carries an id twice. The lane deliberately does not dedup property + lookups, so this MULTIPLIES matched rows -- which is exactly what breaks the + ``rows <= edge frame height`` derivation, so the gate must decline it.""" + nodes, edges = _base_data() + return pd.concat([nodes, nodes.iloc[[4, 5]]], ignore_index=True), edges + + +def _dup_start_node_rows_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + """Duplicates on the PERSON side. That side contributes no property column for the + q4 shape, so it only feeds a semi-join and cannot multiply: the gate still admits.""" + nodes, edges = _base_data() + return pd.concat([nodes, nodes.iloc[[0, 1]]], ignore_index=True), edges + + +def _wide_group_key_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + """``MAX_GROUPS + 1`` cities: the O(1) height bound cannot certify low cardinality even + though every city here shares ONE country. This is the q1-shaped decline in miniature.""" + n_cities = MAX_GROUPS + 1 + # DTYPES ARE EXPLICIT ON PURPOSE. Letting ``age`` fall out as an OBJECT column of ints + # and Nones makes cudf raise MixedTypeError on ingest -- a fixture defect that a + # CPU-only run cannot see, and that showed up only in the RAPIDS image. + persons = pd.DataFrame({ + "id": np.arange(1, 21, dtype="int64"), + "kind": "P", + "age": np.arange(20, 40, dtype="float64"), + "city": pd.Series([None] * 20, dtype="object"), + "country": pd.Series([None] * 20, dtype="object"), + }) + cities = pd.DataFrame({ + "id": np.arange(1000, 1000 + n_cities, dtype="int64"), + "kind": "C", + "age": np.full(n_cities, np.nan, dtype="float64"), + "city": pd.Series([f"city{i}" for i in range(n_cities)], dtype="object"), + "country": pd.Series(["US"] * n_cities, dtype="object"), + }) + rng = np.random.default_rng(3) + edges = pd.DataFrame({ + "s": rng.integers(1, 21, 200).astype("int64"), + "d": rng.integers(1000, 1000 + n_cities, 200).astype("int64"), + "rel": "L", + }) + return pd.concat([persons, cities], ignore_index=True), edges + + +def _exactly_max_groups_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + """``MAX_GROUPS`` cities exactly -- the last admitted height.""" + nodes, edges = _wide_group_key_data() + keep = nodes["kind"].ne("C") | nodes["id"].lt(1000 + MAX_GROUPS) + nodes = nodes[keep].reset_index(drop=True) + edges = edges[edges["d"] < 1000 + MAX_GROUPS].reset_index(drop=True) + return nodes, edges + + +def _many_edges_data() -> Tuple[pd.DataFrame, pd.DataFrame]: + """``MAX_INPUT_ROWS + 1`` edges over a 3-city frame: cardinality is tiny, the ROW bound + is what declines it.""" + n_persons = 50 + persons = pd.DataFrame({ + "id": np.arange(1, n_persons + 1, dtype="int64"), + "kind": "P", + "age": (np.arange(20, 20 + n_persons, dtype="float64") % 60), + "city": pd.Series([None] * n_persons, dtype="object"), + "country": pd.Series([None] * n_persons, dtype="object"), + }) + cities = pd.DataFrame({ + "id": np.array([1000, 1001, 1002], dtype="int64"), + "kind": "C", + "age": np.full(3, np.nan, dtype="float64"), + "city": pd.Series(["LA", "NY", "SF"], dtype="object"), + "country": pd.Series(["US", "US", "MX"], dtype="object"), + }) + m = MAX_INPUT_ROWS + 1 + rng = np.random.default_rng(5) + edges = pd.DataFrame({ + "s": rng.integers(1, n_persons + 1, m).astype("int64"), + "d": rng.integers(1000, 1003, m).astype("int64"), + "rel": "L", + }) + return pd.concat([persons, cities], ignore_index=True), edges + + +_GRAPHS: Dict[str, Callable[[], Tuple[pd.DataFrame, pd.DataFrame]]] = { + "base": _base_data, + "null_group_key": _null_group_key_data, + "all_null_group_key": _all_null_group_key_data, + "empty_edges": _empty_edges_data, + "no_matching_nodes": _no_matching_nodes_data, + "string_ids": _string_ids_data, + "self_loops_parallel": _self_loops_parallel_data, + "dangling_endpoints": _dangling_endpoints_data, + "dup_end_node_rows": _dup_end_node_rows_data, + "dup_start_node_rows": _dup_start_node_rows_data, + "exactly_max_groups": _exactly_max_groups_data, + "wide_group_key": _wide_group_key_data, +} + + +# ------------------------------------------------------------------------- shape corpus + +# ADMITTED on the `base`-shaped graphs: single group key, pure count(*), only the group-key +# alias carries a property, and the fused lane's own order-totality gate is satisfied. +Q_COUNT_STAR = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.city AS city, count(*) AS n " + "ORDER BY n DESC, city ASC LIMIT 3") +Q_COUNT_STAR_NO_LIMIT = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.city AS city, count(*) AS n " + "ORDER BY n DESC, city ASC") +Q_COUNT_ALIAS = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.city AS city, count(p) AS n " + "ORDER BY n DESC, city ASC LIMIT 3") +Q_GROUP_KEY_DESC = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.country AS co, count(*) AS n " + "ORDER BY co DESC") +Q_WHERE_ON_START = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) WHERE p.age >= 30 " + "RETURN c.city AS city, count(*) AS n ORDER BY n DESC, city ASC") +Q_NO_EDGE_FILTER = ( + "MATCH (p {kind:'P'})-[]->(c {kind:'C'}) RETURN c.city AS city, count(*) AS n " + "ORDER BY n DESC, city ASC") +Q_UNFILTERED_ENDS = ( + "MATCH (p)-[{rel:'L'}]->(c) RETURN c.city AS city, count(*) AS n ORDER BY n DESC, city ASC") + +_ADMITTED_SHAPES: List[Tuple[str, str]] = [ + ("count_star_limit", Q_COUNT_STAR), + ("count_star_no_limit", Q_COUNT_STAR_NO_LIMIT), + ("count_alias", Q_COUNT_ALIAS), + ("group_key_desc", Q_GROUP_KEY_DESC), + ("where_on_start", Q_WHERE_ON_START), + ("no_edge_filter", Q_NO_EDGE_FILTER), + ("unfiltered_ends", Q_UNFILTERED_ENDS), +] + +# DECLINED by the gate, for reasons that are NOT the two thresholds. The fused lane still +# serves each of these -- through the unchanged group_by formulation. +Q_AVG = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) WHERE c.country = 'US' " + "RETURN c.city AS city, avg(p.age) AS a ORDER BY a ASC, city ASC LIMIT 5") +Q_SUM = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.city AS city, sum(p.age) AS s " + "ORDER BY s DESC, city ASC") +Q_COUNT_PROP = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.city AS city, count(p.age) AS n " + "ORDER BY n DESC, city ASC") +Q_TWO_GROUP_KEYS = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.country AS co, c.city AS city, " + "count(*) AS n ORDER BY n DESC, co ASC, city ASC") +Q_TWO_AGGS = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.city AS city, count(*) AS n, " + "avg(p.age) AS a ORDER BY city ASC") +Q_SECOND_ALIAS_PROP = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) WHERE p.age >= 0 " + "RETURN c.city AS city, count(p.age) AS n, count(*) AS m ORDER BY city ASC") +Q_GROUP_START_PROP = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN p.age AS age, count(*) AS n " + "ORDER BY age ASC") + +_DECLINED_SHAPES: List[Tuple[str, str]] = [ + ("avg_not_a_count", Q_AVG), + ("sum_not_a_count", Q_SUM), + ("count_over_property", Q_COUNT_PROP), + ("two_group_keys", Q_TWO_GROUP_KEYS), + ("two_aggregates", Q_TWO_AGGS), + ("second_alias_carries_a_property", Q_SECOND_ALIAS_PROP), +] + +_ALL_SHAPES = _ADMITTED_SHAPES + _DECLINED_SHAPES + [("group_start_prop", Q_GROUP_START_PROP)] + +# Shapes whose ORDER BY is not total over the output rows: the FUSED lane declines them +# before the gate is ever consulted, so the gate must record zero calls. +Q_PARTIAL_ORDER = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.city AS city, count(*) AS n " + "ORDER BY n DESC") +Q_NO_ORDER = ( + "MATCH (p {kind:'P'})-[{rel:'L'}]->(c {kind:'C'}) RETURN c.city AS city, count(*) AS n") + + +# ------------------------------------------------------------------------------ helpers + +def _require_polars() -> Any: + return pytest.importorskip("polars") + + +def _require_polars_gpu() -> None: + pl = _require_polars() + pytest.importorskip("cudf_polars") + try: + pl.DataFrame({"a": [1, 2]}).lazy().filter(pl.col("a") > 0).collect( + engine=pl.GPUEngine(raise_on_fail=True) + ) + except Exception as exc: # pragma: no cover - CPU-only CI + pytest.skip(f"cudf_polars installed but the GPU collect probe failed: {exc}") + + +def _graph(engine: str, nodes_df: pd.DataFrame, edges_df: pd.DataFrame) -> Plottable: + if engine == "pandas": + return graphistry.nodes(nodes_df, "id").edges(edges_df, "s", "d") + if engine == "cudf": + cudf = pytest.importorskip("cudf") + try: + _ = cudf.Series([1, 2, 3]) + except Exception as exc: # pragma: no cover - environment-dependent + pytest.skip(f"cudf installed but the runtime is unavailable: {exc}") + return graphistry.nodes(cudf.from_pandas(nodes_df), "id").edges( + cudf.from_pandas(edges_df), "s", "d") + if engine == "polars-gpu": + _require_polars_gpu() + pl = _require_polars() + return graphistry.nodes(pl.from_pandas(nodes_df), "id").edges( + pl.from_pandas(edges_df), "s", "d") + + +def _engine_arg(engine: str) -> str: + return "polars" if engine == "polars-gpu" else engine + + +def _records(result: Plottable) -> Tuple[List[str], List[str], List[Dict[str, Any]]]: + """Row-order AND column-order sensitive comparison value, plus the DTYPE list. + + The dtypes travel because the two formulations produce the count column independently + (``pl.len()`` vs ``value_counts``' own field) -- an equal-valued column at a different + width would be a real divergence that record equality alone cannot see.""" + df = result._nodes + if not isinstance(df, pd.DataFrame): + df = df.to_pandas() if hasattr(df, "to_pandas") else pd.DataFrame(df) + rows = [ + {k: (None if v is None or (isinstance(v, float) and v != v) else v) for k, v in row.items()} + for row in df.to_dict(orient="records") + ] + return [str(c) for c in df.columns], [str(d) for d in df.dtypes], rows + + +def _probe_gate(monkeypatch: pytest.MonkeyPatch) -> List[bool]: + """One entry per gate CALL: True=admitted, False=declined. Empty means the fused lane + never got as far as consulting it.""" + calls: List[bool] = [] + original = gfql_fast_paths_module._low_cardinality_pure_count_plan + + def probe(*args: Any, **kwargs: Any) -> Any: + result = original(*args, **kwargs) + calls.append(result is not None) + return result + + monkeypatch.setattr(gfql_fast_paths_module, "_low_cardinality_pure_count_plan", probe) + return calls + + +def _force_decline(monkeypatch: pytest.MonkeyPatch) -> None: + """Turn the gate off entirely. What runs then IS the unmodified product.""" + monkeypatch.setattr( + gfql_fast_paths_module, + "_low_cardinality_pure_count_plan", + lambda *args, **kwargs: None, + ) + + +def _realized(monkeypatch: pytest.MonkeyPatch) -> List[Dict[str, int]]: + """On every ADMISSION, collect the lane's own work frame and record the REALIZED input + rows and group cardinality -- the two quantities the static bounds claim to bound.""" + seen: List[Dict[str, int]] = [] + original = gfql_fast_paths_module._low_cardinality_pure_count_plan + + def probe(work_lf: Any, **kwargs: Any) -> Any: + result = original(work_lf, **kwargs) + if result is not None: + group_key = list(kwargs["group_keys"])[0] + frame = work_lf.collect() + seen.append({ + "rows": frame.height, + "cardinality": frame.get_column(group_key).n_unique(), + "edge_rows": int(kwargs["edge_rows"]), + }) + return result + + monkeypatch.setattr(gfql_fast_paths_module, "_low_cardinality_pure_count_plan", probe) + return seen + + +# ------------------------------------------------------------------- the soundness claim + +@pytest.mark.parametrize("graph_name", sorted(_GRAPHS)) +@pytest.mark.parametrize("shape_name,query", _ALL_SHAPES) +def test_admitted_shapes_respect_the_measured_bounds( + monkeypatch: pytest.MonkeyPatch, graph_name: str, shape_name: str, query: str +) -> None: + """THE SOUNDNESS TEST. Whenever the gate admits, the REALIZED group cardinality must be + <= MAX_GROUPS and the REALIZED aggregate input rows <= MAX_INPUT_ROWS. + + The bounds are static and O(1) (a node-frame height and an edge-frame height), so this + is where the claim that they are UPPER bounds gets checked against data instead of + argued. An under-estimating bound shows up here as an admission whose realized numbers + are outside the thresholds -- i.e. a shape routed into the formulation the crossover + sweep says is the slower one.""" + _require_polars() + nodes_df, edges_df = _GRAPHS[graph_name]() + graph = _graph("polars", nodes_df, edges_df) + realized = _realized(monkeypatch) + graph.gfql(query, engine="polars") + for entry in realized: + assert entry["cardinality"] <= MAX_GROUPS, ( + f"{graph_name}/{shape_name}: admitted with {entry['cardinality']} groups, over " + f"the committed bound of {MAX_GROUPS} -- the cardinality bound UNDER-estimated" + ) + assert entry["rows"] <= MAX_INPUT_ROWS, ( + f"{graph_name}/{shape_name}: admitted with {entry['rows']} aggregate input " + f"rows, over the committed bound of {MAX_INPUT_ROWS}" + ) + assert entry["rows"] <= entry["edge_rows"], ( + f"{graph_name}/{shape_name}: the property join MULTIPLIED rows " + f"({entry['edge_rows']} edges -> {entry['rows']} rows), so the edge-height row " + "bound does not hold and this shape should have declined" + ) + + +# ------------------------------------------------------- value identity (differential) + +@pytest.mark.parametrize("engine", ["pandas", "polars", "cudf", "polars-gpu"]) +@pytest.mark.parametrize("graph_name", sorted(_GRAPHS)) +@pytest.mark.parametrize("shape_name,query", _ALL_SHAPES) +def test_gate_is_value_identical_to_the_unmodified_product( + monkeypatch: pytest.MonkeyPatch, engine: str, graph_name: str, shape_name: str, query: str +) -> None: + """Differential against the gate forced OFF, which is byte-for-byte the pre-change code + path. Row order, column order, dtypes and values all travel.""" + nodes_df, edges_df = _GRAPHS[graph_name]() + graph = _graph(engine, nodes_df, edges_df) + live = _records(graph.gfql(query, engine=_engine_arg(engine))) + + with monkeypatch.context() as ctx: + _force_decline(ctx) + baseline = _records(graph.gfql(query, engine=_engine_arg(engine))) + + assert live == baseline, f"{engine}/{graph_name}/{shape_name} diverged from the gate-off product" + + +@pytest.mark.parametrize("graph_name", sorted(_GRAPHS)) +@pytest.mark.parametrize("shape_name,query", _ALL_SHAPES) +def test_polars_matches_the_pandas_oracle(graph_name: str, shape_name: str, query: str) -> None: + """A second, independent reference: the pandas branch never enters this lane at all. + + The two duplicate-id graphs are excluded from the VALUE comparison because the polars + branch's property lookup is not deduplicated by node id while the pandas branch's is -- + a pre-existing divergence disclosed by #1823, not something this gate introduces. It + bites on whichever arm carries a property column, hence BOTH fixtures. The gate-off + differential above still covers them, and it is the comparison that matters here.""" + _require_polars() + if graph_name in {"dup_end_node_rows", "dup_start_node_rows"}: + pytest.skip("pre-existing polars-vs-pandas dedup divergence, disclosed in #1823") + nodes_df, edges_df = _GRAPHS[graph_name]() + _, _, pandas_rows = _records(_graph("pandas", nodes_df, edges_df).gfql(query, engine="pandas")) + _, _, polars_rows = _records(_graph("polars", nodes_df, edges_df).gfql(query, engine="polars")) + assert polars_rows == pandas_rows, f"{graph_name}/{shape_name} polars != pandas" + + +# -------------------------------------------------------------------------- engagement + +@pytest.mark.parametrize("shape_name,query", _ADMITTED_SHAPES) +def test_admitted_shapes_are_admitted( + monkeypatch: pytest.MonkeyPatch, shape_name: str, query: str +) -> None: + _require_polars() + nodes_df, edges_df = _base_data() + calls = _probe_gate(monkeypatch) + _graph("polars", nodes_df, edges_df).gfql(query, engine="polars") + assert calls == [True], f"{shape_name} should be admitted, got {calls}" + + +@pytest.mark.parametrize("shape_name,query", _DECLINED_SHAPES) +def test_declined_shapes_are_declined( + monkeypatch: pytest.MonkeyPatch, shape_name: str, query: str +) -> None: + """These reach the gate and are turned away -- the fused lane still answers them, via + the unchanged group_by. A decline is the safe outcome, never a wrong one.""" + _require_polars() + nodes_df, edges_df = _base_data() + calls = _probe_gate(monkeypatch) + _graph("polars", nodes_df, edges_df).gfql(query, engine="polars") + assert calls == [False], f"{shape_name} should be declined, got {calls}" + + +def test_cardinality_bound_declines_a_wide_group_key(monkeypatch: pytest.MonkeyPatch) -> None: + """MAX_GROUPS + 1 city rows. Every one of them carries country 'US', so the TRUE + cardinality of ``c.country`` is 1 -- the O(1) height bound cannot see that, and the + resulting decline is the deliberate cost of not paying for an exact count.""" + _require_polars() + nodes_df, edges_df = _wide_group_key_data() + calls = _probe_gate(monkeypatch) + result = _graph("polars", nodes_df, edges_df).gfql(Q_GROUP_KEY_DESC, engine="polars") + assert calls == [False] + assert len(result._nodes) == 1, "the true cardinality really was 1; the bound was loose" + + +def test_cardinality_bound_admits_exactly_max_groups(monkeypatch: pytest.MonkeyPatch) -> None: + """The boundary is inclusive: a node frame of exactly MAX_GROUPS rows is admitted.""" + _require_polars() + nodes_df, edges_df = _exactly_max_groups_data() + calls = _probe_gate(monkeypatch) + _graph("polars", nodes_df, edges_df).gfql(Q_GROUP_KEY_DESC, engine="polars") + assert calls == [True] + + +def test_row_bound_declines_a_large_edge_frame(monkeypatch: pytest.MonkeyPatch) -> None: + """MAX_INPUT_ROWS + 1 edges over a 3-city frame: cardinality is tiny, so only the ROW + bound can decline this. It must.""" + _require_polars() + nodes_df, edges_df = _many_edges_data() + calls = _probe_gate(monkeypatch) + _graph("polars", nodes_df, edges_df).gfql(Q_GROUP_KEY_DESC, engine="polars") + assert calls == [False] + + +def test_row_bound_admits_exactly_max_input_rows(monkeypatch: pytest.MonkeyPatch) -> None: + _require_polars() + nodes_df, edges_df = _many_edges_data() + edges_df = edges_df.iloc[:MAX_INPUT_ROWS].reset_index(drop=True) + calls = _probe_gate(monkeypatch) + _graph("polars", nodes_df, edges_df).gfql(Q_GROUP_KEY_DESC, engine="polars") + assert calls == [True] + + +def test_duplicate_group_alias_node_ids_decline(monkeypatch: pytest.MonkeyPatch) -> None: + """Duplicate ids in the group-key alias frame make the property join MULTIPLY rows, so + ``rows <= edge frame height`` stops holding. The gate declines rather than reason about + a bound it can no longer prove.""" + _require_polars() + nodes_df, edges_df = _dup_end_node_rows_data() + calls = _probe_gate(monkeypatch) + _graph("polars", nodes_df, edges_df).gfql(Q_COUNT_STAR, engine="polars") + assert calls == [False] + + +def test_duplicate_other_alias_node_ids_still_admit(monkeypatch: pytest.MonkeyPatch) -> None: + """Duplicates on the side that contributes NO property column feed only a semi-join, + which cannot multiply -- so the bound survives and the gate admits. Kept separate from + the test above so a change that collapses the two sides is visible.""" + _require_polars() + nodes_df, edges_df = _dup_start_node_rows_data() + calls = _probe_gate(monkeypatch) + _graph("polars", nodes_df, edges_df).gfql(Q_COUNT_STAR, engine="polars") + assert calls == [True] + + +@pytest.mark.parametrize("shape_name,query", [ + ("partial_order", Q_PARTIAL_ORDER), + ("no_order_by", Q_NO_ORDER), +]) +def test_fused_lane_order_gate_runs_first( + monkeypatch: pytest.MonkeyPatch, shape_name: str, query: str +) -> None: + """The fused lane's own order-totality gate declines these BEFORE any plan is built, so + the low-cardinality gate is never consulted. Recorded so a reordering that consulted it + first -- and thereby changed which shapes the fused lane can serve -- is visible.""" + _require_polars() + nodes_df, edges_df = _base_data() + calls = _probe_gate(monkeypatch) + _graph("polars", nodes_df, edges_df).gfql(query, engine="polars") + assert calls == [], f"{shape_name} reached the gate; it should not have" + + +@pytest.mark.parametrize("engine", ["pandas", "cudf"]) +def test_non_polars_engines_never_reach_the_gate( + monkeypatch: pytest.MonkeyPatch, engine: str +) -> None: + nodes_df, edges_df = _base_data() + graph = _graph(engine, nodes_df, edges_df) + calls = _probe_gate(monkeypatch) + graph.gfql(Q_COUNT_STAR, engine=engine) + assert calls == [], f"{engine} reached a polars-only gate" + + +def test_polars_gpu_engine_reaches_the_gate(monkeypatch: pytest.MonkeyPatch) -> None: + """polars-gpu shares the CPU-collected fused lane, so it is admitted identically. A + SKIP here is a coverage boundary, not a pass.""" + nodes_df, edges_df = _base_data() + graph = _graph("polars-gpu", nodes_df, edges_df) + calls = _probe_gate(monkeypatch) + graph.gfql(Q_COUNT_STAR, engine="polars") + assert calls == [True] + + +# ------------------------------------------------------- unit contract of the gate itself + +def _gate( + *, + group_keys: Sequence[str] = ("city",), + agg_specs: Sequence[Tuple[str, str, Optional[str]]] = (("n", "count", None),), + needed_by_alias: Optional[Dict[str, List[Tuple[str, str]]]] = None, + frames: Optional[Dict[str, Any]] = None, + edge_rows: int = 10, + node_col: str = "id", +) -> Any: + pl = _require_polars() + if needed_by_alias is None: + needed_by_alias = {"p": [], "c": [("city", "city")]} + if frames is None: + frames = { + "p": pl.DataFrame({"id": [1, 2, 3]}), + "c": pl.DataFrame({"id": [10, 11], "city": ["LA", "NY"]}), + } + work = pl.DataFrame({"city": ["LA", "NY", "LA"]}).lazy() + return gfql_fast_paths_module._low_cardinality_pure_count_plan( + work, + node_col=node_col, + group_keys=group_keys, + agg_specs=agg_specs, + needed_by_alias=needed_by_alias, + frames_by_alias=frames, + edge_rows=edge_rows, + ) + + +def test_gate_unit_admits_the_canonical_shape() -> None: + assert _gate() is not None + + +@pytest.mark.parametrize("case,kwargs", [ + ("two_group_keys", {"group_keys": ("city", "country")}), + ("no_group_keys", {"group_keys": ()}), + ("two_aggregates", {"agg_specs": (("n", "count", None), ("m", "count", None))}), + ("no_aggregates", {"agg_specs": ()}), + ("avg_aggregate", {"agg_specs": (("a", "avg", "age"),)}), + # The next two are UNREACHABLE from the cypher surface -- the fast path refuses a + # non-count aggregate without an expression alias long before the fused lane is built + # -- but the gate is a standalone function and its contract is checked here, not + # inherited. Mutation testing found this: dropping ``func != "count"`` from the guard + # SURVIVED the whole suite until these two cases existed. + ("avg_without_an_expression", {"agg_specs": (("a", "avg", None),)}), + ("sum_without_an_expression", {"agg_specs": (("s", "sum", None),)}), + ("count_over_property", {"agg_specs": (("n", "count", "age"),)}), + ("out_alias_equals_group_key", {"agg_specs": (("city", "count", None),)}), + ("edge_rows_over_bound", {"edge_rows": MAX_INPUT_ROWS + 1}), +]) +def test_gate_unit_declines(case: str, kwargs: Dict[str, Any]) -> None: + assert _gate(**kwargs) is None, f"{case} should decline" + + +def test_gate_unit_declines_when_the_group_key_has_no_owning_alias() -> None: + assert _gate(needed_by_alias={"p": [], "c": [("other", "city")]}) is None + + +def test_gate_unit_declines_when_two_aliases_own_the_group_key() -> None: + """DISCLOSED: the ``len(owners) != 1`` guard is PROVABLY REDUNDANT against the check + that immediately follows it, and mutation testing says so -- relaxing it to + ``len(owners) < 1`` survives the entire suite. Two owners means a second alias with a + NON-EMPTY property list, which the ``other alias carries properties`` check declines + anyway, so no input can distinguish the two forms. The guard is kept because it states + the precondition the height bound depends on; this test pins the OUTCOME, which is all + that is observable.""" + pl = _require_polars() + assert _gate( + needed_by_alias={"p": [("city", "city")], "c": [("city", "city")]}, + frames={ + "p": pl.DataFrame({"id": [1, 2], "city": ["LA", "NY"]}), + "c": pl.DataFrame({"id": [10, 11], "city": ["LA", "NY"]}), + }, + ) is None + + +def test_gate_unit_declines_when_the_other_alias_carries_properties() -> None: + pl = _require_polars() + assert _gate( + needed_by_alias={"p": [("age", "age")], "c": [("city", "city")]}, + frames={ + "p": pl.DataFrame({"id": [1, 2], "age": [3, 4]}), + "c": pl.DataFrame({"id": [10, 11], "city": ["LA", "NY"]}), + }, + ) is None + + +def test_gate_unit_declines_a_tall_owner_frame() -> None: + pl = _require_polars() + tall = MAX_GROUPS + 1 + assert _gate(frames={ + "p": pl.DataFrame({"id": [1]}), + "c": pl.DataFrame({"id": list(range(tall)), "city": ["LA"] * tall}), + }) is None + + +def test_gate_unit_admits_an_owner_frame_of_exactly_max_groups() -> None: + pl = _require_polars() + assert _gate(frames={ + "p": pl.DataFrame({"id": [1]}), + "c": pl.DataFrame({"id": list(range(MAX_GROUPS)), "city": ["LA"] * MAX_GROUPS}), + }) is not None + + +def test_gate_unit_declines_duplicate_owner_node_ids() -> None: + pl = _require_polars() + assert _gate(frames={ + "p": pl.DataFrame({"id": [1]}), + "c": pl.DataFrame({"id": [10, 10], "city": ["LA", "NY"]}), + }) is None + + +def test_gate_unit_declines_a_missing_node_id_column() -> None: + pl = _require_polars() + assert _gate(frames={ + "p": pl.DataFrame({"id": [1]}), + "c": pl.DataFrame({"other": [10, 11], "city": ["LA", "NY"]}), + }) is None + + +def test_gate_unit_declines_a_non_polars_owner_frame() -> None: + assert _gate(frames={ + "p": pd.DataFrame({"id": [1]}), + "c": pd.DataFrame({"id": [10, 11], "city": ["LA", "NY"]}), + }) is None + + +def test_gate_unit_serves_a_group_key_literally_named_count() -> None: + """``value_counts`` names its output column ``count`` by default, which would collide. + The lane passes ``name=`` instead of renaming, so this shape is SERVED, and it must + agree with the group_by twin.""" + pl = _require_polars() + work = pl.DataFrame({"count": ["a", "b", "a"]}).lazy() + plan = gfql_fast_paths_module._low_cardinality_pure_count_plan( + work, + node_col="id", + group_keys=["count"], + agg_specs=[("n", "count", None)], + needed_by_alias={"p": [], "c": [("count", "label")]}, + frames_by_alias={ + "p": pl.DataFrame({"id": [1]}), + "c": pl.DataFrame({"id": [10, 11], "label": ["a", "b"]}), + }, + edge_rows=3, + ) + assert plan is not None + twin = work.group_by(["count"], maintain_order=True).agg(pl.len().alias("n")).collect() + got = plan.collect() + assert got.schema == twin.schema + assert sorted(got.rows()) == sorted(twin.rows()) + + +@pytest.mark.parametrize("values,dtype", [ + (["a", None, "b", None, "a"], "String"), + ([None, None], "String"), + ([], "String"), + ([1, 2, 1, 3], "Int64"), + ([True, False, True, None], "Boolean"), + ([1.0, float("nan"), float("nan"), None, 1.0], "Float64"), +]) +def test_gate_unit_matches_the_group_by_twin_on_awkward_keys( + values: List[Any], dtype: str +) -> None: + """Nulls, all-null, empty input, NaN-as-its-own-group and boolean keys: the two + formulations must agree on rows AND on schema, or the routing decision would be a + semantic one.""" + pl = _require_polars() + work = pl.DataFrame({"city": values}, schema={"city": getattr(pl, dtype)}).lazy() + plan = gfql_fast_paths_module._low_cardinality_pure_count_plan( + work, + node_col="id", + group_keys=["city"], + agg_specs=[("n", "count", None)], + needed_by_alias={"p": [], "c": [("city", "city")]}, + frames_by_alias={ + "p": pl.DataFrame({"id": [1]}), + "c": pl.DataFrame({"id": [10, 11], "city": ["LA", "NY"]}), + }, + edge_rows=len(values), + ) + assert plan is not None + twin = work.group_by(["city"], maintain_order=True).agg(pl.len().alias("n")).collect() + got = plan.collect() + assert got.schema == twin.schema + + def canon(rows: List[Tuple[Any, ...]]) -> List[Tuple[Any, ...]]: + """NaN is its own group in BOTH formulations, but ``nan != nan`` would make the + comparison below fail on agreement. Normalize it to a sentinel first.""" + normalized = [ + (("__nan__" if isinstance(r[0], float) and r[0] != r[0] else r[0]), r[1]) + for r in rows + ] + return sorted(normalized, key=lambda r: (r[0] is None, str(r[0]), r[1])) + + assert canon(got.rows()) == canon(twin.rows())