diff --git a/.changes/unreleased/Under the Hood-20230718-200942.yaml b/.changes/unreleased/Under the Hood-20230718-200942.yaml
new file mode 100644
index 00000000000..4f60c45a861
--- /dev/null
+++ b/.changes/unreleased/Under the Hood-20230718-200942.yaml	
@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Refactor jinja templates and relation models in support of new relation framework
+time: 2023-07-18T20:09:42.008424-04:00
+custom:
+  Author: mikealfare
+  Issue: "8128"
diff --git a/.flake8 b/.flake8
index e39b2fa4646..26e20a5d209 100644
--- a/.flake8
+++ b/.flake8
@@ -10,3 +10,5 @@ ignore =
     E741
     E501 # long line checking is done in black
 exclude = test/
+per-file-ignores =
+    */__init__.py: F401
diff --git a/core/dbt/adapters/base/impl.py b/core/dbt/adapters/base/impl.py
index d18c9af7f50..bc709d06135 100644
--- a/core/dbt/adapters/base/impl.py
+++ b/core/dbt/adapters/base/impl.py
@@ -1,5 +1,5 @@
 import abc
-from concurrent.futures import as_completed, Future
+from concurrent.futures import Future, as_completed
 from contextlib import contextmanager
 from datetime import datetime
 from enum import Enum
@@ -20,11 +20,36 @@
     Union,
 )
 
-from dbt.contracts.graph.nodes import ColumnLevelConstraint, ConstraintType, ModelLevelConstraint
-
 import agate
 import pytz
 
+from dbt import deprecations
+from dbt.adapters.base import Column as BaseColumn, Credentials
+from dbt.adapters.base.connections import AdapterResponse, Connection
+from dbt.adapters.base.meta import AdapterMeta, available
+from dbt.adapters.base.relation import BaseRelation, InformationSchema, SchemaSearchMap
+from dbt.adapters.cache import RelationsCache, _make_ref_key_dict
+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 (
+    ColumnLevelConstraint,
+    ConstraintType,
+    ModelLevelConstraint,
+    ResultNode,
+)
+from dbt.contracts.relation import ComponentName
+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 +67,8 @@
     UnexpectedNonTimestampError,
     UnexpectedNullError,
 )
-
-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"
@@ -857,7 +854,7 @@ def quote(cls, identifier: str) -> str:
 
     @available
     def quote_as_configured(self, identifier: str, quote_key: str) -> str:
-        """Quote or do not quote the given identifer as configured in the
+        """Quote or do not quote the given identifier as configured in the
         project config for the quote key.
 
         The quote key should be one of 'database' (on bigquery, 'profile'),
diff --git a/core/dbt/adapters/base/relation.py b/core/dbt/adapters/base/relation.py
index ae4e585d524..6dec9d11361 100644
--- a/core/dbt/adapters/base/relation.py
+++ b/core/dbt/adapters/base/relation.py
@@ -1,25 +1,24 @@
 from collections.abc import Hashable
 from dataclasses import dataclass, field
-from typing import Optional, TypeVar, Any, Type, Dict, Iterator, Tuple, Set
+from typing import Any, Dict, Iterator, Optional, Set, Tuple, Type, TypeVar
 
-from dbt.contracts.graph.nodes import SourceDefinition, ManifestNode, ResultNode, ParsedNode
+from dbt.contracts.graph.nodes import ManifestNode, ParsedNode, ResultNode, SourceDefinition
 from dbt.contracts.relation import (
-    RelationType,
     ComponentName,
-    HasQuoting,
     FakeAPIObject,
-    Policy,
+    HasQuoting,
     Path,
+    Policy,
+    RelationType,
 )
+import dbt.exceptions
 from dbt.exceptions import (
     ApproximateMatchError,
     DbtInternalError,
     MultipleDatabasesNotAllowedError,
 )
 from dbt.node_types import NodeType
-from dbt.utils import filter_null_values, deep_merge, classproperty
-
-import dbt.exceptions
+from dbt.utils import classproperty, deep_merge, filter_null_values
 
 
 Self = TypeVar("Self", bound="BaseRelation")
@@ -31,7 +30,8 @@ class BaseRelation(FakeAPIObject, Hashable):
     type: Optional[RelationType] = None
     quote_character: str = '"'
     # Python 3.11 requires that these use default_factory instead of simple default
-    # ValueError: mutable default <class 'dbt.contracts.relation.Policy'> for field include_policy is not allowed: use default_factory
+    # ValueError: mutable default <class 'dbt.contracts.relation.Policy'> for field include_policy is not allowed:
+    # use default_factory
     include_policy: Policy = field(default_factory=lambda: Policy())
     quote_policy: Policy = field(default_factory=lambda: Policy())
     dbt_created: bool = False
@@ -170,13 +170,14 @@ def without_identifier(self) -> "BaseRelation":
 
     def _render_iterator(self) -> Iterator[Tuple[Optional[ComponentName], Optional[str]]]:
 
-        for key in ComponentName:
+        for member in ComponentName:
             path_part: Optional[str] = None
-            if self.include_policy.get_part(key):
-                path_part = self.path.get_part(key)
-                if path_part is not None and self.quote_policy.get_part(key):
+            component = ComponentName(member)
+            if self.include_policy.get_part(component):
+                path_part = self.path.get_part(component)
+                if path_part is not None and self.quote_policy.get_part(component):
                     path_part = self.quoted(path_part)
-            yield key, path_part
+            yield component, path_part
 
     def render(self) -> str:
         # if there is nothing set, this will return the empty string.
@@ -259,7 +260,7 @@ def create_from(
             return cls.create_from_source(node, **kwargs)
         else:
             # Can't use ManifestNode here because of parameterized generics
-            if not isinstance(node, (ParsedNode)):
+            if not isinstance(node, ParsedNode):
                 raise DbtInternalError(
                     f"type mismatch, expected ManifestNode but got {type(node)}"
                 )
diff --git a/core/dbt/adapters/relation_configs/README.md b/core/dbt/adapters/relation/README.md
similarity index 100%
rename from core/dbt/adapters/relation_configs/README.md
rename to core/dbt/adapters/relation/README.md
diff --git a/core/dbt/adapters/relation/__init__.py b/core/dbt/adapters/relation/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/core/dbt/adapters/relation/models/__init__.py b/core/dbt/adapters/relation/models/__init__.py
new file mode 100644
index 00000000000..68a0e312025
--- /dev/null
+++ b/core/dbt/adapters/relation/models/__init__.py
@@ -0,0 +1,8 @@
+from dbt.adapters.relation.models._relation import (
+    DescribeRelationResults,
+    RelationComponent,
+)
+from dbt.adapters.relation.models._change import (
+    RelationChangeAction,
+    RelationChange,
+)
diff --git a/core/dbt/adapters/relation_configs/config_change.py b/core/dbt/adapters/relation/models/_change.py
similarity index 69%
rename from core/dbt/adapters/relation_configs/config_change.py
rename to core/dbt/adapters/relation/models/_change.py
index ac653fa5210..ad40f23f0c8 100644
--- a/core/dbt/adapters/relation_configs/config_change.py
+++ b/core/dbt/adapters/relation/models/_change.py
@@ -2,19 +2,19 @@
 from dataclasses import dataclass
 from typing import Hashable
 
-from dbt.adapters.relation_configs.config_base import RelationConfigBase
+from dbt.adapters.relation.models._relation import RelationComponent
 from dbt.dataclass_schema import StrEnum
 
 
-class RelationConfigChangeAction(StrEnum):
+class RelationChangeAction(StrEnum):
     alter = "alter"
     create = "create"
     drop = "drop"
 
 
 @dataclass(frozen=True, eq=True, unsafe_hash=True)
-class RelationConfigChange(RelationConfigBase, ABC):
-    action: RelationConfigChangeAction
+class RelationChange(RelationComponent, ABC):
+    action: RelationChangeAction
     context: Hashable  # this is usually a RelationConfig, e.g. IndexConfig, but shouldn't be limited
 
     @property
diff --git a/core/dbt/adapters/relation_configs/config_base.py b/core/dbt/adapters/relation/models/_relation.py
similarity index 80%
rename from core/dbt/adapters/relation_configs/config_base.py
rename to core/dbt/adapters/relation/models/_relation.py
index 9d0cddb0d21..449f2695ff7 100644
--- a/core/dbt/adapters/relation_configs/config_base.py
+++ b/core/dbt/adapters/relation/models/_relation.py
@@ -1,7 +1,8 @@
 from dataclasses import dataclass
-from typing import Union, Dict
+from typing import Any, Dict, Union
 
 import agate
+
 from dbt.utils import filter_null_values
 
 
@@ -18,24 +19,24 @@
     ])
 }
 """
-RelationResults = Dict[str, Union[agate.Row, agate.Table]]
+DescribeRelationResults = Dict[str, Union[agate.Row, agate.Table]]
 
 
 @dataclass(frozen=True)
-class RelationConfigBase:
+class RelationComponent:
     @classmethod
-    def from_dict(cls, kwargs_dict) -> "RelationConfigBase":
+    def from_dict(cls, config_dict: Dict[str, Any]) -> "RelationComponent":
         """
         This assumes the subclass of `RelationConfigBase` is flat, in the sense that no attribute is
         itself another subclass of `RelationConfigBase`. If that's not the case, this should be overriden
         to manually manage that complexity.
 
         Args:
-            kwargs_dict: the dict representation of this instance
+            config_dict: the dict representation of this instance
 
         Returns: the `RelationConfigBase` representation associated with the provided dict
         """
-        return cls(**filter_null_values(kwargs_dict))  # type: ignore
+        return cls(**filter_null_values(config_dict))  # type: ignore
 
     @classmethod
     def _not_implemented_error(cls) -> NotImplementedError:
diff --git a/core/dbt/adapters/relation_configs/__init__.py b/core/dbt/adapters/relation_configs/__init__.py
deleted file mode 100644
index b8c73447a68..00000000000
--- a/core/dbt/adapters/relation_configs/__init__.py
+++ /dev/null
@@ -1,12 +0,0 @@
-from dbt.adapters.relation_configs.config_base import (  # noqa: F401
-    RelationConfigBase,
-    RelationResults,
-)
-from dbt.adapters.relation_configs.config_change import (  # noqa: F401
-    RelationConfigChangeAction,
-    RelationConfigChange,
-)
-from dbt.adapters.relation_configs.config_validation import (  # noqa: F401
-    RelationConfigValidationMixin,
-    RelationConfigValidationRule,
-)
diff --git a/core/dbt/adapters/sql/impl.py b/core/dbt/adapters/sql/impl.py
index b74eb02d991..b72923504c9 100644
--- a/core/dbt/adapters/sql/impl.py
+++ b/core/dbt/adapters/sql/impl.py
@@ -1,17 +1,16 @@
 import agate
-from typing import Any, Optional, Tuple, Type, List
+from typing import Any, List, Optional, Tuple, Type
 
-from dbt.contracts.connection import Connection, AdapterResponse
-from dbt.exceptions import RelationTypeNullError
 from dbt.adapters.base import BaseAdapter, available
+from dbt.adapters.base.relation import BaseRelation
 from dbt.adapters.cache import _make_ref_key_dict
 from dbt.adapters.sql import SQLConnectionManager
+from dbt.contracts.connection import AdapterResponse, Connection
 from dbt.events.functions import fire_event
 from dbt.events.types import ColTypeChange, SchemaCreation, SchemaDrop
+from dbt.exceptions import RelationTypeNullError
 
 
-from dbt.adapters.base.relation import BaseRelation
-
 LIST_RELATIONS_MACRO_NAME = "list_relations_without_caching"
 GET_COLUMNS_IN_RELATION_MACRO_NAME = "get_columns_in_relation"
 LIST_SCHEMAS_MACRO_NAME = "list_schemas"
@@ -222,14 +221,14 @@ def check_schema_exists(self, database: str, schema: str) -> bool:
     def validate_sql(self, sql: str) -> AdapterResponse:
         """Submit the given SQL to the engine for validation, but not execution.
 
-        By default we simply prefix the query with the explain keyword and allow the
+        By default, we simply prefix the query with the explain keyword and allow the
         exceptions thrown by the underlying engine on invalid SQL inputs to bubble up
         to the exception handler. For adjustments to the explain statement - such as
         for adapters that have different mechanisms for hinting at query validation
         or dry-run - callers may be able to override the validate_sql_query macro with
         the addition of an <adapter>__validate_sql implementation.
 
-        :param sql str: The sql to validate
+        :param sql: str The sql to validate
         """
         kwargs = {
             "sql": sql,
diff --git a/core/dbt/adapters/relation_configs/config_validation.py b/core/dbt/adapters/validation.py
similarity index 92%
rename from core/dbt/adapters/relation_configs/config_validation.py
rename to core/dbt/adapters/validation.py
index 17bf74bf3e7..26c3db10f02 100644
--- a/core/dbt/adapters/relation_configs/config_validation.py
+++ b/core/dbt/adapters/validation.py
@@ -5,7 +5,7 @@
 
 
 @dataclass(frozen=True, eq=True, unsafe_hash=True)
-class RelationConfigValidationRule:
+class ValidationRule:
     validation_check: bool
     validation_error: Optional[DbtRuntimeError]
 
@@ -18,12 +18,12 @@ def default_error(self):
 
 
 @dataclass(frozen=True)
-class RelationConfigValidationMixin:
+class ValidationMixin:
     def __post_init__(self):
         self.run_validation_rules()
 
     @property
-    def validation_rules(self) -> Set[RelationConfigValidationRule]:
+    def validation_rules(self) -> Set[ValidationRule]:
         """
         A set of validation rules to run against the object upon creation.
 
diff --git a/core/dbt/include/global_project/macros/adapters/columns.sql b/core/dbt/include/global_project/macros/adapters/columns.sql
deleted file mode 100644
index b5a03ec53b5..00000000000
--- a/core/dbt/include/global_project/macros/adapters/columns.sql
+++ /dev/null
@@ -1,132 +0,0 @@
-{% macro get_columns_in_relation(relation) -%}
-  {{ return(adapter.dispatch('get_columns_in_relation', 'dbt')(relation)) }}
-{% endmacro %}
-
-{% macro default__get_columns_in_relation(relation) -%}
-  {{ exceptions.raise_not_implemented(
-    'get_columns_in_relation macro not implemented for adapter '+adapter.type()) }}
-{% endmacro %}
-
-{# helper for adapter-specific implementations of get_columns_in_relation #}
-{% macro sql_convert_columns_in_relation(table) -%}
-  {% set columns = [] %}
-  {% for row in table %}
-    {% do columns.append(api.Column(*row)) %}
-  {% endfor %}
-  {{ return(columns) }}
-{% endmacro %}
-
-
-{% macro get_empty_subquery_sql(select_sql, select_sql_header=none) -%}
-  {{ return(adapter.dispatch('get_empty_subquery_sql', 'dbt')(select_sql, select_sql_header)) }}
-{% endmacro %}
-
-{#
-  Builds a query that results in the same schema as the given select_sql statement, without necessitating a data scan.
-  Useful for running a query in a 'pre-flight' context, such as model contract enforcement (assert_columns_equivalent macro).
-#}
-{% macro default__get_empty_subquery_sql(select_sql, select_sql_header=none) %}
-    {%- if select_sql_header is not none -%}
-    {{ select_sql_header }}
-    {%- endif -%}
-    select * from (
-        {{ select_sql }}
-    ) as __dbt_sbq
-    where false
-    limit 0
-{% endmacro %}
-
-
-{% macro get_empty_schema_sql(columns) -%}
-  {{ return(adapter.dispatch('get_empty_schema_sql', 'dbt')(columns)) }}
-{% endmacro %}
-
-{% macro default__get_empty_schema_sql(columns) %}
-    {%- set col_err = [] -%}
-    select
-    {% for i in columns %}
-      {%- set col = columns[i] -%}
-      {%- if col['data_type'] is not defined -%}
-        {{ col_err.append(col['name']) }}
-      {%- endif -%}
-      {% set col_name = adapter.quote(col['name']) if col.get('quote') else col['name'] %}
-      cast(null as {{ col['data_type'] }}) as {{ col_name }}{{ ", " if not loop.last }}
-    {%- endfor -%}
-    {%- if (col_err | length) > 0 -%}
-      {{ exceptions.column_type_missing(column_names=col_err) }}
-    {%- endif -%}
-{% endmacro %}
-
-{% macro get_column_schema_from_query(select_sql, select_sql_header=none) -%}
-    {% set columns = [] %}
-    {# -- Using an 'empty subquery' here to get the same schema as the given select_sql statement, without necessitating a data scan.#}
-    {% set sql = get_empty_subquery_sql(select_sql, select_sql_header) %}
-    {% set column_schema = adapter.get_column_schema_from_query(sql) %}
-    {{ return(column_schema) }}
-{% endmacro %}
-
--- here for back compat
-{% macro get_columns_in_query(select_sql) -%}
-  {{ return(adapter.dispatch('get_columns_in_query', 'dbt')(select_sql)) }}
-{% endmacro %}
-
-{% macro default__get_columns_in_query(select_sql) %}
-    {% call statement('get_columns_in_query', fetch_result=True, auto_begin=False) -%}
-        {{ get_empty_subquery_sql(select_sql) }}
-    {% endcall %}
-    {{ return(load_result('get_columns_in_query').table.columns | map(attribute='name') | list) }}
-{% endmacro %}
-
-{% macro alter_column_type(relation, column_name, new_column_type) -%}
-  {{ return(adapter.dispatch('alter_column_type', 'dbt')(relation, column_name, new_column_type)) }}
-{% endmacro %}
-
-{% macro default__alter_column_type(relation, column_name, new_column_type) -%}
-  {#
-    1. Create a new column (w/ temp name and correct type)
-    2. Copy data over to it
-    3. Drop the existing column (cascade!)
-    4. Rename the new column to existing column
-  #}
-  {%- set tmp_column = column_name + "__dbt_alter" -%}
-
-  {% call statement('alter_column_type') %}
-    alter table {{ relation }} add column {{ adapter.quote(tmp_column) }} {{ new_column_type }};
-    update {{ relation }} set {{ adapter.quote(tmp_column) }} = {{ adapter.quote(column_name) }};
-    alter table {{ relation }} drop column {{ adapter.quote(column_name) }} cascade;
-    alter table {{ relation }} rename column {{ adapter.quote(tmp_column) }} to {{ adapter.quote(column_name) }}
-  {% endcall %}
-
-{% endmacro %}
-
-
-{% macro alter_relation_add_remove_columns(relation, add_columns = none, remove_columns = none) -%}
-  {{ return(adapter.dispatch('alter_relation_add_remove_columns', 'dbt')(relation, add_columns, remove_columns)) }}
-{% endmacro %}
-
-{% macro default__alter_relation_add_remove_columns(relation, add_columns, remove_columns) %}
-
-  {% if add_columns is none %}
-    {% set add_columns = [] %}
-  {% endif %}
-  {% if remove_columns is none %}
-    {% set remove_columns = [] %}
-  {% endif %}
-
-  {% set sql -%}
-
-     alter {{ relation.type }} {{ relation }}
-
-            {% for column in add_columns %}
-               add column {{ column.name }} {{ column.data_type }}{{ ',' if not loop.last }}
-            {% endfor %}{{ ',' if add_columns and remove_columns }}
-
-            {% for column in remove_columns %}
-                drop column {{ column.name }}{{ ',' if not loop.last }}
-            {% endfor %}
-
-  {%- endset -%}
-
-  {% do run_query(sql) %}
-
-{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/drop_relation.sql b/core/dbt/include/global_project/macros/adapters/drop_relation.sql
deleted file mode 100644
index bd254c78d51..00000000000
--- a/core/dbt/include/global_project/macros/adapters/drop_relation.sql
+++ /dev/null
@@ -1,44 +0,0 @@
-{% macro drop_relation(relation) -%}
-    {{ return(adapter.dispatch('drop_relation', 'dbt')(relation)) }}
-{% endmacro %}
-
-{% macro default__drop_relation(relation) -%}
-    {% call statement('drop_relation', auto_begin=False) -%}
-        {%- if relation.is_table -%}
-            {{- drop_table(relation) -}}
-        {%- elif relation.is_view -%}
-            {{- drop_view(relation) -}}
-        {%- elif relation.is_materialized_view -%}
-            {{- drop_materialized_view(relation) -}}
-        {%- else -%}
-            drop {{ relation.type }} if exists {{ relation }} cascade
-        {%- endif -%}
-    {%- endcall %}
-{% endmacro %}
-
-
-{% macro drop_table(relation) -%}
-  {{ return(adapter.dispatch('drop_table', 'dbt')(relation)) }}
-{%- endmacro %}
-
-{% macro default__drop_table(relation) -%}
-    drop table if exists {{ relation }} cascade
-{%- endmacro %}
-
-
-{% macro drop_view(relation) -%}
-  {{ return(adapter.dispatch('drop_view', 'dbt')(relation)) }}
-{%- endmacro %}
-
-{% macro default__drop_view(relation) -%}
-    drop view if exists {{ relation }} cascade
-{%- endmacro %}
-
-
-{% macro drop_materialized_view(relation) -%}
-  {{ return(adapter.dispatch('drop_materialized_view', 'dbt')(relation)) }}
-{%- endmacro %}
-
-{% macro default__drop_materialized_view(relation) -%}
-    drop materialized view if exists {{ relation }} cascade
-{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/generic_test_sql/accepted_values.sql b/core/dbt/include/global_project/macros/adapters/generic_test_sql/accepted_values.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/generic_test_sql/accepted_values.sql
rename to core/dbt/include/global_project/macros/adapters/generic_test_sql/accepted_values.sql
diff --git a/core/dbt/include/global_project/macros/generic_test_sql/not_null.sql b/core/dbt/include/global_project/macros/adapters/generic_test_sql/not_null.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/generic_test_sql/not_null.sql
rename to core/dbt/include/global_project/macros/adapters/generic_test_sql/not_null.sql
diff --git a/core/dbt/include/global_project/macros/generic_test_sql/relationships.sql b/core/dbt/include/global_project/macros/adapters/generic_test_sql/relationships.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/generic_test_sql/relationships.sql
rename to core/dbt/include/global_project/macros/adapters/generic_test_sql/relationships.sql
diff --git a/core/dbt/include/global_project/macros/generic_test_sql/unique.sql b/core/dbt/include/global_project/macros/adapters/generic_test_sql/unique.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/generic_test_sql/unique.sql
rename to core/dbt/include/global_project/macros/adapters/generic_test_sql/unique.sql
diff --git a/core/dbt/include/global_project/macros/adapters/metadata.sql b/core/dbt/include/global_project/macros/adapters/metadata.sql
deleted file mode 100644
index 9e45c500a3f..00000000000
--- a/core/dbt/include/global_project/macros/adapters/metadata.sql
+++ /dev/null
@@ -1,74 +0,0 @@
-{% macro get_catalog(information_schema, schemas) -%}
-  {{ return(adapter.dispatch('get_catalog', 'dbt')(information_schema, schemas)) }}
-{%- endmacro %}
-
-{% macro default__get_catalog(information_schema, schemas) -%}
-
-  {% set typename = adapter.type() %}
-  {% set msg -%}
-    get_catalog not implemented for {{ typename }}
-  {%- endset %}
-
-  {{ exceptions.raise_compiler_error(msg) }}
-{% endmacro %}
-
-
-{% macro information_schema_name(database) %}
-  {{ return(adapter.dispatch('information_schema_name', 'dbt')(database)) }}
-{% endmacro %}
-
-{% macro default__information_schema_name(database) -%}
-  {%- if database -%}
-    {{ database }}.INFORMATION_SCHEMA
-  {%- else -%}
-    INFORMATION_SCHEMA
-  {%- endif -%}
-{%- endmacro %}
-
-
-{% macro list_schemas(database) -%}
-  {{ return(adapter.dispatch('list_schemas', 'dbt')(database)) }}
-{% endmacro %}
-
-{% macro default__list_schemas(database) -%}
-  {% set sql %}
-    select distinct schema_name
-    from {{ information_schema_name(database) }}.SCHEMATA
-    where catalog_name ilike '{{ database }}'
-  {% endset %}
-  {{ return(run_query(sql)) }}
-{% endmacro %}
-
-
-{% macro check_schema_exists(information_schema, schema) -%}
-  {{ return(adapter.dispatch('check_schema_exists', 'dbt')(information_schema, schema)) }}
-{% endmacro %}
-
-{% macro default__check_schema_exists(information_schema, schema) -%}
-  {% set sql -%}
-        select count(*)
-        from {{ information_schema.replace(information_schema_view='SCHEMATA') }}
-        where catalog_name='{{ information_schema.database }}'
-          and schema_name='{{ schema }}'
-  {%- endset %}
-  {{ return(run_query(sql)) }}
-{% endmacro %}
-
-
-{% macro list_relations_without_caching(schema_relation) %}
-  {{ return(adapter.dispatch('list_relations_without_caching', 'dbt')(schema_relation)) }}
-{% endmacro %}
-
-{% macro default__list_relations_without_caching(schema_relation) %}
-  {{ exceptions.raise_not_implemented(
-    'list_relations_without_caching macro not implemented for adapter '+adapter.type()) }}
-{% endmacro %}
-
-{% macro get_relations() %}
-  {{ return(adapter.dispatch('get_relations', 'dbt')()) }}
-{% endmacro %}
-
-{% macro default__get_relations() %}
-  {{ exceptions.raise_not_implemented(
-    'get_relations macro not implemented for adapter '+adapter.type()) }}
-{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/freshness.sql b/core/dbt/include/global_project/macros/adapters/metadata/freshness.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/adapters/freshness.sql
rename to core/dbt/include/global_project/macros/adapters/metadata/freshness.sql
diff --git a/core/dbt/include/global_project/macros/adapters/metadata/get_catalog.sql b/core/dbt/include/global_project/macros/adapters/metadata/get_catalog.sql
new file mode 100644
index 00000000000..a7c7708e8be
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/metadata/get_catalog.sql
@@ -0,0 +1,14 @@
+{% macro get_catalog(information_schema, schemas) -%}
+  {{ return(adapter.dispatch('get_catalog', 'dbt')(information_schema, schemas)) }}
+{%- endmacro %}
+
+
+{% macro default__get_catalog(information_schema, schemas) -%}
+
+  {% set typename = adapter.type() %}
+  {% set msg -%}
+    get_catalog not implemented for {{ typename }}
+  {%- endset %}
+
+  {{ exceptions.raise_compiler_error(msg) }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/metadata/get_relations.sql b/core/dbt/include/global_project/macros/adapters/metadata/get_relations.sql
new file mode 100644
index 00000000000..523debbc13b
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/metadata/get_relations.sql
@@ -0,0 +1,9 @@
+{% macro get_relations() %}
+  {{ return(adapter.dispatch('get_relations', 'dbt')()) }}
+{% endmacro %}
+
+
+{% macro default__get_relations() %}
+  {{ exceptions.raise_not_implemented(
+    'get_relations macro not implemented for adapter '+adapter.type()) }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/metadata/information_schema_name.sql b/core/dbt/include/global_project/macros/adapters/metadata/information_schema_name.sql
new file mode 100644
index 00000000000..e84e64d8198
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/metadata/information_schema_name.sql
@@ -0,0 +1,12 @@
+{% macro information_schema_name(database) %}
+  {{ return(adapter.dispatch('information_schema_name', 'dbt')(database)) }}
+{% endmacro %}
+
+
+{% macro default__information_schema_name(database) -%}
+  {%- if database -%}
+    {{ database }}.INFORMATION_SCHEMA
+  {%- else -%}
+    INFORMATION_SCHEMA
+  {%- endif -%}
+{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/metadata/list_relations_without_caching.sql b/core/dbt/include/global_project/macros/adapters/metadata/list_relations_without_caching.sql
new file mode 100644
index 00000000000..9171e1d7538
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/metadata/list_relations_without_caching.sql
@@ -0,0 +1,9 @@
+{% macro list_relations_without_caching(schema_relation) %}
+  {{ return(adapter.dispatch('list_relations_without_caching', 'dbt')(schema_relation)) }}
+{% endmacro %}
+
+
+{% macro default__list_relations_without_caching(schema_relation) %}
+  {{ exceptions.raise_not_implemented(
+    'list_relations_without_caching macro not implemented for adapter '+adapter.type()) }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/validate_sql.sql b/core/dbt/include/global_project/macros/adapters/metadata/validate_sql.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/adapters/validate_sql.sql
rename to core/dbt/include/global_project/macros/adapters/metadata/validate_sql.sql
diff --git a/core/dbt/include/global_project/macros/adapters/persist_docs.sql b/core/dbt/include/global_project/macros/adapters/persist_docs.sql
deleted file mode 100644
index 8749e59f606..00000000000
--- a/core/dbt/include/global_project/macros/adapters/persist_docs.sql
+++ /dev/null
@@ -1,33 +0,0 @@
-{% macro alter_column_comment(relation, column_dict) -%}
-  {{ return(adapter.dispatch('alter_column_comment', 'dbt')(relation, column_dict)) }}
-{% endmacro %}
-
-{% macro default__alter_column_comment(relation, column_dict) -%}
-  {{ exceptions.raise_not_implemented(
-    'alter_column_comment macro not implemented for adapter '+adapter.type()) }}
-{% endmacro %}
-
-
-{% macro alter_relation_comment(relation, relation_comment) -%}
-  {{ return(adapter.dispatch('alter_relation_comment', 'dbt')(relation, relation_comment)) }}
-{% endmacro %}
-
-{% macro default__alter_relation_comment(relation, relation_comment) -%}
-  {{ exceptions.raise_not_implemented(
-    'alter_relation_comment macro not implemented for adapter '+adapter.type()) }}
-{% endmacro %}
-
-
-{% macro persist_docs(relation, model, for_relation=true, for_columns=true) -%}
-  {{ return(adapter.dispatch('persist_docs', 'dbt')(relation, model, for_relation, for_columns)) }}
-{% endmacro %}
-
-{% macro default__persist_docs(relation, model, for_relation, for_columns) -%}
-  {% if for_relation and config.persist_relation_docs() and model.description %}
-    {% do run_query(alter_relation_comment(relation, model.description)) %}
-  {% endif %}
-
-  {% if for_columns and config.persist_column_docs() and model.columns %}
-    {% do run_query(alter_column_comment(relation, model.columns)) %}
-  {% endif %}
-{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation.sql b/core/dbt/include/global_project/macros/adapters/relation.sql
index f0dde7f20f0..76fd3ad8031 100644
--- a/core/dbt/include/global_project/macros/adapters/relation.sql
+++ b/core/dbt/include/global_project/macros/adapters/relation.sql
@@ -32,28 +32,6 @@
 {% endmacro %}
 
 
-{% macro truncate_relation(relation) -%}
-  {{ return(adapter.dispatch('truncate_relation', 'dbt')(relation)) }}
-{% endmacro %}
-
-{% macro default__truncate_relation(relation) -%}
-  {% call statement('truncate_relation') -%}
-    truncate table {{ relation }}
-  {%- endcall %}
-{% endmacro %}
-
-
-{% macro rename_relation(from_relation, to_relation) -%}
-  {{ return(adapter.dispatch('rename_relation', 'dbt')(from_relation, to_relation)) }}
-{% endmacro %}
-
-{% macro default__rename_relation(from_relation, to_relation) -%}
-  {% set target_name = adapter.quote_as_configured(to_relation.identifier, 'identifier') %}
-  {% call statement('rename_relation') -%}
-    alter table {{ from_relation }} rename to {{ target_name }}
-  {%- endcall %}
-{% endmacro %}
-
 
 {% macro get_or_create_relation(database, schema, identifier, type) -%}
   {{ return(adapter.dispatch('get_or_create_relation', 'dbt')(database, schema, identifier, type)) }}
@@ -89,10 +67,3 @@
 {% macro load_relation(relation) %}
     {{ return(load_cached_relation(relation)) }}
 {% endmacro %}
-
-
-{% macro drop_relation_if_exists(relation) %}
-  {% if relation is not none %}
-    {{ adapter.drop_relation(relation) }}
-  {% endif %}
-{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/column/alter.sql b/core/dbt/include/global_project/macros/adapters/relation_components/column/alter.sql
new file mode 100644
index 00000000000..a6d02b47202
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/column/alter.sql
@@ -0,0 +1,55 @@
+{% macro alter_column_type(relation, column_name, new_column_type) -%}
+  {{ return(adapter.dispatch('alter_column_type', 'dbt')(relation, column_name, new_column_type)) }}
+{% endmacro %}
+
+
+{% macro default__alter_column_type(relation, column_name, new_column_type) -%}
+  {#
+    1. Create a new column (w/ temp name and correct type)
+    2. Copy data over to it
+    3. Drop the existing column (cascade!)
+    4. Rename the new column to existing column
+  #}
+  {%- set tmp_column = column_name + "__dbt_alter" -%}
+
+  {% call statement('alter_column_type') %}
+    alter table {{ relation }} add column {{ adapter.quote(tmp_column) }} {{ new_column_type }};
+    update {{ relation }} set {{ adapter.quote(tmp_column) }} = {{ adapter.quote(column_name) }};
+    alter table {{ relation }} drop column {{ adapter.quote(column_name) }} cascade;
+    alter table {{ relation }} rename column {{ adapter.quote(tmp_column) }} to {{ adapter.quote(column_name) }}
+  {% endcall %}
+
+{% endmacro %}
+
+
+{% macro alter_relation_add_remove_columns(relation, add_columns = none, remove_columns = none) -%}
+  {{ return(adapter.dispatch('alter_relation_add_remove_columns', 'dbt')(relation, add_columns, remove_columns)) }}
+{% endmacro %}
+
+
+{% macro default__alter_relation_add_remove_columns(relation, add_columns, remove_columns) %}
+
+  {% if add_columns is none %}
+    {% set add_columns = [] %}
+  {% endif %}
+  {% if remove_columns is none %}
+    {% set remove_columns = [] %}
+  {% endif %}
+
+  {% set sql -%}
+
+     alter {{ relation.type }} {{ relation }}
+
+            {% for column in add_columns %}
+               add column {{ column.name }} {{ column.data_type }}{{ ',' if not loop.last }}
+            {% endfor %}{{ ',' if add_columns and remove_columns }}
+
+            {% for column in remove_columns %}
+                drop column {{ column.name }}{{ ',' if not loop.last }}
+            {% endfor %}
+
+  {%- endset -%}
+
+  {% do run_query(sql) %}
+
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/column/describe_query.sql b/core/dbt/include/global_project/macros/adapters/relation_components/column/describe_query.sql
new file mode 100644
index 00000000000..c789a7eaac5
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/column/describe_query.sql
@@ -0,0 +1,42 @@
+{% macro get_column_schema_from_query(select_sql, select_sql_header=none) -%}
+    {% set columns = [] %}
+    {# -- Using an 'empty subquery' here to get the same schema as the given select_sql statement, without necessitating a data scan.#}
+    {% set sql = get_empty_subquery_sql(select_sql, select_sql_header) %}
+    {% set column_schema = adapter.get_column_schema_from_query(sql) %}
+    {{ return(column_schema) }}
+{% endmacro %}
+
+
+{% macro get_empty_subquery_sql(select_sql, select_sql_header=none) -%}
+  {{ return(adapter.dispatch('get_empty_subquery_sql', 'dbt')(select_sql, select_sql_header)) }}
+{% endmacro %}
+
+
+{#
+  Builds a query that results in the same schema as the given select_sql statement, without necessitating a data scan.
+  Useful for running a query in a 'pre-flight' context, such as model contract enforcement (assert_columns_equivalent macro).
+#}
+{% macro default__get_empty_subquery_sql(select_sql, select_sql_header=none) %}
+    {%- if select_sql_header is not none -%}
+    {{ select_sql_header }}
+    {%- endif -%}
+    select * from (
+        {{ select_sql }}
+    ) as __dbt_sbq
+    where false
+    limit 0
+{% endmacro %}
+
+
+-- here for back compat
+{% macro get_columns_in_query(select_sql) -%}
+  {{ return(adapter.dispatch('get_columns_in_query', 'dbt')(select_sql)) }}
+{% endmacro %}
+
+
+{% macro default__get_columns_in_query(select_sql) %}
+    {% call statement('get_columns_in_query', fetch_result=True, auto_begin=False) -%}
+        {{ get_empty_subquery_sql(select_sql) }}
+    {% endcall %}
+    {{ return(load_result('get_columns_in_query').table.columns | map(attribute='name') | list) }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/column/describe_relation.sql b/core/dbt/include/global_project/macros/adapters/relation_components/column/describe_relation.sql
new file mode 100644
index 00000000000..2ee9a2de9d6
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/column/describe_relation.sql
@@ -0,0 +1,19 @@
+{% macro get_columns_in_relation(relation) -%}
+  {{ return(adapter.dispatch('get_columns_in_relation', 'dbt')(relation)) }}
+{% endmacro %}
+
+
+{% macro default__get_columns_in_relation(relation) -%}
+  {{ exceptions.raise_not_implemented(
+    'get_columns_in_relation macro not implemented for adapter '+adapter.type()) }}
+{% endmacro %}
+
+
+{# helper for adapter-specific implementations of get_columns_in_relation #}
+{% macro sql_convert_columns_in_relation(table) -%}
+  {% set columns = [] %}
+  {% for row in table %}
+    {% do columns.append(api.Column(*row)) %}
+  {% endfor %}
+  {{ return(columns) }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/column/empty_schema.sql b/core/dbt/include/global_project/macros/adapters/relation_components/column/empty_schema.sql
new file mode 100644
index 00000000000..f92a71d1a04
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/column/empty_schema.sql
@@ -0,0 +1,20 @@
+{% macro get_empty_schema_sql(columns) -%}
+  {{ return(adapter.dispatch('get_empty_schema_sql', 'dbt')(columns)) }}
+{% endmacro %}
+
+
+{% macro default__get_empty_schema_sql(columns) %}
+    {%- set col_err = [] -%}
+    select
+    {% for i in columns %}
+      {%- set col = columns[i] -%}
+      {%- if col['data_type'] is not defined -%}
+        {{ col_err.append(col['name']) }}
+      {%- endif -%}
+      {% set col_name = adapter.quote(col['name']) if col.get('quote') else col['name'] %}
+      cast(null as {{ col['data_type'] }}) as {{ col_name }}{{ ", " if not loop.last }}
+    {%- endfor -%}
+    {%- if (col_err | length) > 0 -%}
+      {{ exceptions.column_type_missing(column_names=col_err) }}
+    {%- endif -%}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/comment/alter_column_comment.sql b/core/dbt/include/global_project/macros/adapters/relation_components/comment/alter_column_comment.sql
new file mode 100644
index 00000000000..16b95fbfe7d
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/comment/alter_column_comment.sql
@@ -0,0 +1,9 @@
+{% macro alter_column_comment(relation, column_dict) -%}
+  {{ return(adapter.dispatch('alter_column_comment', 'dbt')(relation, column_dict)) }}
+{% endmacro %}
+
+
+{% macro default__alter_column_comment(relation, column_dict) -%}
+  {{ exceptions.raise_not_implemented(
+    'alter_column_comment macro not implemented for adapter '+adapter.type()) }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/comment/alter_relation_comment.sql b/core/dbt/include/global_project/macros/adapters/relation_components/comment/alter_relation_comment.sql
new file mode 100644
index 00000000000..13801286650
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/comment/alter_relation_comment.sql
@@ -0,0 +1,9 @@
+{% macro alter_relation_comment(relation, relation_comment) -%}
+  {{ return(adapter.dispatch('alter_relation_comment', 'dbt')(relation, relation_comment)) }}
+{% endmacro %}
+
+
+{% macro default__alter_relation_comment(relation, relation_comment) -%}
+  {{ exceptions.raise_not_implemented(
+    'alter_relation_comment macro not implemented for adapter '+adapter.type()) }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/apply_grants.sql b/core/dbt/include/global_project/macros/adapters/relation_components/grant/apply.sql
similarity index 67%
rename from core/dbt/include/global_project/macros/adapters/apply_grants.sql
rename to core/dbt/include/global_project/macros/adapters/relation_components/grant/apply.sql
index 10906e7ffa7..e39890e65a1 100644
--- a/core/dbt/include/global_project/macros/adapters/apply_grants.sql
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/grant/apply.sql
@@ -1,24 +1,6 @@
 {# ------- BOOLEAN MACROS --------- #}
 
-{#
-  -- COPY GRANTS
-  -- When a relational object (view or table) is replaced in this database,
-  -- do previous grants carry over to the new object? This may depend on:
-  --    whether we use alter-rename-swap versus CREATE OR REPLACE
-  --    user-supplied configuration (e.g. copy_grants on Snowflake)
-  -- By default, play it safe, assume TRUE: that grants ARE copied over.
-  -- This means dbt will first "show" current grants and then calculate diffs.
-  -- It may require an additional query than is strictly necessary,
-  -- but better safe than sorry.
-#}
 
-{% macro copy_grants() %}
-    {{ return(adapter.dispatch('copy_grants', 'dbt')()) }}
-{% endmacro %}
-
-{% macro default__copy_grants() %}
-    {{ return(True) }}
-{% endmacro %}
 
 
 {#
@@ -39,49 +21,6 @@
 {%- endmacro -%}
 
 
-{% macro should_revoke(existing_relation, full_refresh_mode=True) %}
-
-    {% if not existing_relation %}
-        {#-- The table doesn't already exist, so no grants to copy over --#}
-        {{ return(False) }}
-    {% elif full_refresh_mode %}
-        {#-- The object is being REPLACED -- whether grants are copied over depends on the value of user config --#}
-        {{ return(copy_grants()) }}
-    {% else %}
-        {#-- The table is being merged/upserted/inserted -- grants will be carried over --#}
-        {{ return(True) }}
-    {% endif %}
-
-{% endmacro %}
-
-{# ------- DCL STATEMENT TEMPLATES --------- #}
-
-{% macro get_show_grant_sql(relation) %}
-    {{ return(adapter.dispatch("get_show_grant_sql", "dbt")(relation)) }}
-{% endmacro %}
-
-{% macro default__get_show_grant_sql(relation) %}
-    show grants on {{ relation }}
-{% endmacro %}
-
-
-{% macro get_grant_sql(relation, privilege, grantees) %}
-    {{ return(adapter.dispatch('get_grant_sql', 'dbt')(relation, privilege, grantees)) }}
-{% endmacro %}
-
-{%- macro default__get_grant_sql(relation, privilege, grantees) -%}
-    grant {{ privilege }} on {{ relation }} to {{ grantees | join(', ') }}
-{%- endmacro -%}
-
-
-{% macro get_revoke_sql(relation, privilege, grantees) %}
-    {{ return(adapter.dispatch('get_revoke_sql', 'dbt')(relation, privilege, grantees)) }}
-{% endmacro %}
-
-{%- macro default__get_revoke_sql(relation, privilege, grantees) -%}
-    revoke {{ privilege }} on {{ relation }} from {{ grantees | join(', ') }}
-{%- endmacro -%}
-
 
 {# ------- RUNTIME APPLICATION --------- #}
 
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/grant/copy.sql b/core/dbt/include/global_project/macros/adapters/relation_components/grant/copy.sql
new file mode 100644
index 00000000000..fb0bacc827b
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/grant/copy.sql
@@ -0,0 +1,20 @@
+{#
+  -- COPY GRANTS
+  -- When a relational object (view or table) is replaced in this database,
+  -- do previous grants carry over to the new object? This may depend on:
+  --    whether we use alter-rename-swap versus CREATE OR REPLACE
+  --    user-supplied configuration (e.g. copy_grants on Snowflake)
+  -- By default, play it safe, assume TRUE: that grants ARE copied over.
+  -- This means dbt will first "show" current grants and then calculate diffs.
+  -- It may require an additional query than is strictly necessary,
+  -- but better safe than sorry.
+#}
+
+{% macro copy_grants() %}
+    {{ return(adapter.dispatch('copy_grants', 'dbt')()) }}
+{% endmacro %}
+
+
+{% macro default__copy_grants() %}
+    {{ return(True) }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/grant/grant.sql b/core/dbt/include/global_project/macros/adapters/relation_components/grant/grant.sql
new file mode 100644
index 00000000000..341c3bf925a
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/grant/grant.sql
@@ -0,0 +1,8 @@
+{% macro get_grant_sql(relation, privilege, grantees) %}
+    {{ return(adapter.dispatch('get_grant_sql', 'dbt')(relation, privilege, grantees)) }}
+{% endmacro %}
+
+
+{%- macro default__get_grant_sql(relation, privilege, grantees) -%}
+    grant {{ privilege }} on {{ relation }} to {{ grantees | join(', ') }}
+{%- endmacro -%}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/grant/revoke.sql b/core/dbt/include/global_project/macros/adapters/relation_components/grant/revoke.sql
new file mode 100644
index 00000000000..2253a1d46e6
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/grant/revoke.sql
@@ -0,0 +1,24 @@
+{% macro get_revoke_sql(relation, privilege, grantees) %}
+    {{ return(adapter.dispatch('get_revoke_sql', 'dbt')(relation, privilege, grantees)) }}
+{% endmacro %}
+
+
+{%- macro default__get_revoke_sql(relation, privilege, grantees) -%}
+    revoke {{ privilege }} on {{ relation }} from {{ grantees | join(', ') }}
+{%- endmacro -%}
+
+
+{% macro should_revoke(existing_relation, full_refresh_mode=True) %}
+
+    {% if not existing_relation %}
+        {#-- The table doesn't already exist, so no grants to copy over --#}
+        {{ return(False) }}
+    {% elif full_refresh_mode %}
+        {#-- The object is being REPLACED -- whether grants are copied over depends on the value of user config --#}
+        {{ return(copy_grants()) }}
+    {% else %}
+        {#-- The table is being merged/upserted/inserted -- grants will be carried over --#}
+        {{ return(True) }}
+    {% endif %}
+
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/grant/show.sql b/core/dbt/include/global_project/macros/adapters/relation_components/grant/show.sql
new file mode 100644
index 00000000000..ed3fe160563
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/grant/show.sql
@@ -0,0 +1,8 @@
+{% macro get_show_grant_sql(relation) %}
+    {{ return(adapter.dispatch("get_show_grant_sql", "dbt")(relation)) }}
+{% endmacro %}
+
+
+{% macro default__get_show_grant_sql(relation) %}
+    show grants on {{ relation }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/indexes.sql b/core/dbt/include/global_project/macros/adapters/relation_components/index/create.sql
similarity index 52%
rename from core/dbt/include/global_project/macros/adapters/indexes.sql
rename to core/dbt/include/global_project/macros/adapters/relation_components/index/create.sql
index b8663a7f971..dc058563612 100644
--- a/core/dbt/include/global_project/macros/adapters/indexes.sql
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/index/create.sql
@@ -2,6 +2,7 @@
   {{ return(adapter.dispatch('get_create_index_sql', 'dbt')(relation, index_dict)) }}
 {% endmacro %}
 
+
 {% macro default__get_create_index_sql(relation, index_dict) -%}
   {% do return(None) %}
 {% endmacro %}
@@ -11,6 +12,7 @@
   {{ adapter.dispatch('create_indexes', 'dbt')(relation) }}
 {%- endmacro %}
 
+
 {% macro default__create_indexes(relation) -%}
   {%- set _indexes = config.get('indexes', default=[]) -%}
 
@@ -21,21 +23,3 @@
     {% endif %}
   {% endfor %}
 {% endmacro %}
-
-
-{% macro get_drop_index_sql(relation, index_name) -%}
-    {{ adapter.dispatch('get_drop_index_sql', 'dbt')(relation, index_name) }}
-{%- endmacro %}
-
-{% macro default__get_drop_index_sql(relation, index_name) -%}
-    {{ exceptions.raise_compiler_error("`get_drop_index_sql has not been implemented for this adapter.") }}
-{%- endmacro %}
-
-
-{% macro get_show_indexes_sql(relation) -%}
-    {{ adapter.dispatch('get_show_indexes_sql', 'dbt')(relation) }}
-{%- endmacro %}
-
-{% macro default__get_show_indexes_sql(relation) -%}
-    {{ exceptions.raise_compiler_error("`get_show_indexes_sql has not been implemented for this adapter.") }}
-{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/index/drop.sql b/core/dbt/include/global_project/macros/adapters/relation_components/index/drop.sql
new file mode 100644
index 00000000000..033eef75414
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/index/drop.sql
@@ -0,0 +1,8 @@
+{% macro get_drop_index_sql(relation, index_name) -%}
+    {{ adapter.dispatch('get_drop_index_sql', 'dbt')(relation, index_name) }}
+{%- endmacro %}
+
+
+{% macro default__get_drop_index_sql(relation, index_name) -%}
+    {{ exceptions.raise_compiler_error("`get_drop_index_sql has not been implemented for this adapter.") }}
+{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/index/show.sql b/core/dbt/include/global_project/macros/adapters/relation_components/index/show.sql
new file mode 100644
index 00000000000..0166486e21c
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/index/show.sql
@@ -0,0 +1,8 @@
+{% macro get_show_indexes_sql(relation) -%}
+    {{ adapter.dispatch('get_show_indexes_sql', 'dbt')(relation) }}
+{%- endmacro %}
+
+
+{% macro default__get_show_indexes_sql(relation) -%}
+    {{ exceptions.raise_compiler_error("`get_show_indexes_sql has not been implemented for this adapter.") }}
+{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/schema.sql b/core/dbt/include/global_project/macros/adapters/relation_components/schema/create.sql
similarity index 50%
rename from core/dbt/include/global_project/macros/adapters/schema.sql
rename to core/dbt/include/global_project/macros/adapters/relation_components/schema/create.sql
index 9e0c7559286..ab3782ab32e 100644
--- a/core/dbt/include/global_project/macros/adapters/schema.sql
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/schema/create.sql
@@ -2,19 +2,9 @@
   {{ adapter.dispatch('create_schema', 'dbt')(relation) }}
 {% endmacro %}
 
+
 {% macro default__create_schema(relation) -%}
   {%- call statement('create_schema') -%}
     create schema if not exists {{ relation.without_identifier() }}
   {% endcall %}
 {% endmacro %}
-
-
-{% macro drop_schema(relation) -%}
-  {{ adapter.dispatch('drop_schema', 'dbt')(relation) }}
-{% endmacro %}
-
-{% macro default__drop_schema(relation) -%}
-  {%- call statement('drop_schema') -%}
-    drop schema if exists {{ relation.without_identifier() }} cascade
-  {% endcall %}
-{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/schema/drop.sql b/core/dbt/include/global_project/macros/adapters/relation_components/schema/drop.sql
new file mode 100644
index 00000000000..fc51b594830
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/schema/drop.sql
@@ -0,0 +1,10 @@
+{% macro drop_schema(relation) -%}
+  {{ adapter.dispatch('drop_schema', 'dbt')(relation) }}
+{% endmacro %}
+
+
+{% macro default__drop_schema(relation) -%}
+  {%- call statement('drop_schema') -%}
+    drop schema if exists {{ relation.without_identifier() }} cascade
+  {% endcall %}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relation_components/schema/show.sql b/core/dbt/include/global_project/macros/adapters/relation_components/schema/show.sql
new file mode 100644
index 00000000000..db285ba16fd
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relation_components/schema/show.sql
@@ -0,0 +1,29 @@
+{% macro list_schemas(database) -%}
+  {{ return(adapter.dispatch('list_schemas', 'dbt')(database)) }}
+{% endmacro %}
+
+
+{% macro default__list_schemas(database) -%}
+  {% set sql %}
+    select distinct schema_name
+    from {{ information_schema_name(database) }}.SCHEMATA
+    where catalog_name ilike '{{ database }}'
+  {% endset %}
+  {{ return(run_query(sql)) }}
+{% endmacro %}
+
+
+{% macro check_schema_exists(information_schema, schema) -%}
+  {{ return(adapter.dispatch('check_schema_exists', 'dbt')(information_schema, schema)) }}
+{% endmacro %}
+
+
+{% macro default__check_schema_exists(information_schema, schema) -%}
+  {% set sql -%}
+        select count(*)
+        from {{ information_schema.replace(information_schema_view='SCHEMATA') }}
+        where catalog_name='{{ information_schema.database }}'
+          and schema_name='{{ schema }}'
+  {%- endset %}
+  {{ return(run_query(sql)) }}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/materializations/models/clone/create_or_replace_clone.sql b/core/dbt/include/global_project/macros/adapters/relations/clone/create.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/clone/create_or_replace_clone.sql
rename to core/dbt/include/global_project/macros/adapters/relations/clone/create.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/materialized_view/alter_materialized_view.sql b/core/dbt/include/global_project/macros/adapters/relations/materialized_view/alter.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/materialized_view/alter_materialized_view.sql
rename to core/dbt/include/global_project/macros/adapters/relations/materialized_view/alter.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/materialized_view/create_materialized_view.sql b/core/dbt/include/global_project/macros/adapters/relations/materialized_view/create.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/materialized_view/create_materialized_view.sql
rename to core/dbt/include/global_project/macros/adapters/relations/materialized_view/create.sql
diff --git a/core/dbt/include/global_project/macros/adapters/relations/materialized_view/drop.sql b/core/dbt/include/global_project/macros/adapters/relations/materialized_view/drop.sql
new file mode 100644
index 00000000000..c10a4993112
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relations/materialized_view/drop.sql
@@ -0,0 +1,8 @@
+{% macro drop_materialized_view(relation) -%}
+  {{ return(adapter.dispatch('drop_materialized_view', 'dbt')(relation)) }}
+{%- endmacro %}
+
+
+{% macro default__drop_materialized_view(relation) -%}
+    drop materialized view if exists {{ relation }} cascade
+{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/materializations/models/materialized_view/get_materialized_view_configuration_changes.sql b/core/dbt/include/global_project/macros/adapters/relations/materialized_view/get_configuration_changes.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/materialized_view/get_materialized_view_configuration_changes.sql
rename to core/dbt/include/global_project/macros/adapters/relations/materialized_view/get_configuration_changes.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/materialized_view/refresh_materialized_view.sql b/core/dbt/include/global_project/macros/adapters/relations/materialized_view/refresh.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/materialized_view/refresh_materialized_view.sql
rename to core/dbt/include/global_project/macros/adapters/relations/materialized_view/refresh.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/materialized_view/replace_materialized_view.sql b/core/dbt/include/global_project/macros/adapters/relations/materialized_view/replace.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/materialized_view/replace_materialized_view.sql
rename to core/dbt/include/global_project/macros/adapters/relations/materialized_view/replace.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/table/columns_spec_ddl.sql b/core/dbt/include/global_project/macros/adapters/relations/table/columns_spec_ddl.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/table/columns_spec_ddl.sql
rename to core/dbt/include/global_project/macros/adapters/relations/table/columns_spec_ddl.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/table/create_table_as.sql b/core/dbt/include/global_project/macros/adapters/relations/table/create.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/table/create_table_as.sql
rename to core/dbt/include/global_project/macros/adapters/relations/table/create.sql
diff --git a/core/dbt/include/global_project/macros/adapters/relations/table/drop.sql b/core/dbt/include/global_project/macros/adapters/relations/table/drop.sql
new file mode 100644
index 00000000000..9f4ac674e81
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relations/table/drop.sql
@@ -0,0 +1,8 @@
+{% macro drop_table(relation) -%}
+  {{ return(adapter.dispatch('drop_table', 'dbt')(relation)) }}
+{%- endmacro %}
+
+
+{% macro default__drop_table(relation) -%}
+    drop table if exists {{ relation }} cascade
+{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/materializations/models/view/create_or_replace_view.sql b/core/dbt/include/global_project/macros/adapters/relations/view/create.sql
similarity index 68%
rename from core/dbt/include/global_project/macros/materializations/models/view/create_or_replace_view.sql
rename to core/dbt/include/global_project/macros/adapters/relations/view/create.sql
index 2846a322e82..84b13488d9f 100644
--- a/core/dbt/include/global_project/macros/materializations/models/view/create_or_replace_view.sql
+++ b/core/dbt/include/global_project/macros/adapters/relations/view/create.sql
@@ -42,3 +42,32 @@
   {{ return({'relations': [target_relation]}) }}
 
 {% endmacro %}
+
+
+{% macro get_create_view_as_sql(relation, sql) -%}
+  {{ adapter.dispatch('get_create_view_as_sql', 'dbt')(relation, sql) }}
+{%- endmacro %}
+
+{% macro default__get_create_view_as_sql(relation, sql) -%}
+  {{ return(create_view_as(relation, sql)) }}
+{% endmacro %}
+
+
+/* {# keep logic under old name for backwards compatibility #} */
+{% macro create_view_as(relation, sql) -%}
+  {{ adapter.dispatch('create_view_as', 'dbt')(relation, sql) }}
+{%- endmacro %}
+
+{% macro default__create_view_as(relation, sql) -%}
+  {%- set sql_header = config.get('sql_header', none) -%}
+
+  {{ sql_header if sql_header is not none }}
+  create view {{ relation }}
+    {% set contract_config = config.get('contract') %}
+    {% if contract_config.enforced %}
+      {{ get_assert_columns_equivalent(sql) }}
+    {%- endif %}
+  as (
+    {{ sql }}
+  );
+{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/adapters/relations/view/drop.sql b/core/dbt/include/global_project/macros/adapters/relations/view/drop.sql
new file mode 100644
index 00000000000..3049a0b08fe
--- /dev/null
+++ b/core/dbt/include/global_project/macros/adapters/relations/view/drop.sql
@@ -0,0 +1,8 @@
+{% macro drop_view(relation) -%}
+  {{ return(adapter.dispatch('drop_view', 'dbt')(relation)) }}
+{%- endmacro %}
+
+
+{% macro default__drop_view(relation) -%}
+    drop view if exists {{ relation }} cascade
+{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/materializations/models/view/helpers.sql b/core/dbt/include/global_project/macros/adapters/relations/view/helpers.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/view/helpers.sql
rename to core/dbt/include/global_project/macros/adapters/relations/view/helpers.sql
diff --git a/core/dbt/include/global_project/macros/utils/any_value.sql b/core/dbt/include/global_project/macros/adapters/utils/any_value.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/any_value.sql
rename to core/dbt/include/global_project/macros/adapters/utils/any_value.sql
diff --git a/core/dbt/include/global_project/macros/utils/array_append.sql b/core/dbt/include/global_project/macros/adapters/utils/array_append.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/array_append.sql
rename to core/dbt/include/global_project/macros/adapters/utils/array_append.sql
diff --git a/core/dbt/include/global_project/macros/utils/array_concat.sql b/core/dbt/include/global_project/macros/adapters/utils/array_concat.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/array_concat.sql
rename to core/dbt/include/global_project/macros/adapters/utils/array_concat.sql
diff --git a/core/dbt/include/global_project/macros/utils/array_construct.sql b/core/dbt/include/global_project/macros/adapters/utils/array_construct.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/array_construct.sql
rename to core/dbt/include/global_project/macros/adapters/utils/array_construct.sql
diff --git a/core/dbt/include/global_project/macros/utils/bool_or.sql b/core/dbt/include/global_project/macros/adapters/utils/bool_or.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/bool_or.sql
rename to core/dbt/include/global_project/macros/adapters/utils/bool_or.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/clone/can_clone_table.sql b/core/dbt/include/global_project/macros/adapters/utils/can_clone_table.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/clone/can_clone_table.sql
rename to core/dbt/include/global_project/macros/adapters/utils/can_clone_table.sql
diff --git a/core/dbt/include/global_project/macros/utils/cast_bool_to_text.sql b/core/dbt/include/global_project/macros/adapters/utils/cast_bool_to_text.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/cast_bool_to_text.sql
rename to core/dbt/include/global_project/macros/adapters/utils/cast_bool_to_text.sql
diff --git a/core/dbt/include/global_project/macros/utils/concat.sql b/core/dbt/include/global_project/macros/adapters/utils/concat.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/concat.sql
rename to core/dbt/include/global_project/macros/adapters/utils/concat.sql
diff --git a/core/dbt/include/global_project/macros/utils/data_types.sql b/core/dbt/include/global_project/macros/adapters/utils/data_types.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/data_types.sql
rename to core/dbt/include/global_project/macros/adapters/utils/data_types.sql
diff --git a/core/dbt/include/global_project/macros/utils/date_trunc.sql b/core/dbt/include/global_project/macros/adapters/utils/date_trunc.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/date_trunc.sql
rename to core/dbt/include/global_project/macros/adapters/utils/date_trunc.sql
diff --git a/core/dbt/include/global_project/macros/utils/dateadd.sql b/core/dbt/include/global_project/macros/adapters/utils/dateadd.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/dateadd.sql
rename to core/dbt/include/global_project/macros/adapters/utils/dateadd.sql
diff --git a/core/dbt/include/global_project/macros/utils/datediff.sql b/core/dbt/include/global_project/macros/adapters/utils/datediff.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/datediff.sql
rename to core/dbt/include/global_project/macros/adapters/utils/datediff.sql
diff --git a/core/dbt/include/global_project/macros/etc/datetime.sql b/core/dbt/include/global_project/macros/adapters/utils/datetime.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/etc/datetime.sql
rename to core/dbt/include/global_project/macros/adapters/utils/datetime.sql
diff --git a/core/dbt/include/global_project/macros/utils/escape_single_quotes.sql b/core/dbt/include/global_project/macros/adapters/utils/escape_single_quotes.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/escape_single_quotes.sql
rename to core/dbt/include/global_project/macros/adapters/utils/escape_single_quotes.sql
diff --git a/core/dbt/include/global_project/macros/utils/except.sql b/core/dbt/include/global_project/macros/adapters/utils/except.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/except.sql
rename to core/dbt/include/global_project/macros/adapters/utils/except.sql
diff --git a/core/dbt/include/global_project/macros/utils/hash.sql b/core/dbt/include/global_project/macros/adapters/utils/hash.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/hash.sql
rename to core/dbt/include/global_project/macros/adapters/utils/hash.sql
diff --git a/core/dbt/include/global_project/macros/utils/intersect.sql b/core/dbt/include/global_project/macros/adapters/utils/intersect.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/intersect.sql
rename to core/dbt/include/global_project/macros/adapters/utils/intersect.sql
diff --git a/core/dbt/include/global_project/macros/utils/last_day.sql b/core/dbt/include/global_project/macros/adapters/utils/last_day.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/last_day.sql
rename to core/dbt/include/global_project/macros/adapters/utils/last_day.sql
diff --git a/core/dbt/include/global_project/macros/utils/length.sql b/core/dbt/include/global_project/macros/adapters/utils/length.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/length.sql
rename to core/dbt/include/global_project/macros/adapters/utils/length.sql
diff --git a/core/dbt/include/global_project/macros/utils/listagg.sql b/core/dbt/include/global_project/macros/adapters/utils/listagg.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/listagg.sql
rename to core/dbt/include/global_project/macros/adapters/utils/listagg.sql
diff --git a/core/dbt/include/global_project/macros/utils/literal.sql b/core/dbt/include/global_project/macros/adapters/utils/literal.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/literal.sql
rename to core/dbt/include/global_project/macros/adapters/utils/literal.sql
diff --git a/core/dbt/include/global_project/macros/utils/position.sql b/core/dbt/include/global_project/macros/adapters/utils/position.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/position.sql
rename to core/dbt/include/global_project/macros/adapters/utils/position.sql
diff --git a/core/dbt/include/global_project/macros/utils/replace.sql b/core/dbt/include/global_project/macros/adapters/utils/replace.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/replace.sql
rename to core/dbt/include/global_project/macros/adapters/utils/replace.sql
diff --git a/core/dbt/include/global_project/macros/utils/right.sql b/core/dbt/include/global_project/macros/adapters/utils/right.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/right.sql
rename to core/dbt/include/global_project/macros/adapters/utils/right.sql
diff --git a/core/dbt/include/global_project/macros/utils/safe_cast.sql b/core/dbt/include/global_project/macros/adapters/utils/safe_cast.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/safe_cast.sql
rename to core/dbt/include/global_project/macros/adapters/utils/safe_cast.sql
diff --git a/core/dbt/include/global_project/macros/utils/split_part.sql b/core/dbt/include/global_project/macros/adapters/utils/split_part.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/utils/split_part.sql
rename to core/dbt/include/global_project/macros/adapters/utils/split_part.sql
diff --git a/core/dbt/include/global_project/macros/adapters/timestamps.sql b/core/dbt/include/global_project/macros/adapters/utils/timestamps.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/adapters/timestamps.sql
rename to core/dbt/include/global_project/macros/adapters/utils/timestamps.sql
diff --git a/core/dbt/include/global_project/macros/materializations/configs.sql b/core/dbt/include/global_project/macros/materializations/_utils/configs.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/configs.sql
rename to core/dbt/include/global_project/macros/materializations/_utils/configs.sql
diff --git a/core/dbt/include/global_project/macros/materializations/hooks.sql b/core/dbt/include/global_project/macros/materializations/_utils/hooks.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/hooks.sql
rename to core/dbt/include/global_project/macros/materializations/_utils/hooks.sql
diff --git a/core/dbt/include/global_project/macros/materializations/_utils/persist_docs.sql b/core/dbt/include/global_project/macros/materializations/_utils/persist_docs.sql
new file mode 100644
index 00000000000..a3a33fc791d
--- /dev/null
+++ b/core/dbt/include/global_project/macros/materializations/_utils/persist_docs.sql
@@ -0,0 +1,14 @@
+
+{% macro persist_docs(relation, model, for_relation=true, for_columns=true) -%}
+  {{ return(adapter.dispatch('persist_docs', 'dbt')(relation, model, for_relation, for_columns)) }}
+{% endmacro %}
+
+{% macro default__persist_docs(relation, model, for_relation, for_columns) -%}
+  {% if for_relation and config.persist_relation_docs() and model.description %}
+    {% do run_query(alter_relation_comment(relation, model.description)) %}
+  {% endif %}
+
+  {% if for_columns and config.persist_column_docs() and model.columns %}
+    {% do run_query(alter_column_comment(relation, model.columns)) %}
+  {% endif %}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/materializations/models/clone/clone.sql b/core/dbt/include/global_project/macros/materializations/clone.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/clone/clone.sql
rename to core/dbt/include/global_project/macros/materializations/clone.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/incremental/column_helpers.sql b/core/dbt/include/global_project/macros/materializations/incremental/column_helpers.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/incremental/column_helpers.sql
rename to core/dbt/include/global_project/macros/materializations/incremental/column_helpers.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/incremental/incremental.sql b/core/dbt/include/global_project/macros/materializations/incremental/incremental.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/incremental/incremental.sql
rename to core/dbt/include/global_project/macros/materializations/incremental/incremental.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/incremental/is_incremental.sql b/core/dbt/include/global_project/macros/materializations/incremental/is_incremental.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/incremental/is_incremental.sql
rename to core/dbt/include/global_project/macros/materializations/incremental/is_incremental.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/incremental/merge.sql b/core/dbt/include/global_project/macros/materializations/incremental/merge.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/incremental/merge.sql
rename to core/dbt/include/global_project/macros/materializations/incremental/merge.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/incremental/on_schema_change.sql b/core/dbt/include/global_project/macros/materializations/incremental/on_schema_change.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/incremental/on_schema_change.sql
rename to core/dbt/include/global_project/macros/materializations/incremental/on_schema_change.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/incremental/strategies.sql b/core/dbt/include/global_project/macros/materializations/incremental/strategies.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/incremental/strategies.sql
rename to core/dbt/include/global_project/macros/materializations/incremental/strategies.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/materialized_view/materialized_view.sql b/core/dbt/include/global_project/macros/materializations/materialized_view.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/materialized_view/materialized_view.sql
rename to core/dbt/include/global_project/macros/materializations/materialized_view.sql
index 015f6cb8585..94b5d936dab 100644
--- a/core/dbt/include/global_project/macros/materializations/models/materialized_view/materialized_view.sql
+++ b/core/dbt/include/global_project/macros/materializations/materialized_view.sql
@@ -22,33 +22,6 @@
 {% endmaterialization %}
 
 
-{% macro materialized_view_setup(backup_relation, intermediate_relation, pre_hooks) %}
-
-    -- backup_relation and intermediate_relation should not already exist in the database
-    -- it's possible these exist because of a previous run that exited unexpectedly
-    {% set preexisting_backup_relation = load_cached_relation(backup_relation) %}
-    {% set preexisting_intermediate_relation = load_cached_relation(intermediate_relation) %}
-
-    -- drop the temp relations if they exist already in the database
-    {{ drop_relation_if_exists(preexisting_backup_relation) }}
-    {{ drop_relation_if_exists(preexisting_intermediate_relation) }}
-
-    {{ run_hooks(pre_hooks, inside_transaction=False) }}
-
-{% endmacro %}
-
-
-{% macro materialized_view_teardown(backup_relation, intermediate_relation, post_hooks) %}
-
-    -- drop the temp relations if they exist to leave the database clean for the next run
-    {{ drop_relation_if_exists(backup_relation) }}
-    {{ drop_relation_if_exists(intermediate_relation) }}
-
-    {{ run_hooks(post_hooks, inside_transaction=False) }}
-
-{% endmacro %}
-
-
 {% macro materialized_view_get_build_sql(existing_relation, target_relation, backup_relation, intermediate_relation) %}
 
     {% set full_refresh_mode = should_full_refresh() %}
@@ -88,6 +61,33 @@
 {% endmacro %}
 
 
+{% macro materialized_view_setup(backup_relation, intermediate_relation, pre_hooks) %}
+
+    -- backup_relation and intermediate_relation should not already exist in the database
+    -- it's possible these exist because of a previous run that exited unexpectedly
+    {% set preexisting_backup_relation = load_cached_relation(backup_relation) %}
+    {% set preexisting_intermediate_relation = load_cached_relation(intermediate_relation) %}
+
+    -- drop the temp relations if they exist already in the database
+    {{ drop_relation_if_exists(preexisting_backup_relation) }}
+    {{ drop_relation_if_exists(preexisting_intermediate_relation) }}
+
+    {{ run_hooks(pre_hooks, inside_transaction=False) }}
+
+{% endmacro %}
+
+
+{% macro materialized_view_teardown(backup_relation, intermediate_relation, post_hooks) %}
+
+    -- drop the temp relations if they exist to leave the database clean for the next run
+    {{ drop_relation_if_exists(backup_relation) }}
+    {{ drop_relation_if_exists(intermediate_relation) }}
+
+    {{ run_hooks(post_hooks, inside_transaction=False) }}
+
+{% endmacro %}
+
+
 {% macro materialized_view_execute_no_op(target_relation) %}
     {% do store_raw_result(
         name="main",
diff --git a/core/dbt/include/global_project/macros/materializations/models/view/create_view_as.sql b/core/dbt/include/global_project/macros/materializations/models/view/create_view_as.sql
deleted file mode 100644
index 41cd196c310..00000000000
--- a/core/dbt/include/global_project/macros/materializations/models/view/create_view_as.sql
+++ /dev/null
@@ -1,27 +0,0 @@
-{% macro get_create_view_as_sql(relation, sql) -%}
-  {{ adapter.dispatch('get_create_view_as_sql', 'dbt')(relation, sql) }}
-{%- endmacro %}
-
-{% macro default__get_create_view_as_sql(relation, sql) -%}
-  {{ return(create_view_as(relation, sql)) }}
-{% endmacro %}
-
-
-/* {# keep logic under old name for backwards compatibility #} */
-{% macro create_view_as(relation, sql) -%}
-  {{ adapter.dispatch('create_view_as', 'dbt')(relation, sql) }}
-{%- endmacro %}
-
-{% macro default__create_view_as(relation, sql) -%}
-  {%- set sql_header = config.get('sql_header', none) -%}
-
-  {{ sql_header if sql_header is not none }}
-  create view {{ relation }}
-    {% set contract_config = config.get('contract') %}
-    {% if contract_config.enforced %}
-      {{ get_assert_columns_equivalent(sql) }}
-    {%- endif %}
-  as (
-    {{ sql }}
-  );
-{%- endmacro %}
diff --git a/core/dbt/include/global_project/macros/materializations/seeds/helpers.sql b/core/dbt/include/global_project/macros/materializations/seed/helpers.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/seeds/helpers.sql
rename to core/dbt/include/global_project/macros/materializations/seed/helpers.sql
diff --git a/core/dbt/include/global_project/macros/materializations/seeds/seed.sql b/core/dbt/include/global_project/macros/materializations/seed/seed.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/seeds/seed.sql
rename to core/dbt/include/global_project/macros/materializations/seed/seed.sql
diff --git a/core/dbt/include/global_project/macros/materializations/snapshots/helpers.sql b/core/dbt/include/global_project/macros/materializations/snapshot/helpers.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/snapshots/helpers.sql
rename to core/dbt/include/global_project/macros/materializations/snapshot/helpers.sql
diff --git a/core/dbt/include/global_project/macros/materializations/snapshots/snapshot.sql b/core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/snapshots/snapshot.sql
rename to core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql
diff --git a/core/dbt/include/global_project/macros/materializations/snapshots/snapshot_merge.sql b/core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/snapshots/snapshot_merge.sql
rename to core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql
diff --git a/core/dbt/include/global_project/macros/materializations/snapshots/strategies.sql b/core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/snapshots/strategies.sql
rename to core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/table/table.sql b/core/dbt/include/global_project/macros/materializations/table.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/table/table.sql
rename to core/dbt/include/global_project/macros/materializations/table.sql
diff --git a/core/dbt/include/global_project/macros/materializations/tests/helpers.sql b/core/dbt/include/global_project/macros/materializations/test/helpers.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/tests/helpers.sql
rename to core/dbt/include/global_project/macros/materializations/test/helpers.sql
diff --git a/core/dbt/include/global_project/macros/materializations/tests/test.sql b/core/dbt/include/global_project/macros/materializations/test/test.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/tests/test.sql
rename to core/dbt/include/global_project/macros/materializations/test/test.sql
diff --git a/core/dbt/include/global_project/macros/materializations/tests/where_subquery.sql b/core/dbt/include/global_project/macros/materializations/test/where_subquery.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/tests/where_subquery.sql
rename to core/dbt/include/global_project/macros/materializations/test/where_subquery.sql
diff --git a/core/dbt/include/global_project/macros/materializations/models/view/view.sql b/core/dbt/include/global_project/macros/materializations/view.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/materializations/models/view/view.sql
rename to core/dbt/include/global_project/macros/materializations/view.sql
diff --git a/core/dbt/include/global_project/macros/relations/drop.sql b/core/dbt/include/global_project/macros/relations/drop.sql
new file mode 100644
index 00000000000..79def8ffef2
--- /dev/null
+++ b/core/dbt/include/global_project/macros/relations/drop.sql
@@ -0,0 +1,25 @@
+{% macro drop_relation(relation) -%}
+    {{ return(adapter.dispatch('drop_relation', 'dbt')(relation)) }}
+{% endmacro %}
+
+
+{% macro default__drop_relation(relation) -%}
+    {% call statement('drop_relation', auto_begin=False) -%}
+        {%- if relation.is_table -%}
+            {{- drop_table(relation) -}}
+        {%- elif relation.is_view -%}
+            {{- drop_view(relation) -}}
+        {%- elif relation.is_materialized_view -%}
+            {{- drop_materialized_view(relation) -}}
+        {%- else -%}
+            drop {{ relation.type }} if exists {{ relation }} cascade
+        {%- endif -%}
+    {%- endcall %}
+{% endmacro %}
+
+
+{% macro drop_relation_if_exists(relation) %}
+  {% if relation is not none %}
+    {{ adapter.drop_relation(relation) }}
+  {% endif %}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/relations/rename.sql b/core/dbt/include/global_project/macros/relations/rename.sql
new file mode 100644
index 00000000000..ddb5aa214ea
--- /dev/null
+++ b/core/dbt/include/global_project/macros/relations/rename.sql
@@ -0,0 +1,11 @@
+{% macro rename_relation(from_relation, to_relation) -%}
+  {{ return(adapter.dispatch('rename_relation', 'dbt')(from_relation, to_relation)) }}
+{% endmacro %}
+
+
+{% macro default__rename_relation(from_relation, to_relation) -%}
+  {% set target_name = adapter.quote_as_configured(to_relation.identifier, 'identifier') %}
+  {% call statement('rename_relation') -%}
+    alter table {{ from_relation }} rename to {{ target_name }}
+  {%- endcall %}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/relations/truncate.sql b/core/dbt/include/global_project/macros/relations/truncate.sql
new file mode 100644
index 00000000000..146c1e5c800
--- /dev/null
+++ b/core/dbt/include/global_project/macros/relations/truncate.sql
@@ -0,0 +1,10 @@
+{% macro truncate_relation(relation) -%}
+  {{ return(adapter.dispatch('truncate_relation', 'dbt')(relation)) }}
+{% endmacro %}
+
+
+{% macro default__truncate_relation(relation) -%}
+  {% call statement('truncate_relation') -%}
+    truncate table {{ relation }}
+  {%- endcall %}
+{% endmacro %}
diff --git a/core/dbt/include/global_project/macros/etc/statement.sql b/core/dbt/include/global_project/macros/statement.sql
similarity index 100%
rename from core/dbt/include/global_project/macros/etc/statement.sql
rename to core/dbt/include/global_project/macros/statement.sql
diff --git a/plugins/postgres/dbt/adapters/postgres/relation.py b/plugins/postgres/dbt/adapters/postgres/relation/__init__.py
similarity index 61%
rename from plugins/postgres/dbt/adapters/postgres/relation.py
rename to plugins/postgres/dbt/adapters/postgres/relation/__init__.py
index 43822efb11f..d7958af5873 100644
--- a/plugins/postgres/dbt/adapters/postgres/relation.py
+++ b/plugins/postgres/dbt/adapters/postgres/relation/__init__.py
@@ -2,19 +2,16 @@
 from typing import Optional, Set, FrozenSet
 
 from dbt.adapters.base.relation import BaseRelation
-from dbt.adapters.relation_configs import (
-    RelationConfigChangeAction,
-    RelationResults,
-)
+from dbt.adapters.relation.models import DescribeRelationResults, RelationChangeAction
 from dbt.context.providers import RuntimeConfigObject
 from dbt.exceptions import DbtRuntimeError
 
-from dbt.adapters.postgres.relation_configs import (
-    PostgresIndexConfig,
-    PostgresIndexConfigChange,
-    PostgresMaterializedViewConfig,
-    PostgresMaterializedViewConfigChangeCollection,
+from dbt.adapters.postgres.relation.models import (
     MAX_CHARACTERS_IN_IDENTIFIER,
+    PostgresIndexRelation,
+    PostgresIndexRelationChange,
+    PostgresMaterializedViewRelation,
+    PostgresMaterializedViewRelationChangeset,
 )
 
 
@@ -36,19 +33,17 @@ def __post_init__(self):
     def relation_max_name_length(self):
         return MAX_CHARACTERS_IN_IDENTIFIER
 
-    def get_materialized_view_config_change_collection(
-        self, relation_results: RelationResults, runtime_config: RuntimeConfigObject
-    ) -> Optional[PostgresMaterializedViewConfigChangeCollection]:
-        config_change_collection = PostgresMaterializedViewConfigChangeCollection()
+    def materialized_view_relation_changeset(
+        self, relation_results: DescribeRelationResults, runtime_config: RuntimeConfigObject
+    ) -> Optional[PostgresMaterializedViewRelationChangeset]:
+        config_change_collection = PostgresMaterializedViewRelationChangeset()
 
-        existing_materialized_view = PostgresMaterializedViewConfig.from_relation_results(
-            relation_results
-        )
-        new_materialized_view = PostgresMaterializedViewConfig.from_model_node(
-            runtime_config.model
+        existing_materialized_view = (
+            PostgresMaterializedViewRelation.from_describe_relation_results(relation_results)
         )
+        new_materialized_view = PostgresMaterializedViewRelation.from_node(runtime_config.model)
 
-        config_change_collection.indexes = self._get_index_config_changes(
+        config_change_collection.indexes = self._index_relation_changes(
             existing_materialized_view.indexes, new_materialized_view.indexes
         )
 
@@ -59,11 +54,11 @@ def get_materialized_view_config_change_collection(
         if config_change_collection.has_changes:
             return config_change_collection
 
-    def _get_index_config_changes(
+    def _index_relation_changes(
         self,
-        existing_indexes: FrozenSet[PostgresIndexConfig],
-        new_indexes: FrozenSet[PostgresIndexConfig],
-    ) -> Set[PostgresIndexConfigChange]:
+        existing_indexes: FrozenSet[PostgresIndexRelation],
+        new_indexes: FrozenSet[PostgresIndexRelation],
+    ) -> Set[PostgresIndexRelationChange]:
         """
         Get the index updates that will occur as a result of a new run
 
@@ -77,14 +72,14 @@ def _get_index_config_changes(
         Returns: a set of index updates in the form {"action": "drop/create", "context": <IndexConfig>}
         """
         drop_changes = set(
-            PostgresIndexConfigChange.from_dict(
-                {"action": RelationConfigChangeAction.drop, "context": index}
+            PostgresIndexRelationChange.from_dict(
+                {"action": RelationChangeAction.drop, "context": index}
             )
             for index in existing_indexes.difference(new_indexes)
         )
         create_changes = set(
-            PostgresIndexConfigChange.from_dict(
-                {"action": RelationConfigChangeAction.create, "context": index}
+            PostgresIndexRelationChange.from_dict(
+                {"action": RelationChangeAction.create, "context": index}
             )
             for index in new_indexes.difference(existing_indexes)
         )
diff --git a/plugins/postgres/dbt/adapters/postgres/relation/models/__init__.py b/plugins/postgres/dbt/adapters/postgres/relation/models/__init__.py
new file mode 100644
index 00000000000..ec195d47386
--- /dev/null
+++ b/plugins/postgres/dbt/adapters/postgres/relation/models/__init__.py
@@ -0,0 +1,11 @@
+from dbt.adapters.postgres.relation.models._index import (
+    PostgresIndexRelation,
+    PostgresIndexRelationChange,
+)
+from dbt.adapters.postgres.relation.models._materialized_view import (
+    PostgresMaterializedViewRelation,
+    PostgresMaterializedViewRelationChangeset,
+)
+from dbt.adapters.postgres.relation.models._policy import (
+    MAX_CHARACTERS_IN_IDENTIFIER,
+)
diff --git a/plugins/postgres/dbt/adapters/postgres/relation_configs/index.py b/plugins/postgres/dbt/adapters/postgres/relation/models/_index.py
similarity index 69%
rename from plugins/postgres/dbt/adapters/postgres/relation_configs/index.py
rename to plugins/postgres/dbt/adapters/postgres/relation/models/_index.py
index 3a072ea4307..cf68243dd6e 100644
--- a/plugins/postgres/dbt/adapters/postgres/relation_configs/index.py
+++ b/plugins/postgres/dbt/adapters/postgres/relation/models/_index.py
@@ -1,16 +1,16 @@
 from dataclasses import dataclass, field
-from typing import Set, FrozenSet
+from typing import Any, Dict, FrozenSet, Set
 
 import agate
+
+from dbt.adapters.relation.models import (
+    RelationChangeAction,
+    RelationChange,
+    RelationComponent,
+)
+from dbt.adapters.validation import ValidationMixin, ValidationRule
 from dbt.dataclass_schema import StrEnum
 from dbt.exceptions import DbtRuntimeError
-from dbt.adapters.relation_configs import (
-    RelationConfigBase,
-    RelationConfigValidationMixin,
-    RelationConfigValidationRule,
-    RelationConfigChangeAction,
-    RelationConfigChange,
-)
 
 
 class PostgresIndexMethod(StrEnum):
@@ -27,7 +27,7 @@ def default(cls) -> "PostgresIndexMethod":
 
 
 @dataclass(frozen=True, eq=True, unsafe_hash=True)
-class PostgresIndexConfig(RelationConfigBase, RelationConfigValidationMixin):
+class PostgresIndexRelation(RelationComponent, ValidationMixin):
     """
     This config fallows the specs found here:
     https://www.postgresql.org/docs/current/sql-createindex.html
@@ -49,9 +49,9 @@ class PostgresIndexConfig(RelationConfigBase, RelationConfigValidationMixin):
     method: PostgresIndexMethod = field(default=PostgresIndexMethod.default(), hash=True)
 
     @property
-    def validation_rules(self) -> Set[RelationConfigValidationRule]:
+    def validation_rules(self) -> Set[ValidationRule]:
         return {
-            RelationConfigValidationRule(
+            ValidationRule(
                 validation_check=self.column_names is not None,
                 validation_error=DbtRuntimeError(
                     "Indexes require at least one column, but none were provided"
@@ -60,7 +60,7 @@ def validation_rules(self) -> Set[RelationConfigValidationRule]:
         }
 
     @classmethod
-    def from_dict(cls, config_dict) -> "PostgresIndexConfig":
+    def from_dict(cls, config_dict: Dict[str, Any]) -> "PostgresIndexRelation":
         # TODO: include the QuotePolicy instead of defaulting to lower()
         kwargs_dict = {
             "name": config_dict.get("name"),
@@ -70,30 +70,33 @@ def from_dict(cls, config_dict) -> "PostgresIndexConfig":
             "unique": config_dict.get("unique"),
             "method": config_dict.get("method"),
         }
-        index: "PostgresIndexConfig" = super().from_dict(kwargs_dict)  # type: ignore
+        index = super().from_dict(kwargs_dict)
+        assert isinstance(index, PostgresIndexRelation)
         return index
 
     @classmethod
-    def parse_model_node(cls, model_node_entry: dict) -> dict:
+    def parse_node(cls, node: Dict[str, Any]) -> Dict[str, Any]:
         config_dict = {
-            "column_names": set(model_node_entry.get("columns", set())),
-            "unique": model_node_entry.get("unique"),
-            "method": model_node_entry.get("type"),
+            "column_names": set(node.get("columns", set())),
+            "unique": node.get("unique"),
+            "method": node.get("type"),
         }
         return config_dict
 
     @classmethod
-    def parse_relation_results(cls, relation_results_entry: agate.Row) -> dict:
+    def parse_describe_relation_results(
+        cls, describe_relation_results: agate.Row
+    ) -> Dict[str, Any]:
         config_dict = {
-            "name": relation_results_entry.get("name"),
-            "column_names": set(relation_results_entry.get("column_names", "").split(",")),
-            "unique": relation_results_entry.get("unique"),
-            "method": relation_results_entry.get("method"),
+            "name": describe_relation_results.get("name"),
+            "column_names": set(describe_relation_results.get("column_names", "").split(",")),
+            "unique": describe_relation_results.get("unique"),
+            "method": describe_relation_results.get("method"),
         }
         return config_dict
 
     @property
-    def as_node_config(self) -> dict:
+    def as_node_config(self) -> Dict[str, Any]:
         """
         Returns: a dictionary that can be passed into `get_create_index_sql()`
         """
@@ -106,7 +109,7 @@ def as_node_config(self) -> dict:
 
 
 @dataclass(frozen=True, eq=True, unsafe_hash=True)
-class PostgresIndexConfigChange(RelationConfigChange, RelationConfigValidationMixin):
+class PostgresIndexRelationChange(RelationChange, ValidationMixin):
     """
     Example of an index change:
     {
@@ -129,33 +132,33 @@ class PostgresIndexConfigChange(RelationConfigChange, RelationConfigValidationMi
     }
     """
 
-    context: PostgresIndexConfig
+    context: PostgresIndexRelation
 
     @property
     def requires_full_refresh(self) -> bool:
         return False
 
     @property
-    def validation_rules(self) -> Set[RelationConfigValidationRule]:
+    def validation_rules(self) -> Set[ValidationRule]:
         return {
-            RelationConfigValidationRule(
+            ValidationRule(
                 validation_check=self.action
-                in {RelationConfigChangeAction.create, RelationConfigChangeAction.drop},
+                in {RelationChangeAction.create, RelationChangeAction.drop},
                 validation_error=DbtRuntimeError(
                     "Invalid operation, only `drop` and `create` changes are supported for indexes."
                 ),
             ),
-            RelationConfigValidationRule(
+            ValidationRule(
                 validation_check=not (
-                    self.action == RelationConfigChangeAction.drop and self.context.name is None
+                    self.action == RelationChangeAction.drop and self.context.name is None
                 ),
                 validation_error=DbtRuntimeError(
                     "Invalid operation, attempting to drop an index with no name."
                 ),
             ),
-            RelationConfigValidationRule(
+            ValidationRule(
                 validation_check=not (
-                    self.action == RelationConfigChangeAction.create
+                    self.action == RelationChangeAction.create
                     and self.context.column_names == set()
                 ),
                 validation_error=DbtRuntimeError(
diff --git a/plugins/postgres/dbt/adapters/postgres/relation/models/_materialized_view.py b/plugins/postgres/dbt/adapters/postgres/relation/models/_materialized_view.py
new file mode 100644
index 00000000000..11e9fa84b15
--- /dev/null
+++ b/plugins/postgres/dbt/adapters/postgres/relation/models/_materialized_view.py
@@ -0,0 +1,113 @@
+from dataclasses import dataclass, field
+from typing import Any, Dict, FrozenSet, List, Set
+
+import agate
+from dbt.adapters.relation.models import DescribeRelationResults, RelationComponent
+from dbt.adapters.validation import ValidationMixin, ValidationRule
+from dbt.contracts.graph.nodes import ModelNode
+from dbt.exceptions import DbtRuntimeError
+
+from dbt.adapters.postgres.relation.models._index import (
+    PostgresIndexRelation,
+    PostgresIndexRelationChange,
+)
+from dbt.adapters.postgres.relation.models._policy import MAX_CHARACTERS_IN_IDENTIFIER
+
+
+@dataclass(frozen=True, eq=True, unsafe_hash=True)
+class PostgresMaterializedViewRelation(RelationComponent, ValidationMixin):
+    """
+    This config follows the specs found here:
+    https://www.postgresql.org/docs/current/sql-creatematerializedview.html
+
+    The following parameters are configurable by dbt:
+    - table_name: name of the materialized view
+    - query: the query that defines the view
+    - indexes: the collection (set) of indexes on the materialized view
+
+    Applicable defaults for non-configurable parameters:
+    - method: `heap`
+    - tablespace_name: `default_tablespace`
+    - with_data: `True`
+    """
+
+    name: str = ""
+    query: str = ""
+    indexes: FrozenSet[PostgresIndexRelation] = field(default_factory=frozenset)
+
+    @property
+    def validation_rules(self) -> Set[ValidationRule]:
+        # index rules get run by default with the mixin
+        return {
+            ValidationRule(
+                validation_check=self.name is None
+                or len(self.name) <= MAX_CHARACTERS_IN_IDENTIFIER,
+                validation_error=DbtRuntimeError(
+                    f"The materialized view name is more than {MAX_CHARACTERS_IN_IDENTIFIER} "
+                    f"characters: {self.name}"
+                ),
+            ),
+        }
+
+    @classmethod
+    def from_dict(cls, config_dict: Dict[str, Any]) -> "PostgresMaterializedViewRelation":
+        kwargs_dict = {
+            "name": config_dict.get("name"),
+            "query": config_dict.get("query"),
+            "indexes": frozenset(
+                PostgresIndexRelation.from_dict(index) for index in config_dict.get("indexes", {})
+            ),
+        }
+        materialized_view = super().from_dict(kwargs_dict)
+        assert isinstance(materialized_view, PostgresMaterializedViewRelation)
+        return materialized_view
+
+    @classmethod
+    def from_node(cls, node: ModelNode) -> "PostgresMaterializedViewRelation":
+        materialized_view_config = cls.parse_node(node)
+        materialized_view = cls.from_dict(materialized_view_config)
+        return materialized_view
+
+    @classmethod
+    def parse_node(cls, node: ModelNode) -> Dict[str, Any]:
+        indexes: List[dict] = node.config.extra.get("indexes", [])
+        config_dict = {
+            "name": node.identifier,
+            "query": node.compiled_code,
+            "indexes": [PostgresIndexRelation.parse_node(index) for index in indexes],
+        }
+        return config_dict
+
+    @classmethod
+    def from_describe_relation_results(
+        cls, describe_relation_results: DescribeRelationResults
+    ) -> "PostgresMaterializedViewRelation":
+        materialized_view_config = cls.parse_describe_relation_results(describe_relation_results)
+        materialized_view = cls.from_dict(materialized_view_config)
+        return materialized_view
+
+    @classmethod
+    def parse_describe_relation_results(
+        cls, describe_relation_results: DescribeRelationResults
+    ) -> Dict[str, Any]:
+        indexes: agate.Table = describe_relation_results.get("indexes", agate.Table(rows={}))
+        config_dict = {
+            "indexes": [
+                PostgresIndexRelation.parse_describe_relation_results(index)
+                for index in indexes.rows
+            ],
+        }
+        return config_dict
+
+
+@dataclass
+class PostgresMaterializedViewRelationChangeset:
+    indexes: Set[PostgresIndexRelationChange] = field(default_factory=set)
+
+    @property
+    def requires_full_refresh(self) -> bool:
+        return any(index.requires_full_refresh for index in self.indexes)
+
+    @property
+    def has_changes(self) -> bool:
+        return self.indexes != set()
diff --git a/plugins/postgres/dbt/adapters/postgres/relation_configs/constants.py b/plugins/postgres/dbt/adapters/postgres/relation/models/_policy.py
similarity index 100%
rename from plugins/postgres/dbt/adapters/postgres/relation_configs/constants.py
rename to plugins/postgres/dbt/adapters/postgres/relation/models/_policy.py
diff --git a/plugins/postgres/dbt/adapters/postgres/relation_configs/__init__.py b/plugins/postgres/dbt/adapters/postgres/relation_configs/__init__.py
deleted file mode 100644
index 9fdb942bfa5..00000000000
--- a/plugins/postgres/dbt/adapters/postgres/relation_configs/__init__.py
+++ /dev/null
@@ -1,11 +0,0 @@
-from dbt.adapters.postgres.relation_configs.constants import (  # noqa: F401
-    MAX_CHARACTERS_IN_IDENTIFIER,
-)
-from dbt.adapters.postgres.relation_configs.index import (  # noqa: F401
-    PostgresIndexConfig,
-    PostgresIndexConfigChange,
-)
-from dbt.adapters.postgres.relation_configs.materialized_view import (  # noqa: F401
-    PostgresMaterializedViewConfig,
-    PostgresMaterializedViewConfigChangeCollection,
-)
diff --git a/plugins/postgres/dbt/adapters/postgres/relation_configs/materialized_view.py b/plugins/postgres/dbt/adapters/postgres/relation_configs/materialized_view.py
deleted file mode 100644
index 15e700e777a..00000000000
--- a/plugins/postgres/dbt/adapters/postgres/relation_configs/materialized_view.py
+++ /dev/null
@@ -1,113 +0,0 @@
-from dataclasses import dataclass, field
-from typing import Set, FrozenSet, List
-
-import agate
-from dbt.adapters.relation_configs import (
-    RelationConfigBase,
-    RelationResults,
-    RelationConfigValidationMixin,
-    RelationConfigValidationRule,
-)
-from dbt.contracts.graph.nodes import ModelNode
-from dbt.exceptions import DbtRuntimeError
-
-from dbt.adapters.postgres.relation_configs.constants import MAX_CHARACTERS_IN_IDENTIFIER
-from dbt.adapters.postgres.relation_configs.index import (
-    PostgresIndexConfig,
-    PostgresIndexConfigChange,
-)
-
-
-@dataclass(frozen=True, eq=True, unsafe_hash=True)
-class PostgresMaterializedViewConfig(RelationConfigBase, RelationConfigValidationMixin):
-    """
-    This config follows the specs found here:
-    https://www.postgresql.org/docs/current/sql-creatematerializedview.html
-
-    The following parameters are configurable by dbt:
-    - table_name: name of the materialized view
-    - query: the query that defines the view
-    - indexes: the collection (set) of indexes on the materialized view
-
-    Applicable defaults for non-configurable parameters:
-    - method: `heap`
-    - tablespace_name: `default_tablespace`
-    - with_data: `True`
-    """
-
-    table_name: str = ""
-    query: str = ""
-    indexes: FrozenSet[PostgresIndexConfig] = field(default_factory=frozenset)
-
-    @property
-    def validation_rules(self) -> Set[RelationConfigValidationRule]:
-        # index rules get run by default with the mixin
-        return {
-            RelationConfigValidationRule(
-                validation_check=self.table_name is None
-                or len(self.table_name) <= MAX_CHARACTERS_IN_IDENTIFIER,
-                validation_error=DbtRuntimeError(
-                    f"The materialized view name is more than {MAX_CHARACTERS_IN_IDENTIFIER} "
-                    f"characters: {self.table_name}"
-                ),
-            ),
-        }
-
-    @classmethod
-    def from_dict(cls, config_dict: dict) -> "PostgresMaterializedViewConfig":
-        kwargs_dict = {
-            "table_name": config_dict.get("table_name"),
-            "query": config_dict.get("query"),
-            "indexes": frozenset(
-                PostgresIndexConfig.from_dict(index) for index in config_dict.get("indexes", {})
-            ),
-        }
-        materialized_view: "PostgresMaterializedViewConfig" = super().from_dict(kwargs_dict)  # type: ignore
-        return materialized_view
-
-    @classmethod
-    def from_model_node(cls, model_node: ModelNode) -> "PostgresMaterializedViewConfig":
-        materialized_view_config = cls.parse_model_node(model_node)
-        materialized_view = cls.from_dict(materialized_view_config)
-        return materialized_view
-
-    @classmethod
-    def parse_model_node(cls, model_node: ModelNode) -> dict:
-        indexes: List[dict] = model_node.config.extra.get("indexes", [])
-        config_dict = {
-            "table_name": model_node.identifier,
-            "query": model_node.compiled_code,
-            "indexes": [PostgresIndexConfig.parse_model_node(index) for index in indexes],
-        }
-        return config_dict
-
-    @classmethod
-    def from_relation_results(
-        cls, relation_results: RelationResults
-    ) -> "PostgresMaterializedViewConfig":
-        materialized_view_config = cls.parse_relation_results(relation_results)
-        materialized_view = cls.from_dict(materialized_view_config)
-        return materialized_view
-
-    @classmethod
-    def parse_relation_results(cls, relation_results: RelationResults) -> dict:
-        indexes: agate.Table = relation_results.get("indexes", agate.Table(rows={}))
-        config_dict = {
-            "indexes": [
-                PostgresIndexConfig.parse_relation_results(index) for index in indexes.rows
-            ],
-        }
-        return config_dict
-
-
-@dataclass
-class PostgresMaterializedViewConfigChangeCollection:
-    indexes: Set[PostgresIndexConfigChange] = field(default_factory=set)
-
-    @property
-    def requires_full_refresh(self) -> bool:
-        return any(index.requires_full_refresh for index in self.indexes)
-
-    @property
-    def has_changes(self) -> bool:
-        return self.indexes != set()
diff --git a/plugins/postgres/dbt/include/postgres/macros/adapters.sql b/plugins/postgres/dbt/include/postgres/macros/adapters.sql
deleted file mode 100644
index 0c56232bb22..00000000000
--- a/plugins/postgres/dbt/include/postgres/macros/adapters.sql
+++ /dev/null
@@ -1,252 +0,0 @@
-{% macro postgres__create_table_as(temporary, relation, sql) -%}
-  {%- set unlogged = config.get('unlogged', default=false) -%}
-  {%- set sql_header = config.get('sql_header', none) -%}
-
-  {{ sql_header if sql_header is not none }}
-
-  create {% if temporary -%}
-    temporary
-  {%- elif unlogged -%}
-    unlogged
-  {%- endif %} table {{ relation }}
-  {% set contract_config = config.get('contract') %}
-  {% if contract_config.enforced %}
-    {{ get_assert_columns_equivalent(sql) }}
-    {{ get_table_columns_and_constraints() }} ;
-    insert into {{ relation }} (
-      {{ adapter.dispatch('get_column_names', 'dbt')() }}
-    )
-    {%- set sql = get_select_subquery(sql) %}
-  {% else %}
-    as
-  {% endif %}
-  (
-    {{ sql }}
-  );
-{%- endmacro %}
-
-{% macro postgres__get_create_index_sql(relation, index_dict) -%}
-  {%- set index_config = adapter.parse_index(index_dict) -%}
-  {%- set comma_separated_columns = ", ".join(index_config.columns) -%}
-  {%- set index_name = index_config.render(relation) -%}
-
-  create {% if index_config.unique -%}
-    unique
-  {%- endif %} index if not exists
-  "{{ index_name }}"
-  on {{ relation }} {% if index_config.type -%}
-    using {{ index_config.type }}
-  {%- endif %}
-  ({{ comma_separated_columns }});
-{%- endmacro %}
-
-{% macro postgres__create_schema(relation) -%}
-  {% if relation.database -%}
-    {{ adapter.verify_database(relation.database) }}
-  {%- endif -%}
-  {%- call statement('create_schema') -%}
-    create schema if not exists {{ relation.without_identifier().include(database=False) }}
-  {%- endcall -%}
-{% endmacro %}
-
-{% macro postgres__drop_schema(relation) -%}
-  {% if relation.database -%}
-    {{ adapter.verify_database(relation.database) }}
-  {%- endif -%}
-  {%- call statement('drop_schema') -%}
-    drop schema if exists {{ relation.without_identifier().include(database=False) }} cascade
-  {%- endcall -%}
-{% endmacro %}
-
-{% macro postgres__get_columns_in_relation(relation) -%}
-  {% call statement('get_columns_in_relation', fetch_result=True) %}
-      select
-          column_name,
-          data_type,
-          character_maximum_length,
-          numeric_precision,
-          numeric_scale
-
-      from {{ relation.information_schema('columns') }}
-      where table_name = '{{ relation.identifier }}'
-        {% if relation.schema %}
-        and table_schema = '{{ relation.schema }}'
-        {% endif %}
-      order by ordinal_position
-
-  {% endcall %}
-  {% set table = load_result('get_columns_in_relation').table %}
-  {{ return(sql_convert_columns_in_relation(table)) }}
-{% endmacro %}
-
-
-{% macro postgres__list_relations_without_caching(schema_relation) %}
-  {% call statement('list_relations_without_caching', fetch_result=True) -%}
-    select
-      '{{ schema_relation.database }}' as database,
-      tablename as name,
-      schemaname as schema,
-      'table' as type
-    from pg_tables
-    where schemaname ilike '{{ schema_relation.schema }}'
-    union all
-    select
-      '{{ schema_relation.database }}' as database,
-      viewname as name,
-      schemaname as schema,
-      'view' as type
-    from pg_views
-    where schemaname ilike '{{ schema_relation.schema }}'
-    union all
-    select
-      '{{ schema_relation.database }}' as database,
-      matviewname as name,
-      schemaname as schema,
-      'materialized_view' as type
-    from pg_matviews
-    where schemaname ilike '{{ schema_relation.schema }}'
-  {% endcall %}
-  {{ return(load_result('list_relations_without_caching').table) }}
-{% endmacro %}
-
-{% macro postgres__information_schema_name(database) -%}
-  {% if database_name -%}
-    {{ adapter.verify_database(database_name) }}
-  {%- endif -%}
-  information_schema
-{%- endmacro %}
-
-{% macro postgres__list_schemas(database) %}
-  {% if database -%}
-    {{ adapter.verify_database(database) }}
-  {%- endif -%}
-  {% call statement('list_schemas', fetch_result=True, auto_begin=False) %}
-    select distinct nspname from pg_namespace
-  {% endcall %}
-  {{ return(load_result('list_schemas').table) }}
-{% endmacro %}
-
-{% macro postgres__check_schema_exists(information_schema, schema) -%}
-  {% if information_schema.database -%}
-    {{ adapter.verify_database(information_schema.database) }}
-  {%- endif -%}
-  {% call statement('check_schema_exists', fetch_result=True, auto_begin=False) %}
-    select count(*) from pg_namespace where nspname = '{{ schema }}'
-  {% endcall %}
-  {{ return(load_result('check_schema_exists').table) }}
-{% endmacro %}
-
-{#
-  Postgres tables have a maximum length of 63 characters, anything longer is silently truncated.
-  Temp and backup relations add a lot of extra characters to the end of table names to ensure uniqueness.
-  To prevent this going over the character limit, the base_relation name is truncated to ensure
-  that name + suffix + uniquestring is < 63 characters.
-#}
-
-{% macro postgres__make_relation_with_suffix(base_relation, suffix, dstring) %}
-    {% if dstring %}
-      {% set dt = modules.datetime.datetime.now() %}
-      {% set dtstring = dt.strftime("%H%M%S%f") %}
-      {% set suffix = suffix ~ dtstring %}
-    {% endif %}
-    {% set suffix_length = suffix|length %}
-    {% set relation_max_name_length = base_relation.relation_max_name_length() %}
-    {% if suffix_length > relation_max_name_length %}
-        {% do exceptions.raise_compiler_error('Relation suffix is too long (' ~ suffix_length ~ ' characters). Maximum length is ' ~ relation_max_name_length ~ ' characters.') %}
-    {% endif %}
-    {% set identifier = base_relation.identifier[:relation_max_name_length - suffix_length] ~ suffix %}
-
-    {{ return(base_relation.incorporate(path={"identifier": identifier })) }}
-
-  {% endmacro %}
-
-{% macro postgres__make_intermediate_relation(base_relation, suffix) %}
-    {{ return(postgres__make_relation_with_suffix(base_relation, suffix, dstring=False)) }}
-{% endmacro %}
-
-{% macro postgres__make_temp_relation(base_relation, suffix) %}
-    {% set temp_relation = postgres__make_relation_with_suffix(base_relation, suffix, dstring=True) %}
-    {{ return(temp_relation.incorporate(path={"schema": none,
-                                              "database": none})) }}
-{% endmacro %}
-
-{% macro postgres__make_backup_relation(base_relation, backup_relation_type, suffix) %}
-    {% set backup_relation = postgres__make_relation_with_suffix(base_relation, suffix, dstring=False) %}
-    {{ return(backup_relation.incorporate(type=backup_relation_type)) }}
-{% endmacro %}
-
-{#
-  By using dollar-quoting like this, users can embed anything they want into their comments
-  (including nested dollar-quoting), as long as they do not use this exact dollar-quoting
-  label. It would be nice to just pick a new one but eventually you do have to give up.
-#}
-{% macro postgres_escape_comment(comment) -%}
-  {% if comment is not string %}
-    {% do exceptions.raise_compiler_error('cannot escape a non-string: ' ~ comment) %}
-  {% endif %}
-  {%- set magic = '$dbt_comment_literal_block$' -%}
-  {%- if magic in comment -%}
-    {%- do exceptions.raise_compiler_error('The string ' ~ magic ~ ' is not allowed in comments.') -%}
-  {%- endif -%}
-  {{ magic }}{{ comment }}{{ magic }}
-{%- endmacro %}
-
-
-{% macro postgres__alter_relation_comment(relation, comment) %}
-  {% set escaped_comment = postgres_escape_comment(comment) %}
-  comment on {{ relation.type }} {{ relation }} is {{ escaped_comment }};
-{% endmacro %}
-
-
-{% macro postgres__alter_column_comment(relation, column_dict) %}
-  {% set existing_columns = adapter.get_columns_in_relation(relation) | map(attribute="name") | list %}
-  {% for column_name in column_dict if (column_name in existing_columns) %}
-    {% set comment = column_dict[column_name]['description'] %}
-    {% set escaped_comment = postgres_escape_comment(comment) %}
-    comment on column {{ relation }}.{{ adapter.quote(column_name) if column_dict[column_name]['quote'] else column_name }} is {{ escaped_comment }};
-  {% endfor %}
-{% endmacro %}
-
-{%- macro postgres__get_show_grant_sql(relation) -%}
-  select grantee, privilege_type
-  from {{ relation.information_schema('role_table_grants') }}
-      where grantor = current_role
-        and grantee != current_role
-        and table_schema = '{{ relation.schema }}'
-        and table_name = '{{ relation.identifier }}'
-{%- endmacro -%}
-
-{% macro postgres__copy_grants() %}
-    {{ return(False) }}
-{% endmacro %}
-
-
-{% macro postgres__get_show_indexes_sql(relation) %}
-    select
-        i.relname                                   as name,
-        m.amname                                    as method,
-        ix.indisunique                              as "unique",
-        array_to_string(array_agg(a.attname), ',')  as column_names
-    from pg_index ix
-    join pg_class i
-        on i.oid = ix.indexrelid
-    join pg_am m
-        on m.oid=i.relam
-    join pg_class t
-        on t.oid = ix.indrelid
-    join pg_namespace n
-        on n.oid = t.relnamespace
-    join pg_attribute a
-        on a.attrelid = t.oid
-        and a.attnum = ANY(ix.indkey)
-    where t.relname = '{{ relation.identifier }}'
-      and n.nspname = '{{ relation.schema }}'
-      and t.relkind in ('r', 'm')
-    group by 1, 2, 3
-    order by 1, 2, 3
-{% endmacro %}
-
-
-{%- macro postgres__get_drop_index_sql(relation, index_name) -%}
-    drop index if exists "{{ relation.schema }}"."{{ index_name }}"
-{%- endmacro -%}
diff --git a/plugins/postgres/dbt/include/postgres/macros/catalog.sql b/plugins/postgres/dbt/include/postgres/macros/metadata/get_catalog.sql
similarity index 100%
rename from plugins/postgres/dbt/include/postgres/macros/catalog.sql
rename to plugins/postgres/dbt/include/postgres/macros/metadata/get_catalog.sql
diff --git a/plugins/postgres/dbt/include/postgres/macros/relations.sql b/plugins/postgres/dbt/include/postgres/macros/metadata/get_relations.sql
similarity index 100%
rename from plugins/postgres/dbt/include/postgres/macros/relations.sql
rename to plugins/postgres/dbt/include/postgres/macros/metadata/get_relations.sql
diff --git a/plugins/postgres/dbt/include/postgres/macros/metadata/information_schema_name.sql b/plugins/postgres/dbt/include/postgres/macros/metadata/information_schema_name.sql
new file mode 100644
index 00000000000..c148ef587c2
--- /dev/null
+++ b/plugins/postgres/dbt/include/postgres/macros/metadata/information_schema_name.sql
@@ -0,0 +1,6 @@
+{% macro postgres__information_schema_name(database) -%}
+  {% if database_name -%}
+    {{ adapter.verify_database(database_name) }}
+  {%- endif -%}
+  information_schema
+{%- endmacro %}
diff --git a/plugins/postgres/dbt/include/postgres/macros/metadata/list_relations_without_caching.sql b/plugins/postgres/dbt/include/postgres/macros/metadata/list_relations_without_caching.sql
new file mode 100644
index 00000000000..9e5d60a35b1
--- /dev/null
+++ b/plugins/postgres/dbt/include/postgres/macros/metadata/list_relations_without_caching.sql
@@ -0,0 +1,28 @@
+{% macro postgres__list_relations_without_caching(schema_relation) %}
+  {% call statement('list_relations_without_caching', fetch_result=True) -%}
+    select
+      '{{ schema_relation.database }}' as database,
+      tablename as name,
+      schemaname as schema,
+      'table' as type
+    from pg_tables
+    where schemaname ilike '{{ schema_relation.schema }}'
+    union all
+    select
+      '{{ schema_relation.database }}' as database,
+      viewname as name,
+      schemaname as schema,
+      'view' as type
+    from pg_views
+    where schemaname ilike '{{ schema_relation.schema }}'
+    union all
+    select
+      '{{ schema_relation.database }}' as database,
+      matviewname as name,
+      schemaname as schema,
+      'materialized_view' as type
+    from pg_matviews
+    where schemaname ilike '{{ schema_relation.schema }}'
+  {% endcall %}
+  {{ return(load_result('list_relations_without_caching').table) }}
+{% endmacro %}
diff --git a/plugins/postgres/dbt/include/postgres/macros/relation.sql b/plugins/postgres/dbt/include/postgres/macros/relation.sql
new file mode 100644
index 00000000000..a95b68256b5
--- /dev/null
+++ b/plugins/postgres/dbt/include/postgres/macros/relation.sql
@@ -0,0 +1,42 @@
+{#
+  Postgres tables have a maximum length of 63 characters, anything longer is silently truncated.
+  Temp and backup relations add a lot of extra characters to the end of table names to ensure uniqueness.
+  To prevent this going over the character limit, the base_relation name is truncated to ensure
+  that name + suffix + uniquestring is < 63 characters.
+#}
+
+
+{% macro postgres__make_relation_with_suffix(base_relation, suffix, dstring) %}
+    {% if dstring %}
+      {% set dt = modules.datetime.datetime.now() %}
+      {% set dtstring = dt.strftime("%H%M%S%f") %}
+      {% set suffix = suffix ~ dtstring %}
+    {% endif %}
+    {% set suffix_length = suffix|length %}
+    {% set relation_max_name_length = base_relation.relation_max_name_length() %}
+    {% if suffix_length > relation_max_name_length %}
+        {% do exceptions.raise_compiler_error('Relation suffix is too long (' ~ suffix_length ~ ' characters). Maximum length is ' ~ relation_max_name_length ~ ' characters.') %}
+    {% endif %}
+    {% set identifier = base_relation.identifier[:relation_max_name_length - suffix_length] ~ suffix %}
+
+    {{ return(base_relation.incorporate(path={"identifier": identifier })) }}
+
+  {% endmacro %}
+
+
+{% macro postgres__make_intermediate_relation(base_relation, suffix) %}
+    {{ return(postgres__make_relation_with_suffix(base_relation, suffix, dstring=False)) }}
+{% endmacro %}
+
+
+{% macro postgres__make_temp_relation(base_relation, suffix) %}
+    {% set temp_relation = postgres__make_relation_with_suffix(base_relation, suffix, dstring=True) %}
+    {{ return(temp_relation.incorporate(path={"schema": none,
+                                              "database": none})) }}
+{% endmacro %}
+
+
+{% macro postgres__make_backup_relation(base_relation, backup_relation_type, suffix) %}
+    {% set backup_relation = postgres__make_relation_with_suffix(base_relation, suffix, dstring=False) %}
+    {{ return(backup_relation.incorporate(type=backup_relation_type)) }}
+{% endmacro %}
diff --git a/plugins/postgres/dbt/include/postgres/macros/relation_components/column.sql b/plugins/postgres/dbt/include/postgres/macros/relation_components/column.sql
new file mode 100644
index 00000000000..8e52963fddc
--- /dev/null
+++ b/plugins/postgres/dbt/include/postgres/macros/relation_components/column.sql
@@ -0,0 +1,20 @@
+{% macro postgres__get_columns_in_relation(relation) -%}
+  {% call statement('get_columns_in_relation', fetch_result=True) %}
+      select
+          column_name,
+          data_type,
+          character_maximum_length,
+          numeric_precision,
+          numeric_scale
+
+      from {{ relation.information_schema('columns') }}
+      where table_name = '{{ relation.identifier }}'
+        {% if relation.schema %}
+        and table_schema = '{{ relation.schema }}'
+        {% endif %}
+      order by ordinal_position
+
+  {% endcall %}
+  {% set table = load_result('get_columns_in_relation').table %}
+  {{ return(sql_convert_columns_in_relation(table)) }}
+{% endmacro %}
diff --git a/plugins/postgres/dbt/include/postgres/macros/relation_components/comment.sql b/plugins/postgres/dbt/include/postgres/macros/relation_components/comment.sql
new file mode 100644
index 00000000000..39e8fa98513
--- /dev/null
+++ b/plugins/postgres/dbt/include/postgres/macros/relation_components/comment.sql
@@ -0,0 +1,31 @@
+{#
+  By using dollar-quoting like this, users can embed anything they want into their comments
+  (including nested dollar-quoting), as long as they do not use this exact dollar-quoting
+  label. It would be nice to just pick a new one but eventually you do have to give up.
+#}
+{% macro postgres_escape_comment(comment) -%}
+  {% if comment is not string %}
+    {% do exceptions.raise_compiler_error('cannot escape a non-string: ' ~ comment) %}
+  {% endif %}
+  {%- set magic = '$dbt_comment_literal_block$' -%}
+  {%- if magic in comment -%}
+    {%- do exceptions.raise_compiler_error('The string ' ~ magic ~ ' is not allowed in comments.') -%}
+  {%- endif -%}
+  {{ magic }}{{ comment }}{{ magic }}
+{%- endmacro %}
+
+
+{% macro postgres__alter_relation_comment(relation, comment) %}
+  {% set escaped_comment = postgres_escape_comment(comment) %}
+  comment on {{ relation.type }} {{ relation }} is {{ escaped_comment }};
+{% endmacro %}
+
+
+{% macro postgres__alter_column_comment(relation, column_dict) %}
+  {% set existing_columns = adapter.get_columns_in_relation(relation) | map(attribute="name") | list %}
+  {% for column_name in column_dict if (column_name in existing_columns) %}
+    {% set comment = column_dict[column_name]['description'] %}
+    {% set escaped_comment = postgres_escape_comment(comment) %}
+    comment on column {{ relation }}.{{ adapter.quote(column_name) if column_dict[column_name]['quote'] else column_name }} is {{ escaped_comment }};
+  {% endfor %}
+{% endmacro %}
diff --git a/plugins/postgres/dbt/include/postgres/macros/relation_components/grant.sql b/plugins/postgres/dbt/include/postgres/macros/relation_components/grant.sql
new file mode 100644
index 00000000000..1125be8356a
--- /dev/null
+++ b/plugins/postgres/dbt/include/postgres/macros/relation_components/grant.sql
@@ -0,0 +1,13 @@
+{%- macro postgres__get_show_grant_sql(relation) -%}
+  select grantee, privilege_type
+  from {{ relation.information_schema('role_table_grants') }}
+      where grantor = current_role
+        and grantee != current_role
+        and table_schema = '{{ relation.schema }}'
+        and table_name = '{{ relation.identifier }}'
+{%- endmacro -%}
+
+
+{% macro postgres__copy_grants() %}
+    {{ return(False) }}
+{% endmacro %}
diff --git a/plugins/postgres/dbt/include/postgres/macros/relation_components/index.sql b/plugins/postgres/dbt/include/postgres/macros/relation_components/index.sql
new file mode 100644
index 00000000000..fdf84a20a76
--- /dev/null
+++ b/plugins/postgres/dbt/include/postgres/macros/relation_components/index.sql
@@ -0,0 +1,45 @@
+{% macro postgres__get_create_index_sql(relation, index_dict) -%}
+  {%- set index_config = adapter.parse_index(index_dict) -%}
+  {%- set comma_separated_columns = ", ".join(index_config.columns) -%}
+  {%- set index_name = index_config.render(relation) -%}
+
+  create {% if index_config.unique -%}
+    unique
+  {%- endif %} index if not exists
+  "{{ index_name }}"
+  on {{ relation }} {% if index_config.type -%}
+    using {{ index_config.type }}
+  {%- endif %}
+  ({{ comma_separated_columns }});
+{%- endmacro %}
+
+
+{% macro postgres__get_show_indexes_sql(relation) %}
+    select
+        i.relname                                   as name,
+        m.amname                                    as method,
+        ix.indisunique                              as "unique",
+        array_to_string(array_agg(a.attname), ',')  as column_names
+    from pg_index ix
+    join pg_class i
+        on i.oid = ix.indexrelid
+    join pg_am m
+        on m.oid=i.relam
+    join pg_class t
+        on t.oid = ix.indrelid
+    join pg_namespace n
+        on n.oid = t.relnamespace
+    join pg_attribute a
+        on a.attrelid = t.oid
+        and a.attnum = ANY(ix.indkey)
+    where t.relname = '{{ relation.identifier }}'
+      and n.nspname = '{{ relation.schema }}'
+      and t.relkind in ('r', 'm')
+    group by 1, 2, 3
+    order by 1, 2, 3
+{% endmacro %}
+
+
+{%- macro postgres__get_drop_index_sql(relation, index_name) -%}
+    drop index if exists "{{ relation.schema }}"."{{ index_name }}"
+{%- endmacro -%}
diff --git a/plugins/postgres/dbt/include/postgres/macros/relation_components/schema.sql b/plugins/postgres/dbt/include/postgres/macros/relation_components/schema.sql
new file mode 100644
index 00000000000..ab17321e7b0
--- /dev/null
+++ b/plugins/postgres/dbt/include/postgres/macros/relation_components/schema.sql
@@ -0,0 +1,40 @@
+{% macro postgres__create_schema(relation) -%}
+  {% if relation.database -%}
+    {{ adapter.verify_database(relation.database) }}
+  {%- endif -%}
+  {%- call statement('create_schema') -%}
+    create schema if not exists {{ relation.without_identifier().include(database=False) }}
+  {%- endcall -%}
+{% endmacro %}
+
+
+{% macro postgres__drop_schema(relation) -%}
+  {% if relation.database -%}
+    {{ adapter.verify_database(relation.database) }}
+  {%- endif -%}
+  {%- call statement('drop_schema') -%}
+    drop schema if exists {{ relation.without_identifier().include(database=False) }} cascade
+  {%- endcall -%}
+{% endmacro %}
+
+
+{% macro postgres__list_schemas(database) %}
+  {% if database -%}
+    {{ adapter.verify_database(database) }}
+  {%- endif -%}
+  {% call statement('list_schemas', fetch_result=True, auto_begin=False) %}
+    select distinct nspname from pg_namespace
+  {% endcall %}
+  {{ return(load_result('list_schemas').table) }}
+{% endmacro %}
+
+
+{% macro postgres__check_schema_exists(information_schema, schema) -%}
+  {% if information_schema.database -%}
+    {{ adapter.verify_database(information_schema.database) }}
+  {%- endif -%}
+  {% call statement('check_schema_exists', fetch_result=True, auto_begin=False) %}
+    select count(*) from pg_namespace where nspname = '{{ schema }}'
+  {% endcall %}
+  {{ return(load_result('check_schema_exists').table) }}
+{% endmacro %}
diff --git a/plugins/postgres/dbt/include/postgres/macros/materializations/materialized_view.sql b/plugins/postgres/dbt/include/postgres/macros/relations/materialized_view.sql
similarity index 94%
rename from plugins/postgres/dbt/include/postgres/macros/materializations/materialized_view.sql
rename to plugins/postgres/dbt/include/postgres/macros/relations/materialized_view.sql
index 6e66e4bcd2c..fcfd953ba15 100644
--- a/plugins/postgres/dbt/include/postgres/macros/materializations/materialized_view.sql
+++ b/plugins/postgres/dbt/include/postgres/macros/relations/materialized_view.sql
@@ -46,7 +46,7 @@
 
 {% macro postgres__get_materialized_view_configuration_changes(existing_relation, new_config) %}
     {% set _existing_materialized_view = postgres__describe_materialized_view(existing_relation) %}
-    {% set _configuration_changes = existing_relation.get_materialized_view_config_change_collection(_existing_materialized_view, new_config) %}
+    {% set _configuration_changes = existing_relation.materialized_view_relation_changeset(_existing_materialized_view, new_config) %}
     {% do return(_configuration_changes) %}
 {% endmacro %}
 
diff --git a/plugins/postgres/dbt/include/postgres/macros/relations/table.sql b/plugins/postgres/dbt/include/postgres/macros/relations/table.sql
new file mode 100644
index 00000000000..c7170aa552f
--- /dev/null
+++ b/plugins/postgres/dbt/include/postgres/macros/relations/table.sql
@@ -0,0 +1,26 @@
+{% macro postgres__create_table_as(temporary, relation, sql) -%}
+  {%- set unlogged = config.get('unlogged', default=false) -%}
+  {%- set sql_header = config.get('sql_header', none) -%}
+
+  {{ sql_header if sql_header is not none }}
+
+  create {% if temporary -%}
+    temporary
+  {%- elif unlogged -%}
+    unlogged
+  {%- endif %} table {{ relation }}
+  {% set contract_config = config.get('contract') %}
+  {% if contract_config.enforced %}
+    {{ get_assert_columns_equivalent(sql) }}
+    {{ get_table_columns_and_constraints() }} ;
+    insert into {{ relation }} (
+      {{ adapter.dispatch('get_column_names', 'dbt')() }}
+    )
+    {%- set sql = get_select_subquery(sql) %}
+  {% else %}
+    as
+  {% endif %}
+  (
+    {{ sql }}
+  );
+{%- endmacro %}
diff --git a/plugins/postgres/dbt/include/postgres/macros/timestamps.sql b/plugins/postgres/dbt/include/postgres/macros/utils/timestamps.sql
similarity index 100%
rename from plugins/postgres/dbt/include/postgres/macros/timestamps.sql
rename to plugins/postgres/dbt/include/postgres/macros/utils/timestamps.sql