From 55dbb6da5e9339c8ef824bfae951d80dd339b3a2 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 7 Jul 2026 18:17:02 +0200 Subject: [PATCH 01/21] 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 bd4609af5963..9bcb6149400f 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -8301,6 +8301,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 37d38b0ef104..d9815ac9821e 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -95,6 +95,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"export_merge_tree_partition_retry_initial_backoff_seconds", 5, 5, "New setting for exponential back-off between failed part export retries in an export partition task"}, {"export_merge_tree_partition_retry_max_backoff_seconds", 300, 300, "New setting capping the exponential back-off between failed part export retries in an export partition task"}, {"export_merge_tree_partition_max_retries", 3, 3, "Obsolete and ignored: export partition tasks now retry retryable failures until the task timeout and fail immediately on non-retryable errors, instead of using a fixed retry budget"}, + {"object_storage_cluster_fallback_if_empty", false, false, "New setting"}, }); addSettingsChanges(settings_changes_history, "26.5", diff --git a/src/Interpreters/Cluster.cpp b/src/Interpreters/Cluster.cpp index 60c5bbed04c5..b952d41af98f 100644 --- a/src/Interpreters/Cluster.cpp +++ b/src/Interpreters/Cluster.cpp @@ -923,6 +923,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 2d707b51265f..575fa945441f 100644 --- a/src/Interpreters/Cluster.h +++ b/src/Interpreters/Cluster.h @@ -273,6 +273,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 3012c7bff735..622f0eff3464 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -64,6 +64,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 @@ -373,7 +374,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]) { @@ -448,7 +464,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)); @@ -743,9 +760,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 9613f9549562..73dfeee0984e 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -112,7 +112,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 fd6258c9c27bfd210e36d77ff3f8942801ce7628 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 8 Jul 2026 13:15:06 +0200 Subject: [PATCH 02/21] 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 622f0eff3464..b4566fa4b21c 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -394,9 +394,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 @@ -404,7 +422,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; @@ -592,6 +609,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 a62bcb0288553e523120c0c026e91f85404a10c8 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 8 Jul 2026 15:43:05 +0200 Subject: [PATCH 03/21] 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 b4566fa4b21c..19cacacdaf38 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -353,6 +353,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, @@ -374,55 +427,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 73dfeee0984e..3b9c3a5231c2 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -111,6 +111,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 9eefd709aba1..81ce58310b1d 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -45,6 +45,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 @@ -743,15 +744,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 377debfa832de7733293d4099611d44357749cdd Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 09:46:06 +0200 Subject: [PATCH 04/21] 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 00978fb7231c..75be1a2c1698 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -328,6 +328,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 ae94c7b00c2e56f349991edc4a0ddd1cddd21a1e Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 13:17:40 +0200 Subject: [PATCH 05/21] 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 75be1a2c1698..348659633039 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -401,6 +401,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 2fbb6b0540335dca377cc63ac5226102cdcb91ea Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 16:09:43 +0200 Subject: [PATCH 06/21] 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 19cacacdaf38..7f085a271f57 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -378,6 +378,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 3b9c3a5231c2..97ad95f8e5d1 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -123,6 +123,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 81ce58310b1d..7073df2feb0a 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -725,17 +725,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 6894bb76d2e1..b14e8a5a2fa7 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -198,6 +198,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 655212fd657cc6c08ec3e5208945dc89905041c8 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 19:53:20 +0200 Subject: [PATCH 07/21] 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 7f085a271f57..26f6264d042f 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -373,7 +373,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(); @@ -391,16 +391,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); } } @@ -431,7 +433,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 97ad95f8e5d1..8fa1bf0d514f 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -126,6 +126,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 7073df2feb0a..4e214df8cbd0 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -760,7 +760,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 b14e8a5a2fa7..f771dab0c9e1 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -200,6 +200,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 c094e5652eb50f3f2ed04f385151740526facbb3 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 9 Jul 2026 22:19:39 +0200 Subject: [PATCH 08/21] 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 4e214df8cbd0..fbccdcee06cf 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -727,12 +727,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; @@ -744,10 +744,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 f771dab0c9e1..3f55b633840d 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; @@ -221,6 +226,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 552a262a5ce2..63560aeeee0c 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 bc3d7237f134..ddf8995cbf48 100644 --- a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp +++ b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp @@ -155,7 +155,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 4f5173a5192351a4a84ae6bf2703ba482c8e03ea Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 19:10:02 +0200 Subject: [PATCH 09/21] Fix after review --- src/Core/Settings.cpp | 2 +- src/Core/SettingsChangesHistory.cpp | 1 + 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(+), 25 deletions(-) diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 9bcb6149400f..f9ed7c61f59d 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -8302,7 +8302,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 d9815ac9821e..874354ddc797 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -96,6 +96,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"export_merge_tree_partition_retry_max_backoff_seconds", 300, 300, "New setting capping the exponential back-off between failed part export retries in an export partition task"}, {"export_merge_tree_partition_max_retries", 3, 3, "Obsolete and ignored: export partition tasks now retry retryable failures until the task timeout and fail immediately on non-retryable errors, instead of using a fixed retry budget"}, {"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.5", diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index 26f6264d042f..56adf3f81253 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -64,7 +64,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 @@ -378,9 +378,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, @@ -396,7 +395,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, @@ -442,7 +441,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; @@ -646,7 +645,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 8fa1bf0d514f..942b7c51fd66 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -115,7 +115,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; @@ -123,7 +123,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 fbccdcee06cf..71da76138c47 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -45,7 +45,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 @@ -744,6 +744,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; @@ -770,7 +773,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 348659633039..ada32f4aa1df 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -328,7 +328,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( @@ -345,7 +345,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 """ ) @@ -365,7 +365,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 """ ) @@ -381,7 +381,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, ) @@ -420,7 +420,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, ) @@ -451,7 +451,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 6e7d2a2e95f6d2fb0a370b50fc035b7dd742a81e Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 19:39:50 +0200 Subject: [PATCH 10/21] 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 f9ed7c61f59d..f3abf195690b 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -8303,7 +8303,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 56adf3f81253..90a550e31c15 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -353,8 +353,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()) @@ -365,6 +378,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. @@ -377,9 +391,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, @@ -395,7 +412,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, @@ -441,7 +459,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 942b7c51fd66..1c6a82b98e24 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -115,7 +115,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; @@ -123,8 +123,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 71da76138c47..2aee00a69751 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -45,7 +45,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 @@ -742,11 +741,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; @@ -773,7 +769,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 3f55b633840d..b4603ea5a3b3 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -203,7 +203,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 66599a30544f0ccc3d878304b521dc434657d681 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 20:23:54 +0200 Subject: [PATCH 11/21] 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 90a550e31c15..9e2e145d7a93 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -361,13 +361,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()) @@ -450,8 +452,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 2aee00a69751..e824f9d85c51 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -760,8 +760,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 ada32f4aa1df..cb081b6c5338 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -1167,6 +1167,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 f8bcb906fda4adf036bab15eded19a0bc98d2675 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 20:29:43 +0200 Subject: [PATCH 12/21] 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 9e2e145d7a93..230e569b73cb 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -366,9 +366,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; @@ -389,7 +392,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 cb081b6c5338..8b76b8aaad4f 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -1734,6 +1734,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 bbe225cc01b3084813fdf5dffb25fe98aeddef84 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 20 Jul 2026 21:27:47 +0200 Subject: [PATCH 13/21] 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 230e569b73cb..f9b5157cc67d 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -366,8 +366,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 @@ -391,11 +391,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 @@ -419,8 +416,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, @@ -456,12 +452,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]) { @@ -533,7 +524,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 1c6a82b98e24..7289f468d25e 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -117,7 +117,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 e824f9d85c51..86f47366ddf1 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -760,12 +760,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 65c59feb128b89c0143e6ea3c51d93b75a77c382 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 10:00:13 +0200 Subject: [PATCH 14/21] 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 f3abf195690b..c68fcfa76d4c 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -8304,7 +8304,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 f9b5157cc67d..082c921dea5a 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -415,13 +415,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 8b76b8aaad4f..3c62736f32de 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -471,6 +471,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 e1b69bacbbfaf54af58c966b8a8fd093992bc392 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 10:52:37 +0200 Subject: [PATCH 15/21] 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 86f47366ddf1..a630325472a9 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -746,9 +746,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 3c62736f32de..10b7f62815d6 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -487,6 +487,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 0fe8c9959ad7928a827fe332800fa5b9e3c37b8b Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 11:22:36 +0200 Subject: [PATCH 16/21] 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 082c921dea5a..e0623bf5bd0d 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -63,6 +63,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; } @@ -367,11 +368,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; @@ -474,7 +473,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 7289f468d25e..011271fc7cee 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -130,7 +130,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 a630325472a9..85e403566ba7 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -398,19 +398,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 b4603ea5a3b3..e696ae5d8d7f 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -205,7 +205,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, @@ -213,11 +214,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 10b7f62815d6..07d7930d6ffb 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -471,6 +471,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""" @@ -521,6 +536,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 062fdcee4689244271e3dcb7212b306d48584eba Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 11:57:40 +0200 Subject: [PATCH 17/21] 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 e0623bf5bd0d..7b0d1ee2a65e 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -357,7 +357,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 @@ -382,7 +401,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. @@ -391,12 +410,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) { @@ -426,6 +445,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, @@ -452,49 +493,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; } @@ -538,12 +573,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 011271fc7cee..365d840fee3a 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -115,6 +115,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. @@ -123,15 +125,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 85e403566ba7..a569d52095f1 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -745,16 +745,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 { @@ -762,28 +752,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 e696ae5d8d7f..303ed5af4945 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -203,10 +203,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 ddf8995cbf48..414fdf416ae7 100644 --- a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp +++ b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp @@ -157,6 +157,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 99ba4a158f85f34b2aff5313066c5077cf44e157 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 21 Jul 2026 17:28:48 +0200 Subject: [PATCH 18/21] 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 c68fcfa76d4c..ded2a468856e 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -8305,6 +8305,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 7b0d1ee2a65e..ee84c71041c7 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -386,8 +386,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()). @@ -403,12 +407,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()); @@ -501,14 +506,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 365d840fee3a..56978724ba06 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -113,7 +113,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 a569d52095f1..4792276d1b9e 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -413,8 +413,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) { @@ -483,8 +483,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 303ed5af4945..344db23241ab 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -208,14 +208,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 07d7930d6ffb..76c3c5be63b9 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -371,6 +371,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""" From 3c5b7bbedbac85679626948f7fd97cce329477be Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Tue, 28 Jul 2026 15:57:27 +0200 Subject: [PATCH 19/21] Set cluster_name_in_settings for ENGINE/DataLake with object_storage_cluster. Align flag semantics with s3()/iceberg() alternative syntax when the cluster name comes from SETTINGS rather than a *Cluster argument. Co-authored-by: Cursor --- src/Databases/DataLake/DatabaseDataLake.cpp | 4 ++++ src/Storages/ObjectStorage/StorageObjectStorageCluster.h | 2 ++ src/Storages/ObjectStorage/registerStorageObjectStorage.cpp | 6 +++++- 3 files changed, 11 insertions(+), 1 deletion(-) diff --git a/src/Databases/DataLake/DatabaseDataLake.cpp b/src/Databases/DataLake/DatabaseDataLake.cpp index ba2fc89c7799..d8df6d2944b6 100644 --- a/src/Databases/DataLake/DatabaseDataLake.cpp +++ b/src/Databases/DataLake/DatabaseDataLake.cpp @@ -747,6 +747,7 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con const UUID table_uuid = catalog_uuid ? parseFromString(*catalog_uuid) : UUIDHelpers::Nil; std::string cluster_name = configuration->isClusterSupported() ? settings[DatabaseDataLakeSetting::object_storage_cluster].value : ""; + const bool cluster_name_from_object_storage_cluster_setting = !cluster_name.empty(); if (cluster_name.empty() && can_use_parallel_replicas && !is_secondary_query) cluster_name = parallel_replicas_cluster_name; @@ -775,6 +776,9 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con if (context_->hasQueryContext() && context_->getSettingsRef()[Setting::log_queries]) context_->getQueryContext()->addQueryFactoriesInfo(Context::QueryLogFactories::Storage, storage_cluster->getName()); + if (cluster_name_from_object_storage_cluster_setting) + storage_cluster->setClusterNameInSettings(true); + storage_cluster->startup(); return storage_cluster; } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 344db23241ab..5338a635c36e 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -224,6 +224,8 @@ class StorageObjectStorageCluster : public IStorageCluster const String engine_name; StorageObjectStorageConfigurationPtr configuration; const ObjectStoragePtr object_storage; + /// True when cluster name comes from SETTINGS (s3()/iceberg() alternative syntax, or + /// ENGINE ... SETTINGS object_storage_cluster / DataLake database setting), not from *Cluster arg. bool cluster_name_in_settings; bool cluster_name_from_function_argument = false; diff --git a/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp b/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp index 4b8276c8c508..d7558e1bcc41 100644 --- a/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp @@ -96,7 +96,7 @@ createStorageObjectStorage(const StorageFactory::Arguments & args, StorageObject /// `{_partition_id}` tables as wildcard (see `initPartitionStrategy`). configuration->is_create_query = args.mode == LoadingStrictnessLevel::CREATE; - return std::make_shared( + auto storage = std::make_shared( cluster_name, configuration, // We only want to perform write actions (e.g. create a container in Azure) when the table is being created, @@ -116,6 +116,10 @@ createStorageObjectStorage(const StorageFactory::Arguments & args, StorageObject /* is_datalake_query */ false, /* is_table_function */ false, /* lazy_init */ false); + /// Same as s3(...)/iceberg() alternative syntax: cluster name comes from SETTINGS, not *Cluster argument. + if (!cluster_name.empty()) + storage->setClusterNameInSettings(true); + return storage; } #endif From 24725668bcf6cf39feaca926590500585a0659a2 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Wed, 29 Jul 2026 13:33:52 +0200 Subject: [PATCH 20/21] Drop redundant cluster_name_in_settings from StorageObjectStorageCluster. AST rewrite and setting-syntax checks only need cluster_name_from_function_argument. Co-authored-by: Cursor --- src/Databases/DataLake/DatabaseDataLake.cpp | 4 ---- .../ObjectStorage/StorageObjectStorageCluster.cpp | 8 ++++---- src/Storages/ObjectStorage/StorageObjectStorageCluster.h | 5 +---- .../ObjectStorage/registerStorageObjectStorage.cpp | 6 +----- .../TableFunctionObjectStorageClusterFallback.cpp | 1 - 5 files changed, 6 insertions(+), 18 deletions(-) diff --git a/src/Databases/DataLake/DatabaseDataLake.cpp b/src/Databases/DataLake/DatabaseDataLake.cpp index d8df6d2944b6..ba2fc89c7799 100644 --- a/src/Databases/DataLake/DatabaseDataLake.cpp +++ b/src/Databases/DataLake/DatabaseDataLake.cpp @@ -747,7 +747,6 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con const UUID table_uuid = catalog_uuid ? parseFromString(*catalog_uuid) : UUIDHelpers::Nil; std::string cluster_name = configuration->isClusterSupported() ? settings[DatabaseDataLakeSetting::object_storage_cluster].value : ""; - const bool cluster_name_from_object_storage_cluster_setting = !cluster_name.empty(); if (cluster_name.empty() && can_use_parallel_replicas && !is_secondary_query) cluster_name = parallel_replicas_cluster_name; @@ -776,9 +775,6 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con if (context_->hasQueryContext() && context_->getSettingsRef()[Setting::log_queries]) context_->getQueryContext()->addQueryFactoriesInfo(Context::QueryLogFactories::Storage, storage_cluster->getName()); - if (cluster_name_from_object_storage_cluster_setting) - storage_cluster->setClusterNameInSettings(true); - storage_cluster->startup(); return storage_cluster; } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 4792276d1b9e..348a255a127c 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -117,7 +117,6 @@ StorageObjectStorageCluster::StorageObjectStorageCluster( cluster_name_, table_id_, getLogger(fmt::format("{}({})", configuration_->getEngineName(), table_id_.table_name))) , configuration{configuration_} , object_storage(object_storage_) - , cluster_name_in_settings(false) { configuration->initPartitionStrategy(partition_by, columns_in_table_or_function_definition, context_); @@ -438,7 +437,7 @@ void StorageObjectStorageCluster::updateQueryToSendIfNeeded( const ContextPtr & context, bool make_cluster_function) { - bool cluster_name_added_to_settings = updateQueryForDistributedEngineIfNeeded(query, context, make_cluster_function); + updateQueryForDistributedEngineIfNeeded(query, context, make_cluster_function); auto * table_function = extractTableFunctionFromSelectQuery(query); if (!table_function) @@ -463,7 +462,8 @@ void StorageObjectStorageCluster::updateQueryToSendIfNeeded( } ASTPtr object_storage_type_arg; - configuration->extractDynamicStorageType(args, context, &object_storage_type_arg, !cluster_name_in_settings && !cluster_name_added_to_settings); + configuration->extractDynamicStorageType( + args, context, &object_storage_type_arg, cluster_name_from_function_argument); ASTPtr settings_temporary_storage = nullptr; for (auto it = args.begin(); it != args.end(); ++it) @@ -477,7 +477,7 @@ void StorageObjectStorageCluster::updateQueryToSendIfNeeded( } } - if (cluster_name_in_settings || cluster_name_added_to_settings || !endsWith(table_function->name, "Cluster")) + if (!cluster_name_from_function_argument || !endsWith(table_function->name, "Cluster")) { configuration->addStructureAndFormatToArgsIfNeeded(args, structure, configuration->getFormat(), context, /*with_structure=*/true); diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 5338a635c36e..4b78d628e2e1 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -62,7 +62,6 @@ class StorageObjectStorageCluster : public IStorageCluster std::optional totalRows(ContextPtr query_context) const override; 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_) { @@ -224,9 +223,7 @@ class StorageObjectStorageCluster : public IStorageCluster const String engine_name; StorageObjectStorageConfigurationPtr configuration; const ObjectStoragePtr object_storage; - /// True when cluster name comes from SETTINGS (s3()/iceberg() alternative syntax, or - /// ENGINE ... SETTINGS object_storage_cluster / DataLake database setting), not from *Cluster arg. - bool cluster_name_in_settings; + /// True for explicit *Cluster(...); false for SETTINGS / ENGINE / DataLake paths. bool cluster_name_from_function_argument = false; /// non-clustered storage to fall back on pure realisation if needed diff --git a/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp b/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp index d7558e1bcc41..4b8276c8c508 100644 --- a/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp @@ -96,7 +96,7 @@ createStorageObjectStorage(const StorageFactory::Arguments & args, StorageObject /// `{_partition_id}` tables as wildcard (see `initPartitionStrategy`). configuration->is_create_query = args.mode == LoadingStrictnessLevel::CREATE; - auto storage = std::make_shared( + return std::make_shared( cluster_name, configuration, // We only want to perform write actions (e.g. create a container in Azure) when the table is being created, @@ -116,10 +116,6 @@ createStorageObjectStorage(const StorageFactory::Arguments & args, StorageObject /* is_datalake_query */ false, /* is_table_function */ false, /* lazy_init */ false); - /// Same as s3(...)/iceberg() alternative syntax: cluster name comes from SETTINGS, not *Cluster argument. - if (!cluster_name.empty()) - storage->setClusterNameInSettings(true); - return storage; } #endif diff --git a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp index 414fdf416ae7..6622a94c4c01 100644 --- a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp +++ b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp @@ -156,7 +156,6 @@ 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); /// BaseCluster marks the storage as *Cluster; alternative syntax must clear that flag. storage->setClusterNameFromFunctionArgument(false); } From 4cb369d4cc54275fef1f2e5aca10d24f4c991524 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Mon, 17 Aug 2026 18:35:34 +0200 Subject: [PATCH 21/21] Fix settings history --- src/Core/SettingsChangesHistory.cpp | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index 874354ddc797..417c9c585861 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -39,6 +39,11 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() /// controls new feature and it's 'true' by default, use 'false' as previous_value). /// It's used to implement `compatibility` setting (see https://github.com/ClickHouse/ClickHouse/issues/35972) /// Note: please check if the key already exists to prevent duplicate entries. + addSettingsChanges(settings_changes_history, "26.6.1.20001.altinityantalya", + { + {"object_storage_cluster_fallback_to_local_if_empty", false, false, "New setting"}, + }); + addSettingsChanges(settings_changes_history, "26.6", { {"analyzer_compatibility_apply_final_to_all_joined_tables", true, false, "Fixed a bug in the analyzer where FINAL on the left-most table of a JOIN was incorrectly applied to the other joined tables as well. previous_value=true so `compatibility` with versions before 26.6 restores the old behavior."}, @@ -95,8 +100,6 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"export_merge_tree_partition_retry_initial_backoff_seconds", 5, 5, "New setting for exponential back-off between failed part export retries in an export partition task"}, {"export_merge_tree_partition_retry_max_backoff_seconds", 300, 300, "New setting capping the exponential back-off between failed part export retries in an export partition task"}, {"export_merge_tree_partition_max_retries", 3, 3, "Obsolete and ignored: export partition tasks now retry retryable failures until the task timeout and fail immediately on non-retryable errors, instead of using a fixed retry budget"}, - {"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.5",