ADAP-850: Support test results as a view (#8653)

* add `store_failures_as` parameter to TestConfig, catch strategy parameter in test materialization
* create test results as views
* updated test expected values for new config option
* break up tests into reusable tests and adapter specific configuration, update test to check for relation type and confirm views update
* move test configuration into base test class
* allow `store_failures_as` to drive whether failures are stored
* update expected test config dicts to include the new default value for store_failures_as
* Add `store_failures_as` config for generic tests
* cover --store-failures on CLI gap
* add generic tests test case for store_failures_as
* update object names for generic test case tests for store_failures_as
* remove unique generic test, it was not testing `store_failures_as`
* pull generic run and assertion into base test class to turn tests into quasi-parameterized tests
* add ephemeral option for store_failures_as, as a way to easily turn off store_failures at the model level
* add compilation error for invalid setting of store_failures_as

---------

Co-authored-by: Doug Beatty <doug.beatty@dbtlabs.com>

.changes/unreleased/Features-20230915-174428.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support storing test failures as views
+time: 2023-09-15T17:44:28.833877-04:00
+custom:
+  Author: mikealfare
+  Issue: "6914"

core/dbt/contracts/graph/model_config.py

@@ -217,7 +217,6 @@ class Hook(dbtClassMixin, Replaceable):
 
 @dataclass
 class BaseConfig(AdditionalPropertiesAllowed, Replaceable):
-
     # enable syntax like: config['key']
     def __getitem__(self, key):
         return self.get(key)
@@ -560,12 +559,61 @@ class TestConfig(NodeAndTestConfig):
     # Annotated is used by mashumaro for jsonschema generation
     severity: Annotated[Severity, Pattern(SEVERITY_PATTERN)] = Severity("ERROR")
     store_failures: Optional[bool] = None
+    store_failures_as: Optional[str] = None
     where: Optional[str] = None
     limit: Optional[int] = None
     fail_calc: str = "count(*)"
     warn_if: str = "!= 0"
     error_if: str = "!= 0"
 
+    def __post_init__(self):
+        """
+        The presence of a setting for `store_failures_as` overrides any existing setting for `store_failures`,
+        regardless of level of granularity. If `store_failures_as` is not set, then `store_failures` takes effect.
+        At the time of implementation, `store_failures = True` would always create a table; the user could not
+        configure this. Hence, if `store_failures = True` and `store_failures_as` is not specified, then it
+        should be set to "table" to mimic the existing functionality.
+
+        A side effect of this overriding functionality is that `store_failures_as="view"` at the project
+        level cannot be turned off at the model level without setting both `store_failures_as` and
+        `store_failures`. The former would cascade down and override `store_failures=False`. The proposal
+        is to include "ephemeral" as a value for `store_failures_as`, which effectively sets
+        `store_failures=False`.
+
+        The exception handling for this is tricky. If we raise an exception here, the entire run fails at
+        parse time. We would rather well-formed models run successfully, leaving only exceptions to be rerun
+        if necessary. Hence, the exception needs to be raised in the test materialization. In order to do so,
+        we need to make sure that we go down the `store_failures = True` route with the invalid setting for
+        `store_failures_as`. This results in the `.get()` defaulted to `True` below, instead of a normal
+        dictionary lookup as is done in the `if` block. Refer to the test materialization for the
+        exception that is raise as a result of an invalid value.
+
+        The intention of this block is to behave as if `store_failures_as` is the only setting,
+        but still allow for backwards compatibility for `store_failures`.
+        See https://github.com/dbt-labs/dbt-core/issues/6914 for more information.
+        """
+
+        # if `store_failures_as` is not set, it gets set by `store_failures`
+        # the settings below mimic existing behavior prior to `store_failures_as`
+        get_store_failures_as_map = {
+            True: "table",
+            False: "ephemeral",
+            None: None,
+        }
+
+        # if `store_failures_as` is set, it dictates what `store_failures` gets set to
+        # the settings below overrides whatever `store_failures` is set to by the user
+        get_store_failures_map = {
+            "ephemeral": False,
+            "table": True,
+            "view": True,
+        }
+
+        if self.store_failures_as is None:
+            self.store_failures_as = get_store_failures_as_map[self.store_failures]
+        else:
+            self.store_failures = get_store_failures_map.get(self.store_failures_as, True)
+
     @classmethod
     def same_contents(cls, unrendered: Dict[str, Any], other: Dict[str, Any]) -> bool:
         """This is like __eq__, except it explicitly checks certain fields."""
@@ -577,6 +625,7 @@ def same_contents(cls, unrendered: Dict[str, Any], other: Dict[str, Any]) -> boo
             "warn_if",
             "error_if",
             "store_failures",
+            "store_failures_as",
         ]
 
         seen = set()

core/dbt/contracts/relation.py

@@ -19,6 +19,7 @@ class RelationType(StrEnum):
     CTE = "cte"
     MaterializedView = "materialized_view"
     External = "external"
+    Ephemeral = "ephemeral"
 
 
 class ComponentName(StrEnum):

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

@@ -6,15 +6,27 @@
 
     {% set identifier = model['alias'] %}
     {% set old_relation = adapter.get_relation(database=database, schema=schema, identifier=identifier) %}
+
+    {% set store_failures_as = config.get('store_failures_as') %}
+    -- if `--store-failures` is invoked via command line and `store_failures_as` is not set,
+    -- config.get('store_failures_as', 'table') returns None, not 'table'
+    {% if store_failures_as == none %}{% set store_failures_as = 'table' %}{% endif %}
+    {% if store_failures_as not in ['table', 'view'] %}
+        {{ exceptions.raise_compiler_error(
+            "'" ~ store_failures_as ~ "' is not a valid value for `store_failures_as`. "
+            "Accepted values are: ['ephemeral', 'table', 'view']"
+        ) }}
+    {% endif %}
+
     {% set target_relation = api.Relation.create(
-        identifier=identifier, schema=schema, database=database, type='table') -%} %}
+        identifier=identifier, schema=schema, database=database, type=store_failures_as) -%} %}
 
     {% if old_relation %}
         {% do adapter.drop_relation(old_relation) %}
     {% endif %}
 
     {% call statement(auto_begin=True) %}
-        {{ create_table_as(False, target_relation, sql) }}
+        {{ get_create_sql(target_relation, sql) }}
     {% endcall %}
 
     {% do relations.append(target_relation) %}

core/dbt/include/global_project/macros/relations/create.sql

@@ -6,7 +6,13 @@
 
 {%- macro default__get_create_sql(relation, sql) -%}
 
-    {%- if relation.is_materialized_view -%}
+    {%- if relation.is_view -%}
+        {{ get_create_view_as_sql(relation, sql) }}
+
+    {%- elif relation.is_table -%}
+        {{ get_create_table_as_sql(False, relation, sql) }}
+
+    {%- elif relation.is_materialized_view -%}
         {{ get_create_materialized_view_as_sql(relation, sql) }}
 
     {%- else -%}

core/dbt/parser/generic_test_builders.py

@@ -101,6 +101,7 @@ class TestBuilder(Generic[Testable]):
         "error_if",
         "fail_calc",
         "store_failures",
+        "store_failures_as",
         "meta",
         "database",
         "schema",
@@ -242,6 +243,10 @@ def severity(self) -> Optional[str]:
     def store_failures(self) -> Optional[bool]:
         return self.config.get("store_failures")
 
+    @property
+    def store_failures_as(self) -> Optional[bool]:
+        return self.config.get("store_failures_as")
+
     @property
     def where(self) -> Optional[str]:
         return self.config.get("where")
@@ -294,6 +299,8 @@ def get_static_config(self):
             config["fail_calc"] = self.fail_calc
         if self.store_failures is not None:
             config["store_failures"] = self.store_failures
+        if self.store_failures_as is not None:
+            config["store_failures_as"] = self.store_failures_as
         if self.meta is not None:
             config["meta"] = self.meta
         if self.database is not None:

tests/adapter/dbt/tests/adapter/store_test_failures_tests/_files.py

@@ -0,0 +1,150 @@
+SEED__CHIPMUNKS = """
+name,shirt
+alvin,red
+simon,blue
+theodore,green
+dave,
+""".strip()
+
+
+MODEL__CHIPMUNKS = """
+{{ config(materialized='table') }}
+select *
+from {{ ref('chipmunks_stage') }}
+"""
+
+
+TEST__VIEW_TRUE = """
+{{ config(store_failures_as="view", store_failures=True) }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__VIEW_FALSE = """
+{{ config(store_failures_as="view", store_failures=False) }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__VIEW_UNSET = """
+{{ config(store_failures_as="view") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__TABLE_TRUE = """
+{{ config(store_failures_as="table", store_failures=True) }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__TABLE_FALSE = """
+{{ config(store_failures_as="table", store_failures=False) }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__TABLE_UNSET = """
+{{ config(store_failures_as="table") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__EPHEMERAL_TRUE = """
+{{ config(store_failures_as="ephemeral", store_failures=True) }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__EPHEMERAL_FALSE = """
+{{ config(store_failures_as="ephemeral", store_failures=False) }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__EPHEMERAL_UNSET = """
+{{ config(store_failures_as="ephemeral") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__UNSET_TRUE = """
+{{ config(store_failures=True) }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__UNSET_FALSE = """
+{{ config(store_failures=False) }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__UNSET_UNSET = """
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__VIEW_UNSET_PASS = """
+{{ config(store_failures_as="view") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'purple'
+"""
+
+
+TEST__ERROR_UNSET = """
+{{ config(store_failures_as="error") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+SCHEMA_YML = """
+version: 2
+
+models:
+  - name: chipmunks
+    columns:
+      - name: name
+        tests:
+          - not_null:
+              store_failures_as: view
+          - accepted_values:
+              store_failures: false
+              store_failures_as: table
+              values:
+                - alvin
+                - simon
+                - theodore
+      - name: shirt
+        tests:
+          - not_null:
+              store_failures: true
+              store_failures_as: view
+"""