dbt bigquery/add sql query status (#1327)

Author: colin-rogers-dbt

dbt-bigquery/.changes/unreleased/Under the Hood-20250908-114952.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Emit SQLQueryStatus during query execution
+time: 2025-09-08T11:49:52.584547-07:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "1327"

dbt-bigquery/src/dbt/adapters/bigquery/connections.py

@@ -5,6 +5,7 @@
 import json
 from multiprocessing.context import SpawnContext
 import re
+import time
 from typing import Dict, Hashable, List, Optional, Tuple, TYPE_CHECKING
 import uuid
 
@@ -34,7 +35,7 @@
     ConnectionState,
 )
 from dbt.adapters.events.logging import AdapterLogger
-from dbt.adapters.events.types import SQLQuery
+from dbt.adapters.events.types import SQLQuery, SQLQueryStatus
 from dbt.adapters.exceptions.connection import FailedToConnectError
 from dbt.adapters.bigquery.clients import create_bigquery_client
 from dbt.adapters.bigquery.credentials import Priority
@@ -607,11 +608,22 @@ def _query_and_results(
                 self._bq_job_link(query_job.location, query_job.project, query_job.job_id)
             )
 
+        pre = time.perf_counter()
         try:
             iterator = query_job.result(max_results=limit)
         except TimeoutError:
             exc = f"Operation did not complete within the designated timeout of {timeout} seconds."
             raise TimeoutError(exc)
+
+        fire_event(
+            SQLQueryStatus(
+                status="OK",
+                elapsed=time.perf_counter() - pre,
+                node_info=get_node_info(),
+                query_id=query_job.job_id,
+            )
+        )
+
         return query_job, iterator
 
     def _labels_from_query_comment(self, comment: str) -> Dict:

dbt-bigquery/tests/unit/test_bigquery_connection_manager.py

@@ -87,6 +87,7 @@ def test_drop_dataset(self):
 
     @patch("dbt.adapters.bigquery.connections.QueryJobConfig")
     def test_query_and_results(self, MockQueryJobConfig):
+        self.mock_client.query.return_value = Mock(job_id="1")
         self.connections._query_and_results(
             self.mock_connection,
             "sql",