base incremental predicate impl

Author: sdebruyn

dbt/include/sqlserver/macros/materializations/models/incremental/merge.sql

@@ -5,49 +5,53 @@
      https://getdbt.slack.com/archives/C50NEBJGG/p1636045535056600
   #}
 
-{% macro sqlserver__get_merge_sql(target, source, unique_key, dest_columns, predicates) %}
-  {{ default__get_merge_sql(target, source, unique_key, dest_columns, predicates) }};
+{% macro sqlserver__get_merge_sql(target, source, unique_key, dest_columns, incremental_predicates) %}
+  {{ default__get_merge_sql(target, source, unique_key, dest_columns, incremental_predicates) }};
 {% endmacro %}
 
-{% macro sqlserver__get_delete_insert_merge_sql(target, source, unique_key, dest_columns, incremental_predicates) %}
-      {% if incremental_predicates %}
-          {{ exceptions.raise_not_implemented('incremental_predicates are not implemented in dbt-sqlserver') }}
-      {% endif %}
-
-      {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}
+{% macro sqlserver__get_insert_overwrite_merge_sql(target, source, dest_columns, predicates, include_sql_header) %}
+  {{ default__get_insert_overwrite_merge_sql(target, source, dest_columns, predicates, include_sql_header) }};
+{% endmacro %}
 
-      {% if unique_key %}
-      {% if unique_key is sequence and unique_key is not string %}
-      delete from {{ target }}
-          where exists (
-                SELECT NULL
-                FROM
-                  {{ source }}
-                WHERE
+{% macro sqlserver__get_delete_insert_merge_sql(target, source, unique_key, dest_columns, incremental_predicates) %}
+    {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}
+
+    {% if unique_key %}
+        {% if unique_key is sequence and unique_key is not string %}
+            delete from {{ target }}
+            where exists (
+                select null
+                from {{ source }}
+                where
                 {% for key in unique_key %}
                     {{ source }}.{{ key }} = {{ target }}.{{ key }}
                     {{ "and " if not loop.last }}
                 {% endfor %}
-            );
-      {% else %}
-      delete from {{ target }}
-          where (
-                {{ unique_key }}) in (
-      select ({{ unique_key }})
-      from {{ source }}
-          );
-
-      {% endif %}
-      {% endif %}
 
-      insert into {{ target }} ({{ dest_cols_csv }})
-          (
-      select {{ dest_cols_csv }}
-      from {{ source }}
-          )
-
-{% endmacro %}
-
-{% macro sqlserver__get_insert_overwrite_merge_sql(target, source, dest_columns, predicates, include_sql_header) %}
-  {{ default__get_insert_overwrite_merge_sql(target, source, dest_columns, predicates, include_sql_header) }};
+            )
+            {% if incremental_predicates %}
+                {% for predicate in incremental_predicates %}
+                    and {{ predicate }}
+                {% endfor %}
+            {% endif %};
+        {% else %}
+            delete from {{ target }}
+            where (
+                {{ unique_key }}) in (
+                select ({{ unique_key }})
+                from {{ source }}
+            )
+            {%- if incremental_predicates %}
+                {% for predicate in incremental_predicates %}
+                    and {{ predicate }}
+                {% endfor %}
+            {%- endif -%};
+        {% endif %}
+    {% endif %}
+
+    insert into {{ target }} ({{ dest_cols_csv }})
+    (
+        select {{ dest_cols_csv }}
+        from {{ source }}
+    )
 {% endmacro %}