Repository: TobikoData/sqlmesh
Pull request #2788 (2024-06-18 12:16:48)

Feat: Limit long lists of indirectly modified models in CLI output
This is an improvement to trim long indirectly modified trees in the output, fixes: #1992 

```
Summary of differences against `dev2`:
Models:
├── Directly Modified:
│   └── sushi__dev2.seed_model
└── Indirectly Modified:
    ├── sushi__dev2.full_model
    ├── ... 30 more ...
    └── sushi__dev2.incremental_model
```

With this update, the names of indirectly modified models will be omitted once their total number exceeds a certain threshold, currently set to 20. This behavior can be overridden with the `verbose` flag, in which case all model names will be shown regardless of their total number.

georgesittas: What about these two?

- `MarkdownConsole.show_model_difference_summary`
- `MarkdownConsole._show_categorized_snapshots`
Themiscodes: Thank you @georgesittas for the review, addressed your comments and extended the functionality for MarkdownConsole as well

Patch:
diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py
index 82fbd712..6baa09a5 100644
--- a/sqlmesh/core/console.py
+++ b/sqlmesh/core/console.py
@@ -60,6 +60,8 @@ class Console(abc.ABC):
     """Abstract base class for defining classes used for displaying information to the user and also interact
     with them when their input is needed."""
 
+    INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD = 20
+
     @abc.abstractmethod
     def start_plan_evaluation(self, plan: Plan) -> None:
         """Indicates that a new evaluation has begun."""
@@ -235,6 +237,17 @@ class Console(abc.ABC):
     def show_row_diff(self, row_diff: RowDiff, show_sample: bool = True) -> None:
         """Show table summary diff."""
 
+    def _limit_model_names(self, tree: Tree, verbose: bool = False) -> Tree:
+        """Trim long indirectly modified model lists below threshold."""
+        modified_length = len(tree.children)
+        if not verbose and modified_length > self.INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD:
+            tree.children = [
+                tree.children[0],
+                Tree(f".... {modified_length-2} more ...."),
+                tree.children[-1],
+            ]
+        return tree
+
 
 def make_progress_bar(message: str, console: t.Optional[RichConsole] = None) -> Progress:
     return Progress(
@@ -728,7 +741,7 @@ class TerminalConsole(Console):
             if direct.children:
                 tree.add(direct)
             if indirect.children:
-                tree.add(indirect)
+                tree.add(self._limit_model_names(indirect, self.verbose))
             if metadata.children:
                 tree.add(metadata)
         if selected_ignored_snapshot_ids:
@@ -795,6 +808,9 @@ class TerminalConsole(Console):
                 indirect_tree.add(
                     f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)}"
                 )
+            if indirect_tree:
+                indirect_tree = self._limit_model_names(indirect_tree, self.verbose)
+
             self._print(tree)
             if not no_prompts:
                 self._get_snapshot_change_category(
@@ -822,6 +838,9 @@ class TerminalConsole(Console):
                 indirect_tree.add(
                     f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)} ({child_category_str})"
                 )
+            if indirect_tree:
+                indirect_tree = self._limit_model_names(indirect_tree, self.verbose)
+
             self._print(Syntax(context_diff.text_diff(snapshot.name), "sql", word_wrap=True))
             self._print(tree)
 
@@ -1550,10 +1569,22 @@ class MarkdownConsole(CaptureTerminalConsole):
                         self._print(f"```diff\n{context_diff.text_diff(snapshot.name)}\n```")
             if indirectly_modified:
                 self._print("\n**Indirectly Modified:**")
-                for snapshot in sorted(indirectly_modified):
+                indirectly_modified = sorted(indirectly_modified)
+                modified_length = len(indirectly_modified)
+                if (
+                    not self.verbose
+                    and modified_length > self.INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD
+                ):
                     self._print(
-                        f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
+                        f"- `{indirectly_modified[0].display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`\n"
+                        f"- `.... {modified_length-2} more ....`\n"
+                        f"- `{indirectly_modified[-1].display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
                     )
+                else:
+                    for snapshot in indirectly_modified:
+                        self._print(
+                            f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
+                        )
             if metadata_modified:
                 self._print("\n**Metadata Updated:**")
                 for snapshot in sorted(metadata_modified):
@@ -1607,6 +1638,8 @@ class MarkdownConsole(CaptureTerminalConsole):
                 indirect_tree.add(
                     f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)} ({child_category_str})"
                 )
+            if indirect_tree:
+                indirect_tree = self._limit_model_names(indirect_tree, self.verbose)
             self._print(f"```diff\n{context_diff.text_diff(snapshot.name)}\n```\n")
             self._print("```\n")
             self._print(tree)
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}'""",