Use common columns for incremental schema changes

CHANGELOG.md

@@ -4,14 +4,16 @@
 N/A
 
 ### Fixes
-- Fix problem with bytes proccessed return None value when the service account used to connect DBT in bigquery had a row policy access.
+- Fix problem with bytes processed return None value when the service account used to connect DBT in bigquery had a row policy access.
 ([#47](https://github.com/dbt-labs/dbt-bigquery/issues/47), [#48](https://github.com/dbt-labs/dbt-bigquery/pull/48))
+- 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
 - Capping `google-api-core` to version `1.31.3` due to `protobuf` dependency conflict ([#53](https://github.com/dbt-labs/dbt-bigquery/pull/53))
 
 ### Contributors
 - [@imartynetz](https://github.com/imartynetz) ([#48](https://github.com/dbt-labs/dbt-bigquery/pull/48))
+- [@Kayrnt](https://github.com/Kayrnt) ([51](https://github.com/dbt-labs/dbt-bigquery/pull/51))
 
 ## dbt-bigquery 1.0.0b2 (October 25, 2021)
 

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

@@ -171,16 +171,25 @@
       {% set build_sql = create_table_as(False, target_relation, sql) %}
 
   {% else %}
+    {% set schema_changes_dict = {} %}
     {% set tmp_relation_exists = false %}
     {% if on_schema_change != 'ignore' %} {# Check first, since otherwise we may not build a temp table #}
       {% do run_query(
         declare_dbt_max_partition(this, partition_by, sql) + create_table_as(True, tmp_relation, sql)
       ) %}
       {% set tmp_relation_exists = true %}
-      {% do process_schema_changes(on_schema_change, tmp_relation, existing_relation) %}
+      {% set schema_changes_dict = process_schema_changes(on_schema_change, tmp_relation, existing_relation) %}
+      {% if on_schema_change == 'append_new_columns' %}
+        {#-- As we append new columns, destination columns are the ones from source --#}
+        {% set dest_columns = schema_changes_dict.get('source_columns', existing_columns) %}
+      {% else %}
+        {#-- Destination columns is the intersection of source and target table --#}
+        {% set dest_columns = schema_changes_dict.get('in_target_and_source', existing_columns) %}
+      {% endif %}
+     {% else %}
+      {% set dest_columns = adapter.get_columns_in_relation(existing_relation) %}
     {% endif %}
-
-    {% set dest_columns = adapter.get_columns_in_relation(existing_relation) %}
+
     {% set build_sql = bq_generate_incremental_build_sql(
         strategy, tmp_relation, target_relation, sql, unique_key, partition_by, partitions, dest_columns, tmp_relation_exists
     ) %}

tests/integration/incremental_schema_tests/models/incremental_append_new_columns.sql

@@ -14,8 +14,7 @@ WITH source_data AS (SELECT * FROM {{ ref('model_a') }} )
 
 SELECT id, 
        cast(field1 as {{string_type}}) as field1, 
-       cast(field2 as {{string_type}}) as field2, 
-       cast(field3 as {{string_type}}) as field3, 
+       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 }} )