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()) 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..dc1e4ff5f7c75 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -3088,3 +3088,218 @@ 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_iceberg_catalog_cluster(started_cluster): + """ + 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}" + 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 and leaves its configuration + uninitialized. Each engine below resolves that table again, then lists files. + + `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_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, + } + settings = table_settings if cluster_mode != "query" else { + **table_settings, + "object_storage_cluster": "cluster_simple", + } + # 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) + if cluster_mode == "database": + create_clickhouse_iceberg_database( + started_cluster, + node, + cluster_catalog_name, + {"object_storage_cluster": "cluster_simple"}, + ) + + try: + 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 {distinct}id FROM {source} WHERE id > 0 ORDER BY id", + settings=settings, + ) == "1\n2\n3\n" + finally: + 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}")