Use common columns for incremental schema changes

CHANGELOG.md

@@ -4,7 +4,7 @@
 N/A
 
 ### Fixes
-N/A
+- When on_schema_change is set, pass common columns as dest_columns in incremental merge macros ([#4144](https://github.com/dbt-labs/dbt-core/issues/4144))
 
 ### Under the hood
 - Add optional profile parameters for atypical local connection setups ([#21](https://github.com/dbt-labs/dbt-snowflake/issues/21), [#36](https://github.com/dbt-labs/dbt-snowflake/pull/36))
@@ -19,6 +19,7 @@ N/A
 
 ### Contributors
 - [@NiallRees](https://github.com/NiallRees) ([#32](https://github.com/dbt-labs/dbt-snowflake/pull/32))
+- [@Kayrnt](https://github.com/Kayrnt) ([#38](https://github.com/dbt-labs/dbt-snowflake/pull/38))
 
 ## dbt-snowflake v1.0.0b1 (October 11, 2021)
 

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

@@ -58,8 +58,11 @@
     {% do adapter.expand_target_column_types(
            from_relation=tmp_relation,
            to_relation=target_relation) %}
-    {% do process_schema_changes(on_schema_change, tmp_relation, existing_relation) %}
-    {% set dest_columns = adapter.get_columns_in_relation(existing_relation) %}
+    {#-- Process schema changes. Returns dict of changes if successful. Use source columns for upserting/merging --#}
+    {% set dest_columns = process_schema_changes(on_schema_change, tmp_relation, existing_relation) %}
+    {% if not dest_columns %}
+      {% set dest_columns = adapter.get_columns_in_relation(existing_relation) %}
+    {% endif %}
     {% set build_sql = dbt_snowflake_get_incremental_sql(strategy, tmp_relation, target_relation, unique_key, dest_columns) %}
 
   {% endif %}

tests/integration/incremental_schema_tests/models/incremental_append_new_columns_remove_one.sql

@@ -0,0 +1,28 @@
+{{
+    config(
+        materialized='incremental',
+        unique_key='id',
+        on_schema_change='append_new_columns'
+    )
+}}
+
+{% set string_type = 'string' if target.type == 'bigquery' else 'varchar(10)' %}
+
+WITH source_data AS (SELECT * FROM {{ ref('model_a') }} )
+
+{% if is_incremental()  %}
+
+SELECT id,
+       cast(field1 as {{string_type}}) as field1,
+       cast(field3 as {{string_type}}) as field3,
+       cast(field4 as {{string_type}}) as field4
+FROM source_data WHERE id NOT IN (SELECT id from {{ this }} )
+
+{% else %}
+
+SELECT id,
+       cast(field1 as {{string_type}}) as field1,
+       cast(field2 as {{string_type}}) as field2
+FROM source_data where id <= 3
+
+{% endif %}

tests/integration/incremental_schema_tests/models/incremental_append_new_columns_remove_one_target.sql

@@ -0,0 +1,19 @@
+{{
+    config(materialized='table')
+}}
+
+{% set string_type = 'string' if target.type == 'bigquery' else 'varchar(10)' %}
+
+with source_data as (
+
+    select * from {{ ref('model_a') }}
+
+)
+
+select id,
+       cast(field1 as {{string_type}}) as field1,
+       cast(CASE WHEN id >  3 THEN NULL ELSE field2 END as {{string_type}}) AS field2,
+       cast(CASE WHEN id <= 3 THEN NULL ELSE field3 END as {{string_type}}) AS field3,
+       cast(CASE WHEN id <= 3 THEN NULL ELSE field4 END as {{string_type}}) AS field4
+
+from source_data

tests/integration/incremental_schema_tests/models/schema.yml

@@ -35,6 +35,20 @@ models:
         tags: [column_level_tag]
         tests:
           - unique
+
+  - name: incremental_append_new_columns_remove_one
+    columns:
+      - name: id
+        tags: [column_level_tag]
+        tests:
+          - unique
+
+  - name: incremental_append_new_columns_remove_one_target
+    columns:
+      - name: id
+        tags: [column_level_tag]
+        tests:
+          - unique
 
   - name: incremental_sync_all_columns
     columns:

tests/integration/incremental_schema_tests/test_incremental_schema.py

@@ -87,6 +87,21 @@ def run_incremental_append_new_columns(self):
         self.list_tests_and_assert(select, exclude, expected)
         self.run_tests_and_assert(select, exclude, expected, compare_source, compare_target)
 
+    def run_incremental_append_new_columns_remove_one(self):
+        select = 'model_a incremental_append_new_columns_remove_one incremental_append_new_columns_remove_one_target'
+        compare_source = 'incremental_append_new_columns_remove_one'
+        compare_target = 'incremental_append_new_columns_remove_one_target'
+        exclude = None
+        expected = [
+            'select_from_a',
+            'select_from_incremental_append_new_columns_remove_one',
+            'select_from_incremental_append_new_columns_remove_one_target',
+            'unique_model_a_id',
+            'unique_incremental_append_new_columns_remove_one_id',
+            'unique_incremental_append_new_columns_remove_one_target_id'
+        ]
+        self.run_tests_and_assert(select, exclude, expected, compare_source, compare_target)
+
     def run_incremental_sync_all_columns(self):
         select = 'model_a incremental_sync_all_columns incremental_sync_all_columns_target'
         compare_source = 'incremental_sync_all_columns'
@@ -116,6 +131,7 @@ def test__snowflake__run_incremental_ignore(self):
     @use_profile('snowflake')
     def test__snowflake__run_incremental_append_new_columns(self):
         self.run_incremental_append_new_columns()
+        self.run_incremental_append_new_columns_remove_one()
 
     @use_profile('snowflake')
     def test__snowflake__run_incremental_sync_all_columns(self):

tests/integration/incremental_schema_tests/tests/select_from_incremental_append_new_columns_remove_one.sql

@@ -0,0 +1 @@
+select * from {{ ref('incremental_append_new_columns_remove_one') }} where false

tests/integration/incremental_schema_tests/tests/select_from_incremental_append_new_columns_remove_one_target.sql

@@ -0,0 +1 @@
+select * from {{ ref('incremental_append_new_columns_remove_one_target') }} where false