Fix issue with check column snapshots in new_record mode. (#1068)

peterallenwebb · web-flow · commit d8ebe134ce0e · 2025-05-13T21:11:58.000Z
diff --git a/dbt-adapters/.changes/unreleased/Fixes-20250506-150345.yaml b/dbt-adapters/.changes/unreleased/Fixes-20250506-150345.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix an issue with snapshots in new_record mode when check columns are used.
+time: 2025-05-06T15:03:45.057076-04:00
+custom:
+    Author: peterallenwebb
+    Issue: "1068"
diff --git a/dbt-adapters/src/dbt/include/global_project/macros/materializations/snapshots/helpers.sql b/dbt-adapters/src/dbt/include/global_project/macros/materializations/snapshots/helpers.sql
@@ -109,7 +109,9 @@
         left outer join snapshotted_data
             on {{ unique_key_join_on(strategy.unique_key, "snapshotted_data", "source_data") }}
             where {{ unique_key_is_null(strategy.unique_key, "snapshotted_data") }}
-            or ({{ unique_key_is_not_null(strategy.unique_key, "snapshotted_data") }} and ({{ strategy.row_changed }})
+            or ({{ unique_key_is_not_null(strategy.unique_key, "snapshotted_data") }} and (
+               {{ strategy.row_changed }} {%- if strategy.hard_deletes == 'new_record' -%} or snapshotted_data.{{ columns.dbt_is_deleted }} = 'True' {% endif %}
+            )
 
         )
 
@@ -129,7 +131,7 @@
         join snapshotted_data
             on {{ unique_key_join_on(strategy.unique_key, "snapshotted_data", "source_data") }}
         where (
-            {{ strategy.row_changed }}
+            {{ strategy.row_changed }}  {%- if strategy.hard_deletes == 'new_record' -%} or snapshotted_data.{{ columns.dbt_is_deleted }} = 'True' {% endif %}
         )
     )
 
diff --git a/dbt-postgres/.changes/unreleased/Under the Hood-20250506-162028.yaml b/dbt-postgres/.changes/unreleased/Under the Hood-20250506-162028.yaml
@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Expand snapshot testing.
+time: 2025-05-06T16:20:28.912537-04:00
+custom:
+    Author: peterallenwebb
+    Issue: "1068"
diff --git a/dbt-postgres/tests/functional/adapter/test_simple_snapshot.py b/dbt-postgres/tests/functional/adapter/test_simple_snapshot.py
@@ -1,3 +1,7 @@
+from dbt.tests.adapter.simple_snapshot.new_record_check_mode import BaseSnapshotNewRecordCheckMode
+from dbt.tests.adapter.simple_snapshot.new_record_timestamp_mode import (
+    BaseSnapshotNewRecordTimestampMode,
+)
 from dbt.tests.adapter.simple_snapshot.test_snapshot import (
     BaseSimpleSnapshot,
     BaseSnapshotCheck,
@@ -10,3 +14,11 @@ class TestSnapshot(BaseSimpleSnapshot):
 
 class TestSnapshotCheck(BaseSnapshotCheck):
     pass
+
+
+class TestSnapshotNewRecordTimestampMode(BaseSnapshotNewRecordTimestampMode):
+    pass
+
+
+class TestSnapshotNewRecordCheckMode(BaseSnapshotNewRecordCheckMode):
+    pass
diff --git a/dbt-redshift/.changes/unreleased/Under the Hood-20250506-162125.yaml b/dbt-redshift/.changes/unreleased/Under the Hood-20250506-162125.yaml
@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Expand snapshot testing.
+time: 2025-05-06T16:21:25.26162-04:00
+custom:
+    Author: peterallenwebb
+    Issue: "1068"
diff --git a/dbt-redshift/tests/functional/adapter/test_simple_snapshot.py b/dbt-redshift/tests/functional/adapter/test_simple_snapshot.py
@@ -1,3 +1,7 @@
+from dbt.tests.adapter.simple_snapshot.new_record_check_mode import BaseSnapshotNewRecordCheckMode
+from dbt.tests.adapter.simple_snapshot.new_record_timestamp_mode import (
+    BaseSnapshotNewRecordTimestampMode,
+)
 from dbt.tests.adapter.simple_snapshot.test_snapshot import BaseSnapshotCheck, BaseSimpleSnapshot
 
 
@@ -7,3 +11,11 @@ class TestSnapshot(BaseSimpleSnapshot):
 
 class TestSnapshotCheck(BaseSnapshotCheck):
     pass
+
+
+class TestSnapshotNewRecordTimestampMode(BaseSnapshotNewRecordTimestampMode):
+    pass
+
+
+class TestSnapshotNewRecordCheckMode(BaseSnapshotNewRecordCheckMode):
+    pass
diff --git a/dbt-snowflake/.changes/unreleased/Under the Hood-20250506-162055.yaml b/dbt-snowflake/.changes/unreleased/Under the Hood-20250506-162055.yaml
@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Expand snapshot testing.
+time: 2025-05-06T16:20:55.476606-04:00
+custom:
+    Author: peterallenwebb
+    Issue: "1068"
diff --git a/dbt-snowflake/tests/functional/adapter/test_simple_snapshot.py b/dbt-snowflake/tests/functional/adapter/test_simple_snapshot.py
@@ -1,4 +1,8 @@
 from dbt.tests.adapter.simple_snapshot.test_snapshot import BaseSnapshotCheck, BaseSimpleSnapshot
+from dbt.tests.adapter.simple_snapshot.new_record_timestamp_mode import (
+    BaseSnapshotNewRecordTimestampMode,
+)
+from dbt.tests.adapter.simple_snapshot.new_record_check_mode import BaseSnapshotNewRecordCheckMode
 
 
 class TestSnapshot(BaseSimpleSnapshot):
@@ -7,3 +11,11 @@ class TestSnapshot(BaseSimpleSnapshot):
 
 class TestSnapshotCheck(BaseSnapshotCheck):
     pass
+
+
+class TestSnapshotNewRecordTimestampMode(BaseSnapshotNewRecordTimestampMode):
+    pass
+
+
+class TestSnapshotNewRecordCheckMode(BaseSnapshotNewRecordCheckMode):
+    pass
diff --git a/dbt-tests-adapter/.changes/unreleased/Under the Hood-20250506-163044.yaml b/dbt-tests-adapter/.changes/unreleased/Under the Hood-20250506-163044.yaml
@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Add new tests for snapshots in new_record mode in combination with check columns.
+time: 2025-05-06T16:30:44.72145-04:00
+custom:
+    Author: peterallenwebb
+    Issue: "1068"
diff --git a/dbt-tests-adapter/src/dbt/tests/adapter/simple_snapshot/new_record_check_mode.py b/dbt-tests-adapter/src/dbt/tests/adapter/simple_snapshot/new_record_check_mode.py
@@ -0,0 +1,27 @@
+import pytest
+
+from dbt.tests.adapter.simple_snapshot.new_record_timestamp_mode import (
+    BaseSnapshotNewRecordTimestampMode,
+)
+
+_snapshots_yml = """
+snapshots:
+  - name: snapshot_actual
+    config:
+      strategy: check
+      check_cols: all
+      hard_deletes: new_record
+"""
+
+_ref_snapshot_sql = """
+select * from {{ ref('snapshot_actual') }}
+"""
+
+
+class BaseSnapshotNewRecordCheckMode(BaseSnapshotNewRecordTimestampMode):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "snapshots.yml": _snapshots_yml,
+            "ref_snapshot.sql": _ref_snapshot_sql,
+        }
diff --git a/dbt-tests-adapter/src/dbt/tests/adapter/simple_snapshot/new_record_timestamp_mode.py b/dbt-tests-adapter/src/dbt/tests/adapter/simple_snapshot/new_record_timestamp_mode.py
@@ -3,20 +3,16 @@
 from dbt.tests.util import check_relations_equal, run_dbt
 
 # Snapshot source data for the tests in this file
-_seed_new_record_mode = """
-BEGIN
-
-create table {database}.{schema}.seed (
+_seed_new_record_mode_statements = [
+    """create table {database}.{schema}.seed (
     id INTEGER,
     first_name VARCHAR(50),
     last_name VARCHAR(50),
     email VARCHAR(50),
     gender VARCHAR(50),
     ip_address VARCHAR(20),
-    updated_at TIMESTAMP WITHOUT TIME ZONE
-);
-
-create table {database}.{schema}.snapshot_expected (
+    updated_at TIMESTAMP WITHOUT TIME ZONE);""",
+    """create table {database}.{schema}.snapshot_expected (
     id INTEGER,
     first_name VARCHAR(50),
     last_name VARCHAR(50),
@@ -31,13 +27,11 @@
     dbt_scd_id     TEXT,
     dbt_updated_at TIMESTAMP WITHOUT TIME ZONE,
     dbt_is_deleted TEXT
-);
-
-
--- seed inserts
---  use the same email for two users to verify that duplicated check_cols values
---  are handled appropriately
-insert into {database}.{schema}.seed (id, first_name, last_name, email, gender, ip_address, updated_at) values
+    );""",
+    # seed inserts
+    #  use the same email for two users to verify that duplicated check_cols values
+    # are handled appropriately
+    """insert into {database}.{schema}.seed (id, first_name, last_name, email, gender, ip_address, updated_at) values
 (1, 'Judith', 'Kennedy', '(not provided)', 'Female', '54.60.24.128', '2015-12-24 12:19:28'),
 (2, 'Arthur', 'Kelly', '(not provided)', 'Male', '62.56.24.215', '2015-10-28 16:22:15'),
 (3, 'Rachel', 'Moreno', 'rmoreno2@msu.edu', 'Female', '31.222.249.23', '2016-04-05 02:05:30'),
@@ -57,11 +51,9 @@
 (17, 'Gary', 'Bishop', 'gbishopg@plala.or.jp', 'Male', '161.108.182.13', '2016-08-29 19:35:20'),
 (18, 'Anna', 'Riley', 'arileyh@nasa.gov', 'Female', '253.31.108.22', '2015-12-11 04:34:27'),
 (19, 'Sarah', 'Knight', 'sknighti@foxnews.com', 'Female', '222.220.3.177', '2016-09-26 00:49:06'),
-(20, 'Phyllis', 'Fox', null, 'Female', '163.191.232.95', '2016-08-21 10:35:19');
-
-
--- populate snapshot table
-insert into {database}.{schema}.snapshot_expected (
+(20, 'Phyllis', 'Fox', null, 'Female', '163.191.232.95', '2016-08-21 10:35:19');""",
+    # populate snapshot table
+    """insert into {database}.{schema}.snapshot_expected (
     id,
     first_name,
     last_name,
@@ -75,7 +67,6 @@
     dbt_scd_id,
     dbt_is_deleted
 )
-
 select
     id,
     first_name,
@@ -90,10 +81,8 @@
     updated_at as dbt_updated_at,
     md5(id || '-' || first_name || '|' || updated_at::text) as dbt_scd_id,
     'False' as dbt_is_deleted
-from {database}.{schema}.seed;
-
-END;
-"""
+from {database}.{schema}.seed;""",
+]
 
 _snapshot_actual_sql = """
 {% snapshot snapshot_actual %}
@@ -123,23 +112,18 @@
 """
 
 
-_invalidate_sql = """
-BEGIN
-
--- update records 11 - 21. Change email and updated_at field
+_invalidate_sql_statements = [
+    """-- Update records 11 - 21. Change email and updated_at field.
 update {schema}.seed set
     updated_at = updated_at + interval '1 hour',
     email      =  case when id = 20 then 'pfoxj@creativecommons.org' else 'new_' || email end
-where id >= 10 and id <= 20;
-
-
--- invalidate records 11 - 21
+where id >= 10 and id <= 20;""",
+    """-- Update the expected snapshot data to reflect the changes we expect to the snapshot on the next run
 update {schema}.snapshot_expected set
     dbt_valid_to   = updated_at + interval '1 hour'
 where id >= 10 and id <= 20;
-
-END;
-"""
+""",
+]
 
 _update_sql = """
 -- insert v2 of the 11 - 21 records
@@ -187,8 +171,22 @@
 select dbt_valid_to, dbt_scd_id, dbt_is_deleted from {schema}.snapshot_actual where id = 1
 """
 
+# SQL to re-insert the deleted record from the snapshot source data
+_reinsert_sql = """
+insert into {database}.{schema}.seed (id, first_name, last_name, email, gender, ip_address, updated_at) values
+(1, 'Judith', 'Kennedy', '(not provided)', 'Female', '54.60.24.128', '2200-01-01 12:00:00');
+"""
+
+# If the re-insertion worked correctly, this should return three rows:
+#   - one for the original record
+#   - one for the deletion interval
+#   - one for the new record
+_reinsert_check_sql = """
+select dbt_valid_from, dbt_valid_to, dbt_scd_id, dbt_is_deleted from {schema}.snapshot_actual where id = 1;
+"""
+
 
-class SnapshotNewRecordMode:
+class BaseSnapshotNewRecordTimestampMode:
     @pytest.fixture(scope="class")
     def snapshots(self):
         return {"snapshot.sql": _snapshot_actual_sql}
@@ -201,12 +199,12 @@ def models(self):
         }
 
     @pytest.fixture(scope="class")
-    def seed_new_record_mode(self):
-        return _seed_new_record_mode
+    def seed_new_record_mode_statements(self):
+        return _seed_new_record_mode_statements
 
     @pytest.fixture(scope="class")
-    def invalidate_sql(self):
-        return _invalidate_sql
+    def invalidate_sql_statements(self):
+        return _invalidate_sql_statements
 
     @pytest.fixture(scope="class")
     def update_sql(self):
@@ -216,14 +214,33 @@ def update_sql(self):
     def delete_sql(self):
         return _delete_sql
 
+    @pytest.fixture(scope="class")
+    def reinsert_sql(self):
+        return _reinsert_sql
+
+    @pytest.fixture(scope="class")
+    def reinsert_check_sql(self):
+        return _reinsert_check_sql
+
     def test_snapshot_new_record_mode(
-        self, project, seed_new_record_mode, invalidate_sql, update_sql
+        self,
+        project,
+        seed_new_record_mode_statements,
+        invalidate_sql_statements,
+        update_sql,
+        delete_sql,
+        reinsert_sql,
+        reinsert_check_sql,
     ):
-        project.run_sql(seed_new_record_mode)
+        for stmt in seed_new_record_mode_statements:
+            project.run_sql(stmt)
+
         results = run_dbt(["snapshot"])
         assert len(results) == 1
 
-        project.run_sql(invalidate_sql)
+        for stmt in invalidate_sql_statements:
+            project.run_sql(stmt)
+
         project.run_sql(update_sql)
 
         results = run_dbt(["snapshot"])
@@ -264,3 +281,9 @@ def test_snapshot_new_record_mode(
             == 1
         )
         assert check_result[0][scd_id] != check_result[1][scd_id]
+
+        project.run_sql(reinsert_sql)
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+        check_result = project.run_sql(reinsert_check_sql, fetch="all")
+        assert len(check_result) == 3

Original file line number	Diff line number	Diff line change
`@@ -109,7 +109,9 @@`
`109`	`109`	`left outer join snapshotted_data`
`110`	`110`	`on {{ unique_key_join_on(strategy.unique_key, "snapshotted_data", "source_data") }}`
`111`	`111`	`where {{ unique_key_is_null(strategy.unique_key, "snapshotted_data") }}`
`112`		`- or ({{ unique_key_is_not_null(strategy.unique_key, "snapshotted_data") }} and ({{ strategy.row_changed }})`
	`112`	`+ or ({{ unique_key_is_not_null(strategy.unique_key, "snapshotted_data") }} and (`
	`113`	`+ {{ strategy.row_changed }} {%- if strategy.hard_deletes == 'new_record' -%} or snapshotted_data.{{ columns.dbt_is_deleted }} = 'True' {% endif %}`
	`114`	`+ )`
`113`	`115`
`114`	`116`	`)`
`115`	`117`
`@@ -129,7 +131,7 @@`
`129`	`131`	`join snapshotted_data`
`130`	`132`	`on {{ unique_key_join_on(strategy.unique_key, "snapshotted_data", "source_data") }}`
`131`	`133`	`where (`
`132`		`- {{ strategy.row_changed }}`
	`134`	`+ {{ strategy.row_changed }} {%- if strategy.hard_deletes == 'new_record' -%} or snapshotted_data.{{ columns.dbt_is_deleted }} = 'True' {% endif %}`
`133`	`135`	`)`
`134`	`136`	`)`
`135`	`137`