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,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,
@@ -365,23 +419,33 @@ 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])
         {
-            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 +504,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 +629,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 +797,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,22 @@ 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:
-    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
@@ -725,15 +726,26 @@ QueryProcessingStage::Enum StorageObjectStorageCluster::getQueryProcessingStage(
     if (!isClusterSupported())
         return QueryProcessingStage::Enum::FetchColumns;
 
-    /// Full query if fall back to pure storage.
-    if (getClusterName(context).empty()  // Not cluster request
-        && context->getSettingsRef()[Setting::object_storage_remote_initiator_cluster].value.empty()) // Not request with remote initiator
+    auto resolved = resolveClusterRead(context);
+    const auto & settings = context->getSettingsRef();
+
+    if (resolved.fallback_to_pure)
     {
-        if (context->getSettingsRef()[Setting::object_storage_remote_initiator])
-            throw Exception(ErrorCodes::BAD_ARGUMENTS,
-                "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments");
+        if (settings[Setting::object_storage_remote_initiator])
+        {
+            if (!resolved.remote_initiator_cluster)
+            {
+                if (!settings[Setting::object_storage_cluster_fallback_if_empty])
+                    throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                        "Setting 'object_storage_remote_initiator' can be used only with 'object_storage_remote_initiator_cluster', 'object_storage_cluster', or cluster name in arguments");
 
-        return QueryProcessingStage::Enum::FetchColumns;
+                return QueryProcessingStage::Enum::FetchColumns;
+            }
+        }
+        else
+        {
+            return QueryProcessingStage::Enum::FetchColumns;
+        }
     }
 
     /// Distributed storage.

tests/integration/test_s3_cluster/test.py

@@ -288,6 +288,80 @@ def test_wrong_cluster(started_cluster):
     assert "not found" in error
 
 
+def test_object_storage_cluster_fallback_if_empty(started_cluster):
+    node = started_cluster.instances["s0_0_0"]
+
+    pure_s3 = node.query(
+        f"""
+    SELECT count(*) from s3(
+        'http://minio1:9001/root/data/{{clickhouse,database}}/*',
+        'minio', '{minio_secret_key}', 'CSV',
+        'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))')"""
+    )
+
+    fallback_s3 = node.query(
+        f"""
+    SELECT count(*) from s3(
+        'http://minio1:9001/root/data/{{clickhouse,database}}/*',
+        'minio', '{minio_secret_key}', 'CSV',
+        'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))')
+    SETTINGS object_storage_cluster = 'non_existing_cluster', object_storage_cluster_fallback_if_empty = 1
+    """
+    )
+
+    assert TSV(pure_s3) == TSV(fallback_s3)
+
+    pure_sum = node.query(
+        f"""
+    SELECT sum(value) from s3(
+        'http://minio1:9001/root/data/{{clickhouse,database}}/*',
+        'minio', '{minio_secret_key}', 'CSV',
+        'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))')"""
+    )
+
+    fallback_sum = node.query(
+        f"""
+    SELECT sum(value) from s3(
+        'http://minio1:9001/root/data/{{clickhouse,database}}/*',
+        'minio', '{minio_secret_key}', 'CSV',
+        'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))')
+    SETTINGS object_storage_cluster = 'non_existing_cluster', object_storage_cluster_fallback_if_empty = 1
+    """
+    )
+
+    assert TSV(pure_sum) == TSV(fallback_sum)
+
+    query_id = uuid.uuid4().hex
+    result = node.query(
+        f"""
+        SELECT * from s3(
+            'http://minio1:9001/root/data/{{clickhouse,database}}/*', 'minio', '{minio_secret_key}', 'CSV',
+            'name String, value UInt32, polygon Array(Array(Tuple(Float64, Float64)))') ORDER BY (name, value, polygon)
+        SETTINGS
+            object_storage_remote_initiator=1,
+            object_storage_cluster='hidden_cluster_with_username_and_password',
+            object_storage_remote_initiator_cluster='cluster_with_dots',
+            object_storage_cluster_fallback_if_empty=1
+        """,
+        query_id=query_id,
+    )
+
+    assert result is not None
+
+    node.query("SYSTEM FLUSH LOGS ON CLUSTER 'cluster_all'")
+    queries = node.query(
+        f"""
+        SELECT count()
+            FROM clusterAllReplicas('cluster_all', system.query_log)
+            WHERE type='QueryFinish' AND initial_query_id='{query_id}'
+            FORMAT TSV
+        """
+    ).splitlines()
+
+    # initial node + remote initiator + 2 subqueries on replicas
+    assert queries == ["4"]
+
+
 def test_ambiguous_join(started_cluster):
     node = started_cluster.instances["s0_0_0"]
     result = node.query(

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

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');