[Bug] Use fully qualified names in rename for tables and views (#1060)

* override create_backup and rename_intermediate to use fully qualified names
* add test cases for backup and intermediate renames

.changes/unreleased/Fixes-20240522-160538.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: 'Rename targets for tables and views use fully qualified names'
+time: 2024-05-22T16:05:38.602074-04:00
+custom:
+  Author: mikealfare
+  Issue: "1031"

dbt/include/snowflake/macros/relations/create_backup.sql

@@ -0,0 +1,12 @@
+{%- macro snowflake__get_create_backup_sql(relation) -%}
+
+    -- get the standard backup name
+    {% set backup_relation = make_backup_relation(relation, relation.type) %}
+
+    -- drop any pre-existing backup
+    {{ get_drop_sql(backup_relation) }};
+
+    -- use `render` to ensure that the fully qualified name is used
+    {{ get_rename_sql(relation, backup_relation.render()) }}
+
+{%- endmacro -%}

dbt/include/snowflake/macros/relations/rename_intermediate.sql

@@ -0,0 +1,9 @@
+{%- macro snowflake__get_rename_intermediate_sql(relation) -%}
+
+    -- get the standard intermediate name
+    {% set intermediate_relation = make_intermediate_relation(relation) %}
+
+    -- use `render` to ensure that the fully qualified name is used
+    {{ get_rename_sql(intermediate_relation, relation.render()) }}
+
+{%- endmacro -%}

dbt/include/snowflake/macros/relations/table/rename.sql

@@ -1,3 +1,13 @@
 {%- macro snowflake__get_rename_table_sql(relation, new_name) -%}
+    /*
+    Rename or move a table to the new name.
+
+    Args:
+        relation: SnowflakeRelation - relation to be renamed
+        new_name: Union[str, SnowflakeRelation] - new name for `relation`
+            if providing a string, the default database/schema will be used if that string is just an identifier
+            if providing a SnowflakeRelation, `render` will be used to produce a fully qualified name
+    Returns: templated string
+    */
     alter table {{ relation }} rename to {{ new_name }}
 {%- endmacro -%}

dbt/include/snowflake/macros/relations/view/rename.sql

@@ -1,3 +1,13 @@
 {%- macro snowflake__get_rename_view_sql(relation, new_name) -%}
+    /*
+    Rename or move a view to the new name.
+
+    Args:
+        relation: SnowflakeRelation - relation to be renamed
+        new_name: Union[str, SnowflakeRelation] - new name for `relation`
+            if providing a string, the default database/schema will be used if that string is just an identifier
+            if providing a SnowflakeRelation, `render` will be used to produce a fully qualified name
+    Returns: templated string
+    */
     alter view {{ relation }} rename to {{ new_name }}
 {%- endmacro -%}

tests/functional/relation_tests/base.py

@@ -0,0 +1,75 @@
+import pytest
+
+from dbt.tests.util import run_dbt, run_dbt_and_capture
+
+
+SEED = """
+id
+0
+1
+2
+""".strip()
+
+
+TABLE = """
+{{ config(materialized="table") }}
+select * from {{ ref('my_seed') }}
+"""
+
+
+VIEW = """
+{{ config(materialized="view") }}
+select * from {{ ref('my_seed') }}
+"""
+
+
+MACRO__GET_CREATE_BACKUP_SQL = """
+{% macro test__get_create_backup_sql(database, schema, identifier, relation_type) -%}
+    {%- set relation = adapter.Relation.create(database=database, schema=schema, identifier=identifier, type=relation_type) -%}
+    {% call statement('test__get_create_backup_sql') -%}
+        {{ get_create_backup_sql(relation) }}
+    {%- endcall %}
+{% endmacro %}"""
+
+
+MACRO__GET_RENAME_INTERMEDIATE_SQL = """
+{% macro test__get_rename_intermediate_sql(database, schema, identifier, relation_type) -%}
+    {%- set relation = adapter.Relation.create(database=database, schema=schema, identifier=identifier, type=relation_type) -%}
+    {% call statement('test__get_rename_intermediate_sql') -%}
+        {{ get_rename_intermediate_sql(relation) }}
+    {%- endcall %}
+{% endmacro %}"""
+
+
+class RelationOperation:
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        yield {"my_seed.csv": SEED}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        yield {
+            "my_table.sql": TABLE,
+            "my_table__dbt_tmp.sql": TABLE,
+            "my_view.sql": VIEW,
+            "my_view__dbt_tmp.sql": VIEW,
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        yield {
+            "test__get_create_backup_sql.sql": MACRO__GET_CREATE_BACKUP_SQL,
+            "test__get_rename_intermediate_sql.sql": MACRO__GET_RENAME_INTERMEDIATE_SQL,
+        }
+
+    @pytest.fixture(scope="class", autouse=True)
+    def setup(self, project):
+        run_dbt(["seed"])
+        run_dbt(["run"])
+
+    def assert_operation(self, project, operation, args, expected_statement):
+        results, logs = run_dbt_and_capture(
+            ["--debug", "run-operation", operation, "--args", str(args)]
+        )
+        assert len(results) == 1
+        assert expected_statement in logs

tests/functional/relation_tests/test_table.py

@@ -0,0 +1,25 @@
+from tests.functional.relation_tests.base import RelationOperation
+
+
+class TestTable(RelationOperation):
+
+    def test_get_create_backup_and_rename_intermediate_sql(self, project):
+        args = {
+            "database": project.database,
+            "schema": project.test_schema,
+            "identifier": "my_table",
+            "relation_type": "table",
+        }
+        expected_statement = (
+            f"alter table {project.database}.{project.test_schema}.my_table "
+            f"rename to {project.database}.{project.test_schema}.my_table__dbt_backup"
+        )
+        self.assert_operation(project, "test__get_create_backup_sql", args, expected_statement)
+
+        expected_statement = (
+            f"alter table {project.database}.{project.test_schema}.my_table__dbt_tmp "
+            f"rename to {project.database}.{project.test_schema}.my_table"
+        )
+        self.assert_operation(
+            project, "test__get_rename_intermediate_sql", args, expected_statement
+        )

tests/functional/relation_tests/test_view.py

@@ -0,0 +1,25 @@
+from tests.functional.relation_tests.base import RelationOperation
+
+
+class TestView(RelationOperation):
+
+    def test_get_create_backup_and_rename_intermediate_sql(self, project):
+        args = {
+            "database": project.database,
+            "schema": project.test_schema,
+            "identifier": "my_view",
+            "relation_type": "view",
+        }
+        expected_statement = (
+            f"alter view {project.database}.{project.test_schema}.my_view "
+            f"rename to {project.database}.{project.test_schema}.my_view__dbt_backup"
+        )
+        self.assert_operation(project, "test__get_create_backup_sql", args, expected_statement)
+
+        expected_statement = (
+            f"alter view {project.database}.{project.test_schema}.my_view__dbt_tmp "
+            f"rename to {project.database}.{project.test_schema}.my_view"
+        )
+        self.assert_operation(
+            project, "test__get_rename_intermediate_sql", args, expected_statement
+        )