Merge branch 'main' into fix/6017-resolve-table-mapping

Author: mday-io

.devcontainer/devcontainer.json

@@ -11,10 +11,7 @@
   "postCreateCommand": "bash .devcontainer/post-create-command.sh",
   "customizations": {
     "vscode": {
-      "extensions": [
-        "ms-python.python",
-        "ms-python.vscode-pylance"
-      ]
+      "extensions": ["ms-python.python", "ms-python.vscode-pylance"]
     }
   },
   "remoteUser": "vscode"

.github/workflows/pr.yaml

@@ -16,6 +16,7 @@ jobs:
     outputs:
       python: ${{ steps.filter.outputs.python }}
       client: ${{ steps.filter.outputs.client }}
+      vscode: ${{ steps.filter.outputs.vscode }}
       ci: ${{ steps.filter.outputs.ci }}
     steps:
       - uses: actions/checkout@v7
@@ -34,6 +35,8 @@ jobs:
               - 'pyproject.toml'
             client:
               - 'web/client/**'
+            vscode:
+              - 'vscode/**'
             ci:
               - '.github/**'
               - 'Makefile'
@@ -188,9 +191,10 @@ jobs:
 
   ui-style:
     needs: [changes]
-    if: false
-      # needs.changes.outputs.client == 'true' || needs.changes.outputs.ci ==
-      # 'true' || github.ref == 'refs/heads/main'
+    if:
+      needs.changes.outputs.client == 'true' || needs.changes.outputs.vscode ==
+      'true' || needs.changes.outputs.ci == 'true' || github.ref ==
+      'refs/heads/main'
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v7
@@ -252,7 +256,17 @@ jobs:
       fail-fast: false
       matrix:
         engine:
-          [duckdb, postgres, mysql, mssql, trino, spark, clickhouse, risingwave, starrocks]
+          [
+            duckdb,
+            postgres,
+            mysql,
+            mssql,
+            trino,
+            spark,
+            clickhouse,
+            risingwave,
+            starrocks,
+          ]
     env:
       PYTEST_XDIST_AUTO_NUM_WORKERS: 2
       SQLMESH__DISABLE_ANONYMIZED_ANALYTICS: '1'
@@ -393,10 +407,13 @@ jobs:
           retention-days: 7
 
   test-vscode:
+    needs: changes
+    if:
+      needs.changes.outputs.vscode == 'true' || needs.changes.outputs.ci ==
+      'true' || github.ref == 'refs/heads/main'
     env:
       PLAYWRIGHT_SKIP_BROWSER_DOWNLOAD: 1
     runs-on: ubuntu-latest
-    if: false
     steps:
       - uses: actions/checkout@v7
       - uses: actions/setup-node@v7
@@ -457,7 +474,19 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        dbt-version: ['1.3', '1.4', '1.5', '1.6', '1.7', '1.8', '1.9', '1.10', '1.11', '1.12']
+        dbt-version:
+          [
+            '1.3',
+            '1.4',
+            '1.5',
+            '1.6',
+            '1.7',
+            '1.8',
+            '1.9',
+            '1.10',
+            '1.11',
+            '1.12',
+          ]
     steps:
       - uses: actions/checkout@v7
       - name: Set up Python

sqlmesh/core/engine_adapter/base.py

@@ -1094,6 +1094,8 @@ def clone_table(
         replace: bool = False,
         exists: bool = True,
         clone_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
+        table_format: t.Optional[str] = None,
+        table_kind: t.Optional[str] = None,
         **kwargs: t.Any,
     ) -> None:
         """Creates a table with the target name by cloning the source table.
@@ -1103,6 +1105,10 @@ def clone_table(
             source_table_name: The name of the source table that should be cloned.
             replace: Whether or not to replace an existing table.
             exists: Indicates whether to include the IF NOT EXISTS check.
+            clone_kwargs: Additional arguments for the CLONE clause.
+            table_format: The table format of the source table, if any. Engines that require
+                format-specific DDL to clone a table use it to derive `table_kind`.
+            table_kind: The kind of table to create. Defaults to `TABLE`.
         """
         if not self.SUPPORTS_CLONING:
             raise NotImplementedError(f"Engine does not support cloning: {type(self)}")
@@ -1111,7 +1117,7 @@ def clone_table(
         self.execute(
             exp.Create(
                 this=exp.to_table(target_table_name),
-                kind="TABLE",
+                kind=table_kind or "TABLE",
                 replace=replace,
                 exists=exists,
                 clone=exp.Clone(
@@ -1214,9 +1220,15 @@ def get_alter_operations(
     def alter_table(
         self,
         alter_expressions: t.Union[t.List[exp.Alter], t.List[TableAlterOperation]],
+        table_format: t.Optional[str] = None,
     ) -> None:
         """
         Performs the alter statements to change the current table into the structure of the target table.
+
+        Args:
+            alter_expressions: The alter operations to apply.
+            table_format: The table format of the target table, if any. Engines that require
+                format-specific DDL to alter a table use it to adjust the generated statements.
         """
         with self.transaction():
             for alter_expression in [

sqlmesh/core/engine_adapter/bigquery.py

@@ -405,6 +405,7 @@ def create_mapping_schema(
     def alter_table(
         self,
         alter_expressions: t.Union[t.List[exp.Alter], t.List[TableAlterOperation]],
+        table_format: t.Optional[str] = None,
     ) -> None:
         """
         Performs the alter statements to change the current table into the structure of the target table,

sqlmesh/core/engine_adapter/clickhouse.py

@@ -699,6 +699,7 @@ def delete_from(self, table_name: TableName, where: t.Union[str, exp.Expr]) -> N
     def alter_table(
         self,
         alter_expressions: t.Union[t.List[exp.Alter], t.List[TableAlterOperation]],
+        table_format: t.Optional[str] = None,
     ) -> None:
         """
         Performs the alter statements to change the current table into the structure of the target table.

sqlmesh/core/engine_adapter/databricks.py

@@ -386,6 +386,8 @@ def clone_table(
         replace: bool = False,
         exists: bool = True,
         clone_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
+        table_format: t.Optional[str] = None,
+        table_kind: t.Optional[str] = None,
         **kwargs: t.Any,
     ) -> None:
         clone_kwargs = clone_kwargs or {}
@@ -395,6 +397,8 @@ def clone_table(
             source_table_name,
             replace=replace,
             clone_kwargs=clone_kwargs,
+            table_format=table_format,
+            table_kind=table_kind,
             **kwargs,
         )
 

sqlmesh/core/engine_adapter/fabric.py

@@ -225,7 +225,9 @@ def set_current_catalog(self, catalog_name: t.Optional[str]) -> None:
         self._target_catalog = target_catalog
 
     def alter_table(
-        self, alter_expressions: t.Union[t.List[exp.Alter], t.List[TableAlterOperation]]
+        self,
+        alter_expressions: t.Union[t.List[exp.Alter], t.List[TableAlterOperation]],
+        table_format: t.Optional[str] = None,
     ) -> None:
         """
         Applies alter expressions to a table. Fabric has limited support for ALTER TABLE,

sqlmesh/core/engine_adapter/snowflake.py

@@ -24,6 +24,7 @@
     SourceQuery,
     set_catalog,
 )
+from sqlmesh.core.schema_diff import TableAlterOperation
 from sqlmesh.utils import optional_import, get_source_columns_to_types
 from sqlmesh.utils.errors import SQLMeshError
 from sqlmesh.utils.pandas import columns_to_types_from_dtypes
@@ -667,6 +668,8 @@ def clone_table(
         replace: bool = False,
         exists: bool = True,
         clone_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
+        table_format: t.Optional[str] = None,
+        table_kind: t.Optional[str] = None,
         **kwargs: t.Any,
     ) -> None:
         # The Snowflake adapter should use the transient property to clone transient tables
@@ -675,14 +678,43 @@ def clone_table(
             if isinstance(table_type, exp.TransientProperty):
                 kwargs["properties"] = exp.Properties(expressions=[table_type])
 
+        # Snowflake rejects `CREATE TABLE ... CLONE` for Iceberg tables, it requires
+        # `CREATE ICEBERG TABLE ... CLONE` instead
+        if table_format and not table_kind:
+            table_kind = f"{table_format.upper()} TABLE"
+
         super().clone_table(
             target_table_name,
             source_table_name,
             replace=replace,
             clone_kwargs=clone_kwargs,
+            table_kind=table_kind,
             **kwargs,
         )
 
+    def alter_table(
+        self,
+        alter_expressions: t.Union[t.List[exp.Alter], t.List[TableAlterOperation]],
+        table_format: t.Optional[str] = None,
+    ) -> None:
+        # Snowflake rejects `ALTER TABLE` for Iceberg tables, it requires
+        # `ALTER ICEBERG TABLE` instead
+        if table_format:
+            table_kind = f"{table_format.upper()} TABLE"
+            resolved_expressions = []
+            for alter_expression in alter_expressions:
+                resolved_expression = (
+                    alter_expression.expression
+                    if isinstance(alter_expression, TableAlterOperation)
+                    else alter_expression.copy()
+                )
+                resolved_expression.set("kind", table_kind)
+                resolved_expressions.append(resolved_expression)
+
+            super().alter_table(resolved_expressions)
+        else:
+            super().alter_table(alter_expressions)
+
     @t.overload
     def _columns_to_types(
         self,

sqlmesh/core/snapshot/evaluator.py

@@ -1104,6 +1104,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,
@@ -2161,7 +2162,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")

sqlmesh/core/test/definition.py

@@ -263,12 +263,9 @@ def assert_equal(
         for col, value in object_sentinel_values.items():
             try:
                 # can't use `isinstance()` here - https://stackoverflow.com/a/68743663/1707525
-                if type(value) is datetime.date:
-                    expected[col] = pd.to_datetime(expected[col]).dt.date
-                elif type(value) is datetime.time:
-                    expected[col] = pd.to_datetime(expected[col]).dt.time
-                elif type(value) is datetime.datetime:
-                    expected[col] = pd.to_datetime(expected[col]).dt.to_pydatetime()
+                value_type = type(value)
+                if value_type in (datetime.date, datetime.time, datetime.datetime):
+                    expected[col] = _parse_expected_datetime_column(expected[col], value_type)
             except Exception as e:
                 from sqlmesh.core.console import get_console
 
@@ -1014,6 +1011,34 @@ def _raise_error(msg: str, path: Path | None = None) -> None:
     raise TestError(f"Failed to run test:\n{msg}")
 
 
+def _parse_expected_datetime_column(series: pd.Series, target_type: type) -> pd.Series:
+    """Convert a series of expected values to python ``date``/``time``/``datetime``.
+
+    Falls back to microsecond resolution when pandas' default nanosecond
+    parsing overflows. SQL ``TIMESTAMP`` columns can carry values outside
+    pandas' default ``datetime64[ns]`` range (1677-09-21..2262-04-11), so
+    unit tests may compare against values like ``0001-01-01`` which are
+    valid in the database but overflow the default resolution.
+    """
+    import pandas as pd
+    from pandas.errors import OutOfBoundsDatetime
+
+    try:
+        parsed = pd.to_datetime(series)
+    except OutOfBoundsDatetime:
+        parsed = series.astype("datetime64[us]")
+
+    if target_type is datetime.date:
+        return parsed.dt.date
+    if target_type is datetime.time:
+        return parsed.dt.time
+    # `Series.dt.to_pydatetime()` returns an `ndarray` in pandas 2.x. Wrap it in a
+    # Series with ``dtype=object`` so pandas does not coerce the values back to
+    # ``pd.Timestamp`` (which would reintroduce the nanosecond overflow this
+    # function exists to avoid).
+    return pd.Series(parsed.dt.to_pydatetime(), index=parsed.index, dtype="object")
+
+
 def _normalize_df_value(value: t.Any) -> t.Any:
     """Normalize data in a pandas dataframe so ruamel and sqlglot can deal with it."""
     import numpy as np