CT 2510 Throw error for duplicate versioned and non versioned model n…

…ames (#7577)

* Check for versioned/unversioned duplicates

* Add new exception DuplicateVersionedUnversionedError

* Changie

* Handle packages when finding versioned and unversioned duplicates

(cherry picked from commit 29f2cfc)

.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

@@ -2145,6 +2145,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

@@ -1094,21 +1094,29 @@ def _check_resource_uniqueness(
 ) -> None:
     names_resources: Dict[str, ManifestNode] = {}
     alias_resources: Dict[str, ManifestNode] = {}
+    versioned_resources: Dict[str, ManifestNode] = {}
+    unversioned_resources: Dict[str, ManifestNode] = {}
+    duplicate_resources: Dict[str, ManifestNode] = {}
 
     for resource, node in manifest.nodes.items():
         if not node.is_relational:
             continue
 
         name = node.name
+        if node.is_versioned:
+            versioned_resources[name] = node
+        else:
+            unversioned_resources[name] = node
+
+        existing_node = names_resources.get(name)
+        if existing_node is not None:
+            duplicate_resources[name] = [existing_node, 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)
         full_node_name = str(relation)
 
-        existing_node = names_resources.get(name)
-        if existing_node is not None and not existing_node.is_versioned:
-            raise dbt.exceptions.DuplicateResourceNameError(existing_node, node)
-
         existing_alias = alias_resources.get(full_node_name)
         if existing_alias is not None:
             raise AmbiguousAliasError(
@@ -1118,6 +1126,25 @@ def _check_resource_uniqueness(
         names_resources[name] = node
         alias_resources[full_node_name] = node
 
+        # Check for versioned models with unversioned duplicates
+        intersection_versioned = set(versioned_resources.keys()).intersection(
+            set(unversioned_resources.keys())
+        )
+        for name in intersection_versioned:
+            versioned_node = versioned_resources[name]
+            unversioned_node = unversioned_resources[name]
+            raise dbt.exceptions.DuplicateVersionedUnversionedError(
+                versioned_node, unversioned_node
+            )
+
+        # Handle base case of multiple unversioned models with same name
+        intersection_unversioned = set(duplicate_resources.keys()) - set(
+            versioned_resources.keys()
+        )
+        for name in intersection_unversioned:
+            existing_node, node = duplicate_resources[name]
+            raise dbt.exceptions.DuplicateResourceNameError(existing_node, 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,13 @@ 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
+
+        # 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: