Allow empty object storage cluster

src/Core/Settings.cpp

@@ -7766,6 +7766,9 @@ Trigger processor to spill data into external storage adpatively. grace join is
 )", EXPERIMENTAL) \
     DECLARE(String, object_storage_cluster, "", R"(
 Cluster to make distributed requests to object storages with alternative syntax.
+)", EXPERIMENTAL) \
+    DECLARE(Bool, object_storage_cluster_fallback_if_empty, false, R"(
+Use non-cluster request if 'object_storage_cluster' is set but empty or unknown.
 )", EXPERIMENTAL) \
     DECLARE(UInt64, object_storage_max_nodes, 0, R"(
 Limit for hosts used for request in object storage cluster table functions - azureBlobStorageCluster, s3Cluster, hdfsCluster, etc.

src/Core/SettingsChangesHistory.cpp

@@ -44,6 +44,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"object_storage_cluster_join_mode", "allow", "allow", "New setting"},
             {"export_merge_tree_partition_task_timeout_seconds", "3600", "86400", "Increase default value to make it more realistic"},
             {"export_merge_tree_part_allow_lossy_cast", false, false, "New setting to gate lossy casts in EXPORT PART/PARTITION behind explicit acknowledgment"},
+            {"object_storage_cluster_fallback_if_empty", false, false, "New setting"},
         });
         addSettingsChanges(settings_changes_history, "26.3",
         {

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<Strings> Cluster::getHostIDs() const
 {
     std::vector<Strings> host_ids;

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<Strings> getHostIDs() const;
 

src/Storages/IStorageCluster.cpp

@@ -57,6 +57,7 @@ namespace Setting
     extern const SettingsBool object_storage_remote_initiator;
     extern const SettingsString object_storage_remote_initiator_cluster;
     extern const SettingsObjectStorageClusterJoinMode object_storage_cluster_join_mode;
+    extern const SettingsBool object_storage_cluster_fallback_if_empty;
 }
 
 namespace ErrorCodes
@@ -344,6 +345,62 @@ 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 = usePureFunctionForRemoteInitiator(context);
+    else
+        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])
+    {
+        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())
+        {
+            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);
+        }
+    }
+
+    return result;
+}
+
 /// The code executes on initiator
 void IStorageCluster::read(
     QueryPlan & query_plan,
@@ -365,23 +422,36 @@ 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;
+
+    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])
         {
-            auto remote_initiator_cluster_name = settings[Setting::object_storage_remote_initiator_cluster].value;
-            if (remote_initiator_cluster_name.empty())
+            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;
+                }
                 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 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 storage_and_context = convertToRemote(resolved.remote_initiator_cluster, context, remote_initiator_cluster_name, query_to_send);
             auto src_distributed = std::dynamic_pointer_cast<StorageDistributed>(storage_and_context.storage);
             auto modified_query_info = query_info;
             modified_query_info.cluster = src_distributed->getCluster();
@@ -440,7 +510,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<ASTSelectQuery &>(), 0));
@@ -564,6 +635,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);
 
@@ -730,9 +803,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);
 }
 
 }

src/Storages/IStorageCluster.h

@@ -109,8 +109,28 @@ 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;
+
+    /// 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:
-    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; }
 

src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp

@@ -42,6 +42,7 @@ namespace Setting
     extern const SettingsInt64 delta_lake_snapshot_end_version;
     extern const SettingsUInt64 lock_object_storage_task_distribution_ms;
     extern const SettingsBool allow_experimental_iceberg_read_optimization;
+    extern const SettingsBool object_storage_cluster_fallback_if_empty;
 }
 
 namespace ErrorCodes
@@ -706,17 +707,31 @@ 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();
+}
+
+bool StorageObjectStorageCluster::useObjectStorageClusterFallbackIfEmpty(ContextPtr context) const
+{
+    if (cluster_name_from_function_argument)
+        return false;
+
+    return cluster_name_in_settings
+        || !getOriginalClusterName().empty()
+        || !context->getSettingsRef()[Setting::object_storage_cluster].value.empty();
 }
 
 QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage(
@@ -725,15 +740,29 @@ 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();
+
+    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 (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.

src/Storages/ObjectStorage/StorageObjectStorageCluster.h

@@ -64,6 +64,11 @@ class StorageObjectStorageCluster : public IStorageCluster
     std::optional<UInt64> 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,6 +201,10 @@ class StorageObjectStorageCluster : public IStorageCluster
         ContextPtr context,
         bool async_insert) override;
 
+    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
@@ -215,6 +224,7 @@ class StorageObjectStorageCluster : public IStorageCluster
     StorageObjectStorageConfigurationPtr configuration;
     const ObjectStoragePtr object_storage;
     bool cluster_name_in_settings;
+    bool cluster_name_from_function_argument = false;
 
     /// non-clustered storage to fall back on pure realisation if needed
     std::shared_ptr<StorageObjectStorage> pure_storage;

src/TableFunctions/TableFunctionObjectStorageCluster.cpp

@@ -90,6 +90,7 @@ StoragePtr TableFunctionObjectStorageCluster<Definition, Configuration, is_data_
             /* is_datalake_query*/ false,
             /* is_table_function */ true,
             /* lazy_init */ true);
+        std::static_pointer_cast<StorageObjectStorageCluster>(storage)->setClusterNameFromFunctionArgument(true);
     }
 
     storage->startup();

src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp

@@ -154,7 +154,10 @@ StoragePtr TableFunctionObjectStorageClusterFallback<Definition, Base>::executeI
     {
         auto result = BaseCluster::executeImpl(ast_function, context, table_name, cached_columns, is_insert_query);
         if (auto storage = typeid_cast<std::shared_ptr<StorageObjectStorageCluster>>(result))
+        {
             storage->setClusterNameInSettings(true);
+            storage->setClusterNameFromFunctionArgument(false);
+        }
         return result;
     }
     else