Gets columns to update from config for BQ and Snowflake

plugins/bigquery/dbt/adapters/bigquery/impl.py

@@ -116,6 +116,7 @@ class BigqueryConfig(AdapterConfig):
     hours_to_expiration: Optional[int] = None
     require_partition_filter: Optional[bool] = None
     partition_expiration_days: Optional[int] = None
+    incrmental_update_columns: Optional[str] = None
 
 
 class BigQueryAdapter(BaseAdapter):

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

@@ -112,7 +112,10 @@
       {% endif %}
       {% set build_sql = create_table_as(False, target_relation, sql) %}
   {% else %}
-     {% set dest_columns = adapter.get_columns_in_relation(existing_relation) %}
+     {% set dest_columns = config.get('incremental_update_columns', none) %}
+     {% if dest_columns is none %}
+         {% set dest_columns = adapter.get_columns_in_relation(existing_relation) %}
+     {% endif %}
 
      {#-- if partitioned, use BQ scripting to get the range of partition values to be updated --#}
      {% if strategy == 'insert_overwrite' %}

plugins/snowflake/dbt/adapters/snowflake/impl.py

@@ -26,6 +26,7 @@ class SnowflakeConfig(AdapterConfig):
     copy_grants: Optional[bool] = None
     snowflake_warehouse: Optional[str] = None
     query_tag: Optional[str] = None
+    incremental_update_columns: Optional[str] = None
 
 
 class SnowflakeAdapter(SQLAdapter):

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

@@ -58,7 +58,10 @@
     {% do adapter.expand_target_column_types(
            from_relation=tmp_relation,
            to_relation=target_relation) %}
-    {% set dest_columns = adapter.get_columns_in_relation(target_relation) %}
+    {% set dest_columns = config.get('update_columns', none) %}
+    {% if dest_columns is none %}
+        {% 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 %}
 

test/integration/001_simple_copy_test/models-snowflake/incremental_update_cols.sql

@@ -0,0 +1,17 @@
+{{
+    config(
+        materialized = "incremental",
+        unique_key = "id",
+        incremental_update_columns = ["email", "ip_address"]
+    )
+}}
+
+
+select *
+from {{ ref('seed') }}
+
+{% if is_incremental() %}
+
+    where id > (select max(id) from {{this}})
+
+{% endif %}