Merge pull request ClickHouse#79599 from korowa/fix-remote-executor-pool-mode

Fix read from remote without hedged requests

Author: azat

src/Client/HedgedConnections.cpp

@@ -18,7 +18,6 @@ namespace Setting
 {
     extern const SettingsBool allow_changing_replica_until_first_data_packet;
     extern const SettingsBool allow_experimental_analyzer;
-    extern const SettingsUInt64 allow_experimental_parallel_reading_from_replicas;
     extern const SettingsUInt64 connections_with_failover_max_tries;
     extern const SettingsDialect dialect;
     extern const SettingsBool fallback_to_stale_replicas_for_distributed_queries;
@@ -54,9 +53,7 @@ HedgedConnections::HedgedConnections(
           timeouts_,
           context_->getSettingsRef()[Setting::connections_with_failover_max_tries].value,
           context_->getSettingsRef()[Setting::fallback_to_stale_replicas_for_distributed_queries].value,
-          context_->getSettingsRef()[Setting::allow_experimental_parallel_reading_from_replicas].value > 0
-            ? context_->getSettingsRef()[Setting::max_parallel_replicas].value
-            : 1,
+          context_->getSettingsRef()[Setting::max_parallel_replicas].value,
           context_->getSettingsRef()[Setting::skip_unavailable_shards].value,
           table_to_check_,
           priority_func)

src/Processors/QueryPlan/ReadFromRemote.cpp

@@ -50,6 +50,7 @@ namespace DB
 {
 namespace Setting
 {
+    extern const SettingsUInt64 allow_experimental_parallel_reading_from_replicas;
     extern const SettingsBool async_query_sending_for_remote;
     extern const SettingsBool async_socket_for_remote;
     extern const SettingsString cluster_for_parallel_replicas;
@@ -600,6 +601,7 @@ void ReadFromRemote::addPipe(Pipes & pipes, const ClusterProxy::SelectStreamFact
     bool add_extremes = false;
     bool async_read = context->getSettingsRef()[Setting::async_socket_for_remote];
     bool async_query_sending = context->getSettingsRef()[Setting::async_query_sending_for_remote];
+    bool parallel_replicas_disabled = context->getSettingsRef()[Setting::allow_experimental_parallel_reading_from_replicas] == 0;
     if (stage == QueryProcessingStage::Complete)
     {
         if (const auto * ast_select = shard.query->as<ASTSelectQuery>())
@@ -690,14 +692,16 @@ void ReadFromRemote::addPipe(Pipes & pipes, const ClusterProxy::SelectStreamFact
             shard.shard_info.pool, query_string, shard.header, context, throttler, scalars, external_tables, stage_to_use, shard.query_plan);
         remote_query_executor->setLogger(log);
 
-        if (context->canUseTaskBasedParallelReplicas())
+        if (context->canUseTaskBasedParallelReplicas() || parallel_replicas_disabled)
         {
             // when doing parallel reading from replicas (ParallelReplicasMode::READ_TASKS) on a shard:
             // establish a connection to a replica on the shard, the replica will instantiate coordinator to manage parallel reading from replicas on the shard.
             // The coordinator will return query result from the shard.
             // Only one coordinator per shard is necessary. Therefore using PoolMode::GET_ONE to establish only one connection per shard.
             // Using PoolMode::GET_MANY for this mode will(can) lead to instantiation of several coordinators (depends on max_parallel_replicas setting)
             // each will execute parallel reading from replicas, so the query result will be multiplied by the number of created coordinators
+            //
+            // In case parallel replicas are disabled, there also should be a single connection to each shard to prevent result duplication
             remote_query_executor->setPoolMode(PoolMode::GET_ONE);
         }
         else

tests/queries/0_stateless/03444_distributed_wo_hedged_requests.reference

@@ -0,0 +1,3 @@
+remote() 1 shard: 100
+remote() 3 shards: 300
+Distributed: 100

tests/queries/0_stateless/03444_distributed_wo_hedged_requests.sql

@@ -0,0 +1,26 @@
+drop table if exists 03444_local;
+drop table if exists 03444_distr;
+
+create table 03444_local (id UInt32) engine = MergeTree order by id as select * from numbers(100);
+create table 03444_distr (id UInt32) engine = Distributed(test_cluster_one_shard_two_replicas, currentDatabase(), 03444_local);
+
+select 'remote() 1 shard: ' || count()
+from remote('127.0.0.1|127.0.0.2|127.0.0.3', currentDatabase(), 03444_local)
+settings prefer_localhost_replica = 0,
+         use_hedged_requests = 0,
+         max_parallel_replicas = 100;
+
+select 'remote() 3 shards: ' || count()
+from remote('127.0.0.1|127.0.0.2,127.0.0.3|127.0.0.4,127.0.0.5|127.0.0.6', currentDatabase(), 03444_local)
+settings prefer_localhost_replica = 0,
+         use_hedged_requests = 0,
+         max_parallel_replicas = 100;
+
+select 'Distributed: ' || count()
+from 03444_distr
+settings prefer_localhost_replica = 0,
+         use_hedged_requests = 0,
+         max_parallel_replicas = 100;
+
+drop table 03444_local;
+drop table 03444_distr;