Repository: TobikoData/sqlmesh
Pull request #2779 (2024-06-17 10:40:22)

Feat: allow overriding the dialects' normalization strategies


Patch:
diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md
index 07e7f227..63a6b323 100644
--- a/docs/guides/configuration.md
+++ b/docs/guides/configuration.md
@@ -893,6 +893,27 @@ Example configuration:
 
 The default model kind is `VIEW` unless overridden with the `kind` key. For more information on model kinds, refer to [model concepts page](../concepts/models/model_kinds.md).
 
+##### Identifier resolution
+
+When a SQL engine receives a query such as `SELECT id FROM "some_table"`, it eventually needs to understand what database objects the identifiers `id` and `"some_table"` correspond to. This process is usually referred to as identifier (or name) resolution.
+
+Different SQL dialects implement different rules when resolving identifiers in queries. For example, certain identifiers may be treated as case-sensitive (e.g. if they're quoted), and a case-insensitive identifier is usually either lowercased or uppercased, before the engine actually looks up what object it corresponds to.
+
+SQLMesh analyzes model queries so that it can extract useful information from them, such as computing Column-Level Lineage. To facilitate this analysis, it _normalizes_ and _quotes_ all identifiers in those queries, [respecting each dialect's resolution rules](https://sqlglot.com/sqlglot/dialects/dialect.html#Dialect.normalize_identifier).
+
+The "normalization strategy", i.e. whether case-insensitive identifiers are lowercased or uppercased, is configurable per dialect. For example, to treat all identifiers as case-sensitive in a BigQuery project, one can do:
+
+=== "YAML"
+
+    ```yaml linenums="1"
+    model_defaults:
+      dialect: "bigquery,normalization_strategy=case_sensitive"
+    ```
+
+This may be useful in cases where the name casing needs to be preserved, since then SQLMesh won't be able to normalize them.
+
+See [here](https://sqlglot.com/sqlglot/dialects/dialect.html#NormalizationStrategy) to learn more about the supported normalization strategies.
+
 #### Model Kinds
 
 Model kinds are required in each model file's `MODEL` DDL statement. They may optionally be used to specify a default kind in the model defaults configuration key.
diff --git a/docs/integrations/airflow.md b/docs/integrations/airflow.md
index 22422729..8fdd4dc4 100644
--- a/docs/integrations/airflow.md
+++ b/docs/integrations/airflow.md
@@ -27,17 +27,18 @@ The name of the module file can be arbitrary, but we recommend something descrip
 
 `SQLMeshAirflow` has two required arguments (`engine_operator` and `default_catalog`). Details on these and additional optional arguments below:
 
-| Argument                        | Description                                                                                                                                                                                                |          Type          | Required |
-|---------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:----------------------:|:--------:|
-| `engine_operator`               | Name or operator to use for creating models. See [Engine Support](#engine-support) for list of options                                                                                                     | string or BaseOperator |    Y     |
-| `default_catalog`               | The default catalog (also called "database" in other engines) to use when models are defined that do not contain a catalog in their name. This should match the default catalog applied by the connection. |         string         |    Y     |
-| `engine_operator_args`          | The dictionary of arguments that will be passed into the evaluate engine operator during its construction. This can be used to customize parameters such as connection ID.                                 |          dict          |    N     |
-| `ddl_engine_operator`           | The type of the Airflow operator that will be used for environment management. These operations are SQL only. `engine_operator` is used if not provided                                                    | string or BaseOperator |    N     |
-| `ddl_engine_operator_args`      | Args to be passed into just the environment management operator. This can be used to customize parameters such as connection ID.                                                                           |          dict          |    N     |
-| `janitor_interval`              | Defines how often the janitor DAG runs. The janitor DAG removes platform-managed DAG instances that are pending deletion from Airflow. Default: 1 hour.                                                    |       timedelta        |    N     |
-| `plan_application_dag_ttl`      | Determines the time-to-live period for finished plan application DAGs. Once this period is exceeded, finished plan application DAGs are deleted by the janitor. Default: 2 days.                           |       timedelta        |    N     |
-| `external_table_sensor_factory` | A factory function that creates a sensor operator for a given signal payload. See [External signals](#external-signals) for more info                                                                      |        function        |    N     |
-| `generate_cadence_dags`         | Whether to generate cadence DAGs for model versions that are currently deployed to production.                                                                                                             |          bool          |    N     |
+| Argument                        | Description                                                                                                                                                                                                     | Type                   | Required |
+|---------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:----------------------:|:--------:|
+| `engine_operator`               | Name or operator to use for creating models. See [Engine Support](#engine-support) for list of options                                                                                                          | string or BaseOperator | Y        |
+| `default_catalog`               | The default catalog (also called "database" in other engines) to use when models are defined that do not contain a catalog in their name. This should match the default catalog applied by the connection.      | string                 | Y        |
+| `engine_operator_args`          | The dictionary of arguments that will be passed into the evaluate engine operator during its construction. This can be used to customize parameters such as connection ID.                                      | dict                   | N        |
+| `ddl_engine_operator`           | The type of the Airflow operator that will be used for environment management. These operations are SQL only. `engine_operator` is used if not provided                                                         | string or BaseOperator | N        |
+| `ddl_engine_operator_args`      | Args to be passed into just the environment management operator. This can be used to customize parameters such as connection ID.                                                                                | dict                   | N        |
+| `janitor_interval`              | Defines how often the janitor DAG runs. The janitor DAG removes platform-managed DAG instances that are pending deletion from Airflow. Default: 1 hour.                                                         | timedelta              | N        |
+| `plan_application_dag_ttl`      | Determines the time-to-live period for finished plan application DAGs. Once this period is exceeded, finished plan application DAGs are deleted by the janitor. Default: 2 days.                                | timedelta              | N        |
+| `external_table_sensor_factory` | A factory function that creates a sensor operator for a given signal payload. See [External signals](#external-signals) for more info                                                                           | function               | N        |
+| `sensor_mode`                   | The mode to use for SQLMesh sensors. Supported values are "poke" and "reschedule". See https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/sensors.html for more details. Default: "reschedule" | string                 | N        |
+| `generate_cadence_dags`         | Whether to generate cadence DAGs for model versions that are currently deployed to production.                                                                                                                  | bool                   | N        |
 
 
 ### State connection
diff --git a/setup.py b/setup.py
index d9c3d579..051f8f1e 100644
--- a/setup.py
+++ b/setup.py
@@ -46,7 +46,7 @@ setup(
         "requests",
         "rich[jupyter]",
         "ruamel.yaml",
-        "sqlglot[rs]~=25.1.0",
+        "sqlglot[rs]~=25.2.0",
     ],
     extras_require={
         "bigquery": [
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/context.py b/sqlmesh/core/context.py
index 17aad875..7e5d17be 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -49,7 +49,7 @@ from shutil import rmtree
 from types import MappingProxyType
 
 import pandas as pd
-from sqlglot import exp
+from sqlglot import Dialect, exp
 from sqlglot.lineage import GraphHTML
 
 from sqlmesh.core import analytics
@@ -325,6 +325,13 @@ class GenericContext(BaseContext, t.Generic[C]):
 
         self.path, self.config = t.cast(t.Tuple[Path, C], next(iter(self.configs.items())))
 
+        # This allows overriding the default dialect's normalization strategy, so for example
+        # one can do `dialect="duckdb,normalization_strategy=lowercase"` and this will be
+        # applied to the DuckDB dialect globally
+        if "normalization_strategy" in str(self.config.dialect):
+            dialect = Dialect.get_or_raise(self.config.dialect)
+            type(dialect).NORMALIZATION_STRATEGY = dialect.normalization_strategy
+
         if self.config.disable_anonymized_analytics:
             analytics.disable_analytics()
 
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)
 
diff --git a/sqlmesh/core/engine_adapter/mixins.py b/sqlmesh/core/engine_adapter/mixins.py
index 793bfc90..c838f9f2 100644
--- a/sqlmesh/core/engine_adapter/mixins.py
+++ b/sqlmesh/core/engine_adapter/mixins.py
@@ -256,7 +256,7 @@ class VarcharSizeWorkaroundMixin(EngineAdapter):
             # redshift and mssql have a bug where CTAS statements have non determistic types. if a limit
             # is applied to a ctas statement, VARCHAR types default to 1 in some instances.
             select_statement = statement.expression.copy()
-            for select_or_union in select_statement.find_all(exp.Select, exp.Union):
+            for select_or_union in select_statement.find_all(exp.Select, exp.SetOperation):
                 select_or_union.set("limit", None)
                 select_or_union.set("where", None)
 
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index 9b2867a8..f5c3e444 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -422,20 +422,20 @@ class _Model(ModelMeta, frozen=True):
         """
         query = self.render_query_or_raise(**render_kwarg).copy()
 
-        for select_or_union in query.find_all(exp.Select, exp.Union):
-            cte = select_or_union.find_ancestor(exp.With, exp.Select, exp.Subquery)
+        for select_or_set_op in query.find_all(exp.Select, exp.SetOperation):
+            cte = select_or_set_op.find_ancestor(exp.With, exp.Select, exp.Subquery)
             skip_limit = isinstance(cte, exp.With) and cte.recursive
 
-            if isinstance(select_or_union, exp.Select) and select_or_union.args.get("from"):
-                select_or_union.where(exp.false(), copy=False)
-                if not skip_limit and not isinstance(select_or_union.parent, exp.Union):
-                    select_or_union.limit(0, copy=False)
+            if isinstance(select_or_set_op, exp.Select) and select_or_set_op.args.get("from"):
+                select_or_set_op.where(exp.false(), copy=False)
+                if not skip_limit and not isinstance(select_or_set_op.parent, exp.SetOperation):
+                    select_or_set_op.limit(0, copy=False)
             elif (
                 not skip_limit
-                and isinstance(select_or_union, exp.Union)
-                and not isinstance(select_or_union.parent, exp.Union)
+                and isinstance(select_or_set_op, exp.SetOperation)
+                and not isinstance(select_or_set_op.parent, exp.SetOperation)
             ):
-                select_or_union.set("limit", exp.Limit(expression=exp.Literal.number(0)))
+                select_or_set_op.set("limit", exp.Limit(expression=exp.Literal.number(0)))
 
         if self.managed_columns:
             query.select(
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 4e863ceb..2ff93e23 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -228,17 +228,22 @@ class QualifiedViewName(PydanticModel, frozen=True):
     def schema_for_environment(
         self, environment_naming_info: EnvironmentNamingInfo, dialect: DialectType = None
     ) -> str:
-        schema = self.schema_name or c.DEFAULT_SCHEMA
+        normalize = environment_naming_info.normalize_name
+
+        if self.schema_name:
+            schema = self.schema_name
+        else:
+            schema = c.DEFAULT_SCHEMA
+            if normalize:
+                schema = normalize_identifiers(schema, dialect=dialect).name
+
         if (
             environment_naming_info.name.lower() != c.PROD
             and environment_naming_info.suffix_target.is_schema
         ):
             env_name = environment_naming_info.name
-            if environment_naming_info.normalize_name:
+            if normalize:
                 env_name = normalize_identifiers(env_name, dialect=dialect).name
-                if not self.schema_name:
-                    # We also normalize the schema if it's the default one
-                    schema = normalize_identifiers(schema, dialect=dialect).name
 
             schema = f"{schema}__{env_name}"
 
diff --git a/sqlmesh/schedulers/airflow/dag_generator.py b/sqlmesh/schedulers/airflow/dag_generator.py
index 9f00c017..ba33f36c 100644
--- a/sqlmesh/schedulers/airflow/dag_generator.py
+++ b/sqlmesh/schedulers/airflow/dag_generator.py
@@ -73,6 +73,7 @@ class SnapshotDagGenerator:
         external_table_sensor_factory: t.Optional[
             t.Callable[[t.Dict[str, t.Any]], BaseSensorOperator]
         ],
+        sensor_mode: str,
         state_reader: StateReader,
     ):
         self._engine_operator = engine_operator
@@ -81,6 +82,7 @@ class SnapshotDagGenerator:
         self._ddl_engine_operator_args = ddl_engine_operator_args or {}
         self._external_table_sensor_factory = external_table_sensor_factory
         self._state_reader = state_reader
+        self._sensor_mode = sensor_mode
 
     def generate_cadence_dags(self, snapshots: t.Iterable[SnapshotIdLike]) -> t.List[DAG]:
         dags = []
@@ -644,6 +646,7 @@ class SnapshotDagGenerator:
                         target_snapshot_info=upstream_snapshot.table_info,
                         this_snapshot=snapshot,
                         task_id=f"{sanitize_name(upstream_snapshot.node.name)}_{upstream_snapshot.version}_high_water_mark_sensor",
+                        mode=self._sensor_mode,
                     )
                 )
 
@@ -664,6 +667,7 @@ class SnapshotDagGenerator:
                 snapshot=snapshot,
                 external_table_sensor_factory=self._external_table_sensor_factory,
                 task_id="external_high_water_mark_sensor",
+                mode=self._sensor_mode,
                 start=start,
                 end=end,
             )
diff --git a/sqlmesh/schedulers/airflow/integration.py b/sqlmesh/schedulers/airflow/integration.py
index 2abc4682..8ab618d1 100644
--- a/sqlmesh/schedulers/airflow/integration.py
+++ b/sqlmesh/schedulers/airflow/integration.py
@@ -65,6 +65,8 @@ class SQLMeshAirflow:
         plan_application_dag_ttl: Determines the time-to-live period for finished plan application DAGs.
             Once this period is exceeded, finished plan application DAGs are deleted by the janitor. Default: 2 days.
         external_table_sensor_factory: A factory function that creates a sensor operator for a given signal payload.
+        sensor_mode: The mode to use for SQLMesh sensors. Supported values are "poke" and "reschedule".
+            See https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/sensors.html for more details. Default: "reschedule".
         generate_cadence_dags: Whether to generate cadence DAGs for model versions that are currently deployed to production.
     """
 
@@ -80,6 +82,7 @@ class SQLMeshAirflow:
         external_table_sensor_factory: t.Optional[
             t.Callable[[t.Dict[str, t.Any]], BaseSensorOperator]
         ] = None,
+        sensor_mode: str = "reschedule",
         generate_cadence_dags: bool = True,
     ):
         if isinstance(engine_operator, str):
@@ -105,6 +108,7 @@ class SQLMeshAirflow:
         self._external_table_sensor_factory = external_table_sensor_factory
         self._generate_cadence_dags = generate_cadence_dags
         self._default_catalog = default_catalog
+        self._sensor_mode = sensor_mode
 
     @classmethod
     def set_default_catalog(cls, default_catalog: str) -> None:
@@ -188,6 +192,7 @@ class SQLMeshAirflow:
             self._ddl_engine_operator,
             self._ddl_engine_operator_args,
             self._external_table_sensor_factory,
+            self._sensor_mode,
             state_reader,
         )
 
diff --git a/sqlmesh/schedulers/airflow/operators/targets.py b/sqlmesh/schedulers/airflow/operators/targets.py
index 76f31ed6..cecd9809 100644
--- a/sqlmesh/schedulers/airflow/operators/targets.py
+++ b/sqlmesh/schedulers/airflow/operators/targets.py
@@ -105,7 +105,7 @@ class BaseTarget(abc.ABC, t.Generic[CP]):
         return payload
 
 
-class SnapshotEvaluationTarget(BaseTarget[commands.EvaluateCommandPayload], PydanticModel):
+class SnapshotEvaluationTarget(PydanticModel, BaseTarget[commands.EvaluateCommandPayload]):
     """The target which contains attributes necessary to evaluate a given snapshot.
 
     Args:
@@ -176,7 +176,7 @@ class SnapshotEvaluationTarget(BaseTarget[commands.EvaluateCommandPayload], Pyda
         return self.execution_time or context["dag_run"].logical_date
 
 
-class SnapshotPromotionTarget(BaseTarget[commands.PromoteCommandPayload], PydanticModel):
+class SnapshotPromotionTarget(PydanticModel, BaseTarget[commands.PromoteCommandPayload]):
     """The target which contains attributes necessary to perform snapshot promotion in a given environment.
 
     The promotion means creation of views associated with the environment which target physical tables
@@ -208,7 +208,7 @@ class SnapshotPromotionTarget(BaseTarget[commands.PromoteCommandPayload], Pydant
         )
 
 
-class SnapshotDemotionTarget(BaseTarget[commands.DemoteCommandPayload], PydanticModel):
+class SnapshotDemotionTarget(PydanticModel, BaseTarget[commands.DemoteCommandPayload]):
     """The target which contains attributes necessary to perform snapshot demotion in a given environment.
 
     The demotion means deletion of views that match names of provided snapshots in the target environment.
@@ -234,7 +234,7 @@ class SnapshotDemotionTarget(BaseTarget[commands.DemoteCommandPayload], Pydantic
         )
 
 
-class SnapshotCleanupTarget(BaseTarget[commands.CleanupCommandPayload], PydanticModel):
+class SnapshotCleanupTarget(PydanticModel, BaseTarget[commands.CleanupCommandPayload]):
     """The target which contains attributes necessary to perform table cleanup of expired snapshots"""
 
     command_type: commands.CommandType = commands.CommandType.CLEANUP
@@ -266,7 +266,7 @@ class SnapshotCleanupTarget(BaseTarget[commands.CleanupCommandPayload], Pydantic
         return command
 
 
-class SnapshotCreateTablesTarget(BaseTarget[commands.CreateTablesCommandPayload], PydanticModel):
+class SnapshotCreateTablesTarget(PydanticModel, BaseTarget[commands.CreateTablesCommandPayload]):
     """The target which creates physical tables for the given set of new snapshots."""
 
     command_type: commands.CommandType = commands.CommandType.CREATE_TABLES
@@ -293,7 +293,7 @@ class SnapshotCreateTablesTarget(BaseTarget[commands.CreateTablesCommandPayload]
         )
 
 
-class SnapshotMigrateTablesTarget(BaseTarget[commands.MigrateTablesCommandPayload], PydanticModel):
+class SnapshotMigrateTablesTarget(PydanticModel, BaseTarget[commands.MigrateTablesCommandPayload]):
     """The target which updates schemas of existing physical tables to bring them in correspondance
     with schemas of target snapshots.
     """


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}'""",
diff --git a/tests/core/test_context.py b/tests/core/test_context.py
index 144397c9..2cbbf4ce 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -735,3 +735,20 @@ def test_disabled_model(copy_to_temp_path):
 
     assert (path[0] / "models" / "disabled.sql").exists()
     assert not context.get_model("sushi.disabled")
+
+
+def test_override_dialect_normalization_strategy():
+    config = Config(
+        model_defaults=ModelDefaultsConfig(dialect="duckdb,normalization_strategy=lowercase")
+    )
+
+    # This has the side-effect of mutating DuckDB globally to override its normalization strategy
+    Context(config=config)
+
+    from sqlglot.dialects import DuckDB
+    from sqlglot.dialects.dialect import NormalizationStrategy
+
+    assert DuckDB.NORMALIZATION_STRATEGY == NormalizationStrategy.LOWERCASE
+
+    # The above change is applied globally so we revert it to avoid breaking other tests
+    DuckDB.NORMALIZATION_STRATEGY = NormalizationStrategy.CASE_INSENSITIVE
diff --git a/tests/schedulers/airflow/test_dag_generator.py b/tests/schedulers/airflow/test_dag_generator.py
index 15f4a896..5e3abdc3 100644
--- a/tests/schedulers/airflow/test_dag_generator.py
+++ b/tests/schedulers/airflow/test_dag_generator.py
@@ -15,7 +15,8 @@ from sqlmesh.core.snapshot import (
 from sqlmesh.schedulers.airflow.dag_generator import SnapshotDagGenerator
 from sqlmesh.schedulers.airflow import common
 from sqlmesh.schedulers.airflow.operators.targets import BaseTarget, SnapshotEvaluationTarget
-from sqlmesh.utils.date import to_datetime
+from sqlmesh.schedulers.airflow.operators.hwm_sensor import HighWaterMarkSensor
+from sqlmesh.utils.date import to_datetime, to_timestamp
 
 
 class TestSubmitOperator(BaseOperator):
@@ -67,6 +68,7 @@ def test_generate_plan_application_dag__batch_index_populated(mocker: MockerFixt
         ddl_engine_operator=TestSubmitOperator,
         ddl_engine_operator_args={},
         external_table_sensor_factory=None,
+        sensor_mode="reschedule",
         state_reader=state_reader_mock,
     )
 
@@ -125,3 +127,45 @@ def test_generate_plan_application_dag__batch_index_populated(mocker: MockerFixt
         assert command is not None
         assert target.batch_index == batch_idx
         assert command.batch_index == batch_idx
+
+
+def test_sensor_mode_override(mocker: MockerFixture, make_snapshot):
+    snapshot_a = make_snapshot(
+        SqlModel(name="a", kind=dict(name="FULL"), query=parse_one("select 1 as a, ds")),
+    )
+    snapshot_a.categorize_as(SnapshotChangeCategory.BREAKING)
+    snapshot_a.unpaused_ts = to_timestamp("2024-01-01")
+
+    snapshot_b = make_snapshot(
+        SqlModel(name="b", kind=dict(name="FULL"), query=parse_one("select a, ds from a")),
+        nodes={snapshot_a.name: snapshot_a.node},
+    )
+    snapshot_b.categorize_as(SnapshotChangeCategory.BREAKING)
+    snapshot_b.unpaused_ts = to_timestamp("2024-01-01")
+
+    state_reader_mock = mocker.Mock()
+    state_reader_mock.get_snapshots.return_value = {
+        snapshot_a.snapshot_id: snapshot_a,
+        snapshot_b.snapshot_id: snapshot_b,
+    }
+
+    generator = SnapshotDagGenerator(
+        engine_operator=TestSubmitOperator,
+        engine_operator_args={},
+        ddl_engine_operator=TestSubmitOperator,
+        ddl_engine_operator_args={},
+        external_table_sensor_factory=None,
+        sensor_mode="poke",
+        state_reader=state_reader_mock,
+    )
+
+    dags = generator.generate_cadence_dags([snapshot_a, snapshot_b])
+    assert len(dags) == 2
+
+    assert len(dags[0].tasks) == 1
+    assert isinstance(dags[0].tasks[0], TestSubmitOperator)
+
+    assert len(dags[1].tasks) == 2
+    assert isinstance(dags[1].tasks[0], HighWaterMarkSensor)
+    assert isinstance(dags[1].tasks[1], TestSubmitOperator)
+    assert dags[1].tasks[0].mode == "poke"