Repository: TobikoData/sqlmesh
Pull request #2790 (2024-06-18 19:17:22)

Fix: Respect columns-to-types mapping provided by a user when replacing a table
When user explicitly specifies the columns-to-types mapping in the `columns` attribute, SQLMesh respects it when creating the snapshot table for the first time.

However, for `FULL` models SQLMesh uses the `replace_query` API during model evaluation, which recreates the target table from scratch using the model's query.  If the upstream types are different from the ones specified in the `columns` attribute, the recreated table might end up having a different set of types than it had at creation time. 

The impact is not limited to just `FULL` models though. SQLMesh relies on the `replace_query` API when restating models that only support full history restatements (eg. `INCREMETNAL_BY_UNIQUE_KEY`, etc. models). Because of this, the column types may change unexpectedly for models of almost any kind.

This fix ensures that SQLMesh coerces column types to the ones provided by a user when creating / replacing a table from a query.


Patch:
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index 03384b70..7f41302e 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -631,6 +631,10 @@ class EngineAdapter:
         with self.transaction(condition=len(source_queries) > 1):
             for i, source_query in enumerate(source_queries):
                 with source_query as query:
+                    if columns_to_types and columns_to_types_known:
+                        query = self._order_projections_and_filter(
+                            query, columns_to_types, coerce_types=True
+                        )
                     if i == 0:
                         self._create_table(
                             schema if schema else table,
@@ -1421,6 +1425,16 @@ class EngineAdapter:
                 prefixed_unmanaged_columns.append(prefixed_col)
             query = (
                 exp.Select()  # type: ignore
+                .select(*table_columns)
+                .from_("static")
+                .union(
+                    exp.select(*table_columns).from_("updated_rows"),
+                    distinct=False,
+                )
+                .union(
+                    exp.select(*table_columns).from_("inserted_rows"),
+                    distinct=False,
+                )
                 .with_(
                     "source",
                     exp.select(exp.true().as_("_exists"), *select_source_columns)
@@ -1559,16 +1573,6 @@ class EngineAdapter:
                     .from_("joined")
                     .where(updated_row_filter),
                 )
-                .select(*table_columns)
-                .from_("static")
-                .union(
-                    exp.select(*table_columns).from_("updated_rows"),
-                    distinct=False,
-                )
-                .union(
-                    exp.select(*table_columns).from_("inserted_rows"),
-                    distinct=False,
-                )
             )
 
             self.replace_query(
@@ -1998,17 +2002,26 @@ class EngineAdapter:
         query: Query,
         columns_to_types: t.Dict[str, exp.DataType],
         where: t.Optional[exp.Expression] = None,
+        coerce_types: bool = False,
     ) -> Query:
         if not isinstance(query, exp.Query) or (
-            not where and query.named_selects == list(columns_to_types)
+            not where and not coerce_types and query.named_selects == list(columns_to_types)
         ):
             return query
 
         query = t.cast(exp.Query, query.copy())
         with_ = query.args.pop("with", None)
-        query = self._select_columns(columns_to_types).from_(
-            query.subquery("_subquery", copy=False), copy=False
-        )
+
+        select_exprs: t.List[exp.Expression] = [
+            exp.column(c, quoted=True) for c in columns_to_types
+        ]
+        if coerce_types and columns_to_types_all_known(columns_to_types):
+            select_exprs = [
+                exp.cast(select_exprs[i], col_tpe).as_(col, quoted=True)
+                for i, (col, col_tpe) in enumerate(columns_to_types.items())
+            ]
+
+        query = exp.select(*select_exprs).from_(query.subquery("_subquery", copy=False), copy=False)
         if where:
             query = query.where(where, copy=False)
 


Test patch:
diff --git a/tests/core/engine_adapter/test_base.py b/tests/core/engine_adapter/test_base.py
index a55fe946..9f6b2278 100644
--- a/tests/core/engine_adapter/test_base.py
+++ b/tests/core/engine_adapter/test_base.py
@@ -443,7 +443,7 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
         """CREATE TABLE IF NOT EXISTS "test_table" ("a" INT COMMENT 'a description', "b" INT) COMMENT='test description'""",
-        """CREATE TABLE IF NOT EXISTS "test_table" ("a" INT COMMENT 'a description', "b" INT) COMMENT='test description' AS SELECT "a", "b" FROM "source_table\"""",
+        """CREATE TABLE IF NOT EXISTS "test_table" ("a" INT COMMENT 'a description', "b" INT) COMMENT='test description' AS SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (SELECT "a", "b" FROM "source_table") AS "_subquery\"""",
         """CREATE TABLE IF NOT EXISTS "test_table" COMMENT='test description' AS SELECT "a", "b" FROM "source_table\"""",
         """COMMENT ON COLUMN "test_table"."a" IS 'a description'""",
         """CREATE OR REPLACE VIEW "test_view" COMMENT='test description' AS SELECT "a", "b" FROM "source_table\"""",
@@ -1154,31 +1154,17 @@ WITH "source" AS (
     "test_UPDATED_at" > "t_test_UPDATED_at"
 )
 SELECT
-  "id",
-  "name",
-  "price",
-  "test_UPDATED_at",
-  "test_valid_from",
-  "test_valid_to"
-FROM "static"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_UPDATED_at",
-  "test_valid_from",
-  "test_valid_to"
-FROM "updated_rows"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_UPDATED_at",
-  "test_valid_from",
-  "test_valid_to"
-FROM "inserted_rows"
+  CAST("id" AS INT) AS "id",
+  CAST("name" AS VARCHAR) AS "name",
+  CAST("price" AS DOUBLE) AS "price",
+  CAST("test_UPDATED_at" AS TIMESTAMP) AS "test_UPDATED_at",
+  CAST("test_valid_from" AS TIMESTAMP) AS "test_valid_from",
+  CAST("test_valid_to" AS TIMESTAMP) AS "test_valid_to"
+FROM (
+  SELECT "id", "name", "price", "test_UPDATED_at", "test_valid_from", "test_valid_to" FROM "static"
+  UNION ALL SELECT "id", "name", "price", "test_UPDATED_at", "test_valid_from", "test_valid_to" FROM "updated_rows"
+  UNION ALL SELECT "id", "name", "price", "test_UPDATED_at", "test_valid_from", "test_valid_to" FROM "inserted_rows"
+) AS "_subquery"
     """
         ).sql()
     )
@@ -1345,31 +1331,17 @@ WITH "source" AS (
     "test_updated_at" > "t_test_updated_at"
 )
 SELECT
-  "id",
-  "name",
-  "price",
-  "test_updated_at",
-  "test_valid_from",
-  "test_valid_to"
-FROM "static"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_updated_at",
-  "test_valid_from",
-  "test_valid_to"
-FROM "updated_rows"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_updated_at",
-  "test_valid_from",
-  "test_valid_to"
-FROM "inserted_rows"
+  CAST("id" AS INT) AS "id",
+  CAST("name" AS VARCHAR) AS "name",
+  CAST("price" AS DOUBLE) AS "price",
+  CAST("test_updated_at" AS TIMESTAMP) AS "test_updated_at",
+  CAST("test_valid_from" AS TIMESTAMP) AS "test_valid_from",
+  CAST("test_valid_to" AS TIMESTAMP) AS "test_valid_to"
+FROM (
+  SELECT "id", "name", "price", "test_updated_at", "test_valid_from", "test_valid_to" FROM "static"
+  UNION ALL SELECT "id", "name", "price", "test_updated_at", "test_valid_from", "test_valid_to" FROM "updated_rows"
+  UNION ALL SELECT "id", "name", "price", "test_updated_at", "test_valid_from", "test_valid_to" FROM "inserted_rows"
+) AS "_subquery"
     """
         ).sql()
     )
@@ -1563,35 +1535,7 @@ WITH "source" AS (
   WHERE
     "test_updated_at" > "t_test_updated_at"
 )
-SELECT
-  "id1",
-  "id2",
-  "name",
-  "price",
-  "test_updated_at",
-  "test_valid_from",
-  "test_valid_to"
-FROM "static"
-UNION ALL
-SELECT
-  "id1",
-  "id2",
-  "name",
-  "price",
-  "test_updated_at",
-  "test_valid_from",
-  "test_valid_to"
-FROM "updated_rows"
-UNION ALL
-SELECT
-  "id1",
-  "id2",
-  "name",
-  "price",
-  "test_updated_at",
-  "test_valid_from",
-  "test_valid_to"
-FROM "inserted_rows"
+SELECT CAST("id1" AS INT) AS "id1", CAST("id2" AS INT) AS "id2", CAST("name" AS VARCHAR) AS "name", CAST("price" AS DOUBLE) AS "price", CAST("test_updated_at" AS TIMESTAMPTZ) AS "test_updated_at", CAST("test_valid_from" AS TIMESTAMPTZ) AS "test_valid_from", CAST("test_valid_to" AS TIMESTAMPTZ) AS "test_valid_to" FROM (SELECT "id1", "id2", "name", "price", "test_updated_at", "test_valid_from", "test_valid_to" FROM "static" UNION ALL SELECT "id1", "id2", "name", "price", "test_updated_at", "test_valid_from", "test_valid_to" FROM "updated_rows" UNION ALL SELECT "id1", "id2", "name", "price", "test_updated_at", "test_valid_from", "test_valid_to" FROM "inserted_rows") AS "_subquery"
 """
         ).sql()
     )
@@ -1771,29 +1715,7 @@ WITH "source" AS (
       )
     )
 )
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_VALID_from",
-  "test_valid_to"
-FROM "static"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_VALID_from",
-  "test_valid_to"
-FROM "updated_rows"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_VALID_from",
-  "test_valid_to"
-FROM "inserted_rows"
+SELECT CAST("id" AS INT) AS "id", CAST("name" AS VARCHAR) AS "name", CAST("price" AS DOUBLE) AS "price", CAST("test_VALID_from" AS TIMESTAMP) AS "test_VALID_from", CAST("test_valid_to" AS TIMESTAMP) AS "test_valid_to" FROM (SELECT "id", "name", "price", "test_VALID_from", "test_valid_to" FROM "static" UNION ALL SELECT "id", "name", "price", "test_VALID_from", "test_valid_to" FROM "updated_rows" UNION ALL SELECT "id", "name", "price", "test_VALID_from", "test_valid_to" FROM "inserted_rows") AS "_subquery"
     """
         ).sql()
     )
@@ -1976,29 +1898,7 @@ WITH "source" AS (
       )
     )
 )
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_valid_from",
-  "test_valid_to"
-FROM "static"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_valid_from",
-  "test_valid_to"
-FROM "updated_rows"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_valid_from",
-  "test_valid_to"
-FROM "inserted_rows"
+SELECT CAST("id" AS INT) AS "id", CAST("name" AS VARCHAR) AS "name", CAST("price" AS DOUBLE) AS "price", CAST("test_valid_from" AS TIMESTAMP) AS "test_valid_from", CAST("test_valid_to" AS TIMESTAMP) AS "test_valid_to" FROM (SELECT "id", "name", "price", "test_valid_from", "test_valid_to" FROM "static" UNION ALL SELECT "id", "name", "price", "test_valid_from", "test_valid_to" FROM "updated_rows" UNION ALL SELECT "id", "name", "price", "test_valid_from", "test_valid_to" FROM "inserted_rows") AS "_subquery"
     """
         ).sql()
     )
@@ -2192,29 +2092,7 @@ WITH "source" AS (
       )
     )
 )
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_valid_from",
-  "test_valid_to"
-FROM "static"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_valid_from",
-  "test_valid_to"
-FROM "updated_rows"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_valid_from",
-  "test_valid_to"
-FROM "inserted_rows"
+SELECT CAST("id" AS INT) AS "id", CAST("name" AS VARCHAR) AS "name", CAST("price" AS DOUBLE) AS "price", CAST("test_valid_from" AS TIMESTAMP) AS "test_valid_from", CAST("test_valid_to" AS TIMESTAMP) AS "test_valid_to" FROM (SELECT "id", "name", "price", "test_valid_from", "test_valid_to" FROM "static" UNION ALL SELECT "id", "name", "price", "test_valid_from", "test_valid_to" FROM "updated_rows" UNION ALL SELECT "id", "name", "price", "test_valid_from", "test_valid_to" FROM "inserted_rows") AS "_subquery"
     """
         ).sql()
     )
@@ -2392,43 +2270,30 @@ WITH "source" AS (
       )
     )
 )
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_valid_from",
-  "test_valid_to"
-FROM "static"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_valid_from",
-  "test_valid_to"
-FROM "updated_rows"
-UNION ALL
-SELECT
-  "id",
-  "name",
-  "price",
-  "test_valid_from",
-  "test_valid_to"
-FROM "inserted_rows"
+SELECT CAST("id" AS INT) AS "id", CAST("name" AS VARCHAR) AS "name", CAST("price" AS DOUBLE) AS "price", CAST("test_valid_from" AS TIMESTAMP) AS "test_valid_from", CAST("test_valid_to" AS TIMESTAMP) AS "test_valid_to" FROM (SELECT "id", "name", "price", "test_valid_from", "test_valid_to" FROM "static" UNION ALL SELECT "id", "name", "price", "test_valid_from", "test_valid_to" FROM "updated_rows" UNION ALL SELECT "id", "name", "price", "test_valid_from", "test_valid_to" FROM "inserted_rows") AS "_subquery"
     """
         ).sql()
     )
 
 
-def test_replace_query(make_mocked_engine_adapter: t.Callable):
+def test_replace_query(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
     adapter = make_mocked_engine_adapter(EngineAdapter)
+    columns_mock = mocker.patch.object(adapter, "columns")
+    columns_mock.return_value = None
+
     adapter.replace_query(
         "test_table", parse_one("SELECT a FROM tbl"), {"a": exp.DataType.build("INT")}
     )
+    adapter.replace_query("test_table", parse_one("SELECT a FROM tbl"))
+    adapter.replace_query(
+        "test_table", parse_one("SELECT a FROM tbl"), {"a": exp.DataType.build("UNKNOWN")}
+    )
 
     # TODO: Shouldn't we enforce that `a` is casted to an int?
     assert to_sql_calls(adapter) == [
-        'CREATE OR REPLACE TABLE "test_table" AS SELECT "a" FROM "tbl"'
+        'CREATE OR REPLACE TABLE "test_table" AS SELECT CAST("a" AS INT) AS "a" FROM (SELECT "a" FROM "tbl") AS "_subquery"',
+        'CREATE OR REPLACE TABLE "test_table" AS SELECT "a" FROM "tbl"',
+        'CREATE OR REPLACE TABLE "test_table" AS SELECT "a" FROM "tbl"',
     ]
 
 
@@ -2442,9 +2307,9 @@ def test_replace_query_pandas(make_mocked_engine_adapter: t.Callable):
     )
 
     assert to_sql_calls(adapter) == [
-        'CREATE OR REPLACE TABLE "test_table" AS SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (VALUES (1, 4)) AS "t"("a", "b")',
-        'INSERT INTO "test_table" ("a", "b") SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (VALUES (2, 5)) AS "t"("a", "b")',
-        'INSERT INTO "test_table" ("a", "b") SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (VALUES (3, 6)) AS "t"("a", "b")',
+        'CREATE OR REPLACE TABLE "test_table" AS SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (VALUES (1, 4)) AS "t"("a", "b")) AS "_subquery"',
+        'INSERT INTO "test_table" ("a", "b") SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (VALUES (2, 5)) AS "t"("a", "b")) AS "_subquery"',
+        'INSERT INTO "test_table" ("a", "b") SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (VALUES (3, 6)) AS "t"("a", "b")) AS "_subquery"',
     ]
 
 
@@ -2508,7 +2373,7 @@ def test_replace_query_self_referencing_not_exists_known(
 
     assert to_sql_calls(adapter) == [
         'CREATE TABLE IF NOT EXISTS "test" ("a" INT)',
-        'CREATE OR REPLACE TABLE "test" AS SELECT "a" FROM "test"',
+        'CREATE OR REPLACE TABLE "test" AS SELECT CAST("a" AS INT) AS "a" FROM (SELECT "a" FROM "test") AS "_subquery"',
     ]
 
 
@@ -2585,7 +2450,7 @@ def test_ctas_pandas(make_mocked_engine_adapter: t.Callable):
     adapter.ctas("new_table", df)
 
     assert to_sql_calls(adapter) == [
-        'CREATE TABLE IF NOT EXISTS "new_table" AS SELECT CAST("a" AS BIGINT) AS "a", CAST("b" AS BIGINT) AS "b" FROM (VALUES (1, 4), (2, 5), (3, 6)) AS "t"("a", "b")'
+        'CREATE TABLE IF NOT EXISTS "new_table" AS SELECT CAST("a" AS BIGINT) AS "a", CAST("b" AS BIGINT) AS "b" FROM (SELECT CAST("a" AS BIGINT) AS "a", CAST("b" AS BIGINT) AS "b" FROM (VALUES (1, 4), (2, 5), (3, 6)) AS "t"("a", "b")) AS "_subquery"'
     ]
 
 
@@ -2743,7 +2608,7 @@ def test_insert_overwrite_by_partition_query(
 
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
-        'CREATE TABLE "test_schema"."__temp_test_table_abcdefgh" AS SELECT "a", "ds", "b" FROM "tbl"',
+        'CREATE TABLE "test_schema"."__temp_test_table_abcdefgh" AS SELECT CAST("a" AS INT) AS "a", CAST("ds" AS DATETIME) AS "ds", CAST("b" AS BOOLEAN) AS "b" FROM (SELECT "a", "ds", "b" FROM "tbl") AS "_subquery"',
         expected_delete_stmt,
         'INSERT INTO "test_schema"."test_table" ("a", "ds", "b") SELECT "a", "ds", "b" FROM "test_schema"."__temp_test_table_abcdefgh"',
         'DROP TABLE IF EXISTS "test_schema"."__temp_test_table_abcdefgh"',
diff --git a/tests/core/engine_adapter/test_bigquery.py b/tests/core/engine_adapter/test_bigquery.py
index a2d6a246..2906e1fc 100644
--- a/tests/core/engine_adapter/test_bigquery.py
+++ b/tests/core/engine_adapter/test_bigquery.py
@@ -223,7 +223,9 @@ def test_replace_query(make_mocked_engine_adapter: t.Callable, mocker: MockerFix
     )
 
     sql_calls = _to_sql_calls(execute_mock)
-    assert sql_calls == ["CREATE OR REPLACE TABLE `test_table` AS SELECT `a` FROM `tbl`"]
+    assert sql_calls == [
+        "CREATE OR REPLACE TABLE `test_table` AS SELECT CAST(`a` AS INT64) AS `a` FROM (SELECT `a` FROM `tbl`) AS `_subquery`"
+    ]
 
 
 def test_replace_query_pandas(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
@@ -292,7 +294,7 @@ def test_replace_query_pandas(make_mocked_engine_adapter: t.Callable, mocker: Mo
     ]
     sql_calls = _to_sql_calls(execute_mock)
     assert sql_calls == [
-        "CREATE OR REPLACE TABLE `test_table` AS SELECT `a`, `b` FROM `project`.`dataset`.`temp_table`",
+        "CREATE OR REPLACE TABLE `test_table` AS SELECT CAST(`a` AS INT64) AS `a`, CAST(`b` AS INT64) AS `b` FROM (SELECT `a`, `b` FROM `project`.`dataset`.`temp_table`) AS `_subquery`",
         "DROP TABLE IF EXISTS `project`.`dataset`.`temp_table`",
     ]
 
@@ -636,7 +638,7 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
     assert sql_calls == [
         f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='{truncated_column_comment}'), `b` INT64) OPTIONS (description='{truncated_table_comment}')",
         "CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='\\\\'), `b` INT64) OPTIONS (description='\\\\')",
-        f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='{truncated_column_comment}'), `b` INT64) OPTIONS (description='{truncated_table_comment}') AS SELECT `a`, `b` FROM `source_table`",
+        f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='{truncated_column_comment}'), `b` INT64) OPTIONS (description='{truncated_table_comment}') AS SELECT CAST(`a` AS INT64) AS `a`, CAST(`b` AS INT64) AS `b` FROM (SELECT `a`, `b` FROM `source_table`) AS `_subquery`",
         f"CREATE OR REPLACE VIEW `test_table` OPTIONS (description='{truncated_table_comment}') AS SELECT `a`, `b` FROM `source_table`",
         f"ALTER TABLE `test_table` SET OPTIONS(description = '{truncated_table_comment}')",
     ]
diff --git a/tests/core/engine_adapter/test_databricks.py b/tests/core/engine_adapter/test_databricks.py
index 5d6e2ed3..a825af44 100644
--- a/tests/core/engine_adapter/test_databricks.py
+++ b/tests/core/engine_adapter/test_databricks.py
@@ -24,7 +24,7 @@ def test_replace_query_not_exists(mocker: MockFixture, make_mocked_engine_adapte
     )
 
     assert to_sql_calls(adapter) == [
-        "CREATE TABLE IF NOT EXISTS `test_table` AS SELECT `a` FROM `tbl`",
+        "CREATE TABLE IF NOT EXISTS `test_table` AS SELECT CAST(`a` AS INT) AS `a` FROM (SELECT `a` FROM `tbl`) AS `_subquery`",
     ]
 
 
@@ -55,7 +55,7 @@ def test_replace_query_pandas_not_exists(
     )
 
     assert to_sql_calls(adapter) == [
-        "CREATE TABLE IF NOT EXISTS `test_table` AS SELECT CAST(`a` AS INT) AS `a`, CAST(`b` AS INT) AS `b` FROM VALUES (1, 4), (2, 5), (3, 6) AS `t`(`a`, `b`)",
+        "CREATE TABLE IF NOT EXISTS `test_table` AS SELECT CAST(`a` AS INT) AS `a`, CAST(`b` AS INT) AS `b` FROM (SELECT CAST(`a` AS INT) AS `a`, CAST(`b` AS INT) AS `b` FROM VALUES (1, 4), (2, 5), (3, 6) AS `t`(`a`, `b`)) AS `_subquery`",
     ]
 
 
@@ -132,7 +132,7 @@ def test_insert_overwrite_by_partition_query(
 
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
-        "CREATE TABLE `test_schema`.`temp_test_table_abcdefgh` AS SELECT `a`, `ds`, `b` FROM `tbl`",
+        "CREATE TABLE `test_schema`.`temp_test_table_abcdefgh` AS SELECT CAST(`a` AS INT) AS `a`, CAST(`ds` AS TIMESTAMP) AS `ds`, CAST(`b` AS BOOLEAN) AS `b` FROM (SELECT `a`, `ds`, `b` FROM `tbl`) AS `_subquery`",
         "INSERT INTO `test_schema`.`test_table` REPLACE WHERE CONCAT_WS('__SQLMESH_DELIM__', DATE_TRUNC('MONTH', `ds`), `b`) IN (SELECT DISTINCT CONCAT_WS('__SQLMESH_DELIM__', DATE_TRUNC('MONTH', `ds`), `b`) FROM `test_schema`.`temp_test_table_abcdefgh`) SELECT `a`, `ds`, `b` FROM `test_schema`.`temp_test_table_abcdefgh`",
         "DROP TABLE IF EXISTS `test_schema`.`temp_test_table_abcdefgh`",
     ]
diff --git a/tests/core/engine_adapter/test_mixins.py b/tests/core/engine_adapter/test_mixins.py
index 2f8b1c74..57803427 100644
--- a/tests/core/engine_adapter/test_mixins.py
+++ b/tests/core/engine_adapter/test_mixins.py
@@ -33,7 +33,9 @@ def test_logical_merge(make_mocked_engine_adapter: t.Callable, mocker: MockerFix
 
     adapter.cursor.execute.assert_has_calls(
         [
-            call('''CREATE TABLE "temporary" AS SELECT "id", "ts", "val" FROM "source"'''),
+            call(
+                '''CREATE TABLE "temporary" AS SELECT CAST("id" AS INT) AS "id", CAST("ts" AS TIMESTAMP) AS "ts", CAST("val" AS INT) AS "val" FROM (SELECT "id", "ts", "val" FROM "source") AS "_subquery"'''
+            ),
             call("""DELETE FROM "target" WHERE "id" IN (SELECT "id" FROM "temporary")"""),
             call(
                 'INSERT INTO "target" ("id", "ts", "val") SELECT DISTINCT ON ("id") "id", "ts", "val" FROM "temporary"'
@@ -56,7 +58,9 @@ def test_logical_merge(make_mocked_engine_adapter: t.Callable, mocker: MockerFix
 
     adapter.cursor.execute.assert_has_calls(
         [
-            call('''CREATE TABLE "temporary" AS SELECT "id", "ts", "val" FROM "source"'''),
+            call(
+                '''CREATE TABLE "temporary" AS SELECT CAST("id" AS INT) AS "id", CAST("ts" AS TIMESTAMP) AS "ts", CAST("val" AS INT) AS "val" FROM (SELECT "id", "ts", "val" FROM "source") AS "_subquery"'''
+            ),
             call(
                 """DELETE FROM "target" WHERE CONCAT_WS('__SQLMESH_DELIM__', "id", "ts") IN (SELECT CONCAT_WS('__SQLMESH_DELIM__', "id", "ts") FROM "temporary")"""
             ),
diff --git a/tests/core/engine_adapter/test_mysql.py b/tests/core/engine_adapter/test_mysql.py
index 1feccf79..f00c5332 100644
--- a/tests/core/engine_adapter/test_mysql.py
+++ b/tests/core/engine_adapter/test_mysql.py
@@ -56,7 +56,7 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
         f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT COMMENT '{truncated_column_comment}', `b` INT) COMMENT='{truncated_table_comment}'",
-        f"CREATE TABLE IF NOT EXISTS `test_table` COMMENT='{truncated_table_comment}' AS SELECT `a`, `b` FROM `source_table`",
+        f"CREATE TABLE IF NOT EXISTS `test_table` COMMENT='{truncated_table_comment}' AS SELECT CAST(`a` AS SIGNED) AS `a`, CAST(`b` AS SIGNED) AS `b` FROM (SELECT `a`, `b` FROM `source_table`) AS `_subquery`",
         f"ALTER TABLE `test_table` MODIFY `a` INT COMMENT '{truncated_column_comment}'",
         "CREATE OR REPLACE VIEW `test_view` AS SELECT `a`, `b` FROM `source_table`",
         f"ALTER TABLE `test_table` COMMENT = '{truncated_table_comment}'",
diff --git a/tests/core/engine_adapter/test_snowflake.py b/tests/core/engine_adapter/test_snowflake.py
index 9df09e32..9b35d95c 100644
--- a/tests/core/engine_adapter/test_snowflake.py
+++ b/tests/core/engine_adapter/test_snowflake.py
@@ -138,7 +138,7 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
         """CREATE TABLE IF NOT EXISTS "test_table" ("a" INT COMMENT 'a column description', "b" INT) COMMENT='table description'""",
-        """CREATE TABLE IF NOT EXISTS "test_table" ("a" INT COMMENT 'a column description', "b" INT) COMMENT='table description' AS SELECT "a", "b" FROM "source_table\"""",
+        """CREATE TABLE IF NOT EXISTS "test_table" ("a" INT COMMENT 'a column description', "b" INT) COMMENT='table description' AS SELECT CAST("a" AS INT) AS "a", CAST("b" AS INT) AS "b" FROM (SELECT "a", "b" FROM "source_table") AS "_subquery\"""",
         """CREATE OR REPLACE VIEW "test_view" COMMENT='table description' AS SELECT "a", "b" FROM "source_table\"""",
         """ALTER VIEW "test_view" ALTER COLUMN "a" COMMENT 'a column description'""",
         """COMMENT ON TABLE "test_table" IS 'table description'""",
diff --git a/tests/core/engine_adapter/test_spark.py b/tests/core/engine_adapter/test_spark.py
index 1316a292..152c412c 100644
--- a/tests/core/engine_adapter/test_spark.py
+++ b/tests/core/engine_adapter/test_spark.py
@@ -82,7 +82,7 @@ def test_replace_query_table_properties_not_exists(
     )
 
     assert to_sql_calls(adapter) == [
-        "CREATE TABLE IF NOT EXISTS `test_table` USING ICEBERG PARTITIONED BY (`colb`) TBLPROPERTIES ('a'=1) AS SELECT 1 AS `cola`, '2' AS `colb`, '3' AS `colc`",
+        "CREATE TABLE IF NOT EXISTS `test_table` USING ICEBERG PARTITIONED BY (`colb`) TBLPROPERTIES ('a'=1) AS SELECT CAST(`cola` AS INT) AS `cola`, CAST(`colb` AS STRING) AS `colb`, CAST(`colc` AS STRING) AS `colc` FROM (SELECT 1 AS `cola`, '2' AS `colb`, '3' AS `colc`) AS `_subquery`",
         "INSERT INTO `test_table` SELECT * FROM `test_table`",
     ]
 
@@ -178,7 +178,7 @@ def test_replace_query_not_exists(mocker: MockerFixture, make_mocked_engine_adap
     )
 
     assert to_sql_calls(adapter) == [
-        "CREATE TABLE IF NOT EXISTS `test_table` AS SELECT `a` FROM `tbl`",
+        "CREATE TABLE IF NOT EXISTS `test_table` AS SELECT CAST(`a` AS INT) AS `a` FROM (SELECT `a` FROM `tbl`) AS `_subquery`",
     ]
 
 
@@ -213,7 +213,7 @@ def test_replace_query_pandas_not_exists(
     )
 
     assert to_sql_calls(adapter) == [
-        "CREATE TABLE IF NOT EXISTS `test_table` AS SELECT CAST(`a` AS INT) AS `a`, CAST(`b` AS INT) AS `b` FROM VALUES (1, 4), (2, 5), (3, 6) AS `t`(`a`, `b`)",
+        "CREATE TABLE IF NOT EXISTS `test_table` AS SELECT CAST(`a` AS INT) AS `a`, CAST(`b` AS INT) AS `b` FROM (SELECT CAST(`a` AS INT) AS `a`, CAST(`b` AS INT) AS `b` FROM VALUES (1, 4), (2, 5), (3, 6) AS `t`(`a`, `b`)) AS `_subquery`",
     ]
 
 
@@ -280,7 +280,7 @@ def test_replace_query_self_ref_not_exists(
     assert to_sql_calls(adapter) == [
         "CREATE TABLE IF NOT EXISTS `db`.`table` (`col` INT)",
         "CREATE SCHEMA IF NOT EXISTS `db`",
-        "CREATE TABLE IF NOT EXISTS `db`.`temp_table_abcdefgh` AS SELECT `col` FROM `db`.`table`",
+        "CREATE TABLE IF NOT EXISTS `db`.`temp_table_abcdefgh` AS SELECT CAST(`col` AS INT) AS `col` FROM (SELECT `col` FROM `db`.`table`) AS `_subquery`",
         "INSERT OVERWRITE TABLE `db`.`table` (`col`) SELECT `col` + 1 AS `col` FROM `db`.`temp_table_abcdefgh`",
         "DROP TABLE IF EXISTS `db`.`temp_table_abcdefgh`",
     ]
@@ -324,7 +324,7 @@ def test_replace_query_self_ref_exists(
         # This is something temp_table does and isn't really needed. The contract with `replace_query` is that
         # the schema for the table already exists
         "CREATE SCHEMA IF NOT EXISTS `db`",
-        f"CREATE TABLE IF NOT EXISTS `db`.`temp_table_{temp_table_id}` AS SELECT `col` FROM `db`.`table`",
+        f"CREATE TABLE IF NOT EXISTS `db`.`temp_table_{temp_table_id}` AS SELECT CAST(`col` AS INT) AS `col` FROM (SELECT `col` FROM `db`.`table`) AS `_subquery`",
         "INSERT OVERWRITE TABLE `db`.`table` (`col`) SELECT `col` + 1 AS `col` FROM `db`.`temp_table_abcdefgh`",
         f"DROP TABLE IF EXISTS `db`.`temp_table_{temp_table_id}`",
     ]
@@ -604,7 +604,7 @@ def test_scd_type_2_by_time(
     assert to_sql_calls(adapter) == [
         "CREATE TABLE IF NOT EXISTS `db`.`target` (`id` INT, `name` STRING, `price` DOUBLE, `test_updated_at` TIMESTAMP, `test_valid_from` TIMESTAMP, `test_valid_to` TIMESTAMP)",
         "CREATE SCHEMA IF NOT EXISTS `db`",
-        "CREATE TABLE IF NOT EXISTS `db`.`temp_target_abcdefgh` AS SELECT `id`, `name`, `price`, `test_updated_at`, `test_valid_from`, `test_valid_to` FROM `db`.`target`",
+        "CREATE TABLE IF NOT EXISTS `db`.`temp_target_abcdefgh` AS SELECT CAST(`id` AS INT) AS `id`, CAST(`name` AS STRING) AS `name`, CAST(`price` AS DOUBLE) AS `price`, CAST(`test_updated_at` AS TIMESTAMP) AS `test_updated_at`, CAST(`test_valid_from` AS TIMESTAMP) AS `test_valid_from`, CAST(`test_valid_to` AS TIMESTAMP) AS `test_valid_to` FROM (SELECT `id`, `name`, `price`, `test_updated_at`, `test_valid_from`, `test_valid_to` FROM `db`.`target`) AS `_subquery`",
         parse_one(
             """WITH `source` AS (
   SELECT
@@ -910,7 +910,7 @@ def test_comments_hive(mocker: MockerFixture, make_mocked_engine_adapter: t.Call
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
         f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT COMMENT '{truncated_column_comment}', `b` INT) COMMENT '{truncated_table_comment}'",
-        f"CREATE TABLE IF NOT EXISTS `test_table` COMMENT '{truncated_table_comment}' AS SELECT `a`, `b` FROM `source_table`",
+        f"CREATE TABLE IF NOT EXISTS `test_table` COMMENT '{truncated_table_comment}' AS SELECT CAST(`a` AS INT) AS `a`, CAST(`b` AS INT) AS `b` FROM (SELECT `a`, `b` FROM `source_table`) AS `_subquery`",
         f"ALTER TABLE `test_table` ALTER COLUMN `a` COMMENT '{truncated_column_comment}'",
         f"CREATE OR REPLACE VIEW test_view COMMENT '{truncated_table_comment}' AS SELECT a, b FROM source_table",
         f"COMMENT ON TABLE `test_table` IS '{truncated_table_comment}'",
@@ -968,7 +968,7 @@ def test_comments_iceberg(mocker: MockerFixture, make_mocked_engine_adapter: t.C
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
         f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT COMMENT '{long_column_comment}', `b` INT) COMMENT '{long_table_comment}'",
-        f"CREATE TABLE IF NOT EXISTS `test_table` COMMENT '{long_table_comment}' AS SELECT `a`, `b` FROM `source_table`",
+        f"CREATE TABLE IF NOT EXISTS `test_table` COMMENT '{long_table_comment}' AS SELECT CAST(`a` AS INT) AS `a`, CAST(`b` AS INT) AS `b` FROM (SELECT `a`, `b` FROM `source_table`) AS `_subquery`",
         f"ALTER TABLE `test_table` ALTER COLUMN `a` COMMENT '{long_column_comment}'",
         f"CREATE OR REPLACE VIEW test_view COMMENT '{long_table_comment}' AS SELECT a, b FROM source_table",
         f"COMMENT ON TABLE `test_table` IS '{long_table_comment}'",
@@ -1021,7 +1021,7 @@ def test_replace_query_with_wap_self_reference(
     assert sql_calls == [
         "CREATE TABLE IF NOT EXISTS `catalog`.`schema`.`table` (`a` INT)",
         "CREATE SCHEMA IF NOT EXISTS `catalog`.`schema`",
-        "CREATE TABLE IF NOT EXISTS `catalog`.`schema`.`temp_branch_wap_12345_abcdefgh` USING ICEBERG AS SELECT `a` FROM `catalog`.`schema`.`table`.`branch_wap_12345`",
+        "CREATE TABLE IF NOT EXISTS `catalog`.`schema`.`temp_branch_wap_12345_abcdefgh` USING ICEBERG AS SELECT CAST(`a` AS INT) AS `a` FROM (SELECT `a` FROM `catalog`.`schema`.`table`.`branch_wap_12345`) AS `_subquery`",
         "INSERT OVERWRITE TABLE `catalog`.`schema`.`table`.`branch_wap_12345` (`a`) SELECT 1 AS `a` FROM `catalog`.`schema`.`temp_branch_wap_12345_abcdefgh`",
         "DROP TABLE IF EXISTS `catalog`.`schema`.`temp_branch_wap_12345_abcdefgh`",
     ]
diff --git a/tests/core/engine_adapter/test_trino.py b/tests/core/engine_adapter/test_trino.py
index b453b889..0ad7662e 100644
--- a/tests/core/engine_adapter/test_trino.py
+++ b/tests/core/engine_adapter/test_trino.py
@@ -301,7 +301,7 @@ def test_comments_hive(mocker: MockerFixture, make_mocked_engine_adapter: t.Call
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
         f"""CREATE TABLE IF NOT EXISTS "test_table" ("a" INTEGER COMMENT '{truncated_column_comment}', "b" INTEGER) COMMENT '{truncated_table_comment}'""",
-        f"""CREATE TABLE IF NOT EXISTS "test_table" COMMENT '{truncated_table_comment}' AS SELECT "a", "b" FROM "source_table\"""",
+        f"""CREATE TABLE IF NOT EXISTS "test_table" COMMENT '{truncated_table_comment}' AS SELECT CAST("a" AS INTEGER) AS "a", CAST("b" AS INTEGER) AS "b" FROM (SELECT "a", "b" FROM "source_table") AS "_subquery\"""",
         f"""COMMENT ON COLUMN "test_table"."a" IS '{truncated_column_comment}'""",
         """CREATE OR REPLACE VIEW test_view AS SELECT a, b FROM source_table""",
         f"""COMMENT ON VIEW "test_view" IS '{truncated_table_comment}'""",
@@ -369,7 +369,7 @@ def test_comments_iceberg_delta(
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
         f"""CREATE TABLE IF NOT EXISTS "test_table" ("a" INTEGER COMMENT '{long_column_comment}', "b" INTEGER) COMMENT '{long_table_comment}'""",
-        f"""CREATE TABLE IF NOT EXISTS "test_table" COMMENT '{long_table_comment}' AS SELECT "a", "b" FROM "source_table\"""",
+        f"""CREATE TABLE IF NOT EXISTS "test_table" COMMENT '{long_table_comment}' AS SELECT CAST("a" AS INTEGER) AS "a", CAST("b" AS INTEGER) AS "b" FROM (SELECT "a", "b" FROM "source_table") AS "_subquery\"""",
         f"""COMMENT ON COLUMN "test_table"."a" IS '{long_column_comment}'""",
         """CREATE OR REPLACE VIEW test_view AS SELECT a, b FROM source_table""",
         f"""COMMENT ON VIEW "test_view" IS '{long_table_comment}'""",