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,8 +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:
-            auth = BasicAuthentication(self.user, self.password)
+            auth = BasicAuthentication(self.user, self.password or "")
         elif self.method.is_kerberos:
             if self.keytab:
                 os.environ["KRB5_CLIENT_KTNAME"] = self.keytab
@@ -2032,9 +2033,11 @@ def _static_connection_kwargs(self) -> t.Dict[str, t.Any]:
         elif self.method.is_oauth:
             auth = OAuth2Authentication()
         elif self.method.is_jwt:
-            auth = JWTAuthentication(self.jwt_token)
+            auth = JWTAuthentication(self.jwt_token or "")
         elif self.method.is_certificate:
-            auth = CertificateAuthentication(self.client_certificate, self.client_private_key)
+            auth = CertificateAuthentication(
+                self.client_certificate or "", self.client_private_key or ""
+            )
         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

@@ -2109,7 +2109,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")