`ClickhouseEngineAdapter._exchange_tables` silently swallows non-`NOT_IMPLEMENTED` swap failures → FULL model reports success over stale data

[davidstravito]

Version: 0.236.1 (also present in 0.230.1, 0.236.2 and current main)
Engine: ClickHouse

What happens

ClickhouseEngineAdapter._exchange_tables (sqlmesh/core/engine_adapter/clickhouse.py)
wraps EXCHANGE TABLES in except DatabaseError but only acts on NOT_IMPLEMENTED
(fall back to a non-atomic rename). There is no else: raise, so any other
DatabaseError from the exchange is caught and discarded:

try:
    self.execute(
        f"EXCHANGE TABLES {old_table_sql} AND {new_table_sql}{self._on_cluster_sql()}"
    )
except DatabaseError as e:
    if "NOT_IMPLEMENTED" in str(e):
        throwaway_table_name = self._get_temp_table(old_table_name)
        self._rename_table(old_table_name, throwaway_table_name)
        self._rename_table(new_table_name, old_table_name)
        self.drop_table(throwaway_table_name)
    # <-- no else: the exception is swallowed

The caller _insert_overwrite_by_condition then drops the freshly-computed temp table in
its finally, so the stale target table stays live and the evaluation returns
success.

Why it matters

ClickHouse sets SUPPORTS_REPLACE_TABLE = False, so a steady-state FULL model's
replace_query routes through _insert_overwrite_by_condition → _exchange_tables. A
transient/permission/keeper error during the swap therefore leaves the old data in place
while SQLMesh records the interval as built. Audits do not catch it: the scheduler
evaluates, then audits against the same (now stale) table, then records the interval — so
an audit validates the stale data too. The failure is completely silent.

Expected

A swap failure that is not NOT_IMPLEMENTED should propagate, failing the evaluation
(and, in a DAG, skipping dependents) rather than reporting success over stale data.

Fix

Re-raise anything that is not NOT_IMPLEMENTED:

     except DatabaseError as e:
-        if "NOT_IMPLEMENTED" in str(e):
-            # ... non-atomic rename fallback ...
-            throwaway_table_name = self._get_temp_table(old_table_name)
-            self._rename_table(old_table_name, throwaway_table_name)
-            self._rename_table(new_table_name, old_table_name)
-            self.drop_table(throwaway_table_name)
+        if "NOT_IMPLEMENTED" not in str(e):
+            raise
+        # ... non-atomic rename fallback (old ClickHouse / non-Atomic engine) ...
+        throwaway_table_name = self._get_temp_table(old_table_name)
+        self._rename_table(old_table_name, throwaway_table_name)
+        self._rename_table(new_table_name, old_table_name)
+        self.drop_table(throwaway_table_name)

Happy to open a PR with a unit test covering the three branches (success, NOT_IMPLEMENTED
rename, re-raise). We currently work around it with an engine-adapter subclass that re-raises.

[davidstravito]

For anyone hitting this before a fix lands, this is the workaround we run in production-shadow: a subclass that mirrors _exchange_tables with the condition inverted, plus a connection config that builds it.

from sqlglot import exp
from sqlmesh.core.config.connection import ClickhouseConnectionConfig
from sqlmesh.core.engine_adapter import EngineAdapter
from sqlmesh.core.engine_adapter.clickhouse import ClickhouseEngineAdapter


class StrictExchangeClickhouseEngineAdapter(ClickhouseEngineAdapter):
    def _exchange_tables(self, old_table_name, new_table_name) -> None:
        from clickhouse_connect.driver.exceptions import DatabaseError

        old_sql = exp.to_table(old_table_name).sql(dialect=self.dialect, identify=True)
        new_sql = exp.to_table(new_table_name).sql(dialect=self.dialect, identify=True)
        try:
            self.execute(f"EXCHANGE TABLES {old_sql} AND {new_sql}{self._on_cluster_sql()}")
        except DatabaseError as e:
            if "NOT_IMPLEMENTED" not in str(e):
                raise
            throwaway_table_name = self._get_temp_table(old_table_name)
            self._rename_table(old_table_name, throwaway_table_name)
            self._rename_table(new_table_name, old_table_name)
            self.drop_table(throwaway_table_name)


class StrictExchangeClickhouseConnectionConfig(ClickhouseConnectionConfig):
    # GatewayConfig.connection keeps a ConnectionConfig instance as-is, so this
    # override survives validation when the config is built in Python.
    @property
    def _engine_adapter(self) -> type[EngineAdapter]:
        return StrictExchangeClickhouseEngineAdapter

Used as GatewayConfig(connection=StrictExchangeClickhouseConnectionConfig(host=..., ...)) in config.py. The upstream fix is the same inversion inside ClickhouseEngineAdapter._exchange_tables. The tests we use cover three cases: a successful exchange, a NOT_IMPLEMENTED error that falls back to rename, and any other DatabaseError re-raising (mocking execute to raise clickhouse_connect.driver.exceptions.DatabaseError).