From 9aad315f609a92a0b6121772e8ab113be07d1449 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 12 Oct 2026 13:41:16 +0200 Subject: [PATCH 1/8] Fix cluster reads of Iceberg tables reached through Alias. DataLakeCatalog builds a fresh storage on every lookup, so object_storage_cluster file listing ran before the snapshot was loaded and threw "Metadata is not initialized". Co-authored-by: Cursor --- .../StorageObjectStorageSource.cpp | 2 + .../integration/test_database_iceberg/test.py | 62 +++++++ ...01_alias_iceberg_catalog_cluster.reference | 24 +++ .../05301_alias_iceberg_catalog_cluster.sh | 174 ++++++++++++++++++ 4 files changed, 262 insertions(+) create mode 100644 tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.reference create mode 100755 tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.sh diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp index ab23503dfc413..f86d0b4b2757b 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp @@ -411,6 +411,8 @@ std::shared_ptr StorageObjectStorageSource::createFileIterator( if (configuration->isDataLakeConfiguration() && (!storage_metadata || !storage_metadata->datalake_table_state.has_value())) { + configuration->lazyInitializeIfNeeded(object_storage, local_context); + if (auto state = configuration->getTableStateSnapshot(local_context)) { auto fixed_metadata = storage_metadata diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 5270a2b5a3bd1..c428f8e2bd6b8 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -3088,3 +3088,65 @@ def test_catalog_listing_error_surfaces_in_system_tables(started_cluster): assert "table_x" in result node.query(f"DROP DATABASE IF EXISTS {CATALOG_NAME}") + + +def test_alias_to_datalake_catalog_object_storage_cluster(started_cluster): + """ + A DataLakeCatalog lookup builds a fresh storage. Query analysis initializes one + copy, then an Alias read looks the table up again. Cluster planning lists files + from that second copy in StorageObjectStorageCluster::getTaskIteratorExtension. + + Direct catalog reads and count() through the alias succeed. A distributed scan + through the alias must return the same rows. + """ + node = started_cluster.instances["node1"] + + test_ref = f"test_alias_osc_{uuid.uuid4().hex[:8]}" + namespace = f"{test_ref}_ns" + table_name = f"{test_ref}_t" + catalog_table = f"{CATALOG_NAME}.`{namespace}.{table_name}`" + alias_name = f"{test_ref}_alias" + table_settings = { + "allow_database_iceberg": 1, + "allow_insert_into_iceberg": 1, + "write_full_path_in_iceberg_metadata": 1, + } + cluster_settings = { + **table_settings, + "object_storage_cluster": "cluster_simple", + } + expected = "1\n2\n3\n" + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + create_clickhouse_iceberg_table( + started_cluster, node, namespace, table_name, "(id Int64)" + ) + node.query(f"INSERT INTO {catalog_table} VALUES (1), (2), (3);", settings=table_settings) + + node.query(f"DROP TABLE IF EXISTS {alias_name}") + node.query( + f"CREATE TABLE {alias_name} ENGINE = Alias('{CATALOG_NAME}', '{namespace}.{table_name}')" + ) + + try: + assert node.query( + f"SELECT id FROM {catalog_table} ORDER BY id", + settings=cluster_settings, + ) == expected + + assert node.query( + f"SELECT id FROM {alias_name} ORDER BY id", + settings=table_settings, + ) == expected + + assert node.query( + f"SELECT count() FROM {alias_name}", + settings=cluster_settings, + ) == "3\n" + + assert node.query( + f"SELECT id FROM {alias_name} WHERE id > 0 ORDER BY id", + settings=cluster_settings, + ) == expected + finally: + node.query(f"DROP TABLE IF EXISTS {alias_name}") diff --git a/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.reference b/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.reference new file mode 100644 index 0000000000000..70c90fda37ee0 --- /dev/null +++ b/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.reference @@ -0,0 +1,24 @@ +local-direct-filtered +67 2026-10-07 06:36:00.000000 +local-alias-filtered +67 2026-10-07 06:36:00.000000 +local-alias-limit1 +2026-10-06 14:00:00.000000 s0 +local-alias-trivcount +1000 +osc-direct-filtered +67 2026-10-07 06:36:00.000000 +osc-alias-filtered +67 2026-10-07 06:36:00.000000 +osc-alias-limit1 +2026-10-06 14:00:00.000000 s0 +osc-alias-trivcount +1000 +pr-direct-filtered +67 2026-10-07 06:36:00.000000 +pr-alias-filtered +67 2026-10-07 06:36:00.000000 +pr-alias-limit1 +2026-10-06 14:00:00.000000 s0 +pr-alias-trivcount +1000 diff --git a/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.sh b/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.sh new file mode 100755 index 0000000000000..cb272ca8b49b9 --- /dev/null +++ b/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.sh @@ -0,0 +1,174 @@ +#!/usr/bin/env bash +# Tags: no-fasttest, no-replicated-database +# Tag no-replicated-database: IcebergLocal is non-replicated. +# +# Customer repro: an Alias over a DataLakeCatalog Iceberg table. +# The catalog is a local REST stub in front of an IcebergLocal table. +# A filtered scan and a LIMIT through the alias must match a direct catalog read +# with no cluster, with object_storage_cluster, and with parallel replicas. +# count() through the alias is the control that already succeeded before the fix. + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +set -euo pipefail + +WH="${CLICKHOUSE_USER_FILES_UNIQUE}/wh" +TABLE_PATH="${WH}/pnl/pnl_t" +ICE_DB="${CLICKHOUSE_DATABASE}_ice" +PNL_DB="${CLICKHOUSE_DATABASE}_pnl" +STUB_PY="${CLICKHOUSE_TMP}/alias_iceberg_stub_${CLICKHOUSE_TEST_UNIQUE_NAME}.py" +STUB_LOG="${CLICKHOUSE_TMP}/alias_iceberg_stub_${CLICKHOUSE_TEST_UNIQUE_NAME}.log" +STUB_PID="" + +STUB_PORT=$(python3 -c 'import socket; s=socket.socket(); s.bind(("127.0.0.1", 0)); print(s.getsockname()[1]); s.close()') + +cleanup() { + ${CLICKHOUSE_CLIENT} --query "DROP TABLE IF EXISTS ${PNL_DB}.pnl_t" >/dev/null 2>&1 || true + ${CLICKHOUSE_CLIENT} --query "DROP DATABASE IF EXISTS ${PNL_DB}" >/dev/null 2>&1 || true + ${CLICKHOUSE_CLIENT} --query "DROP DATABASE IF EXISTS ${ICE_DB}" >/dev/null 2>&1 || true + ${CLICKHOUSE_CLIENT} --query "DROP TABLE IF EXISTS src" >/dev/null 2>&1 || true + if [[ -n "${STUB_PID}" ]]; then + kill "${STUB_PID}" >/dev/null 2>&1 || true + wait "${STUB_PID}" >/dev/null 2>&1 || true + fi + rm -rf "${WH}" "${STUB_PY}" "${STUB_LOG}" +} +trap cleanup EXIT + +mkdir -p "${WH}/pnl" "$(dirname "${STUB_PY}")" + +cat > "${STUB_PY}" <<'PY' +import glob +import json +import os +import sys +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +WH = sys.argv[1] +PORT = int(sys.argv[2]) + + +def tables(ns): + return sorted( + os.path.basename(p) + for p in glob.glob(os.path.join(WH, ns, "*")) + if os.path.isdir(os.path.join(p, "metadata")) + ) + + +def newest_meta(ns, t): + files = glob.glob(os.path.join(WH, ns, t, "metadata", "*.metadata.json")) + files.sort(key=os.path.getmtime) + return files[-1] + + +class H(BaseHTTPRequestHandler): + def _send(self, code, obj): + body = json.dumps(obj).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + path = self.path.split("?")[0].rstrip("/") + parts = path.split("/")[2:] if path.startswith("/v1") else None + if parts is None: + return self._send(404, {"error": {"message": "not found", "type": "NotFound", "code": 404}}) + if parts == ["config"]: + return self._send(200, {"defaults": {}, "overrides": {}}) + if parts == ["namespaces"]: + if "parent=" in self.path: + return self._send(200, {"namespaces": []}) + nss = sorted(os.path.basename(p) for p in glob.glob(os.path.join(WH, "*")) if os.path.isdir(p)) + return self._send(200, {"namespaces": [[n] for n in nss]}) + if len(parts) == 2 and parts[0] == "namespaces": + return self._send(200, {"namespace": [parts[1]], "properties": {}}) + if len(parts) == 3 and parts[0] == "namespaces" and parts[2] == "tables": + return self._send(200, {"identifiers": [{"namespace": [parts[1]], "name": t} for t in tables(parts[1])]}) + if len(parts) == 4 and parts[0] == "namespaces" and parts[2] == "tables": + ns, t = parts[1], parts[3] + if t not in tables(ns): + return self._send(404, {"error": {"message": "no table", "type": "NoSuchTableException", "code": 404}}) + m = newest_meta(ns, t) + with open(m) as f: + meta = json.load(f) + if "://" not in meta.get("location", ""): + meta["location"] = "file://" + meta["location"] + return self._send(200, {"metadata-location": "file://" + m, "metadata": meta, "config": {}}) + return self._send(404, {"error": {"message": "not found", "type": "NotFound", "code": 404}}) + + def log_message(self, *a): + pass + + +ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever() +PY + +python3 "${STUB_PY}" "${WH}" "${STUB_PORT}" >"${STUB_LOG}" 2>&1 & +STUB_PID=$! + +for _ in $(seq 1 50); do + if python3 -c "import socket; socket.create_connection(('127.0.0.1', ${STUB_PORT}), 0.2).close()" 2>/dev/null; then + break + fi + sleep 0.1 +done + +${CLICKHOUSE_CLIENT} --allow_insert_into_iceberg 1 --query " + CREATE TABLE src ( + TimeSlice DateTime64(6, 'UTC'), + TradeDate Date32, + Strategy String, + NetPL Float64 + ) ENGINE = IcebergLocal('${TABLE_PATH}') + PARTITION BY TradeDate +" + +${CLICKHOUSE_CLIENT} --allow_insert_into_iceberg 1 --query " + INSERT INTO src + SELECT + toDateTime64('2026-10-06 14:00:00', 6, 'UTC') + number * 60, + toDate32('2026-10-06') + (number % 3), + 's' || toString(number % 5), + number / 7 + FROM numbers(1000) +" + +${CLICKHOUSE_CLIENT} --allow_database_iceberg 1 --query " + CREATE DATABASE ${ICE_DB} ENGINE = DataLakeCatalog('http://127.0.0.1:${STUB_PORT}/v1') + SETTINGS + catalog_type = 'rest', + auth_header = 'Authorization: Bearer sandboxonlytesttoken0001', + warehouse = 'warehouse', + vended_credentials = false +" + +${CLICKHOUSE_CLIENT} --query "CREATE DATABASE ${PNL_DB}" +${CLICKHOUSE_CLIENT} --query "CREATE TABLE ${PNL_DB}.pnl_t ENGINE = Alias(${ICE_DB}.\`pnl.pnl_t\`)" + +FILT="WHERE TradeDate = '2026-10-06' AND Strategy = 's1'" +OSC="SETTINGS object_storage_cluster = 'test_shard_localhost'" +PR="SETTINGS enable_parallel_replicas = 1, cluster_for_parallel_replicas = 'test_shard_localhost', parallel_replicas_for_cluster_engines = 1, max_parallel_replicas = 2, parallel_replicas_for_non_replicated_merge_tree = 1" + +run() { + local tag="$1" + local sql="$2" + echo "${tag}" + ${CLICKHOUSE_CLIENT} --query "${sql}" +} + +for mode in local osc pr; do + case "${mode}" in + local) S="" ;; + osc) S="${OSC}" ;; + pr) S="${PR}" ;; + esac + run "${mode}-direct-filtered" "SELECT count(), toTimeZone(max(TimeSlice), 'UTC') FROM ${ICE_DB}.\`pnl.pnl_t\` ${FILT} ${S}" + run "${mode}-alias-filtered" "SELECT count(), toTimeZone(max(TimeSlice), 'UTC') FROM ${PNL_DB}.pnl_t ${FILT} ${S}" + run "${mode}-alias-limit1" "SELECT toTimeZone(TimeSlice, 'UTC'), Strategy FROM ${PNL_DB}.pnl_t ORDER BY TimeSlice LIMIT 1 ${S}" + run "${mode}-alias-trivcount" "SELECT count() FROM ${PNL_DB}.pnl_t ${S}" +done From 6b711b230222edce9b2db66f6bd3d5a8cc9c0b24 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 12 Oct 2026 15:38:11 +0200 Subject: [PATCH 2/8] Cover catalog wrappers that re-read Iceberg tables with object_storage_cluster. Alias, Buffer, Merge, materialized views, loop, and Distributed all resolve a fresh catalog storage, so each one is checked the same way for local, query, and database cluster settings. Co-authored-by: Cursor --- .../integration/test_database_iceberg/test.py | 125 +++++++++++++----- 1 file changed, 93 insertions(+), 32 deletions(-) diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index c428f8e2bd6b8..a46b335ce85f2 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -3090,63 +3090,124 @@ def test_catalog_listing_error_surfaces_in_system_tables(started_cluster): node.query(f"DROP DATABASE IF EXISTS {CATALOG_NAME}") -def test_alias_to_datalake_catalog_object_storage_cluster(started_cluster): +def _create_catalog_wrapper(node, engine, name, catalog_db, namespace, table_name, alias_name): + catalog_table = f"{catalog_db}.`{namespace}.{table_name}`" + qualified_name = f"{namespace}.{table_name}" + if engine == "Alias": + node.query(f"CREATE TABLE {name} ENGINE = Alias('{catalog_db}', '{qualified_name}')") + elif engine == "Buffer": + node.query( + f"CREATE TABLE {name} (id Int64) ENGINE = Buffer('{catalog_db}', '{qualified_name}', " + "1, 86400, 86400, 1000000, 1000000, 1000000000, 1000000000)" + ) + elif engine == "Merge": + node.query( + f"CREATE TABLE {name} (id Int64) ENGINE = Merge('{catalog_db}', '^{namespace}[.]{table_name}$')" + ) + elif engine == "MergeAlias": + node.query(f"CREATE TABLE {name} (id Int64) ENGINE = Merge(currentDatabase(), '^{alias_name}$')") + elif engine == "MaterializedView": + node.query(f"CREATE MATERIALIZED VIEW {name} TO {catalog_table} AS SELECT id FROM {alias_name}") + elif engine == "Distributed": + node.query( + f"CREATE TABLE {name} (id Int64) " + f"ENGINE = Distributed('cluster_simple', currentDatabase(), '{alias_name}')" + ) + else: + raise AssertionError(f"Unexpected engine {engine}") + + +@pytest.mark.parametrize( + "engine", + ["Alias", "Buffer", "Merge", "MergeAlias", "MaterializedView", "Loop", "Distributed"], +) +@pytest.mark.parametrize("cluster_mode", ["local", "query", "database"]) +def test_datalake_catalog_wrapper_object_storage_cluster(started_cluster, engine, cluster_mode): """ - A DataLakeCatalog lookup builds a fresh storage. Query analysis initializes one - copy, then an Alias read looks the table up again. Cluster planning lists files - from that second copy in StorageObjectStorageCluster::getTaskIteratorExtension. + A DataLakeCatalog lookup builds a fresh storage and leaves its configuration + uninitialized. Each engine below resolves that table again, then lists files. - Direct catalog reads and count() through the alias succeed. A distributed scan - through the alias must return the same rows. + cluster_mode chooses where object_storage_cluster comes from: + local keeps it unset, query sets it on the scan, database sets it on the catalog. """ + if engine == "Loop" and cluster_mode != "local": + pytest.skip("loop has no cluster table function") + node = started_cluster.instances["node1"] + node2 = started_cluster.instances["node2"] - test_ref = f"test_alias_osc_{uuid.uuid4().hex[:8]}" + test_ref = f"test_wrapper_osc_{uuid.uuid4().hex[:8]}" namespace = f"{test_ref}_ns" table_name = f"{test_ref}_t" catalog_table = f"{CATALOG_NAME}.`{namespace}.{table_name}`" + cluster_catalog_name = f"{test_ref}_catalog_cluster" + catalog_db = cluster_catalog_name if cluster_mode == "database" else CATALOG_NAME alias_name = f"{test_ref}_alias" + wrapper_name = alias_name if engine == "Alias" else f"{test_ref}_wrapper" table_settings = { "allow_database_iceberg": 1, "allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1, } - cluster_settings = { + settings = table_settings if cluster_mode != "query" else { **table_settings, "object_storage_cluster": "cluster_simple", } - expected = "1\n2\n3\n" + # A clustered Buffer destination reports WithMergeableState, so Buffer plans + # the in-memory branch as a second read of the same query. + distinct = "DISTINCT " if engine == "Buffer" and cluster_mode != "local" else "" + nodes = [node, node2] if engine == "Distributed" else [node] + created = {replica: [] for replica in nodes} create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) create_clickhouse_iceberg_table( started_cluster, node, namespace, table_name, "(id Int64)" ) node.query(f"INSERT INTO {catalog_table} VALUES (1), (2), (3);", settings=table_settings) - - node.query(f"DROP TABLE IF EXISTS {alias_name}") - node.query( - f"CREATE TABLE {alias_name} ENGINE = Alias('{CATALOG_NAME}', '{namespace}.{table_name}')" - ) + if cluster_mode == "database": + create_clickhouse_iceberg_database( + started_cluster, + node, + cluster_catalog_name, + {"object_storage_cluster": "cluster_simple"}, + ) try: - assert node.query( - f"SELECT id FROM {catalog_table} ORDER BY id", - settings=cluster_settings, - ) == expected - - assert node.query( - f"SELECT id FROM {alias_name} ORDER BY id", - settings=table_settings, - ) == expected - - assert node.query( - f"SELECT count() FROM {alias_name}", - settings=cluster_settings, - ) == "3\n" + for replica in nodes: + if replica is node2: + catalog_settings = ( + {"object_storage_cluster": "cluster_simple"} if cluster_mode == "database" else {} + ) + create_clickhouse_iceberg_database( + started_cluster, replica, catalog_db, catalog_settings + ) + if engine in ("Alias", "MergeAlias", "MaterializedView", "Distributed"): + replica.query( + f"CREATE TABLE {alias_name} ENGINE = Alias('{catalog_db}', '{namespace}.{table_name}')" + ) + created[replica].append(alias_name) + if engine != "Alias" and engine != "Loop" and replica is node: + _create_catalog_wrapper( + replica, engine, wrapper_name, catalog_db, namespace, table_name, alias_name + ) + created[replica].append(wrapper_name) + if engine == "Loop": + source = ( + f"(SELECT id FROM loop('{catalog_db}', '{namespace}.{table_name}') " + "WHERE id > 0 LIMIT 3)" + ) + else: + source = wrapper_name assert node.query( - f"SELECT id FROM {alias_name} WHERE id > 0 ORDER BY id", - settings=cluster_settings, - ) == expected + f"SELECT {distinct}id FROM {source} WHERE id > 0 ORDER BY id", + settings=settings, + ) == "1\n2\n3\n" finally: - node.query(f"DROP TABLE IF EXISTS {alias_name}") + for replica, names in created.items(): + for name in reversed(names): + replica.query(f"DROP TABLE IF EXISTS {name}") + if engine == "Distributed": + node2.query(f"DROP DATABASE IF EXISTS {catalog_db}") + if cluster_mode == "database": + node.query(f"DROP DATABASE IF EXISTS {cluster_catalog_name}") From 3e8533385411498b6ee926ab3e49f3d7eae7f15a Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 12 Oct 2026 16:15:42 +0200 Subject: [PATCH 3/8] Move the alias Iceberg catalog scan into the integration tests. The customer filter, limit, and count checks now run against the REST catalog and cluster_simple, including parallel replicas, so the stateless REST stub is no longer needed. Co-authored-by: Cursor --- .../integration/test_database_iceberg/test.py | 92 +++++++++ ...01_alias_iceberg_catalog_cluster.reference | 24 --- .../05301_alias_iceberg_catalog_cluster.sh | 174 ------------------ 3 files changed, 92 insertions(+), 198 deletions(-) delete mode 100644 tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.reference delete mode 100755 tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.sh diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index a46b335ce85f2..64bb13d0056cf 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -3090,6 +3090,98 @@ def test_catalog_listing_error_surfaces_in_system_tables(started_cluster): node.query(f"DROP DATABASE IF EXISTS {CATALOG_NAME}") +def test_alias_iceberg_catalog_cluster(started_cluster): + """ + Customer repro moved from the stateless suite. An Alias over a catalog Iceberg + table must apply the same filter as a direct catalog read: local, with + object_storage_cluster, and with parallel replicas. count() is the path that + already succeeded before snapshot initialization on the cluster file listing. + """ + node = started_cluster.instances["node1"] + + test_ref = f"test_alias_repro_{uuid.uuid4().hex[:8]}" + namespace = f"{test_ref}_pnl" + table_name = "pnl_t" + catalog_table = f"{CATALOG_NAME}.`{namespace}.{table_name}`" + pnl_db = f"{test_ref}_db" + table_settings = { + "allow_database_iceberg": 1, + "allow_insert_into_iceberg": 1, + "write_full_path_in_iceberg_metadata": 1, + } + modes = { + "local": table_settings, + "osc": {**table_settings, "object_storage_cluster": "cluster_simple"}, + "pr": { + **table_settings, + "enable_parallel_replicas": 1, + "cluster_for_parallel_replicas": "cluster_simple", + "parallel_replicas_for_cluster_engines": 1, + "max_parallel_replicas": 2, + "parallel_replicas_for_non_replicated_merge_tree": 1, + }, + } + filtered = "WHERE TradeDate = '2026-10-06' AND Strategy = 's1'" + filtered_expected = "67\t2026-10-07 06:36:00.000000\n" + limit_expected = "2026-10-06 14:00:00.000000\ts0\n" + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + node.query( + f""" + CREATE TABLE {catalog_table} + ( + TimeSlice DateTime64(6, 'UTC'), + TradeDate Date32, + Strategy String, + NetPL Float64 + ) + ENGINE = IcebergS3('http://minio1:9001/warehouse-rest/{namespace}/', '{minio_access_key}', '{minio_secret_key}') + PARTITION BY TradeDate + """, + settings={ + "allow_experimental_database_iceberg": 1, + "write_full_path_in_iceberg_metadata": 1, + }, + ) + node.query( + f""" + INSERT INTO {catalog_table} + SELECT + toDateTime64('2026-10-06 14:00:00', 6, 'UTC') + number * 60, + toDate32('2026-10-06') + (number % 3), + 's' || toString(number % 5), + number / 7 + FROM numbers(1000) + """, + settings=table_settings, + ) + node.query(f"CREATE DATABASE {pnl_db}") + node.query(f"CREATE TABLE {pnl_db}.pnl_t ENGINE = Alias({CATALOG_NAME}.`{namespace}.{table_name}`)") + + try: + for settings in modes.values(): + assert node.query( + f"SELECT count(), toTimeZone(max(TimeSlice), 'UTC') FROM {catalog_table} {filtered}", + settings=settings, + ) == filtered_expected + assert node.query( + f"SELECT count(), toTimeZone(max(TimeSlice), 'UTC') FROM {pnl_db}.pnl_t {filtered}", + settings=settings, + ) == filtered_expected + assert node.query( + f"SELECT toTimeZone(TimeSlice, 'UTC'), Strategy FROM {pnl_db}.pnl_t ORDER BY TimeSlice LIMIT 1", + settings=settings, + ) == limit_expected + assert node.query( + f"SELECT count() FROM {pnl_db}.pnl_t", + settings=settings, + ) == "1000\n" + finally: + node.query(f"DROP TABLE IF EXISTS {pnl_db}.pnl_t") + node.query(f"DROP DATABASE IF EXISTS {pnl_db}") + drop_clickhouse_iceberg_table(node, namespace, table_name, if_exists=True) + + def _create_catalog_wrapper(node, engine, name, catalog_db, namespace, table_name, alias_name): catalog_table = f"{catalog_db}.`{namespace}.{table_name}`" qualified_name = f"{namespace}.{table_name}" diff --git a/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.reference b/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.reference deleted file mode 100644 index 70c90fda37ee0..0000000000000 --- a/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.reference +++ /dev/null @@ -1,24 +0,0 @@ -local-direct-filtered -67 2026-10-07 06:36:00.000000 -local-alias-filtered -67 2026-10-07 06:36:00.000000 -local-alias-limit1 -2026-10-06 14:00:00.000000 s0 -local-alias-trivcount -1000 -osc-direct-filtered -67 2026-10-07 06:36:00.000000 -osc-alias-filtered -67 2026-10-07 06:36:00.000000 -osc-alias-limit1 -2026-10-06 14:00:00.000000 s0 -osc-alias-trivcount -1000 -pr-direct-filtered -67 2026-10-07 06:36:00.000000 -pr-alias-filtered -67 2026-10-07 06:36:00.000000 -pr-alias-limit1 -2026-10-06 14:00:00.000000 s0 -pr-alias-trivcount -1000 diff --git a/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.sh b/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.sh deleted file mode 100755 index cb272ca8b49b9..0000000000000 --- a/tests/queries/0_stateless/05301_alias_iceberg_catalog_cluster.sh +++ /dev/null @@ -1,174 +0,0 @@ -#!/usr/bin/env bash -# Tags: no-fasttest, no-replicated-database -# Tag no-replicated-database: IcebergLocal is non-replicated. -# -# Customer repro: an Alias over a DataLakeCatalog Iceberg table. -# The catalog is a local REST stub in front of an IcebergLocal table. -# A filtered scan and a LIMIT through the alias must match a direct catalog read -# with no cluster, with object_storage_cluster, and with parallel replicas. -# count() through the alias is the control that already succeeded before the fix. - -CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -# shellcheck source=../shell_config.sh -. "$CUR_DIR"/../shell_config.sh - -set -euo pipefail - -WH="${CLICKHOUSE_USER_FILES_UNIQUE}/wh" -TABLE_PATH="${WH}/pnl/pnl_t" -ICE_DB="${CLICKHOUSE_DATABASE}_ice" -PNL_DB="${CLICKHOUSE_DATABASE}_pnl" -STUB_PY="${CLICKHOUSE_TMP}/alias_iceberg_stub_${CLICKHOUSE_TEST_UNIQUE_NAME}.py" -STUB_LOG="${CLICKHOUSE_TMP}/alias_iceberg_stub_${CLICKHOUSE_TEST_UNIQUE_NAME}.log" -STUB_PID="" - -STUB_PORT=$(python3 -c 'import socket; s=socket.socket(); s.bind(("127.0.0.1", 0)); print(s.getsockname()[1]); s.close()') - -cleanup() { - ${CLICKHOUSE_CLIENT} --query "DROP TABLE IF EXISTS ${PNL_DB}.pnl_t" >/dev/null 2>&1 || true - ${CLICKHOUSE_CLIENT} --query "DROP DATABASE IF EXISTS ${PNL_DB}" >/dev/null 2>&1 || true - ${CLICKHOUSE_CLIENT} --query "DROP DATABASE IF EXISTS ${ICE_DB}" >/dev/null 2>&1 || true - ${CLICKHOUSE_CLIENT} --query "DROP TABLE IF EXISTS src" >/dev/null 2>&1 || true - if [[ -n "${STUB_PID}" ]]; then - kill "${STUB_PID}" >/dev/null 2>&1 || true - wait "${STUB_PID}" >/dev/null 2>&1 || true - fi - rm -rf "${WH}" "${STUB_PY}" "${STUB_LOG}" -} -trap cleanup EXIT - -mkdir -p "${WH}/pnl" "$(dirname "${STUB_PY}")" - -cat > "${STUB_PY}" <<'PY' -import glob -import json -import os -import sys -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer - -WH = sys.argv[1] -PORT = int(sys.argv[2]) - - -def tables(ns): - return sorted( - os.path.basename(p) - for p in glob.glob(os.path.join(WH, ns, "*")) - if os.path.isdir(os.path.join(p, "metadata")) - ) - - -def newest_meta(ns, t): - files = glob.glob(os.path.join(WH, ns, t, "metadata", "*.metadata.json")) - files.sort(key=os.path.getmtime) - return files[-1] - - -class H(BaseHTTPRequestHandler): - def _send(self, code, obj): - body = json.dumps(obj).encode() - self.send_response(code) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - def do_GET(self): - path = self.path.split("?")[0].rstrip("/") - parts = path.split("/")[2:] if path.startswith("/v1") else None - if parts is None: - return self._send(404, {"error": {"message": "not found", "type": "NotFound", "code": 404}}) - if parts == ["config"]: - return self._send(200, {"defaults": {}, "overrides": {}}) - if parts == ["namespaces"]: - if "parent=" in self.path: - return self._send(200, {"namespaces": []}) - nss = sorted(os.path.basename(p) for p in glob.glob(os.path.join(WH, "*")) if os.path.isdir(p)) - return self._send(200, {"namespaces": [[n] for n in nss]}) - if len(parts) == 2 and parts[0] == "namespaces": - return self._send(200, {"namespace": [parts[1]], "properties": {}}) - if len(parts) == 3 and parts[0] == "namespaces" and parts[2] == "tables": - return self._send(200, {"identifiers": [{"namespace": [parts[1]], "name": t} for t in tables(parts[1])]}) - if len(parts) == 4 and parts[0] == "namespaces" and parts[2] == "tables": - ns, t = parts[1], parts[3] - if t not in tables(ns): - return self._send(404, {"error": {"message": "no table", "type": "NoSuchTableException", "code": 404}}) - m = newest_meta(ns, t) - with open(m) as f: - meta = json.load(f) - if "://" not in meta.get("location", ""): - meta["location"] = "file://" + meta["location"] - return self._send(200, {"metadata-location": "file://" + m, "metadata": meta, "config": {}}) - return self._send(404, {"error": {"message": "not found", "type": "NotFound", "code": 404}}) - - def log_message(self, *a): - pass - - -ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever() -PY - -python3 "${STUB_PY}" "${WH}" "${STUB_PORT}" >"${STUB_LOG}" 2>&1 & -STUB_PID=$! - -for _ in $(seq 1 50); do - if python3 -c "import socket; socket.create_connection(('127.0.0.1', ${STUB_PORT}), 0.2).close()" 2>/dev/null; then - break - fi - sleep 0.1 -done - -${CLICKHOUSE_CLIENT} --allow_insert_into_iceberg 1 --query " - CREATE TABLE src ( - TimeSlice DateTime64(6, 'UTC'), - TradeDate Date32, - Strategy String, - NetPL Float64 - ) ENGINE = IcebergLocal('${TABLE_PATH}') - PARTITION BY TradeDate -" - -${CLICKHOUSE_CLIENT} --allow_insert_into_iceberg 1 --query " - INSERT INTO src - SELECT - toDateTime64('2026-10-06 14:00:00', 6, 'UTC') + number * 60, - toDate32('2026-10-06') + (number % 3), - 's' || toString(number % 5), - number / 7 - FROM numbers(1000) -" - -${CLICKHOUSE_CLIENT} --allow_database_iceberg 1 --query " - CREATE DATABASE ${ICE_DB} ENGINE = DataLakeCatalog('http://127.0.0.1:${STUB_PORT}/v1') - SETTINGS - catalog_type = 'rest', - auth_header = 'Authorization: Bearer sandboxonlytesttoken0001', - warehouse = 'warehouse', - vended_credentials = false -" - -${CLICKHOUSE_CLIENT} --query "CREATE DATABASE ${PNL_DB}" -${CLICKHOUSE_CLIENT} --query "CREATE TABLE ${PNL_DB}.pnl_t ENGINE = Alias(${ICE_DB}.\`pnl.pnl_t\`)" - -FILT="WHERE TradeDate = '2026-10-06' AND Strategy = 's1'" -OSC="SETTINGS object_storage_cluster = 'test_shard_localhost'" -PR="SETTINGS enable_parallel_replicas = 1, cluster_for_parallel_replicas = 'test_shard_localhost', parallel_replicas_for_cluster_engines = 1, max_parallel_replicas = 2, parallel_replicas_for_non_replicated_merge_tree = 1" - -run() { - local tag="$1" - local sql="$2" - echo "${tag}" - ${CLICKHOUSE_CLIENT} --query "${sql}" -} - -for mode in local osc pr; do - case "${mode}" in - local) S="" ;; - osc) S="${OSC}" ;; - pr) S="${PR}" ;; - esac - run "${mode}-direct-filtered" "SELECT count(), toTimeZone(max(TimeSlice), 'UTC') FROM ${ICE_DB}.\`pnl.pnl_t\` ${FILT} ${S}" - run "${mode}-alias-filtered" "SELECT count(), toTimeZone(max(TimeSlice), 'UTC') FROM ${PNL_DB}.pnl_t ${FILT} ${S}" - run "${mode}-alias-limit1" "SELECT toTimeZone(TimeSlice, 'UTC'), Strategy FROM ${PNL_DB}.pnl_t ORDER BY TimeSlice LIMIT 1 ${S}" - run "${mode}-alias-trivcount" "SELECT count() FROM ${PNL_DB}.pnl_t ${S}" -done From 15eceeb40ba13a11cbcce5e7f90a3d4b82da7e04 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 12 Oct 2026 16:48:40 +0200 Subject: [PATCH 4/8] Format ClickHouse names in the Iceberg alias test docstrings. Related: https://github.com/Altinity/ClickHouse/pull/2550 Co-authored-by: Cursor --- tests/integration/test_database_iceberg/test.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 64bb13d0056cf..9e0bbc0e4e79a 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -3092,10 +3092,10 @@ def test_catalog_listing_error_surfaces_in_system_tables(started_cluster): def test_alias_iceberg_catalog_cluster(started_cluster): """ - Customer repro moved from the stateless suite. An Alias over a catalog Iceberg - table must apply the same filter as a direct catalog read: local, with - object_storage_cluster, and with parallel replicas. count() is the path that - already succeeded before snapshot initialization on the cluster file listing. + An `Alias` over a catalog `Iceberg` table must apply the same filter as a direct + catalog read: local, with `object_storage_cluster`, and with parallel replicas. + `count` is the path that already succeeded before snapshot initialization on the + cluster file listing. """ node = started_cluster.instances["node1"] @@ -3216,10 +3216,10 @@ def _create_catalog_wrapper(node, engine, name, catalog_db, namespace, table_nam @pytest.mark.parametrize("cluster_mode", ["local", "query", "database"]) def test_datalake_catalog_wrapper_object_storage_cluster(started_cluster, engine, cluster_mode): """ - A DataLakeCatalog lookup builds a fresh storage and leaves its configuration + A `DataLakeCatalog` lookup builds a fresh storage and leaves its configuration uninitialized. Each engine below resolves that table again, then lists files. - cluster_mode chooses where object_storage_cluster comes from: + `cluster_mode` chooses where `object_storage_cluster` comes from: local keeps it unset, query sets it on the scan, database sets it on the catalog. """ if engine == "Loop" and cluster_mode != "local": From 2a09ee317a90d65f72165bbb9f1c4d407ec41c48 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 12 Oct 2026 17:01:29 +0200 Subject: [PATCH 5/8] Format `Buffer` and `WithMergeableState` in the wrapper test comment. Related: https://github.com/Altinity/ClickHouse/pull/2550 Co-authored-by: Cursor --- tests/integration/test_database_iceberg/test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 9e0bbc0e4e79a..dc1e4ff5f7c75 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -3245,7 +3245,7 @@ def test_datalake_catalog_wrapper_object_storage_cluster(started_cluster, engine **table_settings, "object_storage_cluster": "cluster_simple", } - # A clustered Buffer destination reports WithMergeableState, so Buffer plans + # A clustered `Buffer` destination reports `WithMergeableState`, so `Buffer` plans # the in-memory branch as a second read of the same query. distinct = "DISTINCT " if engine == "Buffer" and cluster_mode != "local" else "" nodes = [node, node2] if engine == "Distributed" else [node] From abc516e8d5491482d026d54a960c6b272a9db2a7 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 13 Oct 2026 10:03:08 +0200 Subject: [PATCH 6/8] Keep the catalog storage alive for an `Alias` snapshot. `StorageSnapshot` only referenced the target, and `DataLakeCatalog` drops that storage when `getStorageSnapshot` returns. The analyzer then read freed memory under ASan. https: //altinity-build-artifacts.s3.amazonaws.com/json.html?PR=2550&sha=2a09ee317a90d65f72165bbb9f1c4d407ec41c48&name_0=PR&name_1=Integration%20tests%20%28amd_asan_ubsan%2C%20db%20disk%2C%20old%20analyzer%2C%208%2F8%29 Related: https://github.com/Altinity/ClickHouse/pull/2550 Co-authored-by: Cursor --- .../Optimizations/optimizeUseAggregateProjection.cpp | 2 +- .../Optimizations/optimizeUseNormalProjection.cpp | 2 +- src/Processors/QueryPlan/ReadFromMergeTree.cpp | 4 ++-- src/Storages/ObjectStorage/StorageObjectStorage.cpp | 6 +++--- src/Storages/StorageAlias.cpp | 12 ++++++++++-- src/Storages/StorageSnapshot.cpp | 11 ++++++++++- src/Storages/StorageSnapshot.h | 11 +++++++++++ 7 files changed, 38 insertions(+), 10 deletions(-) diff --git a/src/Processors/QueryPlan/Optimizations/optimizeUseAggregateProjection.cpp b/src/Processors/QueryPlan/Optimizations/optimizeUseAggregateProjection.cpp index 0a0cbe26fdccc..fb3942b047829 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizeUseAggregateProjection.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizeUseAggregateProjection.cpp @@ -873,7 +873,7 @@ std::optional optimizeUseAggregateProjections( { chassert(parent_reading_select_result); auto storage_snapshot = reading->getStorageSnapshot(); - auto proj_snapshot = std::make_shared(storage_snapshot->storage, best_candidate->projection->metadata); + auto proj_snapshot = storage_snapshot->cloneWithMetadata(best_candidate->projection->metadata); auto projection_query_info = query_info; projection_query_info.prewhere_info = nullptr; projection_query_info.row_level_filter = nullptr; diff --git a/src/Processors/QueryPlan/Optimizations/optimizeUseNormalProjection.cpp b/src/Processors/QueryPlan/Optimizations/optimizeUseNormalProjection.cpp index 14e48cea12a20..fdcf8b0a5d775 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizeUseNormalProjection.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizeUseNormalProjection.cpp @@ -399,7 +399,7 @@ std::optional optimizeUseNormalProjections( } auto storage_snapshot = reading->getStorageSnapshot(); - auto proj_snapshot = std::make_shared(storage_snapshot->storage, best_candidate->projection->metadata); + auto proj_snapshot = storage_snapshot->cloneWithMetadata(best_candidate->projection->metadata); /// Enables PREWHERE on projections to improve read efficiency and leverage query condition cache. if (query.dag && query.filter_node) diff --git a/src/Processors/QueryPlan/ReadFromMergeTree.cpp b/src/Processors/QueryPlan/ReadFromMergeTree.cpp index 21b014bd72ff8..12cf6364c255a 100644 --- a/src/Processors/QueryPlan/ReadFromMergeTree.cpp +++ b/src/Processors/QueryPlan/ReadFromMergeTree.cpp @@ -3927,7 +3927,7 @@ void ReadFromMergeTree::initializePipeline(QueryPipelineBuilder & pipeline, [[ma SingleProjectionIndexReader( std::make_shared( empty_mutations_snapshot, - std::make_shared(storage_snapshot->storage, read_info.projection->metadata), + storage_snapshot->cloneWithMetadata(read_info.projection->metadata), read_info.prewhere_info, actions_settings, reader_settings, @@ -4654,7 +4654,7 @@ void ReadFromMergeTree::createReadTasksForTextIndex(const UsefulSkipIndexes & sk } } - storage_snapshot = std::make_shared(storage_snapshot->storage, std::move(new_metadata)); + storage_snapshot = storage_snapshot->cloneWithMetadata(std::move(new_metadata)); if (output_header != nullptr) { diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index 2a9552981e0ff..b0acc944dfe1f 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -580,7 +580,7 @@ void StorageObjectStorage::read( { auto stripped = std::make_shared(*storage_snapshot->metadata); stripped->datalake_table_state.reset(); - storage_snapshot = std::make_shared(*this, std::move(stripped)); + storage_snapshot = storage_snapshot->cloneWithMetadata(std::move(stripped)); } }); @@ -616,8 +616,8 @@ void StorageObjectStorage::read( rebuilt->columns, local_context, format_settings, configuration->getPartitionStrategyType())); } - storage_snapshot = std::make_shared( - *this, std::make_shared(std::move(pinned))); + storage_snapshot = storage_snapshot->cloneWithMetadata( + std::make_shared(std::move(pinned))); } } diff --git a/src/Storages/StorageAlias.cpp b/src/Storages/StorageAlias.cpp index d6d3567665990..54dbd205688b6 100644 --- a/src/Storages/StorageAlias.cpp +++ b/src/Storages/StorageAlias.cpp @@ -350,12 +350,20 @@ std::optional StorageAlias::distributedWrite(const ASTInsertQuery StorageSnapshotPtr StorageAlias::getStorageSnapshot(const StorageMetadataPtr & metadata_snapshot, ContextPtr query_context) const { - return getTargetTable()->getStorageSnapshot(metadata_snapshot, query_context); + auto target = getTargetTable(); + auto snapshot = target->getStorageSnapshot(metadata_snapshot, query_context); + /// `DataLakeCatalog` builds a new storage on every lookup. The snapshot keeps only a reference, + /// so it must own that `shared_ptr` or the analyzer reads freed memory. + snapshot->storage_holder = std::move(target); + return snapshot; } StorageSnapshotPtr StorageAlias::getStorageSnapshotWithoutData(const StorageMetadataPtr & metadata_snapshot, ContextPtr query_context) const { - return getTargetTable()->getStorageSnapshotWithoutData(metadata_snapshot, query_context); + auto target = getTargetTable(); + auto snapshot = target->getStorageSnapshotWithoutData(metadata_snapshot, query_context); + snapshot->storage_holder = std::move(target); + return snapshot; } void StorageAlias::rename(const String & /* new_path_to_table_data */, const StorageID & new_table_id) diff --git a/src/Storages/StorageSnapshot.cpp b/src/Storages/StorageSnapshot.cpp index 7abaf6688033c..92815364ea605 100644 --- a/src/Storages/StorageSnapshot.cpp +++ b/src/Storages/StorageSnapshot.cpp @@ -39,7 +39,16 @@ StorageSnapshot::StorageSnapshot( std::shared_ptr StorageSnapshot::clone(DataPtr data_) const { - return std::make_shared(storage, metadata, std::move(data_)); + auto result = std::make_shared(storage, metadata, std::move(data_)); + result->storage_holder = storage_holder; + return result; +} + +std::shared_ptr StorageSnapshot::cloneWithMetadata(StorageMetadataPtr metadata_) const +{ + auto result = std::make_shared(storage, std::move(metadata_)); + result->storage_holder = storage_holder; + return result; } ColumnsDescription StorageSnapshot::getAllColumnsDescription() const diff --git a/src/Storages/StorageSnapshot.h b/src/Storages/StorageSnapshot.h index a3635fbf64414..0480608d0ea6c 100644 --- a/src/Storages/StorageSnapshot.h +++ b/src/Storages/StorageSnapshot.h @@ -1,6 +1,9 @@ #pragma once +#include #include +#include + namespace DB { @@ -19,6 +22,11 @@ struct StorageSnapshot const IStorage & storage; const StorageMetadataPtr metadata; + /// Owns `storage` when nothing else does. Empty for catalog-resident tables. + /// `Alias` over `DataLakeCatalog` fills it: each lookup returns a new storage, + /// and this snapshot is the only object that outlives that lookup. + StoragePtr storage_holder; + /// Additional data, on which set of columns may depend. /// E.g. data parts in MergeTree, list of blocks in Memory, etc. struct Data @@ -40,6 +48,9 @@ struct StorageSnapshot std::shared_ptr clone(DataPtr data_) const; + /// Same `storage` and `storage_holder`, replaced metadata. Drops `data`. + std::shared_ptr cloneWithMetadata(StorageMetadataPtr metadata_) const; + /// Get columns description ColumnsDescription getAllColumnsDescription() const; From bd359eba7db709ed5bdca5244d71cd747aba8b22 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 13 Oct 2026 10:23:57 +0200 Subject: [PATCH 7/8] Revert "Keep the catalog storage alive for an `Alias` snapshot." This reverts commit abc516e8d5491482d026d54a960c6b272a9db2a7. --- .../Optimizations/optimizeUseAggregateProjection.cpp | 2 +- .../Optimizations/optimizeUseNormalProjection.cpp | 2 +- src/Processors/QueryPlan/ReadFromMergeTree.cpp | 4 ++-- src/Storages/ObjectStorage/StorageObjectStorage.cpp | 6 +++--- src/Storages/StorageAlias.cpp | 12 ++---------- src/Storages/StorageSnapshot.cpp | 11 +---------- src/Storages/StorageSnapshot.h | 11 ----------- 7 files changed, 10 insertions(+), 38 deletions(-) diff --git a/src/Processors/QueryPlan/Optimizations/optimizeUseAggregateProjection.cpp b/src/Processors/QueryPlan/Optimizations/optimizeUseAggregateProjection.cpp index fb3942b047829..0a0cbe26fdccc 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizeUseAggregateProjection.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizeUseAggregateProjection.cpp @@ -873,7 +873,7 @@ std::optional optimizeUseAggregateProjections( { chassert(parent_reading_select_result); auto storage_snapshot = reading->getStorageSnapshot(); - auto proj_snapshot = storage_snapshot->cloneWithMetadata(best_candidate->projection->metadata); + auto proj_snapshot = std::make_shared(storage_snapshot->storage, best_candidate->projection->metadata); auto projection_query_info = query_info; projection_query_info.prewhere_info = nullptr; projection_query_info.row_level_filter = nullptr; diff --git a/src/Processors/QueryPlan/Optimizations/optimizeUseNormalProjection.cpp b/src/Processors/QueryPlan/Optimizations/optimizeUseNormalProjection.cpp index fdcf8b0a5d775..14e48cea12a20 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizeUseNormalProjection.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizeUseNormalProjection.cpp @@ -399,7 +399,7 @@ std::optional optimizeUseNormalProjections( } auto storage_snapshot = reading->getStorageSnapshot(); - auto proj_snapshot = storage_snapshot->cloneWithMetadata(best_candidate->projection->metadata); + auto proj_snapshot = std::make_shared(storage_snapshot->storage, best_candidate->projection->metadata); /// Enables PREWHERE on projections to improve read efficiency and leverage query condition cache. if (query.dag && query.filter_node) diff --git a/src/Processors/QueryPlan/ReadFromMergeTree.cpp b/src/Processors/QueryPlan/ReadFromMergeTree.cpp index 12cf6364c255a..21b014bd72ff8 100644 --- a/src/Processors/QueryPlan/ReadFromMergeTree.cpp +++ b/src/Processors/QueryPlan/ReadFromMergeTree.cpp @@ -3927,7 +3927,7 @@ void ReadFromMergeTree::initializePipeline(QueryPipelineBuilder & pipeline, [[ma SingleProjectionIndexReader( std::make_shared( empty_mutations_snapshot, - storage_snapshot->cloneWithMetadata(read_info.projection->metadata), + std::make_shared(storage_snapshot->storage, read_info.projection->metadata), read_info.prewhere_info, actions_settings, reader_settings, @@ -4654,7 +4654,7 @@ void ReadFromMergeTree::createReadTasksForTextIndex(const UsefulSkipIndexes & sk } } - storage_snapshot = storage_snapshot->cloneWithMetadata(std::move(new_metadata)); + storage_snapshot = std::make_shared(storage_snapshot->storage, std::move(new_metadata)); if (output_header != nullptr) { diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index b0acc944dfe1f..2a9552981e0ff 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -580,7 +580,7 @@ void StorageObjectStorage::read( { auto stripped = std::make_shared(*storage_snapshot->metadata); stripped->datalake_table_state.reset(); - storage_snapshot = storage_snapshot->cloneWithMetadata(std::move(stripped)); + storage_snapshot = std::make_shared(*this, std::move(stripped)); } }); @@ -616,8 +616,8 @@ void StorageObjectStorage::read( rebuilt->columns, local_context, format_settings, configuration->getPartitionStrategyType())); } - storage_snapshot = storage_snapshot->cloneWithMetadata( - std::make_shared(std::move(pinned))); + storage_snapshot = std::make_shared( + *this, std::make_shared(std::move(pinned))); } } diff --git a/src/Storages/StorageAlias.cpp b/src/Storages/StorageAlias.cpp index 54dbd205688b6..d6d3567665990 100644 --- a/src/Storages/StorageAlias.cpp +++ b/src/Storages/StorageAlias.cpp @@ -350,20 +350,12 @@ std::optional StorageAlias::distributedWrite(const ASTInsertQuery StorageSnapshotPtr StorageAlias::getStorageSnapshot(const StorageMetadataPtr & metadata_snapshot, ContextPtr query_context) const { - auto target = getTargetTable(); - auto snapshot = target->getStorageSnapshot(metadata_snapshot, query_context); - /// `DataLakeCatalog` builds a new storage on every lookup. The snapshot keeps only a reference, - /// so it must own that `shared_ptr` or the analyzer reads freed memory. - snapshot->storage_holder = std::move(target); - return snapshot; + return getTargetTable()->getStorageSnapshot(metadata_snapshot, query_context); } StorageSnapshotPtr StorageAlias::getStorageSnapshotWithoutData(const StorageMetadataPtr & metadata_snapshot, ContextPtr query_context) const { - auto target = getTargetTable(); - auto snapshot = target->getStorageSnapshotWithoutData(metadata_snapshot, query_context); - snapshot->storage_holder = std::move(target); - return snapshot; + return getTargetTable()->getStorageSnapshotWithoutData(metadata_snapshot, query_context); } void StorageAlias::rename(const String & /* new_path_to_table_data */, const StorageID & new_table_id) diff --git a/src/Storages/StorageSnapshot.cpp b/src/Storages/StorageSnapshot.cpp index 92815364ea605..7abaf6688033c 100644 --- a/src/Storages/StorageSnapshot.cpp +++ b/src/Storages/StorageSnapshot.cpp @@ -39,16 +39,7 @@ StorageSnapshot::StorageSnapshot( std::shared_ptr StorageSnapshot::clone(DataPtr data_) const { - auto result = std::make_shared(storage, metadata, std::move(data_)); - result->storage_holder = storage_holder; - return result; -} - -std::shared_ptr StorageSnapshot::cloneWithMetadata(StorageMetadataPtr metadata_) const -{ - auto result = std::make_shared(storage, std::move(metadata_)); - result->storage_holder = storage_holder; - return result; + return std::make_shared(storage, metadata, std::move(data_)); } ColumnsDescription StorageSnapshot::getAllColumnsDescription() const diff --git a/src/Storages/StorageSnapshot.h b/src/Storages/StorageSnapshot.h index 0480608d0ea6c..a3635fbf64414 100644 --- a/src/Storages/StorageSnapshot.h +++ b/src/Storages/StorageSnapshot.h @@ -1,9 +1,6 @@ #pragma once -#include #include -#include - namespace DB { @@ -22,11 +19,6 @@ struct StorageSnapshot const IStorage & storage; const StorageMetadataPtr metadata; - /// Owns `storage` when nothing else does. Empty for catalog-resident tables. - /// `Alias` over `DataLakeCatalog` fills it: each lookup returns a new storage, - /// and this snapshot is the only object that outlives that lookup. - StoragePtr storage_holder; - /// Additional data, on which set of columns may depend. /// E.g. data parts in MergeTree, list of blocks in Memory, etc. struct Data @@ -48,9 +40,6 @@ struct StorageSnapshot std::shared_ptr clone(DataPtr data_) const; - /// Same `storage` and `storage_holder`, replaced metadata. Drops `data`. - std::shared_ptr cloneWithMetadata(StorageMetadataPtr metadata_) const; - /// Get columns description ColumnsDescription getAllColumnsDescription() const; From 43952dafab9f5b1926d9d81f35fdbe28d635b181 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 14 Oct 2026 11:44:43 +0200 Subject: [PATCH 8/8] Keep the unwrapped storage alive in `parallelReplicasEnabledForStorage`. `getTargetTable` returns a `StoragePtr` that can be the only owner of a catalog storage. Destroying it before `isMergeTree` was a heap-use-after-free under AddressSanitizer. Related: https://github.com/Altinity/ClickHouse/pull/2550 https: //altinity-build-artifacts.s3.amazonaws.com/json.html?PR=2550&sha=bb444f10dd1de59c9ae3adcf33540371cedf6c5b&name_0=PR&name_1=Integration%20tests%20%28amd_asan_ubsan%2C%20db%20disk%2C%20old%20analyzer%2C%208%2F8%29 Co-authored-by: Cursor --- src/Planner/PlannerJoinTree.cpp | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 1633d044ab470..c67eca26efccf 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -991,13 +991,16 @@ std::unique_ptr createComputeAliasColumnsStep( bool parallelReplicasEnabledForStorage(const StoragePtr & current_storage, const ContextPtr & context, const Settings & query_settings) { const auto * table_ptr = current_storage.get(); + /// `getTargetTable` and `getUnderlyingMergeTreeStorageForParallelReplicas` return a `StoragePtr` by value. + /// A catalog lookup can be the only owner of that storage, so it has to stay alive for the checks below. + StoragePtr underlying_storage; if (query_settings[Setting::parallel_replicas_allow_view_over_mergetree]) { const auto * view = typeid_cast(current_storage.get()); if (view) { - auto underlying_storage = view->getUnderlyingMergeTreeStorageForParallelReplicas(context); + underlying_storage = view->getUnderlyingMergeTreeStorageForParallelReplicas(context); if (!underlying_storage) return false; @@ -1015,7 +1018,8 @@ bool parallelReplicasEnabledForStorage(const StoragePtr & current_storage, const if (mv->isRefreshable()) return false; - table_ptr = mv->getTargetTable().get(); + underlying_storage = mv->getTargetTable(); + table_ptr = underlying_storage.get(); } if (!table_ptr->isMergeTree())