fix: Validate and autoformat query tag values (#1253)

<!-- Please review our pull request review process in CONTRIBUTING.md
before your proceed. -->

Resolves #

<!---
Include the number of the issue addressed by this PR above if
applicable.

  Example:
    resolves #1234

Please review our pull request review process in CONTRIBUTING.md before
your proceed.
-->

### Description

- Auto-escape special characters in query tag values with a warning
message
- Throw error when query tag value (after applying escapes) exceeds 128
length limit

### Checklist

- [ ] I have run this code in development and it appears to resolve the
stated issue
- [ ] This PR includes tests, or tests are not required/relevant for
this PR
- [ ] I have updated the `CHANGELOG.md` and added information about my
change to the "dbt-databricks next" section.

Author: ericj-db

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## dbt-databricks 1.11.1 (TBD)
 
+### Under the hood
+
+- Add validation for query tag value length and auto-escape special characters
+
 ## dbt-databricks 1.11.0 (Oct 29, 2025)
 
 ### Features

dbt/adapters/databricks/utils.py

@@ -151,6 +151,24 @@ def validate_query_tags(tags: dict[str, str], source: str = "") -> None:
                 f"Reserved keys are: {', '.join(sorted(QueryTagsUtils.RESERVED_KEYS))}"
             )
 
+        # Escape commas, colons, and backslashes in tag values
+        for key in tags.keys():
+            value = tags[key]
+            if re.search(r"[\\,:]", value):
+                logger.warning(
+                    f"{source_prefix}Query tag value for key '{key}' contains unescaped "
+                    f"character(s): {value}. Escaping..."
+                )
+                tags[key] = value.replace("\\", "\\\\").replace(",", "\\,").replace(":", "\\:")
+
+        # Validate that no tag value exceeds 128 characters
+        long_values = {k: v for k, v in tags.items() if len(v) > 128}
+        if long_values:
+            raise DbtValidationError(
+                f"{source_prefix}Query tag values must be at most 128 characters. "
+                f"Following keys have values exceeding the limit: {', '.join(long_values.keys())}."
+            )
+
         # Check tag limit
         if len(tags) > QueryTagsUtils.MAX_TAGS:
             raise DbtValidationError(

tests/unit/test_query_tags.py

@@ -84,6 +84,60 @@ def test_validate_query_tags_valid(self):
         # Should not raise any exception
         QueryTagsUtils.validate_query_tags(tags)
 
+    def test_validate_query_tags_escapes_comma(self):
+        """Test that commas in tag values are escaped."""
+        tags = {"team": "marketing,sales"}
+        QueryTagsUtils.validate_query_tags(tags)
+        assert tags["team"] == "marketing\\,sales"
+
+    def test_validate_query_tags_escapes_colon(self):
+        """Test that colons in tag values are escaped."""
+        tags = {"description": "project:alpha"}
+        QueryTagsUtils.validate_query_tags(tags)
+        assert tags["description"] == "project\\:alpha"
+
+    def test_validate_query_tags_escapes_backslash(self):
+        """Test that backslashes in tag values are escaped."""
+        tags = {"path": "folder\\subfolder"}
+        QueryTagsUtils.validate_query_tags(tags)
+        assert tags["path"] == "folder\\\\subfolder"
+
+    def test_validate_query_tags_escapes_multiple_special_chars(self):
+        """Test that multiple special characters are all escaped."""
+        tags = {"complex": "value:with,comma\\and\\backslash"}
+        QueryTagsUtils.validate_query_tags(tags)
+        assert tags["complex"] == "value\\:with\\,comma\\\\and\\\\backslash"
+
+    def test_validate_query_tags_multiple_values_too_long(self):
+        tags = {
+            "long_tag_1": "a" * 129,
+            "short_tag": "ok",
+            "long_tag_2": "b" * 130,
+        }
+        expected_msg = (
+            "Query tag values must be at most 128 characters. "
+            "Following keys have values exceeding the limit: "
+            "long_tag_1, long_tag_2."
+        )
+        with pytest.raises(DbtValidationError, match=expected_msg):
+            QueryTagsUtils.validate_query_tags(tags)
+
+    def test_validate_query_tags_value_after_escaping_too_long(self):
+        """Test validation fails when tag value exceeds 128 chars after escaping."""
+        # Create a value that's 127 characters but will exceed 128 after escaping
+        # We need at least 64 commas (each comma becomes \\, adding 1 char)
+        value_with_commas = ",".join(["a"] * 64)  # Creates "a,a,a..." with 63 commas
+        # After escaping, each comma becomes \\, so length increases by 63
+        tags = {"tag_with_commas": value_with_commas}
+
+        # First it escapes, then it validates length
+        expected_msg = (
+            "Query tag values must be at most 128 characters. "
+            "Following keys have values exceeding the limit: tag_with_commas."
+        )
+        with pytest.raises(DbtValidationError, match=expected_msg):
+            QueryTagsUtils.validate_query_tags(tags)
+
     def test_merge_query_tags_precedence(self):
         """Test that model tags override connection tags."""
         connection_tags = {"team": "marketing", "cost_center": "3000"}