perf: apply only new or changed databricks_tags on table re-runs

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

dbt/adapters/databricks/impl.py

@@ -1160,6 +1160,39 @@ 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_replacement_tag_changes(
+        self, relation: DatabricksRelation, model: RelationConfig
+    ) -> dict[str, Union[dict[str, str], dict[str, dict[str, str]]]]:
+        """Reconcile tags retained by CREATE OR REPLACE TABLE.
+
+        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:
         if config.get("table_format") == constants.ICEBERG_TABLE_FORMAT:

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 %}
+{% endmacro %}

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

@@ -4,11 +4,14 @@
   {%- 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') %}
   {% 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. Shallow clones are excluded because
+     their table type cannot be changed in place, so they are dropped and recreated. #}
+  {%- set replaced_in_place = existing_relation and not existing_relation.is_shallow_clone 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.is_shallow_clone or 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.is_shallow_clone or 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,12 +64,7 @@
     {% if language=="python" %}
       {% do apply_tblproperties(target_relation, tblproperties) %}
     {% 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) }}
 
     {% do persist_docs(target_relation, model, for_relation=language=='python') %}
 

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,11 +16,7 @@
   {% endcall %}
 
   {{ apply_alter_constraints(target_relation) }}
-  {{ 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) }}
 
   {% call statement('merge into target') %}
     insert into {{ target_relation }} by name select * from {{ intermediate_relation }}
@@ -139,4 +134,4 @@
   {%- else -%}
     {{ create_python_intermediate_table(relation, compiled_code) }}
   {%- endif -%}
-{% endmacro %}
+{% endmacro %}

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") }}

tests/functional/adapter/dbt_clone/fixtures.py

@@ -30,3 +30,23 @@
 
 select 1 as id
 """
+
+tagged_table_model_sql = """
+{{ config(
+    materialized = 'table',
+    databricks_tags = {'classification': 'internal'},
+) }}
+
+select 1 as id
+"""
+
+tagged_table_model_schema_yml = """
+version: 2
+
+models:
+  - name: tagged_table_model
+    columns:
+      - name: id
+        databricks_tags:
+          pii: 'false'
+"""

tests/functional/adapter/dbt_clone/test_dbt_clone.py

@@ -337,3 +337,53 @@ def test_rebuild_over_shallow_clone(self, project, unique_schema, other_schema,
             fetch="all",
         )
         assert [row[0] for row in rows] == [1]
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestRebuildOverShallowCloneAppliesAllTags(BaseClone, CleanupMixin):
+    """Rebuilding a tagged table over a shallow clone drops the clone, so the resulting table
+    inherits no tags and every configured tag must be applied rather than diffed away."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "tagged_table_model.sql": fixtures.tagged_table_model_sql,
+            "schema.yml": fixtures.tagged_table_model_schema_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {}
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {}
+
+    def test_rebuild_over_shallow_clone_applies_all_tags(
+        self, project, unique_schema, other_schema
+    ):
+        project.create_test_schema(other_schema)
+        run_dbt(["run"])
+        self.copy_state(project.project_root)
+
+        # Clone into the other schema so the target starts life as a shallow clone.
+        run_dbt(["clone", "--state", "state", "--target", "otherschema"])
+        assert _table_type(project, other_schema, "tagged_table_model") == "MANAGED_SHALLOW_CLONE"
+
+        run_dbt(["run", "--target", "otherschema", "--full-refresh", "-s", "tagged_table_model"])
+        assert _table_type(project, other_schema, "tagged_table_model") == "MANAGED"
+
+        table_tags = project.run_sql(
+            "select tag_name, tag_value from `system`.`information_schema`.`table_tags`"
+            f" where schema_name = '{other_schema}' and table_name = 'tagged_table_model'",
+            fetch="all",
+        )
+        assert {(row[0], row[1]) for row in table_tags} == {("classification", "internal")}
+
+        column_tags = project.run_sql(
+            "select column_name, tag_name, tag_value from"
+            " `system`.`information_schema`.`column_tags`"
+            f" where schema_name = '{other_schema}' and table_name = 'tagged_table_model'",
+            fetch="all",
+        )
+        assert {(row[0], row[1], row[2]) for row in column_tags} == {("id", "pii", "false")}

tests/functional/adapter/tags/fixtures.py

@@ -81,3 +81,83 @@ 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
+"""
+
+# 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
+"""