diff --git a/.claude/skills/slayer-query.md b/.claude/skills/slayer-query.md index e1fb5171..dbba3297 100644 --- a/.claude/skills/slayer-query.md +++ b/.claude/skills/slayer-query.md @@ -22,6 +22,8 @@ A `SlayerQuery` is a JSON/dict object. The same shape works across the REST API, `order[].column` is the short alias (`count`, `revenue_sum`) — not the colon form. +**Ordering by something you don't project.** `order` may name an undeclared column/aggregate ("top-N by X, show only Y, Z"). An **aggregate** (`amount:sum`, `customers.revenue:sum`) is computed hidden, sorted on, and stripped from the result. A **raw row column** is orderable only in a raw-rows query (`distinct_dimension_values: false`); in a grouped/dedup query it's rejected (HTTP 400 — add it to `dimensions` or order by an aggregate of it). A **joined** row column and an inline **transform/composite** (`change(amount:sum)`) are also rejected — project it, or declare it as a measure and order by that. + **Dim-only queries deduplicate.** A query with no measures and at least one dimension or time-dimension auto-emits `GROUP BY ` and returns the distinct combinations. The `GROUP BY` is applied before `LIMIT`, so a row cap can't silently drop unique tuples. To opt out, set `"distinct_dimension_values": false` on the query — emits raw rows (no top-level `GROUP BY`), with WHERE / ORDER BY / LIMIT applied as usual. Any measure reference in `measures` / `filters` / `order` raises `DistinctDimensionValuesError` in this mode. ## Measures — colon aggregation diff --git a/DECISIONS.md b/DECISIONS.md index 87adc3a9..96c3486a 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -79,3 +79,4 @@ implementation detail. Include issue refs when known. - 2026-08-03 — CTE-name collision detection stays case-sensitive for now (DEV-1713 Codex review, deferred to DEV-1726): `AliasAllocator` / `assert_unique_cte_names` compare CTE names by exact string, but generated CTE names are emitted unquoted and case-fold on Postgres/Snowflake/Redshift — so two user measure aliases differing only in case, both generating CTEs, can still collide there. Deferred rather than fixed in Stage 9 because it is pre-existing (pre-DEV-1713 the names weren't deduped at all), an edge case, and the correct fix is dialect-aware (fold only for case-folding dialects; a blanket case-insensitive dedup would wrongly merge genuinely-distinct names on case-sensitive BigQuery / quoted SQLite). Tracked in DEV-1726. - 2026-08-03 — Merge resolution, Stage 4 × Stage 9 (DEV-1708 × DEV-1713): the user-approved derived-shared-grain raise (DEV-1708) wins over Stage 9's combined test vehicle — D3's dotted final-stage keys do NOT by themselves make a plain derived joined dim legal as cross-model shared grain (the CTE join-back rendering is still unbuilt; full support remains DEV-1495). The two Stage-9 Codex-F6 alias/key-agreement tests split their derived-dim and cross-model-aggregate coverage into separate queries. - 2026-08-03 — Dialect-aware CTE-name case-folding (DEV-1726): `AliasAllocator` gains `folds_case` (resolved via `naming.dialect_folds_case`, threaded by the single `SQLGenerator._new_allocator` factory — the only construction site, test-pinned) and `assert_unique_cte_names` folds per dialect. Comparison-only folding with `str.lower()` (not `casefold`; sqlglot `normalize_identifier` parity) — allocated names keep original case, so output changes only when a genuine fold-collision forces a `_2` walk. Fold set = every registry dialect EXCEPT ClickHouse; unknown strings stay exact. Issue-text corrections (GoogleSQL docs + sqlglot + empirics): BigQuery FOLDS (CTE names are query aliases, case-insensitive — only real table names are CS) and SQLite/DuckDB fold even quoted names; MySQL/T-SQL fold deliberately despite config-dependence (folding is rename-only-safe, not folding leaves the bug live on majority configs). The belt folds regardless of quoting — over-strict by design on allocator-sanitized output, never a general SQL validator. Public result keys are reserved, never allocator-minted, so folding can never rename them. +- 2026-08-03 — Order-only hidden slots + plan-time order/partition validations (DEV-1712, DEV-1703 Stage 8). An ORDER BY ref that is not a declared dimension/measure is classified at **plan time** in `stage_planner.plan_query` (right after `_bucket_slots`): an **aggregate** (local or cross-model) always materialises hidden and orders — never rejected; a **local row column** is allowed only in a raw-rows query (`distinct_dimension_values=False`, no grouping) and emits a SPLIT `orders.` reference in the generator's `_apply_order_limit_from_planned` (the old `NotImplementedError` becomes a defensive internal assertion — the plan-time pass guarantees only that shape reaches it); a **grouped** local row column raises `ValueError` (not in GROUP BY — add it to dims or order by an aggregate of it); a **joined** row column raises `UnresolvableOrderColumnError`; an inline **transform/composite** (`change(amount:sum)`, only reachable as `raw_formula` — composite arithmetic is unexpressible via `OrderItem.column`, Pydantic rejects it) raises `ValueError` pointing at "declare it as a measure" — full support deferred to **DEV-1733** with strict-xfail future tests + a matching worktree. The grouping predicate is planner-semantic: `bool(agg_slots) or (dims/tds present and distinct_dimension_values)` — a hidden order aggregate that induces grouping counts, so a row column in that same query is correctly rejected. **Hidden cross-model aggregate trim** (DEV-1495 bug 2): an order-only CMA gets `hidden=True`/`public_alias=None` from the planner; the generator's combined-projection loop skips it (`public_aliases=[]`) **only when there is no transform chain** — a hidden CMA feeding a `cumsum(...)` step must stay projected for the step CTE to consume (the transform outer-wrap does the public-vs-hidden trim there). Its ORDER BY term is CTE-qualified (`_cm_*.""`) since the bare alias is no longer projected. The DEV-1495-bug-1 malformed-alias half (`orders.customers._sum`) was already fixed by Stage 9's naming module — only the projection leak remained. **partition_by grain guard** (DEV-1497): a pre-intern pass (`planning.rewrite_rank_partition_keys`, mirroring `lower_sugar_transforms`' identity-preserving rebuild) validates every rank-family (`rank`/`dense_rank`/`percent_rank`/`ntile`) `partition_by` key resolves to a query dimension/time-dimension by **exact ValueKey membership** — the typed binder resolves `partition_by` to a ValueKey before validation, so the legacy string-matching ambiguity can't arise. A time-dimension **source column** is rewritten to its `TimeTruncKey` so `PARTITION BY` uses the truncated bucket (not the raw timestamp, which had silently widened the GROUP BY grain and emitted a duplicate alias); a non-dimension raises `ValueError` naming the transform + column + available dims (the legacy `enrichment._resolve_rank_partition` message, restored). The 7 DEV-1645 Flavor-A ORDER-BY unit pins (split-not-composite for unprojected sort keys; `UnresolvableOrderColumnError` for joined sort keys) were made green by porting main's legacy `_OrderColRef`/`_order_split_sql`/`_resolve_order_column` fix (lost in the Stage-0 merge) into the legacy generator — throwaway parity, deleted with the legacy stack in Stage 11 — so `tests/parity_xfails.py` is now empty (the DEV-1485 end-state). Deliberate divergence from main: the typed pipeline rejects a grouped raw-row order at plan time (HTTP 400) instead of emitting SQL the database rejects at execution. A follow-up review sweep (PR #274) extended the same joined-ORDER-BY host-local guard into the legacy `_resolve_order_column` and rejected an unprojected sort key in the CTE-wrapped `_apply_pagination_to_sql`; added a partition ambiguity guard for a time column carried at two granularities. diff --git a/docs/concepts/formulas.md b/docs/concepts/formulas.md index e09835ea..a193d61c 100644 --- a/docs/concepts/formulas.md +++ b/docs/concepts/formulas.md @@ -230,7 +230,7 @@ Combine with a filter to get "top N": **Ranking within a partition (`partition_by=`):** -To rank within groups instead of across the whole result set, pass `partition_by=` referencing one or more **query dimensions** (or time dimensions). The columns must already be grouped on — partitioning by a column that's not a dimension errors at enrichment time. +To rank within groups instead of across the whole result set, pass `partition_by=` referencing one or more **query dimensions** (or time dimensions). The columns must already be grouped on — partitioning by a column that's not a dimension errors at plan time (HTTP 400). Naming a query time-dimension partitions by its truncated bucket, not the raw timestamp. ```json { diff --git a/docs/concepts/queries.md b/docs/concepts/queries.md index a8008bd1..3ecfd8e0 100644 --- a/docs/concepts/queries.md +++ b/docs/concepts/queries.md @@ -107,6 +107,27 @@ A sort specification: `column` is the short alias (`status`, `revenue_sum`, `*:c Via MCP: `{"column": "*:count", "direction": "desc"}` +### Ordering by something you don't project + +`order` may reference a column or aggregate that is **not** declared as a dimension/measure — the classic "top-N by metric X, display only Y, Z" pattern: + +```json +{"source_model": "orders", "dimensions": ["status"], "measures": [{"formula": "*:count"}], + "order": [{"column": "amount:sum", "direction": "desc"}], "limit": 10} +``` + +The `amount:sum` aggregate is computed as a hidden column, sorted on, and **stripped from the result** — the response projects only `status` and `_count`. This works for local aggregates, cross-model aggregates (`customers.revenue:sum`), and inner-stage columns re-aggregated in a later DAG stage (`customers__revenue_sum:max`). + +What each shape of an *undeclared* order target does: + +| Order target | Behavior | +| --- | --- | +| An aggregate (`amount:sum`, `customers.revenue:sum`) | Computed hidden, sorted on, stripped from the result. Always allowed. | +| A raw row column, in a **raw-rows** query (`distinct_dimension_values: false`, no measures) | Sorted on directly (`ORDER BY orders.created_at`). | +| A raw row column, in an **aggregated / dedup** query | Rejected (HTTP 400): it isn't in the `GROUP BY`. Add it to `dimensions`, or order by an aggregate of it (`created_at:max`). | +| A **joined** row column (`customers.regions.name`) not projected | Rejected (HTTP 400): project it (add to `dimensions`) or order by a projected field. | +| An inline **transform / composite** (`change(amount:sum)`, `revenue:sum / cnt:sum`) not declared | Rejected (HTTP 400): declare it as a measure (optionally with a `name`) and order by that. | + ## Response Query results are returned as a `SlayerResponse`: diff --git a/slayer/engine/planning.py b/slayer/engine/planning.py index b909c7af..2a57aa70 100644 --- a/slayer/engine/planning.py +++ b/slayer/engine/planning.py @@ -25,7 +25,7 @@ from __future__ import annotations -from typing import Dict, FrozenSet, List, Optional +from typing import Callable, Dict, FrozenSet, List, Optional from pydantic import BaseModel, ConfigDict, Field @@ -54,6 +54,7 @@ column_path, normalize_scalar, ) +from slayer.core.formula import RANK_FAMILY_TRANSFORMS from slayer.core.refs import agg_kwarg_canonical_str, canonical_agg_name from slayer.engine.binding import BoundExpr, BoundFilter from slayer.engine.planned import SlotId, ValueSlot @@ -401,6 +402,63 @@ def lower_sugar_transforms(key: ValueKey) -> ValueKey: return key +def rewrite_rank_partition_keys( # NOSONAR(S3776) — sequential isinstance dispatch over the closed ValueKey union; each branch is the per-type identity-preserving rebuild contract, mirroring lower_sugar_transforms. Extracting per-type helpers would scatter the contract across the module. + key: ValueKey, *, rewrite_fn: Callable[[TransformKey], FrozenSet], +) -> ValueKey: + """Walk ``key``; for every rank-family ``TransformKey`` carrying an + explicit ``partition_by`` (non-empty ``partition_keys``), replace those + keys with ``rewrite_fn(transform_key)`` (DEV-1497). + + ``rewrite_fn`` receives the whole ``TransformKey`` and returns a new + ``frozenset`` of partition keys — validating that each resolves to a query + dimension / time-dimension and rewriting a time-dimension source column to + its ``TimeTruncKey`` bucket. It may raise ``ValueError`` for a partition + column that is not a query dimension. + + Identity-preserving (mirrors :func:`lower_sugar_transforms`): parents are + rebuilt only where a child changed, so this runs BEFORE interning without + churning unrelated slots. Reaches rank transforms nested in composite + measures (``ArithmeticKey`` / ``ScalarCallKey``) and in filter predicates + (comparisons are ``ArithmeticKey``). + """ + def _rec(k: ValueKey) -> ValueKey: + return rewrite_rank_partition_keys(key=k, rewrite_fn=rewrite_fn) + + if isinstance(key, TransformKey): + new_input = _rec(key.input) + new_pk = key.partition_keys + if key.op in RANK_FAMILY_TRANSFORMS and key.partition_keys: + new_pk = rewrite_fn(key) + if new_input is key.input and new_pk == key.partition_keys: + return key + return key.model_copy(update={"input": new_input, "partition_keys": new_pk}) + if isinstance(key, ArithmeticKey): + new_ops = tuple(_rec(op) for op in key.operands) + unchanged = all(a is b for a, b in zip(new_ops, key.operands)) + return key if unchanged else ArithmeticKey(op=key.op, operands=new_ops) + if isinstance(key, ScalarCallKey): + rewritable = _SLOTTABLE_KIND + (ArithmeticKey, ScalarCallKey, BetweenKey) + new_args = tuple( + _rec(a) if isinstance(a, rewritable) else a for a in key.args + ) + unchanged = all(a is b for a, b in zip(new_args, key.args)) + return key if unchanged else ScalarCallKey(name=key.name, args=new_args) + if isinstance(key, BetweenKey): + new_col, new_low, new_high = _rec(key.column), _rec(key.low), _rec(key.high) + unchanged = ( + new_col is key.column and new_low is key.low and new_high is key.high + ) + return key if unchanged else BetweenKey( + column=new_col, low=new_low, high=new_high, + ) + if isinstance(key, InKey): + new_col = _rec(key.column) + return key if new_col is key.column else InKey( + column=new_col, values=key.values, negated=key.negated, + ) + return key + + def desugar_change_pct(key: TransformKey) -> ArithmeticKey: """``change_pct(x)`` → ``(x - time_shift(x, periods=-1)) / NULLIF(time_shift(x, periods=-1), 0)``. diff --git a/slayer/engine/stage_planner.py b/slayer/engine/stage_planner.py index 2b07d4c7..172e4c6c 100644 --- a/slayer/engine/stage_planner.py +++ b/slayer/engine/stage_planner.py @@ -32,6 +32,7 @@ from slayer.core.errors import ( AmbiguousReferenceError, UnknownReferenceError, + UnresolvableOrderColumnError, ) from slayer.core.keys import ( AggregateKey, @@ -86,6 +87,7 @@ _iter_slot_deps, filter_referenced_slot_ids, lower_sugar_transforms, + rewrite_rank_partition_keys, ) from slayer.engine.source_bundle import ( ResolvedSourceBundle, @@ -175,6 +177,27 @@ def _attach_time_keys( return key +def _partition_key_display(pk: ValueKey) -> str: + """Human-readable name of a rank ``partition_by`` key for error messages + (DEV-1497). Local refs surface as the bare leaf; joined refs keep the + dotted path.""" + if isinstance(pk, ColumnKey): + return ".".join([*pk.path, pk.leaf]) + if isinstance(pk, ColumnSqlKey): + return ".".join([*pk.path, pk.column_name]) + if isinstance(pk, TimeTruncKey): + return _partition_key_display(pk.column) + return str(pk) + + +def _row_key_path(key: ValueKey) -> tuple: + """Join path of a ROW value key (``()`` for local, non-empty for joined). + Unwraps a ``TimeTruncKey`` to its underlying column.""" + if isinstance(key, TimeTruncKey): + return _row_key_path(key.column) + return tuple(getattr(key, "path", ())) + + def _find_unresolved_time_needing_op(key: ValueKey) -> Optional[str]: """Return the op name of the first time-needing TransformKey reached that has ``time_key is None``, or ``None`` if every time-needing @@ -369,9 +392,26 @@ def plan_query( # NOSONAR(S3776) — planner entry-point dispatcher. The DEV-15 bound_filter_texts.append(f) order_specs = [] + # Host identity for the qualifier check below — the source model for a + # ``ModelScope``, the stage relation name (``s1``) for a downstream + # ``StageSchema`` (so a self-qualified ``s1.metric`` order stays host-local, + # Codex). Same resolution ``_host_model_name`` uses everywhere else. + _order_host_name = _host_model_name(scope) for o in (query.order or []): col_name = o.column.name full_name = o.column.full_name + # An order ref qualified with a FOREIGN model (``owners.status`` when + # the host is ``orders``) must not resolve to a same-named local column + # via the bare-leaf shortcut — otherwise a joined sort key silently + # binds to the local column and sorts by the wrong field (Codex). The + # bare-name lookups below apply only to unqualified refs or refs + # qualified with the host itself; a foreign-qualified ref falls through + # to the dotted/flattened/`bind_expr` paths, where a truly-joined ref + # is then rejected by the plan-time order validation. + _order_qualifier = getattr(o.column, "model", None) + _order_host_local = ( + _order_qualifier is None or _order_qualifier == _order_host_name + ) # Prefer declared-measure alias resolution over model-scope # binding (DEV-1450 stage 7b.8 — gap fix): aggregate canonical # aliases like ``amount_sum`` are not columns on the model, so @@ -386,7 +426,7 @@ def plan_query( # NOSONAR(S3776) — planner entry-point dispatcher. The DEV-15 # fall back to binding the preserved colon/path ``raw_formula`` # so the order key interns onto the same cross-model aggregate # slot (P2/P4) rather than raising. - if col_name in declared_alias_to_bound: + if _order_host_local and col_name in declared_alias_to_bound: bo = declared_alias_to_bound[col_name] elif full_name in declared_alias_to_bound: bo = declared_alias_to_bound[full_name] @@ -397,7 +437,7 @@ def plan_query( # NOSONAR(S3776) — planner entry-point dispatcher. The DEV-15 # written in dotted form must intern onto that same declared # slot rather than binding the raw column as a fresh slot. bo = declared_alias_to_bound[_flatten_dotted(full_name)] - elif f"_{col_name}" in declared_alias_to_bound: + elif _order_host_local and f"_{col_name}" in declared_alias_to_bound: # ``*:count`` surfaces as the alias ``_count`` (the ``*`` is # dropped, the leading ``_`` kept as a marker); users naturally # order by the bare ``count``. Mirror the legacy @@ -545,6 +585,90 @@ def plan_query( # NOSONAR(S3776) — planner entry-point dispatcher. The DEV-15 for spec in order_specs ] + # DEV-1497: validate that every rank-family ``partition_by`` column resolves + # to a query dimension / time-dimension, and rewrite a time-dimension source + # column to its truncated-bucket ``TimeTruncKey`` (partition by the bucket, + # not the raw timestamp — which would silently widen the grain). Runs BEFORE + # interning so a rewritten key never leaves a stale slot behind (identity is + # only touched on the rewritten rank transform). + _dim_dms = declared_measures[:n_dims] + _td_dms = declared_measures[n_dims:n_dims + n_tds] + _dim_key_set = {dm.bound.value_key for dm in _dim_dms} + # A source column carrying two time-dimension granularities (``created_at`` + # at both month and day) maps to two distinct ``TimeTruncKey`` buckets — a + # bare ``partition_by=created_at`` is then ambiguous, so track those columns + # and reject rather than silently pick whichever bucket comes last. + _td_by_source: Dict[ValueKey, TimeTruncKey] = {} + _td_ambiguous_sources: set = set() + for dm in _td_dms: + vk = dm.bound.value_key + if not isinstance(vk, TimeTruncKey): + continue + # Ambiguous only when the SAME source column already mapped to a + # DIFFERENT bucket (a different granularity) — two identical + # ``created_at:month`` declarations resolve to one bucket, not a clash. + if vk.column in _td_by_source and _td_by_source[vk.column] != vk: + _td_ambiguous_sources.add(vk.column) + _td_by_source[vk.column] = vk + _td_key_set = set(_td_by_source.values()) + _available_dims = [dm.declared_name for dm in (*_dim_dms, *_td_dms)] + + def _validate_partition_keys(tk: TransformKey) -> frozenset: + new_pks = [] + for pk in tk.partition_keys: + if pk in _dim_key_set or pk in _td_key_set: + new_pks.append(pk) # already a query dim / td bucket + elif pk in _td_ambiguous_sources: + raise ValueError( + f"Transform '{tk.op}': partition_by column " + f"'{_partition_key_display(pk)}' is ambiguous — it is a " + f"time dimension at multiple granularities. Partition by a " + f"single query dimension instead." + ) + elif pk in _td_by_source: + new_pks.append(_td_by_source[pk]) # td source col -> bucket + else: + raise ValueError( + f"Transform '{tk.op}': partition_by column " + f"'{_partition_key_display(pk)}' is not a query dimension. " + f"Add it to dimensions/time_dimensions, or choose one of: " + f"{', '.join(_available_dims) or '(none)'}." + ) + return frozenset(new_pks) + + def _rw(vk: ValueKey) -> ValueKey: + return rewrite_rank_partition_keys(vk, rewrite_fn=_validate_partition_keys) + + declared_measures = [ + DeclaredMeasure( + bound=BinderBoundExpr(value_key=_rw(dm.bound.value_key)), + declared_name=dm.declared_name, + public_name=dm.public_name, + label=dm.label, + canonical_alias=dm.canonical_alias, + type=dm.type, + format=dm.format, + description=dm.description, + ) + for dm in declared_measures + ] + _rewritten_filters = [] + for bf in bound_filters: + bf_vk = _rw(bf.value_key) + _rewritten_filters.append(BoundFilter( + value_key=bf_vk, + phase=bf.phase, + referenced_keys=tuple(walk_value_keys(bf_vk)), + )) + bound_filters = _rewritten_filters + order_specs = [ + OrderSpec( + bound=BinderBoundExpr(value_key=_rw(spec.bound.value_key)), + direction=spec.direction, + ) + for spec in order_specs + ] + source_col_names = _source_column_names(scope) host_model_name = _host_model_name(scope) @@ -560,6 +684,56 @@ def plan_query( # NOSONAR(S3776) — planner entry-point dispatcher. The DEV-15 projection.registry.slots, ) + # DEV-1712 (Law 2): plan-time classification of every ORDER BY target that + # is not a declared/public slot. Order-only AGGREGATES (local or + # cross-model) always materialise and order — never rejected. The rest: + # * joined row column -> UnresolvableOrderColumnError (the sort + # scope is relocated where the joined table is unbound); + # * local row column, grouped -> ValueError (no valid SQL — the column + # isn't in GROUP BY; add it to dims, or order by an aggregate of it); + # * local row column, ungrouped -> allowed (split emission in the + # generator, ``_apply_order_limit_from_planned``); + # * transform / composite -> ValueError (deferred to DEV-1733; + # declare it as a measure and order by that). + _has_grouping = bool(agg_slots) or ( + bool(query.dimensions or query.time_dimensions) + and query.distinct_dimension_values + ) + for spec in order_specs: + okey = spec.bound.value_key + osid = projection.registry.find_by_key(okey) + if osid is not None and not projection.registry.get(osid).hidden: + continue # declared / projected output — orders on a real column + if isinstance(okey, AggregateKey): + continue # hidden aggregate (local base or cross-model CTE) + if isinstance(okey, (ColumnKey, ColumnSqlKey, TimeTruncKey)): + disp = _partition_key_display(okey) + path = _row_key_path(okey) + if path: + # ``UnresolvableOrderColumnError`` formats ``qualifier.column``; + # pass the bare leaf as ``column`` and the joined path as the + # qualifier so the message reads ``customers.regions.name`` and + # not a duplicated ``customers.customers.regions.name``. + raise UnresolvableOrderColumnError( + column=disp.rsplit(".", 1)[-1], qualifier=".".join(path), + ) + if _has_grouping: + raise ValueError( + f"ORDER BY column '{disp}' is a row column that this " + f"aggregated query does not project, so it is not in the " + f"GROUP BY. Add it to dimensions/time_dimensions, or order " + f"by an aggregate of it (e.g. '{disp}:max')." + ) + continue # ungrouped local row column -> split emission + # TransformKey / ArithmeticKey / ScalarCallKey — an inline transform or + # composite expression that is only referenced in ORDER BY. + raise ValueError( + "ORDER BY references a transform / composite expression that is " + "not a declared measure. Declare it as a measure (optionally with " + "a name) and order by that measure. (DEV-1733: materialising inline " + "transform / composite ORDER BY targets is not yet supported.)" + ) + # Build filters_by_phase in legacy WHERE order: # 1. date_range bound filters (bound_filters[:n_date_range]) # 2. model.filters (text_filter_entries) diff --git a/slayer/sql/generator.py b/slayer/sql/generator.py index d2e79c6a..5282dbfe 100644 --- a/slayer/sql/generator.py +++ b/slayer/sql/generator.py @@ -8,7 +8,7 @@ import copy import logging import re -from typing import Any, Dict, List, Literal, Optional, Set, Tuple, Union +from typing import Any, Dict, List, Literal, NamedTuple, Optional, Set, Tuple, Union import sqlglot from sqlglot import exp @@ -22,7 +22,7 @@ ) from pydantic import BaseModel, ConfigDict, field_validator -from slayer.core.errors import AggregationNotAllowedError +from slayer.core.errors import AggregationNotAllowedError, UnresolvableOrderColumnError from slayer.core.keys import _reroot_path_ref, reroot_aggregate_key from slayer.core.models import Aggregation from slayer.core.refs import agg_kwarg_canonical_str @@ -54,6 +54,24 @@ from slayer.sql.stage_wrapper import build_flat_rename_wrapper +class _OrderColRef(NamedTuple): + """DEV-1645: resolved ORDER BY key. ``is_alias`` distinguishes a projected + output alias (emit whole-quoted, e.g. ``"orders.revenue_sum"``) from a + table.column fallback (emit SPLIT, e.g. ``ranked."time_mark"``), so a sort + on an unprojected/renamed column references the underlying FROM-scope column + instead of a nonexistent composite identifier. + + Ported from ``origin/main`` for the LEGACY enrichment pipeline (this stack is + deleted in DEV-1485 Stage 11); the typed pipeline enforces the same policy + independently in ``_apply_order_limit_from_planned`` + the plan-time order + validation. + """ + text: str # whole resolved string (used for base_cols membership) + is_alias: bool # True => projected output alias; False => table.column fallback + qualifier: str | None # fallback only: FROM-scope alias + column: str | None # fallback only: underlying column short name + + class ResolvedAggKwarg(BaseModel): """DEV-1706 — a resolved parametric-aggregation kwarg value (2-kind tag). @@ -1391,13 +1409,14 @@ def _assemble_combined_sql(self, enriched: EnrichedQuery, } for order_item in enriched.order: col = order_item.column - col_name = self._resolve_order_column(col=col, enriched=enriched) + ref = self._resolve_order_column(col=col, enriched=enriched) direction = "ASC" if order_item.direction == "asc" else "DESC" - qcol = self._quote_ident(col_name) # DEV-1716: dialect-quoted - if col_name in base_cols: - order_parts.append(f'_base.{qcol} {direction}') + if ref.is_alias and ref.text in base_cols: + order_parts.append(f'_base.{self._quote_ident(ref.text)} {direction}') + elif ref.is_alias: + order_parts.append(f'{self._quote_ident(ref.text)} {direction}') else: - order_parts.append(f'{qcol} {direction}') + order_parts.append(f'{self._order_split_sql(ref)} {direction}') sql += "\nORDER BY " + ", ".join(order_parts) if enriched.limit is not None: sql += f"\nLIMIT {enriched.limit}" @@ -1407,14 +1426,28 @@ def _assemble_combined_sql(self, enriched: EnrichedQuery, return sql def _apply_pagination_to_sql(self, enriched: EnrichedQuery, sql: str) -> str: - """Apply ORDER BY, LIMIT, OFFSET to a raw SQL string.""" + """Apply ORDER BY, LIMIT, OFFSET to a raw SQL string. + + This wrapper is only ever applied over the CTE-wrapped computed-column + assembly (its single caller builds ``WITH … SELECT … FROM ``), + so the outer FROM is a CTE — a SPLIT ``.`` reference + would name a table unbound in this scope. An unprojected (non-alias) + sort key is therefore unresolvable here (unlike the base-SELECT applier + ``_apply_order_limit``, where the split IS bound). Reject it rather than + emit invalid SQL — consistent with the typed pipeline's plan-time guard. + """ if enriched.order: order_parts = [] for order_item in enriched.order: col = order_item.column - col_name = SQLGenerator._resolve_order_column(col=col, enriched=enriched) + ref = SQLGenerator._resolve_order_column(col=col, enriched=enriched) direction = "ASC" if order_item.direction == "asc" else "DESC" - order_parts.append(f'{self._quote_ident(col_name)} {direction}') # DEV-1716 + if ref.is_alias: + order_parts.append(f'{self._quote_ident(ref.text)} {direction}') + else: + raise UnresolvableOrderColumnError( + column=ref.column, qualifier=ref.qualifier, + ) sql += "\nORDER BY " + ", ".join(order_parts) if enriched.limit is not None: sql += f"\nLIMIT {enriched.limit}" @@ -2331,8 +2364,14 @@ def _apply_order_limit(self, select: exp.Select, enriched: EnrichedQuery) -> exp if enriched.order: for order_item in enriched.order: col = order_item.column - col_name = self._resolve_order_column(col=col, enriched=enriched) - order_col = exp.Column(this=exp.to_identifier(col_name, quoted=True)) + ref = self._resolve_order_column(col=col, enriched=enriched) + if ref.is_alias: + order_col = exp.Column(this=exp.to_identifier(ref.text, quoted=True)) + else: + order_col = exp.Column( + this=self._to_ident(ref.column), + table=exp.to_identifier(ref.qualifier), + ) ascending = order_item.direction == "asc" select = select.order_by(self._ordered(order_col, ascending=ascending)) @@ -2344,20 +2383,41 @@ def _apply_order_limit(self, select: exp.Select, enriched: EnrichedQuery) -> exp return select + def _order_split_sql(self, ref: _OrderColRef) -> str: + """DEV-1645: emit a non-projected ORDER BY key as a SPLIT + ``qualifier.column`` reference (mixed-case-quoted), not one + composite-quoted token.""" + col = exp.Column(this=self._to_ident(ref.column), table=exp.to_identifier(ref.qualifier)) + return col.sql(dialect=self.dialect) + @staticmethod - def _resolve_order_column(col, enriched: EnrichedQuery) -> str: - """Resolve an order column reference to the correct enriched alias. + def _resolve_order_column(col, enriched: EnrichedQuery) -> _OrderColRef: + """Resolve an order column reference to a discriminated result (DEV-1645). Users refer to columns by their short name (e.g., ``count``, ``revenue_sum``). The enriched query stores fully qualified aliases (e.g., ``orders._count``, ``orders.revenue_sum``). This method - matches the user-provided name against all enriched columns and - returns the matching alias. If no match is found, the name is - qualified with the model name as a fallback. - - For ``*:count`` results, the internal name is ``_count`` but users - refer to it as ``count``. A fallback check for ``_name`` handles - this case. + matches the user-provided name against all enriched columns. + + When it matches a projected alias, the result carries ``is_alias=True`` + and the caller emits it whole-quoted (``"orders.revenue_sum"`` — that IS + the real output column name via ``AS "orders.revenue_sum"``). + + When no projected alias matches (renamed via ``columns:``, or an + inner-stage dim the outer stage dropped), the result carries + ``is_alias=False`` with ``qualifier``/``column`` set, and the caller + emits a SPLIT ``qualifier.column`` reference that resolves against the + FROM-scope table — instead of the old composite ``"."`` + token that Postgres rejects as UndefinedColumn. + + A joined qualifier (anything other than the base model) is rejected with + ``UnresolvableOrderColumnError``: the compiler's outer-wrapping layers + (measure CTEs, pagination, the first/last ranked subquery, projection + trimming) relocate the ORDER BY into a scope where the joined table is + unbound, so emitting a reference there would produce invalid SQL. + + For ``*:count`` results, the internal name is ``_count`` but users refer + to it as ``count``. A fallback check for ``_name`` handles this case. """ user_name = col.name model_prefix = col.model or enriched.model_name @@ -2379,24 +2439,39 @@ def _resolve_order_column(col, enriched: EnrichedQuery) -> str: # Custom field names (e.g., {"formula": "x:count_distinct", "name": "my_name"}) alias_lookup.update(enriched.field_name_aliases) + # A ref qualified with a FOREIGN model (``owners.status`` when the base + # model is ``orders``) must not resolve to a same-named local column via + # the bare-name / ``_name`` lookups — that silently sorts by the wrong + # field. Only unqualified refs, or refs qualified with the base model + # itself, take the bare shortcuts; a foreign qualifier falls through to + # the ``.`` qualified match and then the joined-qualifier + # rejection below. Mirrors the typed pipeline's plan-time guard. + host_local = model_prefix == enriched.model_name + # Direct match on the user-provided name - if user_name in alias_lookup: - return alias_lookup[user_name] + if host_local and user_name in alias_lookup: + return _OrderColRef(alias_lookup[user_name], True, None, None) # Qualified match for cross-model measures: # col.model="customers", col.name="revenue_sum" → "customers.revenue_sum" if col.model: qualified = f"{col.model}.{col.name}" if qualified in alias_lookup: - return alias_lookup[qualified] + return _OrderColRef(alias_lookup[qualified], True, None, None) # Fallback for *:count → _count: user says "count", internal is "_count" prefixed = f"_{user_name}" - if prefixed in alias_lookup: - return alias_lookup[prefixed] - - # Fallback: qualify with model prefix - return f"{model_prefix}.{user_name}" + if host_local and prefixed in alias_lookup: + return _OrderColRef(alias_lookup[prefixed], True, None, None) + + # Fallback: a non-projected order key is only safe to emit as a split + # reference against the BASE-model alias. A joined qualifier (anything + # other than the base model) is rejected — even when a filter pulls the + # join into the base FROM, the outer-wrapping layers relocate the ORDER + # BY into a scope where the joined table is unbound. + if model_prefix != enriched.model_name: + raise UnresolvableOrderColumnError(column=user_name, qualifier=model_prefix) + return _OrderColRef(f"{model_prefix}.{user_name}", False, model_prefix, user_name) # ------------------------------------------------------------------ # FROM / JOIN building @@ -6083,10 +6158,24 @@ def _render_outer_composite(cslot) -> str: canonical_alias = canonical_alias_for_plan[plan.aggregate_slot_id] agg_col_alias = agg_col_alias_for_plan[plan.aggregate_slot_id] cte_name = _cte_name_from_alias("_cm_", canonical_alias) - public_aliases = self._public_aliases_for_cross_model_agg( - slot=agg_slot, - source_relation=source_relation, - canonical_alias=canonical_alias, + # DEV-1495 bug 2 / DEV-1712: an order-by-only (hidden) cross-model + # aggregate never surfaces in the combined projection — its CTE is + # still joined below, and the ORDER BY references it CTE-qualified + # (``hidden_cma_order_ref``). Trimming it keeps the outer SELECT to + # the user-declared columns (Law 2 projection boundary). Only when + # there is NO transform chain: a hidden CMA feeding a transform + # layer (``cumsum(customers.revenue:sum)``) must stay projected so + # the step CTE can consume it — the transform outer wrap does the + # public-vs-hidden trim in that path. + trim_hidden = plan.hidden and not planned_query.transform_layers + public_aliases = ( + [] + if trim_hidden + else self._public_aliases_for_cross_model_agg( + slot=agg_slot, + source_relation=source_relation, + canonical_alias=canonical_alias, + ) ) for pub in public_aliases: if pub == agg_col_alias: @@ -6196,6 +6285,22 @@ def _render_outer_composite(cslot) -> str: # level. ORDER BY columns must be qualified — ``_base`` columns # use ``_base."..."``, cross-model columns use the bare alias # (only present on one side). + # DEV-1712 / DEV-1495 bug 2: hidden (order-only) cross-model aggregates + # are trimmed from the projection above, so their ORDER BY term must be + # CTE-qualified (``_cm_*.""``) rather than the bare + # combined-SELECT alias. + hidden_cma_order_ref: Dict[str, str] = {} + for plan in planned_query.cross_model_aggregate_plans: + # Only CMAs actually trimmed from the projection (hidden + no + # transform chain) need the CTE-qualified ORDER BY reference. + if not (plan.hidden and not planned_query.transform_layers): + continue + _canon = canonical_alias_for_plan[plan.aggregate_slot_id] + _agg_col = agg_col_alias_for_plan[plan.aggregate_slot_id] + _cte = _cte_name_from_alias("_cm_", _canon) + hidden_cma_order_ref[plan.aggregate_slot_id] = ( + f'{_cte}.{self._quote_ident(_agg_col)}' + ) order_sql = self._build_combined_order_by_sql( planned_query=planned_query, slots_by_id=slots_by_id, @@ -6204,6 +6309,7 @@ def _render_outer_composite(cslot) -> str: bare_order_slot_ids=set(order_only_local_ids), outer_composite_aliases=outer_composite_order_alias_by_sid, outer_composite_expressions=outer_composite_order_expressions, + hidden_cma_order_ref=hidden_cma_order_ref, ) if order_sql: sql += "\n" + order_sql @@ -7559,6 +7665,7 @@ def _build_combined_order_by_sql( bare_order_slot_ids: Optional[Set[str]] = None, outer_composite_aliases: Optional[Dict[str, str]] = None, outer_composite_expressions: Optional[Dict[str, str]] = None, + hidden_cma_order_ref: Optional[Dict[str, str]] = None, ) -> Optional[str]: """Build the ORDER BY clause for the combined SELECT. @@ -7586,6 +7693,7 @@ def _build_combined_order_by_sql( bare_ids = bare_order_slot_ids or set() outer_aliases = outer_composite_aliases or {} outer_expressions = outer_composite_expressions or {} + hidden_cma_refs = hidden_cma_order_ref or {} parts: List[str] = [] for entry in planned_query.order: slot = slots_by_id.get(entry.slot_id) @@ -7600,6 +7708,7 @@ def _build_combined_order_by_sql( bare_ids=bare_ids, outer_aliases=outer_aliases, outer_expressions=outer_expressions, + hidden_cma_refs=hidden_cma_refs, ) if term is not None: parts.append(term) @@ -7618,6 +7727,7 @@ def _resolve_combined_order_term( bare_ids: Set[str], outer_aliases: Dict[str, str], outer_expressions: Optional[Dict[str, str]] = None, + hidden_cma_refs: Optional[Dict[str, str]] = None, ) -> Optional[str]: """Resolve one ``OrderEntry`` to its ``"alias" `` term. @@ -7633,6 +7743,12 @@ def _resolve_combined_order_term( """ direction = "ASC" if entry.direction == "asc" else "DESC" if entry.slot_id in cma_slot_ids: + # DEV-1712: a HIDDEN (order-only) cross-model aggregate is trimmed + # from the combined projection, so the bare alias no longer names a + # projected column — reference the CTE-qualified column instead. + hidden_ref = (hidden_cma_refs or {}).get(entry.slot_id) + if hidden_ref is not None: + return f'{hidden_ref} {direction}' alias = cm_alias_for_plan.get(entry.slot_id) if alias is None: return None @@ -10902,7 +11018,7 @@ def _apply_order_limit_from_planned( # NOSONAR(S3776) — per-order-entry slot- targets out of ``base_render_order``, preserving today's ``NotImplementedError``). """ - from slayer.core.keys import AggregateKey + from slayer.core.keys import AggregateKey, ColumnKey, ColumnSqlKey, TimeTruncKey for order_entry in planned_query.order: slot = slots_by_id.get(order_entry.slot_id) @@ -10929,18 +11045,85 @@ def _apply_order_limit_from_planned( # NOSONAR(S3776) — per-order-entry slot- self._ordered(order_col, ascending=ascending), ) continue - # Hidden ROW / transform / cross-model / composite ORDER - # targets aren't materialised in the local-only SELECT — - # preserved as NotImplementedError. DEV-1501's - # ``aggregates_only=True`` keeps hidden ROW targets out - # of ``base_render_order``; hidden composite-aggregate - # ORDER BY is rejected at the ``OrderItem`` input - # validation layer. + # DEV-1712 (Law 2, split emission): a hidden LOCAL ROW column + # ordered in an UNGROUPED query. The plan-time order validation + # (``plan_query``) guarantees the only hidden ROW slot that + # reaches here is a local (empty-path) column in a query with no + # GROUP BY — grouped row columns and joined columns are rejected + # up front, aggregates take the branch above. Emit a SPLIT + # ``.`` reference (mixed-case-aware) against + # the base FROM scope, identical to how the column would render + # if it were a projected dimension. + key = slot.key + row_key = key.column if isinstance(key, TimeTruncKey) else key + # A bare LOCAL column — split-emit the qualified column ref. + if ( + source_model is not None + and isinstance(row_key, ColumnKey) + and not row_key.path + ): + order_col = self._joined_or_local_dim_expr( + path=(), leaf=row_key.leaf, source_model=source_model, + source_relation=source_relation, bundle=bundle, + ) + ascending = order_entry.direction == "asc" + select = select.order_by( + self._ordered(order_col, ascending=ascending), + ) + continue + # A LOCAL DERIVED column (``ColumnSqlKey``, path empty): resolve + # its ``Column.sql`` through a throwaway host scope. That both + # anchors the expansion AND surfaces whether the SQL crosses a + # join. A hidden order-only derived column is NOT projected, so + # its join was never pulled into the base FROM — ordering on it + # would reference an unbound table. Reject that (project it), + # rather than emit invalid SQL; a non-crossing derived column + # (e.g. a bare mixed-case identifier) orders on its expression. + if ( + source_model is not None + and bundle is not None + and isinstance(row_key, ColumnSqlKey) + and not row_key.path + ): + # Detect join crossing via a throwaway scope (register-only); + # the resolved expr is discarded — its expansion lacks the + # DEV-1645 mixed-case quoting the planned-dim helper applies. + allocator = self._new_allocator() + scope = ScopeFrame( + scope_id=allocator.next_scope_id(source_relation), + root_model=source_model, + root_relation=source_relation, + bundle=bundle, + dialect=self._dialect, + allocator=allocator, + ) + scope.resolve(row_key) + if scope.join_paths: + # The derived column IS local (``orders.cust_region``); + # it merely depends on an unpulled join. Report its own + # qualified name, not a fabricated ``customers.cust_region``. + raise UnresolvableOrderColumnError( + column=row_key.column_name, qualifier=source_relation, + ) + # Non-crossing local derived column — emit through the + # planned-dim helper so the expansion is quoted identically + # to a projected dimension (mixed-case-safe). + order_col = self._joined_or_local_dim_expr( + path=(), leaf=row_key.column_name, + source_model=source_model, + source_relation=source_relation, bundle=bundle, + ) + ascending = order_entry.direction == "asc" + select = select.order_by( + self._ordered(order_col, ascending=ascending), + ) + continue + # Defensive: any other hidden shape should have been rejected at + # plan time (transform / composite / joined / grouped-row). raise NotImplementedError( - f"DEV-1450 stage 7b.10+: ORDER BY references a " - f"hidden slot (id={slot.id!r}, key=" - f"{type(slot.key).__name__}) not materialised in " - f"the local-only SELECT. Deferred to a later slice." + f"ORDER BY references a hidden slot (id={slot.id!r}, key=" + f"{type(slot.key).__name__}) that was not resolved at plan " + f"time — this is an internal invariant violation." ) # DEV-1713: resolve to the SAME full alias the projection emits — # a joined ROW dimension projects under the DOTTED result key diff --git a/tests/parity_xfails.py b/tests/parity_xfails.py index 25fef025..5b24e8d6 100644 --- a/tests/parity_xfails.py +++ b/tests/parity_xfails.py @@ -9,20 +9,9 @@ Keyed by exact pytest node id so strict-xfail can never mask a passing test. """ -PARITY_XFAILS: dict[str, str] = { - 'tests/test_dev1645_invalid_postgres_sql.py::TestFlavorAOrderByUnprojected::test_orderby_nonprojected_column_emits_split_not_composite': "DEV-1712 (Stage 8): DEV-1645 ORDER BY policies (unprojected/joined sort keys).", - 'tests/test_dev1645_invalid_postgres_sql.py::TestFlavorAOrderByUnprojected::test_orderby_nonprojected_mixed_case_column_split_and_quoted': "DEV-1712 (Stage 8): DEV-1645 ORDER BY policies (unprojected/joined sort keys).", - 'tests/test_dev1645_invalid_postgres_sql.py::TestFlavorAOrderByUnprojected::test_orderby_split_key_keeps_asc_limit_offset': "DEV-1712 (Stage 8): DEV-1645 ORDER BY policies (unprojected/joined sort keys).", - 'tests/test_dev1645_invalid_postgres_sql.py::TestFlavorAOrderByUnprojected::test_orderby_unresolvable_joined_column_rejected': "DEV-1712 (Stage 8): DEV-1645 ORDER BY policies (unprojected/joined sort keys).", - 'tests/test_dev1645_invalid_postgres_sql.py::TestFlavorAOrderByUnprojected::test_orderby_joined_column_rejected_even_when_filter_pulls_join_in': "DEV-1712 (Stage 8): DEV-1645 ORDER BY policies (unprojected/joined sort keys).", - 'tests/test_dev1645_invalid_postgres_sql.py::TestFlavorAOrderByUnprojected::test_orderby_joined_column_rejected_in_cte_wrapped_scope': "DEV-1712 (Stage 8): DEV-1645 ORDER BY policies (unprojected/joined sort keys).", - 'tests/test_dev1645_invalid_postgres_sql.py::TestFlavorAOrderByUnprojected::test_orderby_joined_column_rejected_in_first_last_ranked_scope': "DEV-1712 (Stage 8): DEV-1645 ORDER BY policies (unprojected/joined sort keys).", - # NOTE: DEV-1645 mixed-case *identifier* quoting (Flavor B) landed early in - # DEV-1706 Stage 2 — it is a hard dependency of the DEV-1686 reserved-word - # fix (a reserved-model join key such as ``grant.merchantId`` must emit - # ``"grant"."merchantId"``). Those pins were removed here; the DEV-1645 - # ORDER-BY *placement* policies above remain for Stage 8 (DEV-1712). - # Flavor-B mixed-case identifier execution — landed in DEV-1706 Stage 2 - # (see the note above); un-pinned. Flavor-A ORDER-BY stays for Stage 8. - 'tests/integration/test_integration_postgres.py::TestDev1645ValidPostgres::test_flavor_a_orderby_nonprojected_column_executes': "DEV-1712 (Stage 8): DEV-1645 ORDER BY unprojected column policy.", -} +# DEV-1712 (Stage 8) landed the final entries — the DEV-1645 Flavor-A ORDER BY +# placement policies (split-not-composite for unprojected sort keys; +# UnresolvableOrderColumnError for joined/unresolvable sort keys) and the typed +# hidden-slot / partition_by validations. With those absorbed the registry is +# empty, which is the DEV-1485 (Stage 11) end-state the gate checks for. +PARITY_XFAILS: dict[str, str] = {} diff --git a/tests/test_dev1712_order_only_hidden_slots.py b/tests/test_dev1712_order_only_hidden_slots.py new file mode 100644 index 00000000..f5155569 --- /dev/null +++ b/tests/test_dev1712_order_only_hidden_slots.py @@ -0,0 +1,854 @@ +"""DEV-1712 Stage 8 — order-only hidden slots + plan-time validations. + +Covers the typed-pipeline contract for ORDER BY refs that are NOT declared +as dimensions/measures, plus the ``rank(..., partition_by=X)`` grain guard. +The behaviour table (a ref reaching ORDER BY that matches no declared/public +slot): + + target (hidden) | has_grouping = False (raw rows) | has_grouping = True (grouped) + ------------------------ | ------------------------------- | ---------------------------------- + local aggregate | works (agg induces grouping) | materialise hidden, order, strip + cross-model aggregate | works | hidden CMA plan, trimmed, CTE order + local row column | split emission (orders.col) | ValueError (add to dims / order agg) + joined row column | UnresolvableOrderColumnError | UnresolvableOrderColumnError + transform (change/…) | ValueError -> declare (DEV-1733)| same + +``has_grouping`` = any aggregating measure OR (dims/time-dims present AND +``distinct_dimension_values``). An order-only aggregate never needs +rejection — it always materialises and orders. Composite arithmetic +(``a:sum / b:sum``) is not expressible as an ``OrderItem.column`` at all +(Pydantic rejects it), pinned below. + +Refs: DEV-1712 (this stage), DEV-1472 (order-only hidden slots), DEV-1495 +bug 2 (cross-model leak — the canonical repro lives in +``tests/test_projection_trim.py``), DEV-1497 (partition_by guard), DEV-1645 +(split-not-composite ORDER BY), DEV-1733 (deferred transform/composite +order targets). +""" +from __future__ import annotations + +import re +import sqlite3 + +import pydantic +import pytest +import sqlglot +from sqlglot import exp + +from slayer.core.enums import DataType +from slayer.core.errors import UnresolvableOrderColumnError +from slayer.core.models import Column, DatasourceConfig, ModelJoin, ModelMeasure, SlayerModel +from slayer.core.query import ColumnRef, OrderItem, SlayerQuery, TimeDimension +from slayer.engine.query_engine import SlayerQueryEngine +from slayer.storage.yaml_storage import YAMLStorage + + +# --------------------------------------------------------------------------- +# SQL-inspection helpers (implementation-agnostic — walk the AST). +# --------------------------------------------------------------------------- +def _outermost_select(sql: str, *, dialect: str = "postgres") -> exp.Select: + parsed = sqlglot.parse_one(sql, dialect=dialect) + assert isinstance(parsed, exp.Select), f"not a SELECT:\n{sql}" + return parsed + + +def _outer_select_columns(sql: str, *, dialect: str = "postgres") -> list[str]: + """Alias names projected by the OUTERMOST SELECT.""" + parsed = _outermost_select(sql, dialect=dialect) + return [proj.alias_or_name for proj in parsed.expressions] + + +def _all_aliases_in_sql(sql: str) -> list[str]: + return re.findall(r'"([^"]+)"', sql) + + +def _outer_order_by_columns(sql: str, *, dialect: str = "postgres") -> list[tuple[str, str]]: + """Return ``(table, name)`` for each Column in the outermost ORDER BY. + + A SPLIT reference (``orders.created_at``) parses to a Column with + ``table='orders'`` and ``name='created_at'`` (no dot in the name). A + COMPOSITE reference (the buggy ``"orders.created_at"`` single quoted + token) parses to a Column with ``table=''`` and ``name`` containing a + dot. The distinction is exactly the DEV-1645 Flavor-A fix. + """ + parsed = _outermost_select(sql, dialect=dialect) + order = parsed.args.get("order") + if order is None: + return [] + out: list[tuple[str, str]] = [] + for ordered in order.expressions: + col = ordered.this + if isinstance(col, exp.Column): + out.append((col.table, col.name)) + return out + + +def _outer_order_by_names(sql: str, *, dialect: str = "postgres") -> list[str]: + """Leaf identifier names referenced by the outermost ORDER BY.""" + parsed = _outermost_select(sql, dialect=dialect) + order = parsed.args.get("order") + if order is None: + return [] + names: list[str] = [] + for ordered in order.expressions: + col = ordered.this + names.append(col.name if isinstance(col, exp.Column) else col.sql(dialect=dialect)) + return names + + +# --------------------------------------------------------------------------- +# Fixtures — orders -> customers -> regions. +# --------------------------------------------------------------------------- +async def _save_models(storage: YAMLStorage) -> SlayerModel: + await storage.save_model(SlayerModel( + name="regions", sql_table="regions", data_source="test", + columns=[ + Column(name="id", sql="id", type=DataType.INT, primary_key=True), + Column(name="name", sql="name", type=DataType.TEXT), + ], + )) + await storage.save_model(SlayerModel( + name="customers", sql_table="customers", data_source="test", + columns=[ + Column(name="id", sql="id", type=DataType.INT, primary_key=True), + Column(name="region_id", sql="region_id", type=DataType.INT), + Column(name="region", sql="region", type=DataType.TEXT), + Column(name="revenue", sql="lifetime_revenue", type=DataType.DOUBLE), + ], + joins=[ModelJoin(target_model="regions", join_pairs=[["region_id", "id"]])], + )) + # Second join target exposing the SAME leaf names (``region`` / ``revenue``) + # as customers — used for the diamond same-leaf/same-agg cross-model test + # (target-path identity) and the two-same-leaf-dims path. + await storage.save_model(SlayerModel( + name="suppliers", sql_table="suppliers", data_source="test", + columns=[ + Column(name="id", sql="id", type=DataType.INT, primary_key=True), + Column(name="region", sql="region", type=DataType.TEXT), + Column(name="revenue", sql="supplier_revenue", type=DataType.DOUBLE), + ], + )) + orders = SlayerModel( + name="orders", sql_table="orders", data_source="test", + default_time_dimension="created_at", + columns=[ + Column(name="id", sql="id", type=DataType.INT, primary_key=True), + Column(name="customer_id", sql="customer_id", type=DataType.INT), + Column(name="supplier_id", sql="supplier_id", type=DataType.INT), + Column(name="status", sql="status", type=DataType.TEXT), + Column(name="created_at", sql="created_at", type=DataType.TIMESTAMP), + # Mixed-case physical identifier — exercises DEV-1645 split-quoting. + Column(name="activity_ts", sql="ActivityTs", type=DataType.TIMESTAMP), + Column(name="amount", sql="amount", type=DataType.DOUBLE), + # DEV-1503 host-rooted isolation: a local aggregate source whose + # ``filter`` references a joined table. + Column(name="flagged_amount", sql="amount", type=DataType.DOUBLE, + filter="customers.region = 'West'"), + # A LOCAL derived column whose ``sql`` crosses the customers join — + # ``ColumnSqlKey(path=())`` but its expansion pulls a join. + Column(name="cust_region", sql="customers.region", type=DataType.TEXT), + ], + joins=[ + ModelJoin(target_model="customers", join_pairs=[["customer_id", "id"]]), + ModelJoin(target_model="suppliers", join_pairs=[["supplier_id", "id"]]), + ], + ) + await storage.save_model(orders) + return orders + + +@pytest.fixture +async def engine(tmp_path): + storage = YAMLStorage(base_dir=str(tmp_path)) + await storage.save_datasource(DatasourceConfig(name="test", type="sqlite", database=":memory:")) + orders = await _save_models(storage) + return SlayerQueryEngine(storage=storage), orders + + +async def _sql(engine_and_model, query: SlayerQuery) -> str: + engine, _ = engine_and_model + resp = await engine.execute(query, dry_run=True) + return resp.sql or "" + + +@pytest.fixture +async def exec_engine(tmp_path): + """On-disk SQLite seeded for execution / top-N ordering assertions.""" + db_path = tmp_path / "t.db" + conn = sqlite3.connect(str(db_path)) + conn.executescript( + """ + CREATE TABLE regions (id INTEGER PRIMARY KEY, name TEXT); + CREATE TABLE customers (id INTEGER PRIMARY KEY, region_id INTEGER, + region TEXT, lifetime_revenue REAL); + CREATE TABLE suppliers (id INTEGER PRIMARY KEY, region TEXT, + supplier_revenue REAL); + CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, + supplier_id INTEGER, status TEXT, created_at TEXT, + ActivityTs TEXT, amount REAL); + INSERT INTO regions VALUES (10,'West'),(11,'East'); + INSERT INTO customers VALUES (100,10,'West',500.0),(101,11,'East',700.0); + INSERT INTO suppliers VALUES (200,'West',1000.0),(201,'East',2000.0); + INSERT INTO orders VALUES + (1,100,200,'paid','2025-01-01','2025-01-05',10.0), + (2,100,200,'paid','2025-01-02','2025-01-06',40.0), + (3,101,201,'open','2025-02-01','2025-02-05',20.0), + (4,101,201,'open','2025-02-02','2025-02-06',5.0); + """ + ) + conn.commit() + conn.close() + storage = YAMLStorage(base_dir=str(tmp_path / "store")) + await storage.save_datasource( + DatasourceConfig(name="test", type="sqlite", database=str(db_path)) + ) + await _save_models(storage) + return SlayerQueryEngine(storage=storage) + + +# =========================================================================== +# Group 1 — local row column, ungrouped (dedup off) -> SPLIT emission. +# =========================================================================== +class TestUngroupedRowColumnSplit: + async def test_ungrouped_row_column_order_emits_split(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + distinct_dimension_values=False, + order=[OrderItem(column=ColumnRef(name="created_at"), direction="desc")], + ) + sql = await _sql(engine, query) + cols = _outer_order_by_columns(sql) + assert cols, f"no ORDER BY column found.\nSQL:\n{sql}" + table, name = cols[0] + # SPLIT reference: qualified by the base table, leaf is a single + # identifier — NOT the composite ``"orders.created_at"`` token. + split_msg = f"ORDER BY must be the SPLIT reference orders.created_at, got '{table}.{name}'.\nSQL:\n{sql}" + assert table == "orders", split_msg + assert name == "created_at", split_msg + assert "." not in name, split_msg + assert '"orders.created_at"' not in sql, f"composite token leaked.\nSQL:\n{sql}" + + async def test_ungrouped_mixed_case_row_column_split_quoted(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + distinct_dimension_values=False, + order=[OrderItem(column=ColumnRef(name="activity_ts"), direction="desc")], + ) + sql = await _sql(engine, query) + # The mixed-case physical identifier must be quoted (survive Postgres + # case-folding) and referenced split (qualifier + quoted leaf) in the + # ORDER BY itself — not merely present somewhere in the SQL. + assert '"ActivityTs"' in sql, f"mixed-case leaf must be quoted.\nSQL:\n{sql}" + cols = _outer_order_by_columns(sql) + assert cols, f"no ORDER BY column found.\nSQL:\n{sql}" + table, name = cols[0] + mc_msg = f"ORDER BY must be the split reference orders.\"ActivityTs\", got '{table}.{name}'.\nSQL:\n{sql}" + assert table == "orders", mc_msg + assert name == "ActivityTs", mc_msg + + async def test_split_key_preserves_asc_and_limit(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + distinct_dimension_values=False, + order=[OrderItem(column=ColumnRef(name="created_at"), direction="asc")], + limit=10, + ) + sql = await _sql(engine, query) + parsed = _outermost_select(sql) + order = parsed.args.get("order") + assert order is not None + ordered = order.expressions[0] + assert not ordered.args.get("desc"), f"expected ASC.\nSQL:\n{sql}" + assert "LIMIT 10" in sql.upper() or parsed.args.get("limit") is not None + + +# =========================================================================== +# Group 2 — grouped row column -> plan-time ValueError (D-B). +# =========================================================================== +class TestGroupedRowColumnRejected: + async def test_dedup_on_row_column_order_raises(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], # dedup ON (default) -> GROUP BY + order=[OrderItem(column=ColumnRef(name="created_at"), direction="desc")], + ) + with pytest.raises(ValueError) as ei: + await _sql(engine, query) + msg = str(ei.value).lower() + assert "created_at" in msg + assert "dimension" in msg or "aggregate" in msg, ( + f"error should guide the user to project it or order by an " + f"aggregate. got: {ei.value}" + ) + + async def test_measures_present_row_column_order_raises(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="amount:sum")], + order=[OrderItem(column=ColumnRef(name="created_at"), direction="desc")], + ) + with pytest.raises(ValueError): + await _sql(engine, query) + + +# =========================================================================== +# Group 3 — joined row column -> UnresolvableOrderColumnError. +# =========================================================================== +class TestJoinedRowColumnRejected: + async def test_joined_row_column_ungrouped_raises(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + distinct_dimension_values=False, + order=[OrderItem(column=ColumnRef(name="customers.region"), direction="desc")], + ) + with pytest.raises(UnresolvableOrderColumnError): + await _sql(engine, query) + + async def test_joined_row_column_grouped_raises(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="*:count")], + order=[OrderItem(column=ColumnRef(name="customers.region"), direction="desc")], + ) + with pytest.raises(UnresolvableOrderColumnError): + await _sql(engine, query) + + async def test_joined_reject_message_does_not_duplicate_qualifier(self, engine) -> None: + """The rejection message names the column once (``customers.region``), + not a duplicated ``customers.customers.region`` (CodeRabbit).""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + distinct_dimension_values=False, + order=[OrderItem(column=ColumnRef(name="customers.region"), direction="desc")], + ) + with pytest.raises(UnresolvableOrderColumnError) as ei: + await _sql(engine, query) + msg = str(ei.value) + assert "customers.region" in msg + assert "customers.customers" not in msg + + async def test_ungrouped_order_by_derived_crossing_column_rejected(self, engine) -> None: + """A hidden order-only LOCAL DERIVED column whose ``Column.sql`` crosses + a join (``cust_region`` = ``customers.region``) is not projected, so its + join is never pulled into the base FROM. Ordering on it must be rejected + rather than emit an unbound ``ORDER BY`` (CodeRabbit / T3).""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + distinct_dimension_values=False, + order=[OrderItem(column=ColumnRef(name="cust_region"), direction="desc")], + ) + with pytest.raises(UnresolvableOrderColumnError): + await _sql(engine, query) + + async def test_joined_order_ref_colliding_local_leaf_raises(self, tmp_path) -> None: + """A joined order ref whose LEAF collides with a local declared + dimension (``owners.status`` vs a local ``status``) must NOT silently + bind to the local column and sort by the wrong field — it is a joined + ref and is rejected (Codex / DEV-1712).""" + storage = YAMLStorage(base_dir=str(tmp_path)) + await storage.save_datasource( + DatasourceConfig(name="test", type="sqlite", database=":memory:") + ) + await storage.save_model(SlayerModel( + name="owners", sql_table="owners", data_source="test", + columns=[ + Column(name="id", sql="id", type=DataType.INT, primary_key=True), + Column(name="status", sql="status", type=DataType.TEXT), + ], + )) + await storage.save_model(SlayerModel( + name="orders", sql_table="orders", data_source="test", + columns=[ + Column(name="id", sql="id", type=DataType.INT, primary_key=True), + Column(name="owner_id", sql="owner_id", type=DataType.INT), + Column(name="status", sql="status", type=DataType.TEXT), + Column(name="amount", sql="amount", type=DataType.DOUBLE), + ], + joins=[ModelJoin(target_model="owners", join_pairs=[["owner_id", "id"]])], + )) + engine = SlayerQueryEngine(storage=storage) + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], # local status + measures=[ModelMeasure(formula="*:count")], + order=[OrderItem(column="owners.status", direction="desc")], # joined + ) + with pytest.raises(UnresolvableOrderColumnError): + resp = await engine.execute(query, dry_run=True) + # If it did not raise, it must at least NOT have silently sorted by + # the local column (which would be the bug). + assert "owners" in (resp.sql or ""), ( + f"joined order ref silently bound to the local column.\n" + f"SQL:\n{resp.sql}" + ) + + +# =========================================================================== +# Group 4 — local aggregate hidden order (regression; must stay green). +# =========================================================================== +class TestLocalAggregateHiddenOrder: + async def test_local_aggregate_order_only_hidden_and_ordered(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="*:count")], + order=[OrderItem(column="amount:sum", direction="desc")], + limit=3, + ) + sql = await _sql(engine, query) + outer = _outer_select_columns(sql) + assert outer == ["orders.status", "orders._count"], ( + f"hidden amount_sum must not project.\ngot: {outer}\nSQL:\n{sql}" + ) + assert "orders.amount_sum" in _all_aliases_in_sql(sql) + assert _outer_order_by_names(sql) == ["orders.amount_sum"], ( + f"outer ORDER BY must name the hidden alias.\nSQL:\n{sql}" + ) + + async def test_ungrouped_input_local_aggregate_order_still_works(self, engine) -> None: + """has_grouping=False INPUT (dedup off, no measures) but the ORDER BY + target is a local aggregate: the aggregate induces GROUP BY on the + declared dims, so it materialises + orders like any grouped query — + it never falls into the row-column split/reject branch.""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + distinct_dimension_values=False, + order=[OrderItem(column="amount:sum", direction="desc")], + ) + sql = await _sql(engine, query) + assert _outer_select_columns(sql) == ["orders.status"], f"SQL:\n{sql}" + assert _outer_order_by_names(sql) == ["orders.amount_sum"], f"SQL:\n{sql}" + + async def test_dev1472_joined_dim_repro(self, engine) -> None: + """The literal DEV-1472 repro — joined dim + *:count + order by a + local aggregate. Already valid single-stage; pinned as regression.""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="customers.region")], + measures=[ModelMeasure(formula="*:count")], + order=[OrderItem(column="amount:sum")], + limit=10, + ) + sql = await _sql(engine, query) + outer = _outer_select_columns(sql) + assert outer == ["orders.customers.region", "orders._count"], ( + f"got: {outer}\nSQL:\n{sql}" + ) + assert "orders.amount_sum" in _all_aliases_in_sql(sql) + + async def test_order_matches_declared_measure_no_hidden_slot(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="amount:sum", name="rev")], + order=[OrderItem(column="rev", direction="desc")], + limit=5, + ) + sql = await _sql(engine, query) + # Declared alias -> composite quoted reference is correct (it IS the + # projected output column ``AS "orders.rev"``); no outer trim wrapper. + assert '"orders.rev"' in sql + assert _outer_select_columns(sql) == ["orders.status", "orders.rev"] + + async def test_declared_transform_plus_hidden_agg_order(self, engine) -> None: + """A declared rank() transform plus an order-only hidden aggregate: + the hidden aggregate threads through the transform chain and orders + at the outer wrap. Codex F7 positive case — declared transforms must + NOT disable a valid hidden aggregate order target.""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="rank(amount:sum)", name="rk")], + order=[OrderItem(column="id:count", direction="desc")], + ) + sql = await _sql(engine, query) + outer = _outer_select_columns(sql) + assert outer == ["orders.status", "orders.rk"], ( + f"hidden id_count must not project.\ngot: {outer}\nSQL:\n{sql}" + ) + assert _outer_order_by_names(sql) == ["orders.id_count"] + + +# =========================================================================== +# Group 5 — execution: top-N by a hidden metric + response-column strip. +# =========================================================================== +class TestExecutionHiddenOrder: + async def test_execute_top_n_by_hidden_aggregate(self, exec_engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="*:count")], + order=[OrderItem(column="amount:sum", direction="desc")], + ) + resp = await exec_engine.execute(query) + # paid: 10+40=50, open: 20+5=25 -> paid first (desc). + statuses = [r["orders.status"] for r in resp.data] + assert statuses == ["paid", "open"], f"rows: {resp.data}" + # Hidden order slot must be stripped from the response columns. + assert "orders.amount_sum" not in resp.columns, f"columns: {resp.columns}" + assert all("amount_sum" not in k for k in resp.data[0]), f"row: {resp.data[0]}" + + async def test_hidden_order_slot_absent_from_attributes(self, exec_engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="*:count")], + order=[OrderItem(column="amount:sum", direction="desc")], + ) + resp = await exec_engine.execute(query) + keys = set(resp.attributes.dimensions) | set(resp.attributes.measures) + assert not any("amount_sum" in k for k in keys), f"attributes leaked: {keys}" + + +# =========================================================================== +# Group 6 — cross-model hidden aggregate order (DEV-1495 bug 2 neighbourhood). +# =========================================================================== +class TestCrossModelHiddenOrder: + async def test_two_hidden_cross_model_aggs_distinct_ctes(self, engine) -> None: + """Two order-only cross-model aggregates over the same leaf but + different aggregations (``customers.revenue:sum`` and + ``customers.revenue:max``) get DISTINCT hidden CTE aliases; neither + leaks into the outer projection; the outer ORDER BY names them + unambiguously.""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + order=[ + OrderItem(column="customers.revenue:sum", direction="desc"), + OrderItem(column="customers.revenue:max", direction="asc"), + ], + limit=3, + ) + sql = await _sql(engine, query) + assert _outer_select_columns(sql) == ["orders.status"], ( + f"cross-model aggregates must be hidden.\nSQL:\n{sql}" + ) + aliases = _all_aliases_in_sql(sql) + assert any("revenue_sum" in a for a in aliases), f"aliases: {aliases}\n{sql}" + assert any("revenue_max" in a for a in aliases), f"aliases: {aliases}\n{sql}" + order_names = _outer_order_by_names(sql) + assert any("revenue_sum" in n for n in order_names), order_names + assert any("revenue_max" in n for n in order_names), order_names + # No inline aggregate call in the outermost ORDER BY. + parsed = _outermost_select(sql) + order = parsed.args.get("order") + if order is not None: + assert "SUM(" not in order.sql(dialect="postgres").upper() + + async def test_cross_model_hidden_order_postgres_shape(self, engine) -> None: + """Postgres-dialect shape (F9): the hidden cross-model aggregate lives + in a CTE, is absent from the outer projection, and the ORDER BY names + its alias with no inline aggregate.""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + order=[OrderItem(column="customers.revenue:sum", direction="desc")], + limit=3, + ) + engine_obj, _ = engine + resp = await engine_obj.execute(query, dry_run=True) + sql = resp.sql or "" + # sanity: parses as postgres + sqlglot.parse_one(sql, dialect="postgres") + assert "revenue_sum" in "".join(_all_aliases_in_sql(sql)) + assert _outer_select_columns(sql) == ["orders.status"] + + async def test_diamond_same_leaf_same_agg_distinct_targets(self, engine) -> None: + """Diamond: two DIFFERENT join targets (``customers`` and + ``suppliers``) exposing the SAME leaf under the SAME aggregation + (``customers.revenue:sum`` vs ``suppliers.revenue:sum``). Target-path + identity must keep them DISTINCT — distinct CTE aliases, both hidden, + both named unambiguously in the ORDER BY. (Codex F4/F6: guards against + an AggregateKey identity that omits the target path.)""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + order=[ + OrderItem(column="customers.revenue:sum", direction="desc"), + OrderItem(column="suppliers.revenue:sum", direction="asc"), + ], + limit=3, + ) + sql = await _sql(engine, query) + assert _outer_select_columns(sql) == ["orders.status"], ( + f"both cross-model aggregates must be hidden.\nSQL:\n{sql}" + ) + aliases = "\n".join(_all_aliases_in_sql(sql)) + assert "customers.revenue_sum" in aliases or "customers__revenue_sum" in aliases, aliases + assert "suppliers.revenue_sum" in aliases or "suppliers__revenue_sum" in aliases, aliases + order_names = _outer_order_by_names(sql) + assert any("customers" in n and "revenue_sum" in n for n in order_names), order_names + assert any("suppliers" in n and "revenue_sum" in n for n in order_names), order_names + + async def test_ungrouped_input_cross_model_aggregate_order_hidden(self, engine) -> None: + """has_grouping=False INPUT (dedup off) ordering by a cross-model + aggregate: still hidden, never projected.""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + distinct_dimension_values=False, + order=[OrderItem(column="customers.revenue:sum", direction="desc")], + ) + sql = await _sql(engine, query) + assert _outer_select_columns(sql) == ["orders.status"], f"SQL:\n{sql}" + + async def test_execute_cross_model_hidden_order_stripped(self, exec_engine) -> None: + """Execution: an order-only cross-model aggregate is absent from the + response columns, row keys, and attribute metadata — pinning the + hidden CrossModelAggregatePlan (hidden=True / public_alias=None) path + at the response level, distinct from the local-aggregate path.""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="*:count")], + order=[OrderItem(column="customers.revenue:sum", direction="desc")], + ) + resp = await exec_engine.execute(query) + assert all("revenue_sum" not in c for c in resp.columns), f"columns: {resp.columns}" + assert all("revenue_sum" not in k for k in resp.data[0]), f"row: {resp.data[0]}" + keys = set(resp.attributes.dimensions) | set(resp.attributes.measures) + assert not any("revenue_sum" in k for k in keys), f"attributes leaked: {keys}" + + +# =========================================================================== +# Group 7 — partition_by grain guard (DEV-1497). +# =========================================================================== +class TestPartitionByGuard: + async def test_partition_by_bare_dim_accepted(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="rank(amount:sum, partition_by=status)", name="rk")], + ) + sql = await _sql(engine, query) + assert "PARTITION BY" in sql.upper() + # grain must NOT widen: the only GROUP BY key is status. + assert "customer_id" not in sql + + async def test_partition_by_qualified_dim_accepted(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="rank(amount:sum, partition_by=orders.status)", name="rk")], + ) + sql = await _sql(engine, query) + assert "PARTITION BY" in sql.upper() + + async def test_partition_by_dotted_joined_dim_accepted(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="customers.region")], + measures=[ModelMeasure( + formula="rank(amount:sum, partition_by=customers.region)", name="rk")], + ) + sql = await _sql(engine, query) + assert "PARTITION BY" in sql.upper() + + async def test_partition_by_time_dimension_uses_bucket(self, engine) -> None: + """partition_by naming a query time-dimension must partition by the + TRUNCATED bucket, not the raw column — and must not add the raw + column to GROUP BY (grain widening) nor emit a duplicate alias.""" + query = SlayerQuery( + source_model="orders", + time_dimensions=[TimeDimension(dimension="created_at", granularity="month")], + measures=[ModelMeasure( + formula="rank(amount:sum, partition_by=created_at)", name="rk")], + ) + sql = await _sql(engine, query) + # (a) No column in the base SELECT is a raw bare-column projection of + # created_at — the DEV-1497 grain-widening signature was a second + # ``orders.created_at AS "orders.created_at"`` projection (raw column) + # alongside the STRFTIME/DATE_TRUNC bucket, plus a raw GROUP BY key. + tree = sqlglot.parse_one(sql, dialect="sqlite") + base_cte0 = next( + (cte.this for cte in tree.find_all(exp.CTE) if cte.alias == "base"), None + ) + assert base_cte0 is not None, f"expected a `base` CTE.\nSQL:\n{sql}" + raw_created_projections = [ + p for p in base_cte0.expressions + if isinstance(p.unalias(), exp.Column) and p.unalias().name == "created_at" + ] + assert not raw_created_projections, ( + f"base SELECT projects the RAW created_at column alongside the " + f"bucket — grain widened.\nSQL:\n{sql}" + ) + # (b) The base GROUP BY must contain a TRUNCATION expression for + # created_at (the bucket), never a bare column reference to the raw + # timestamp. On SQLite that truncation is STRFTIME / DATE. + base_cte = next( + (cte.this for cte in tree.find_all(exp.CTE) if cte.alias == "base"), None + ) + assert base_cte is not None, f"expected a `base` CTE.\nSQL:\n{sql}" + group = base_cte.args.get("group") + assert group is not None, f"base has no GROUP BY.\nSQL:\n{sql}" + group_exprs = group.expressions + # No GROUP BY key may be a bare column whose name is the raw created_at. + for g in group_exprs: + assert not (isinstance(g, exp.Column) and g.name == "created_at"), ( + f"GROUP BY includes the RAW created_at column — grain widened " + f"past the month bucket.\nSQL:\n{sql}" + ) + # (c) The RANK() PARTITION BY must reference the bucket alias, not add + # an independent partition column. + window = next(iter(tree.find_all(exp.Window)), None) + assert window is not None, f"no window function found.\nSQL:\n{sql}" + partition = window.args.get("partition_by") or [] + assert partition, f"RANK() has no PARTITION BY.\nSQL:\n{sql}" + part_sql = " ".join(p.sql(dialect="sqlite") for p in partition) + assert "created_at" in part_sql, f"PARTITION BY lost the bucket.\nSQL:\n{sql}" + + async def test_partition_by_non_dim_raises_rank(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ + ModelMeasure(formula="amount:sum"), + ModelMeasure(formula="rank(amount:sum, partition_by=customer_id)", name="rk"), + ], + ) + with pytest.raises(ValueError) as ei: + await _sql(engine, query) + msg = str(ei.value) + assert "partition_by" in msg + assert "customer_id" in msg + assert "rank" in msg.lower(), f"error should name the transform: {msg}" + assert "status" in msg, "error should list available dimensions" + + async def test_partition_by_ambiguous_time_dim_granularity_raises(self, engine) -> None: + """A source column carried by two time-dimension granularities + (``created_at`` at month AND day) makes a bare ``partition_by=created_at`` + ambiguous — reject rather than silently pick a bucket (CodeRabbit / T1).""" + query = SlayerQuery( + source_model="orders", + time_dimensions=[ + TimeDimension(dimension="created_at", granularity="month"), + TimeDimension(dimension="created_at", granularity="day"), + ], + measures=[ModelMeasure( + formula="rank(amount:sum, partition_by=created_at)", name="rk")], + ) + with pytest.raises(ValueError) as ei: + await _sql(engine, query) + assert "ambiguous" in str(ei.value).lower() + + async def test_partition_by_same_granularity_time_dim_not_flagged_ambiguous(self, engine) -> None: + """Two time-dimension declarations at the SAME granularity are one + bucket, not competing granularities — the partition-ambiguity guard must + NOT misfire (Codex round 4). The query is still invalid (the two + ``created_at`` columns collide on their downstream name), but the error + must be that collision, not a bogus 'ambiguous granularities'.""" + query = SlayerQuery( + source_model="orders", + time_dimensions=[ + TimeDimension(dimension="created_at", granularity="month", label="A"), + TimeDimension(dimension="created_at", granularity="month", label="B"), + ], + measures=[ModelMeasure( + formula="rank(amount:sum, partition_by=created_at)", name="rk")], + ) + with pytest.raises(ValueError) as ei: + await _sql(engine, query) + assert "ambiguous" not in str(ei.value).lower() + + async def test_partition_by_non_dim_raises_ntile(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure( + formula="ntile(amount:sum, n=4, partition_by=customer_id)", name="q")], + ) + with pytest.raises(ValueError) as ei: + await _sql(engine, query) + msg = str(ei.value) + assert "customer_id" in msg + assert "ntile" in msg.lower(), f"error should name the transform: {msg}" + + +# =========================================================================== +# Group 8 — transform / composite order targets: deferred (DEV-1733). +# =========================================================================== +class TestTransformOrderDeferred: + async def test_transform_order_only_raises_actionable(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + time_dimensions=[TimeDimension(dimension="created_at", granularity="month")], + measures=[ModelMeasure(formula="amount:sum")], + order=[OrderItem(column="change(amount:sum)", direction="desc")], + ) + with pytest.raises(ValueError) as ei: + await _sql(engine, query) + msg = str(ei.value).lower() + assert "measure" in msg, f"error should tell the user to declare it as a measure: {ei.value}" + + async def test_cumsum_order_only_raises(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + time_dimensions=[TimeDimension(dimension="created_at", granularity="month")], + measures=[ModelMeasure(formula="amount:sum")], + order=[OrderItem(column="cumsum(amount:sum)", direction="desc")], + ) + with pytest.raises(ValueError): + await _sql(engine, query) + + def test_composite_order_string_rejected_at_construction(self) -> None: + """A composite arithmetic ORDER BY string is not expressible as an + ``OrderItem.column`` — Pydantic rejects it before it ever reaches the + engine. Pins the boundary so DEV-1733 knows the entry point that must + change to support it.""" + with pytest.raises(pydantic.ValidationError): + OrderItem(column="amount:sum / id:count", direction="desc") + + @pytest.mark.xfail( + strict=True, + reason=( + "DEV-1733: order-only transform refs (change(...)/cumsum(...) in " + "ORDER BY, not declared as a measure) are deferred — Stage 8 " + "raises a clean ValueError. Auto-promotes when DEV-1733 lands " + "hidden TransformKey materialisation for ORDER BY." + ), + ) + async def test_transform_order_only_materializes_FUTURE(self, engine) -> None: + query = SlayerQuery( + source_model="orders", + time_dimensions=[TimeDimension(dimension="created_at", granularity="month")], + measures=[ModelMeasure(formula="amount:sum")], + order=[OrderItem(column="change(amount:sum)", direction="desc")], + ) + sql = await _sql(engine, query) + # Future contract: the change() value is materialised (hidden) and the + # outermost ORDER BY references it — not dropped, not projected. + assert _outer_select_columns(sql) == ["orders.created_at", "orders.amount_sum"] + assert _outer_order_by_names(sql), f"ORDER BY must be present.\nSQL:\n{sql}" + + +# =========================================================================== +# Group 9 — host-rooted isolated aggregate, order-only (DEV-1503 x Law 2). +# =========================================================================== +class TestHostRootedIsolatedHiddenOrder: + async def test_order_only_filtered_local_aggregate_hidden(self, engine) -> None: + """A local aggregate whose ``Column.filter`` crosses a join + (``flagged_amount`` filtered on ``customers.region``) is host-rooted + isolated. Used order-only it must stay hidden — not projected in the + outer SELECT — while still driving the ORDER BY.""" + query = SlayerQuery( + source_model="orders", + dimensions=[ColumnRef(name="status")], + measures=[ModelMeasure(formula="*:count")], + order=[OrderItem(column="flagged_amount:sum", direction="desc")], + limit=3, + ) + sql = await _sql(engine, query) + outer = _outer_select_columns(sql) + assert outer == ["orders.status", "orders._count"], ( + f"filtered-local aggregate must be hidden.\ngot: {outer}\nSQL:\n{sql}" + ) + assert any("flagged_amount_sum" in a for a in _all_aliases_in_sql(sql)) diff --git a/tests/test_nested_dag_cross_stage_refs.py b/tests/test_nested_dag_cross_stage_refs.py index 0f26c266..124f1720 100644 --- a/tests/test_nested_dag_cross_stage_refs.py +++ b/tests/test_nested_dag_cross_stage_refs.py @@ -1240,18 +1240,15 @@ async def test_unrenamed_intercepted_cmm_colon_filter_does_not_get_rewritten_as_ finally: tmp.cleanup() - @pytest.mark.skip( - reason="DEV-1472: hidden-slot ORDER BY (order-only ref not in " - "dimensions/measures) is documented as deferred in the typed pipeline " - "(stage 7b.10+). Re-enable when the gap closes." - ) - async def test_intercepted_cmm_order_only_no_qfield_registers_alias( + async def test_intercepted_cmm_order_only_reaggregated_resolves( self, ) -> None: - """DEV-1450 typed contract: outer ORDER BY may reference the - flat inner alias directly, even when the column is not - re-projected by the outer query. The order resolver must find - the flat column on s1 and emit a valid `s1.` ref.""" + """DEV-1712 (D-D variant 1): an AGGREGATED outer stage may order by + an EXPLICIT re-aggregation of an inner-stage column + (``customers__revenue_sum:max``) that is not re-projected. It + materialises as a hidden aggregate on s1 and orders at the outer + wrap — valid grouped SQL, resolving on s1, never a bare `customers` + table.""" engine, tmp = await _engine_with_join_chain() try: inner = SlayerQuery( @@ -1260,31 +1257,147 @@ async def test_intercepted_cmm_order_only_no_qfield_registers_alias( dimensions=["customers.regions.name"], measures=[{"formula": "customers.revenue:sum"}], ) - # Outer: order-only ref to the inner flat alias, NOT in - # the projection. Must still resolve. outer = SlayerQuery( source_model="s1", dimensions=["customers__regions__name"], measures=[{"formula": "*:count"}], - order=[{"column": "customers__revenue_sum"}], + order=[{"column": "customers__revenue_sum:max"}], ) resp = await engine.execute(query=[inner, outer], dry_run=True) sql = resp.sql or "" outer_select = _outermost_select(sql) + # Outer projects EXACTLY the two declared columns — the hidden + # re-aggregate must not surface. + projected = [p.alias_or_name for p in outer_select.expressions] + assert projected == ["s1.customers__regions__name", "s1._count"], ( + f"outer projection leaked the hidden re-aggregate.\n" + f"got: {projected}\nSQL:\n{sql}" + ) + # The outer ORDER BY names the hidden re-aggregate (a MAX over the + # inner column), never a bare `customers` table. order = outer_select.args.get("order") assert order is not None, f"Outer ORDER BY missing.\nSQL:\n{sql}" order_cols = list(order.find_all(exp.Column)) - order_col_names = {c.name for c in order_cols} - # Must resolve to a column on s1 (or a quoted projection - # alias) — NOT a bare `customers` table that doesn't exist - # in the outer scope. assert "customers" not in {c.table for c in order_cols if c.table}, ( - f"ORDER BY must not reference a bare `customers` table.\n" - f"got: {order_col_names}\nSQL:\n{sql}" + f"ORDER BY must not reference a bare `customers` table.\nSQL:\n{sql}" + ) + order_names = {c.name for c in order_cols} + assert any("revenue_sum" in n and "max" in n.lower() for n in order_names), ( + f"ORDER BY must reference the hidden MAX re-aggregate.\n" + f"got: {order_names}\nSQL:\n{sql}" ) finally: tmp.cleanup() + async def test_intercepted_cmm_order_only_bare_grouped_raises( + self, + ) -> None: + """DEV-1712 (D-D variant 2): an AGGREGATED outer stage ordering by a + BARE inner-stage row column (``customers__revenue_sum``, not + re-aggregated) is not in the outer GROUP BY — no valid SQL exists, so + the pipeline rejects it at plan time (the D-B row-column-under-grouping + contract).""" + engine, tmp = await _engine_with_join_chain() + try: + inner = SlayerQuery( + name="s1", + source_model="orders", + dimensions=["customers.regions.name"], + measures=[{"formula": "customers.revenue:sum"}], + ) + outer = SlayerQuery( + source_model="s1", + dimensions=["customers__regions__name"], + measures=[{"formula": "*:count"}], + order=[{"column": "customers__revenue_sum"}], + ) + with pytest.raises(ValueError) as ei: + await engine.execute(query=[inner, outer], dry_run=True) + msg = str(ei.value) + assert "customers__revenue_sum" in msg, ( + f"error must name the offending order column.\ngot: {msg}" + ) + assert "dimension" in msg.lower() or "aggregate" in msg.lower(), ( + f"error must guide the user (add as dimension / order by an " + f"aggregate).\ngot: {msg}" + ) + finally: + tmp.cleanup() + + async def test_intercepted_cmm_order_only_bare_ungrouped_splits( + self, + ) -> None: + """DEV-1712 (D-D variant 3): a NON-aggregating outer stage (dedup off) + ordering by a bare inner-stage column emits a valid SPLIT reference + against the s1 rowset — resolving on s1, never a bare `customers` + table.""" + engine, tmp = await _engine_with_join_chain() + try: + inner = SlayerQuery( + name="s1", + source_model="orders", + dimensions=["customers.regions.name"], + measures=[{"formula": "customers.revenue:sum"}], + ) + outer = SlayerQuery( + source_model="s1", + dimensions=["customers__regions__name"], + distinct_dimension_values=False, + order=[{"column": "customers__revenue_sum"}], + ) + resp = await engine.execute(query=[inner, outer], dry_run=True) + sql = resp.sql or "" + outer_select = _outermost_select(sql) + order = outer_select.args.get("order") + assert order is not None, f"Outer ORDER BY missing.\nSQL:\n{sql}" + order_cols = list(order.find_all(exp.Column)) + assert order_cols, f"ORDER BY has no column.\nSQL:\n{sql}" + # SPLIT reference qualified by the s1 stage rowset, single-identifier + # leaf — never a bare `customers` table, never a composite token. + col = order_cols[0] + split_msg = ( + f"ORDER BY must be the split reference " + f"s1.customers__revenue_sum, got '{col.table}.{col.name}'.\n" + f"SQL:\n{sql}" + ) + assert col.table == "s1", split_msg + assert col.name == "customers__revenue_sum", split_msg + finally: + tmp.cleanup() + + async def test_intercepted_cmm_order_only_self_qualified_resolves( + self, + ) -> None: + """DEV-1712 (Codex): a downstream stage may SELF-qualify an order ref + with its own stage name (``s1.customers__revenue_sum``). The + qualifier-aware host-local check must recognise the stage relation as + the host, so it resolves identically to the bare form — not + misclassified as a foreign-joined ref.""" + engine, tmp = await _engine_with_join_chain() + try: + inner = SlayerQuery( + name="s1", + source_model="orders", + dimensions=["customers.regions.name"], + measures=[{"formula": "customers.revenue:sum"}], + ) + outer = SlayerQuery( + source_model="s1", + dimensions=["customers__regions__name"], + distinct_dimension_values=False, + order=[{"column": "s1.customers__revenue_sum"}], + ) + resp = await engine.execute(query=[inner, outer], dry_run=True) + sql = resp.sql or "" + order = _outermost_select(sql).args.get("order") + assert order is not None, f"Outer ORDER BY missing.\nSQL:\n{sql}" + cols = list(order.find_all(exp.Column)) + assert cols, f"ORDER BY has no column.\nSQL:\n{sql}" + assert cols[0].table == "s1", f"SQL:\n{sql}" + assert cols[0].name == "customers__revenue_sum", f"SQL:\n{sql}" + finally: + tmp.cleanup() + # =========================================================================== diff --git a/tests/test_projection_trim.py b/tests/test_projection_trim.py index f75d2e46..1954b92d 100644 --- a/tests/test_projection_trim.py +++ b/tests/test_projection_trim.py @@ -1123,15 +1123,6 @@ async def test_post_filter_plus_order_plus_trim( # Group M — Cross-model / isolated ORDER BY hoisted, not projected. # =========================================================================== class TestCrossModelOrderBy: - @pytest.mark.xfail( - strict=True, - reason=( - "DEV-1495 (Bug 2): an order-by-only cross-model aggregate " - "(``customers.revenue:sum``, no declared measure) LEAKS into the " - "outer projection as a malformed ``orders.customers._sum`` instead " - "of staying a hidden slot. Auto-promotes when DEV-1495 is fixed." - ), - ) async def test_order_by_cross_model_agg_hoisted_not_projected( self, orders_customers_engine, ) -> None: diff --git a/tests/test_sql_generator.py b/tests/test_sql_generator.py index 8ba02e93..66ee0d33 100644 --- a/tests/test_sql_generator.py +++ b/tests/test_sql_generator.py @@ -2053,15 +2053,6 @@ async def test_rank_partition_by_time_dimension( in _norm(sql) ) - @pytest.mark.xfail( - strict=True, - reason=( - "DEV-1497: the typed pipeline does not validate that a rank " - "partition_by column is a query dimension — it silently adds it " - "to the base GROUP BY (changing result grain) instead of raising. " - "Auto-promotes when the validation is restored." - ), - ) async def test_partition_by_must_be_a_query_dimension( self, generator: SQLGenerator, orders_model: SlayerModel ) -> None: @@ -7161,11 +7152,12 @@ async def test_dim_only_dedup_with_hidden_order_first_last( async def test_hidden_row_order_target_raises_nyi( self, generator: SQLGenerator ) -> None: - """ORDER BY a non-projected ROW column (e.g. ``customer_id``) is - not a supported shape and must raise NotImplementedError — both - today and after Change 2. Guards against broad ``include_order= - True`` accidentally materialising hidden row slots and silently - changing GROUP BY grain. + """ORDER BY a non-projected ROW column (e.g. ``customer_id``) in an + aggregated query is refused. DEV-1712 (Stage 8) replaced the earlier + ``NotImplementedError`` with a clear plan-time ``ValueError`` (the + column is not in the GROUP BY; add it to dimensions or order by an + aggregate of it) — the query is still rejected, never silently + grain-widened. """ m = SlayerModel( name="orders", sql_table="orders", data_source="test", @@ -7183,8 +7175,9 @@ async def test_hidden_row_order_target_raises_nyi( dimensions=[ColumnRef(name="status")], order=[OrderItem(column="customer_id", direction="asc")], ) - with pytest.raises(NotImplementedError): + with pytest.raises(ValueError) as ei: await engine.execute(query, dry_run=True) + assert "customer_id" in str(ei.value) async def test_hidden_simple_aggregate_in_having( self, generator: SQLGenerator