Check for dbt_valid_to_current config for deleted records to avoid duplication (#1241)

Author: jeremyyeo

dbt-adapters/.changes/unreleased/Fixes-20250903-152102.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix duplicates when hard_deletes new_record is used with custom dbt_valid_to_current
+time: 2025-09-03T15:21:02.112323+12:00
+custom:
+    Author: jeremyyeo
+    Issue: "1240"

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

@@ -158,7 +158,12 @@
             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
+                and
+                {% if config.get('dbt_valid_to_current') -%}
+                    snapshotted_data.{{ columns.dbt_valid_to }} = {{ config.get('dbt_valid_to_current') }}
+                {%- else -%}
+                    snapshotted_data.{{ columns.dbt_valid_to }} is null
+                {%- endif %}
             )
             {%- endif %}
     )
@@ -172,10 +177,10 @@
 
         select
             'insert' as dbt_change_type,
-            {#
+            {#/*
                 If a column has been added to the source it won't yet exist in the
                 snapshotted table so we insert a null value as a placeholder for the column.
-             #}
+             */#}
             {%- for col in source_sql_cols -%}
             {%- if col.name in snapshotted_cols -%}
             snapshotted_data.{{ adapter.quote(col.column) }},
@@ -202,7 +207,12 @@
         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
+            and
+            {% if config.get('dbt_valid_to_current') -%}
+                snapshotted_data.{{ columns.dbt_valid_to }} = {{ config.get('dbt_valid_to_current') }}
+            {%- else -%}
+                snapshotted_data.{{ columns.dbt_valid_to }} is null
+            {%- endif %}
             )
 
     )

dbt-postgres/tests/functional/adapter/test_simple_snapshot.py

@@ -2,6 +2,9 @@
 from dbt.tests.adapter.simple_snapshot.new_record_timestamp_mode import (
     BaseSnapshotNewRecordTimestampMode,
 )
+from dbt.tests.adapter.simple_snapshot.new_record_dbt_valid_to_current import (
+    BaseSnapshotNewRecordDbtValidToCurrent,
+)
 from dbt.tests.adapter.simple_snapshot.test_snapshot import (
     BaseSimpleSnapshot,
     BaseSnapshotCheck,
@@ -22,3 +25,7 @@ class TestSnapshotNewRecordTimestampMode(BaseSnapshotNewRecordTimestampMode):
 
 class TestSnapshotNewRecordCheckMode(BaseSnapshotNewRecordCheckMode):
     pass
+
+
+class TestSnapshotNewRecordDbtValidToCurrent(BaseSnapshotNewRecordDbtValidToCurrent):
+    pass

dbt-tests-adapter/.changes/unreleased/Fixes-20250903-152930.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix duplicates when hard_deletes new_record is used with custom dbt_valid_to_current
+time: 2025-09-03T15:29:30.515546+12:00
+custom:
+    Author: jeremyyeo
+    Issue: "1240"

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

@@ -0,0 +1,74 @@
+import pytest
+
+from dbt.tests.util import run_dbt
+
+_seed_new_record_mode_statements = [
+    "create table {database}.{schema}.seed (id INTEGER, first_name VARCHAR(50));",
+    "insert into {database}.{schema}.seed (id, first_name) values (1, 'Judith'), (2, 'Arthur');",
+]
+
+_snapshot_actual_sql = """
+{% snapshot snapshot_actual %}
+    select * from {{target.database}}.{{target.schema}}.seed
+{% endsnapshot %}
+"""
+
+_delete_sql = """
+delete from {database}.{schema}.seed where id = 1
+"""
+
+# If the deletion worked correctly, this should return one row (and not more) where dbt_is_deleted is True
+_delete_check_sql = """
+select dbt_scd_id from {schema}.snapshot_actual where id = 1 and dbt_is_deleted = 'True'
+"""
+
+_snapshots_yml = """
+snapshots:
+  - name: snapshot_actual
+    config:
+      unique_key: id
+      strategy: check
+      check_cols: all
+      hard_deletes: new_record
+      dbt_valid_to_current: "date('9999-12-31')"
+"""
+
+
+class BaseSnapshotNewRecordDbtValidToCurrent:
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {"snapshot.sql": _snapshot_actual_sql}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"snapshots.yml": _snapshots_yml}
+
+    @pytest.fixture(scope="class")
+    def seed_new_record_mode_statements(self):
+        return _seed_new_record_mode_statements
+
+    @pytest.fixture(scope="class")
+    def delete_sql(self):
+        return _delete_sql
+
+    def test_snapshot_new_record_mode(
+        self,
+        project,
+        seed_new_record_mode_statements,
+        delete_sql,
+    ):
+        for stmt in seed_new_record_mode_statements:
+            project.run_sql(stmt)
+
+        # Snapshot once to get the initial snapshot
+        run_dbt(["snapshot"])
+
+        # Remove the record from the source data
+        project.run_sql(delete_sql)
+
+        # Snapshot twice in a row checking that the deleted record is not duplicated in the snapshot
+        run_dbt(["snapshot"])
+        run_dbt(["snapshot"])
+
+        check_result = project.run_sql(_delete_check_sql, fetch="all")
+        assert len(check_result) == 1