From 319ca135dd5a0e7f22bb66884cc594efd840e5b7 Mon Sep 17 00:00:00 2001
From: Shubham Dhal <shubham.dhal@databricks.com>
Date: Wed, 1 Jul 2026 17:50:38 +0530
Subject: [PATCH 1/3] fix: apply only new or changed databricks_tags on table
 re-runs

---
 CHANGELOG.md                                  |   1 +
 dbt/adapters/databricks/impl.py               |  32 +++++
 .../macros/materializations/table.sql         |  20 ++-
 .../macros/relations/table/create.sql         |  12 +-
 tests/functional/adapter/tags/fixtures.py     |  59 +++++++++
 .../tags/test_table_tag_fetch_skips.py        | 119 ++++++++++++++++++
 6 files changed, 235 insertions(+), 8 deletions(-)
 create mode 100644 tests/functional/adapter/tags/test_table_tag_fetch_skips.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index bb8f00a62..a8abdc1fd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,7 @@
 ### Fixes
 - Honor the `expression` field on `primary_key` constraints on the V1 materialization path. A primary key declared with `expression: RELY` (or any trailing clause) previously had its expression silently dropped. ([#1551](https://github.com/databricks/dbt-databricks/pull/1551))
 - Apply column-level `databricks_tags` for incremental models on the V1 materialization path ([#1520](https://github.com/databricks/dbt-databricks/pull/1520) closes [#1307](https://github.com/databricks/dbt-databricks/issues/1307))
+- Stop the `table` materialization from re-applying every `databricks_tags` and column-level `databricks_tags` on every run. Because `CREATE OR REPLACE TABLE` preserves existing tags, a re-run now diffs the configured tags against the current server tags and issues `SET TAGS` only for those that are new or changed (matching the incremental path); tag-free tables skip the metadata fetch entirely ([#1308](https://github.com/databricks/dbt-databricks/issues/1308))
 - 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))
 - Fix PK/FK constraints declaring an `expression` (e.g. `RELY`) being dropped and re-added on every incremental run. **Regression:** changing the `expression` on an existing PK/FK (`RELY`↔`NORELY`, or an expression-form FK's target) is no longer applied on incremental runs — use `--full-refresh`. ([#1552](https://github.com/databricks/dbt-databricks/pull/1552) closes [#1513](https://github.com/databricks/dbt-databricks/issues/1513))
 - Honor `incremental_apply_config_changes` in the V1 incremental merge path, allowing users to skip metadata diff queries (tags, column_tags, constraints, column_masks, tblproperties, describe_extended) when set to `false`. Matches the existing V2 behavior. ([1467](https://github.com/databricks/dbt-databricks/pull/1467) partially solves [#1402](https://github.com/databricks/dbt-databricks/issues/1402))
diff --git a/dbt/adapters/databricks/impl.py b/dbt/adapters/databricks/impl.py
index c0a966e61..9ea34e31a 100644
--- a/dbt/adapters/databricks/impl.py
+++ b/dbt/adapters/databricks/impl.py
@@ -1133,6 +1133,38 @@ def build_catalog_relation(self, model: RelationConfig) -> Optional[CatalogRelat
     def get_column_tags_from_model(self, model: RelationConfig) -> Optional[ColumnTagsConfig]:
         return ColumnTagsProcessor.from_relation_config(model)
 
+    @available
+    def get_table_tags_changes(
+        self, relation: DatabricksRelation, model: RelationConfig
+    ) -> dict[str, str]:
+        """Table tags to set: only those new or changed vs the server."""
+        desired = TagsProcessor.from_relation_config(model)
+        if not desired.set_tags:
+            return {}
+        # Defer the hive_metastore error to apply_tags.
+        if relation.is_hive_metastore():
+            return desired.set_tags
+        existing_rows = self.execute_macro("fetch_tags", kwargs={"relation": relation})
+        existing = TagsProcessor.from_relation_results({"information_schema.tags": existing_rows})
+        diff = desired.get_diff(existing)
+        return diff.set_tags if diff else {}
+
+    @available
+    def get_column_tags_changes(
+        self, relation: DatabricksRelation, model: RelationConfig
+    ) -> Optional[ColumnTagsConfig]:
+        """Column tags to set: only those new or changed vs the server."""
+        desired = ColumnTagsProcessor.from_relation_config(model)
+        if not desired.set_column_tags:
+            return None
+        if relation.is_hive_metastore():
+            return desired
+        existing_rows = self.execute_macro("fetch_column_tags", kwargs={"relation": relation})
+        existing = ColumnTagsProcessor.from_relation_results(
+            {"information_schema.column_tags": existing_rows}
+        )
+        return desired.get_diff(existing)
+
     @available
     def resolve_file_format(self, config: BaseConfig) -> str:
         if config.get("table_format") == constants.ICEBERG_TABLE_FORMAT:
diff --git a/dbt/include/databricks/macros/materializations/table.sql b/dbt/include/databricks/macros/materializations/table.sql
index 157c86993..6191f26e6 100644
--- a/dbt/include/databricks/macros/materializations/table.sql
+++ b/dbt/include/databricks/macros/materializations/table.sql
@@ -9,6 +9,9 @@
   {% set existing_relation = adapter.get_relation(database=database, schema=schema, identifier=identifier, needs_information=True) %}
   {% set target_relation = this.incorporate(type='table') %}
   {% set compiled_code = adapter.clean_sql(compiled_code) %}
+  {# True when the relation is replaced in place (negation of the drop conditions below);
+     a fresh create or drop+recreate inherits no tags. #}
+  {%- set replaced_in_place = existing_relation and existing_relation.type == 'table' and existing_relation.can_be_replaced and adapter.resolve_file_format(config) in ('delta', 'iceberg') -%}
 
   {% if adapter.get_behavior_flag_no_warn('use_materialization_v2') %}
     {% set intermediate_relation = make_intermediate_relation(target_relation) %}
@@ -25,10 +28,10 @@
       {% if safe_create and existing_relation.can_be_renamed %}
         {{ safe_relation_replace(existing_relation, staging_relation, intermediate_relation, compiled_code) }}
       {% else %}
-        {% if existing_relation and (existing_relation.type != 'table' or not (existing_relation.can_be_replaced and adapter.resolve_file_format(config) in ('delta', 'iceberg'))) -%}
+        {% if existing_relation and not replaced_in_place -%}
           {{ adapter.drop_relation(existing_relation) }}
         {%- endif %}
-        {{ create_table_at(target_relation, intermediate_relation, compiled_code) }}
+        {{ create_table_at(target_relation, intermediate_relation, compiled_code, replaced_in_place=replaced_in_place) }}
       {% endif %}
     {% endif %}
 
@@ -46,7 +49,7 @@
     -- setup: if the target relation already exists, drop it
     -- in case if the existing and future table is delta or iceberg, we want to do a
     -- create or replace table instead of dropping, so we don't have the table unavailable
-    {% if existing_relation and (existing_relation.type != 'table' or not (existing_relation.can_be_replaced and adapter.resolve_file_format(config) in ('delta', 'iceberg'))) -%}
+    {% if existing_relation and not replaced_in_place -%}
       {{ adapter.drop_relation(existing_relation) }}
     {%- endif %}
 
@@ -61,9 +64,16 @@
     {% if language=="python" %}
       {% do apply_tblproperties(target_relation, tblproperties) %}
     {% endif %}
-    {%- do apply_tags(target_relation, tags) -%}
+    {%- if replaced_in_place -%}
+      {# Replace preserves tags, so apply only new/changed ones. #}
+      {%- set tags_to_set = adapter.get_table_tags_changes(target_relation, config.model) -%}
+      {%- set column_tags = adapter.get_column_tags_changes(target_relation, config.model) -%}
+    {%- else -%}
+      {%- set tags_to_set = tags -%}
+      {%- set column_tags = adapter.get_column_tags_from_model(config.model) -%}
+    {%- endif -%}
+    {%- do apply_tags(target_relation, tags_to_set) -%}
 
-    {% 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 %}
diff --git a/dbt/include/databricks/macros/relations/table/create.sql b/dbt/include/databricks/macros/relations/table/create.sql
index 281dcfbba..015c42247 100644
--- a/dbt/include/databricks/macros/relations/table/create.sql
+++ b/dbt/include/databricks/macros/relations/table/create.sql
@@ -1,5 +1,4 @@
-{% macro create_table_at(relation, intermediate_relation, compiled_code) %}
-  {% set tags = config.get('databricks_tags') %}
+{% macro create_table_at(relation, intermediate_relation, compiled_code, replaced_in_place=false) %}
   {% set model_columns = model.get('columns', []) %}
   {% set existing_columns = adapter.get_columns_in_relation(intermediate_relation) %}
   {% set contract_config = config.get('contract') %}
@@ -17,8 +16,15 @@
   {% endcall %}
 
   {{ apply_alter_constraints(target_relation) }}
+  {%- if replaced_in_place -%}
+    {# Replace preserves tags, so apply only new/changed ones; otherwise apply all. #}
+    {% set tags = adapter.get_table_tags_changes(target_relation, config.model) %}
+    {% set column_tags = adapter.get_column_tags_changes(target_relation, config.model) %}
+  {%- else -%}
+    {% set tags = config.get('databricks_tags') %}
+    {% set column_tags = adapter.get_column_tags_from_model(config.model) %}
+  {%- endif -%}
   {{ 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 %}
diff --git a/tests/functional/adapter/tags/fixtures.py b/tests/functional/adapter/tags/fixtures.py
index dd7294b53..6ce4cb71a 100644
--- a/tests/functional/adapter/tags/fixtures.py
+++ b/tests/functional/adapter/tags/fixtures.py
@@ -81,3 +81,62 @@ def model(dbt, spark):
     select cast(1 as bigint) as id, 'hello' as msg, 'blue' as color
 {% endsnapshot %}
 """
+
+# A plain table model; tag config comes from the schema fixtures below.
+metadata_fetch_table_sql = """
+{{ config(
+    materialized = 'table',
+) }}
+
+select cast(1 as bigint) as id
+"""
+
+metadata_fetch_no_tags_schema = """
+version: 2
+
+models:
+  - name: metadata_fetch_table
+    columns:
+      - name: id
+"""
+
+metadata_fetch_table_tags_schema = """
+version: 2
+
+models:
+  - name: metadata_fetch_table
+    config:
+      databricks_tags:
+        classification: internal
+    columns:
+      - name: id
+"""
+
+metadata_fetch_column_tags_schema = """
+version: 2
+
+models:
+  - name: metadata_fetch_table
+    columns:
+      - name: id
+        databricks_tags:
+          classification: internal
+"""
+
+# A view, later reconfigured to a tagged table to force a drop+recreate.
+metadata_fetch_view_first_sql = """
+{{ config(
+    materialized = 'view',
+) }}
+
+select cast(1 as bigint) as id
+"""
+
+metadata_fetch_table_with_tags_sql = """
+{{ config(
+    materialized = 'table',
+    databricks_tags = {'classification': 'internal'},
+) }}
+
+select cast(1 as bigint) as id
+"""
diff --git a/tests/functional/adapter/tags/test_table_tag_fetch_skips.py b/tests/functional/adapter/tags/test_table_tag_fetch_skips.py
new file mode 100644
index 000000000..17152c272
--- /dev/null
+++ b/tests/functional/adapter/tags/test_table_tag_fetch_skips.py
@@ -0,0 +1,119 @@
+import pytest
+from dbt.artifacts.schemas.results import RunStatus
+from dbt.tests import util
+
+from tests.functional.adapter.fixtures import (
+    MaterializationV2Mixin,
+    fail_if_tag_and_column_tag_fetch_called_macros,
+)
+from tests.functional.adapter.tags import fixtures
+
+
+class BaseTableMetadataFetch:
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {"fail_if_tag_fetch_called.sql": fail_if_tag_and_column_tag_fetch_called_macros}
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestTableMetadataFetchSkips(BaseTableMetadataFetch):
+    """A tag-free table must never fetch tag metadata."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "metadata_fetch_table.sql": fixtures.metadata_fetch_table_sql,
+            "schema.yml": fixtures.metadata_fetch_no_tags_schema,
+        }
+
+    def test_table_runs_never_fetch_tags(self, project):
+        # No tags configured, so neither the create nor the re-run fetches.
+        util.run_dbt(["run"])
+        util.run_dbt(["run"])
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestTableMetadataFetchRequiresTableTags(BaseTableMetadataFetch):
+    """A re-run with table tags fetches to diff against the server."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "metadata_fetch_table.sql": fixtures.metadata_fetch_table_sql,
+            "schema.yml": fixtures.metadata_fetch_table_tags_schema,
+        }
+
+    def test_rerun_fetches_table_tags(self, project):
+        # Fresh create applies all tags; the re-run fetches to diff.
+        util.run_dbt(["run"])
+
+        run_execution_results = util.run_dbt(["run"], expect_pass=False)
+        assert len(run_execution_results.results) == 1
+        result = run_execution_results.results[0]
+        assert result.status == RunStatus.Error
+        assert "tags should not be called" in result.message
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestTableMetadataFetchRequiresColumnTags(BaseTableMetadataFetch):
+    """A re-run with column tags fetches to diff per column."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "metadata_fetch_table.sql": fixtures.metadata_fetch_table_sql,
+            "schema.yml": fixtures.metadata_fetch_column_tags_schema,
+        }
+
+    def test_rerun_fetches_column_tags(self, project):
+        util.run_dbt(["run"])
+
+        run_execution_results = util.run_dbt(["run"], expect_pass=False)
+        assert len(run_execution_results.results) == 1
+        result = run_execution_results.results[0]
+        assert result.status == RunStatus.Error
+        assert "tags should not be called" in result.message
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestTableDropRecreateAppliesAllTags(BaseTableMetadataFetch):
+    """A dropped+recreated relation inherits no tags, so all are applied (no diff, no fetch)."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"metadata_fetch_table.sql": fixtures.metadata_fetch_view_first_sql}
+
+    def test_drop_recreate_applies_all_tags_without_fetch(self, project):
+        # View (no fetch) reconfigured to a tagged table: the recreate applies all tags, no fetch.
+        util.run_dbt(["run"])
+        util.write_file(
+            fixtures.metadata_fetch_table_with_tags_sql, "models", "metadata_fetch_table.sql"
+        )
+        util.run_dbt(["run"])
+        results = project.run_sql(
+            "select tag_name, tag_value from `system`.`information_schema`.`table_tags`"
+            " where schema_name = '{schema}' and table_name='metadata_fetch_table'",
+            fetch="all",
+        )
+        assert set((row[0], row[1]) for row in results) == {("classification", "internal")}
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestTableMetadataFetchRequiresColumnTagsV2(MaterializationV2Mixin, BaseTableMetadataFetch):
+    """Same on the v2 create_table_at path."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "metadata_fetch_table.sql": fixtures.metadata_fetch_table_sql,
+            "schema.yml": fixtures.metadata_fetch_column_tags_schema,
+        }
+
+    def test_rerun_fetches_column_tags(self, project):
+        util.run_dbt(["run"])
+
+        run_execution_results = util.run_dbt(["run"], expect_pass=False)
+        assert len(run_execution_results.results) == 1
+        result = run_execution_results.results[0]
+        assert result.status == RunStatus.Error
+        assert "tags should not be called" in result.message

From c28183328384cd4cd06a81fbfba07ec0b1092d1d Mon Sep 17 00:00:00 2001
From: Shubham Dhal <shubham.dhal@databricks.com>
Date: Sun, 9 Aug 2026 06:45:27 +0000
Subject: [PATCH 2/3] test: cover changed tag values and the v2 table-tags path

Address review gaps on the table tag-diff change:

- Assert a *changed* tag value still reaches the server on a re-run, for
  both table- and column-level tags. This is the diff path's one real
  failure mode and was previously uncovered; the existing tests only
  assert whether a metadata fetch occurred.
- Add a v2 `create_table_at` case for table-level tags. v2 coverage was
  column-tags only, and `use_materialization_v2` defaults to false, so the
  unmarked classes all exercise v1.
- Make `TestTableDropRecreateAppliesAllTags` rerun-safe. It rewrites a
  model via `write_file`, so without the mixin a `--reruns` retry inherits
  the converted table and stops testing view->table conversion.

Also link the PR in the changelog entry and describe the diff's gating
condition rather than naming file formats: `resolve_file_format` maps
`table_format='iceberg'` to delta or parquet, so it does not return the
literal 'iceberg' the gate tests for.
---
 CHANGELOG.md                                  |  2 +-
 tests/functional/adapter/tags/fixtures.py     | 21 +++++
 .../tags/test_table_tag_fetch_skips.py        | 90 ++++++++++++++++++-
 3 files changed, 110 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a8abdc1fd..19f677de9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,7 +8,7 @@
 ### Fixes
 - Honor the `expression` field on `primary_key` constraints on the V1 materialization path. A primary key declared with `expression: RELY` (or any trailing clause) previously had its expression silently dropped. ([#1551](https://github.com/databricks/dbt-databricks/pull/1551))
 - Apply column-level `databricks_tags` for incremental models on the V1 materialization path ([#1520](https://github.com/databricks/dbt-databricks/pull/1520) closes [#1307](https://github.com/databricks/dbt-databricks/issues/1307))
-- Stop the `table` materialization from re-applying every `databricks_tags` and column-level `databricks_tags` on every run. Because `CREATE OR REPLACE TABLE` preserves existing tags, a re-run now diffs the configured tags against the current server tags and issues `SET TAGS` only for those that are new or changed (matching the incremental path); tag-free tables skip the metadata fetch entirely ([#1308](https://github.com/databricks/dbt-databricks/issues/1308))
+- Stop the `table` materialization from re-applying every `databricks_tags` and column-level `databricks_tags` on every run. When an existing table is replaced in place (`CREATE OR REPLACE`, which preserves its tags), a re-run now diffs the configured tags against the current server tags and issues `SET TAGS` only for those that are new or changed (matching the incremental path); fresh creates and drop+recreates still apply all tags, and tag-free tables skip the metadata fetch entirely ([#1572](https://github.com/databricks/dbt-databricks/pull/1572) resolves [#1308](https://github.com/databricks/dbt-databricks/issues/1308))
 - 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))
 - Fix PK/FK constraints declaring an `expression` (e.g. `RELY`) being dropped and re-added on every incremental run. **Regression:** changing the `expression` on an existing PK/FK (`RELY`↔`NORELY`, or an expression-form FK's target) is no longer applied on incremental runs — use `--full-refresh`. ([#1552](https://github.com/databricks/dbt-databricks/pull/1552) closes [#1513](https://github.com/databricks/dbt-databricks/issues/1513))
 - Honor `incremental_apply_config_changes` in the V1 incremental merge path, allowing users to skip metadata diff queries (tags, column_tags, constraints, column_masks, tblproperties, describe_extended) when set to `false`. Matches the existing V2 behavior. ([1467](https://github.com/databricks/dbt-databricks/pull/1467) partially solves [#1402](https://github.com/databricks/dbt-databricks/issues/1402))
diff --git a/tests/functional/adapter/tags/fixtures.py b/tests/functional/adapter/tags/fixtures.py
index 6ce4cb71a..47b528683 100644
--- a/tests/functional/adapter/tags/fixtures.py
+++ b/tests/functional/adapter/tags/fixtures.py
@@ -140,3 +140,24 @@ def model(dbt, spark):
 
 select cast(1 as bigint) as id
 """
+
+# Same model with a different tag value, to check the diff still applies real changes.
+metadata_fetch_table_with_changed_tags_sql = """
+{{ config(
+    materialized = 'table',
+    databricks_tags = {'classification': 'confidential'},
+) }}
+
+select cast(1 as bigint) as id
+"""
+
+metadata_fetch_changed_column_tags_schema = """
+version: 2
+
+models:
+  - name: metadata_fetch_table
+    columns:
+      - name: id
+        databricks_tags:
+          classification: confidential
+"""
diff --git a/tests/functional/adapter/tags/test_table_tag_fetch_skips.py b/tests/functional/adapter/tags/test_table_tag_fetch_skips.py
index 17152c272..9dba62886 100644
--- a/tests/functional/adapter/tags/test_table_tag_fetch_skips.py
+++ b/tests/functional/adapter/tags/test_table_tag_fetch_skips.py
@@ -4,6 +4,7 @@
 
 from tests.functional.adapter.fixtures import (
     MaterializationV2Mixin,
+    RerunSafeMixin,
     fail_if_tag_and_column_tag_fetch_called_macros,
 )
 from tests.functional.adapter.tags import fixtures
@@ -76,13 +77,17 @@ def test_rerun_fetches_column_tags(self, project):
 
 
 @pytest.mark.skip_profile("databricks_cluster")
-class TestTableDropRecreateAppliesAllTags(BaseTableMetadataFetch):
+class TestTableDropRecreateAppliesAllTags(RerunSafeMixin, BaseTableMetadataFetch):
     """A dropped+recreated relation inherits no tags, so all are applied (no diff, no fetch)."""
 
     @pytest.fixture(scope="class")
     def models(self):
         return {"metadata_fetch_table.sql": fixtures.metadata_fetch_view_first_sql}
 
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("metadata_fetch_table",)
+
     def test_drop_recreate_applies_all_tags_without_fetch(self, project):
         # View (no fetch) reconfigured to a tagged table: the recreate applies all tags, no fetch.
         util.run_dbt(["run"])
@@ -98,9 +103,30 @@ def test_drop_recreate_applies_all_tags_without_fetch(self, project):
         assert set((row[0], row[1]) for row in results) == {("classification", "internal")}
 
 
+@pytest.mark.skip_profile("databricks_cluster")
+class TestTableMetadataFetchRequiresTableTagsV2(MaterializationV2Mixin, BaseTableMetadataFetch):
+    """Table tags on the v2 create_table_at path."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "metadata_fetch_table.sql": fixtures.metadata_fetch_table_sql,
+            "schema.yml": fixtures.metadata_fetch_table_tags_schema,
+        }
+
+    def test_rerun_fetches_table_tags(self, project):
+        util.run_dbt(["run"])
+
+        run_execution_results = util.run_dbt(["run"], expect_pass=False)
+        assert len(run_execution_results.results) == 1
+        result = run_execution_results.results[0]
+        assert result.status == RunStatus.Error
+        assert "tags should not be called" in result.message
+
+
 @pytest.mark.skip_profile("databricks_cluster")
 class TestTableMetadataFetchRequiresColumnTagsV2(MaterializationV2Mixin, BaseTableMetadataFetch):
-    """Same on the v2 create_table_at path."""
+    """Column tags on the v2 create_table_at path."""
 
     @pytest.fixture(scope="class")
     def models(self):
@@ -117,3 +143,63 @@ def test_rerun_fetches_column_tags(self, project):
         result = run_execution_results.results[0]
         assert result.status == RunStatus.Error
         assert "tags should not be called" in result.message
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestChangedTableTagValueIsApplied(RerunSafeMixin):
+    """The diff must still apply a tag whose value changed."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"metadata_fetch_table.sql": fixtures.metadata_fetch_table_with_tags_sql}
+
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("metadata_fetch_table",)
+
+    def test_changed_tag_value_reaches_server(self, project):
+        util.run_dbt(["run"])
+        util.write_file(
+            fixtures.metadata_fetch_table_with_changed_tags_sql,
+            "models",
+            "metadata_fetch_table.sql",
+        )
+        util.run_dbt(["run"])
+
+        results = project.run_sql(
+            "select tag_name, tag_value from `system`.`information_schema`.`table_tags`"
+            " where schema_name = '{schema}' and table_name='metadata_fetch_table'",
+            fetch="all",
+        )
+        assert set((row[0], row[1]) for row in results) == {("classification", "confidential")}
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestChangedColumnTagValueIsApplied(RerunSafeMixin):
+    """Same for a column tag whose value changed."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "metadata_fetch_table.sql": fixtures.metadata_fetch_table_sql,
+            "schema.yml": fixtures.metadata_fetch_column_tags_schema,
+        }
+
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("metadata_fetch_table",)
+
+    def test_changed_column_tag_value_reaches_server(self, project):
+        util.run_dbt(["run"])
+        util.write_file(fixtures.metadata_fetch_changed_column_tags_schema, "models", "schema.yml")
+        util.run_dbt(["run"])
+
+        results = project.run_sql(
+            "select column_name, tag_name, tag_value from"
+            " `system`.`information_schema`.`column_tags`"
+            " where schema_name = '{schema}' and table_name='metadata_fetch_table'",
+            fetch="all",
+        )
+        assert set((row[0], row[1], row[2]) for row in results) == {
+            ("id", "classification", "confidential")
+        }

From 2c3ae4cd708e36fffc52b5dec25851917c98495a Mon Sep 17 00:00:00 2001
From: Shubham Dhal <shubham.dhal@databricks.com>
Date: Fri, 11 Sep 2026 17:32:01 +0530
Subject: [PATCH 3/3] Consolidate tag reconciliation for table replacements

---
 CHANGELOG.md                                  |  5 +-
 dbt/adapters/databricks/impl.py               | 57 ++++++++-------
 .../incremental/incremental.sql               | 23 +++---
 .../macros/materializations/table.sql         | 15 +---
 .../macros/relations/table/create.sql         | 15 +---
 .../databricks/macros/relations/tags.sql      | 15 ++++
 .../tags/test_table_tag_fetch_skips.py        | 73 +++++++++++++++----
 .../unit/macros/relations/test_tags_macros.py | 41 +++++++++++
 tests/unit/test_adapter.py                    | 59 +++++++++++++++
 9 files changed, 218 insertions(+), 85 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e90333886..370aba637 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,9 @@
 ## dbt-databricks 1.12.6 (TBD)
 
+### Fixes
+
+- Avoid redundant tag application when tag configurations match after in-place table rebuilds and incremental full-refresh replacements; skip tag metadata reads when no tags are configured ([#1572](https://github.com/databricks/dbt-databricks/pull/1572) resolves [#1308](https://github.com/databricks/dbt-databricks/issues/1308))
+
 ### Under the Hood
 
 - Document serverless environment configuration for Python models (thanks @TangoEnSkai!) ([#1649](https://github.com/databricks/dbt-databricks/pull/1649) resolves [#1055](https://github.com/databricks/dbt-databricks/issues/1055))
@@ -22,7 +26,6 @@
 - Redact all `credential` and `encryption` clauses in logged SQL, regardless of keyword case (thanks @SreeramaYeshwanthGowd!) ([#1610](https://github.com/databricks/dbt-databricks/pull/1610) resolves [#1609](https://github.com/databricks/dbt-databricks/issues/1609))
 - Stop `delete+insert` with a composite `unique_key` from deleting unmatched rows on DBR below 17.1 (thanks @SreeramaYeshwanthGowd!) ([#1612](https://github.com/databricks/dbt-databricks/pull/1612) resolves [#1611](https://github.com/databricks/dbt-databricks/issues/1611))
 - Escape single quotes in relation comments so materialized views and streaming tables with an apostrophe in the description can be created (thanks @SreeramaYeshwanthGowd!) ([#1613](https://github.com/databricks/dbt-databricks/pull/1613) resolves [#1251](https://github.com/databricks/dbt-databricks/issues/1251))
-- Stop the `table` materialization from re-applying every `databricks_tags` and column-level `databricks_tags` on every run. When an existing table is replaced in place (`CREATE OR REPLACE`, which preserves its tags), a re-run now diffs the configured tags against the current server tags and issues `SET TAGS` only for those that are new or changed (matching the incremental path); fresh creates and drop+recreates still apply all tags, and tag-free tables skip the metadata fetch entirely ([#1572](https://github.com/databricks/dbt-databricks/pull/1572) resolves [#1308](https://github.com/databricks/dbt-databricks/issues/1308))
 
 ### Under the Hood
 
diff --git a/dbt/adapters/databricks/impl.py b/dbt/adapters/databricks/impl.py
index 3fa8d7a63..fff92955a 100644
--- a/dbt/adapters/databricks/impl.py
+++ b/dbt/adapters/databricks/impl.py
@@ -1161,36 +1161,37 @@ def get_column_tags_from_model(self, model: RelationConfig) -> Optional[ColumnTa
         return ColumnTagsProcessor.from_relation_config(model)
 
     @available
-    def get_table_tags_changes(
+    def get_table_replacement_tag_changes(
         self, relation: DatabricksRelation, model: RelationConfig
-    ) -> dict[str, str]:
-        """Table tags to set: only those new or changed vs the server."""
-        desired = TagsProcessor.from_relation_config(model)
-        if not desired.set_tags:
-            return {}
-        # Defer the hive_metastore error to apply_tags.
-        if relation.is_hive_metastore():
-            return desired.set_tags
-        existing_rows = self.execute_macro("fetch_tags", kwargs={"relation": relation})
-        existing = TagsProcessor.from_relation_results({"information_schema.tags": existing_rows})
-        diff = desired.get_diff(existing)
-        return diff.set_tags if diff else {}
+    ) -> dict[str, Union[dict[str, str], dict[str, dict[str, str]]]]:
+        """Reconcile tags retained by CREATE OR REPLACE TABLE.
 
-    @available
-    def get_column_tags_changes(
-        self, relation: DatabricksRelation, model: RelationConfig
-    ) -> Optional[ColumnTagsConfig]:
-        """Column tags to set: only those new or changed vs the server."""
-        desired = ColumnTagsProcessor.from_relation_config(model)
-        if not desired.set_column_tags:
-            return None
-        if relation.is_hive_metastore():
-            return desired
-        existing_rows = self.execute_macro("fetch_column_tags", kwargs={"relation": relation})
-        existing = ColumnTagsProcessor.from_relation_results(
-            {"information_schema.column_tags": existing_rows}
-        )
-        return desired.get_diff(existing)
+        Table rebuilds and incremental full-refresh replacements bypass the normal ALTER
+        changeset flow. This Jinja API reuses the Python tag processors and component diffs
+        while fetching only tag metadata, avoiding a full relation-config read after replacement.
+        """
+        tags = TagsProcessor.from_relation_config(model)
+        column_tags = ColumnTagsProcessor.from_relation_config(model)
+        table_tags_to_set = tags.set_tags
+        column_tags_to_set = column_tags.set_column_tags
+        # Preserve the existing UC-only errors in the apply macros.
+        if not relation.is_hive_metastore():
+            if tags.requires_server_metadata_for_diff():
+                rows = self.execute_macro("fetch_tags", kwargs={"relation": relation})
+                existing = TagsProcessor.from_relation_results({"information_schema.tags": rows})
+                tags_diff = tags.get_diff(existing)
+                table_tags_to_set = tags_diff.set_tags if tags_diff else {}
+            if column_tags.requires_server_metadata_for_diff():
+                rows = self.execute_macro("fetch_column_tags", kwargs={"relation": relation})
+                existing_columns = ColumnTagsProcessor.from_relation_results(
+                    {"information_schema.column_tags": rows}
+                )
+                column_tags_diff = column_tags.get_diff(existing_columns)
+                column_tags_to_set = column_tags_diff.set_column_tags if column_tags_diff else {}
+        return {
+            "table_tags": table_tags_to_set,
+            "column_tags": column_tags_to_set,
+        }
 
     @available
     def resolve_file_format(self, config: BaseConfig) -> str:
diff --git a/dbt/include/databricks/macros/materializations/incremental/incremental.sql b/dbt/include/databricks/macros/materializations/incremental/incremental.sql
index e79859822..fee874964 100644
--- a/dbt/include/databricks/macros/materializations/incremental/incremental.sql
+++ b/dbt/include/databricks/macros/materializations/incremental/incremental.sql
@@ -49,7 +49,10 @@
           {% do adapter.drop_relation(existing_relation) %}
         {% endif %}
         {{ log("Replacing target relation") }}
-        {{ create_table_at(target_relation, intermediate_relation, compiled_code) }}
+        {{ create_table_at(
+            target_relation, intermediate_relation, compiled_code,
+            replaced_in_place=is_replaceable and not existing_relation.is_shallow_clone
+        ) }}
       {% endif %}
     {%- else -%}
       {{ log("Existing relation found, proceeding with incremental work")}}
@@ -93,7 +96,6 @@
 
   {% else %}
     {%- set tblproperties = config.get('tblproperties') -%}
-    {%- set tags = config.get('databricks_tags') -%}
     {% set temp_relation = make_temp_relation(target_relation) %}
     {% set incremental_predicates = config.get('predicates') or config.get('incremental_predicates') %}
     {%- set unique_key = config.get('unique_key') -%}
@@ -107,11 +109,7 @@
         {{ create_table_as(False, target_relation, compiled_code, language) }}
       {%- 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 %}
+      {{ reconcile_tags(target_relation) }}
       {%- if language == 'python' -%}
         {%- do apply_tblproperties(target_relation, tblproperties) %}
       {%- endif -%}
@@ -129,11 +127,10 @@
       {% if not existing_relation.is_view %}
         {% 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 %}
+      {{ reconcile_tags(
+          target_relation,
+          replaced_in_place=is_replaceable_format and not existing_relation.is_shallow_clone
+      ) }}
       {% do persist_docs(target_relation, model, for_relation=language=='python') %}
     {%- else -%}
       {#-- Set Overwrite Mode to DYNAMIC for subsequent incremental operations --#}
@@ -261,4 +258,4 @@
     {%- set configuration_changes = model_config.get_changeset(existing_config) -%}
     {{ apply_config_changeset(target_relation, model, configuration_changes, existing_relation) }}
   {% endif %}
-{% endmacro %}
\ No newline at end of file
+{% endmacro %}
diff --git a/dbt/include/databricks/macros/materializations/table.sql b/dbt/include/databricks/macros/materializations/table.sql
index 6137ba161..3d3fc8e3d 100644
--- a/dbt/include/databricks/macros/materializations/table.sql
+++ b/dbt/include/databricks/macros/materializations/table.sql
@@ -4,7 +4,6 @@
   {%- set identifier = model['alias'] -%}
   {%- set grant_config = config.get('grants') -%}
   {%- set tblproperties = config.get('tblproperties') -%}
-  {%- set tags = config.get('databricks_tags') -%}
   {%- set safe_create = config.get('use_safer_relation_operations', False) %}
   {% set existing_relation = adapter.get_relation(database=database, schema=schema, identifier=identifier, needs_information=True) %}
   {% set target_relation = this.incorporate(type='table') %}
@@ -65,19 +64,7 @@
     {% if language=="python" %}
       {% do apply_tblproperties(target_relation, tblproperties) %}
     {% endif %}
-    {%- if replaced_in_place -%}
-      {# Replace preserves tags, so apply only new/changed ones. #}
-      {%- set tags_to_set = adapter.get_table_tags_changes(target_relation, config.model) -%}
-      {%- set column_tags = adapter.get_column_tags_changes(target_relation, config.model) -%}
-    {%- else -%}
-      {%- set tags_to_set = tags -%}
-      {%- set column_tags = adapter.get_column_tags_from_model(config.model) -%}
-    {%- endif -%}
-    {%- do apply_tags(target_relation, tags_to_set) -%}
-
-    {% if column_tags and column_tags.set_column_tags %}
-      {{ apply_column_tags(target_relation, column_tags) }}
-    {% endif %}
+    {{ reconcile_tags(target_relation, replaced_in_place) }}
 
     {% do persist_docs(target_relation, model, for_relation=language=='python') %}
 
diff --git a/dbt/include/databricks/macros/relations/table/create.sql b/dbt/include/databricks/macros/relations/table/create.sql
index 015c42247..2aef97e46 100644
--- a/dbt/include/databricks/macros/relations/table/create.sql
+++ b/dbt/include/databricks/macros/relations/table/create.sql
@@ -16,18 +16,7 @@
   {% endcall %}
 
   {{ apply_alter_constraints(target_relation) }}
-  {%- if replaced_in_place -%}
-    {# Replace preserves tags, so apply only new/changed ones; otherwise apply all. #}
-    {% set tags = adapter.get_table_tags_changes(target_relation, config.model) %}
-    {% set column_tags = adapter.get_column_tags_changes(target_relation, config.model) %}
-  {%- else -%}
-    {% set tags = config.get('databricks_tags') %}
-    {% set column_tags = adapter.get_column_tags_from_model(config.model) %}
-  {%- endif -%}
-  {{ apply_tags(target_relation, tags) }}
-  {% if column_tags and column_tags.set_column_tags %}
-    {{ apply_column_tags(target_relation, column_tags) }}
-  {% endif %}
+  {{ reconcile_tags(target_relation, replaced_in_place) }}
 
   {% call statement('merge into target') %}
     insert into {{ target_relation }} by name select * from {{ intermediate_relation }}
@@ -145,4 +134,4 @@
   {%- else -%}
     {{ create_python_intermediate_table(relation, compiled_code) }}
   {%- endif -%}
-{% endmacro %}
\ No newline at end of file
+{% endmacro %}
diff --git a/dbt/include/databricks/macros/relations/tags.sql b/dbt/include/databricks/macros/relations/tags.sql
index 49defdf44..6c426756a 100644
--- a/dbt/include/databricks/macros/relations/tags.sql
+++ b/dbt/include/databricks/macros/relations/tags.sql
@@ -1,3 +1,18 @@
+{% macro reconcile_tags(relation, replaced_in_place=false) -%}
+  {%- if replaced_in_place -%}
+    {%- set changes = adapter.get_table_replacement_tag_changes(relation, config.model) -%}
+    {%- set tags = changes['table_tags'] -%}
+    {%- set column_tags = {'set_column_tags': changes['column_tags']} -%}
+  {%- else -%}
+    {%- set tags = config.get('databricks_tags') -%}
+    {%- set column_tags = adapter.get_column_tags_from_model(config.model) -%}
+  {%- endif -%}
+  {%- do apply_tags(relation, tags) -%}
+  {%- if column_tags and column_tags.set_column_tags -%}
+    {%- do apply_column_tags(relation, column_tags) -%}
+  {%- endif -%}
+{%- endmacro %}
+
 {% macro fetch_tags(relation) -%}
   {% if relation.is_hive_metastore() %}
     {{ exceptions.raise_compiler_error("Tags are only supported for Unity Catalog") }}
diff --git a/tests/functional/adapter/tags/test_table_tag_fetch_skips.py b/tests/functional/adapter/tags/test_table_tag_fetch_skips.py
index 9dba62886..8a5dd0f9f 100644
--- a/tests/functional/adapter/tags/test_table_tag_fetch_skips.py
+++ b/tests/functional/adapter/tags/test_table_tag_fetch_skips.py
@@ -157,21 +157,37 @@ def models(self):
     def relations_to_reset(self):
         return ("metadata_fetch_table",)
 
-    def test_changed_tag_value_reaches_server(self, project):
-        util.run_dbt(["run"])
-        util.write_file(
-            fixtures.metadata_fetch_table_with_changed_tags_sql,
-            "models",
-            "metadata_fetch_table.sql",
-        )
-        util.run_dbt(["run"])
+    @pytest.fixture(scope="class", params=["table", "incremental"])
+    def materialization(self, request):
+        return request.param
 
+    def assert_tags(self, project, value):
         results = project.run_sql(
             "select tag_name, tag_value from `system`.`information_schema`.`table_tags`"
             " where schema_name = '{schema}' and table_name='metadata_fetch_table'",
             fetch="all",
         )
-        assert set((row[0], row[1]) for row in results) == {("classification", "confidential")}
+        assert {(row[0], row[1]) for row in results} == {("classification", value)}
+
+    def test_changed_tag_value_reaches_server(self, project, materialization):
+        util.write_file(
+            fixtures.metadata_fetch_table_with_tags_sql.replace("'table'", repr(materialization)),
+            "models",
+            "metadata_fetch_table.sql",
+        )
+        util.run_dbt(["run"])
+        self.assert_tags(project, "internal")
+        util.run_dbt(["run", "--full-refresh"])
+        self.assert_tags(project, "internal")
+        util.write_file(
+            fixtures.metadata_fetch_table_with_changed_tags_sql.replace(
+                "'table'", repr(materialization)
+            ),
+            "models",
+            "metadata_fetch_table.sql",
+        )
+        util.run_dbt(["run", "--full-refresh"])
+        self.assert_tags(project, "confidential")
 
 
 @pytest.mark.skip_profile("databricks_cluster")
@@ -189,17 +205,42 @@ def models(self):
     def relations_to_reset(self):
         return ("metadata_fetch_table",)
 
-    def test_changed_column_tag_value_reaches_server(self, project):
-        util.run_dbt(["run"])
-        util.write_file(fixtures.metadata_fetch_changed_column_tags_schema, "models", "schema.yml")
-        util.run_dbt(["run"])
+    @pytest.fixture(scope="class", params=["table", "incremental"])
+    def materialization(self, request):
+        return request.param
 
+    def assert_tags(self, project, value):
         results = project.run_sql(
             "select column_name, tag_name, tag_value from"
             " `system`.`information_schema`.`column_tags`"
             " where schema_name = '{schema}' and table_name='metadata_fetch_table'",
             fetch="all",
         )
-        assert set((row[0], row[1], row[2]) for row in results) == {
-            ("id", "classification", "confidential")
-        }
+        assert {(row[0], row[1], row[2]) for row in results} == {("id", "classification", value)}
+
+    def test_changed_column_tag_value_reaches_server(self, project, materialization):
+        util.write_file(fixtures.metadata_fetch_column_tags_schema, "models", "schema.yml")
+        util.write_file(
+            fixtures.metadata_fetch_table_sql.replace("'table'", repr(materialization)),
+            "models",
+            "metadata_fetch_table.sql",
+        )
+        util.run_dbt(["run"])
+        self.assert_tags(project, "internal")
+        util.run_dbt(["run", "--full-refresh"])
+        self.assert_tags(project, "internal")
+        util.write_file(fixtures.metadata_fetch_changed_column_tags_schema, "models", "schema.yml")
+        util.run_dbt(["run", "--full-refresh"])
+        self.assert_tags(project, "confidential")
+
+
+class TestChangedTableTagValueIsAppliedV2(
+    MaterializationV2Mixin, TestChangedTableTagValueIsApplied
+):
+    pass
+
+
+class TestChangedColumnTagValueIsAppliedV2(
+    MaterializationV2Mixin, TestChangedColumnTagValueIsApplied
+):
+    pass
diff --git a/tests/unit/macros/relations/test_tags_macros.py b/tests/unit/macros/relations/test_tags_macros.py
index f56afbac3..da0c70742 100644
--- a/tests/unit/macros/relations/test_tags_macros.py
+++ b/tests/unit/macros/relations/test_tags_macros.py
@@ -1,3 +1,5 @@
+from unittest.mock import Mock
+
 import pytest
 
 from dbt.adapters.databricks.relation import DatabricksRelationType
@@ -5,6 +7,45 @@
 
 
 class TestTagsMacros(MacroTestBase):
+    @pytest.fixture
+    def default_context(self):
+        context = super().default_context.__wrapped__(self)
+        context["apply_column_tags"] = Mock(return_value="")
+        context["statement"] = Mock(side_effect=lambda name, caller: caller())
+        return context
+
+    @pytest.mark.parametrize("replaced_in_place", [False, True])
+    @pytest.mark.parametrize("has_changes", [False, True])
+    def test_reconcile_tags(self, template_bundle, config, replaced_in_place, has_changes):
+        context = template_bundle.context
+        adapter = context["adapter"]
+        relation = template_bundle.relation
+        relation.is_hive_metastore.return_value = False
+        config["databricks_tags"] = {"classification": "internal"}
+        desired_columns = {"set_column_tags": {"id": {"pii": "false"}}}
+        adapter.get_column_tags_from_model.return_value = desired_columns
+        adapter.get_table_replacement_tag_changes.return_value = {
+            "table_tags": {"classification": "internal"} if has_changes else {},
+            "column_tags": desired_columns["set_column_tags"] if has_changes else {},
+        }
+
+        self.render_bundle(template_bundle, "reconcile_tags", replaced_in_place)
+
+        if replaced_in_place:
+            adapter.get_table_replacement_tag_changes.assert_called_once_with(
+                relation, context["config"].model
+            )
+            adapter.get_column_tags_from_model.assert_not_called()
+        else:
+            adapter.get_table_replacement_tag_changes.assert_not_called()
+            adapter.get_column_tags_from_model.assert_called_once_with(context["config"].model)
+        if has_changes or not replaced_in_place:
+            context["statement"].assert_called_once()
+            context["apply_column_tags"].assert_called_once_with(relation, desired_columns)
+        else:
+            context["statement"].assert_not_called()
+            context["apply_column_tags"].assert_not_called()
+
     @pytest.fixture
     def template_name(self) -> str:
         return "tags.sql"
diff --git a/tests/unit/test_adapter.py b/tests/unit/test_adapter.py
index 92520bca8..c0691d05f 100644
--- a/tests/unit/test_adapter.py
+++ b/tests/unit/test_adapter.py
@@ -123,6 +123,65 @@ def _get_config(
         return config_from_parts_or_dicts(self.project_cfg, self.profile_cfg)
 
 
+class TestChangedTags:
+    @pytest.mark.parametrize("table_tags", [False, True])
+    @pytest.mark.parametrize("column_tags", [False, True])
+    @pytest.mark.parametrize("hms", [False, True])
+    def test_fetches_only_configured_tags(self, table_tags, column_tags, hms):
+        adapter = Mock()
+        relation = Mock()
+        relation.is_hive_metastore.return_value = hms
+        model = Mock()
+        model.config.extra = {
+            "databricks_tags": {"classification": "internal"} if table_tags else {}
+        }
+        model.columns = (
+            {"id": {"_extra": {"databricks_tags": {"pii": "false"}}}} if column_tags else {}
+        )
+        adapter.execute_macro.side_effect = lambda name, **kwargs: agate.Table(
+            [],
+            ["tag_name", "tag_value"]
+            if name == "fetch_tags"
+            else ["column_name", "tag_name", "tag_value"],
+        )
+
+        result = DatabricksAdapter.get_table_replacement_tag_changes(adapter, relation, model)
+
+        assert result == {
+            "table_tags": {"classification": "internal"} if table_tags else {},
+            "column_tags": {"id": {"pii": "false"}} if column_tags else {},
+        }
+        expected = (
+            []
+            if hms
+            else (
+                (["fetch_tags"] if table_tags else [])
+                + (["fetch_column_tags"] if column_tags else [])
+            )
+        )
+        assert [call.args[0] for call in adapter.execute_macro.call_args_list] == expected
+
+    def test_unchanged_tags_are_empty(self):
+        adapter = Mock()
+        relation = Mock()
+        relation.is_hive_metastore.return_value = False
+        model = Mock()
+        model.config.extra = {"databricks_tags": {"classification": "internal"}}
+        model.columns = {"ID": {"_extra": {"databricks_tags": {"pii": "false"}}}}
+        adapter.execute_macro.side_effect = [
+            agate.Table([("classification", "internal")], ["tag_name", "tag_value"]),
+            agate.Table(
+                [("id", "pii", "false")],
+                ["column_name", "tag_name", "tag_value"],
+                column_types=[agate.Text(), agate.Text(), agate.Text()],
+            ),
+        ]
+        assert DatabricksAdapter.get_table_replacement_tag_changes(adapter, relation, model) == {
+            "table_tags": {},
+            "column_tags": {},
+        }
+
+
 class TestDatabricksAdapter(DatabricksAdapterBase):
     @pytest.fixture(autouse=True)
     def _stub_spog_probe(self):