Fix insert_overwrite handling of schema update (#1058)

Author: ericj-db

CHANGELOG.md

@@ -1,4 +1,10 @@
-## dbt-databricks 1.10.3 (TBD)
+## dbt-databricks 1.10.4 (TBD)
+
+### Fixes
+
+- Fix bug where schema update causes insert_overwrite strategy to fail on subsequent runs ([1057](https://github.com/databricks/dbt-databricks/issues/1057))
+
+## dbt-databricks 1.10.3 (June 4, 2025)
 
 ### Features
 

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

@@ -25,15 +25,22 @@
     {{ return(get_insert_overwrite_sql(source, target)) }}
 {% endmacro %}
 
-
 {% macro get_insert_overwrite_sql(source_relation, target_relation) %}
-
-    {%- set dest_columns = adapter.get_columns_in_relation(target_relation) -%}
-    {%- set dest_cols_csv = dest_columns | map(attribute='quoted') | join(', ') -%}
+    {%- set dest_columns = adapter.get_columns_in_relation(target_relation) | map(attribute='quoted') | list -%}
+    {%- set source_columns = adapter.get_columns_in_relation(source_relation) | map(attribute='quoted') | list -%}
+    {%- set common_columns = [] -%}
+    {%- for dest_col in dest_columns -%}
+      {%- if dest_col in source_columns -%}
+        {%- do common_columns.append(dest_col) -%}
+      {%- else -%}
+        {%- do common_columns.append('DEFAULT') -%}
+      {%- endif -%}
+    {%- endfor -%}
+    {%- set dest_cols_csv = dest_columns | join(', ') -%}
+    {%- set source_cols_csv = common_columns | join(', ') -%}
     insert overwrite table {{ target_relation }}
     {{ partition_cols(label="partition") }}
-    select {{dest_cols_csv}} from {{ source_relation }}
-
+    select {{source_cols_csv}} from {{ source_relation }}
 {% endmacro %}
 
 {% macro get_replace_where_sql(args_dict) -%}

tests/functional/adapter/incremental/fixtures.py

@@ -223,6 +223,12 @@
 3,anyway
 """
 
+upsert_expected_no_msg = """id,msg
+1,hello
+2,null
+3,null
+"""
+
 upsert_expected = """id,msg
 1,hello
 2,yo
@@ -303,6 +309,26 @@
 {% endif %}
 """
 
+update_schema_model = """
+{{ config(
+    materialized = 'incremental'
+) }}
+
+{% if not is_incremental() %}
+
+select cast(1 as bigint) as id, 'hello' as msg
+union all
+select cast(2 as bigint) as id, 'goodbye' as msg
+
+{% else %}
+
+select cast(2 as bigint) as id
+union all
+select cast(3 as bigint) as id
+
+{% endif %}
+"""
+
 upsert_model = """
 {{ config(
     materialized = 'incremental',

tests/functional/adapter/incremental/test_incremental_strategies.py

@@ -120,6 +120,34 @@ def test_incremental(self, project):
         util.check_relations_equal(project.adapter, ["overwrite_model", "upsert_expected"])
 
 
+@pytest.mark.skip_profile("databricks_uc_sql_endpoint")
+class TestInsertOverwriteChangeSchema(InsertOverwriteBase):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "overwrite_model.sql": fixtures.update_schema_model,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "models": {
+                "+incremental_strategy": "insert_overwrite",
+                "+partition_by": "id",
+            },
+        }
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {
+            "overwrite_expected.csv": fixtures.upsert_expected_no_msg,
+        }
+
+    def test_incremental(self, project):
+        self.seed_and_run_twice()
+        util.check_relations_equal(project.adapter, ["overwrite_model", "overwrite_expected"])
+
+
 # Only runs under SQL warehouse profile, but overrides compute at model level
 @pytest.mark.skip_profile("databricks_uc_cluster", "databricks_cluster")
 class TestInsertOverwriteWithModelComputeOverride(IncrementalBase):