Add missing location param to BigQueryUpdateTableSchemaOperator (#40237)

Author: shahar1

airflow/providers/google/cloud/operators/bigquery.py

@@ -2719,6 +2719,7 @@ def __init__(
         project_id: str = PROVIDE_PROJECT_ID,
         gcp_conn_id: str = "google_cloud_default",
         impersonation_chain: str | Sequence[str] | None = None,
+        location: str | None = None,
         **kwargs,
     ) -> None:
         self.schema_fields_updates = schema_fields_updates
@@ -2728,12 +2729,12 @@ def __init__(
         self.project_id = project_id
         self.gcp_conn_id = gcp_conn_id
         self.impersonation_chain = impersonation_chain
+        self.location = location
         super().__init__(**kwargs)
 
     def execute(self, context: Context):
         bq_hook = BigQueryHook(
-            gcp_conn_id=self.gcp_conn_id,
-            impersonation_chain=self.impersonation_chain,
+            gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain, location=self.location
         )
 
         table = bq_hook.update_table_schema(

tests/providers/google/cloud/operators/test_bigquery.py

@@ -458,9 +458,16 @@ def test_execute(self, mock_hook):
             dataset_id=TEST_DATASET,
             table_id=TEST_TABLE_ID,
             project_id=TEST_GCP_PROJECT_ID,
+            location=TEST_DATASET_LOCATION,
+            impersonation_chain=["[email protected]"],
         )
         operator.execute(context=MagicMock())
 
+        mock_hook.assert_called_once_with(
+            gcp_conn_id=GCP_CONN_ID,
+            impersonation_chain=["[email protected]"],
+            location=TEST_DATASET_LOCATION,
+        )
         mock_hook.return_value.update_table_schema.assert_called_once_with(
             schema_fields_updates=schema_field_updates,
             include_policy_tags=False,