Prototype: test materialization that allows materialization to be set in the config()

core/dbt/contracts/graph/model_config.py

@@ -583,12 +583,6 @@ def same_contents(cls, unrendered: Dict[str, Any], other: Dict[str, Any]) -> boo
                     return False
         return True
 
-    @classmethod
-    def validate(cls, data):
-        super().validate(data)
-        if data.get("materialized") and data.get("materialized") != "test":
-            raise ValidationError("A test must have a materialized value of 'test'")
-
 
 @dataclass
 class EmptySnapshotConfig(NodeConfig):

core/dbt/include/global_project/macros/materializations/helpers.sql

@@ -0,0 +1,13 @@
+{% macro get_materialization_macro(materialization_type) -%}
+      -- Reuse a fundamental materialization type (like table, view, or materialized_view)
+      -- TODO: support actual dispatch for materialization macros
+      -- See related tracking ticket: https://github.com/dbt-labs/dbt-core/issues/7799
+      {% set base_search_name = "materialization_" ~ materialization_type ~ "_" %}
+      {% set search_name = base_search_name ~ adapter.type() %}
+
+      {% if not search_name in context %}
+          {% set search_name = base_search_name ~ "default" %}
+      {% endif %}
+      {% set materialization_macro = context[search_name] %}
+      {% do return(materialization_macro) %}
+{%- endmacro %}

core/dbt/include/global_project/macros/materializations/tests/test.sql

@@ -2,41 +2,47 @@
 
   {% set relations = [] %}
 
-  {% if should_store_failures() %}
-
-    {% set identifier = model['alias'] %}
-    {% set old_relation = adapter.get_relation(database=database, schema=schema, identifier=identifier) %}
-    {% set target_relation = api.Relation.create(
-        identifier=identifier, schema=schema, database=database, type='table') -%} %}
-
-    {% if old_relation %}
-        {% do adapter.drop_relation(old_relation) %}
-    {% endif %}
-
-    {% call statement(auto_begin=True) %}
-        {{ create_table_as(False, target_relation, sql) }}
-    {% endcall %}
+  -- default SQL to use for the relation to test; will be modified depending on materialization_type
+  {% set main_sql = sql %}
 
-    {% do relations.append(target_relation) %}
+  -- default value
+  {% set materialization_type = "ephemeral" %}
+  {{ log("materialization_type: " ~ materialization_type ~ " (default)", True) }}
 
-    {% set main_sql %}
-        select *
-        from {{ target_relation }}
-    {% endset %}
+  -- default value if storing failures
+  {% if should_store_failures() %}
+      {% set materialization_type = "table" %}
+  {% endif %}
+  {{ log("materialization_type: " ~ materialization_type ~ " (after considering should_store_failures())", True) }}
 
-    {{ adapter.commit() }}
+  -- override the default (but only if configured to do so)
+  {% set materialization_type = config.get("materialized") or materialization_type %}
+  {{ log("materialization_type: " ~ materialization_type ~ " (after considering test config())", True) }}
 
-  {% else %}
+  -- only allow certain materializations for now
+  {% if materialization_type not in ["test", "ephemeral", "table", "view", "materialized_view"] %}
+    {{ exceptions.raise_compiler_error("Invalid `materialization_type`. Got: " ~ materialization_type) }}
+  {% endif %}
 
-      {% set main_sql = sql %}
+  -- only a few of the allowed materializations actually create database objects
+  {% if materialization_type in ["table", "view", "materialized_view"] %}
+      {%- set target_relation = this.incorporate(type=materialization_type) -%}
+      {%- set materialization_macro = get_materialization_macro(materialization_type) -%}
+      {% set relations = materialization_macro() %}
 
+      {% set main_sql %}
+          select *
+          from {{ target_relation }}
+      {% endset %}
   {% endif %}
 
   {% set limit = config.get('limit') %}
   {% set fail_calc = config.get('fail_calc') %}
   {% set warn_if = config.get('warn_if') %}
   {% set error_if = config.get('error_if') %}
 
+  {{ log("main_sql: " ~ main_sql, True) }}
+
   {% call statement('main', fetch_result=True) -%}
 
     {{ get_test_sql(main_sql, fail_calc, warn_if, error_if, limit)}}

core/dbt/task/test.py

@@ -95,9 +95,8 @@ def before_execute(self):
 
     def execute_test(self, test: TestNode, manifest: Manifest) -> TestResultData:
         context = generate_runtime_model_context(test, self.config, manifest)
-
         materialization_macro = manifest.find_materialization_macro_by_name(
-            self.config.project_name, test.get_materialization(), self.adapter.type()
+            self.config.project_name, "test", self.adapter.type()
         )
 
         if materialization_macro is None: