Delete rows with matching unique key

CHANGELOG.md

@@ -1,3 +1,7 @@
+# dbt-utils v0.9.0
+## Features
+- Add `unique_key` configuration option to `insert_by_period` materialization ([#478](https://github.com/dbt-labs/dbt-utils/pull/478))
+
 # dbt-utils v0.8.0
 ## 🚨 Breaking changes
 - dbt ONE POINT OH is here! This version of dbt-utils requires _any_ version (minor and patch) of v1, which means far less need for compatibility releases in the future. 

README.md

@@ -1078,7 +1078,8 @@ Progress is logged in the command line for easy monitoring.
     period = "day",
     timestamp_field = "created_at",
     start_date = "2018-01-01",
-    stop_date = "2018-06-01")
+    stop_date = "2018-06-01",
+    unique_key = "id")
 }}
 
 with events as (
@@ -1098,11 +1099,11 @@ with events as (
 * `timestamp_field`: the column name of the timestamp field that will be used to break the model into smaller queries
 * `start_date`: literal date or timestamp - generally choose a date that is earlier than the start of your data
 * `stop_date`: literal date or timestamp (default=current_timestamp)
+* `unique_key`: optional key to use to deduplicate records, this is the same as `unique_key` in [incremental models](https://docs.getdbt.com/docs/building-a-dbt-project/building-models/configuring-incremental-models#defining-a-uniqueness-constraint-optional)
 
 **Caveats:**
 * This materialization is compatible with dbt 0.10.1.
 * This materialization has been written for Redshift.
-* This materialization can only be used for a model where records are not expected to change after they are created.
 * Any model post-hooks that use `{{ this }}` will fail using this materialization. For example:
 ```yaml
 models:

macros/materializations/insert_by_period_materialization.sql

@@ -55,6 +55,7 @@
   {%- set start_date = config.require('start_date') -%}
   {%- set stop_date = config.get('stop_date') or '' -%}}
   {%- set period = config.get('period') or 'week' -%}
+  {%- set unique_key = config.get('unique_key') -%}
 
   {%- if sql.find('__PERIOD_FILTER__') == -1 -%}
     {%- set error_message -%}
@@ -143,6 +144,13 @@
                                          to_relation=target_relation)}}
     {%- set name = 'main-' ~ i -%}
     {% call statement(name, fetch_result=True) -%}
+      {%- if unique_key is not none %}
+      delete from {{ target_relation }}
+      where ({{ unique_key }}) in (
+        select ({{ unique_key }})
+        from {{ tmp_relation.include(schema=False) }}
+      );
+      {%- endif %}
       insert into {{target_relation}} ({{target_cols_csv}})
       (
           select