diff --git a/docs/guides/history.md b/docs/guides/history.md new file mode 100644 index 0000000000..1505e406a5 --- /dev/null +++ b/docs/guides/history.md @@ -0,0 +1,122 @@ +# Plan history guide + +Sometimes something looks off about how SQLMesh behaved during a plan: a model materialized with unexpected data, a step seemed to hang, or a backfill took far longer than expected. Answering "what did SQLMesh actually run, and did it succeed?" usually means context-switching into the warehouse's own query history UI and manually correlating queries by time. + +The `sqlmesh history` command does that correlation for you. Give it a plan id and it prints every SQL statement SQLMesh executed for that plan, in order, along with each statement's status, duration, and rows/bytes processed — or export the SQL to a file for offline inspection. + +`history` is read-only. It doesn't touch SQLMesh state or your data; it only reads the query engine's own job/query history for the queries SQLMesh ran. + +!!! note + `sqlmesh history` currently supports **BigQuery only**. Running it against a project configured with another engine prints a message that the engine isn't yet supported. + +## How it works + +Every job SQLMesh runs for a plan — both query jobs and the load jobs used to ingest Python-model dataframes and seeds — is tagged with that plan's correlation id as a BigQuery job label. `sqlmesh history` looks up the plan (either the one you specify or one you pick from a menu), then queries BigQuery's job history for every job carrying that label and renders the results chronologically. + +Because it's driven by the query engine's job history rather than SQLMesh state, it only shows what actually executed against the warehouse. It's the fastest way to answer "did this step run, how long did it take, and if it failed, why?" without leaving your terminal. + +## Selecting a plan + +### Interactively + +Run `sqlmesh history` with no arguments to get a numbered menu of recent plans, most recent first: + +```bash linenums="1" +$ sqlmesh history + +Select a plan to inspect: +[1] prod plan 3f9a2c1e applied 2026-07-10 02:03PM (42 models) +[2] prod plan a17b90ff applied 2026-07-09 09:12AM (42 models) +[3] dev_sung plan 9c02d5e1 applied 2026-07-10 01:50PM (3 models) +``` + +Each row shows the environment, the first 8 characters of the plan id, when the plan was applied (or `in progress` if it hasn't finished), and the number of models it touched. The menu is built from SQLMesh state — the current and previous plan for each environment. Type a number to inspect that plan's history. + +Use `--environment`/`--env` to restrict the menu to a single environment, which is useful when a lot of plans have been applied recently: + +```bash linenums="1" +$ sqlmesh history --env prod +``` + +### By plan id + +If you already know the plan id (for example, from `sqlmesh plan` output or the interactive menu), pass it directly to skip the menu: + +```bash linenums="1" +$ sqlmesh history 3f9a2c1e +``` + +You only need to type enough of the plan id to uniquely identify it — the 8-character short id shown in the menu is usually enough. + +## Reading the output + +`sqlmesh history ` prints a table with one row per query, in the order SQLMesh ran them: + +```bash linenums="1" +$ sqlmesh history 3f9a2c1e + +History · plan 3f9a2c1e · prod · 9 queries · 7 ✓ 1 ✗ 1 running + +Time Status Operation Target Duration Bytes/Rows +──────── ────── ───────────── ──────────────────────────────── ──────── ─────────────── +14:03:11 ✓ CREATE TABLE sqlmesh__sushi.orders__2837a1c 1200 ms - +14:03:14 ✓ INSERT sqlmesh__sushi.orders__2837a1c 14800 ms 8,912,441 bytes +14:03:22 ✓ LOAD sqlmesh__sushi.customers__4b90fc 640 ms 12,004 bytes +14:03:29 ✗ INSERT sqlmesh__sushi.items__9f01b2d 420 ms - + └─ Column not found: name PRICE cannot be resolved in sushi.items [at 1:34] +14:03:31 … MERGE sqlmesh__sushi.order_items__7c3a - - + +1 failed · re-run with `-o failures.sql` to export the SQL. +``` + +(In your terminal this is rendered as a bordered table.) The header line summarizes the plan: the short plan id, environment, total job count, and a breakdown of how many succeeded (`✓`), failed (`✗`), or are still running (`…`). + +- **Time** — when the job started. +- **Status** — `✓` succeeded, `✗` failed, `…` still running. +- **Operation** — the leading SQL keyword (`CREATE TABLE`, `INSERT`, `MERGE`, `CREATE VIEW`, `ALTER TABLE`, `LOAD`, etc.), so you can scan the kind of work each job did without reading the full statement. +- **Target** — the physical table the job wrote to, so you can tell which model (and which step) each row belongs to. This is how you follow a Python model whose steps are interleaved with SQL models, or attribute a `LOAD`/`INSERT`/`MERGE` to the right table. Shown as `-` for jobs with no destination (for example a standalone audit `SELECT`). +- **Duration** — wall-clock time the job took. Shown as `-` for jobs that haven't finished. +- **Bytes/Rows** — bytes processed if BigQuery reported them, otherwise rows affected, otherwise `-`. + +A failed job is followed by an indented line with the error message BigQuery returned, so you can see why it failed without opening the warehouse UI. + +Because the rows are ordered by execution time, the steps of a **Python model** (its physical `CREATE TABLE`, the `LOAD` of its dataframe, and the `INSERT`/`MERGE` that publishes it) slot chronologically in between the SQL models' steps, and **hooks** — model pre/post-statements and environment `before_all`/`after_all` statements — appear as their own rows in position. Use the **Target** column to tell them apart. + +## Exporting the SQL + +Pass `-o`/`--output-file` to write the executed SQL to a file instead of printing the table. This is handy for replaying a failed step locally, diffing what ran between two plans, or attaching the SQL to a bug report: + +```bash linenums="1" +$ sqlmesh history 3f9a2c1e -o plan_3f9a2c1e.sql +``` + +Each query becomes one entry, preceded by a header comment with the timestamp, status, duration, and (for failures) the error: + +```sql linenums="1" +-- [2026-07-10T14:03:29] FAILED (420 ms) error: Syntax error: unexpected identifier 'PRICE' at [1:34] +INSERT INTO sqlmesh__sushi.items__9f01 SELECT ...; +``` + +The file is one statement per entry, in execution order, so it can be read top to bottom or replayed a statement at a time. + +## Limitations + +- **BigQuery only.** Other engines aren't supported yet; `history` prints a clear message rather than partial or incorrect results. +- **Requires `bigquery.jobs.listAll`.** Reading every job in the project (not just your own) needs this permission. If it's missing, `history` fails with: + + ``` + Error: Permission denied reading BigQuery job history for project 'acme-analytics'. Reading all jobs in the project requires the 'bigquery.jobs.listAll' permission. + + To grant access, run: + gcloud projects add-iam-policy-binding acme-analytics --member='user:YOUR_EMAIL' --role='roles/bigquery.resourceViewer' + + Docs: https://cloud.google.com/bigquery/docs/information-schema-jobs#required_permissions + ``` + +- **Ingestion latency and retention.** BigQuery's job history views have some delay before a query appears and don't retain jobs forever, so a plan applied moments ago or a very old plan may show no rows. +- **Signals and pure-Python logic aren't shown.** Signals are Python readiness checks that gate whether a model runs; they execute no SQL, so they don't appear as rows. Their effect shows up as an *absence* — a signal-blocked model simply has no `INSERT` for that interval. To inspect what a signal is holding back, use [`sqlmesh check_intervals`](../reference/cli.md#check_intervals), which reports pending intervals while respecting signals. Likewise, any purely in-memory work inside a Python model (API calls, dataframe transforms) runs no SQL and won't appear — only the jobs that touch the warehouse do. +- **Plans only.** Scheduled `sqlmesh run` executions aren't included, only `sqlmesh plan` applications. + +## See also + +The [CLI reference](../reference/cli.md#history) has the full list of options. diff --git a/docs/reference/cli.md b/docs/reference/cli.md index b65f8256ac..402733962a 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -251,6 +251,27 @@ Options: --help Show this message and exit. ``` +## history + +``` +Usage: sqlmesh history [OPTIONS] [PLAN_ID] + + Show the query engine history of everything SQLMesh ran for a plan. + + Scheduled `sqlmesh run` executions are not included; only plans are shown. + +Options: + --environment, --env TEXT Restrict the plan menu to this environment. + -o, --output-file FILE Export the executed SQL to a file instead of + printing. + --help Show this message and exit. +``` + +`history` is a read-only debugging command: it doesn't touch your state or your data, it only reads the query engine's own query history for the queries SQLMesh ran under a plan. See the [plan history guide](../guides/history.md) for a walkthrough with example output, or `sqlmesh history --help` for the full option list above. + +!!! note + Only available for BigQuery in this version. Other engines print a message that they're not yet supported. + ## info ``` diff --git a/mkdocs.yml b/mkdocs.yml index 368fb6690a..d4985e398d 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -32,6 +32,7 @@ nav: - SQLMesh tools: - guides/vscode.md - guides/tablediff.md + - guides/history.md - guides/linter.md - guides/ui.md - Advanced usage: diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py index b3c7a7027b..16b2e190f7 100644 --- a/sqlmesh/cli/main.py +++ b/sqlmesh/cli/main.py @@ -33,6 +33,7 @@ "create_external_models", "destroy", "environments", + "history", "invalidate", "janitor", "migrate", @@ -1187,6 +1188,35 @@ def environments(obj: Context) -> None: obj.print_environment_names() +@cli.command("history") +@click.argument("plan_id", required=False) +@click.option( + "--environment", + "--env", + help="Restrict the plan menu to this environment.", +) +@click.option( + "-o", + "--output-file", + type=click.Path(dir_okay=False, writable=True, path_type=Path), + help="Export the executed SQL to a file instead of printing.", +) +@click.pass_obj +@error_handler +@cli_analytics +def history( + obj: Context, + plan_id: t.Optional[str], + environment: t.Optional[str], + output_file: t.Optional[Path], +) -> None: + """Show the query engine history of everything SQLMesh ran for a plan. + + Scheduled `sqlmesh run` executions are not included; only plans are shown. + """ + obj.plan_history(plan_id=plan_id, environment=environment, output_file=output_file) + + @cli.command("lint") @click.option( "--models", diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py index 8af837b08a..439b665343 100644 --- a/sqlmesh/core/console.py +++ b/sqlmesh/core/console.py @@ -49,6 +49,7 @@ from sqlmesh.utils.errors import ( PythonModelEvalError, NodeAuditsErrors, + SQLMeshError, format_destructive_change_msg, format_additive_change_msg, ) @@ -64,6 +65,8 @@ from sqlmesh.core.table_diff import TableDiff, RowDiff, SchemaDiff from sqlmesh.core.config.connection import ConnectionConfig from sqlmesh.core.state_sync import Versions + from sqlmesh.core.engine_adapter.shared import QueryHistoryRecord + from sqlmesh.core.environment import Environment LayoutWidget = t.TypeVar("LayoutWidget", bound=t.Union[widgets.VBox, widgets.HBox]) @@ -227,6 +230,19 @@ def print_environments(self, environments_summary: t.List[EnvironmentSummary]) - def show_intervals(self, snapshot_intervals: t.Dict[Snapshot, SnapshotIntervals]) -> None: """Show ready intervals""" + @abc.abstractmethod + def select_plan(self, environments: t.List[Environment]) -> t.Tuple[str, Environment]: + """Prompts the user to select a plan to inspect from the given environments.""" + + @abc.abstractmethod + def show_history( + self, + records: t.List[QueryHistoryRecord], + plan_id: str, + environment: t.Optional[Environment] = None, + ) -> None: + """Show the query engine's history of everything SQLMesh ran for a plan.""" + class DifferenceConsole(abc.ABC): """Console for displaying environment differences""" @@ -879,6 +895,19 @@ def print_environments(self, environments_summary: t.List[EnvironmentSummary]) - def show_intervals(self, snapshot_intervals: t.Dict[Snapshot, SnapshotIntervals]) -> None: pass + def select_plan(self, environments: t.List[Environment]) -> t.Tuple[str, Environment]: + raise SQLMeshError( + "Cannot select a plan interactively with this console; pass plan_id explicitly." + ) + + def show_history( + self, + records: t.List[QueryHistoryRecord], + plan_id: str, + environment: t.Optional[Environment] = None, + ) -> None: + pass + def show_linter_violations( self, violations: t.List[RuleViolation], model: Model, is_error: bool = False ) -> None: @@ -918,6 +947,41 @@ def make_progress_bar( ) +_SQL_OPERATION_KEYWORDS = ( + "CREATE OR REPLACE TABLE", + "CREATE OR REPLACE VIEW", + "CREATE TABLE", + "CREATE VIEW", + "CREATE SCHEMA", + "INSERT", + "MERGE", + "UPDATE", + "DELETE", + "ALTER", + "DROP", + "SELECT", + "LOAD", +) + + +def _sql_operation(sql: str) -> str: + """Best-effort extraction of the leading SQL keyword(s) for display, e.g. "CREATE TABLE". + + Strips SQLMesh's leading correlation id comment first. This is a simple prefix match, not a + real SQL parser - good enough for a debugging table, not for anything that needs to be exact. + """ + text = " ".join(sql.strip().split()) + if text.startswith("/*"): + end = text.find("*/") + if end != -1: + text = text[end + 2 :].strip() + upper = text.upper() + for keyword in _SQL_OPERATION_KEYWORDS: + if upper.startswith(keyword): + return keyword + return upper.split(" ", 1)[0] if upper else "UNKNOWN" + + class TerminalConsole(Console): """A rich based implementation of the console.""" @@ -2722,6 +2786,91 @@ def show_intervals(self, snapshot_intervals: t.Dict[Snapshot, SnapshotIntervals] if incomplete.children: self._print(incomplete) + def select_plan(self, environments: t.List[Environment]) -> t.Tuple[str, Environment]: + if not environments: + raise SQLMeshError("No environments found in state to inspect.") + + ordered = sorted(environments, key=lambda e: e.finalized_ts or 0, reverse=True) + + labels = [] + for env in ordered: + applied = time_like_to_str(env.finalized_ts) if env.finalized_ts else "in progress" + labels.append( + f"{env.name} plan {env.plan_id[:8]} applied {applied} " + f"({len(env.snapshots)} models)" + ) + + response = self._prompt( + "\n".join([f"[{i + 1}] {label}" for i, label in enumerate(labels)]), + show_choices=False, + choices=[f"{i + 1}" for i in range(len(ordered))], + ) + chosen = ordered[int(response) - 1] + return chosen.plan_id, chosen + + def show_history( + self, + records: t.List[QueryHistoryRecord], + plan_id: str, + environment: t.Optional[Environment] = None, + ) -> None: + succeeded = sum(1 for r in records if r.status == "success") + failed = sum(1 for r in records if r.status == "failed") + running = sum(1 for r in records if r.status == "running") + + env_label = f" · {environment.name}" if environment is not None else "" + header = ( + f"[b]History · plan {plan_id[:8]}{env_label}[/b] · " + f"{len(records)} queries · {succeeded} ✓ {failed} ✗ {running} running" + ) + + if not records: + self.log_status_update( + f"{header}\n[yellow]No queries found for this plan in the engine's history " + "(the plan id may be incorrect, it may have aged out of the history window, " + "or nothing ran).[/yellow]" + ) + return + + glyph = { + "success": "[green]✓[/green]", + "failed": "[red]✗[/red]", + "running": "[yellow]…[/yellow]", + } + table = Table(title=header) + table.add_column("Time", no_wrap=True) + table.add_column("Status", no_wrap=True) + table.add_column("Operation", no_wrap=True) + table.add_column("Target") + table.add_column("Duration", justify="right", no_wrap=True) + table.add_column("Bytes/Rows", justify="right", no_wrap=True) + + for record in records: + started = record.started_at.strftime("%H:%M:%S") if record.started_at else "-" + duration = f"{record.duration_ms} ms" if record.duration_ms is not None else "-" + if record.bytes_processed is not None: + size = f"{record.bytes_processed:,} bytes" + elif record.rows is not None: + size = f"{record.rows:,} rows" + else: + size = "-" + table.add_row( + started, + glyph.get(record.status, record.status), + _sql_operation(record.sql), + record.target or "-", + duration, + size, + ) + if record.error: + table.add_row("", "", f"[red]{record.error}[/red]", "", "", "") + + self._print(table) + if failed: + self.log_status_update( + f"[red]{failed} failed[/red] · re-run with `-o failures.sql` to export the SQL." + ) + def print_connection_config(self, config: ConnectionConfig, title: str = "Connection") -> None: tree = Tree(f"[b]{title}:[/b]") tree.add(f"Type: [bold cyan]{config.type_}[/bold cyan]") diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py index 5902977331..43f5ef1deb 100644 --- a/sqlmesh/core/context.py +++ b/sqlmesh/core/context.py @@ -118,7 +118,7 @@ filter_tests_by_patterns, ) from sqlmesh.core.user import User -from sqlmesh.utils import UniqueKeyDict, Verbosity +from sqlmesh.utils import CorrelationId, UniqueKeyDict, Verbosity from sqlmesh.utils.concurrency import concurrent_apply_to_values from sqlmesh.utils.dag import DAG from sqlmesh.utils.date import ( @@ -2458,6 +2458,57 @@ def check_intervals( return results + @python_api_analytics + def plan_history( + self, + plan_id: t.Optional[str] = None, + environment: t.Optional[str] = None, + output_file: t.Optional[Path] = None, + ) -> None: + """Show the query engine history of everything SQLMesh ran for a plan. + + Args: + plan_id: The plan to inspect. If None, an interactive menu of known plans is shown. + environment: Restrict the plan menu to this environment. Defaults to all environments. + output_file: If set, export the executed SQL to this file instead of printing it. + """ + environments = self.state_reader.get_environments() + if environment: + environments = [env for env in environments if env.name == environment] + if not environments: + raise SQLMeshError(f"Environment '{environment}' was not found.") + + env: t.Optional[Environment] + if plan_id is None: + plan_id, env = self.console.select_plan(environments) + else: + env = next( + (e for e in environments if plan_id in (e.plan_id, e.previous_plan_id)), + None, + ) + + # NOTE: uses the default gateway adapter; per-environment gateways are a follow-up. + records = self.engine_adapter.query_history(CorrelationId.from_plan_id(plan_id)) + + if output_file: + lines: t.List[str] = [] + for record in records: + started = record.started_at.isoformat() if record.started_at else "unknown" + header = f"-- [{started}] {record.status.upper()}" + if record.duration_ms is not None: + header += f" ({record.duration_ms} ms)" + if record.error: + header += f" error: {record.error}" + lines.append(header) + lines.append(f"{record.sql.strip().rstrip(';')};\n") + try: + output_file.write_text("\n".join(lines)) + except OSError as e: + raise SQLMeshError(f"Failed to write history to '{output_file}': {e}") from e + self.console.log_success(f"Exported {len(records)} queries to {output_file}") + else: + self.console.show_history(records, plan_id, env) + @python_api_analytics def migrate(self) -> None: """Migrates SQLMesh to the current running version. diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py index 54a27c0920..4ee6075683 100644 --- a/sqlmesh/core/engine_adapter/base.py +++ b/sqlmesh/core/engine_adapter/base.py @@ -35,6 +35,7 @@ DataObjectType, EngineRunMode, InsertOverwriteStrategy, + QueryHistoryRecord, SourceQuery, set_catalog, ) @@ -122,6 +123,7 @@ class EngineAdapter: MAX_IDENTIFIER_LENGTH: t.Optional[int] = None ATTACH_CORRELATION_ID = True SUPPORTS_QUERY_EXECUTION_TRACKING = False + SUPPORTS_QUERY_HISTORY = False SUPPORTS_METADATA_TABLE_LAST_MODIFIED_TS = False RESOLVE_TABLE_REFS_IN_PHYSICAL_PROPERTIES: t.FrozenSet[str] = frozenset() """Physical property keys whose values may contain logical model references that @@ -2635,6 +2637,13 @@ def _attach_correlation_id(self, sql: str) -> str: return f"/* {self.correlation_id} */ {sql}" return sql + def query_history(self, correlation_id: CorrelationId) -> t.List[QueryHistoryRecord]: + """Return the queries executed under the given correlation id from the engine's history. + + Only supported on engines that expose a queryable query history (SUPPORTS_QUERY_HISTORY). + """ + raise SQLMeshError(f"Query history is not supported for the '{self.dialect}' engine.") + def _log_sql( self, sql: str, diff --git a/sqlmesh/core/engine_adapter/bigquery.py b/sqlmesh/core/engine_adapter/bigquery.py index d136445114..04333309b4 100644 --- a/sqlmesh/core/engine_adapter/bigquery.py +++ b/sqlmesh/core/engine_adapter/bigquery.py @@ -19,15 +19,16 @@ CatalogSupport, DataObject, DataObjectType, + QueryHistoryRecord, SourceQuery, set_catalog, InsertOverwriteStrategy, ) from sqlmesh.core.node import IntervalUnit from sqlmesh.core.schema_diff import TableAlterOperation, NestedSupport -from sqlmesh.utils import optional_import, get_source_columns_to_types +from sqlmesh.utils import optional_import, get_source_columns_to_types, CorrelationId from sqlmesh.utils.date import to_datetime -from sqlmesh.utils.errors import SQLMeshError +from sqlmesh.utils.errors import SQLMeshError, QueryHistoryPermissionError from sqlmesh.utils.pandas import columns_to_types_from_dtypes if t.TYPE_CHECKING: @@ -74,6 +75,7 @@ class BigQueryEngineAdapter(ClusteredByMixin, RowDiffMixin, GrantsFromInfoSchema MAX_TABLE_COMMENT_LENGTH = 1024 MAX_COLUMN_COMMENT_LENGTH = 1024 SUPPORTS_QUERY_EXECUTION_TRACKING = True + SUPPORTS_QUERY_HISTORY = True SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["SCHEMA"] INSERT_OVERWRITE_STRATEGY = InsertOverwriteStrategy.MERGE @@ -130,6 +132,16 @@ def bigframe(self) -> t.Optional[BigframeSession]: return bigframes.connect(context=options) return None + def _correlation_labels(self) -> t.Dict[str, str]: + """BigQuery job labels tagging a job with the current correlation id (keys must be lowercase). + + Used for both query jobs (`_job_params`) and dataframe/seed load jobs so that everything + SQLMesh runs for a plan is discoverable via `query_history`. + """ + if self.correlation_id: + return {self.correlation_id.job_type.value.lower(): self.correlation_id.job_id} + return {} + @property def _job_params(self) -> t.Dict[str, t.Any]: from sqlmesh.core.config.connection import BigQueryPriority @@ -144,10 +156,9 @@ def _job_params(self) -> t.Dict[str, t.Any]: params["maximum_bytes_billed"] = self._extra_config.get("maximum_bytes_billed") if self._extra_config.get("reservation") is not None: params["reservation"] = self._extra_config.get("reservation") - if self.correlation_id: - # BigQuery label keys must be lowercase - key = self.correlation_id.job_type.value.lower() - params["labels"] = {key: self.correlation_id.job_id} + labels = self._correlation_labels() + if labels: + params["labels"] = labels return params @property @@ -470,6 +481,139 @@ def fetchall( ) return list(self._query_data) + def _query_history_location(self, project: str) -> t.Optional[str]: + """Best-effort BigQuery region for INFORMATION_SCHEMA.JOBS: the connection's configured + location, falling back to the location of the first dataset in the project.""" + if self.client.location: + return self.client.location + datasets = list(self._db_call(self.client.list_datasets, project=project, max_results=1)) + if datasets: + return self._get_bq_dataset_location(project, datasets[0].dataset_id) + return None + + def query_history(self, correlation_id: CorrelationId) -> t.List[QueryHistoryRecord]: + """Return the queries executed under `correlation_id` from BigQuery's INFORMATION_SCHEMA.JOBS. + + Jobs are matched on the label SQLMesh attaches to every job it runs (see `_correlation_labels`). + Both query jobs and dataframe/seed load jobs are captured. Reading project-wide job history + requires the `bigquery.jobs.listAll` permission. + + Note: signals (Python readiness checks) and pure-Python model logic execute no SQL, so they + do not appear here. + """ + from google.api_core.exceptions import Forbidden + + project = self.get_current_catalog() + location = self._query_history_location(project) if project else None + if not project or not location: + raise SQLMeshError( + "Cannot read BigQuery query history: could not determine the project or its " + "region. Set the 'location' field on the BigQuery gateway connection and retry." + ) + + # https://cloud.google.com/bigquery/docs/information-schema-jobs + jobs_table = exp.to_table( + f"`{project}`.`region-{location.lower()}`.INFORMATION_SCHEMA.JOBS", + dialect=self.dialect, + ) + # BigQuery label keys must be lowercase, matching how `_job_params` sets them. + label_key = correlation_id.job_type.value.lower() + label_match = exp.Exists( + this=exp.select(exp.Literal.number(1)) + .from_( + exp.Unnest( + expressions=[exp.column("labels")], + alias=exp.TableAlias(columns=[exp.to_identifier("l")]), + ) + ) + .where( + exp.and_( + exp.column("key", table="l").eq(exp.Literal.string(label_key)), + exp.column("value", table="l").eq(exp.Literal.string(correlation_id.job_id)), + ) + ) + ) + query = ( + exp.select( + exp.column("start_time"), + exp.column("end_time"), + exp.column("query"), + exp.column("state"), + exp.column("message", table="error_result").as_("error_message"), + exp.column("total_bytes_processed"), + exp.column("job_id"), + exp.column("dataset_id", table="destination_table").as_("dest_dataset"), + exp.column("table_id", table="destination_table").as_("dest_table"), + ) + .from_(jobs_table) + .where(label_match) + # COALESCE so queued jobs (NULL start_time) sort chronologically, not first. + .order_by(exp.func("COALESCE", exp.column("start_time"), exp.column("creation_time"))) + ) + + try: + rows = self.fetchall(query) + except Forbidden as e: + raise QueryHistoryPermissionError( + f"Permission denied reading BigQuery job history for project '{project}'. " + "Reading all jobs in the project requires the 'bigquery.jobs.listAll' permission.", + remediation=( + f"gcloud projects add-iam-policy-binding {project} " + "--member='user:YOUR_EMAIL' --role='roles/bigquery.resourceViewer'" + ), + docs_url="https://cloud.google.com/bigquery/docs/information-schema-jobs#required_permissions", + ) from e + + records: t.List[QueryHistoryRecord] = [] + for ( + start_time, + end_time, + sql, + state, + error_message, + total_bytes, + job_id, + dest_dataset, + dest_table, + ) in rows: + # BigQuery routes SELECTs to anonymous result tables (dataset "_...", table "anon..."); + # only surface a real destination the statement actually wrote to. + target = None + if ( + dest_table + and dest_dataset + and not dest_dataset.startswith("_") + and not dest_table.startswith("anon") + ): + target = f"{dest_dataset}.{dest_table}" + if not sql: + # Load jobs (dataframe/seed loads) carry no SQL text. + sql = f"LOAD INTO {target}" if target else "LOAD" + if error_message: + status = "failed" + elif state in ("RUNNING", "PENDING"): + status = "running" + else: + status = "success" + duration_ms = ( + int((end_time - start_time).total_seconds() * 1000) + if start_time and end_time + else None + ) + records.append( + QueryHistoryRecord( + started_at=start_time, + sql=sql, + status=status, + duration_ms=duration_ms, + bytes_processed=int(total_bytes) if total_bytes is not None else None, + error=error_message, + query_id=job_id, + target=target, + ) + ) + return records + def _split_alter_expressions( self, alter_expressions: t.List[exp.Alter], @@ -582,6 +726,10 @@ def __load_pandas_to_table( from google.cloud import bigquery job_config = bigquery.job.LoadJobConfig(schema=self.__get_bq_schema(columns_to_types)) + labels = self._correlation_labels() + if labels: + # Label load jobs too, so Python-model dataframe loads and seeds show up in query_history. + job_config.labels = labels if replace: job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE logger.info(f"Loading dataframe to BigQuery. Table Path: {table.path}") diff --git a/sqlmesh/core/engine_adapter/shared.py b/sqlmesh/core/engine_adapter/shared.py index ba0e1fa619..191fe900a3 100644 --- a/sqlmesh/core/engine_adapter/shared.py +++ b/sqlmesh/core/engine_adapter/shared.py @@ -5,6 +5,8 @@ import logging import types import typing as t +from dataclasses import dataclass +from datetime import datetime from enum import Enum from pydantic import Field @@ -299,6 +301,24 @@ def __exit__( return None +@dataclass(frozen=True) +class QueryHistoryRecord: + """A single query SQLMesh executed, reconstructed from the engine's query history. + + Returned by `EngineAdapter.query_history` and rendered by the `sqlmesh history` command. + """ + + started_at: t.Optional[datetime] + sql: str + status: str # "success", "failed", or "running" + duration_ms: t.Optional[int] = None + bytes_processed: t.Optional[int] = None + rows: t.Optional[int] = None + error: t.Optional[str] = None + query_id: t.Optional[str] = None + target: t.Optional[str] = None # physical table the statement wrote to, if any + + def set_catalog(override_mapping: t.Optional[t.Dict[str, CatalogSupport]] = None) -> t.Callable: def set_catalog_decorator( func: t.Callable, diff --git a/sqlmesh/utils/errors.py b/sqlmesh/utils/errors.py index ca3e1bfb05..94816a0a0e 100644 --- a/sqlmesh/utils/errors.py +++ b/sqlmesh/utils/errors.py @@ -190,6 +190,27 @@ class EngineAdapterError(SQLMeshError): pass +class QueryHistoryPermissionError(EngineAdapterError): + def __init__( + self, + message: str, + remediation: t.Optional[str] = None, + docs_url: t.Optional[str] = None, + ) -> None: + super().__init__(message) + self.remediation = remediation + self.docs_url = docs_url + + def __str__(self) -> str: + # The CLI only surfaces str(exc), so fold the remediation and docs into the message. + parts = [super().__str__()] + if self.remediation: + parts.append(f"\n\nTo grant access, run:\n {self.remediation}") + if self.docs_url: + parts.append(f"\n\nDocs: {self.docs_url}") + return "".join(parts) + + class UnsupportedCatalogOperationError(EngineAdapterError): pass diff --git a/tests/core/engine_adapter/integration/test_integration_bigquery.py b/tests/core/engine_adapter/integration/test_integration_bigquery.py index 0a6dd6b2a4..0a1e017803 100644 --- a/tests/core/engine_adapter/integration/test_integration_bigquery.py +++ b/tests/core/engine_adapter/integration/test_integration_bigquery.py @@ -469,3 +469,38 @@ def test_correlation_id_in_job_labels(ctx: TestContext): labels = adapter._job_params.get("labels") correlation_id = CorrelationId.from_plan_id(plan.plan_id) assert labels == {correlation_id.job_type.value.lower(): correlation_id.job_id} + + +def test_query_history(ctx: TestContext): + from sqlmesh.utils.errors import QueryHistoryPermissionError + + model_name = ctx.table("test") + + sqlmesh = ctx.create_context() + sqlmesh.upsert_model( + load_sql_based_model(d.parse(f"MODEL (name {model_name}, kind FULL); SELECT 1 AS col")) + ) + + plan_evaluator = BuiltInPlanEvaluator( + sqlmesh.state_sync, + sqlmesh.snapshot_evaluator, + sqlmesh.create_scheduler, + sqlmesh.default_catalog, + ) + plan: Plan = sqlmesh.plan_builder("prod", skip_tests=True).build() + plan_evaluator.evaluate(plan.to_evaluatable()) + adapter = t.cast(BigQueryEngineAdapter, plan_evaluator.snapshot_evaluator.adapter) + + correlation_id = CorrelationId.from_plan_id(plan.plan_id) + try: + # Exercises region resolution, the INFORMATION_SCHEMA.JOBS query, and read permissions + # against a real project. Empty-tolerant: JOBS has ingestion latency, so the jobs from the + # plan we just evaluated may not be queryable yet. + records = adapter.query_history(correlation_id) + except QueryHistoryPermissionError as e: + pytest.skip(f"Service account lacks BigQuery job-history access: {e}") + + assert isinstance(records, list) + for record in records: + assert record.status in ("success", "failed", "running") + assert record.sql diff --git a/tests/core/engine_adapter/test_base.py b/tests/core/engine_adapter/test_base.py index 2b9bcc665f..294e9549e3 100644 --- a/tests/core/engine_adapter/test_base.py +++ b/tests/core/engine_adapter/test_base.py @@ -15,7 +15,7 @@ from sqlmesh.core.engine_adapter import EngineAdapter, EngineAdapterWithIndexSupport from sqlmesh.core.engine_adapter.shared import InsertOverwriteStrategy, DataObject from sqlmesh.core.schema_diff import SchemaDiffer, TableAlterOperation, NestedSupport -from sqlmesh.utils import columns_to_types_to_struct +from sqlmesh.utils import columns_to_types_to_struct, CorrelationId from sqlmesh.utils.date import to_ds from sqlmesh.utils.errors import SQLMeshError, UnsupportedCatalogOperationError from tests.core.engine_adapter import to_sql_calls @@ -4170,3 +4170,13 @@ def test_get_current_grants_config_not_implemented(make_mocked_engine_adapter: t with pytest.raises(NotImplementedError): adapter._get_current_grants_config(relation) + + +def test_query_history_not_supported(make_mocked_engine_adapter: t.Callable): + adapter = make_mocked_engine_adapter(EngineAdapter, dialect="postgres") + assert adapter.SUPPORTS_QUERY_HISTORY is False + + with pytest.raises( + SQLMeshError, match="Query history is not supported for the 'postgres' engine." + ): + adapter.query_history(CorrelationId.from_plan_id("abc123")) diff --git a/tests/core/engine_adapter/test_bigquery.py b/tests/core/engine_adapter/test_bigquery.py index 134f144df1..5eb66ad2fa 100644 --- a/tests/core/engine_adapter/test_bigquery.py +++ b/tests/core/engine_adapter/test_bigquery.py @@ -13,10 +13,10 @@ import sqlmesh.core.dialect as d from sqlmesh.core.engine_adapter import BigQueryEngineAdapter from sqlmesh.core.engine_adapter.bigquery import select_partitions_expr -from sqlmesh.core.engine_adapter.shared import DataObjectType +from sqlmesh.core.engine_adapter.shared import DataObjectType, QueryHistoryRecord from sqlmesh.core.node import IntervalUnit -from sqlmesh.utils import AttributeDict -from sqlmesh.utils.errors import SQLMeshError +from sqlmesh.utils import AttributeDict, CorrelationId +from sqlmesh.utils.errors import QueryHistoryPermissionError, SQLMeshError pytestmark = [pytest.mark.bigquery, pytest.mark.engine] @@ -1380,3 +1380,164 @@ def test_sync_grants_config_no_schema( with pytest.raises(ValueError, match="Table test_table does not have a schema \\(dataset\\)"): adapter.sync_grants_config(relation, new_grants_config) + + +def test_query_history_builds_labels_filter_sql( + make_mocked_engine_adapter: t.Callable, mocker: MockerFixture +): + adapter = make_mocked_engine_adapter(BigQueryEngineAdapter) + fetchall_mock = mocker.patch.object(adapter, "fetchall", return_value=[]) + mocker.patch.object(adapter, "get_current_catalog", return_value="project") + mocker.patch.object(adapter.client, "location", "us-central1") + + adapter.query_history(CorrelationId.from_plan_id("abc123")) + + fetchall_mock.assert_called_once() + executed_query = fetchall_mock.call_args[0][0] + executed_sql = executed_query.sql(dialect="bigquery") + expected_sql = ( + "SELECT start_time, end_time, query, state, error_result.message AS error_message, " + "total_bytes_processed, job_id, destination_table.dataset_id AS dest_dataset, " + "destination_table.table_id AS dest_table " + "FROM `project`.`region-us-central1`.`INFORMATION_SCHEMA.JOBS` AS JOBS " + "WHERE EXISTS(SELECT 1 FROM UNNEST(labels) AS l WHERE l.key = 'sqlmesh_plan' AND l.value = 'abc123') " + "ORDER BY COALESCE(start_time, creation_time)" + ) + assert executed_sql == expected_sql + + +@pytest.mark.parametrize( + "state, error_message, expected_status", + [ + ("DONE", None, "success"), + ("DONE", "Syntax error at line 1", "failed"), + ("RUNNING", None, "running"), + ("PENDING", None, "running"), + ], +) +def test_query_history_maps_job_status( + make_mocked_engine_adapter: t.Callable, + mocker: MockerFixture, + state: str, + error_message: t.Optional[str], + expected_status: str, +): + adapter = make_mocked_engine_adapter(BigQueryEngineAdapter) + start_time = datetime(2024, 1, 1, 10, 0, 0) + end_time = datetime(2024, 1, 1, 10, 0, 5) + mocker.patch.object( + adapter, + "fetchall", + return_value=[ + ( + start_time, + end_time, + "SELECT 1", + state, + error_message, + 1024, + "job-1", + "sushi", + "orders__abc", + ), + ], + ) + mocker.patch.object(adapter, "get_current_catalog", return_value="project") + mocker.patch.object(adapter.client, "location", "us-central1") + + records = adapter.query_history(CorrelationId.from_plan_id("abc123")) + + assert len(records) == 1 + record = records[0] + assert isinstance(record, QueryHistoryRecord) + assert record.status == expected_status + assert record.error == error_message + assert record.started_at == start_time + assert record.duration_ms == 5000 + assert record.bytes_processed == 1024 + assert record.query_id == "job-1" + assert record.sql == "SELECT 1" + assert record.target == "sushi.orders__abc" + + +def test_query_history_captures_load_jobs_and_skips_anon( + make_mocked_engine_adapter: t.Callable, mocker: MockerFixture +): + start_time = datetime(2024, 1, 1, 10, 0, 0) + end_time = datetime(2024, 1, 1, 10, 0, 3) + adapter = make_mocked_engine_adapter(BigQueryEngineAdapter) + mocker.patch.object( + adapter, + "fetchall", + return_value=[ + # load job: no SQL text, real destination -> becomes a LOAD row with a target + (start_time, end_time, None, "DONE", None, 4096, "job-load", "sushi", "seed_data__v1"), + # plain SELECT routed to an anonymous result table -> no target surfaced + (start_time, end_time, "SELECT 1", "DONE", None, 10, "job-anon", "_scratch", "anon9f8"), + ], + ) + mocker.patch.object(adapter, "get_current_catalog", return_value="project") + mocker.patch.object(adapter.client, "location", "us-central1") + + records = adapter.query_history(CorrelationId.from_plan_id("abc123")) + + assert records[0].sql == "LOAD INTO sushi.seed_data__v1" + assert records[0].target == "sushi.seed_data__v1" + assert records[1].target is None + + +def test_query_history_forbidden_raises_permission_error( + make_mocked_engine_adapter: t.Callable, mocker: MockerFixture +): + from google.api_core.exceptions import Forbidden + + adapter = make_mocked_engine_adapter(BigQueryEngineAdapter) + mocker.patch.object(adapter, "fetchall", side_effect=Forbidden("permission denied")) + mocker.patch.object(adapter, "get_current_catalog", return_value="project") + mocker.patch.object(adapter.client, "location", "us-central1") + + with pytest.raises(QueryHistoryPermissionError) as exc_info: + adapter.query_history(CorrelationId.from_plan_id("abc123")) + + assert "project" in str(exc_info.value) + assert exc_info.value.remediation + assert "project" in exc_info.value.remediation + assert exc_info.value.docs_url == ( + "https://cloud.google.com/bigquery/docs/information-schema-jobs#required_permissions" + ) + # The CLI only shows str(exc), so the remediation + docs must be folded into it. + assert exc_info.value.remediation in str(exc_info.value) + assert exc_info.value.docs_url in str(exc_info.value) + + +def test_query_history_missing_location_raises( + make_mocked_engine_adapter: t.Callable, mocker: MockerFixture +): + adapter = make_mocked_engine_adapter(BigQueryEngineAdapter) + mocker.patch.object(adapter, "get_current_catalog", return_value="project") + mocker.patch.object(adapter.client, "location", None) + mocker.patch.object(adapter, "_db_call", side_effect=lambda func, *a, **k: func(*a, **k)) + # No configured location and no datasets to resolve one from -> clear error. + mocker.patch.object(adapter.client, "list_datasets", return_value=[]) + + with pytest.raises(SQLMeshError, match="Cannot read BigQuery query history"): + adapter.query_history(CorrelationId.from_plan_id("abc123")) + + +def test_query_history_resolves_location_from_dataset( + make_mocked_engine_adapter: t.Callable, mocker: MockerFixture +): + adapter = make_mocked_engine_adapter(BigQueryEngineAdapter) + mocker.patch.object(adapter, "get_current_catalog", return_value="project") + mocker.patch.object(adapter.client, "location", None) + mocker.patch.object(adapter, "_db_call", side_effect=lambda func, *a, **k: func(*a, **k)) + mocker.patch.object( + adapter.client, "list_datasets", return_value=[mocker.Mock(dataset_id="analytics")] + ) + mocker.patch.object(adapter, "_get_bq_dataset_location", return_value="EU") + fetchall_mock = mocker.patch.object(adapter, "fetchall", return_value=[]) + + adapter.query_history(CorrelationId.from_plan_id("abc123")) + + # region resolved from the dataset and lowercased + assert "region-eu" in fetchall_mock.call_args[0][0].sql(dialect="bigquery") diff --git a/tests/core/test_console.py b/tests/core/test_console.py index f899713235..cca38cf160 100644 --- a/tests/core/test_console.py +++ b/tests/core/test_console.py @@ -1,4 +1,21 @@ -from sqlmesh.core.console import MarkdownConsole +import typing as t +from datetime import datetime +from io import StringIO + +import pytest +from rich.console import Console as RichConsole + +from sqlmesh.core.console import MarkdownConsole, TerminalConsole +from sqlmesh.core.engine_adapter.shared import QueryHistoryRecord +from sqlmesh.core.environment import Environment +from sqlmesh.utils.errors import SQLMeshError + + +def _create_test_console(width: t.Optional[int] = None) -> t.Tuple[StringIO, TerminalConsole]: + """Creates a console and buffer for validating console output.""" + console_output = StringIO() + console = RichConsole(file=console_output, force_terminal=True, width=width) + return console_output, TerminalConsole(console=console) def test_markdown_console_warning_block(): @@ -129,3 +146,72 @@ def test_markdown_console_error_block(): ) assert console.consume_captured_errors() == "" + + +def test_show_history(): + output, console = _create_test_console(width=200) + plan_id = "plan-id-123" + records = [ + QueryHistoryRecord( + started_at=datetime(2024, 1, 1, 10, 0, 0), + sql="CREATE TABLE foo AS SELECT 1", + status="success", + duration_ms=1500, + bytes_processed=2048, + target="sushi.orders__2837", + ), + QueryHistoryRecord( + started_at=datetime(2024, 1, 1, 10, 0, 5), + sql="INSERT INTO foo VALUES (1)", + status="failed", + error="Column mismatch", + target="sushi.items__9f01", + ), + QueryHistoryRecord( + started_at=None, + sql="MERGE INTO foo USING bar ON foo.id = bar.id", + status="running", + ), + ] + + console.show_history(records, plan_id, environment=None) + + printed = output.getvalue() + assert plan_id[:8] in printed + assert "3 queries" in printed + assert "CREATE TABLE" in printed + assert "INSERT" in printed + assert "MERGE" in printed + assert "Column mismatch" in printed + assert "re-run with" in printed + # target column attributes each step to its physical table + assert "Target" in printed + assert "sushi.orders__2837" in printed + + +def test_show_history_empty(): + output, console = _create_test_console() + console.show_history([], "plan-id-123", environment=None) + assert "No queries found" in output.getvalue() + # hints that a wrong plan id is a possible cause (debugging aid) + assert "incorrect" in output.getvalue() + + +def test_select_plan(mocker): + _, console = _create_test_console() + environments = [ + Environment(name="dev", start_at="2023-01-01", plan_id="plan-a", snapshots=[]), + Environment(name="prod", start_at="2023-01-01", plan_id="plan-b", snapshots=[]), + ] + mocker.patch.object(console, "_prompt", return_value="2") + + plan_id, env = console.select_plan(environments) + + assert plan_id == "plan-b" + assert env is environments[1] + + +def test_select_plan_empty_raises(): + _, console = _create_test_console() + with pytest.raises(SQLMeshError, match="No environments found"): + console.select_plan([]) diff --git a/tests/core/test_context.py b/tests/core/test_context.py index 365d31d3fd..916368e10a 100644 --- a/tests/core/test_context.py +++ b/tests/core/test_context.py @@ -35,6 +35,7 @@ from sqlmesh.core.console import create_console, get_console from sqlmesh.core.dialect import parse, schema_ from sqlmesh.core.engine_adapter.duckdb import DuckDBEngineAdapter +from sqlmesh.core.engine_adapter.shared import QueryHistoryRecord from sqlmesh.core.environment import Environment, EnvironmentNamingInfo, EnvironmentStatements from sqlmesh.core.plan.definition import Plan from sqlmesh.core.macros import MacroEvaluator, RuntimeStage @@ -469,12 +470,12 @@ def test_multi_gateway_catalog_aware_and_unsupported(tmp_path: Path, mocker): # Both models must have 3-level FQNs so MappingSchema nesting is uniform. # count(".") == 2 means 3 parts (catalog.db.table), i.e. a 3-level FQN. - assert duckdb_model.fqn.count(".") == 2, ( - f"Expected 3-level FQN for duckdb model, got: {duckdb_model.fqn}" - ) - assert ch_model.fqn.count(".") == 2, ( - f"Expected 3-level FQN for ch model, got: {ch_model.fqn}" - ) # 3 parts = 2 dots + assert ( + duckdb_model.fqn.count(".") == 2 + ), f"Expected 3-level FQN for duckdb model, got: {duckdb_model.fqn}" + assert ( + ch_model.fqn.count(".") == 2 + ), f"Expected 3-level FQN for ch model, got: {ch_model.fqn}" # 3 parts = 2 dots # Both models loaded into the same MappingSchema must not raise a nesting SchemaError. from sqlglot.schema import MappingSchema @@ -633,12 +634,12 @@ def test_multi_gateway_virtual_catalog_create_schema_strips_prefix(tmp_path: Pat # Both models must have 3-level FQNs (catalog.db.table → 2 dots) so MappingSchema nesting # is uniform and does not raise a SchemaError. - assert ch_model.fqn.count(".") == 2, ( - f"Expected 3-level FQN for ClickHouse model, got: {ch_model.fqn}" - ) - assert duckdb_model.fqn.count(".") == 2, ( - f"Expected 3-level FQN for DuckDB model, got: {duckdb_model.fqn}" - ) + assert ( + ch_model.fqn.count(".") == 2 + ), f"Expected 3-level FQN for ClickHouse model, got: {ch_model.fqn}" + assert ( + duckdb_model.fqn.count(".") == 2 + ), f"Expected 3-level FQN for DuckDB model, got: {duckdb_model.fqn}" from sqlglot.schema import MappingSchema @@ -672,9 +673,9 @@ def _capture_create_schema( assert len(create_schema_calls) == 1, "Expected exactly one _create_schema call" passed_schema = create_schema_calls[0] # The virtual catalog prefix must NOT appear in the SQL sent to the wire. - assert "__clickhouse_gw__" not in passed_schema, ( - f"Virtual catalog prefix should be stripped before reaching _create_schema, got: {passed_schema!r}" - ) + assert ( + "__clickhouse_gw__" not in passed_schema + ), f"Virtual catalog prefix should be stripped before reaching _create_schema, got: {passed_schema!r}" @pytest.mark.fast @@ -3004,6 +3005,72 @@ def test_check_intervals(sushi_context, mocker): assert tuple(intervals.values())[0].intervals +def test_plan_history_environment_not_found(sushi_context): + with pytest.raises(SQLMeshError, match="Environment 'dev' was not found"): + sushi_context.plan_history(environment="dev") + + +def test_plan_history_with_explicit_plan_id(sushi_context, mocker): + prod_env = sushi_context.state_reader.get_environment("prod") + records = [QueryHistoryRecord(started_at=None, sql="SELECT 1", status="success")] + query_history_mock = mocker.patch.object( + sushi_context.engine_adapter, "query_history", return_value=records + ) + show_history_mock = mocker.patch.object(sushi_context.console, "show_history") + + sushi_context.plan_history(plan_id=prod_env.plan_id) + + query_history_mock.assert_called_once() + assert query_history_mock.call_args[0][0].job_id == prod_env.plan_id + show_history_mock.assert_called_once_with(records, prod_env.plan_id, prod_env) + + +def test_plan_history_prompts_when_plan_id_missing(sushi_context, mocker): + prod_env = sushi_context.state_reader.get_environment("prod") + mocker.patch.object(sushi_context.engine_adapter, "query_history", return_value=[]) + select_plan_mock = mocker.patch.object( + sushi_context.console, "select_plan", return_value=(prod_env.plan_id, prod_env) + ) + show_history_mock = mocker.patch.object(sushi_context.console, "show_history") + + sushi_context.plan_history() + + select_plan_mock.assert_called_once() + show_history_mock.assert_called_once_with([], prod_env.plan_id, prod_env) + + +def test_plan_history_output_file(sushi_context, mocker, tmp_path): + prod_env = sushi_context.state_reader.get_environment("prod") + records = [ + QueryHistoryRecord( + started_at=datetime(2024, 1, 1, 10, 0, 0), + sql="SELECT 1;", + status="success", + duration_ms=10, + ), + QueryHistoryRecord(started_at=None, sql="SELECT 2", status="failed", error="boom"), + ] + mocker.patch.object(sushi_context.engine_adapter, "query_history", return_value=records) + log_success_mock = mocker.patch.object(sushi_context.console, "log_success") + + output_file = tmp_path / "history.sql" + sushi_context.plan_history(plan_id=prod_env.plan_id, output_file=output_file) + + content = output_file.read_text() + assert "SELECT 1;" in content + assert "SELECT 2;" in content + assert "boom" in content + log_success_mock.assert_called_once() + + +def test_plan_history_not_supported_by_engine(sushi_context): + prod_env = sushi_context.state_reader.get_environment("prod") + with pytest.raises( + SQLMeshError, match="Query history is not supported for the 'duckdb' engine." + ): + sushi_context.plan_history(plan_id=prod_env.plan_id) + + def test_audit(): context = Context(config=Config()) @@ -3696,9 +3763,9 @@ def test_uppercase_gateway_external_models(tmp_path): for model in context_uppercase.models.values() if model.name == "test_db.uppercase_gateway_table" ] - assert len(gateway_specific_models) == 1, ( - f"External model with lowercase gateway name should be found with uppercase gateway. Found {len(gateway_specific_models)} models" - ) + assert ( + len(gateway_specific_models) == 1 + ), f"External model with lowercase gateway name should be found with uppercase gateway. Found {len(gateway_specific_models)} models" # Verify external model without gateway is also found no_gateway_models = [ @@ -3706,9 +3773,9 @@ def test_uppercase_gateway_external_models(tmp_path): for model in context_uppercase.models.values() if model.name == "test_db.no_gateway_table" ] - assert len(no_gateway_models) == 1, ( - f"External model without gateway should be found. Found {len(no_gateway_models)} models" - ) + assert ( + len(no_gateway_models) == 1 + ), f"External model without gateway should be found. Found {len(no_gateway_models)} models" # Check that the column types are properly loaded (not UNKNOWN) external_model = gateway_specific_models[0] @@ -3729,9 +3796,9 @@ def test_uppercase_gateway_external_models(tmp_path): if model.name == "test_db.uppercase_gateway_table" ] # This should work but might fail if case sensitivity is not handled correctly - assert len(gateway_specific_models_mixed) == 1, ( - f"External model should be found regardless of gateway parameter case. Found {len(gateway_specific_models_mixed)} models" - ) + assert ( + len(gateway_specific_models_mixed) == 1 + ), f"External model should be found regardless of gateway parameter case. Found {len(gateway_specific_models_mixed)} models" # Test a case that should demonstrate the potential issue: # Create another external model file with uppercase gateway name in the YAML @@ -3765,9 +3832,9 @@ def test_uppercase_gateway_external_models(tmp_path): for model in context_reloaded.models.values() if model.name == "test_db.uppercase_in_yaml" ] - assert len(uppercase_in_yaml_models) == 1, ( - f"External model with uppercase gateway in YAML should be found. Found {len(uppercase_in_yaml_models)} models" - ) + assert ( + len(uppercase_in_yaml_models) == 1 + ), f"External model with uppercase gateway in YAML should be found. Found {len(uppercase_in_yaml_models)} models" def test_plan_no_start_configured():