From 70e431e9a3635598dd8f3a2977587b5bbbcbcd0e Mon Sep 17 00:00:00 2001 From: Iaroslav Zeigerman Date: Fri, 11 Apr 2025 09:26:21 -0700 Subject: [PATCH] A few fixes --- sqlmesh/core/context.py | 4 ++-- sqlmesh/core/context_diff.py | 8 ++++++-- sqlmesh/core/state_sync/db/facade.py | 6 ++---- 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py index f9e9bdc87d..af746c0cbc 100644 --- a/sqlmesh/core/context.py +++ b/sqlmesh/core/context.py @@ -363,7 +363,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 @@ -2222,7 +2222,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() diff --git a/sqlmesh/core/context_diff.py b/sqlmesh/core/context_diff.py index eb05a61155..c7e9dea9a0 100644 --- a/sqlmesh/core/context_diff.py +++ b/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 diff --git a/sqlmesh/core/state_sync/db/facade.py b/sqlmesh/core/state_sync/db/facade.py index e5f7493e2a..9e021995eb 100644 --- a/sqlmesh/core/state_sync/db/facade.py +++ b/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)