fix: apply column_tags on V1 incremental materialization (#1520)

## Description

V1 (`use_materialization_v2: false`, the **default**) incremental models
silently dropped column-level `databricks_tags`. The V1 branch of
`incremental/incremental.sql` never called `apply_column_tags` on its
create, full-refresh/recreate, or merge-changeset paths — while V1
`table` create (fixed in #1307) and the entire V2 path both apply them.
The model builds successfully but
`system.information_schema.column_tags` stays empty, with no error or
warning.

This applies column tags on all three V1 incremental paths, mirroring
`table.sql` and `apply_config_changeset`. Follow-up to #1307, which
fixed the identical gap for the `table` materialization only.

## Tests
- `column_tags/test_column_tags.py::TestColumnTagsIncrementalV1` —
create (run 1) + merge-change (run 2); fails on `main`, passes here.
-
`column_tags/test_column_tags.py::TestColumnTagsIncrementalV1FullRefresh`
— column tags survive a `dbt run --full-refresh`.
- `tests/unit/macros/relations/components/test_column_tags_macros.py` —
`apply_column_tags` / `alter_set_column_tags` render (incl. the view →
`ALTER TABLE` special case) and the empty-diff no-op (idempotency).

## Checklist
- [x] Changelog entry added (`### Fixes`)
- [x] Functional + unit/macro tests added

Author: sd-db

CHANGELOG.md

@@ -5,6 +5,7 @@
 - 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))
 
 ### Under the Hood

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

@@ -108,6 +108,10 @@
       {%- endcall -%}
       {% do persist_constraints(target_relation, model) %}
       {% do apply_tags(target_relation, tags) %}
+      {% 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 %}
       {%- if language == 'python' -%}
         {%- do apply_tblproperties(target_relation, tblproperties) %}
       {%- endif -%}
@@ -126,6 +130,10 @@
         {% do persist_constraints(target_relation, model) %}
       {% endif %}
       {% do apply_tags(target_relation, tags) %}
+      {% 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 persist_docs(target_relation, model, for_relation=language=='python') %}
     {%- else -%}
       {#-- Set Overwrite Mode to DYNAMIC for subsequent incremental operations --#}
@@ -179,6 +187,7 @@
         {% set liquid_clustering = _configuration_changes.changes.get("liquid_clustering") %}
         {% set row_filter = _configuration_changes.changes.get("row_filter") %}
         {% set constraints = _configuration_changes.changes.get("constraints") %}
+        {% set column_tags = _configuration_changes.changes.get("column_tags") %}
         {% if tags is not none %}
           {% do apply_tags(target_relation, tags.set_tags) %}
         {%- endif -%}
@@ -191,6 +200,9 @@
         {% if row_filter is not none %}
           {{ apply_row_filter(target_relation, row_filter) }}
         {% endif %}
+        {% if column_tags %}
+          {{ apply_column_tags(target_relation, column_tags) }}
+        {% endif %}
         {#- Incremental constraint application requires information_schema access (see fetch_*_constraints macros) -#}
         {% set contract_config = config.get('contract') %}
         {% if constraints and contract_config and contract_config.enforced and not target_relation.is_hive_metastore() %}

tests/functional/adapter/column_tags/test_column_tags.py

@@ -2,7 +2,11 @@
 from dbt.tests import util
 
 from tests.functional.adapter.column_tags import fixtures
-from tests.functional.adapter.fixtures import MaterializationV2Mixin, RerunSafeMixin
+from tests.functional.adapter.fixtures import (
+    MaterializationV1Mixin,
+    MaterializationV2Mixin,
+    RerunSafeMixin,
+)
 
 
 class ColumnTagsMixin(RerunSafeMixin, MaterializationV2Mixin):
@@ -124,3 +128,66 @@ class TestColumnTagsTableV1(ColumnTagsMixin):
     def project_config_update(self):
         # Override MaterializationV2Mixin to test the V1 (default) materialization path
         return {}
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestColumnTagsIncrementalV1(ColumnTagsMixin):
+    relation_type = "incremental"
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        # Override MaterializationV2Mixin to test the V1 (default) materialization path.
+        # Exercises both the V1 incremental create path (run 1) and the V1 merge-time
+        # configuration changeset (run 2 adds a tag to a previously-untagged column).
+        return {}
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestColumnTagsIncrementalV1FullRefresh(RerunSafeMixin, MaterializationV1Mixin):
+    """A `dbt run --full-refresh` of a V1 incremental model must re-apply column tags.
+
+    Full-refresh recreates the relation from scratch (the recreate branch of the V1
+    incremental materialization), so any column tags must be re-applied there or they
+    are lost on every full refresh.
+    """
+
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("base_model",)
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "base_model.sql": fixtures.base_model_sql,
+            "schema.yml": fixtures.initial_column_tag_model.replace(
+                "materialized: table", "materialized: incremental"
+            ),
+        }
+
+    def _column_tags(self, project):
+        rows = project.run_sql(
+            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
+            """,
+            fetch="all",
+        )
+        return {(row[0], row[1], row[2]) for row in rows}
+
+    def test_column_tags_survive_full_refresh(self, project):
+        expected = {
+            ("account_number", "pii", "true"),
+            ("account_number", "sensitive", "true"),
+            ("account_number", "key_only", ""),
+            ("account_number", "null_value", ""),
+        }
+        # Initial run creates the relation (create path applies the tags).
+        util.run_dbt(["run"])
+        assert self._column_tags(project) == expected
+        # Full refresh recreates the relation; the recreate path must re-apply the tags.
+        util.run_dbt(["run", "--full-refresh"])
+        assert self._column_tags(project) == expected

tests/unit/macros/relations/components/test_column_tags_macros.py

@@ -0,0 +1,61 @@
+import pytest
+
+from dbt.adapters.databricks.relation import DatabricksRelationType
+from dbt.adapters.databricks.relation_configs.column_tags import ColumnTagsConfig
+from tests.unit.macros.base import MacroTestBase
+
+
+class TestColumnTagsMacros(MacroTestBase):
+    @pytest.fixture
+    def template_name(self) -> str:
+        return "column_tags.sql"
+
+    @pytest.fixture
+    def macro_folders_to_load(self) -> list:
+        return ["macros/relations/components", "macros/relations", "macros"]
+
+    @pytest.fixture
+    def passthrough_statement(self, template_bundle):
+        """Make `{% call statement('main') %}…{% endcall %}` render its body.
+
+        The base mock returns the statement label and discards the body; for the
+        apply_* wrappers we need the inner ALTER SQL so we can assert on it. Also
+        force the UC path (apply_column_tags raises on hive_metastore).
+        """
+        template_bundle.context["statement"] = lambda label, caller=None: (
+            caller() if caller else label
+        )
+        template_bundle.relation.is_hive_metastore = lambda: False
+        return template_bundle
+
+    def test_alter_set_column_tags_table(self, template_bundle):
+        sql = self.render_bundle(
+            template_bundle, "alter_set_column_tags", "email", {"pii": "true", "team": "growth"}
+        )
+        assert "alter table `some_database`.`some_schema`.`some_table`" in sql
+        assert "alter column `email` set tags ('pii' = 'true', 'team' = 'growth')" in sql
+
+    def test_alter_set_column_tags_view_uses_alter_table(self, template_bundle):
+        # ALTER VIEW cannot set column tags, so the macro must emit ALTER TABLE for views.
+        template_bundle.relation.type = DatabricksRelationType.View
+        sql = self.render_bundle(template_bundle, "alter_set_column_tags", "email", {"pii": "true"})
+        assert "alter table" in sql
+        assert "alter view" not in sql
+        assert "alter column `email` set tags ('pii' = 'true')" in sql
+
+    def test_apply_column_tags_applies_each_column(self, passthrough_statement):
+        config = ColumnTagsConfig(
+            set_column_tags={"email": {"pii": "true"}, "ssn": {"pii": "true"}}
+        )
+        sql = self.render_bundle(passthrough_statement, "apply_column_tags", config)
+        assert "alter column `email` set tags ('pii' = 'true')" in sql
+        assert "alter column `ssn` set tags ('pii' = 'true')" in sql
+
+    def test_apply_column_tags_noop_when_empty(self, passthrough_statement):
+        # Idempotency safety net: when get_diff returns an empty/None changeset, the
+        # {% if column_tags %} gate skips apply entirely; and even if apply_column_tags
+        # is reached with nothing to set, it must emit no ALTER. This pins the latter.
+        config = ColumnTagsConfig(set_column_tags={})
+        sql = self.render_bundle(passthrough_statement, "apply_column_tags", config)
+        assert "alter" not in sql
+        assert "set tags" not in sql