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/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,42 @@
     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, models as materialization_models
+from dbt.adapters.protocol import AdapterConfig, ConnectionManagerProtocol
+from dbt.adapters.relation import RelationFactory, models as relation_models
+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,
+    ResultNode,
+)
+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 +73,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 +225,14 @@ class BaseAdapter(metaclass=AdapterMeta):
         ConstraintType.foreign_key: ConstraintSupport.ENFORCED,
     }
 
+    @property
+    def relation_factory(self):
+        return RelationFactory(relation_models={})
+
+    @property
+    def materialization_factory(self):
+        return MaterializationFactory(relation_factory=self.relation_factory)
+
     def __init__(self, config):
         self.config = config
         self.cache = RelationsCache()
@@ -1168,7 +1179,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
 
@@ -1420,6 +1431,148 @@ 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_runtime_config(
+        self, runtime_config, materialization_type: materialization_models.MaterializationType
+    ) -> materialization_models.Materialization:
+        """
+        Produce a `Materialization` instance along with whatever associated `Relation` and `RelationRef`
+        instances are needed.
+
+        Args:
+            runtime_config: the `config` (`RuntimeConfigObject`) in the global jinja context
+            materialization_type: the name of the materialization
+
+        Returns:
+            a `Materialization` instance that contains all the information required to execute the materialization
+        """
+        existing_relation_ref = self._get_existing_relation_ref_from_model_node(
+            runtime_config.model
+        )
+
+        materialization = self.materialization_factory.make_from_runtime_config(
+            runtime_config, materialization_type, existing_relation_ref
+        )
+
+        return materialization
+
+    def _get_existing_relation_ref_from_model_node(
+        self, model_node
+    ) -> Optional[relation_models.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=model_node.database,
+            schema=model_node.schema,
+            identifier=model_node.identifier,
+        )
+
+        if existing_base_relation:
+            existing_relation_ref = self.relation_factory.make_ref(
+                name=existing_base_relation.identifier,
+                schema_name=existing_base_relation.schema,
+                database_name=existing_base_relation.database,
+                relation_type=existing_base_relation.type,
+            )
+        else:
+            existing_relation_ref = None
+
+        return existing_relation_ref
+
+    @available
+    def make_changeset(
+        self,
+        existing_relation: relation_models.Relation,
+        target_relation: relation_models.Relation,
+    ) -> relation_models.RelationChangeset:
+        """
+
+        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: relation_models.Relation) -> str:
+        base_relation = self.base_relation_from_relation_model(relation)
+        return self.cache_added(base_relation)
+
+    @available
+    def cache_dropped_relation_model(self, relation: relation_models.Relation) -> str:
+        base_relation = self.base_relation_from_relation_model(relation)
+        return self.cache_dropped(base_relation)
+
+    @available
+    def cache_renamed_relation_model(
+        self, relation: relation_models.Relation, 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, relation_models.Relation]) -> bool:
+        """
+        Convenient for templating, given the mix of `BaseRelation` and `Relation`
+        """
+        return isinstance(relation, BaseRelation)
+
+    @available
+    def is_relation_model(self, relation: Union[BaseRelation, relation_models.Relation]) -> bool:
+        """
+        Convenient for templating, given the mix of `BaseRelation` and `Relation`
+        """
+        return isinstance(relation, relation_models.Relation)
+
+    @available
+    def base_relation_from_relation_model(
+        self, relation: relation_models.Relation
+    ) -> 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
+        """
+        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 (