Declare compatibility for previous artifact versions (#5346)

* Add functional test

* Add is_compatible_version logic

* Add changelog entry

* Fix mypy

.changes/unreleased/Fixes-20220608-080914.yaml

@@ -0,0 +1,8 @@
+kind: Fixes
+body: 'Define compatibility for older manifest versions when using state: selection
+  methods'
+time: 2022-06-08T08:09:14.321735+02:00
+custom:
+  Author: jtcohen6
+  Issue: "5213"
+  PR: "5346"

core/dbt/contracts/graph/manifest.py

@@ -1135,6 +1135,10 @@ class WritableManifest(ArtifactMixin):
         )
     )
 
+    @classmethod
+    def compatible_previous_versions(self):
+        return [("manifest", 4)]
+
     def __post_serialize__(self, dct):
         for unique_id, node in dct["nodes"].items():
             if "config_call_dict" in node:

core/dbt/contracts/util.py

@@ -201,6 +201,14 @@ def json_schema(cls, embeddable: bool = False) -> Dict[str, Any]:
             result["$id"] = str(cls.dbt_schema_version)
         return result
 
+    @classmethod
+    def is_compatible_version(cls, schema_version):
+        compatible_versions = [str(cls.dbt_schema_version)]
+        if hasattr(cls, "compatible_previous_versions"):
+            for name, version in cls.compatible_previous_versions():
+                compatible_versions.append(str(SchemaVersion(name, version)))
+        return str(schema_version) in compatible_versions
+
     @classmethod
     def read_and_check_versions(cls, path: str):
         try:
@@ -217,7 +225,7 @@ def read_and_check_versions(cls, path: str):
             if "metadata" in data and "dbt_schema_version" in data["metadata"]:
                 previous_schema_version = data["metadata"]["dbt_schema_version"]
                 # cls.dbt_schema_version is a SchemaVersion object
-                if str(cls.dbt_schema_version) != previous_schema_version:
+                if not cls.is_compatible_version(previous_schema_version):
                     raise IncompatibleSchemaException(
                         expected=str(cls.dbt_schema_version), found=previous_schema_version
                     )

tests/functional/artifacts/test_previous_version_state.py

@@ -0,0 +1,50 @@
+import pytest
+import os
+from dbt.tests.util import run_dbt
+from dbt.exceptions import IncompatibleSchemaException
+
+models__my_model_sql = """
+select 1 as id
+"""
+
+
+class TestPreviousVersionState:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"my_model.sql": models__my_model_sql}
+
+    def compare_previous_state(
+        self,
+        project,
+        compare_manifest_version,
+        expect_pass,
+    ):
+        state_path = os.path.join(project.test_data_dir, f"previous/{compare_manifest_version}")
+        cli_args = [
+            "list",
+            "--select",
+            "state:modified",
+            "--state",
+            state_path,
+        ]
+        if expect_pass:
+            results = run_dbt(cli_args, expect_pass=expect_pass)
+            assert len(results) == 0
+        else:
+            with pytest.raises(IncompatibleSchemaException):
+                run_dbt(cli_args, expect_pass=expect_pass)
+
+    def test_compare_state_v5(self, project):
+        self.compare_previous_state(project, "v5", True)
+
+    def test_compare_state_v4(self, project):
+        self.compare_previous_state(project, "v4", True)
+
+    def test_compare_state_v3(self, project):
+        self.compare_previous_state(project, "v3", False)
+
+    def test_compare_state_v2(self, project):
+        self.compare_previous_state(project, "v2", False)
+
+    def test_compare_state_v1(self, project):
+        self.compare_previous_state(project, "v1", False)