From be3bce68b52f0fbe98af4c15931680b9850bc560 Mon Sep 17 00:00:00 2001 From: whimo Date: Tue, 18 Aug 2026 23:44:55 +0000 Subject: [PATCH 1/2] Fix BigQuery outer-wrap ORDER BY and dataset-scoped introspection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The DEV-1444 outer wrap stripped the ORDER BY qualifier with `col.set("table", None)`. BigQuery still renders a qualifier slot for such a column, so any query combining a computed measure with an ORDER BY failed with `400 Syntax error: Invalid empty identifier`. Replacing the node is not enough either: BigQuery parses a quoted dotted alias into one part per segment, so the model prefix would be dropped. `_outer_order_column` now keeps the longest part-suffix the inner SELECT actually projects; other dialects are byte-identical. Per-table introspection fell back to the bare `information_schema.columns`, which BigQuery resolves at project level — unreadable by a dataset-scoped service account, so every table 403'd and drift reported the whole datasource for deletion. Qualify by dataset, and treat an all-tables introspection failure as "unknown" (`IntrospectionUnavailable`) instead of "everything was dropped". Co-Authored-By: Claude Opus 5 (1M context) --- DECISIONS.md | 1 + slayer/engine/introspect_utils.py | 14 ++++++- slayer/engine/schema_drift.py | 26 +++++++++--- slayer/sql/dialects/base.py | 23 +++++++++-- slayer/storage/type_refinement.py | 11 ++++- tests/dialects/test_bigquery.py | 68 ++++++++++++++++++++++++++++++- tests/test_ingestion.py | 15 +++++++ tests/test_schema_drift_error.py | 49 ++++++++++++++++++++++ 8 files changed, 193 insertions(+), 14 deletions(-) diff --git a/DECISIONS.md b/DECISIONS.md index 391739bd..9dbf0a97 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -77,3 +77,4 @@ implementation detail. Include issue refs when known. - 2026-08-12 — Formula measure referencing a sibling saved measure now emits valid SQL regardless of measure order (DEV-1779). A saved formula (`habit_score = order_count / unique_customers`) inline-expands at parse time to leaf colon refs (`id:count / customer:count_distinct`), so when the formula measure is enriched BEFORE a referenced sibling, its expression SQL freezes the sibling's canonical alias (`orders.id_count`); the later direct selection of that sibling renames the base-CTE column to the declared name (`orders.order_count`) and the frozen reference dangled — invalid SQL on Postgres, silently-NULL on SQLite (double-quote-as-string-literal). The DEV-1444 provenance-merge only reconciled the forward order (sibling declared first). Fix makes the rename atomic via one `_repoint_alias(prev, new)` helper called at BOTH rename sites (local-agg and cross-model-intercept): it sweeps every `known_aliases` value, the `measure_canonical_key_to_alias` provenance index, and — the new part — the already-frozen carriers `EnrichedExpression.sql` (exact quoted-token replace; the closing quote makes `"orders.id_count"` never match `"orders.id_count_2"`) and `EnrichedTransform.measure_alias` (so `cumsum(order_count)` and `change_pct` desugaring follow the rename too). Quoted-token string replacement is SQL-token-blind but safe here because arithmetic expression SQL is compiler-produced and never embeds a single-quoted literal containing a double-quoted alias — same invariant `_resolve_sql` already relies on. Defense-in-depth: the SQL generator's CTE-layering loop previously emitted an unresolved expression (and silently DROPPED an unresolved self-join `time_shift`) when it stalled, so a regression of this class reached the DB as broken SQL; it now raises a precise `ValueError` naming the computed column / transform and the missing alias for expressions AND all transform types. `_deps_available` gates in-loop addition, so anything still pending is genuinely unresolved — no false-positive raise. - 2026-08-12 — Dotted dimension join-path binding (DEV-1780): a dotted dimension/time-dimension path resolves only when every hop is a direct join. Previously a hop that was not a direct join fell through leniently — the enriched dim kept its `A__B` alias in SELECT/GROUP BY but `_resolve_joins` emitted no join, shipping invalid SQL (unbound table alias). Filters and cross-model measures already rejected such paths; only dimensions/time-dimensions had the hole (the shared `_resolve_dotted_dim_with_stage_fallback` lenient branch). Fix is an engine routing pre-pass (`SlayerQueryEngine._route_dotted_dimension_refs`, run in `_enrich` before `enrich_query`, gated on `enforce_join_binding and source_model_origin is None`): it normalizes root-prefixes via `strip_source_model_prefix`, then for each dotted ref tries the explicit direct-join walk and, on `_NoJoinError`, routes via a datasource-scoped `JoinGraph`. A SHORT FORM (one model segment, e.g. `Consumer.name`) with exactly ONE route to the target auto-resolves — the ref is rewritten to the full routed path, so the result key is the full path (`root.Subscription.Customer.Consumer.name`), consistent with "joined dims keep the full path". Ambiguous (≥2 routes), unreachable (0), and explicit multi-hop chains with a broken hop are REJECTED with `UnresolvableDimensionJoinError(SlayerError, ValueError)` (mirrors the DEV-1645 `UnresolvableOrderColumnError` reject-don't-emit-invalid-SQL doctrine); the message suggests the short form when the target is uniquely reachable, else the shortest deterministic full path (`JoinGraph.shortest_path`), else nothing. `JoinGraph.count_simple_paths(root, target, cap=2)` classifies routes — it counts ALL simple paths (a 2-hop + 3-hop route is genuinely ambiguous; auto-picking the shorter would silently change join semantics), reverse-reachability-pruned and cycle-guarded. The rewrite map is also applied to matching `OrderItem.column` refs and `main_time_dimension` so dependent references stay consistent. Deliberate limits (conservative, prefer reject over a wrong route): routing runs only within a single datasource (`model.data_source` truthy; the graph is datasource-scoped) and is deferred when named-query stages are in scope (their virtual models aren't in the stored graph) — those refs fall through to the guard. A post-`_resolve_joins` safety-net guard in `enrich_query` (same gate) raises `UnresolvableDimensionJoinError` for any dim/time-dim whose alias is absent from `resolved_joins`, guaranteeing the invariant even for direct `enrich_query` callers; the re-rooted cross-model CTE enrichment passes `enforce_join_binding=False` (it legitimately carries source-local shared dims like `orders.status` that never bind to a base-table join). Out of scope: the multi-stage lenient cross-stage fall-through (`test_unresolvable_dotted_ref_falls_through`, where distinguishing a genuine error from a re-rooting artifact is unsolved) and leaf-column-missing-on-a-valid-path (the alias IS bound there — a different failure class). - 2026-08-16 — FK-derived joins name the MODEL, not the live object (DEV-1688 / DEV-1741 / #279). Model names strip `__` (reserved for join paths), so an FK to `reports__patient__drug` used to persist a join targeting a model that cannot exist; `_generate_joins` now takes the live→model map, and a target whose object was skipped on a sanitization collision drops its join rather than dangling. Stores written before the fix self-heal on the normal re-ingest path rather than via a schema migration — no version bump, and the repair demands the sanitized target AND identical `join_pairs` to match a freshly-generated join, so it can only rename the join the bug produced (name-only matching would collapse `a__b` and `a___b` onto one target and trip the duplicate-target guard, turning a merely-dangling store into a failed re-ingest). A store that never re-ingests keeps a join that was already broken. +- 2026-08-18 — Two BigQuery-only failures fixed. (1) The DEV-1444 outer wrap re-parses already-emitted SQL, and BigQuery parses a quoted dotted alias (`` `orders.created_at` ``) into one part per segment, so the old qualifier strip (`col.set("table", None)`) both emitted an empty backtick pair — `400 Syntax error: Invalid empty identifier` on any computed measure plus ORDER BY — and, once that was fixed by replacing the node, silently dropped the model prefix, turning a syntax error into an unresolved name. `SqlDialect._outer_order_column` therefore re-resolves the column by picking the LONGEST part-suffix the inner SELECT actually projects (checked as a quoted identifier against `inner_sql`, which at that point still carries canonical dotted aliases — `rewrite_emitted_sql` mangles them afterwards). That subsumes the `_base.`-qualified form `_assemble_combined_sql` emits and keeps Postgres/DuckDB/MySQL output byte-identical, including the untouched bare-column and no-match fallbacks; hidden ORDER-BY hoists resolve too, which a `public`-list match would have missed. (2) `_get_columns_fallback` queried the bare `information_schema.columns`, which BigQuery resolves as `.information_schema.columns` — a project-level view a dataset-scoped service account (the normal least-privilege setup) cannot read, so every per-table introspection 403'd. It now qualifies by dataset (`` ``.INFORMATION_SCHEMA.COLUMNS ``), taken from `schema` or from the dotted table name, and drops the now-redundant `table_schema` predicate. Related hardening: `_live_schema_for_datasource` raises `IntrospectionUnavailable` when EVERY table in a datasource failed rather than returning an empty map, because empty is indistinguishable from "every table was dropped" — drift then reported a `WholeModelDelete` for every model in the tenant off one credential error, and `--force-clean` would act on it. `_collect_sql_table_diffs` catches it and returns no verdict; type refinement catches it and keeps the persisted types. diff --git a/slayer/engine/introspect_utils.py b/slayer/engine/introspect_utils.py index 17ea153d..5abbd99e 100644 --- a/slayer/engine/introspect_utils.py +++ b/slayer/engine/introspect_utils.py @@ -76,10 +76,20 @@ def _get_columns_fallback( schema: Optional[str], ) -> List[Dict]: """Get columns via INFORMATION_SCHEMA when Inspector.get_columns() fails.""" + source = "information_schema.columns" + if getattr(getattr(sa_engine, "dialect", None), "name", "") == "bigquery": + # BigQuery only exposes INFORMATION_SCHEMA per dataset; the bare name + # resolves to a project-level view a dataset-scoped account cannot read. + dataset = schema + if "." in table_name: + dataset, table_name = table_name.rsplit(".", 1) + if dataset: + source = f"`{dataset}`.INFORMATION_SCHEMA.COLUMNS" + schema = None if schema: sql = ( "SELECT column_name, data_type " - "FROM information_schema.columns " + f"FROM {source} " "WHERE table_name = :table_name " "AND table_schema = :schema " "ORDER BY ordinal_position" @@ -88,7 +98,7 @@ def _get_columns_fallback( else: sql = ( "SELECT column_name, data_type " - "FROM information_schema.columns " + f"FROM {source} " "WHERE table_name = :table_name " "ORDER BY ordinal_position" ) diff --git a/slayer/engine/schema_drift.py b/slayer/engine/schema_drift.py index aa534198..e5d31fed 100644 --- a/slayer/engine/schema_drift.py +++ b/slayer/engine/schema_drift.py @@ -1657,6 +1657,11 @@ def compute_datasource_drops( # =========================================================================== +class IntrospectionUnavailable(Exception): + """Every table in the datasource failed to introspect, so the live schema + is unknown — callers must not read that as "everything was dropped".""" + + def _live_schema_for_datasource( *, datasource: DatasourceConfig, @@ -1699,6 +1704,11 @@ def _live_schema_for_datasource( datasource.name, exc, ) + if table_names and not out: + raise IntrospectionUnavailable( + f"failed to introspect every table in datasource " + f"{datasource.name!r} ({len(table_names)} table(s))" + ) return out finally: # Same rationale as ``ingest_datasource``: this is a one-shot @@ -2058,11 +2068,17 @@ async def _collect_sql_table_diffs( # Honour the datasource's configured schema_name so non-default-schema # datasources diff against the right table set; otherwise SQLAlchemy # introspects the default and produces false WholeModelDeletes. - live_tables = await asyncio.to_thread( - _live_schema_for_datasource, - datasource=datasource, - schema=datasource.schema_name or None, - ) + try: + live_tables = await asyncio.to_thread( + _live_schema_for_datasource, + datasource=datasource, + schema=datasource.schema_name or None, + ) + except IntrospectionUnavailable as exc: + # Unknown live schema — reporting every model for deletion here would + # hand ``--force-clean`` a whole tenant on a transient credential error. + logger.warning("validate_models: skipping drift verdict: %s", exc) + return {} probe_drifts_by_model = await _sqlite_probe_drifts_for_models( datasource=datasource, sql_table_models=sql_table_models, diff --git a/slayer/sql/dialects/base.py b/slayer/sql/dialects/base.py index 985440ba..206adaf4 100644 --- a/slayer/sql/dialects/base.py +++ b/slayer/sql/dialects/base.py @@ -498,9 +498,11 @@ def emit_outer_wrap( return base out = base if order is not None: - for col in order.find_all(exp.Column): - if col.args.get("table") is not None: - col.set("table", None) + for col in list(order.find_all(exp.Column)): + if len(col.parts) > 1: + col.replace( + self._outer_order_column(col=col, inner_sql=inner_sql) + ) out += "\n" + order.sql(dialect=self.sqlglot_name, pretty=True) if limit is not None: out += "\n" + limit.sql(dialect=self.sqlglot_name, pretty=True) @@ -508,6 +510,21 @@ def emit_outer_wrap( out += "\n" + offset_arg.sql(dialect=self.sqlglot_name, pretty=True) return out + def _outer_order_column(self, *, col: exp.Column, inner_sql: str) -> exp.Column: + """Re-resolve a qualified ORDER BY column against the ``_outer`` scope. + + BigQuery parses a quoted dotted alias (`` `orders.created_at` ``) into + one part per segment, so clearing the ``table`` arg would both drop the + model prefix and leave an empty qualifier; instead keep the longest + part-suffix the inner SELECT actually projects. + """ + names = [p.name for p in col.parts] + for i in range(len(names)): + candidate = ".".join(names[i:]) + if self.quote_identifier(candidate) in inner_sql: + return exp.Column(this=exp.Identifier(this=candidate, quoted=True)) + return exp.Column(this=col.parts[-1].copy()) + # DEV-1756 identifier-length fitting. Aliases stay canonical inside SLayer, # fitted only on emission and restored on the result keys. diff --git a/slayer/storage/type_refinement.py b/slayer/storage/type_refinement.py index eae213da..7a0069be 100644 --- a/slayer/storage/type_refinement.py +++ b/slayer/storage/type_refinement.py @@ -319,9 +319,16 @@ def refine_dict_with_live_schema(d: dict, datasource: DatasourceConfig) -> bool: return False # Local import to avoid circular import at module load time. - from slayer.engine.schema_drift import _live_schema_for_datasource + from slayer.engine.schema_drift import ( + IntrospectionUnavailable, + _live_schema_for_datasource, + ) - live = _live_schema_for_datasource(datasource=datasource) + try: + live = _live_schema_for_datasource(datasource=datasource) + except IntrospectionUnavailable: + # Persisted types are the safe fallback when the live schema is unknown. + return False table = live.get(sql_table) if table is None: return False diff --git a/tests/dialects/test_bigquery.py b/tests/dialects/test_bigquery.py index 5ffb6ec7..ed4a2b6a 100644 --- a/tests/dialects/test_bigquery.py +++ b/tests/dialects/test_bigquery.py @@ -18,10 +18,11 @@ from unittest.mock import patch import pytest +import sqlglot from slayer.core.enums import DataType, TimeGranularity -from slayer.core.models import Column, DatasourceConfig, SlayerModel -from slayer.core.query import ColumnRef, SlayerQuery +from slayer.core.models import Column, DatasourceConfig, ModelMeasure, SlayerModel +from slayer.core.query import ColumnRef, OrderItem, SlayerQuery, TimeDimension from slayer.engine.enriched import EnrichedQuery from slayer.engine.enrichment import enrich_query from slayer.engine.query_engine import SlayerQueryEngine, _sql_client_cache_key @@ -881,3 +882,66 @@ def test_build_engine_oauth_validates_before_importing_optional_driver() -> None pytest.raises(ValueError, match="is not valid JSON"), ): dialect.build_engine(ds, connection_string="bigquery://p/d") + + +# --------------------------------------------------------------------------- +# Outer-wrap ORDER BY — BigQuery parses a quoted dotted alias into one part +# per segment, so the qualifier strip must rebuild the whole alias. +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "order_sql", + [ + "SELECT 1 FROM t ORDER BY `orders.created_at` DESC", + "SELECT 1 FROM t ORDER BY `_base`.`orders.created_at` DESC", + ], +) +def test_bigquery_outer_wrap_order_by_keeps_full_alias(order_sql: str) -> None: + """No empty backtick qualifier, and the alias keeps its model prefix so it + resolves against the ``_outer`` scope.""" + order = sqlglot.parse_one(order_sql, dialect="bigquery").args["order"] + out = BigqueryDialect().emit_outer_wrap( + inner_sql="SELECT `orders.created_at` AS `orders.created_at`, 1 AS x FROM t", + public=["orders.created_at"], + order=order, + limit=None, + offset_arg=None, + ) + assert "``" not in out, f"empty identifier emitted: {out}" + assert "ORDER BY\n `orders.created_at` DESC" in out, out + + +async def test_bigquery_computed_measure_with_order_by_resolves(tmp_path) -> None: + """End-to-end: the ORDER BY alias matches the mangled outer projection.""" + model = SlayerModel( + name="orders", + sql_table="orders", + data_source="bq", + default_time_dimension="created_at", + columns=[ + Column(name="id", sql="id", type=DataType.DOUBLE, primary_key=True), + Column(name="created_at", sql="created_at", type=DataType.TIMESTAMP), + Column(name="revenue", sql="amount", type=DataType.DOUBLE), + Column(name="quantity", sql="quantity", type=DataType.DOUBLE), + ], + ) + query = SlayerQuery( + source_model="orders", + time_dimensions=[ + TimeDimension(dimension="created_at", granularity=TimeGranularity.MONTH) + ], + measures=[ModelMeasure(formula="revenue:sum / quantity:sum", name="aov")], + order=[OrderItem(column="created_at", direction="desc")], + limit=6, + ) + enriched = await enrich_query( + query=query, + model=model, + resolve_dimension_via_joins=_noop_async, + resolve_cross_model_measure=_noop_async, + resolve_join_target=_noop_async, + ) + sql = SQLGenerator(dialect="bigquery").generate(enriched=enriched, render_mode="outer") + assert "``" not in sql, f"empty identifier emitted: {sql}" + assert "ORDER BY\n `orders___created_at` DESC" in sql, sql diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py index aa3e2064..c8e3dca6 100644 --- a/tests/test_ingestion.py +++ b/tests/test_ingestion.py @@ -2,6 +2,7 @@ import os import tempfile +from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest @@ -87,6 +88,20 @@ def test_with_schema(self): params = args[1] if len(args) > 1 else kwargs assert params == {"table_name": "orders", "schema": "public"} + def test_bigquery_uses_dataset_qualified_information_schema(self): + """A dataset-scoped BigQuery account cannot read the project-level view.""" + engine, conn = _setup_mock_engine([("id", "INTEGER")]) + engine.dialect = SimpleNamespace(name="bigquery") + _get_columns_fallback( + sa_engine=engine, table_name="core.mart__kpis", schema=None, + ) + + args, kwargs = conn.execute.call_args + sql_str = str(args[0]) + assert "`core`.INFORMATION_SCHEMA.COLUMNS" in sql_str + params = args[1] if len(args) > 1 else kwargs + assert params == {"table_name": "mart__kpis"} + def test_no_fstring_interpolation(self): """Ensure table_name/schema values never appear literally in the SQL text.""" engine, conn = _setup_mock_engine([]) diff --git a/tests/test_schema_drift_error.py b/tests/test_schema_drift_error.py index 7b075117..917970bd 100644 --- a/tests/test_schema_drift_error.py +++ b/tests/test_schema_drift_error.py @@ -233,3 +233,52 @@ async def test_touched_includes_join_targets(self, workspace: Path) -> None: # Both source_model and the join target must be reported as touched. assert "orders" in exc.value.models assert "customers" in exc.value.models + + +class TestIntrospectionUnavailable: + """A datasource whose every table fails to introspect is 'unknown', not + 'everything was dropped' — otherwise ``--force-clean`` deletes a tenant.""" + + def _ds_with_one_table(self, tmpdir: str) -> DatasourceConfig: + import sqlalchemy as sa + + db_path = str(Path(tmpdir) / "live.db") + engine = sa.create_engine(f"sqlite:///{db_path}") + with engine.connect() as c: + c.execute(sa.text("CREATE TABLE t (id INTEGER PRIMARY KEY)")) + c.commit() + return DatasourceConfig(name="live", type="sqlite", database=db_path) + + def test_live_schema_raises_when_every_table_fails(self) -> None: + from slayer.engine.schema_drift import ( + IntrospectionUnavailable, + _live_schema_for_datasource, + ) + + with tempfile.TemporaryDirectory() as tmpdir: + ds = self._ds_with_one_table(tmpdir) + with patch( + "slayer.engine.schema_drift._introspect_one_table", + side_effect=PermissionError("403 Access Denied"), + ), pytest.raises(IntrospectionUnavailable): + _live_schema_for_datasource(datasource=ds) + + async def test_drift_verdict_skipped_when_introspection_unavailable(self) -> None: + from slayer.engine.schema_drift import _collect_sql_table_diffs + + with tempfile.TemporaryDirectory() as tmpdir: + ds = self._ds_with_one_table(tmpdir) + model = SlayerModel( + name="t", sql_table="t", data_source="live", + columns=[Column(name="id", sql="id", type=DataType.INT)], + ) + with patch( + "slayer.engine.schema_drift._introspect_one_table", + side_effect=PermissionError("403 Access Denied"), + ): + diffs = await _collect_sql_table_diffs( + datasource=ds, + sql_table_models=[model], + available_in_ds={"t"}, + ) + assert diffs == {}, f"must not recommend deletes: {diffs}" From bac6247bdd20377ede330e0017b3acfe9d006e0f Mon Sep 17 00:00:00 2001 From: whimo Date: Wed, 19 Aug 2026 00:01:01 +0000 Subject: [PATCH 2/2] Address review: quote the dataset, prefer public aliases, hoist imports - Build the BigQuery dataset-qualified FROM via a sqlglot AST so a hostile dataset name stays inside one quoted identifier (CodeRabbit). - `_outer_order_column` now resolves against the public alias list first and only falls back to scanning `inner_sql` for hidden ORDER BY hoists, so a qualified source column with a different projected alias resolves to the name the outer scope exposes. - Extract `_info_schema_columns_query` (Sonar S3776) and drop the `list()` around `find_all` in favour of `transform` (Sonar S7504). - Hoist function-local test imports. Co-Authored-By: Claude Opus 5 (1M context) --- slayer/engine/introspect_utils.py | 51 +++++++++++++++++++------------ slayer/sql/dialects/base.py | 30 ++++++++++++------ tests/dialects/test_bigquery.py | 17 +++++++++++ tests/test_ingestion.py | 17 +++++++++++ tests/test_schema_drift_error.py | 17 +++++------ 5 files changed, 93 insertions(+), 39 deletions(-) diff --git a/slayer/engine/introspect_utils.py b/slayer/engine/introspect_utils.py index 5abbd99e..fe30306a 100644 --- a/slayer/engine/introspect_utils.py +++ b/slayer/engine/introspect_utils.py @@ -16,6 +16,7 @@ from typing import Dict, List, Optional import sqlalchemy as sa +from sqlglot import exp from slayer.core.enums import DataType @@ -70,12 +71,13 @@ def _parse_info_schema_is_float(data_type_str: str) -> bool: return True # No precision/scale info, default to float -def _get_columns_fallback( +def _info_schema_columns_query( + *, sa_engine: sa.Engine, table_name: str, schema: Optional[str], -) -> List[Dict]: - """Get columns via INFORMATION_SCHEMA when Inspector.get_columns() fails.""" +) -> tuple[str, Dict]: + """Build the parameterized INFORMATION_SCHEMA.columns query for one table.""" source = "information_schema.columns" if getattr(getattr(sa_engine, "dialect", None), "name", "") == "bigquery": # BigQuery only exposes INFORMATION_SCHEMA per dataset; the bare name @@ -84,25 +86,34 @@ def _get_columns_fallback( if "." in table_name: dataset, table_name = table_name.rsplit(".", 1) if dataset: - source = f"`{dataset}`.INFORMATION_SCHEMA.COLUMNS" + # sqlglot quotes/escapes the dataset, which is config-supplied. + source = exp.Table( + this=exp.to_identifier("COLUMNS"), + db=exp.to_identifier("INFORMATION_SCHEMA"), + catalog=exp.to_identifier(dataset, quoted=True), + ).sql(dialect="bigquery") schema = None + sql = ( + "SELECT column_name, data_type " + f"FROM {source} " + "WHERE table_name = :table_name " + ) + params = {"table_name": table_name} if schema: - sql = ( - "SELECT column_name, data_type " - f"FROM {source} " - "WHERE table_name = :table_name " - "AND table_schema = :schema " - "ORDER BY ordinal_position" - ) - params = {"table_name": table_name, "schema": schema} - else: - sql = ( - "SELECT column_name, data_type " - f"FROM {source} " - "WHERE table_name = :table_name " - "ORDER BY ordinal_position" - ) - params = {"table_name": table_name} + sql += "AND table_schema = :schema " + params["schema"] = schema + return sql + "ORDER BY ordinal_position", params + + +def _get_columns_fallback( + sa_engine: sa.Engine, + table_name: str, + schema: Optional[str], +) -> List[Dict]: + """Get columns via INFORMATION_SCHEMA when Inspector.get_columns() fails.""" + sql, params = _info_schema_columns_query( + sa_engine=sa_engine, table_name=table_name, schema=schema, + ) with sa_engine.connect() as conn: rows = conn.execute(sa.text(sql), params).fetchall() result = [] diff --git a/slayer/sql/dialects/base.py b/slayer/sql/dialects/base.py index 206adaf4..8babac7d 100644 --- a/slayer/sql/dialects/base.py +++ b/slayer/sql/dialects/base.py @@ -498,11 +498,15 @@ def emit_outer_wrap( return base out = base if order is not None: - for col in list(order.find_all(exp.Column)): - if len(col.parts) > 1: - col.replace( - self._outer_order_column(col=col, inner_sql=inner_sql) + order = order.transform( + lambda node: ( + self._outer_order_column( + col=node, public=public, inner_sql=inner_sql, ) + if isinstance(node, exp.Column) and len(node.parts) > 1 + else node + ) + ) out += "\n" + order.sql(dialect=self.sqlglot_name, pretty=True) if limit is not None: out += "\n" + limit.sql(dialect=self.sqlglot_name, pretty=True) @@ -510,17 +514,25 @@ def emit_outer_wrap( out += "\n" + offset_arg.sql(dialect=self.sqlglot_name, pretty=True) return out - def _outer_order_column(self, *, col: exp.Column, inner_sql: str) -> exp.Column: + def _outer_order_column( + self, *, col: exp.Column, public: Sequence[str], inner_sql: str, + ) -> exp.Column: """Re-resolve a qualified ORDER BY column against the ``_outer`` scope. BigQuery parses a quoted dotted alias (`` `orders.created_at` ``) into one part per segment, so clearing the ``table`` arg would both drop the model prefix and leave an empty qualifier; instead keep the longest - part-suffix the inner SELECT actually projects. + part-suffix that names a column of the outer scope. ``public`` is the + authoritative half of that scope; the ``inner_sql`` scan is the fallback + for an ORDER BY over a hidden hoist, which is projected but not public. """ - names = [p.name for p in col.parts] - for i in range(len(names)): - candidate = ".".join(names[i:]) + candidates = [ + ".".join(p.name for p in col.parts[i:]) for i in range(len(col.parts)) + ] + for candidate in candidates: + if candidate in public: + return exp.Column(this=exp.Identifier(this=candidate, quoted=True)) + for candidate in candidates: if self.quote_identifier(candidate) in inner_sql: return exp.Column(this=exp.Identifier(this=candidate, quoted=True)) return exp.Column(this=col.parts[-1].copy()) diff --git a/tests/dialects/test_bigquery.py b/tests/dialects/test_bigquery.py index ed4a2b6a..cb7efeb5 100644 --- a/tests/dialects/test_bigquery.py +++ b/tests/dialects/test_bigquery.py @@ -945,3 +945,20 @@ async def test_bigquery_computed_measure_with_order_by_resolves(tmp_path) -> Non sql = SQLGenerator(dialect="bigquery").generate(enriched=enriched, render_mode="outer") assert "``" not in sql, f"empty identifier emitted: {sql}" assert "ORDER BY\n `orders___created_at` DESC" in sql, sql + + +def test_bigquery_outer_wrap_order_by_prefers_projected_alias() -> None: + """A qualified source column whose projected alias differs must resolve to + the alias the ``_outer`` scope actually exposes.""" + order = sqlglot.parse_one( + "SELECT 1 FROM t ORDER BY `_base`.`orders.created_at` DESC", + dialect="bigquery", + ).args["order"] + out = BigqueryDialect().emit_outer_wrap( + inner_sql="SELECT `_base`.`orders.created_at` AS `created_at` FROM _base", + public=["created_at"], + order=order, + limit=None, + offset_arg=None, + ) + assert "ORDER BY\n `created_at` DESC" in out, out diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py index c8e3dca6..4c151847 100644 --- a/tests/test_ingestion.py +++ b/tests/test_ingestion.py @@ -102,6 +102,23 @@ def test_bigquery_uses_dataset_qualified_information_schema(self): params = args[1] if len(args) > 1 else kwargs assert params == {"table_name": "mart__kpis"} + def test_bigquery_dataset_is_quoted_and_escaped(self): + """A hostile dataset name cannot break out of the identifier.""" + engine, conn = _setup_mock_engine([]) + engine.dialect = SimpleNamespace(name="bigquery") + _get_columns_fallback( + sa_engine=engine, + table_name="orders", + schema="evil` UNION SELECT 1,2 FROM `x", + ) + + sql_str = str(conn.execute.call_args[0][0]) + # The payload stays inside ONE quoted identifier, its backticks doubled. + assert ( + "FROM `evil`` UNION SELECT 1,2 FROM ``x`.INFORMATION_SCHEMA.COLUMNS" + in sql_str + ), sql_str + def test_no_fstring_interpolation(self): """Ensure table_name/schema values never appear literally in the SQL text.""" engine, conn = _setup_mock_engine([]) diff --git a/tests/test_schema_drift_error.py b/tests/test_schema_drift_error.py index 917970bd..94d141b5 100644 --- a/tests/test_schema_drift_error.py +++ b/tests/test_schema_drift_error.py @@ -15,6 +15,7 @@ from unittest.mock import patch import pytest +import sqlalchemy as sa from slayer.core.enums import DataType from slayer.core.errors import SchemaDriftError @@ -26,7 +27,12 @@ ) from slayer.core.query import SlayerQuery from slayer.engine.query_engine import SlayerQueryEngine -from slayer.engine.schema_drift import WholeModelDelete +from slayer.engine.schema_drift import ( + IntrospectionUnavailable, + WholeModelDelete, + _collect_sql_table_diffs, + _live_schema_for_datasource, +) from slayer.storage.yaml_storage import YAMLStorage @@ -240,8 +246,6 @@ class TestIntrospectionUnavailable: 'everything was dropped' — otherwise ``--force-clean`` deletes a tenant.""" def _ds_with_one_table(self, tmpdir: str) -> DatasourceConfig: - import sqlalchemy as sa - db_path = str(Path(tmpdir) / "live.db") engine = sa.create_engine(f"sqlite:///{db_path}") with engine.connect() as c: @@ -250,11 +254,6 @@ def _ds_with_one_table(self, tmpdir: str) -> DatasourceConfig: return DatasourceConfig(name="live", type="sqlite", database=db_path) def test_live_schema_raises_when_every_table_fails(self) -> None: - from slayer.engine.schema_drift import ( - IntrospectionUnavailable, - _live_schema_for_datasource, - ) - with tempfile.TemporaryDirectory() as tmpdir: ds = self._ds_with_one_table(tmpdir) with patch( @@ -264,8 +263,6 @@ def test_live_schema_raises_when_every_table_fails(self) -> None: _live_schema_for_datasource(datasource=ds) async def test_drift_verdict_skipped_when_introspection_unavailable(self) -> None: - from slayer.engine.schema_drift import _collect_sql_table_diffs - with tempfile.TemporaryDirectory() as tmpdir: ds = self._ds_with_one_table(tmpdir) model = SlayerModel(