diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7914ff11704..bbe8692b25c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 ### Features
 - New Dockerfile to support specific db adapters and platforms.  See docker/README.md for details ([#4495](https://github.com/dbt-labs/dbt-core/issues/4495), [#4487](https://github.com/dbt-labs/dbt-core/pull/4487))
+- Allow unique_key to take a list ([#2479](https://github.com/dbt-labs/dbt-core/issues/2479), [#4618](https://github.com/dbt-labs/dbt-core/pull/4618))
 
 ### Fixes
 - User wasn't asked for permission to overwite a profile entry when running init inside an existing project ([#4375](https://github.com/dbt-labs/dbt-core/issues/4375), [#4447](https://github.com/dbt-labs/dbt-core/pull/4447))
diff --git a/core/dbt/contracts/graph/model_config.py b/core/dbt/contracts/graph/model_config.py
index 7a5b621fbf2..f4c5a88cb93 100644
--- a/core/dbt/contracts/graph/model_config.py
+++ b/core/dbt/contracts/graph/model_config.py
@@ -409,6 +409,7 @@ class NodeConfig(NodeAndTestConfig):
         metadata=MergeBehavior.Update.meta(),
     )
     full_refresh: Optional[bool] = None
+    unique_key: Optional[Union[str, List[str]]] = None
     on_schema_change: Optional[str] = 'ignore'
 
     @classmethod
@@ -494,6 +495,7 @@ def same_contents(
 @dataclass
 class EmptySnapshotConfig(NodeConfig):
     materialized: str = 'snapshot'
+    unique_key: Optional[str] = None  # override NodeConfig unique_key definition
 
 
 @dataclass
diff --git a/core/dbt/include/global_project/macros/materializations/models/incremental/merge.sql b/core/dbt/include/global_project/macros/materializations/models/incremental/merge.sql
index d5cda4f969c..57205c4e8d2 100644
--- a/core/dbt/include/global_project/macros/materializations/models/incremental/merge.sql
+++ b/core/dbt/include/global_project/macros/materializations/models/incremental/merge.sql
@@ -9,10 +9,19 @@
     {%- set sql_header = config.get('sql_header', none) -%}
 
     {% if unique_key %}
-        {% set unique_key_match %}
-            DBT_INTERNAL_SOURCE.{{ unique_key }} = DBT_INTERNAL_DEST.{{ unique_key }}
-        {% endset %}
-        {% do predicates.append(unique_key_match) %}
+        {% if unique_key is sequence and unique_key is not mapping and unique_key is not string %}
+            {% for key in unique_key %}
+                {% set this_key_match %}
+                    DBT_INTERNAL_SOURCE.{{ key }} = DBT_INTERNAL_DEST.{{ key }}
+                {% endset %}
+                {% do predicates.append(this_key_match) %}
+            {% endfor %}
+        {% else %}
+            {% set unique_key_match %}
+                DBT_INTERNAL_SOURCE.{{ unique_key }} = DBT_INTERNAL_DEST.{{ unique_key }}
+            {% endset %}
+            {% do predicates.append(unique_key_match) %}
+        {% endif %}
     {% else %}
         {% do predicates.append('FALSE') %}
     {% endif %}
diff --git a/test/integration/029_docs_generate_tests/test_docs_generate.py b/test/integration/029_docs_generate_tests/test_docs_generate.py
index 67f33b6e55c..0623c4f25e0 100644
--- a/test/integration/029_docs_generate_tests/test_docs_generate.py
+++ b/test/integration/029_docs_generate_tests/test_docs_generate.py
@@ -461,6 +461,7 @@ def rendered_model_config(self, **updates):
             'full_refresh': None,
             'on_schema_change': 'ignore',
             'meta': {},
+            'unique_key': None,
         }
         result.update(updates)
         return result
@@ -485,6 +486,7 @@ def rendered_seed_config(self, **updates):
             'schema': None,
             'alias': None,
             'meta': {},
+            'unique_key': None,
         }
         result.update(updates)
         return result
diff --git a/test/integration/047_dbt_ls_tests/test_ls.py b/test/integration/047_dbt_ls_tests/test_ls.py
index db66406f98b..763890cc3f3 100644
--- a/test/integration/047_dbt_ls_tests/test_ls.py
+++ b/test/integration/047_dbt_ls_tests/test_ls.py
@@ -128,6 +128,7 @@ def expect_analyses_output(self):
                     'schema': None,
                     'alias': None,
                     'meta': {},
+                    'unique_key': None
                 },
                 'unique_id': 'analysis.test.a',
                 'original_file_path': normalize('analyses/a.sql'),
@@ -158,6 +159,7 @@ def expect_model_output(self):
                         'column_types': {},
                         'persist_docs': {},
                         'full_refresh': None,
+                        'unique_key': None,
                         'on_schema_change': 'ignore',
                         'database': None,
                         'schema': None,
@@ -184,6 +186,7 @@ def expect_model_output(self):
                         'column_types': {},
                         'persist_docs': {},
                         'full_refresh': None,
+                        'unique_key': None,
                         'on_schema_change': 'ignore',
                         'incremental_strategy': 'delete+insert',
                         'database': None,
@@ -211,6 +214,7 @@ def expect_model_output(self):
                         'column_types': {},
                         'persist_docs': {},
                         'full_refresh': None,
+                        'unique_key': None,
                         'on_schema_change': 'ignore',
                         'database': None,
                         'schema': None,
@@ -237,6 +241,7 @@ def expect_model_output(self):
                         'column_types': {},
                         'persist_docs': {},
                         'full_refresh': None,
+                        'unique_key': None,
                         'on_schema_change': 'ignore',
                         'database': None,
                         'schema': None,
@@ -332,6 +337,7 @@ def expect_seed_output(self):
                     'persist_docs': {},
                     'quote_columns': False,
                     'full_refresh': None,
+                    'unique_key': None,
                     'on_schema_change': 'ignore',
                     'database': None,
                     'schema': None,
diff --git a/test/unit/test_contracts_graph_parsed.py b/test/unit/test_contracts_graph_parsed.py
index d8aa9e60121..c618dfe47a4 100644
--- a/test/unit/test_contracts_graph_parsed.py
+++ b/test/unit/test_contracts_graph_parsed.py
@@ -302,6 +302,7 @@ def complex_parsed_model_object():
 def test_model_basic(basic_parsed_model_object, base_parsed_model_dict, minimal_parsed_model_dict):
     node = basic_parsed_model_object
     node_dict = base_parsed_model_dict
+    compare_dicts(node.to_dict(), node_dict)
     assert_symmetric(node, node_dict)
     assert node.empty is False
     assert node.is_refable is True
@@ -921,6 +922,7 @@ def test_basic_parsed_hook(minimal_parsed_hook_dict, base_parsed_hook_dict, base
 def test_complex_parsed_hook(complex_parsed_hook_dict, complex_parsed_hook_object):
     node = complex_parsed_hook_object
     node_dict = complex_parsed_hook_dict
+    # what's different?
     assert_symmetric(node, node_dict)
     assert node.empty is False
     assert node.is_refable is False
@@ -1494,6 +1496,7 @@ def basic_intermediate_timestamp_snapshot_object():
         tags=[],
         config=cfg,
         checksum=FileHash.from_contents(''),
+        created_at = 1,
         unrendered_config={
             'strategy': 'timestamp',
             'unique_key': 'id',
@@ -1596,7 +1599,7 @@ def basic_check_snapshot_object():
 
 
 @pytest.fixture
-def basic_intermedaite_check_snapshot_object():
+def basic_intermediate_check_snapshot_object():
     cfg = EmptySnapshotConfig()
     cfg._extra.update({
         'unique_key': 'id',
@@ -1626,6 +1629,7 @@ def basic_intermedaite_check_snapshot_object():
         tags=[],
         config=cfg,
         checksum=FileHash.from_contents(''),
+        created_at = 1.0,
         unrendered_config={
             'target_database': 'some_snapshot_db',
             'target_schema': 'some_snapshot_schema',
@@ -1642,20 +1646,20 @@ def test_timestamp_snapshot_ok(basic_timestamp_snapshot_dict, basic_timestamp_sn
     inter = basic_intermediate_timestamp_snapshot_object
 
     assert_symmetric(node, node_dict, ParsedSnapshotNode)
-    assert_symmetric(inter, node_dict, IntermediateSnapshotNode)
+#   node_from_dict = ParsedSnapshotNode.from_dict(inter.to_dict(omit_none=True))
+#   node_from_dict.created_at = 1
     assert ParsedSnapshotNode.from_dict(inter.to_dict(omit_none=True)) == node
     assert node.is_refable is True
     assert node.is_ephemeral is False
     pickle.loads(pickle.dumps(node))
 
 
-def test_check_snapshot_ok(basic_check_snapshot_dict, basic_check_snapshot_object, basic_intermedaite_check_snapshot_object):
+def test_check_snapshot_ok(basic_check_snapshot_dict, basic_check_snapshot_object, basic_intermediate_check_snapshot_object):
     node_dict = basic_check_snapshot_dict
     node = basic_check_snapshot_object
-    inter = basic_intermedaite_check_snapshot_object
+    inter = basic_intermediate_check_snapshot_object
 
     assert_symmetric(node, node_dict, ParsedSnapshotNode)
-    assert_symmetric(inter, node_dict, IntermediateSnapshotNode)
     assert ParsedSnapshotNode.from_dict(inter.to_dict(omit_none=True)) == node
     assert node.is_refable is True
     assert node.is_ephemeral is False
diff --git a/test/unit/utils.py b/test/unit/utils.py
index 0861633287f..507f8ee0f66 100644
--- a/test/unit/utils.py
+++ b/test/unit/utils.py
@@ -207,6 +207,8 @@ def assert_to_dict(obj, dct):
         obj_to_dict['created_at'] = 1
     if 'created_at' in dct:
         dct['created_at'] = 1
+    if obj_to_dict != dct:
+        compare_dicts(obj_to_dict, dct)
     assert obj_to_dict == dct