[Backport 1.5.latest] fix v0 ref resolution

.changes/unreleased/Fixes-20230419-220910.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix v0 ref resolution
+time: 2023-04-19T22:09:10.155137-04:00
+custom:
+  Author: MichelleArk
+  Issue: "7408"

core/dbt/contracts/graph/manifest.py

@@ -177,7 +177,7 @@ def add_node(self, node: ManifestNode):
             if node.name not in self.storage:
                 self.storage[node.name] = {}
 
-            if node.resource_type in self._versioned_types and node.version:
+            if node.is_versioned:
                 if node.search_name not in self.storage:
                     self.storage[node.search_name] = {}
                 self.storage[node.search_name][node.package_name] = node.unique_id

core/dbt/contracts/graph/nodes.py

@@ -668,8 +668,8 @@ class ModelNode(CompiledNode):
     latest_version: Optional[NodeVersion] = None
 
     @property
-    def is_latest_version(self):
-        return self.version and self.version == self.latest_version
+    def is_latest_version(self) -> bool:
+        return self.version is not None and self.version == self.latest_version
 
     @property
     def search_name(self):

core/dbt/parser/schemas.py

@@ -1068,7 +1068,9 @@ def parse_patch(self, block: TargetBlock[UnparsedModelUpdate], refs: ParserRef)
         else:
             assert isinstance(self.yaml.file, SchemaSourceFile)
             source_file: SchemaSourceFile = self.yaml.file
-            latest_version = target.latest_version or max(versions).v
+            latest_version = (
+                target.latest_version if target.latest_version is not None else max(versions).v
+            )
             for unparsed_version in versions:
                 versioned_model_name = (
                     unparsed_version.defined_in or f"{block.name}_{unparsed_version.formatted_v}"

test/unit/test_manifest.py

@@ -1399,6 +1399,20 @@ def _refable_parameter_sets():
                 version=None,
                 expected=None,
             ),
+            FindNodeSpec(
+                nodes=[MockNode("root", "my_model", version="0", is_latest_version=False)],
+                sources=[],
+                package="root",
+                version=None,
+                expected=None,
+            ),
+            FindNodeSpec(
+                nodes=[MockNode("root", "my_model", version="0", is_latest_version=True)],
+                sources=[],
+                package="root",
+                version=None,
+                expected=("root", "my_model", "0"),
+            ),
             # a source with that name exists, but not a refable
             FindNodeSpec(
                 nodes=[],

test/unit/test_parser.py

@@ -259,9 +259,6 @@ def assertEqualNodes(node_one, node_two):
 models:
     - name: my_model
       description: A description of my model
-      config:
-        materialized: table
-        sql_header: test_sql_header
       tests:
         - unique:
             column_name: color
@@ -280,6 +277,31 @@ def assertEqualNodes(node_one, node_two):
           columns:
             - include: '*'
             - name: extra
+        - v: 2
+          columns:
+            - include: '*'
+              exclude: ['location_id']
+            - name: extra
+"""
+
+MULTIPLE_TABLE_VERSIONED_MODEL = """
+models:
+    - name: my_model
+      description: A description of my model
+      config:
+        materialized: table
+        sql_header: test_sql_header
+      columns:
+        - name: color
+          description: The color value
+        - name: location_id
+          data_type: int
+      versions:
+        - v: 1
+          defined_in: arbitrary_file_name
+          columns:
+            - include: '*'
+            - name: extra
         - v: 2
           config:
             materialized: view
@@ -289,6 +311,26 @@ def assertEqualNodes(node_one, node_two):
             - name: extra
 """
 
+MULTIPLE_TABLE_VERSIONED_MODEL_V0 = """
+models:
+    - name: my_model
+      versions:
+        - v: 0
+          defined_in: arbitrary_file_name
+        - v: 2
+"""
+
+
+MULTIPLE_TABLE_VERSIONED_MODEL_V0_LATEST_VERSION = """
+models:
+    - name: my_model
+      latest_version: 0
+      versions:
+        - v: 0
+          defined_in: arbitrary_file_name
+        - v: 2
+"""
+
 
 SINGLE_TABLE_SOURCE_PATCH = """
 sources:
@@ -554,7 +596,7 @@ def test__parse_basic_model_tests(self):
         self.assertEqual(self.parser.manifest.files[file_id].node_patches, ["model.root.my_model"])
 
 
-class SchemaParserVersionedModelsTest(SchemaParserTest):
+class SchemaParserVersionedModels(SchemaParserTest):
     def setUp(self):
         super().setUp()
         my_model_v1_node = MockNode(
@@ -604,11 +646,7 @@ def test__parse_versioned_model_tests(self):
         self.assert_has_manifest_lengths(self.parser.manifest, nodes=5)
 
         all_nodes = sorted(self.parser.manifest.nodes.values(), key=lambda n: n.unique_id)
-        tests = []
-        for node in all_nodes:
-            if node.resource_type != NodeType.Test:
-                continue
-            tests.append(node)
+        tests = [node for node in all_nodes if node.resource_type == NodeType.Test]
 
         # test on color column on my_model v1
         self.assertEqual(tests[0].config.severity, "WARN")
@@ -682,6 +720,84 @@ def test__parse_versioned_model_tests(self):
             ["model.snowplow.my_model.v1", "model.snowplow.my_model.v2"],
         )
 
+    def test__parsed_versioned_models(self):
+        block = self.file_block_for(MULTIPLE_TABLE_VERSIONED_MODEL, "test_one.yml")
+        self.parser.manifest.files[block.file.file_id] = block.file
+        dct = yaml_from_file(block.file)
+        self.parser.parse_file(block, dct)
+        self.assert_has_manifest_lengths(self.parser.manifest, nodes=2)
+
+        all_nodes = sorted(self.parser.manifest.nodes.values(), key=lambda n: n.unique_id)
+        models = [node for node in all_nodes if node.resource_type == NodeType.Model]
+
+        # test v1 model
+        parsed_node_patch_v1 = models[0].patch.call_args_list[0][0][0]
+        self.assertEqual(models[0].unique_id, "model.snowplow.my_model.v1")
+        self.assertEqual(parsed_node_patch_v1.version, 1)
+        self.assertEqual(parsed_node_patch_v1.latest_version, 2)
+        self.assertEqual(
+            list(parsed_node_patch_v1.columns.keys()), ["color", "location_id", "extra"]
+        )
+        self.assertEqual(
+            parsed_node_patch_v1.config, {"materialized": "table", "sql_header": "test_sql_header"}
+        )
+
+        # test v2 model
+        parsed_node_patch_v2 = models[1].patch.call_args_list[0][0][0]
+        self.assertEqual(models[1].unique_id, "model.snowplow.my_model.v2")
+        self.assertEqual(parsed_node_patch_v2.version, 2)
+        self.assertEqual(parsed_node_patch_v2.latest_version, 2)
+        self.assertEqual(list(parsed_node_patch_v2.columns.keys()), ["color", "extra"])
+        self.assertEqual(
+            parsed_node_patch_v2.config, {"materialized": "view", "sql_header": "test_sql_header"}
+        )
+
+    def test__parsed_versioned_models_v0(self):
+        block = self.file_block_for(MULTIPLE_TABLE_VERSIONED_MODEL_V0, "test_one.yml")
+        self.parser.manifest.files[block.file.file_id] = block.file
+        dct = yaml_from_file(block.file)
+        self.parser.parse_file(block, dct)
+        self.assert_has_manifest_lengths(self.parser.manifest, nodes=2)
+
+        all_nodes = sorted(self.parser.manifest.nodes.values(), key=lambda n: n.unique_id)
+        models = [node for node in all_nodes if node.resource_type == NodeType.Model]
+
+        # test v0 model
+        parsed_node_patch_v1 = models[0].patch.call_args_list[0][0][0]
+        self.assertEqual(models[0].unique_id, "model.snowplow.my_model.v0")
+        self.assertEqual(parsed_node_patch_v1.version, 0)
+        self.assertEqual(parsed_node_patch_v1.latest_version, 2)
+
+        # test v2 model
+        parsed_node_patch_v2 = models[1].patch.call_args_list[0][0][0]
+        self.assertEqual(models[1].unique_id, "model.snowplow.my_model.v2")
+        self.assertEqual(parsed_node_patch_v2.version, 2)
+        self.assertEqual(parsed_node_patch_v2.latest_version, 2)
+
+    def test__parsed_versioned_models_v0_latest_version(self):
+        block = self.file_block_for(
+            MULTIPLE_TABLE_VERSIONED_MODEL_V0_LATEST_VERSION, "test_one.yml"
+        )
+        self.parser.manifest.files[block.file.file_id] = block.file
+        dct = yaml_from_file(block.file)
+        self.parser.parse_file(block, dct)
+        self.assert_has_manifest_lengths(self.parser.manifest, nodes=2)
+
+        all_nodes = sorted(self.parser.manifest.nodes.values(), key=lambda n: n.unique_id)
+        models = [node for node in all_nodes if node.resource_type == NodeType.Model]
+
+        # test v0 model
+        parsed_node_patch_v1 = models[0].patch.call_args_list[0][0][0]
+        self.assertEqual(models[0].unique_id, "model.snowplow.my_model.v0")
+        self.assertEqual(parsed_node_patch_v1.version, 0)
+        self.assertEqual(parsed_node_patch_v1.latest_version, 0)
+
+        # test v2 model
+        parsed_node_patch_v2 = models[1].patch.call_args_list[0][0][0]
+        self.assertEqual(models[1].unique_id, "model.snowplow.my_model.v2")
+        self.assertEqual(parsed_node_patch_v2.version, 2)
+        self.assertEqual(parsed_node_patch_v2.latest_version, 0)
+
 
 sql_model = """
 {{ config(materialized="table") }}

test/unit/utils.py

@@ -338,8 +338,8 @@ def MockNode(package, name, resource_type=None, **kwargs):
     search_name = name if version is None else f"{name}.v{version}"
     unique_id = (
         f"{str(resource_type)}.{package}.{name}"
-        if version is None
-        else f"{str(resource_type)}.{package}.{name}.v{version}"
+        # if version is None
+        # else f"{str(resource_type)}.{package}.{name}.v{version}"
     )
     node = mock.MagicMock(
         __class__=cls,

tests/functional/graph_selection/fixtures.py

@@ -51,6 +51,7 @@
   - name: versioned
     latest_version: 2
     versions:
+      - v: 0
       - v: 1
       - v: 2
       - v: 3