Support new column in source table with hard_deletes = new_record (#1201)

Author: colin-rogers-dbt

dbt-adapters/.changes/unreleased/Fixes-20250715-125933.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: 'Update dbt snapshot staging table creation to handle new source columns when
+  using hard_deletes == ''new_record'' '
+time: 2025-07-15T12:59:33.467839-07:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "852"

dbt-adapters/src/dbt/include/global_project/macros/adapters/columns.sql

@@ -16,6 +16,13 @@
   {{ return(columns) }}
 {% endmacro %}
 
+{%- macro get_list_of_column_names(columns) -%}
+  {% set col_names = [] %}
+  {% for col in columns %}
+    {% do col_names.append(col.name) %}
+  {% endfor %}
+  {{ return(col_names) }}
+{% endmacro %}
 
 {% macro get_empty_subquery_sql(select_sql, select_sql_header=none) -%}
   {{ return(adapter.dispatch('get_empty_subquery_sql', 'dbt')(select_sql, select_sql_header)) }}

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

@@ -165,14 +165,23 @@
     {%- endif %}
 
     {%- if strategy.hard_deletes == 'new_record' %}
+        {% set snapshotted_cols = get_list_of_column_names(get_columns_in_relation(target_relation)) %}
         {% set source_sql_cols = get_column_schema_from_query(source_sql) %}
     ,
     deletion_records as (
 
         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) }},
+            {%- else -%}
+            NULL as {{ adapter.quote(col.column) }},
+            {%- endif -%}
             {% endfor -%}
             {%- if strategy.unique_key | is_list -%}
                 {%- for key in strategy.unique_key -%}

dbt-snowflake/tests/functional/adapter/test_ephemeral_snapshot_hard_deletes.py

@@ -0,0 +1,7 @@
+from dbt.tests.adapter.simple_snapshot.test_ephemeral_snapshot_hard_deletes import (
+    BaseSnapshotEphemeralHardDeletes,
+)
+
+
+class TestSnapshotEphemeralHardDeletes(BaseSnapshotEphemeralHardDeletes):
+    pass

dbt-tests-adapter/.changes/unreleased/Features-20250715-131036.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: 'Add BaseSnapshotEphemeralHardDeletes test '
+time: 2025-07-15T13:10:36.059488-07:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "1201"

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

@@ -0,0 +1,113 @@
+import pytest
+
+from dbt.tests.util import run_dbt
+
+
+# Source table creation statement
+_source_create_sql = """
+create table {database}.{schema}.src_customers (
+    id INTEGER,
+    first_name VARCHAR(50),
+    last_name VARCHAR(50),
+    email VARCHAR(50),
+    updated_at TIMESTAMP
+);
+"""
+
+# Initial data for source table
+_source_insert_sql = """
+insert into {database}.{schema}.src_customers (id, first_name, last_name, email, updated_at) values
+(1, 'John', 'Doe', '[email protected]', '2023-01-01 10:00:00'),
+(2, 'Jane', 'Smith', '[email protected]', '2023-01-02 11:00:00'),
+(3, 'Bob', 'Johnson', '[email protected]', '2023-01-03 12:00:00');
+"""
+
+# SQL to add a dummy column to source table (simulating schema change)
+_source_alter_sql = """
+alter table {database}.{schema}.src_customers add column dummy_column VARCHAR(50) default 'dummy_value';
+"""
+
+# Sources YAML configuration
+_sources_yml = """
+version: 2
+
+sources:
+  - name: test_source
+    schema: "{{ target.schema }}"
+    tables:
+      - name: src_customers
+"""
+
+# Ephemeral model that references the source
+_ephemeral_customers_sql = """
+{{ config(materialized='ephemeral') }}
+
+select * from {{ source('test_source', 'src_customers') }}
+"""
+
+# Snapshots YAML configuration with hard_deletes: new_record
+_snapshots_yml = """
+snapshots:
+  - name: snapshot_customers
+    relation: ref('ephemeral_customers')
+    config:
+      unique_key: id
+      strategy: check
+      check_cols: all
+      hard_deletes: new_record
+"""
+
+# Test model to query the snapshot (for verification)
+_ref_snapshot_sql = """
+select * from {{ ref('snapshot_customers') }}
+"""
+
+
+class BaseSnapshotEphemeralHardDeletes:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "_sources.yml": _sources_yml,
+            "ephemeral_customers.sql": _ephemeral_customers_sql,
+            "snapshots.yml": _snapshots_yml,
+            "ref_snapshot.sql": _ref_snapshot_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def source_create_sql(self):
+        return _source_create_sql
+
+    @pytest.fixture(scope="class")
+    def source_insert_sql(self):
+        return _source_insert_sql
+
+    @pytest.fixture(scope="class")
+    def source_alter_sql(self):
+        return _source_alter_sql
+
+    def test_ephemeral_snapshot_hard_deletes(
+        self, project, source_create_sql, source_insert_sql, source_alter_sql
+    ):
+
+        project.run_sql(
+            source_create_sql.format(database=project.database, schema=project.test_schema)
+        )
+        project.run_sql(
+            source_insert_sql.format(database=project.database, schema=project.test_schema)
+        )
+
+        results = run_dbt(["snapshot"])
+        assert results is not None
+        assert len(results) == 1  # type: ignore
+
+        snapshot_result = project.run_sql(
+            "select count(*) as row_count from snapshot_customers", fetch="one"
+        )
+        assert snapshot_result[0] == 3  # Should have 3 rows from initial data
+
+        project.run_sql(
+            source_alter_sql.format(database=project.database, schema=project.test_schema)
+        )
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1  # type: ignore