Repository: dbt-labs/dbt-core (language: python)

Issue:
[Feature] Allows snapshots to have a list as a unique key
### Is this your first time submitting a feature request?

- [X] I have read the [expectations for open source contributors](https://docs.getdbt.com/docs/contributing/oss-expectations)
- [X] I have searched the existing issues, and I could not find an existing issue for this feature
- [X] I am requesting a straightforward extension of existing dbt functionality, rather than a Big Idea better suited to a discussion

### Describe the feature

This a clone of https://github.com/dbt-labs/dbt-adapters/issues/181

Right now the 'unique_key' for a snapshot must be a single column or an expression. If a user needs a combination of columns to be unique then they can use the expression to concatenate multiple columns together how they like.

Ideally users would be able to enter a list of columns, just like they can do for incremental models.

### Describe alternatives you've considered

I considered adding additional parameters instead of changing the existing unique_key parameter. I think using the existing parameter is best because it's the most consistent with how incremental tables work.

### Who will this benefit?

Snapshot users who work in Postgres or Redshift can take advantage of performance optimizations in their databases with this change.

There are performance benefits from using multiple columns rather than an expression. In Postgres users would be able to pass in partition keys or indexed columns to speed up their snapshots, without needing to make an additional expression index. In Redshift users can pass in a separate distribution key to also get performance benefits.

These benefits can't be realized unless dbt allows the unique key to natively be a combination of columns, rather than a concatenation/combination expression.

### Are you interested in contributing this feature?

Yes, I have a PR in dbt-adapters and in dbt-core coming shortly

### Anything else?

_No response_


Discussion / hints:
Thanks so much for opening - agree that ideally users would be able to enter a list of columns for `unique_key` for snapshots, just like they can do for incremental models.

Will queue up your PRs for review from our team
@graciegoheen Any chance the PR will be merged soon?

Fix patch:
diff --git a/.changes/unreleased/Features-20241001-134051.yaml b/.changes/unreleased/Features-20241001-134051.yaml
new file mode 100644
index 00000000000..60ada51ece3
--- /dev/null
+++ b/.changes/unreleased/Features-20241001-134051.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: Enable use of multi-column unique key in snapshots
+time: 2024-10-01T13:40:51.297529-04:00
+custom:
+  Author: gshank
+  Issue: "9992"
diff --git a/core/dbt/artifacts/resources/v1/snapshot.py b/core/dbt/artifacts/resources/v1/snapshot.py
index 062b6a62814..1a7b9344ca0 100644
--- a/core/dbt/artifacts/resources/v1/snapshot.py
+++ b/core/dbt/artifacts/resources/v1/snapshot.py
@@ -19,7 +19,7 @@ class SnapshotMetaColumnNames(dbtClassMixin):
 class SnapshotConfig(NodeConfig):
     materialized: str = "snapshot"
     strategy: Optional[str] = None
-    unique_key: Optional[str] = None
+    unique_key: Optional[Union[str, List[str]]] = None
     target_schema: Optional[str] = None
     target_database: Optional[str] = None
     updated_at: Optional[str] = None
diff --git a/core/setup.py b/core/setup.py
index b7a8dabd14e..b787ce8a923 100644
--- a/core/setup.py
+++ b/core/setup.py
@@ -71,7 +71,7 @@
         "dbt-extractor>=0.5.0,<=0.6",
         "dbt-semantic-interfaces>=0.7.3,<0.8",
         # Minor versions for these are expected to be backwards-compatible
-        "dbt-common>=1.9.0,<2.0",
+        "dbt-common>=1.11.0,<2.0",
         "dbt-adapters>=1.7.0,<2.0",
         # ----
         # Expect compatibility with all new versions of these packages, so lower bounds only.
diff --git a/schemas/dbt/manifest/v12.json b/schemas/dbt/manifest/v12.json
index 9cb7f732e62..cebc0970b9d 100644
--- a/schemas/dbt/manifest/v12.json
+++ b/schemas/dbt/manifest/v12.json
@@ -6540,6 +6540,12 @@
                       {
                         "type": "string"
                       },
+                      {
+                        "type": "array",
+                        "items": {
+                          "type": "string"
+                        }
+                      },
                       {
                         "type": "null"
                       }
@@ -16425,6 +16431,12 @@
                             {
                               "type": "string"
                             },
+                            {
+                              "type": "array",
+                              "items": {
+                                "type": "string"
+                              }
+                            },
                             {
                               "type": "null"
                             }


Test patch:
diff --git a/tests/functional/adapter/simple_snapshot/fixtures.py b/tests/functional/adapter/simple_snapshot/fixtures.py
new file mode 100644
index 00000000000..cec28a7d64d
--- /dev/null
+++ b/tests/functional/adapter/simple_snapshot/fixtures.py
@@ -0,0 +1,430 @@
+create_seed_sql = """
+create table {schema}.seed (
+    id INTEGER,
+    first_name VARCHAR(50),
+    last_name VARCHAR(50),
+    email VARCHAR(50),
+    gender VARCHAR(50),
+    ip_address VARCHAR(20),
+    updated_at TIMESTAMP
+);
+"""
+
+create_snapshot_expected_sql = """
+create table {schema}.snapshot_expected (
+    id INTEGER,
+    first_name VARCHAR(50),
+    last_name VARCHAR(50),
+    email VARCHAR(50),
+    gender VARCHAR(50),
+    ip_address VARCHAR(20),
+
+    -- snapshotting fields
+    updated_at TIMESTAMP,
+    test_valid_from TIMESTAMP,
+    test_valid_to   TIMESTAMP,
+    test_scd_id     TEXT,
+    test_updated_at TIMESTAMP
+);
+"""
+
+
+seed_insert_sql = """
+-- seed inserts
+--  use the same email for two users to verify that duplicated check_cols values
+--  are handled appropriately
+insert into {schema}.seed (id, first_name, last_name, email, gender, ip_address, updated_at) values
+(1, 'Judith', 'Kennedy', '(not provided)', 'Female', '54.60.24.128', '2015-12-24 12:19:28'),
+(2, 'Arthur', 'Kelly', '(not provided)', 'Male', '62.56.24.215', '2015-10-28 16:22:15'),
+(3, 'Rachel', 'Moreno', 'rmoreno2@msu.edu', 'Female', '31.222.249.23', '2016-04-05 02:05:30'),
+(4, 'Ralph', 'Turner', 'rturner3@hp.com', 'Male', '157.83.76.114', '2016-08-08 00:06:51'),
+(5, 'Laura', 'Gonzales', 'lgonzales4@howstuffworks.com', 'Female', '30.54.105.168', '2016-09-01 08:25:38'),
+(6, 'Katherine', 'Lopez', 'klopez5@yahoo.co.jp', 'Female', '169.138.46.89', '2016-08-30 18:52:11'),
+(7, 'Jeremy', 'Hamilton', 'jhamilton6@mozilla.org', 'Male', '231.189.13.133', '2016-07-17 02:09:46'),
+(8, 'Heather', 'Rose', 'hrose7@goodreads.com', 'Female', '87.165.201.65', '2015-12-29 22:03:56'),
+(9, 'Gregory', 'Kelly', 'gkelly8@trellian.com', 'Male', '154.209.99.7', '2016-03-24 21:18:16'),
+(10, 'Rachel', 'Lopez', 'rlopez9@themeforest.net', 'Female', '237.165.82.71', '2016-08-20 15:44:49'),
+(11, 'Donna', 'Welch', 'dwelcha@shutterfly.com', 'Female', '103.33.110.138', '2016-02-27 01:41:48'),
+(12, 'Russell', 'Lawrence', 'rlawrenceb@qq.com', 'Male', '189.115.73.4', '2016-06-11 03:07:09'),
+(13, 'Michelle', 'Montgomery', 'mmontgomeryc@scientificamerican.com', 'Female', '243.220.95.82', '2016-06-18 16:27:19'),
+(14, 'Walter', 'Castillo', 'wcastillod@pagesperso-orange.fr', 'Male', '71.159.238.196', '2016-10-06 01:55:44'),
+(15, 'Robin', 'Mills', 'rmillse@vkontakte.ru', 'Female', '172.190.5.50', '2016-10-31 11:41:21'),
+(16, 'Raymond', 'Holmes', 'rholmesf@usgs.gov', 'Male', '148.153.166.95', '2016-10-03 08:16:38'),
+(17, 'Gary', 'Bishop', 'gbishopg@plala.or.jp', 'Male', '161.108.182.13', '2016-08-29 19:35:20'),
+(18, 'Anna', 'Riley', 'arileyh@nasa.gov', 'Female', '253.31.108.22', '2015-12-11 04:34:27'),
+(19, 'Sarah', 'Knight', 'sknighti@foxnews.com', 'Female', '222.220.3.177', '2016-09-26 00:49:06'),
+(20, 'Phyllis', 'Fox', null, 'Female', '163.191.232.95', '2016-08-21 10:35:19');
+"""
+
+
+populate_snapshot_expected_sql = """
+-- populate snapshot table
+insert into {schema}.snapshot_expected (
+    id,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    test_valid_from,
+    test_valid_to,
+    test_updated_at,
+    test_scd_id
+)
+
+select
+    id,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    -- fields added by snapshotting
+    updated_at as test_valid_from,
+    null::timestamp as test_valid_to,
+    updated_at as test_updated_at,
+    md5(id || '-' || first_name || '|' || updated_at::text) as test_scd_id
+from {schema}.seed;
+"""
+
+populate_snapshot_expected_valid_to_current_sql = """
+-- populate snapshot table
+insert into {schema}.snapshot_expected (
+    id,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    test_valid_from,
+    test_valid_to,
+    test_updated_at,
+    test_scd_id
+)
+
+select
+    id,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    -- fields added by snapshotting
+    updated_at as test_valid_from,
+    date('2099-12-31') as test_valid_to,
+    updated_at as test_updated_at,
+    md5(id || '-' || first_name || '|' || updated_at::text) as test_scd_id
+from {schema}.seed;
+"""
+
+snapshot_actual_sql = """
+{% snapshot snapshot_actual %}
+
+    {{
+        config(
+            unique_key='id || ' ~ "'-'" ~ ' || first_name',
+        )
+    }}
+
+    select * from {{target.database}}.{{target.schema}}.seed
+
+{% endsnapshot %}
+"""
+
+snapshots_yml = """
+snapshots:
+  - name: snapshot_actual
+    config:
+      strategy: timestamp
+      updated_at: updated_at
+      snapshot_meta_column_names:
+          dbt_valid_to: test_valid_to
+          dbt_valid_from: test_valid_from
+          dbt_scd_id: test_scd_id
+          dbt_updated_at: test_updated_at
+"""
+
+snapshots_no_column_names_yml = """
+snapshots:
+  - name: snapshot_actual
+    config:
+      strategy: timestamp
+      updated_at: updated_at
+"""
+
+ref_snapshot_sql = """
+select * from {{ ref('snapshot_actual') }}
+"""
+
+
+invalidate_sql = """
+-- update records 11 - 21. Change email and updated_at field
+update {schema}.seed set
+    updated_at = updated_at + interval '1 hour',
+    email      =  case when id = 20 then 'pfoxj@creativecommons.org' else 'new_' || email end
+where id >= 10 and id <= 20;
+
+
+-- invalidate records 11 - 21
+update {schema}.snapshot_expected set
+    test_valid_to   = updated_at + interval '1 hour'
+where id >= 10 and id <= 20;
+
+"""
+
+update_sql = """
+-- insert v2 of the 11 - 21 records
+
+insert into {schema}.snapshot_expected (
+    id,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    test_valid_from,
+    test_valid_to,
+    test_updated_at,
+    test_scd_id
+)
+
+select
+    id,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    -- fields added by snapshotting
+    updated_at as test_valid_from,
+    null::timestamp as test_valid_to,
+    updated_at as test_updated_at,
+    md5(id || '-' || first_name || '|' || updated_at::text) as test_scd_id
+from {schema}.seed
+where id >= 10 and id <= 20;
+"""
+
+# valid_to_current fixtures
+
+snapshots_valid_to_current_yml = """
+snapshots:
+  - name: snapshot_actual
+    config:
+      strategy: timestamp
+      updated_at: updated_at
+      dbt_valid_to_current: "date('2099-12-31')"
+      snapshot_meta_column_names:
+          dbt_valid_to: test_valid_to
+          dbt_valid_from: test_valid_from
+          dbt_scd_id: test_scd_id
+          dbt_updated_at: test_updated_at
+"""
+
+update_with_current_sql = """
+-- insert v2 of the 11 - 21 records
+
+insert into {schema}.snapshot_expected (
+    id,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    test_valid_from,
+    test_valid_to,
+    test_updated_at,
+    test_scd_id
+)
+
+select
+    id,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    -- fields added by snapshotting
+    updated_at as test_valid_from,
+    date('2099-12-31') as test_valid_to,
+    updated_at as test_updated_at,
+    md5(id || '-' || first_name || '|' || updated_at::text) as test_scd_id
+from {schema}.seed
+where id >= 10 and id <= 20;
+"""
+
+
+# multi-key snapshot fixtures
+
+create_multi_key_seed_sql = """
+create table {schema}.seed (
+    id1 INTEGER,
+    id2 INTEGER,
+    first_name VARCHAR(50),
+    last_name VARCHAR(50),
+    email VARCHAR(50),
+    gender VARCHAR(50),
+    ip_address VARCHAR(20),
+    updated_at TIMESTAMP
+);
+"""
+
+
+create_multi_key_snapshot_expected_sql = """
+create table {schema}.snapshot_expected (
+    id1 INTEGER,
+    id2 INTEGER,
+    first_name VARCHAR(50),
+    last_name VARCHAR(50),
+    email VARCHAR(50),
+    gender VARCHAR(50),
+    ip_address VARCHAR(20),
+
+    -- snapshotting fields
+    updated_at TIMESTAMP,
+    test_valid_from TIMESTAMP,
+    test_valid_to   TIMESTAMP,
+    test_scd_id     TEXT,
+    test_updated_at TIMESTAMP
+);
+"""
+
+seed_multi_key_insert_sql = """
+-- seed inserts
+--  use the same email for two users to verify that duplicated check_cols values
+--  are handled appropriately
+insert into {schema}.seed (id1, id2, first_name, last_name, email, gender, ip_address, updated_at) values
+(1, 100,  'Judith', 'Kennedy', '(not provided)', 'Female', '54.60.24.128', '2015-12-24 12:19:28'),
+(2, 200, 'Arthur', 'Kelly', '(not provided)', 'Male', '62.56.24.215', '2015-10-28 16:22:15'),
+(3, 300, 'Rachel', 'Moreno', 'rmoreno2@msu.edu', 'Female', '31.222.249.23', '2016-04-05 02:05:30'),
+(4, 400, 'Ralph', 'Turner', 'rturner3@hp.com', 'Male', '157.83.76.114', '2016-08-08 00:06:51'),
+(5, 500, 'Laura', 'Gonzales', 'lgonzales4@howstuffworks.com', 'Female', '30.54.105.168', '2016-09-01 08:25:38'),
+(6, 600, 'Katherine', 'Lopez', 'klopez5@yahoo.co.jp', 'Female', '169.138.46.89', '2016-08-30 18:52:11'),
+(7, 700, 'Jeremy', 'Hamilton', 'jhamilton6@mozilla.org', 'Male', '231.189.13.133', '2016-07-17 02:09:46'),
+(8, 800, 'Heather', 'Rose', 'hrose7@goodreads.com', 'Female', '87.165.201.65', '2015-12-29 22:03:56'),
+(9, 900, 'Gregory', 'Kelly', 'gkelly8@trellian.com', 'Male', '154.209.99.7', '2016-03-24 21:18:16'),
+(10, 1000, 'Rachel', 'Lopez', 'rlopez9@themeforest.net', 'Female', '237.165.82.71', '2016-08-20 15:44:49'),
+(11, 1100, 'Donna', 'Welch', 'dwelcha@shutterfly.com', 'Female', '103.33.110.138', '2016-02-27 01:41:48'),
+(12, 1200, 'Russell', 'Lawrence', 'rlawrenceb@qq.com', 'Male', '189.115.73.4', '2016-06-11 03:07:09'),
+(13, 1300, 'Michelle', 'Montgomery', 'mmontgomeryc@scientificamerican.com', 'Female', '243.220.95.82', '2016-06-18 16:27:19'),
+(14, 1400, 'Walter', 'Castillo', 'wcastillod@pagesperso-orange.fr', 'Male', '71.159.238.196', '2016-10-06 01:55:44'),
+(15, 1500, 'Robin', 'Mills', 'rmillse@vkontakte.ru', 'Female', '172.190.5.50', '2016-10-31 11:41:21'),
+(16, 1600, 'Raymond', 'Holmes', 'rholmesf@usgs.gov', 'Male', '148.153.166.95', '2016-10-03 08:16:38'),
+(17, 1700, 'Gary', 'Bishop', 'gbishopg@plala.or.jp', 'Male', '161.108.182.13', '2016-08-29 19:35:20'),
+(18, 1800, 'Anna', 'Riley', 'arileyh@nasa.gov', 'Female', '253.31.108.22', '2015-12-11 04:34:27'),
+(19, 1900, 'Sarah', 'Knight', 'sknighti@foxnews.com', 'Female', '222.220.3.177', '2016-09-26 00:49:06'),
+(20, 2000, 'Phyllis', 'Fox', null, 'Female', '163.191.232.95', '2016-08-21 10:35:19');
+"""
+
+populate_multi_key_snapshot_expected_sql = """
+-- populate snapshot table
+insert into {schema}.snapshot_expected (
+    id1,
+    id2,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    test_valid_from,
+    test_valid_to,
+    test_updated_at,
+    test_scd_id
+)
+
+select
+    id1,
+    id2,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    -- fields added by snapshotting
+    updated_at as test_valid_from,
+    null::timestamp as test_valid_to,
+    updated_at as test_updated_at,
+    md5(id1::text || '|' || id2::text || '|' || updated_at::text) as test_scd_id
+from {schema}.seed;
+"""
+
+model_seed_sql = """
+select * from {{target.database}}.{{target.schema}}.seed
+"""
+
+snapshots_multi_key_yml = """
+snapshots:
+  - name: snapshot_actual
+    relation: "ref('seed')"
+    config:
+      strategy: timestamp
+      updated_at: updated_at
+      unique_key:
+        - id1
+        - id2
+      snapshot_meta_column_names:
+          dbt_valid_to: test_valid_to
+          dbt_valid_from: test_valid_from
+          dbt_scd_id: test_scd_id
+          dbt_updated_at: test_updated_at
+"""
+
+invalidate_multi_key_sql = """
+-- update records 11 - 21. Change email and updated_at field
+update {schema}.seed set
+    updated_at = updated_at + interval '1 hour',
+    email      =  case when id1 = 20 then 'pfoxj@creativecommons.org' else 'new_' || email end
+where id1 >= 10 and id1 <= 20;
+
+
+-- invalidate records 11 - 21
+update {schema}.snapshot_expected set
+    test_valid_to   = updated_at + interval '1 hour'
+where id1 >= 10 and id1 <= 20;
+
+"""
+
+update_multi_key_sql = """
+-- insert v2 of the 11 - 21 records
+
+insert into {schema}.snapshot_expected (
+    id1,
+    id2,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    test_valid_from,
+    test_valid_to,
+    test_updated_at,
+    test_scd_id
+)
+
+select
+    id1,
+    id2,
+    first_name,
+    last_name,
+    email,
+    gender,
+    ip_address,
+    updated_at,
+    -- fields added by snapshotting
+    updated_at as test_valid_from,
+    null::timestamp as test_valid_to,
+    updated_at as test_updated_at,
+    md5(id1::text || '|' || id2::text || '|' || updated_at::text) as test_scd_id
+from {schema}.seed
+where id1 >= 10 and id1 <= 20;
+"""
diff --git a/tests/functional/adapter/simple_snapshot/test_various_configs.py b/tests/functional/adapter/simple_snapshot/test_various_configs.py
new file mode 100644
index 00000000000..d288d2934df
--- /dev/null
+++ b/tests/functional/adapter/simple_snapshot/test_various_configs.py
@@ -0,0 +1,277 @@
+import datetime
+
+import pytest
+
+from dbt.tests.util import (
+    check_relations_equal,
+    get_manifest,
+    run_dbt,
+    run_dbt_and_capture,
+    run_sql_with_adapter,
+    update_config_file,
+)
+from tests.functional.adapter.simple_snapshot.fixtures import (
+    create_multi_key_seed_sql,
+    create_multi_key_snapshot_expected_sql,
+    create_seed_sql,
+    create_snapshot_expected_sql,
+    invalidate_multi_key_sql,
+    invalidate_sql,
+    model_seed_sql,
+    populate_multi_key_snapshot_expected_sql,
+    populate_snapshot_expected_sql,
+    populate_snapshot_expected_valid_to_current_sql,
+    ref_snapshot_sql,
+    seed_insert_sql,
+    seed_multi_key_insert_sql,
+    snapshot_actual_sql,
+    snapshots_multi_key_yml,
+    snapshots_no_column_names_yml,
+    snapshots_valid_to_current_yml,
+    snapshots_yml,
+    update_multi_key_sql,
+    update_sql,
+    update_with_current_sql,
+)
+
+
+class BaseSnapshotColumnNames:
+    @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,
+            "ref_snapshot.sql": ref_snapshot_sql,
+        }
+
+    def test_snapshot_column_names(self, project):
+        project.run_sql(create_seed_sql)
+        project.run_sql(create_snapshot_expected_sql)
+        project.run_sql(seed_insert_sql)
+        project.run_sql(populate_snapshot_expected_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+
+        project.run_sql(invalidate_sql)
+        project.run_sql(update_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+
+        # run_dbt(["test"])
+        check_relations_equal(project.adapter, ["snapshot_actual", "snapshot_expected"])
+
+
+class TestSnapshotColumnNames(BaseSnapshotColumnNames):
+    pass
+
+
+class BaseSnapshotColumnNamesFromDbtProject:
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {"snapshot.sql": snapshot_actual_sql}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "snapshots.yml": snapshots_no_column_names_yml,
+            "ref_snapshot.sql": ref_snapshot_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "snapshots": {
+                "test": {
+                    "+snapshot_meta_column_names": {
+                        "dbt_valid_to": "test_valid_to",
+                        "dbt_valid_from": "test_valid_from",
+                        "dbt_scd_id": "test_scd_id",
+                        "dbt_updated_at": "test_updated_at",
+                    }
+                }
+            }
+        }
+
+    def test_snapshot_column_names_from_project(self, project):
+        project.run_sql(create_seed_sql)
+        project.run_sql(create_snapshot_expected_sql)
+        project.run_sql(seed_insert_sql)
+        project.run_sql(populate_snapshot_expected_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+
+        project.run_sql(invalidate_sql)
+        project.run_sql(update_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+
+        # run_dbt(["test"])
+        check_relations_equal(project.adapter, ["snapshot_actual", "snapshot_expected"])
+
+
+class TestSnapshotColumnNamesFromDbtProject(BaseSnapshotColumnNamesFromDbtProject):
+    pass
+
+
+class BaseSnapshotInvalidColumnNames:
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {"snapshot.sql": snapshot_actual_sql}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "snapshots.yml": snapshots_no_column_names_yml,
+            "ref_snapshot.sql": ref_snapshot_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "snapshots": {
+                "test": {
+                    "+snapshot_meta_column_names": {
+                        "dbt_valid_to": "test_valid_to",
+                        "dbt_valid_from": "test_valid_from",
+                        "dbt_scd_id": "test_scd_id",
+                        "dbt_updated_at": "test_updated_at",
+                    }
+                }
+            }
+        }
+
+    def test_snapshot_invalid_column_names(self, project):
+        project.run_sql(create_seed_sql)
+        project.run_sql(create_snapshot_expected_sql)
+        project.run_sql(seed_insert_sql)
+        project.run_sql(populate_snapshot_expected_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+        manifest = get_manifest(project.project_root)
+        snapshot_node = manifest.nodes["snapshot.test.snapshot_actual"]
+        snapshot_node.config.snapshot_meta_column_names == {
+            "dbt_valid_to": "test_valid_to",
+            "dbt_valid_from": "test_valid_from",
+            "dbt_scd_id": "test_scd_id",
+            "dbt_updated_at": "test_updated_at",
+        }
+
+        project.run_sql(invalidate_sql)
+        project.run_sql(update_sql)
+
+        # Change snapshot_meta_columns and look for an error
+        different_columns = {
+            "snapshots": {
+                "test": {
+                    "+snapshot_meta_column_names": {
+                        "dbt_valid_to": "test_valid_to",
+                        "dbt_updated_at": "test_updated_at",
+                    }
+                }
+            }
+        }
+        update_config_file(different_columns, "dbt_project.yml")
+
+        results, log_output = run_dbt_and_capture(["snapshot"], expect_pass=False)
+        assert len(results) == 1
+        assert "Compilation Error in snapshot snapshot_actual" in log_output
+        assert "Snapshot target is missing configured columns" in log_output
+
+
+class TestSnapshotInvalidColumnNames(BaseSnapshotInvalidColumnNames):
+    pass
+
+
+class BaseSnapshotDbtValidToCurrent:
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {"snapshot.sql": snapshot_actual_sql}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "snapshots.yml": snapshots_valid_to_current_yml,
+            "ref_snapshot.sql": ref_snapshot_sql,
+        }
+
+    def test_valid_to_current(self, project):
+        project.run_sql(create_seed_sql)
+        project.run_sql(create_snapshot_expected_sql)
+        project.run_sql(seed_insert_sql)
+        project.run_sql(populate_snapshot_expected_valid_to_current_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+
+        original_snapshot = run_sql_with_adapter(
+            project.adapter,
+            "select id, test_scd_id, test_valid_to from {schema}.snapshot_actual",
+            "all",
+        )
+        assert original_snapshot[0][2] == datetime.datetime(2099, 12, 31, 0, 0)
+        assert original_snapshot[9][2] == datetime.datetime(2099, 12, 31, 0, 0)
+
+        project.run_sql(invalidate_sql)
+        project.run_sql(update_with_current_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+
+        updated_snapshot = run_sql_with_adapter(
+            project.adapter,
+            "select id, test_scd_id, test_valid_to from {schema}.snapshot_actual",
+            "all",
+        )
+        assert updated_snapshot[0][2] == datetime.datetime(2099, 12, 31, 0, 0)
+        # Original row that was updated now has a non-current (2099/12/31) date
+        assert updated_snapshot[9][2] == datetime.datetime(2016, 8, 20, 16, 44, 49)
+        # Updated row has a current date
+        assert updated_snapshot[20][2] == datetime.datetime(2099, 12, 31, 0, 0)
+
+        check_relations_equal(project.adapter, ["snapshot_actual", "snapshot_expected"])
+
+
+class TestSnapshotDbtValidToCurrent(BaseSnapshotDbtValidToCurrent):
+    pass
+
+
+# This uses snapshot_meta_column_names, yaml-only snapshot def,
+# and multiple keys
+class BaseSnapshotMultiUniqueKey:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "seed.sql": model_seed_sql,
+            "snapshots.yml": snapshots_multi_key_yml,
+            "ref_snapshot.sql": ref_snapshot_sql,
+        }
+
+    def test_multi_column_unique_key(self, project):
+        project.run_sql(create_multi_key_seed_sql)
+        project.run_sql(create_multi_key_snapshot_expected_sql)
+        project.run_sql(seed_multi_key_insert_sql)
+        project.run_sql(populate_multi_key_snapshot_expected_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+
+        project.run_sql(invalidate_multi_key_sql)
+        project.run_sql(update_multi_key_sql)
+
+        results = run_dbt(["snapshot"])
+        assert len(results) == 1
+
+        # run_dbt(["test"])
+        check_relations_equal(project.adapter, ["snapshot_actual", "snapshot_expected"])
+
+
+class TestSnapshotMultiUniqueKey(BaseSnapshotMultiUniqueKey):
+    pass