Merge branch 'main' into 1.10.latest

ericj-db · ericj-db · commit 4cc5413f97ff · 2025-08-07T16:08:43.000-07:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,4 +1,12 @@
-## dbt-databricks 1.10.9 (TBD)
+## dbt-databricks 1.10.10 (TBD)
+
+## dbt-databricks 1.10.9 (August 7, 2025)
+
+### Features
+- Support column tags for views using `ALTER TABLE`
+
+### Under the hood
+- Revert `REPLACE USING` syntax being used for insert overwrite ([1025](https://github.com/databricks/dbt-databricks/issues/1025))
 
 ## dbt-databricks 1.10.8 (August 4, 2025)
 
diff --git a/dbt/adapters/databricks/__version__.py b/dbt/adapters/databricks/__version__.py
@@ -1 +1 @@
-version = "1.10.8"
+version = "1.10.9"
diff --git a/dbt/adapters/databricks/relation_configs/view.py b/dbt/adapters/databricks/relation_configs/view.py
@@ -8,6 +8,7 @@
     DatabricksRelationConfigBase,
 )
 from dbt.adapters.databricks.relation_configs.column_comments import ColumnCommentsProcessor
+from dbt.adapters.databricks.relation_configs.column_tags import ColumnTagsProcessor
 from dbt.adapters.databricks.relation_configs.comment import CommentProcessor
 from dbt.adapters.databricks.relation_configs.query import QueryProcessor
 from dbt.adapters.databricks.relation_configs.tags import TagsProcessor
@@ -21,6 +22,7 @@ class ViewConfig(DatabricksRelationConfigBase):
         QueryProcessor,
         CommentProcessor,
         ColumnCommentsProcessor,
+        ColumnTagsProcessor,
     ]
 
     def get_changeset(self, existing: Self) -> Optional[DatabricksRelationChangeSet]:
diff --git a/dbt/include/databricks/macros/materializations/incremental/incremental.sql b/dbt/include/databricks/macros/materializations/incremental/incremental.sql
@@ -52,7 +52,7 @@
     {%- else -%}
       {{ log("Existing relation found, proceeding with incremental work")}}
       {#-- Set Overwrite Mode to DYNAMIC for subsequent incremental operations --#}
-      {%- if incremental_strategy == 'insert_overwrite' and partition_by and adapter.compare_dbr_version(16, 3) < 0 -%}
+      {%- if incremental_strategy == 'insert_overwrite' and partition_by -%}
         {{ set_overwrite_mode('DYNAMIC') }}
       {%- endif -%}
       {#-- Relation must be merged --#}
@@ -118,7 +118,7 @@
       {% do persist_docs(target_relation, model, for_relation=language=='python') %}
     {%- else -%}
       {#-- Set Overwrite Mode to DYNAMIC for subsequent incremental operations --#}
-      {%- if incremental_strategy == 'insert_overwrite' and partition_by and adapter.compare_dbr_version(16, 3) < 0 -%}
+      {%- if incremental_strategy == 'insert_overwrite' and partition_by -%}
         {{ set_overwrite_mode('DYNAMIC') }}
       {%- endif -%}
       {#-- Relation must be merged --#}
@@ -177,7 +177,7 @@
     {{ run_hooks(post_hooks) }}
   {%- endif -%}
 
-  {%- if incremental_strategy == 'insert_overwrite' and not full_refresh and adapter.compare_dbr_version(16, 3) < 0 -%}
+  {%- if incremental_strategy == 'insert_overwrite' and not full_refresh -%}
     {{ set_overwrite_mode('STATIC') }}
   {%- endif -%}
 
@@ -191,7 +191,7 @@
       set spark.sql.sources.partitionOverwriteMode = {{ value }}
     {%- endcall -%}
   {% else %}
-    {{ exceptions.warn("INSERT OVERWRITE is supported on SQL warehouses with DBR 16.3+. On older DBR versions, this strategy would be equivalent to using the table materialization.") }}
+    {{ exceptions.warn("INSERT OVERWRITE is only properly supported on all-purpose clusters.  On SQL Warehouses, this strategy would be equivalent to using the table materialization.") }}
   {% endif %}
 {% endmacro %}
 
@@ -216,4 +216,4 @@
     {%- set configuration_changes = model_config.get_changeset(existing_config) -%}
     {{ apply_config_changeset(target_relation, model, configuration_changes) }}
   {% endif %}
-{% endmacro %}
+{% endmacro %}
diff --git a/dbt/include/databricks/macros/materializations/incremental/strategies.sql b/dbt/include/databricks/macros/materializations/incremental/strategies.sql
@@ -38,32 +38,9 @@
     {%- endfor -%}
     {%- set dest_cols_csv = dest_columns | join(', ') -%}
     {%- set source_cols_csv = common_columns | join(', ') -%}
-    
-    {%- if adapter.compare_dbr_version(16, 3) >= 0 -%}
-        {{ get_insert_replace_using_sql(source_relation, target_relation, source_cols_csv) }}
-    {%- else -%}
-        {#-- Use traditional INSERT OVERWRITE for older DBR versions --#}
-        insert overwrite table {{ target_relation }}
-        {{ partition_cols(label="partition") }}
-        select {{ source_cols_csv }} from {{ source_relation }}
-    {%- endif -%}
-{% endmacro %}
-
-{% macro get_insert_replace_using_sql(source_relation, target_relation, source_cols_csv) %}
-    {%- set partition_by = config.get('partition_by') -%}
-    {%- if partition_by -%}
-        {%- if partition_by is string -%}
-            {%- set partition_by = [partition_by] -%}
-        {%- endif -%}
-        {%- set partition_cols_csv = partition_by | join(', ') -%}
-        insert into table {{ target_relation }}
-        replace using ({{ partition_cols_csv }})
-        select {{ source_cols_csv }} from {{ source_relation }}
-    {%- else -%}
-        {#-- Fallback to regular insert if no partitions defined --#}
-        insert overwrite table {{ target_relation }}
-        select {{ source_cols_csv }} from {{ source_relation }}
-    {%- endif -%}
+    insert overwrite table {{ target_relation }}
+    {{ partition_cols(label="partition") }}
+    select {{source_cols_csv}} from {{ source_relation }}
 {% endmacro %}
 
 {% macro get_replace_where_sql(args_dict) -%}
diff --git a/dbt/include/databricks/macros/materializations/incremental/validate.sql b/dbt/include/databricks/macros/materializations/incremental/validate.sql
@@ -24,6 +24,12 @@
     You can only choose this strategy when file_format is set to 'delta'
   {%- endset %}
 
+  {% set invalid_insert_overwrite_endpoint_msg -%}
+    Invalid incremental strategy provided: {{ raw_strategy }}
+    You cannot use this strategy when connecting via warehouse
+    Use the 'merge' or 'replace_where' strategy instead
+  {%- endset %}
+
   {% if raw_strategy not in adapter.valid_incremental_strategies() %}
     {{ log("WARNING - You are using an unsupported incremental strategy: " ~ raw_strategy) }}
     {{ log("You can ignore this warning if you are using a custom incremental strategy") }}
diff --git a/dbt/include/databricks/macros/materializations/view.sql b/dbt/include/databricks/macros/materializations/view.sql
@@ -31,6 +31,10 @@
         {{ get_create_view_as_sql(target_relation, sql) }}
       {%- endcall %}
       {{ 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 %}
     {% endif %}
     {% set should_revoke = should_revoke(exists_as_view, full_refresh_mode=True) %}
     {% do apply_grants(target_relation, grant_config, should_revoke=True) %}
@@ -57,6 +61,11 @@
 
     {%- 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 %}
+
     {{ run_hooks(post_hooks) }}
   {% endif %}
 
@@ -69,6 +78,11 @@
   {% set tags = config.get('databricks_tags') %}
   {{ execute_multiple_statements(get_replace_sql(existing_relation, target_relation, sql)) }}
   {%- 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 %}
 {% endmacro %}
 
 {% macro relation_should_be_altered(existing_relation) %}
diff --git a/dbt/include/databricks/macros/relations/components/column_tags.sql b/dbt/include/databricks/macros/relations/components/column_tags.sql
@@ -34,7 +34,12 @@
 {%- endmacro -%}
 
 {% macro alter_set_column_tags(relation, column, tags) -%}
-  ALTER {{ relation.type | replace('_', ' ') }} {{ relation.render() }}
+  {# ALTER VIEW does not support setting column tags, but ALTER TABLE works for views #}
+  {%- if relation.type == 'view' -%}
+    ALTER TABLE {{ relation.render() }}
+  {%- else -%}
+    ALTER {{ relation.type | replace('_', ' ') }} {{ relation.render() }}
+  {%- endif -%}
   ALTER COLUMN `{{ column }}`
   SET TAGS (
     {%- for tag_name, tag_value in tags.items() -%}
diff --git a/dbt/include/databricks/macros/relations/view/create.sql b/dbt/include/databricks/macros/relations/view/create.sql
@@ -2,9 +2,6 @@
   {% if column_mask_exists() %}
     {% do exceptions.raise_compiler_error("Column masks are not supported for views.") %}
   {% endif %}
-  {% if column_tags_exist() %}
-    {% do exceptions.raise_compiler_error("Column tags are not supported for views.") %}
-  {% endif %}
   {{ log("Creating view " ~ relation) }}
   create or replace view {{ relation.render() }}
   {%- if config.persist_column_docs() -%}
diff --git a/tests/functional/adapter/column_tags/fixtures.py b/tests/functional/adapter/column_tags/fixtures.py
@@ -2,7 +2,7 @@
 select 1 as id, 'abc123' as account_number
 """
 
-model_with_column_tags = """
+initial_column_tag_model = """
 version: 2
 models:
   - name: base_model
@@ -16,6 +16,22 @@
           sensitive: "true"
 """
 
+updated_column_tag_model = """
+version: 2
+models:
+  - name: base_model
+    config:
+        materialized: table
+    columns:
+      - name: id
+        databricks_tags:
+          pii: "false"
+      - name: account_number
+        databricks_tags:
+          pii: "true"
+          sensitive: "true"
+"""
+
 column_tags_seed = """
 id,account_number
 1,'1234567890'
diff --git a/tests/functional/adapter/column_tags/test_column_tags.py b/tests/functional/adapter/column_tags/test_column_tags.py
@@ -6,67 +6,76 @@
 
 
 class ColumnTagsMixin(MaterializationV2Mixin):
+    @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", f"materialized: {self.relation_type}"
+            ),
+        }
+
     def test_column_tags(self, project):
         util.run_dbt(["run"])
 
-        # Check that column tags were applied
-        tags = project.run_sql(
-            f"""
+        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
-            """,
-            fetch="all",
-        )
+            """
 
+        # Check that column tags were applied
+        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
 
+        # Run a second time with an updated model
+        util.write_file(
+            fixtures.updated_column_tag_model.replace(
+                "materialized: table", f"materialized: {self.relation_type}"
+            ),
+            "models",
+            "schema.yml",
+        )
+        util.run_dbt(["run"])
+
+        # Check that column tags were updated
+        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
 
 
 @pytest.mark.skip_profile("databricks_cluster")
 class TestColumnTagsTable(ColumnTagsMixin):
-    @pytest.fixture(scope="class")
-    def models(self):
-        return {
-            "base_model.sql": fixtures.base_model_sql,
-            "schema.yml": fixtures.model_with_column_tags,
-        }
+    relation_type = "table"
 
 
 @pytest.mark.skip_profile("databricks_cluster")
 class TestColumnTagsIncremental(ColumnTagsMixin):
-    @pytest.fixture(scope="class")
-    def models(self):
-        return {
-            "base_model.sql": fixtures.base_model_sql,
-            "schema.yml": fixtures.model_with_column_tags.replace(
-                "materialized: table", "materialized: incremental"
-            ),
-        }
+    relation_type = "incremental"
 
 
 @pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
 class TestColumnTagsMaterializedView(ColumnTagsMixin):
-    @pytest.fixture(scope="class")
-    def models(self):
-        return {
-            "base_model.sql": fixtures.base_model_sql,
-            "schema.yml": fixtures.model_with_column_tags.replace(
-                "materialized: table", "materialized: materialized_view"
-            ),
-        }
+    relation_type = "materialized_view"
 
 
 @pytest.mark.skip_profile("databricks_cluster", "databricks_uc_cluster")
 class TestStreamingTableColumnTags(ColumnTagsMixin):
+    relation_type = "streaming_table"
+
     @pytest.fixture(scope="class")
     def seeds(self):
         return {
@@ -77,7 +86,7 @@ def seeds(self):
     def models(self):
         return {
             "base_model.sql": fixtures.base_model_streaming_table,
-            "schema.yml": fixtures.model_with_column_tags.replace(
+            "schema.yml": fixtures.initial_column_tag_model.replace(
                 "materialized: table", "materialized: streaming_table"
             ),
         }
@@ -88,16 +97,5 @@ def setup_streaming_table_seed(self, project):
 
 
 @pytest.mark.skip_profile("databricks_cluster")
-class TestViewColumnTagsFailure(MaterializationV2Mixin):
-    @pytest.fixture(scope="class")
-    def models(self):
-        return {
-            "base_model.sql": fixtures.base_model_sql,
-            "schema.yml": fixtures.model_with_column_tags.replace(
-                "materialized: table", "materialized: view"
-            ),
-        }
-
-    def test_view_column_tags_failure(self, project):
-        result = util.run_dbt(["run"], expect_pass=False)
-        assert "Column tags are not supported" in result.results[0].message
+class TestColumnTagsView(ColumnTagsMixin):
+    relation_type = "view"
diff --git a/tests/functional/adapter/incremental/test_incremental_strategies.py b/tests/functional/adapter/incremental/test_incremental_strategies.py
diff --git a/tests/unit/macros/materializations/incremental/test_insert_overwrite_macros.py b/tests/unit/macros/materializations/incremental/test_insert_overwrite_macros.py

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-version = "1.10.8"`
	`1`	`+version = "1.10.9"`
Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@`
`8`	`8`	`DatabricksRelationConfigBase,`
`9`	`9`	`)`
`10`	`10`	`from dbt.adapters.databricks.relation_configs.column_comments import ColumnCommentsProcessor`
	`11`	`+from dbt.adapters.databricks.relation_configs.column_tags import ColumnTagsProcessor`
`11`	`12`	`from dbt.adapters.databricks.relation_configs.comment import CommentProcessor`
`12`	`13`	`from dbt.adapters.databricks.relation_configs.query import QueryProcessor`
`13`	`14`	`from dbt.adapters.databricks.relation_configs.tags import TagsProcessor`
`@@ -21,6 +22,7 @@ class ViewConfig(DatabricksRelationConfigBase):`
`21`	`22`	`QueryProcessor,`
`22`	`23`	`CommentProcessor,`
`23`	`24`	`ColumnCommentsProcessor,`
	`25`	`+ ColumnTagsProcessor,`
`24`	`26`	`]`
`25`	`27`
`26`	`28`	`def get_changeset(self, existing: Self) -> Optional[DatabricksRelationChangeSet]:`