feat: support cloning Iceberg tables on Snowflake by passing table format to the engine adapter.

Author: sineline

sqlmesh/core/engine_adapter/base.py

@@ -1077,10 +1077,11 @@ def clone_table(
             raise NotImplementedError(f"Engine does not support cloning: {type(self)}")
 
         kwargs.pop("rendered_physical_properties", None)
+        table_kind = kwargs.pop("table_kind", "TABLE")
         self.execute(
             exp.Create(
                 this=exp.to_table(target_table_name),
-                kind="TABLE",
+                kind=table_kind,
                 replace=replace,
                 exists=exists,
                 clone=exp.Clone(

sqlmesh/core/engine_adapter/snowflake.py

@@ -675,6 +675,12 @@ def clone_table(
             if isinstance(table_type, exp.TransientProperty):
                 kwargs["properties"] = exp.Properties(expressions=[table_type])
 
+        # Snowflake requires CREATE ICEBERG TABLE ... CLONE for Iceberg tables
+        # instead of the regular CREATE TABLE ... CLONE
+        table_format = kwargs.pop("table_format", None)
+        if table_format and isinstance(table_format, str) and table_format.upper() == "ICEBERG":
+            kwargs["table_kind"] = f"{table_format.upper()} TABLE"
+
         super().clone_table(
             target_table_name,
             source_table_name,

sqlmesh/core/snapshot/evaluator.py

@@ -1085,6 +1085,7 @@ def _clone_snapshot_in_dev(
                 target_table_name,
                 snapshot.table_name(),
                 rendered_physical_properties=rendered_physical_properties,
+                table_format=snapshot.model.table_format,
             )
             self._migrate_target_table(
                 target_table_name=target_table_name,