Parse query comment and use as bigquery job labels.

core/dbt/contracts/connection.py

@@ -215,6 +215,7 @@ def to_target_dict(self):
 class QueryComment(dbtClassMixin):
     comment: str = DEFAULT_QUERY_COMMENT
     append: bool = False
+    job_label: bool = False
 
 
 class AdapterRequiredConfig(HasCredentials, Protocol):

plugins/bigquery/dbt/adapters/bigquery/connections.py

@@ -1,3 +1,5 @@
+import json
+import re
 from contextlib import contextmanager
 from dataclasses import dataclass
 from functools import lru_cache
@@ -305,12 +307,24 @@ def raw_execute(self, sql, fetch=False, *, use_legacy_sql=False):
 
         logger.debug('On {}: {}', conn.name, sql)
 
-        job_params = {'use_legacy_sql': use_legacy_sql}
+        labels = {}
 
         if active_user:
-            job_params['labels'] = {
-                'dbt_invocation_id': active_user.invocation_id
-            }
+            labels['dbt_invocation_id'] = active_user.invocation_id
+
+        if self.profile.query_comment.job_label:
+            try:
+                comment_labels = json.loads(
+                    self.query_header.comment.query_comment
+                )
+                labels.update({
+                    _sanitize_label(key): _sanitize_label(str(value))
+                    for key, value in comment_labels.items()
+                })
+            except (TypeError, ValueError):
+                pass
+
+        job_params = {'use_legacy_sql': use_legacy_sql, 'labels': labels}
 
         priority = conn.credentials.priority
         if priority == Priority.Batch:
@@ -573,3 +587,14 @@ def _is_retryable(error):
             e['reason'] == 'rateLimitExceeded' for e in error.errors):
         return True
     return False
+
+
+_SANITIZE_LABEL_PATTERN = re.compile(r"[^a-z0-9_-]")
+
+
+def _sanitize_label(value: str, max_length: int = 63) -> str:
+    """Return a legal value for a BigQuery label."""
+    value = value.lower()
+    value = _SANITIZE_LABEL_PATTERN.sub("_", value)
+    value = value[: max_length - 1]
+    return value