Feature/materialized views/adap 608

.changes/unreleased/Features-20230629-033005.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Establish framework for materialized views and materialization change management
+time: 2023-06-29T03:30:05.527325-04:00
+custom:
+  Author: mikealfare
+  Issue: "6911"

.flake8

@@ -10,3 +10,5 @@ ignore =
     E741
     E501 # long line checking is done in black
 exclude = test/
+per-file-ignores =
+    */__init__.py: F401

core/dbt/adapters/README.md

@@ -28,3 +28,19 @@ Defines various interfaces for various adapter objects. Helps mypy correctly res
 ## `reference_keys.py`
 
 Configures naming scheme for cache elements to be universal.
+
+## Validation
+
+- `ValidationMixin`
+- `ValidationRule`
+
+These classes live in `validation.py`, outside of `relation` because they don't pertain specifically to `Relation`.
+However, they are only currently used by `Relation`.
+`ValidationMixin` provides optional validation mechanics that can be applied to either `Relation`, `RelationComponent`,
+or `RelationChange` subclasses. To implement `ValidationMixin`, include it as a subclass in your `Relation`-like
+object and add a method `validation_rules()` that returns a set of `ValidationRule` objects.
+A `ValidationRule` is a combination of a `validation_check`, something that should always evaluate to `True`
+in expected scenarios (i.e. a `False` is an invalid configuration), and an optional `validation_error`,
+an instance of `DbtRuntimeError` that should be raised in the event the `validation_check` fails.
+While optional, it's recommended that the `validation_error` be provided for clearer transparency to the end user
+as the default does not know why the `validation_check` failed.

core/dbt/adapters/base/__init__.py

@@ -1,19 +1,17 @@
-# these are all just exports, #noqa them so flake8 will be happy
-
 # TODO: Should we still include this in the `adapters` namespace?
-from dbt.contracts.connection import Credentials  # noqa: F401
-from dbt.adapters.base.meta import available  # noqa: F401
-from dbt.adapters.base.connections import BaseConnectionManager  # noqa: F401
-from dbt.adapters.base.relation import (  # noqa: F401
+from dbt.contracts.connection import Credentials
+from dbt.adapters.base.meta import available
+from dbt.adapters.base.connections import BaseConnectionManager
+from dbt.adapters.base.relation import (
     BaseRelation,
     RelationType,
     SchemaSearchMap,
 )
-from dbt.adapters.base.column import Column  # noqa: F401
-from dbt.adapters.base.impl import (  # noqa: F401
+from dbt.adapters.base.column import Column
+from dbt.adapters.base.impl import (
     AdapterConfig,
     BaseAdapter,
     PythonJobHelper,
     ConstraintSupport,
 )
-from dbt.adapters.base.plugin import AdapterPlugin  # noqa: F401
+from dbt.adapters.base.plugin import AdapterPlugin

core/dbt/adapters/base/impl.py

@@ -3,8 +3,8 @@
 from contextlib import contextmanager
 from datetime import datetime
 from enum import Enum
-import time
 from itertools import chain
+import time
 from typing import (
     Any,
     Callable,
@@ -20,11 +20,46 @@
     Union,
 )
 
-from dbt.contracts.graph.nodes import ColumnLevelConstraint, ConstraintType, ModelLevelConstraint
-
 import agate
 import pytz
 
+from dbt import deprecations
+from dbt.adapters.base import Credentials, Column as BaseColumn
+from dbt.adapters.base.connections import AdapterResponse, Connection
+from dbt.adapters.base.meta import AdapterMeta, available
+from dbt.adapters.base.relation import (
+    ComponentName,
+    BaseRelation,
+    InformationSchema,
+    SchemaSearchMap,
+)
+from dbt.adapters.cache import RelationsCache, _make_ref_key_dict
+from dbt.adapters.materialization import MaterializationFactory
+from dbt.adapters.materialization.models import Materialization
+from dbt.adapters.protocol import AdapterConfig, ConnectionManagerProtocol
+from dbt.adapters.relation import RelationFactory
+from dbt.adapters.relation.models import Relation as RelationModel, RelationChangeset, RelationRef
+from dbt.clients.agate_helper import empty_table, merge_tables, table_from_rows
+from dbt.clients.jinja import MacroGenerator
+from dbt.contracts.graph.manifest import MacroManifest, Manifest
+from dbt.contracts.graph.nodes import (
+    ColumnLevelConstraint,
+    ConstraintType,
+    ModelLevelConstraint,
+    ParsedNode,
+    ResultNode,
+)
+from dbt.contracts.relation import RelationType
+from dbt.events.functions import fire_event, warn_or_error
+from dbt.events.types import (
+    CacheMiss,
+    CatalogGenerationError,
+    CodeExecution,
+    CodeExecutionStatus,
+    ConstraintNotEnforced,
+    ConstraintNotSupported,
+    ListRelations,
+)
 from dbt.exceptions import (
     DbtInternalError,
     DbtRuntimeError,
@@ -42,36 +77,8 @@
     UnexpectedNonTimestampError,
     UnexpectedNullError,
 )
+from dbt.utils import AttrDict, cast_to_str, filter_null_values, executor
 
-from dbt.adapters.protocol import AdapterConfig, ConnectionManagerProtocol
-from dbt.clients.agate_helper import empty_table, merge_tables, table_from_rows
-from dbt.clients.jinja import MacroGenerator
-from dbt.contracts.graph.manifest import Manifest, MacroManifest
-from dbt.contracts.graph.nodes import ResultNode
-from dbt.events.functions import fire_event, warn_or_error
-from dbt.events.types import (
-    CacheMiss,
-    ListRelations,
-    CodeExecution,
-    CodeExecutionStatus,
-    CatalogGenerationError,
-    ConstraintNotSupported,
-    ConstraintNotEnforced,
-)
-from dbt.utils import filter_null_values, executor, cast_to_str, AttrDict
-
-from dbt.adapters.base.connections import Connection, AdapterResponse
-from dbt.adapters.base.meta import AdapterMeta, available
-from dbt.adapters.base.relation import (
-    ComponentName,
-    BaseRelation,
-    InformationSchema,
-    SchemaSearchMap,
-)
-from dbt.adapters.base import Column as BaseColumn
-from dbt.adapters.base import Credentials
-from dbt.adapters.cache import RelationsCache, _make_ref_key_dict
-from dbt import deprecations
 
 GET_CATALOG_MACRO_NAME = "get_catalog"
 FRESHNESS_MACRO_NAME = "collect_freshness"
@@ -222,6 +229,26 @@ class BaseAdapter(metaclass=AdapterMeta):
         ConstraintType.foreign_key: ConstraintSupport.ENFORCED,
     }
 
+    @property
+    def relation_factory(self) -> RelationFactory:
+        """
+        It's common to overwrite `Relation` instances in an adapter. In those cases this method
+        should also be overridden to register those new `Relation` instances.
+        """
+        return RelationFactory()
+
+    @property
+    def materialization_factory(self) -> MaterializationFactory:
+        """
+        It's common to overwrite `Relation` instances in an adapter. In those cases `self.relation_factory`
+        should also be overridden to register those new `Relation` instances. Take the adapter's setting
+        to override the default in `MaterializationFactory`.
+
+        It's uncommon to overwrite `Materialization` instances. In those cases the adapter should
+        override this method to override the default in `MaterializationFactory`.
+        """
+        return MaterializationFactory(relation_factory=self.relation_factory)
+
     def __init__(self, config):
         self.config = config
         self.cache = RelationsCache()
@@ -1177,7 +1204,7 @@ def post_model_hook(self, config: Mapping[str, Any], context: Any) -> None:
         available in the materialization context). It should be considered
         read-only.
 
-        The second parameter is the value returned by pre_mdoel_hook.
+        The second parameter is the value returned by pre_model_hook.
         """
         pass
 
@@ -1429,6 +1456,146 @@ def render_model_constraint(cls, constraint: ModelLevelConstraint) -> Optional[s
         else:
             return None
 
+    """
+    Pass-through methods to access `MaterializationFactory` and `RelationFactory` functionality
+    """
+
+    @available
+    def make_materialization_from_node(self, node: ParsedNode) -> Materialization:
+        """
+        Produce a `Materialization` instance along with whatever associated `Relation` and `RelationRef`
+        instances are needed.
+
+        *Note:* The node that comes in could be any one of `ParsedNode`, `CompiledNode`, or `ModelNode`. We
+        need at least a `ParsedNode` to process a materialization in general, and at least a `CompiledNode`
+        to process a materialization that requires a query.
+
+        Args:
+            node: `model` or `config.model` in the global jinja context
+
+        Returns:
+            a `Materialization` instance that contains all the information required to execute the materialization
+        """
+        existing_relation_ref = self._get_existing_relation_ref_from_node(node)
+        return self.materialization_factory.make_from_node(node, existing_relation_ref)
+
+    def _get_existing_relation_ref_from_node(self, node: ParsedNode) -> Optional[RelationRef]:
+        """
+        We need to get `existing_relation_ref` from `Adapter` because we need access to a bunch of `cache`
+        things, in particular `get_relations`.
+
+        TODO: if we refactor the interaction between `Adapter` and `cache`, the calculation of `existing_relation_ref`
+        could be moved here, which is a more intuitive spot (like `target_relation`) for it
+        (and removes the concern of creating a `RelationRef` from `Adapter` where it doesn't belong
+        """
+        existing_base_relation: BaseRelation = self.get_relation(
+            database=node.database,
+            schema=node.schema,
+            identifier=node.identifier,
+        )
+
+        # mypy thinks existing_base_relation's identifiers are all optional because of IncludePolicy
+        if existing_base_relation:
+            existing_relation_ref = self.relation_factory.make_ref(
+                name=existing_base_relation.identifier,  # type: ignore
+                schema_name=existing_base_relation.schema,  # type: ignore
+                database_name=existing_base_relation.database,  # type: ignore
+                relation_type=existing_base_relation.type,  # type: ignore
+            )
+        else:
+            existing_relation_ref = None
+
+        return existing_relation_ref
+
+    @available
+    def make_changeset(
+        self, existing_relation: RelationModel, target_relation: RelationModel
+    ) -> RelationChangeset:
+        """
+        Generate a changeset between two relations. This gets used in macros like `alter_template()` to
+        determine what changes are needed, or if a full refresh is needed.
+
+        Note that while this could be determined on `Materialization`, the fact that it gets called
+        in `alter_template()`, which only takes in two `Relation` instances, means that it needs to live
+        separately from `Materialization`.
+
+        Args:
+            existing_relation: the current implementation of the relation in the database
+            target_relation: the new implementation that should exist in the database going forward
+
+        Returns:
+            a `RelationChangeset` instance that collects all the changes required to turn `existing_relation`
+            into `target_relation`
+        """
+        return self.relation_factory.make_changeset(existing_relation, target_relation)
+
+    """
+    Implementation of cache methods for `Relation` instances (versus `BaseRelation` instances)
+    """
+
+    @available
+    def cache_created_relation_model(self, relation: RelationModel) -> str:
+        base_relation = self.base_relation_from_relation_model(relation)
+        return self.cache_added(base_relation)
+
+    @available
+    def cache_dropped_relation_model(self, relation: RelationModel) -> str:
+        base_relation = self.base_relation_from_relation_model(relation)
+        return self.cache_dropped(base_relation)
+
+    @available
+    def cache_renamed_relation_model(self, relation: RelationModel, new_name: str) -> str:
+        from_relation = self.base_relation_from_relation_model(relation)
+        to_relation = from_relation.incorporate(path={"identifier": new_name})
+        return self.cache_renamed(from_relation, to_relation)
+
+    """
+    Methods to swap back and forth between `Relation` and `BaseRelation` instances
+    """
+
+    @available
+    def is_base_relation(self, relation: Union[BaseRelation, RelationModel]) -> bool:
+        """
+        Convenient for templating, given the mix of `BaseRelation` and `Relation`
+        """
+        return isinstance(relation, BaseRelation)
+
+    @available
+    def is_relation_model(self, relation: Union[BaseRelation, RelationModel]) -> bool:
+        """
+        Convenient for templating, given the mix of `BaseRelation` and `Relation`
+        """
+        return isinstance(relation, RelationModel)
+
+    @available
+    def base_relation_from_relation_model(self, relation: RelationModel) -> BaseRelation:
+        """
+        Produce a `BaseRelation` instance from a `Relation` instance. This is primarily done to
+        reuse existing functionality based on `BaseRelation` while working with `Relation` instances.
+
+        Useful in combination with `is_relation_model`/`is_base_relation`
+
+        Args:
+            relation: a `Relation` instance or subclass to be converted
+
+        Returns:
+            a converted `BaseRelation` instance
+        """
+        try:
+            relation_type = RelationType(relation.type)
+        except ValueError:
+            relation_type = RelationType.External
+
+        base_relation: BaseRelation = self.Relation.create(
+            database=relation.database_name,
+            schema=relation.schema_name,
+            identifier=relation.name,
+            quote_policy=self.relation_factory.render_policy.quote_policy,
+            type=relation_type,
+        )
+        assert isinstance(base_relation, BaseRelation)  # mypy
+        return base_relation
+
 
 COLUMNS_EQUAL_SQL = """
 with diff_count as (