feat(clickhouse): support DataLake catalog source mappings

.github/workflows/pr.yaml

@@ -267,6 +267,7 @@ jobs:
             trino,
             spark,
             clickhouse,
+            clickhouse-datalake,
             risingwave,
             starrocks,
           ]
@@ -293,6 +294,9 @@ jobs:
         run: |
           source .venv/bin/activate
           make ${{ matrix.engine }}-test
+      - name: Stop DataLake services
+        if: ${{ always() && matrix.engine == 'clickhouse-datalake' }}
+        run: make clickhouse-datalake-down
       - name: Upload test results
         uses: actions/upload-artifact@v7
         if: ${{ !cancelled() }}

Makefile

@@ -1,4 +1,5 @@
 .PHONY: docs
+.PHONY: clickhouse-datalake-up clickhouse-datalake-test clickhouse-datalake-down
 
 ifdef UV
     PIP := uv pip
@@ -192,6 +193,21 @@ engine-%-up: engine-%-install engine-docker-%-up
 engine-%-down:
 	docker compose -f ./tests/core/engine_adapter/integration/docker/compose.${*}.yaml down -v
 
+CLICKHOUSE_DATALAKE_COMPOSE = docker compose --project-name sqlmesh-clickhouse-datalake -f ./tests/core/engine_adapter/integration/docker/compose.clickhouse-datalake.yaml
+
+clickhouse-datalake-up:
+	$(CLICKHOUSE_DATALAKE_COMPOSE) up -d --wait --wait-timeout 120 clickhouse minio-init rest
+	$(CLICKHOUSE_DATALAKE_COMPOSE) run --build --rm seed
+
+clickhouse-datalake-test:
+	docker info >/dev/null
+	$(MAKE) engine-clickhouse-install
+	$(MAKE) clickhouse-datalake-up
+	SQLMESH_RUN_CLICKHOUSE_DATALAKE=1 pytest -n 0 -m "docker and clickhouse_datalake" tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py --junitxml=test-results/junit-clickhouse-datalake.xml
+
+clickhouse-datalake-down:
+	$(CLICKHOUSE_DATALAKE_COMPOSE) down -v --remove-orphans
+
 ##################
 # Docker Engines #
 ##################

docs/integrations/engines/clickhouse.md

@@ -470,6 +470,57 @@ gateways:
 
 With this configuration, ClickHouse models will appear as `ch_virtual.mydb.mytable` in plan output instead of `__clickhouse__.mydb.mytable`.
 
+### Reading catalog-backed physical tables
+
+When ClickHouse exposes a source catalog through a [`DataLakeCatalog`](https://clickhouse.com/docs/engines/database-engines/datalake), map that source catalog to its ClickHouse database in the connection configuration:
+
+```yaml
+gateways:
+  clickhouse:
+    connection:
+      type: clickhouse
+      host: my-clickhouse-host
+      username: default
+      datalake_catalog_mapping:
+        lake: iceberg
+```
+
+SQLMesh renders `lake.analytics.orders` as `iceberg."analytics.orders"`. It combines the source
+schema and table into one quoted table name because ClickHouse uses a two-level namespace. The
+`iceberg` database must already exist and use the `DataLakeCatalog` engine. SQLMesh does not create
+the database or manage its metastore or storage permissions.
+
+For parsed queries that read data without writing the result to a ClickHouse table, SQLMesh rewrites
+mapped source catalogs and passes other foreign catalog-qualified references through unchanged.
+During rendering, SQLMesh strips the configured virtual catalog prefix, even before injection. For
+`INSERT ... SELECT` and other query materializations, SQLMesh rewrites mapped sources and validates
+the remaining catalog-qualified sources. With a non-empty mapping, each source catalog must appear
+in the mapping. When SQLMesh has injected a virtual catalog, a source can also use that catalog.
+SQLMesh raises an error for other catalogs. Without a non-empty mapping or an injected virtual
+catalog, SQLMesh does not apply DataLake mapping or catalog validation during materialization.
+
+This mapping rewrites parsed query sources. It does not rewrite table names used for metadata
+operations such as `columns()` or `table_exists()`. The `sqlmesh create_external_models` command also
+uses the source name as supplied when it fetches column types, so it cannot infer columns for a
+source named `lake.analytics.orders` through the mapping. In a ClickHouse-only project, registering
+that three-part logical name as an external model can also fail SQLMesh's schema nesting check,
+because ClickHouse project models use two-part names.
+
+To register column metadata until mapped names are supported there, use ClickHouse's physical
+two-part name in both the external model and the query. For example, use
+`iceberg."analytics.orders"` for the table exposed by database `iceberg`. This matches ClickHouse's
+`database."schema.table"` name and keeps the model at ClickHouse's two-part nesting level. The
+mapping still lets SQLMesh render queries that use mapped logical names. Without registered
+column metadata, SQLMesh cannot fully optimize those queries or produce column-level lineage.
+
+An injected virtual catalog takes precedence over a mapping key with the same name. SQLMesh treats
+that name as its virtual catalog. Use different names for the virtual catalog and DataLakeCatalog
+source catalogs. The mapping applies to query source references. It does not rewrite SQL strings
+passed directly to the ClickHouse adapter or map `INSERT` targets.
+
+View creation rejects a source that uses another catalog when a virtual catalog is injected, even
+if the mapping includes that source. Mapped-source views work when no virtual catalog is injected.
+
 ## Local/Built-in Scheduler
 
 **Engine Adapter Type**: `clickhouse`
@@ -498,3 +549,4 @@ With this configuration, ClickHouse models will appear as `ch_virtual.mydb.mytab
 | `connection_settings`     | Additional [connection settings](https://clickhouse.com/docs/integrations/python#settings-argument)                                                                                                                                                                             |  dict  |    N     |
 | `connection_pool_options` | Additional [options](https://clickhouse.com/docs/integrations/python#customizing-the-http-connection-pool)                                                                                                                                         for the HTTP connection pool |  dict  |    N     |
 | `virtual_catalog`         | Override the virtual catalog name used when ClickHouse runs alongside a catalog-aware gateway (e.g. Trino). Defaults to `__{gateway_name}__`. See [Multi-gateway setup](#multi-gateway-setup) for details.                                                                      | string |    N     |
+| `datalake_catalog_mapping` | Map source catalogs to ClickHouse DataLakeCatalog databases. See [Reading catalog-backed physical tables](#reading-catalog-backed-physical-tables).                                                                                                                  | dict  |    N     |

pyproject.toml

@@ -273,6 +273,7 @@ markers = [
     "athena: test for Athena",
     "bigquery: test for BigQuery",
     "clickhouse: test for Clickhouse (standalone mode / cluster mode)",
+    "clickhouse_datalake: test for ClickHouse DataLakeCatalog mapping",
     "clickhouse_cloud: test for Clickhouse (cloud mode)",
     "databricks: test for Databricks",
     "db2: test for Db2",

sqlmesh/core/config/connection.py

@@ -2244,6 +2244,7 @@ class ClickhouseConnectionConfig(ConnectionConfig):
     port: t.Optional[int] = None
     cluster: t.Optional[str] = None
     virtual_catalog: t.Optional[str] = None
+    datalake_catalog_mapping: t.Dict[str, str] = Field(default_factory=dict)
     connect_timeout: int = 10
     send_receive_timeout: int = 300
     query_limit: int = 0
@@ -2291,6 +2292,21 @@ def validate_virtual_catalog(cls, v: t.Optional[str]) -> t.Optional[str]:
             )
         return v
 
+    @field_validator("datalake_catalog_mapping")
+    def validate_datalake_catalog_mapping(cls, v: t.Dict[str, str]) -> t.Dict[str, str]:
+        for source_catalog, target_database in v.items():
+            if not source_catalog.strip() or "." in source_catalog:
+                raise ConfigError(
+                    "datalake_catalog_mapping keys must be non-empty single identifiers "
+                    f"(got: {source_catalog!r})"
+                )
+            if not target_database.strip() or "." in target_database:
+                raise ConfigError(
+                    "datalake_catalog_mapping values must be non-empty ClickHouse database "
+                    f"identifiers (got: {target_database!r})"
+                )
+        return v
+
     @property
     def _connection_kwargs_keys(self) -> t.Set[str]:
         kwargs = {
@@ -2357,6 +2373,7 @@ def _extra_engine_config(self) -> t.Dict[str, t.Any]:
             "cluster": self.cluster,
             "cloud_mode": self.cloud_mode,
             "virtual_catalog": self.virtual_catalog,
+            "datalake_catalog_mapping": self.datalake_catalog_mapping,
         }
 
     @property

sqlmesh/core/engine_adapter/clickhouse.py

@@ -5,6 +5,7 @@
 import re
 from functools import cached_property
 from sqlglot import exp, maybe_parse
+from sqlglot.optimizer.scope import Scope, find_all_in_scope, traverse_scope
 from sqlmesh.core.dialect import to_schema
 from sqlmesh.core.engine_adapter.mixins import LogicalMergeMixin
 from sqlmesh.core.engine_adapter.base import EngineAdapterWithIndexSupport
@@ -20,6 +21,7 @@
 )
 from sqlmesh.core.schema_diff import TableAlterOperation
 from sqlmesh.utils import get_source_columns_to_types
+from sqlmesh.utils.errors import SQLMeshError
 
 if t.TYPE_CHECKING:
     import pandas as pd
@@ -71,17 +73,164 @@ def inject_virtual_catalog(self, gateway: str) -> None:
 
     def _to_sql(self, expression: exp.Expr, quote: bool = True, **kwargs: t.Any) -> str:
         """Render SQL without the virtual catalog, which ClickHouse does not support."""
-        if self._default_catalog and any(self._virtual_catalog_references(expression)):
+        virtual_catalog = self._default_catalog or self._extra_config.get("virtual_catalog")
+        if virtual_catalog and any(self._virtual_catalog_references(expression)):
             expression = expression.copy()
             for reference in list(self._virtual_catalog_references(expression)):
                 reference.set("catalog", None)
+        if isinstance(expression, exp.Query):
+            expression = self._map_catalog_references(expression, preserve_unmapped=True)
+        elif isinstance(expression, exp.Delete):
+            where = expression.args.get("where")
+            if where and any(where.find_all(exp.Query)):
+                expression = expression.copy()
+                where = expression.args["where"]
+                expression.set(
+                    "where",
+                    where.transform(
+                        lambda node: self._map_catalog_references(node)
+                        if isinstance(node, exp.Query)
+                        else node,
+                        copy=False,
+                    ),
+                )
+        elif isinstance(expression, exp.Insert) and isinstance(expression.expression, exp.Query):
+            expression = expression.copy()
+            expression.set("expression", self._map_catalog_references(expression.expression))
         return super()._to_sql(expression, quote=quote, **kwargs)
 
-    def _virtual_catalog_references(self, expression: exp.Expr) -> t.Iterator[exp.Expr]:
+    def _virtual_catalog_references(
+        self, expression: exp.Expr
+    ) -> t.Iterator[t.Union[exp.Table, exp.Column]]:
+        virtual_catalog = self._default_catalog or self._extra_config.get("virtual_catalog")
         return (
             reference
-            for reference in expression.find_all(exp.Table, exp.Column)
-            if reference.text("catalog") == self._default_catalog
+            for reference in self._catalog_references(expression)
+            if virtual_catalog and reference.text("catalog") == virtual_catalog
+        )
+
+    def _get_source_queries(
+        self,
+        query_or_df: QueryOrDF,
+        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]],
+        target_table: TableName,
+        *,
+        batch_size: t.Optional[int] = None,
+        source_columns: t.Optional[t.List[str]] = None,
+    ) -> t.List[SourceQuery]:
+        if isinstance(query_or_df, exp.Query):
+            query_or_df = self._map_catalog_references(query_or_df)
+        return super()._get_source_queries(
+            query_or_df,
+            target_columns_to_types,
+            target_table,
+            batch_size=batch_size,
+            source_columns=source_columns,
+        )
+
+    def _map_catalog_references(self, query: Query, *, preserve_unmapped: bool = False) -> Query:
+        mapping = self._extra_config.get("datalake_catalog_mapping") or {}
+        if not self._default_catalog and not mapping:
+            return query
+
+        mapped_catalogs = {self._default_catalog, *mapping}
+        query = query.copy()
+        for reference in self._catalog_references(query):
+            if not reference.catalog or (
+                preserve_unmapped and reference.catalog not in mapped_catalogs
+            ):
+                continue
+            if isinstance(reference, exp.Table):
+                self._map_catalog_reference(reference)
+                continue
+
+            table = self._map_catalog_reference(
+                exp.Table(
+                    this=reference.args["table"].copy(),
+                    db=reference.args["db"].copy(),
+                    catalog=reference.args["catalog"].copy(),
+                )
+            )
+            for column_key, table_key in (("table", "this"), ("db", "db"), ("catalog", "catalog")):
+                value = table.args.get(table_key)
+                reference.set(column_key, value.copy() if value is not None else None)
+        return query
+
+    def _catalog_references(self, expression: exp.Expr) -> t.List[t.Union[exp.Table, exp.Column]]:
+        """Use query scopes to distinguish qualified columns from nested-field paths."""
+        tables = list(expression.find_all(exp.Table))
+        references: t.List[t.Union[exp.Table, exp.Column]] = list(tables)
+        root_tables = list(find_all_in_scope(expression, exp.Table))
+        scoped_columns = set()
+        virtual_catalog = self._default_catalog or self._extra_config.get("virtual_catalog")
+
+        def matches_source_table(column: exp.Column, table: exp.Table) -> bool:
+            if (column.catalog, column.db, column.table) == (table.catalog, table.db, table.name):
+                return True
+            # ClickHouse strips the virtual catalog from DML targets before SQL rendering.
+            return (
+                column.catalog == virtual_catalog
+                and not table.catalog
+                and (column.db, column.table) == (table.db, table.name)
+            )
+
+        def add_column_reference(column: exp.Column, scope: t.Optional[Scope]) -> None:
+            if not column.catalog:
+                return
+            current_scope = scope
+            while current_scope is not None:
+                # The nearest alias or CTE name takes precedence as a nested-field path.
+                if column.catalog in current_scope.sources:
+                    return
+                if any(
+                    isinstance(source, exp.Table) and matches_source_table(column, source)
+                    for source in current_scope.sources.values()
+                ):
+                    references.append(column)
+                    return
+                current_scope = current_scope.parent
+
+            if column.catalog in {table.alias_or_name for table in root_tables}:
+                return
+            if any(matches_source_table(column, table) for table in root_tables):
+                references.append(column)
+
+        for scope in traverse_scope(expression):
+            for column in scope.find_all(exp.Column):
+                scoped_columns.add(id(column))
+                add_column_reference(column, scope)
+
+        for column in find_all_in_scope(expression, exp.Column):
+            if id(column) not in scoped_columns:
+                add_column_reference(column, None)
+        return references
+
+    def _map_catalog_reference(self, table: exp.Table) -> exp.Table:
+        """Map catalog.schema.table to a ClickHouse DataLakeCatalog database.table reference."""
+        catalog = table.catalog
+        if not catalog:
+            return table
+        if catalog == self._default_catalog:
+            table.set("catalog", None)
+            return table
+
+        target_database = (self._extra_config.get("datalake_catalog_mapping") or {}).get(catalog)
+        if target_database:
+            if not table.db:
+                raise SQLMeshError(
+                    f"ClickHouse DataLakeCatalog mapping for '{catalog}' requires a "
+                    "three-level catalog.schema.table reference"
+                )
+            source_table = f"{table.db}.{table.name}"
+            table.set("catalog", None)
+            table.set("db", exp.to_identifier(target_database))
+            table.set("this", exp.to_identifier(source_table, quoted=True))
+            return table
+
+        raise SQLMeshError(
+            f"{self.dialect} requires that all catalog operations be against the virtual "
+            f"catalog {self._default_catalog!r} or a configured DataLakeCatalog mapping. "
+            f"Provided catalog: {catalog}"
         )
 
     @property

tests/core/engine_adapter/integration/docker/clickhouse-datalake/Dockerfile

@@ -0,0 +1,12 @@
+# SPDX-License-Identifier: Apache-2.0
+
+FROM python:3.12.13-slim-bookworm@sha256:4766d8b510c428e595d74b9cc5bbb2fae8e26316fffb4adc89908d79aacd58a2
+
+WORKDIR /fixture
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY seed.py .
+
+CMD ["python", "seed.py"]