SQLMesh
diff --git a/‎sqlmesh/core/context.py‎
Lines changed: 1 addition & 1 deletion b/‎sqlmesh/core/context.py‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎sqlmesh/core/context_diff.py‎
Lines changed: 2 additions & 0 deletions b/‎sqlmesh/core/context_diff.py‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎sqlmesh/core/snapshot/evaluator.py‎
Lines changed: 2 additions & 0 deletions b/‎sqlmesh/core/snapshot/evaluator.py‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎tests/core/test_context.py‎
Lines changed: 2 additions & 8 deletions b/‎tests/core/test_context.py‎
Lines changed: 2 additions & 8 deletions
diff --git a/‎tests/core/test_integration.py‎
Lines changed: 16 additions & 1 deletion b/‎tests/core/test_integration.py‎
Lines changed: 16 additions & 1 deletion
@@ -2226,7 +2226,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."""
+        """Returns the default catalogs for each engine adapter."""
         if self._default_catalog_per_gateway is None:
             self._default_catalog_per_gateway = {
                 name: adapter.default_catalog
 
@@ -270,6 +270,8 @@ def create_no_diff(cls, environment: str, state_reader: StateReader) -> ContextD
             previous_requirements=env.requirements,
             requirements=env.requirements,
             previous_environment_statements=[],
+            previous_gateway_managed_virtual_layer=env.gateway_managed,
+            gateway_managed_virtual_layer=env.gateway_managed,
         )
 
     @property
 
@@ -239,6 +239,7 @@ def promote(
                 )
                 tables_by_gateway[gateway].append(table)
 
+        # A schema can be shared across multiple engines, so we need to group by gateway
         for gateway, tables in tables_by_gateway.items():
             self._create_schemas(tables=tables, gateway=gateway)
 
@@ -337,6 +338,7 @@ def _get_data_objects(
 
         with self.concurrent_context():
             existing_objects: t.Set[str] = set()
+            # A schema can be shared across multiple engines, so we need to group tables by both gateway and schema
             for gateway, tables_by_schema in tables_by_gateway_and_schema.items():
                 objs_for_gateway = {
                     obj
 
@@ -327,21 +327,15 @@ def test_evaluate_limit():
 def test_gateway_specific_adapters(copy_to_temp_path, mocker):
     path = copy_to_temp_path("examples/sushi")
     ctx = Context(paths=path, config="isolated_systems_config", gateway="prod")
-    assert len(ctx._engine_adapters) == 1
+    assert len(ctx._engine_adapters) == 3
     assert ctx.engine_adapter == ctx._engine_adapters["prod"]
-
-    with pytest.raises(SQLMeshError):
-        assert ctx._get_engine_adapter("non_existing")
-
-    # This will create the requested engine adapter
     assert ctx._get_engine_adapter("dev") == ctx._engine_adapters["dev"]
 
     ctx = Context(paths=path, config="isolated_systems_config")
-    assert len(ctx._engine_adapters) == 1
+    assert len(ctx._engine_adapters) == 3
     assert ctx.engine_adapter == ctx._engine_adapters["dev"]
 
     ctx = Context(paths=path, config="isolated_systems_config")
-
     assert len(ctx.engine_adapters) == 3
     assert ctx.engine_adapter == ctx._get_engine_adapter()
     assert ctx._get_engine_adapter("test") == ctx._engine_adapters["test"]
 
@@ -11,6 +11,7 @@
 import pytest
 from pathlib import Path
 import os
+from sqlmesh.utils.concurrency import NodeExecutionFailedError
 import time_machine
 from pytest_mock.plugin import MockerFixture
 from sqlglot import exp
@@ -4497,7 +4498,7 @@ def test_multi(mocker):
 
 @use_terminal_console
 def test_multi_virtual_layer(mocker):
-    context = Context(paths=["examples/multi_virtual_layer"])
+    context = Context(paths=["tests/fixtures/multi_virtual_layer"])
 
     local_db = "db.duckdb"
     if os.path.exists(local_db):
@@ -4569,6 +4570,20 @@ def test_multi_virtual_layer(mocker):
         == "     item_id  global_one  macro_one extra\n0  gateway_2          88          1     c"
     )
 
+    # Changing the flag should show a diff
+    context.gateway_managed_virtual_layer = False
+    plan = context.plan_builder().build()
+    assert not plan.requires_backfill
+    assert (
+        plan.context_diff.previous_gateway_managed_virtual_layer
+        != plan.context_diff.gateway_managed_virtual_layer
+    )
+    assert plan.context_diff.has_changes
+
+    # This should error since the default_gateway won't have access to create the view on a non-shared catalog
+    with pytest.raises(NodeExecutionFailedError, match=r"Execution failed for node SnapshotId*"):
+        context.apply(plan)
+
     if os.path.exists(local_db):
         os.remove(local_db)
Original file line number	Diff line number	Diff line change
@@ -270,6 +270,8 @@ def create_no_diff(cls, environment: str, state_reader: StateReader) -> ContextD
270	270	previous_requirements=env.requirements,
271	271	requirements=env.requirements,
272	272	previous_environment_statements=[],
	273	+ previous_gateway_managed_virtual_layer=env.gateway_managed,
	274	+ gateway_managed_virtual_layer=env.gateway_managed,
273	275	)
274	276
275	277	@property