feat(bigquery): Add support for flexible job reservations

dbt-bigquery/.changes/unreleased/Features-20250725-141647.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add support for job reservation config
+time: 2025-07-25T14:16:47.155349+01:00
+custom:
+    Author: dejii
+    Issue: "1228"

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

@@ -263,6 +263,10 @@ def raw_execute(
         if maximum_bytes_billed is not None and maximum_bytes_billed != 0:
             job_params["maximum_bytes_billed"] = maximum_bytes_billed
 
+        reservation = conn.credentials.reservation
+        if reservation:
+            job_params["reservation"] = reservation
+
         with self.exception_handler(sql):
             job_id = self.generate_job_id()
 

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

@@ -59,6 +59,7 @@ class BigQueryCredentials(Credentials):
     priority: Optional[Priority] = None
     maximum_bytes_billed: Optional[int] = None
     impersonate_service_account: Optional[str] = None
+    reservation: Optional[str] = None
 
     job_retry_deadline_seconds: Optional[int] = None
     job_retries: Optional[int] = 1
@@ -164,6 +165,7 @@ def _connection_keys(self):
             "dataproc_cluster_name",
             "gcs_bucket",
             "dataproc_batch",
+            "reservation",
         )
 
     @classmethod

dbt-bigquery/tests/unit/test_bigquery_adapter.py

@@ -100,6 +100,14 @@ def setUp(self):
                     "threads": 1,
                     "impersonate_service_account": "dummyaccount@dbt.iam.gserviceaccount.com",
                 },
+                "reservation": {
+                    "type": "bigquery",
+                    "method": "oauth",
+                    "project": "dbt-unit-000000",
+                    "schema": "dummy_schema",
+                    "threads": 1,
+                    "reservation": "projects/dbt-unit-000000/locations/US/reservations/test-reservation",
+                },
                 "oauth-credentials-token": {
                     "type": "bigquery",
                     "method": "oauth-secrets",
@@ -407,6 +415,22 @@ def test_acquire_connection_maximum_bytes_billed(self, mock_open_connection):
         connection.handle
         mock_open_connection.assert_called_once()
 
+    @patch("dbt.adapters.bigquery.BigQueryConnectionManager.open", return_value=_bq_conn())
+    def test_acquire_connection_reservation(self, mock_open_connection):
+        adapter = self.get_adapter("reservation")
+        try:
+            connection = adapter.acquire_connection("dummy")
+            self.assertEqual(connection.type, "bigquery")
+            self.assertEqual(
+                connection.credentials.reservation,
+                "projects/dbt-unit-000000/locations/US/reservations/test-reservation",
+            )
+        except dbt_common.exceptions.base.DbtValidationError as e:
+            self.fail("got DbtValidationError: {}".format(str(e)))
+        mock_open_connection.assert_not_called()
+        connection.handle
+        mock_open_connection.assert_called_once()
+
     def test_cancel_open_connections_empty(self):
         adapter = self.get_adapter("oauth")
         self.assertEqual(len(list(adapter.cancel_open_connections())), 0)