CT 2510 Throw error for duplicate versioned and non versioned model names (dbt-labs#7577)

* Check for versioned/unversioned duplicates

* Add new exception DuplicateVersionedUnversionedError

* Changie

* Handle packages when finding versioned and unversioned duplicates

Author: gshank

.changes/unreleased/Fixes-20230509-165007.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Throw error for duplicated versioned and unversioned models
+time: 2023-05-09T16:50:07.530882-04:00
+custom:
+  Author: gshank
+  Issue: "7487"

core/dbt/exceptions.py

@@ -2200,6 +2200,26 @@ def get_message(self) -> str:
         return msg
 
 
+class DuplicateVersionedUnversionedError(ParsingError):
+    def __init__(self, versioned_node, unversioned_node):
+        self.versioned_node = versioned_node
+        self.unversioned_node = unversioned_node
+        super().__init__(msg=self.get_message())
+
+    def get_message(self) -> str:
+        msg = f"""
+dbt found versioned and unversioned models with the name "{self.versioned_node.name}".
+
+Since these resources have the same name, dbt will be unable to find the correct resource
+when looking for ref('{self.versioned_node.name}').
+
+To fix this, change the name of the unversioned resource
+{self.unversioned_node.unique_id} ({self.unversioned_node.original_file_path})
+or add the unversioned model to the versions in {self.versioned_node.patch_path}
+    """.strip()
+        return msg
+
+
 class PropertyYMLError(CompilationError):
     def __init__(self, path: str, issue: str):
         self.path = path

core/dbt/parser/manifest.py

@@ -1267,11 +1267,19 @@ def _check_resource_uniqueness(
     config: RuntimeConfig,
 ) -> None:
     alias_resources: Dict[str, ManifestNode] = {}
+    name_resources: Dict[str, Dict] = {}
 
     for resource, node in manifest.nodes.items():
         if not node.is_relational:
             continue
 
+        if node.package_name not in name_resources:
+            name_resources[node.package_name] = {"ver": {}, "unver": {}}
+        if node.is_versioned:
+            name_resources[node.package_name]["ver"][node.name] = node
+        else:
+            name_resources[node.package_name]["unver"][node.name] = node
+
         # the full node name is really defined by the adapter's relation
         relation_cls = get_relation_class_by_name(config.credentials.type)
         relation = relation_cls.create_from(config=config, node=node)
@@ -1285,6 +1293,18 @@ def _check_resource_uniqueness(
 
         alias_resources[full_node_name] = node
 
+    for ver_unver_dict in name_resources.values():
+        versioned_names = ver_unver_dict["ver"].keys()
+        unversioned_names = ver_unver_dict["unver"].keys()
+        intersection_versioned = set(versioned_names).intersection(set(unversioned_names))
+        if intersection_versioned:
+            for name in intersection_versioned:
+                versioned_node = ver_unver_dict["ver"][name]
+                unversioned_node = ver_unver_dict["unver"][name]
+                raise dbt.exceptions.DuplicateVersionedUnversionedError(
+                    versioned_node, unversioned_node
+                )
+
 
 def _warn_for_unused_resource_config_paths(manifest: Manifest, config: RuntimeConfig) -> None:
     resource_fqns: Mapping[str, PathSet] = manifest.get_resource_fqns()

tests/functional/partial_parsing/test_partial_parsing.py

@@ -71,7 +71,7 @@
     groups_schema_yml_two_groups_private_orders_invalid_access,
 )
 
-from dbt.exceptions import CompilationError, ParsingError
+from dbt.exceptions import CompilationError, ParsingError, DuplicateVersionedUnversionedError
 from dbt.contracts.files import ParseFileType
 from dbt.contracts.results import TestStatus
 import re
@@ -358,6 +358,14 @@ def test_pp_versioned_models(self, project):
         write_file(model_two_sql, project.project_root, "models", "model_one_different.sql")
         results = run_dbt(["--partial-parse", "run"])
         assert len(results) == 3
+        manifest = get_manifest(project.project_root)
+        assert len(manifest.nodes) == 3
+        print(f"--- nodes: {manifest.nodes.keys()}")
+
+        # create a new model_one in model_one.sql and re-parse
+        write_file(model_one_sql, project.project_root, "models", "model_one.sql")
+        with pytest.raises(DuplicateVersionedUnversionedError):
+            run_dbt(["parse"])
 
 
 class TestSources: