Fix(snowflake): Support cloning transient tables for dev previews (#4155)

Author: themisvaltinos

sqlmesh/core/engine_adapter/base.py

@@ -890,6 +890,8 @@ def clone_table(
         """
         if not self.SUPPORTS_CLONING:
             raise NotImplementedError(f"Engine does not support cloning: {type(self)}")
+
+        kwargs.pop("rendered_physical_properties", None)
         self.execute(
             exp.Create(
                 this=exp.to_table(target_table_name),

sqlmesh/core/engine_adapter/snowflake.py

@@ -469,3 +469,25 @@ def _create_column_comments(
                 f"Column comments for table '{table.alias_or_name}' not registered - this may be due to limited permissions.",
                 exc_info=True,
             )
+
+    def clone_table(
+        self,
+        target_table_name: TableName,
+        source_table_name: TableName,
+        replace: bool = False,
+        clone_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
+        **kwargs: t.Any,
+    ) -> None:
+        # The Snowflake adapter should use the transient property to clone transient tables
+        if physical_properties := kwargs.get("rendered_physical_properties"):
+            table_type = self._pop_creatable_type_from_properties(physical_properties)
+            if isinstance(table_type, exp.TransientProperty):
+                kwargs["properties"] = exp.Properties(expressions=[table_type])
+
+        super().clone_table(
+            target_table_name,
+            source_table_name,
+            replace=replace,
+            clone_kwargs=clone_kwargs,
+            **kwargs,
+        )

sqlmesh/core/snapshot/evaluator.py

@@ -830,7 +830,12 @@ def _create_snapshot(
                 )
 
                 try:
-                    adapter.clone_table(target_table_name, snapshot.table_name(), replace=True)
+                    adapter.clone_table(
+                        target_table_name,
+                        snapshot.table_name(),
+                        replace=True,
+                        rendered_physical_properties=rendered_physical_properties,
+                    )
                     alter_expressions = adapter.get_alter_expressions(
                         target_table_name, tmp_table_name
                     )

tests/core/engine_adapter/test_snowflake.py

@@ -7,6 +7,7 @@
 
 import sqlmesh.core.dialect as d
 from sqlmesh.core.dialect import normalize_model_name
+from sqlmesh.core.engine_adapter.base import EngineAdapter
 from sqlmesh.core.model import load_sql_based_model
 from sqlmesh.core.engine_adapter import SnowflakeEngineAdapter
 from sqlmesh.core.model.definition import SqlModel
@@ -634,3 +635,34 @@ def test_create_view(make_mocked_engine_adapter: t.Callable):
         'CREATE OR REPLACE VIEW "test_view" COPY GRANTS AS SELECT 1',
         'CREATE VIEW "test_view" AS SELECT 1',
     ]
+
+
+def test_clone_table(mocker: MockerFixture, make_mocked_engine_adapter: t.Callable):
+    mocker.patch("sqlmesh.core.engine_adapter.snowflake.SnowflakeEngineAdapter.set_current_catalog")
+    adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter, default_catalog="test_catalog")
+    adapter.clone_table("target_table", "source_table")
+    adapter.cursor.execute.assert_called_once_with(
+        'CREATE TABLE "target_table" CLONE "source_table"'
+    )
+
+    # Validate with transient type we create the clone table accordingly
+    rendered_physical_properties = {
+        "creatable_type": exp.column("transient"),
+    }
+    adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter, default_catalog="test_catalog")
+    adapter.clone_table(
+        "target_table", "source_table", rendered_physical_properties=rendered_physical_properties
+    )
+    adapter.cursor.execute.assert_called_once_with(
+        'CREATE TRANSIENT TABLE "target_table" CLONE "source_table"'
+    )
+
+    # Validate other engine adapters would work as usual even when we pass the properties
+    adapter = make_mocked_engine_adapter(EngineAdapter, default_catalog="test_catalog")
+    adapter.SUPPORTS_CLONING = True
+    adapter.clone_table(
+        "target_table", "source_table", rendered_physical_properties=rendered_physical_properties
+    )
+    adapter.cursor.execute.assert_called_once_with(
+        'CREATE TABLE "target_table" CLONE "source_table"'
+    )

tests/core/test_snapshot_evaluator.py

@@ -1503,6 +1503,7 @@ def test_create_clone_in_dev(mocker: MockerFixture, adapter_mock, make_snapshot)
         f"sqlmesh__test_schema.test_schema__test_model__{snapshot.version}__dev",
         f"sqlmesh__test_schema.test_schema__test_model__{snapshot.version}",
         replace=True,
+        rendered_physical_properties={},
     )
 
     adapter_mock.get_alter_expressions.assert_called_once_with(
@@ -1601,6 +1602,7 @@ def test_drop_clone_in_dev_when_migration_fails(mocker: MockerFixture, adapter_m
         f"sqlmesh__test_schema.test_schema__test_model__{snapshot.version}__dev",
         f"sqlmesh__test_schema.test_schema__test_model__{snapshot.version}",
         replace=True,
+        rendered_physical_properties={},
     )
 
     adapter_mock.get_alter_expressions.assert_called_once_with(