Repository: TobikoData/sqlmesh
Pull request #2767 (2024-06-12 21:27:27)

add mssql support for airflow
adding support for microsoft mssql in the airflow scheduler.

izeigerman: Minor doc comment, looks great otherwise 👍  Feel free to graduate this PR from being a draft.

Patch:
diff --git a/docs/guides/scheduling.md b/docs/guides/scheduling.md
index ffd2128d..8f19b5fd 100644
--- a/docs/guides/scheduling.md
+++ b/docs/guides/scheduling.md
@@ -56,7 +56,7 @@ To integrate with [Airflow](../integrations/airflow.md), ensure that you meet th
         for dag in sqlmesh_airflow.dags:
             globals()[dag.dag_id] = dag
 
-    The example above uses `spark` as the engine of choice. Other engines can be configured instead by providing a corresponding string as an argument to the `SQLMeshAirflow` constructor. Supported strings are `"spark"`, `"databricks"`, `"snowflake"`, `"bigquery"`, `"redshift"` and `"trino"`. See the [Airflow Cluster Configuration](../integrations/airflow.md#airflow-cluster-configuration) for full list of arguments and their descriptions.
+    The example above uses `spark` as the engine of choice. Other engines can be configured instead by providing a corresponding string as an argument to the `SQLMeshAirflow` constructor. Supported strings are `"spark"`, `"databricks"`, `"snowflake"`, `"bigquery"`, `"redshift"`, `"trino"` and `"mssql"`. See the [Airflow Cluster Configuration](../integrations/airflow.md#airflow-cluster-configuration) for full list of arguments and their descriptions.
 
 After setup is completed, the `sqlmesh_janitor_dag` DAG should become available in the Airflow UI when filtered by the `sqlmesh` tag:
 
diff --git a/docs/integrations/airflow.md b/docs/integrations/airflow.md
index 3412248c..22422729 100644
--- a/docs/integrations/airflow.md
+++ b/docs/integrations/airflow.md
@@ -116,6 +116,7 @@ SQLMesh supports a variety of engines in Airflow. Support for each engine is pro
 
 * [BigQuery](engines/bigquery.md#airflow-scheduler)
 * [Databricks](engines/databricks.md#airflow-scheduler)
+* [MSSQL](engines/mssql.md#airflow-scheduler)
 * [Postgres](engines/postgres.md#airflow-scheduler)
 * [Redshift](engines/redshift.md#airflow-scheduler)
 * [Snowflake](engines/snowflake.md#airflow-scheduler)
diff --git a/docs/integrations/engines/mssql.md b/docs/integrations/engines/mssql.md
index 1650319d..32dbc019 100644
--- a/docs/integrations/engines/mssql.md
+++ b/docs/integrations/engines/mssql.md
@@ -24,3 +24,27 @@ pip install "sqlmesh[mssql]"
 | `appname`         | The application name to use for the connection               |    string    |    N     |
 | `conn_properties` | The list of connection properties                            | list[string] |    N     |
 | `autocommit`      | Is autocommit mode enabled. Default: false                   |     bool     |    N     |
+
+## Airflow Scheduler
+**Engine Name:** `mssql`
+
+The SQLMesh MsSql Operator is similar to the [MsSqlOperator](https://airflow.apache.org/docs/apache-airflow-providers-microsoft-mssql/stable/_api/airflow/providers/microsoft/mssql/operators/mssql/index.html), and relies on the same [MsSqlHook](https://airflow.apache.org/docs/apache-airflow-providers-microsoft-mssql/stable/_api/airflow/providers/microsoft/mssql/hooks/mssql/index.html) implementation.
+
+To enable support for this operator, the Airflow Microsoft MSSQL provider package should be installed on the target Airflow cluster along with SQLMesh with the mssql extra:
+```
+pip install "apache-airflow-providers-microsoft-mssql"
+pip install "sqlmesh[mssql]"
+```
+
+The operator requires an [Airflow connection](https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html) to determine the target MSSQL account. Refer to [MSSQL connection](https://airflow.apache.org/docs/apache-airflow-providers-microsoft-mssql/stable/connections/mssql.html) for more details.
+
+By default, the connection ID is set to `mssql_default`, but can be overridden using the `engine_operator_args` parameter to the `SQLMeshAirflow` instance as in the example below:
+```python linenums="1"
+sqlmesh_airflow = SQLMeshAirflow(
+    "mssql",
+    default_catalog="<database name>",
+    engine_operator_args={
+        "mssql_conn_id": "<Connection ID>"
+    },
+)
+```
\ No newline at end of file
diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py
index e386484e..a46ace67 100644
--- a/sqlmesh/core/console.py
+++ b/sqlmesh/core/console.py
@@ -38,6 +38,7 @@ from sqlmesh.utils.date import time_like_to_str, to_date, yesterday_ds
 if t.TYPE_CHECKING:
     import ipywidgets as widgets
 
+    from sqlglot.dialects.dialect import DialectType
     from sqlmesh.core.context_diff import ContextDiff
     from sqlmesh.core.plan import Plan, PlanBuilder
     from sqlmesh.core.table_diff import RowDiff, SchemaDiff
@@ -223,7 +224,11 @@ class TerminalConsole(Console):
     """A rich based implementation of the console."""
 
     def __init__(
-        self, console: t.Optional[RichConsole] = None, verbose: bool = False, **kwargs: t.Any
+        self,
+        console: t.Optional[RichConsole] = None,
+        verbose: bool = False,
+        dialect: DialectType = None,
+        **kwargs: t.Any,
     ) -> None:
         self.console: RichConsole = console or srich.console
 
@@ -250,6 +255,7 @@ class TerminalConsole(Console):
         self.loading_status: t.Dict[uuid.UUID, Status] = {}
 
         self.verbose = verbose
+        self.dialect = dialect
 
     def _print(self, value: t.Any, **kwargs: t.Any) -> None:
         self.console.print(value, **kwargs)
@@ -308,7 +314,9 @@ class TerminalConsole(Console):
 
     def start_snapshot_evaluation_progress(self, snapshot: Snapshot) -> None:
         if self.evaluation_model_progress and snapshot.name not in self.evaluation_model_tasks:
-            display_name = snapshot.display_name(self.environment_naming_info, self.default_catalog)
+            display_name = snapshot.display_name(
+                self.environment_naming_info, self.default_catalog, dialect=self.dialect
+            )
             self.evaluation_model_tasks[snapshot.name] = self.evaluation_model_progress.add_task(
                 f"Evaluating {display_name}...",
                 view_name=display_name,
@@ -328,7 +336,7 @@ class TerminalConsole(Console):
 
             if duration_ms:
                 self.evaluation_progress_live.console.print(
-                    f"[{batch_idx + 1}/{total_batches}] {snapshot.display_name(self.environment_naming_info, self.default_catalog)} [green]evaluated[/green] in {(duration_ms / 1000.0):.2f}s"
+                    f"[{batch_idx + 1}/{total_batches}] {snapshot.display_name(self.environment_naming_info, self.default_catalog, dialect=self.dialect)} [green]evaluated[/green] in {(duration_ms / 1000.0):.2f}s"
                 )
 
             self.evaluation_total_progress.update(
@@ -389,7 +397,7 @@ class TerminalConsole(Console):
         if self.creation_progress is not None and self.creation_task is not None:
             if self.verbose:
                 self.creation_progress.live.console.print(
-                    f"{snapshot.display_name(self.environment_naming_info, self.default_catalog)} [green]created[/green]"
+                    f"{snapshot.display_name(self.environment_naming_info, self.default_catalog, dialect=self.dialect)} [green]created[/green]"
                 )
             self.creation_progress.update(self.creation_task, refresh=True, advance=1)
 
@@ -444,7 +452,7 @@ class TerminalConsole(Console):
             if self.verbose:
                 action_str = "[green]promoted[/green]" if promoted else "[yellow]demoted[/yellow]"
                 self.promotion_progress.live.console.print(
-                    f"{snapshot.display_name(self.environment_naming_info, self.default_catalog)} {action_str}"
+                    f"{snapshot.display_name(self.environment_naming_info, self.default_catalog, dialect=self.dialect)} {action_str}"
                 )
             self.promotion_progress.update(self.promotion_task, refresh=True, advance=1)
 
@@ -594,7 +602,7 @@ class TerminalConsole(Console):
         for s_id in ignored_snapshot_ids:
             snapshot = snapshots[s_id]
             ignored.add(
-                f"[ignored]{snapshot.display_name(environment_naming_info, default_catalog)} ({snapshot.get_latest(start_date(snapshot, snapshots.values()))})"
+                f"[ignored]{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)} ({snapshot.get_latest(start_date(snapshot, snapshots.values()))})"
             )
         return ignored
 
@@ -646,7 +654,7 @@ class TerminalConsole(Console):
             for s_id in added_snapshot_ids:
                 snapshot = context_diff.snapshots[s_id]
                 added_tree.add(
-                    f"[added]{snapshot.display_name(environment_naming_info, default_catalog)}"
+                    f"[added]{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}"
                 )
             tree.add(added_tree)
         if removed_snapshot_ids:
@@ -654,7 +662,7 @@ class TerminalConsole(Console):
             for s_id in removed_snapshot_ids:
                 snapshot_table_info = context_diff.removed_snapshots[s_id]
                 removed_tree.add(
-                    f"[removed]{snapshot_table_info.display_name(environment_naming_info, default_catalog)}"
+                    f"[removed]{snapshot_table_info.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}"
                 )
             tree.add(removed_tree)
         if modified_snapshot_ids:
@@ -664,7 +672,7 @@ class TerminalConsole(Console):
             for s_id in modified_snapshot_ids:
                 name = s_id.name
                 display_name = context_diff.snapshots[s_id].display_name(
-                    environment_naming_info, default_catalog
+                    environment_naming_info, default_catalog, dialect=self.dialect
                 )
                 if context_diff.directly_modified(name):
                     direct.add(
@@ -738,7 +746,7 @@ class TerminalConsole(Console):
             if not no_diff:
                 self.show_sql(plan.context_diff.text_diff(snapshot.name))
             tree = Tree(
-                f"[bold][direct]Directly Modified: {snapshot.display_name(plan.environment_naming_info, default_catalog)}"
+                f"[bold][direct]Directly Modified: {snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)}"
             )
             indirect_tree = None
 
@@ -748,7 +756,7 @@ class TerminalConsole(Console):
                     indirect_tree = Tree("[indirect]Indirectly Modified Children:")
                     tree.add(indirect_tree)
                 indirect_tree.add(
-                    f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog)}"
+                    f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)}"
                 )
             self._print(tree)
             if not no_prompts:
@@ -765,7 +773,7 @@ class TerminalConsole(Console):
 
             category_str = SNAPSHOT_CHANGE_CATEGORY_STR[snapshot.change_category]
             tree = Tree(
-                f"[bold][direct]Directly Modified: {snapshot.display_name(plan.environment_naming_info, default_catalog)} ({category_str})"
+                f"[bold][direct]Directly Modified: {snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)} ({category_str})"
             )
             indirect_tree = None
             for child_sid in sorted(plan.indirectly_modified.get(snapshot.snapshot_id, set())):
@@ -775,7 +783,7 @@ class TerminalConsole(Console):
                     tree.add(indirect_tree)
                 child_category_str = SNAPSHOT_CHANGE_CATEGORY_STR[child_snapshot.change_category]
                 indirect_tree.add(
-                    f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog)} ({child_category_str})"
+                    f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)} ({child_category_str})"
                 )
             self._print(Syntax(context_diff.text_diff(snapshot.name), "sql", word_wrap=True))
             self._print(tree)
@@ -796,7 +804,7 @@ class TerminalConsole(Console):
                 preview_modifier = " ([orange1]preview[/orange1])"
 
             backfill.add(
-                f"{snapshot.display_name(plan.environment_naming_info, default_catalog)}: {missing.format_intervals(snapshot.node.interval_unit)}{preview_modifier}"
+                f"{snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)}: {missing.format_intervals(snapshot.node.interval_unit)}{preview_modifier}"
             )
         self._print(backfill)
 
@@ -1030,7 +1038,9 @@ class TerminalConsole(Console):
         default_catalog: t.Optional[str],
         use_rich_formatting: bool = True,
     ) -> t.Dict[SnapshotChangeCategory, str]:
-        direct = snapshot.display_name(environment_naming_info, default_catalog)
+        direct = snapshot.display_name(
+            environment_naming_info, default_catalog, dialect=self.dialect
+        )
         if use_rich_formatting:
             direct = f"[direct]{direct}[/direct]"
         indirect = "indirectly modified children"
@@ -1082,6 +1092,7 @@ class NotebookMagicConsole(TerminalConsole):
         self,
         display: t.Optional[t.Callable] = None,
         console: t.Optional[RichConsole] = None,
+        dialect: DialectType = None,
         **kwargs: t.Any,
     ) -> None:
         import ipywidgets as widgets
@@ -1094,6 +1105,8 @@ class NotebookMagicConsole(TerminalConsole):
         self.missing_dates_output = widgets.Output()
         self.dynamic_options_after_categorization_output = widgets.VBox()
 
+        self.dialect = dialect
+
     def _show_missing_dates(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
         self._add_to_dynamic_options(self.missing_dates_output)
         self.missing_dates_output.outputs = ()
@@ -1442,7 +1455,7 @@ class MarkdownConsole(CaptureTerminalConsole):
             self._print("\n**Added Models:**")
             for snapshot in sorted(added_snapshot_models):
                 self._print(
-                    f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
+                    f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
                 )
 
         added_snapshot_audits = {s for s in added_snapshots if s.is_audit}
@@ -1450,7 +1463,7 @@ class MarkdownConsole(CaptureTerminalConsole):
             self._print("\n**Added Standalone Audits:**")
             for snapshot in sorted(added_snapshot_audits):
                 self._print(
-                    f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
+                    f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
                 )
 
         removed_snapshot_table_infos = {
@@ -1463,7 +1476,7 @@ class MarkdownConsole(CaptureTerminalConsole):
             self._print("\n**Removed Models:**")
             for snapshot_table_info in sorted(removed_model_snapshot_table_infos):
                 self._print(
-                    f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog)}`"
+                    f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
                 )
 
         removed_audit_snapshot_table_infos = {s for s in removed_snapshot_table_infos if s.is_audit}
@@ -1471,7 +1484,7 @@ class MarkdownConsole(CaptureTerminalConsole):
             self._print("\n**Removed Standalone Audits:**")
             for snapshot_table_info in sorted(removed_audit_snapshot_table_infos):
                 self._print(
-                    f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog)}`"
+                    f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
                 )
 
         modified_snapshots = {
@@ -1494,7 +1507,7 @@ class MarkdownConsole(CaptureTerminalConsole):
                 self._print("\n**Directly Modified:**")
                 for snapshot in sorted(directly_modified):
                     self._print(
-                        f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
+                        f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
                     )
                     if not no_diff:
                         self._print(f"```diff\n{context_diff.text_diff(snapshot.name)}\n```")
@@ -1502,20 +1515,20 @@ class MarkdownConsole(CaptureTerminalConsole):
                 self._print("\n**Indirectly Modified:**")
                 for snapshot in sorted(indirectly_modified):
                     self._print(
-                        f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
+                        f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
                     )
             if metadata_modified:
                 self._print("\n**Metadata Updated:**")
                 for snapshot in sorted(metadata_modified):
                     self._print(
-                        f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
+                        f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
                     )
         if ignored_snapshot_ids:
             self._print("\n**Ignored Models (Expected Plan Start):**")
             for s_id in sorted(ignored_snapshot_ids):
                 snapshot = context_diff.snapshots[s_id]
                 self._print(
-                    f"- `{snapshot.display_name(environment_naming_info, default_catalog)}` ({snapshot.get_latest(start_date(snapshot, context_diff.snapshots.values()))})"
+                    f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}` ({snapshot.get_latest(start_date(snapshot, context_diff.snapshots.values()))})"
                 )
 
     def _show_missing_dates(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
@@ -1534,7 +1547,7 @@ class MarkdownConsole(CaptureTerminalConsole):
                 preview_modifier = " (**preview**)"
 
             self._print(
-                f"* `{snapshot.display_name(plan.environment_naming_info, default_catalog)}`: {missing.format_intervals(snapshot.node.interval_unit)}{preview_modifier}"
+                f"* `{snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)}`: {missing.format_intervals(snapshot.node.interval_unit)}{preview_modifier}"
             )
 
     def _show_categorized_snapshots(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
@@ -1545,7 +1558,7 @@ class MarkdownConsole(CaptureTerminalConsole):
 
             category_str = SNAPSHOT_CHANGE_CATEGORY_STR[snapshot.change_category]
             tree = Tree(
-                f"[bold][direct]Directly Modified: {snapshot.display_name(plan.environment_naming_info, default_catalog)} ({category_str})"
+                f"[bold][direct]Directly Modified: {snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)} ({category_str})"
             )
             indirect_tree = None
             for child_sid in sorted(plan.indirectly_modified.get(snapshot.snapshot_id, set())):
@@ -1555,7 +1568,7 @@ class MarkdownConsole(CaptureTerminalConsole):
                     tree.add(indirect_tree)
                 child_category_str = SNAPSHOT_CHANGE_CATEGORY_STR[child_snapshot.change_category]
                 indirect_tree.add(
-                    f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog)} ({child_category_str})"
+                    f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)} ({child_category_str})"
                 )
             self._print(f"```diff\n{context_diff.text_diff(snapshot.name)}\n```\n")
             self._print("```\n")
@@ -1625,7 +1638,7 @@ class DatabricksMagicConsole(CaptureTerminalConsole):
     def start_snapshot_evaluation_progress(self, snapshot: Snapshot) -> None:
         if not self.evaluation_batch_progress.get(snapshot.snapshot_id):
             display_name = snapshot.display_name(
-                self.evaluation_environment_naming_info, self.default_catalog
+                self.evaluation_environment_naming_info, self.default_catalog, dialect=self.dialect
             )
             self.evaluation_batch_progress[snapshot.snapshot_id] = (display_name, 0)
             print(f"Starting '{display_name}', Total batches: {self.evaluation_batches[snapshot]}")
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index 92a3a0d1..e68c536d 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -309,7 +309,6 @@ class GenericContext(BaseContext, t.Generic[C]):
         console: t.Optional[Console] = None,
         users: t.Optional[t.List[User]] = None,
     ):
-        self.console = console or get_console()
         self.configs = (
             config if isinstance(config, dict) else load_configs(config, self.CONFIG_TYPE, paths)
         )
@@ -332,13 +331,15 @@ class GenericContext(BaseContext, t.Generic[C]):
         self.gateway = gateway
         self._scheduler = self.config.get_scheduler(self.gateway)
         self.environment_ttl = self.config.environment_ttl
-        self.pinned_environments = Environment.normalize_names(self.config.pinned_environments)
+        self.pinned_environments = Environment.sanitize_names(self.config.pinned_environments)
         self.auto_categorize_changes = self.config.plan.auto_categorize_changes
 
         self._connection_config = self.config.get_connection(self.gateway)
         self.concurrent_tasks = concurrent_tasks or self._connection_config.concurrent_tasks
         self._engine_adapter = engine_adapter or self._connection_config.create_engine_adapter()
 
+        self.console = console or get_console(dialect=self._engine_adapter.dialect)
+
         self._test_connection_config = self.config.get_test_connection(
             self.gateway, self.default_catalog, default_catalog_dialect=self.engine_adapter.DIALECT
         )
@@ -1015,7 +1016,7 @@ class GenericContext(BaseContext, t.Generic[C]):
             The plan builder.
         """
         environment = environment or self.config.default_target_environment
-        environment = Environment.normalize_name(environment)
+        environment = Environment.sanitize_name(environment)
         is_dev = environment != c.PROD
 
         if skip_backfill and not no_gaps and not is_dev:
@@ -1213,7 +1214,7 @@ class GenericContext(BaseContext, t.Generic[C]):
             True if there are changes, False otherwise.
         """
         environment = environment or self.config.default_target_environment
-        environment = Environment.normalize_name(environment)
+        environment = Environment.sanitize_name(environment)
         context_diff = self._context_diff(environment)
         self.console.show_model_difference_summary(
             context_diff,
@@ -1221,6 +1222,7 @@ class GenericContext(BaseContext, t.Generic[C]):
                 self.config.environment_catalog_mapping,
                 name=environment,
                 suffix_target=self.config.environment_suffix_target,
+                normalize_name=context_diff.normalize_environment_name,
             ),
             self.default_catalog,
             no_diff=not detailed,
@@ -1870,9 +1872,13 @@ class GenericContext(BaseContext, t.Generic[C]):
         force_no_diff: bool = False,
         ensure_finalized_snapshots: bool = False,
     ) -> ContextDiff:
-        environment = Environment.normalize_name(environment)
+        environment = Environment.sanitize_name(environment)
         if force_no_diff:
-            return ContextDiff.create_no_diff(environment)
+            return ContextDiff.create_no_diff(
+                self.state_reader.get_environment(environment.lower())
+                or EnvironmentNamingInfo(name=environment)
+            )
+
         return ContextDiff.create(
             environment,
             snapshots=snapshots or self.snapshots,
diff --git a/sqlmesh/core/context_diff.py b/sqlmesh/core/context_diff.py
index 2f74f7eb..4e5d129a 100644
--- a/sqlmesh/core/context_diff.py
+++ b/sqlmesh/core/context_diff.py
@@ -21,6 +21,7 @@ from sqlmesh.utils.errors import SQLMeshError
 from sqlmesh.utils.pydantic import PydanticModel
 
 if t.TYPE_CHECKING:
+    from sqlmesh.core.environment import EnvironmentNamingInfo
     from sqlmesh.core.state_sync import StateReader
 
 logger = logging.getLogger(__name__)
@@ -39,6 +40,8 @@ class ContextDiff(PydanticModel):
     """Whether the target environment is new."""
     is_unfinalized_environment: bool
     """Whether the currently stored environment record is in unfinalized state."""
+    normalize_environment_name: bool
+    """Whether the environment name should be normalized."""
     create_from: str
     """The name of the environment the target environment will be created from if new."""
     added: t.Set[SnapshotId]
@@ -162,6 +165,7 @@ class ContextDiff(PydanticModel):
             environment=environment,
             is_new_environment=is_new_environment,
             is_unfinalized_environment=bool(env and not env.finalized_ts),
+            normalize_environment_name=is_new_environment or bool(env and env.normalize_name),
             create_from=create_from,
             added=added,
             removed_snapshots=removed,
@@ -174,7 +178,7 @@ class ContextDiff(PydanticModel):
         )
 
     @classmethod
-    def create_no_diff(cls, environment: str) -> ContextDiff:
+    def create_no_diff(cls, environment: EnvironmentNamingInfo) -> ContextDiff:
         """Create a no-op ContextDiff object.
 
         Args:
@@ -184,9 +188,10 @@ class ContextDiff(PydanticModel):
             The ContextDiff object.
         """
         return ContextDiff(
-            environment=environment,
+            environment=environment.name,
             is_new_environment=False,
             is_unfinalized_environment=False,
+            normalize_environment_name=environment.normalize_name,
             create_from="",
             added=set(),
             removed_snapshots={},
diff --git a/sqlmesh/core/environment.py b/sqlmesh/core/environment.py
index 2640a697..e28956bc 100644
--- a/sqlmesh/core/environment.py
+++ b/sqlmesh/core/environment.py
@@ -24,41 +24,48 @@ class EnvironmentNamingInfo(PydanticModel):
         name: The name of the environment.
         suffix_target: Indicates whether to append the environment name to the schema or table name.
         catalog_name_override: The name of the catalog to use for this environment if an override was provided
-
+        normalize_name: Indicates whether the environment's name will be normalized. For example, if it's
+            `dev`, then it will become `DEV` when targeting Snowflake.
     """
 
     name: str = c.PROD
     suffix_target: EnvironmentSuffixTarget = Field(default=EnvironmentSuffixTarget.SCHEMA)
     catalog_name_override: t.Optional[str] = None
+    normalize_name: bool = True
 
     @field_validator("name", mode="before")
     @classmethod
-    def _normalize_name(cls, v: str) -> str:
+    def _sanitize_name(cls, v: str) -> str:
         return word_characters_only(v).lower()
 
+    @field_validator("normalize_name", mode="before")
+    @classmethod
+    def _validate_normalize_name(cls, v: t.Any) -> bool:
+        return True if v is None else bool(v)
+
     @t.overload
     @classmethod
-    def normalize_name(cls, v: str) -> str: ...
+    def sanitize_name(cls, v: str) -> str: ...
 
     @t.overload
     @classmethod
-    def normalize_name(cls, v: Environment) -> Environment: ...
+    def sanitize_name(cls, v: Environment) -> Environment: ...
 
     @classmethod
-    def normalize_name(cls, v: str | Environment) -> str | Environment:
+    def sanitize_name(cls, v: str | Environment) -> str | Environment:
         """
-        Normalizes the environment name so we create names that are valid names for database objects.
+        Sanitizes the environment name so we create names that are valid names for database objects.
         This means alphanumeric and underscores only. Invalid characters are replaced with underscores.
         """
         if isinstance(v, Environment):
             return v
         if not isinstance(v, str):
             raise TypeError(f"Expected str or Environment, got {type(v).__name__}")
-        return cls._normalize_name(v)
+        return cls._sanitize_name(v)
 
     @classmethod
-    def normalize_names(cls, values: t.Iterable[str]) -> t.Set[str]:
-        return {cls.normalize_name(value) for value in values}
+    def sanitize_names(cls, values: t.Iterable[str]) -> t.Set[str]:
+        return {cls.sanitize_name(value) for value in values}
 
     @classmethod
     def from_environment_catalog_mapping(
@@ -143,4 +150,5 @@ class Environment(EnvironmentNamingInfo):
             name=self.name,
             suffix_target=self.suffix_target,
             catalog_name_override=self.catalog_name_override,
+            normalize_name=self.normalize_name,
         )
diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py
index 88cf6826..ced9f394 100644
--- a/sqlmesh/core/plan/builder.py
+++ b/sqlmesh/core/plan/builder.py
@@ -137,6 +137,7 @@ class PlanBuilder:
             environment_catalog_mapping or {},
             name=self._context_diff.environment,
             suffix_target=environment_suffix_target,
+            normalize_name=self._context_diff.normalize_environment_name,
         )
 
         self._latest_plan: t.Optional[Plan] = None
diff --git a/sqlmesh/core/selector.py b/sqlmesh/core/selector.py
index 54618d77..e1c6bd66 100644
--- a/sqlmesh/core/selector.py
+++ b/sqlmesh/core/selector.py
@@ -65,10 +65,10 @@ class Selector:
         Returns:
             A dictionary of models.
         """
-        target_env = self._state_reader.get_environment(Environment.normalize_name(target_env_name))
+        target_env = self._state_reader.get_environment(Environment.sanitize_name(target_env_name))
         if not target_env and fallback_env_name:
             target_env = self._state_reader.get_environment(
-                Environment.normalize_name(fallback_env_name)
+                Environment.sanitize_name(fallback_env_name)
             )
 
         env_models: t.Dict[str, Model] = {}
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index b3981e11..4e863ceb 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -10,6 +10,7 @@ from functools import cached_property, lru_cache
 from pydantic import Field
 from sqlglot import exp
 from sqlglot.helper import seq_get
+from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 
 from sqlmesh.core import constants as c
 from sqlmesh.core.audit import BUILT_IN_AUDITS, Audit, ModelAudit, StandaloneAudit
@@ -43,6 +44,7 @@ else:
     from typing_extensions import Annotated
 
 if t.TYPE_CHECKING:
+    from sqlglot.dialects.dialect import DialectType
     from sqlmesh.core.environment import EnvironmentNamingInfo
 
 Interval = t.Tuple[int, int]
@@ -204,13 +206,17 @@ class QualifiedViewName(PydanticModel, frozen=True):
     schema_name: t.Optional[str] = None
     table: str
 
-    def for_environment(self, environment_naming_info: EnvironmentNamingInfo) -> str:
-        return exp.table_name(self.table_for_environment(environment_naming_info))
+    def for_environment(
+        self, environment_naming_info: EnvironmentNamingInfo, dialect: DialectType = None
+    ) -> str:
+        return exp.table_name(self.table_for_environment(environment_naming_info, dialect=dialect))
 
-    def table_for_environment(self, environment_naming_info: EnvironmentNamingInfo) -> exp.Table:
+    def table_for_environment(
+        self, environment_naming_info: EnvironmentNamingInfo, dialect: DialectType = None
+    ) -> exp.Table:
         return exp.table_(
-            self.table_name_for_environment(environment_naming_info),
-            db=self.schema_for_environment(environment_naming_info),
+            self.table_name_for_environment(environment_naming_info, dialect=dialect),
+            db=self.schema_for_environment(environment_naming_info, dialect=dialect),
             catalog=self.catalog_for_environment(environment_naming_info),
         )
 
@@ -219,22 +225,39 @@ class QualifiedViewName(PydanticModel, frozen=True):
     ) -> t.Optional[str]:
         return environment_naming_info.catalog_name_override or self.catalog
 
-    def schema_for_environment(self, environment_naming_info: EnvironmentNamingInfo) -> str:
+    def schema_for_environment(
+        self, environment_naming_info: EnvironmentNamingInfo, dialect: DialectType = None
+    ) -> str:
         schema = self.schema_name or c.DEFAULT_SCHEMA
         if (
             environment_naming_info.name.lower() != c.PROD
             and environment_naming_info.suffix_target.is_schema
         ):
-            schema = f"{schema}__{environment_naming_info.name}"
+            env_name = environment_naming_info.name
+            if environment_naming_info.normalize_name:
+                env_name = normalize_identifiers(env_name, dialect=dialect).name
+                if not self.schema_name:
+                    # We also normalize the schema if it's the default one
+                    schema = normalize_identifiers(schema, dialect=dialect).name
+
+            schema = f"{schema}__{env_name}"
+
         return schema
 
-    def table_name_for_environment(self, environment_naming_info: EnvironmentNamingInfo) -> str:
+    def table_name_for_environment(
+        self, environment_naming_info: EnvironmentNamingInfo, dialect: DialectType = None
+    ) -> str:
         table = self.table
         if (
             environment_naming_info.name.lower() != c.PROD
             and environment_naming_info.suffix_target.is_table
         ):
-            table = f"{table}__{environment_naming_info.name}"
+            env_name = environment_naming_info.name
+            if environment_naming_info.normalize_name:
+                env_name = normalize_identifiers(env_name, dialect=dialect).name
+
+            table = f"{table}__{env_name}"
+
         return table
 
 
@@ -314,14 +337,17 @@ class SnapshotInfoMixin(ModelKindMixin):
         return (*self.previous_versions, self.data_version)[-c.DATA_VERSION_LIMIT :]
 
     def display_name(
-        self, environment_naming_info: EnvironmentNamingInfo, default_catalog: t.Optional[str]
+        self,
+        environment_naming_info: EnvironmentNamingInfo,
+        default_catalog: t.Optional[str],
+        dialect: DialectType = None,
     ) -> str:
         """
         Returns the model name as a qualified view name.
         This is just used for presenting information back to the user and `qualified_view_name` should be used
         when wanting a view name in all other cases.
         """
-        return display_name(self, environment_naming_info, default_catalog)
+        return display_name(self, environment_naming_info, default_catalog, dialect=dialect)
 
     def data_hash_matches(self, other: t.Optional[SnapshotInfoMixin | SnapshotDataVersion]) -> bool:
         return other is not None and self.fingerprint.data_hash == other.fingerprint.data_hash
@@ -1338,6 +1364,7 @@ def display_name(
     snapshot_info_like: t.Union[SnapshotInfoLike, SnapshotInfoMixin],
     environment_naming_info: EnvironmentNamingInfo,
     default_catalog: t.Optional[str],
+    dialect: DialectType = None,
 ) -> str:
     """
     Returns the model name as a qualified view name.
@@ -1356,7 +1383,7 @@ def display_name(
         schema_name=view_name.db or None,
         table=view_name.name,
     )
-    return qvn.for_environment(environment_naming_info)
+    return qvn.for_environment(environment_naming_info, dialect=dialect)
 
 
 def fingerprint_from_node(
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 5f39a15e..9f683bbc 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -195,7 +195,9 @@ class SnapshotEvaluator:
         """
         self._create_schemas(
             [
-                s.qualified_view_name.table_for_environment(environment_naming_info)
+                s.qualified_view_name.table_for_environment(
+                    environment_naming_info, dialect=self.adapter.dialect
+                )
                 for s in target_snapshots
                 if s.is_model and not s.is_symbolic
             ]
@@ -1098,7 +1100,9 @@ class EmbeddedStrategy(SymbolicStrategy):
         table_name: str,
         snapshot: Snapshot,
     ) -> None:
-        target_name = view_name.for_environment(environment_naming_info)
+        target_name = view_name.for_environment(
+            environment_naming_info, dialect=self.adapter.dialect
+        )
         logger.info("Dropping view '%s' for non-materialized table", target_name)
         self.adapter.drop_view(target_name, cascade=False)
 
@@ -1111,7 +1115,9 @@ class PromotableStrategy(EvaluationStrategy):
         table_name: str,
         snapshot: Snapshot,
     ) -> None:
-        target_name = view_name.for_environment(environment_naming_info)
+        target_name = view_name.for_environment(
+            environment_naming_info, dialect=self.adapter.dialect
+        )
         is_prod = environment_naming_info.name.lower() == c.PROD
         logger.info("Updating view '%s' to point at table '%s'", target_name, table_name)
         self.adapter.create_view(
@@ -1127,7 +1133,9 @@ class PromotableStrategy(EvaluationStrategy):
         view_name: QualifiedViewName,
         environment_naming_info: EnvironmentNamingInfo,
     ) -> None:
-        target_name = view_name.for_environment(environment_naming_info)
+        target_name = view_name.for_environment(
+            environment_naming_info, dialect=self.adapter.dialect
+        )
         logger.info("Dropping view '%s'", target_name)
         self.adapter.drop_view(target_name, cascade=False)
 
diff --git a/sqlmesh/core/state_sync/common.py b/sqlmesh/core/state_sync/common.py
index 3f6610dc..12da604d 100644
--- a/sqlmesh/core/state_sync/common.py
+++ b/sqlmesh/core/state_sync/common.py
@@ -40,7 +40,9 @@ def cleanup_expired_views(
     for expired_catalog, expired_schema in {
         (
             snapshot.qualified_view_name.catalog_for_environment(environment.naming_info),
-            snapshot.qualified_view_name.schema_for_environment(environment.naming_info),
+            snapshot.qualified_view_name.schema_for_environment(
+                environment.naming_info, dialect=adapter.dialect
+            ),
         )
         for environment in expired_schema_environments
         for snapshot in environment.snapshots
@@ -58,7 +60,9 @@ def cleanup_expired_views(
         except Exception as e:
             logger.warning("Falied to drop the expired environment schema '%s': %s", schema, e)
     for expired_view in {
-        snapshot.qualified_view_name.for_environment(environment.naming_info)
+        snapshot.qualified_view_name.for_environment(
+            environment.naming_info, dialect=adapter.dialect
+        )
         for environment in expired_table_environments
         for snapshot in environment.snapshots
         if snapshot.is_model and not snapshot.is_symbolic
diff --git a/sqlmesh/core/state_sync/engine_adapter.py b/sqlmesh/core/state_sync/engine_adapter.py
index 8ce39089..13e35337 100644
--- a/sqlmesh/core/state_sync/engine_adapter.py
+++ b/sqlmesh/core/state_sync/engine_adapter.py
@@ -141,6 +141,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             "suffix_target": exp.DataType.build("text"),
             "catalog_name_override": exp.DataType.build("text"),
             "previous_finalized_snapshots": exp.DataType.build("text"),
+            "normalize_name": exp.DataType.build("boolean"),
         }
 
         self._interval_columns_to_types = {
@@ -1394,6 +1395,7 @@ def _environment_to_df(environment: Environment) -> pd.DataFrame:
                     if environment.previous_finalized_snapshots is not None
                     else None
                 ),
+                "normalize_name": environment.normalize_name,
             }
         ]
     )
diff --git a/sqlmesh/integrations/github/cicd/controller.py b/sqlmesh/integrations/github/cicd/controller.py
index 2067f98e..3a9e6808 100644
--- a/sqlmesh/integrations/github/cicd/controller.py
+++ b/sqlmesh/integrations/github/cicd/controller.py
@@ -357,7 +357,7 @@ class GithubController:
 
     @property
     def pr_environment_name(self) -> str:
-        return Environment.normalize_name(
+        return Environment.sanitize_name(
             "_".join(
                 [
                     self.bot_config.pr_environment_name or self._event.pull_request_info.repo,
diff --git a/sqlmesh/migrations/v0052_add_normalize_name_in_environment_naming_info.py b/sqlmesh/migrations/v0052_add_normalize_name_in_environment_naming_info.py
new file mode 100644
index 00000000..617c2985
--- /dev/null
+++ b/sqlmesh/migrations/v0052_add_normalize_name_in_environment_naming_info.py
@@ -0,0 +1,27 @@
+"""Add flag that controls whether environment names will be normalized."""
+
+from sqlglot import exp
+
+
+def migrate(state_sync, **kwargs):  # type: ignore
+    engine_adapter = state_sync.engine_adapter
+    environments_table = "_environments"
+    if state_sync.schema:
+        environments_table = f"{state_sync.schema}.{environments_table}"
+
+    alter_table_exp = exp.AlterTable(
+        this=exp.to_table(environments_table),
+        actions=[
+            exp.ColumnDef(
+                this=exp.to_column("normalize_name"),
+                kind=exp.DataType.build("boolean"),
+            )
+        ],
+    )
+    engine_adapter.execute(alter_table_exp)
+
+    state_sync.engine_adapter.update_table(
+        environments_table,
+        {"normalize_name": False},
+        where=exp.true(),
+    )
diff --git a/sqlmesh/schedulers/airflow/operators/mssql.py b/sqlmesh/schedulers/airflow/operators/mssql.py
new file mode 100644
index 00000000..067c6902
--- /dev/null
+++ b/sqlmesh/schedulers/airflow/operators/mssql.py
@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import typing as t
+
+from airflow.models import BaseOperator
+from airflow.providers.microsoft.mssql.hooks.mssql import MsSqlHook
+from airflow.utils.context import Context
+
+from sqlmesh.schedulers.airflow.operators.targets import BaseTarget
+
+
+class SQLMeshMsSqlOperator(BaseOperator):
+    """The operator that evaluates a SQLMesh model snapshot on a mssql target
+
+    Args:
+        target: The target that will be executed by this operator instance.
+        mssql_conn_id: The Airflow connection id for the mssql target.
+    """
+
+    def __init__(
+        self,
+        *,
+        target: BaseTarget,
+        mssql_conn_id: str = MsSqlHook.default_conn_name,
+        **kwargs: t.Any,
+    ) -> None:
+        super().__init__(**kwargs)
+        self._target = target
+        self._mssql_conn_id = mssql_conn_id
+        self._hook_params = kwargs
+
+    def get_db_hook(self) -> MsSqlHook:
+        """Gets the mssql Hook which contains the DB API connection object"""
+        return MsSqlHook(self._mssql_conn_id, **self._hook_params)
+
+    def execute(self, context: Context) -> None:
+        """Executes the desired target against the configured mssql connection"""
+        self._target.execute(context, lambda: self.get_db_hook().get_conn(), "mssql")
diff --git a/sqlmesh/schedulers/airflow/util.py b/sqlmesh/schedulers/airflow/util.py
index 6db4e533..9009bd45 100644
--- a/sqlmesh/schedulers/airflow/util.py
+++ b/sqlmesh/schedulers/airflow/util.py
@@ -169,6 +169,10 @@ def discover_engine_operator(name: str, sql_only: bool = False) -> t.Type[BaseOp
             from sqlmesh.schedulers.airflow.operators.trino import SQLMeshTrinoOperator
 
             return SQLMeshTrinoOperator
+        if name == "mssql":
+            from sqlmesh.schedulers.airflow.operators.mssql import SQLMeshMsSqlOperator
+
+            return SQLMeshMsSqlOperator
     except ImportError:
         raise SQLMeshError(f"Failed to automatically discover an operator for '{name}'.'")
 


Test patch:
diff --git a/tests/core/test_context.py b/tests/core/test_context.py
index c12862fe..52c7ce00 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -285,6 +285,31 @@ def test_plan_execution_time():
     )
 
 
+def test_env_and_default_schema_normalization(mocker: MockerFixture):
+    from sqlglot.dialects import DuckDB
+    from sqlglot.dialects.dialect import NormalizationStrategy
+
+    mocker.patch.object(DuckDB, "NORMALIZATION_STRATEGY", NormalizationStrategy.UPPERCASE)
+
+    context = Context(config=Config())
+    context.upsert_model(
+        load_sql_based_model(
+            parse(
+                """
+                MODEL(
+                    name x,
+                    kind FULL
+                );
+
+                SELECT 1 AS c
+                """
+            )
+        )
+    )
+    context.plan("dev", auto_apply=True, no_prompts=True)
+    assert list(context.fetchdf('select c from "DEFAULT__DEV"."X"')["c"])[0] == 1
+
+
 def test_clear_caches(tmp_path: pathlib.Path):
     models_dir = tmp_path / "models"
 
@@ -461,6 +486,7 @@ def test_physical_schema_override(copy_to_temp_path: t.Callable) -> None:
 @pytest.mark.slow
 def test_janitor(sushi_context, mocker: MockerFixture) -> None:
     adapter_mock = mocker.MagicMock()
+    adapter_mock.dialect = "duckdb"
     state_sync_mock = mocker.MagicMock()
     state_sync_mock.delete_expired_environments.return_value = [
         Environment(
diff --git a/tests/core/test_environment.py b/tests/core/test_environment.py
index 7ddbdd8b..42b75b0c 100644
--- a/tests/core/test_environment.py
+++ b/tests/core/test_environment.py
@@ -3,13 +3,13 @@ import pytest
 from sqlmesh.core.environment import Environment, EnvironmentNamingInfo
 
 
-def test_normalize_name():
-    assert Environment.normalize_name("12foo!#%@") == "12foo____"
-    assert Environment.normalize_name("__test@$@%") == "__test____"
-    assert Environment.normalize_name("test") == "test"
-    assert Environment.normalize_name("-test_") == "_test_"
+def test_sanitize_name():
+    assert Environment.sanitize_name("12foo!#%@") == "12foo____"
+    assert Environment.sanitize_name("__test@$@%") == "__test____"
+    assert Environment.sanitize_name("test") == "test"
+    assert Environment.sanitize_name("-test_") == "_test_"
     with pytest.raises(TypeError, match="Expected str or Environment, got int"):
-        Environment.normalize_name(123)
+        Environment.sanitize_name(123)
 
 
 @pytest.mark.parametrize(
diff --git a/tests/core/test_plan.py b/tests/core/test_plan.py
index 2a17f73a..64fcf074 100644
--- a/tests/core/test_plan.py
+++ b/tests/core/test_plan.py
@@ -68,6 +68,7 @@ def test_forward_only_plan_sets_version(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -120,6 +121,7 @@ def test_forward_only_dev(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -178,6 +180,7 @@ def test_forward_only_metadata_change_dev(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -219,6 +222,7 @@ def test_forward_only_plan_added_models(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot_b.snapshot_id},
         removed_snapshots={},
@@ -265,6 +269,7 @@ def test_paused_forward_only_parent(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -295,6 +300,7 @@ def test_forward_only_plan_allow_destructive_models(
         environment="prod",
         is_new_environment=False,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -357,6 +363,7 @@ def test_forward_only_plan_allow_destructive_models(
         environment="prod",
         is_new_environment=False,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -413,6 +420,7 @@ def test_forward_only_model_on_destructive_change(
         environment="prod",
         is_new_environment=False,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -465,6 +473,7 @@ def test_forward_only_model_on_destructive_change(
         environment="prod",
         is_new_environment=False,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -542,6 +551,7 @@ def test_forward_only_model_on_destructive_change(
         environment="prod",
         is_new_environment=False,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -579,6 +589,7 @@ def test_forward_only_model_on_destructive_change_no_column_types(
         environment="prod",
         is_new_environment=False,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -618,6 +629,7 @@ def test_missing_intervals_lookback(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         modified_snapshots={},
@@ -771,6 +783,7 @@ def test_restate_symbolic_model(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -803,6 +816,7 @@ def test_restate_seed_model(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -825,6 +839,7 @@ def test_restate_missing_model(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -852,6 +867,7 @@ def test_new_snapshots_with_restatements(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -883,6 +899,7 @@ def test_end_validation(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot_a.snapshot_id},
         removed_snapshots={},
@@ -946,6 +963,7 @@ def test_forward_only_revert_not_allowed(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1002,6 +1020,7 @@ def test_forward_only_plan_seed_models(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1031,6 +1050,7 @@ def test_start_inference(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot_b.snapshot_id},
         removed_snapshots={},
@@ -1070,6 +1090,7 @@ def test_auto_categorization(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1108,6 +1129,7 @@ def test_auto_categorization_missing_schema_downstream(make_snapshot, mocker: Mo
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1142,6 +1164,7 @@ def test_broken_references(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={snapshot_a.snapshot_id: snapshot_a.table_info},
@@ -1175,6 +1198,7 @@ def test_broken_references_external_model(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={snapshot_a.snapshot_id: snapshot_a.table_info},
@@ -1214,6 +1238,7 @@ def test_effective_from(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1276,6 +1301,7 @@ def test_new_environment_no_changes(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1314,6 +1340,7 @@ def test_new_environment_with_changes(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1398,6 +1425,7 @@ def test_forward_only_models(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1440,6 +1468,7 @@ def test_forward_only_models_model_kind_changed(make_snapshot, mocker: MockerFix
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1487,6 +1516,7 @@ def test_indirectly_modified_forward_only_model(make_snapshot, mocker: MockerFix
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1546,6 +1576,7 @@ def test_added_model_with_forward_only_parent(make_snapshot, mocker: MockerFixtu
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot_b.snapshot_id},
         removed_snapshots={},
@@ -1579,6 +1610,7 @@ def test_added_forward_only_model(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot_a.snapshot_id, snapshot_b.snapshot_id},
         removed_snapshots={},
@@ -1615,6 +1647,7 @@ def test_disable_restatement(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=False,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1668,6 +1701,7 @@ def test_revert_to_previous_value(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1884,6 +1918,7 @@ def test_add_restatements(
         environment="test_environment",
         is_new_environment=False,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -1951,6 +1986,7 @@ def test_dev_plan_depends_past(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot.snapshot_id, snapshot_child.snapshot_id, unrelated_snapshot.snapshot_id},
         removed_snapshots={},
@@ -2054,6 +2090,7 @@ def test_dev_plan_depends_past_non_deployable(make_snapshot, mocker: MockerFixtu
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot_child.snapshot_id, unrelated_snapshot.snapshot_id},
         removed_snapshots={},
@@ -2130,6 +2167,7 @@ def test_models_selected_for_backfill(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=False,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot_b.snapshot_id},
         removed_snapshots={},
@@ -2188,6 +2226,7 @@ def test_categorized_uncategorized(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -2234,6 +2273,7 @@ def test_environment_previous_finalized_snapshots(make_snapshot, mocker: MockerF
         environment="test_environment",
         is_new_environment=False,
         is_unfinalized_environment=True,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot_b.snapshot_id},
         removed_snapshots={snapshot_c.snapshot_id: snapshot_c.table_info},
@@ -2293,6 +2333,7 @@ def test_metadata_change(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added=set(),
         removed_snapshots={},
@@ -2329,6 +2370,7 @@ def test_plan_start_when_preview_enabled(make_snapshot, mocker: MockerFixture):
         environment="test_environment",
         is_new_environment=True,
         is_unfinalized_environment=False,
+        normalize_environment_name=True,
         create_from="prod",
         added={snapshot.snapshot_id},
         removed_snapshots={},
diff --git a/tests/core/test_state_sync.py b/tests/core/test_state_sync.py
index b7738e88..71c876ce 100644
--- a/tests/core/test_state_sync.py
+++ b/tests/core/test_state_sync.py
@@ -1886,6 +1886,7 @@ def test_cleanup_expired_views(
     mocker: MockerFixture, state_sync: EngineAdapterStateSync, make_snapshot: t.Callable
 ):
     adapter = mocker.MagicMock()
+    adapter.dialect = None
     snapshot_a = make_snapshot(SqlModel(name="catalog.schema.a", query=parse_one("select 1, ds")))
     snapshot_a.categorize_as(SnapshotChangeCategory.BREAKING)
     snapshot_b = make_snapshot(SqlModel(name="catalog.schema.b", query=parse_one("select 1, ds")))
diff --git a/tests/schedulers/airflow/test_client.py b/tests/schedulers/airflow/test_client.py
index 10278698..ea086a63 100644
--- a/tests/schedulers/airflow/test_client.py
+++ b/tests/schedulers/airflow/test_client.py
@@ -156,6 +156,7 @@ def test_apply_plan(mocker: MockerFixture, snapshot: Snapshot):
                 }
             ],
             "suffix_target": "schema",
+            "normalize_name": True,
         },
         "no_gaps": False,
         "skip_backfill": False,