deprecate materialization overrides from imported packages

core/dbt/contracts/graph/manifest.py

@@ -23,6 +23,7 @@
 )
 from typing_extensions import Protocol
 
+from dbt import deprecations
 from dbt import tracking
 from dbt.contracts.graph.nodes import (
     BaseNode,
@@ -570,11 +571,15 @@ def __lt__(self, other: object) -> bool:
 
 
 class CandidateList(List[M]):
-    def last(self) -> Optional[Macro]:
+    def last_candidate(self) -> Optional[MacroCandidate]:
         if not self:
             return None
         self.sort()
-        return self[-1].macro
+        return self[-1]
+
+    def last(self) -> Optional[Macro]:
+        last_candidate = self.last_candidate()
+        return last_candidate.macro if last_candidate is not None else None
 
 
 def _get_locality(macro: Macro, root_project_name: str, internal_packages: Set[str]) -> Locality:
@@ -930,7 +935,24 @@ def find_materialization_macro_by_name(
                 for specificity, atype in enumerate(self._get_parent_adapter_types(adapter_type))
             )
         )
-        return candidates.last()
+        core_candidates = [
+            candidate for candidate in candidates if candidate.locality == Locality.Core
+        ]
+
+        materialization_candidate = candidates.last_candidate()
+        # If an imported materialization macro was found that also had a core candidate, fire a deprecation
+        if (
+            materialization_candidate is not None
+            and materialization_candidate.locality == Locality.Imported
+            and core_candidates
+        ):
+            deprecations.warn(
+                "package-materialization-override",
+                package_name=materialization_candidate.macro.package_name,
+                materialization_name=materialization_name,
+            )
+
+        return materialization_candidate.macro if materialization_candidate else None
 
     def get_resource_fqns(self) -> Mapping[str, PathSet]:
         resource_fqns: Dict[str, Set[Tuple[str, ...]]] = {}

core/dbt/deprecations.py

@@ -118,6 +118,11 @@ def show(self, *args, **kwargs) -> None:
             active_deprecations.add(self.name)
 
 
+class PackageMaterializationOverrideDeprecation(DBTDeprecation):
+    _name = "package-materialization-override"
+    _event = "PackageMaterializationOverrideDeprecation"
+
+
 def renamed_env_var(old_name: str, new_name: str):
     class EnvironmentVariableRenamed(DBTDeprecation):
         _name = f"environment-variable-renamed:{old_name}"
@@ -157,6 +162,7 @@ def warn(name, *args, **kwargs):
     CollectFreshnessReturnSignature(),
     TestsConfigDeprecation(),
     ProjectFlagsMovedDeprecation(),
+    PackageMaterializationOverrideDeprecation(),
 ]
 
 deprecations: Dict[str, DBTDeprecation] = {d.name: d for d in deprecations_list}

core/dbt/events/core_types.proto

@@ -422,6 +422,17 @@ message TotalModelNamesWithSpacesDeprecation {
     string level = 3;
 }
 
+// D016
+message PackageMaterializationOverrideDeprecation {
+    string package_name = 1;
+    string materialization_name = 2;
+}
+
+message PackageMaterializationOverrideDeprecationMsg {
+    CoreEventInfo info = 1;
+    PackageMaterializationOverrideDeprecation data = 2;
+}
+
 message TotalModelNamesWithSpacesDeprecationMsg {
     CoreEventInfo info = 1;
     TotalModelNamesWithSpacesDeprecation data = 2;

core/dbt/events/types.py

@@ -450,6 +450,16 @@ def message(self) -> str:
         return line_wrap_message(description)
 
 
+class PackageMaterializationOverrideDeprecation(WarnLevel):
+    def code(self) -> str:
+        return "D016"
+
+    def message(self) -> str:
+        description = f"Package '{self.package_name}' is overriding the built-in materialization '{self.materialization_name}' from an installed package, which will be deprecated in future versions of dbt."
+
+        return line_wrap_message(description)
+
+
 # =======================================================
 # I - Project parsing
 # =======================================================

tests/functional/materializations/test_custom_materialization.py

@@ -36,6 +36,7 @@ def packages(self):
     def test_default_dependency(self, project, override_view_default_dep):
         run_dbt(["deps"])
         # this should error because the override is buggy
+        # TODO: assert deprecation fired
         run_dbt(["run"], expect_pass=False)