diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index a11dc2ef96ac..dfb82607bc75 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7772,6 +7772,12 @@ 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_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/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index ff4dea11f189..00277f58856c 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -44,6 +44,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"object_storage_cluster_join_mode", "allow", "allow", "New setting"}, {"export_merge_tree_partition_task_timeout_seconds", "3600", "86400", "Increase default value to make it more realistic"}, {"export_merge_tree_part_allow_lossy_cast", false, false, "New setting to gate lossy casts in EXPORT PART/PARTITION behind explicit acknowledgment"}, + {"object_storage_cluster_fallback_to_local_if_empty", false, false, "New setting"}, {"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"}, diff --git a/src/Interpreters/Cluster.cpp b/src/Interpreters/Cluster.cpp index 95f0559748fc..cd735a250e96 100644 --- a/src/Interpreters/Cluster.cpp +++ b/src/Interpreters/Cluster.cpp @@ -909,6 +909,14 @@ Cluster::Cluster(Cluster::SubclusterTag, const Cluster & from, const std::vector initMisc(); } +size_t Cluster::getAllNodeCount() const +{ + size_t count = 0; + for (const auto & shard : shards_info) + count += shard.getAllNodeCount(); + return count; +} + std::vector Cluster::getHostIDs() const { std::vector host_ids; diff --git a/src/Interpreters/Cluster.h b/src/Interpreters/Cluster.h index b20f989da2d2..6e3413c2bdb6 100644 --- a/src/Interpreters/Cluster.h +++ b/src/Interpreters/Cluster.h @@ -264,6 +264,9 @@ class Cluster /// The number of all shards. size_t getShardCount() const { return shards_info.size(); } + /// The number of all nodes. + size_t getAllNodeCount() const; + /// Returns an array of arrays of strings in the format 'escaped_host_name:port' for all replicas of all shards in the cluster. std::vector getHostIDs() const; diff --git a/src/Storages/IStorageCluster.cpp b/src/Storages/IStorageCluster.cpp index bcf2efaa3840..50c895515f67 100644 --- a/src/Storages/IStorageCluster.cpp +++ b/src/Storages/IStorageCluster.cpp @@ -56,7 +56,9 @@ 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; } namespace ErrorCodes @@ -344,6 +346,124 @@ void IStorageCluster::updateQueryWithJoinToSendIfNeeded( } } +bool IStorageCluster::shouldFallbackToLocalOnEmptyCluster(ContextPtr context) const +{ + return context->getSettingsRef()[Setting::object_storage_cluster_fallback_to_local_if_empty] + && 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 +{ + /// 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 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()). + ResolvedClusterRead result; + + if (!isClusterSupported()) + { + result.fallback_to_pure = true; + return result; + } + + auto cluster_name_from_settings = getClusterName(context); + const auto & settings = context->getSettingsRef(); + result.local_fallback = shouldFallbackToLocalOnEmptyCluster(context); + + /// 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()); + + const bool try_resolve_with_local_fallback + = !defer_object_storage_cluster_resolution + && !result.fallback_to_pure + && result.local_fallback; + + if (try_resolve_with_local_fallback) + { + 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 (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()) + { + /// 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); + } + } + + 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, @@ -365,32 +485,48 @@ void IStorageCluster::read( const auto & settings = context->getSettingsRef(); ASTPtr query_to_send = query_info.query; - if (cluster_name_from_settings.empty()) + auto resolved = resolveClusterRead(context); + ClusterPtr cluster = resolved.object_storage_cluster; + + if (resolved.fallback_to_pure) { - if (settings[Setting::object_storage_remote_initiator]) + if (shouldReadLocallyOnFallbackToPure(resolved, context)) { - auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; - if (remote_initiator_cluster_name.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"); - - /// 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 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); + readFallBackToPure(query_plan, column_names, storage_snapshot, query_info, context, processed_stage, max_block_size, num_streams); return; } - readFallBackToPure(query_plan, column_names, storage_snapshot, query_info, context, processed_stage, max_block_size, num_streams); + auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value; + + /// 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 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; + } + + 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; } @@ -430,17 +566,28 @@ 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); - 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); + /// 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); + + 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; } - 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)); @@ -564,6 +711,8 @@ SinkToStoragePtr IStorageCluster::write( { auto cluster_name_from_settings = getClusterName(context); + // 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); @@ -730,9 +879,18 @@ ContextPtr ReadFromCluster::updateSettings(const Settings & settings) return new_context; } -ClusterPtr IStorageCluster::getClusterImpl(ContextPtr context, const String & cluster_name_, size_t max_hosts) +ClusterPtr IStorageCluster::getClusterImpl(ContextPtr context, const String & cluster_name_, size_t max_hosts, bool allow_null) { - return context->getCluster(cluster_name_)->getClusterWithReplicasAsShards(context->getSettingsRef(), /* max_replicas_from_shard */ 0, max_hosts); + ClusterPtr cluster = nullptr; + if (allow_null) + { + cluster = context->tryGetCluster(cluster_name_); + if (!cluster || !cluster->getAllNodeCount()) + return nullptr; + } + else + cluster = context->getCluster(cluster_name_); + return cluster->getClusterWithReplicasAsShards(context->getSettingsRef(), /* max_replicas_from_shard */ 0, max_hosts); } } diff --git a/src/Storages/IStorageCluster.h b/src/Storages/IStorageCluster.h index 53ca742a994a..9138a7f9f949 100644 --- a/src/Storages/IStorageCluster.h +++ b/src/Storages/IStorageCluster.h @@ -109,8 +109,47 @@ class IStorageCluster : public IStorage NamesAndTypesList hive_partition_columns_to_read_from_file_path; + struct ResolvedClusterRead + { + /// 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; + /// 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. + ClusterPtr remote_initiator_cluster; + }; + + ResolvedClusterRead resolveClusterRead(ContextPtr context) const; + + /// 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, setting-syntax storage, and non-empty object_storage_cluster. + bool shouldFallbackToLocalOnEmptyCluster(ContextPtr context) const; + + /// 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: - 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; } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 17704602a57f..5e9ecd909529 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -392,19 +392,23 @@ bool StorageObjectStorageCluster::updateQueryForDistributedEngineIfNeeded(ASTPtr table_expression->table_function = function_ast_ptr; table_expression->children[0] = function_ast_ptr; - if (!make_cluster_function) - return false; - auto cluster_name = getClusterName(context); if (cluster_name.empty()) { - throw Exception( - ErrorCodes::LOGICAL_ERROR, - "Can't be here without cluster name, no cluster name in query {}", - query->formatForLogging()); + /// Pure remote path without a local object_storage_cluster (remote may define it). + if (make_cluster_function) + { + throw Exception( + ErrorCodes::LOGICAL_ERROR, + "Can't be here without cluster name, no cluster name in query {}", + query->formatForLogging()); + } + return false; } + /// Inject OSC into 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) { @@ -473,8 +477,10 @@ void StorageObjectStorageCluster::updateQueryToSendIfNeeded( if (make_cluster_function) { - /// Convert to old-stype *Cluster table function. - /// This allows to use old clickhouse versions in cluster. + /// Convert to *Cluster for the non-deferred clustered path (initiator belongs to the + /// object-storage cluster). Not used under remote_initiator_cluster deferral, which must + /// keep plain s3()/iceberg() with object_storage_cluster as a query setting. + /// *Cluster also helps older ClickHouse versions that lack the object_storage_cluster setting. static std::unordered_map function_to_cluster_function = { {"s3", "s3Cluster"}, {"azureBlobStorage", "azureBlobStorageCluster"}, @@ -706,17 +712,21 @@ 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; table engine and alternative-syntax use the setting path. if (!isClusterSupported()) return ""; + if (cluster_name_from_function_argument) + 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(); } QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage( @@ -725,18 +735,10 @@ 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 - { - 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"); - + auto resolved = resolveClusterRead(context); + 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 bcaf293ad4e7..864bb3a67781 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; @@ -196,17 +201,20 @@ class StorageObjectStorageCluster : public IStorageCluster ContextPtr context, bool async_insert) override; + bool usesObjectStorageClusterSettingSyntax() const override { return !cluster_name_from_function_argument; } + /* 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' - 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 on the non-deferred path). Returns true if cluster name was added to settings. */ bool updateQueryForDistributedEngineIfNeeded(ASTPtr & query, ContextPtr context, bool make_cluster_function); @@ -215,6 +223,7 @@ class StorageObjectStorageCluster : public IStorageCluster StorageObjectStorageConfigurationPtr configuration; const ObjectStoragePtr object_storage; bool cluster_name_in_settings; + bool cluster_name_from_function_argument = false; /// non-clustered storage to fall back on pure realisation if needed std::shared_ptr pure_storage; diff --git a/src/TableFunctions/TableFunctionObjectStorageCluster.cpp b/src/TableFunctions/TableFunctionObjectStorageCluster.cpp index 4f813e516f9a..21f217dc3c9b 100644 --- a/src/TableFunctions/TableFunctionObjectStorageCluster.cpp +++ b/src/TableFunctions/TableFunctionObjectStorageCluster.cpp @@ -90,6 +90,7 @@ StoragePtr TableFunctionObjectStorageCluster(storage)->setClusterNameFromFunctionArgument(true); } storage->startup(); diff --git a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp index 98d8dc43c85a..0853bccf1091 100644 --- a/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp +++ b/src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp @@ -154,7 +154,11 @@ 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); + } return result; } else diff --git a/tests/integration/test_s3_cluster/test.py b/tests/integration/test_s3_cluster/test.py index d4d91b923782..2d78d312aa62 100644 --- a/tests/integration/test_s3_cluster/test.py +++ b/tests/integration/test_s3_cluster/test.py @@ -288,6 +288,289 @@ def test_wrong_cluster(started_cluster): assert "not found" in error +def test_object_storage_cluster_fallback_to_local_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_to_local_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_to_local_if_empty = 1 + """ + ) + + 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""" + 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_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() + + # 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_to_local_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_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() + + # 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""" + 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() + + # 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) + + # 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") + + def test_ambiguous_join(started_cluster): node = started_cluster.instances["s0_0_0"] result = node.query( @@ -983,6 +1266,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( @@ -1463,6 +1812,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 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 new file mode 100644 index 000000000000..5682404d3d11 --- /dev/null +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.reference @@ -0,0 +1,30 @@ +pure +10 45 +case1 unknown OSC without fallback still errors +case1 unknown OSC with fallback runs locally +10 45 +valid OSC with fallback unchanged +10 45 +s3Cluster ignores fallback +s3Cluster ignores fallback even with OSC setting +s3Cluster with valid argument unchanged +10 45 +case2 unknown OSC + RI without RI-cluster with fallback runs locally +10 45 +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 +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 +285 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 new file mode 100644 index 000000000000..48841c4d5baf --- /dev/null +++ b/tests/queries/0_stateless/04303_object_storage_cluster_fallback_to_local_if_empty.sql @@ -0,0 +1,121 @@ +-- 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. +-- Covered for both s3() table function and S3 table engine. + +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 '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 '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 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 '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 '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 '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 '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 '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', + object_storage_cluster_fallback_to_local_if_empty = 1, + object_storage_remote_initiator = 1, + object_storage_remote_initiator_cluster = 'test_shard_localhost'; + +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', + 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 } + +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; + +SELECT 'pure aggregate'; +SELECT sum(x * x) FROM s3('http://localhost:11111/test/04303_object_storage_cluster_fallback.tsv', 'TSV', 'x UInt32');