feat: Add support for ALTER ICEBERG TABLE in Snowflake by passing table format to the alter_table method.

Author: sineline

sqlmesh/core/engine_adapter/base.py

@@ -1184,6 +1184,7 @@ def get_alter_operations(
     def alter_table(
         self,
         alter_expressions: t.Union[t.List[exp.Alter], t.List[TableAlterOperation]],
+        **kwargs: t.Any,
     ) -> None:
         """
         Performs the alter statements to change the current table into the structure of the target table.

sqlmesh/core/engine_adapter/snowflake.py

@@ -689,6 +689,31 @@ def clone_table(
             **kwargs,
         )
 
+    def alter_table(
+        self,
+        alter_expressions: t.Union[t.List[exp.Alter], t.List["TableAlterOperation"]],
+        **kwargs: t.Any,
+    ) -> None:
+        # Snowflake requires ALTER ICEBERG TABLE instead of ALTER TABLE for Iceberg tables
+        table_format = kwargs.pop("table_format", None)
+        if table_format and isinstance(table_format, str) and table_format.upper() == "ICEBERG":
+            from sqlmesh.core.schema_diff import TableAlterOperation
+
+            resolved_expressions = []
+            for x in alter_expressions:
+                if isinstance(x, TableAlterOperation):
+                    alter_expr = x.expression
+                else:
+                    alter_expr = x
+                alter_expr.args["kind"] = f"{table_format.upper()} TABLE"
+                resolved_expressions.append(alter_expr)
+
+            with self.transaction():
+                for alter_expr in resolved_expressions:
+                    self.execute(alter_expr)
+        else:
+            super().alter_table(alter_expressions, **kwargs)
+
     @t.overload
     def _columns_to_types(
         self,

sqlmesh/core/snapshot/evaluator.py

@@ -2109,7 +2109,7 @@ def migrate(
         _check_additive_schema_change(
             snapshot, alter_operations, kwargs["allow_additive_snapshots"]
         )
-        self.adapter.alter_table(alter_operations)
+        self.adapter.alter_table(alter_operations, table_format=snapshot.model.table_format)
 
         # Apply grants after schema migration
         deployability_index = kwargs.get("deployability_index")