ADAP-869: Support atomic replace in replace macro (#8539)

* move config changes into alter.sql in alignment with other adapters
* move shared relations macros to relations root
* move single models files to models root
* add table to replace
* move create file into relation directory
* implement replace for postgres
* move column specific macros into column directory
* add unit test for can_be_replaced
* update renameable_relations and replaceable_relations to frozensets to set defaults
* fixed tests for new defaults

.changes/unreleased/Features-20230831-204804.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support atomic replace in the global replace macro
+time: 2023-08-31T20:48:04.098933-04:00
+custom:
+  Author: mikealfare
+  Issue: "8539"

core/dbt/adapters/base/relation.py

@@ -1,6 +1,6 @@
 from collections.abc import Hashable
 from dataclasses import dataclass, field
-from typing import Optional, TypeVar, Any, Type, Dict, Iterator, Tuple, Set, List
+from typing import Optional, TypeVar, Any, Type, Dict, Iterator, Tuple, Set, FrozenSet
 
 from dbt.contracts.graph.nodes import SourceDefinition, ManifestNode, ResultNode, ParsedNode
 from dbt.contracts.relation import (
@@ -36,9 +36,9 @@ class BaseRelation(FakeAPIObject, Hashable):
     quote_policy: Policy = field(default_factory=lambda: Policy())
     dbt_created: bool = False
     # register relation types that can be renamed for the purpose of replacing relations using stages and backups
-    renameable_relations: List[str] = field(
-        default_factory=lambda: [RelationType.Table, RelationType.View]
-    )
+    renameable_relations: FrozenSet[str] = frozenset()
+    # register relation types that are replaceable, i.e. they have "create or replace" capability
+    replaceable_relations: FrozenSet[str] = frozenset()
 
     def _is_exactish_match(self, field: ComponentName, value: str) -> bool:
         if self.dbt_created and self.quote_policy.get_part(field) is False:
@@ -290,9 +290,13 @@ def create(
         return cls.from_dict(kwargs)
 
     @property
-    def can_be_renamed(self):
+    def can_be_renamed(self) -> bool:
         return self.type in self.renameable_relations
 
+    @property
+    def can_be_replaced(self) -> bool:
+        return self.type in self.replaceable_relations
+
     def __repr__(self) -> str:
         return "<{} {}>".format(self.__class__.__name__, self.render())
 

core/dbt/include/global_project/macros/relations/materialized_view/alter.sql

@@ -28,3 +28,28 @@
 ) %}
     {{ exceptions.raise_compiler_error("Materialized views have not been implemented for this adapter.") }}
 {% endmacro %}
+
+
+{% macro get_materialized_view_configuration_changes(existing_relation, new_config) %}
+    /* {#
+    It's recommended that configuration changes be formatted as follows:
+    {"<change_category>": [{"action": "<name>", "context": ...}]}
+
+    For example:
+    {
+        "indexes": [
+            {"action": "drop", "context": "index_abc"},
+            {"action": "create", "context": {"columns": ["column_1", "column_2"], "type": "hash", "unique": True}},
+        ],
+    }
+
+    Either way, `get_materialized_view_configuration_changes` needs to align with `get_alter_materialized_view_as_sql`.
+    #} */
+    {{- log('Determining configuration changes on: ' ~ existing_relation) -}}
+    {%- do return(adapter.dispatch('get_materialized_view_configuration_changes', 'dbt')(existing_relation, new_config)) -%}
+{% endmacro %}
+
+
+{% macro default__get_materialized_view_configuration_changes(existing_relation, new_config) %}
+    {{ exceptions.raise_compiler_error("Materialized views have not been implemented for this adapter.") }}
+{% endmacro %}

core/dbt/include/global_project/macros/relations/materialized_view/replace.sql

@@ -0,0 +1,10 @@
+{% macro get_replace_materialized_view_sql(relation, sql) %}
+    {{- adapter.dispatch('get_replace_materialized_view_sql', 'dbt')(relation, sql) -}}
+{% endmacro %}
+
+
+{% macro default__get_replace_materialized_view_sql(relation, sql) %}
+    {{ exceptions.raise_compiler_error(
+        "`get_replace_materialized_view_sql` has not been implemented for this adapter."
+    ) }}
+{% endmacro %}

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

@@ -6,14 +6,29 @@
 
 {% macro default__get_replace_sql(existing_relation, target_relation, sql) %}
 
+    {# /* use a create or replace statement if possible */ #}
+
+    {% set is_replaceable = existing_relation.type == target_relation_type and existing_relation.is_replaceable %}
+
+    {% if is_replaceable and existing_relation.is_view %}
+        {{ get_replace_view_sql(target_relation, sql) }}
+
+    {% elif is_replaceable and existing_relation.is_table %}
+        {{ get_replace_table_sql(target_relation, sql) }}
+
+    {% elif is_replaceable and existing_relation.is_materialized_view %}
+        {{ get_replace_materialized_view_sql(target_relation, sql) }}
+
+    {# /* a create or replace statement is not possible, so try to stage and/or backup to be safe */ #}
+
     {# /* create target_relation as an intermediate relation, then swap it out with the existing one using a backup */ #}
-    {%- if target_relation.can_be_renamed and existing_relation.can_be_renamed -%}
+    {%- elif target_relation.can_be_renamed and existing_relation.can_be_renamed -%}
         {{ get_create_intermediate_sql(target_relation, sql) }};
         {{ get_create_backup_sql(existing_relation) }};
         {{ get_rename_intermediate_sql(target_relation) }};
         {{ get_drop_backup_sql(existing_relation) }}
 
-    {# /* create target_relation as an intermediate relation, then swap it out with the existing one using drop */ #}
+    {# /* create target_relation as an intermediate relation, then swap it out with the existing one without using a backup */ #}
     {%- elif target_relation.can_be_renamed -%}
         {{ get_create_intermediate_sql(target_relation, sql) }};
         {{ get_drop_sql(existing_relation) }};

core/dbt/include/global_project/macros/relations/table/replace.sql

@@ -0,0 +1,10 @@
+{% macro get_replace_table_sql(relation, sql) %}
+    {{- adapter.dispatch('get_replace_table_sql', 'dbt')(relation, sql) -}}
+{% endmacro %}
+
+
+{% macro default__get_replace_table_sql(relation, sql) %}
+    {{ exceptions.raise_compiler_error(
+        "`get_replace_table_sql` has not been implemented for this adapter."
+    ) }}
+{% endmacro %}

core/dbt/include/global_project/macros/materializations/models/view/create_or_replace_view.sql → core/dbt/include/global_project/macros/relations/view/replace.sql

@@ -1,3 +1,15 @@
+{% macro get_replace_view_sql(relation, sql) %}
+    {{- adapter.dispatch('get_replace_view_sql', 'dbt')(relation, sql) -}}
+{% endmacro %}
+
+
+{% macro default__get_replace_view_sql(relation, sql) %}
+    {{ exceptions.raise_compiler_error(
+        "`get_replace_view_sql` has not been implemented for this adapter."
+    ) }}
+{% endmacro %}
+
+
 /* {#
        Core materialization implementation. BigQuery and Snowflake are similar
        because both can use `create or replace view` where the resulting view's columns
@@ -42,3 +54,13 @@
   {{ return({'relations': [target_relation]}) }}
 
 {% endmacro %}
+
+
+{% macro handle_existing_table(full_refresh, old_relation) %}
+    {{ adapter.dispatch('handle_existing_table', 'dbt')(full_refresh, old_relation) }}
+{% endmacro %}
+
+{% macro default__handle_existing_table(full_refresh, old_relation) %}
+    {{ log("Dropping relation " ~ old_relation ~ " because it is of type " ~ old_relation.type) }}
+    {{ adapter.drop_relation(old_relation) }}
+{% endmacro %}

plugins/postgres/dbt/adapters/postgres/relation.py

@@ -21,11 +21,19 @@
 
 @dataclass(frozen=True, eq=False, repr=False)
 class PostgresRelation(BaseRelation):
-    relations_that_can_be_renamed = [
-        RelationType.View,
-        RelationType.Table,
-        RelationType.MaterializedView,
-    ]
+    renameable_relations = frozenset(
+        {
+            RelationType.View,
+            RelationType.Table,
+            RelationType.MaterializedView,
+        }
+    )
+    replaceable_relations = frozenset(
+        {
+            RelationType.View,
+            RelationType.Table,
+        }
+    )
 
     def __post_init__(self):
         # Check for length of Postgres table/view names.

plugins/postgres/dbt/include/postgres/macros/relations/materialized_view/alter.sql

@@ -41,3 +41,10 @@
     {%- endfor -%}
 
 {%- endmacro -%}
+
+
+{% 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) %}
+    {% do return(_configuration_changes) %}
+{% endmacro %}

plugins/postgres/dbt/include/postgres/macros/relations/table/replace.sql

@@ -0,0 +1,17 @@
+{% macro postgres__get_replace_table_sql(relation, sql) -%}
+
+    {%- set sql_header = config.get('sql_header', none) -%}
+    {{ sql_header if sql_header is not none }}
+
+    create or replace table {{ relation }}
+        {% set contract_config = config.get('contract') %}
+        {% if contract_config.enforced %}
+            {{ get_assert_columns_equivalent(sql) }}
+            {{ get_table_columns_and_constraints() }}
+            {%- set sql = get_select_subquery(sql) %}
+        {% endif %}
+    as (
+        {{ sql }}
+    );
+
+{%- endmacro %}

plugins/postgres/dbt/include/postgres/macros/relations/view/replace.sql

@@ -0,0 +1,15 @@
+{% macro postgres__get_replace_view_sql(relation, sql) -%}
+
+    {%- set sql_header = config.get('sql_header', none) -%}
+    {{ sql_header if sql_header is not none }}
+
+    create or replace view {{ relation }}
+        {% set contract_config = config.get('contract') %}
+        {% if contract_config.enforced %}
+            {{ get_assert_columns_equivalent(sql) }}
+        {%- endif %}
+    as (
+        {{ sql }}
+    );
+
+{%- endmacro %}

tests/unit/test_relation.py

@@ -1,3 +1,5 @@
+from dataclasses import replace
+
 import pytest
 
 from dbt.adapters.base import BaseRelation
@@ -13,4 +15,28 @@
 )
 def test_can_be_renamed(relation_type, result):
     my_relation = BaseRelation.create(type=relation_type)
+    my_relation = replace(my_relation, renameable_relations=frozenset({RelationType.View}))
     assert my_relation.can_be_renamed is result
+
+
+def test_can_be_renamed_default():
+    my_relation = BaseRelation.create(type=RelationType.View)
+    assert my_relation.can_be_renamed is False
+
+
+@pytest.mark.parametrize(
+    "relation_type,result",
+    [
+        (RelationType.View, True),
+        (RelationType.External, False),
+    ],
+)
+def test_can_be_replaced(relation_type, result):
+    my_relation = BaseRelation.create(type=relation_type)
+    my_relation = replace(my_relation, replaceable_relations=frozenset({RelationType.View}))
+    assert my_relation.can_be_replaced is result
+
+
+def test_can_be_replaced_default():
+    my_relation = BaseRelation.create(type=RelationType.View)
+    assert my_relation.can_be_replaced is False