feat: add column level tags for snapshots

CHANGELOG.md

@@ -55,6 +55,7 @@
 - Warn when `contract.enforced: true` is set on a `materialized_view` model ([#1279](https://github.com/databricks/dbt-databricks/issues/1279))
 - Fix `materialized_view` models with `databricks_tags` silently going stale on `dbt run`. `MaterializedViewAPI._describe_relation` was not fetching `information_schema.tags`, so existing tags always parsed as empty, producing a spurious tag diff that routed the materialization to `ALTER ... SET TAGS` instead of `REFRESH MATERIALIZED VIEW` ([#1419](https://github.com/databricks/dbt-databricks/issues/1419))
 - Fix `dbt docs generate` failing with `RuntimeError: Tables contain columns with the same names ... but different types` during catalog merge across schemas ([#1392](https://github.com/databricks/dbt-databricks/issues/1392))
+- Fix snapshots not applying databricks_tags on columns ([#1441](https://github.com/databricks/dbt-databricks/issues/1441))
 
 ## dbt-databricks 1.11.7 (Apr 17, 2026)
 

dbt/include/databricks/macros/materializations/snapshot.sql

@@ -106,6 +106,11 @@
     {%- do apply_tags(target_relation, tags) -%}
   {%- endif -%}
 
+  {% set column_tags = adapter.get_column_tags_from_model(config.model) %}
+  {% if column_tags and column_tags.set_column_tags %}
+    {{ apply_column_tags(target_relation, column_tags) }}
+  {% endif %}
+
   {% do apply_grants(target_relation, grant_config, should_revoke=should_revoke) %}
 
   {% do persist_docs(target_relation, model) %}

tests/functional/adapter/column_tags/fixtures.py

@@ -44,3 +44,44 @@
 base_model_streaming_table = """
 SELECT * FROM stream {{ ref('base_model_seed') }}
 """
+
+snapshot_column_tag_sql = """
+{% snapshot snapshot %}
+    {{
+        config(
+            target_database=database,
+            target_schema=schema,
+            unique_key='id',
+            strategy='check',
+            check_cols=['account_number'],
+        )
+    }}
+    select 1 as id, 'abc123' as account_number
+{% endsnapshot %}
+"""
+
+initial_snapshot_column_tag_schema = """
+version: 2
+snapshots:
+  - name: snapshot
+    columns:
+      - name: id
+      - name: account_number
+        databricks_tags:
+          pii: "true"
+          sensitive: "true"
+"""
+
+updated_snapshot_column_tag_schema = """
+version: 2
+snapshots:
+  - name: snapshot
+    columns:
+      - name: id
+        databricks_tags:
+          pii: "false"
+      - name: account_number
+        databricks_tags:
+          pii: "true"
+          sensitive: "true"
+"""

tests/functional/adapter/column_tags/test_snapshot_column_tags.py

@@ -0,0 +1,47 @@
+import pytest
+from dbt.tests import util
+
+from tests.functional.adapter.column_tags import fixtures
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestSnapshotColumnTags:
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {"snapshot.sql": fixtures.snapshot_column_tag_sql}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": fixtures.initial_snapshot_column_tag_schema}
+
+    def test_snapshot_column_tags(self, project):
+        util.run_dbt(["snapshot"])
+
+        column_tags_query = f"""
+            SELECT column_name, tag_name, tag_value
+            FROM `system`.`information_schema`.`column_tags`
+            WHERE catalog_name = '{project.database}'
+              AND schema_name = '{project.test_schema}'
+              AND table_name = 'snapshot'
+            ORDER BY column_name, tag_name
+            """
+
+        tags = project.run_sql(column_tags_query, fetch="all")
+        expected_tags = {
+            ("account_number", "pii", "true"),
+            ("account_number", "sensitive", "true"),
+        }
+        actual_tags = {(row[0], row[1], row[2]) for row in tags}
+        assert actual_tags == expected_tags
+
+        util.write_file(fixtures.updated_snapshot_column_tag_schema, "models", "schema.yml")
+        util.run_dbt(["snapshot"])
+
+        tags = project.run_sql(column_tags_query, fetch="all")
+        expected_tags = {
+            ("id", "pii", "false"),
+            ("account_number", "pii", "true"),
+            ("account_number", "sensitive", "true"),
+        }
+        actual_tags = {(row[0], row[1], row[2]) for row in tags}
+        assert actual_tags == expected_tags