From 93e7bdc0dc0f03033ac8d7fa469a8e9c18b7391c Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 7 Jul 2026 18:17:02 +0200 Subject: [PATCH 01/18] Setting 'object_storage_cluster_fallback_if_empty' --- src/Core/Settings.cpp | 3 +++ src/Core/SettingsChangesHistory.cpp | 1 + src/Interpreters/Cluster.cpp | 8 +++++++ src/Interpreters/Cluster.h | 3 +++ src/Storages/IStorageCluster.cpp | 34 +++++++++++++++++++++++++---- src/Storages/IStorageCluster.h | 4 +++- 6 files changed, 48 insertions(+), 5 deletions(-) diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 7a9089880d96..baad3512d7b8 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7766,6 +7766,9 @@ Trigger processor to spill data into external storage adpatively. grace join is )", EXPERIMENTAL) \ DECLARE(String, object_storage_cluster, "", R"( Cluster to make distributed requests to object storages with alternative syntax. +)", EXPERIMENTAL) \ + DECLARE(Bool, object_storage_cluster_fallback_if_empty, false, R"( +Use non-cluster request if 'object_storage_cluster' is set but empty or unknown. )", EXPERIMENTAL) \ DECLARE(UInt64, object_storage_max_nodes, 0, R"( Limit for hosts used for request in object storage cluster table functions - azureBlobStorageCluster, s3Cluster, hdfsCluster, etc. diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 3a72ec667415..027224aeaecd 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -44,6 +44,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"object_storage_cluster_join_mode", "allow", "allow", "New setting"}, {"export_merge_tree_partition_task_timeout_seconds", "3600", "86400", "Increase default value to make it more realistic"}, {"export_merge_tree_part_allow_lossy_cast", false, false, "New setting to gate lossy casts in EXPORT PART/PARTITION behind explicit acknowledgment"}, + {"object_storage_cluster_fallback_if_empty", false, false, "New setting"}, }); addSettingsChanges(settings_changes_history, "26.3", { diff --git a/src/Interpreters/Cluster.cpp b/src/Interpreters/Cluster.cpp index 95f0559748fc..cd735a250e96 100644 --- a/src/Interpreters/Cluster.cpp +++ b/src/Interpreters/Cluster.cpp @@ -909,6 +909,14 @@ Cluster::Cluster(Cluster::SubclusterTag, const Cluster & from, const std::vector initMisc(); } +size_t Cluster::getAllNodeCount() const +{ + size_t count = 0; + for (const auto & shard : shards_info) + count += shard.getAllNodeCount(); + return count; +} + std::vector Cluster::getHostIDs() const { std::vector host_ids; diff --git a/src/Interpreters/Cluster.h b/src/Interpreters/Cluster.h index b20f989da2d2..6e3413c2bdb6 100644 --- a/src/Interpreters/Cluster.h +++ b/src/Interpreters/Cluster.h @@ -264,6 +264,9 @@ class Cluster /// The number of all shards. size_t getShardCount() const { return shards_info.size(); } + /// The number of all nodes. + size_t getAllNodeCount() const; + /// Returns an array of arrays of strings in the format 'escaped_host_name:port' for all replicas of all shards in the cluster. std::vector getHostIDs() const; diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index bcf2efaa3840..a17b07f6dc0c 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -57,6 +57,7 @@ namespace Setting extern const SettingsBool object_storage_remote_initiator; extern const SettingsString object_storage_remote_initiator_cluster; extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; + extern const SettingsBool object_storage_cluster_fallback_if_empty; } namespace ErrorCodes @@ -365,7 +366,22 @@ void IStorageCluster::read( const auto & settings = context->getSettingsRef(); ASTPtr query_to_send = query_info.query; - if (cluster_name_from_settings.empty()) + ClusterPtr cluster = nullptr; + + bool fallback_to_pure = cluster_name_from_settings.empty(); + + if (!fallback_to_pure && settings[Setting::object_storage_cluster_fallback_if_empty]) + { + cluster = getClusterImpl( + context, + cluster_name_from_settings, + isObjectStorage() ? settings[Setting::object_storage_max_nodes] : 0, + /*allow_null*/ true); + if (!cluster) + fallback_to_pure = true; + } + + if (fallback_to_pure) { if (settings[Setting::object_storage_remote_initiator]) { @@ -440,7 +456,8 @@ void IStorageCluster::read( return; } - auto cluster = getClusterImpl(context, cluster_name_from_settings, isObjectStorage() ? settings[Setting::object_storage_max_nodes] : 0); + if (!cluster) + cluster = getClusterImpl(context, cluster_name_from_settings, isObjectStorage() ? settings[Setting::object_storage_max_nodes] : 0); RestoreQualifiedNamesVisitor::Data data; data.distributed_table = DatabaseAndTableWithAlias(*getTableExpression(query_to_send->as(), 0)); @@ -730,9 +747,18 @@ ContextPtr ReadFromCluster::updateSettings(const Settings & settings) return new_context; } -ClusterPtr IStorageCluster::getClusterImpl(ContextPtr context, const String & cluster_name_, size_t max_hosts) +ClusterPtr IStorageCluster::getClusterImpl(ContextPtr context, const String & cluster_name_, size_t max_hosts, bool allow_null) { - return context->getCluster(cluster_name_)->getClusterWithReplicasAsShards(context->getSettingsRef(), /* max_replicas_from_shard */ 0, max_hosts); + ClusterPtr cluster = nullptr; + if (allow_null) + { + cluster = context->tryGetCluster(cluster_name_); + if (!cluster || !cluster->getAllNodeCount()) + return nullptr; + } + else + cluster = context->getCluster(cluster_name_); + return cluster->getClusterWithReplicasAsShards(context->getSettingsRef(), /* max_replicas_from_shard */ 0, max_hosts); } } diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 53ca742a994a..5fc368034646 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -110,7 +110,9 @@ class IStorageCluster : public IStorage NamesAndTypesList hive_partition_columns_to_read_from_file_path; private: - static ClusterPtr getClusterImpl(ContextPtr context, const String & cluster_name_, size_t max_hosts = 0); + // With 'allow_null=true' returns nullptr when cluster does not exist or empty + // With 'allow_null=false' throws exception + static ClusterPtr getClusterImpl(ContextPtr context, const String & cluster_name_, size_t max_hosts = 0, bool allow_null = false); virtual bool isClusterSupported() const { return true; } From 0297ff2a2919c5acc691438642ff2a1bc3f2affb Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 8 Jul 2026 13:15:06 +0200 Subject: [PATCH 02/18] Fix some issues --- src/Storages/IStorageCluster.cpp | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index a17b07f6dc0c..509d0e9e41fd 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -386,9 +386,27 @@ void IStorageCluster::read( if (settings[Setting::object_storage_remote_initiator]) { auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; - if (remote_initiator_cluster_name.empty()) + ClusterPtr remote_initiator_cluster; + if (!remote_initiator_cluster_name.empty()) + { + remote_initiator_cluster = getClusterImpl( + context, + remote_initiator_cluster_name, + /*max_hosts*/ 0, + /*allow_null*/ settings[Setting::object_storage_cluster_fallback_if_empty]); + } + if (remote_initiator_cluster_name.empty() || !remote_initiator_cluster) + { + if (settings[Setting::object_storage_cluster_fallback_if_empty]) + { + readFallBackToPure(query_plan, column_names, storage_snapshot, query_info, context, processed_stage, max_block_size, num_streams); + return; + } + // remote_initiator_cluster can be nullptr only when object_storage_cluster_fallback_if_empty is set + // so this exception is thrown only with empty remote_initiator_cluster_name throw Exception(ErrorCodes::BAD_ARGUMENTS, "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments"); + } /// rewrite query to execute `remote('remote_host', s3(...))` /// remote_host can execute query itself or make on-cluster query depends on own `object_storage_cluster` setting @@ -396,7 +414,6 @@ void IStorageCluster::read( updateQueryWithJoinToSendIfNeeded(query_to_send, query_info, context); updateQueryToSendIfNeeded(query_to_send, storage_snapshot, context, /*make_cluster_function*/ false); - auto remote_initiator_cluster = getClusterImpl(context, remote_initiator_cluster_name); auto storage_and_context = convertToRemote(remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send); auto src_distributed = std::dynamic_pointer_cast(storage_and_context.storage); auto modified_query_info = query_info; @@ -581,6 +598,8 @@ SinkToStoragePtr IStorageCluster::write( { auto cluster_name_from_settings = getClusterName(context); + // Intentionally do not apply object_storage_cluster_fallback_if_empty here. + // Cluster write is not supported; applying fallback would make INSERT depend on cluster availability. if (cluster_name_from_settings.empty()) return writeFallBackToPure(query, metadata_snapshot, context, async_insert); From 8bac60257926ab771158a118d55a404aac6093f8 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 8 Jul 2026 15:43:05 +0200 Subject: [PATCH 03/18] Align cluster fallback planning with read path for object storage. Share cluster resolution via resolveClusterRead so getQueryProcessingStage matches read when object_storage_cluster_fallback_if_empty is enabled, and skip local object_storage_cluster lookup when object_storage_remote_initiator and object_storage_remote_initiator_cluster are both set. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 89 +++++++++++++------ src/Storages/IStorageCluster.h | 12 +++ .../StorageObjectStorageCluster.cpp | 26 ++++-- 3 files changed, 91 insertions(+), 36 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 509d0e9e41fd..ed9b60cdbc35 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -345,6 +345,59 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( } } +IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(ContextPtr context) const +{ + ResolvedClusterRead result; + + if (!isClusterSupported()) + { + result.fallback_to_pure = true; + return result; + } + + auto cluster_name_from_settings = getClusterName(context); + const auto & settings = context->getSettingsRef(); + + /// When both remote-initiator settings are set, object_storage_cluster may be defined only on the remote node. + /// In this case object_storage_cluster must not be resolved locally. + const bool defer_object_storage_cluster_resolution + = settings[Setting::object_storage_remote_initiator] + && !settings[Setting::object_storage_remote_initiator_cluster].value.empty(); + + if (defer_object_storage_cluster_resolution) + result.fallback_to_pure = false; + else + result.fallback_to_pure = cluster_name_from_settings.empty(); + + if (!defer_object_storage_cluster_resolution + && !result.fallback_to_pure + && settings[Setting::object_storage_cluster_fallback_if_empty]) + { + result.object_storage_cluster = getClusterImpl( + context, + cluster_name_from_settings, + isObjectStorage() ? settings[Setting::object_storage_max_nodes] : 0, + /*allow_null*/ true); + if (!result.object_storage_cluster) + result.fallback_to_pure = true; + } + + if (result.fallback_to_pure && settings[Setting::object_storage_remote_initiator]) + { + auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; + if (!remote_initiator_cluster_name.empty()) + { + result.remote_initiator_cluster = getClusterImpl( + context, + remote_initiator_cluster_name, + /*max_hosts*/ 0, + /*allow_null*/ settings[Setting::object_storage_cluster_fallback_if_empty]); + } + } + + return result; +} + /// The code executes on initiator void IStorageCluster::read( QueryPlan & query_plan, @@ -366,55 +419,33 @@ void IStorageCluster::read( const auto & settings = context->getSettingsRef(); ASTPtr query_to_send = query_info.query; - ClusterPtr cluster = nullptr; - - bool fallback_to_pure = cluster_name_from_settings.empty(); - - if (!fallback_to_pure && settings[Setting::object_storage_cluster_fallback_if_empty]) - { - cluster = getClusterImpl( - context, - cluster_name_from_settings, - isObjectStorage() ? settings[Setting::object_storage_max_nodes] : 0, - /*allow_null*/ true); - if (!cluster) - fallback_to_pure = true; - } + auto resolved = resolveClusterRead(context); + ClusterPtr cluster = resolved.object_storage_cluster; - if (fallback_to_pure) + if (resolved.fallback_to_pure) { if (settings[Setting::object_storage_remote_initiator]) { - auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; - ClusterPtr remote_initiator_cluster; - if (!remote_initiator_cluster_name.empty()) - { - remote_initiator_cluster = getClusterImpl( - context, - remote_initiator_cluster_name, - /*max_hosts*/ 0, - /*allow_null*/ settings[Setting::object_storage_cluster_fallback_if_empty]); - } - if (remote_initiator_cluster_name.empty() || !remote_initiator_cluster) + if (!resolved.remote_initiator_cluster) { if (settings[Setting::object_storage_cluster_fallback_if_empty]) { readFallBackToPure(query_plan, column_names, storage_snapshot, query_info, context, processed_stage, max_block_size, num_streams); return; } - // remote_initiator_cluster can be nullptr only when object_storage_cluster_fallback_if_empty is set - // so this exception is thrown only with empty remote_initiator_cluster_name throw Exception(ErrorCodes::BAD_ARGUMENTS, "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments"); } + auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; + /// rewrite query to execute `remote('remote_host', s3(...))` /// remote_host can execute query itself or make on-cluster query depends on own `object_storage_cluster` setting updateConfigurationIfNeeded(context); updateQueryWithJoinToSendIfNeeded(query_to_send, query_info, context); updateQueryToSendIfNeeded(query_to_send, storage_snapshot, context, /*make_cluster_function*/ false); - auto storage_and_context = convertToRemote(remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send); + auto storage_and_context = convertToRemote(resolved.remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send); auto src_distributed = std::dynamic_pointer_cast(storage_and_context.storage); auto modified_query_info = query_info; modified_query_info.cluster = src_distributed->getCluster(); diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 5fc368034646..9188c543efdd 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -109,6 +109,18 @@ class IStorageCluster : public IStorage NamesAndTypesList hive_partition_columns_to_read_from_file_path; + struct ResolvedClusterRead + { + /// True when read() should use readFallBackToPure() or remote-initiator fallback branch. + bool fallback_to_pure = false; + /// Pre-resolved object-storage cluster when object_storage_cluster_fallback_if_empty prefetch was done. + ClusterPtr object_storage_cluster; + /// Resolved remote-initiator cluster when object_storage_remote_initiator is enabled in fallback branch. + ClusterPtr remote_initiator_cluster; + }; + + ResolvedClusterRead resolveClusterRead(ContextPtr context) const; + private: // With 'allow_null=true' returns nullptr when cluster does not exist or empty // With 'allow_null=false' throws exception diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 17704602a57f..6f3be429351d 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -42,6 +42,7 @@ namespace Setting extern const SettingsInt64 delta_lake_snapshot_end_version; extern const SettingsUInt64 lock_object_storage_task_distribution_ms; extern const SettingsBool allow_experimental_iceberg_read_optimization; + extern const SettingsBool object_storage_cluster_fallback_if_empty; } namespace ErrorCodes @@ -725,15 +726,26 @@ QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( if (!isClusterSupported()) return QueryProcessingStage::Enum::FetchColumns; - /// Full query if fall back to pure storage. - if (getClusterName(context).empty() // Not cluster request - && context->getSettingsRef()[Setting::object_storage_remote_initiator_cluster].value.empty()) // Not request with remote initiator + auto resolved = resolveClusterRead(context); + const auto & settings = context->getSettingsRef(); + + if (resolved.fallback_to_pure) { - if (context->getSettingsRef()[Setting::object_storage_remote_initiator]) - throw Exception(ErrorCodes::BAD_ARGUMENTS, - "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments"); + if (settings[Setting::object_storage_remote_initiator]) + { + if (!resolved.remote_initiator_cluster) + { + if (!settings[Setting::object_storage_cluster_fallback_if_empty]) + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments"); - return QueryProcessingStage::Enum::FetchColumns; + return QueryProcessingStage::Enum::FetchColumns; + } + } + else + { + return QueryProcessingStage::Enum::FetchColumns; + } } /// Distributed storage. From bc7aacf2e4a40f6847f694db2059aa19309da639 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 09:46:06 +0200 Subject: [PATCH 04/18] Add tests for object_storage_cluster_fallback_if_empty. Cover pure fallback on unknown cluster, aggregate planning, remote-initiator interaction, and integration scenarios with locally unknown object_storage_cluster. Co-authored-by: Cursor --- tests/integration/test_s3_cluster/test.py | 74 +++++++++++++++++++ ...torage_cluster_fallback_if_empty.reference | 13 ++++ ...ject_storage_cluster_fallback_if_empty.sql | 37 ++++++++++ 3 files changed, 124 insertions(+) create mode 100644 tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference create mode 100644 tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index d4d91b923782..c332cc71c1e7 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -288,6 +288,80 @@ def test_wrong_cluster(started_cluster): assert "not found" in error +def test_object_storage_cluster_fallback_if_empty(started_cluster): + node = started_cluster.instances["s0_0_0"] + + pure_s3 = node.query( + f""" + SELECT count(*) from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', + 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))')""" + ) + + fallback_s3 = node.query( + f""" + SELECT count(*) from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', + 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') + SETTINGS object_storage_cluster = 'non_existing_cluster', object_storage_cluster_fallback_if_empty = 1 + """ + ) + + assert TSV(pure_s3) == TSV(fallback_s3) + + pure_sum = node.query( + f""" + SELECT sum(value) from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', + 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))')""" + ) + + fallback_sum = node.query( + f""" + SELECT sum(value) from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', + 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') + SETTINGS object_storage_cluster = 'non_existing_cluster', object_storage_cluster_fallback_if_empty = 1 + """ + ) + + assert TSV(pure_sum) == TSV(fallback_sum) + + query_id = uuid.uuid4().hex + result = node.query( + f""" + SELECT * from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') ORDER BY (name, value, polygon) + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster='hidden_cluster_with_username_and_password', + object_storage_remote_initiator_cluster='cluster_with_dots', + object_storage_cluster_fallback_if_empty=1 + """, + query_id=query_id, + ) + + assert result is not None + + node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster_all'") + queries = node.query( + f""" + SELECT count() + FROM clusterAllReplicas('cluster_all', system.query_log) + WHERE type='QueryFinish' AND initial_query_id='{query_id}' + FORMAT TSV + """ + ).splitlines() + + # initial node + remote initiator + 2 subqueries on replicas + assert queries == ["4"] + + def test_ambiguous_join(started_cluster): node = started_cluster.instances["s0_0_0"] result = node.query( diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference new file mode 100644 index 000000000000..388ccd963c32 --- /dev/null +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference @@ -0,0 +1,13 @@ +pure +10 45 +unknown cluster without fallback +unknown cluster with fallback +10 45 +valid cluster with fallback +10 45 +remote initiator unresolved with fallback +10 45 +aggregate with fallback +285 +pure aggregate +285 diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql new file mode 100644 index 000000000000..0b31408dc06c --- /dev/null +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql @@ -0,0 +1,37 @@ +-- Tags: no-fasttest +-- Tag no-fasttest: Depends on Minio + +SET enable_analyzer = 1; + +INSERT INTO FUNCTION s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SELECT number FROM numbers(10) +SETTINGS s3_truncate_on_insert = 1; + +SELECT 'pure'; +SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32'); + +SELECT 'unknown cluster without fallback'; +SELECT count() FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS object_storage_cluster = 'non_existent_cluster_04303'; -- { serverError CLUSTER_DOESNT_EXIST } + +SELECT 'unknown cluster with fallback'; +SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_if_empty = 1; + +SELECT 'valid cluster with fallback'; +SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS object_storage_cluster = 'test_shard_localhost', object_storage_cluster_fallback_if_empty = 1; + +SELECT 'remote initiator unresolved with fallback'; +SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS + object_storage_cluster = 'non_existent_cluster_04303', + object_storage_cluster_fallback_if_empty = 1, + object_storage_remote_initiator = 1; + +SELECT 'aggregate with fallback'; +SELECT sum(x * x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_if_empty = 1; + +SELECT 'pure aggregate'; +SELECT sum(x * x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32'); From 1ad4ce96522b64c1b5891be45e68b7c562dbf43a Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 13:17:40 +0200 Subject: [PATCH 05/18] Extend fallback tests for remote initiator with unknown cluster. Cover stateless and integration cases where object_storage_cluster is missing locally and remote initiator falls back to non-cluster execution. Co-authored-by: Cursor --- tests/integration/test_s3_cluster/test.py | 70 +++++++++++++++++++ ...torage_cluster_fallback_if_empty.reference | 2 + ...ject_storage_cluster_fallback_if_empty.sql | 8 +++ 3 files changed, 80 insertions(+) diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index c332cc71c1e7..bec89362ca78 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -361,6 +361,76 @@ def test_object_storage_cluster_fallback_if_empty(started_cluster): # initial node + remote initiator + 2 subqueries on replicas assert queries == ["4"] + pure_count = node.query( + f""" + SELECT count(*) from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', + 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))')""" + ) + + query_id = uuid.uuid4().hex + remote_initiator_count = node.query( + f""" + SELECT count(*) from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', + 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster='non_existing_cluster', + object_storage_remote_initiator_cluster='cluster_with_dots_and_user', + object_storage_cluster_fallback_if_empty=1 + """, + query_id=query_id, + ) + + assert TSV(pure_count) == TSV(remote_initiator_count) + + node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster_all'") + queries = node.query( + f""" + SELECT count() + FROM clusterAllReplicas('cluster_all', system.query_log) + WHERE type='QueryFinish' AND initial_query_id='{query_id}' + FORMAT TSV + """ + ).splitlines() + + # initial node + remote initiator. + # object_storage_cluster is not exist on remote initiator, so it will not be able to run subqueries on replicas. + assert queries == ["2"] + + query_id = uuid.uuid4().hex + result = node.query( + f""" + SELECT * from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') ORDER BY (name, value, polygon) + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster='non_existing_cluster', + object_storage_remote_initiator_cluster='cluster_with_dots', + object_storage_cluster_fallback_if_empty=1 + """, + query_id=query_id, + ) + + assert result is not None + + node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster_all'") + queries = node.query( + f""" + SELECT count() + FROM clusterAllReplicas('cluster_all', system.query_log) + WHERE type='QueryFinish' AND initial_query_id='{query_id}' + FORMAT TSV + """ + ).splitlines() + + # initial node + remote initiator. + assert queries == ["2"] + def test_ambiguous_join(started_cluster): node = started_cluster.instances["s0_0_0"] diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference index 388ccd963c32..e84971718ae6 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference @@ -7,6 +7,8 @@ valid cluster with fallback 10 45 remote initiator unresolved with fallback 10 45 +remote initiator with non-existent cluster +10 45 aggregate with fallback 285 pure aggregate diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql index 0b31408dc06c..422cfbeef8a7 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql @@ -29,6 +29,14 @@ SETTINGS object_storage_cluster_fallback_if_empty = 1, object_storage_remote_initiator = 1; +SELECT 'remote initiator with non-existent cluster'; +SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS + object_storage_cluster = 'non_existent_cluster_04303', + object_storage_cluster_fallback_if_empty = 1, + object_storage_remote_initiator = 1, + object_storage_remote_initiator_cluster = 'test_shard_localhost'; + SELECT 'aggregate with fallback'; SELECT sum(x * x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_if_empty = 1; From 195f687769583e017d3e024dc077a1e30a816a2d Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 16:09:43 +0200 Subject: [PATCH 06/18] Do not apply object_storage_cluster_fallback_if_empty to explicit *Cluster functions. Distinguish s3() with object_storage_cluster setting from s3Cluster() argument so fallback works for the former but explicit cluster names still fail when unknown. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 1 + src/Storages/IStorageCluster.h | 3 +++ .../StorageObjectStorageCluster.cpp | 25 ++++++++++++++----- .../StorageObjectStorageCluster.h | 2 ++ ...torage_cluster_fallback_if_empty.reference | 4 +++ ...ject_storage_cluster_fallback_if_empty.sql | 12 +++++++++ 6 files changed, 41 insertions(+), 6 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index ed9b60cdbc35..e4f095b8bb49 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -370,6 +370,7 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context result.fallback_to_pure = cluster_name_from_settings.empty(); if (!defer_object_storage_cluster_resolution + && useObjectStorageClusterFallbackIfEmpty(context) && !result.fallback_to_pure && settings[Setting::object_storage_cluster_fallback_if_empty]) { diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 9188c543efdd..6da119eee894 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -121,6 +121,9 @@ class IStorageCluster : public IStorage ResolvedClusterRead resolveClusterRead(ContextPtr context) const; + /// Apply object_storage_cluster_fallback_if_empty only for storages that take cluster name from the setting. + virtual bool useObjectStorageClusterFallbackIfEmpty(ContextPtr /* context */) const { return false; } + private: // With 'allow_null=true' returns nullptr when cluster does not exist or empty // With 'allow_null=false' throws exception diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 6f3be429351d..6c00db88ff52 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -707,17 +707,30 @@ SinkToStoragePtr StorageObjectStorageCluster::writeFallBackToPure( String StorageObjectStorageCluster::getClusterName(ContextPtr context) const { /// StorageObjectStorageCluster is always created for cluster or non-cluster variants. - /// User can specify cluster name in table definition or in setting `object_storage_cluster` - /// only for several queries. When it specified in both places, priority is given to the query setting. - /// When it is empty, non-cluster realization is used. + /// User can specify cluster name in table definition, in *Cluster table function argument, + /// or in setting `object_storage_cluster` for s3()/iceberg() alternative syntax. + /// Explicit *Cluster argument has priority over query setting; alternative-syntax requests use the setting. if (!isClusterSupported()) return ""; + if (!cluster_name_in_settings && !getOriginalClusterName().empty()) + return getOriginalClusterName(); + auto cluster_name_from_settings = context->getSettingsRef()[Setting::object_storage_cluster].value; - if (cluster_name_from_settings.empty()) - cluster_name_from_settings = getOriginalClusterName(); - return cluster_name_from_settings; + if (!cluster_name_from_settings.empty()) + return cluster_name_from_settings; + + return getOriginalClusterName(); +} + +bool StorageObjectStorageCluster::useObjectStorageClusterFallbackIfEmpty(ContextPtr context) const +{ + if (!cluster_name_in_settings && !getOriginalClusterName().empty()) + return false; + + return cluster_name_in_settings + || !context->getSettingsRef()[Setting::object_storage_cluster].value.empty(); } QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index bcaf293ad4e7..0192b3f46001 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -196,6 +196,8 @@ class StorageObjectStorageCluster : public IStorageCluster ContextPtr context, bool async_insert) override; + bool useObjectStorageClusterFallbackIfEmpty(ContextPtr context) const override; + /* In case the table was created with `object_storage_cluster` setting, modify the AST query object so that it uses the table function implementation diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference index e84971718ae6..bc3d0c602799 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference @@ -5,6 +5,10 @@ unknown cluster with fallback 10 45 valid cluster with fallback 10 45 +explicit cluster function with fallback +explicit cluster function overrides setting +explicit cluster function overrides setting with valid cluster +10 45 remote initiator unresolved with fallback 10 45 remote initiator with non-existent cluster diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql index 422cfbeef8a7..cd3488ecb576 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql @@ -22,6 +22,18 @@ SELECT 'valid cluster with fallback'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'test_shard_localhost', object_storage_cluster_fallback_if_empty = 1; +SELECT 'explicit cluster function with fallback'; +SELECT count() FROM s3Cluster('non_existent_cluster_04303', 'http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS object_storage_cluster_fallback_if_empty = 1; -- { serverError CLUSTER_DOESNT_EXIST } + +SELECT 'explicit cluster function overrides setting'; +SELECT count() FROM s3Cluster('non_existent_cluster_04303', 'http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS object_storage_cluster_fallback_if_empty = 1, object_storage_cluster = 'non_existent_cluster_04303_2'; -- { serverError CLUSTER_DOESNT_EXIST } + +SELECT 'explicit cluster function overrides setting with valid cluster'; +SELECT count(), sum(x) FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_if_empty = 1; + SELECT 'remote initiator unresolved with fallback'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS From 92a51421bdd6681b141aada2fb3567677ce4c720 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 19:53:20 +0200 Subject: [PATCH 07/18] Send pure s3() to remote initiator without converting to cluster first. Distinguish alternative-syntax table functions from table reads so remote initiator keeps s3() for fallback and uses cluster path for Iceberg tables. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 13 +++++++++---- src/Storages/IStorageCluster.h | 3 +++ .../ObjectStorage/StorageObjectStorageCluster.cpp | 5 ++++- .../ObjectStorage/StorageObjectStorageCluster.h | 2 ++ 4 files changed, 18 insertions(+), 5 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index e4f095b8bb49..dc1f4c0e2f27 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -365,7 +365,7 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context && !settings[Setting::object_storage_remote_initiator_cluster].value.empty(); if (defer_object_storage_cluster_resolution) - result.fallback_to_pure = false; + result.fallback_to_pure = usePureFunctionForRemoteInitiator(context); else result.fallback_to_pure = cluster_name_from_settings.empty(); @@ -383,16 +383,18 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context result.fallback_to_pure = true; } - if (result.fallback_to_pure && settings[Setting::object_storage_remote_initiator]) + if (settings[Setting::object_storage_remote_initiator]) { auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; if (!remote_initiator_cluster_name.empty()) { + const bool allow_null = settings[Setting::object_storage_cluster_fallback_if_empty] + && (result.fallback_to_pure || usePureFunctionForRemoteInitiator(context)); result.remote_initiator_cluster = getClusterImpl( context, remote_initiator_cluster_name, /*max_hosts*/ 0, - /*allow_null*/ settings[Setting::object_storage_cluster_fallback_if_empty]); + allow_null); } } @@ -423,7 +425,10 @@ void IStorageCluster::read( auto resolved = resolveClusterRead(context); ClusterPtr cluster = resolved.object_storage_cluster; - if (resolved.fallback_to_pure) + const bool send_pure_function_to_remote_initiator + = settings[Setting::object_storage_remote_initiator] && usePureFunctionForRemoteInitiator(context); + + if (resolved.fallback_to_pure || send_pure_function_to_remote_initiator) { if (settings[Setting::object_storage_remote_initiator]) { diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 6da119eee894..bb42227dd2f0 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -124,6 +124,9 @@ class IStorageCluster : public IStorage /// Apply object_storage_cluster_fallback_if_empty only for storages that take cluster name from the setting. virtual bool useObjectStorageClusterFallbackIfEmpty(ContextPtr /* context */) const { return false; } + /// True for s3()/iceberg() alternative syntax (cluster name from object_storage_cluster setting, not *Cluster argument). + virtual bool usePureFunctionForRemoteInitiator(ContextPtr /* context */) const { return false; } + private: // With 'allow_null=true' returns nullptr when cluster does not exist or empty // With 'allow_null=false' throws exception diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 6c00db88ff52..cc82b331034e 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -742,7 +742,10 @@ QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( auto resolved = resolveClusterRead(context); const auto & settings = context->getSettingsRef(); - if (resolved.fallback_to_pure) + const bool send_pure_function_to_remote_initiator + = settings[Setting::object_storage_remote_initiator] && usePureFunctionForRemoteInitiator(context); + + if (resolved.fallback_to_pure || send_pure_function_to_remote_initiator) { if (settings[Setting::object_storage_remote_initiator]) { diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 0192b3f46001..165ad89b48a4 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -198,6 +198,8 @@ class StorageObjectStorageCluster : public IStorageCluster bool useObjectStorageClusterFallbackIfEmpty(ContextPtr context) const override; + bool usePureFunctionForRemoteInitiator(ContextPtr /* context */) const override { return cluster_name_in_settings; } + /* In case the table was created with `object_storage_cluster` setting, modify the AST query object so that it uses the table function implementation From 13eaae5c8fd7b9d5c2a3dedbef6dfb6873dcb645 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 22:19:39 +0200 Subject: [PATCH 08/18] Allow object_storage_cluster_fallback_if_empty for tables with engine cluster setting. Track explicit *Cluster function arguments separately from table engine and query object_storage_cluster settings so fallback works for persistent tables. Co-authored-by: Cursor --- src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp | 7 ++++--- src/Storages/ObjectStorage/StorageObjectStorageCluster.h | 6 ++++++ src/TableFunctions/TableFunctionObjectStorageCluster.cpp | 1 + .../TableFunctionObjectStorageClusterFallback.cpp | 3 +++ 4 files changed, 14 insertions(+), 3 deletions(-) diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index cc82b331034e..2085cdb7a5f4 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -709,12 +709,12 @@ String StorageObjectStorageCluster::getClusterName(ContextPtr context) const /// StorageObjectStorageCluster is always created for cluster or non-cluster variants. /// User can specify cluster name in table definition, in *Cluster table function argument, /// or in setting `object_storage_cluster` for s3()/iceberg() alternative syntax. - /// Explicit *Cluster argument has priority over query setting; alternative-syntax requests use the setting. + /// Explicit *Cluster argument has priority over query setting; table engine and alternative-syntax use the setting path. if (!isClusterSupported()) return ""; - if (!cluster_name_in_settings && !getOriginalClusterName().empty()) + if (cluster_name_from_function_argument) return getOriginalClusterName(); auto cluster_name_from_settings = context->getSettingsRef()[Setting::object_storage_cluster].value; @@ -726,10 +726,11 @@ String StorageObjectStorageCluster::getClusterName(ContextPtr context) const bool StorageObjectStorageCluster::useObjectStorageClusterFallbackIfEmpty(ContextPtr context) const { - if (!cluster_name_in_settings && !getOriginalClusterName().empty()) + if (cluster_name_from_function_argument) return false; return cluster_name_in_settings + || !getOriginalClusterName().empty() || !context->getSettingsRef()[Setting::object_storage_cluster].value.empty(); } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 165ad89b48a4..1d09bb33a8c4 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -64,6 +64,11 @@ class StorageObjectStorageCluster : public IStorageCluster std::optional totalBytes(ContextPtr query_context) const override; void setClusterNameInSettings(bool cluster_name_in_settings_) { cluster_name_in_settings = cluster_name_in_settings_; } + void setClusterNameFromFunctionArgument(bool cluster_name_from_function_argument_) + { + cluster_name_from_function_argument = cluster_name_from_function_argument_; + } + String getClusterName(ContextPtr context) const override; QueryProcessingStage::Enum getQueryProcessingStage(ContextPtr, QueryProcessingStage::Enum, const StorageSnapshotPtr &, SelectQueryInfo &) const override; @@ -219,6 +224,7 @@ class StorageObjectStorageCluster : public IStorageCluster StorageObjectStorageConfigurationPtr configuration; const ObjectStoragePtr object_storage; bool cluster_name_in_settings; + bool cluster_name_from_function_argument = false; /// non-clustered storage to fall back on pure realisation if needed std::shared_ptr pure_storage; diff --git a/src/TableFunctions/TableFunctionObjectStorageCluster.cpp b/src/TableFunctions/TableFunctionObjectStorageCluster.cpp index 4f813e516f9a..21f217dc3c9b 100644 --- a/src/TableFunctions/TableFunctionObjectStorageCluster.cpp +++ b/src/TableFunctions/TableFunctionObjectStorageCluster.cpp @@ -90,6 +90,7 @@ StoragePtr TableFunctionObjectStorageCluster(storage)->setClusterNameFromFunctionArgument(true); } storage->startup(); diff --git a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp index 98d8dc43c85a..148cef62aff6 100644 --- a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp +++ b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp @@ -154,7 +154,10 @@ StoragePtr TableFunctionObjectStorageClusterFallback::executeI { auto result = BaseCluster::executeImpl(ast_function, context, table_name, cached_columns, is_insert_query); if (auto storage = typeid_cast>(result)) + { storage->setClusterNameInSettings(true); + storage->setClusterNameFromFunctionArgument(false); + } return result; } else From 26cb88aed1cb914cd9fa8bba4da7307f946757f5 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 19:10:02 +0200 Subject: [PATCH 09/18] Fix after review --- src/Core/Settings.cpp | 2 +- src/Core/SettingsChangesHistory.cpp | 2 +- src/Storages/IStorageCluster.cpp | 11 +++++------ src/Storages/IStorageCluster.h | 4 ++-- .../StorageObjectStorageCluster.cpp | 7 +++++-- tests/integration/test_s3_cluster/test.py | 12 ++++++------ ..._object_storage_cluster_fallback_if_empty.sql | 16 ++++++++-------- 7 files changed, 28 insertions(+), 26 deletions(-) diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index baad3512d7b8..5b3424d347a4 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7767,7 +7767,7 @@ Trigger processor to spill data into external storage adpatively. grace join is DECLARE(String, object_storage_cluster, "", R"( Cluster to make distributed requests to object storages with alternative syntax. )", EXPERIMENTAL) \ - DECLARE(Bool, object_storage_cluster_fallback_if_empty, false, R"( + DECLARE(Bool, object_storage_cluster_fallback_to_local_if_empty, false, R"( Use non-cluster request if 'object_storage_cluster' is set but empty or unknown. )", EXPERIMENTAL) \ DECLARE(UInt64, object_storage_max_nodes, 0, R"( diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 027224aeaecd..a6265ebb403b 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -44,7 +44,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"object_storage_cluster_join_mode", "allow", "allow", "New setting"}, {"export_merge_tree_partition_task_timeout_seconds", "3600", "86400", "Increase default value to make it more realistic"}, {"export_merge_tree_part_allow_lossy_cast", false, false, "New setting to gate lossy casts in EXPORT PART/PARTITION behind explicit acknowledgment"}, - {"object_storage_cluster_fallback_if_empty", false, false, "New setting"}, + {"object_storage_cluster_fallback_to_local_if_empty", false, false, "New setting"}, }); addSettingsChanges(settings_changes_history, "26.3", { diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index dc1f4c0e2f27..bdf828751f92 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -57,7 +57,7 @@ namespace Setting extern const SettingsBool object_storage_remote_initiator; extern const SettingsString object_storage_remote_initiator_cluster; extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; - extern const SettingsBool object_storage_cluster_fallback_if_empty; + extern const SettingsBool object_storage_cluster_fallback_to_local_if_empty; } namespace ErrorCodes @@ -370,9 +370,8 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context result.fallback_to_pure = cluster_name_from_settings.empty(); if (!defer_object_storage_cluster_resolution - && useObjectStorageClusterFallbackIfEmpty(context) && !result.fallback_to_pure - && settings[Setting::object_storage_cluster_fallback_if_empty]) + && useObjectStorageClusterFallbackIfEmpty(context)) { result.object_storage_cluster = getClusterImpl( context, @@ -388,7 +387,7 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; if (!remote_initiator_cluster_name.empty()) { - const bool allow_null = settings[Setting::object_storage_cluster_fallback_if_empty] + const bool allow_null = settings[Setting::object_storage_cluster_fallback_to_local_if_empty] && (result.fallback_to_pure || usePureFunctionForRemoteInitiator(context)); result.remote_initiator_cluster = getClusterImpl( context, @@ -434,7 +433,7 @@ void IStorageCluster::read( { if (!resolved.remote_initiator_cluster) { - if (settings[Setting::object_storage_cluster_fallback_if_empty]) + if (settings[Setting::object_storage_cluster_fallback_to_local_if_empty]) { readFallBackToPure(query_plan, column_names, storage_snapshot, query_info, context, processed_stage, max_block_size, num_streams); return; @@ -635,7 +634,7 @@ SinkToStoragePtr IStorageCluster::write( { auto cluster_name_from_settings = getClusterName(context); - // Intentionally do not apply object_storage_cluster_fallback_if_empty here. + // Intentionally do not apply object_storage_cluster_fallback_to_local_if_empty here. // Cluster write is not supported; applying fallback would make INSERT depend on cluster availability. if (cluster_name_from_settings.empty()) return writeFallBackToPure(query, metadata_snapshot, context, async_insert); diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index bb42227dd2f0..e9fad0fd273f 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -113,7 +113,7 @@ class IStorageCluster : public IStorage { /// True when read() should use readFallBackToPure() or remote-initiator fallback branch. bool fallback_to_pure = false; - /// Pre-resolved object-storage cluster when object_storage_cluster_fallback_if_empty prefetch was done. + /// Pre-resolved object-storage cluster when object_storage_cluster_fallback_to_local_if_empty prefetch was done. ClusterPtr object_storage_cluster; /// Resolved remote-initiator cluster when object_storage_remote_initiator is enabled in fallback branch. ClusterPtr remote_initiator_cluster; @@ -121,7 +121,7 @@ class IStorageCluster : public IStorage ResolvedClusterRead resolveClusterRead(ContextPtr context) const; - /// Apply object_storage_cluster_fallback_if_empty only for storages that take cluster name from the setting. + /// Apply object_storage_cluster_fallback_to_local_if_empty only for storages that take cluster name from the setting. virtual bool useObjectStorageClusterFallbackIfEmpty(ContextPtr /* context */) const { return false; } /// True for s3()/iceberg() alternative syntax (cluster name from object_storage_cluster setting, not *Cluster argument). diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 2085cdb7a5f4..e645c7edf256 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -42,7 +42,7 @@ namespace Setting extern const SettingsInt64 delta_lake_snapshot_end_version; extern const SettingsUInt64 lock_object_storage_task_distribution_ms; extern const SettingsBool allow_experimental_iceberg_read_optimization; - extern const SettingsBool object_storage_cluster_fallback_if_empty; + extern const SettingsBool object_storage_cluster_fallback_to_local_if_empty; } namespace ErrorCodes @@ -726,6 +726,9 @@ String StorageObjectStorageCluster::getClusterName(ContextPtr context) const bool StorageObjectStorageCluster::useObjectStorageClusterFallbackIfEmpty(ContextPtr context) const { + if (!context->getSettingsRef()[Setting::object_storage_cluster_fallback_to_local_if_empty]) + return false; + if (cluster_name_from_function_argument) return false; @@ -752,7 +755,7 @@ QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( { if (!resolved.remote_initiator_cluster) { - if (!settings[Setting::object_storage_cluster_fallback_if_empty]) + if (!settings[Setting::object_storage_cluster_fallback_to_local_if_empty]) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments"); diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index bec89362ca78..ec4190209af9 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -288,7 +288,7 @@ def test_wrong_cluster(started_cluster): assert "not found" in error -def test_object_storage_cluster_fallback_if_empty(started_cluster): +def test_object_storage_cluster_fallback_to_local_if_empty(started_cluster): node = started_cluster.instances["s0_0_0"] pure_s3 = node.query( @@ -305,7 +305,7 @@ def test_object_storage_cluster_fallback_if_empty(started_cluster): 'http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV', 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') - SETTINGS object_storage_cluster = 'non_existing_cluster', object_storage_cluster_fallback_if_empty = 1 + SETTINGS object_storage_cluster = 'non_existing_cluster', object_storage_cluster_fallback_to_local_if_empty = 1 """ ) @@ -325,7 +325,7 @@ def test_object_storage_cluster_fallback_if_empty(started_cluster): 'http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV', 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') - SETTINGS object_storage_cluster = 'non_existing_cluster', object_storage_cluster_fallback_if_empty = 1 + SETTINGS object_storage_cluster = 'non_existing_cluster', object_storage_cluster_fallback_to_local_if_empty = 1 """ ) @@ -341,7 +341,7 @@ def test_object_storage_cluster_fallback_if_empty(started_cluster): object_storage_remote_initiator=1, object_storage_cluster='hidden_cluster_with_username_and_password', object_storage_remote_initiator_cluster='cluster_with_dots', - object_storage_cluster_fallback_if_empty=1 + object_storage_cluster_fallback_to_local_if_empty=1 """, query_id=query_id, ) @@ -380,7 +380,7 @@ def test_object_storage_cluster_fallback_if_empty(started_cluster): object_storage_remote_initiator=1, object_storage_cluster='non_existing_cluster', object_storage_remote_initiator_cluster='cluster_with_dots_and_user', - object_storage_cluster_fallback_if_empty=1 + object_storage_cluster_fallback_to_local_if_empty=1 """, query_id=query_id, ) @@ -411,7 +411,7 @@ def test_object_storage_cluster_fallback_if_empty(started_cluster): object_storage_remote_initiator=1, object_storage_cluster='non_existing_cluster', object_storage_remote_initiator_cluster='cluster_with_dots', - object_storage_cluster_fallback_if_empty=1 + object_storage_cluster_fallback_to_local_if_empty=1 """, query_id=query_id, ) diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql index cd3488ecb576..8f909c432599 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql @@ -16,42 +16,42 @@ SETTINGS object_storage_cluster = 'non_existent_cluster_04303'; -- { serverError SELECT 'unknown cluster with fallback'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') -SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_if_empty = 1; +SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_to_local_if_empty = 1; SELECT 'valid cluster with fallback'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') -SETTINGS object_storage_cluster = 'test_shard_localhost', object_storage_cluster_fallback_if_empty = 1; +SETTINGS object_storage_cluster = 'test_shard_localhost', object_storage_cluster_fallback_to_local_if_empty = 1; SELECT 'explicit cluster function with fallback'; SELECT count() FROM s3Cluster('non_existent_cluster_04303', 'http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') -SETTINGS object_storage_cluster_fallback_if_empty = 1; -- { serverError CLUSTER_DOESNT_EXIST } +SETTINGS object_storage_cluster_fallback_to_local_if_empty = 1; -- { serverError CLUSTER_DOESNT_EXIST } SELECT 'explicit cluster function overrides setting'; SELECT count() FROM s3Cluster('non_existent_cluster_04303', 'http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') -SETTINGS object_storage_cluster_fallback_if_empty = 1, object_storage_cluster = 'non_existent_cluster_04303_2'; -- { serverError CLUSTER_DOESNT_EXIST } +SETTINGS object_storage_cluster_fallback_to_local_if_empty = 1, object_storage_cluster = 'non_existent_cluster_04303_2'; -- { serverError CLUSTER_DOESNT_EXIST } SELECT 'explicit cluster function overrides setting with valid cluster'; SELECT count(), sum(x) FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') -SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_if_empty = 1; +SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_to_local_if_empty = 1; SELECT 'remote initiator unresolved with fallback'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303', - object_storage_cluster_fallback_if_empty = 1, + object_storage_cluster_fallback_to_local_if_empty = 1, object_storage_remote_initiator = 1; SELECT 'remote initiator with non-existent cluster'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303', - object_storage_cluster_fallback_if_empty = 1, + object_storage_cluster_fallback_to_local_if_empty = 1, object_storage_remote_initiator = 1, object_storage_remote_initiator_cluster = 'test_shard_localhost'; SELECT 'aggregate with fallback'; SELECT sum(x * x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') -SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_if_empty = 1; +SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_to_local_if_empty = 1; SELECT 'pure aggregate'; SELECT sum(x * x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32'); From 988e463d75952197a11a47b139f37b035b4c15f1 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 19:39:50 +0200 Subject: [PATCH 10/18] Simplify object_storage_cluster_fallback_to_local_if_empty control flow. Split storage policy from the setting check and route all local-fallback decisions through one helper so the resolve/read path is easier to follow. Co-authored-by: Cursor --- src/Core/Settings.cpp | 3 ++- src/Storages/IStorageCluster.cpp | 26 ++++++++++++++++--- src/Storages/IStorageCluster.h | 10 ++++--- .../StorageObjectStorageCluster.cpp | 8 ++---- .../StorageObjectStorageCluster.h | 2 +- ...ster_fallback_to_local_if_empty.reference} | 0 ...ge_cluster_fallback_to_local_if_empty.sql} | 0 7 files changed, 34 insertions(+), 15 deletions(-) rename tests/queries/0_stateless/{04303_object_storage_cluster_fallback_if_empty.reference => 04303_object_storage_cluster_fallback_to_local_if_empty.reference} (100%) rename tests/queries/0_stateless/{04303_object_storage_cluster_fallback_if_empty.sql => 04303_object_storage_cluster_fallback_to_local_if_empty.sql} (100%) diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 23a51beca8ac..e3bb87e4206f 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7774,7 +7774,8 @@ Trigger processor to spill data into external storage adpatively. grace join is Cluster to make distributed requests to object storages with alternative syntax. )", EXPERIMENTAL) \ DECLARE(Bool, object_storage_cluster_fallback_to_local_if_empty, false, R"( -Use non-cluster request if 'object_storage_cluster' is set but empty or unknown. +Execute the read locally if 'object_storage_cluster' is set but the cluster is empty or unknown. +Does not apply to explicit *Cluster table functions. Does not apply to writes. )", EXPERIMENTAL) \ DECLARE(UInt64, object_storage_max_nodes, 0, R"( Limit for hosts used for request in object storage cluster table functions - azureBlobStorageCluster, s3Cluster, hdfsCluster, etc. diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index bdf828751f92..b1d009e72061 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -345,8 +345,21 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( } } +bool IStorageCluster::shouldFallbackToLocalOnEmptyCluster(ContextPtr context) const +{ + return context->getSettingsRef()[Setting::object_storage_cluster_fallback_to_local_if_empty] + && allowsLocalFallbackOnEmptyObjectStorageCluster(context); +} + IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(ContextPtr context) const { + /// Decision matrix for object_storage_cluster_fallback_to_local_if_empty: + /// - s3()/iceberg() (or ENGINE ... SETTINGS object_storage_cluster) + empty/unknown cluster + setting + /// -> read locally (fallback_to_pure). + /// - s3Cluster(...)/explicit *Cluster argument -> never fall back locally. + /// - object_storage_remote_initiator=1 with object_storage_remote_initiator_cluster set + /// -> defer object_storage_cluster resolution to the remote initiator. + /// - writes -> never apply this fallback (see write()). ResolvedClusterRead result; if (!isClusterSupported()) @@ -357,6 +370,7 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context auto cluster_name_from_settings = getClusterName(context); const auto & settings = context->getSettingsRef(); + const bool local_fallback = shouldFallbackToLocalOnEmptyCluster(context); /// When both remote-initiator settings are set, object_storage_cluster may be defined only on the remote node. /// In this case object_storage_cluster must not be resolved locally. @@ -369,9 +383,12 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context else result.fallback_to_pure = cluster_name_from_settings.empty(); - if (!defer_object_storage_cluster_resolution + const bool try_resolve_with_local_fallback + = !defer_object_storage_cluster_resolution && !result.fallback_to_pure - && useObjectStorageClusterFallbackIfEmpty(context)) + && local_fallback; + + if (try_resolve_with_local_fallback) { result.object_storage_cluster = getClusterImpl( context, @@ -387,7 +404,8 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; if (!remote_initiator_cluster_name.empty()) { - const bool allow_null = settings[Setting::object_storage_cluster_fallback_to_local_if_empty] + /// Allow a missing remote-initiator cluster only when we would fall back to a pure/local read anyway. + const bool allow_null = local_fallback && (result.fallback_to_pure || usePureFunctionForRemoteInitiator(context)); result.remote_initiator_cluster = getClusterImpl( context, @@ -433,7 +451,7 @@ void IStorageCluster::read( { if (!resolved.remote_initiator_cluster) { - if (settings[Setting::object_storage_cluster_fallback_to_local_if_empty]) + if (shouldFallbackToLocalOnEmptyCluster(context)) { readFallBackToPure(query_plan, column_names, storage_snapshot, query_info, context, processed_stage, max_block_size, num_streams); return; diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index e9fad0fd273f..c5f60519dc28 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -113,7 +113,7 @@ class IStorageCluster : public IStorage { /// True when read() should use readFallBackToPure() or remote-initiator fallback branch. bool fallback_to_pure = false; - /// Pre-resolved object-storage cluster when object_storage_cluster_fallback_to_local_if_empty prefetch was done. + /// Pre-resolved object-storage cluster when local-fallback prefetch was done. ClusterPtr object_storage_cluster; /// Resolved remote-initiator cluster when object_storage_remote_initiator is enabled in fallback branch. ClusterPtr remote_initiator_cluster; @@ -121,8 +121,12 @@ class IStorageCluster : public IStorage ResolvedClusterRead resolveClusterRead(ContextPtr context) const; - /// Apply object_storage_cluster_fallback_to_local_if_empty only for storages that take cluster name from the setting. - virtual bool useObjectStorageClusterFallbackIfEmpty(ContextPtr /* context */) const { return false; } + /// Storage policy: may apply object_storage_cluster_fallback_to_local_if_empty. + /// True for alternative syntax / table engine settings; false for explicit *Cluster(...). + virtual bool allowsLocalFallbackOnEmptyObjectStorageCluster(ContextPtr /* context */) const { return false; } + + /// Setting enabled and storage allows local fallback on empty/unknown object_storage_cluster. + bool shouldFallbackToLocalOnEmptyCluster(ContextPtr context) const; /// True for s3()/iceberg() alternative syntax (cluster name from object_storage_cluster setting, not *Cluster argument). virtual bool usePureFunctionForRemoteInitiator(ContextPtr /* context */) const { return false; } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index e645c7edf256..06a039a8cf0e 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -42,7 +42,6 @@ namespace Setting extern const SettingsInt64 delta_lake_snapshot_end_version; extern const SettingsUInt64 lock_object_storage_task_distribution_ms; extern const SettingsBool allow_experimental_iceberg_read_optimization; - extern const SettingsBool object_storage_cluster_fallback_to_local_if_empty; } namespace ErrorCodes @@ -724,11 +723,8 @@ String StorageObjectStorageCluster::getClusterName(ContextPtr context) const return getOriginalClusterName(); } -bool StorageObjectStorageCluster::useObjectStorageClusterFallbackIfEmpty(ContextPtr context) const +bool StorageObjectStorageCluster::allowsLocalFallbackOnEmptyObjectStorageCluster(ContextPtr context) const { - if (!context->getSettingsRef()[Setting::object_storage_cluster_fallback_to_local_if_empty]) - return false; - if (cluster_name_from_function_argument) return false; @@ -755,7 +751,7 @@ QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( { if (!resolved.remote_initiator_cluster) { - if (!settings[Setting::object_storage_cluster_fallback_to_local_if_empty]) + if (!shouldFallbackToLocalOnEmptyCluster(context)) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments"); diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 1d09bb33a8c4..d452cba8cb7d 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -201,7 +201,7 @@ class StorageObjectStorageCluster : public IStorageCluster ContextPtr context, bool async_insert) override; - bool useObjectStorageClusterFallbackIfEmpty(ContextPtr context) const override; + bool allowsLocalFallbackOnEmptyObjectStorageCluster(ContextPtr context) const override; bool usePureFunctionForRemoteInitiator(ContextPtr /* context */) const override { return cluster_name_in_settings; } diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference similarity index 100% rename from tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.reference rename to tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql similarity index 100% rename from tests/queries/0_stateless/04303_object_storage_cluster_fallback_if_empty.sql rename to tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql From 20fa7358f13fac7f605c7b5fa0f3b7959d58145b Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 20:23:54 +0200 Subject: [PATCH 11/18] Fix remote initiator with only object_storage_cluster for alternative syntax. Pure-send to a remote initiator requires object_storage_remote_initiator_cluster; otherwise keep the clustered path that defaults the initiator cluster to object_storage_cluster. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 12 ++-- .../StorageObjectStorageCluster.cpp | 4 +- tests/integration/test_s3_cluster/test.py | 66 +++++++++++++++++++ 3 files changed, 77 insertions(+), 5 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index b1d009e72061..a64269cf7ccb 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -353,13 +353,15 @@ bool IStorageCluster::shouldFallbackToLocalOnEmptyCluster(ContextPtr context) co IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(ContextPtr context) const { - /// Decision matrix for object_storage_cluster_fallback_to_local_if_empty: + /// Decision matrix for object_storage_cluster_fallback_to_local_if_empty / remote initiator: /// - s3()/iceberg() (or ENGINE ... SETTINGS object_storage_cluster) + empty/unknown cluster + setting /// -> read locally (fallback_to_pure). /// - s3Cluster(...)/explicit *Cluster argument -> never fall back locally. /// - object_storage_remote_initiator=1 with object_storage_remote_initiator_cluster set - /// -> defer object_storage_cluster resolution to the remote initiator. - /// - writes -> never apply this fallback (see write()). + /// -> defer object_storage_cluster resolution; send pure s3()/iceberg() to the remote initiator. + /// - object_storage_remote_initiator=1 with only object_storage_cluster (no remote_initiator_cluster) + /// -> clustered remote path: default initiator cluster to object_storage_cluster, send *Cluster. + /// - writes -> never apply local fallback (see write()). ResolvedClusterRead result; if (!isClusterSupported()) @@ -442,8 +444,10 @@ void IStorageCluster::read( auto resolved = resolveClusterRead(context); ClusterPtr cluster = resolved.object_storage_cluster; + /// Pure-send only when remote_initiator_cluster was resolved (requires object_storage_remote_initiator_cluster). + /// Alternative syntax with only object_storage_cluster must use the clustered remote-initiator path below. const bool send_pure_function_to_remote_initiator - = settings[Setting::object_storage_remote_initiator] && usePureFunctionForRemoteInitiator(context); + = resolved.remote_initiator_cluster && usePureFunctionForRemoteInitiator(context); if (resolved.fallback_to_pure || send_pure_function_to_remote_initiator) { diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 06a039a8cf0e..7992e4037b85 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -742,8 +742,10 @@ QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( auto resolved = resolveClusterRead(context); const auto & settings = context->getSettingsRef(); + /// Pure-send only when remote_initiator_cluster was resolved (requires object_storage_remote_initiator_cluster). + /// Alternative syntax with only object_storage_cluster must use the clustered remote-initiator path. const bool send_pure_function_to_remote_initiator - = settings[Setting::object_storage_remote_initiator] && usePureFunctionForRemoteInitiator(context); + = resolved.remote_initiator_cluster && usePureFunctionForRemoteInitiator(context); if (resolved.fallback_to_pure || send_pure_function_to_remote_initiator) { diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index ec4190209af9..ea20018722d7 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -1127,6 +1127,72 @@ def test_object_storage_remote_initiator(started_cluster): "s0_1_0\tfoo"] +def test_object_storage_remote_initiator_with_object_storage_cluster_only(started_cluster): + """Alternative syntax with object_storage_cluster + remote_initiator, without remote_initiator_cluster. + + Must default the initiator cluster to object_storage_cluster and send a clustered request, + not throw BAD_ARGUMENTS or fall back to a local read. + """ + node = started_cluster.instances["s0_0_0"] + + query_id = uuid.uuid4().hex + result = node.query( + f""" + SELECT * from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') ORDER BY (name, value, polygon) + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster='cluster_remote' + """, + query_id=query_id, + ) + + assert result is not None + + node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster_all'") + queries = node.query( + f""" + SELECT count() + FROM clusterAllReplicas('cluster_all', system.query_log) + WHERE type='QueryFinish' AND initial_query_id='{query_id}' + FORMAT TSV + """ + ).splitlines() + + # initial node + remote initiator + 2 subqueries on replicas + assert queries == ["4"] + + # Same config must not fall back locally when fallback setting is enabled. + query_id = uuid.uuid4().hex + result = node.query( + f""" + SELECT * from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') ORDER BY (name, value, polygon) + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster='cluster_remote', + object_storage_cluster_fallback_to_local_if_empty=1 + """, + query_id=query_id, + ) + + assert result is not None + + node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster_all'") + queries = node.query( + f""" + SELECT count() + FROM clusterAllReplicas('cluster_all', system.query_log) + WHERE type='QueryFinish' AND initial_query_id='{query_id}' + FORMAT TSV + """ + ).splitlines() + + assert queries == ["4"] + + def test_remote_hedged(started_cluster): node = started_cluster.instances["s0_0_0"] pure_s3 = node.query( From 86ff083fb399562aa3bd824bba65cc80a59083fe Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 20:29:43 +0200 Subject: [PATCH 12/18] Restore pure remote path for ENGINE tables without object_storage_cluster. Under remote-initiator deferral, an empty local cluster name must fall back to pure send so ENGINE=S3/Iceberg without object_storage_cluster does not hit LOGICAL_ERROR. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 8 +++-- tests/integration/test_s3_cluster/test.py | 38 +++++++++++++++++++++++ 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index a64269cf7ccb..cb0158167144 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -358,9 +358,12 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context /// -> read locally (fallback_to_pure). /// - s3Cluster(...)/explicit *Cluster argument -> never fall back locally. /// - object_storage_remote_initiator=1 with object_storage_remote_initiator_cluster set - /// -> defer object_storage_cluster resolution; send pure s3()/iceberg() to the remote initiator. + /// -> defer object_storage_cluster resolution; send pure s3()/iceberg() to the remote initiator + /// for alternative syntax, or when the local cluster name is empty (ENGINE without object_storage_cluster). /// - object_storage_remote_initiator=1 with only object_storage_cluster (no remote_initiator_cluster) /// -> clustered remote path: default initiator cluster to object_storage_cluster, send *Cluster. + /// - ENGINE/Iceberg with a local object_storage_cluster + remote_initiator_cluster + /// -> clustered remote path (send *Cluster to the remote initiator). /// - writes -> never apply local fallback (see write()). ResolvedClusterRead result; @@ -381,7 +384,8 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context && !settings[Setting::object_storage_remote_initiator_cluster].value.empty(); if (defer_object_storage_cluster_resolution) - result.fallback_to_pure = usePureFunctionForRemoteInitiator(context); + result.fallback_to_pure + = cluster_name_from_settings.empty() || usePureFunctionForRemoteInitiator(context); else result.fallback_to_pure = cluster_name_from_settings.empty(); diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index ea20018722d7..9efcecb04cec 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -1673,6 +1673,44 @@ def test_object_storage_remote_initiator_without_cluster_function(started_cluste assert users[0] in ["c2.s0_0_0\tdefault", "c2.s0_0_1\tdefault"] assert users[1:] == ["s0_0_0\tdefault"] + # ENGINE table without object_storage_cluster must also take the pure remote path. + node.query("DROP TABLE IF EXISTS engine_remote_initiator_no_cluster") + node.query( + f""" + CREATE TABLE engine_remote_initiator_no_cluster + (name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))) + ENGINE=S3('http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV') + """ + ) + + query_id = uuid.uuid4().hex + result = node.query( + """ + SELECT * FROM engine_remote_initiator_no_cluster ORDER BY (name, value, polygon) + SETTINGS + object_storage_remote_initiator=1, + object_storage_remote_initiator_cluster='cluster_with_dots' + """, + query_id=query_id, + ) + + assert result is not None + + node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster_all'") + queries = node.query( + f""" + SELECT count() + FROM clusterAllReplicas('cluster_all', system.query_log) + WHERE type='QueryFinish' AND initial_query_id='{query_id}' + FORMAT TSV + """ + ).splitlines() + + # initial node + remote initiator + assert queries == ["2"] + + node.query("DROP TABLE IF EXISTS engine_remote_initiator_no_cluster") + # Remove initiator without cluster request # but with `object_storage_cluster` specified for user on remote cluster query_id = uuid.uuid4().hex From 9e1b15fbcddd8a36649d5ca61fc7e08d1c307e16 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 21:27:47 +0200 Subject: [PATCH 13/18] Simplify cluster read routing after remote-initiator fixes. Drop the redundant pure-send gate, flatten fallback_to_pure, and reuse the already resolved remote-initiator cluster on the clustered path. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 26 +++++++------------ src/Storages/IStorageCluster.h | 2 +- .../StorageObjectStorageCluster.cpp | 7 +---- 3 files changed, 12 insertions(+), 23 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index cb0158167144..59fdee293785 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -358,8 +358,8 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context /// -> read locally (fallback_to_pure). /// - s3Cluster(...)/explicit *Cluster argument -> never fall back locally. /// - object_storage_remote_initiator=1 with object_storage_remote_initiator_cluster set - /// -> defer object_storage_cluster resolution; send pure s3()/iceberg() to the remote initiator - /// for alternative syntax, or when the local cluster name is empty (ENGINE without object_storage_cluster). + /// -> defer object_storage_cluster resolution; fallback_to_pure when the local cluster name is empty + /// or for alternative syntax (pure s3()/iceberg() sent to the remote initiator). /// - object_storage_remote_initiator=1 with only object_storage_cluster (no remote_initiator_cluster) /// -> clustered remote path: default initiator cluster to object_storage_cluster, send *Cluster. /// - ENGINE/Iceberg with a local object_storage_cluster + remote_initiator_cluster @@ -383,11 +383,8 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context = settings[Setting::object_storage_remote_initiator] && !settings[Setting::object_storage_remote_initiator_cluster].value.empty(); - if (defer_object_storage_cluster_resolution) - result.fallback_to_pure - = cluster_name_from_settings.empty() || usePureFunctionForRemoteInitiator(context); - else - result.fallback_to_pure = cluster_name_from_settings.empty(); + result.fallback_to_pure = cluster_name_from_settings.empty() + || (defer_object_storage_cluster_resolution && usePureFunctionForRemoteInitiator(context)); const bool try_resolve_with_local_fallback = !defer_object_storage_cluster_resolution @@ -411,8 +408,7 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context if (!remote_initiator_cluster_name.empty()) { /// Allow a missing remote-initiator cluster only when we would fall back to a pure/local read anyway. - const bool allow_null = local_fallback - && (result.fallback_to_pure || usePureFunctionForRemoteInitiator(context)); + const bool allow_null = local_fallback && result.fallback_to_pure; result.remote_initiator_cluster = getClusterImpl( context, remote_initiator_cluster_name, @@ -448,12 +444,7 @@ void IStorageCluster::read( auto resolved = resolveClusterRead(context); ClusterPtr cluster = resolved.object_storage_cluster; - /// Pure-send only when remote_initiator_cluster was resolved (requires object_storage_remote_initiator_cluster). - /// Alternative syntax with only object_storage_cluster must use the clustered remote-initiator path below. - const bool send_pure_function_to_remote_initiator - = resolved.remote_initiator_cluster && usePureFunctionForRemoteInitiator(context); - - if (resolved.fallback_to_pure || send_pure_function_to_remote_initiator) + if (resolved.fallback_to_pure) { if (settings[Setting::object_storage_remote_initiator]) { @@ -525,7 +516,10 @@ void IStorageCluster::read( auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; if (remote_initiator_cluster_name.empty()) remote_initiator_cluster_name = cluster_name_from_settings; - auto remote_initiator_cluster = getClusterImpl(context, remote_initiator_cluster_name); + /// Prefer the cluster already resolved in resolveClusterRead (when remote_initiator_cluster was set). + ClusterPtr remote_initiator_cluster = resolved.remote_initiator_cluster; + if (!remote_initiator_cluster) + remote_initiator_cluster = getClusterImpl(context, remote_initiator_cluster_name); auto storage_and_context = convertToRemote(remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send); auto src_distributed = std::dynamic_pointer_cast(storage_and_context.storage); auto modified_query_info = query_info; diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index c5f60519dc28..da1f158009b0 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -115,7 +115,7 @@ class IStorageCluster : public IStorage bool fallback_to_pure = false; /// Pre-resolved object-storage cluster when local-fallback prefetch was done. ClusterPtr object_storage_cluster; - /// Resolved remote-initiator cluster when object_storage_remote_initiator is enabled in fallback branch. + /// Resolved remote-initiator cluster when object_storage_remote_initiator_cluster is set. ClusterPtr remote_initiator_cluster; }; diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 7992e4037b85..bef23ef4c3c0 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -742,12 +742,7 @@ QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( auto resolved = resolveClusterRead(context); const auto & settings = context->getSettingsRef(); - /// Pure-send only when remote_initiator_cluster was resolved (requires object_storage_remote_initiator_cluster). - /// Alternative syntax with only object_storage_cluster must use the clustered remote-initiator path. - const bool send_pure_function_to_remote_initiator - = resolved.remote_initiator_cluster && usePureFunctionForRemoteInitiator(context); - - if (resolved.fallback_to_pure || send_pure_function_to_remote_initiator) + if (resolved.fallback_to_pure) { if (settings[Setting::object_storage_remote_initiator]) { From 4ba13df9e09f4879c1317f03235d5a594189150f Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 10:00:13 +0200 Subject: [PATCH 14/18] Do not soft-fail a missing object_storage_remote_initiator_cluster. Local fallback applies only to object_storage_cluster; a bad remote-initiator cluster must still report CLUSTER_DOESNT_EXIST. Co-authored-by: Cursor --- src/Core/Settings.cpp | 2 +- src/Storages/IStorageCluster.cpp | 7 +++---- tests/integration/test_s3_cluster/test.py | 16 ++++++++++++++++ ...torage_cluster_fallback_to_local_if_empty.sql | 8 ++++++++ 4 files changed, 28 insertions(+), 5 deletions(-) diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index e3bb87e4206f..6f15fb2d247d 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7775,7 +7775,7 @@ Cluster to make distributed requests to object storages with alternative syntax. )", EXPERIMENTAL) \ DECLARE(Bool, object_storage_cluster_fallback_to_local_if_empty, false, R"( Execute the read locally if 'object_storage_cluster' is set but the cluster is empty or unknown. -Does not apply to explicit *Cluster table functions. Does not apply to writes. +Does not apply to explicit *Cluster table functions, to 'object_storage_remote_initiator_cluster', or to writes. )", EXPERIMENTAL) \ DECLARE(UInt64, object_storage_max_nodes, 0, R"( Limit for hosts used for request in object storage cluster table functions - azureBlobStorageCluster, s3Cluster, hdfsCluster, etc. diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 59fdee293785..73e2cbc575f4 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -407,13 +407,12 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; if (!remote_initiator_cluster_name.empty()) { - /// Allow a missing remote-initiator cluster only when we would fall back to a pure/local read anyway. - const bool allow_null = local_fallback && result.fallback_to_pure; + /// Never soft-fail a missing/empty remote-initiator cluster via + /// object_storage_cluster_fallback_to_local_if_empty: that setting applies only to object_storage_cluster. result.remote_initiator_cluster = getClusterImpl( context, remote_initiator_cluster_name, - /*max_hosts*/ 0, - allow_null); + /*max_hosts*/ 0); } } diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index 9efcecb04cec..2e5f898df8b3 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -431,6 +431,22 @@ def test_object_storage_cluster_fallback_to_local_if_empty(started_cluster): # initial node + remote initiator. assert queries == ["2"] + # A missing remote-initiator cluster must not be masked by local OSC fallback. + error = node.query_and_get_error( + f""" + SELECT count(*) from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', + 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster='cluster_remote', + object_storage_remote_initiator_cluster='non_existing_remote_initiator_cluster', + object_storage_cluster_fallback_to_local_if_empty=1 + """ + ) + assert "not found" in error or "CLUSTER_DOESNT_EXIST" in error or "doesn't exist" in error.lower() + def test_ambiguous_join(started_cluster): node = started_cluster.instances["s0_0_0"] diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql index 8f909c432599..8db3e2cc4c71 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql @@ -49,6 +49,14 @@ SETTINGS object_storage_remote_initiator = 1, object_storage_remote_initiator_cluster = 'test_shard_localhost'; +SELECT 'remote initiator cluster missing does not use local fallback'; +SELECT count() FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS + object_storage_cluster = 'test_shard_localhost', + object_storage_cluster_fallback_to_local_if_empty = 1, + object_storage_remote_initiator = 1, + object_storage_remote_initiator_cluster = 'non_existent_remote_initiator_04303'; -- { serverError CLUSTER_DOESNT_EXIST } + SELECT 'aggregate with fallback'; SELECT sum(x * x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_to_local_if_empty = 1; From 4b41b1db185436acad47791f3bfacfc12ccf2adb Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 10:52:37 +0200 Subject: [PATCH 15/18] Require non-empty object_storage_cluster for local fallback. Empty OSC with remote_initiator and no remote_initiator_cluster must keep BAD_ARGUMENTS even when fallback is enabled; extend tests for cases 1-3 and this regression. Co-authored-by: Cursor --- .../StorageObjectStorageCluster.cpp | 6 ++-- tests/integration/test_s3_cluster/test.py | 36 +++++++++++++++++++ ...uster_fallback_to_local_if_empty.reference | 19 +++++----- ...age_cluster_fallback_to_local_if_empty.sql | 35 +++++++++++++----- 4 files changed, 76 insertions(+), 20 deletions(-) diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index bef23ef4c3c0..bf9067d133bc 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -728,9 +728,9 @@ bool StorageObjectStorageCluster::allowsLocalFallbackOnEmptyObjectStorageCluster if (cluster_name_from_function_argument) return false; - return cluster_name_in_settings - || !getOriginalClusterName().empty() - || !context->getSettingsRef()[Setting::object_storage_cluster].value.empty(); + /// Only when a non-empty object_storage_cluster was requested (query setting or table engine). + /// Empty OSC + remote_initiator without remote_initiator_cluster must keep BAD_ARGUMENTS. + return !getClusterName(context).empty(); } QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index 2e5f898df8b3..c4bb354eb116 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -447,6 +447,42 @@ def test_object_storage_cluster_fallback_to_local_if_empty(started_cluster): ) assert "not found" in error or "CLUSTER_DOESNT_EXIST" in error or "doesn't exist" in error.lower() + # Empty OSC + RI without RI-cluster must keep BAD_ARGUMENTS even with fallback enabled. + error = node.query_and_get_error( + f""" + SELECT count(*) from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', + 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster_fallback_to_local_if_empty=1 + """ + ) + assert "BAD_ARGUMENTS" in error or "object_storage_remote_initiator" in error + + # Case 1 for ENGINE: unknown OSC in table SETTINGS + fallback -> local. + node.query("DROP TABLE IF EXISTS engine_osc_fallback") + node.query( + f""" + CREATE TABLE engine_osc_fallback + (name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))) + ENGINE=S3('http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV') + SETTINGS object_storage_cluster='non_existing_cluster' + """ + ) + error = node.query_and_get_error("SELECT count(*) FROM engine_osc_fallback") + assert "not found" in error or "CLUSTER_DOESNT_EXIST" in error or "doesn't exist" in error.lower() + + engine_fallback_count = node.query( + """ + SELECT count(*) FROM engine_osc_fallback + SETTINGS object_storage_cluster_fallback_to_local_if_empty=1 + """ + ) + assert TSV(pure_count) == TSV(engine_fallback_count) + node.query("DROP TABLE IF EXISTS engine_osc_fallback") + def test_ambiguous_join(started_cluster): node = started_cluster.instances["s0_0_0"] diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference index bc3d0c602799..2a4c5257b765 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference @@ -1,18 +1,21 @@ pure 10 45 -unknown cluster without fallback -unknown cluster with fallback +case1 unknown OSC without fallback still errors +case1 unknown OSC with fallback runs locally 10 45 -valid cluster with fallback +valid OSC with fallback unchanged 10 45 -explicit cluster function with fallback -explicit cluster function overrides setting -explicit cluster function overrides setting with valid cluster +s3Cluster ignores fallback +s3Cluster ignores fallback even with OSC setting +s3Cluster with valid argument unchanged 10 45 -remote initiator unresolved with fallback +case2 unknown OSC + RI without RI-cluster with fallback runs locally 10 45 -remote initiator with non-existent cluster +empty OSC + RI without RI-cluster with fallback still BAD_ARGUMENTS +empty OSC + RI without RI-cluster without fallback still BAD_ARGUMENTS +case3-like unknown OSC + RI-cluster with fallback (pure send, remote local) 10 45 +missing RI-cluster is not masked by OSC fallback aggregate with fallback 285 pure aggregate diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql index 8db3e2cc4c71..62d3903c7c45 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql @@ -1,5 +1,12 @@ -- Tags: no-fasttest -- Tag no-fasttest: Depends on Minio +-- +-- object_storage_cluster_fallback_to_local_if_empty applies only to non-cluster table functions / table engines. +-- Intended behavior changes (setting=1): +-- 1) OSC non-empty but unknown/empty cluster, RI=0 -> local instead of error +-- 2) OSC non-empty but unknown/empty cluster, RI=1, RI-cluster empty -> local instead of error +-- 3) local OSC empty, RI=1, valid RI-cluster; remote OSC unknown/empty -> remote non-distributed instead of error +-- Other cases must keep pre-setting behavior. SET enable_analyzer = 1; @@ -10,38 +17,48 @@ SETTINGS s3_truncate_on_insert = 1; SELECT 'pure'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32'); -SELECT 'unknown cluster without fallback'; +SELECT 'case1 unknown OSC without fallback still errors'; SELECT count() FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303'; -- { serverError CLUSTER_DOESNT_EXIST } -SELECT 'unknown cluster with fallback'; +SELECT 'case1 unknown OSC with fallback runs locally'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_to_local_if_empty = 1; -SELECT 'valid cluster with fallback'; +SELECT 'valid OSC with fallback unchanged'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'test_shard_localhost', object_storage_cluster_fallback_to_local_if_empty = 1; -SELECT 'explicit cluster function with fallback'; +SELECT 's3Cluster ignores fallback'; SELECT count() FROM s3Cluster('non_existent_cluster_04303', 'http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster_fallback_to_local_if_empty = 1; -- { serverError CLUSTER_DOESNT_EXIST } -SELECT 'explicit cluster function overrides setting'; +SELECT 's3Cluster ignores fallback even with OSC setting'; SELECT count() FROM s3Cluster('non_existent_cluster_04303', 'http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster_fallback_to_local_if_empty = 1, object_storage_cluster = 'non_existent_cluster_04303_2'; -- { serverError CLUSTER_DOESNT_EXIST } -SELECT 'explicit cluster function overrides setting with valid cluster'; +SELECT 's3Cluster with valid argument unchanged'; SELECT count(), sum(x) FROM s3Cluster('test_shard_localhost', 'http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_to_local_if_empty = 1; -SELECT 'remote initiator unresolved with fallback'; +SELECT 'case2 unknown OSC + RI without RI-cluster with fallback runs locally'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_to_local_if_empty = 1, object_storage_remote_initiator = 1; -SELECT 'remote initiator with non-existent cluster'; +SELECT 'empty OSC + RI without RI-cluster with fallback still BAD_ARGUMENTS'; +SELECT count() FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS + object_storage_remote_initiator = 1, + object_storage_cluster_fallback_to_local_if_empty = 1; -- { serverError BAD_ARGUMENTS } + +SELECT 'empty OSC + RI without RI-cluster without fallback still BAD_ARGUMENTS'; +SELECT count() FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') +SETTINGS object_storage_remote_initiator = 1; -- { serverError BAD_ARGUMENTS } + +SELECT 'case3-like unknown OSC + RI-cluster with fallback (pure send, remote local)'; SELECT count(), sum(x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303', @@ -49,7 +66,7 @@ SETTINGS object_storage_remote_initiator = 1, object_storage_remote_initiator_cluster = 'test_shard_localhost'; -SELECT 'remote initiator cluster missing does not use local fallback'; +SELECT 'missing RI-cluster is not masked by OSC fallback'; SELECT count() FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'test_shard_localhost', From afbd078a12c414af1847deffcb03ed06acd54d10 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 11:22:36 +0200 Subject: [PATCH 16/18] Apply object_storage_cluster fallback to ENGINE with remote initiator. Pure-send ENGINE tables under remote-initiator deferral (preserving OSC in SETTINGS/context) so fallback matches s3() alternative syntax; extend TF and ENGINE tests for cases 1-3. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 19 ++++- src/Storages/IStorageCluster.h | 2 +- .../StorageObjectStorageCluster.cpp | 18 ++-- .../StorageObjectStorageCluster.h | 7 +- tests/integration/test_s3_cluster/test.py | 84 +++++++++++++++++++ ...uster_fallback_to_local_if_empty.reference | 8 ++ ...age_cluster_fallback_to_local_if_empty.sql | 39 +++++++++ 7 files changed, 162 insertions(+), 15 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 73e2cbc575f4..6cb710f09e10 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -56,6 +56,7 @@ namespace Setting extern const SettingsUInt64 object_storage_max_nodes; extern const SettingsBool object_storage_remote_initiator; extern const SettingsString object_storage_remote_initiator_cluster; + extern const SettingsString object_storage_cluster; extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode; extern const SettingsBool object_storage_cluster_fallback_to_local_if_empty; } @@ -359,11 +360,9 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context /// - s3Cluster(...)/explicit *Cluster argument -> never fall back locally. /// - object_storage_remote_initiator=1 with object_storage_remote_initiator_cluster set /// -> defer object_storage_cluster resolution; fallback_to_pure when the local cluster name is empty - /// or for alternative syntax (pure s3()/iceberg() sent to the remote initiator). + /// or for non-*Cluster forms (pure s3()/iceberg()/ENGINE sent to the remote initiator). /// - object_storage_remote_initiator=1 with only object_storage_cluster (no remote_initiator_cluster) /// -> clustered remote path: default initiator cluster to object_storage_cluster, send *Cluster. - /// - ENGINE/Iceberg with a local object_storage_cluster + remote_initiator_cluster - /// -> clustered remote path (send *Cluster to the remote initiator). /// - writes -> never apply local fallback (see write()). ResolvedClusterRead result; @@ -466,7 +465,19 @@ void IStorageCluster::read( updateQueryWithJoinToSendIfNeeded(query_to_send, query_info, context); updateQueryToSendIfNeeded(query_to_send, storage_snapshot, context, /*make_cluster_function*/ false); - auto storage_and_context = convertToRemote(resolved.remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send); + /// ENGINE tables keep object_storage_cluster on the storage, not in the initiator query context. + /// Propagate it into the context copied for remote() so the remote node sees the same OSC as alt-syntax. + auto context_for_remote = context; + const auto object_storage_cluster_name = getClusterName(context); + if (!object_storage_cluster_name.empty() + && context->getSettingsRef()[Setting::object_storage_cluster].value.empty()) + { + auto ctx = Context::createCopy(context); + ctx->setSetting("object_storage_cluster", object_storage_cluster_name); + context_for_remote = ctx; + } + + auto storage_and_context = convertToRemote(resolved.remote_initiator_cluster, context_for_remote, remote_initiator_cluster_name, query_to_send); auto src_distributed = std::dynamic_pointer_cast(storage_and_context.storage); auto modified_query_info = query_info; modified_query_info.cluster = src_distributed->getCluster(); diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index da1f158009b0..8540d8a56fe2 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -128,7 +128,7 @@ class IStorageCluster : public IStorage /// Setting enabled and storage allows local fallback on empty/unknown object_storage_cluster. bool shouldFallbackToLocalOnEmptyCluster(ContextPtr context) const; - /// True for s3()/iceberg() alternative syntax (cluster name from object_storage_cluster setting, not *Cluster argument). + /// True for non-*Cluster forms (alternative syntax and table engines): send pure s3()/iceberg() to the remote initiator. virtual bool usePureFunctionForRemoteInitiator(ContextPtr /* context */) const { return false; } private: diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index bf9067d133bc..0fadd8b20053 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -392,19 +392,23 @@ bool StorageObjectStorageCluster::updateQueryForDistributedEngineIfNeeded(ASTPtr table_expression->table_function = function_ast_ptr; table_expression->children[0] = function_ast_ptr; - if (!make_cluster_function) - return false; - auto cluster_name = getClusterName(context); if (cluster_name.empty()) { - throw Exception( - ErrorCodes::LOGICAL_ERROR, - "Can't be here without cluster name, no cluster name in query {}", - query->formatForLogging()); + /// Pure remote path without a local object_storage_cluster (remote may define it). + if (make_cluster_function) + { + throw Exception( + ErrorCodes::LOGICAL_ERROR, + "Can't be here without cluster name, no cluster name in query {}", + query->formatForLogging()); + } + return false; } + /// Inject OSC into SETTINGS for both pure and *Cluster rewrite paths. + /// Pure send must preserve ENGINE object_storage_cluster so the remote can distribute or fall back. auto settings = select_query->settings(); if (settings) { diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index d452cba8cb7d..7bfca12e1ad1 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -203,7 +203,8 @@ class StorageObjectStorageCluster : public IStorageCluster bool allowsLocalFallbackOnEmptyObjectStorageCluster(ContextPtr context) const override; - bool usePureFunctionForRemoteInitiator(ContextPtr /* context */) const override { return cluster_name_in_settings; } + /// Pure s3()/iceberg() (not *Cluster) for remote initiator: alternative syntax and table engines. + bool usePureFunctionForRemoteInitiator(ContextPtr /* context */) const override { return !cluster_name_from_function_argument; } /* In case the table was created with `object_storage_cluster` setting, @@ -211,11 +212,11 @@ class StorageObjectStorageCluster : public IStorageCluster by mapping the engine name to table function name and setting `object_storage_cluster`. For table like CREATE TABLE table ENGINE=S3(...) SETTINGS object_storage_cluster='cluster' - coverts request + converts request SELECT * FROM table to SELECT * FROM s3(...) SETTINGS object_storage_cluster='cluster' - to make distributed request over cluster 'cluster'. + (and optionally to s3Cluster when make_cluster_function is true). Returns true if cluster name was added to settings. */ bool updateQueryForDistributedEngineIfNeeded(ASTPtr & query, ContextPtr context, bool make_cluster_function); diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index c4bb354eb116..80fdbad6d0cd 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -431,6 +431,21 @@ def test_object_storage_cluster_fallback_to_local_if_empty(started_cluster): # initial node + remote initiator. assert queries == ["2"] + # Without fallback, unknown OSC + RI-cluster still errors on remote (table function). + error = node.query_and_get_error( + f""" + SELECT count(*) from s3( + 'http://minio1:9001/root/data/{{clickhouse,database}}/*', + 'minio', '{minio_secret_key}', 'CSV', + 'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster='non_existing_cluster', + object_storage_remote_initiator_cluster='cluster_with_dots' + """ + ) + assert "not found" in error or "CLUSTER_DOESNT_EXIST" in error or "doesn't exist" in error.lower() + # A missing remote-initiator cluster must not be masked by local OSC fallback. error = node.query_and_get_error( f""" @@ -481,6 +496,75 @@ def test_object_storage_cluster_fallback_to_local_if_empty(started_cluster): """ ) assert TSV(pure_count) == TSV(engine_fallback_count) + + # Case 2 for ENGINE: unknown OSC + RI without RI-cluster + fallback -> local. + engine_case2_count = node.query( + """ + SELECT count(*) FROM engine_osc_fallback + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster_fallback_to_local_if_empty=1 + """ + ) + assert TSV(pure_count) == TSV(engine_case2_count) + + # ENGINE empty OSC + RI without RI-cluster + fallback must keep BAD_ARGUMENTS. + node.query("DROP TABLE IF EXISTS engine_no_osc_ri") + node.query( + f""" + CREATE TABLE engine_no_osc_ri + (name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))) + ENGINE=S3('http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV') + """ + ) + error = node.query_and_get_error( + """ + SELECT count(*) FROM engine_no_osc_ri + SETTINGS + object_storage_remote_initiator=1, + object_storage_cluster_fallback_to_local_if_empty=1 + """ + ) + assert "BAD_ARGUMENTS" in error or "object_storage_remote_initiator" in error + node.query("DROP TABLE IF EXISTS engine_no_osc_ri") + + # Case 3-like for ENGINE: unknown OSC + RI-cluster + fallback -> pure send, remote local. + query_id = uuid.uuid4().hex + engine_case3_count = node.query( + """ + SELECT count(*) FROM engine_osc_fallback + SETTINGS + object_storage_remote_initiator=1, + object_storage_remote_initiator_cluster='cluster_with_dots', + object_storage_cluster_fallback_to_local_if_empty=1 + """, + query_id=query_id, + ) + assert TSV(pure_count) == TSV(engine_case3_count) + + node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster_all'") + queries = node.query( + f""" + SELECT count() + FROM clusterAllReplicas('cluster_all', system.query_log) + WHERE type='QueryFinish' AND initial_query_id='{query_id}' + FORMAT TSV + """ + ).splitlines() + # initial node + remote initiator (remote cannot distribute: OSC unknown, fallback local) + assert queries == ["2"] + + # Without fallback, ENGINE unknown OSC + RI-cluster still errors on remote. + error = node.query_and_get_error( + """ + SELECT count(*) FROM engine_osc_fallback + SETTINGS + object_storage_remote_initiator=1, + object_storage_remote_initiator_cluster='cluster_with_dots' + """ + ) + assert "not found" in error or "CLUSTER_DOESNT_EXIST" in error or "doesn't exist" in error.lower() + node.query("DROP TABLE IF EXISTS engine_osc_fallback") diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference index 2a4c5257b765..5682404d3d11 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference @@ -16,6 +16,14 @@ empty OSC + RI without RI-cluster without fallback still BAD_ARGUMENTS case3-like unknown OSC + RI-cluster with fallback (pure send, remote local) 10 45 missing RI-cluster is not masked by OSC fallback +engine case1 unknown OSC without fallback still errors +engine case1 unknown OSC with fallback runs locally +10 45 +engine case2 unknown OSC + RI without RI-cluster with fallback runs locally +10 45 +engine case3-like unknown OSC + RI-cluster with fallback +10 45 +engine empty OSC + RI without RI-cluster with fallback still BAD_ARGUMENTS aggregate with fallback 285 pure aggregate diff --git a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql index 62d3903c7c45..48841c4d5baf 100644 --- a/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql @@ -7,6 +7,7 @@ -- 2) OSC non-empty but unknown/empty cluster, RI=1, RI-cluster empty -> local instead of error -- 3) local OSC empty, RI=1, valid RI-cluster; remote OSC unknown/empty -> remote non-distributed instead of error -- Other cases must keep pre-setting behavior. +-- Covered for both s3() table function and S3 table engine. SET enable_analyzer = 1; @@ -74,6 +75,44 @@ SETTINGS object_storage_remote_initiator = 1, object_storage_remote_initiator_cluster = 'non_existent_remote_initiator_04303'; -- { serverError CLUSTER_DOESNT_EXIST } +DROP TABLE IF EXISTS engine_osc_fallback_04303; +CREATE TABLE engine_osc_fallback_04303 (x UInt32) +ENGINE = S3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV') +SETTINGS object_storage_cluster = 'non_existent_cluster_04303'; + +SELECT 'engine case1 unknown OSC without fallback still errors'; +SELECT count() FROM engine_osc_fallback_04303; -- { serverError CLUSTER_DOESNT_EXIST } + +SELECT 'engine case1 unknown OSC with fallback runs locally'; +SELECT count(), sum(x) FROM engine_osc_fallback_04303 +SETTINGS object_storage_cluster_fallback_to_local_if_empty = 1; + +SELECT 'engine case2 unknown OSC + RI without RI-cluster with fallback runs locally'; +SELECT count(), sum(x) FROM engine_osc_fallback_04303 +SETTINGS + object_storage_cluster_fallback_to_local_if_empty = 1, + object_storage_remote_initiator = 1; + +SELECT 'engine case3-like unknown OSC + RI-cluster with fallback'; +SELECT count(), sum(x) FROM engine_osc_fallback_04303 +SETTINGS + object_storage_cluster_fallback_to_local_if_empty = 1, + object_storage_remote_initiator = 1, + object_storage_remote_initiator_cluster = 'test_shard_localhost'; + +DROP TABLE IF EXISTS engine_no_osc_04303; +CREATE TABLE engine_no_osc_04303 (x UInt32) +ENGINE = S3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV'); + +SELECT 'engine empty OSC + RI without RI-cluster with fallback still BAD_ARGUMENTS'; +SELECT count() FROM engine_no_osc_04303 +SETTINGS + object_storage_remote_initiator = 1, + object_storage_cluster_fallback_to_local_if_empty = 1; -- { serverError BAD_ARGUMENTS } + +DROP TABLE engine_no_osc_04303; +DROP TABLE engine_osc_fallback_04303; + SELECT 'aggregate with fallback'; SELECT sum(x * x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32') SETTINGS object_storage_cluster = 'non_existent_cluster_04303', object_storage_cluster_fallback_to_local_if_empty = 1; From e0c3a5d716e2ede2c9e7617ae0f9b878b9f0c5a8 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 11:57:40 +0200 Subject: [PATCH 17/18] Simplify object_storage_cluster fallback routing helpers. Merge the two policy hooks into usesObjectStorageClusterSettingSyntax, share local-vs-remote fallback decisions between read and getQueryProcessingStage, and dedupe remote-initiator send. Co-authored-by: Cursor --- src/Storages/IStorageCluster.cpp | 138 ++++++++++++------ src/Storages/IStorageCluster.h | 27 +++- .../StorageObjectStorageCluster.cpp | 33 +---- .../StorageObjectStorageCluster.h | 5 +- ...leFunctionObjectStorageClusterFallback.cpp | 1 + 5 files changed, 115 insertions(+), 89 deletions(-) diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 6cb710f09e10..b5c9c588dcca 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -349,7 +349,26 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( bool IStorageCluster::shouldFallbackToLocalOnEmptyCluster(ContextPtr context) const { return context->getSettingsRef()[Setting::object_storage_cluster_fallback_to_local_if_empty] - && allowsLocalFallbackOnEmptyObjectStorageCluster(context); + && usesObjectStorageClusterSettingSyntax() + && !getClusterName(context).empty(); +} + +bool IStorageCluster::shouldReadLocallyOnFallbackToPure(const ResolvedClusterRead & resolved, ContextPtr context) const +{ + if (!resolved.fallback_to_pure) + return false; + + if (!context->getSettingsRef()[Setting::object_storage_remote_initiator]) + return true; + + if (resolved.remote_initiator_cluster) + return false; + + if (resolved.local_fallback) + return true; + + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments"); } IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(ContextPtr context) const @@ -374,7 +393,7 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context auto cluster_name_from_settings = getClusterName(context); const auto & settings = context->getSettingsRef(); - const bool local_fallback = shouldFallbackToLocalOnEmptyCluster(context); + result.local_fallback = shouldFallbackToLocalOnEmptyCluster(context); /// When both remote-initiator settings are set, object_storage_cluster may be defined only on the remote node. /// In this case object_storage_cluster must not be resolved locally. @@ -383,12 +402,12 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context && !settings[Setting::object_storage_remote_initiator_cluster].value.empty(); result.fallback_to_pure = cluster_name_from_settings.empty() - || (defer_object_storage_cluster_resolution && usePureFunctionForRemoteInitiator(context)); + || (defer_object_storage_cluster_resolution && usesObjectStorageClusterSettingSyntax()); const bool try_resolve_with_local_fallback = !defer_object_storage_cluster_resolution && !result.fallback_to_pure - && local_fallback; + && result.local_fallback; if (try_resolve_with_local_fallback) { @@ -418,6 +437,28 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context return result; } +void IStorageCluster::readFromRemoteInitiator( + QueryPlan & query_plan, + const Names & column_names, + const StorageSnapshotPtr & storage_snapshot, + SelectQueryInfo & query_info, + ContextPtr context, + QueryProcessingStage::Enum processed_stage, + size_t max_block_size, + size_t num_streams, + ASTPtr query_to_send, + ClusterPtr remote_initiator_cluster, + const String & remote_initiator_cluster_name) +{ + auto storage_and_context = convertToRemote(remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send); + auto src_distributed = std::dynamic_pointer_cast(storage_and_context.storage); + auto modified_query_info = query_info; + modified_query_info.cluster = src_distributed->getCluster(); + auto new_storage_snapshot = storage_and_context.storage->getStorageSnapshot(storage_snapshot->metadata, storage_and_context.context); + storage_and_context.storage->read( + query_plan, column_names, new_storage_snapshot, modified_query_info, storage_and_context.context, processed_stage, max_block_size, num_streams); +} + /// The code executes on initiator void IStorageCluster::read( QueryPlan & query_plan, @@ -444,49 +485,43 @@ void IStorageCluster::read( if (resolved.fallback_to_pure) { - if (settings[Setting::object_storage_remote_initiator]) + if (shouldReadLocallyOnFallbackToPure(resolved, context)) { - if (!resolved.remote_initiator_cluster) - { - if (shouldFallbackToLocalOnEmptyCluster(context)) - { - readFallBackToPure(query_plan, column_names, storage_snapshot, query_info, context, processed_stage, max_block_size, num_streams); - return; - } - throw Exception(ErrorCodes::BAD_ARGUMENTS, - "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments"); - } - - auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; - - /// rewrite query to execute `remote('remote_host', s3(...))` - /// remote_host can execute query itself or make on-cluster query depends on own `object_storage_cluster` setting - updateConfigurationIfNeeded(context); - updateQueryWithJoinToSendIfNeeded(query_to_send, query_info, context); - updateQueryToSendIfNeeded(query_to_send, storage_snapshot, context, /*make_cluster_function*/ false); + readFallBackToPure(query_plan, column_names, storage_snapshot, query_info, context, processed_stage, max_block_size, num_streams); + return; + } - /// ENGINE tables keep object_storage_cluster on the storage, not in the initiator query context. - /// Propagate it into the context copied for remote() so the remote node sees the same OSC as alt-syntax. - auto context_for_remote = context; - const auto object_storage_cluster_name = getClusterName(context); - if (!object_storage_cluster_name.empty() - && context->getSettingsRef()[Setting::object_storage_cluster].value.empty()) - { - auto ctx = Context::createCopy(context); - ctx->setSetting("object_storage_cluster", object_storage_cluster_name); - context_for_remote = ctx; - } + auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; - auto storage_and_context = convertToRemote(resolved.remote_initiator_cluster, context_for_remote, remote_initiator_cluster_name, query_to_send); - auto src_distributed = std::dynamic_pointer_cast(storage_and_context.storage); - auto modified_query_info = query_info; - modified_query_info.cluster = src_distributed->getCluster(); - auto new_storage_snapshot = storage_and_context.storage->getStorageSnapshot(storage_snapshot->metadata, storage_and_context.context); - storage_and_context.storage->read(query_plan, column_names, new_storage_snapshot, modified_query_info, storage_and_context.context, processed_stage, max_block_size, num_streams); - return; + /// rewrite query to execute `remote('remote_host', s3(...))` + /// remote_host can execute query itself or make on-cluster query depends on own `object_storage_cluster` setting + updateConfigurationIfNeeded(context); + updateQueryWithJoinToSendIfNeeded(query_to_send, query_info, context); + updateQueryToSendIfNeeded(query_to_send, storage_snapshot, context, /*make_cluster_function*/ false); + + /// ENGINE tables keep object_storage_cluster on the storage, not in the initiator query context. + /// Propagate it into the context copied for remote() so the remote node sees the same OSC as alt-syntax. + auto context_for_remote = context; + if (!cluster_name_from_settings.empty() + && context->getSettingsRef()[Setting::object_storage_cluster].value.empty()) + { + auto ctx = Context::createCopy(context); + ctx->setSetting("object_storage_cluster", cluster_name_from_settings); + context_for_remote = ctx; } - readFallBackToPure(query_plan, column_names, storage_snapshot, query_info, context, processed_stage, max_block_size, num_streams); + readFromRemoteInitiator( + query_plan, + column_names, + storage_snapshot, + query_info, + context_for_remote, + processed_stage, + max_block_size, + num_streams, + query_to_send, + resolved.remote_initiator_cluster, + remote_initiator_cluster_name); return; } @@ -530,12 +565,19 @@ void IStorageCluster::read( ClusterPtr remote_initiator_cluster = resolved.remote_initiator_cluster; if (!remote_initiator_cluster) remote_initiator_cluster = getClusterImpl(context, remote_initiator_cluster_name); - auto storage_and_context = convertToRemote(remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send); - auto src_distributed = std::dynamic_pointer_cast(storage_and_context.storage); - auto modified_query_info = query_info; - modified_query_info.cluster = src_distributed->getCluster(); - auto new_storage_snapshot = storage_and_context.storage->getStorageSnapshot(storage_snapshot->metadata, storage_and_context.context); - storage_and_context.storage->read(query_plan, column_names, new_storage_snapshot, modified_query_info, storage_and_context.context, processed_stage, max_block_size, num_streams); + + readFromRemoteInitiator( + query_plan, + column_names, + storage_snapshot, + query_info, + context, + processed_stage, + max_block_size, + num_streams, + query_to_send, + remote_initiator_cluster, + remote_initiator_cluster_name); return; } diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 8540d8a56fe2..6f0dccd7721e 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -113,6 +113,8 @@ class IStorageCluster : public IStorage { /// True when read() should use readFallBackToPure() or remote-initiator fallback branch. bool fallback_to_pure = false; + /// Cached shouldFallbackToLocalOnEmptyCluster(context). + bool local_fallback = false; /// Pre-resolved object-storage cluster when local-fallback prefetch was done. ClusterPtr object_storage_cluster; /// Resolved remote-initiator cluster when object_storage_remote_initiator_cluster is set. @@ -121,15 +123,28 @@ class IStorageCluster : public IStorage ResolvedClusterRead resolveClusterRead(ContextPtr context) const; - /// Storage policy: may apply object_storage_cluster_fallback_to_local_if_empty. - /// True for alternative syntax / table engine settings; false for explicit *Cluster(...). - virtual bool allowsLocalFallbackOnEmptyObjectStorageCluster(ContextPtr /* context */) const { return false; } + /// True for alternative syntax / table engines (object_storage_cluster setting path). + /// False for explicit *Cluster(...) and for non-object-storage IStorageCluster subclasses. + virtual bool usesObjectStorageClusterSettingSyntax() const { return false; } - /// Setting enabled and storage allows local fallback on empty/unknown object_storage_cluster. + /// Setting enabled, setting-syntax storage, and non-empty object_storage_cluster. bool shouldFallbackToLocalOnEmptyCluster(ContextPtr context) const; - /// True for non-*Cluster forms (alternative syntax and table engines): send pure s3()/iceberg() to the remote initiator. - virtual bool usePureFunctionForRemoteInitiator(ContextPtr /* context */) const { return false; } + /// Shared by read() and getQueryProcessingStage: local pure path vs throw vs remote pure send. + bool shouldReadLocallyOnFallbackToPure(const ResolvedClusterRead & resolved, ContextPtr context) const; + + void readFromRemoteInitiator( + QueryPlan & query_plan, + const Names & column_names, + const StorageSnapshotPtr & storage_snapshot, + SelectQueryInfo & query_info, + ContextPtr context, + QueryProcessingStage::Enum processed_stage, + size_t max_block_size, + size_t num_streams, + ASTPtr query_to_send, + ClusterPtr remote_initiator_cluster, + const String & remote_initiator_cluster_name); private: // With 'allow_null=true' returns nullptr when cluster does not exist or empty diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 0fadd8b20053..8a96ed9b75ac 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -727,16 +727,6 @@ String StorageObjectStorageCluster::getClusterName(ContextPtr context) const return getOriginalClusterName(); } -bool StorageObjectStorageCluster::allowsLocalFallbackOnEmptyObjectStorageCluster(ContextPtr context) const -{ - if (cluster_name_from_function_argument) - return false; - - /// Only when a non-empty object_storage_cluster was requested (query setting or table engine). - /// Empty OSC + remote_initiator without remote_initiator_cluster must keep BAD_ARGUMENTS. - return !getClusterName(context).empty(); -} - QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( ContextPtr context, QueryProcessingStage::Enum to_stage, const StorageSnapshotPtr & storage_snapshot, SelectQueryInfo & query_info) const { @@ -744,28 +734,9 @@ QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( return QueryProcessingStage::Enum::FetchColumns; auto resolved = resolveClusterRead(context); - const auto & settings = context->getSettingsRef(); - - if (resolved.fallback_to_pure) - { - if (settings[Setting::object_storage_remote_initiator]) - { - if (!resolved.remote_initiator_cluster) - { - if (!shouldFallbackToLocalOnEmptyCluster(context)) - throw Exception(ErrorCodes::BAD_ARGUMENTS, - "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments"); - - return QueryProcessingStage::Enum::FetchColumns; - } - } - else - { - return QueryProcessingStage::Enum::FetchColumns; - } - } + if (shouldReadLocallyOnFallbackToPure(resolved, context)) + return QueryProcessingStage::Enum::FetchColumns; - /// Distributed storage. return IStorageCluster::getQueryProcessingStage(context, to_stage, storage_snapshot, query_info); } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 7bfca12e1ad1..4ed91e12e476 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -201,10 +201,7 @@ class StorageObjectStorageCluster : public IStorageCluster ContextPtr context, bool async_insert) override; - bool allowsLocalFallbackOnEmptyObjectStorageCluster(ContextPtr context) const override; - - /// Pure s3()/iceberg() (not *Cluster) for remote initiator: alternative syntax and table engines. - bool usePureFunctionForRemoteInitiator(ContextPtr /* context */) const override { return !cluster_name_from_function_argument; } + bool usesObjectStorageClusterSettingSyntax() const override { return !cluster_name_from_function_argument; } /* In case the table was created with `object_storage_cluster` setting, diff --git a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp index 148cef62aff6..0853bccf1091 100644 --- a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp +++ b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp @@ -156,6 +156,7 @@ StoragePtr TableFunctionObjectStorageClusterFallback::executeI if (auto storage = typeid_cast>(result)) { storage->setClusterNameInSettings(true); + /// BaseCluster marks the storage as *Cluster; alternative syntax must clear that flag. storage->setClusterNameFromFunctionArgument(false); } return result; From e8b4414a110d52d258ed794be402fddd11a9ba02 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 17:28:48 +0200 Subject: [PATCH 18/18] Document why remote-initiator deferral keeps plain s3() with query SETTINGS. object_storage_cluster may be unknown locally and defined on the remote (or the reverse); *Cluster would bake the name into the function argument and skip remote fallback. Co-authored-by: Cursor --- src/Core/Settings.cpp | 2 ++ src/Storages/IStorageCluster.cpp | 19 ++++++++++++------- src/Storages/IStorageCluster.h | 2 +- .../StorageObjectStorageCluster.cpp | 10 ++++++---- .../StorageObjectStorageCluster.h | 5 +++-- tests/integration/test_s3_cluster/test.py | 3 +++ 6 files changed, 27 insertions(+), 14 deletions(-) diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 6f15fb2d247d..dfb82607bc75 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7776,6 +7776,8 @@ Cluster to make distributed requests to object storages with alternative syntax. DECLARE(Bool, object_storage_cluster_fallback_to_local_if_empty, false, R"( Execute the read locally if 'object_storage_cluster' is set but the cluster is empty or unknown. Does not apply to explicit *Cluster table functions, to 'object_storage_remote_initiator_cluster', or to writes. +With remote initiator + remote_initiator_cluster, the initiator sends plain s3()/iceberg() and passes +object_storage_cluster as a query setting so the remote can resolve or fall back (OSC may be unknown locally). )", EXPERIMENTAL) \ DECLARE(UInt64, object_storage_max_nodes, 0, R"( Limit for hosts used for request in object storage cluster table functions - azureBlobStorageCluster, s3Cluster, hdfsCluster, etc. diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index b5c9c588dcca..50c895515f67 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -378,8 +378,12 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context /// -> read locally (fallback_to_pure). /// - s3Cluster(...)/explicit *Cluster argument -> never fall back locally. /// - object_storage_remote_initiator=1 with object_storage_remote_initiator_cluster set - /// -> defer object_storage_cluster resolution; fallback_to_pure when the local cluster name is empty - /// or for non-*Cluster forms (pure s3()/iceberg()/ENGINE sent to the remote initiator). + /// -> defer object_storage_cluster resolution on the initiator. For setting-syntax / ENGINE, + /// always send plain s3()/iceberg() to the remote initiator with object_storage_cluster as a + /// query SETTINGS value (not *Cluster). OSC may be unknown locally and defined only on the + /// remote (or the reverse); *Cluster would bake the name into the function argument and + /// would also ignore object_storage_cluster_fallback_to_local_if_empty on the remote. + /// Empty local OSC is omitted so the remote can supply its own. /// - object_storage_remote_initiator=1 with only object_storage_cluster (no remote_initiator_cluster) /// -> clustered remote path: default initiator cluster to object_storage_cluster, send *Cluster. /// - writes -> never apply local fallback (see write()). @@ -395,12 +399,13 @@ IStorageCluster::ResolvedClusterRead IStorageCluster::resolveClusterRead(Context const auto & settings = context->getSettingsRef(); result.local_fallback = shouldFallbackToLocalOnEmptyCluster(context); - /// When both remote-initiator settings are set, object_storage_cluster may be defined only on the remote node. - /// In this case object_storage_cluster must not be resolved locally. + /// Defer OSC resolution when a separate remote-initiator cluster is set: the initiator must not + /// require a locally known object_storage_cluster (it may exist only on the remote node). const bool defer_object_storage_cluster_resolution = settings[Setting::object_storage_remote_initiator] && !settings[Setting::object_storage_remote_initiator_cluster].value.empty(); + /// Under deferral, setting-syntax / ENGINE always take the pure remote path (plain s3() + query SETTINGS). result.fallback_to_pure = cluster_name_from_settings.empty() || (defer_object_storage_cluster_resolution && usesObjectStorageClusterSettingSyntax()); @@ -493,14 +498,14 @@ void IStorageCluster::read( auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; - /// rewrite query to execute `remote('remote_host', s3(...))` - /// remote_host can execute query itself or make on-cluster query depends on own `object_storage_cluster` setting + /// Send remote('host', s3(...)) with object_storage_cluster in query SETTINGS (not s3Cluster). + /// The remote may resolve OSC, fall back locally, or use its own OSC when local OSC is empty. updateConfigurationIfNeeded(context); updateQueryWithJoinToSendIfNeeded(query_to_send, query_info, context); updateQueryToSendIfNeeded(query_to_send, storage_snapshot, context, /*make_cluster_function*/ false); /// ENGINE tables keep object_storage_cluster on the storage, not in the initiator query context. - /// Propagate it into the context copied for remote() so the remote node sees the same OSC as alt-syntax. + /// Propagate it into the context copied for remote() so the remote sees the same OSC as alt-syntax. auto context_for_remote = context; if (!cluster_name_from_settings.empty() && context->getSettingsRef()[Setting::object_storage_cluster].value.empty()) diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 6f0dccd7721e..9138a7f9f949 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -111,7 +111,7 @@ class IStorageCluster : public IStorage struct ResolvedClusterRead { - /// True when read() should use readFallBackToPure() or remote-initiator fallback branch. + /// True for local pure read, or for pure s3()/iceberg() sent to a remote initiator (not *Cluster). bool fallback_to_pure = false; /// Cached shouldFallbackToLocalOnEmptyCluster(context). bool local_fallback = false; diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 8a96ed9b75ac..5e9ecd909529 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -407,8 +407,8 @@ bool StorageObjectStorageCluster::updateQueryForDistributedEngineIfNeeded(ASTPtr return false; } - /// Inject OSC into SETTINGS for both pure and *Cluster rewrite paths. - /// Pure send must preserve ENGINE object_storage_cluster so the remote can distribute or fall back. + /// Inject OSC into SELECT SETTINGS (query setting, not a table-function argument). + /// On the pure remote-initiator path this preserves ENGINE OSC so the remote can distribute or fall back. auto settings = select_query->settings(); if (settings) { @@ -477,8 +477,10 @@ void StorageObjectStorageCluster::updateQueryToSendIfNeeded( if (make_cluster_function) { - /// Convert to old-stype *Cluster table function. - /// This allows to use old clickhouse versions in cluster. + /// Convert to *Cluster for the non-deferred clustered path (initiator belongs to the + /// object-storage cluster). Not used under remote_initiator_cluster deferral, which must + /// keep plain s3()/iceberg() with object_storage_cluster as a query setting. + /// *Cluster also helps older ClickHouse versions that lack the object_storage_cluster setting. static std::unordered_map function_to_cluster_function = { {"s3", "s3Cluster"}, {"azureBlobStorage", "azureBlobStorageCluster"}, diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 4ed91e12e476..864bb3a67781 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -206,14 +206,15 @@ class StorageObjectStorageCluster : public IStorageCluster /* In case the table was created with `object_storage_cluster` setting, modify the AST query object so that it uses the table function implementation - by mapping the engine name to table function name and setting `object_storage_cluster`. + by mapping the engine name to table function name and setting `object_storage_cluster` + as a query SETTINGS value (not a table-function argument). For table like CREATE TABLE table ENGINE=S3(...) SETTINGS object_storage_cluster='cluster' converts request SELECT * FROM table to SELECT * FROM s3(...) SETTINGS object_storage_cluster='cluster' - (and optionally to s3Cluster when make_cluster_function is true). + (and optionally to s3Cluster when make_cluster_function is true on the non-deferred path). Returns true if cluster name was added to settings. */ bool updateQueryForDistributedEngineIfNeeded(ASTPtr & query, ContextPtr context, bool make_cluster_function); diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index 80fdbad6d0cd..2d78d312aa62 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -331,6 +331,9 @@ def test_object_storage_cluster_fallback_to_local_if_empty(started_cluster): assert TSV(pure_sum) == TSV(fallback_sum) + # Asymmetric OSC: cluster name exists only on the remote initiator nodes (hidden_clusters.xml). + # Initiator must send plain s3() with object_storage_cluster as a query setting (not s3Cluster), + # so the remote can resolve the swarm even though the local node does not know that cluster. query_id = uuid.uuid4().hex result = node.query( f"""