fix: Add @@ prefix to system query tag keys (#1264)

<!-- 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

Add `@@` prefix to system query tag keys (i.e. reserved keys) for
internal observability

<img width="280" height="177" alt="Screenshot 2025-11-17 at 9 47 33 AM"
src="https://github.com/user-attachments/assets/2926d2b4-8cd3-420a-b68c-8b103a3cdbd7"
/>

### 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.

---------

Co-authored-by: Ben Cassell <98852248+benc-db@users.noreply.github.com>
Co-authored-by: Ben Cassell <ben.cassell@databricks.com>

Author: ericj-db

CHANGELOG.md

@@ -16,6 +16,7 @@
 ### Under the hood
 
 - Add validation for query tag value length and auto-escape special characters
+- Add `@@` prefix to system query tag keys
 
 ## dbt-databricks 1.11.0 (Oct 29, 2025)
 

dbt/adapters/databricks/connections.py

@@ -579,16 +579,20 @@ def get_merged_query_tags(
 
         # Default tags that will be applied to all queries
         default_tags = {
-            "dbt_databricks_version": __version__,
-            "dbt_core_version": version("dbt-core"),
+            QueryTagsUtils.DBT_DATABRICKS_VERSION_QUERY_TAG_KEY: __version__,
+            QueryTagsUtils.DBT_CORE_VERSION_QUERY_TAG_KEY: version("dbt-core"),
         }
 
         # Default tags that will only exists for queries tied to a specific model
         if query_header_context:
             if hasattr(query_header_context, "model_name") and query_header_context.model_name:
-                default_tags["dbt_model_name"] = query_header_context.model_name
+                default_tags[QueryTagsUtils.DBT_MODEL_NAME_QUERY_TAG_KEY] = (
+                    query_header_context.model_name
+                )
             if hasattr(query_header_context, "materialized") and query_header_context.materialized:
-                default_tags["dbt_materialized"] = query_header_context.materialized
+                default_tags[QueryTagsUtils.DBT_MATERIALIZED_QUERY_TAG_KEY] = (
+                    query_header_context.materialized
+                )
 
         # Parse connection tags from JSON string
         connection_tags = (

dbt/adapters/databricks/utils.py

@@ -102,12 +102,17 @@ def is_cluster_http_path(http_path: str, cluster_id: Optional[str]) -> bool:
 class QueryTagsUtils:
     """Utility class for handling query tags merging and validation."""
 
+    DBT_MODEL_NAME_QUERY_TAG_KEY = "@@dbt_model_name"
+    DBT_CORE_VERSION_QUERY_TAG_KEY = "@@dbt_core_version"
+    DBT_DATABRICKS_VERSION_QUERY_TAG_KEY = "@@dbt_databricks_version"
+    DBT_MATERIALIZED_QUERY_TAG_KEY = "@@dbt_materialized"
+
     # Reserved query tag keys that cannot be overridden
     RESERVED_KEYS = {
-        "dbt_model_name",
-        "dbt_core_version",
-        "dbt_databricks_version",
-        "dbt_materialized",
+        DBT_MODEL_NAME_QUERY_TAG_KEY,
+        DBT_CORE_VERSION_QUERY_TAG_KEY,
+        DBT_DATABRICKS_VERSION_QUERY_TAG_KEY,
+        DBT_MATERIALIZED_QUERY_TAG_KEY,
     }
 
     # Maximum number of query tags allowed

tests/functional/adapter/aliases/test_aliases.py

@@ -28,7 +28,7 @@ def macros(self):
 
 
 class TestDatabricksSameAliasDifferentSchemas(BaseSameAliasDifferentSchemas):
-    @pytest.fixture(autouse=True)
+    @pytest.fixture(scope="class", autouse=True)
     def clean_up(self, project):
         yield
         with project.adapter.connection_named("__test"):
@@ -53,7 +53,7 @@ def macros(self):
 
 
 class TestDatabricksSameAliasDifferentDatabases(BaseSameAliasDifferentDatabases):
-    @pytest.fixture(autouse=True)
+    @pytest.fixture(scope="class", autouse=True)
     def clean_up(self, project):
         yield
         with project.adapter.connection_named("__test"):

tests/functional/adapter/dbt_clone/test_dbt_clone.py

@@ -7,7 +7,7 @@
 
 
 class CleanupMixin:
-    @pytest.fixture(autouse=True)
+    @pytest.fixture(scope="class", autouse=True)
     def clean_up(self, project):
         yield
         with project.adapter.connection_named("__test"):

tests/functional/adapter/incremental/test_incremental_unique_id.py

@@ -1,13 +1,34 @@
+import pytest
 from dbt.tests.adapter.incremental.test_incremental_unique_id import (
     BaseIncrementalUniqueKey,
 )
 
 from tests.functional.adapter.fixtures import MaterializationV2Mixin
 
 
-class TestUniqueKeyDatabricks(BaseIncrementalUniqueKey):
+class CleanupMixin:
+    """Override the upstream clean_up fixture to use class scope.
+
+    The upstream fixture uses function scope (default), which causes it to drop
+    the schema after each test method. This creates a race condition in parallel
+    test execution where the schema is dropped while other test methods are still
+    using it. Using class scope ensures cleanup only happens after all test methods
+    in the class complete.
+    """
+
+    @pytest.fixture(scope="class", autouse=True)
+    def clean_up(self, project):
+        yield
+        with project.adapter.connection_named("__test"):
+            relation = project.adapter.Relation.create(
+                database=project.database, schema=project.test_schema
+            )
+            project.adapter.drop_schema(relation)
+
+
+class TestUniqueKeyDatabricks(CleanupMixin, BaseIncrementalUniqueKey):
     pass
 
 
-class TestUniqueKeyDatabricksV2(MaterializationV2Mixin, BaseIncrementalUniqueKey):
+class TestUniqueKeyDatabricksV2(CleanupMixin, MaterializationV2Mixin, BaseIncrementalUniqueKey):
     pass

tests/unit/test_query_tags.py

@@ -68,7 +68,7 @@ def test_parse_query_tags_non_dict(self):
 
     def test_validate_query_tags_reserved_keys(self):
         """Test validation fails for reserved keys."""
-        tags = {"dbt_model_name": "test", "team": "marketing"}
+        tags = {"@@dbt_model_name": "test", "team": "marketing"}
         with pytest.raises(DbtValidationError, match="Cannot use reserved query tag keys"):
             QueryTagsUtils.validate_query_tags(tags)
 
@@ -165,7 +165,7 @@ def test_merge_query_tags_reserved_key_conflict(self):
         assert "dbt_model_name" in result
 
         # But if user tries to use a reserved key, it should fail
-        model_tags_with_reserved = {"dbt_model_name": "override_attempt"}
+        model_tags_with_reserved = {"@@dbt_model_name": "override_attempt"}
         with pytest.raises(DbtValidationError, match="Cannot use reserved query tag keys"):
             QueryTagsUtils.merge_query_tags(connection_tags, model_tags_with_reserved, default_tags)
 
@@ -220,10 +220,10 @@ def test_get_merged_query_tags(self):
         assert result["project"] == "analytics"  # From model
 
         # Check default tags
-        assert "dbt_databricks_version" in result
-        assert "dbt_core_version" in result
-        assert result["dbt_materialized"] == "table"
-        assert result["dbt_model_name"] == "test_model"
+        assert "@@dbt_databricks_version" in result
+        assert "@@dbt_core_version" in result
+        assert result["@@dbt_materialized"] == "table"
+        assert result["@@dbt_model_name"] == "test_model"
 
     def test_get_merged_query_tags_no_model_tags(self):
         """Test getting merged query tags with no model tags."""
@@ -242,10 +242,10 @@ def test_get_merged_query_tags_no_model_tags(self):
         assert result["team"] == "marketing"
 
         # Check default tags
-        assert "dbt_databricks_version" in result
-        assert "dbt_core_version" in result
-        assert result["dbt_materialized"] == "view"
-        assert result["dbt_model_name"] == "test_model"
+        assert "@@dbt_databricks_version" in result
+        assert "@@dbt_core_version" in result
+        assert result["@@dbt_materialized"] == "view"
+        assert result["@@dbt_model_name"] == "test_model"
 
     def test_get_merged_query_tags_no_connection_tags(self):
         """Test getting merged query tags with no connection tags."""
@@ -264,10 +264,10 @@ def test_get_merged_query_tags_no_connection_tags(self):
         assert result["project"] == "analytics"
 
         # Check default tags
-        assert "dbt_databricks_version" in result
-        assert "dbt_core_version" in result
-        assert result["dbt_materialized"] == "incremental"
-        assert result["dbt_model_name"] == "test_model"
+        assert "@@dbt_databricks_version" in result
+        assert "@@dbt_core_version" in result
+        assert result["@@dbt_materialized"] == "incremental"
+        assert result["@@dbt_model_name"] == "test_model"
 
     def test_get_merged_query_tags_no_tags_at_all(self):
         """Test getting merged query tags with no tags from any source."""
@@ -283,10 +283,10 @@ def test_get_merged_query_tags_no_tags_at_all(self):
         result = QueryConfigUtils.get_merged_query_tags(query_header_context, creds)
 
         # Check default tags only
-        assert "dbt_databricks_version" in result
-        assert "dbt_core_version" in result
-        assert result["dbt_materialized"] == "table"
-        assert result["dbt_model_name"] == "test_model"
+        assert "@@dbt_databricks_version" in result
+        assert "@@dbt_core_version" in result
+        assert result["@@dbt_materialized"] == "table"
+        assert result["@@dbt_model_name"] == "test_model"
 
         # Should only have default tags
         assert len(result) == 4