From 95a9761fb0699888a5f5f1aa11c61590fb60b122 Mon Sep 17 00:00:00 2001
From: Ben Knight <Benjamin.Knight@wvt.nhs.uk>
Date: Wed, 20 May 2026 10:17:48 +0000
Subject: [PATCH] Fix snapshot meta column name overrides on second run (#639).

The existing-target branch was reading `snapshot_meta_column_names` as
the full source of column names, so a partial override (e.g. only
`dbt_valid_to` + `dbt_valid_from`) dropped the other defaults and the
next snapshot run failed with `KeyError: 'dbt_scd_id'`. Start from the
defaults dict and merge in each provided override key.
---
 .../materializations/snapshots/snapshot.sql   | 10 ++++-
 .../adapter/dbt/test_snapshot_configs.py      | 38 +++++++++++++++++++
 2 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/dbt/include/sqlserver/macros/materializations/snapshots/snapshot.sql b/dbt/include/sqlserver/macros/materializations/snapshots/snapshot.sql
index 237384534..2f2eaa9ad 100644
--- a/dbt/include/sqlserver/macros/materializations/snapshots/snapshot.sql
+++ b/dbt/include/sqlserver/macros/materializations/snapshots/snapshot.sql
@@ -52,7 +52,15 @@
 
   {% else %}
 
-      {% set columns = config.get("snapshot_meta_column_names") or get_snapshot_table_column_names() %}
+      {% set columns = get_snapshot_table_column_names() %}
+      {% set meta = config.get("snapshot_meta_column_names") %}
+      {% if meta %}
+          {% if meta.dbt_valid_from %}{% do columns.update({"dbt_valid_from": meta.dbt_valid_from}) %}{% endif %}
+          {% if meta.dbt_valid_to %}{% do columns.update({"dbt_valid_to": meta.dbt_valid_to}) %}{% endif %}
+          {% if meta.dbt_scd_id %}{% do columns.update({"dbt_scd_id": meta.dbt_scd_id}) %}{% endif %}
+          {% if meta.dbt_updated_at %}{% do columns.update({"dbt_updated_at": meta.dbt_updated_at}) %}{% endif %}
+          {% if meta.dbt_is_deleted %}{% do columns.update({"dbt_is_deleted": meta.dbt_is_deleted}) %}{% endif %}
+      {% endif %}
       {{ adapter.valid_snapshot_target(target_relation, columns) }}
       {% set build_or_select_sql = snapshot_staging_table(strategy, temp_snapshot_relation, target_relation) %}
       {% set staging_table = build_snapshot_staging_table(strategy, temp_snapshot_relation, target_relation) %}
diff --git a/tests/functional/adapter/dbt/test_snapshot_configs.py b/tests/functional/adapter/dbt/test_snapshot_configs.py
index 4aaa1c75c..5cf39bb50 100644
--- a/tests/functional/adapter/dbt/test_snapshot_configs.py
+++ b/tests/functional/adapter/dbt/test_snapshot_configs.py
@@ -779,3 +779,41 @@ def test_multi_column_unique_key(self, project):
 
 class TestBaseSnapshotMultiUniqueKey(BaseSnapshotMultiUniqueKey):
     pass
+
+
+snapshot_partial_column_names_yml = """
+snapshots:
+  - name: snapshot_actual
+    config:
+      strategy: timestamp
+      updated_at: updated_at
+      snapshot_meta_column_names:
+          dbt_valid_to: EffectiveEndDate
+          dbt_valid_from: EffectiveStartDate
+"""
+
+
+class TestSnapshotPartialColumnNames:
+    """Regression test for #639: partial snapshot_meta_column_names fails on second run."""
+
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {"snapshot.sql": snapshot_actual_sql}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "snapshots.yml": snapshot_partial_column_names_yml,
+            "ref_snapshot.sql": ref_snapshot_sql,
+        }
+
+    def test_partial_column_names_second_run(self, project):
+        project.run_sql(create_seed_sql)
+        project.run_sql(seed_insert_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+
+        # Second run should succeed — this is where #639 failed
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1