Feature/incremental predicates

.changes/unreleased/Features-20220823-093457.yaml

@@ -0,0 +1,7 @@
+kind: Features
+body: incremental predicates
+time: 2022-08-23T09:34:57.026688-05:00
+custom:
+  Author: dave-connors-3
+  Issue: "435"
+  PR: "436"

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

@@ -13,6 +13,7 @@
   {%- set partition_by = config.get('partition_by', none) -%}
   {%- set language = model['language'] -%}
   {%- set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='ignore') -%}
+  {%- set incremental_predicates = config.get('predicates', none) or config.get('incremental_predicates', none) -%}
   {%- set target_relation = this -%}
   {%- set existing_relation = load_relation(this) -%}
   {%- set tmp_relation = make_temp_relation(this) -%}
@@ -54,7 +55,7 @@
     {%- endcall -%}
     {%- do process_schema_changes(on_schema_change, tmp_relation, existing_relation) -%}
     {%- call statement('main') -%}
-      {{ dbt_spark_get_incremental_sql(strategy, tmp_relation, target_relation, unique_key) }}
+      {{ dbt_spark_get_incremental_sql(strategy, tmp_relation, target_relation, unique_key, incremental_predicates) }}
     {%- endcall -%}
     {%- if language == 'python' -%}
       {#--

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

@@ -19,9 +19,9 @@
 {% endmacro %}
 
 
-{% macro spark__get_merge_sql(target, source, unique_key, dest_columns, predicates=none) %}
+{% macro spark__get_merge_sql(target, source, unique_key, dest_columns, incremental_predicates) %}
   {# need dest_columns for merge_exclude_columns, default to use "*" #}
-  {%- set predicates = [] if predicates is none else [] + predicates -%}
+  {%- set predicates = [] if incremental_predicates is none else [] + incremental_predicates -%}
   {%- set dest_columns = adapter.get_columns_in_relation(target) -%}
   {%- set merge_update_columns = config.get('merge_update_columns') -%}
   {%- set merge_exclude_columns = config.get('merge_exclude_columns') -%}
@@ -62,7 +62,7 @@
 {% endmacro %}
 
 
-{% macro dbt_spark_get_incremental_sql(strategy, source, target, unique_key) %}
+{% macro dbt_spark_get_incremental_sql(strategy, source, target, unique_key, incremental_predicates) %}
   {%- if strategy == 'append' -%}
     {#-- insert new records into existing table, without updating or overwriting #}
     {{ get_insert_into_sql(source, target) }}
@@ -71,7 +71,7 @@
     {{ get_insert_overwrite_sql(source, target) }}
   {%- elif strategy == 'merge' -%}
   {#-- merge all columns with databricks delta - schema changes are handled for us #}
-    {{ get_merge_sql(target, source, unique_key, dest_columns=none, predicates=none) }}
+    {{ get_merge_sql(target, source, unique_key, dest_columns=none, incremental_predicates=incremental_predicates) }}
   {%- else -%}
     {% set no_sql_for_strategy_msg -%}
       No known SQL for the incremental strategy provided: {{ strategy }}

tests/functional/adapter/test_incremental_predicates.py

@@ -0,0 +1,67 @@
+import pytest
+from dbt.tests.adapter.incremental.test_incremental_predicates import BaseIncrementalPredicates
+
+
+models__spark_incremental_predicates_sql = """
+{{ config(
+    materialized = 'incremental',
+    unique_key = 'id'
+) }}
+
+{% if not is_incremental() %}
+
+select cast(1 as bigint) as id, 'hello' as msg, 'blue' as color
+union all
+select cast(2 as bigint) as id, 'goodbye' as msg, 'red' as color
+
+{% else %}
+
+-- merge will not happen on the above record where id = 2, so new record will fall to insert
+select cast(1 as bigint) as id, 'hey' as msg, 'blue' as color
+union all
+select cast(2 as bigint) as id, 'yo' as msg, 'green' as color
+union all
+select cast(3 as bigint) as id, 'anyway' as msg, 'purple' as color
+
+{% endif %}
+"""
+
+@pytest.mark.skip_profile('spark_session', 'apache_spark')
+class TestIncrementalPredicatesMergeSpark(BaseIncrementalPredicates):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "models": { 
+                "+incremental_predicates": [
+                    "dbt_internal_dest.id != 2"
+                ],
+                "+incremental_strategy": "merge",
+                "+file_format": "delta"
+            }
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "delete_insert_incremental_predicates.sql": models__spark_incremental_predicates_sql
+        }
+
+@pytest.mark.skip_profile('spark_session', 'apache_spark')
+class TestPredicatesMergeSpark(BaseIncrementalPredicates):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "models": { 
+                "+predicates": [
+                    "dbt_internal_dest.id != 2"
+                ],
+                "+incremental_strategy": "merge",
+                "+file_format": "delta"
+            }
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "delete_insert_incremental_predicates.sql": models__spark_incremental_predicates_sql
+        }