Add BigQuery resource tags support via resource_tags config option (#1177)

tnk-ysk · colin-rogers-dbt · VersusFacit · web-flow · commit 3325afe0b2cd · 2025-08-06T11:37:34.000-07:00
Co-authored-by: Colin Rogers &lt;111200756+colin-rogers-dbt@users.noreply.github.com&gt;
Co-authored-by: Mila Page &lt;67295367+VersusFacit@users.noreply.github.com&gt;
diff --git a/dbt-bigquery/.changes/unreleased/Features-20250627-181243.yaml b/dbt-bigquery/.changes/unreleased/Features-20250627-181243.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: Add support for BigQuery resource tags via `resource_tags` configuration option
+time: 2025-06-27T18:12:43.109922+09:00
+custom:
+    Author: tnk-ysk
+    Issue: 565 577
diff --git a/dbt-bigquery/src/dbt/adapters/bigquery/impl.py b/dbt-bigquery/src/dbt/adapters/bigquery/impl.py
@@ -783,6 +783,9 @@ def get_common_options(
         if labels:
             opts["labels"] = list(labels.items())  # type: ignore[assignment]
 
+        if resource_tags := config.get("resource_tags"):
+            opts["tags"] = list(resource_tags.items())  # type: ignore[assignment]
+
         return opts
 
     @available.parse(lambda *a, **k: {})
diff --git a/dbt-bigquery/src/dbt/adapters/bigquery/relation_configs/_options.py b/dbt-bigquery/src/dbt/adapters/bigquery/relation_configs/_options.py
@@ -1,5 +1,6 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta
+from types import MappingProxyType
 from typing import Any, Dict, Optional
 
 from dbt.adapters.relation_configs import RelationConfigChange
@@ -11,7 +12,7 @@
 from dbt.adapters.bigquery.utility import bool_setting, float_setting, sql_escape
 
 
-@dataclass(frozen=True, eq=True, unsafe_hash=True)
+@dataclass(frozen=True, eq=True)
 class BigQueryOptionsConfig(BigQueryBaseRelationConfig):
     """
     This config manages materialized view options. See the following for more information:
@@ -24,7 +25,21 @@ class BigQueryOptionsConfig(BigQueryBaseRelationConfig):
     max_staleness: Optional[str] = None
     kms_key_name: Optional[str] = None
     description: Optional[str] = None
-    labels: Optional[Dict[str, str]] = None
+    labels: MappingProxyType = field(default_factory=lambda: MappingProxyType({}))
+    tags: MappingProxyType = field(default_factory=lambda: MappingProxyType({}))
+
+    def __hash__(self) -> int:
+        """Custom hash method to handle unhashable dict fields."""
+        hashable_fields = []
+        for field_name, field_value in self.__dict__.items():
+            if isinstance(field_value, (dict, MappingProxyType)):
+                # Convert dict/MappingProxyType to sorted tuple of items for hashing
+                hashable_fields.append(
+                    (field_name, tuple(sorted(field_value.items())) if field_value else None)
+                )
+            else:
+                hashable_fields.append((field_name, field_value))
+        return hash(tuple(hashable_fields))
 
     def as_ddl_dict(self) -> Dict[str, Any]:
         """
@@ -61,6 +76,7 @@ def array(x):
             "kms_key_name": string,
             "description": escaped_string,
             "labels": array,
+            "tags": array,
         }
 
         def formatted_option(name: str) -> Optional[Any]:
@@ -88,6 +104,7 @@ def from_dict(cls, config_dict: Dict[str, Any]) -> Self:
             "kms_key_name": None,
             "description": None,
             "labels": None,
+            "tags": None,
         }
 
         def formatted_setting(name: str) -> Any:
@@ -131,6 +148,10 @@ def parse_relation_config(cls, relation_config: RelationConfig) -> Dict[str, Any
         if not relation_config.config.persist_docs:  # type:ignore
             del config_dict["description"]
 
+        # Handle resource_tags if present
+        if resource_tags := relation_config.config.extra.get("resource_tags"):  # type:ignore
+            config_dict.update({"tags": resource_tags})
+
         return config_dict
 
     @classmethod
@@ -147,6 +168,10 @@ def parse_bq_table(cls, table: BigQueryTable) -> Dict[str, Any]:
         if labels := table.labels:
             config_dict.update({"labels": labels})
 
+        # Handle tags if they exist on the table
+        if hasattr(table, "resource_tags") and table.resource_tags:
+            config_dict.update({"tags": table.resource_tags})
+
         if encryption_configuration := table.encryption_configuration:
             config_dict.update({"kms_key_name": encryption_configuration.kms_key_name})
         return config_dict
diff --git a/dbt-bigquery/tests/unit/test_bigquery_adapter.py b/dbt-bigquery/tests/unit/test_bigquery_adapter.py
@@ -932,6 +932,59 @@ def test_get_common_options_empty(self):
         actual = adapter.get_common_options(mock_config, node={}, temporary=False)
         self.assertEqual(expected, actual)
 
+    def test_get_common_options_resource_tags(self):
+        adapter = self.get_adapter("oauth")
+        mock_config = create_autospec(RuntimeConfigObject)
+        config = {
+            "resource_tags": {"test-project/env": "dev", "test-project/team": "data"},
+        }
+        mock_config.get.side_effect = lambda name, default=None: config.get(name, default)
+
+        expected = {"tags": [("test-project/env", "dev"), ("test-project/team", "data")]}
+        actual = adapter.get_common_options(mock_config, node={}, temporary=False)
+        # Compare as sets since tag order depends on dictionary ordering
+        self.assertEqual(set(actual["tags"]), set(expected["tags"]))
+
+    def test_get_common_options_resource_tags_empty(self):
+        adapter = self.get_adapter("oauth")
+        mock_config = create_autospec(RuntimeConfigObject)
+        config = {
+            "resource_tags": {},
+        }
+        mock_config.get.side_effect = lambda name, default=None: config.get(name, default)
+
+        expected = {}
+        actual = adapter.get_common_options(mock_config, node={}, temporary=False)
+        self.assertEqual(expected, actual)
+
+    def test_get_common_options_resource_tags_and_labels(self):
+        adapter = self.get_adapter("oauth")
+        mock_config = create_autospec(RuntimeConfigObject)
+        config = {
+            "labels": {"label_key": "label_value"},
+            "resource_tags": {"test-project/tag_key": "tag_value"},
+        }
+        mock_config.get.side_effect = lambda name, default=None: config.get(name, default)
+
+        expected_labels = [("label_key", "label_value")]
+        expected_tags = [("test-project/tag_key", "tag_value")]
+        actual = adapter.get_common_options(mock_config, node={}, temporary=False)
+
+        self.assertEqual(set(actual["labels"]), set(expected_labels))
+        self.assertEqual(set(actual["tags"]), set(expected_tags))
+
+    def test_get_common_options_resource_tags_none(self):
+        adapter = self.get_adapter("oauth")
+        mock_config = create_autospec(RuntimeConfigObject)
+        config = {
+            "resource_tags": None,
+        }
+        mock_config.get.side_effect = lambda name, default=None: config.get(name, default)
+
+        expected = {}
+        actual = adapter.get_common_options(mock_config, node={}, temporary=False)
+        self.assertEqual(expected, actual)
+
 
 class TestBigQueryFilterCatalog(unittest.TestCase):
     def test__catalog_filter_table(self):
diff --git a/dbt-bigquery/tests/unit/test_bigquery_options_config.py b/dbt-bigquery/tests/unit/test_bigquery_options_config.py
@@ -0,0 +1,111 @@
+import unittest
+
+from dbt.adapters.bigquery.relation_configs._options import BigQueryOptionsConfig
+
+
+class TestBigQueryOptionsConfig(unittest.TestCase):
+    """Unit tests for BigQuery options config with tags"""
+
+    def test_from_dict_with_tags(self):
+        """Test creating BigQueryOptionsConfig from dict with tags"""
+        config_dict = {
+            "enable_refresh": True,
+            "refresh_interval_minutes": 45,
+            "description": "Test materialized view",
+            "labels": {"env": "dev", "version": "1.0"},
+            "tags": {"test-project/team": "data", "test-project/cost_center": "analytics"},
+        }
+
+        options = BigQueryOptionsConfig.from_dict(config_dict)
+
+        # Verify all fields are correctly set
+        self.assertTrue(options.enable_refresh)
+        self.assertEqual(options.refresh_interval_minutes, 45)
+        self.assertEqual(options.description, "Test materialized view")
+        self.assertEqual(options.labels, {"env": "dev", "version": "1.0"})
+        self.assertEqual(
+            options.tags, {"test-project/team": "data", "test-project/cost_center": "analytics"}
+        )
+
+    def test_as_ddl_dict_with_tags(self):
+        """Test generating DDL dict with tags"""
+        options = BigQueryOptionsConfig(
+            enable_refresh=True,
+            refresh_interval_minutes=30,
+            description="Test MV",
+            labels={"env": "test"},
+            tags={"test-project/team": "data"},
+        )
+
+        ddl_dict = options.as_ddl_dict()
+
+        # Verify the output format
+        self.assertTrue(ddl_dict["enable_refresh"])
+        self.assertEqual(ddl_dict["refresh_interval_minutes"], 30)
+        self.assertEqual(ddl_dict["description"], '"""Test MV"""')
+
+        # Check labels and tags are properly formatted as arrays of tuples
+        expected_labels = [("env", "test")]
+        expected_tags = [("test-project/team", "data")]
+
+        self.assertEqual(ddl_dict["labels"], expected_labels)
+        self.assertEqual(ddl_dict["tags"], expected_tags)
+
+    def test_tags_none_not_included_in_ddl(self):
+        """Test that None tags are not included in DDL dict"""
+        options = BigQueryOptionsConfig(enable_refresh=True, tags=None)
+
+        ddl_dict = options.as_ddl_dict()
+
+        # tags should not be in the output when None
+        self.assertNotIn("tags", ddl_dict)
+        self.assertIn("enable_refresh", ddl_dict)
+
+    def test_parse_relation_config_with_resource_tags(self):
+        """Test parse_relation_config method handles resource_tags correctly"""
+        from unittest.mock import Mock
+
+        # Mock relation config with resource_tags
+        mock_relation_config = Mock()
+        mock_config = Mock()
+        mock_config.extra = {
+            "enable_refresh": True,
+            "refresh_interval_minutes": 30,
+            "description": "Test view",
+            "labels": {"env": "test"},
+            "resource_tags": {"test-project/team": "data", "test-project/project": "analytics"},
+        }
+        mock_config.persist_docs = True
+        mock_relation_config.config = mock_config
+
+        config_dict = BigQueryOptionsConfig.parse_relation_config(mock_relation_config)
+
+        # Verify resource_tags gets mapped to tags
+        self.assertEqual(
+            config_dict["tags"], {"test-project/team": "data", "test-project/project": "analytics"}
+        )
+        self.assertEqual(config_dict["labels"], {"env": "test"})
+        self.assertEqual(config_dict["description"], "Test view")
+        self.assertTrue(config_dict["enable_refresh"])
+
+    def test_parse_relation_config_without_resource_tags(self):
+        """Test parse_relation_config when resource_tags is not present"""
+        from unittest.mock import Mock
+
+        # Mock relation config without resource_tags
+        mock_relation_config = Mock()
+        mock_config = Mock()
+        mock_config.extra = {"enable_refresh": False, "labels": {"env": "prod"}}
+        mock_config.persist_docs = True
+        mock_relation_config.config = mock_config
+
+        config_dict = BigQueryOptionsConfig.parse_relation_config(mock_relation_config)
+
+        # Verify tags is not in the config_dict when resource_tags is not present
+        self.assertNotIn("tags", config_dict)
+        self.assertEqual(config_dict["labels"], {"env": "prod"})
+        self.assertFalse(config_dict["enable_refresh"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/dbt-bigquery/tests/unit/test_bigquery_relation_configs.py b/dbt-bigquery/tests/unit/test_bigquery_relation_configs.py
