fix: support setting location in BigQuery Airflow op (#1558)

Author: eakmanrq

docs/integrations/engines/bigquery.md

@@ -46,7 +46,7 @@ pip install "sqlmesh[bigquery]"
 
 ### Connection info
 
-The operator requires an [Airflow connection](https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html) to determine the target BigQuery account. Please see [GoogleBaseHook](https://airflow.apache.org/docs/apache-airflow-providers-google/stable/_api/airflow/providers/google/common/hooks/base_google/index.html#airflow.providers.google.common.hooks.base_google.GoogleBaseHook) and [GCP connection](https://airflow.apache.org/docs/apache-airflow-providers-google/stable/connections/gcp.html)for more details. The goal is to create a working GCP connection on Airflow. The only difference is that you should use the `sqlmesh_google_cloud_bigquery_default` (by default) connection ID instead of the `google_cloud_default` one in the Airflow guide.
+The operator requires an [Airflow connection](https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html) to determine the target BigQuery account. Please see [GoogleBaseHook](https://airflow.apache.org/docs/apache-airflow-providers-google/stable/_api/airflow/providers/google/common/hooks/base_google/index.html#airflow.providers.google.common.hooks.base_google.GoogleBaseHook) and [GCP connection](https://airflow.apache.org/docs/apache-airflow-providers-google/stable/connections/gcp.html)for more details. Use the `sqlmesh_google_cloud_bigquery_default` (by default) connection ID instead of the `google_cloud_default` one in the Airflow guide.
 
 By default, the connection ID is set to `sqlmesh_google_cloud_bigquery_default`, but it can be overridden using the `engine_operator_args` parameter to the `SQLMeshAirflow` instance as in the example below:
 ```python linenums="1"
@@ -58,6 +58,20 @@ sqlmesh_airflow = SQLMeshAirflow(
 )
 ```
 
+#### Optional Arguments
+
+* `location`: Sets the default location for datasets and tables. If not set, BigQuery defaults to US for new datasets. See `location` in [Connection options](#connection-options) for more details.
+
+```python linenums="1"
+sqlmesh_airflow = SQLMeshAirflow(
+    "bigquery",
+    engine_operator_args={
+        "bigquery_conn_id": "<Connection ID>",
+        "location": "<location>"
+    },
+)
+```
+
 ## Connection Methods
 - [oauth](https://google-auth.readthedocs.io/en/master/reference/google.auth.html#google.auth.default) (default)
     - Related Credential Configuration:

sqlmesh/schedulers/airflow/hooks/bigquery.py

@@ -31,6 +31,7 @@ def __init__(
         gcp_conn_id: str = default_conn_name,
         delegate_to: str | None = None,
         impersonation_chain: str | t.Sequence[str] | None = None,
+        location: t.Optional[str] = None,
         **kwargs: t.Any,
     ) -> None:
         GoogleBaseHook.__init__(
@@ -39,9 +40,11 @@ def __init__(
             delegate_to=delegate_to,
             impersonation_chain=impersonation_chain,
         )
+        self.location = location
 
     def get_conn(self) -> Connection:
         """Returns a BigQuery DBAPI connection object."""
+        from google.api_core.client_info import ClientInfo
         from google.cloud.bigquery import Client
         from google.cloud.bigquery.dbapi import Connection
 
@@ -50,5 +53,10 @@ def get_conn(self) -> Connection:
             creds, project_id = self._get_credentials_and_project_id()  # type: ignore
         except AttributeError:
             creds, project_id = self.get_credentials_and_project_id()  # type: ignore
-        client = Client(project=project_id, credentials=creds)
+        client = Client(
+            project=project_id,
+            credentials=creds,
+            location=self.location,
+            client_info=ClientInfo(user_agent="sqlmesh"),
+        )
         return Connection(client=client)