feat(bigquery): Add support for flexible job reservations

[dejii]

resolves #1228

Problem

BigQuery recently introduced the ability to assign reservations at query runtime. Previously, dbt did not expose any way to specify a reservation when submitting a query job.

Solution

This PR adds support for setting the reservation field on a BigQuery job if one is provided in the adapter credentials. The value is optional and does not change default behaviour for existing users.

This unlocks the ability to define multiple targets in profiles.yml, each configured with a different reservation. This allows teams to route different workloads such as business-critical vs. non-critical jobs to separate slot reservations within the same project.

Checklist

• I have read the contributing guide and understand what's expected of me
• I have run this code in development and it appears to resolve the stated issue
• This PR includes tests, or tests are not required/relevant for this PR
• This PR has no interface changes (e.g. macros, cli, logs, json artifacts, config files, adapter interface, etc) or this PR has already received feedback and approval from Product or DX

[austinweisgrau]

Just in case anyone else is anxiously awaiting this to be merged but don't want to bundle this feature branch into prod, you can also monkey-patch this in if you're using the python runner rather than the dbt CLI.

This looks like

from dbt.adapters.bigquery import connections
from dbt.adapters.bigquery.connections import (
    Priority,
    QueryPriority,
    SQLQuery,
    fire_event,
    get_invocation_id,
    get_node_info,
)

def raw_execute(
    self,
    sql,
    use_legacy_sql=False,
    limit: int | None = None,
    dry_run: bool = False,
):
    conn = self.get_thread_connection()

    fire_event(SQLQuery(conn_name=conn.name, sql=sql, node_info=get_node_info()))

    labels = self.get_labels_from_query_comment()

    labels["dbt_invocation_id"] = get_invocation_id()

    job_params = {
        "use_legacy_sql": use_legacy_sql,
        "labels": labels,
        "dry_run": dry_run,
    }

    priority = conn.credentials.priority
    if priority == Priority.Batch:
        job_params["priority"] = QueryPriority.BATCH
    else:
        job_params["priority"] = QueryPriority.INTERACTIVE

    maximum_bytes_billed = conn.credentials.maximum_bytes_billed
    if maximum_bytes_billed is not None and maximum_bytes_billed != 0:
        job_params["maximum_bytes_billed"] = maximum_bytes_billed

    # Here's the important line
    job_params["reservation"] = "/projects/your/reservation/id"

    with self.exception_handler(sql):
        job_id = self.generate_job_id()

        return self._query_and_results(
            conn,
            sql,
            job_params,
            job_id,
            limit=limit,
        )

# Monkey patch the method
connections.BigQueryConnectionManager.raw_execute = raw_execute

then you can use

from dbt.cli.main import dbtRunner

dbtRunner().invoke(...)

[austinweisgrau]

I believe this is an incomplete solution. After using this method in production for about a month, it looks like only about half of our dbt query activity in BigQuery is being called with this method and put on the slot reservation. dbt seems to be using another method for a substantial portion of its query activity.