From e846234119f8756448ca1de59f7b92977d11bef3 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 28 Oct 2026 15:34:41 +0100 Subject: [PATCH 1/4] Reuse the resolved query tree for `object_storage_cluster` headers Re-analyzing the original AST resolved a `DataLakeCatalog` table again and repeated the REST catalog lookup. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 7 +++- .../integration/test_database_iceberg/test.py | 35 +++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 7e972f2802038..9fa48d316ee39 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -730,7 +730,12 @@ void IStorageCluster::read( if (settings[Setting::allow_experimental_analyzer]) { - sample_block = InterpreterSelectQueryAnalyzer::getSampleBlock(query_to_send, context, SelectQueryOptions(processed_stage)); + /// For an unchanged query, calculate the header from the resolved query tree. Re-analyzing + /// the original AST resolves DataLakeCatalog tables once more and repeats a remote catalog lookup. + if (query_to_send == query_info.query && query_info.query_tree) + sample_block = getHeaderForProcessingStage(column_names, storage_snapshot, query_info, context, processed_stage); + else + sample_block = InterpreterSelectQueryAnalyzer::getSampleBlock(query_to_send, context, SelectQueryOptions(processed_stage)); } else { diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 430f3d593f656..1d8457ab7602d 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -1441,6 +1441,41 @@ def test_cluster_select(started_cluster): assert node2.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`", settings={"parallel_replicas_for_cluster_engines": 1, "enable_parallel_replicas": 2, "cluster_for_parallel_replicas": "cluster_simple"}) == 'pablo\n' +def test_object_storage_cluster_loads_rest_catalog_table_once(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_object_storage_cluster_{uuid.uuid4().hex}" + table_name = f"{test_ref}_table" + namespace = f"{test_ref}_namespace" + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(namespace) + table = create_table(catalog, namespace, table_name) + table.append(pa.Table.from_pylist([generate_record()])) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + query_id = f"object_storage_cluster_catalog_{uuid.uuid4().hex}" + assert ( + node.query( + f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` " + "SETTINGS object_storage_cluster = 'cluster_simple' FORMAT Null", + query_id=query_id, + ) + == "" + ) + + node.query("SYSTEM FLUSH LOGS system.query_log") + assert ( + node.query( + "SELECT ProfileEvents['DataLakeRestCatalogGetTableMetadata'] " + "FROM system.query_log " + f"WHERE query_id = '{query_id}' AND type = 'QueryFinish' AND is_initial_query" + ) + == "1\n" + ) + + def test_on_cluster_ddl_rejected_for_datalake_catalog(started_cluster): node1 = started_cluster.instances["node1"] node2 = started_cluster.instances["node2"] From dbe0f46f39f1bfcf7b604420338d6c7ef3216ee5 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 28 Oct 2026 15:59:31 +0100 Subject: [PATCH 2/4] Limit catalog header reuse to `DataLakeCatalog` tables Keep existing header analysis for all other clustered storage reads. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 2 +- src/Storages/IStorageCluster.h | 2 ++ src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp | 1 + src/Storages/ObjectStorage/StorageObjectStorageCluster.h | 3 +++ 4 files changed, 7 insertions(+), 1 deletion(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 9fa48d316ee39..e5611ba5290b3 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -732,7 +732,7 @@ void IStorageCluster::read( { /// For an unchanged query, calculate the header from the resolved query tree. Re-analyzing /// the original AST resolves DataLakeCatalog tables once more and repeats a remote catalog lookup. - if (query_to_send == query_info.query && query_info.query_tree) + if (canReuseResolvedQueryTreeForHeader() && query_to_send == query_info.query && query_info.query_tree) sample_block = getHeaderForProcessingStage(column_names, storage_snapshot, query_info, context, processed_stage); else sample_block = InterpreterSelectQueryAnalyzer::getSampleBlock(query_to_send, context, SelectQueryOptions(processed_stage)); diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index c592a259e12b0..9479fd19cf1bd 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -81,6 +81,8 @@ class IStorageCluster : public IStorage protected: virtual void updateBeforeRead(const ContextPtr &) {} void updateQueryWithJoinToSendIfNeeded(ASTPtr & query_to_send, SelectQueryInfo query_info, const ContextPtr & context); + /// `DataLakeCatalog` tables need this to avoid resolving the catalog table a second time for the header. + virtual bool canReuseResolvedQueryTreeForHeader() const { return false; } virtual void updateConfigurationIfNeeded(ContextPtr /* context */) {} diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 545122e68f0f4..f686b67d37725 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -126,6 +126,7 @@ StorageObjectStorageCluster::StorageObjectStorageCluster( cluster_name_, table_id_, getLogger(fmt::format("{}({})", configuration_->getEngineName(), table_id_.table_name))) , configuration{configuration_} , object_storage(object_storage_) + , is_datalake_catalog_table(is_datalake_query) { configuration->initPartitionStrategy(partition_by, columns_in_table_or_function_definition, context_); configuration->check(context_); diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 9a4a4fc53122e..cac884a204ce6 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -203,6 +203,7 @@ class StorageObjectStorageCluster : public IStorageCluster /// Whether this query is served by `pure_storage` instead of being distributed over a cluster. bool readsFromPureStorage(ContextPtr context) const; + bool canReuseResolvedQueryTreeForHeader() const override { return is_datalake_catalog_table; } void readFallBackToPure( QueryPlan & query_plan, @@ -243,6 +244,8 @@ class StorageObjectStorageCluster : public IStorageCluster const ObjectStoragePtr object_storage; /// True for explicit *Cluster(...); false for SETTINGS / ENGINE / DataLake paths. bool cluster_name_from_function_argument = false; + /// True only for tables resolved through a `DataLakeCatalog` database. + bool is_datalake_catalog_table = false; /// non-clustered storage to fall back on pure realisation if needed std::shared_ptr pure_storage; From f281aab6b4cbb5c3d632e3c09eabb5f33b6f6f3f Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 29 Oct 2026 11:48:49 +0100 Subject: [PATCH 3/4] Reuse the resolved left-scan tree for `DataLakeCatalog` joins Re-analyzing the rewritten shard AST resolved the catalog table a second time. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 24 ++++++++---- src/Storages/IStorageCluster.h | 9 ++++- .../integration/test_database_iceberg/test.py | 39 +++++++++++++++++++ 3 files changed, 62 insertions(+), 10 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index e5611ba5290b3..8d79f7d1375af 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -441,7 +441,8 @@ or (object_storage_cluster_join_mode='global') void IStorageCluster::updateQueryWithJoinToSendIfNeeded( ASTPtr & query_to_send, SelectQueryInfo query_info, - const ContextPtr & context) + const ContextPtr & context, + QueryTreeNodePtr * resolved_query_tree) { auto object_storage_cluster_join_mode = context->getSettingsRef()[Setting::object_storage_cluster_join_mode]; switch (object_storage_cluster_join_mode) @@ -453,7 +454,7 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( if (!needsInitiatorLocalJoin(info)) return; - rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context); + rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context, resolved_query_tree); return; } case ObjectStorageClusterJoinMode::GLOBAL: @@ -464,7 +465,7 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( if (queryHasRightOrFullJoin(query_to_send)) { auto info = getQueryJoinInfoFromAST(query_to_send); - rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context); + rewriteQueryForInitiatorLocalJoin(query_to_send, query_info, info, context, resolved_query_tree); } return; } @@ -726,13 +727,16 @@ void IStorageCluster::read( SharedHeader sample_block; - updateQueryWithJoinToSendIfNeeded(query_to_send, query_info, context); + /// A join kept on the initiator replaces `query_to_send` but already has a resolved left-scan tree. + QueryTreeNodePtr resolved_query_tree; + updateQueryWithJoinToSendIfNeeded(query_to_send, query_info, context, &resolved_query_tree); if (settings[Setting::allow_experimental_analyzer]) { - /// For an unchanged query, calculate the header from the resolved query tree. Re-analyzing - /// the original AST resolves DataLakeCatalog tables once more and repeats a remote catalog lookup. - if (canReuseResolvedQueryTreeForHeader() && query_to_send == query_info.query && query_info.query_tree) + /// Re-analyzing the shard AST resolves a `DataLakeCatalog` table again and repeats a remote catalog lookup. + if (canReuseResolvedQueryTreeForHeader() && resolved_query_tree) + sample_block = InterpreterSelectQueryAnalyzer::getSampleBlock(resolved_query_tree, context, SelectQueryOptions(processed_stage)); + else if (canReuseResolvedQueryTreeForHeader() && query_to_send == query_info.query && query_info.query_tree) sample_block = getHeaderForProcessingStage(column_names, storage_snapshot, query_info, context, processed_stage); else sample_block = InterpreterSelectQueryAnalyzer::getSampleBlock(query_to_send, context, SelectQueryOptions(processed_stage)); @@ -1065,7 +1069,8 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( ASTPtr & query_to_send, const SelectQueryInfo & query_info, const QueryTreeInfo & info, - const ContextPtr & context) + const ContextPtr & context, + QueryTreeNodePtr * resolved_query_tree) { /// Analyzer path: reuse `extractLeftTableExpression` and `buildQueryToReadColumnsFromTableExpression` /// (same helpers the planner uses when wrapping `IStorageCluster` in a subquery). @@ -1107,6 +1112,9 @@ void IStorageCluster::rewriteQueryForInitiatorLocalJoin( rewritten_query_node.getPrewhere() = query_node.getPrewhere(); rewritten_query_node.getWhere() = query_node.getWhere(); + if (resolved_query_tree) + *resolved_query_tree = rewritten_query_tree; + query_to_send = queryNodeToDistributedSelectQuery(rewritten_query_tree); return; } diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 9479fd19cf1bd..5bbca00713c77 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -80,7 +80,11 @@ class IStorageCluster : public IStorage protected: virtual void updateBeforeRead(const ContextPtr &) {} - void updateQueryWithJoinToSendIfNeeded(ASTPtr & query_to_send, SelectQueryInfo query_info, const ContextPtr & context); + void updateQueryWithJoinToSendIfNeeded( + ASTPtr & query_to_send, + SelectQueryInfo query_info, + const ContextPtr & context, + QueryTreeNodePtr * resolved_query_tree = nullptr); /// `DataLakeCatalog` tables need this to avoid resolving the catalog table a second time for the header. virtual bool canReuseResolvedQueryTreeForHeader() const { return false; } @@ -186,7 +190,8 @@ class IStorageCluster : public IStorage ASTPtr & query_to_send, const SelectQueryInfo & query_info, const QueryTreeInfo & info, - const ContextPtr & context); + const ContextPtr & context, + QueryTreeNodePtr * resolved_query_tree); }; diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 1d8457ab7602d..11d117e67be1f 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -1476,6 +1476,45 @@ def test_object_storage_cluster_loads_rest_catalog_table_once(started_cluster): ) +def test_object_storage_cluster_join_loads_rest_catalog_table_once(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_object_storage_cluster_join_{uuid.uuid4().hex}" + table_name = f"{test_ref}_table" + namespace = f"{test_ref}_namespace" + memory_table = f"{test_ref}_memory" + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(namespace) + table = create_table(catalog, namespace, table_name) + table.append(pa.Table.from_pylist([generate_record()])) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + node.query(f"CREATE TABLE {memory_table} (symbol String) ENGINE = Memory") + node.query(f"INSERT INTO {memory_table} VALUES ('kek')") + + query_id = f"object_storage_cluster_join_catalog_{uuid.uuid4().hex}" + assert ( + node.query( + f"SELECT t.symbol FROM {CATALOG_NAME}.`{namespace}.{table_name}` AS t " + f"INNER JOIN {memory_table} AS m ON t.symbol = m.symbol " + "SETTINGS object_storage_cluster = 'cluster_simple'", + query_id=query_id, + ) + == "kek\n" + ) + + node.query("SYSTEM FLUSH LOGS system.query_log") + assert ( + node.query( + "SELECT ProfileEvents['DataLakeRestCatalogGetTableMetadata'] " + "FROM system.query_log " + f"WHERE query_id = '{query_id}' AND type = 'QueryFinish' AND is_initial_query" + ) + == "1\n" + ) + + def test_on_cluster_ddl_rejected_for_datalake_catalog(started_cluster): node1 = started_cluster.instances["node1"] node2 = started_cluster.instances["node2"] From 6a9a4cbe5f81a1877c67177ca8e4a757b9444850 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 29 Oct 2026 16:07:26 +0100 Subject: [PATCH 4/4] Renumber `DataLakeCatalog` cluster headers to the shard `__table1` alias The resolved tree kept the outer `__tableN` alias, so a non-leftmost catalog table did not match the shard block. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 10 ++++- .../integration/test_database_iceberg/test.py | 39 +++++++++++++++++++ 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 8d79f7d1375af..c8e1633adc790 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -3,6 +3,7 @@ #include #include +#include #include #include @@ -737,7 +738,14 @@ void IStorageCluster::read( if (canReuseResolvedQueryTreeForHeader() && resolved_query_tree) sample_block = InterpreterSelectQueryAnalyzer::getSampleBlock(resolved_query_tree, context, SelectQueryOptions(processed_stage)); else if (canReuseResolvedQueryTreeForHeader() && query_to_send == query_info.query && query_info.query_tree) - sample_block = getHeaderForProcessingStage(column_names, storage_snapshot, query_info, context, processed_stage); + { + /// A shard re-analyzes this query and `createUniqueAliasesIfNecessary` restarts `__tableN` at 1. + /// The resolved tree still has the outer alias (`__table2` when this table is not leftmost). + auto query_info_for_header = query_info; + query_info_for_header.query_tree = query_info.query_tree->clone(); + createUniqueAliasesIfNecessary(query_info_for_header.query_tree, context); + sample_block = getHeaderForProcessingStage(column_names, storage_snapshot, query_info_for_header, context, processed_stage); + } else sample_block = InterpreterSelectQueryAnalyzer::getSampleBlock(query_to_send, context, SelectQueryOptions(processed_stage)); } diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 11d117e67be1f..6f4834209be8f 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -1515,6 +1515,45 @@ def test_object_storage_cluster_join_loads_rest_catalog_table_once(started_clust ) +def test_object_storage_cluster_right_join_loads_rest_catalog_table_once(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_object_storage_cluster_right_join_{uuid.uuid4().hex}" + table_name = f"{test_ref}_table" + namespace = f"{test_ref}_namespace" + memory_table = f"{test_ref}_memory" + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(namespace) + table = create_table(catalog, namespace, table_name) + table.append(pa.Table.from_pylist([generate_record()])) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + node.query(f"CREATE TABLE {memory_table} (symbol String) ENGINE = Memory") + node.query(f"INSERT INTO {memory_table} VALUES ('kek')") + + query_id = f"object_storage_cluster_right_join_catalog_{uuid.uuid4().hex}" + assert ( + node.query( + f"SELECT t.symbol FROM {memory_table} AS m " + f"INNER JOIN {CATALOG_NAME}.`{namespace}.{table_name}` AS t ON m.symbol = t.symbol " + "SETTINGS object_storage_cluster = 'cluster_simple'", + query_id=query_id, + ) + == "kek\n" + ) + + node.query("SYSTEM FLUSH LOGS system.query_log") + assert ( + node.query( + "SELECT ProfileEvents['DataLakeRestCatalogGetTableMetadata'] " + "FROM system.query_log " + f"WHERE query_id = '{query_id}' AND type = 'QueryFinish' AND is_initial_query" + ) + == "1\n" + ) + + def test_on_cluster_ddl_rejected_for_datalake_catalog(started_cluster): node1 = started_cluster.instances["node1"] node2 = started_cluster.instances["node2"]