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,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,88 @@
+<!-- SPDX-License-Identifier: CC-BY-4.0 -->
+
+# Run the ClickHouse DataLake integration tests
+
+This fixture tests ClickHouse `DataLakeCatalog` with a local Iceberg REST catalog and MinIO. Run it 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
+```
+
+The PR CI Docker matrix runs this target as `clickhouse-datalake`. It uploads the JUnit
+report at `test-results/junit-clickhouse-datalake.xml` and stops the services after the run.
+
+If Docker runs under a named context, set `DOCKER_CONTEXT` for each Make command. For example, with a context named `colima`:
+
+```bash
+DOCKER_CONTEXT=colima make clickhouse-datalake-test
+```
+
+Replace `colima` with the name of your context. This setting applies only to the command and leaves Docker's global context unchanged. Use it with the `up` and `down` targets too.
+
+The test target checks the Docker daemon, installs the ClickHouse test dependencies, 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 seeds the Iceberg table, then 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
+```
+
+Pytest skips the module if you omit the flag. The test target checks Docker with `docker info`. The startup target fails if Compose cannot reach the daemon.
+
+## Fixture behavior
+
+The Compose project is named `sqlmesh-clickhouse-datalake`. It publishes ClickHouse HTTP only on `127.0.0.1:18123`. The REST catalog and MinIO have no host ports. The fixture bucket has no public policy. The services use local fixture credentials, and the tests do not read shared integration profiles 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 `/v1`. ClickHouse connects to `http://rest:8181/v1`.
+
+The script creates `analytics.events` at `s3://warehouse-rest/data/sqlmesh/events`. It appends two rows, then one row, and verifies three rows across two snapshots. On rerun, it drops the table, deletes objects under that location, and creates the table again.
+
+The tests check the ClickHouse version and read the seeded table. They also cover:
+
+- Mapped reads with qualified tables, aliases, qualified columns, and CTEs, with and without virtual catalog injection.
+- Preserving the original query AST.
+- CTAS, append, and replace materializations.
+- Rejection of an unknown source before CTAS creates its target.
+- Stripping the injected virtual catalog from local reads.
+
+View creation accepts a mapped source without virtual catalog injection. With injection, it still raises the single-catalog error.
+
+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 upstream Compose MinIO image references were unavailable during version verification, so this fixture uses pinned Chainguard images.

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