Include hard-deletes when making snapshot

It sets dbt_valid_to to the current snapshot time.

core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql

@@ -65,6 +65,14 @@
         from snapshot_query
     ),
 
+    deletes_source_data as (
+
+        select 
+            *,
+            {{ strategy.unique_key }} as dbt_unique_key
+        from snapshot_query
+    ),
+
     insertions as (
 
         select
@@ -97,11 +105,29 @@
         and (
             {{ strategy.row_changed }}
         )
+    ),
+
+    deletes as (
+
+        select
+            'delete' as dbt_change_type,
+            source_data.*,
+            {{ snapshot_get_time() }} as dbt_valid_from,
+            {{ snapshot_get_time() }} as dbt_updated_at,
+            {{ snapshot_get_time() }} as dbt_valid_to,
+            snapshotted_data.dbt_scd_id
+
+        from snapshotted_data
+        left join deletes_source_data as source_data on snapshotted_data.dbt_unique_key = source_data.dbt_unique_key
+        where snapshotted_data.dbt_valid_to is null
+        and source_data.dbt_unique_key is null
     )
 
     select * from insertions
     union all
     select * from updates
+    union all
+    select * from deletes
 
 {%- endmacro %}
 

core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql

@@ -17,6 +17,12 @@
         then update
         set dbt_valid_to = DBT_INTERNAL_SOURCE.dbt_valid_to
 
+    when matched
+     and DBT_INTERNAL_DEST.dbt_valid_to is null
+     and DBT_INTERNAL_SOURCE.dbt_change_type = 'delete'
+        then update
+        set dbt_valid_to = DBT_INTERNAL_SOURCE.dbt_valid_to
+
     when not matched
      and DBT_INTERNAL_SOURCE.dbt_change_type = 'insert'
         then insert ({{ insert_cols_csv }})

plugins/postgres/dbt/include/postgres/macros/materializations/snapshot_merge.sql

@@ -9,6 +9,13 @@
       and DBT_INTERNAL_SOURCE.dbt_change_type::text = 'update'::text
       and {{ target }}.dbt_valid_to is null;
 
+    update {{ target }}
+    set dbt_valid_to = DBT_INTERNAL_SOURCE.dbt_valid_to
+    from {{ source }} as DBT_INTERNAL_SOURCE
+    where DBT_INTERNAL_SOURCE.dbt_scd_id::text = {{ target }}.dbt_scd_id::text
+      and DBT_INTERNAL_SOURCE.dbt_change_type::text = 'delete'::text
+      and {{ target }}.dbt_valid_to is null;
+
     insert into {{ target }} ({{ insert_cols_csv }})
     select {% for column in insert_cols -%}
         DBT_INTERNAL_SOURCE.{{ column }} {%- if not loop.last %}, {%- endif %}