feat(history): attribute each job to its target table; capture load jobs

Follow-up to `sqlmesh history` addressing how interwoven Python models,
hooks, and signals appear:

- Add a `Target` column showing the physical table each job wrote to
  (from BigQuery's INFORMATION_SCHEMA.JOBS destination_table). This makes
  interleaved Python/SQL model steps and pre/post-hook statements legible
  instead of an anonymous stream of INSERT/CREATE. Anonymous SELECT result
  tables are omitted.
- Label BigQuery load jobs (LoadJobConfig) with the correlation id, so
  Python-model dataframe loads and seed loads appear as `LOAD` rows instead
  of being invisible. Factored the label out into `_correlation_labels`,
  shared by query and load jobs.
- Document that signals (Python readiness checks) and pure-Python model
  logic run no SQL and don't appear; point to `check_intervals`.

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

Author: sungchun12

docs/guides/history.md

@@ -11,7 +11,7 @@ The `sqlmesh history` command does that correlation for you. Give it a plan id a
 
 ## 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.
+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.
 
@@ -55,28 +55,32 @@ You only need to type enough of the plan id to uniquely identify it — the
 ```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.
+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.
 ```
 
-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 (`…`).
+(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 query started.
+- **Time** — when the job started.
 - **Status** — `✓` succeeded, `✗` failed, `…` still running.
-- **Operation** — 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** — wall-clock time the query took. Shown as `-` for queries that haven't finished.
+- **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 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.
+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
 
@@ -110,7 +114,7 @@ The file is one statement per entry, in execution order, so it can be read top t
     ```
 
 - **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 — like loading a seed model's dataframe into BigQuery — use a load job rather than a labeled query job, so they won't appear in the history.
+- **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

sqlmesh/core/console.py

@@ -960,6 +960,7 @@ def make_progress_bar(
     "ALTER",
     "DROP",
     "SELECT",
+    "LOAD",
 )
 
 
@@ -2840,6 +2841,7 @@ def show_history(
         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)
 
@@ -2856,11 +2858,12 @@ def show_history(
                 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]", "", "")
+                table.add_row("", "", f"[red]{record.error}[/red]", "", "", "")
 
         self._print(table)
         if failed:

sqlmesh/core/engine_adapter/bigquery.py

@@ -132,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
@@ -146,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
@@ -485,11 +494,12 @@ def _query_history_location(self, project: str) -> t.Optional[str]:
     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 query it runs (see `_job_params`).
-        Reading project-wide job history requires the `bigquery.jobs.listAll` permission.
+        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: only labeled query jobs are captured. Dataframe/seed load jobs
-        (`load_table_from_dataframe`) are not labeled, so they do not appear here.
+        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
 
@@ -532,6 +542,8 @@ def query_history(self, correlation_id: CorrelationId) -> t.List[QueryHistoryRec
                 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)
@@ -553,7 +565,30 @@ def query_history(self, correlation_id: CorrelationId) -> t.List[QueryHistoryRec
             ) from e
 
         records: t.List[QueryHistoryRecord] = []
-        for start_time, end_time, sql, state, error_message, total_bytes, job_id in rows:
+        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"):
@@ -574,6 +609,7 @@ def query_history(self, correlation_id: CorrelationId) -> t.List[QueryHistoryRec
                     bytes_processed=int(total_bytes) if total_bytes is not None else None,
                     error=error_message,
                     query_id=job_id,
+                    target=target,
                 )
             )
         return records
@@ -690,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}")

sqlmesh/core/engine_adapter/shared.py

@@ -316,6 +316,7 @@ class QueryHistoryRecord:
     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:

tests/core/engine_adapter/test_bigquery.py

@@ -1397,7 +1397,9 @@ def test_query_history_builds_labels_filter_sql(
     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 FROM `project`.`region-us-central1`.`INFORMATION_SCHEMA.JOBS` AS JOBS "
+        "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)"
     )
@@ -1427,7 +1429,17 @@ def test_query_history_maps_job_status(
         adapter,
         "fetchall",
         return_value=[
-            (start_time, end_time, "SELECT 1", state, error_message, 1024, "job-1"),
+            (
+                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")
@@ -1445,6 +1457,33 @@ def test_query_history_maps_job_status(
     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(

tests/core/test_console.py

@@ -11,10 +11,10 @@
 from sqlmesh.utils.errors import SQLMeshError
 
 
-def _create_test_console() -> t.Tuple[StringIO, TerminalConsole]:
+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)
+    console = RichConsole(file=console_output, force_terminal=True, width=width)
     return console_output, TerminalConsole(console=console)
 
 
@@ -149,7 +149,7 @@ def test_markdown_console_error_block():
 
 
 def test_show_history():
-    output, console = _create_test_console()
+    output, console = _create_test_console(width=200)
     plan_id = "plan-id-123"
     records = [
         QueryHistoryRecord(
@@ -158,12 +158,14 @@ def test_show_history():
             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,
@@ -182,6 +184,9 @@ def test_show_history():
     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():