Repository: dbt-labs/dbt-core
Pull request: Include hard-deletes when making snapshot
Continues the work of the closed PR #2355, which resolves #249. 

### Description

The idea is to select ids from the snapshotted table which no longer exist in the source table, and update these to set `dbt_valid_to` to the current timestamp. Marking them as hard deleted. 

As of now, I didn't write any integration tests yet as I'm having difficulties running them locally. Any pointers on how these work would be nice. Running integration test (postgres) locally currently fails, but I'm unsure why. 

```
scheduling tests via LoadScheduling

test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestSimpleSnapshotFiles::test__postgres_ref_snapshot 
test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestSimpleSnapshotFiles::test__postgres__simple_snapshot 
test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestCustomSnapshotFiles::test__postgres__simple_custom_snapshot 
test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestSimpleColumnSnapshotFiles::test_postgres_renamed_source 
[gw1] PASSED test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestSimpleSnapshotFiles::test__postgres_ref_snapshot 
test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestNamespacedCustomSnapshotFiles::test__postgres__simple_custom_snapshot_namespaced 
[gw2] FAILED test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestSimpleColumnSnapshotFiles::test_postgres_renamed_source 

============================================================================================ FAILURES =============================================================================================
___________________________________________________________________ TestSimpleColumnSnapshotFiles.test_postgres_renamed_source ____________________________________________________________________
[gw2] linux -- Python 3.6.9 /usr/app/.tox/integration-postgres-py36/bin/python

self = <test_simple_snapshot.TestSimpleColumnSnapshotFiles testMethod=test_postgres_renamed_source>

    @use_profile('postgres')
    def test_postgres_renamed_source(self):
>       self._run_snapshot_test()

test/integration/004_simple_snapshot_test/test_simple_snapshot.py:158: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
test/integration/004_simple_snapshot_test/test_simple_snapshot.py:135: in _run_snapshot_test
    self.run_dbt(['snapshot', '--vars', '{seed_name: seed_newcol}'])
test/integration/base.py:578: in run_dbt
    "dbt exit state did not match expected")
E   AssertionError: False != True : dbt exit state did not match expected
-------------------------------------------------------------------------------------- Captured logbook call --------------------------------------------------------------------------------------
[DEBUG] dbt: Acquiring new postgres connection "__test".
[DEBUG] dbt: Acquiring new postgres connection "__test".
[DEBUG] dbt: test connection "__test" executing: DROP SCHEMA IF EXISTS "test15998113637441561482_simple_snapshot_004" CASCADE
[DEBUG] dbt: Opening a new connection, currently in state init
[DEBUG] dbt: On __test: Close
[DEBUG] dbt: Acquiring new postgres connection "__test".
[DEBUG] dbt: Acquiring new postgres connection "__test".
[DEBUG] dbt: test connection "__test" executing: CREATE SCHEMA "test15998113637441561482_simple_snapshot_004"
[DEBUG] dbt: Opening a new connection, currently in state closed
[DEBUG] dbt: On __test: Close
[INFO] dbt: Invoking dbt with ['--strict', '--test-new-parser', 'seed', '--profiles-dir', '/tmp/dbt-int-test-1dn56bze', '--log-cache-events']
[INFO] dbt: Invoking dbt with ['--strict', '--test-new-parser', 'snapshot', '--profiles-dir', '/tmp/dbt-int-test-1dn56bze', '--log-cache-events']
[DEBUG] dbt: Acquiring new postgres connection "__test".
[DEBUG] dbt: test connection "__test" executing: select * from dbt.test15998113637441561482_simple_snapshot_004.my_snapshot
[DEBUG] dbt: Opening a new connection, currently in state init
[DEBUG] dbt: On __test: Close
[INFO] dbt: Invoking dbt with ['--strict', '--test-new-parser', 'snapshot', '--vars', '{seed_name: seed_newcol}', '--profiles-dir', '/tmp/dbt-int-test-1dn56bze', '--log-cache-events']
[DEBUG] dbt: Acquiring new postgres connection "__test".
[DEBUG] dbt: Acquiring new postgres connection "__test".
[DEBUG] dbt: test connection "__test" executing: DROP SCHEMA IF EXISTS "test15998113637441561482_simple_snapshot_004" CASCADE
[DEBUG] dbt: Opening a new connection, currently in state closed
[DEBUG] dbt: On __test: Close
[DEBUG] dbt: Connection '__test' was properly closed.
===================================================================================== slowest test durations ======================================================================================
9.03s call     test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestSimpleColumnSnapshotFiles::test_postgres_renamed_source
6.44s call     test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestSimpleSnapshotFiles::test__postgres_ref_snapshot

(0.00 durations hidden.  Use -vv to show these durations.)
===================================================================================== short test summary info =====================================================================================
FAILED test/integration/004_simple_snapshot_test/test_simple_snapshot.py::TestSimpleColumnSnapshotFiles::test_postgres_renamed_source - AssertionError: False != True : dbt exit state did not m...
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! xdist.dsession.Interrupted: stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
============================================================================= 1 failed, 1 passed in 60.78s (0:01:00) ==============================================================================
ERROR: InvocationError for command /bin/bash -c '/usr/app/.tox/integration-postgres-py36/bin/python -m pytest --durations 0 -v -m profile_postgres -s -x -m profile_postgres test/integration/004_simple_snapshot_test/test_simple_snapshot.py -n4 test/integration/*' (exited with code 2)
_____________________________________________________________________________________________ summary _____________________________________________________________________________________________
ERROR:   integration-postgres-py36: commands failed
```

### Checklist
 - [x] I have signed the [CLA](https://docs.getdbt.com/docs/contributor-license-agreements)
 - [x] I have run this code in development and it appears to resolve the stated issue
 - [x] This PR includes tests, or tests are not required/relevant for this PR
 - [x] I have updated the `CHANGELOG.md` and added information about my change to the "dbt next" section.


Problem statement:
snapshots should handle hard deletes
Currently, `dbt archive` stores a series of state changes for records in covered tables, but it does not store a state change for deleted records. Hard deletes in an OLTP database will likely be one of the primary use cases for archival, and we therefore want to make sure that we're capturing record deletion as a state change.

Hints from issue discussion:
this might need to be a completely separate run through the table, which would eat up additional resources. if that turns out to be true, we should allow people to opt out of this in configuration if they know the source data never performs deletes.
let me have a think about this. definitely a good idea
Just a note that this is a big problem for us, for some of our tables, but the issue isn't likely to be solvable by `dbt archive`. One example is a table with checklist preferences per user, 10 million rows or so, I think. If we use our usual ELT pattern, this would require us to _copy_ that entire table into our warehouse, multiple times per day, which is prohibitively expensive to do with Stitch or whatever. 

To get fully reliable data, we'll almost certainly have to solve the problem upstream, either by converting the table to soft deletes, or adding an audit table of deleted rows so we can do an anti-join later.
@jthandy is this something you're still interested in? My instinct is that there are many different ways for hard deletes to be represented in warehouse. Depending on your replication, you might see:
 - a `__deleted_at` column value from your ETL tool
   - you wouldn't need to capture this with archival
 - the record never gets updated again
   - there's no way to know that a hard-delete happened in the source
 - the record is _actually_ deleted from the table in the warehouse
   - we could determine this in `archive`, but I'm unsure how common this paradigm is

Lmk if you have a different idea of how this would work
@drewbanin it wasn't super-clear in my original issue writeup, but I meant hard deletes in the table on the warehouse side. this can occur for potentially two reasons:

1. the table is fully re-replicated from scratch from the source data on a periodic basis and there are hard deletes on the source side, so records disappear from the table within the warehouse.
2. the table is incrementally replicated using binlogs and the deletes are actually propagated. while this isn't something that the ETL providers we use do (instead they would set a deleted flag), it's certainly possible that an ETL pipeline would be designed this way.

I don't think this is an issue with burning criticality, but I have personally run into this a couple of times in client work. if we're touching archive right now we should consider this for inclusion. I won't be offended if we decide not to implement it, but if we don't implement it then we should probably #wontfix it.
A regular archival job coupled with a join back to the source table will be able to detect hard deletes that happen on the warehouse side. Might be worth revisiting this after [archival blocks](https://github.com/fishtown-analytics/dbt/issues/1175) are fully implemented. The new archival materialization will work with pluggable "strategies", and this could very well be one of them
Hi!

Pitching in from a question I made in the #support slack channel https://getdbt.slack.com/archives/C2JRRQDTL/p1563527481312500

I believe this feature would add great value in certain use cases. I been involved in multiple cases where this would solve a common problem (and also make the case for using dbt stronger since the platform would handle this instead of in the Extract processes)

I´ve built many analytic platforms on top of ERP systems (often industry specific niche ERP´s) . A common process nowadays is to extract/replicate the source systems on a daily basis and from those replicas build downstream analytic models (perfect use cases for dbt...). Often there is a need for history tracking, including tracking hard deletes in the source systems - objects (orders, customers, actions) since the source systems more often than not does not track history.

The data volumes I´ve come across in these cases are small to moderate. For tracking hard deletes you´ll have to replicate the entire source table and compare to what is already loaded, so it might get tough with huge amounts of data. However, in these cases, full extracts/replicas are made anyway.

So I´ll definitely vote for further investegation of this issue :) Although agreeing with jthandy above that it might not be a burning issue



Thanks for following up @alangsbo! 

I'm imagining this query taking some sort of time range as an argument. If a given record was last loaded more than X hours/days since the most recent loaded_at date, then dbt can consider it "deleted". Is that approximately what you had in mind?


I was thinking about a simple join back to the land/replicated table from source system. Assuming that table gets dropped and reloaded every day, so that delted rows simply are missing. If the backwards join on specified key columns does not yield any rows. It is deleted.

I guessing you suggestion assumes some kind of persistence of rows in land area, that would also be a good solution where the history would be kept also in the land layer.

So maybe both? :)

Br/A
Ah! Ok - most of the ETL tools we used here will upsert new records into source tables. I didn't think about the drop + reload approach. I think that a join from the snapshot table back to the source table is a good approach here.

Commit message: Update changelog

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2303cfefa2f..8b988ed1b4e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,10 @@
 ### Features
 - dbt will compare configurations using the un-rendered form of the config block in dbt_project.yml ([#2713](https://github.com/fishtown-analytics/dbt/issues/2713), [#2735](https://github.com/fishtown-analytics/dbt/pull/2735))
 - Added state and defer arguments to the RPC client, matching the CLI ([#2678](https://github.com/fishtown-analytics/dbt/issues/2678), [#2736](https://github.com/fishtown-analytics/dbt/pull/2736))
+- Added ability to snapshot hard-deleted records (opt-in with `invalidate_hard_deletes` config option). ([#249](https://github.com/fishtown-analytics/dbt/issues/249), [#2749](https://github.com/fishtown-analytics/dbt/pull/2749))
+
+Contributors:
+- [@joelluijmes](https://github.com/joelluijmes) ([#2749](https://github.com/fishtown-analytics/dbt/pull/2749))
 
 ## dbt 0.18.0 (September 03, 2020)
 
diff --git a/core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql b/core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql
index 04d2112fe51..2ed2343b27b 100644
--- a/core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql
+++ b/core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql
@@ -65,6 +65,16 @@
         from snapshot_query
     ),
 
+    {%- if strategy.invalidate_hard_deletes %}
+    deletes_source_data as (
+
+        select 
+            *,
+            {{ strategy.unique_key }} as dbt_unique_key
+        from snapshot_query
+    ),
+    {% endif %}
+
     insertions as (
 
         select
@@ -99,9 +109,33 @@
         )
     )
 
+    {%- if strategy.invalidate_hard_deletes -%}
+    ,
+
+    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
+    )
+    {%- endif %}
+
     select * from insertions
     union all
     select * from updates
+    {%- if strategy.invalidate_hard_deletes %}
+    union all
+    select * from deletes
+    {%- endif %}
 
 {%- endmacro %}
 
diff --git a/core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql b/core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql
index e4370196db3..74bfa2b32d3 100644
--- a/core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql
+++ b/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 }})
diff --git a/core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql b/core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql
index 94915868061..e5bdaa569b5 100644
--- a/core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql
+++ b/core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql
@@ -66,6 +66,7 @@
 {% macro snapshot_timestamp_strategy(node, snapshotted_rel, current_rel, config, target_exists) %}
     {% set primary_key = config['unique_key'] %}
     {% set updated_at = config['updated_at'] %}
+    {% set invalidate_hard_deletes = config['invalidate_hard_deletes'] %}
 
     {#/*
         The snapshot relation might not have an {{ updated_at }} value if the
@@ -86,7 +87,8 @@
         "unique_key": primary_key,
         "updated_at": updated_at,
         "row_changed": row_changed_expr,
-        "scd_id": scd_id_expr
+        "scd_id": scd_id_expr,
+        "invalidate_hard_deletes": invalidate_hard_deletes
     }) %}
 {% endmacro %}
 
@@ -131,6 +133,8 @@
 {% macro snapshot_check_strategy(node, snapshotted_rel, current_rel, config, target_exists) %}
     {% set check_cols_config = config['check_cols'] %}
     {% set primary_key = config['unique_key'] %}
+    {% set invalidate_hard_deletes = config['invalidate_hard_deletes'] %}
+    
     {% set select_current_time -%}
         select {{ snapshot_get_time() }} as snapshot_start
     {%- endset %}
@@ -173,6 +177,7 @@
         "unique_key": primary_key,
         "updated_at": updated_at,
         "row_changed": row_changed_expr,
-        "scd_id": scd_id_expr
+        "scd_id": scd_id_expr,
+        "invalidate_hard_deletes": invalidate_hard_deletes
     }) %}
 {% endmacro %}
diff --git a/plugins/postgres/dbt/include/postgres/macros/materializations/snapshot_merge.sql b/plugins/postgres/dbt/include/postgres/macros/materializations/snapshot_merge.sql
index 6f8e7cbc6a5..ba016c83a6a 100644
--- a/plugins/postgres/dbt/include/postgres/macros/materializations/snapshot_merge.sql
+++ b/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 %}
diff --git a/test/integration/004_simple_snapshot_test/test-snapshots-pg/snapshot.sql b/test/integration/004_simple_snapshot_test/test-snapshots-pg/snapshot.sql
index 70be6d4ed89..ae5aac087ac 100644
--- a/test/integration/004_simple_snapshot_test/test-snapshots-pg/snapshot.sql
+++ b/test/integration/004_simple_snapshot_test/test-snapshots-pg/snapshot.sql
@@ -9,6 +9,11 @@
             updated_at='updated_at',
         )
     }}
+
+    {% if var('invalidate_hard_deletes', 'false') | as_bool %}
+        {{ config(invalidate_hard_deletes=True) }}
+    {% endif %}
+
     select * from {{target.database}}.{{target.schema}}.seed
 
 {% endsnapshot %}
diff --git a/test/integration/004_simple_snapshot_test/test_simple_snapshot.py b/test/integration/004_simple_snapshot_test/test_simple_snapshot.py
index fc3e1be3840..756632b45de 100644
--- a/test/integration/004_simple_snapshot_test/test_simple_snapshot.py
+++ b/test/integration/004_simple_snapshot_test/test_simple_snapshot.py
@@ -1,4 +1,5 @@
 from test.integration.base import DBTIntegrationTest, use_profile
+from datetime import datetime
 import dbt.exceptions
 
 
@@ -754,3 +755,44 @@ def test__postgres__changing_strategy(self):
 
         results = self.run_dbt(['test'])
         self.assertEqual(len(results), 1)
+
+
+class TestSnapshotHardDelete(BaseSimpleSnapshotTest):
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            "data-paths": ['data'],
+            "snapshot-paths": ['test-snapshots-pg'],
+            'macro-paths': ['macros'],
+        }
+
+    @use_profile('postgres')
+    def test__postgres__snapshot_hard_delete(self):
+        # This test uses the same seed data, containing 20 records of which we hard delete the last 10.
+        # These deleted records set the dbt_valid_to to time the snapshot was ran.
+
+        self.dbt_run_seed_snapshot()
+        self.assert_expected()
+
+        database = self.default_database
+        self.run_sql(
+            'delete from {}.{}.seed where id >= 10;'.format(database, self.unique_schema())
+        )
+
+        begin_snapshot_datetime = datetime.utcnow()
+
+        results = self.run_dbt(['snapshot', '--vars', '{invalidate_hard_deletes: true}'])
+        self.assertEqual(len(results), self.NUM_SNAPSHOT_MODELS)
+
+        results = self.run_sql(
+            'select * from {}.{}.snapshot_actual'.format(database, self.unique_schema()),
+            fetch='all'
+        )
+
+        self.assertEqual(len(results), 20)
+        for result in results[10:]:
+            # result is a tuple, the dbt_valid_to column is the latest
+            self.assertIsInstance(result[-1], datetime)
+            self.assertGreaterEqual(result[-1], begin_snapshot_datetime)
diff --git a/test/rpc/test_snapshots.py b/test/rpc/test_snapshots.py
index 0a313ad0452..44fbfa3031e 100644
--- a/test/rpc/test_snapshots.py
+++ b/test/rpc/test_snapshots.py
@@ -19,7 +19,7 @@
             updated_at='updated_at',
         )
     }}
-    select 1 as id, '2019-10-31 23:59:40' as updated_at
+    select 1 as id, cast('2019-10-31 23:59:40' as timestamp) as updated_at
 
 {% endsnapshot %}
 '''
@@ -36,7 +36,7 @@
             updated_at='updated_at',
         )
     }}
-    select 2 as id, '2019-10-31 23:59:40' as updated_at
+    select 2 as id, cast('2019-10-31 23:59:40' as timestamp) as updated_at
 
 {% endsnapshot %}
 '''
@@ -169,4 +169,4 @@ def test_rpc_snapshot_state(
         results = querier.async_wait_for_result(
             querier.snapshot(state='./state', select=['state:modified']),
         )
-        assert len(results['results']) == 0
+        assert len(results['results']) == 0
\ No newline at end of file


Code review comments:
diff_hunk: @@ -66,6 +66,7 @@
 {% macro snapshot_timestamp_strategy(node, snapshotted_rel, current_rel, config, target_exists) %}
     {% set primary_key = config['unique_key'] %}
     {% set updated_at = config['updated_at'] %}
+    {% set invalidate_hard_deletes = config['invalidate_hard_deletes'] %}
path: core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql
text: @user1:
I think this should be `config.get('invalidate_hard_deletes', false)` - `config` is a dict here, and we don't want to promise that `invalidate_hard_deletes` is present.

I think this works currently because dbt uses a non-strict `undefined`, which is false-ish. But I don't want to risk these `undefined` values escaping the context, they cause annoying json serialization issues down the line.

The other two configs are required, so they don't need to use `.get` and can use the regular `[...]` syntax.