🐛 Destination BigQuery: Handle embedded project ID in dataset ID during normalization

airbyte-integrations/bases/base-normalization/normalization/transform_config/transform.py

@@ -128,10 +128,24 @@ def get_ssh_altered_config(config: Dict[str, Any], port_key: str = "port", host_
     def transform_bigquery(config: Dict[str, Any]):
         print("transform_bigquery")
         # https://docs.getdbt.com/reference/warehouse-profiles/bigquery-profile
+
+        project_id = config["project_id"]
+        dataset_id = config["dataset_id"]
+
+        if ":" in config["dataset_id"]:
+            splits = config["dataset_id"].split(":")
+            if len(splits) > 2:
+                raise ValueError("Invalid format for dataset ID (expected at most one colon)")
+            project_id, dataset_id = splits
+            if project_id != config["project_id"]:
+                raise ValueError(
+                    f"Project ID in dataset ID did not match explicitly-provided project ID: {project_id} and {config['project_id']}"
+                )
+
         dbt_config = {
             "type": "bigquery",
-            "project": config["project_id"],
-            "dataset": config["dataset_id"],
+            "project": project_id,
+            "dataset": dataset_id,
             "priority": config.get("transformation_priority", "interactive"),
             "threads": 8,
             "retries": 3,

airbyte-integrations/bases/base-normalization/unit_tests/test_transform_config.py

@@ -174,6 +174,41 @@ def test_transform_bigquery_no_credentials(self):
         assert expected_output == actual_output
         assert extract_schema(actual_output) == "my_dataset_id"
 
+    def test_transform_bigquery_with_embedded_project_id(self):
+        input = {"project_id": "my_project_id", "dataset_id": "my_project_id:my_dataset_id"}
+
+        actual_output = TransformConfig().transform_bigquery(input)
+        expected_output = {
+            "type": "bigquery",
+            "method": "oauth",
+            "project": "my_project_id",
+            "dataset": "my_dataset_id",
+            "priority": "interactive",
+            "retries": 3,
+            "threads": 8,
+        }
+
+        assert expected_output == actual_output
+        assert extract_schema(actual_output) == "my_dataset_id"
+
+    def test_transform_bigquery_with_embedded_mismatched_project_id(self):
+        input = {"project_id": "my_project_id", "dataset_id": "bad_project_id:my_dataset_id"}
+
+        try:
+            TransformConfig().transform_bigquery(input)
+            assert False, "transform_bigquery should have raised an exception"
+        except ValueError:
+            pass
+
+    def test_transform_bigquery_with_invalid_format(self):
+        input = {"project_id": "my_project_id", "dataset_id": "foo:bar:baz"}
+
+        try:
+            TransformConfig().transform_bigquery(input)
+            assert False, "transform_bigquery should have raised an exception"
+        except ValueError:
+            pass
+
     def test_transform_postgres(self):
         input = {
             "host": "airbyte.io",