fix: apply column_tags when updating a view via ALTER

CHANGELOG.md

@@ -5,8 +5,10 @@
 - Add catalogs.yml v2 support (requires `use_catalogs_v2: true` in dbt-core) ([1440](https://github.com/databricks/dbt-databricks/pull/1440))
 
 ### Fixes
-- Apply column-level `databricks_tags` for incremental models on the V1 materialization path (`use_materialization_v2: false`, the default). They were silently dropped at create and on subsequent tag changes; the V1 incremental materialization now applies them, matching the `table` materialization and the V2 path. ([#1520](https://github.com/databricks/dbt-databricks/pull/1520) closes [#1307](https://github.com/databricks/dbt-databricks/issues/1307))
+
 - Raise a `DbtRuntimeError` when a Python model job run terminates with a non-success `result_state` (e.g. `FAILED`/`TIMEDOUT`) instead of returning silently ([#1477](https://github.com/databricks/dbt-databricks/pull/1477))
+- Apply column-level `databricks_tags` for incremental models on the V1 materialization path (`use_materialization_v2: false`, the default). They were silently dropped at create and on subsequent tag changes; the V1 incremental materialization now applies them, matching the `table` materialization and the V2 path. ([#1520](https://github.com/databricks/dbt-databricks/pull/1520) closes [#1307](https://github.com/databricks/dbt-databricks/issues/1307))
+- Fix column-level `databricks_tags` on Unity Catalog views updated via `ALTER` (`view_update_via_alter: true`), including runs where column tags were the only change and skipping reapplication when tags are already correct ([#1526](https://github.com/databricks/dbt-databricks/pull/1526) closes [#1525](https://github.com/databricks/dbt-databricks/issues/1525))
 
 ### Under the Hood
 

dbt/adapters/databricks/impl.py

@@ -1432,6 +1432,16 @@ def _describe_relation(
         else:
             results["information_schema.tags"] = None
 
+        column_tag_config = (
+            model_config.config.get(ColumnTagsProcessor.name) if model_config else None
+        )
+        if column_tag_config is None or column_tag_config.requires_server_metadata_for_diff():
+            results["information_schema.column_tags"] = adapter.execute_macro(
+                "fetch_column_tags", kwargs=kwargs
+            )
+        else:
+            results["information_schema.column_tags"] = None
+
         if adapter.is_describe_as_json_supported(relation):
             json_metadata = adapter.fetch_json_metadata(relation)
             results["information_schema.views"] = (

dbt/include/databricks/macros/relations/view/alter.sql

@@ -8,6 +8,7 @@
   {% set tblproperties = changes.get("tblproperties") %}
   {% set query = changes.get("query") %}
   {% set column_comments = changes.get("column_comments") %}
+  {% set column_tags = changes.get("column_tags") %}
   {% if tags %}
     {{ apply_tags(target_relation, tags.set_tags) }}
   {% endif %}
@@ -20,4 +21,7 @@
   {% if column_comments %}
     {{ alter_column_comments(target_relation, column_comments.comments) }}
   {% endif %}
+  {% if column_tags %}
+    {{ apply_column_tags(target_relation, column_tags) }}
+  {% endif %}
 {% endmacro %}

tests/functional/adapter/column_tags/test_column_tags.py

@@ -120,6 +120,43 @@ class TestColumnTagsView(ColumnTagsMixin):
     relation_type = "view"
 
 
+@pytest.mark.skip_profile("databricks_cluster")
+class TestColumnTagsViewUpdateViaAlter(ColumnTagsMixin):
+    relation_type = "view"
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        # With view_update_via_alter, a subsequent run whose only change is its column
+        # tags takes the alter_view path instead of a full replace; the changed tags
+        # must be applied there too.
+        return {
+            "flags": {"use_materialization_v2": True},
+            "models": {"+view_update_via_alter": True},
+        }
+
+    def test_unchanged_rerun_keeps_column_tags(self, project):
+        util.run_dbt(["run"])
+        util.run_dbt(["run"])
+
+        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 = 'base_model'
+            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"),
+            ("account_number", "key_only", ""),
+            ("account_number", "null_value", ""),
+        }
+        actual_tags = {(row[0], row[1], row[2]) for row in tags}
+        assert actual_tags == expected_tags
+
+
 @pytest.mark.skip_profile("databricks_cluster")
 class TestColumnTagsTableV1(ColumnTagsMixin):
     relation_type = "table"

tests/unit/macros/relations/test_view_macros.py

@@ -52,6 +52,7 @@ def mocks(self, context):
         context["apply_tags"] = Mock()
         context["apply_tblproperties"] = Mock()
         context["alter_query"] = Mock()
+        context["apply_column_tags"] = Mock()
 
     def render_alter_view(self, template_bundle, changes):
         return self.run_macro(
@@ -84,3 +85,11 @@ def test_macros__alter_view_with_query(self, context, template_bundle):
         context["apply_tags"].assert_not_called()
         context["apply_tblproperties"].assert_not_called()
         context["alter_query"].assert_called_once()
+
+    def test_macros__alter_view_with_column_tags(self, context, template_bundle):
+        column_tags = Mock()
+        self.render_alter_view(template_bundle, {"column_tags": column_tags})
+        context["apply_tags"].assert_not_called()
+        context["apply_tblproperties"].assert_not_called()
+        context["alter_query"].assert_not_called()
+        context["apply_column_tags"].assert_called_once_with(template_bundle.relation, column_tags)

tests/unit/test_adapter.py

@@ -1539,8 +1539,16 @@ def _create_incremental_config(
         )
 
     @staticmethod
-    def _create_view_config(tags: dict[str, str] | None = None) -> ViewConfig:
-        return ViewConfig(config={TagsProcessor.name: TagsConfig(set_tags=tags or {})})
+    def _create_view_config(
+        tags: dict[str, str] | None = None,
+        column_tags: dict[str, dict[str, str]] | None = None,
+    ) -> ViewConfig:
+        return ViewConfig(
+            config={
+                TagsProcessor.name: TagsConfig(set_tags=tags or {}),
+                ColumnTagsProcessor.name: ColumnTagsConfig(set_column_tags=column_tags or {}),
+            }
+        )
 
     @staticmethod
     def _create_mv_config(tags: dict[str, str] | None = None) -> MaterializedViewConfig:
@@ -1687,6 +1695,40 @@ def test_view_describe_relation_fetches_tag_query_when_tags_present(self):
         assert "fetch_tags" in called_macro_names
         assert "get_view_description" in called_macro_names
 
+    def test_view_describe_relation_skips_column_tag_query_without_tags(self):
+        adapter = self._create_adapter()
+        relation = self._create_view_relation()
+        relation_config = self._create_view_config()
+
+        results = ViewAPI._describe_relation(adapter, relation, relation_config)
+
+        assert results["information_schema.column_tags"] is None
+        called_macro_names = self._called_macro_names(adapter)
+        assert "fetch_column_tags" not in called_macro_names
+
+    def test_view_describe_relation_fetches_column_tag_query_when_present(self):
+        adapter = self._create_adapter()
+        relation = self._create_view_relation()
+        relation_config = self._create_view_config(
+            column_tags={"id": {"classification": "internal"}}
+        )
+
+        results = ViewAPI._describe_relation(adapter, relation, relation_config)
+
+        assert results["information_schema.column_tags"] == "fetch_column_tags_result"
+        called_macro_names = self._called_macro_names(adapter)
+        assert "fetch_column_tags" in called_macro_names
+
+    def test_view_describe_relation_fetches_column_tags_when_relation_config_is_none(self):
+        adapter = self._create_adapter()
+        relation = self._create_view_relation()
+
+        results = ViewAPI._describe_relation(adapter, relation, None)
+
+        assert results["information_schema.column_tags"] == "fetch_column_tags_result"
+        called_macro_names = self._called_macro_names(adapter)
+        assert "fetch_column_tags" in called_macro_names
+
     def test_mv_describe_relation_skips_tag_query_without_tags(self):
         adapter = self._create_adapter()
         relation = self._create_mv_relation()