From 05f814126690db19cdcfc1ec58e9a3ae149d7b06 Mon Sep 17 00:00:00 2001
From: brucearctor <5032356+brucearctor@users.noreply.github.com>
Date: Sun, 29 Mar 2026 23:11:29 -0700
Subject: [PATCH] fix: prevent default catalog leak into catalog-unsupported
 gateways (#5748)

When a multi-gateway project uses a default gateway with a catalog
(e.g. Trino), that catalog was silently prepended to model names
targeting secondary gateways that don't support catalogs (e.g.
ClickHouse), causing UnsupportedCatalogOperationError at evaluation.

Root cause: catalog-unsupported adapters return None for
default_catalog and are never added to default_catalog_per_gateway.
The model loader cannot distinguish 'no catalog' from 'not checked',
so the global default_catalog leaks through.

Fix: explicitly register catalog-unsupported gateways with empty
string in the per-gateway dict. The model loader's 'is not None'
check picks this up and overrides default_catalog to '', preventing
the leak.

Signed-off-by: Bruce Arctor <brucearctor@users.noreply.github.com>
---
 sqlmesh/core/config/scheduler.py |  2 +
 tests/core/test_model.py         | 64 ++++++++++++++++++++++++++++++++
 2 files changed, 66 insertions(+)

diff --git a/sqlmesh/core/config/scheduler.py b/sqlmesh/core/config/scheduler.py
index 970defee62..1911ff2241 100644
--- a/sqlmesh/core/config/scheduler.py
+++ b/sqlmesh/core/config/scheduler.py
@@ -141,6 +141,8 @@ def get_default_catalog_per_gateway(self, context: GenericContext) -> t.Dict[str
         for gateway, adapter in context.engine_adapters.items():
             if catalog := adapter.default_catalog:
                 default_catalogs_per_gateway[gateway] = catalog
+            elif adapter.catalog_support.is_unsupported:
+                default_catalogs_per_gateway[gateway] = ""
         return default_catalogs_per_gateway
 
 
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 81707c075f..e485707f36 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -7186,6 +7186,70 @@ def test_gateway_macro_jinja() -> None:
     assert model.render_query_or_raise().sql() == "SELECT 'in_memory' AS \"gateway_jinja\""
 
 
+def test_default_catalog_not_leaked_to_unsupported_gateway() -> None:
+    """Regression test for https://github.com/SQLMesh/sqlmesh/issues/5748.
+
+    When a model targets a gateway that does not support catalogs (e.g. ClickHouse),
+    the default gateway's catalog must not leak into the model's FQN. The scheduler
+    registers such gateways with empty string in default_catalog_per_gateway, and
+    the model loader should use that to override the global default_catalog.
+    """
+    expressions = d.parse(
+        """
+        MODEL (
+            name my_schema.my_model,
+            kind VIEW,
+            gateway clickhouse_gw,
+            dialect clickhouse,
+        );
+        SELECT 1 AS id
+        """
+    )
+
+    # Simulate a multi-gateway setup: default gateway has catalog "example_catalog",
+    # but clickhouse_gw is catalog-unsupported (registered with empty string).
+    models = load_sql_based_models(
+        expressions,
+        get_variables=lambda gw: {},
+        dialect="clickhouse",
+        default_catalog="example_catalog",
+        default_catalog_per_gateway={
+            "default_gw": "example_catalog",
+            "clickhouse_gw": "",
+        },
+    )
+
+    assert len(models) == 1
+    model_result = models[0]
+    # The catalog should be empty — not the leaked "example_catalog"
+    assert model_result.catalog != "example_catalog"
+    # The FQN should be a two-part name, not three-part with the leaked catalog
+    assert "example_catalog" not in model_result.fqn
+
+    # Verify the positive case: without the per-gateway override, catalog leaks
+    models_leaked = load_sql_based_models(
+        d.parse(
+            """
+            MODEL (
+                name my_schema.my_model2,
+                kind VIEW,
+                gateway clickhouse_gw,
+                dialect clickhouse,
+            );
+            SELECT 1 AS id
+            """
+        ),
+        get_variables=lambda gw: {},
+        dialect="clickhouse",
+        default_catalog="example_catalog",
+        # No entry for clickhouse_gw — catalog will leak
+        default_catalog_per_gateway={"default_gw": "example_catalog"},
+    )
+    assert len(models_leaked) == 1
+    # Without the fix, example_catalog leaks into the model name
+    assert "example_catalog" in models_leaked[0].fqn
+
+
 def test_gateway_python_model(mocker: MockerFixture) -> None:
     @model(
         "test_gateway_python_model",