diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a8680673d..81bad70bc2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - **GFQL execution context is declared rather than attached at runtime**: the private `_gfql_*` per-execution fields (index policy and registry, row-pipeline base graph, carried seed nodes, edge aliases, shortest-path backend, and the indexed-bindings handoff) are now declared on `Plottable` with defaults on `PlotterBase`, instead of being set with `setattr` and read back with `getattr(..., default)`. No public API or behaviour change; internal call sites are typed, and hand-rolled `Plottable` stand-ins must now construct the full context. ### Fixed +- **`sum()`/`avg()` over a non-numeric column now raise the same typed error on every engine — polars no longer answers `avg()` with a silent `null`, and pandas no longer answers `sum()` with the string CONCATENATION**: the aggregate kernels are written four times (pandas/cuDF row pipeline, native polars row pipeline, the OLAP single-hop grouped fast path, and the fused lazy lane #1823 added in front of it) and each had inherited its host library's opinion about non-numeric input, so the SAME query answered differently depending on the engine: `avg(n.name)` raised on pandas but returned `null` on polars, while `sum(n.name)` returned `'abac'` on pandas and leaked a raw `polars.exceptions.InvalidOperationError` through the GFQL surface. "Match the other engine" was not available, because the two engines were wrong in OPPOSITE directions — so the contract is pinned to Cypher instead. openCypher/Neo4j declares `avg(input)` and `sum(input)` over `INTEGER | FLOAT | DURATION` only (neo4j/docs-cypher `functions/aggregating.adoc`) and enforces it: Neo4j 5.26.26 answers `RETURN avg(r.s)` over strings with *"AVG(...) can only handle numerical values, duration, or null."* and `sum(date(...))` with *"Type mismatch: expected Float, Integer or Duration but was Date"*; Kuzu 0.11.3 rejects both at bind time. A non-numeric input is therefore a **`GFQLTypeError` (E302) naming the aggregate and the user's output alias**, on every engine and on both the row pipeline and the fast path. **This is a deliberate contract change on the pandas engine**: `sum()` used to return the concatenation, which is a silent wrong answer with no meaning in Cypher, and it was already inconsistent with `avg()`, which raised. Also brought to the same contract, all found by a full aggregate × dtype differential sweep rather than one at a time: `sum` over temporal and categorical columns (polars returned `null`, pandas raised), `sum`/`avg` over an all-null column (now `0` / `null` per Neo4j's *"`sum(null)` returns `0`"*, substituted rather than delegated — the host kernels answer that case `0`/`''`/`NaT`/`TypeError` depending on dtype, and polars raised), and `min`/`max`/`collect`/`count(DISTINCT)` over a **categorical** column, which Cypher accepts as `ANY` but which pandas rejected outright and cuDF answered with a *category label instead of a count*. A 180-cell polars × cuDF × polars-gpu matrix over 6 aggregates × 10 dtype columns (`_MATRIX_AGGS` × `_MATRIX_COLS`, each compared against the pandas oracle on BOTH the value and the error class) now shows **zero divergences**. Error-path only: the guards read a dtype and do not touch value computation on any served shape. +- **A raw `polars.exceptions.*` could reach the caller from the native polars row pipeline**: on the pandas/cuDF surface `execute_call` wraps any kernel exception as `GFQLTypeError(E303)`, but the native polars path runs BEFORE `execute_call` and so skipped that wrapper entirely — `polars.exceptions.InvalidOperationError: `sum` operation not supported for dtype `str`` was reaching users verbatim, third-party class and all. Native row ops now funnel polars errors through the same wrapper with the same code and message shape, preserving the polars text as the exception cause. - **Three BOUNDED variable-length shapes returned a silently different count on `engine='polars'` (#1787)**: the native polars `rows(binding_ops=...)` builder rebuilds a variable-length segment from the raw matching edge table, and for three bounded shapes that rebuild produces a different edge multiplicity than the pandas oracle — with no error, just a different number. They now raise `NotImplementedError` like every other shape this lowering cannot reproduce, which is a deliberate behaviour change: these were *served* before, so a silent wrong answer becomes a loud, actionable error, and `engine='pandas'` still answers all of them. The three: (1) directed `-[*k..m]->` with `min_hops >= 3`, and `min_hops >= 2` when the segment starts from a filtered seed — `max_reached_hop` in `compute/hop.py` is a dedup-by-node BFS eccentricity rather than a longest-walk length, so the oracle prunes to empty where the rebuild expands a different edge multiset; (2) the DEGENERATE undirected window `-[*1..1]-` / `-[*1]-`, which halved the count — it resolves to `min == max == 1` and so is not "multihop", yet pandas still routes it through the variable-length hop, which is exactly why it slipped past the existing gates (the gate therefore keys on an explicit variable-length window, not on `is_multihop`); (3) undirected `-[*1..k]-` that does not start from the full node set (filtered seed, or a non-first segment), where the doubled-pair expansion over-counts. Every neighbouring shape stays native and is pinned as such — unseeded `min_hops <= 2`, the directed twin of each undirected decline, the directed degenerate window, and the plain `-[]-` edge — so the gate is a scalpel rather than a blanket refusal of variable-length segments. Same root-cause family as the unbounded shapes #1781 declined; the gate should shrink again once the multiplicity is reconstructible. Found by differential fuzzing against the pandas oracle, and pinned by an engine-parametrized suite (pandas / polars / cuDF / polars-gpu) that encodes the intended per-engine behaviour — the polars engines must decline exactly where the pandas-API engines must answer — plus a seeded fuzz that fails if the gate declines too much. - **`rows(table='edges')` silently returned the wrong table after a NAMED traversal**: the named-middle rewrite — which turns `[...named ops..., rows()]` into `rows(binding_ops=...)` so a Cypher multi-alias `RETURN` lowers to a bindings table — skipped itself when the call already carried `binding_ops`, `source` or `alias_endpoints`, but not when it carried a non-default `table`. So naming any op in the middle changed which table came back: an odd-length middle silently produced the BINDINGS table instead of the requested edges (no error, `table=` simply ignored), and an even-length middle — a path ending on an edge, e.g. `(person)-[r:KNOWS]-` — produced a non-alternating op list and raised "require ... a single connected alternating node/edge path". An unnamed middle was unaffected, which is what made it look shape-specific rather than naming-specific. Both chain surfaces are fixed (the generic chain and the native polars chain carry the rewrite independently, so fixing one left the other wrong). Note the guard keys on a NON-DEFAULT table: `rows()` declares `table='nodes'` and always emits it, so an explicit `rows(table='nodes')` is indistinguishable from a bare `rows()` and still rewrites — pinned as a known limitation rather than left to be rediscovered. - **Indexed bindings bypass returned the WHOLE edge table for `rows(table='edges')`**: the bypass exists to SKIP the canonical traversal, so whatever it hands the suffix is the pre-traversal graph. That is sound for a bindings table — the indexed path bag already is the answer — but the gate declined only for `source` / `alias_endpoints` / `alias_prefilters`, not for a non-default `table`, so a `rows(table='edges')` after a named middle read the full edge frame instead of the traversal-narrowed one (measured on a 12-edge fixture: 12 rows instead of 3, on pandas, cuDF and native polars, and the wrong values survive a following `select`). Both gates now decline a non-default `table` (`chain._plan_indexed_middle` and the native polars `_try_indexed_middle_polars`), matching the named-middle rewrite's guard. This became reachable only once that rewrite stopped firing for a non-default `table` — before it, the rewrite converted the call on BOTH the indexed and the scan path, so the two agreed on the wrong table. Same class as the earlier "indexed bypass could return every node for `rows()` over an unnamed pattern". diff --git a/bin/test-polars.sh b/bin/test-polars.sh index d05a63e684..d765fff687 100755 --- a/bin/test-polars.sh +++ b/bin/test-polars.sh @@ -34,6 +34,9 @@ POLARS_TEST_FILES=( # only ever run here (the file has no module-level importorskip, so nothing else flags it) graphistry/tests/compute/gfql/test_exec_context_scoping.py graphistry/tests/compute/gfql/test_engine_polars_cypher_conformance.py + # aggregate x dtype type contract: the polars params of this file are the ONLY lane where the + # native polars aggregate guard and the raw-polars-exception wrap are exercised + graphistry/tests/compute/gfql/test_aggregate_type_contract.py graphistry/tests/compute/gfql/test_engine_polars_conformance_matrix.py graphistry/tests/compute/gfql/test_polars_string_predicate_nonstring.py graphistry/tests/compute/gfql/cypher/test_order_by_null_placement.py diff --git a/docs/source/gfql/spec/cypher_mapping.md b/docs/source/gfql/spec/cypher_mapping.md index 754611ab5c..c85b2408a5 100644 --- a/docs/source/gfql/spec/cypher_mapping.md +++ b/docs/source/gfql/spec/cypher_mapping.md @@ -437,10 +437,25 @@ analysis = g.gfql([ | `LIMIT 10` | `limit(10)` | Row cap | | `WHERE ` | `where_rows(expr="...")` | Scalar expression subset | | `count(*)` | `group_by(keys=[...], aggregations=[("cnt", "count")])` | Grouped count | -| `sum(n.val)` | `group_by(..., aggregations=[("total", "sum", "val")])` | Grouped sum | +| `sum(n.val)` | `group_by(..., aggregations=[("total", "sum", "val")])` | Grouped sum; see [Aggregate input types](#aggregate-input-types) | | `collect(n.x)` | `group_by(..., aggregations=[("xs", "collect", "x")])` | Nulls excluded from collection | | Named patterns | `rows(source="alias")` | Scope row table to a named match alias | +### Aggregate input types + +`sum` and `avg` accept `INTEGER`, `FLOAT` and `DURATION`, matching Cypher — **plus `BOOLEAN`, which +is a deliberate GFQL extension.** Neo4j rejects it (*"expected Float, Integer or Duration but was +Boolean"*); GFQL accepts it because summing an indicator column is idiomatic in the dataframe +surface GFQL also serves, and every engine already agrees on the answer. `sum` over a boolean +counts the true values; `avg` gives their fraction. + +Any other input type **raises**. This is stricter than earlier releases, where a string column +returned its *concatenation* on pandas and leaked a raw polars error on polars — wrong in two +different directions. + +Empty and all-null inputs follow Cypher rather than SQL: `sum` returns **0**, `avg` returns +**null**. + ## Key Differences | Feature | Python | Wire Protocol | diff --git a/graphistry/compute/gfql/agg_types.py b/graphistry/compute/gfql/agg_types.py new file mode 100644 index 0000000000..d97cba1d3f --- /dev/null +++ b/graphistry/compute/gfql/agg_types.py @@ -0,0 +1,188 @@ +"""ONE definition of which GFQL aggregates accept which column types, shared by every engine. + +WHY ONE MODULE: the aggregate kernels are written three times (pandas/cuDF row pipeline, native +polars row pipeline, OLAP three-hop fast path). Each was inheriting its host dataframe library's +opinion about non-numeric input, so the SAME query returned a value on one engine and raised on +another -- ``avg()`` raised ``GFQLTypeError`` on pandas but silently returned +``null`` on polars, and ``sum()`` returned the string CONCATENATION on pandas but +leaked a raw ``polars.exceptions.InvalidOperationError`` on polars. Both directions are wrong, +so "match the other engine" was not available: the contract had to be pinned to Cypher. + +THE CONTRACT (openCypher / Neo4j, verified against two independent implementations): + + ``avg(input)`` / ``sum(input)`` accept ``INTEGER | FLOAT | DURATION`` (and ``null``) ONLY. + Neo4j declares exactly that signature for both functions + (neo4j/docs-cypher ``modules/ROOT/pages/functions/aggregating.adoc``: "Returns the average + of a set of ``INTEGER``, ``FLOAT`` or ``DURATION`` values", ``input : INTEGER | FLOAT | + DURATION``; same for ``sum()``), and enforces it at runtime -- Neo4j 5.26.26 answers + ``RETURN avg(r.s)`` over strings with "AVG(...) can only handle numerical values, duration, + or null." and ``RETURN sum(date(...))`` with "Type mismatch: expected Float, Integer or + Duration but was Date". Kuzu 0.11.3 rejects the same two at bind time ("Function AVG did + not receive correct arguments: Actual: (STRING)"). + => a non-numeric column is a TYPE ERROR, never a null and never a concatenation. + + ``min(input)`` / ``max(input)`` / ``count(input)`` / ``collect(input)`` accept ``ANY`` + (same doc: ``input : ANY``; openCypher TCK ``Aggregation2`` scenarios 7-12 cover + ``min()``/``max()`` over strings, lists and mixed values). + => these must NOT raise on strings/categoricals. + + Nulls are excluded from every aggregate; ``sum`` over an empty-or-all-null set is ``0`` and + ``avg`` over one is ``null`` (Neo4j "Considerations": "``sum(null)`` returns ``0``", + "``avg(null)`` returns ``null``"; confirmed live on 5.26.26). + => an all-null column carries no type evidence and must NOT be rejected. + +DELIBERATE GFQL EXTENSION, not an oversight: ``sum``/``avg`` over BOOLEAN is a type error in +Neo4j ("expected Float, Integer or Duration but was Boolean") but is accepted here on every +engine, because summing an indicator column is idiomatic in the dataframe surface GFQL also +serves and both engines already agreed on it. It is recorded here so the divergence is a choice +with a reason rather than an accident. + +Each engine classifies its OWN dtypes (a pandas dtype and a polars ``DataType`` are not +comparable) and then funnels into the one raiser below, so the diagnostic text, the error class +and the ``ErrorCode`` cannot drift between engines. +""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Final, FrozenSet, NoReturn, Optional + +from graphistry.compute.exceptions import ErrorCode, GFQLTypeError + +if TYPE_CHECKING: + import polars as pl + + from graphistry.compute.typing import SeriesT + + +#: Aggregates Cypher restricts to ``INTEGER | FLOAT | DURATION``. ``mean`` is GFQL's internal +#: spelling of ``avg`` (see ``GFQL_GROUPBY_AGG_METHODS``), so both names must gate. +GFQL_NUMERIC_ONLY_AGGREGATIONS: Final[FrozenSet[str]] = frozenset({"sum", "avg", "mean"}) + + +def _describe_agg_input(column: str, alias: Optional[str]) -> str: + """How to point the user at the offending value in THEIR query text. + + The cypher lowering materializes ``avg(n.score)``'s argument into an internal + ``__cypher_agg__`` column before the group-by, so naming the raw column would hand the user + a name that appears nowhere in their query. When the column is one of those temporaries, name + the aggregate's output alias instead -- that one they wrote (``... AS score_avg``). + """ + if alias and column.startswith("__") and column.endswith("__"): + return f"the argument of {alias!r}" + return f"column {column!r}" + + +def raise_non_numeric_aggregation( + func: str, column: str, dtype: str, alias: Optional[str] = None +) -> NoReturn: + """The single diagnostic for "this aggregate needs numbers and this input has none". + + Names the OPERATION and the INPUT (and the offending type): an aggregate's output alias is + usually not its input's column name, so "type error" alone leaves the user grepping. + """ + target = _describe_agg_input(column, alias) + raise GFQLTypeError( + ErrorCode.E302, + f"Aggregation {func}() requires numeric or duration values, " + f"but {target} has type {dtype}", + field=column, + value=dtype, + suggestion=( + f"Cypher restricts {func}() to INTEGER/FLOAT/DURATION; " + f"use count()/collect()/min()/max() over {target}, or cast it to a number" + ), + ) + + +def numeric_agg_all_null_value(func: str) -> Optional[int]: + """The Cypher answer for a numeric-only aggregate over an ALL-NULL input: ``0`` / ``null``. + + Callers apply this BEFORE the dtype check, because an all-null column carries no type + evidence and so can never be a type error. It also has to bypass the host kernels entirely: + pandas answers an all-null column with ``0``/``NaN`` when it is ``object`` but with ``''``, + ``NaT`` or a ``TypeError`` once it is typed (``string``/``category``/``datetime64``), and + polars raises for both ``str`` and ``null`` dtypes. One substitution, one answer. + """ + return 0 if func == "sum" else None + + +def pandas_dtype_is_numeric_for_agg(series: "SeriesT") -> bool: + """True when the pandas/cuDF dtype ITSELF proves the column is a valid sum/avg input. + + The cheap, O(1), hot-path answer: a column that passes here needs no data inspection at all + -- no null scan, no value sampling -- so an ordinary numeric aggregate pays a string check + and nothing else. Everything that fails here (object, string, categorical, temporal) is + already headed for a slow or erroring path, which is where the O(n) questions get asked. + ``timedelta`` is Cypher's ``DURATION``; ``bool`` is the documented GFQL extension. + """ + dtype_txt = str(getattr(series, "dtype", "")).lower() + if "interval" in dtype_txt or "datetime" in dtype_txt or "period" in dtype_txt: + return False + if "timedelta" in dtype_txt or "duration" in dtype_txt: + return True + if dtype_txt in {"bool", "boolean"}: + return True + return any(token in dtype_txt for token in ("int", "float", "double", "decimal")) + + +def pandas_non_numeric_agg_dtype(series: "SeriesT") -> Optional[str]: + """Dtype label if this pandas/cuDF column must be REJECTED by ``sum``/``avg``, else ``None``. + + Deliberately a DENY list keyed on positively-identified non-numeric types (string dtype, + categorical, datetime/date) rather than an allow list of numerics: pandas ``object`` columns + routinely carry numbers through the cypher property path, and demanding positive numeric + proof would start rejecting queries that compute correctly today. ``timedelta64`` is Cypher's + ``DURATION`` and is allowed. All-null columns are the callers' job + (:func:`numeric_agg_all_null_value`), applied before this. + """ + dtype = getattr(series, "dtype", None) + dtype_txt = str(dtype).lower() + if "datetime" in dtype_txt or dtype_txt in {"date32[day][pyarrow]", "date64[ms][pyarrow]"}: + return str(dtype) + if "category" in dtype_txt: + return str(dtype) + # Prefix match, not a fixed set: pandas spells its string dtype differently across versions and + # storages -- `object` on pandas 2, `str` by default on pandas 3, plus `string`, + # `string[pyarrow]`, `string[python]`, `str[pyarrow]`. A missed spelling here fails OPEN + # (delegates to the kernel, restoring the concatenation), so the check is deliberately broad. + if dtype_txt.startswith("str") or dtype_txt.startswith("large_string"): + return str(dtype) + if dtype_txt == "object" and _object_series_is_str_like(series): + return "object (strings)" + return None + + +def _object_series_is_str_like(series: "SeriesT") -> bool: + """Every non-null value in a bounded head sample is a ``str`` (and there is at least one). + + Bounded so the check costs the same on a 10-row and a 10M-row frame, and empty-after-dropna + returns False so all-null columns fall through to the ``sum(null) == 0`` contract. + """ + if not hasattr(series, "dropna"): + return isinstance(series, str) + from graphistry.Engine import series_to_pylist + values = series_to_pylist(series.dropna().head(128)) + return len(values) > 0 and all(isinstance(v, str) for v in values) + + +def polars_non_numeric_agg_dtype(dtype: "Optional[pl.DataType]") -> Optional[str]: + """Dtype label if this polars column must be REJECTED by ``sum``/``avg``, else ``None``. + + Mirrors :func:`pandas_non_numeric_agg_dtype` as a DENY list over the same type families so + the two engines cannot classify one column differently: String/Categorical/Enum, temporal + Date/Datetime/Time, and the composite dtypes (List/Array/Struct/Binary/Object). ``Duration`` + is Cypher's ``DURATION`` and is allowed; ``Null`` never reaches here because an all-null + column is short-circuited by :func:`numeric_agg_all_null_value` first. + """ + import polars as pl + + from .lazy.engine.polars.dtypes import is_stringlike + + if dtype is None: + return None + if is_stringlike(dtype): + return str(dtype) + for name in ("Date", "Datetime", "Time", "List", "Array", "Struct", "Binary", "Object"): + candidate = getattr(pl, name, None) + if candidate is not None and (dtype == candidate or isinstance(dtype, candidate)): + return str(dtype) + return None diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index 17466b2ea1..521bd27533 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -7,7 +7,7 @@ (no silent pandas fallback). Deferred: variable-length/multi-hop edge sub-cases, some undirected multi-edge combos, node query=. """ -from typing import TYPE_CHECKING, Any, List, Optional, Tuple, cast +from typing import TYPE_CHECKING, Any, List, Optional, Tuple, Type, cast from typing_extensions import TypedDict @@ -29,6 +29,21 @@ from .reserved_columns import CHAIN_NODE_HOP +def _polars_error_types() -> Tuple[Type[BaseException], ...]: + """The polars exception hierarchy root, as an ``except`` target. + + A tuple (not the class) so the empty tuple is available as the fail-closed answer when an + older polars has no ``PolarsError`` base: ``except ()`` matches nothing, which degrades to + today's behaviour rather than swallowing something unrelated. Evaluated only while an + exception is being matched, so it costs nothing on the success path. + """ + import polars as pl + base = getattr(pl.exceptions, "PolarsError", None) + if isinstance(base, type) and issubclass(base, BaseException): + return (base,) + return () + + def _semi(df: "PolarsT", ids_df: "PolarsT", df_col: str, id_col: str) -> "PolarsT": """Rows of df whose df_col is present in ids_df[id_col] (vectorized semi-join). @@ -571,6 +586,20 @@ def _run_calls_polars(g_cur, calls, start_nodes, base_graph, middle): field="function", value=fn_name, ) from validation_error + except _polars_error_types() as polars_error: + # A THIRD-PARTY exception must never be the GFQL surface. On the pandas/cuDF side + # `execute_call` already wraps any kernel exception as GFQLTypeError(E303) — the + # native polars path runs BEFORE execute_call and so skipped that wrapper entirely, + # letting e.g. `polars.exceptions.InvalidOperationError: \`sum\` operation not + # supported for dtype \`str\`` reach the caller verbatim. Same code, same message + # shape as the pandas surface; the polars text is preserved as the cause. + fn_name = getattr(op, "function", None) + raise GFQLTypeError( + ErrorCode.E303, + f"Error executing '{fn_name}': {polars_error}", + field="function", + value=fn_name, + ) from polars_error if native is not None: g_cur = native continue diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index e4e45f0910..08414c1913 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -33,6 +33,12 @@ # Engine-neutral wire-format payload types (ASTCall.params). Shapes are safelist-validated # (gfql/call/validation.py) before reaching these helpers, so the runtime isinstance/len # checks below are defense-in-depth, not the contract. +from graphistry.compute.gfql.agg_types import ( + GFQL_NUMERIC_ONLY_AGGREGATIONS, + numeric_agg_all_null_value, + polars_non_numeric_agg_dtype, + raise_non_numeric_aggregation, +) from graphistry.compute.gfql.call.support import AggSpec, OrderKey, SelectItem from .dtypes import is_float as _dtype_is_float, is_int as _dtype_is_int, is_numeric as _dtype_is_numeric, is_stringlike as _dtype_is_stringlike # Same-package sibling holding the var-length specializations. Safe at module scope: @@ -854,7 +860,9 @@ def order_by_polars(g: Plottable, keys: Sequence[OrderKey]) -> Optional[Plottabl # Native aggs: count/sum/avg/min/max/count_distinct/collect/collect_distinct; stdev/percentile # etc. return None → caller declines (NIE). -def _agg_expr(func: str, expr: Optional[str], columns: Sequence[str], alias: str, schema: Optional[Mapping[str, "pl.DataType"]] = None) -> Optional[pl.Expr]: +def _agg_expr(func: str, expr: Optional[str], columns: Sequence[str], alias: str, + schema: Optional[Mapping[str, "pl.DataType"]] = None, + is_all_null: Optional[Callable[[str], bool]] = None) -> Optional[pl.Expr]: import polars as pl func = func.lower() if func == "count" and (expr is None or expr == "*"): @@ -862,6 +870,15 @@ def _agg_expr(func: str, expr: Optional[str], columns: Sequence[str], alias: str if not isinstance(expr, str) or expr not in columns: return None col = pl.col(expr) + dtype = schema.get(expr) if schema is not None else None + if dtype is not None and _dtype_is_stringlike(dtype) and dtype != pl.String: + # Categorical/Enum is a STRING column to cypher (its categories are the values). Twin of + # the pandas row pipeline's decategorize, and load-bearing on older polars: 1.35.2 (the + # RAPIDS 26.02 image) PANICS in the rust core on a grouped min/max over a Categorical + # (`categorical.rs: not implemented`), which escapes as a pyo3 PanicException -- not even + # a polars exception, so nothing on the python side can wrap it. Casting to String makes + # the aggregate well-defined on every polars version AND matches what pandas returns. + col = col.cast(pl.String) # pandas aggs skip NaN (skipna); polars skips only NULL and treats NaN as a value (NaN == NaN # is True, so self-inequality can't detect it). For FLOAT columns convert in-query NaN -> null # first so every agg matches the oracle (pandas sum([nan, 1]) == 1 vs raw polars == nan). @@ -869,6 +886,26 @@ def _agg_expr(func: str, expr: Optional[str], columns: Sequence[str], alias: str # NaN created mid-query (e.g. 0.0/0.0). if schema is not None and _dtype_is_float(schema.get(expr)): col = col.fill_nan(None) + if func in GFQL_NUMERIC_ONLY_AGGREGATIONS and schema is not None: + # DTYPE FIRST, data second -- deliberately, for cost. Cypher restricts sum()/avg() to + # INTEGER|FLOAT|DURATION (see gfql/agg_types.py for the sources), and that verdict is a + # schema lookup; only a column the SCHEMA already rejects is worth an O(n) null scan. A + # numeric column -- every served aggregate -- therefore pays nothing here. + if dtype == pl.Null: + # all-null by construction: `sum`/`mean` are unsupported on `null` dtype in polars, + # while cypher says 0 / null. + return pl.lit(numeric_agg_all_null_value(func)).alias(alias) + dtype_label = polars_non_numeric_agg_dtype(dtype) + if dtype_label is not None: + # An ALL-NULL column carries no type evidence, so it is never a type error: cypher + # answers `sum(null)` with 0 and `avg(null)` with null whatever the declared type, + # and pandas already did (an all-None pandas object column arrives here typed + # `String`). Both would otherwise raise -- `sum`/`mean` are unsupported on `str`. + if is_all_null is not None and is_all_null(expr): + return pl.lit(numeric_agg_all_null_value(func)).alias(alias) + # Raise, don't return None: None is an NIE-decline that falls back to the pandas + # kernel, which would then ANSWER the same wrong-typed query. + raise_non_numeric_aggregation(func, expr, dtype_label, alias) if func == "count": return col.count().alias(alias) if func == "sum": @@ -932,7 +969,13 @@ def group_by_polars( alias = str(spec[0]) func = str(spec[1]) expr = spec[2] if len(spec) == 3 else None - lowered = _agg_expr(func, expr, cols, alias, table.schema) + # Passed as a CALLABLE, not a precomputed flag: the null scan is O(n) and is only ever + # consulted for a column the dtype check has already rejected, so a normal numeric + # aggregate never runs it. + def _is_all_null(col_name: str) -> bool: + return table.height > 0 and table[col_name].null_count() == table.height + + lowered = _agg_expr(func, expr, cols, alias, table.schema, _is_all_null) if lowered is None: return None aggs.append(lowered) diff --git a/graphistry/compute/gfql/row/pipeline.py b/graphistry/compute/gfql/row/pipeline.py index 5aec80b971..dde90da329 100644 --- a/graphistry/compute/gfql/row/pipeline.py +++ b/graphistry/compute/gfql/row/pipeline.py @@ -30,6 +30,13 @@ is_order_aggregate_alias_ast, order_expr_ast_static_supported, ) +from graphistry.compute.gfql.agg_types import ( + GFQL_NUMERIC_ONLY_AGGREGATIONS, + numeric_agg_all_null_value, + pandas_dtype_is_numeric_for_agg, + pandas_non_numeric_agg_dtype, + raise_non_numeric_aggregation, +) from graphistry.compute.gfql.language_defs import ( GFQL_COMPARISON_BINARY_OP_NAMES, GFQL_COMPARISON_BINARY_OPS, @@ -5116,6 +5123,14 @@ def _build_grouped(group_df: Any) -> Any: tmp_col = f"{tmp_col}_x" table_df = table_df.assign(**{tmp_col: expr_values}) expr_col = tmp_col + if "category" in str(table_df[expr_col].dtype).lower(): + # A categorical column is a STRING column to cypher (its categories are the + # values), but the host kernels disagree per-aggregate: pandas raises on + # grouped min/max/sum/mean over a categorical, and cuDF additionally raises + # on collect and -- worse -- answers count(DISTINCT) with a CATEGORY LABEL + # instead of a count. Decategorize once, up front, so every aggregate below + # sees the string column cypher says it is. + table_df = table_df.assign(**{expr_col: table_df[expr_col].astype("string")}) grouped = _make_grouped(table_df, [expr_col]) if func in {"collect", "collect_distinct"}: # collect() ignores null entries; compute collection on @@ -5153,9 +5168,39 @@ def _build_grouped(group_df: Any) -> Any: method_name = GFQL_GROUPBY_AGG_METHODS.get(func) if method_name is None: raise ValueError(f"unsupported group_by aggregation function: {func!r}") - if ( - func in {"min", "max"} - and RowPipelineMixin._gfql_series_object_non_null_str_like(table_df[expr_col]) + all_null_numeric = False + if func in GFQL_NUMERIC_ONLY_AGGREGATIONS: + # Cypher restricts sum()/avg() to INTEGER|FLOAT|DURATION -- see + # gfql/agg_types.py for the sources. Without this, pandas answered + # sum() with the CONCATENATION ('abac'), a silent wrong + # answer, while avg() raised only incidentally (pandas' + # own TypeError, rewrapped as an E201 "parameter error" naming neither + # the column nor the operation). + # DTYPE FIRST, data second: a column the dtype PROVES numeric is served + # with no data inspection at all, so an ordinary numeric aggregate -- + # every served one -- pays only a dtype string check here. Anything else + # is already on a slow or erroring path, which is where the O(n) + # all-null question gets asked. + if not pandas_dtype_is_numeric_for_agg(table_df[expr_col]): + all_null_numeric = ( + len(table_df) > 0 and bool(table_df[expr_col].isna().all()) + ) + if not all_null_numeric: + dtype_label = pandas_non_numeric_agg_dtype(table_df[expr_col]) + if dtype_label is not None: + raise_non_numeric_aggregation(func, expr_col, dtype_label, alias) + if all_null_numeric: + # cypher sum(null)==0 / avg(null)==null, substituted rather than computed: + # pandas answers an all-null column 0/NaN when object but ''/NaT/TypeError + # once typed, so the kernel cannot be trusted to produce it. + agg_df = out_df[key_cols].copy() + agg_df[alias] = self._gfql_broadcast_scalar( + agg_df, numeric_agg_all_null_value(func) + ) + out_df = out_df.merge(agg_df, on=key_cols, how="left", sort=False) + continue + if func in {"min", "max"} and ( + RowPipelineMixin._gfql_series_object_non_null_str_like(table_df[expr_col]) ): table_df = table_df.assign(**{expr_col: table_df[expr_col].astype("string")}) grouped = _make_grouped(table_df, [expr_col]) diff --git a/graphistry/compute/gfql_fast_paths.py b/graphistry/compute/gfql_fast_paths.py index b6d8e116e7..75afe43171 100644 --- a/graphistry/compute/gfql_fast_paths.py +++ b/graphistry/compute/gfql_fast_paths.py @@ -40,6 +40,14 @@ parse_where_json, ) from graphistry.compute.exceptions import ErrorCode, GFQLValidationError +from graphistry.compute.gfql.agg_types import ( + GFQL_NUMERIC_ONLY_AGGREGATIONS, + numeric_agg_all_null_value, + pandas_dtype_is_numeric_for_agg, + pandas_non_numeric_agg_dtype, + polars_non_numeric_agg_dtype, + raise_non_numeric_aggregation, +) from graphistry.compute.gfql.cypher.parser import parse_cypher from graphistry.compute.gfql.cypher.lowering import ( ConnectedMatchJoinPlan, @@ -1781,6 +1789,11 @@ def _single_hop_grouped_aggregate_fused_polars( * a projected output column that collides with an edge endpoint column or with the internal lookup key; * an aggregate the expression builder cannot translate; + * a Categorical/Enum property column -- the twin CASTS those to ``String`` before it + groups or aggregates, and this lane does not; + * ``sum``/``avg`` over a non-numeric or all-null column -- the twin owns the one cypher + aggregate type contract (``gfql/agg_types.py``), and declining keeps that contract from + depending on which of the two lanes served the query; * **a result row order that ORDER BY does not fully determine** -- i.e. no ORDER BY, or an ORDER BY that does not mention every group key. When the sort is total over the output rows (group keys are unique per row, so naming them all makes it total), @@ -1792,6 +1805,8 @@ def _single_hop_grouped_aggregate_fused_polars( """ import polars as pl + from graphistry.compute.gfql.lazy.engine.polars.dtypes import is_stringlike + if not isinstance(start_nodes, pl.DataFrame) or not isinstance(end_nodes, pl.DataFrame): return None if not isinstance(edges, pl.DataFrame): @@ -1807,15 +1822,36 @@ def _single_hop_grouped_aggregate_fused_polars( end_alias: _GROUPED_AGG_LOOKUP_KEY_FMT.format(alias=end_alias), } reserved_out_cols = {src_col, dst_col} | set(lookup_keys.values()) + prop_dtypes: Dict[str, Any] = {} for alias, node_frame in ((start_alias, start_nodes), (end_alias, end_nodes)): for out_col, prop in needed_by_alias.get(alias, ()): if prop not in node_frame.columns: return None if out_col in reserved_out_cols: return None + prop_dtype = node_frame.schema[prop] + # Categorical/Enum is a STRING column to cypher, and the eager twin CASTS it to + # `String` before aggregating or grouping -- both to match the row pipelines and to + # dodge the polars 1.35.2 rust panic on a grouped min/max over a Categorical. This + # lane does not cast, so it DECLINES rather than answer with a different dtype. + if is_stringlike(prop_dtype) and prop_dtype != pl.String: + return None + prop_dtypes[out_col] = prop_dtype agg_exprs: List["pl.Expr"] = [] for out_alias, func, expr_col in agg_specs: + # ONE cypher type contract for sum()/avg() whichever lane serves the query (see + # gfql/agg_types.py). A non-numeric or all-null input is the eager twin's job -- it + # raises GFQLTypeError(E302), or substitutes cypher's `0`/`null` for an all-null column + # -- so decline instead of letting `.collect()` surface a raw `polars.exceptions.*`. + if func in GFQL_NUMERIC_ONLY_AGGREGATIONS and expr_col is not None: + agg_dtype = prop_dtypes.get(expr_col) + if ( + agg_dtype is None + or agg_dtype == pl.Null + or polars_non_numeric_agg_dtype(agg_dtype) is not None + ): + return None if func == "count" and expr_col is None: agg_exprs.append(pl.len().alias(out_alias)) elif expr_col is None: @@ -2077,7 +2113,33 @@ def join_props_polars(work_df: Any, alias: str, node_df: Any, edge_col: str) -> if work is None: return None agg_exprs = [] + # Categorical/Enum is a STRING column to cypher; cast it BEFORE any aggregate, matching + # the row pipelines on both engines (and dodging the polars 1.35.2 rust panic on a + # grouped min/max over a Categorical). Done on the frame, so `pl.col(...)` below is + # already the string column. + from graphistry.compute.gfql.lazy.engine.polars.dtypes import is_stringlike + cat_cols = [c for c, dt in work.schema.items() if is_stringlike(dt) and dt != pl.String] + if cat_cols: + work = work.with_columns([pl.col(c).cast(pl.String) for c in cat_cols]) + work_schema = work.schema for alias, func, expr_alias in aggregations: + # Same Cypher sum()/avg() type contract the row-pipeline kernels enforce (see + # gfql/agg_types.py). This fast path reimplements the aggregates, so without the + # guard here the exact same query answers differently depending on whether it + # matched the fast-path shape. + if func in GFQL_NUMERIC_ONLY_AGGREGATIONS and expr_alias is not None: + # Dtype first, data second -- see the row-pipeline twin: only a column the schema + # already rejects is worth the O(n) null scan. + dtype_label = polars_non_numeric_agg_dtype(work_schema.get(expr_alias)) + if work_schema.get(expr_alias) == pl.Null or ( + dtype_label is not None + and work.height > 0 + and work[expr_alias].null_count() == work.height + ): + agg_exprs.append(pl.lit(numeric_agg_all_null_value(func)).alias(alias)) + continue + if dtype_label is not None: + raise_non_numeric_aggregation(func, expr_alias, dtype_label, alias) if func == "count" and (expr_alias is None or with_items[expr_alias][1] is None): agg_exprs.append(pl.len().alias(alias)) elif func == "count" and expr_alias is not None: @@ -2141,6 +2203,18 @@ def join_props_df(work_df: DataFrameT, alias: str, node_df: DataFrameT, edge_col grouped = work.groupby(group_keys, sort=False) out_df = grouped.size().reset_index(name="__gfql_group_size__")[group_keys] for alias, func, expr_alias in aggregations: + # Twin of the polars branch above: one Cypher sum()/avg() type contract, enforced on + # whichever engine runs the fast path (see gfql/agg_types.py). + if func in GFQL_NUMERIC_ONLY_AGGREGATIONS and expr_alias is not None: + if not pandas_dtype_is_numeric_for_agg(work[expr_alias]): + if len(work) > 0 and bool(work[expr_alias].isna().all()): + agg_df = out_df[group_keys].copy() + agg_df[alias] = numeric_agg_all_null_value(func) + out_df = cast(DataFrameT, out_df.merge(agg_df, on=group_keys, how="left", sort=False)) + continue + dtype_label = pandas_non_numeric_agg_dtype(work[expr_alias]) + if dtype_label is not None: + raise_non_numeric_aggregation(func, expr_alias, dtype_label, alias) if func == "count" and (expr_alias is None or with_items[expr_alias][1] is None): agg_df = grouped.size().reset_index(name=alias) elif func == "count" and expr_alias is not None: diff --git a/graphistry/tests/compute/gfql/test_aggregate_type_contract.py b/graphistry/tests/compute/gfql/test_aggregate_type_contract.py new file mode 100644 index 0000000000..e3016e8148 --- /dev/null +++ b/graphistry/tests/compute/gfql/test_aggregate_type_contract.py @@ -0,0 +1,523 @@ +"""The Cypher type contract for GFQL aggregates, enforced identically on every engine. + +WHY THIS FILE EXISTS: the aggregate kernels are written three times (pandas/cuDF row pipeline, +native polars row pipeline, OLAP three-hop fast path) and each had inherited its host library's +opinion about non-numeric input, so the SAME query answered differently per engine: + + ``avg()`` pandas raised, polars returned a silent ``null`` + ``sum()`` pandas returned the CONCATENATION ('abac'), + polars leaked a raw ``polars.exceptions.InvalidOperationError`` + +Both directions are wrong, so the fix could not be "match the other engine" -- the contract is +pinned to Cypher (see ``graphistry/compute/gfql/agg_types.py`` for the Neo4j 5.26.26 and Kuzu +0.11.3 receipts). This module is the executable form of that contract: a positive lane (what MUST +compute), a negative lane (what MUST raise, with which typed error), and a differential matrix +over aggregate x dtype that fails on ANY pandas-vs-other-engine disagreement. + +ENGINE COVERAGE: parametrized over pandas / polars / cudf / polars-gpu. The GPU engines SKIP with +an explicit reason when their stack is absent -- run the GPU lane on a GPU box +(``graphistry/test-rapids-official:26.02-gfql-polars`` with ``--gpus all``) to close them; a +skipped GPU param states its own boundary in the pytest report rather than passing quietly. +Deliberately does NOT use ``available_nonpandas_engines()``: that helper SHRINKS the parametrization +silently when a stack is missing, so a lane can vanish without any signal. +""" +import datetime + +import pandas as pd +import pytest + +import graphistry +from graphistry.compute.exceptions import ErrorCode, GFQLTypeError + +pl = pytest.importorskip("polars") + + +ALL_ENGINES = ["pandas", "polars", "cudf", "polars-gpu"] + + +def _require_engine(engine: str) -> None: + """Skip with a NAMED reason so an absent GPU stack is visible in the report, not silent.""" + if engine == "cudf": + pytest.importorskip("cudf", reason="cudf engine lane requires a GPU box (--gpus all)") + if engine == "polars-gpu": + pytest.importorskip("cudf", reason="polars-gpu lane requires a GPU box (--gpus all)") + import importlib.util + if importlib.util.find_spec("cudf_polars") is None: + pytest.skip("polars-gpu lane requires cudf_polars (RAPIDS 26.02+ image)") + + +# Every column here is exercised by BOTH the positive and the negative lane; which side a column +# lands on is exactly the contract under test. +_STR = ["a", "b", "a", "c", "b", "d"] +# Column names are deliberately multi-letter and distinct from the pattern aliases used below +# (a / b / e) and from the edge endpoint columns: a node column named `b` shadows the `(b)` alias +# in `MATCH (a)-[e]->(b)` and silently aggregates the wrong values. +_NUMERIC_COLS = { + "int_col": [1, 2, 3, 4, 5, 6], + "float_col": [1.5, 2.5, 3.5, 4.5, 5.5, 6.5], + "bool_col": [True, False, True, True, False, False], + "dur_col": pd.to_timedelta([1, 2, 3, 4, 5, 6], unit="D"), + "nullint_col": [1, None, 3, None, 5, 6], +} +_NON_NUMERIC_COLS = { + "str_col": _STR, + "nullstr_col": ["a", None, "a", None, "b", None], + "cat_col": pd.Categorical(_STR), + "date_col": pd.to_datetime([f"2020-01-0{i}" for i in range(1, 7)]), +} +_ALL_NULL_COLS = {"allnull_col": [None] * 6} + + +def _graph(): + data = {"id": list(range(6)), "grp": ["x", "x", "x", "y", "y", "y"]} + data.update(_NUMERIC_COLS) + data.update(_NON_NUMERIC_COLS) + data.update(_ALL_NULL_COLS) + nodes = pd.DataFrame(data) + edges = pd.DataFrame({"src": [0, 1, 2, 3], "dst": [1, 2, 3, 4]}) + return graphistry.nodes(nodes, "id").edges(edges, "src", "dst") + + +def _cells(df): + """Engine-neutral, dtype-neutral cell values: [(col, value), ...] rows, sorted. + + Values are normalized (numpy/polars scalars -> python, NaN/NaT -> None, temporal -> iso, + list-likes -> list) so the matrix compares SEMANTICS. Numeric repr width (int64 vs uint32) is + a separate, already-decided concern and must not masquerade as a contract violation here. + """ + import numpy as np + + def norm(v): + if isinstance(v, np.generic): + v = v.item() + if isinstance(v, (pd.Timestamp, datetime.date, datetime.datetime, np.datetime64)): + return None if pd.isna(v) else str(pd.Timestamp(v)) + if isinstance(v, (pd.Timedelta, datetime.timedelta, np.timedelta64)): + # nanoseconds, NOT str(): pandas renders a Timedelta "4 days 00:00:00" and polars + # hands back a python timedelta whose str() is "4 days, 0:00:00" -- a repr gap that + # would read as a value divergence. + return None if pd.isna(v) else int(pd.Timedelta(v).value) + if isinstance(v, str): + return v + if isinstance(v, (list, tuple, np.ndarray, pd.Series)): + return [norm(x) for x in list(v)] + if v is None: + return None + if isinstance(v, bool): + return bool(v) # NOT folded into the numeric tag below: True must never equal 1 + if isinstance(v, (int, float)): + # One tag for int and float: whether an aggregate lands as int64 or float64 is the + # separate, already-decided nullable-merge dtype question (#1796 BU1), and letting it + # fail here would bury the type-contract signal this matrix exists to carry. + return None if v != v else ("num", round(float(v), 9)) + return None if pd.isna(v) else v + + if df is None: + return None + if "polars" in type(df).__module__: + cols, records = df.columns, df.to_dicts() + else: + if hasattr(df, "to_pandas"): # cudf + df = df.to_pandas() + cols, records = list(df.columns), [row for _, row in df.iterrows()] + return sorted(tuple((c, norm(row[c])) for c in cols) for row in records) + + +def _run(g, query, engine): + """('ok', cells) | ('raise', ExcTypeName). NOT a try/except that hides failures: the class + name IS the assertion payload, so an engine swapping a value for an error still fails.""" + try: + return ("ok", _cells(g.gfql(query, engine=engine)._nodes)) + except Exception as exc: # noqa: BLE001 - the exception TYPE is the thing under test + return ("raise", type(exc).__module__.split(".")[0] + "." + type(exc).__name__) + + +# -------------------------------------------------------------------------------------- +# NEGATIVE lane: what MUST raise, and with which typed error +# -------------------------------------------------------------------------------------- + +@pytest.mark.parametrize("engine", ALL_ENGINES) +@pytest.mark.parametrize("func", ["sum", "avg"]) +@pytest.mark.parametrize("col", sorted(_NON_NUMERIC_COLS)) +@pytest.mark.parametrize("grouped", [True, False]) +def test_numeric_only_aggregate_over_non_numeric_column_raises(engine, func, col, grouped): + """Neo4j: "SUM(...) can only handle numerical values, duration, or null." -- a GFQLTypeError, + never a null (polars' old answer for avg) and never a concatenation (pandas' old answer for + sum), on every engine and both the grouped and the whole-table shape.""" + _require_engine(engine) + g = _graph() + query = (f"MATCH (n) RETURN n.grp AS grp, {func}(n.{col}) AS agg_out ORDER BY grp" if grouped + else f"MATCH (n) RETURN {func}(n.{col}) AS agg_out") + with pytest.raises(GFQLTypeError) as excinfo: + g.gfql(query, engine=engine) + assert excinfo.value.code == ErrorCode.E302 + message = str(excinfo.value) + assert f"{func}()" in message, message # names the OPERATION + assert "agg_out" in message, message # names the user's alias + assert "numeric or duration" in message, message + + +@pytest.mark.parametrize("engine", ["polars", "polars-gpu"]) +def test_no_raw_polars_exception_reaches_the_gfql_surface(engine): + """``sum()`` used to surface ``polars.exceptions.InvalidOperationError`` + verbatim. A third-party exception class is never the GFQL surface -- the pandas side has + always been wrapped by execute_call, and the native polars path must match.""" + _require_engine(engine) + with pytest.raises(GFQLTypeError) as excinfo: + _graph().gfql("MATCH (n) RETURN sum(n.str_col) AS agg_out", engine=engine) + assert "polars" not in type(excinfo.value).__module__ + + +@pytest.mark.parametrize("engine", ["polars", "polars-gpu"]) +def test_native_polars_row_op_wraps_any_polars_error(engine, monkeypatch): + """The choke point itself, independent of the dtype guard above: ANY polars error raised + inside a native row op is rewrapped as GFQLTypeError(E303) -- the same code/message shape + execute_call gives the pandas surface -- instead of escaping as polars.exceptions.*. + + Injected rather than provoked on purpose: the dtype guard now prevents the one polars error + this surface was known to raise, so a "find a query that still breaks polars" test would be + testing today's gap list rather than the invariant. The invariant is that the NEXT such error + is wrapped too.""" + _require_engine(engine) + import graphistry.compute.gfql.lazy.engine.polars.chain as polars_chain + + def boom(*_args, **_kwargs): + raise pl.exceptions.ComputeError("injected polars failure") + + monkeypatch.setattr(polars_chain, "_try_native_row_op", boom) + with pytest.raises(GFQLTypeError) as excinfo: + _graph().gfql("MATCH (n) RETURN n.grp AS grp, count(n.int_col) AS agg_out", engine=engine) + assert excinfo.value.code == ErrorCode.E303 + assert "injected polars failure" in str(excinfo.value) + assert "polars" not in type(excinfo.value).__module__ + + +def test_polars_error_base_is_resolvable(): + """``_polars_error_types()`` returning () would make the wrap above match nothing and fail + open -- silently restoring the raw-exception leak.""" + from graphistry.compute.gfql.lazy.engine.polars.chain import _polars_error_types + types = _polars_error_types() + assert types, "polars error base not found -- the except clause would match nothing" + assert issubclass(pl.exceptions.InvalidOperationError, types[0]) + + +# -------------------------------------------------------------------------------------- +# POSITIVE lane: what MUST still compute +# -------------------------------------------------------------------------------------- + +@pytest.mark.parametrize("engine", ALL_ENGINES) +@pytest.mark.parametrize("func", ["sum", "avg"]) +@pytest.mark.parametrize("col", sorted(_NUMERIC_COLS)) +def test_numeric_only_aggregate_over_numeric_column_computes(engine, func, col): + """The guard must not become a blanket rejection: INTEGER / FLOAT / DURATION all aggregate, + and so does BOOLEAN -- a deliberate GFQL extension over Cypher (see agg_types.py).""" + _require_engine(engine) + g = _graph() + got = _run(g, f"MATCH (n) RETURN n.grp AS grp, {func}(n.{col}) AS agg_out ORDER BY grp", engine) + assert got[0] == "ok", got + assert len(got[1]) == 2 + + +@pytest.mark.parametrize("engine", ALL_ENGINES) +@pytest.mark.parametrize("func", ["min", "max", "count", "collect"]) +@pytest.mark.parametrize("col", sorted(_NON_NUMERIC_COLS)) +def test_any_typed_aggregate_accepts_non_numeric_columns(engine, func, col): + """Cypher declares min/max/count/collect over ``ANY`` (openCypher TCK Aggregation2 [7]-[12] + covers min/max over strings, lists and mixed values). Tightening sum/avg must not tighten + these -- and pandas' grouped min/max over a CATEGORICAL, which used to raise, now answers.""" + _require_engine(engine) + g = _graph() + got = _run(g, f"MATCH (n) RETURN n.grp AS grp, {func}(n.{col}) AS agg_out ORDER BY grp", engine) + assert got[0] == "ok", got + + +@pytest.mark.parametrize("engine", ALL_ENGINES) +def test_min_max_over_categorical_returns_lexicographic_values(engine): + """Concrete values, not just "did not raise": pandas raised here before, so a regression + could otherwise hide behind an empty/None result.""" + _require_engine(engine) + g = _graph() + got = _run(g, "MATCH (n) RETURN n.grp AS grp, min(n.cat_col) AS lo, max(n.cat_col) AS hi ORDER BY grp", + engine) + assert got == ("ok", [(("grp", "x"), ("lo", "a"), ("hi", "b")), + (("grp", "y"), ("lo", "b"), ("hi", "d"))]), got + + +@pytest.mark.parametrize("engine", ["polars", "polars-gpu"]) +def test_categorical_is_decategorized_before_aggregating_on_polars(engine): + """The RESULT DTYPE, not just the values: a categorical must reach the aggregate as a plain + string column. Pinned as a dtype because the failure it prevents is version-dependent and + therefore invisible to a value assertion on a modern polars -- polars 1.35.2 (the RAPIDS + 26.02 image) PANICS in the rust core on a grouped min/max over a Categorical + (`categorical.rs: not implemented`), and a pyo3 PanicException is not even a polars + exception, so no python-side wrapper can turn it into a GFQL error.""" + _require_engine(engine) + out = _graph().gfql( + "MATCH (n) RETURN n.grp AS grp, min(n.cat_col) AS lo ORDER BY grp", engine=engine) + assert str(out._nodes.schema["lo"]) == "String", out._nodes.schema + + +@pytest.mark.parametrize("engine", ["polars", "polars-gpu"]) +def test_polars_native_null_dtype_column_follows_the_all_null_contract(engine): + """A polars-NATIVE all-null column carries dtype ``Null`` (a pandas all-None object column + arrives typed ``String`` instead), and polars refuses `sum`/`mean` on it outright. Distinct + input, same cypher answer: 0 / null.""" + _require_engine(engine) + nodes = pl.DataFrame({"id": [0, 1, 2, 3], "grp": ["x", "x", "y", "y"], + "nul": pl.Series("nul", [None] * 4, dtype=pl.Null)}) + edges = pl.DataFrame({"src": [0, 1], "dst": [1, 2]}) + assert nodes.schema["nul"] == pl.Null + g = graphistry.nodes(nodes, "id").edges(edges, "src", "dst") + got = _run(g, "MATCH (n) RETURN n.grp AS grp, sum(n.nul) AS s, avg(n.nul) AS a ORDER BY grp", + engine) + assert got == ("ok", [(("grp", "x"), ("s", ("num", 0.0)), ("a", None)), + (("grp", "y"), ("s", ("num", 0.0)), ("a", None))]), got + + +@pytest.mark.parametrize("engine", ALL_ENGINES) +def test_all_null_column_sums_to_zero_and_averages_to_null(engine): + """Neo4j: "``sum(null)`` returns ``0``", "``avg(null)`` returns ``null``" -- verified live on + 5.26.26. An all-null column carries NO type evidence, so it is never a type error; it also + cannot be delegated to the host kernels, which answer it 0 / '' / NaT / TypeError depending + on dtype (pandas) or raise for both str and null dtypes (polars).""" + _require_engine(engine) + g = _graph() + got = _run(g, "MATCH (n) RETURN n.grp AS grp, sum(n.allnull_col) AS s, avg(n.allnull_col) AS a ORDER BY grp", engine) + assert got == ("ok", [(("grp", "x"), ("s", ("num", 0.0)), ("a", None)), + (("grp", "y"), ("s", ("num", 0.0)), ("a", None))]), got + + +@pytest.mark.parametrize("engine", ALL_ENGINES) +def test_empty_result_is_not_treated_as_all_null(engine): + """A 0-row frame is "no data", not "all null": the all-null substitution must NOT fire, or + the 0-row schema fill loses the source dtype (an untyped None gives object, which upcasts + sum/avg and escapes via UNION ALL -- the failure mode + test_connected_join_empty_edge_aggregate_keeps_numeric_dtype guards).""" + _require_engine(engine) + g = _graph() + got = _run(g, "MATCH (n) WHERE n.int_col > 9999 RETURN sum(n.int_col) AS s", engine) + assert got[0] == "ok", got + + +# -------------------------------------------------------------------------------------- +# DIFFERENTIAL matrix: aggregate x dtype, pandas oracle vs every other engine +# -------------------------------------------------------------------------------------- + +_MATRIX_COLS = sorted({**_NUMERIC_COLS, **_NON_NUMERIC_COLS, **_ALL_NULL_COLS}) +_MATRIX_AGGS = ["count", "sum", "avg", "min", "max", "collect"] + + +@pytest.mark.parametrize("engine", ["polars", "cudf", "polars-gpu"]) +@pytest.mark.parametrize("func", _MATRIX_AGGS) +@pytest.mark.parametrize("col", _MATRIX_COLS) +def test_aggregate_dtype_matrix_matches_pandas_oracle(engine, func, col): + """The whole aggregate x dtype cross-product must agree with pandas on BOTH axes an engine + can disagree on: the VALUE when it computes, and the ERROR CLASS when it raises. This is the + gate that would have caught the original two divergences -- and the eight sibling cells the + sweep turned up alongside them (sum over date/categorical/all-null, avg over + nullable-string/categorical, min/max over categorical).""" + _require_engine(engine) + g = _graph() + query = f"MATCH (n) RETURN n.grp AS grp, {func}(n.{col}) AS agg_out ORDER BY grp" + oracle = _run(g, query, "pandas") + got = _run(g, query, engine) + assert got == oracle, f"{func}(n.{col}) on {engine}: {got} != pandas {oracle}" + + +# -------------------------------------------------------------------------------------- +# The OLAP single-hop grouped fast path -- a SEPARATE aggregate implementation +# -------------------------------------------------------------------------------------- + +def _fast_path_graph(): + """The shape ``_execute_single_hop_grouped_aggregate_fast_path`` actually accepts: labelled + endpoints + a labelled edge + a direct grouped RETURN. Verified engaged by the spy below -- + a fast-path test written on a shape the fast path declines tests nothing.""" + nodes = pd.DataFrame({ + "id": [0, 1, 2, 10, 11], + "node_type": ["Person", "Person", "Person", "City", "City"], + "age": [20, 30, 40, None, None], + "nick": ["ann", "bob", "cat", None, None], + "city": [None, None, None, "NYC", "LA"], + }) + edges = pd.DataFrame({"s": [0, 1, 2], "d": [10, 10, 11], "rel": ["LIVES_IN"] * 3}) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +# NOTE the placeholders are / and substitution is str.replace, NOT str.format: the +# query text contains cypher property maps ({node_type:'Person'}), which .format() would try to +# interpolate -- it raises KeyError('node_type') before the query is ever run. +_FAST_PATH_QUERY = ( + "MATCH (p {node_type:'Person'})-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "RETURN c.city AS city, () AS agg_out ORDER BY city" +) + + +def _fast_path_query(agg: str, arg: str) -> str: + return _FAST_PATH_QUERY.replace("", agg).replace("", arg) + + +def _run_watching_fast_path(g, query, engine): + """Run, and report whether the fast path was ENTERED (not merely whether it returned rows). + + Entered-ness is recorded before the call so a fast path that raises still counts -- the + negative lane needs exactly that, and a post-hoc `is not None` check would record nothing. + """ + import graphistry.compute.gfql_unified as gu + entered = [] + original = gu._execute_single_hop_grouped_aggregate_fast_path + + def spy(*args, **kwargs): + entered.append(True) + return original(*args, **kwargs) + + gu._execute_single_hop_grouped_aggregate_fast_path = spy + try: + result = _run(g, query, engine) # BEFORE reading `entered`: a tuple literal would + return bool(entered), result # evaluate bool(entered) first and always report False + finally: + gu._execute_single_hop_grouped_aggregate_fast_path = original + + +@pytest.mark.parametrize("engine", ALL_ENGINES) +def test_fast_path_is_actually_engaged_by_this_shape(engine): + """Canary for the two tests below: if the fast path stops accepting this shape they would + silently start testing the ordinary row pipeline instead, and pass for the wrong reason.""" + _require_engine(engine) + entered, got = _run_watching_fast_path( + _fast_path_graph(), _fast_path_query("avg", "p.age"), engine) + assert entered, "fast path not engaged -- the fast-path lane below would be testing nothing" + assert got[0] == "ok", got + + +@pytest.mark.parametrize("engine", ALL_ENGINES) +@pytest.mark.parametrize("func", ["sum", "avg"]) +def test_fast_path_rejects_non_numeric_aggregates(engine, func): + """The fast path reimplements the aggregates, so without its own guard whether an ill-typed + aggregate is caught would depend on whether the query happened to match the fast-path shape. + Both engine branches of the fast path are covered (pandas/cuDF and polars).""" + _require_engine(engine) + entered, got = _run_watching_fast_path( + _fast_path_graph(), _fast_path_query(func, "p.nick"), engine) + assert entered, "fast path not engaged -- this test would not be exercising its guard" + assert got == ("raise", "graphistry.GFQLTypeError"), got + + +# (func, arg) pairs the fast path ACCEPTS. A SOLE min()/max() aggregate does not compile at all +# ("Cypher row lowering cannot ..."), and aggregating the group key itself declines the fast path +# and then fails in the row pipeline -- both pre-existing, both unrelated to types, so the lane is +# pinned to the pairs that actually reach the code under test (min/max still get fast-path +# coverage via the multi-aggregate RETURN in the pinned-value test below). +_FAST_PATH_CASES = [("sum", "p.age"), ("avg", "p.age"), ("count", "p.age"), + ("sum", "p.nick"), ("avg", "p.nick"), ("count", "p.nick")] + + +@pytest.mark.parametrize("engine", ["polars", "cudf", "polars-gpu"]) +@pytest.mark.parametrize("func,arg", _FAST_PATH_CASES) +def test_fast_path_matches_pandas_oracle(engine, func, arg): + """The fast path has an engine split of its own (a polars branch and a pandas/cuDF branch), + so its two aggregate implementations get the same differential treatment as the row + pipeline's -- over a numeric, a string and the group-key column.""" + _require_engine(engine) + g = _fast_path_graph() + query = _fast_path_query(func, arg) + oracle_entered, oracle = _run_watching_fast_path(g, query, "pandas") + entered, got = _run_watching_fast_path(g, query, engine) + assert oracle_entered and entered, "fast path not engaged on both engines" + assert got == oracle, f"{func}({arg}) on {engine}: {got} != pandas {oracle}" + + +@pytest.mark.parametrize("engine", ALL_ENGINES) +def test_fast_path_numeric_aggregates_keep_their_values(engine): + """Pinned values, so the guard cannot be "passed" by an implementation that stopped + computing: the fast path must still answer the numeric aggregates it was written for.""" + _require_engine(engine) + entered, got = _run_watching_fast_path( + _fast_path_graph(), + "MATCH (p {node_type:'Person'})-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "RETURN c.city AS city, sum(p.age) AS s, avg(p.age) AS a, max(p.nick) AS m ORDER BY city", + engine) + assert entered + assert got == ("ok", [(("city", "LA"), ("s", ("num", 40.0)), ("a", ("num", 40.0)), ("m", "cat")), + (("city", "NYC"), ("s", ("num", 50.0)), ("a", ("num", 25.0)), ("m", "bob"))]), got + + +# -------------------------------------------------------------------------------------- +# The classifiers themselves +# -------------------------------------------------------------------------------------- + +def test_pandas_fast_numeric_gate_admits_exactly_the_kernel_safe_dtypes(): + """The O(1) hot-path gate. A FALSE POSITIVE here is the dangerous direction -- it would send + a non-numeric column straight to the host kernel, restoring the concatenation.""" + from graphistry.compute.gfql.agg_types import pandas_dtype_is_numeric_for_agg as numeric + assert numeric(pd.Series([1, 2])) + assert numeric(pd.Series([1.5])) + assert numeric(pd.Series([1], dtype="Int64")) + assert numeric(pd.Series([True])) + assert numeric(pd.Series([1], dtype="boolean")) + assert numeric(pd.to_timedelta([1], unit="D").to_series()) + assert not numeric(pd.Series(["a"])) + assert not numeric(pd.Series(["a"], dtype="object")) + assert not numeric(pd.Series(["a"], dtype="string")) + assert not numeric(pd.Series(pd.Categorical(["a"]))) + assert not numeric(pd.Series(pd.to_datetime(["2020-01-01"]))) + assert not numeric(pd.Series([None, None])) # object: no type evidence + assert not numeric(pd.Series(pd.period_range("2020", periods=1, freq="D"))) + assert not numeric(pd.Series(pd.interval_range(0, 2, periods=1))) + + +def test_pandas_classifier_rejects_strings_and_admits_numbers_and_durations(): + from graphistry.compute.gfql.agg_types import pandas_non_numeric_agg_dtype as reject + assert reject(pd.Series([1, 2, 3])) is None + assert reject(pd.Series([1.0, 2.0])) is None + assert reject(pd.Series([True, False])) is None + assert reject(pd.to_timedelta([1, 2], unit="D").to_series()) is None + assert reject(pd.Series([1, 2], dtype="object")) is None # numbers boxed in object + # The LABEL is the dtype's own repr and pandas changes it across versions (a bare + # pd.Series(["a"]) is `object` on pandas 2 and `str` on pandas 3), so assert the VERDICT and + # that the label carries the dtype -- pinning the exact text tests pandas, not this contract. + assert reject(pd.Series(["a", "b"], dtype="object")) == "object (strings)" + assert reject(pd.Series(["a", "b"])) is not None + assert reject(pd.Series(["a"], dtype="string")) == "string" + # every string spelling pandas has used across versions/storages must reject; a missed one + # fails OPEN (delegates to the kernel and restores the concatenation) + for spelling in ["object", "str", "string", "string[python]", "string[pyarrow]"]: + try: + series = pd.Series(["a", "b"], dtype=spelling) + except (TypeError, ValueError): + continue # dtype not available in this pandas/pyarrow build + assert reject(series) is not None, spelling + assert reject(pd.Series(pd.Categorical(["a"]))) is not None + assert reject(pd.Series(pd.to_datetime(["2020-01-01"]))) is not None + + +def test_polars_classifier_rejects_strings_and_admits_numbers_and_durations(): + from graphistry.compute.gfql.agg_types import polars_non_numeric_agg_dtype as reject + assert reject(pl.Int64) is None + assert reject(pl.Float64) is None + assert reject(pl.Boolean) is None + assert reject(pl.Duration) is None + assert reject(pl.Null) is None # all-null: no type evidence, never a type error + assert reject(None) is None + assert reject(pl.String) == "String" + assert reject(pl.Categorical) is not None + assert reject(pl.Date) is not None + assert reject(pl.Datetime) is not None + assert reject(pl.List(pl.Int64)) is not None + + +def test_all_null_substitution_values_follow_cypher(): + from graphistry.compute.gfql.agg_types import numeric_agg_all_null_value + assert numeric_agg_all_null_value("sum") == 0 + assert numeric_agg_all_null_value("avg") is None + assert numeric_agg_all_null_value("mean") is None + + +def test_numeric_only_aggregation_set_covers_both_spellings(): + """``avg`` is the cypher name and ``mean`` GFQL's internal one (GFQL_GROUPBY_AGG_METHODS maps + avg -> mean); a set holding only one of them would leave the other unguarded.""" + from graphistry.compute.gfql.agg_types import GFQL_NUMERIC_ONLY_AGGREGATIONS + from graphistry.compute.gfql.language_defs import GFQL_GROUPBY_AGG_METHODS + assert GFQL_NUMERIC_ONLY_AGGREGATIONS == {"sum", "avg", "mean"} + assert set(GFQL_GROUPBY_AGG_METHODS) - GFQL_NUMERIC_ONLY_AGGREGATIONS == { + "count", "count_distinct", "min", "max" + }