fix(clickhouse): warn at plan time when virtual catalog triggers re-materialization

When a catalog-aware gateway is added to an existing ClickHouse-only
project, ClickHouse model FQNs change from 2-level to 3-level, causing
SQLMesh to treat them as brand-new snapshots. This triggers full
re-materialization — including historical backfills for incremental models —
with no user-visible signal before apply.

Add _warn_if_virtual_catalog_rematerialization() called from plan()
between build() and console.plan(). It detects when new 3-level CH
snapshots map to existing 2-level names in the current environment and
emits a console warning listing the affected models with a plain-language
explanation of the cost (FULL = recreate once, IBTR = full backfill).

Signed-off-by: Michael Day <michael.day@cloudkitchens.com>
Signed-off-by: mday-io <mdaytn@gmail.com>

Author: mday-io

sqlmesh/core/context.py

@@ -1445,6 +1445,8 @@ def plan(
 
         plan = plan_builder.build()
 
+        self._warn_if_virtual_catalog_rematerialization(plan)
+
         if no_auto_categorization or plan.uncategorized:
             # Prompts are required if the auto categorization is disabled
             # or if there are any uncategorized snapshots in the plan
@@ -2743,6 +2745,61 @@ def _run_plan_tests(self, skip_tests: bool = False) -> t.Optional[ModelTextTestR
             return result
         return None
 
+    def _warn_if_virtual_catalog_rematerialization(self, plan: "Plan") -> None:
+        """Warn when ClickHouse models appear as new snapshots solely because a virtual catalog
+        prefix was added to their FQNs after a catalog-aware gateway joined the project.
+
+        This situation causes every previously-applied ClickHouse model to be treated as brand-new
+        by SQLMesh, triggering full re-materialization and historical backfills. Emitting a warning
+        before the plan is displayed gives users a chance to understand the cost before applying.
+        """
+        from sqlglot import exp
+
+        # Collect the set of old 2-level snapshot names from the current environment so we can
+        # detect which new 3-level names are renames rather than genuinely new models.
+        old_names: t.Set[str] = set()
+        for s_id in plan.context_diff.removed_snapshots:
+            old_names.add(s_id.name)
+        for name in plan.context_diff.snapshots_by_name:
+            old_names.add(name)
+
+        affected: t.List[t.Tuple[str, str]] = []  # (new_3level_name, old_2level_name)
+
+        for gateway, adapter in self.engine_adapters.items():
+            if not adapter.supports_virtual_catalog() or not adapter._default_catalog:
+                continue
+            virtual_catalog = adapter._default_catalog
+
+            for snapshot in plan.new_snapshots:
+                table = exp.to_table(snapshot.name)
+                if table.catalog != virtual_catalog:
+                    continue
+                # Reconstruct the 2-level name that would have been used before injection.
+                old_name = f"{table.db}.{table.name}"
+                if old_name in old_names:
+                    affected.append((snapshot.name, old_name))
+
+        if not affected:
+            return
+
+        max_display = 10
+        model_lines = "\n".join(
+            f"  - {new_name}  (was: {old_name})" for new_name, old_name in affected[:max_display]
+        )
+        if len(affected) > max_display:
+            model_lines += f"\n  ... and {len(affected) - max_display} more"
+
+        self.console.log_warning(
+            "ClickHouse models are being re-materialized due to virtual catalog FQN change.\n\n"
+            "The following ClickHouse models appear as new because their fully-qualified\n"
+            "names changed from 2-level (db.table) to 3-level (__gateway__.db.table):\n\n"
+            f"{model_lines}\n\n"
+            "FULL models will be recreated once. INCREMENTAL_BY_TIME_RANGE models will\n"
+            "require a full historical backfill from their configured start date.\n\n"
+            "This is a one-time cost when first adding a catalog-aware gateway to an\n"
+            "existing ClickHouse project. To proceed, run `sqlmesh apply`."
+        )
+
     @property
     def _model_tables(self) -> t.Dict[str, str]:
         """Mapping of model name to physical table name.

tests/core/test_context.py

@@ -677,6 +677,137 @@ def _capture_create_schema(
     )
 
 
+@pytest.mark.fast
+def test_warn_if_virtual_catalog_rematerialization_emits_warning(mocker):
+    """_warn_if_virtual_catalog_rematerialization must emit a log_warning when new snapshots have
+    3-level FQNs that map to existing 2-level FQNs in the current environment, indicating that the
+    virtual catalog prefix was added to previously-applied ClickHouse models."""
+    from unittest.mock import MagicMock
+
+    from sqlmesh.core.engine_adapter.clickhouse import ClickhouseEngineAdapter
+    from sqlmesh.core.snapshot.definition import SnapshotId
+
+    # Build a minimal Context with no models.
+    ctx = Context(config=Config())
+
+    # Create a ClickHouse adapter with a virtual catalog already injected.
+    ch_adapter = ClickhouseEngineAdapter(
+        lambda *a, **k: mocker.NonCallableMock(),
+        dialect="clickhouse",
+    )
+    ch_adapter._default_catalog = "__ch_gw__"
+
+    # Override engine_adapters so the context sees our prepared adapter.
+    mocker.patch.object(
+        type(ctx), "engine_adapters", new_callable=PropertyMock, return_value={"ch_gw": ch_adapter}
+    )
+
+    # Build a mock snapshot with a 3-level name that has the virtual catalog prefix.
+    new_snapshot = MagicMock()
+    new_snapshot.name = "__ch_gw__.mydb.my_table"
+
+    # The old 2-level name must appear in snapshots_by_name so we detect the rename.
+    old_snapshot_id = SnapshotId(name="mydb.my_table", identifier="abc123")
+
+    context_diff = MagicMock()
+    context_diff.new_snapshots = {new_snapshot.name: new_snapshot}
+    context_diff.removed_snapshots = {}
+    context_diff.snapshots_by_name = {"mydb.my_table": MagicMock()}
+
+    plan = MagicMock()
+    plan.new_snapshots = [new_snapshot]
+    plan.context_diff = context_diff
+
+    warning_mock = mocker.patch.object(ctx.console, "log_warning")
+
+    ctx._warn_if_virtual_catalog_rematerialization(plan)
+
+    warning_mock.assert_called_once()
+    warning_text = warning_mock.call_args[0][0]
+    assert "__ch_gw__" in warning_text
+    assert "mydb.my_table" in warning_text
+
+
+@pytest.mark.fast
+def test_warn_if_virtual_catalog_rematerialization_no_warning_when_genuinely_new(mocker):
+    """_warn_if_virtual_catalog_rematerialization must NOT warn when there is no matching old
+    2-level name — i.e. the model is a brand-new model, not a renamed existing one."""
+    from unittest.mock import MagicMock
+
+    from sqlmesh.core.engine_adapter.clickhouse import ClickhouseEngineAdapter
+
+    ctx = Context(config=Config())
+
+    ch_adapter = ClickhouseEngineAdapter(
+        lambda *a, **k: mocker.NonCallableMock(),
+        dialect="clickhouse",
+    )
+    ch_adapter._default_catalog = "__ch_gw__"
+
+    mocker.patch.object(
+        type(ctx), "engine_adapters", new_callable=PropertyMock, return_value={"ch_gw": ch_adapter}
+    )
+
+    new_snapshot = MagicMock()
+    new_snapshot.name = "__ch_gw__.mydb.brand_new_table"
+
+    context_diff = MagicMock()
+    context_diff.new_snapshots = {new_snapshot.name: new_snapshot}
+    context_diff.removed_snapshots = {}
+    # No matching old name.
+    context_diff.snapshots_by_name = {}
+
+    plan = MagicMock()
+    plan.new_snapshots = [new_snapshot]
+    plan.context_diff = context_diff
+
+    warning_mock = mocker.patch.object(ctx.console, "log_warning")
+
+    ctx._warn_if_virtual_catalog_rematerialization(plan)
+
+    warning_mock.assert_not_called()
+
+
+@pytest.mark.fast
+def test_warn_if_virtual_catalog_rematerialization_no_warning_without_virtual_catalog(mocker):
+    """_warn_if_virtual_catalog_rematerialization must NOT warn when the ClickHouse adapter has no
+    virtual catalog injected (i.e. _default_catalog is None)."""
+    from unittest.mock import MagicMock
+
+    from sqlmesh.core.engine_adapter.clickhouse import ClickhouseEngineAdapter
+
+    ctx = Context(config=Config())
+
+    ch_adapter = ClickhouseEngineAdapter(
+        lambda *a, **k: mocker.NonCallableMock(),
+        dialect="clickhouse",
+    )
+    # No virtual catalog injected — adapter stays at 2-level mode.
+    assert ch_adapter._default_catalog is None
+
+    mocker.patch.object(
+        type(ctx), "engine_adapters", new_callable=PropertyMock, return_value={"ch_gw": ch_adapter}
+    )
+
+    new_snapshot = MagicMock()
+    new_snapshot.name = "mydb.my_table"
+
+    context_diff = MagicMock()
+    context_diff.new_snapshots = {new_snapshot.name: new_snapshot}
+    context_diff.removed_snapshots = {}
+    context_diff.snapshots_by_name = {}
+
+    plan = MagicMock()
+    plan.new_snapshots = [new_snapshot]
+    plan.context_diff = context_diff
+
+    warning_mock = mocker.patch.object(ctx.console, "log_warning")
+
+    ctx._warn_if_virtual_catalog_rematerialization(plan)
+
+    warning_mock.assert_not_called()
+
+
 def test_plan_execution_time():
     context = Context(config=Config())
     context.upsert_model(