Merge pull request #670 from Benjamin-Knight/fix/#639-snapshot-partial-column-names

Fix snapshot meta column name overrides on second run (#639).

Author: axellpadilla

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) %}

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