Fix Snapshot behavior in new_record mode for records already marked as deleted (reopened) (#1174)

Co-authored-by: Colin Rogers <[email protected]>

Author: lpillmann

dbt-adapters/.changes/unreleased/Fixes-20250626-112659.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix Snapshot to avoid inserting redundant entries for records already marked as deleted
+time: 2025-06-26T11:26:59.01969873+02:00
+custom:
+    Author: lpillmann
+    Issue: "654"

dbt-adapters/src/dbt/include/global_project/macros/materializations/snapshots/helpers.sql

@@ -153,6 +153,14 @@
         left join deletes_source_data as source_data
             on {{ unique_key_join_on(strategy.unique_key, "snapshotted_data", "source_data") }}
             where {{ unique_key_is_null(strategy.unique_key, "source_data") }}
+
+            {%- if strategy.hard_deletes == 'new_record' %}
+            and not (
+                --avoid updating the record's valid_to if the latest entry is marked as deleted
+                snapshotted_data.{{ columns.dbt_is_deleted }} = 'True'
+                and snapshotted_data.{{ columns.dbt_valid_to }} is null
+            )
+            {%- endif %}
     )
     {%- endif %}
 
@@ -182,6 +190,11 @@
         left join deletes_source_data as source_data
             on {{ unique_key_join_on(strategy.unique_key, "snapshotted_data", "source_data") }}
         where {{ unique_key_is_null(strategy.unique_key, "source_data") }}
+        and not (
+            --avoid inserting a new record if the latest one is marked as deleted
+            snapshotted_data.{{ columns.dbt_is_deleted }} = 'True'
+            and snapshotted_data.{{ columns.dbt_valid_to }} is null
+            )
 
     )
     {%- endif %}

dbt-tests-adapter/.changes/unreleased/Under the Hood-20250627-092211.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Improve test coverage for snapshots with new record mode
+time: 2025-06-27T09:22:11.280507585+02:00
+custom:
+    Author: lpillmann
+    Issue: "654"

dbt-tests-adapter/src/dbt/tests/adapter/simple_snapshot/new_record_timestamp_mode.py

@@ -248,7 +248,7 @@ def test_snapshot_new_record_mode(
 
         check_relations_equal(project.adapter, ["snapshot_actual", "snapshot_expected"])
 
-        project.run_sql(_delete_sql)
+        project.run_sql(delete_sql)
 
         results = run_dbt(["snapshot"])
         assert len(results) == 1
@@ -282,8 +282,28 @@ def test_snapshot_new_record_mode(
         )
         assert check_result[0][scd_id] != check_result[1][scd_id]
 
+        # run snapshot with the same source data; neither insert nor update should happen
+        run_dbt(["snapshot"])
+        assert len(results) == 1
+        check_result = project.run_sql(_delete_check_sql, fetch="all")
+        assert len(check_result) == 2
+
+        # insert the record back and run the snapshot again; update and insert expected
         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
+
+        # delete it once again and run the snapshot; update and insert expected
+        project.run_sql(delete_sql)
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+        check_result = project.run_sql(_delete_check_sql, fetch="all")
+        assert len(check_result) == 4
+
+        # run snapshot with the same source data; neither insert nor update should happen
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+        check_result = project.run_sql(_delete_check_sql, fetch="all")
+        assert len(check_result) == 4