A few fixes (#4129)

Author: izeigerman

sqlmesh/core/context.py

@@ -364,7 +364,7 @@ def __init__(
         self._environment_statements: t.List[EnvironmentStatements] = []
         self._excluded_requirements: t.Set[str] = set()
         self._default_catalog: t.Optional[str] = None
-        self._default_catalog_per_gateway: t.Dict[str, str] = {}
+        self._default_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None
         self._linters: t.Dict[str, Linter] = {}
         self._loaded: bool = False
 
@@ -2227,7 +2227,7 @@ def engine_adapters(self) -> t.Dict[str, EngineAdapter]:
     @cached_property
     def default_catalog_per_gateway(self) -> t.Dict[str, str]:
         """Returns the catalogs for each engine adapter in a multi virtual layer setup when the catalog isn't shared."""
-        if self.gateway_managed_virtual_layer:
+        if self._default_catalog_per_gateway is None:
             self._default_catalog_per_gateway = {
                 name: adapter.default_catalog
                 for name, adapter in self.engine_adapters.items()

sqlmesh/core/context_diff.py

@@ -53,7 +53,9 @@ class ContextDiff(PydanticModel):
     """Whether the currently stored environment record is in unfinalized state."""
     normalize_environment_name: bool
     """Whether the environment name should be normalized."""
-    gateway_managed_virtual_layer: bool = False
+    previous_gateway_managed_virtual_layer: bool
+    """Whether the previous environment's virtual layer's views were created by the model specified gateways."""
+    gateway_managed_virtual_layer: bool
     """Whether the virtual layer's views will be created by the model specified gateways."""
     create_from: str
     """The name of the environment the target environment will be created from if new."""
@@ -121,7 +123,7 @@ def create(
         env = state_reader.get_environment(environment)
 
         create_from_env_exists = False
-        if env is None or env.expired or env.gateway_managed != gateway_managed_virtual_layer:
+        if env is None or env.expired:
             env = state_reader.get_environment(create_from.lower())
 
             if not env and create_from != c.PROD:
@@ -229,6 +231,7 @@ def create(
             diff_rendered=diff_rendered,
             previous_environment_statements=previous_environment_statements,
             environment_statements=environment_statements,
+            previous_gateway_managed_virtual_layer=env.gateway_managed if env else False,
             gateway_managed_virtual_layer=gateway_managed_virtual_layer,
         )
 
@@ -277,6 +280,7 @@ def has_changes(self) -> bool:
             or self.is_unfinalized_environment
             or self.has_requirement_changes
             or self.has_environment_statements_changes
+            or self.previous_gateway_managed_virtual_layer != self.gateway_managed_virtual_layer
         )
 
     @property

sqlmesh/core/state_sync/db/facade.py

@@ -199,10 +199,7 @@ def promote(
                 )
                 != table_infos[name].qualified_view_name.for_environment(environment.naming_info)
             }
-            if (
-                not existing_environment.expired
-                and existing_environment.gateway_managed == environment.gateway_managed
-            ):
+            if not existing_environment.expired:
                 if environment.previous_plan_id != existing_environment.plan_id:
                     raise ConflictingPlanError(
                         f"Plan '{environment.plan_id}' is no longer valid for the target environment '{environment.name}'. "
@@ -229,6 +226,7 @@ def promote(
             existing_environment
             and existing_environment.finalized_ts
             and not existing_environment.expired
+            and existing_environment.gateway_managed == environment.gateway_managed
         ):
             # Only promote new snapshots.
             added_table_infos -= set(existing_environment.promoted_snapshots)