feat(clickhouse): support DataLake catalog source mappings

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
+
+clickhouse-datalake-down:
+	$(CLICKHOUSE_DATALAKE_COMPOSE) down -v --remove-orphans
+
 ##################
 # Docker Engines #
 ##################

docs/integrations/engines/clickhouse.md

@@ -470,6 +470,43 @@ 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.
+
+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 +535,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

@@ -20,6 +20,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 +72,97 @@ 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.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]:
+        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
+            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 table in query.find_all(exp.Table):
+            if not table.catalog or (preserve_unmapped and table.catalog not in mapped_catalogs):
+                continue
+            self._map_catalog_reference(table)
+        for column in query.find_all(exp.Column):
+            if not column.catalog or (preserve_unmapped and column.catalog not in mapped_catalogs):
+                continue
+            table = self._map_catalog_reference(
+                exp.Table(
+                    this=column.args["table"].copy(),
+                    db=column.args["db"].copy(),
+                    catalog=column.args["catalog"].copy(),
+                )
+            )
+            for column_key, table_key in (("table", "this"), ("db", "db"), ("catalog", "catalog")):
+                value = table.args.get(table_key)
+                column.set(column_key, value.copy() if value is not None else None)
+        return query
+
+    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"]

tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md

@@ -0,0 +1,85 @@
+<!-- SPDX-License-Identifier: CC-BY-4.0 -->
+
+# Run the ClickHouse DataLake integration tests
+
+This fixture runs ClickHouse's `DataLakeCatalog` against a local Iceberg REST catalog and MinIO. Run the commands from the repository root with a recent Docker Compose v2 release that supports `up --wait --wait-timeout`.
+
+## Run the tests
+
+Run the complete fixture and test module:
+
+```bash
+make clickhouse-datalake-test
+```
+
+If your running local Docker daemon is available through a named context, set `DOCKER_CONTEXT` for the command. For example, if the context is named `colima`:
+
+```bash
+DOCKER_CONTEXT=colima make clickhouse-datalake-test
+```
+
+Replace `colima` with the name of your running context. This setting applies only to this command and does not change Docker's global context. Use the same prefix with the `up` and `down` targets when needed.
+
+The target checks the Docker daemon first. It then installs the repository's ClickHouse test environment and starts the dedicated Compose project.
+
+Compose waits up to 120 seconds for these checks:
+
+- MinIO can list `local/warehouse-rest` with `mc ls`.
+- The REST catalog accepts a TCP connection on port `8181`.
+- ClickHouse returns a result for `SELECT 1`.
+
+Before MinIO creates the bucket, its init service retries `mc ready` up to 30 times. The target then seeds the Iceberg table and runs [the integration tests](../../test_integration_clickhouse_datalake.py) with `SQLMESH_RUN_CLICKHOUSE_DATALAKE=1`.
+
+To start and seed the services without running pytest, use:
+
+```bash
+make clickhouse-datalake-up
+```
+
+To run pytest directly after startup, install the ClickHouse test dependencies and set the opt-in flag:
+
+```bash
+make engine-clickhouse-install
+SQLMESH_RUN_CLICKHOUSE_DATALAKE=1 pytest -n 0 -m "docker and clickhouse_datalake" tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py
+```
+
+Running pytest without the flag skips the module. The test target fails at its `docker info` preflight if the daemon is unavailable. The startup target fails when Compose connects to an unavailable daemon. Either failure means the live tests did not run.
+
+## Fixture behavior
+
+The Compose project is named `sqlmesh-clickhouse-datalake`. It publishes ClickHouse HTTP on `127.0.0.1:18123`. The REST catalog and MinIO have no host ports. Test traffic uses the project network and fixture credentials. The bucket has no public policy. The REST catalog, seeder, and ClickHouse use only local fixture credentials. The upstream integration Compose stack makes its bucket public. This fixture leaves its bucket private. The tests do not read the shared integration profile or production credentials.
+
+The Tabulario REST service uses `CATALOG_S3_PATH__STYLE__ACCESS=true` for path-style S3 access.
+
+The [seed script](seed.py) connects to the REST catalog at `http://rest:8181`. PyIceberg adds the `/v1` path. ClickHouse connects to `http://rest:8181/v1`.
+
+The script creates the `analytics.events` Iceberg table at `s3://warehouse-rest/data/sqlmesh/events`. It appends two rows, then one row, and checks that the table contains three rows across two snapshots. On rerun, it drops the table and clears objects under that table location before creating the table again.
+
+The tests verify the ClickHouse server version and read the seeded table. They cover mapped reads with qualified tables, aliases, qualified columns, and CTEs, both with and without an injected virtual catalog.
+
+Other cases cover AST immutability, raw SQL strings, CTAS, append and replace materializations, and rejection of an unknown source before target creation. They also check that local reads strip an injected virtual catalog. A mapped view succeeds without catalog injection. The same source fails when a virtual catalog is injected.
+
+Run `make clickhouse-datalake-down` to stop the services. The target removes their containers, the project network, attached volumes, and orphan containers.
+
+The next `make clickhouse-datalake-test` starts a clean project and reseeds the table.
+
+## Pinned versions and source
+
+The DataLakeCatalog setup and Iceberg seed flow follow ClickHouse's [`v25.8.16.34-lts` integration test](https://github.com/ClickHouse/ClickHouse/blob/v25.8.16.34-lts/tests/integration/test_database_iceberg/test.py) and [REST catalog Compose file](https://github.com/ClickHouse/ClickHouse/blob/v25.8.16.34-lts/tests/integration/compose/docker_compose_iceberg_rest_catalog.yml).
+
+ClickHouse pins PyIceberg `0.8.1` and PyArrow `17.0.0` in its [integration runner requirements](https://github.com/ClickHouse/ClickHouse/blob/v25.8.16.34-lts/ci/docker/integration/runner/requirements.txt). This fixture also pins MinIO's Python client to `7.2.3` for cleanup.
+
+The [fixture Compose file](../compose.clickhouse-datalake.yaml) pins these images by digest:
+
+- `clickhouse/clickhouse-server:25.8.16.34` (`sha256:86e6acceeacb49202fd694b77b4f7fcd6e022b52a29b0568fbfb4b05e8f6b2e7`). This version matches the upstream test tag `v25.8.16.34-lts`.
+- `tabulario/iceberg-rest:1.6.0` (`sha256:3b7d31bdfec626b68e97531c9778a1b9119659e456fe28545a49f6aa6a9ce472`).
+- `cgr.dev/chainguard/minio` (`sha256:039800e64ec7247d2fde7cff3697e964f6fe20b6d7d2c46aa7d82cc63355d512`) and `cgr.dev/chainguard/minio-client:latest-dev` (`sha256:fb635b967f5f32424391150dac4985cc1aea78fb0a1ce031c01952cf661a13ec`).
+- Seeder base image `python:3.12.13-slim-bookworm` (`sha256:4766d8b510c428e595d74b9cc5bbb2fae8e26316fffb4adc89908d79aacd58a2`). Its package pins are in [requirements.txt](requirements.txt).
+
+The MinIO images use Chainguard registry digests because the MinIO image references in the upstream Compose file were unavailable during version verification.
+
+## Live validation status
+
+The pinned fixture passed on a local ARM64 Docker daemon. The default Docker context was unavailable in this environment, so the commands used `DOCKER_CONTEXT=colima` to reach the running Colima daemon. This per-command setting did not change the global Docker context.
+
+The seed script verified three rows across two Iceberg snapshots. The opt-in integration module passed all 19 cases with no skips, including after a clean `down` and `up` repeat. An unauthenticated request to the MinIO bucket endpoint returned HTTP `403 AccessDenied`, confirming that the bucket stayed private.

tests/core/engine_adapter/integration/docker/clickhouse-datalake/requirements.txt

@@ -0,0 +1,5 @@
+# SPDX-License-Identifier: Apache-2.0
+
+pyiceberg==0.8.1
+pyarrow==17.0.0
+minio==7.2.3