feat(cli): add sqlmesh history to inspect a plan's query engine history

Adds a read-only `sqlmesh history [PLAN_ID]` command that reconstructs
everything SQLMesh ran for a plan by querying the query engine's native
query history (matched on the correlation id SQLMesh attaches to every
query), rendering it chronologically with success/failure, duration, and
bytes -- or exporting the SQL with `-o`. This removes the need to context
-switch into the warehouse's query history or SQLMesh state when debugging.

- BigQuery (v1): INFORMATION_SCHEMA.JOBS, matched on the `sqlmesh_plan`
  job label; region resolved dynamically from the project; a clear,
  actionable permission error (gcloud remediation + docs) on jobs.listAll.
- Interactive numbered plan menu sourced from state, or pass a plan id.
- Other engines raise a clear "not supported" message.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: sungchun12

docs/guides/history.md

@@ -0,0 +1,118 @@
+# 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 query SQLMesh runs for a plan is tagged with that plan's correlation id — a BigQuery job label on the query job itself. `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 <plan_id>` prints a table with one row per query, in the order SQLMesh ran them:
+
+```bash linenums="1"
+$ sqlmesh history 3f9a2c1e
+
+                    History · plan 3f9a2c1e · 41 queries · 38 ✓  2 ✗  1 running
+┏━━━━━━━━━━┳━━━━━━━━┳━━━━━━━━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━━━┓
+┃ Time     ┃ Status ┃ Operation      ┃ Duration ┃   Bytes/Rows ┃
+┡━━━━━━━━━━╇━━━━━━━━╇━━━━━━━━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━━━┩
+│ 14:03:11 │ ✓      │ CREATE TABLE   │ 1200 ms  │  1,048,576 bytes │
+│ 14:03:14 │ ✓      │ INSERT         │ 14800 ms │  8,912,441 bytes │
+│ 14:03:29 │ ✗      │ INSERT         │ 420 ms   │            - │
+│          │        │ Syntax error: unexpected identifier 'PRICE' at [1:34] │       │              │
+│ 14:04:02 │ …      │ MERGE          │ -        │            - │
+└──────────┴────────┴────────────────┴──────────┴──────────────┘
+2 failed · re-run with `-o failures.sql` to export the SQL.
+```
+
+The header line summarizes the plan: the short plan id, total query count, and a breakdown of how many queries succeeded (`✓`), failed (`✗`), or are still running (`…`).
+
+- **Time** &mdash; when the query started.
+- **Status** &mdash; `✓` succeeded, `✗` failed, `…` still running.
+- **Operation** &mdash; the leading SQL keyword (`CREATE TABLE`, `INSERT`, `MERGE`, `CREATE VIEW`, `ALTER TABLE`, etc.), so you can scan for the kind of work each query did without reading the full statement.
+- **Duration** &mdash; wall-clock time the query took. Shown as `-` for queries that haven't finished.
+- **Bytes/Rows** &mdash; bytes processed if BigQuery reported them, otherwise rows affected, otherwise `-`.
+
+A failed query is followed by an indented line with the error message BigQuery returned, so you can see why it failed without opening the warehouse UI.
+
+## 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.
+- **Labeled query jobs only.** SQLMesh labels the query jobs it runs, but some operations &mdash; like loading a seed model's dataframe into BigQuery &mdash; use a load job rather than a labeled query job, so they won't appear in the history.
+- **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.

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
 
 ```

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:

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",

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,40 @@ 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",
+)
+
+
+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 +2785,89 @@ 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("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),
+                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]")