feat: support gateway virtual layer catalogs

Signed-off-by: Joseph Finlayson <joseph.finlayson@gmail.com>

Author: josephfinlayson

docs/guides/multi_engine.md

@@ -134,6 +134,101 @@ If your project's engines don’t have a mutually accessible catalog or your raw
 
 To enable this, set `gateway_managed_virtual_layer` to `true` in your configuration. By default, this flag is set to false.
 
+Each gateway normally uses the model's catalog for both physical snapshot tables and virtual-layer
+views. For an unqualified model name, that is the gateway connection's default catalog. Set the
+gateway's `virtual_layer_catalog` when those layers should use different catalogs. This setting does
+not change the physical snapshot location; it publishes that gateway's model views in
+`virtual_layer_catalog`.
+
+This is useful when multiple projects share SQLMesh state and one production environment, but
+each project owns a separate physical catalog and publishes stable model names through a separate
+catalog. Configure the route once on the gateway rather than on each model:
+
+=== "YAML"
+
+    ```yaml linenums="1"
+    gateways:
+      project_a:
+        connection:
+          type: databricks
+          catalog: project_a_physical
+          server_hostname: <server_hostname>
+          http_path: <http_path>
+          access_token: <access_token>
+        virtual_layer_catalog: project_a_published
+      project_b:
+        connection:
+          type: databricks
+          catalog: project_b_physical
+          server_hostname: <server_hostname>
+          http_path: <http_path>
+          access_token: <access_token>
+        virtual_layer_catalog: project_b_published
+
+    model_defaults:
+      dialect: databricks
+      gateway: project_a
+
+    gateway_managed_virtual_layer: true
+    ```
+
+=== "Python"
+
+    ```python linenums="1"
+    from sqlmesh.core.config import Config, DatabricksConnectionConfig, GatewayConfig, ModelDefaultsConfig
+
+    config = Config(
+        gateways={
+            "project_a": GatewayConfig(
+                connection=DatabricksConnectionConfig(
+                    catalog="project_a_physical",
+                    server_hostname="<server_hostname>",
+                    http_path="<http_path>",
+                    access_token="<access_token>",
+                ),
+                virtual_layer_catalog="project_a_published",
+            ),
+            "project_b": GatewayConfig(
+                connection=DatabricksConnectionConfig(
+                    catalog="project_b_physical",
+                    server_hostname="<server_hostname>",
+                    http_path="<http_path>",
+                    access_token="<access_token>",
+                ),
+                virtual_layer_catalog="project_b_published",
+            ),
+        },
+        model_defaults=ModelDefaultsConfig(dialect="databricks", gateway="project_a"),
+        gateway_managed_virtual_layer=True,
+    )
+    ```
+
+The example focuses on catalog routing. When separate workflows operate on the projects, configure
+them to use the same state backend or scheduler so they read and update the same SQLMesh
+environment. See the [multi-repository guide](multi_repo.md) for the shared-state configuration.
+
+For a model named `sales.orders` using the `project_a` gateway, this configuration stores
+versioned snapshot tables under `project_a_physical` and creates the environment's `sales.orders`
+view under `project_a_published`. SQLMesh resolves and stores this route with the snapshot, so a
+later plan that loads only one project retains the routes of models loaded from shared state.
+
+The gateway route is the base catalog for normal environment naming. Existing naming settings
+retain their precedence:
+
+- A matching `environment_catalog_mapping` target overrides `virtual_layer_catalog`.
+- With `environment_suffix_target: catalog`, a development environment suffix is appended to
+  `virtual_layer_catalog`.
+- If neither setting changes the catalog, `virtual_layer_catalog` is used as configured.
+
+Changing only `virtual_layer_catalog` is a metadata-only model change. Applying the plan creates
+the view at its new location and removes the view at its previous location without rebuilding the
+physical snapshot table. Ensure the gateway can create and drop objects in both catalogs during a
+route change. All composed project configurations that define the same gateway name must agree on
+its `virtual_layer_catalog`; SQLMesh rejects conflicting routes.
+
+`virtual_layer_catalog` is gateway-level configuration. It cannot be set in a `MODEL` block or in
+`model_defaults`. External models are source definitions and do not inherit this publishing route.
+
 #### Example: Redshift + Athena + Snowflake
 
 Consider a scenario where you need to create a project with models in Redshift, Athena and Snowflake, where each engine hosts its models' virtual layer views.

docs/reference/configuration.md

@@ -186,6 +186,7 @@ NOTE: Spark and Trino engines may not be used for the state connection.
 | `connection`       | The data warehouse connection for core SQLMesh functions.                                                                                                                       | [connection configuration](#connection)                      | N (if [`default_connection`](#default-connectionsscheduler) specified) |
 | `state_connection` | The data warehouse connection where SQLMesh will store internal information about the project. (Default: `connection` if using builtin scheduler, otherwise scheduler database) | [connection configuration](#connection)                      | N                                                                      |
 | `state_schema`     | The name of the schema where state information should be stored. (Default: `sqlmesh`)                                                                                           | string                                                       | N                                                                      |
+| `virtual_layer_catalog` | The catalog where this gateway publishes virtual-layer views. This does not change the catalog used for physical snapshot tables. Intended for use with [`gateway_managed_virtual_layer`](../guides/multi_engine.md#gateway-managed-virtual-layer). | string | N |
 | `test_connection`  | The data warehouse connection SQLMesh will use to execute tests. (Default: `connection`)                                                                                        | [connection configuration](#connection)                      | N                                                                      |
 | `scheduler`        | The scheduler SQLMesh will use to execute tests. (Default: `builtin`)                                                                                                           | [scheduler configuration](#scheduler)                        | N                                                                      |
 | `variables`        | The gateway-specific variables which override the root-level [variables](#variables) by key.                                                                                    | dict[string, int \| float \| bool \| string \| list \| dict] | N                                                                      |

sqlmesh/core/config/gateway.py

@@ -11,6 +11,8 @@
     connection_config_validator,
 )
 from sqlmesh.core.config.scheduler import SchedulerConfig, scheduler_config_validator
+from sqlmesh.utils.errors import ConfigError
+from sqlmesh.utils.pydantic import field_validator
 
 
 class GatewayConfig(BaseConfig):
@@ -25,6 +27,8 @@ class GatewayConfig(BaseConfig):
         scheduler: The scheduler configuration.
         state_schema: Schema name to use for the state tables. If None or empty string are provided
             then no schema name is used and therefore the default schema defined for the connection will be used
+        virtual_layer_catalog: Optional catalog in which this gateway publishes virtual-layer views.
+            This does not change the catalog used for physical snapshot tables.
         variables: A dictionary of gateway-specific variables that can be used in models / macros. This overrides
             root-level variables by key.
     """
@@ -34,9 +38,17 @@ class GatewayConfig(BaseConfig):
     test_connection: t.Optional[SerializableConnectionConfig] = None
     scheduler: t.Optional[SchedulerConfig] = None
     state_schema: t.Optional[str] = c.SQLMESH
+    virtual_layer_catalog: t.Optional[str] = None
     variables: t.Dict[str, t.Any] = {}
     model_defaults: t.Optional[ModelDefaultsConfig] = None
 
     _connection_config_validator = connection_config_validator
     _scheduler_config_validator = scheduler_config_validator
     _variables_validator = variables_validator
+
+    @field_validator("virtual_layer_catalog")
+    @classmethod
+    def _validate_virtual_layer_catalog(cls, value: t.Optional[str]) -> t.Optional[str]:
+        if value is not None and not value.strip():
+            raise ConfigError("virtual_layer_catalog cannot be an empty string.")
+        return value

sqlmesh/core/context.py

@@ -2897,6 +2897,23 @@ def default_catalog_per_gateway(self) -> t.Dict[str, str]:
         """Returns the default catalogs for each engine adapter."""
         return self._scheduler.get_default_catalog_per_gateway(self)
 
+    @cached_property
+    def virtual_layer_catalog_per_gateway(self) -> t.Dict[str, str]:
+        """Returns configured virtual-layer catalogs keyed by gateway name."""
+        catalogs: t.Dict[str, str] = {}
+        for config in self.configs.values():
+            for gateway_name, gateway in config.gateways.items():
+                if not gateway.virtual_layer_catalog:
+                    continue
+                existing = catalogs.get(gateway_name)
+                if existing is not None and existing != gateway.virtual_layer_catalog:
+                    raise ConfigError(
+                        f"Gateway '{gateway_name}' has conflicting virtual_layer_catalog values "
+                        f"'{existing}' and '{gateway.virtual_layer_catalog}'."
+                    )
+                catalogs[gateway_name] = gateway.virtual_layer_catalog
+        return catalogs
+
     @property
     def concurrent_tasks(self) -> int:
         if self._concurrent_tasks is None:

sqlmesh/core/loader.py

@@ -3,6 +3,7 @@
 import abc
 import glob
 import itertools
+import json
 import linecache
 import os
 import re
@@ -600,6 +601,8 @@ def _load_sql_models(
                 infer_names=self.config.model_naming.infer_names,
                 signal_definitions=signals,
                 default_catalog_per_gateway=self.context.default_catalog_per_gateway,
+                virtual_layer_catalog_per_gateway=self.context.virtual_layer_catalog_per_gateway,
+                selected_gateway=self.context.selected_gateway,
                 virtual_environment_mode=self.config.virtual_environment_mode,
             )
 
@@ -680,6 +683,8 @@ def _load_python_models(
                             audit_definitions=audits,
                             signal_definitions=signals,
                             default_catalog_per_gateway=self.context.default_catalog_per_gateway,
+                            virtual_layer_catalog_per_gateway=self.context.virtual_layer_catalog_per_gateway,
+                            selected_gateway=self.context.selected_gateway,
                             virtual_environment_mode=self.config.virtual_environment_mode,
                         ):
                             if model.enabled:
@@ -951,5 +956,12 @@ def _model_cache_entry_id(self, model_path: Path) -> str:
                     # model's python environment if the @gateway macro variable is
                     # used in the model
                     self._loader.context.gateway or self._loader.config.default_gateway_name,
+                    # A model's virtual-layer catalog can come from another composed
+                    # project's gateway configuration, so the local config fingerprint
+                    # alone is not sufficient to invalidate this cache entry.
+                    json.dumps(
+                        self._loader.context.virtual_layer_catalog_per_gateway,
+                        sort_keys=True,
+                    ),
                 ]
             )

sqlmesh/core/model/decorator.py

@@ -71,6 +71,11 @@ def __init__(self, name: t.Optional[str] = None, is_sql: bool = False, **kwargs:
 
         if "default_catalog" in kwargs:
             raise ConfigError("`default_catalog` cannot be set on a per-model basis.")
+        if "virtual_layer_catalog" in kwargs:
+            raise ConfigError(
+                "`virtual_layer_catalog` cannot be set on a per-model basis. "
+                "Configure it on the model's gateway instead."
+            )
 
         self.columns = {
             column_name: (
@@ -97,6 +102,8 @@ def models(
         module_path: Path,
         dialect: t.Optional[str] = None,
         default_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None,
+        virtual_layer_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None,
+        selected_gateway: t.Optional[str] = None,
         **loader_kwargs: t.Any,
     ) -> t.List[Model]:
         blueprints = self.kwargs.pop("blueprints", None)
@@ -138,6 +145,8 @@ def models(
             module_path=module_path,
             dialect=dialect,
             default_catalog_per_gateway=default_catalog_per_gateway,
+            virtual_layer_catalog_per_gateway=virtual_layer_catalog_per_gateway,
+            selected_gateway=selected_gateway,
             **loader_kwargs,
         )
 
@@ -160,6 +169,7 @@ def model(
         infer_names: t.Optional[bool] = False,
         blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
         virtual_environment_mode: VirtualEnvironmentMode = VirtualEnvironmentMode.default,
+        resolved_virtual_layer_catalog: t.Optional[str] = None,
     ) -> Model:
         """Get the model registered by this function."""
         env: t.Dict[str, t.Tuple[t.Any, t.Optional[bool]]] = {}
@@ -237,6 +247,7 @@ def model(
             "signal_definitions": signal_definitions,
             "blueprint_variables": blueprint_variables,
             "virtual_environment_mode": virtual_environment_mode,
+            "resolved_virtual_layer_catalog": resolved_virtual_layer_catalog,
             **rendered_fields,
         }
 

sqlmesh/core/model/definition.py

@@ -242,7 +242,12 @@ def render_definition(
                             value=exp.to_table(field_value, dialect=self.dialect),
                         )
                     )
-                elif field_name not in ("default_catalog", "enabled", "ignored_rules_"):
+                elif field_name not in (
+                    "default_catalog",
+                    "enabled",
+                    "ignored_rules_",
+                    "virtual_layer_catalog",
+                ):
                     expressions.append(
                         exp.Property(
                             this=field_info.alias or field_name,
@@ -1239,6 +1244,10 @@ def metadata_hash(self) -> str:
                 self.grants_target_layer,
             ]
 
+            # Preserve existing metadata hashes when no gateway publishing route is configured.
+            if virtual_layer_catalog := getattr(self, "virtual_layer_catalog", None):
+                metadata.append(virtual_layer_catalog)
+
             for key, value in (self.virtual_properties or {}).items():
                 metadata.append(key)
                 metadata.append(gen(value))
@@ -2063,6 +2072,8 @@ def create_models_from_blueprints(
     module_path: Path = Path(),
     dialect: DialectType = None,
     default_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None,
+    virtual_layer_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None,
+    selected_gateway: t.Optional[str] = None,
     **loader_kwargs: t.Any,
 ) -> t.List[Model]:
     model_blueprints: t.List[Model] = []
@@ -2105,6 +2116,13 @@ def create_models_from_blueprints(
                 # default from the primary gateway doesn't leak into this model's name.
                 loader_kwargs["default_catalog"] = None
 
+        effective_gateway = gateway_name or selected_gateway
+        loader_kwargs["resolved_virtual_layer_catalog"] = (
+            virtual_layer_catalog_per_gateway.get(effective_gateway)
+            if virtual_layer_catalog_per_gateway and effective_gateway
+            else None
+        )
+
         model_blueprints.append(
             loader(
                 path=path,
@@ -2126,6 +2144,8 @@ def load_sql_based_models(
     module_path: Path = Path(),
     dialect: DialectType = None,
     default_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None,
+    virtual_layer_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None,
+    selected_gateway: t.Optional[str] = None,
     **loader_kwargs: t.Any,
 ) -> t.List[Model]:
     gateway: t.Optional[exp.Expr] = None
@@ -2170,6 +2190,8 @@ def load_sql_based_models(
         module_path=module_path,
         dialect=dialect,
         default_catalog_per_gateway=default_catalog_per_gateway,
+        virtual_layer_catalog_per_gateway=virtual_layer_catalog_per_gateway,
+        selected_gateway=selected_gateway,
         **loader_kwargs,
     )
 
@@ -2581,8 +2603,16 @@ def _create_model(
     variables: t.Optional[t.Dict[str, t.Any]] = None,
     blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
     use_original_sql: bool = False,
+    resolved_virtual_layer_catalog: t.Optional[str] = None,
     **kwargs: t.Any,
 ) -> Model:
+    if "virtual_layer_catalog" in kwargs:
+        raise_config_error(
+            "`virtual_layer_catalog` cannot be set on a per-model basis. "
+            "Configure it on the model's gateway instead.",
+            path,
+        )
+
     validate_extra_and_required_fields(
         klass,
         {"name", *kwargs} - {"grain", "table_properties"},
@@ -2614,6 +2644,7 @@ def _create_model(
         # external_models.yaml, so it must remain explicit rather than inheriting the
         # gateway used to execute managed models in the project.
         defaults.pop("gateway", None)
+        resolved_virtual_layer_catalog = None
     if not issubclass(klass, SqlModel):
         defaults.pop("optimize_query", None)
 
@@ -2687,6 +2718,7 @@ def _create_model(
             "dialect": dialect,
             "depends_on": depends_on,
             "physical_schema_override": physical_schema_override,
+            "virtual_layer_catalog": resolved_virtual_layer_catalog,
             **kwargs,
         },
     )

sqlmesh/core/model/meta.py

@@ -114,6 +114,9 @@ class ModelMeta(_Node):
     enabled: bool = True
     physical_version: t.Optional[str] = None
     gateway: t.Optional[str] = None
+    # Resolved from GatewayConfig during loading. This is persisted with the model so virtual
+    # layer routing remains available when the originating project configuration is not loaded.
+    virtual_layer_catalog: t.Optional[str] = None
     optimize_query: t.Optional[bool] = None
     ignored_rules_: t.Optional[t.Set[str]] = Field(
         default=None, exclude=True, alias="ignored_rules"