Repository: TobikoData/sqlmesh
Pull request #2064 (2024-01-31 19:02:03)

Feat: allow configuring default formatting behavior in the UI via sql…
…mesh config. In subsequent iterations perhaps this serves as the "default options" while a modal allows overriding them for the session. But as a general source of truth, I think its better here since you have consistency across the entire team as well as version control. 

```yaml
default_gateway: duckdb
gateways:
  duckdb:
    connection:
      type: duckdb

model_defaults:
  dialect: 'duckdb'

format:
  pad: 4
  indent: 4
  leading_commas: true
  max_text_width: 120

ui:
  format_on_save: true
```

CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/TobikoData/sqlmesh?pullRequest=2064) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our [Contributor License Agreement](https://cla-assistant.io/TobikoData/sqlmesh?pullRequest=2064) before we can accept your contribution.<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/TobikoData/sqlmesh?pullRequest=2064) it.</sub>
z3z1ma: I wondered the same, personally I think top level would be best. Then the CLI flags would just serve as overrides.
This would ensure uniformity between both uses of SQLMesh without having to ensure your CLI flags match ui opts. 
izeigerman: @z3z1ma @mykalmax  +1 to making format settings top-level
z3z1ma: @izeigerman I cleaned it up a bit, renamed the newline option to make its intent clear, and made sure format_options is a top level config applicable to both the CLI and UI, and lastly that CLI flags take precedence over the config settings as would be expected.
z3z1ma: Seems like a pydantic inconsistency on v1?
mykalmax: do we actually need an option to override global `format` by passing flags to CLI, every `override` will be adding number of changes to the project repo and maybe be `overridden` by someone running format without `override` flags later

should we then do the same for `ui` part for config ?
(bringing back) 
```yaml
ui:
  format_on_save: true
  format_options:
    pad: 4
    indent: 4
    leading_commas: true
    max_text_width: 120
```
z3z1ma: @mykalmax I think the user-specific config in the home directory supported by sqlmesh will already solve for user specific overrides in the ui technically. i think longer term, there should be a modal in the UI with options preset based on the `format` config. And they can tweak there in theory. 
z3z1ma: I also think CLI precedence is typical enough to expect in a tool, it allows for faster iteration -- or imperative flows which dont use the config.py/yaml. I dont think its necessary to constrain it personally 🤷 
z3z1ma: Thinking about it more @mykalmax , 
I could go either way really on dropping the cli flags and using just the yaml honestly
 
Since there is some simplification in the code then. And is easier to grok than stacking CLI flags. And its uniform with UI by default as you said. 
mykalmax: @z3z1ma would it make more sense to have `format` top level with `enable: true` as default (and `false` to disable formatting), then drop `ui` from config and if `format` is set for a project `format-on-save` on ui just be reading options from config.
Remove all overrides (flags), so only way to change/set format is to change it in config for entire project (or disable).
In this case if projects needs `formatting`: pass options to config and it will be consistent in cli and ui and if formatting not needed , just disable for entire project.
z3z1ma: I think thats fine to consolidate and drop flags if everyone agrees @mykalmax ! 

I would maybe nest the ui format-on-save option though.

```yaml
format:
  pad: 4
  indent: 4
  normalize: false
  ui:
    enabled: true # auto format on save
```

I think being able to run `sqlmesh format` in the CLI and the UI **auto-formatting** files on save are 2 different concerns and the one worth disabling is the uncontrollable implicit one (the UI). The CLI is by contrast a very explicit action and a user running it, typically wants to run it. And if they dont want to run it, they dont run it. We don't need to make it a no-op for the sake of consistency with UI autoformat.

I will await decision. (can always pivot later 😄)
mykalmax: @izeigerman @z3z1ma my suggestion was to drop `ui:` from `config` at all and if formatting is `enabled` for a project `UI` just picks up settings from `format:` in `config`. 
But with option to opt-out from `auto-formatting` on save on `UI` it should be separate top level `ui:` to avoid nesting.
Sorry for confusion 
```yaml
...

format:
  pad: 4
  indent: 4
  leading_commas: true
  max_text_width: 120

ui:
  format_on_save: true

...
```
z3z1ma: @izeigerman can you retrigger test. Seems to be hanging? Or I can do another commit I guess

Patch:
diff --git a/examples/airflow/docker_compose_decorator.py b/examples/airflow/docker_compose_decorator.py
index babf3727..21d7fda8 100644
--- a/examples/airflow/docker_compose_decorator.py
+++ b/examples/airflow/docker_compose_decorator.py
@@ -108,13 +108,13 @@ for airflow_component in ["airflow-scheduler", "airflow-worker"]:
             raise RuntimeError(
                 "Tried to use Databricks Airflow Engine operator but did not define `DATABRICKS_SERVER_HOSTNAME`, `DATABRICKS_TOKEN`, `DATABRICKS_HTTP_PATH`"
             )
-        environment_variables[
-            "AIRFLOW_CONN_DATABRICKS_DEFAULT"
-        ] = "databricks://${DATABRICKS_SERVER_HOSTNAME}?token=${DATABRICKS_TOKEN}&http_path=${DATABRICKS_HTTP_PATH}"
+        environment_variables["AIRFLOW_CONN_DATABRICKS_DEFAULT"] = (
+            "databricks://${DATABRICKS_SERVER_HOSTNAME}?token=${DATABRICKS_TOKEN}&http_path=${DATABRICKS_HTTP_PATH}"
+        )
     if os.getenv("DEMO_GITHUB_PAT"):
-        environment_variables[
-            "AIRFLOW_CONN_GITHUB_DEFAULT"
-        ] = '{"conn_type": "github", "password": "${DEMO_GITHUB_PAT}"}'
+        environment_variables["AIRFLOW_CONN_GITHUB_DEFAULT"] = (
+            '{"conn_type": "github", "password": "${DEMO_GITHUB_PAT}"}'
+        )
     docker_compose["services"][airflow_component]["environment"].update(environment_variables)
 
 
diff --git a/setup.py b/setup.py
index 48c1d371..7dbf8da8 100644
--- a/setup.py
+++ b/setup.py
@@ -63,7 +63,7 @@ setup(
             "autoflake==1.7.7",
             "agate==1.6.3",
             "beautifulsoup4",
-            "black==22.6.0",
+            "black==24.1.1",
             "dbt-bigquery",
             "dbt-core",
             "dbt-duckdb>=1.4.2",
@@ -74,7 +74,7 @@ setup(
             "google-cloud-bigquery",
             "google-cloud-bigquery-storage",
             "isort==5.10.1",
-            "mypy~=1.4.0",
+            "mypy~=1.8.0",
             # Pendulum 3.0.0 contains a breaking change for Airflow.
             # To test if this is fixed with future versions, check if this line works:
             # https://github.com/apache/airflow/blob/main/airflow/settings.py#L59
diff --git a/sqlmesh/__init__.py b/sqlmesh/__init__.py
index ab545b70..8006475a 100644
--- a/sqlmesh/__init__.py
+++ b/sqlmesh/__init__.py
@@ -1,6 +1,7 @@
 """
 .. include:: ../README.md
 """
+
 from __future__ import annotations
 
 import glob
diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py
index 006134bc..e95fb74d 100644
--- a/sqlmesh/cli/main.py
+++ b/sqlmesh/cli/main.py
@@ -201,27 +201,26 @@ def evaluate(
     help="Transpile project models to the specified dialect.",
 )
 @click.option(
-    "--newline",
+    "--append-newline",
     is_flag=True,
-    help="Include a new line at the end of each file.",
+    help="Include a newline at the end of each file.",
+    default=None,
 )
 @click.option(
     "--normalize",
     is_flag=True,
     help="Whether or not to normalize identifiers to lowercase.",
-    default=False,
+    default=None,
 )
 @click.option(
     "--pad",
     type=int,
     help="Determines the pad size in a formatted string.",
-    default=2,
 )
 @click.option(
     "--indent",
     type=int,
     help="Determines the indentation size in a formatted string.",
-    default=2,
 )
 @click.option(
     "--normalize-functions",
@@ -232,21 +231,18 @@ def evaluate(
     "--leading-comma",
     is_flag=True,
     help="Determines whether or not the comma is leading or trailing in select expressions. Default is trailing.",
-    default=False,
+    default=None,
 )
 @click.option(
     "--max-text-width",
     type=int,
     help="The max number of characters in a segment before creating new lines in pretty mode.",
-    default=80,
 )
 @click.pass_context
 @error_handler
-def format(
-    ctx: click.Context, transpile: t.Optional[str] = None, newline: bool = False, **kwargs: t.Any
-) -> None:
+def format(ctx: click.Context, **kwargs: t.Any) -> None:
     """Format all SQL models."""
-    ctx.obj.format(transpile, newline, **{k: v for k, v in kwargs.items() if v is not None})
+    ctx.obj.format(**{k: v for k, v in kwargs.items() if v is not None})
 
 
 @cli.command("diff")
diff --git a/sqlmesh/core/config/format.py b/sqlmesh/core/config/format.py
new file mode 100644
index 00000000..2b8fd215
--- /dev/null
+++ b/sqlmesh/core/config/format.py
@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import typing as t
+
+from sqlmesh.core.config.base import BaseConfig
+
+
+class FormatConfig(BaseConfig):
+    """The format configuration for SQL code.
+
+    Args:
+        normalize: Whether to normalize the SQL code or not.
+        pad: The number of spaces to use for padding.
+        indent: The number of spaces to use for indentation.
+        normalize_functions: Whether or not to normalize all function names. Possible values are: 'upper', 'lower'
+        leading_comma: Whether to use leading commas or not.
+        max_text_width: The maximum text width in a segment before creating new lines.
+        append_newline: Whether to append a newline to the end of the file or not.
+    """
+
+    normalize: bool = False
+    pad: int = 2
+    indent: int = 2
+    normalize_functions: t.Optional[str] = None
+    leading_comma: bool = False
+    max_text_width: int = 80
+    append_newline: bool = False
+
+    @property
+    def generator_options(self) -> t.Dict[str, t.Any]:
+        """Options which can be passed through to the SQLGlot Generator class.
+
+        Returns:
+            The generator options.
+        """
+        return self.dict(exclude={"append_newline"})
diff --git a/sqlmesh/core/config/root.py b/sqlmesh/core/config/root.py
index c96dfa9b..b95eaf0f 100644
--- a/sqlmesh/core/config/root.py
+++ b/sqlmesh/core/config/root.py
@@ -19,10 +19,12 @@ from sqlmesh.core.config.connection import (
     DuckDBConnectionConfig,
     connection_config_validator,
 )
+from sqlmesh.core.config.format import FormatConfig
 from sqlmesh.core.config.gateway import GatewayConfig
 from sqlmesh.core.config.model import ModelDefaultsConfig
 from sqlmesh.core.config.run import RunConfig
 from sqlmesh.core.config.scheduler import BuiltInSchedulerConfig, SchedulerConfig
+from sqlmesh.core.config.ui import UIConfig
 from sqlmesh.core.loader import Loader, SqlMeshLoader
 from sqlmesh.core.notification_target import NotificationTarget
 from sqlmesh.core.user import User
@@ -64,6 +66,8 @@ class Config(BaseConfig):
         environment_suffix_target: Indicates whether to append the environment name to the schema or table name.
         default_target_environment: The name of the environment that will be the default target for the `sqlmesh plan` and `sqlmesh run` commands.
         log_limit: The default number of logs to keep.
+        format: The formatting options for SQL code.
+        ui: The UI configuration for SQLMesh.
     """
 
     gateways: t.Dict[str, GatewayConfig] = {"": GatewayConfig()}
@@ -97,6 +101,8 @@ class Config(BaseConfig):
     log_limit: int = c.DEFAULT_LOG_LIMIT
     cicd_bot: t.Optional[CICDBotConfig] = None
     run: RunConfig = RunConfig()
+    format: FormatConfig = FormatConfig()
+    ui: UIConfig = UIConfig()
 
     _FIELD_UPDATE_STRATEGY: t.ClassVar[t.Dict[str, UpdateStrategy]] = {
         "gateways": UpdateStrategy.KEY_UPDATE,
@@ -108,6 +114,8 @@ class Config(BaseConfig):
         "pinned_environments": UpdateStrategy.EXTEND,
         "physical_schema_override": UpdateStrategy.KEY_UPDATE,
         "run": UpdateStrategy.NESTED_UPDATE,
+        "format": UpdateStrategy.NESTED_UPDATE,
+        "ui": UpdateStrategy.NESTED_UPDATE,
         "loader_kwargs": UpdateStrategy.KEY_UPDATE,
     }
 
diff --git a/sqlmesh/core/config/ui.py b/sqlmesh/core/config/ui.py
new file mode 100644
index 00000000..e24dc433
--- /dev/null
+++ b/sqlmesh/core/config/ui.py
@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from sqlmesh.core.config.base import BaseConfig
+
+
+class UIConfig(BaseConfig):
+    """The UI configuration for SQLMesh.
+
+    Args:
+        format_on_save: Whether to format the SQL code on save or not.
+    """
+
+    format_on_save: bool = True
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index 10952e4a..f354eec2 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -30,6 +30,7 @@ context = Context(path="example")
 context.test()
 ```
 """
+
 from __future__ import annotations
 
 import abc
@@ -518,16 +519,14 @@ class Context(BaseContext):
     @t.overload
     def get_model(
         self, model_or_snapshot: ModelOrSnapshot, raise_if_missing: Literal[True] = True
-    ) -> Model:
-        ...
+    ) -> Model: ...
 
     @t.overload
     def get_model(
         self,
         model_or_snapshot: ModelOrSnapshot,
         raise_if_missing: Literal[False] = False,
-    ) -> t.Optional[Model]:
-        ...
+    ) -> t.Optional[Model]: ...
 
     def get_model(
         self, model_or_snapshot: ModelOrSnapshot, raise_if_missing: bool = False
@@ -557,20 +556,17 @@ class Context(BaseContext):
         return model
 
     @t.overload
-    def get_snapshot(self, node_or_snapshot: NodeOrSnapshot) -> t.Optional[Snapshot]:
-        ...
+    def get_snapshot(self, node_or_snapshot: NodeOrSnapshot) -> t.Optional[Snapshot]: ...
 
     @t.overload
     def get_snapshot(
         self, node_or_snapshot: NodeOrSnapshot, raise_if_missing: Literal[True]
-    ) -> Snapshot:
-        ...
+    ) -> Snapshot: ...
 
     @t.overload
     def get_snapshot(
         self, node_or_snapshot: NodeOrSnapshot, raise_if_missing: Literal[False]
-    ) -> t.Optional[Snapshot]:
-        ...
+    ) -> t.Optional[Snapshot]: ...
 
     def get_snapshot(
         self, node_or_snapshot: NodeOrSnapshot, raise_if_missing: bool = False
@@ -741,7 +737,10 @@ class Context(BaseContext):
         return df
 
     def format(
-        self, transpile: t.Optional[str] = None, newline: bool = False, **kwargs: t.Any
+        self,
+        transpile: t.Optional[str] = None,
+        append_newline: t.Optional[bool] = None,
+        **kwargs: t.Any,
     ) -> None:
         """Format all SQL models."""
         for model in self._models.values():
@@ -760,11 +759,15 @@ class Context(BaseContext):
                                     value=exp.Literal.string(transpile or model.dialect),
                                 )
                             )
+                format = self.config_for_node(model).format
+                opts = {**format.generator_options, **kwargs}
                 file.seek(0)
                 file.write(
-                    format_model_expressions(expressions, transpile or model.dialect, **kwargs)
+                    format_model_expressions(expressions, transpile or model.dialect, **opts)
                 )
-                if newline:
+                if append_newline is None:
+                    append_newline = format.append_newline
+                if append_newline:
                     file.write("\n")
                 file.truncate()
 
@@ -1371,7 +1374,10 @@ class Context(BaseContext):
                 f"\nFailure in audit {error.audit.name} ({error.audit._path})."
             )
             self.console.log_status_update(f"Got {error.count} results, expected 0.")
-            self.console.show_sql(f"{error.query}")
+            if error.query:
+                self.console.show_sql(
+                    f"{error.query.sql(dialect=self.snapshot_evaluator.adapter.dialect)}"
+                )
 
         self.console.log_status_update("Done.")
 
@@ -1559,13 +1565,15 @@ class Context(BaseContext):
         If a snapshot has not been versioned yet, its view name will be returned.
         """
         return {
-            fqn: snapshot.table_name()
-            if snapshot.version
-            else snapshot.qualified_view_name.for_environment(
-                EnvironmentNamingInfo.from_environment_catalog_mapping(
-                    self.config.environment_catalog_mapping,
-                    name=c.PROD,
-                    suffix_target=self.config.environment_suffix_target,
+            fqn: (
+                snapshot.table_name()
+                if snapshot.version
+                else snapshot.qualified_view_name.for_environment(
+                    EnvironmentNamingInfo.from_environment_catalog_mapping(
+                        self.config.environment_catalog_mapping,
+                        name=c.PROD,
+                        suffix_target=self.config.environment_suffix_target,
+                    )
                 )
             )
             for fqn, snapshot in self.snapshots.items()
diff --git a/sqlmesh/core/context_diff.py b/sqlmesh/core/context_diff.py
index f19015da..7643d3dd 100644
--- a/sqlmesh/core/context_diff.py
+++ b/sqlmesh/core/context_diff.py
@@ -9,6 +9,7 @@ The SQLMesh CLI diff command uses ContextDiff to determine what to visualize.
 When creating a ContextDiff object, SQLMesh will compare the snapshots from one environment with those of
 another remote environment and determine if nodes have been added, removed, or modified.
 """
+
 from __future__ import annotations
 
 import typing as t
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index f0f60533..222fe03e 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -6,6 +6,7 @@ generalize its functionality to different engines that have Python Database API
 connections. Rather than executing queries directly against your data stores, SQLMesh components such as
 the SnapshotEvaluator delegate them to engine adapters so these components can be engine-agnostic.
 """
+
 from __future__ import annotations
 
 import contextlib
@@ -247,14 +248,12 @@ class EngineAdapter:
     @t.overload
     def _columns_to_types(
         self, query_or_df: DF, columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None
-    ) -> t.Dict[str, exp.DataType]:
-        ...
+    ) -> t.Dict[str, exp.DataType]: ...
 
     @t.overload
     def _columns_to_types(
         self, query_or_df: Query, columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None
-    ) -> t.Optional[t.Dict[str, exp.DataType]]:
-        ...
+    ) -> t.Optional[t.Dict[str, exp.DataType]]: ...
 
     def _columns_to_types(
         self, query_or_df: QueryOrDF, columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None
@@ -343,10 +342,12 @@ class EngineAdapter:
                 ) as temp_table:
                     for source_query in source_queries:
                         source_query.add_transform(
-                            lambda node: temp_table  # type: ignore
-                            if isinstance(node, exp.Table)
-                            and quote_identifiers(node) == quote_identifiers(target_table)
-                            else node
+                            lambda node: (  # type: ignore
+                                temp_table  # type: ignore
+                                if isinstance(node, exp.Table)
+                                and quote_identifiers(node) == quote_identifiers(target_table)
+                                else node
+                            )
                         )
                     return self._insert_overwrite_by_condition(
                         target_table,
@@ -495,6 +496,17 @@ class EngineAdapter:
             kwargs: Optional create table properties.
         """
         table = exp.to_table(table_name)
+
+        if not columns_to_types_all_known(columns_to_types):
+            # It is ok if the columns types are not known if the table already exists and IF NOT EXISTS is set
+            if exists and self.table_exists(table_name):
+                return
+            raise SQLMeshError(
+                "Cannot create a table without knowing the column types. "
+                "Try casting the columns to an expected type or defining the columns in the model metadata. "
+                f"Columns to types: {columns_to_types}"
+            )
+
         primary_key_expression = (
             [exp.PrimaryKey(expressions=[exp.to_column(k) for k in primary_key])]
             if primary_key and self.SUPPORTS_INDEXES
@@ -554,11 +566,13 @@ class EngineAdapter:
                 exp.ColumnDef(
                     this=exp.to_identifier(column),
                     kind=None if is_view else kind,  # don't include column data type for views
-                    constraints=self._build_col_comment_exp(column, column_descriptions)
-                    if column_descriptions
-                    and engine_supports_schema_comments
-                    and self.comments_enabled
-                    else None,
+                    constraints=(
+                        self._build_col_comment_exp(column, column_descriptions)
+                        if column_descriptions
+                        and engine_supports_schema_comments
+                        and self.comments_enabled
+                        else None
+                    ),
                 )
                 for column, kind in columns_to_types.items()
             ]
@@ -660,9 +674,11 @@ class EngineAdapter:
                 exists=exists,
                 replace=replace,
                 columns_to_types=columns_to_types,
-                table_description=table_description
-                if self.COMMENT_CREATION_TABLE.supports_schema_def and self.comments_enabled
-                else None,
+                table_description=(
+                    table_description
+                    if self.COMMENT_CREATION_TABLE.supports_schema_def and self.comments_enabled
+                    else None
+                ),
                 **kwargs,
             )
         )
@@ -834,9 +850,11 @@ class EngineAdapter:
 
         create_view_properties = self._build_view_properties_exp(
             create_kwargs.pop("table_properties", None),
-            table_description
-            if self.COMMENT_CREATION_VIEW.supports_schema_def and self.comments_enabled
-            else None,
+            (
+                table_description
+                if self.COMMENT_CREATION_VIEW.supports_schema_def and self.comments_enabled
+                else None
+            ),
         )
         if create_view_properties:
             for view_property in create_view_properties.expressions:
@@ -975,13 +993,11 @@ class EngineAdapter:
 
     @t.overload
     @classmethod
-    def _escape_json(cls, value: Query) -> Query:
-        ...
+    def _escape_json(cls, value: Query) -> Query: ...
 
     @t.overload
     @classmethod
-    def _escape_json(cls, value: str) -> str:
-        ...
+    def _escape_json(cls, value: str) -> str: ...
 
     @classmethod
     def _escape_json(cls, value: Query | str) -> Query | str:
@@ -993,9 +1009,11 @@ class EngineAdapter:
             if isinstance(value, str):
                 return double_escape(value)
             return value.transform(
-                lambda e: exp.Literal.string(double_escape(e.name))
-                if isinstance(e, exp.Literal) and e.args["is_string"]
-                else e
+                lambda e: (
+                    exp.Literal.string(double_escape(e.name))
+                    if isinstance(e, exp.Literal) and e.args["is_string"]
+                    else e
+                )
             )
         return value
 
@@ -1118,9 +1136,11 @@ class EngineAdapter:
                             query,
                             table,
                             # Change once Databricks supports REPLACE WHERE with columns
-                            columns=list(columns_to_types)
-                            if not insert_overwrite_strategy.is_replace_where
-                            else None,
+                            columns=(
+                                list(columns_to_types)
+                                if not insert_overwrite_strategy.is_replace_where
+                                else None
+                            ),
                             overwrite=insert_overwrite_strategy.is_insert_overwrite,
                         )
                         if insert_overwrite_strategy.is_replace_where:
@@ -1136,9 +1156,11 @@ class EngineAdapter:
     ) -> None:
         if contains_json and properties:
             properties = {
-                k: self._escape_json(v)
-                if isinstance(v, (str, exp.Subqueryable, exp.DerivedTable))
-                else v
+                k: (
+                    self._escape_json(v)
+                    if isinstance(v, (str, exp.Subqueryable, exp.DerivedTable))
+                    else v
+                )
                 for k, v in properties.items()
             }
         self.execute(exp.update(table_name, properties, where=where))
@@ -1620,9 +1642,11 @@ class EngineAdapter:
             for e in ensure_list(expressions):
                 sql = t.cast(
                     str,
-                    self._to_sql(e, quote=quote_identifiers, **to_sql_kwargs)
-                    if isinstance(e, exp.Expression)
-                    else e,
+                    (
+                        self._to_sql(e, quote=quote_identifiers, **to_sql_kwargs)
+                        if isinstance(e, exp.Expression)
+                        else e
+                    ),
                 )
                 self._log_sql(sql)
                 self._execute(sql, **kwargs)
diff --git a/sqlmesh/core/engine_adapter/mssql.py b/sqlmesh/core/engine_adapter/mssql.py
index 93dad48f..875b8fa9 100644
--- a/sqlmesh/core/engine_adapter/mssql.py
+++ b/sqlmesh/core/engine_adapter/mssql.py
@@ -1,6 +1,5 @@
 """Contains MSSQLEngineAdapter."""
 
-
 from __future__ import annotations
 
 import typing as t
diff --git a/sqlmesh/core/engine_adapter/shared.py b/sqlmesh/core/engine_adapter/shared.py
index edeeac3a..1c98d40f 100644
--- a/sqlmesh/core/engine_adapter/shared.py
+++ b/sqlmesh/core/engine_adapter/shared.py
@@ -266,9 +266,11 @@ def set_catalog(override_mapping: t.Optional[t.Dict[str, CatalogSupport]] = None
 
             obj, container, key = t.cast(
                 t.Tuple[t.Union[str, exp.Table], t.Union[t.Dict, t.List], t.Union[int, str]],
-                (kwargs.get(target_name), kwargs, target_name)
-                if kwargs.get(target_name)
-                else (list_args[target_pos], list_args, target_pos),
+                (
+                    (kwargs.get(target_name), kwargs, target_name)
+                    if kwargs.get(target_name)
+                    else (list_args[target_pos], list_args, target_pos)
+                ),
             )
             to_expression_func = t.cast(
                 t.Callable[[t.Union[str, exp.Table]], exp.Table],
diff --git a/sqlmesh/core/engine_adapter/spark.py b/sqlmesh/core/engine_adapter/spark.py
index 24e2fac1..5e4232ec 100644
--- a/sqlmesh/core/engine_adapter/spark.py
+++ b/sqlmesh/core/engine_adapter/spark.py
@@ -224,14 +224,12 @@ class SparkEngineAdapter(GetCurrentCatalogFromFunctionMixin, HiveMetastoreTableP
     @t.overload
     def _columns_to_types(
         self, query_or_df: DF, columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None
-    ) -> t.Dict[str, exp.DataType]:
-        ...
+    ) -> t.Dict[str, exp.DataType]: ...
 
     @t.overload
     def _columns_to_types(
         self, query_or_df: Query, columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None
-    ) -> t.Optional[t.Dict[str, exp.DataType]]:
-        ...
+    ) -> t.Optional[t.Dict[str, exp.DataType]]: ...
 
     def _columns_to_types(
         self, query_or_df: QueryOrDF, columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None
@@ -328,9 +326,11 @@ class SparkEngineAdapter(GetCurrentCatalogFromFunctionMixin, HiveMetastoreTableP
                 catalog=self.get_current_catalog(),
                 schema=schema_name,
                 name=row["tableName"],
-                type=DataObjectType.VIEW
-                if "Type: VIEW" in row["information"]
-                else DataObjectType.TABLE,
+                type=(
+                    DataObjectType.VIEW
+                    if "Type: VIEW" in row["information"]
+                    else DataObjectType.TABLE
+                ),
             )
             for row in results  # type: ignore
         ]
diff --git a/sqlmesh/core/environment.py b/sqlmesh/core/environment.py
index 719e9bc3..1bdade63 100644
--- a/sqlmesh/core/environment.py
+++ b/sqlmesh/core/environment.py
@@ -38,13 +38,11 @@ class EnvironmentNamingInfo(PydanticModel):
 
     @t.overload
     @classmethod
-    def normalize_name(cls, v: str) -> str:
-        ...
+    def normalize_name(cls, v: str) -> str: ...
 
     @t.overload
     @classmethod
-    def normalize_name(cls, v: Environment) -> Environment:
-        ...
+    def normalize_name(cls, v: Environment) -> Environment: ...
 
     @classmethod
     def normalize_name(cls, v: str | Environment) -> str | Environment:
diff --git a/sqlmesh/core/metric/definition.py b/sqlmesh/core/metric/definition.py
index c48d621f..0c968672 100644
--- a/sqlmesh/core/metric/definition.py
+++ b/sqlmesh/core/metric/definition.py
@@ -31,9 +31,11 @@ def load_metric_ddl(
     metric = MetricMeta(
         **{
             "dialect": dialect,
-            "description": "\n".join(comment.strip() for comment in expression.comments)
-            if expression.comments
-            else None,
+            "description": (
+                "\n".join(comment.strip() for comment in expression.comments)
+                if expression.comments
+                else None
+            ),
             **{prop.name.lower(): prop.args.get("value") for prop in expression.expressions},
             **kwargs,
         }
@@ -156,9 +158,11 @@ class Metric(MetricMeta, frozen=True):
         """
         return {
             t.cast(exp.Expression, agg.parent).transform(
-                lambda node: exp.column(node.this, table=remove_namespace(node))
-                if isinstance(node, exp.Column) and node.table
-                else node
+                lambda node: (
+                    exp.column(node.this, table=remove_namespace(node))
+                    if isinstance(node, exp.Column) and node.table
+                    else node
+                )
             ): _get_measure_and_dim_tables(agg)
             for agg in self.expanded.find_all(exp.AggFunc)
         }
diff --git a/sqlmesh/core/model/decorator.py b/sqlmesh/core/model/decorator.py
index 7bf84d9c..f6ed5679 100644
--- a/sqlmesh/core/model/decorator.py
+++ b/sqlmesh/core/model/decorator.py
@@ -32,16 +32,18 @@ class model(registry_decorator):
         # Make sure that argument values are expressions in order to pass validation in ModelMeta.
         calls = self.kwargs.pop("audits", [])
         self.kwargs["audits"] = [
-            (call, {})
-            if isinstance(call, str)
-            else (
-                call[0],
-                {
-                    arg_key: exp.convert(
-                        tuple(arg_value) if isinstance(arg_value, list) else arg_value
-                    )
-                    for arg_key, arg_value in call[1].items()
-                },
+            (
+                (call, {})
+                if isinstance(call, str)
+                else (
+                    call[0],
+                    {
+                        arg_key: exp.convert(
+                            tuple(arg_value) if isinstance(arg_value, list) else arg_value
+                        )
+                        for arg_key, arg_value in call[1].items()
+                    },
+                )
             )
             for call in calls
         ]
@@ -50,9 +52,11 @@ class model(registry_decorator):
             raise ConfigError("`default_catalog` cannot be set on a per-model basis.")
 
         self.columns = {
-            column_name: column_type
-            if isinstance(column_type, exp.DataType)
-            else exp.DataType.build(str(column_type))
+            column_name: (
+                column_type
+                if isinstance(column_type, exp.DataType)
+                else exp.DataType.build(str(column_type))
+            )
             for column_name, column_type in self.kwargs.pop("columns", {}).items()
         }
 
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index d9fa255f..ee0d9ae8 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -1421,9 +1421,9 @@ def load_sql_based_model(
 
     meta_fields: t.Dict[str, t.Any] = {
         "dialect": dialect,
-        "description": "\n".join(comment.strip() for comment in meta.comments)
-        if meta.comments
-        else None,
+        "description": (
+            "\n".join(comment.strip() for comment in meta.comments) if meta.comments else None
+        ),
         **{prop.name.lower(): prop.args.get("value") for prop in meta.expressions},
         **kwargs,
     }
diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py
index 5e185fc9..ac2a0e90 100644
--- a/sqlmesh/core/plan/evaluator.py
+++ b/sqlmesh/core/plan/evaluator.py
@@ -13,6 +13,7 @@ At a high level, when a plan is evaluated, SQLMesh will:
 
 Refer to `sqlmesh.core.plan`.
 """
+
 import abc
 import logging
 import typing as t
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index 36420fcf..a327fdd9 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -345,7 +345,7 @@ class Scheduler:
                     if len(p_intervals) > 1:
                         upstream_dependencies.append((p_sid.name, terminal_node))
                     else:
-                        for (i, interval) in enumerate(p_intervals):
+                        for i, interval in enumerate(p_intervals):
                             upstream_dependencies.append((p_sid.name, (interval, i)))
 
             for i, interval in enumerate(intervals):
diff --git a/sqlmesh/core/selector.py b/sqlmesh/core/selector.py
index 1ddbd9a8..80c879ba 100644
--- a/sqlmesh/core/selector.py
+++ b/sqlmesh/core/selector.py
@@ -13,6 +13,7 @@ from sqlmesh.core.model import Model
 from sqlmesh.core.state_sync import StateReader
 from sqlmesh.utils import UniqueKeyDict
 from sqlmesh.utils.dag import DAG
+from sqlmesh.utils.git import GitClient
 
 logger = logging.getLogger(__name__)
 
@@ -32,6 +33,7 @@ class Selector:
         self._context_path = context_path
         self._default_catalog = default_catalog
         self._dialect = dialect
+        self._git_client = GitClient(context_path)
         self.__models_by_tag: t.Optional[t.Dict[str, t.Set[str]]] = None
 
         if dag is None:
@@ -41,15 +43,6 @@ class Selector:
         else:
             self._dag = dag
 
-    @property
-    def _models_by_tag(self) -> t.Dict[str, t.Set[str]]:
-        if self.__models_by_tag is None:
-            self.__models_by_tag = defaultdict(set)
-            for model in self._models.values():
-                for tag in model.tags:
-                    self.__models_by_tag[tag.lower()].add(model.fqn)
-        return self.__models_by_tag
-
     def select_models(
         self,
         model_selections: t.Iterable[str],
@@ -121,34 +114,101 @@ class Selector:
 
         return models
 
-    @staticmethod
-    def _get_value_and_dependency_inclusion(value: str) -> t.Tuple[str, bool, bool]:
-        include_upstream = False
-        include_downstream = False
-        if value[0] == "+":
-            value = value[1:]
-            include_upstream = True
-        if value[-1] == "+":
-            value = value[:-1]
-            include_downstream = True
-        return value, include_upstream, include_downstream
-
-    def _get_models(
-        self,
-        model_name: str,
-        include_upstream: bool,
-        include_downstream: bool,
-        models: t.Optional[t.Dict[str, Model]] = None,
+    def expand_model_selections(
+        self, model_selections: t.Iterable[str], models: t.Optional[t.Dict[str, Model]] = None
     ) -> t.Set[str]:
-        result = {model_name}
-        if include_upstream:
-            models = models or self._models
-            result.update([u for u in self._dag.upstream(model_name) if u in models])
-        if include_downstream:
-            result.update(self._dag.downstream(model_name))
-        return result
+        """Expands a set of model selections into a set of model names.
 
-    def _expand_model_tag(self, tag_selection: str) -> t.Set[str]:
+        Args:
+            model_selections: A set of model selections.
+
+        Returns:
+            A set of model names.
+        """
+        results: t.Set[str] = set()
+        models = models or self._models
+        models_by_tags: t.Optional[t.Dict[str, t.Set[str]]] = None
+
+        for selection in model_selections:
+            sub_results: t.Optional[t.Set[str]] = None
+
+            def add_sub_results(sr: t.Set[str]) -> None:
+                nonlocal sub_results
+                if sub_results is None:
+                    sub_results = sr
+                else:
+                    sub_results &= sr
+
+            sub_selections = [s.strip() for s in selection.split("&")]
+            for sub_selection in sub_selections:
+                if not sub_selection:
+                    continue
+
+                if sub_selection.startswith("tag:"):
+                    if models_by_tags is None:
+                        models_by_tag = defaultdict(set)
+                        for model in models.values():
+                            for tag in model.tags:
+                                models_by_tag[tag.lower()].add(model.fqn)
+                    add_sub_results(
+                        self._expand_model_tag(sub_selection[4:], models, models_by_tag)
+                    )
+                elif sub_selection.startswith("git:"):
+                    add_sub_results(self._expand_git(sub_selection[4:]))
+                else:
+                    add_sub_results(self._expand_model_name(sub_selection, models))
+
+            if sub_results:
+                results.update(sub_results)
+            else:
+                logger.warning(f"Expression '{selection}' doesn't match any models.")
+
+        return results
+
+    def _expand_git(self, target_branch: str) -> t.Set[str]:
+        git_modified_files = {
+            *self._git_client.list_untracked_files(),
+            *self._git_client.list_changed_files(target_branch=target_branch),
+        }
+        matched_models = {m.fqn for m in self._models.values() if m._path in git_modified_files}
+
+        if not matched_models:
+            logger.warning(f"Expression 'git:{target_branch}' doesn't match any models.")
+
+        return matched_models
+
+    def _expand_model_name(self, selection: str, models: t.Dict[str, Model]) -> t.Set[str]:
+        results = set()
+
+        (
+            selection,
+            include_upstream,
+            include_downstream,
+        ) = self._get_value_and_dependency_inclusion(selection.lower())
+
+        matched_models = set()
+
+        if "*" in selection:
+            for model in models.values():
+                if fnmatch.fnmatchcase(model.name, selection):
+                    matched_models.add(model.fqn)
+        else:
+            model_fqn = normalize_model_name(selection, self._default_catalog, self._dialect)
+            if model_fqn in models:
+                matched_models.add(model_fqn)
+
+        if not matched_models:
+            logger.warning(f"Expression '{selection}' doesn't match any models.")
+
+        for model_fqn in matched_models:
+            results.update(
+                self._get_models(model_fqn, include_upstream, include_downstream, models)
+            )
+        return results
+
+    def _expand_model_tag(
+        self, tag_selection: str, models: t.Dict[str, Model], models_by_tag: t.Dict[str, t.Set[str]]
+    ) -> t.Set[str]:
         """
         Expands a set of model tags into a set of model names.
         The tag matching is case-insensitive and supports wildcards and + prefix and suffix to
@@ -169,66 +229,43 @@ class Selector:
         ) = self._get_value_and_dependency_inclusion(tag_selection.lower())
 
         if "*" in selection:
-            for model_tag in self._models_by_tag:
+            for model_tag in models_by_tag:
                 if fnmatch.fnmatchcase(model_tag, selection):
                     matched_tags.add(model_tag)
-        elif selection in self._models_by_tag:
+        elif selection in models_by_tag:
             matched_tags.add(selection)
 
         if not matched_tags:
             logger.warning(f"Expression 'tag:{tag_selection}' doesn't match any models.")
 
         for tag in matched_tags:
-            for model in self._models_by_tag[tag]:
-                result.update(self._get_models(model, include_upstream, include_downstream))
+            for model in models_by_tag[tag]:
+                result.update(self._get_models(model, include_upstream, include_downstream, models))
 
         return result
 
-    def expand_model_selections(
-        self, model_selections: t.Iterable[str], models: t.Optional[t.Dict[str, Model]] = None
+    def _get_models(
+        self,
+        model_name: str,
+        include_upstream: bool,
+        include_downstream: bool,
+        models: t.Dict[str, Model],
     ) -> t.Set[str]:
-        """Expands a set of model selections into a set of model names.
-
-        Args:
-            model_selections: A set of model selections.
-
-        Returns:
-            A set of model names.
-        """
-        results: t.Set[str] = set()
-        models = models or self._models
-
-        for selection in model_selections:
-            if not selection:
-                continue
-
-            if selection.startswith("tag:"):
-                results.update(self._expand_model_tag(selection[4:]))
-                continue
-
-            (
-                selection,
-                include_upstream,
-                include_downstream,
-            ) = self._get_value_and_dependency_inclusion(selection.lower())
-
-            matched_models = set()
-
-            if "*" in selection:
-                for model in models.values():
-                    if fnmatch.fnmatchcase(model.name, selection):
-                        matched_models.add(model.fqn)
-            else:
-                model_fqn = normalize_model_name(selection, self._default_catalog, self._dialect)
-                if model_fqn in models:
-                    matched_models.add(model_fqn)
-
-            if not matched_models:
-                logger.warning(f"Expression '{selection}' doesn't match any models.")
-
-            for model_fqn in matched_models:
-                results.update(
-                    self._get_models(model_fqn, include_upstream, include_downstream, models=models)
-                )
+        result = {model_name}
+        if include_upstream:
+            result.update([u for u in self._dag.upstream(model_name) if u in models])
+        if include_downstream:
+            result.update(self._dag.downstream(model_name))
+        return result
 
-        return results
+    @staticmethod
+    def _get_value_and_dependency_inclusion(value: str) -> t.Tuple[str, bool, bool]:
+        include_upstream = False
+        include_downstream = False
+        if value[0] == "+":
+            value = value[1:]
+            include_upstream = True
+        if value[-1] == "+":
+            value = value[:-1]
+            include_downstream = True
+        return value, include_upstream, include_downstream
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 83c00254..dac7290b 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1072,9 +1072,11 @@ class DeployabilityIndex(PydanticModel, frozen=True):
         # Transforming into strings because the serialization of sets of objects / lists is broken in Pydantic.
         return frozenset(
             {
-                cls._snapshot_id_key(snapshot_id)
-                if isinstance(snapshot_id, SnapshotId)
-                else snapshot_id
+                (
+                    cls._snapshot_id_key(snapshot_id)
+                    if isinstance(snapshot_id, SnapshotId)
+                    else snapshot_id
+                )
                 for snapshot_id in v
             }
         )
@@ -1241,9 +1243,11 @@ def display_name(
         return snapshot_info_like.name
     view_name = exp.to_table(snapshot_info_like.name)
     qvn = QualifiedViewName(
-        catalog=view_name.catalog
-        if view_name.catalog and view_name.catalog != default_catalog
-        else None,
+        catalog=(
+            view_name.catalog
+            if view_name.catalog and view_name.catalog != default_catalog
+            else None
+        ),
         schema_name=view_name.db or None,
         table=view_name.name,
     )
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 47dd87bb..3fa00c8a 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -19,6 +19,7 @@ has been evaluated to check for data quality issues.
 
 For more information about audits, see `sqlmesh.core.audit`.
 """
+
 from __future__ import annotations
 
 import abc
@@ -545,9 +546,11 @@ class SnapshotEvaluator:
                 and snapshot.is_incremental_by_time_range
             ):
                 query_or_df = reduce(
-                    lambda a, b: pd.concat([a, b], ignore_index=True)  # type: ignore
-                    if self.adapter.is_pandas_df(a)
-                    else a.union_all(b),  # type: ignore
+                    lambda a, b: (
+                        pd.concat([a, b], ignore_index=True)  # type: ignore
+                        if self.adapter.is_pandas_df(a)
+                        else a.union_all(b)  # type: ignore
+                    ),  # type: ignore
                     queries_or_dfs,
                 )
                 apply(query_or_df, index=0)
diff --git a/sqlmesh/core/state_sync/__init__.py b/sqlmesh/core/state_sync/__init__.py
index 5c64f4f5..8662a1dd 100644
--- a/sqlmesh/core/state_sync/__init__.py
+++ b/sqlmesh/core/state_sync/__init__.py
@@ -13,6 +13,7 @@ implies, it only allows for read-only operations on snapshots and environment st
 The provided `sqlmesh.core.state_sync.EngineAdapterStateSync` leverages an existing engine
 adapter to read and write state to the underlying data store.
 """
+
 from sqlmesh.core.state_sync.base import StateReader, StateSync, Versions
 from sqlmesh.core.state_sync.cache import CachingStateSync
 from sqlmesh.core.state_sync.common import CommonStateSyncMixin, cleanup_expired_views
diff --git a/sqlmesh/core/state_sync/cache.py b/sqlmesh/core/state_sync/cache.py
index 5338580d..ef92dc98 100644
--- a/sqlmesh/core/state_sync/cache.py
+++ b/sqlmesh/core/state_sync/cache.py
@@ -29,7 +29,7 @@ class CachingStateSync(DelegatingStateSync):
         ttl: The number of seconds a snapshot should be cached.
     """
 
-    def __init__(self, state_sync: StateSync, ttl: int = 30):
+    def __init__(self, state_sync: StateSync, ttl: int = 120):
         super().__init__(state_sync)
         # The cache can contain a snapshot or False or None.
         # False means that the snapshot does not exist in the state sync but has been requested before
diff --git a/sqlmesh/core/state_sync/common.py b/sqlmesh/core/state_sync/common.py
index d2419e03..a440561b 100644
--- a/sqlmesh/core/state_sync/common.py
+++ b/sqlmesh/core/state_sync/common.py
@@ -182,9 +182,9 @@ class CommonStateSyncMixin(StateSync):
         return PromotionResult(
             added=sorted(added_table_infos),
             removed=list(removed),
-            removed_environment_naming_info=existing_environment.naming_info
-            if removed and existing_environment
-            else None,
+            removed_environment_naming_info=(
+                existing_environment.naming_info if removed and existing_environment else None
+            ),
         )
 
     @transactional()
diff --git a/sqlmesh/core/state_sync/engine_adapter.py b/sqlmesh/core/state_sync/engine_adapter.py
index 0f6a36b0..8d14c26c 100644
--- a/sqlmesh/core/state_sync/engine_adapter.py
+++ b/sqlmesh/core/state_sync/engine_adapter.py
@@ -13,6 +13,7 @@ implies, it only allows for read-only operations on snapshots and environment st
 The provided `sqlmesh.core.state_sync.EngineAdapterStateSync` leverages an existing engine
 adapter to read and write state to the underlying data store.
 """
+
 from __future__ import annotations
 
 import contextlib
@@ -275,7 +276,9 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             SnapshotId(name=name, identifier=identifier): SnapshotNameVersion(
                 name=name, version=version
             )
-            for name, identifier, version in self.engine_adapter.fetchall(expired_query)
+            for name, identifier, version in self.engine_adapter.fetchall(
+                expired_query, quote_identifiers=True
+            )
         }
         if not expired_candidates:
             return []
@@ -1080,9 +1083,11 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         updated_environments = []
         for environment in environments:
             snapshots = [
-                new_snapshots[snapshot_id_mapping[info.snapshot_id]].table_info
-                if info.snapshot_id in snapshot_id_mapping
-                else info
+                (
+                    new_snapshots[snapshot_id_mapping[info.snapshot_id]].table_info
+                    if info.snapshot_id in snapshot_id_mapping
+                    else info
+                )
                 for info in environment.snapshots
             ]
 
@@ -1244,11 +1249,11 @@ def _environment_to_df(environment: Environment) -> pd.DataFrame:
                 "previous_plan_id": environment.previous_plan_id,
                 "expiration_ts": environment.expiration_ts,
                 "finalized_ts": environment.finalized_ts,
-                "promoted_snapshot_ids": json.dumps(
-                    [s.dict() for s in environment.promoted_snapshot_ids]
-                )
-                if environment.promoted_snapshot_ids is not None
-                else None,
+                "promoted_snapshot_ids": (
+                    json.dumps([s.dict() for s in environment.promoted_snapshot_ids])
+                    if environment.promoted_snapshot_ids is not None
+                    else None
+                ),
                 "suffix_target": environment.suffix_target.value,
                 "catalog_name_override": environment.catalog_name_override,
             }
diff --git a/sqlmesh/core/table_diff.py b/sqlmesh/core/table_diff.py
index bda05056..092b1ebf 100644
--- a/sqlmesh/core/table_diff.py
+++ b/sqlmesh/core/table_diff.py
@@ -299,19 +299,23 @@ class TableDiff:
                 }
                 if self.source != self.source_alias and self.target != self.target_alias:
                     joined_renamed_cols = {
-                        c: n.replace(
-                            "s__", f"{self.source_alias.upper() if self.source_alias else ''}__"
+                        c: (
+                            n.replace(
+                                "s__", f"{self.source_alias.upper() if self.source_alias else ''}__"
+                            )
+                            if n.startswith("s__")
+                            else n
                         )
-                        if n.startswith("s__")
-                        else n
                         for c, n in joined_renamed_cols.items()
                     }
                     joined_renamed_cols = {
-                        c: n.replace(
-                            "t__", f"{self.target_alias.upper() if self.target_alias else ''}__"
+                        c: (
+                            n.replace(
+                                "t__", f"{self.target_alias.upper() if self.target_alias else ''}__"
+                            )
+                            if n.startswith("t__")
+                            else n
                         )
-                        if n.startswith("t__")
-                        else n
                         for c, n in joined_renamed_cols.items()
                     }
                 joined_sample = sample[sample["rows_joined"] == 1][joined_sample_cols]
diff --git a/sqlmesh/dbt/adapter.py b/sqlmesh/dbt/adapter.py
index 916a1d0a..1e070388 100644
--- a/sqlmesh/dbt/adapter.py
+++ b/sqlmesh/dbt/adapter.py
@@ -233,9 +233,11 @@ class RuntimeAdapter(BaseAdapter):
                 identifier=do.name,
                 quote_policy=self.quote_policy,
                 # DBT relation types aren't snake case and instead just one word without spaces so we remove underscores
-                type=RelationType.External
-                if do.type.is_unknown
-                else RelationType(do.type.lower().replace("_", "")),
+                type=(
+                    RelationType.External
+                    if do.type.is_unknown
+                    else RelationType(do.type.lower().replace("_", ""))
+                ),
             )
             for do in self.engine_adapter.get_data_objects(schema)
         ]
diff --git a/sqlmesh/dbt/model.py b/sqlmesh/dbt/model.py
index 1e1bec2a..4b95e857 100644
--- a/sqlmesh/dbt/model.py
+++ b/sqlmesh/dbt/model.py
@@ -232,9 +232,11 @@ class ModelConfig(BaseModelConfig):
                 valid_to_name="dbt_valid_to",
                 updated_at_name=self.updated_at,
                 updated_at_as_valid_from=True,
-                time_data_type=exp.DataType.build("TIMESTAMPTZ")
-                if target.type == "bigquery"
-                else exp.DataType.build("TIMESTAMP"),
+                time_data_type=(
+                    exp.DataType.build("TIMESTAMPTZ")
+                    if target.type == "bigquery"
+                    else exp.DataType.build("TIMESTAMP")
+                ),
             )
         raise ConfigError(f"{materialization.value} materialization not supported.")
 
diff --git a/sqlmesh/magics.py b/sqlmesh/magics.py
index 253511e3..4995e2ee 100644
--- a/sqlmesh/magics.py
+++ b/sqlmesh/magics.py
@@ -618,26 +618,26 @@ class SQLMeshMagics(Magics):
         help="Transpile project models to the specified dialect.",
     )
     @argument(
-        "--newline",
+        "--append-newline",
         action="store_true",
-        help="The output dialect of the sql string.",
+        help="Whether or not to append a newline to the end of the file.",
+        default=None,
     )
     @argument(
         "--normalize",
         action="store_true",
         help="Whether or not to normalize identifiers to lowercase.",
+        default=None,
     )
     @argument(
         "--pad",
         type=int,
         help="Determines the pad size in a formatted string.",
-        default=2,
     )
     @argument(
         "--indent",
         type=int,
         help="Determines the indentation size in a formatted string.",
-        default=2,
     )
     @argument(
         "--normalize-functions",
@@ -648,19 +648,19 @@ class SQLMeshMagics(Magics):
         "--leading-comma",
         action="store_true",
         help="Determines whether or not the comma is leading or trailing in select expressions. Default is trailing.",
+        default=None,
     )
     @argument(
         "--max-text-width",
         type=int,
         help="The max number of characters in a segment before creating new lines in pretty mode.",
-        default=80,
     )
     @line_magic
     @pass_sqlmesh_context
     def format(self, context: Context, line: str) -> None:
         """Format all SQL models."""
         args = parse_argstring(self.format, line)
-        context.format(**vars(args))
+        context.format(**{k: v for k, v in vars(args).items() if v is not None})
 
     @magic_arguments()
     @argument("environment", type=str, help="The environment to diff local state against.")
diff --git a/sqlmesh/migrations/v0001_init.py b/sqlmesh/migrations/v0001_init.py
index f25a4384..778c36bc 100644
--- a/sqlmesh/migrations/v0001_init.py
+++ b/sqlmesh/migrations/v0001_init.py
@@ -3,6 +3,7 @@
 If a migration alters the payload of any pydantic models, you should not actually use them because
 the running model may not be able to load them. Make sure that these migration files are standalone.
 """
+
 from sqlglot import exp
 
 from sqlmesh.utils.migration import index_text_type
diff --git a/sqlmesh/migrations/v0003_move_batch_size.py b/sqlmesh/migrations/v0003_move_batch_size.py
index 81fdc9aa..9d6d7ce4 100644
--- a/sqlmesh/migrations/v0003_move_batch_size.py
+++ b/sqlmesh/migrations/v0003_move_batch_size.py
@@ -1,4 +1,5 @@
 """Move batch_size from the model and into the kind."""
+
 import json
 
 from sqlglot import exp
diff --git a/sqlmesh/migrations/v0004_environmnent_add_finalized_at.py b/sqlmesh/migrations/v0004_environmnent_add_finalized_at.py
index 0b74c65c..ad228abb 100644
--- a/sqlmesh/migrations/v0004_environmnent_add_finalized_at.py
+++ b/sqlmesh/migrations/v0004_environmnent_add_finalized_at.py
@@ -1,4 +1,5 @@
 """Add support for environment finalization."""
+
 from sqlglot import exp
 
 
diff --git a/sqlmesh/migrations/v0005_create_seed_table.py b/sqlmesh/migrations/v0005_create_seed_table.py
index a57b1678..1e1e7dc3 100644
--- a/sqlmesh/migrations/v0005_create_seed_table.py
+++ b/sqlmesh/migrations/v0005_create_seed_table.py
@@ -1,4 +1,5 @@
 """Create a dedicated table to store the content of seeds."""
+
 from sqlglot import exp
 
 from sqlmesh.utils.migration import index_text_type
diff --git a/sqlmesh/migrations/v0007_env_table_info_to_kind.py b/sqlmesh/migrations/v0007_env_table_info_to_kind.py
index 107a61aa..6e0ac94a 100644
--- a/sqlmesh/migrations/v0007_env_table_info_to_kind.py
+++ b/sqlmesh/migrations/v0007_env_table_info_to_kind.py
@@ -1,4 +1,5 @@
 """Change environments because snapshot table info now stores model kind name."""
+
 import json
 import zlib
 
diff --git a/sqlmesh/migrations/v0008_create_intervals_table.py b/sqlmesh/migrations/v0008_create_intervals_table.py
index 44150242..0746febc 100644
--- a/sqlmesh/migrations/v0008_create_intervals_table.py
+++ b/sqlmesh/migrations/v0008_create_intervals_table.py
@@ -1,4 +1,5 @@
 """Create a dedicated table to store snapshot intervals."""
+
 from sqlglot import exp
 
 from sqlmesh.utils.migration import index_text_type
diff --git a/sqlmesh/migrations/v0009_remove_pre_post_hooks.py b/sqlmesh/migrations/v0009_remove_pre_post_hooks.py
index 5ea4c4bc..79aab446 100644
--- a/sqlmesh/migrations/v0009_remove_pre_post_hooks.py
+++ b/sqlmesh/migrations/v0009_remove_pre_post_hooks.py
@@ -1,4 +1,5 @@
 """Remove pre- / post- hooks from existing snapshots."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0011_add_model_kind_name.py b/sqlmesh/migrations/v0011_add_model_kind_name.py
index 64b33aef..d7fa976e 100644
--- a/sqlmesh/migrations/v0011_add_model_kind_name.py
+++ b/sqlmesh/migrations/v0011_add_model_kind_name.py
@@ -1,4 +1,5 @@
 """Add the kind_name column to the snapshots table."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0012_update_jinja_expressions.py b/sqlmesh/migrations/v0012_update_jinja_expressions.py
index 56babdd8..14e04d6a 100644
--- a/sqlmesh/migrations/v0012_update_jinja_expressions.py
+++ b/sqlmesh/migrations/v0012_update_jinja_expressions.py
@@ -1,4 +1,5 @@
 """Fix expressions that contain jinja."""
+
 import json
 import typing as t
 
diff --git a/sqlmesh/migrations/v0013_serde_using_model_dialects.py b/sqlmesh/migrations/v0013_serde_using_model_dialects.py
index 42f4dbf9..26f9c1e7 100644
--- a/sqlmesh/migrations/v0013_serde_using_model_dialects.py
+++ b/sqlmesh/migrations/v0013_serde_using_model_dialects.py
@@ -1,4 +1,5 @@
 """Serialize SQL using the dialect of each model."""
+
 import json
 import typing as t
 
diff --git a/sqlmesh/migrations/v0015_environment_add_promoted_snapshot_ids.py b/sqlmesh/migrations/v0015_environment_add_promoted_snapshot_ids.py
index a41e8a90..d8d963cb 100644
--- a/sqlmesh/migrations/v0015_environment_add_promoted_snapshot_ids.py
+++ b/sqlmesh/migrations/v0015_environment_add_promoted_snapshot_ids.py
@@ -1,4 +1,5 @@
 """Include a set of snapshot IDs filtered for promotion."""
+
 from sqlglot import exp
 
 
diff --git a/sqlmesh/migrations/v0016_fix_windows_path.py b/sqlmesh/migrations/v0016_fix_windows_path.py
index 50a93520..01e06e1f 100644
--- a/sqlmesh/migrations/v0016_fix_windows_path.py
+++ b/sqlmesh/migrations/v0016_fix_windows_path.py
@@ -1,4 +1,5 @@
 """Fix paths that have a Windows forward slash in them."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0017_fix_windows_seed_path.py b/sqlmesh/migrations/v0017_fix_windows_seed_path.py
index 011390e1..44675aa6 100644
--- a/sqlmesh/migrations/v0017_fix_windows_seed_path.py
+++ b/sqlmesh/migrations/v0017_fix_windows_seed_path.py
@@ -1,4 +1,5 @@
 """Fix seed paths that have a Windows forward slash in them."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0018_rename_snapshot_model_to_node.py b/sqlmesh/migrations/v0018_rename_snapshot_model_to_node.py
index b5ec32e9..c3119841 100644
--- a/sqlmesh/migrations/v0018_rename_snapshot_model_to_node.py
+++ b/sqlmesh/migrations/v0018_rename_snapshot_model_to_node.py
@@ -1,4 +1,5 @@
 """Replace snapshot model field with node."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0019_add_env_suffix_target.py b/sqlmesh/migrations/v0019_add_env_suffix_target.py
index d3008a8d..b3007b45 100644
--- a/sqlmesh/migrations/v0019_add_env_suffix_target.py
+++ b/sqlmesh/migrations/v0019_add_env_suffix_target.py
@@ -1,4 +1,5 @@
 """Add support for environment suffix target."""
+
 from sqlglot import exp
 
 
diff --git a/sqlmesh/migrations/v0020_remove_redundant_attributes_from_dbt_models.py b/sqlmesh/migrations/v0020_remove_redundant_attributes_from_dbt_models.py
index b99eaa26..938e7d42 100644
--- a/sqlmesh/migrations/v0020_remove_redundant_attributes_from_dbt_models.py
+++ b/sqlmesh/migrations/v0020_remove_redundant_attributes_from_dbt_models.py
@@ -1,4 +1,5 @@
 """Remove redundant attributes from dbt models."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0021_fix_table_properties.py b/sqlmesh/migrations/v0021_fix_table_properties.py
index 068d06cb..a7d6b8b2 100644
--- a/sqlmesh/migrations/v0021_fix_table_properties.py
+++ b/sqlmesh/migrations/v0021_fix_table_properties.py
@@ -1,4 +1,5 @@
 """Fix table properties that have extra quoting due to a bug."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0022_move_project_to_model.py b/sqlmesh/migrations/v0022_move_project_to_model.py
index 59f5c510..b50d870a 100644
--- a/sqlmesh/migrations/v0022_move_project_to_model.py
+++ b/sqlmesh/migrations/v0022_move_project_to_model.py
@@ -1,4 +1,5 @@
 """Move project attr from snapshot to model."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0023_fix_added_models_with_forward_only_parents.py b/sqlmesh/migrations/v0023_fix_added_models_with_forward_only_parents.py
index 03dbe4f1..6ae64955 100644
--- a/sqlmesh/migrations/v0023_fix_added_models_with_forward_only_parents.py
+++ b/sqlmesh/migrations/v0023_fix_added_models_with_forward_only_parents.py
@@ -1,4 +1,5 @@
 """Fix snapshots of added models with forward only parents."""
+
 import json
 import typing as t
 
diff --git a/sqlmesh/migrations/v0024_replace_model_kind_name_enum_with_value.py b/sqlmesh/migrations/v0024_replace_model_kind_name_enum_with_value.py
index 00d7f880..81864bbf 100644
--- a/sqlmesh/migrations/v0024_replace_model_kind_name_enum_with_value.py
+++ b/sqlmesh/migrations/v0024_replace_model_kind_name_enum_with_value.py
@@ -1,4 +1,5 @@
 """Replace snapshot model_kind_name enum with value."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0025_fix_intervals_and_missing_change_category.py b/sqlmesh/migrations/v0025_fix_intervals_and_missing_change_category.py
index 9adc5fbf..694cbd10 100644
--- a/sqlmesh/migrations/v0025_fix_intervals_and_missing_change_category.py
+++ b/sqlmesh/migrations/v0025_fix_intervals_and_missing_change_category.py
@@ -1,4 +1,5 @@
 """Normalize intervals and fix missing change category."""
+
 import json
 import zlib
 
diff --git a/sqlmesh/migrations/v0026_remove_dialect_from_seed.py b/sqlmesh/migrations/v0026_remove_dialect_from_seed.py
index aba687ca..957cde9a 100644
--- a/sqlmesh/migrations/v0026_remove_dialect_from_seed.py
+++ b/sqlmesh/migrations/v0026_remove_dialect_from_seed.py
@@ -1,4 +1,5 @@
 """Remove dialect from seeds."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0027_minute_interval_to_five.py b/sqlmesh/migrations/v0027_minute_interval_to_five.py
index 36f5383f..ce59066a 100644
--- a/sqlmesh/migrations/v0027_minute_interval_to_five.py
+++ b/sqlmesh/migrations/v0027_minute_interval_to_five.py
@@ -1,4 +1,5 @@
 """Change any interval unit of minute to five_minute."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0028_add_plan_dags_table.py b/sqlmesh/migrations/v0028_add_plan_dags_table.py
index 46372429..d8e67f60 100644
--- a/sqlmesh/migrations/v0028_add_plan_dags_table.py
+++ b/sqlmesh/migrations/v0028_add_plan_dags_table.py
@@ -1,4 +1,5 @@
 """Creates the '_plan_dags' table if Airflow is used."""
+
 from sqlglot import exp
 
 from sqlmesh.utils.migration import index_text_type
diff --git a/sqlmesh/migrations/v0029_generate_schema_types_using_dialect.py b/sqlmesh/migrations/v0029_generate_schema_types_using_dialect.py
index 7be9eac0..39fd527f 100644
--- a/sqlmesh/migrations/v0029_generate_schema_types_using_dialect.py
+++ b/sqlmesh/migrations/v0029_generate_schema_types_using_dialect.py
@@ -1,4 +1,5 @@
 """Generate mapping schema data types using the corresponding model's dialect."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0030_update_unrestorable_snapshots.py b/sqlmesh/migrations/v0030_update_unrestorable_snapshots.py
index 066b0091..ca5ca247 100644
--- a/sqlmesh/migrations/v0030_update_unrestorable_snapshots.py
+++ b/sqlmesh/migrations/v0030_update_unrestorable_snapshots.py
@@ -1,4 +1,5 @@
 """Update unrestorable snapshots."""
+
 import json
 import typing as t
 from collections import defaultdict
diff --git a/sqlmesh/migrations/v0031_remove_dbt_target_fields.py b/sqlmesh/migrations/v0031_remove_dbt_target_fields.py
index 90d0ab77..8ac6c949 100644
--- a/sqlmesh/migrations/v0031_remove_dbt_target_fields.py
+++ b/sqlmesh/migrations/v0031_remove_dbt_target_fields.py
@@ -1,4 +1,5 @@
 """Remove dbt target fields from snapshots outside of limited list of approved fields"""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0032_add_sqlmesh_version.py b/sqlmesh/migrations/v0032_add_sqlmesh_version.py
index 58588ed4..0b17f4a0 100644
--- a/sqlmesh/migrations/v0032_add_sqlmesh_version.py
+++ b/sqlmesh/migrations/v0032_add_sqlmesh_version.py
@@ -1,4 +1,5 @@
 """Add new 'sqlmesh_version' column to the version state table."""
+
 from sqlglot import exp
 
 
diff --git a/sqlmesh/migrations/v0033_mysql_fix_blob_text_type.py b/sqlmesh/migrations/v0033_mysql_fix_blob_text_type.py
index 574ecf60..5660240d 100644
--- a/sqlmesh/migrations/v0033_mysql_fix_blob_text_type.py
+++ b/sqlmesh/migrations/v0033_mysql_fix_blob_text_type.py
@@ -1,4 +1,5 @@
 """Use LONGTEXT type for blob fields in MySQL."""
+
 from sqlglot import exp
 
 
diff --git a/sqlmesh/migrations/v0034_add_default_catalog.py b/sqlmesh/migrations/v0034_add_default_catalog.py
index 32c51269..40be1b59 100644
--- a/sqlmesh/migrations/v0034_add_default_catalog.py
+++ b/sqlmesh/migrations/v0034_add_default_catalog.py
@@ -41,9 +41,9 @@ def normalize_mapping_schema(mapping_schema: t.Dict, dialect: str) -> t.Dict:
     normalized_mapping_schema = {}
     for key, value in mapping_schema.items():
         if isinstance(value, dict):
-            normalized_mapping_schema[
-                normalize_model_name(key, None, dialect)
-            ] = normalize_mapping_schema(value, dialect)
+            normalized_mapping_schema[normalize_model_name(key, None, dialect)] = (
+                normalize_mapping_schema(value, dialect)
+            )
         else:
             normalized_mapping_schema[key] = value
     return normalized_mapping_schema
@@ -331,7 +331,11 @@ def migrate(state_sync, default_catalog: t.Optional[str], **kwargs):  # type: ig
         )
 
     new_seeds = []
-    for (name, identifier, content,) in engine_adapter.fetchall(
+    for (
+        name,
+        identifier,
+        content,
+    ) in engine_adapter.fetchall(
         exp.select(
             "name",
             "identifier",
diff --git a/sqlmesh/migrations/v0035_add_catalog_name_override.py b/sqlmesh/migrations/v0035_add_catalog_name_override.py
index e6725fbb..34cfc8f3 100644
--- a/sqlmesh/migrations/v0035_add_catalog_name_override.py
+++ b/sqlmesh/migrations/v0035_add_catalog_name_override.py
@@ -1,4 +1,5 @@
 """Add support for environment catalog name override."""
+
 from sqlglot import exp
 
 
diff --git a/sqlmesh/migrations/v0037_remove_dbt_is_incremental_macro.py b/sqlmesh/migrations/v0037_remove_dbt_is_incremental_macro.py
index 18489fce..e65bdf22 100644
--- a/sqlmesh/migrations/v0037_remove_dbt_is_incremental_macro.py
+++ b/sqlmesh/migrations/v0037_remove_dbt_is_incremental_macro.py
@@ -1,4 +1,5 @@
 """Remove dbt is_incremental macro"""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0038_add_expiration_ts_to_snapshot.py b/sqlmesh/migrations/v0038_add_expiration_ts_to_snapshot.py
index 636eab99..047bca59 100644
--- a/sqlmesh/migrations/v0038_add_expiration_ts_to_snapshot.py
+++ b/sqlmesh/migrations/v0038_add_expiration_ts_to_snapshot.py
@@ -1,4 +1,5 @@
 """Add the expiration_ts column to the snapshots table."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/migrations/v0039_include_environment_in_plan_dag_spec.py b/sqlmesh/migrations/v0039_include_environment_in_plan_dag_spec.py
index a1c3c69a..db8f6cc0 100644
--- a/sqlmesh/migrations/v0039_include_environment_in_plan_dag_spec.py
+++ b/sqlmesh/migrations/v0039_include_environment_in_plan_dag_spec.py
@@ -1,4 +1,5 @@
 """Include environment in plan dag spec."""
+
 import json
 
 import pandas as pd
diff --git a/sqlmesh/schedulers/airflow/dag_generator.py b/sqlmesh/schedulers/airflow/dag_generator.py
index 4fe7ccc8..e9b4e393 100644
--- a/sqlmesh/schedulers/airflow/dag_generator.py
+++ b/sqlmesh/schedulers/airflow/dag_generator.py
@@ -316,9 +316,9 @@ class SnapshotDagGenerator:
             python_callable=promotion_update_state_task,
             op_kwargs={
                 "environment": environment,
-                "no_gaps_snapshot_names": request.no_gaps_snapshot_names
-                if request.no_gaps
-                else set(),
+                "no_gaps_snapshot_names": (
+                    request.no_gaps_snapshot_names if request.no_gaps else set()
+                ),
             },
         )
 
diff --git a/sqlmesh/schedulers/airflow/operators/spark_submit.py b/sqlmesh/schedulers/airflow/operators/spark_submit.py
index 7a0664fd..c8e54031 100644
--- a/sqlmesh/schedulers/airflow/operators/spark_submit.py
+++ b/sqlmesh/schedulers/airflow/operators/spark_submit.py
@@ -139,9 +139,9 @@ class SQLMeshSparkSubmitOperator(BaseOperator):
             "default_catalog": self._target.default_catalog,
             "command_type": command_type.value if command_type else None,
             "ddl_concurrent_tasks": ddl_concurrent_tasks,
-            "payload_path": command_payload_file_path.split("/")[-1]
-            if command_payload_file_path
-            else None,
+            "payload_path": (
+                command_payload_file_path.split("/")[-1] if command_payload_file_path else None
+            ),
         }
         return SparkSubmitHook(
             conf=spark_conf,
diff --git a/sqlmesh/schedulers/airflow/operators/targets.py b/sqlmesh/schedulers/airflow/operators/targets.py
index 2f25c743..031d5b1a 100644
--- a/sqlmesh/schedulers/airflow/operators/targets.py
+++ b/sqlmesh/schedulers/airflow/operators/targets.py
@@ -119,9 +119,9 @@ class SnapshotEvaluationTarget(BaseTarget[commands.EvaluateCommandPayload], Pyda
     """
 
     command_type: commands.CommandType = commands.CommandType.EVALUATE
-    command_handler: t.Callable[
-        [SnapshotEvaluator, commands.EvaluateCommandPayload], None
-    ] = commands.evaluate
+    command_handler: t.Callable[[SnapshotEvaluator, commands.EvaluateCommandPayload], None] = (
+        commands.evaluate
+    )
     ddl_concurrent_tasks: int = 1
 
     snapshot: Snapshot
@@ -196,9 +196,9 @@ class SnapshotPromotionTarget(BaseTarget[commands.PromoteCommandPayload], Pydant
     """
 
     command_type: commands.CommandType = commands.CommandType.PROMOTE
-    command_handler: t.Callable[
-        [SnapshotEvaluator, commands.PromoteCommandPayload], None
-    ] = commands.promote
+    command_handler: t.Callable[[SnapshotEvaluator, commands.PromoteCommandPayload], None] = (
+        commands.promote
+    )
 
     snapshots: t.List[Snapshot]
     environment_naming_info: EnvironmentNamingInfo
@@ -224,9 +224,9 @@ class SnapshotDemotionTarget(BaseTarget[commands.DemoteCommandPayload], Pydantic
     """
 
     command_type: commands.CommandType = commands.CommandType.DEMOTE
-    command_handler: t.Callable[
-        [SnapshotEvaluator, commands.DemoteCommandPayload], None
-    ] = commands.demote
+    command_handler: t.Callable[[SnapshotEvaluator, commands.DemoteCommandPayload], None] = (
+        commands.demote
+    )
 
     snapshots: t.List[SnapshotTableInfo]
     environment_naming_info: EnvironmentNamingInfo
@@ -243,9 +243,9 @@ class SnapshotCleanupTarget(BaseTarget[commands.CleanupCommandPayload], Pydantic
     """The target which contains attributes necessary to perform table cleanup of expired snapshots"""
 
     command_type: commands.CommandType = commands.CommandType.CLEANUP
-    command_handler: t.Callable[
-        [SnapshotEvaluator, commands.CleanupCommandPayload], None
-    ] = commands.cleanup
+    command_handler: t.Callable[[SnapshotEvaluator, commands.CleanupCommandPayload], None] = (
+        commands.cleanup
+    )
     ddl_concurrent_tasks: int = 1
 
     @provide_session
@@ -275,9 +275,9 @@ class SnapshotCreateTablesTarget(BaseTarget[commands.CreateTablesCommandPayload]
     """The target which creates physical tables for the given set of new snapshots."""
 
     command_type: commands.CommandType = commands.CommandType.CREATE_TABLES
-    command_handler: t.Callable[
-        [SnapshotEvaluator, commands.CreateTablesCommandPayload], None
-    ] = commands.create_tables
+    command_handler: t.Callable[[SnapshotEvaluator, commands.CreateTablesCommandPayload], None] = (
+        commands.create_tables
+    )
 
     new_snapshots: t.List[Snapshot]
     ddl_concurrent_tasks: int
@@ -302,9 +302,9 @@ class SnapshotMigrateTablesTarget(BaseTarget[commands.MigrateTablesCommandPayloa
     """
 
     command_type: commands.CommandType = commands.CommandType.MIGRATE_TABLES
-    command_handler: t.Callable[
-        [SnapshotEvaluator, commands.MigrateTablesCommandPayload], None
-    ] = commands.migrate_tables
+    command_handler: t.Callable[[SnapshotEvaluator, commands.MigrateTablesCommandPayload], None] = (
+        commands.migrate_tables
+    )
 
     snapshots: t.List[Snapshot]
     ddl_concurrent_tasks: int
diff --git a/sqlmesh/schedulers/airflow/plan.py b/sqlmesh/schedulers/airflow/plan.py
index 5d5a01d1..bea326e5 100644
--- a/sqlmesh/schedulers/airflow/plan.py
+++ b/sqlmesh/schedulers/airflow/plan.py
@@ -147,9 +147,7 @@ def create_plan_dag_spec(
             if deployability_index_for_creation.is_representative(s)
         }
         if request.no_gaps and not request.is_dev
-        else None
-        if request.no_gaps
-        else set()
+        else None if request.no_gaps else set()
     )
 
     return common.PlanDagSpec(
diff --git a/sqlmesh/utils/dag.py b/sqlmesh/utils/dag.py
index 3c04ee81..69c45855 100644
--- a/sqlmesh/utils/dag.py
+++ b/sqlmesh/utils/dag.py
@@ -5,6 +5,7 @@ A DAG, or directed acyclic graph, is a graph where the edges are directional and
 all the edges pointing in the same direction. SQLMesh uses a DAG to keep track of a project's models. This
 allows SQLMesh to easily determine a model's lineage and to identify upstream and downstream dependencies.
 """
+
 from __future__ import annotations
 
 import typing as t
diff --git a/sqlmesh/utils/git.py b/sqlmesh/utils/git.py
new file mode 100644
index 00000000..4634329d
--- /dev/null
+++ b/sqlmesh/utils/git.py
@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import subprocess
+import typing as t
+from functools import cached_property
+from pathlib import Path
+
+
+class GitClient:
+    def __init__(self, repo: str | Path):
+        self._work_dir = Path(repo)
+
+    def list_untracked_files(self) -> t.List[Path]:
+        return self._execute_list_output(["ls-files", "--others", "--exclude-standard"])
+
+    def list_changed_files(self, target_branch: str = "main") -> t.List[Path]:
+        return self._execute_list_output(["diff", "--name-only", "--diff-filter=d", target_branch])
+
+    def _execute_list_output(self, commands: t.List[str]) -> t.List[Path]:
+        return [(self._git_root / o).absolute() for o in self._execute(commands).split("\n") if o]
+
+    def _execute(self, commands: t.List[str]) -> str:
+        result = subprocess.run(["git"] + commands, cwd=self._work_dir, stdout=subprocess.PIPE)
+        return result.stdout.decode("utf-8").strip()
+
+    @cached_property
+    def _git_root(self) -> Path:
+        return Path(self._execute(["rev-parse", "--show-toplevel"]))
diff --git a/sqlmesh/utils/yaml.py b/sqlmesh/utils/yaml.py
index 425ce7ec..21dc6c0a 100644
--- a/sqlmesh/utils/yaml.py
+++ b/sqlmesh/utils/yaml.py
@@ -47,13 +47,11 @@ def load(
 
 
 @t.overload
-def dump(value: t.Any, stream: io.IOBase) -> None:
-    ...
+def dump(value: t.Any, stream: io.IOBase) -> None: ...
 
 
 @t.overload
-def dump(value: t.Any) -> str:
-    ...
+def dump(value: t.Any) -> str: ...
 
 
 def dump(value: t.Any, stream: t.Optional[io.IOBase] = None) -> t.Optional[str]:
diff --git a/web/client/src/App.tsx b/web/client/src/App.tsx
index 7fed0076..5c620f5c 100644
--- a/web/client/src/App.tsx
+++ b/web/client/src/App.tsx
@@ -6,18 +6,47 @@ import Footer from './library/pages/root/Footer'
 import { getBrowserRouter } from './routes'
 import Loading from '@components/loading/Loading'
 import Spinner from '@components/logo/Spinner'
-import { useApiMeta } from './api'
+import { useApiEnvironments, useApiMeta, useApiPlanRun } from './api'
 import { useStoreContext } from '@context/context'
 import { useStorePlan } from '@context/plan'
 import { EnumPlanAction, ModelPlanAction } from '@models/plan-action'
-import { isNil, isTrue } from './utils'
+import { isNil, isObjectNotEmpty, isTrue } from './utils'
+import { useChannelEvents } from '@api/channels'
+import { type PlanOverviewTracker } from '@models/tracker-plan-overview'
+import {
+  type Environments,
+  type EnvironmentsEnvironments,
+  Status,
+} from '@api/client'
+import { type PlanApplyTracker } from '@models/tracker-plan-apply'
+import { type PlanCancelTracker } from '@models/tracker-plan-cancel'
 
 export default function App(): JSX.Element {
+  const environment = useStoreContext(s => s.environment)
   const version = useStoreContext(s => s.version)
   const modules = useStoreContext(s => s.modules)
   const setVersion = useStoreContext(s => s.setVersion)
-  const setPlanAction = useStorePlan(s => s.setPlanAction)
   const setModules = useStoreContext(s => s.setModules)
+  const addRemoteEnvironments = useStoreContext(s => s.addRemoteEnvironments)
+
+  const planOverview = useStorePlan(s => s.planOverview)
+  const planApply = useStorePlan(s => s.planApply)
+  const planCancel = useStorePlan(s => s.planCancel)
+  const setPlanOverview = useStorePlan(s => s.setPlanOverview)
+  const setPlanApply = useStorePlan(s => s.setPlanApply)
+  const setPlanCancel = useStorePlan(s => s.setPlanCancel)
+  const setPlanAction = useStorePlan(s => s.setPlanAction)
+
+  const { refetch: getEnvironments, cancel: cancelRequestEnvironments } =
+    useApiEnvironments()
+  const { refetch: planRun } = useApiPlanRun(environment.name, {
+    planOptions: {
+      skip_tests: true,
+      include_unmodified: true,
+    },
+  })
+
+  const channel = useChannelEvents()
 
   const {
     refetch: getMeta,
@@ -26,6 +55,20 @@ export default function App(): JSX.Element {
   } = useApiMeta()
 
   useEffect(() => {
+    const channelPlanOverview = channel<any>(
+      'plan-overview',
+      updatePlanOverviewTracker,
+    )
+    const channelPlanApply = channel<any>('plan-apply', updatePlanApplyTracker)
+    const channelPlanCancel = channel<any>(
+      'plan-cancel',
+      updatePlanCancelTracker,
+    )
+
+    channelPlanOverview.subscribe()
+    channelPlanApply.subscribe()
+    channelPlanCancel.subscribe()
+
     void getMeta().then(({ data }) => {
       setVersion(data?.version)
       setModules(Array.from(new Set(modules.concat(data?.modules ?? []))))
@@ -37,11 +80,64 @@ export default function App(): JSX.Element {
       }
     })
 
+    void getEnvironments().then(({ data }) => updateEnviroments(data))
+
     return () => {
+      channelPlanOverview.unsubscribe()
+      channelPlanApply.unsubscribe()
+      channelPlanCancel.unsubscribe()
+
       void cancelRequestMeta()
+      void cancelRequestEnvironments()
     }
   }, [])
 
+  function updatePlanOverviewTracker(data: PlanOverviewTracker): void {
+    planOverview.update(data)
+
+    setPlanOverview(planOverview)
+  }
+
+  function updatePlanCancelTracker(data: PlanCancelTracker): void {
+    planCancel.update(data)
+
+    setPlanCancel(planCancel)
+  }
+
+  function updatePlanApplyTracker(data: PlanApplyTracker): void {
+    planApply.update(data, planOverview)
+
+    const isFinished =
+      isTrue(data.meta?.done) && data.meta?.status !== Status.init
+
+    if (isFinished) {
+      void getEnvironments().then(({ data }) => {
+        planCancel.reset()
+
+        updateEnviroments(data)
+
+        void planRun()
+
+        setPlanCancel(planCancel)
+      })
+    }
+
+    setPlanApply(planApply)
+  }
+
+  function updateEnviroments(data: Optional<Environments>): void {
+    const { environments, default_target_environment, pinned_environments } =
+      data ?? {}
+
+    if (isObjectNotEmpty<EnvironmentsEnvironments>(environments)) {
+      addRemoteEnvironments(
+        Object.values(environments),
+        default_target_environment,
+        pinned_environments,
+      )
+    }
+  }
+
   const router = getBrowserRouter(modules)
   const isLoadingMeta = isNil(version) || isFetchingMeta
 
diff --git a/web/client/src/library/components/documentation/Documentation.tsx b/web/client/src/library/components/documentation/Documentation.tsx
index 38e18e6f..de2614fe 100644
--- a/web/client/src/library/components/documentation/Documentation.tsx
+++ b/web/client/src/library/components/documentation/Documentation.tsx
@@ -11,9 +11,10 @@ import {
   truncate,
 } from '@utils/index'
 import clsx from 'clsx'
-import { type ModelSQLMeshModel } from '@models/sqlmesh-model'
+import { ModelSQLMeshModel } from '@models/sqlmesh-model'
 import { ModelColumns } from '@components/graph/Graph'
 import { useApiModel } from '@api/index'
+import { useStoreContext } from '@context/context'
 
 const Documentation = function Documentation({
   model,
@@ -30,6 +31,8 @@ const Documentation = function Documentation({
   withDescription?: boolean
   withColumns?: boolean
 }): JSX.Element {
+  const setLastSelectedModel = useStoreContext(s => s.setLastSelectedModel)
+
   const { refetch: getModel, cancel: cancelRequestModel } = useApiModel(
     model.name,
   )
@@ -37,6 +40,8 @@ const Documentation = function Documentation({
   useEffect(() => {
     void getModel().then(({ data }) => {
       model.update(data)
+
+      setLastSelectedModel(new ModelSQLMeshModel(model))
     })
 
     return () => {
@@ -100,7 +105,7 @@ const Documentation = function Documentation({
             className="max-h-[15rem]"
             nodeId={model.name}
             columns={model.columns}
-            disabled={model?.isModelPython}
+            disabled={model.isModelPython}
             withHandles={false}
             withSource={false}
             withDescription={true}
diff --git a/web/client/src/library/components/editor/Editor.tsx b/web/client/src/library/components/editor/Editor.tsx
index f5e581f4..a99bf86f 100644
--- a/web/client/src/library/components/editor/Editor.tsx
+++ b/web/client/src/library/components/editor/Editor.tsx
@@ -9,7 +9,7 @@ import EditorInspector from './EditorInspector'
 import EditorPreview from './EditorPreview'
 import { type EditorTab, useStoreEditor } from '~/context/editor'
 import clsx from 'clsx'
-import { EnumFileExtensions } from '@models/file'
+import { EnumFileExtensions, ModelFile } from '@models/file'
 import { useLineageFlow } from '@components/graph/context'
 import { CodeEditorRemoteFile, CodeEditorDefault } from './EditorCode'
 import { useDefaultKeymapsEditorTab, useSQLMeshModelExtensions } from './hooks'
@@ -19,6 +19,7 @@ import { type Table } from 'apache-arrow'
 import { type KeyBinding } from '@codemirror/view'
 import { useStoreContext } from '@context/context'
 import { useIDE } from '~/library/pages/ide/context'
+import { ModelDirectory } from '@models/directory'
 
 function Editor(): JSX.Element {
   const tab = useStoreEditor(s => s.tab)
@@ -49,8 +50,10 @@ function EditorMain({ tab }: { tab: EditorTab }): JSX.Element {
   const isModel = useStoreContext(s => s.isModel)
 
   const files = useStoreProject(s => s.files)
+  const selectedFile = useStoreProject(s => s.selectedFile)
   const setSelectedFile = useStoreProject(s => s.setSelectedFile)
 
+  const tabs = useStoreEditor(s => s.tabs)
   const direction = useStoreEditor(s => s.direction)
   const engine = useStoreEditor(s => s.engine)
   const previewTable = useStoreEditor(s => s.previewTable)
@@ -60,6 +63,10 @@ function EditorMain({ tab }: { tab: EditorTab }): JSX.Element {
   const setPreviewTable = useStoreEditor(s => s.setPreviewTable)
   const setPreviewDiff = useStoreEditor(s => s.setPreviewDiff)
   const setDialects = useStoreEditor(s => s.setDialects)
+  const replaceTab = useStoreEditor(s => s.replaceTab)
+  const createTab = useStoreEditor(s => s.createTab)
+  const selectTab = useStoreEditor(s => s.selectTab)
+  const addTab = useStoreEditor(s => s.addTab)
 
   const { setManuallySelectedColumn } = useLineageFlow()
 
@@ -94,7 +101,12 @@ function EditorMain({ tab }: { tab: EditorTab }): JSX.Element {
   const handleEngineWorkerMessage = useCallback(
     (e: MessageEvent): void => {
       if (e.data.topic === 'dialects') {
+        const model = models.get(tab.file.path)
+
+        tab.dialect = model?.dialect ?? ''
+
         setDialects(e.data.payload)
+        refreshTab(tab)
       }
 
       if (e.data.topic === 'format') {
@@ -113,7 +125,7 @@ function EditorMain({ tab }: { tab: EditorTab }): JSX.Element {
       tab.file.content = value
       tab.isSaved = isFalse(tab.file.isChanged)
 
-      setTimeout(() => refreshTab(tab), 200)
+      refreshTab(tab)
     },
     [tab.id],
   )
@@ -123,18 +135,14 @@ function EditorMain({ tab }: { tab: EditorTab }): JSX.Element {
 
     setIsOpenInspector(false)
 
+    if (isNil(selectedFile)) {
+      setSelectedFile(tab?.file)
+    }
+
     return () => {
       engine.removeEventListener('message', handleEngineWorkerMessage)
     }
-  }, [handleEngineWorkerMessage])
-
-  useEffect(() => {
-    const model = models.get(tab.file.path)
-
-    tab.dialect = model?.dialect ?? ''
-
-    refreshTab(tab)
-  }, [tab.id, models, files])
+  }, [tab.id])
 
   useEffect(() => {
     setPreviewQuery(undefined)
@@ -142,6 +150,31 @@ function EditorMain({ tab }: { tab: EditorTab }): JSX.Element {
     setPreviewDiff(undefined)
   }, [tab.id, tab.file.fingerprint])
 
+  useEffect(() => {
+    if (
+      isNil(selectedFile) ||
+      tab?.file === selectedFile ||
+      selectedFile instanceof ModelDirectory
+    )
+      return
+
+    const newTab = createTab(selectedFile)
+    const shouldReplaceTab =
+      isNotNil(tab) &&
+      tab.file instanceof ModelFile &&
+      isFalse(tab.file.isChanged) &&
+      tab.file.isRemote &&
+      isFalse(tabs.has(selectedFile))
+
+    if (shouldReplaceTab) {
+      replaceTab(tab, newTab)
+    } else {
+      addTab(newTab)
+    }
+
+    selectTab(newTab)
+  }, [selectedFile])
+
   useEffect(() => {
     setPreviewDiff(undefined)
   }, [environment])
diff --git a/web/client/src/library/components/editor/EditorInspector.tsx b/web/client/src/library/components/editor/EditorInspector.tsx
index 09caae0f..2fdb7a00 100644
--- a/web/client/src/library/components/editor/EditorInspector.tsx
+++ b/web/client/src/library/components/editor/EditorInspector.tsx
@@ -22,7 +22,6 @@ import { type ModelSQLMeshModel } from '@models/sqlmesh-model'
 import {
   useApiEvaluate,
   useApiFetchdf,
-  useApiModel,
   useApiRender,
   useApiTableDiff,
 } from '@api/index'
@@ -92,20 +91,6 @@ function InspectorModel({
     .filter(({ isRemote }) => isRemote)
     .map(({ name }) => ({ text: name, value: name }))
 
-  const { refetch: getModel, cancel: cancelRequestModel } = useApiModel(
-    model.name,
-  )
-
-  useEffect(() => {
-    void getModel().then(({ data }) => {
-      model.update(data)
-    })
-
-    return () => {
-      cancelRequestModel()
-    }
-  }, [model.name])
-
   return (
     <Tab.Group>
       <div className="flex w-full items-center">
diff --git a/web/client/src/library/components/editor/EditorTabs.tsx b/web/client/src/library/components/editor/EditorTabs.tsx
index d5b4ff1b..67cc656c 100644
--- a/web/client/src/library/components/editor/EditorTabs.tsx
+++ b/web/client/src/library/components/editor/EditorTabs.tsx
@@ -5,21 +5,16 @@ import { Button } from '../button/Button'
 import clsx from 'clsx'
 import { type EditorTab, useStoreEditor } from '~/context/editor'
 import { useStoreProject } from '@context/project'
-import { ModelDirectory } from '@models/directory'
-import { ModelFile } from '@models/file'
-import { isFalse, isNil, isNotNil } from '@utils/index'
+import { isNil } from '@utils/index'
 import { useStoreContext } from '@context/context'
 
 export default function EditorTabs(): JSX.Element {
-  const selectedFile = useStoreProject(s => s.selectedFile)
   const setSelectedFile = useStoreProject(s => s.setSelectedFile)
 
-  const tab = useStoreEditor(s => s.tab)
   const tabs = useStoreEditor(s => s.tabs)
   const addTab = useStoreEditor(s => s.addTab)
   const selectTab = useStoreEditor(s => s.selectTab)
   const createTab = useStoreEditor(s => s.createTab)
-  const replaceTab = useStoreEditor(s => s.replaceTab)
 
   const [tabsLocal, tabsRemote] = useMemo(() => {
     const local: EditorTab[] = []
@@ -38,37 +33,6 @@ export default function EditorTabs(): JSX.Element {
     return [local, remote]
   }, [tabs])
 
-  useEffect(() => {
-    if (
-      isNil(selectedFile) ||
-      tab?.file === selectedFile ||
-      selectedFile instanceof ModelDirectory
-    )
-      return
-
-    const newTab = createTab(selectedFile)
-    const shouldReplaceTab =
-      isNotNil(tab) &&
-      tab.file instanceof ModelFile &&
-      isFalse(tab.file.isChanged) &&
-      tab.file.isRemote &&
-      isFalse(tabs.has(selectedFile))
-
-    if (shouldReplaceTab) {
-      replaceTab(tab, newTab)
-    } else {
-      addTab(newTab)
-    }
-
-    selectTab(newTab)
-  }, [selectedFile])
-
-  useEffect(() => {
-    if (isNil(selectedFile)) {
-      setSelectedFile(tab?.file)
-    }
-  }, [tab?.id])
-
   function addTabAndSelect(): void {
     const tab = createTab()
 
diff --git a/web/client/src/library/components/plan/Plan.tsx b/web/client/src/library/components/plan/Plan.tsx
index 1ea5da8f..f52e5d3b 100644
--- a/web/client/src/library/components/plan/Plan.tsx
+++ b/web/client/src/library/components/plan/Plan.tsx
@@ -1,4 +1,4 @@
-import { isFalse, isNil, isTrue } from '~/utils'
+import { isNil, isTrue } from '~/utils'
 import { useStorePlan } from '~/context/plan'
 import { useApiPlanRun, useApiPlanApply, useApiCancelPlan } from '~/api'
 import PlanHeader from './PlanHeader'
@@ -6,16 +6,11 @@ import PlanActions from './PlanActions'
 import PlanOptions from './PlanOptions'
 import { EnumPlanActions, usePlanDispatch } from './context'
 import { useApplyPayload, usePlanPayload } from './hooks'
-import Loading from '@components/loading/Loading'
-import Spinner from '@components/logo/Spinner'
-import { EnumVariant } from '~/types/enum'
 import PlanApplyStageTracker from './PlanApplyStageTracker'
 import { EnumPlanAction, ModelPlanAction } from '@models/plan-action'
 import { useStoreProject } from '@context/project'
 import { useIDE } from '~/library/pages/ide/context'
 import { useStoreContext } from '@context/context'
-import { Transition } from '@headlessui/react'
-import PlanActionsDescription from './PlanActionsDescription'
 import { Modules } from '@api/client'
 
 function Plan(): JSX.Element {
@@ -27,6 +22,7 @@ function Plan(): JSX.Element {
 
   const planOverviewTracker = useStorePlan(s => s.planOverview)
   const planApplyTracker = useStorePlan(s => s.planApply)
+  const planCancelTracker = useStorePlan(s => s.planCancel)
   const planAction = useStorePlan(s => s.planAction)
   const setPlanAction = useStorePlan(s => s.setPlanAction)
 
@@ -55,7 +51,7 @@ function Plan(): JSX.Element {
   }
 
   function reset(): void {
-    planApplyTracker.reset()
+    resetPlanTrackers()
 
     cleanUp()
     clearErrors()
@@ -86,97 +82,46 @@ function Plan(): JSX.Element {
   }
 
   function apply(): void {
+    resetPlanTrackers()
+
     dispatch([{ type: EnumPlanActions.ResetTestsReport }])
 
     planApply().catch(console.log)
   }
 
   function run(): void {
-    setTests(undefined)
+    resetPlanTrackers()
 
-    planOverviewTracker.reset()
-    planApplyTracker.reset()
+    setTests(undefined)
 
     dispatch([{ type: EnumPlanActions.ResetTestsReport }])
 
     void planRun()
   }
 
+  function resetPlanTrackers(): void {
+    planOverviewTracker.reset()
+    planApplyTracker.reset()
+    planCancelTracker.reset()
+  }
+
   return (
-    <div className="flex flex-col w-full h-full overflow-hidden">
-      <PlanHeader />
-      <div className="w-full h-full px-4 overflow-y-scroll hover:scrollbar scrollbar--vertical">
-        <Transition
-          appear
-          show
-          enter="transition ease duration-700 transform"
-          enterFrom="opacity-0 -translate-y-full"
-          enterTo="opacity-100 translate-y-0"
-          leave="transition ease duration-1000 transform"
-          leaveFrom="opacity-100 translate-y-0"
-          leaveTo="opacity-0 -translate-y-full"
-        >
-          <PlanOptions />
-        </Transition>
-        <Transition
-          appear
-          show
-          enter="transition ease duration-700 transform"
-          enterFrom="opacity-0 scale-95"
-          enterTo="opacity-100 scale-100"
-          leave="transition ease duration-1000 transform"
-          leaveFrom="opacity-100 scale-100"
-          leaveTo="opacity-0 scale-95"
-        >
-          {planAction.isCancelling ? (
-            <CancellingPlanApply />
-          ) : (
-            <PlanApplyStageTracker />
-          )}
-        </Transition>
+    <div className="flex flex-col w-full h-full">
+      {environment.isProd && <PlanHeader />}
+      <div className="w-full h-full flex flex-col pt-2 pl-4 pr-2 overflow-y-scroll hover:scrollbar scrollbar--vertical">
+        <PlanOptions />
+        <PlanApplyStageTracker />
       </div>
       {modules.includes(Modules.plans) && (
-        <>
-          <Transition
-            appear
-            show={isFalse(planAction.isDone)}
-            enter="transition ease duration-1000 transform"
-            enterFrom="opacity-0 translate-y-full"
-            enterTo="opacity-100 translate-y-0"
-            leave="transition ease duration-500 transform"
-            leaveFrom="opacity-100 translate-y-0"
-            leaveTo="opacity-0 translate-y-full"
-          >
-            <PlanActionsDescription />
-          </Transition>
-          <PlanActions
-            apply={apply}
-            run={run}
-            cancel={cancel}
-            reset={reset}
-          />
-        </>
+        <PlanActions
+          apply={apply}
+          run={run}
+          cancel={cancel}
+          reset={reset}
+        />
       )}
     </div>
   )
 }
 
 export default Plan
-
-function CancellingPlanApply(): JSX.Element {
-  return (
-    <div className="w-full h-full p-4">
-      <div className="w-full h-full flex justify-center items-center p-4 bg-warning-10 rounded-lg overflow-hidden">
-        <Loading className="inline-block">
-          <Spinner
-            variant={EnumVariant.Warning}
-            className="w-3 h-3 border border-neutral-10 mr-4"
-          />
-          <h3 className="text-2xl text-warning-500 font-bold">
-            Cancelling Plan...
-          </h3>
-        </Loading>
-      </div>
-    </div>
-  )
-}
diff --git a/web/client/src/library/components/plan/PlanActions.tsx b/web/client/src/library/components/plan/PlanActions.tsx
index 2da2474f..6a31778d 100644
--- a/web/client/src/library/components/plan/PlanActions.tsx
+++ b/web/client/src/library/components/plan/PlanActions.tsx
@@ -1,12 +1,13 @@
 import { type MouseEvent } from 'react'
 import useActiveFocus from '~/hooks/useActiveFocus'
 import { EnumSize, EnumVariant } from '~/types/enum'
-import { includes, isFalse } from '~/utils'
+import { isFalse } from '~/utils'
 import { Button } from '../button/Button'
 import { EnumPlanAction, ModelPlanAction } from '@models/plan-action'
 import { useStorePlan } from '@context/plan'
 import { useStoreContext } from '@context/context'
 import { AddEnvironment, SelectEnvironemnt } from '~/library/pages/root/Page'
+import { Transition } from '@headlessui/react'
 
 export default function PlanActions({
   run,
@@ -25,6 +26,7 @@ export default function PlanActions({
   const setShowConfirmation = useStoreContext(s => s.setShowConfirmation)
 
   const planAction = useStorePlan(s => s.planAction)
+  const planCancel = useStorePlan(s => s.planCancel)
 
   const setFocus = useActiveFocus<HTMLButtonElement>()
 
@@ -90,83 +92,112 @@ export default function PlanActions({
     run()
   }
 
+  const showPlanAction =
+    isFalse(planAction.isCancelling) &&
+    isFalse(planAction.isDone) &&
+    (planAction.isProcessing ? isFalse(planCancel.isSuccessed) : true)
+
   return (
     <>
-      <div className="flex justify-between px-4 pb-2">
+      <div className="flex justify-between pt-2 pl-4 pr-2 pb-2">
         <div className="flex w-full items-center">
-          {(planAction.isRun || planAction.isRunning) && (
-            <Button
-              disabled={planAction.isRunning}
-              onClick={handleRun}
-              ref={setFocus}
-              variant={EnumVariant.Primary}
-              autoFocus
-            >
-              <span>
-                {ModelPlanAction.getActionDisplayName(planAction, [
-                  EnumPlanAction.RunningTask,
-                  EnumPlanAction.Running,
-                  EnumPlanAction.Run,
-                ])}
-              </span>
-            </Button>
-          )}
-          {(planAction.isApply || planAction.isApplying) && (
-            <Button
-              onClick={handleApply}
-              disabled={planAction.isApplying}
-              ref={setFocus}
-              variant={EnumVariant.Primary}
-            >
-              {ModelPlanAction.getActionDisplayName(
-                planAction,
-                [
-                  EnumPlanAction.Applying,
-                  EnumPlanAction.ApplyBackfill,
-                  EnumPlanAction.ApplyVirtual,
-                  EnumPlanAction.ApplyChangesAndBackfill,
-                  EnumPlanAction.ApplyMetadata,
-                ],
-                'Apply',
-              )}
-            </Button>
-          )}
-          {planAction.isProcessing && (
-            <Button
-              onClick={handleCancel}
-              variant={EnumVariant.Danger}
-              className="justify-self-end"
-              disabled={planAction.isCancelling}
-            >
-              {ModelPlanAction.getActionDisplayName(
-                planAction,
-                [EnumPlanAction.Cancelling],
-                'Cancel',
-              )}
-            </Button>
-          )}
+          <Transition
+            appear
+            show={showPlanAction}
+            enter="transition ease duration-300 transform"
+            enterFrom="opacity-0 scale-95"
+            enterTo="opacity-100 scale-100"
+            leave="transition ease duration-300 transform"
+            leaveFrom="opacity-100 scale-100"
+            leaveTo="opacity-0 scale-95"
+            className="trasition-all duration-300 ease-in-out"
+          >
+            {showPlanAction && (
+              <Button
+                disabled={
+                  planAction.isProcessing ||
+                  planCancel.isSuccessed ||
+                  planAction.isDone
+                }
+                onClick={
+                  planAction.isRun
+                    ? handleRun
+                    : planCancel.isSuccessed
+                    ? undefined
+                    : handleApply
+                }
+                ref={setFocus}
+                variant={
+                  planCancel.isSuccessed
+                    ? EnumVariant.Danger
+                    : EnumVariant.Primary
+                }
+                autoFocus
+                className="trasition-all duration-300 ease-in-out"
+              >
+                <span>
+                  {ModelPlanAction.getActionDisplayName(
+                    planAction,
+                    planCancel.isSuccessed
+                      ? []
+                      : [
+                          EnumPlanAction.RunningTask,
+                          EnumPlanAction.Running,
+                          EnumPlanAction.Run,
+                          EnumPlanAction.Applying,
+                          EnumPlanAction.ApplyBackfill,
+                          EnumPlanAction.ApplyVirtual,
+                          EnumPlanAction.ApplyChangesAndBackfill,
+                          EnumPlanAction.ApplyMetadata,
+                        ],
+                    planCancel.isSuccessed ? 'Canceled' : 'Done',
+                  )}
+                </span>
+              </Button>
+            )}
+          </Transition>
         </div>
         <div className="flex items-center">
-          {[planAction.isProcessing, planAction.isRun].every(isFalse) && (
+          <Transition
+            appear
+            show={planAction.isProcessing || isFalse(planAction.isRun)}
+            enter="transition ease duration-300 transform"
+            enterFrom="opacity-0 scale-95"
+            enterTo="opacity-100 scale-100"
+            leave="transition ease duration-300 transform"
+            leaveFrom="opacity-100 scale-100"
+            leaveTo="opacity-0 scale-95"
+            className="trasition-all duration-300 ease-in-out"
+          >
             <Button
-              onClick={handleReset}
-              variant={EnumVariant.Neutral}
-              disabled={includes(
-                [
-                  EnumPlanAction.Running,
-                  EnumPlanAction.Applying,
-                  EnumPlanAction.Cancelling,
-                ],
-                planAction.value,
-              )}
+              onClick={
+                planAction.isRunning || planAction.isApplying
+                  ? handleCancel
+                  : handleReset
+              }
+              variant={
+                planAction.isProcessing
+                  ? EnumVariant.Danger
+                  : EnumVariant.Neutral
+              }
+              disabled={
+                planAction.isCancelling ||
+                (planAction.isProcessing && planCancel.isSuccessed)
+              }
             >
               {ModelPlanAction.getActionDisplayName(
                 planAction,
-                [],
-                'Start Over',
+                planAction.isProcessing && isFalse(planCancel.isSuccessed)
+                  ? [EnumPlanAction.Cancelling]
+                  : [],
+                planAction.isProcessing
+                  ? planCancel.isSuccessed
+                    ? 'Finishing Cancellation...'
+                    : 'Cancel'
+                  : 'Start Over',
               )}
             </Button>
-          )}
+          </Transition>
         </div>
       </div>
     </>
diff --git a/web/client/src/library/components/plan/PlanActionsDescription.tsx b/web/client/src/library/components/plan/PlanActionsDescription.tsx
deleted file mode 100644
index cb22a086..00000000
--- a/web/client/src/library/components/plan/PlanActionsDescription.tsx
+++ /dev/null
@@ -1,100 +0,0 @@
-import { useStoreContext } from '~/context/context'
-import { usePlan } from './context'
-import { isFalse, isNotNil, isStringEmptyOrNil, isTrue } from '@utils/index'
-import Banner from '@components/banner/Banner'
-import { EnumSize, EnumVariant } from '~/types/enum'
-
-export default function PlanActionsDescription(): JSX.Element {
-  const {
-    start,
-    end,
-    skip_tests,
-    skip_backfill,
-    no_gaps,
-    no_auto_categorization,
-    forward_only,
-    restate_models,
-    include_unmodified,
-    auto_apply,
-    create_from,
-  } = usePlan()
-
-  const environment = useStoreContext(s => s.environment)
-
-  return (
-    <div className="text-xs px-4 py-2">
-      <Banner
-        variant={EnumVariant.Warning}
-        size={EnumSize.sm}
-      >
-        <p className="text-sm">
-          <span>Plan for</span>
-          <b className="text-primary-500 font-bold mx-1">{environment.name}</b>
-          <span className="inline-block mr-1">environment</span>
-          {isFalse(environment.isProd) && isNotNil(create_from) && (
-            <span className="inline-block mr-1">
-              <span>based of</span>
-              <b className="text-primary-500 font-bold mx-1">{create_from}</b>
-            </span>
-          )}
-          {isTrue(auto_apply) && (
-            <span className="inline-block mr-1">
-              and <b>Apply</b> automatically
-            </span>
-          )}
-          {isTrue(skip_tests) && (
-            <span className="inline-block mr-1">
-              without <b>Tests</b>
-            </span>
-          )}
-          {
-            <span className="inline-block mr-1">
-              <span>from&nbsp;</span>
-              <b>
-                {isFalse(isStringEmptyOrNil(start))
-                  ? start
-                  : 'the begining of history'}
-              </b>
-            </span>
-          }
-          {
-            <span className="inline-block mr-1">
-              till <b>{isFalse(isStringEmptyOrNil(start)) ? end : 'today'}</b>
-            </span>
-          }
-          {isTrue(no_gaps) && (
-            <span className="inline-block mr-1">
-              with <b>No Gaps</b>
-            </span>
-          )}
-          {isTrue(skip_backfill) && (
-            <span className="inline-block mr-1">
-              without <b>Backfills</b>
-            </span>
-          )}
-          {isTrue(forward_only) && (
-            <span className="inline-block mr-1">
-              all changes will be <b>Forward Only</b>
-            </span>
-          )}
-          {isTrue(no_auto_categorization) && (
-            <span className="inline-block mr-1">
-              also set <b>Change Category</b> manually
-            </span>
-          )}
-          {isFalse(isStringEmptyOrNil(restate_models)) && (
-            <span className="inline-block mr-1">
-              and restate the following models <b>{restate_models}</b>
-            </span>
-          )}
-          {isTrue(include_unmodified) && (
-            <span className="inline-block mr-1">
-              and with views for all models in the target development
-              environment
-            </span>
-          )}
-        </p>
-      </Banner>
-    </div>
-  )
-}
diff --git a/web/client/src/library/components/plan/PlanApplyStageTracker.tsx b/web/client/src/library/components/plan/PlanApplyStageTracker.tsx
index a0cd8597..ef4c9de8 100644
--- a/web/client/src/library/components/plan/PlanApplyStageTracker.tsx
+++ b/web/client/src/library/components/plan/PlanApplyStageTracker.tsx
@@ -44,65 +44,81 @@ export default function PlanApplyStageTracker(): JSX.Element {
 
   const planApply = useStorePlan(s => s.planApply)
   const planOverview = useStorePlan(s => s.planOverview)
+  const planCancel = useStorePlan(s => s.planCancel)
   const planAction = useStorePlan(s => s.planAction)
 
-  const { plan_options } = getPlanOverviewDetails(planApply, planOverview)
+  const { plan_options } = getPlanOverviewDetails(
+    planApply,
+    planOverview,
+    planCancel,
+  )
 
   const hasTestsDetails = isNotNil(tests) && Boolean(tests.total)
   const showTestsMessage =
     isNotNil(tests) && Boolean(tests.message) && isFalse(hasTestsDetails)
 
+  const hasFailedTests = isNotNil(tests) && Boolean(tests.failures)
+  const showChangesAndBackfills =
+    isFalse(planAction.isProcessing) &&
+    isFalse(planAction.isDone) &&
+    isFalse(planCancel.isFinished)
+
   return (
     <Transition
+      appear
       show={isFalse(planAction.isRun)}
-      enter="transition ease duration-500 transform"
+      enter="transition ease duration-300 transform"
       enterFrom="opacity-0 scale-95"
       enterTo="opacity-100 scale-100"
-      leave="transition ease duration-500 transform"
+      leave="transition ease duration-300 transform"
       leaveFrom="opacity-100 scale-100"
       leaveTo="opacity-0 scale-95"
+      className="my-2"
     >
-      <div className="mt-8 mb-4">
-        <StageChanges isOpen={isFalse(planAction.isApplying)} />
-        <StageBackfills isOpen={isFalse(planAction.isApplying)} />
-        {isTrue(plan_options?.skip_tests) ? (
-          <Banner
-            className="flex items-center mb-1"
-            size={EnumSize.sm}
-            hasBackground={false}
-          >
-            <CheckIcon className="w-5 mr-2" />
-            <Banner.Label className="mr-2 text-sm">Tests Skipped</Banner.Label>
-          </Banner>
-        ) : hasTestsDetails ? (
-          <StageTestsFailed
-            isOpen={true}
-            report={tests}
-          />
-        ) : showTestsMessage ? (
-          <StageTestsCompleted report={tests} />
-        ) : (
-          <Banner
-            className="flex items-center mb-1"
-            size={EnumSize.sm}
-            hasBackground={false}
-          >
-            <CheckIcon className="w-5 mr-2" />
-            <Banner.Label className="mr-2 text-sm">No Tests</Banner.Label>
-          </Banner>
-        )}
-        <StageValidate />
-        <StageVirtualUpdate />
+      <StageChanges isOpen={showChangesAndBackfills} />
+      <StageBackfills isOpen={showChangesAndBackfills} />
+      {isTrue(plan_options?.skip_tests) ? (
+        <Banner
+          className="flex items-center mb-1"
+          size={EnumSize.sm}
+          hasBackground={false}
+        >
+          <CheckIcon className="w-4 mr-2" />
+          <Banner.Label className="mr-2 text-sm">Tests Skipped</Banner.Label>
+        </Banner>
+      ) : hasTestsDetails ? (
+        <StageTestsFailed
+          isOpen={true}
+          report={tests}
+        />
+      ) : showTestsMessage ? (
+        <StageTestsCompleted report={tests} />
+      ) : (
+        <Banner
+          className="flex items-center mb-1"
+          size={EnumSize.sm}
+          hasBackground={false}
+        >
+          <CheckIcon className="w-4 mr-2" />
+          <Banner.Label className="mr-2 text-sm">No Tests</Banner.Label>
+        </Banner>
+      )}
+      <StageVirtualUpdate />
+      {(hasFailedTests || planApply.shouldShowEvaluation) && (
         <StageEvaluate
           start={planApply.evaluationStart}
-          end={planApply.evaluationEnd}
+          end={
+            planApply.isFinished
+              ? planApply.evaluationEnd ?? planCancel.meta?.end
+              : undefined
+          }
         >
           <StageCreation />
           <StageRestate />
           <StageBackfill />
           <StagePromote />
         </StageEvaluate>
-      </div>
+      )}
     </Transition>
   )
 }
@@ -110,92 +126,69 @@ export default function PlanApplyStageTracker(): JSX.Element {
 function StageChanges({ isOpen = false }: { isOpen?: boolean }): JSX.Element {
   const planApply = useStorePlan(s => s.planApply)
   const planOverview = useStorePlan(s => s.planOverview)
-  const planAction = useStorePlan(s => s.planAction)
+  const planCancel = useStorePlan(s => s.planCancel)
 
   const { meta, stageChanges, hasChanges } = getPlanOverviewDetails(
     planApply,
     planOverview,
+    planCancel,
   )
-
-  const showChanges = planAction.isRunning || isTrue(hasChanges)
+  const tempMeta = stageChanges?.meta ?? meta
+  const showChanges = tempMeta?.status === Status.init || isTrue(hasChanges)
 
   return showChanges ? (
-    <Transition
-      show={showChanges}
-      enter="transition ease duration-700 transform"
-      enterFrom="opacity-0 scale-95"
-      enterTo="opacity-100 scale-100"
-      leave="transition ease duration-1000 transform"
-      leaveFrom="opacity-100 scale-100"
-      leaveTo="opacity-0 scale-95"
-    >
-      <Stage
-        meta={stageChanges?.meta ?? meta ?? { status: Status.init }}
-        states={['Changes', 'Failed Getting Changes', 'Getting Changes...']}
-        isOpen={isOpen}
-        panel={<PlanChanges />}
-      />
-    </Transition>
+    <Stage
+      meta={tempMeta}
+      states={['Changes', 'Failed Getting Changes', 'Getting Changes...']}
+      isOpen={isOpen && isTrue(hasChanges)}
+      panel={<PlanChanges />}
+    />
   ) : (
     <Banner
       className="flex items-center mb-1"
       size={EnumSize.sm}
       hasBackground={false}
     >
-      <CheckIcon className="w-5 mr-2" />
+      <CheckIcon className="w-4 mr-2" />
       <Banner.Label className="mr-2 text-sm">No Changes</Banner.Label>
     </Banner>
   )
 }
 
-function StageBackfills({ isOpen }: { isOpen?: boolean }): JSX.Element {
+function StageBackfills({ isOpen = false }: { isOpen?: boolean }): JSX.Element {
   const planApply = useStorePlan(s => s.planApply)
   const planOverview = useStorePlan(s => s.planOverview)
-  const planAction = useStorePlan(s => s.planAction)
+  const planCancel = useStorePlan(s => s.planCancel)
 
   const { meta, stageBackfills, backfills, hasBackfills } =
-    getPlanOverviewDetails(planApply, planOverview)
-
-  const showBackfills = planAction.isRunning || isTrue(hasBackfills)
+    getPlanOverviewDetails(planApply, planOverview, planCancel)
+  const tempMeta = stageBackfills?.meta ?? meta
+  const showBackfills = tempMeta?.status === Status.init || isTrue(hasBackfills)
 
   return showBackfills ? (
-    <Transition
-      show={showBackfills}
-      enter="transition ease duration-700 transform"
-      enterFrom="opacity-0 scale-95"
-      enterTo="opacity-100 scale-100"
-      leave="transition ease duration-1000 transform"
-      leaveFrom="opacity-100 scale-100"
-      leaveTo="opacity-0 scale-95"
-    >
-      <Stage
-        meta={stageBackfills?.meta ?? meta ?? { status: Status.init }}
-        states={[
-          'Backfills',
-          'Failed Getting Backfills',
-          'Getting Backfills...',
-        ]}
-        isOpen={isOpen}
-        panel={
-          <PlanChangePreview
-            headline={`Models ${backfills.length}`}
+    <Stage
+      meta={tempMeta}
+      states={['Backfills', 'Failed Getting Backfills', 'Getting Backfills...']}
+      isOpen={isOpen && isTrue(hasBackfills)}
+      panel={
+        <PlanChangePreview
+          headline={`Models ${backfills.length}`}
+          type={EnumPlanChangeType.Default}
+        >
+          <PlanChangePreview.Default
             type={EnumPlanChangeType.Default}
-          >
-            <PlanChangePreview.Default
-              type={EnumPlanChangeType.Default}
-              changes={backfills}
-            />
-          </PlanChangePreview>
-        }
-      />
-    </Transition>
+            changes={backfills}
+          />
+        </PlanChangePreview>
+      }
+    />
   ) : (
     <Banner
       className="flex items-center mb-1"
       size={EnumSize.sm}
       hasBackground={false}
     >
-      <CheckIcon className="w-5 mr-2" />
+      <CheckIcon className="w-4 mr-2" />
       <Banner.Label className="mr-2 text-sm">No Backfills</Banner.Label>
     </Banner>
   )
@@ -238,25 +231,6 @@ function StageTestsFailed({
   )
 }
 
-function StageValidate(): JSX.Element {
-  const planApply = useStorePlan(s => s.planApply)
-  const planOverview = useStorePlan(s => s.planOverview)
-
-  const { stageValidation } = getPlanOverviewDetails(planApply, planOverview)
-
-  return (
-    <Stage
-      meta={stageValidation?.meta}
-      states={[
-        'Plan Validated',
-        'Plan Validation Failed',
-        'Validating Plan...',
-      ]}
-      showDetails={false}
-    />
-  )
-}
-
 function StageEvaluate({
   start,
   end,
@@ -266,54 +240,42 @@ function StageEvaluate({
   end?: PlanOverviewStageTrackerEnd
   children: React.ReactNode
 }): JSX.Element {
-  const tests = useStoreProject(s => s.tests)
-  const planApply = useStorePlan(s => s.planApply)
-
   const elStageEvaluate = useRef<HTMLDivElement>(null)
 
-  const hasFailedTests = isNotNil(tests) && Boolean(tests.failures)
+  useEffect(() => {
+    setTimeout(() => {
+      elStageEvaluate.current?.scrollIntoView({
+        behavior: 'smooth',
+        block: 'start',
+      })
+    }, 500)
+  }, [])
 
   return (
-    <Transition
-      show={hasFailedTests || planApply.shouldShowEvaluation}
-      enter="transition ease duration-500 transform delay-[1000ms]"
-      enterFrom="opacity-0 scale-95"
-      enterTo="opacity-100 scale-100"
-      leave="transition ease duration-500 transform"
-      leaveFrom="opacity-100 scale-100"
-      leaveTo="opacity-0 scale-95"
+    <div
+      ref={elStageEvaluate}
+      className="pt-4 pb-2 text-xs"
     >
-      <div
-        ref={elStageEvaluate}
-        className="pt-6 pb-2"
-      >
-        {isNotNil(start) && (
-          <>
-            <small className="text-neutral-500 block px-4 mb-1">
-              Evaluation started at{' '}
-              <b>{toDateFormat(new Date(start), 'yyyy-mm-dd hh-mm-ss')}</b>
-            </small>
-            <Divider />
-            <small className="text-neutral-500 block px-4 mt-1">
-              Given a plan, it pushes snapshots into the state and then kicks
-              off the backfill process for all affected snapshots. Once backfill
-              is done, snapshots that are part of the plan are promoted in the
-              environment targeted by this plan.
-            </small>
-          </>
-        )}
-        <div className="py-2">{children}</div>
-        {isNotNil(end) && (
-          <>
-            <Divider />
-            <small className="text-neutral-500 block px-4 mt-1">
-              Evaluation stopped at{' '}
-              <b>{toDateFormat(new Date(end), 'yyyy-mm-dd hh-mm-ss')}</b>
-            </small>
-          </>
-        )}
-      </div>
-    </Transition>
+      {isNotNil(start) && (
+        <>
+          <span className="text-neutral-500 block px-4 mb-1">
+            Evaluation started at{' '}
+            {toDateFormat(new Date(start), 'yyyy-mm-dd hh-mm-ss')}
+          </span>
+          <Divider />
+        </>
+      )}
+      <div className="py-2">{children}</div>
+      {isNotNil(end) && (
+        <>
+          <Divider />
+          <span className="text-neutral-500 block px-4 mt-1">
+            Evaluation stopped at{' '}
+            {toDateFormat(new Date(end), 'yyyy-mm-dd hh-mm-ss')}
+          </span>
+        </>
+      )}
+    </div>
   )
 }
 
@@ -372,7 +334,7 @@ function StageRestate(): JSX.Element {
       size={EnumSize.sm}
       hasBackground={false}
     >
-      <CheckIcon className="w-5 mr-2" />
+      <CheckIcon className="w-4 mr-2" />
       <Banner.Label className="mr-2 text-sm">No Models To Restate</Banner.Label>
     </Banner>
   ) : (
@@ -388,23 +350,12 @@ function StageRestate(): JSX.Element {
 }
 
 function StageBackfill(): JSX.Element {
-  const elStageBackfill = useRef<HTMLDivElement>(null)
-
   const planApply = useStorePlan(s => s.planApply)
   const planAction = useStorePlan(s => s.planAction)
 
   const environment = planApply.environment
   const stageBackfill = planApply.stageBackfill
 
-  useEffect(() => {
-    setTimeout(() => {
-      elStageBackfill.current?.scrollIntoView({
-        behavior: 'smooth',
-        block: 'start',
-      })
-    }, 1000)
-  }, [])
-
   const { change_categorization } = usePlan()
 
   const categories = useMemo(
@@ -472,49 +423,47 @@ function StageBackfill(): JSX.Element {
   if (isNil(stageBackfill) || isNil(environment)) return <></>
 
   return (
-    <div ref={elStageBackfill}>
-      <Stage
-        meta={stageBackfill.meta}
-        states={[
-          'Intervals Backfilled',
-          'Intervals Backfilling Failed',
-          'Backfilling Intervals...',
-        ]}
-        showDetails={true}
-        isOpen={true}
-        shouldCollapse={false}
-      >
-        <TasksOverview tasks={tasks}>
-          {({ total, completed, models, completedBatches, totalBatches }) => (
-            <>
-              <TasksOverview.Summary
-                headline="Target Environment"
-                environment={environment}
-                completed={completed}
-                total={total}
-                completedBatches={completedBatches}
-                totalBatches={totalBatches}
-                updateType={planAction.isApplyVirtual ? 'Virtual' : 'Backfill'}
+    <Stage
+      meta={stageBackfill.meta}
+      states={[
+        'Intervals Backfilled',
+        'Intervals Backfilling Failed',
+        'Backfilling Intervals...',
+      ]}
+      showDetails={true}
+      isOpen={true}
+      shouldCollapse={false}
+    >
+      <TasksOverview tasks={tasks}>
+        {({ total, completed, models, completedBatches, totalBatches }) => (
+          <>
+            <TasksOverview.Summary
+              headline="Target Environment"
+              environment={environment}
+              completed={completed}
+              total={total}
+              completedBatches={completedBatches}
+              totalBatches={totalBatches}
+              updateType={planAction.isApplyVirtual ? 'Virtual' : 'Backfill'}
+            />
+            {isNotNil(models) && (
+              <TasksOverview.Details
+                models={models}
+                added={planApply.added}
+                removed={planApply.removed}
+                direct={planApply.direct}
+                indirect={planApply.indirect}
+                metadata={planApply.metadata}
+                queue={planApply.queue}
+                showBatches={true}
+                showVirtualUpdate={planAction.isApplyVirtual}
+                showProgress={true}
               />
-              {isNotNil(models) && (
-                <TasksOverview.Details
-                  models={models}
-                  added={planApply.added}
-                  removed={planApply.removed}
-                  direct={planApply.direct}
-                  indirect={planApply.indirect}
-                  metadata={planApply.metadata}
-                  queue={planApply.queue}
-                  showBatches={true}
-                  showVirtualUpdate={planAction.isApplyVirtual}
-                  showProgress={true}
-                />
-              )}
-            </>
-          )}
-        </TasksOverview>
-      </Stage>
-    </div>
+            )}
+          </>
+        )}
+      </TasksOverview>
+    </Stage>
   )
 }
 
@@ -572,10 +521,10 @@ function StageVirtualUpdate(): JSX.Element {
 
   const planApply = useStorePlan(s => s.planApply)
   const planOverview = useStorePlan(s => s.planOverview)
-  const isVirtualUpdate = planOverview.isLatest
-    ? isTrue(planApply.overview?.isVirtualUpdate)
-    : planOverview.isVirtualUpdate
-  const isUpdated = isTrue(planApply.stagePromote?.meta?.done)
+
+  const isVirtualUpdate =
+    planApply.overview?.isVirtualUpdate ?? planOverview.isVirtualUpdate
+  const isUpdated = isTrue(planApply.isFinished)
 
   return isVirtualUpdate ? (
     <Stage
@@ -599,9 +548,10 @@ function StageVirtualUpdate(): JSX.Element {
 function PlanChanges(): JSX.Element {
   const planOverview = useStorePlan(s => s.planOverview)
   const planApply = useStorePlan(s => s.planApply)
+  const planCancel = useStorePlan(s => s.planCancel)
 
   const { hasChanges, added, removed, direct, indirect, metadata } =
-    getPlanOverviewDetails(planApply, planOverview)
+    getPlanOverviewDetails(planApply, planOverview, planCancel)
 
   return (
     <div className="w-full my-2">
@@ -694,7 +644,7 @@ function Stage({
 
   const hasChildren = isNotNil(panel) || isNotNil(children)
 
-  const [open, setOpen] = useState(false)
+  const [open, setOpen] = useState(isOpen)
 
   useEffect(() => {
     if (isNil(elTrigger.current)) return
@@ -707,35 +657,34 @@ function Stage({
   }, [elTrigger, planOverview, planApply, shouldCollapse])
 
   useEffect(() => {
-    setOpen(isOpen || (meta?.status !== Status.init && hasChildren))
-  }, [meta?.status, hasChildren, isOpen])
-
-  if (isNil(meta)) return <></>
+    setOpen(isOpen && hasChildren)
+  }, [isOpen, hasChildren])
 
   const variant =
-    meta.status === Status.success
+    meta?.status === Status.success
       ? EnumVariant.Success
-      : meta.status === Status.fail
+      : meta?.status === Status.fail
       ? EnumVariant.Danger
       : EnumVariant.Info
   const [titleSuccess, titleFail, titleDefault] = states
   const text =
-    meta.status === Status.success
+    meta?.status === Status.success
       ? titleSuccess
-      : meta.status === Status.fail
+      : meta?.status === Status.fail
       ? titleFail
       : titleDefault
 
   return (
     <Transition
       appear
-      show
-      enter="transition-all ease-in duration-500 delay-[200ms]"
-      enterFrom="opacity-0 translate-y-6"
-      enterTo="opacity-100 translate-y-0"
-      leave="transition-all ease-out duration-200"
-      leaveFrom="opacity-100"
-      leaveTo="opacity-0"
+      show={isNotNil(meta)}
+      enter="transition ease duration-300 transform"
+      enterFrom="opacity-0 scale-95"
+      enterTo="opacity-100 scale-100"
+      leave="transition ease duration-300 transform"
+      leaveFrom="opacity-100 scale-100"
+      leaveTo="opacity-0 scale-95"
+      className="my-2"
     >
       <Disclosure>
         <Banner
@@ -756,16 +705,7 @@ function Stage({
           >
             {isNil(trigger) ? (
               <>
-                <Transition
-                  show={meta.status === Status.init}
-                  className="flex items-center h-full"
-                  enter="transition-all ease-in-out duration-500"
-                  enterFrom="opacity-0 translate-y-6"
-                  enterTo="opacity-100 translate-y-0"
-                  leave="transition-all ease-in-out duration-200"
-                  leaveFrom="opacity-100"
-                  leaveTo="opacity-0"
-                >
+                {meta?.status === Status.init && (
                   <Loading
                     text={text}
                     hasSpinner
@@ -773,43 +713,36 @@ function Stage({
                     variant={EnumVariant.Primary}
                     className="w-full"
                   />
-                </Transition>
-                <Transition
-                  show={meta.status !== Status.init}
-                  className="flex items-center h-full"
-                  enter="transition-all ease-in-out duration-400 delay-[300ms]"
-                  enterFrom="opacity-0 translate-y-6"
-                  enterTo="opacity-100 translate-y-0"
-                  leave="transition-all ease-in-out duration-200"
-                  leaveFrom="opacity-100"
-                  leaveTo="opacity-0"
-                >
-                  {showDetails ? (
-                    <>
-                      {open ? (
-                        <MinusCircleIcon className="w-6 mr-2" />
-                      ) : (
-                        <PlusCircleIcon className="w-6 mr-2" />
-                      )}
-                    </>
-                  ) : (
-                    <>
-                      {meta.status === Status.success && (
-                        <CheckIcon className="w-6 mr-2" />
-                      )}
-                      {meta.status === Status.fail && (
-                        <ExclamationCircleIcon className="w-6 mr-2" />
-                      )}
-                    </>
-                  )}
-                  <Banner.Label className="mr-2 text-sm w-full">
-                    <Title
-                      text={text}
-                      size={EnumSize.sm}
-                      variant={variant}
-                    />
-                  </Banner.Label>
-                </Transition>
+                )}
+                {meta?.status !== Status.init && (
+                  <div className="flex items-center h-full">
+                    {showDetails ? (
+                      <>
+                        {open ? (
+                          <MinusCircleIcon className="min-w-4 max-w-4 mr-2" />
+                        ) : (
+                          <PlusCircleIcon className="min-w-4 max-w-4 mr-2" />
+                        )}
+                      </>
+                    ) : (
+                      <>
+                        {meta?.status === Status.success && (
+                          <CheckIcon className="min-w-4 max-w-4 mr-2" />
+                        )}
+                        {meta?.status === Status.fail && (
+                          <ExclamationCircleIcon className="min-w-4 max-w-4 mr-2" />
+                        )}
+                      </>
+                    )}
+                    <Banner.Label className="mr-2 text-sm w-full">
+                      <Title
+                        text={text}
+                        size={EnumSize.sm}
+                        variant={variant}
+                      />
+                    </Banner.Label>
+                  </div>
+                )}
               </>
             ) : (
               trigger
@@ -817,13 +750,15 @@ function Stage({
           </Disclosure.Button>
         </Banner>
         <Transition
+          appear
           show={open}
-          enter="transition-all ease-in-out duration-300 delay-[200ms]"
-          enterFrom="opacity-0 translate-y-6"
-          enterTo="opacity-100 translate-y-0"
-          leave="transition-all ease-in-out duration-200"
-          leaveFrom="opacity-100"
-          leaveTo="opacity-0"
+          enter="transition ease duration-300 transform"
+          enterFrom="opacity-0 scale-95"
+          enterTo="opacity-100 scale-100"
+          leave="transition ease duration-300 transform"
+          leaveFrom="opacity-100 scale-100"
+          leaveTo="opacity-0 scale-95"
+          className="trasition-all duration-300 ease-in-out"
         >
           <Disclosure.Panel
             static
@@ -841,7 +776,7 @@ function Stage({
                 {children}
               </div>
             )}
-            {meta.status !== Status.fail && panel}
+            {meta?.status !== Status.fail && panel}
           </Disclosure.Panel>
         </Transition>
       </Disclosure>
diff --git a/web/client/src/library/components/plan/PlanHeader.tsx b/web/client/src/library/components/plan/PlanHeader.tsx
index f6d44df9..92b487e3 100644
--- a/web/client/src/library/components/plan/PlanHeader.tsx
+++ b/web/client/src/library/components/plan/PlanHeader.tsx
@@ -3,31 +3,33 @@ import { useStoreContext } from '~/context/context'
 import { EnumVariant } from '~/types/enum'
 import { Disclosure } from '@headlessui/react'
 import { MinusCircleIcon, PlusCircleIcon } from '@heroicons/react/24/solid'
-import { useStorePlan } from '@context/plan'
-import { isFalse } from '@utils/index'
 
 export default function PlanHeader(): JSX.Element {
   const environment = useStoreContext(s => s.environment)
 
-  const planApply = useStorePlan(s => s.planApply)
-
   return (
     <div className="flex flex-col px-4 py-2 w-full">
-      {environment.isInitialProd && isFalse(planApply.isFinished) && (
-        <Banner variant={EnumVariant.Warning}>
-          <Disclosure defaultOpen={false}>
-            {({ open }) => (
-              <>
-                <Disclosure.Button className="w-full flex items-center justify-between">
-                  <Banner.Label className="w-full">
-                    Initializing Prod Environment
-                  </Banner.Label>
-                  {open ? (
-                    <MinusCircleIcon className="w-5 text-warning-500" />
-                  ) : (
-                    <PlusCircleIcon className="w-5 text-warning-500" />
-                  )}
-                </Disclosure.Button>
+      <Banner variant={EnumVariant.Warning}>
+        <Disclosure defaultOpen={false}>
+          {({ open }) => (
+            <>
+              <Disclosure.Button className="w-full flex items-center justify-between">
+                <Banner.Label className="w-full">
+                  {environment.isInitialProd
+                    ? 'Initializing Prod Environment'
+                    : 'Prod Environment'}
+                </Banner.Label>
+                {environment.isInitialProd && (
+                  <>
+                    {open ? (
+                      <MinusCircleIcon className="w-5 text-warning-500" />
+                    ) : (
+                      <PlusCircleIcon className="w-5 text-warning-500" />
+                    )}
+                  </>
+                )}
+              </Disclosure.Button>
+              {environment.isInitialProd && (
                 <Disclosure.Panel className="px-2 text-sm mt-2">
                   <Banner.Description>
                     Prod will be completely backfilled in order to ensure there
@@ -36,11 +38,11 @@ export default function PlanHeader(): JSX.Element {
                     deploying to production.
                   </Banner.Description>
                 </Disclosure.Panel>
-              </>
-            )}
-          </Disclosure>
-        </Banner>
-      )}
+              )}
+            </>
+          )}
+        </Disclosure>
+      </Banner>
     </div>
   )
 }
diff --git a/web/client/src/library/components/plan/help.ts b/web/client/src/library/components/plan/help.ts
index e0425131..78b58dc7 100644
--- a/web/client/src/library/components/plan/help.ts
+++ b/web/client/src/library/components/plan/help.ts
@@ -1,6 +1,7 @@
 import { type ModelPlanOverviewTracker } from '@models/tracker-plan-overview'
 import { isArrayNotEmpty, isNotNil } from '../../../utils'
 import { type ModelPlanApplyTracker } from '@models/tracker-plan-apply'
+import { type ModelPlanCancelTracker } from '@models/tracker-plan-cancel'
 
 export function isModified<T extends object>(modified?: T): boolean {
   return Object.values(modified ?? {}).some(isArrayNotEmpty)
@@ -28,10 +29,12 @@ type PlanOverviewDetails = Pick<
 export function getPlanOverviewDetails(
   planApply: ModelPlanApplyTracker,
   planOverview: ModelPlanOverviewTracker,
+  planCancel: ModelPlanCancelTracker,
 ): PlanOverviewDetails {
   const isLatest =
-    planApply.isFinished &&
-    (planOverview.isLatest || planOverview.isRunning) &&
+    ((planApply.isFinished &&
+      (planOverview.isLatest || planOverview.isRunning)) ||
+      planCancel.isFinished) &&
     isNotNil(planApply.overview)
   const overview = isLatest ? planApply.overview : planOverview
   const plan = isLatest ? planApply : planOverview
diff --git a/web/client/src/library/components/report/ReportErrors.tsx b/web/client/src/library/components/report/ReportErrors.tsx
index 9aed97de..f5315d7f 100644
--- a/web/client/src/library/components/report/ReportErrors.tsx
+++ b/web/client/src/library/components/report/ReportErrors.tsx
@@ -40,8 +40,8 @@ export default function ReportErrors(): JSX.Element {
             className={clsx(
               'block ml-1 px-3 py-1 first-child:ml-0 rounded-full whitespace-nowrap font-bold text-xs text-center',
               hasError
-                ? 'border-danger-10 text-danger-500 bg-danger-10 cursor-pointer'
-                : 'border-neutral-10 cursor-default text-neutral-500 bg-neutral-10 dark:text-neutral-300',
+                ? 'text-danger-500 bg-danger-10 dark:bg-danger-20 cursor-pointer'
+                : 'bg-neutral-5 dark:bg-neutral-20 cursor-default text-neutral-500 dark:text-neutral-300',
             )}
           >
             {hasError ? (
diff --git a/web/client/src/library/components/tasksOverview/TasksOverview.tsx b/web/client/src/library/components/tasksOverview/TasksOverview.tsx
index de085495..12f5b0de 100644
--- a/web/client/src/library/components/tasksOverview/TasksOverview.tsx
+++ b/web/client/src/library/components/tasksOverview/TasksOverview.tsx
@@ -14,6 +14,7 @@ import { type EnvironmentName } from '~/models/environment'
 import { EnumPlanChangeType, type PlanChangeType } from '../plan/context'
 import Title from '@components/title/Title'
 import { type ModelSQLMeshChangeDisplay } from '@models/sqlmesh-change-display'
+import { Transition } from '@headlessui/react'
 
 const TasksOverview = function TasksOverview({
   children,
@@ -159,8 +160,17 @@ function TasksDetails({
 }): JSX.Element {
   return (
     <>
-      {isArrayNotEmpty(queue) && (
-        <div className="p-4 mt-6 shadow-lg rounded-lg">
+      <Transition
+        show={isArrayNotEmpty(queue)}
+        enter="transition ease duration-300 transform"
+        enterFrom="opacity-0 scale-95"
+        enterTo="opacity-100 scale-100"
+        leave="transition ease duration-300 transform"
+        leaveFrom="opacity-100 scale-100"
+        leaveTo="opacity-0 scale-95"
+        className="trasition-all duration-300 ease-in-out"
+      >
+        <div className="px-4 pt-3 pb-2 mt-4 shadow-lg rounded-lg">
           <Title text="Currently in proccess" />
           <Tasks models={queue}>
             {task => (
@@ -228,7 +238,7 @@ function TasksDetails({
             )}
           </Tasks>
         </div>
-      )}
+      </Transition>
       <TasksBlock className={className}>
         <Tasks models={models}>
           {task => (
@@ -343,7 +353,7 @@ function Tasks({
   return (
     <ul
       className={clsx(
-        'rounded-lg py-4 overflow-auto text-prose hover:scrollbar scrollbar--vertical scrollbar--horizontal',
+        'rounded-lg pt-4 overflow-auto text-prose hover:scrollbar scrollbar--vertical scrollbar--horizontal',
         className,
       )}
     >
diff --git a/web/client/src/library/pages/docs/Content.tsx b/web/client/src/library/pages/docs/Content.tsx
index fb51a547..866ba929 100644
--- a/web/client/src/library/pages/docs/Content.tsx
+++ b/web/client/src/library/pages/docs/Content.tsx
@@ -44,7 +44,7 @@ export default function Content(): JSX.Element {
     }
 
     setLastSelectedModel(model)
-  }, [model])
+  }, [model, model?.sql])
 
   const modelExtensions = isNil(model)
     ? []
@@ -100,7 +100,7 @@ export default function Content(): JSX.Element {
                     path={model.path}
                   >
                     {({ file }) => (
-                      <Tab.Group defaultIndex={model.isModelSQL ? 1 : 0}>
+                      <Tab.Group>
                         <TabList
                           list={
                             [
@@ -122,10 +122,7 @@ export default function Content(): JSX.Element {
                             />
                           </Tab.Panel>
                           {model.isModelSQL && (
-                            <Tab.Panel
-                              unmount={false}
-                              className="w-full h-full"
-                            >
+                            <Tab.Panel className="w-full h-full">
                               <CodeEditorDefault
                                 type={EnumFileExtensions.SQL}
                                 content={model.sql ?? ''}
diff --git a/web/client/src/library/pages/ide/IDE.tsx b/web/client/src/library/pages/ide/IDE.tsx
index 3ba7f110..bc9ad7a8 100644
--- a/web/client/src/library/pages/ide/IDE.tsx
+++ b/web/client/src/library/pages/ide/IDE.tsx
@@ -2,33 +2,18 @@ import React, { useEffect } from 'react'
 import {
   useApiModels,
   useApiFiles,
-  useApiEnvironments,
   useApiPlanRun,
   useApiPlanApply,
   useApiCancelPlan,
 } from '../../../api'
 import { useChannelEvents } from '../../../api/channels'
-import {
-  isArrayEmpty,
-  isFalse,
-  isNil,
-  isNotNil,
-  isObjectNotEmpty,
-  isTrue,
-} from '~/utils'
+import { includes, isArrayEmpty, isFalse, isNil, isNotNil } from '~/utils'
 import { useStoreContext } from '~/context/context'
 import { Outlet, useLocation, useNavigate } from 'react-router-dom'
 import { EnumRoutes } from '~/routes'
 import { type Tests, useStoreProject } from '@context/project'
 import { EnumErrorKey, type ErrorIDE, useIDE } from './context'
-import {
-  type Directory,
-  type Model,
-  type Environments,
-  type EnvironmentsEnvironments,
-  Status,
-  Modules,
-} from '@api/client'
+import { type Directory, type Model, Status, Modules } from '@api/client'
 import { Button } from '@components/button/Button'
 import Container from '@components/container/Container'
 import { useStoreEditor, createLocalFile } from '@context/editor'
@@ -41,12 +26,11 @@ import {
   EnumFileExplorerChange,
   type FileExplorerChange,
 } from '@components/fileExplorer/context'
-import { type PlanOverviewTracker } from '@models/tracker-plan-overview'
-import { type PlanApplyTracker } from '@models/tracker-plan-apply'
-import { type PlanCancelTracker } from '@models/tracker-plan-cancel'
 import { EnumPlanAction, ModelPlanAction } from '@models/plan-action'
 import { useStorePlan } from '@context/plan'
 
+let planActionUpdateTimeoutId: Optional<number>
+
 export default function PageIDE(): JSX.Element {
   const location = useLocation()
   const navigate = useNavigate()
@@ -60,8 +44,8 @@ export default function PageIDE(): JSX.Element {
   const confirmations = useStoreContext(s => s.confirmations)
   const removeConfirmation = useStoreContext(s => s.removeConfirmation)
   const environment = useStoreContext(s => s.environment)
+
   const setModels = useStoreContext(s => s.setModels)
-  const addRemoteEnvironments = useStoreContext(s => s.addRemoteEnvironments)
 
   const planOverview = useStorePlan(s => s.planOverview)
   const planApply = useStorePlan(s => s.planApply)
@@ -95,8 +79,7 @@ export default function PageIDE(): JSX.Element {
   // all pages have access to models and files
   const { refetch: getModels, cancel: cancelRequestModels } = useApiModels()
   const { refetch: getFiles, cancel: cancelRequestFiles } = useApiFiles()
-  const { refetch: getEnvironments, cancel: cancelRequestEnvironments } =
-    useApiEnvironments()
+
   const { isFetching: isFetchingPlanApply } = useApiPlanApply(environment.name)
   const { refetch: cancelPlanOrPlanApply, isFetching: isFetchingPlanCancel } =
     useApiCancelPlan()
@@ -115,15 +98,7 @@ export default function PageIDE(): JSX.Element {
     const channelModels = channel<Model[]>('models', updateModels)
     const channelTests = channel<Tests>('tests', updateTests)
     const channelErrors = channel<ErrorIDE>('errors', displayErrors)
-    const channelPlanOverview = channel<any>(
-      'plan-overview',
-      updatePlanOverviewTracker,
-    )
-    const channelPlanApply = channel<any>('plan-apply', updatePlanApplyTracker)
-    const channelPlanCancel = channel<any>(
-      'plan-cancel',
-      updatePlanCancelTracker,
-    )
+
     const channelFile = channel<{
       changes: Array<{
         change: FileExplorerChange
@@ -226,8 +201,6 @@ export default function PageIDE(): JSX.Element {
       setProject(project)
     })
 
-    void getEnvironments().then(({ data }) => updateEnviroments(data))
-
     void getModels().then(({ data }) => {
       updateModels(data as Model[])
     })
@@ -235,26 +208,19 @@ export default function PageIDE(): JSX.Element {
     channelFormatFile.subscribe()
     channelModels.subscribe()
     channelErrors.subscribe()
-    channelPlanOverview.subscribe()
-    channelPlanApply.subscribe()
     channelFile.subscribe()
-    channelPlanCancel.subscribe()
     channelTests.subscribe()
 
     return () => {
       void cancelRequestModels()
       void cancelRequestFiles()
-      void cancelRequestEnvironments()
       void cancelRequestPlan()
 
       channelFormatFile.unsubscribe()
       channelTests.unsubscribe()
       channelModels.unsubscribe()
       channelErrors.unsubscribe()
-      channelPlanOverview.unsubscribe()
-      channelPlanApply.unsubscribe()
       channelFile.unsubscribe()
-      channelPlanCancel.unsubscribe()
     }
   }, [])
 
@@ -280,30 +246,35 @@ export default function PageIDE(): JSX.Element {
       isFalse(planAction.isRunningTask)
     ) {
       planApply.reset()
+      planCancel.reset()
 
       void planRun()
     }
   }, [models, environment])
 
   useEffect(() => {
-    planOverview.isFetching = isFetchingPlanRun
+    if (isFetchingPlanRun) {
+      planOverview.isFetching = true
 
-    setPlanOverview(planOverview)
-  }, [isFetchingPlanRun])
+      setPlanOverview(planOverview)
+    }
 
-  useEffect(() => {
-    planOverview.isFetching = isFetchingPlanApply
+    if (isFetchingPlanApply) {
+      planApply.isFetching = true
 
-    setPlanApply(planApply)
-  }, [isFetchingPlanApply])
+      setPlanApply(planApply)
+    }
 
-  useEffect(() => {
-    planOverview.isFetching = isFetchingPlanCancel
+    if (isFetchingPlanCancel) {
+      planCancel.isFetching = true
 
-    setPlanCancel(planCancel)
-  }, [isFetchingPlanCancel])
+      setPlanCancel(planCancel)
+    }
+  }, [isFetchingPlanRun, isFetchingPlanApply, isFetchingPlanCancel])
 
   useEffect(() => {
+    clearTimeout(planActionUpdateTimeoutId)
+
     const value = ModelPlanAction.getPlanAction({
       planOverview,
       planApply,
@@ -312,9 +283,31 @@ export default function PageIDE(): JSX.Element {
 
     // This inconsistency can happen when we receive flag from the backend
     // that some task is runninng but not yet recived SSE event with data
-    if (value === EnumPlanAction.Done && planAction.isRunningTask) return
+    if (
+      (value === EnumPlanAction.Done && planAction.isRunningTask) ||
+      value === planAction.value
+    )
+      return
+
+    // Delaying the update of the plan action to avoid flickering
+    planActionUpdateTimeoutId = setTimeout(
+      () => {
+        setPlanAction(new ModelPlanAction({ value }))
 
-    setPlanAction(new ModelPlanAction({ value }))
+        planActionUpdateTimeoutId = undefined
+      },
+      includes(
+        [
+          EnumPlanAction.Running,
+          EnumPlanAction.RunningTask,
+          EnumPlanAction.Cancelling,
+          EnumPlanAction.Applying,
+        ],
+        value,
+      )
+        ? 0
+        : 500,
+    ) as unknown as number
   }, [planOverview, planApply, planCancel])
 
   function updateModels(models?: Model[]): void {
@@ -336,48 +329,6 @@ export default function PageIDE(): JSX.Element {
     setTests(tests)
   }
 
-  function updatePlanOverviewTracker(data: PlanOverviewTracker): void {
-    planOverview.update(data)
-
-    setPlanOverview(planOverview)
-  }
-
-  function updatePlanCancelTracker(data: PlanCancelTracker): void {
-    planCancel.update(data)
-
-    setPlanCancel(planCancel)
-  }
-
-  function updatePlanApplyTracker(data: PlanApplyTracker): void {
-    planApply.update(data, planOverview)
-
-    const isFinished =
-      isTrue(data.meta?.done) && data.meta?.status !== Status.init
-
-    if (isFinished) {
-      void getEnvironments().then(({ data }) => {
-        void planRun()
-
-        updateEnviroments(data)
-      })
-    }
-
-    setPlanApply(planApply)
-  }
-
-  function updateEnviroments(data: Optional<Environments>): void {
-    const { environments, default_target_environment, pinned_environments } =
-      data ?? {}
-
-    if (isObjectNotEmpty<EnvironmentsEnvironments>(environments)) {
-      addRemoteEnvironments(
-        Object.values(environments),
-        default_target_environment,
-        pinned_environments,
-      )
-    }
-  }
-
   function restoreEditorTabsFromSaved(files: ModelFile[]): void {
     if (isArrayEmpty(storedTabs)) return
 
diff --git a/web/client/src/library/pages/root/Page.tsx b/web/client/src/library/pages/root/Page.tsx
index 90dc20f0..e55aeb9b 100644
--- a/web/client/src/library/pages/root/Page.tsx
+++ b/web/client/src/library/pages/root/Page.tsx
@@ -58,9 +58,10 @@ import {
 import { Menu, Transition, Popover } from '@headlessui/react'
 import { type ModelEnvironment } from '@models/environment'
 import { type ModelSQLMeshChangeDisplay } from '@models/sqlmesh-change-display'
-import { Fragment, useState, type MouseEvent } from 'react'
+import { Fragment, useState, type MouseEvent, useEffect } from 'react'
 import Input from '@components/input/Input'
 import ReportErrors from '@components/report/ReportErrors'
+import { ModelPlanOverviewTracker } from '@models/tracker-plan-overview'
 
 export default function Page({
   sidebar,
@@ -154,81 +155,73 @@ function EnvironmentDetails(): JSX.Element {
       ) : (
         <>
           {withPlanModule && (
-            <>
-              {planAction.isProcessing ? (
-                <ModuleLink
-                  title={planAction.displayStatus(planOverview)}
-                  to={EnumRoutes.Plan}
-                  icon={
-                    <Spinner
-                      variant={EnumVariant.Success}
-                      className="w-3"
-                    />
-                  }
-                  iconActive={
-                    <Spinner
-                      variant={EnumVariant.Success}
-                      className="w-3"
-                    />
-                  }
-                  before={
-                    <p className="block mx-1 text-xs text-success-100 font-bold">
-                      {planAction.displayStatus(planOverview)}
-                    </p>
-                  }
-                  className="px-2 py-1 bg-success-500"
-                  classActive="bg-success-500"
-                />
-              ) : (
-                <ModuleLink
-                  title="Plan"
-                  to={EnumRoutes.Plan}
-                  icon={
-                    <OutlinePlayCircleIcon
-                      className={clsx(
-                        'w-5',
-                        planOverview.isFailed
-                          ? 'text-danger-500 dark:text-danger-100'
-                          : 'text-success-100',
-                      )}
-                    />
-                  }
-                  iconActive={
-                    <PlayCircleIcon
-                      className={clsx(
-                        'w-5',
-                        planOverview.isFailed
-                          ? 'text-danger-500 dark:text-danger-100'
-                          : 'text-success-100',
-                      )}
-                    />
-                  }
-                  before={
-                    <p
-                      className={clsx(
-                        'block mx-1 text-xs font-bold',
-                        planOverview.isFailed
-                          ? 'text-danger-500 dark:text-danger-100'
-                          : 'text-success-100',
-                      )}
-                    >
-                      Plan
-                    </p>
-                  }
+            <ModuleLink
+              title={
+                planAction.isProcessing
+                  ? planAction.displayStatus(planOverview)
+                  : 'Plan'
+              }
+              to={EnumRoutes.Plan}
+              icon={
+                planAction.isProcessing ? (
+                  <Spinner
+                    variant={EnumVariant.Success}
+                    className="w-3 py-1"
+                  />
+                ) : (
+                  <OutlinePlayCircleIcon
+                    className={clsx(
+                      'w-5',
+                      planOverview.isFailed
+                        ? 'text-danger-500 dark:text-danger-100'
+                        : 'text-success-100',
+                    )}
+                  />
+                )
+              }
+              iconActive={
+                planAction.isProcessing ? (
+                  <Spinner
+                    variant={EnumVariant.Success}
+                    className="w-3 py-1"
+                  />
+                ) : (
+                  <PlayCircleIcon
+                    className={clsx(
+                      'w-5',
+                      planOverview.isFailed
+                        ? 'text-danger-500 dark:text-danger-100'
+                        : 'text-success-100',
+                    )}
+                  />
+                )
+              }
+              before={
+                <p
                   className={clsx(
-                    'px-2',
-                    planOverview.isFailed
-                      ? 'bg-danger-10 dark:bg-danger-500'
-                      : 'bg-success-500',
-                  )}
-                  classActive={clsx(
+                    'block mx-1 text-xs font-bold',
                     planOverview.isFailed
-                      ? 'bg-danger-10 dark:bg-danger-500'
-                      : 'bg-success-500',
+                      ? 'text-danger-500 dark:text-danger-100'
+                      : 'text-success-100',
                   )}
-                />
+                >
+                  {planAction.isProcessing
+                    ? planAction.displayStatus(planOverview)
+                    : 'Plan'}
+                </p>
+              }
+              className={clsx(
+                'px-2',
+                planOverview.isFailed
+                  ? 'bg-danger-10 dark:bg-danger-500'
+                  : 'bg-success-500',
               )}
-            </>
+              classActive={clsx(
+                planOverview.isFailed
+                  ? 'bg-danger-10 dark:bg-danger-500'
+                  : 'bg-success-500',
+              )}
+            />
           )}
           <SelectEnvironemnt
             className="border-none h-6 "
@@ -240,8 +233,9 @@ function EnvironmentDetails(): JSX.Element {
               environment.isInitialProd
             }
           />
-          {isFalse(planOverview.isLatest) &&
-            isTrue(planOverview.hasChanges) && <PlanChanges />}
+          {[Modules['plan-progress'], Modules.plans].some(m =>
+            modules.includes(m),
+          ) && <PlanChanges />}
           {modules.includes(Modules.errors) && <ReportErrors />}
         </>
       )}
@@ -383,62 +377,79 @@ function ActionStatus({
 function PlanChanges(): JSX.Element {
   const planOverview = useStorePlan(s => s.planOverview)
 
-  return (
+  const [planOverviewTracker, setPlanOverviewTracker] =
+    useState<ModelPlanOverviewTracker>(planOverview)
+
+  useEffect(() => {
+    if (isFalse(planOverview.isEmpty) && planOverview.isFinished) {
+      setPlanOverviewTracker(new ModelPlanOverviewTracker(planOverview))
+    }
+  }, [planOverview])
+
+  const shouldShow =
+    (isFalse(planOverview.isLatest) || planOverview.isFetching) &&
+    planOverviewTracker.hasUpdates
+
+  return shouldShow ? (
     <span className="flex group items-center bg-neutral-10 px-1 py-1 rounded-full">
-      {isTrue(planOverview.hasChanges) && (
+      {isTrue(planOverviewTracker.hasChanges) && (
         <p className="flex text-xs ml-2 mr-2 dark:text-neutral-300">Changes</p>
       )}
-      {isArrayNotEmpty(planOverview.added) && (
+      {isArrayNotEmpty(planOverviewTracker.added) && (
         <ChangesPreview
           headline="Added"
           type={EnumPlanChangeType.Add}
-          changes={planOverview.added}
+          changes={planOverviewTracker.added}
           className="-mr-2 group-hover:mr-0 z-[5]"
         />
       )}
-      {isArrayNotEmpty(planOverview.direct) && (
+      {isArrayNotEmpty(planOverviewTracker.direct) && (
         <ChangesPreview
           headline="Directly Modified"
           type={EnumPlanChangeType.Direct}
-          changes={planOverview.direct}
+          changes={planOverviewTracker.direct}
           className="-mr-2 group-hover:mr-0 z-[4]"
         />
       )}
-      {isArrayNotEmpty(planOverview.indirect) && (
+      {isArrayNotEmpty(planOverviewTracker.indirect) && (
         <ChangesPreview
           headline="Indirectly Modified"
           type={EnumPlanChangeType.Indirect}
-          changes={planOverview.indirect}
+          changes={planOverviewTracker.indirect}
           className="-mr-2 group-hover:mr-0 z-[3]"
         />
       )}
-      {isArrayNotEmpty(planOverview.metadata) && (
+      {isArrayNotEmpty(planOverviewTracker.metadata) && (
         <ChangesPreview
           headline="Metadata"
           type={EnumPlanChangeType.Metadata}
-          changes={planOverview.metadata}
+          changes={planOverviewTracker.metadata}
           className="-mr-2 group-hover:mr-0 z-[2]"
         />
       )}
-      {isArrayNotEmpty(planOverview.removed) && (
+      {isArrayNotEmpty(planOverviewTracker.removed) && (
         <ChangesPreview
           headline="Removed"
           type={EnumPlanChangeType.Remove}
-          changes={planOverview.removed}
+          changes={planOverviewTracker.removed}
           className="-mr-2 group-hover:mr-0 z-[1]"
         />
       )}
-      {isArrayNotEmpty(planOverview.backfills) ? (
+      {isArrayNotEmpty(planOverviewTracker.backfills) ? (
         <ChangesPreview
           headline="Backfills"
           type={EnumPlanChangeType.Default}
-          changes={planOverview.backfills}
+          changes={planOverviewTracker.backfills}
           className="ml-4 group-hover:ml-2 z-[6]"
         />
       ) : (
         <div className="ml-2 group-hover:ml-2 z-[6]"></div>
       )}
     </span>
+  ) : (
+    <span className="block ml-1 px-3 py-1 first-child:ml-0 rounded-full whitespace-nowrap font-bold text-xs text-center bg-neutral-5 dark:bg-neutral-20 cursor-default text-neutral-500 dark:text-neutral-300">
+      No Changes
+    </span>
   )
 }
 
diff --git a/web/client/src/models/plan-action.ts b/web/client/src/models/plan-action.ts
index a0fbddd6..c4af976e 100644
--- a/web/client/src/models/plan-action.ts
+++ b/web/client/src/models/plan-action.ts
@@ -106,13 +106,14 @@ export class ModelPlanAction<
   }
 
   displayStatus(planOverview: ModelPlanOverviewTracker): string {
-    if (this.isRunningTask) return 'Running Task...'
+    if (this.isCancelling) return 'Cancelling Plan...'
     if (this.isApplying) return 'Applying Plan...'
     if (this.isRunning && isNil(planOverview.hasChanges))
       return 'Getting Changes...'
     if (this.isRunning && isNil(planOverview.hasBackfills))
       return 'Getting Backfills...'
     if (this.isRunning) return 'Checking Plan...'
+    if (this.isRunningTask) return 'Running Task...'
 
     return 'Plan'
   }
@@ -177,10 +178,13 @@ export class ModelPlanAction<
     const isRunningPlan = planOverview.isRunning
     const isRunningApply = planApply.isRunning
     const isRunningCancel = planCancel.isRunning
+    const isFinished =
+      (planApply.isFinished && planOverview.isFinished) || planCancel.isFinished
 
-    if (isRunningPlan) return EnumPlanAction.Running
     if (isRunningCancel) return EnumPlanAction.Cancelling
     if (isRunningApply) return EnumPlanAction.Applying
+    if (isRunningPlan) return EnumPlanAction.Running
+    if (isFinished) return EnumPlanAction.Done
     if (planOverview.isVirtualUpdate) return EnumPlanAction.ApplyVirtual
     if (planOverview.isMetadataUpdate) return EnumPlanAction.ApplyMetadata
     if (planOverview.isBackfillUpdate) return EnumPlanAction.ApplyBackfill
diff --git a/web/client/src/models/tracker-plan-cancel.ts b/web/client/src/models/tracker-plan-cancel.ts
index 9098a7e5..d49f5365 100644
--- a/web/client/src/models/tracker-plan-cancel.ts
+++ b/web/client/src/models/tracker-plan-cancel.ts
@@ -13,8 +13,9 @@ export class ModelPlanCancelTracker extends ModelPlanTracker<PlanCancelTracker>
 
   get isCancelling(): boolean {
     return (
-      isFalseOrNil(this._current?.cancel?.meta?.done) &&
-      this._current?.cancel?.meta?.status === Status.init
+      (isFalseOrNil(this.current?.meta?.done) &&
+        this.current?.meta?.status === Status.init) ||
+      this.isFetching
     )
   }
 
diff --git a/web/client/src/models/tracker-plan-overview.ts b/web/client/src/models/tracker-plan-overview.ts
index 5fe8fca7..2ce1cb0d 100644
--- a/web/client/src/models/tracker-plan-overview.ts
+++ b/web/client/src/models/tracker-plan-overview.ts
@@ -4,7 +4,7 @@ import {
   type PlanStageChanges,
 } from '@api/client'
 import { ModelPlanTracker, type PlanTracker } from './tracker-plan'
-import { isArrayNotEmpty, isFalse, isNil, isNotNil } from '@utils/index'
+import { isArrayNotEmpty, isFalse, isNil, isNotNil, isTrue } from '@utils/index'
 import {
   type InitialChangeDisplay,
   ModelSQLMeshChangeDisplay,
@@ -82,6 +82,10 @@ export class ModelPlanOverviewTracker
       : isArrayNotEmpty(this._current.backfills?.models)
   }
 
+  get hasUpdates(): boolean {
+    return isTrue(this.hasChanges) || isTrue(this.hasBackfills)
+  }
+
   get backfills(): ModelSQLMeshChangeDisplay[] {
     return this._backfills
   }
diff --git a/web/server/api/endpoints/commands.py b/web/server/api/endpoints/commands.py
index 6127aade..a059bf48 100644
--- a/web/server/api/endpoints/commands.py
+++ b/web/server/api/endpoints/commands.py
@@ -198,19 +198,19 @@ def _run_plan_apply(
 ) -> None:
     """Run plan apply"""
     plan_options = plan_options or models.PlanOptions()
-
+    tracker_apply = models.PlanApplyStageTracker(environment=environment, plan_options=plan_options)
+    api_console.start_plan_tracker(tracker_apply)
     plan_builder = get_plan_builder(context, plan_options, environment, plan_dates)
     plan = plan_builder.build()
+    tracker_apply.start = plan.start
+    tracker_apply.end = plan.end
+    api_console.start_plan_tracker(tracker_apply)
 
     if categories is not None:
         for new, _ in plan.context_diff.modified_snapshots.values():
             if plan.is_new_snapshot(new) and new.name in categories:
                 plan_builder.set_choice(new, categories[new.name])
         plan = plan_builder.build()
-
-    tracker_apply = models.PlanApplyStageTracker(environment=environment, plan_options=plan_options)
-    tracker_apply.start = plan.start
-    tracker_apply.end = plan.end
     api_console.start_plan_tracker(tracker_apply)
     try:
         context.apply(plan, circuit_breaker)
diff --git a/web/server/api/endpoints/files.py b/web/server/api/endpoints/files.py
index c538a04e..9a1f0bba 100644
--- a/web/server/api/endpoints/files.py
+++ b/web/server/api/endpoints/files.py
@@ -67,18 +67,22 @@ async def write_file(
         replace_file(settings.project_path / path, settings.project_path / path_or_new_path)
     else:
         full_path = settings.project_path / path
-        format_file_status = models.FormatFileStatus(
-            status=models.Status.INIT, path=path_or_new_path
-        )
-        if content and Path(path_or_new_path).suffix == ".sql":
+        config = context.config_for_path(Path(path_or_new_path))
+        if config.ui.format_on_save and content and Path(path_or_new_path).suffix == ".sql":
+            format_file_status = models.FormatFileStatus(
+                status=models.Status.INIT, path=path_or_new_path
+            )
             path_to_model_mapping = await get_path_to_model_mapping(settings=settings)
             model = path_to_model_mapping.get(Path(full_path))
-            default_dialect = context.config_for_path(Path(path_or_new_path)).dialect
+            default_dialect = config.dialect
             dialect = model.dialect if model and model.is_sql else default_dialect
-
             try:
                 expressions = parse(content, default_dialect=default_dialect)
-                content = format_model_expressions(expressions, dialect)
+                content = format_model_expressions(
+                    expressions, dialect, **config.format.generator_options
+                )
+                if config.format.append_newline:
+                    content += "\n"
                 format_file_status.status = models.Status.SUCCESS
             except Exception:
                 format_file_status.status = models.Status.FAIL
@@ -87,7 +91,9 @@ async def write_file(
                     origin="API -> files -> write_file",
                 ).to_dict()
                 api_console.log_event(event=models.EventName.WARNINGS, data=error)
-        api_console.log_event(event=models.EventName.FORMAT_FILE, data=format_file_status.dict())
+            api_console.log_event(
+                event=models.EventName.FORMAT_FILE, data=format_file_status.dict()
+            )
         full_path.write_text(content, encoding="utf-8")
 
     response.status_code = HTTP_204_NO_CONTENT
diff --git a/web/server/api/endpoints/plan.py b/web/server/api/endpoints/plan.py
index cfd93eb6..2a467128 100644
--- a/web/server/api/endpoints/plan.py
+++ b/web/server/api/endpoints/plan.py
@@ -62,16 +62,17 @@ async def cancel_plan(
             origin="API -> plan -> cancel_plan",
         )
 
+    tracker = models.PlanCancelStageTracker()
+    api_console.start_plan_tracker(tracker)
+    tracker_stage_cancel = models.PlanStageCancel()
+    tracker.add_stage(stage=models.PlanStage.cancel, data=tracker_stage_cancel)
     request.app.state.circuit_breaker.set()
     request.app.state.task.cancel()
     try:
         await request.app.state.task
     except asyncio.CancelledError:
         pass
-    tracker = models.PlanCancelStageTracker()
-    api_console.start_plan_tracker(tracker)
-    tracker_stage_cancel = models.PlanStageCancel()
-    tracker.add_stage(stage=models.PlanStage.cancel, data=tracker_stage_cancel)
+
     tracker_stage_cancel.stop(success=True)
     api_console.stop_plan_tracker(tracker)
     response.status_code = HTTP_204_NO_CONTENT
diff --git a/web/server/console.py b/web/server/console.py
index 9485f5e1..c4cb06e2 100644
--- a/web/server/console.py
+++ b/web/server/console.py
@@ -201,13 +201,14 @@ class ApiConsole(TerminalConsole):
             self.plan_apply_stage_tracker.stop(success=success)
             self.log_event_plan_apply()
             self.plan_apply_stage_tracker = None
+            self.plan_overview_stage_tracker = None
         elif (
             isinstance(tracker, models.PlanOverviewStageTracker)
             and self.plan_overview_stage_tracker
         ):
             self.plan_overview_stage_tracker.stop(success=success)
             self.log_event_plan_overview()
-            self.plan_overview_stage_tracker = None
+            self.plan_apply_stage_tracker = None
         elif isinstance(tracker, models.PlanCancelStageTracker) and self.plan_cancel_stage_tracker:
             self.plan_cancel_stage_tracker.stop(success=success)
             self.log_event_plan_cancel()
@@ -267,9 +268,9 @@ class ApiConsole(TerminalConsole):
     def log_event_plan_overview(self) -> None:
         self.log_event(
             event=models.EventName.PLAN_OVERVIEW,
-            data=self.plan_overview_stage_tracker.dict()
-            if self.plan_overview_stage_tracker
-            else {},
+            data=(
+                self.plan_overview_stage_tracker.dict() if self.plan_overview_stage_tracker else {}
+            ),
         )
 
     def log_event_plan_cancel(self) -> None:


Test patch:
diff --git a/sqlmesh/core/test/result.py b/sqlmesh/core/test/result.py
index 18a3ea3c..5a8da783 100644
--- a/sqlmesh/core/test/result.py
+++ b/sqlmesh/core/test/result.py
@@ -8,8 +8,9 @@ class ModelTextTestResult(unittest.TextTestResult):
     def addFailure(
         self,
         test: unittest.TestCase,
-        err: tuple[type[BaseException], BaseException, types.TracebackType]
-        | tuple[None, None, None],
+        err: (
+            tuple[type[BaseException], BaseException, types.TracebackType] | tuple[None, None, None]
+        ),
     ) -> None:
         """Called when the test case test signals a failure.
 
diff --git a/tests/core/engine_adapter/test_base.py b/tests/core/engine_adapter/test_base.py
index 2cbb6641..b82a6746 100644
--- a/tests/core/engine_adapter/test_base.py
+++ b/tests/core/engine_adapter/test_base.py
@@ -15,7 +15,7 @@ from sqlmesh.core.engine_adapter import EngineAdapter, EngineAdapterWithIndexSup
 from sqlmesh.core.engine_adapter.shared import InsertOverwriteStrategy
 from sqlmesh.core.schema_diff import SchemaDiffer, TableAlterOperation
 from sqlmesh.utils.date import to_ds
-from sqlmesh.utils.errors import UnsupportedCatalogOperationError
+from sqlmesh.utils.errors import SQLMeshError, UnsupportedCatalogOperationError
 from tests.core.engine_adapter import to_sql_calls
 
 pytestmark = pytest.mark.engine
@@ -454,14 +454,14 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
 
     adapter_no_comments.create_table(
         "test_table",
-        {"a": "int", "b": "int"},
+        {"a": exp.DataType.build("int"), "b": exp.DataType.build("int")},
         table_description="test description",
         column_descriptions={"a": "a description"},
     )
 
     sql_calls = to_sql_calls(adapter_no_comments)
     assert sql_calls == [
-        """CREATE TABLE IF NOT EXISTS "test_table" ("a" int, "b" int)""",
+        """CREATE TABLE IF NOT EXISTS "test_table" ("a" INT, "b" INT)""",
     ]
 
 
@@ -1371,6 +1371,70 @@ def test_replace_query_pandas(make_mocked_engine_adapter: t.Callable):
     ]
 
 
+def test_replace_query_self_referencing_not_exists_unknown(
+    make_mocked_engine_adapter: t.Callable, mocker: MockerFixture
+):
+    adapter = make_mocked_engine_adapter(EngineAdapter)
+
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.base.EngineAdapter.table_exists",
+        return_value=False,
+    )
+
+    with pytest.raises(
+        SQLMeshError,
+        match="Cannot create a table without knowing the column types.*",
+    ):
+        adapter.replace_query(
+            "test",
+            parse_one("SELECT a FROM test"),
+            columns_to_types={"a": exp.DataType.build("UNKNOWN")},
+        )
+
+
+def test_replace_query_self_referencing_exists(
+    make_mocked_engine_adapter: t.Callable, mocker: MockerFixture
+):
+    adapter = make_mocked_engine_adapter(EngineAdapter)
+
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.base.EngineAdapter.table_exists",
+        return_value=True,
+    )
+
+    adapter.replace_query(
+        "test",
+        parse_one("SELECT a FROM test"),
+        columns_to_types={"a": exp.DataType.build("UNKNOWN")},
+    )
+
+    assert to_sql_calls(adapter) == [
+        'CREATE OR REPLACE TABLE "test" AS SELECT "a" FROM "test"',
+    ]
+
+
+def test_replace_query_self_referencing_not_exists_known(
+    make_mocked_engine_adapter: t.Callable, mocker: MockerFixture
+):
+    adapter = make_mocked_engine_adapter(EngineAdapter)
+
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.base.EngineAdapter.table_exists",
+        return_value=False,
+    )
+
+    adapter.replace_query(
+        "test",
+        parse_one("SELECT a FROM test"),
+        columns_to_types={"a": exp.DataType.build("INT")},
+    )
+
+    assert to_sql_calls(adapter) == [
+        'CREATE TABLE IF NOT EXISTS "test" ("a" INT)',
+        'CREATE OR REPLACE TABLE "test" AS SELECT "a" FROM "test"',
+    ]
+
+
 def test_create_table_like(make_mocked_engine_adapter: t.Callable):
     adapter = make_mocked_engine_adapter(EngineAdapter)
 
diff --git a/tests/core/engine_adapter/test_bigquery.py b/tests/core/engine_adapter/test_bigquery.py
index a9019dd4..dd003c71 100644
--- a/tests/core/engine_adapter/test_bigquery.py
+++ b/tests/core/engine_adapter/test_bigquery.py
@@ -317,7 +317,7 @@ def test_create_table_date_partition(
     )
     adapter.create_table(
         "test_table",
-        {"a": "int", "b": "int"},
+        {"a": exp.DataType.build("int"), "b": exp.DataType.build("int")},
         partitioned_by=partition_by_cols,
         partition_interval_unit=IntervalUnit.DAY,
         clustered_by=["b"],
@@ -325,7 +325,7 @@ def test_create_table_date_partition(
 
     sql_calls = _to_sql_calls(execute_mock)
     assert sql_calls == [
-        f"CREATE TABLE IF NOT EXISTS `test_table` (`a` int, `b` int) PARTITION BY {partition_by_statement} CLUSTER BY `b`"
+        f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64, `b` INT64) PARTITION BY {partition_by_statement} CLUSTER BY `b`"
     ]
 
 
@@ -572,7 +572,7 @@ def test_create_table_table_options(make_mocked_engine_adapter: t.Callable, mock
 
     adapter.create_table(
         "test_table",
-        {"a": "int", "b": "int"},
+        {"a": exp.DataType.build("int"), "b": exp.DataType.build("int")},
         table_properties={
             "partition_expiration_days": exp.convert(7),
         },
@@ -580,7 +580,7 @@ def test_create_table_table_options(make_mocked_engine_adapter: t.Callable, mock
 
     sql_calls = _to_sql_calls(execute_mock)
     assert sql_calls == [
-        "CREATE TABLE IF NOT EXISTS `test_table` (`a` int, `b` int) OPTIONS (partition_expiration_days=7)"
+        "CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64, `b` INT64) OPTIONS (partition_expiration_days=7)"
     ]
 
 
diff --git a/tests/core/engine_adapter/test_integration.py b/tests/core/engine_adapter/test_integration.py
index c5f0a9b6..6a149bee 100644
--- a/tests/core/engine_adapter/test_integration.py
+++ b/tests/core/engine_adapter/test_integration.py
@@ -861,9 +861,9 @@ def test_replace_query(ctx: TestContext):
         ctx.engine_adapter.replace_query(
             table,
             ctx.input_data(replace_data),
-            columns_to_types=ctx.columns_to_types
-            if ctx.dialect in ["spark", "databricks"]
-            else None,
+            columns_to_types=(
+                ctx.columns_to_types if ctx.dialect in ["spark", "databricks"] else None
+            ),
         )
         results = ctx.get_metadata_results()
         assert len(results.views) == 0
@@ -914,9 +914,9 @@ def test_replace_query_batched(ctx: TestContext):
         ctx.engine_adapter.replace_query(
             table,
             ctx.input_data(replace_data),
-            columns_to_types=ctx.columns_to_types
-            if ctx.dialect in ["spark", "databricks"]
-            else None,
+            columns_to_types=(
+                ctx.columns_to_types if ctx.dialect in ["spark", "databricks"] else None
+            ),
         )
         results = ctx.get_metadata_results()
         assert len(results.views) == 0
@@ -1491,9 +1491,7 @@ def test_sushi(ctx: TestContext):
         ) -> None:
             layer_objects = context.engine_adapter.get_data_objects(schema_name)
             layer_models = {
-                x.name.split("__")[1]
-                if is_physical_layer
-                else x.name: {
+                x.name.split("__")[1] if is_physical_layer else x.name: {
                     "table_name": x.name,
                     "is_view": x.type == DataObjectType.VIEW,
                 }
@@ -1572,9 +1570,7 @@ def test_sushi(ctx: TestContext):
         ) -> None:
             layer_objects = context.engine_adapter.get_data_objects(schema_name)
             layer_models = {
-                x.name.split("__")[1]
-                if is_physical_layer
-                else x.name: {
+                x.name.split("__")[1] if is_physical_layer else x.name: {
                     "table_name": x.name,
                     "is_view": x.type == DataObjectType.VIEW,
                 }
diff --git a/tests/core/engine_adapter/test_spark.py b/tests/core/engine_adapter/test_spark.py
index 5097a963..c4216b8b 100644
--- a/tests/core/engine_adapter/test_spark.py
+++ b/tests/core/engine_adapter/test_spark.py
@@ -238,10 +238,6 @@ def test_replace_query_pandas_exists(make_mocked_engine_adapter: t.Callable, moc
 def test_replace_query_self_ref_not_exists(
     make_mocked_engine_adapter: t.Callable, mocker: MockerFixture, make_temp_table_name: t.Callable
 ):
-    mocker.patch(
-        "sqlmesh.core.engine_adapter.spark.SparkEngineAdapter.table_exists",
-        return_value=False,
-    )
     mocker.patch(
         "sqlmesh.core.engine_adapter.spark.SparkEngineAdapter.get_current_catalog",
         lambda self: "spark_catalog",
@@ -266,13 +262,25 @@ def test_replace_query_self_ref_not_exists(
         return_value={"col": exp.DataType(this=exp.DataType.Type.INT)},
     )
 
+    def check_table_exists(table_name: exp.Table) -> bool:
+        for sql in to_sql_calls(adapter):
+            if f"CREATE TABLE IF NOT EXISTS {table_name.sql(dialect=adapter.dialect)}" in sql:
+                return True
+        return False
+
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.spark.SparkEngineAdapter.table_exists",
+        side_effect=check_table_exists,
+    )
+
     adapter.replace_query(table_name, parse_one(f"SELECT col + 1 AS col FROM {table_name}"))
 
     assert to_sql_calls(adapter) == [
         "CREATE TABLE IF NOT EXISTS `db`.`table` (`col` INT)",
-        # This second query doesn't do anything but we can't just run this because we need the table
-        # to exist since it is self-referencing
-        "CREATE TABLE IF NOT EXISTS `db`.`table` AS SELECT `col` + 1 AS `col` FROM `db`.`table`",
+        "CREATE SCHEMA IF NOT EXISTS `db`",
+        "CREATE TABLE IF NOT EXISTS `db`.`temp_table_abcdefgh` AS SELECT `col` FROM `db`.`table`",
+        "INSERT OVERWRITE TABLE `db`.`table` (`col`) SELECT `col` + 1 AS `col` FROM `db`.`temp_table_abcdefgh`",
+        "DROP TABLE IF EXISTS `db`.`temp_table_abcdefgh`",
     ]
 
 
@@ -325,24 +333,28 @@ def test_create_table_table_options(make_mocked_engine_adapter: t.Callable):
 
     adapter.create_table(
         "test_table",
-        {"a": "int", "b": "int"},
+        {"a": exp.DataType.build("int"), "b": exp.DataType.build("int")},
         table_properties={
             "test.conf.key": exp.convert("value"),
         },
     )
 
     assert to_sql_calls(adapter) == [
-        "CREATE TABLE IF NOT EXISTS `test_table` (`a` int, `b` int) TBLPROPERTIES ('test.conf.key'='value')",
+        "CREATE TABLE IF NOT EXISTS `test_table` (`a` INT, `b` INT) TBLPROPERTIES ('test.conf.key'='value')",
     ]
 
 
 def test_create_state_table(make_mocked_engine_adapter: t.Callable):
     adapter = make_mocked_engine_adapter(SparkEngineAdapter)
 
-    adapter.create_state_table("test_table", {"a": "int", "b": "int"}, primary_key=["a"])
+    adapter.create_state_table(
+        "test_table",
+        {"a": exp.DataType.build("int"), "b": exp.DataType.build("int")},
+        primary_key=["a"],
+    )
 
     assert to_sql_calls(adapter) == [
-        "CREATE TABLE IF NOT EXISTS `test_table` (`a` int, `b` int) PARTITIONED BY (`a`)",
+        "CREATE TABLE IF NOT EXISTS `test_table` (`a` INT, `b` INT) PARTITIONED BY (`a`)",
     ]
 
 
@@ -542,6 +554,11 @@ def test_spark_struct_complex_to_col_to_types(type_name, spark_type):
 def test_scd_type_2(
     make_mocked_engine_adapter: t.Callable, make_temp_table_name: t.Callable, mocker: MockerFixture
 ):
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.spark.SparkEngineAdapter.table_exists",
+        return_value=False,
+    )
+
     adapter = make_mocked_engine_adapter(SparkEngineAdapter)
     adapter.spark.catalog.currentCatalog.return_value = "spark_catalog"
     adapter.spark.catalog.currentDatabase.return_value = "default"
@@ -551,6 +568,17 @@ def test_scd_type_2(
     temp_table_id = "abcdefgh"
     temp_table_mock.return_value = make_temp_table_name(table_name, temp_table_id)
 
+    def check_table_exists(table_name: exp.Table) -> bool:
+        for sql in to_sql_calls(adapter):
+            if f"CREATE TABLE IF NOT EXISTS {table_name.sql(dialect=adapter.dialect)}" in sql:
+                return True
+        return False
+
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.spark.SparkEngineAdapter.table_exists",
+        side_effect=check_table_exists,
+    )
+
     adapter.scd_type_2(
         target_table="db.target",
         source_table=t.cast(
@@ -573,7 +601,6 @@ def test_scd_type_2(
 
     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)",
-        "DESCRIBE `db`.`target`",
         "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`",
         parse_one(
@@ -850,13 +877,13 @@ def test_create_table_with_wap(make_mocked_engine_adapter: t.Callable, mocker: M
 
     adapter.create_table(
         "catalog.schema.table.branch_wap_12345",
-        {"a": "int"},
+        {"a": exp.DataType.build("int")},
         storage_format="ICEBERG",
     )
 
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
-        "CREATE TABLE IF NOT EXISTS `catalog`.`schema`.`table` (`a` int) USING ICEBERG",
+        "CREATE TABLE IF NOT EXISTS `catalog`.`schema`.`table` (`a` INT) USING ICEBERG",
         "INSERT INTO `catalog`.`schema`.`table` SELECT * FROM `catalog`.`schema`.`table`",
     ]
 
diff --git a/tests/core/test_scheduler.py b/tests/core/test_scheduler.py
index 47d40e93..5b548d49 100644
--- a/tests/core/test_scheduler.py
+++ b/tests/core/test_scheduler.py
@@ -199,7 +199,10 @@ def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapsho
                 ((to_datetime("2023-01-04"), to_datetime("2023-01-05")), 2),
             ),
         },
-        (incremental_self_snapshot.name, ((to_datetime(0), to_datetime(0)), -1),): set(
+        (
+            incremental_self_snapshot.name,
+            ((to_datetime(0), to_datetime(0)), -1),
+        ): set(
             [
                 (
                     incremental_self_snapshot.name,
diff --git a/tests/core/test_selector.py b/tests/core/test_selector.py
index 032fcca6..b3740b0e 100644
--- a/tests/core/test_selector.py
+++ b/tests/core/test_selector.py
@@ -213,7 +213,7 @@ def test_select_models_missing_env(mocker: MockerFixture, make_snapshot):
 
 
 @pytest.mark.parametrize(
-    "tags, tag_selections, output",
+    "model_defs, selections, output",
     [
         # Direct matching only
         (
@@ -335,18 +335,42 @@ def test_select_models_missing_env(mocker: MockerFixture, make_snapshot):
             ["tag:tag1", "model2"],
             {'"model1"', '"model2"'},
         ),
+        # Intersection of tags and model names
+        (
+            [
+                ("model1", "tag1", None),
+                ("model2", "tag1", {"model1"}),
+                ("model3", "tag2", {"model1"}),
+                ("model4", "tag1", None),
+            ],
+            ["tag:tag1 & model1+"],
+            {'"model1"', '"model2"'},
+        ),
+        # Intersection of tags and model names (order doesn't matter)
+        (
+            [
+                ("model1", "tag1", None),
+                ("model2", "tag1", {"model1"}),
+                ("model3", "tag2", {"model1"}),
+                ("model4", "tag1", None),
+            ],
+            ["model1+ & tag:tag1"],
+            {'"model1"', '"model2"'},
+        ),
     ],
 )
-def test_expand_model_tags(mocker: MockerFixture, make_snapshot, tags, tag_selections, output):
+def test_expand_model_selections(
+    mocker: MockerFixture, make_snapshot, model_defs, selections, output
+):
     models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
-    for (model_name, tag, depends_on) in tags:
+    for model_name, tag, depends_on in model_defs:
         model = SqlModel(
             name=model_name, query=d.parse_one("SELECT 1 AS a"), depends_on=depends_on, tags=[tag]
         )
         models[model.fqn] = model
 
     selector = Selector(mocker.Mock(), models)
-    assert selector.expand_model_selections(tag_selections) == output
+    assert selector.expand_model_selections(selections) == output
 
 
 def test_select_models_with_external_parent(mocker: MockerFixture):
diff --git a/tests/integrations/github/cicd/fixtures.py b/tests/integrations/github/cicd/fixtures.py
index 806bb6c4..056bbdba 100644
--- a/tests/integrations/github/cicd/fixtures.py
+++ b/tests/integrations/github/cicd/fixtures.py
@@ -83,9 +83,11 @@ def make_controller(mocker: MockerFixture) -> t.Callable:
         return GithubController(
             paths=["examples/sushi"],
             token="abc",
-            event=GithubEvent.from_path(event_path)
-            if isinstance(event_path, str)
-            else GithubEvent.from_obj(event_path),
+            event=(
+                GithubEvent.from_path(event_path)
+                if isinstance(event_path, str)
+                else GithubEvent.from_obj(event_path)
+            ),
             client=client,
         )
 
diff --git a/tests/integrations/github/cicd/test_integration.py b/tests/integrations/github/cicd/test_integration.py
index 2ce495db..db230804 100644
--- a/tests/integrations/github/cicd/test_integration.py
+++ b/tests/integrations/github/cicd/test_integration.py
@@ -2,6 +2,7 @@
 These integration tests are for testing integrating with SQLMesh and not integrating with Github.
 Therefore Github calls are still mocked but context is fully evaluated.
 """
+
 import os
 import pathlib
 import typing as t
diff --git a/tests/schedulers/airflow/test_plan.py b/tests/schedulers/airflow/test_plan.py
index ed9be085..337d476f 100644
--- a/tests/schedulers/airflow/test_plan.py
+++ b/tests/schedulers/airflow/test_plan.py
@@ -172,9 +172,11 @@ def test_create_plan_dag_spec(
         forward_only=True,
         dag_start_ts=to_timestamp("2023-01-01"),
         no_gaps_snapshot_names=expected_no_gaps_snapshot_names,
-        deployability_index_for_creation=DeployabilityIndex.all_deployable()
-        if not paused_forward_only
-        else DeployabilityIndex.none_deployable(),
+        deployability_index_for_creation=(
+            DeployabilityIndex.all_deployable()
+            if not paused_forward_only
+            else DeployabilityIndex.none_deployable()
+        ),
     )
 
     state_sync_mock.get_snapshots.assert_called_once()