Repository: TobikoData/sqlmesh
Pull request #2783 (2024-06-17 17:07:07)

Fix: Shortcuts and some styling for editor
Fix jumping cursor when user typing
Run custom query on Cmd + Enter
Show suggestion in uppercase in user typing in uppercase


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/docs/integrations/engines/trino.md b/docs/integrations/engines/trino.md
index e88050cc..ff45e56a 100644
--- a/docs/integrations/engines/trino.md
+++ b/docs/integrations/engines/trino.md
@@ -129,10 +129,11 @@ gateway_name:
 
 === "Basic Auth"
 
-    | Option     | Description                              |  Type  | Required |
-    |------------|------------------------------------------|:------:|:--------:|
-    | `method`   | `basic`                                  | string |    Y     |
-    | `password` | The password to use when authenticating. | string |    Y     |
+    | Option     | Description                                                                                                                                                                  |  Type  | Required |
+    | ---------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | :----: | :------: |
+    | `method`   | `basic`                                                                                                                                                                      | string |    Y     |
+    | `password` | The password to use when authenticating.                                                                                                                                     | string |    Y     |
+    | `verify`   | Boolean flag for whether SSL verification should occur. Default: [trinodb Python client](https://github.com/trinodb/trino-python-client) default (`true` as of this writing) |  bool  |    N     |
 
 
     ```yaml linenums="1"
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/config/connection.py b/sqlmesh/core/config/connection.py
index f5dd644f..343bfbb8 100644
--- a/sqlmesh/core/config/connection.py
+++ b/sqlmesh/core/config/connection.py
@@ -1185,6 +1185,7 @@ class TrinoConnectionConfig(ConnectionConfig):
     timezone: t.Optional[str] = None
     # Basic/LDAP
     password: t.Optional[str] = None
+    verify: t.Optional[bool] = None  # disable SSL verification (ignored if `cert` is provided)
     # LDAP
     impersonation_user: t.Optional[str] = None
     # Kerberos
@@ -1309,7 +1310,7 @@ class TrinoConnectionConfig(ConnectionConfig):
             "auth": auth,
             "user": self.impersonation_user or self.user,
             "max_attempts": self.retries,
-            "verify": self.cert,
+            "verify": self.cert if self.cert is not None else self.verify,
             "source": "sqlmesh",
         }
 
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..2ef6b7ad 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -182,10 +182,18 @@ class EngineAdapter:
         batch_size = self.DEFAULT_BATCH_SIZE if batch_size is None else batch_size
         if isinstance(query_or_df, (exp.Query, exp.DerivedTable)):
             return [SourceQuery(query_factory=lambda: query_or_df)]  # type: ignore
+
         if not columns_to_types:
             raise SQLMeshError(
-                "It is expected that if a DF is passed in then columns_to_types is set"
+                "It is expected that if a DataFrame is passed in then columns_to_types is set"
+            )
+        if query_or_df.empty:
+            raise SQLMeshError(
+                "Cannot construct source query from an empty DataFrame. This error is commonly "
+                "related to Python models that produce no data. For such models, consider yielding "
+                "from an empty generator if the resulting set is empty, i.e. use `yield from ()`."
             )
+
         return self._df_to_source_queries(
             query_or_df, columns_to_types, batch_size, target_table=target_table
         )
@@ -631,6 +639,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 +1433,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 +1581,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 +2010,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.
     """
diff --git a/web/client/openapi.json b/web/client/openapi.json
index 65313d81..48d0c0f9 100644
--- a/web/client/openapi.json
+++ b/web/client/openapi.json
@@ -871,6 +871,7 @@
             "title": "Stack"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["timestamp", "message", "origin"],
         "title": "ApiExceptionPayload"
@@ -897,6 +898,7 @@
           },
           "batches": { "type": "integer", "title": "Batches" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name", "interval", "batches"],
         "title": "BackfillDetails"
@@ -931,6 +933,7 @@
             "title": "Interval"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name", "completed", "total", "start"],
         "title": "BackfillTask"
@@ -1059,6 +1062,7 @@
             ]
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name", "diff"],
         "title": "ChangeDirect"
@@ -1079,6 +1083,7 @@
             "default": []
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name"],
         "title": "ChangeDisplay"
@@ -1099,6 +1104,7 @@
             "default": []
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name"],
         "title": "ChangeIndirect"
@@ -1112,6 +1118,7 @@
             "title": "Description"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "type"],
         "title": "Column"
@@ -1133,6 +1140,7 @@
             "default": []
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path"],
         "title": "Directory"
@@ -1150,6 +1158,11 @@
             "anyOf": [{ "type": "string" }, { "type": "null" }],
             "title": "Catalog Name Override"
           },
+          "normalize_name": {
+            "type": "boolean",
+            "title": "Normalize Name",
+            "default": true
+          },
           "snapshots": {
             "items": { "$ref": "#/components/schemas/SnapshotTableInfo" },
             "type": "array",
@@ -1244,6 +1257,7 @@
             "default": ""
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "Environments"
       },
@@ -1282,6 +1296,7 @@
           },
           "limit": { "type": "integer", "title": "Limit", "default": 1000 }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["model", "start", "end", "execution_time"],
         "title": "EvaluateInput"
@@ -1291,6 +1306,7 @@
           "sql": { "type": "string", "title": "Sql" },
           "limit": { "type": "integer", "title": "Limit", "default": 1000 }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["sql"],
         "title": "FetchdfInput"
@@ -1309,6 +1325,7 @@
             "title": "Content"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path"],
         "title": "File"
@@ -1358,6 +1375,7 @@
             "title": "Models"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["models"],
         "title": "LineageColumn"
@@ -1371,6 +1389,7 @@
             "default": false
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["version"],
         "title": "Meta"
@@ -1407,6 +1426,7 @@
           },
           "hash": { "type": "string", "title": "Hash" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": [
           "name",
@@ -1526,6 +1546,7 @@
             "title": "Annotated"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "ModelDetails"
       },
@@ -1534,6 +1555,7 @@
         "enum": [
           "INCREMENTAL_BY_TIME_RANGE",
           "INCREMENTAL_BY_UNIQUE_KEY",
+          "INCREMENTAL_BY_PARTITION",
           "INCREMENTAL_UNMANAGED",
           "FULL",
           "SCD_TYPE_2",
@@ -1573,6 +1595,7 @@
             "default": []
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "ModelsDiff"
       },
@@ -1656,6 +1679,7 @@
             ]
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanApplyStageTracker"
       },
@@ -1701,6 +1725,7 @@
             ]
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanCancelStageTracker"
       },
@@ -1729,6 +1754,7 @@
             "title": "End"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanDates"
       },
@@ -1778,6 +1804,7 @@
             "default": false
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanOptions"
       },
@@ -1835,6 +1862,7 @@
             ]
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanOverviewStageTracker"
       },
@@ -1857,6 +1885,7 @@
             "default": {}
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageBackfill"
       },
@@ -1874,6 +1903,7 @@
             "title": "Models"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageBackfills"
       },
@@ -1881,6 +1911,7 @@
         "properties": {
           "meta": { "$ref": "#/components/schemas/TrackableMeta" }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageCancel"
       },
@@ -1914,6 +1945,7 @@
           },
           "meta": { "$ref": "#/components/schemas/TrackableMeta" }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageChanges"
       },
@@ -1923,6 +1955,7 @@
           "total_tasks": { "type": "integer", "title": "Total Tasks" },
           "num_tasks": { "type": "integer", "title": "Num Tasks" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["total_tasks", "num_tasks"],
         "title": "PlanStageCreation"
@@ -1937,6 +1970,7 @@
             "title": "Target Environment"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["total_tasks", "num_tasks", "target_environment"],
         "title": "PlanStagePromote"
@@ -1945,6 +1979,7 @@
         "properties": {
           "meta": { "$ref": "#/components/schemas/TrackableMeta" }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageRestate"
       },
@@ -1952,11 +1987,13 @@
         "properties": {
           "meta": { "$ref": "#/components/schemas/TrackableMeta" }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageValidation"
       },
       "Query": {
         "properties": { "sql": { "type": "string", "title": "Sql" } },
+        "additionalProperties": false,
         "type": "object",
         "required": ["sql"],
         "title": "Query"
@@ -1967,6 +2004,7 @@
           "expression": { "type": "string", "title": "Expression" },
           "unique": { "type": "boolean", "title": "Unique" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "expression", "unique"],
         "title": "Reference"
@@ -2021,6 +2059,7 @@
             "title": "Dialect"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["model"],
         "title": "RenderInput"
@@ -2039,6 +2078,7 @@
           "target_count": { "type": "integer", "title": "Target Count" },
           "count_pct_change": { "type": "number", "title": "Count Pct Change" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": [
           "source",
@@ -2081,6 +2121,7 @@
             "title": "Modified"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": [
           "source",
@@ -2095,9 +2136,9 @@
       },
       "SnapshotChangeCategory": {
         "type": "integer",
-        "enum": [1, 2, 3, 4, 5],
+        "enum": [1, 2, 3, 4, 5, 6],
         "title": "SnapshotChangeCategory",
-        "description": "Values are ordered by decreasing severity and that ordering is required.\n\nBREAKING: The change requires that snapshot modified and downstream dependencies be rebuilt\nNON_BREAKING: The change requires that only the snapshot modified be rebuilt\nFORWARD_ONLY: The change requires no rebuilding\nINDIRECT_BREAKING: The change was caused indirectly and is breaking.\nINDIRECT_NON_BREAKING: The change was caused indirectly by a non-breaking change."
+        "description": "Values are ordered by decreasing severity and that ordering is required.\n\nBREAKING: The change requires that snapshot modified and downstream dependencies be rebuilt\nNON_BREAKING: The change requires that only the snapshot modified be rebuilt\nFORWARD_ONLY: The change requires no rebuilding\nINDIRECT_BREAKING: The change was caused indirectly and is breaking.\nINDIRECT_NON_BREAKING: The change was caused indirectly by a non-breaking change.\nMETADATA: The change was caused by a metadata update."
       },
       "SnapshotDataVersion": {
         "properties": {
@@ -2222,6 +2263,7 @@
             "title": "On"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["schema_diff", "row_diff", "on"],
         "title": "TableDiff"
@@ -2231,6 +2273,7 @@
           "name": { "type": "string", "title": "Name" },
           "path": { "type": "string", "format": "path", "title": "Path" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path"],
         "title": "TestCase"
@@ -2241,6 +2284,7 @@
           "path": { "type": "string", "format": "path", "title": "Path" },
           "tb": { "type": "string", "title": "Tb" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path", "tb"],
         "title": "TestErrorOrFailure"
@@ -2269,6 +2313,7 @@
             "title": "Successes"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["tests_run", "failures", "errors", "skipped", "successes"],
         "title": "TestResult"
@@ -2279,6 +2324,7 @@
           "path": { "type": "string", "format": "path", "title": "Path" },
           "reason": { "type": "string", "title": "Reason" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path", "reason"],
         "title": "TestSkipped"
@@ -2296,6 +2342,7 @@
           },
           "done": { "type": "boolean", "title": "Done", "default": false }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "TrackableMeta"
       },
diff --git a/web/client/package-lock.json b/web/client/package-lock.json
index 49954465..7bd0f0c6 100644
--- a/web/client/package-lock.json
+++ b/web/client/package-lock.json
@@ -8,13 +8,14 @@
       "name": "tobiko",
       "version": "0.0.0",
       "dependencies": {
-        "@codemirror/commands": "^6.2.5",
-        "@codemirror/lang-python": "^6.1.3",
-        "@codemirror/lang-sql": "^6.5.4",
-        "@codemirror/language": "^6.9.0",
-        "@codemirror/legacy-modes": "^6.3.3",
-        "@codemirror/state": "^6.2.1",
-        "@codemirror/view": "^6.17.0",
+        "@codemirror/autocomplete": "^6.16.2",
+        "@codemirror/commands": "^6.6.0",
+        "@codemirror/lang-python": "^6.1.6",
+        "@codemirror/lang-sql": "^6.6.4",
+        "@codemirror/language": "^6.10.2",
+        "@codemirror/legacy-modes": "^6.4.0",
+        "@codemirror/state": "^6.4.1",
+        "@codemirror/view": "^6.28.1",
         "@headlessui/react": "^1.7.17",
         "@heroicons/react": "^2.0.18",
         "@radix-ui/react-context-menu": "^2.1.4",
@@ -27,6 +28,7 @@
         "@uiw/react-codemirror": "^4.21.12",
         "apache-arrow": "^13.0.0",
         "clsx": "^2.0.0",
+        "diff": "^5.2.0",
         "elkjs": "^0.8.2",
         "pluralize": "^8.0.0",
         "react": "^18.2.0",
@@ -44,6 +46,7 @@
         "@testing-library/jest-dom": "^6.1.2",
         "@testing-library/react": "^14.0.0",
         "@testing-library/user-event": "^14.4.3",
+        "@types/diff": "^5.2.1",
         "@types/pluralize": "^0.0.30",
         "@types/react": "^18.2.21",
         "@types/react-dom": "^18.2.7",
@@ -381,8 +384,9 @@
       }
     },
     "node_modules/@codemirror/autocomplete": {
-      "version": "6.11.1",
-      "license": "MIT",
+      "version": "6.16.2",
+      "resolved": "https://registry.npmjs.org/@codemirror/autocomplete/-/autocomplete-6.16.2.tgz",
+      "integrity": "sha512-MjfDrHy0gHKlPWsvSsikhO1+BOh+eBHNgfH1OXs1+DAf30IonQldgMM3kxLDTG9ktE7kDLaA1j/l7KMPA4KNfw==",
       "dependencies": {
         "@codemirror/language": "^6.0.0",
         "@codemirror/state": "^6.0.0",
@@ -397,41 +401,48 @@
       }
     },
     "node_modules/@codemirror/commands": {
-      "version": "6.3.2",
-      "license": "MIT",
+      "version": "6.6.0",
+      "resolved": "https://registry.npmjs.org/@codemirror/commands/-/commands-6.6.0.tgz",
+      "integrity": "sha512-qnY+b7j1UNcTS31Eenuc/5YJB6gQOzkUoNmJQc0rznwqSRpeaWWpjkWy2C/MPTcePpsKJEM26hXrOXl1+nceXg==",
       "dependencies": {
         "@codemirror/language": "^6.0.0",
-        "@codemirror/state": "^6.2.0",
-        "@codemirror/view": "^6.0.0",
+        "@codemirror/state": "^6.4.0",
+        "@codemirror/view": "^6.27.0",
         "@lezer/common": "^1.1.0"
       }
     },
     "node_modules/@codemirror/lang-python": {
-      "version": "6.1.3",
-      "license": "MIT",
+      "version": "6.1.6",
+      "resolved": "https://registry.npmjs.org/@codemirror/lang-python/-/lang-python-6.1.6.tgz",
+      "integrity": "sha512-ai+01WfZhWqM92UqjnvorkxosZ2aq2u28kHvr+N3gu012XqY2CThD67JPMHnGceRfXPDBmn1HnyqowdpF57bNg==",
       "dependencies": {
         "@codemirror/autocomplete": "^6.3.2",
         "@codemirror/language": "^6.8.0",
+        "@codemirror/state": "^6.0.0",
+        "@lezer/common": "^1.2.1",
         "@lezer/python": "^1.1.4"
       }
     },
     "node_modules/@codemirror/lang-sql": {
-      "version": "6.5.4",
-      "license": "MIT",
+      "version": "6.6.4",
+      "resolved": "https://registry.npmjs.org/@codemirror/lang-sql/-/lang-sql-6.6.4.tgz",
+      "integrity": "sha512-n+FVfKGut+frOvor9dU5pFUalcP614WBNQ9IT1kOUj1t6LFLjWHi2I9DdxXnJuxqFV9jTyYF79coDV3ilSJqCw==",
       "dependencies": {
         "@codemirror/autocomplete": "^6.0.0",
         "@codemirror/language": "^6.0.0",
         "@codemirror/state": "^6.0.0",
+        "@lezer/common": "^1.2.0",
         "@lezer/highlight": "^1.0.0",
         "@lezer/lr": "^1.0.0"
       }
     },
     "node_modules/@codemirror/language": {
-      "version": "6.9.3",
-      "license": "MIT",
+      "version": "6.10.2",
+      "resolved": "https://registry.npmjs.org/@codemirror/language/-/language-6.10.2.tgz",
+      "integrity": "sha512-kgbTYTo0Au6dCSc/TFy7fK3fpJmgHDv1sG1KNQKJXVi+xBTEeBPY/M30YXiU6mMXeH+YIDLsbrT4ZwNRdtF+SA==",
       "dependencies": {
         "@codemirror/state": "^6.0.0",
-        "@codemirror/view": "^6.0.0",
+        "@codemirror/view": "^6.23.0",
         "@lezer/common": "^1.1.0",
         "@lezer/highlight": "^1.0.0",
         "@lezer/lr": "^1.0.0",
@@ -439,8 +450,9 @@
       }
     },
     "node_modules/@codemirror/legacy-modes": {
-      "version": "6.3.3",
-      "license": "MIT",
+      "version": "6.4.0",
+      "resolved": "https://registry.npmjs.org/@codemirror/legacy-modes/-/legacy-modes-6.4.0.tgz",
+      "integrity": "sha512-5m/K+1A6gYR0e+h/dEde7LoGimMjRtWXZFg4Lo70cc8HzjSdHe3fLwjWMR0VRl5KFT1SxalSap7uMgPKF28wBA==",
       "dependencies": {
         "@codemirror/language": "^6.0.0"
       }
@@ -464,8 +476,9 @@
       }
     },
     "node_modules/@codemirror/state": {
-      "version": "6.3.3",
-      "license": "MIT"
+      "version": "6.4.1",
+      "resolved": "https://registry.npmjs.org/@codemirror/state/-/state-6.4.1.tgz",
+      "integrity": "sha512-QkEyUiLhsJoZkbumGZlswmAhA7CBU02Wrz7zvH4SrcifbsqwlXShVXg65f3v/ts57W3dqyamEriMhij1Z3Zz4A=="
     },
     "node_modules/@codemirror/theme-one-dark": {
       "version": "6.1.2",
@@ -478,10 +491,11 @@
       }
     },
     "node_modules/@codemirror/view": {
-      "version": "6.22.1",
-      "license": "MIT",
+      "version": "6.28.1",
+      "resolved": "https://registry.npmjs.org/@codemirror/view/-/view-6.28.1.tgz",
+      "integrity": "sha512-BUWr+zCJpMkA/u69HlJmR+YkV4yPpM81HeMkOMZuwFa8iM5uJdEPKAs1icIRZKkKmy0Ub1x9/G3PQLTXdpBxrQ==",
       "dependencies": {
-        "@codemirror/state": "^6.1.4",
+        "@codemirror/state": "^6.4.0",
         "style-mod": "^4.1.0",
         "w3c-keyname": "^2.2.4"
       }
@@ -766,8 +780,9 @@
       }
     },
     "node_modules/@lezer/common": {
-      "version": "1.1.2",
-      "license": "MIT"
+      "version": "1.2.1",
+      "resolved": "https://registry.npmjs.org/@lezer/common/-/common-1.2.1.tgz",
+      "integrity": "sha512-yemX0ZD2xS/73llMZIK6KplkjIjf2EvAHcinDi/TfJ9hS25G0388+ClHt6/3but0oOxinTcQHJLDXh6w1crzFQ=="
     },
     "node_modules/@lezer/highlight": {
       "version": "1.2.0",
@@ -2630,6 +2645,13 @@
         "@types/d3-selection": "*"
       }
     },
+    "node_modules/@types/diff": {
+      "version": "5.2.1",
+      "resolved": "https://registry.npmjs.org/@types/diff/-/diff-5.2.1.tgz",
+      "integrity": "sha512-uxpcuwWJGhe2AR1g8hD9F5OYGCqjqWnBUQFD8gMZsDbv8oPHzxJF6iMO6n8Tk0AdzlxoaaoQhOYlIg/PukVU8g==",
+      "dev": true,
+      "license": "MIT"
+    },
     "node_modules/@types/es-aggregate-error": {
       "version": "1.0.6",
       "dev": true,
@@ -3627,7 +3649,9 @@
       }
     },
     "node_modules/caniuse-lite": {
-      "version": "1.0.30001566",
+      "version": "1.0.30001636",
+      "resolved": "https://registry.npmjs.org/caniuse-lite/-/caniuse-lite-1.0.30001636.tgz",
+      "integrity": "sha512-bMg2vmr8XBsbL6Lr0UHXy/21m84FTxDLWn2FSqMd5PrlbMxwJlQnC2YWYxVgp66PZE+BBNF2jYQUBKCo1FDeZg==",
       "dev": true,
       "funding": [
         {
@@ -4126,6 +4150,14 @@
       "version": "1.2.2",
       "license": "Apache-2.0"
     },
+    "node_modules/diff": {
+      "version": "5.2.0",
+      "resolved": "https://registry.npmjs.org/diff/-/diff-5.2.0.tgz",
+      "integrity": "sha512-uIFDxqpRZGZ6ThOk84hEfqWoHx2devRFvpTZcTHur85vImfaxUbTW9Ryh4CpCuDnToOP1CEtXKIgytHBPVff5A==",
+      "engines": {
+        "node": ">=0.3.1"
+      }
+    },
     "node_modules/diff-sequences": {
       "version": "29.6.3",
       "dev": true,
diff --git a/web/client/package.json b/web/client/package.json
index d66ab1aa..b56ec9fc 100644
--- a/web/client/package.json
+++ b/web/client/package.json
@@ -16,13 +16,14 @@
     "generate:api": "orval --config ./orval.config.ts"
   },
   "dependencies": {
-    "@codemirror/commands": "^6.2.5",
-    "@codemirror/lang-python": "^6.1.3",
-    "@codemirror/lang-sql": "^6.5.4",
-    "@codemirror/language": "^6.9.0",
-    "@codemirror/legacy-modes": "^6.3.3",
-    "@codemirror/state": "^6.2.1",
-    "@codemirror/view": "^6.17.0",
+    "@codemirror/autocomplete": "^6.16.2",
+    "@codemirror/commands": "^6.6.0",
+    "@codemirror/lang-python": "^6.1.6",
+    "@codemirror/lang-sql": "^6.6.4",
+    "@codemirror/language": "^6.10.2",
+    "@codemirror/legacy-modes": "^6.4.0",
+    "@codemirror/state": "^6.4.1",
+    "@codemirror/view": "^6.28.1",
     "@headlessui/react": "^1.7.17",
     "@heroicons/react": "^2.0.18",
     "@radix-ui/react-context-menu": "^2.1.4",
@@ -35,6 +36,7 @@
     "@uiw/react-codemirror": "^4.21.12",
     "apache-arrow": "^13.0.0",
     "clsx": "^2.0.0",
+    "diff": "^5.2.0",
     "elkjs": "^0.8.2",
     "pluralize": "^8.0.0",
     "react": "^18.2.0",
@@ -52,6 +54,7 @@
     "@testing-library/jest-dom": "^6.1.2",
     "@testing-library/react": "^14.0.0",
     "@testing-library/user-event": "^14.4.3",
+    "@types/diff": "^5.2.1",
     "@types/pluralize": "^0.0.30",
     "@types/react": "^18.2.21",
     "@types/react-dom": "^18.2.7",
diff --git a/web/client/src/index.css b/web/client/src/index.css
index c6fd970c..6e0a5233 100644
--- a/web/client/src/index.css
+++ b/web/client/src/index.css
@@ -222,7 +222,7 @@ html[mode='dark'] {
   --color-logo-lighter: var(--color-neutral-100);
 
   --color-editor-text: var(--color-neutral-300);
-  --color-editor-active-line-text: var(--color-neutral-100);
+  --color-editor-active-line-text: var(--color-neutral-200);
 
   --color-graph-edge-secondary: var(--color-primary-500);
   --color-graph-edge-main: var(--color-primary-20);
@@ -250,7 +250,7 @@ html[mode='light'] {
   --color-logo-lighter: var(--color-neutral-300);
 
   --color-editor-text: var(--color-neutral-600);
-  --color-editor-active-line-text: var(--color-neutral-800);
+  --color-editor-active-line-text: var(--color-neutral-700);
 
   --color-graph-edge-secondary: var(--color-secondary-500);
   --color-graph-edge-main: var(--color-secondary-20);
diff --git a/web/client/src/library/components/editor/Editor.css b/web/client/src/library/components/editor/Editor.css
index e36aca16..3a0efa73 100644
--- a/web/client/src/library/components/editor/Editor.css
+++ b/web/client/src/library/components/editor/Editor.css
@@ -2,98 +2,106 @@
 .cm-editor .cm-content {
   background-color: var(--color-editor);
   color: var(--color-editor-text);
+  font-weight: 400;
+  font-size: 0.85rem;
 }
-
 .cm-editor .cm-gutters {
   background-color: var(--color-editor);
   border: none;
   padding-left: 0.25rem;
   padding-right: 0.5rem;
 }
-
-.cm-editor .cm-activeLineGutter {
-  background-color: var(--color-theme-lighter);
-  color: var(--color-editor-active-line-text);
-}
-
+.cm-editor .cm-activeLineGutter,
 .cm-editor .cm-activeLine {
-  background-color: var(--color-theme-lighter);
   color: var(--color-editor-active-line-text);
 }
-
 .cm-editor .cm-line::selection,
 .cm-editor .cm-line > *::selection {
   background-color: var(--color-brand-100) !important;
   color: var(--color-brand-700) !important;
 }
-
 .cm-editor .cm-tooltip.cm-tooltip-autocomplete {
-  border: 1px solid var(--color-neutral-300) !important;
-  border-radius: 0.25rem;
+  margin-top: 0.25rem;
 }
-
-.cm-editor .cm-tooltip-autocomplete ul li[aria-selected='true'] {
-  background: var(--color-primary-10);
+.cm-editor .cm-tooltip.cm-tooltip-autocomplete > ul {
+  border-radius: 0.35rem;
+  padding: 0.25rem 0.25rem;
   color: var(--color-text);
 }
-
+.cm-editor .cm-tooltip.cm-tooltip-autocomplete > ul > li {
+  padding: 0.25rem 0.5rem;
+  cursor: pointer;
+  border-radius: 0.3rem;
+  border: 1px solid transparent;
+  font-weight: normal;
+  display: flex;
+  align-items: center;
+}
+.cm-editor .cm-tooltip-autocomplete > ul > li[aria-selected='true'] {
+  color: var(--color-text);
+}
+.cm-editor
+  .cm-tooltip.cm-tooltip-autocomplete
+  > ul
+  > li
+  .cm-completionMatchedText {
+  opacity: 1;
+  font-weight: bold;
+  text-decoration: none;
+}
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon-column,
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon-model {
-  margin: 0.125rem 1.25rem 0.125rem 0;
+  margin-right: 1.5rem;
 }
-
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon::before {
   display: inline-block;
   padding: 0.125rem 0.25rem;
   border-radius: 0.25rem;
 }
-
+.cm-editor .cm-tooltip-autocomplete .cm-completionIcon {
+  display: none;
+}
+.cm-editor .cm-tooltip-autocomplete .cm-completionIcon-column,
+.cm-editor .cm-tooltip-autocomplete .cm-completionIcon-model {
+  display: block;
+  opacity: 1;
+}
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon-column::before {
   content: 'col';
-  color: var(--color-brand-600);
-  background: var(--color-brand-10);
+  font-weight: bold;
 }
-
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon-model::before {
   content: 'mdl';
-  color: var(--color-primary-500);
-  background: var(--color-primary-10);
+  font-weight: bold;
 }
-
 .cm-editor .cm-scroller {
   font-family: inherit;
 }
-
 .cm-editor .cm-scroller::-webkit-scrollbar,
 .cm-editor .cm-tooltip.cm-tooltip-autocomplete ul::-webkit-scrollbar {
   height: var(--scrollbar-size);
   width: var(--scrollbar-size);
 }
-
 .cm-editor .cm-scroller::-webkit-scrollbar-track,
 .cm-editor .cm-tooltip.cm-tooltip-autocomplete ul::-webkit-scrollbar-track {
   background: transparent;
 }
-
 .cm-editor .cm-scroller::-webkit-scrollbar-thumb,
 .cm-editor .cm-tooltip.cm-tooltip-autocomplete ul::-webkit-scrollbar-thumb {
-  background: var(--color-brand);
+  background: var(--color-neutral-40);
   border-radius: var(--scrollbar-radius);
 }
-
 .cm-editor .cm-tooltip {
   border: none !important;
   outline: none !important;
   background: var(--color-theme) !important;
 }
-
 .sqlmesh-expression {
   background: var(--color-secondary-10);
   box-shadow:
     inset 0 -1px 0 0 var(--color-secondary-500),
     inset 0 1px 0 0 var(--color-secondary-500);
 }
-
 .sqlmesh-model,
 .sqlmesh-model > span {
   display: inline-block;
@@ -106,11 +114,9 @@
   cursor: pointer;
   box-shadow: inset 0 -2px 0 0 var(--color-primary-300);
 }
-
 .sqlmesh-model.--is-action-mode:hover {
   opacity: 0.8;
 }
-
 .sqlmesh-model.--is-active-model,
 .sqlmesh-model.--is-active-model:hover,
 .sqlmesh-model.--is-active-model > span,
@@ -119,7 +125,6 @@
   cursor: default;
   opacity: 1;
 }
-
 .sqlmesh-model__column.--is-original,
 .sqlmesh-model__column.--is-original > span,
 .sqlmesh-model__column.--is-alias,
@@ -128,7 +133,6 @@
   color: var(--color-accent-300);
   font-weight: bold;
 }
-
 .sqlmesh-model__column.--is-active-model.--is-alias,
 .sqlmesh-model__column.--is-active-model.--is-alias > span {
   display: inline-block;
@@ -144,3 +148,83 @@
 .sqlmesh-model__column.--is-action-mode:hover {
   opacity: 0.8;
 }
+
+[mode='dark'] .cm-editor .cm-activeLineGutter,
+[mode='dark'] .cm-editor .cm-activeLine {
+  background-color: var(--color-neutral-30);
+}
+[mode='dark']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  .cm-completionIcon-model::before {
+  color: var(--color-secondary-300);
+  background: var(--color-secondary-20);
+}
+[mode='dark']
+  .cm-editor
+  .cm-tooltip.cm-tooltip-autocomplete
+  > ul
+  > li
+  .cm-completionMatchedText {
+  color: var(--color-accent-200);
+  background: var(--color-accent-50);
+}
+[mode='dark']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  .cm-completionIcon-column::before {
+  color: var(--color-brand-500);
+  background: var(--color-brand-10);
+}
+[mode='dark'] .cm-editor .cm-tooltip.cm-tooltip-autocomplete > ul {
+  border: 1px solid var(--color-neutral-50) !important;
+  box-shadow: 0 2px 3px 0 var(--color-neutral-20);
+}
+[mode='dark']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  > ul
+  > li[aria-selected='true'] {
+  background: var(--color-neutral-30);
+  border: 1px solid var(--color-neutral-50) !important;
+}
+
+[mode='light'] .cm-editor .cm-activeLineGutter,
+[mode='light'] .cm-editor .cm-activeLine {
+  background-color: var(--color-neutral-10);
+}
+[mode='light']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  .cm-completionIcon-model::before {
+  color: var(--color-secondary-500);
+  background: var(--color-secondary-10);
+}
+[mode='light']
+  .cm-editor
+  .cm-tooltip.cm-tooltip-autocomplete
+  > ul
+  > li
+  .cm-completionMatchedText {
+  color: var(--color-accent-500);
+  background: var(--color-accent-5);
+}
+[mode='light']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  .cm-completionIcon-column::before {
+  color: var(--color-brand-500);
+  background: var(--color-brand-20);
+}
+[mode='light'] .cm-editor .cm-tooltip.cm-tooltip-autocomplete > ul {
+  border: 1px solid var(--color-neutral-20) !important;
+  box-shadow: 0 2px 3px 0 var(--color-neutral-20);
+}
+[mode='light']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  > ul
+  > li[aria-selected='true'] {
+  background: var(--color-neutral-5);
+  border: 1px solid var(--color-neutral-10) !important;
+}
diff --git a/web/client/src/library/components/editor/Editor.tsx b/web/client/src/library/components/editor/Editor.tsx
index 9a4a06fe..6e108d5f 100644
--- a/web/client/src/library/components/editor/Editor.tsx
+++ b/web/client/src/library/components/editor/Editor.tsx
@@ -95,8 +95,10 @@ function EditorMain({ tab }: { tab: EditorTab }): JSX.Element {
     return [
       ...defaultKeymapsEditorTab,
       {
-        key: 'Shift-Mod-Enter',
+        key: 'Mod-Enter',
+        win: 'Ctrl-Enter',
         preventDefault: true,
+        stopPropagation: true,
         run(view) {
           const sql = view.state.doc.toString()
 
diff --git a/web/client/src/library/components/editor/EditorCode.tsx b/web/client/src/library/components/editor/EditorCode.tsx
index fb437b0a..eb594649 100644
--- a/web/client/src/library/components/editor/EditorCode.tsx
+++ b/web/client/src/library/components/editor/EditorCode.tsx
@@ -1,9 +1,9 @@
 import { useCallback, useEffect, useMemo, useState } from 'react'
 import CodeMirror from '@uiw/react-codemirror'
-import { type KeyBinding, keymap } from '@codemirror/view'
-import { type Extension } from '@codemirror/state'
+import { type EditorView, type KeyBinding, keymap } from '@codemirror/view'
+import { type ChangeSpec, type Extension } from '@codemirror/state'
 import { useApiFileByPath, useMutationApiSaveFile } from '~/api'
-import { debounceSync, isNil } from '~/utils'
+import { debounceSync, isNil, isNotNil } from '~/utils'
 import { useStoreContext } from '~/context/context'
 import { useStoreEditor } from '~/context/editor'
 import {
@@ -31,7 +31,9 @@ import {
   SQLMeshDialect,
   SQLMeshDialectCleanUp,
 } from './extensions/SQLMeshDialect'
-import { indentMore } from '@codemirror/commands'
+import { indentMore, defaultKeymap, historyKeymap } from '@codemirror/commands'
+import { EMPTY_STRING } from '@components/search/help'
+import { diffChars } from 'diff'
 
 export { CodeEditorDefault, CodeEditorRemoteFile }
 
@@ -55,25 +57,15 @@ function CodeEditorDefault({
   const { mode } = useColorScheme()
 
   const models = useStoreContext(s => s.models)
-
   const engine = useStoreEditor(s => s.engine)
   const dialects = useStoreEditor(s => s.dialects)
 
-  const [value, setValue] = useState(content)
-
+  const [editorView, setEditorView] = useState<Optional<EditorView>>()
   const [dialectOptions, setDialectOptions] = useState<{
     types: string
     keywords: string
   }>()
 
-  const debouncedChange = useCallback(
-    debounceSync(function handleChange(value): void {
-      setValue(value)
-      onChange?.(value)
-    }, 500),
-    [],
-  )
-
   const handleEngineWorkerMessage = useCallback((e: MessageEvent): void => {
     if (e.data.topic === 'dialect') {
       setDialectOptions(e.data.payload)
@@ -96,26 +88,34 @@ function CodeEditorDefault({
     () => dialects.map(d => d.dialect_title),
     [dialects],
   )
-  const extensionKeymap = useMemo(
-    () =>
-      keymap.of(
-        [
-          ...(keymaps ?? []),
-          [
-            {
-              key: 'Tab',
-              preventDefault: true,
-              run(e: any) {
-                return isNil(completionStatus(e.state))
-                  ? indentMore(e)
-                  : acceptCompletion(e)
-              },
-            },
-          ],
-        ].flat(),
-      ),
-    [keymaps],
-  )
+  const extensionKeymap = useMemo(() => {
+    const keys = [
+      defaultKeymap,
+      historyKeymap,
+      ...(keymaps ?? []),
+      [
+        {
+          key: 'Tab',
+          preventDefault: true,
+          run(e: any) {
+            return isNil(completionStatus(e.state))
+              ? indentMore(e)
+              : acceptCompletion(e)
+          },
+        },
+      ],
+    ]
+      .flat()
+      .reduce<Record<string, KeyBinding>>((acc, binding) => {
+        if (isNotNil(binding.key)) {
+          acc[binding.key] = binding
+        }
+
+        return acc
+      }, {})
+
+    return keymap.of(Object.values(keys))
+  }, [keymaps])
   const allModels = useMemo(() => Array.from(models.values()), [models])
   const allModelsNames = useMemo(
     () =>
@@ -191,8 +191,49 @@ function CodeEditorDefault({
   }, [dialect])
 
   useEffect(() => {
-    setValue(content)
-  }, [content])
+    updateEditor(content)
+  }, [editorView, content, extensionsAll])
+
+  function updateEditor(newContent: string = ''): void {
+    if (isNil(editorView)) return
+
+    const state = editorView.state
+    const currentContent = state.doc.toString()
+
+    if (currentContent === newContent) return
+
+    const changes: ChangeSpec[] = []
+
+    if (currentContent === EMPTY_STRING) {
+      changes.push({ from: 0, to: 0, insert: newContent })
+    } else {
+      const items = diffChars(currentContent, newContent)
+
+      let from = 0
+
+      items.forEach(item => {
+        const count = item.count ?? 0
+
+        if (isNotNil(item.removed)) {
+          const to = from + count
+          const change = { from, to }
+
+          from = to
+
+          changes.push(change)
+        } else if (isNotNil(item.added)) {
+          changes.push({ from, to: from, insert: item.value })
+        } else {
+          from = from + count
+        }
+      })
+    }
+
+    editorView.dispatch({
+      changes,
+      scrollIntoView: true,
+    })
+  }
 
   return (
     <div className={clsx('flex w-full h-full', className)}>
@@ -200,14 +241,16 @@ function CodeEditorDefault({
         height="100%"
         width="100%"
         className={clsx('flex w-full h-full font-mono text-xs', className)}
-        value={value}
         indentWithTab={false}
         extensions={extensionsAll}
-        onChange={debouncedChange}
+        onChange={onChange}
         readOnly={isNil(onChange)}
+        onCreateEditor={setEditorView}
         basicSetup={{
           autocompletion: false,
+          defaultKeymap: false,
         }}
+        autoFocus
       />
     </div>
   )
diff --git a/web/client/src/library/components/editor/EditorFooter.tsx b/web/client/src/library/components/editor/EditorFooter.tsx
index 7a6f7125..44ef59ff 100644
--- a/web/client/src/library/components/editor/EditorFooter.tsx
+++ b/web/client/src/library/components/editor/EditorFooter.tsx
@@ -50,7 +50,7 @@ export default function EditorFooter({ tab }: { tab: EditorTab }): JSX.Element {
   }
 
   return (
-    <div className="flex w-full items-center px-2 min-h-[2rem]">
+    <div className="flex w-full items-center px-2 min-h-[2rem] overflow-hidden">
       {tab.file.isRemote && (
         <EditorIndicator
           className="mr-2"
diff --git a/web/client/src/library/components/editor/extensions/SQLMeshDialect.ts b/web/client/src/library/components/editor/extensions/SQLMeshDialect.ts
index 38928131..b6fddd30 100644
--- a/web/client/src/library/components/editor/extensions/SQLMeshDialect.ts
+++ b/web/client/src/library/components/editor/extensions/SQLMeshDialect.ts
@@ -3,6 +3,7 @@ import {
   completeFromList,
   type CompletionContext,
   type Completion,
+  type CompletionResult,
 } from '@codemirror/autocomplete'
 import { keywordCompletionSource, SQLDialect } from '@codemirror/lang-sql'
 import { LanguageSupport } from '@codemirror/language'
@@ -139,28 +140,19 @@ export const SQLMeshDialect: ExtensionSQLMeshDialect = function SQLMeshDialect(
           suggestions.push(...allColumnsNames)
         }
 
-        const wordLastChar = getFirstChar(word.text)
-        const isUpperCase =
-          isStringEmptyOrNil(wordLastChar) &&
-          wordLastChar === wordLastChar.toUpperCase()
+        const wordLastChar = word.text.trim().slice(-1)
+        const completionResult = keywordCompletionSource(
+          lang,
+          wordLastChar === wordLastChar.toUpperCase(),
+        )(ctx) as CompletionResult
 
-        if (isUpperCase) {
-          suggestions = suggestions.map(suggestion => ({
-            ...suggestion,
-            label: suggestion.label.toUpperCase(),
-          }))
+        if (isNotNil(completionResult)) {
+          const options = completionResult.options ?? []
+          completionResult.options = options.concat(suggestions)
+          completionResult.map = () => null // Updates suggestions to switch between lower and upper case
         }
 
-        const source = isArrayEmpty(suggestions)
-          ? keywordCompletionSource(lang, isUpperCase)(ctx)
-          : {
-              from: word.from,
-              to: word.to,
-              options: suggestions,
-              validFor: /^\w*$/,
-            }
-
-        return source
+        return completionResult
       },
     }),
   ])
@@ -262,18 +254,7 @@ function matchWordWithSpacesAfter(
 }
 
 function maybeSuggestion(suggestions: Completion[], word: string): boolean {
-  return (
-    word.length > 1 &&
-    suggestions.some(suggestion =>
-      suggestion.label.toLowerCase().includes(word.toLowerCase()),
-    )
+  return suggestions.some(suggestion =>
+    suggestion.label.toLowerCase().includes(word.toLowerCase()),
   )
 }
-
-function getFirstChar(word: string): string {
-  for (const char of word) {
-    if (/[a-zA-Z]/.test(char)) return char
-  }
-
-  return ''
-}
diff --git a/web/client/src/library/components/editor/extensions/index.ts b/web/client/src/library/components/editor/extensions/index.ts
index a2b0a87b..7dd557a8 100644
--- a/web/client/src/library/components/editor/extensions/index.ts
+++ b/web/client/src/library/components/editor/extensions/index.ts
@@ -10,6 +10,7 @@ import {
 } from '@codemirror/view'
 import { type ModelSQLMeshModel } from '@models/sqlmesh-model'
 import SQLMeshModel from './SQLMeshModel'
+import { isNil } from '@utils/index'
 
 export { events, HoverTooltip, SQLMeshModel, SQLMeshExpression }
 
@@ -36,7 +37,7 @@ function HoverTooltip(models: Map<string, ModelSQLMeshModel>): Extension {
       const modelName = view.state.doc.sliceString(start, end)
       const model = models.get(modelName)
 
-      if (model == null) return null
+      if (isNil(model)) return null
 
       return {
         pos: start,
diff --git a/web/client/src/library/components/editor/hooks.ts b/web/client/src/library/components/editor/hooks.ts
index 0982bf40..ae4bb08f 100644
--- a/web/client/src/library/components/editor/hooks.ts
+++ b/web/client/src/library/components/editor/hooks.ts
@@ -106,7 +106,7 @@ function useSQLMeshModelExtensions(
     )
 
     function handleEventModelClick(event: MouseEvent): void {
-      if (event.ctrlKey) {
+      if (event.metaKey) {
         const model = findModel(event, models)
 
         if (isNil(model)) return
@@ -116,7 +116,7 @@ function useSQLMeshModelExtensions(
     }
 
     function handleEventlColumnClick(event: MouseEvent): void {
-      if (event.ctrlKey) {
+      if (event.metaKey) {
         if (isNil(model)) return
 
         const column = findColumn(event, model)
@@ -127,28 +127,24 @@ function useSQLMeshModelExtensions(
       }
     }
 
-    return isNil(model)
-      ? []
-      : ([
-          models.size > 0 && isActionMode && HoverTooltip(models),
-          events({
-            keydown: e => {
-              if (e.ctrlKey) {
-                setIsActionMode(true)
-              }
-            },
-            keyup: e => {
-              if (isFalse(e.ctrlKey)) {
-                setIsActionMode(false)
-              }
-            },
-          }),
-          isNotNil(handleModelClick) &&
-            events({ click: handleEventModelClick }),
-          isNotNil(handleModelColumn) &&
-            events({ click: handleEventlColumnClick }),
-          SQLMeshModel(models, columns, isActionMode, model),
-        ].filter(Boolean) as Extension[])
+    return [
+      models.size > 0 && isActionMode && HoverTooltip(models),
+      events({
+        keydown: e => {
+          if (e.metaKey) {
+            setIsActionMode(true)
+          }
+        },
+        keyup: e => {
+          if (isFalse(e.metaKey)) {
+            setIsActionMode(false)
+          }
+        },
+      }),
+      isNotNil(handleModelClick) && events({ click: handleEventModelClick }),
+      isNotNil(handleModelColumn) && events({ click: handleEventlColumnClick }),
+      SQLMeshModel(models, columns, isActionMode, model),
+    ].filter(Boolean) as Extension[]
   }, [handleModelClick, handleModelColumn, model, models, files, isActionMode])
 
   return extensions
diff --git a/web/client/src/library/pages/root/Navigation.tsx b/web/client/src/library/pages/root/Navigation.tsx
index 84116c3b..6fc0cf4b 100644
--- a/web/client/src/library/pages/root/Navigation.tsx
+++ b/web/client/src/library/pages/root/Navigation.tsx
@@ -16,6 +16,7 @@ export default function PageNavigation(): JSX.Element {
       className={clsx(
         'relative min-w-[10rem] px-2 min-h-8 max-h-8 w-full flex items-center',
         modules.showHistoryNavigation ? 'justify-between' : 'justify-end',
+        isFetchingModels && 'overflow-hidden',
       )}
     >
       {modules.showHistoryNavigation && <HistoryNavigation />}


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..733ad9b8 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -7,6 +7,8 @@ from unittest.mock import PropertyMock, call, patch
 
 import freezegun
 import pytest
+import pandas as pd
+from pathlib import Path
 from pytest_mock.plugin import MockerFixture
 from sqlglot import parse_one
 from sqlglot.errors import SchemaError
@@ -24,7 +26,7 @@ from sqlmesh.core.config import (
 from sqlmesh.core.context import Context
 from sqlmesh.core.dialect import parse, schema_
 from sqlmesh.core.environment import Environment
-from sqlmesh.core.model import load_sql_based_model
+from sqlmesh.core.model import load_sql_based_model, model
 from sqlmesh.core.model.kind import ModelKindName
 from sqlmesh.core.plan import BuiltInPlanEvaluator, PlanBuilder
 from sqlmesh.utils.date import (
@@ -34,7 +36,7 @@ from sqlmesh.utils.date import (
     to_timestamp,
     yesterday_ds,
 )
-from sqlmesh.utils.errors import ConfigError
+from sqlmesh.utils.errors import ConfigError, SQLMeshError
 from tests.utils.test_filesystem import create_temp_file
 
 
@@ -285,6 +287,32 @@ def test_plan_execution_time():
     )
 
 
+def test_python_model_empty_df_raises(sushi_context, capsys):
+    @model(
+        "memory.sushi.test_model",
+        columns={"col": "int"},
+        kind=dict(name=ModelKindName.FULL),
+    )
+    def entrypoint(context, **kwargs):
+        yield pd.DataFrame({"col": []})
+
+    test_model = model.get_registry()["memory.sushi.test_model"].model(
+        module_path=Path("."), path=Path(".")
+    )
+    sushi_context.upsert_model(test_model)
+
+    capsys.readouterr()
+    with pytest.raises(SQLMeshError):
+        sushi_context.plan(no_prompts=True, auto_apply=True)
+
+    assert (
+        "Cannot construct source query from an empty \nDataFrame. This error "
+        "is commonly related to Python models that produce no data.\nFor such "
+        "models, consider yielding from an empty generator if the resulting set "
+        "\nis empty, i.e. use `yield from ()`"
+    ) in capsys.readouterr().out
+
+
 def test_env_and_default_schema_normalization(mocker: MockerFixture):
     from sqlglot.dialects import DuckDB
     from sqlglot.dialects.dialect import NormalizationStrategy
@@ -735,3 +763,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/dbt/test_integration.py b/tests/dbt/test_integration.py
index fc3a10bd..a7838897 100644
--- a/tests/dbt/test_integration.py
+++ b/tests/dbt/test_integration.py
@@ -192,11 +192,12 @@ test_config = config"""
         values: t.List[t.Tuple[int, str, str]],
     ):
         df = create_df(values, self.source_schema)
-        adapter.replace_query(
-            "sushi.raw_marketing",
-            df,
-            columns_to_types=columns_to_types_from_df(df),
-        )
+        columns_to_types = columns_to_types_from_df(df)
+
+        if values:
+            adapter.replace_query("sushi.raw_marketing", df, columns_to_types=columns_to_types)
+        else:
+            adapter.create_table("sushi.raw_marketing", columns_to_types=columns_to_types)
 
     def _normalize_dbt_dataframe(
         self,
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"