Fix read from remote without hedged requests

In case of disabled parallel reading from replicas, hedged connections
will read from single replica, but when hedged requests are disabled,
the query will use `max_parallel_replicas`.

This patch enforces client to use single replica in case of disabled
parallel replicas reading with `PoolMode::GET_ONE`, and removes this
logic from `HedgedConnections`.

Author: korowa

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;