Add supported languages to materializations (#5695)

* Add supported languages to materializations

* Add changie entry

* Linting

* add more error and only get supported language for materialization macro, update schema

* fix test and add more check

Co-authored-by: Chenyu Li <chenyu.li@dbtlabs.com>

.changes/unreleased/Under the Hood-20220822-103739.yaml

@@ -0,0 +1,7 @@
+kind: Under the Hood
+body: Add supported languages to materializations
+time: 2022-08-22T10:37:39.50743-05:00
+custom:
+  Author: stu-k
+  Issue: "5569"
+  PR: "5695"

core/dbt/clients/jinja.py

@@ -37,6 +37,10 @@
     UndefinedMacroException,
 )
 from dbt import flags
+from dbt.node_types import ModelLanguage
+
+
+SUPPORTED_LANG_ARG = jinja2.nodes.Name("supported_languages", "param")
 
 
 def _linecache_inject(source, write):
@@ -364,9 +368,20 @@ def parse(self, parser):
                 value = parser.parse_expression()
                 adapter_name = value.value
 
+            elif target.name == "supported_languages":
+                target.set_ctx("param")
+                node.args.append(target)
+                parser.stream.expect("assign")
+                languages = parser.parse_expression()
+                node.defaults.append(languages)
+
             else:
                 invalid_materialization_argument(materialization_name, target.name)
 
+        if SUPPORTED_LANG_ARG not in node.args:
+            node.args.append(SUPPORTED_LANG_ARG)
+            node.defaults.append(jinja2.nodes.List([jinja2.nodes.Const("sql")]))
+
         node.name = get_materialization_macro_name(materialization_name, adapter_name)
 
         node.body = parser.parse_statements(("name:endmaterialization",), drop_needle=True)
@@ -632,3 +647,21 @@ def _convert_function(value: Any, keypath: Tuple[Union[str, int], ...]) -> Any:
     # when the test node was created in _parse_generic_test.
     kwargs = deep_map_render(_convert_function, node.test_metadata.kwargs)
     context[GENERIC_TEST_KWARGS_NAME] = kwargs
+
+
+def get_supported_languages(node: jinja2.nodes.Macro) -> List[ModelLanguage]:
+    if "materialization" not in node.name:
+        raise_compiler_error("Only materialization macros can be used with this function")
+
+    no_kwargs = not node.defaults
+    no_langs_found = SUPPORTED_LANG_ARG not in node.args
+
+    if no_kwargs or no_langs_found:
+        raise_compiler_error(f"No supported_languages found in materialization macro {node.name}")
+
+    lang_idx = node.args.index(SUPPORTED_LANG_ARG)
+    # indexing defaults from the end
+    # since supported_languages is a kwarg, and kwargs are at always after args
+    return [
+        ModelLanguage[item.value] for item in node.defaults[-(len(node.args) - lang_idx)].items
+    ]

core/dbt/contracts/graph/parsed.py

@@ -41,7 +41,7 @@
 from dbt.contracts.util import Replaceable, AdditionalPropertiesMixin
 from dbt.exceptions import warn_or_error
 from dbt import flags
-from dbt.node_types import NodeType
+from dbt.node_types import ModelLanguage, NodeType
 
 
 from .model_config import (
@@ -516,6 +516,7 @@ class ParsedMacro(UnparsedBaseNode, HasUniqueID):
     patch_path: Optional[str] = None
     arguments: List[MacroArgument] = field(default_factory=list)
     created_at: float = field(default_factory=lambda: time.time())
+    supported_languages: Optional[List[ModelLanguage]] = None
 
     def patch(self, patch: ParsedMacroPatch):
         self.patch_path: Optional[str] = patch.file_id

core/dbt/parser/macros.py

@@ -80,13 +80,16 @@ def parse_unparsed_macros(self, base_node: UnparsedMacro) -> Iterable[ParsedMacr
                     f"Found multiple macros in {block.full_block}, expected 1", node=base_node
                 )
 
-            macro_name = macro_nodes[0].name
+            macro = macro_nodes[0]
 
-            if not macro_name.startswith(MACRO_PREFIX):
+            if not macro.name.startswith(MACRO_PREFIX):
                 continue
 
-            name: str = macro_name.replace(MACRO_PREFIX, "")
+            name: str = macro.name.replace(MACRO_PREFIX, "")
             node = self.parse_macro(block, base_node, name)
+            # get supported_languages for materialization macro
+            if "materialization" in name:
+                node.supported_languages = jinja.get_supported_languages(macro)
             yield node
 
     def parse_file(self, block: FileBlock):

core/dbt/task/run.py

@@ -25,6 +25,7 @@
     CompilationException,
     InternalException,
     RuntimeException,
+    ValidationException,
     missing_materialization,
 )
 from dbt.events.functions import fire_event, get_invocation_id
@@ -266,6 +267,15 @@ def execute(self, model, manifest):
             )
         context_config = context["config"]
 
+        mat_has_supported_langs = hasattr(materialization_macro, "supported_languages")
+        model_lang_supported = model.language in materialization_macro.supported_languages
+        if mat_has_supported_langs and not model_lang_supported:
+            str_langs = [str(lang) for lang in materialization_macro.supported_languages]
+            raise ValidationException(
+                f'Materialization "{materialization_macro.name}" only supports languages {str_langs}; '
+                f'got "{model.language}"'
+            )
+
         hook_ctx = self.adapter.pre_model_hook(context_config)
         try:
             result = MacroGenerator(materialization_macro, context)()