feat: Implement special DDL generation for Snowflake Iceberg tables with PARTITION BY to correctly handle property ordering and CTAS limitations.

Author: sineline

sqlmesh/core/config/connection.py

@@ -2013,7 +2013,9 @@ def _static_connection_kwargs(self) -> t.Dict[str, t.Any]:
             OAuth2Authentication,
         )
 
+        auth: t.Any
         if self.method.is_basic or self.method.is_ldap:
+            assert self.password is not None
             auth = BasicAuthentication(self.user, self.password)
         elif self.method.is_kerberos:
             if self.keytab:
@@ -2032,8 +2034,11 @@ def _static_connection_kwargs(self) -> t.Dict[str, t.Any]:
         elif self.method.is_oauth:
             auth = OAuth2Authentication()
         elif self.method.is_jwt:
+            assert self.jwt_token is not None
             auth = JWTAuthentication(self.jwt_token)
         elif self.method.is_certificate:
+            assert self.client_certificate is not None
+            assert self.client_private_key is not None
             auth = CertificateAuthentication(self.client_certificate, self.client_private_key)
         else:
             auth = None

sqlmesh/core/engine_adapter/base.py

@@ -295,8 +295,8 @@ def _get_source_queries(
                         )
                         for c in target_columns_to_types
                     ]
-                    query_factory = (
-                        lambda: exp.Select()
+                    query_factory = lambda: (
+                        exp.Select()
                         .select(*select_columns)
                         .from_(query_or_df.subquery("select_source_columns"))
                     )
@@ -1184,7 +1184,6 @@ 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

@@ -42,6 +42,7 @@
         SnowparkSession,
     )
     from sqlmesh.core.node import IntervalUnit
+    from sqlmesh.core.schema_diff import TableAlterOperation
 
 
 @set_catalog(
@@ -689,7 +690,7 @@ def clone_table(
             **kwargs,
         )
 
-    def alter_table(
+    def alter_table(  # type: ignore[override]
         self,
         alter_expressions: t.Union[t.List[exp.Alter], t.List["TableAlterOperation"]],
         **kwargs: t.Any,
@@ -712,7 +713,7 @@ def alter_table(
                 for alter_expr in resolved_expressions:
                     self.execute(alter_expr)
         else:
-            super().alter_table(alter_expressions, **kwargs)
+            super().alter_table(alter_expressions)
 
     @t.overload
     def _columns_to_types(

sqlmesh/core/snapshot/evaluator.py

@@ -1081,11 +1081,14 @@ def _clone_snapshot_in_dev(
 
         try:
             logger.info(f"Cloning table '{source_table_name}' into '{target_table_name}'")
+            clone_kwargs: t.Dict[str, t.Any] = {}
+            if snapshot.model.table_format:
+                clone_kwargs["table_format"] = snapshot.model.table_format
             adapter.clone_table(
                 target_table_name,
                 snapshot.table_name(),
                 rendered_physical_properties=rendered_physical_properties,
-                table_format=snapshot.model.table_format,
+                **clone_kwargs,
             )
             self._migrate_target_table(
                 target_table_name=target_table_name,
@@ -2109,7 +2112,10 @@ def migrate(
         _check_additive_schema_change(
             snapshot, alter_operations, kwargs["allow_additive_snapshots"]
         )
-        self.adapter.alter_table(alter_operations, table_format=snapshot.model.table_format)
+        alter_kwargs: t.Dict[str, t.Any] = {}
+        if snapshot.model.table_format:
+            alter_kwargs["table_format"] = snapshot.model.table_format
+        self.adapter.alter_table(alter_operations, **alter_kwargs)
 
         # Apply grants after schema migration
         deployability_index = kwargs.get("deployability_index")