From e813efe166eb2623eb05b880b7cbb9b2484e59fd Mon Sep 17 00:00:00 2001 From: Michael Day Date: Sat, 17 Oct 2026 20:50:12 -0400 Subject: [PATCH 1/3] feat(clickhouse): support DataLake catalog source mappings Signed-off-by: Michael Day --- Makefile | 16 + docs/integrations/engines/clickhouse.md | 38 +++ pyproject.toml | 1 + sqlmesh/core/config/connection.py | 17 + sqlmesh/core/engine_adapter/clickhouse.py | 85 ++++- .../docker/clickhouse-datalake/Dockerfile | 12 + .../docker/clickhouse-datalake/README.md | 85 +++++ .../clickhouse-datalake/requirements.txt | 5 + .../docker/clickhouse-datalake/seed.py | 122 +++++++ .../docker/compose.clickhouse-datalake.yaml | 105 ++++++ .../test_integration_clickhouse_datalake.py | 297 ++++++++++++++++ tests/core/engine_adapter/test_clickhouse.py | 321 ++++++++++++++++++ 12 files changed, 1102 insertions(+), 2 deletions(-) create mode 100644 tests/core/engine_adapter/integration/docker/clickhouse-datalake/Dockerfile create mode 100644 tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md create mode 100644 tests/core/engine_adapter/integration/docker/clickhouse-datalake/requirements.txt create mode 100644 tests/core/engine_adapter/integration/docker/clickhouse-datalake/seed.py create mode 100644 tests/core/engine_adapter/integration/docker/compose.clickhouse-datalake.yaml create mode 100644 tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py diff --git a/Makefile b/Makefile index 0d8571b33f..75c1b491ff 100644 --- a/Makefile +++ b/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 # ################## diff --git a/docs/integrations/engines/clickhouse.md b/docs/integrations/engines/clickhouse.md index 995d48e5f1..938064484c 100644 --- a/docs/integrations/engines/clickhouse.md +++ b/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 | diff --git a/pyproject.toml b/pyproject.toml index 378d3ffe43..f3f1929bd9 100644 --- a/pyproject.toml +++ b/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", diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py index 29cb1ef467..7a65e87d72 100644 --- a/sqlmesh/core/config/connection.py +++ b/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 diff --git a/sqlmesh/core/engine_adapter/clickhouse.py b/sqlmesh/core/engine_adapter/clickhouse.py index e385452bf3..91b2b03e87 100644 --- a/sqlmesh/core/engine_adapter/clickhouse.py +++ b/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 diff --git a/tests/core/engine_adapter/integration/docker/clickhouse-datalake/Dockerfile b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/Dockerfile new file mode 100644 index 0000000000..5b98992d0f --- /dev/null +++ b/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"] diff --git a/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md new file mode 100644 index 0000000000..d221ba5f7c --- /dev/null +++ b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md @@ -0,0 +1,85 @@ + + +# 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. diff --git a/tests/core/engine_adapter/integration/docker/clickhouse-datalake/requirements.txt b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/requirements.txt new file mode 100644 index 0000000000..3470a1fb13 --- /dev/null +++ b/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 diff --git a/tests/core/engine_adapter/integration/docker/clickhouse-datalake/seed.py b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/seed.py new file mode 100644 index 0000000000..874de046f4 --- /dev/null +++ b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/seed.py @@ -0,0 +1,122 @@ +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import os +import time +from typing import Any + +import pyarrow as pa # type: ignore[import-untyped] +from minio import Minio # type: ignore[import-not-found] +from pyiceberg.catalog import load_catalog # type: ignore[import-not-found] +from pyiceberg.schema import Schema # type: ignore[import-not-found] +from pyiceberg.types import LongType, NestedField, StringType # type: ignore[import-not-found] + + +CATALOG_NAME = "demo" +NAMESPACE = "analytics" +TABLE_NAME = "events" +TABLE_IDENTIFIER = f"{NAMESPACE}.{TABLE_NAME}" +BUCKET = "warehouse-rest" +OBJECT_PREFIX = "data/sqlmesh/events/" +TABLE_LOCATION = "s3://warehouse-rest/data/sqlmesh/events" +ROWS = [ + {"id": 1, "value": "alpha"}, + {"id": 2, "value": "beta"}, + {"id": 3, "value": "gamma"}, +] + +SCHEMA = Schema( + NestedField(field_id=1, name="id", field_type=LongType(), required=True), + NestedField(field_id=2, name="value", field_type=StringType(), required=True), +) +ARROW_SCHEMA = pa.schema( + [ + pa.field("id", pa.int64(), nullable=False), + pa.field("value", pa.string(), nullable=False), + ] +) + + +def _load_catalog() -> Any: + last_error: Exception | None = None + for attempt in range(30): + try: + catalog = load_catalog( + CATALOG_NAME, + **{ + "uri": os.environ["ICEBERG_REST_URL"], + "type": "rest", + "s3.endpoint": os.environ["S3_ENDPOINT"], + "s3.access-key-id": os.environ["S3_ACCESS_KEY_ID"], + "s3.secret-access-key": os.environ["S3_SECRET_ACCESS_KEY"], + "s3.region": "us-east-1", + # PyArrow's pinned S3FileSystem defaults to path-style addressing. + }, + ) + catalog.list_namespaces() + return catalog + except Exception as ex: + last_error = ex + if attempt == 29: + break + time.sleep(1) + + raise RuntimeError( + "Iceberg REST catalog did not become ready after 30 attempts" + ) from last_error + + +def _clear_table_location() -> None: + client = Minio( + "minio:9000", + access_key=os.environ["S3_ACCESS_KEY_ID"], + secret_key=os.environ["S3_SECRET_ACCESS_KEY"], + secure=False, + ) + for attempt in range(30): + try: + if client.bucket_exists(BUCKET): + break + except Exception: + if attempt == 29: + raise + time.sleep(1) + else: + raise RuntimeError(f"MinIO bucket {BUCKET!r} was not created") + + for item in client.list_objects(BUCKET, prefix=OBJECT_PREFIX, recursive=True): + if item.object_name: + client.remove_object(BUCKET, item.object_name) + + +def main() -> None: + catalog = _load_catalog() + namespaces = catalog.list_namespaces() + if (NAMESPACE,) not in namespaces: + catalog.create_namespace(NAMESPACE) + + if (NAMESPACE, TABLE_NAME) in catalog.list_tables(NAMESPACE): + catalog.drop_table(TABLE_IDENTIFIER) + _clear_table_location() + + table = catalog.create_table( + identifier=TABLE_IDENTIFIER, + schema=SCHEMA, + location=TABLE_LOCATION, + ) + table.append(pa.Table.from_pylist(ROWS[:2], schema=ARROW_SCHEMA)) + table.append(pa.Table.from_pylist(ROWS[2:], schema=ARROW_SCHEMA)) + + actual_rows = sorted(table.scan().to_arrow().to_pylist(), key=lambda row: row["id"]) + snapshot_count = len(table.metadata.snapshots) + if actual_rows != ROWS or snapshot_count != 2: + raise RuntimeError( + f"Expected 3 rows in 2 snapshots; got rows={actual_rows!r}, snapshots={snapshot_count}" + ) + + print(f"Seeded {TABLE_IDENTIFIER} at {TABLE_LOCATION}: 3 rows, 2 snapshots") + + +if __name__ == "__main__": + main() diff --git a/tests/core/engine_adapter/integration/docker/compose.clickhouse-datalake.yaml b/tests/core/engine_adapter/integration/docker/compose.clickhouse-datalake.yaml new file mode 100644 index 0000000000..afeca3e86b --- /dev/null +++ b/tests/core/engine_adapter/integration/docker/compose.clickhouse-datalake.yaml @@ -0,0 +1,105 @@ +# SPDX-License-Identifier: Apache-2.0 + +services: + minio: + image: cgr.dev/chainguard/minio@sha256:039800e64ec7247d2fde7cff3697e964f6fe20b6d7d2c46aa7d82cc63355d512 + command: ["server", "/data", "--console-address", ":9001"] + environment: + MINIO_ROOT_USER: minio + MINIO_ROOT_PASSWORD: localpwd + + minio-init: + image: cgr.dev/chainguard/minio-client:latest-dev@sha256:fb635b967f5f32424391150dac4985cc1aea78fb0a1ce031c01952cf661a13ec + depends_on: + minio: + condition: service_started + environment: + MINIO_ROOT_USER: minio + MINIO_ROOT_PASSWORD: localpwd + entrypoint: ["/bin/sh", "-c"] + command: + - | + set -eu + attempt=0 + until mc alias set local http://minio:9000 "$${MINIO_ROOT_USER}" "$${MINIO_ROOT_PASSWORD}" >/dev/null 2>&1 && mc ready local >/dev/null 2>&1; do + attempt=$$((attempt + 1)) + if [ "$$attempt" -ge 30 ]; then + echo 'MinIO did not become ready after 30 attempts' >&2 + exit 1 + fi + sleep 1 + done + mc mb --ignore-existing local/warehouse-rest + exec sleep infinity + healthcheck: + test: ["CMD-SHELL", "mc ls local/warehouse-rest >/dev/null"] + interval: 2s + timeout: 2s + retries: 30 + start_period: 1s + + rest: + image: tabulario/iceberg-rest:1.6.0@sha256:3b7d31bdfec626b68e97531c9778a1b9119659e456fe28545a49f6aa6a9ce472 + depends_on: + minio-init: + condition: service_healthy + environment: + AWS_ACCESS_KEY_ID: minio + AWS_SECRET_ACCESS_KEY: localpwd + AWS_REGION: us-east-1 + CATALOG_WAREHOUSE: s3://warehouse-rest/ + CATALOG_IO__IMPL: org.apache.iceberg.aws.s3.S3FileIO + CATALOG_S3_ENDPOINT: http://minio:9000 + CATALOG_S3_PATH__STYLE__ACCESS: "true" + healthcheck: + test: ["CMD", "bash", "-c", "echo > /dev/tcp/localhost/8181"] + interval: 2s + timeout: 2s + retries: 30 + start_period: 15s + + clickhouse: + image: clickhouse/clickhouse-server:25.8.16.34@sha256:86e6acceeacb49202fd694b77b4f7fcd6e022b52a29b0568fbfb4b05e8f6b2e7 + depends_on: + rest: + condition: service_healthy + minio-init: + condition: service_healthy + environment: + CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: "1" + CLICKHOUSE_USER: clickhouse + CLICKHOUSE_PASSWORD: clickhouse + CLICKHOUSE_DB: default + ports: + - "127.0.0.1:18123:8123" + healthcheck: + test: + [ + "CMD", + "clickhouse-client", + "--user", + "clickhouse", + "--password", + "clickhouse", + "--query", + "SELECT 1", + ] + interval: 2s + timeout: 2s + retries: 30 + start_period: 15s + + seed: + build: + context: ./clickhouse-datalake + dockerfile: Dockerfile + depends_on: + rest: + condition: service_healthy + minio-init: + condition: service_healthy + environment: + ICEBERG_REST_URL: http://rest:8181 + S3_ENDPOINT: http://minio:9000 + S3_ACCESS_KEY_ID: minio + S3_SECRET_ACCESS_KEY: localpwd diff --git a/tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py b/tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py new file mode 100644 index 0000000000..e2e23dcd5a --- /dev/null +++ b/tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py @@ -0,0 +1,297 @@ +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import os +import typing as t + +import pytest +from sqlglot import exp, parse_one + +from sqlmesh.core.config.connection import ClickhouseConnectionConfig +from sqlmesh.core.engine_adapter.clickhouse import ClickhouseEngineAdapter +from sqlmesh.utils.errors import SQLMeshError + + +pytestmark = [pytest.mark.docker, pytest.mark.clickhouse_datalake] + +_RUN_FLAG = "SQLMESH_RUN_CLICKHOUSE_DATALAKE" +_CLICKHOUSE_VERSION = "25.8.16.34" +_REST_PASSWORD = "localpwd" +_EXPECTED_ROWS = [(1, "alpha"), (2, "beta"), (3, "gamma")] +_COLUMN_TYPES = { + "id": exp.DataType.build("Int64", dialect="clickhouse"), + "value": exp.DataType.build("String", dialect="clickhouse"), +} +_SOURCE_QUERY = "SELECT id, value FROM lake.analytics.events ORDER BY id" + + +def _new_adapter() -> ClickhouseEngineAdapter: + connection_config = ClickhouseConnectionConfig( + host="127.0.0.1", + port=18123, + username="clickhouse", + password="clickhouse", + connection_settings={"allow_database_iceberg": True}, + datalake_catalog_mapping={"lake": "iceberg"}, + concurrent_tasks=1, + register_comments=False, + ) + adapter = connection_config.create_engine_adapter() + assert isinstance(adapter, ClickhouseEngineAdapter) + return adapter + + +@pytest.fixture(scope="module") +def engine_adapter() -> t.Iterator[ClickhouseEngineAdapter]: + if os.environ.get(_RUN_FLAG) != "1": + pytest.skip(f"Set {_RUN_FLAG}=1 after starting the dedicated DataLake Compose stack") + + adapter = _new_adapter() + try: + version = adapter.fetchone("SELECT version()") + assert version is not None and version[0] == _CLICKHOUSE_VERSION + + # Keep the official ClickHouse test's setting and DDL sequence. The mapped database is + # local to this Compose project and points at its private MinIO bucket. + adapter.execute("SET allow_database_iceberg=true") + adapter.execute( + "CREATE DATABASE IF NOT EXISTS iceberg " + "ENGINE = DataLakeCatalog('http://rest:8181/v1', 'minio', " + f"'{_REST_PASSWORD}') " + "SETTINGS catalog_type = 'rest', warehouse = 'demo', " + "storage_endpoint = 'http://minio:9000/warehouse-rest'" + ) + physical_query = parse_one( + 'SELECT id, value FROM iceberg."analytics.events" ORDER BY id', + dialect="clickhouse", + ) + assert adapter.fetchall(physical_query) == _EXPECTED_ROWS + + yield adapter + finally: + adapter.close() + + +@pytest.fixture(scope="module") +def injected_engine_adapter( + engine_adapter: ClickhouseEngineAdapter, +) -> t.Iterator[ClickhouseEngineAdapter]: + adapter = _new_adapter() + adapter.inject_virtual_catalog("clickhouse_datalake") + try: + yield adapter + finally: + adapter.close() + + +@pytest.mark.parametrize( + ("adapter_fixture", "query_sql"), + [ + pytest.param("engine_adapter", _SOURCE_QUERY, id="uninjected-qualified-table"), + pytest.param( + "injected_engine_adapter", + _SOURCE_QUERY, + id="injected-qualified-table", + ), + pytest.param( + "engine_adapter", + "SELECT source.id, source.value " + "FROM lake.analytics.events AS source ORDER BY source.id", + id="uninjected-table-alias", + ), + pytest.param( + "injected_engine_adapter", + "SELECT source.id, source.value " + "FROM lake.analytics.events AS source ORDER BY source.id", + id="injected-table-alias", + ), + pytest.param( + "engine_adapter", + "SELECT lake.analytics.events.id, lake.analytics.events.value " + "FROM lake.analytics.events ORDER BY lake.analytics.events.id", + id="uninjected-qualified-columns", + ), + pytest.param( + "injected_engine_adapter", + "SELECT lake.analytics.events.id, lake.analytics.events.value " + "FROM lake.analytics.events ORDER BY lake.analytics.events.id", + id="injected-qualified-columns", + ), + pytest.param( + "engine_adapter", + "WITH source AS (SELECT id, value FROM lake.analytics.events) " + "SELECT id, value FROM source ORDER BY id", + id="uninjected-cte", + ), + pytest.param( + "injected_engine_adapter", + "WITH source AS (SELECT id, value FROM lake.analytics.events) " + "SELECT id, value FROM source ORDER BY id", + id="injected-cte", + ), + ], +) +def test_reads_mapped_source_queries_without_mutating_ast( + request: pytest.FixtureRequest, adapter_fixture: str, query_sql: str +) -> None: + adapter = request.getfixturevalue(adapter_fixture) + query = parse_one(query_sql, dialect="clickhouse") + original_query = query.copy() + + assert adapter.fetchall(query) == _EXPECTED_ROWS + assert query == original_query + + +def test_raw_string_sql_is_not_rewritten(engine_adapter: ClickhouseEngineAdapter) -> None: + from clickhouse_connect.driver.exceptions import DatabaseError + + with pytest.raises(DatabaseError) as error: + engine_adapter.fetchall(_SOURCE_QUERY) + assert error.value.code in {62, 81} + + +@pytest.mark.parametrize("adapter_fixture", ["engine_adapter", "injected_engine_adapter"]) +def test_ctas_reads_mapped_source(request: pytest.FixtureRequest, adapter_fixture: str) -> None: + adapter = request.getfixturevalue(adapter_fixture) + target = "default.sqlmesh_datalake_ctas" + adapter.drop_table(target, exists=True) + try: + adapter.ctas( + target, + parse_one(_SOURCE_QUERY, dialect="clickhouse"), + target_columns_to_types=_COLUMN_TYPES, + ) + + assert ( + adapter.fetchall( + parse_one(f"SELECT id, value FROM {target} ORDER BY id", dialect="clickhouse") + ) + == _EXPECTED_ROWS + ) + finally: + adapter.drop_table(target, exists=True) + + +@pytest.mark.parametrize("adapter_fixture", ["engine_adapter", "injected_engine_adapter"]) +def test_insert_append_reads_mapped_source( + request: pytest.FixtureRequest, adapter_fixture: str +) -> None: + adapter = request.getfixturevalue(adapter_fixture) + target = "default.sqlmesh_datalake_insert" + adapter.drop_table(target, exists=True) + try: + adapter.create_table(target, _COLUMN_TYPES, exists=False) + adapter.insert_append( + target, + parse_one(_SOURCE_QUERY, dialect="clickhouse"), + target_columns_to_types=_COLUMN_TYPES, + ) + + assert ( + adapter.fetchall( + parse_one(f"SELECT id, value FROM {target} ORDER BY id", dialect="clickhouse") + ) + == _EXPECTED_ROWS + ) + finally: + adapter.drop_table(target, exists=True) + + +@pytest.mark.parametrize("adapter_fixture", ["engine_adapter", "injected_engine_adapter"]) +def test_replace_query_reads_mapped_source( + request: pytest.FixtureRequest, adapter_fixture: str +) -> None: + adapter = request.getfixturevalue(adapter_fixture) + target = "default.sqlmesh_datalake_replace" + adapter.drop_table(target, exists=True) + try: + adapter.ctas( + target, + parse_one("SELECT 99 AS id, 'stale' AS value", dialect="clickhouse"), + target_columns_to_types=_COLUMN_TYPES, + ) + adapter.replace_query( + target, + parse_one(_SOURCE_QUERY, dialect="clickhouse"), + target_columns_to_types=_COLUMN_TYPES, + ) + + assert ( + adapter.fetchall( + parse_one(f"SELECT id, value FROM {target} ORDER BY id", dialect="clickhouse") + ) + == _EXPECTED_ROWS + ) + finally: + adapter.drop_table(target, exists=True) + + +def test_unknown_materialization_source_fails_before_target_creation( + engine_adapter: ClickhouseEngineAdapter, +) -> None: + target = "default.sqlmesh_datalake_unknown_source" + engine_adapter.drop_table(target, exists=True) + + try: + with pytest.raises(SQLMeshError, match="Provided catalog: foreign"): + engine_adapter.ctas( + target, + parse_one("SELECT id, value FROM foreign.analytics.events", dialect="clickhouse"), + target_columns_to_types=_COLUMN_TYPES, + ) + + assert not engine_adapter.table_exists(target) + finally: + engine_adapter.drop_table(target, exists=True) + + +def test_uninjected_view_maps_source(engine_adapter: ClickhouseEngineAdapter) -> None: + view = "default.sqlmesh_datalake_view" + engine_adapter.drop_view(view, ignore_if_not_exists=True) + try: + engine_adapter.create_view(view, parse_one(_SOURCE_QUERY, dialect="clickhouse")) + + assert ( + engine_adapter.fetchall( + parse_one(f"SELECT id, value FROM {view} ORDER BY id", dialect="clickhouse") + ) + == _EXPECTED_ROWS + ) + finally: + engine_adapter.drop_view(view, ignore_if_not_exists=True) + + +def test_injected_view_rejects_mapped_source( + injected_engine_adapter: ClickhouseEngineAdapter, +) -> None: + adapter = injected_engine_adapter + view = "default.sqlmesh_datalake_injected_view" + adapter.drop_view(view, ignore_if_not_exists=True) + try: + with pytest.raises(SQLMeshError, match="all catalog operations"): + adapter.create_view(view, parse_one(_SOURCE_QUERY, dialect="clickhouse")) + assert not adapter.table_exists(view) + finally: + adapter.drop_view(view, ignore_if_not_exists=True) + + +def test_injected_virtual_catalog_is_stripped_from_reads( + injected_engine_adapter: ClickhouseEngineAdapter, +) -> None: + target = "default.sqlmesh_datalake_virtual_catalog" + injected_engine_adapter.drop_table(target, exists=True) + try: + injected_engine_adapter.ctas( + target, + parse_one("SELECT 1 AS id, 'local' AS value", dialect="clickhouse"), + target_columns_to_types=_COLUMN_TYPES, + ) + virtual_query = parse_one( + "SELECT id, value FROM __clickhouse_datalake__.default.sqlmesh_datalake_virtual_catalog", + dialect="clickhouse", + ) + + assert injected_engine_adapter.fetchall(virtual_query) == [(1, "local")] + finally: + injected_engine_adapter.drop_table(target, exists=True) diff --git a/tests/core/engine_adapter/test_clickhouse.py b/tests/core/engine_adapter/test_clickhouse.py index e0355f456d..cde0367336 100644 --- a/tests/core/engine_adapter/test_clickhouse.py +++ b/tests/core/engine_adapter/test_clickhouse.py @@ -2092,3 +2092,324 @@ def test_creating_a_database_invalidates_the_engine_cache( assert "new_db" not in adapter._database_engine_cache assert "_has_replicated_database" not in adapter.__dict__ + + +def test_clickhouse_datalake_catalog_mapping_config_is_forwarded(): + from sqlmesh.core.config.connection import ClickhouseConnectionConfig + + mapping = {"lake": "iceberg"} + config = ClickhouseConnectionConfig( + host="localhost", username="user", datalake_catalog_mapping=mapping + ) + + assert config._extra_engine_config["datalake_catalog_mapping"] == mapping + + +@pytest.mark.parametrize( + "mapping", + [ + {"": "iceberg"}, + {"lake.schema": "iceberg"}, + {"lake": ""}, + {"lake": "iceberg.db"}, + ], +) +def test_clickhouse_datalake_catalog_mapping_rejects_invalid_identifiers( + mapping: t.Dict[str, str], +): + from sqlmesh.core.config.connection import ClickhouseConnectionConfig + from sqlmesh.utils.errors import ConfigError + + with pytest.raises(ConfigError, match="datalake_catalog_mapping"): + ClickhouseConnectionConfig( + host="localhost", username="user", datalake_catalog_mapping=mapping + ) + + +@pytest.mark.parametrize("inject_catalog", [False, True]) +@pytest.mark.parametrize( + "query_sql, expected_sql", + [ + ( + "SELECT e.id FROM lake.source.events AS e", + 'SELECT "e"."id" FROM "iceberg"."source.events" AS "e"', + ), + ( + "SELECT lake.source.events.id FROM lake.source.events", + 'SELECT "iceberg"."source.events"."id" FROM "iceberg"."source.events"', + ), + ( + "WITH events AS (SELECT id FROM lake.source.events) " + "SELECT id FROM events UNION ALL SELECT id FROM lake.source.other", + 'WITH "events" AS (SELECT "id" FROM "iceberg"."source.events") ' + 'SELECT "id" FROM "events" UNION ALL SELECT "id" FROM ' + '"iceberg"."source.other"', + ), + ( + "SELECT nested.id FROM (SELECT id FROM lake.source.events) AS nested", + 'SELECT "nested"."id" FROM (SELECT "id" FROM "iceberg"."source.events") AS "nested"', + ), + ( + "SELECT lake.source.events.* FROM lake.source.events", + 'SELECT "iceberg"."source.events".* FROM "iceberg"."source.events"', + ), + ], +) +def test_clickhouse_datalake_mapping_rewrites_direct_query_sources( + make_mocked_engine_adapter: t.Callable, + inject_catalog: bool, + query_sql: str, + expected_sql: str, +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, datalake_catalog_mapping={"lake": "iceberg"} + ) + if inject_catalog: + adapter.inject_virtual_catalog("clickhouse_gw") + + query = parse_one(query_sql, dialect="clickhouse") + original = query.copy() + adapter.fetchall(query, quote_identifiers=True) + + assert to_sql_calls(adapter) == [expected_sql] + assert query == original + + +@pytest.mark.parametrize("mapping", [{}, {"lake": "iceberg"}]) +def test_clickhouse_datalake_mapping_preserves_unmapped_foreign_read_catalogs( + make_mocked_engine_adapter: t.Callable, mapping: t.Dict[str, str] +): + adapter = make_mocked_engine_adapter(ClickhouseEngineAdapter, datalake_catalog_mapping=mapping) + adapter.inject_virtual_catalog("clickhouse_gw") + query = parse_one( + "SELECT foreign_catalog.source.events.id FROM foreign_catalog.source.events", + dialect="clickhouse", + ) + original = query.copy() + + adapter.fetchall(query, quote_identifiers=True) + + assert to_sql_calls(adapter) == [ + 'SELECT "foreign_catalog"."source"."events"."id" FROM "foreign_catalog"."source"."events"' + ] + assert query == original + + +def test_clickhouse_datalake_mapping_rendering_is_repeatable_and_does_not_mutate_ast( + make_mocked_engine_adapter: t.Callable, +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, datalake_catalog_mapping={"lake": "iceberg"} + ) + query = parse_one("SELECT lake.source.events.id FROM lake.source.events", dialect="clickhouse") + original = query.copy() + + first = adapter._to_sql(query) + second = adapter._to_sql(query) + + assert ( + first == second == ('SELECT "iceberg"."source.events"."id" FROM "iceberg"."source.events"') + ) + assert query == original + + +@pytest.mark.parametrize("inject_catalog", [False, True]) +def test_clickhouse_datalake_mapping_rewrites_insert_select_source( + make_mocked_engine_adapter: t.Callable, inject_catalog: bool +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, datalake_catalog_mapping={"lake": "iceberg"} + ) + if inject_catalog: + adapter.inject_virtual_catalog("clickhouse_gw") + target = "__clickhouse_gw__.analytics.target" if inject_catalog else "analytics.target" + statement = parse_one( + f"INSERT INTO {target} (id) SELECT lake.source.events.id FROM lake.source.events", + dialect="clickhouse", + ) + original = statement.copy() + + adapter.execute(statement) + + assert to_sql_calls(adapter) == [ + 'INSERT INTO "analytics"."target" ("id") ' + 'SELECT "iceberg"."source.events"."id" FROM "iceberg"."source.events"' + ] + assert statement == original + + +def test_clickhouse_insert_values_keeps_generic_virtual_catalog_rendering( + make_mocked_engine_adapter: t.Callable, +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, datalake_catalog_mapping={"lake": "iceberg"} + ) + adapter.inject_virtual_catalog("clickhouse_gw") + statement = parse_one( + "INSERT INTO __clickhouse_gw__.analytics.target (id) VALUES (1)", + dialect="clickhouse", + ) + original = statement.copy() + + adapter.execute(statement) + + assert to_sql_calls(adapter) == ['INSERT INTO "analytics"."target" ("id") VALUES ((1))'] + assert statement == original + + +@pytest.mark.parametrize("inject_catalog", [False, True]) +@pytest.mark.parametrize("operation", ["ctas", "insert", "replace"]) +def test_clickhouse_datalake_mapping_rewrites_materialized_query_sources( + make_mocked_engine_adapter: t.Callable, inject_catalog: bool, operation: str +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, datalake_catalog_mapping={"lake": "iceberg"} + ) + if inject_catalog: + adapter.inject_virtual_catalog("clickhouse_gw") + target = "__clickhouse_gw__.analytics.target" if inject_catalog else "analytics.target" + query = parse_one("SELECT lake.source.events.id FROM lake.source.events", dialect="clickhouse") + original = query.copy() + columns_to_types = {"id": exp.DataType.build("Int64", dialect="clickhouse")} + + if operation == "ctas": + adapter.ctas(target, query, columns_to_types) + elif operation == "insert": + adapter.insert_append(target, query, columns_to_types, track_rows_processed=False) + else: + adapter.get_data_object = lambda *args, **kwargs: None # type: ignore[method-assign] + adapter.replace_query(target, query, columns_to_types) + + rendered = "\n".join(to_sql_calls(adapter)) + assert '"iceberg"."source.events"' in rendered + assert '"lake"."source"."events"' not in rendered + assert query == original + + +@pytest.mark.parametrize("operation", ["insert", "ctas", "replace", "view"]) +def test_clickhouse_datalake_mapping_rejects_unconfigured_write_sources( + make_mocked_engine_adapter: t.Callable, operation: str +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, datalake_catalog_mapping={"lake": "iceberg"} + ) + query = parse_one("SELECT id FROM foreign_catalog.source.events", dialect="clickhouse") + columns_to_types = {"id": exp.DataType.build("Int64", dialect="clickhouse")} + + with pytest.raises(SQLMeshError, match="Provided catalog: foreign_catalog"): + if operation == "insert": + adapter.insert_append( + "analytics.target", query, columns_to_types, track_rows_processed=False + ) + elif operation == "ctas": + adapter.ctas("analytics.target", query, columns_to_types) + elif operation == "replace": + adapter.get_data_object = lambda *args, **kwargs: None # type: ignore[method-assign] + adapter.replace_query("analytics.target", query, columns_to_types) + else: + adapter.create_view("analytics.target", query) + + +def test_clickhouse_datalake_mapping_rejects_foreign_write_source_when_virtual_catalog_exists( + make_mocked_engine_adapter: t.Callable, +): + adapter = make_mocked_engine_adapter(ClickhouseEngineAdapter) + adapter.inject_virtual_catalog("clickhouse_gw") + query = parse_one("SELECT id FROM foreign_catalog.source.events", dialect="clickhouse") + + with pytest.raises(SQLMeshError, match="Provided catalog: foreign_catalog"): + adapter._get_source_queries(query, None, "analytics.target") + + +def test_clickhouse_without_mapping_or_virtual_catalog_preserves_query_source( + make_mocked_engine_adapter: t.Callable, +): + adapter = make_mocked_engine_adapter(ClickhouseEngineAdapter) + query = parse_one("SELECT id FROM foreign_catalog.source.events", dialect="clickhouse") + original = query.copy() + + source_queries = adapter._get_source_queries(query, None, "analytics.target") + + assert source_queries[0].query_factory() == original + assert query == original + + +def test_clickhouse_datalake_mapping_gives_virtual_catalog_precedence_over_mapping_key( + make_mocked_engine_adapter: t.Callable, +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, + virtual_catalog="css_data", + datalake_catalog_mapping={"css_data": "iceberg"}, + ) + adapter.inject_virtual_catalog("clickhouse_gw") + query = parse_one("SELECT id FROM css_data.source.events", dialect="clickhouse") + + source_queries = adapter._get_source_queries(query, None, "analytics.target") + rendered = adapter._to_sql(source_queries[0].query_factory()) + + assert rendered == 'SELECT "id" FROM "source"."events"' + + +def test_clickhouse_datalake_mapping_rewrites_create_view_query_source( + make_mocked_engine_adapter: t.Callable, +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, datalake_catalog_mapping={"lake": "iceberg"} + ) + query = parse_one("SELECT id FROM lake.source.events", dialect="clickhouse") + original = query.copy() + + adapter.create_view("analytics.events_view", query) + + assert any('"iceberg"."source.events"' in call for call in to_sql_calls(adapter)) + assert query == original + + +def test_clickhouse_datalake_mapping_does_not_bypass_injected_create_view_catalog_validation( + make_mocked_engine_adapter: t.Callable, +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, datalake_catalog_mapping={"lake": "iceberg"} + ) + adapter.inject_virtual_catalog("clickhouse_gw") + query = parse_one("SELECT id FROM lake.source.events", dialect="clickhouse") + original = query.copy() + + with pytest.raises( + SQLMeshError, + match="requires that all catalog operations be against a single catalog: __clickhouse_gw__", + ): + adapter.create_view("analytics.events_view", query) + + assert to_sql_calls(adapter) == [] + assert query == original + + +@pytest.mark.parametrize("mapping", [{}, {"configured_catalog": "iceberg"}]) +def test_virtual_catalog_execute_uses_configured_catalog_fallback( + make_mocked_engine_adapter: t.Callable, mapping: t.Dict[str, str] +): + adapter = make_mocked_engine_adapter( + ClickhouseEngineAdapter, + virtual_catalog="configured_catalog", + datalake_catalog_mapping=mapping, + ) + query = parse_one( + "SELECT configured_catalog.mydb.source.id, other_catalog.otherdb.source.id, " + "'configured_catalog.literal' FROM configured_catalog.mydb.source " + "JOIN other_catalog.otherdb.source ON configured_catalog.mydb.source.id = " + "other_catalog.otherdb.source.id", + dialect="clickhouse", + ) + original = query.copy() + + adapter.execute(query) + + assert to_sql_calls(adapter) == [ + 'SELECT "mydb"."source"."id", "other_catalog"."otherdb"."source"."id", ' + '\'configured_catalog.literal\' FROM "mydb"."source" JOIN ' + '"other_catalog"."otherdb"."source" ON "mydb"."source"."id" = ' + '"other_catalog"."otherdb"."source"."id"' + ] + assert query == original From fcf348e768acc0e190ffbcb53ccbf61b58f74818 Mon Sep 17 00:00:00 2001 From: Michael Day Date: Mon, 19 Oct 2026 08:13:15 -0400 Subject: [PATCH 2/3] docs(clickhouse): simplify DataLake test instructions Signed-off-by: Michael Day --- .../docker/clickhouse-datalake/README.md | 37 ++++++++++--------- 1 file changed, 19 insertions(+), 18 deletions(-) diff --git a/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md index d221ba5f7c..83b94419fa 100644 --- a/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md +++ b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md @@ -2,7 +2,7 @@ # 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`. +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 @@ -12,15 +12,15 @@ Run the complete fixture and test module: 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`: +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 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. +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 target checks the Docker daemon first. It then installs the repository's ClickHouse test environment and starts the dedicated Compose project. +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: @@ -28,7 +28,7 @@ Compose waits up to 120 seconds for these checks: - 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`. +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: @@ -43,21 +43,28 @@ 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. +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 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 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 the `/v1` path. ClickHouse connects to `http://rest:8181/v1`. +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 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 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 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. +The tests check the ClickHouse version and read the seeded table. They also cover: -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. +- Mapped reads with qualified tables, aliases, qualified columns, and CTEs, with and without virtual catalog injection. +- Preserving the original query AST. +- Passing raw SQL strings through unchanged. +- 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. @@ -76,10 +83,4 @@ The [fixture Compose file](../compose.clickhouse-datalake.yaml) pins these image - `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. +The upstream Compose MinIO image references were unavailable during version verification, so this fixture uses pinned Chainguard images. From 71f1f02c16c0c65f604134001b05716d3b5adfaa Mon Sep 17 00:00:00 2001 From: Michael Day Date: Mon, 19 Oct 2026 08:30:12 -0400 Subject: [PATCH 3/3] test(clickhouse): strengthen DataLake contracts and run fixture in CI Signed-off-by: Michael Day --- .github/workflows/pr.yaml | 4 +++ Makefile | 2 +- .../docker/clickhouse-datalake/README.md | 4 ++- .../test_integration_clickhouse_datalake.py | 8 ------ tests/core/engine_adapter/test_clickhouse.py | 26 ++++++++++--------- 5 files changed, 22 insertions(+), 22 deletions(-) diff --git a/.github/workflows/pr.yaml b/.github/workflows/pr.yaml index 2c18fe1e04..1e6a93f0ad 100644 --- a/.github/workflows/pr.yaml +++ b/.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() }} diff --git a/Makefile b/Makefile index 75c1b491ff..b5aebb53b5 100644 --- a/Makefile +++ b/Makefile @@ -203,7 +203,7 @@ 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 + 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 diff --git a/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md index 83b94419fa..fc2be35f3f 100644 --- a/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md +++ b/tests/core/engine_adapter/integration/docker/clickhouse-datalake/README.md @@ -12,6 +12,9 @@ Run the complete fixture and test module: 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 @@ -59,7 +62,6 @@ The tests check the ClickHouse version and read the seeded table. They also cove - Mapped reads with qualified tables, aliases, qualified columns, and CTEs, with and without virtual catalog injection. - Preserving the original query AST. -- Passing raw SQL strings through unchanged. - CTAS, append, and replace materializations. - Rejection of an unknown source before CTAS creates its target. - Stripping the injected virtual catalog from local reads. diff --git a/tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py b/tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py index e2e23dcd5a..190be5659a 100644 --- a/tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py +++ b/tests/core/engine_adapter/integration/test_integration_clickhouse_datalake.py @@ -143,14 +143,6 @@ def test_reads_mapped_source_queries_without_mutating_ast( assert query == original_query -def test_raw_string_sql_is_not_rewritten(engine_adapter: ClickhouseEngineAdapter) -> None: - from clickhouse_connect.driver.exceptions import DatabaseError - - with pytest.raises(DatabaseError) as error: - engine_adapter.fetchall(_SOURCE_QUERY) - assert error.value.code in {62, 81} - - @pytest.mark.parametrize("adapter_fixture", ["engine_adapter", "injected_engine_adapter"]) def test_ctas_reads_mapped_source(request: pytest.FixtureRequest, adapter_fixture: str) -> None: adapter = request.getfixturevalue(adapter_fixture) diff --git a/tests/core/engine_adapter/test_clickhouse.py b/tests/core/engine_adapter/test_clickhouse.py index cde0367336..3d989f00f0 100644 --- a/tests/core/engine_adapter/test_clickhouse.py +++ b/tests/core/engine_adapter/test_clickhouse.py @@ -2195,22 +2195,24 @@ def test_clickhouse_datalake_mapping_preserves_unmapped_foreign_read_catalogs( assert query == original -def test_clickhouse_datalake_mapping_rendering_is_repeatable_and_does_not_mutate_ast( - make_mocked_engine_adapter: t.Callable, +@pytest.mark.parametrize("fetch_method", ["fetchall", "fetchdf"]) +def test_clickhouse_datalake_mapping_leaves_raw_sql_strings_unchanged( + make_mocked_engine_adapter: t.Callable, fetch_method: str ): adapter = make_mocked_engine_adapter( ClickhouseEngineAdapter, datalake_catalog_mapping={"lake": "iceberg"} ) - query = parse_one("SELECT lake.source.events.id FROM lake.source.events", dialect="clickhouse") - original = query.copy() + query = "SELECT id, value FROM lake.analytics.events ORDER BY id" - first = adapter._to_sql(query) - second = adapter._to_sql(query) + if fetch_method == "fetchall": + adapter.fetchall(query) + adapter.cursor.execute.assert_called_once_with(query) + else: + import pandas as pd - assert ( - first == second == ('SELECT "iceberg"."source.events"."id" FROM "iceberg"."source.events"') - ) - assert query == original + adapter.cursor.client.query_df.return_value = pd.DataFrame() + adapter.fetchdf(query) + adapter.cursor.client.query_df.assert_called_once_with(query, use_extended_dtypes=True) @pytest.mark.parametrize("inject_catalog", [False, True]) @@ -2222,7 +2224,7 @@ def test_clickhouse_datalake_mapping_rewrites_insert_select_source( ) if inject_catalog: adapter.inject_virtual_catalog("clickhouse_gw") - target = "__clickhouse_gw__.analytics.target" if inject_catalog else "analytics.target" + target = "lake.analytics.target" statement = parse_one( f"INSERT INTO {target} (id) SELECT lake.source.events.id FROM lake.source.events", dialect="clickhouse", @@ -2232,7 +2234,7 @@ def test_clickhouse_datalake_mapping_rewrites_insert_select_source( adapter.execute(statement) assert to_sql_calls(adapter) == [ - 'INSERT INTO "analytics"."target" ("id") ' + 'INSERT INTO "lake"."analytics"."target" ("id") ' 'SELECT "iceberg"."source.events"."id" FROM "iceberg"."source.events"' ] assert statement == original