Feat(dbt): Add support for adapter.rename_relation

docs/integrations/dbt.md

@@ -355,7 +355,7 @@ SQLMesh supports running dbt projects using the majority of dbt jinja methods, i
 | builtins    | modules        | source       |         |
 | config      | print          | statement    |         |
 
-\* `adapter.rename_relation` and `adapter.expand_target_column_types` are not currently supported.
+\* `adapter.expand_target_column_types` is not currently supported.
 
 ## Unsupported dbt jinja methods
 
@@ -364,7 +364,6 @@ The dbt jinja methods that are not currently supported are:
 * debug
 * selected_sources
 * adapter.expand_target_column_types
-* adapter.rename_relation
 * graph.nodes.values
 * graph.metrics.values
 

sqlmesh/dbt/adapter.py

@@ -15,7 +15,6 @@
 
 if t.TYPE_CHECKING:
     import agate
-    import pandas as pd
     from dbt.adapters.base import BaseRelation
     from dbt.adapters.base.column import Column
     from dbt.adapters.base.impl import AdapterResponse
@@ -86,6 +85,10 @@ def drop_schema(self, relation: BaseRelation) -> None:
     def drop_relation(self, relation: BaseRelation) -> None:
         """Drops a relation (table) in the target database."""
 
+    @abc.abstractmethod
+    def rename_relation(self, from_relation: BaseRelation, to_relation: BaseRelation) -> None:
+        """Renames a relation (table) in the target database."""
+
     @abc.abstractmethod
     def execute(
         self, sql: str, auto_begin: bool = False, fetch: bool = False
@@ -210,6 +213,9 @@ def drop_schema(self, relation: BaseRelation) -> None:
     def drop_relation(self, relation: BaseRelation) -> None:
         self._raise_parsetime_adapter_call_error("drop relation")
 
+    def rename_relation(self, from_relation: BaseRelation, to_relation: BaseRelation) -> None:
+        self._raise_parsetime_adapter_call_error("rename relation")
+
     def execute(
         self, sql: str, auto_begin: bool = False, fetch: bool = False
     ) -> t.Tuple[AdapterResponse, agate.Table]:
@@ -349,6 +355,12 @@ def drop_relation(self, relation: BaseRelation) -> None:
         if relation.schema is not None and relation.identifier is not None:
             self.engine_adapter.drop_table(self._normalize(self._relation_to_table(relation)))
 
+    def rename_relation(self, from_relation: BaseRelation, to_relation: BaseRelation) -> None:
+        old_table_name = self._normalize(self._relation_to_table(from_relation))
+        new_table_name = self._normalize(self._relation_to_table(to_relation))
+
+        self.engine_adapter.rename_table(old_table_name, new_table_name)
+
     def execute(
         self, sql: str, auto_begin: bool = False, fetch: bool = False
     ) -> t.Tuple[AdapterResponse, agate.Table]:

tests/dbt/test_adapter.py

@@ -81,6 +81,19 @@ def test_adapter_relation(sushi_test_project: Project, runtime_renderer: t.Calla
         == "[]"
     )
 
+    renderer("""
+        {%- set old_relation = adapter.get_relation(
+              database=None,
+              schema='foo',
+              identifier='bar') -%}
+
+        {%- set backup_relation = api.Relation.create(schema='foo', identifier='bar__backup') -%}
+
+        {% do adapter.rename_relation(old_relation, backup_relation) %}
+    """)
+    assert not engine_adapter.table_exists("foo.bar")
+    assert engine_adapter.table_exists("foo.bar__backup")
+
 
 def test_bigquery_get_columns_in_relation(
     sushi_test_project: Project,