Merge pull request #9673 from azat/processors-max_distributed_connections

alexey-milovidov · web-flow · commit 9a2760d9143e · 2020-03-24T08:32:15.000+03:00
Fix max_distributed_connections
diff --git a/dbms/programs/server/TCPHandler.cpp b/dbms/programs/server/TCPHandler.cpp
@@ -265,7 +265,7 @@ void TCPHandler::runImpl()
                 state.io.onFinish();
             }
             else if (state.io.pipeline.initialized())
-                processOrdinaryQueryWithProcessors(query_context->getSettingsRef().max_threads);
+                processOrdinaryQueryWithProcessors();
             else
                 processOrdinaryQuery();
 
@@ -544,13 +544,10 @@ void TCPHandler::processOrdinaryQuery()
 }
 
 
-void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)
+void TCPHandler::processOrdinaryQueryWithProcessors()
 {
     auto & pipeline = state.io.pipeline;
 
-    /// Reduce the number of threads to recommended value.
-    num_threads = std::min(num_threads, pipeline.getNumThreads());
-
     /// Send header-block, to allow client to prepare output format for data to send.
     {
         auto & header = pipeline.getHeader();
@@ -585,7 +582,7 @@ void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)
 
             try
             {
-                executor->execute(num_threads);
+                executor->execute(pipeline.getNumThreads());
             }
             catch (...)
             {
diff --git a/dbms/programs/server/TCPHandler.h b/dbms/programs/server/TCPHandler.h
@@ -171,7 +171,7 @@ class TCPHandler : public Poco::Net::TCPServerConnection
     /// Process a request that does not require the receiving of data blocks from the client
     void processOrdinaryQuery();
 
-    void processOrdinaryQueryWithProcessors(size_t num_threads);
+    void processOrdinaryQueryWithProcessors();
 
     void processTablesStatusRequest();
 
diff --git a/dbms/src/Interpreters/InterpreterSelectQuery.cpp b/dbms/src/Interpreters/InterpreterSelectQuery.cpp
@@ -467,7 +467,6 @@ QueryPipeline InterpreterSelectQuery::executeWithProcessors()
 {
     QueryPipeline query_pipeline;
     executeImpl(query_pipeline, input, std::move(input_pipe), query_pipeline);
-    query_pipeline.setMaxThreads(max_streams);
     query_pipeline.addInterpreterContext(context);
     query_pipeline.addStorageHolder(storage);
     return query_pipeline;
@@ -1301,6 +1300,7 @@ void InterpreterSelectQuery::executeFetchColumns(
     {
         is_remote = true;
         max_streams = settings.max_distributed_connections;
+        pipeline.setMaxThreads(max_streams);
     }
 
     UInt64 max_block_size = settings.max_block_size;
@@ -1325,6 +1325,7 @@ void InterpreterSelectQuery::executeFetchColumns(
     {
         max_block_size = std::max(UInt64(1), limit_length + limit_offset);
         max_streams = 1;
+        pipeline.setMaxThreads(max_streams);
     }
 
     if (!max_block_size)
diff --git a/dbms/src/Interpreters/InterpreterSelectQuery.h b/dbms/src/Interpreters/InterpreterSelectQuery.h
@@ -92,6 +92,8 @@ class InterpreterSelectQuery : public IInterpreter
 
     ASTPtr getQuery() const { return query_ptr; }
 
+    size_t getMaxStreams() const { return max_streams; }
+
 private:
     InterpreterSelectQuery(
         const ASTPtr & query_ptr_,
@@ -122,6 +124,9 @@ class InterpreterSelectQuery : public IInterpreter
         BlockInputStreamPtr stream_with_non_joined_data;
         bool union_stream = false;
 
+        /// Cache value of InterpreterSelectQuery::max_streams
+        size_t max_threads = 1;
+
         BlockInputStreamPtr & firstStream() { return streams.at(0); }
 
         template <typename Transform>
@@ -147,6 +152,10 @@ class InterpreterSelectQuery : public IInterpreter
 
         bool hasDelayedStream() const { return stream_with_non_joined_data != nullptr; }
         bool initialized() const { return !streams.empty(); }
+
+        /// Compatibility with QueryPipeline (Processors)
+        void   setMaxThreads(size_t max_threads_) { max_threads = max_threads_; }
+        size_t getNumThreads() const { return max_threads; }
     };
 
     template <typename TPipeline>
diff --git a/dbms/src/Interpreters/InterpreterSelectWithUnionQuery.cpp b/dbms/src/Interpreters/InterpreterSelectWithUnionQuery.cpp
@@ -34,7 +34,8 @@ InterpreterSelectWithUnionQuery::InterpreterSelectWithUnionQuery(
     const Names & required_result_column_names)
     : options(options_),
     query_ptr(query_ptr_),
-    context(std::make_shared<Context>(context_))
+    context(std::make_shared<Context>(context_)),
+    max_streams(context->getSettingsRef().max_threads)
 {
     const auto & ast = query_ptr->as<ASTSelectWithUnionQuery &>();
 
@@ -196,14 +197,23 @@ BlockInputStreams InterpreterSelectWithUnionQuery::executeWithMultipleStreams(Qu
         parent_pipeline.addInterpreterContext(context);
     }
 
+    /// Update max_streams due to:
+    /// - max_distributed_connections for Distributed() engine
+    /// - max_streams_to_max_threads_ratio
+    ///
+    /// XXX: res.pipeline.getMaxThreads() cannot be used since it is capped to
+    ///      number of streams, which is empty for non-Processors case.
+    max_streams = (*std::min_element(nested_interpreters.begin(), nested_interpreters.end(), [](const auto &a, const auto &b)
+    {
+        return a->getMaxStreams() < b->getMaxStreams();
+    }))->getMaxStreams();
+
     return nested_streams;
 }
 
 
 BlockIO InterpreterSelectWithUnionQuery::execute()
 {
-    const Settings & settings = context->getSettingsRef();
-
     BlockIO res;
     BlockInputStreams nested_streams = executeWithMultipleStreams(res.pipeline);
     BlockInputStreamPtr result_stream;
@@ -219,7 +229,7 @@ BlockIO InterpreterSelectWithUnionQuery::execute()
     }
     else
     {
-        result_stream = std::make_shared<UnionBlockInputStream>(nested_streams, nullptr, settings.max_threads);
+        result_stream = std::make_shared<UnionBlockInputStream>(nested_streams, nullptr, max_streams);
         nested_streams.clear();
     }
 
diff --git a/dbms/src/Interpreters/InterpreterSelectWithUnionQuery.h b/dbms/src/Interpreters/InterpreterSelectWithUnionQuery.h
@@ -56,6 +56,8 @@ class InterpreterSelectWithUnionQuery : public IInterpreter
 
     Block result_header;
 
+    size_t max_streams = 1;
+
     static Block getCommonHeaderForUnion(const Blocks & headers);
 };
 
diff --git a/dbms/src/Interpreters/executeQuery.cpp b/dbms/src/Interpreters/executeQuery.cpp
@@ -751,7 +751,7 @@ void executeQuery(
 
             {
                 auto executor = pipeline.execute();
-                executor->execute(context.getSettingsRef().max_threads);
+                executor->execute(pipeline.getNumThreads());
             }
         }
     }
diff --git a/dbms/src/Storages/StorageDistributed.cpp b/dbms/src/Storages/StorageDistributed.cpp
@@ -427,7 +427,7 @@ Pipes StorageDistributed::read(
         if (!smaller_cluster)
         {
             LOG_DEBUG(log, "Reading from " << table_id.getNameForLogs() <<
-                           (has_sharding_key ? "" : "(no sharding key)") << ": "
+                           (has_sharding_key ? "" : " (no sharding key)") << ": "
                            "Unable to figure out irrelevant shards from WHERE/PREWHERE clauses - "
                            "the query will be sent to all shards of the cluster");
 
diff --git a/dbms/tests/queries/0_stateless/01085_max_distributed_connections.reference b/dbms/tests/queries/0_stateless/01085_max_distributed_connections.reference
@@ -0,0 +1,2 @@
+0
+0
diff --git a/dbms/tests/queries/0_stateless/01085_max_distributed_connections.sh b/dbms/tests/queries/0_stateless/01085_max_distributed_connections.sh
@@ -0,0 +1,13 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. $CURDIR/../shell_config.sh
+
+opts=(
+    --max_distributed_connections 2
+    --max_threads 1
+    --query "SELECT sleepEachRow(1) FROM remote('127.{2,3}', system.one)"
+)
+# 1.8 less then 2 seconds, but long enough to cover possible load peaks
+# "$@" left to pass manual options (like --experimental_use_processors 0) during manual testing
+timeout 1.8s $CLICKHOUSE_CLIENT "${opts[@]}" "$@"
diff --git a/dbms/tests/queries/0_stateless/01085_max_distributed_connections_http.reference b/dbms/tests/queries/0_stateless/01085_max_distributed_connections_http.reference
@@ -0,0 +1,2 @@
+0
+0
diff --git a/dbms/tests/queries/0_stateless/01085_max_distributed_connections_http.sh b/dbms/tests/queries/0_stateless/01085_max_distributed_connections_http.sh
@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. $CURDIR/../shell_config.sh
+
+query="SELECT sleepEachRow(1) FROM remote('127.{2,3}', system.one)"
+# 1.8 less then 2 seconds, but long enough to cover possible load peaks
+timeout 1.8s ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}&max_distributed_connections=2&max_threads=1" -d "$query"

Original file line number	Diff line number	Diff line change
`@@ -265,7 +265,7 @@ void TCPHandler::runImpl()`
`265`	`265`	`state.io.onFinish();`
`266`	`266`	`}`
`267`	`267`	`else if (state.io.pipeline.initialized())`
`268`		`- processOrdinaryQueryWithProcessors(query_context->getSettingsRef().max_threads);`
	`268`	`+ processOrdinaryQueryWithProcessors();`
`269`	`269`	`else`
`270`	`270`	`processOrdinaryQuery();`
`271`	`271`
`@@ -544,13 +544,10 @@ void TCPHandler::processOrdinaryQuery()`
`544`	`544`	`}`
`545`	`545`
`546`	`546`
`547`		`-void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)`
	`547`	`+void TCPHandler::processOrdinaryQueryWithProcessors()`
`548`	`548`	`{`
`549`	`549`	`auto & pipeline = state.io.pipeline;`
`550`	`550`
`551`		`- /// Reduce the number of threads to recommended value.`
`552`		`- num_threads = std::min(num_threads, pipeline.getNumThreads());`
`553`		`-`
`554`	`551`	`/// Send header-block, to allow client to prepare output format for data to send.`
`555`	`552`	`{`
`556`	`553`	`auto & header = pipeline.getHeader();`
`@@ -585,7 +582,7 @@ void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)`
`585`	`582`
`586`	`583`	`try`
`587`	`584`	`{`
`588`		`- executor->execute(num_threads);`
	`585`	`+ executor->execute(pipeline.getNumThreads());`
`589`	`586`	`}`
`590`	`587`	`catch (...)`
`591`	`588`	`{`
Original file line number	Diff line number	Diff line change
`@@ -467,7 +467,6 @@ QueryPipeline InterpreterSelectQuery::executeWithProcessors()`
`467`	`467`	`{`
`468`	`468`	`QueryPipeline query_pipeline;`
`469`	`469`	`executeImpl(query_pipeline, input, std::move(input_pipe), query_pipeline);`
`470`		`- query_pipeline.setMaxThreads(max_streams);`
`471`	`470`	`query_pipeline.addInterpreterContext(context);`
`472`	`471`	`query_pipeline.addStorageHolder(storage);`
`473`	`472`	`return query_pipeline;`
`@@ -1301,6 +1300,7 @@ void InterpreterSelectQuery::executeFetchColumns(`
`1301`	`1300`	`{`
`1302`	`1301`	`is_remote = true;`
`1303`	`1302`	`max_streams = settings.max_distributed_connections;`
	`1303`	`+ pipeline.setMaxThreads(max_streams);`
`1304`	`1304`	`}`
`1305`	`1305`
`1306`	`1306`	`UInt64 max_block_size = settings.max_block_size;`
`@@ -1325,6 +1325,7 @@ void InterpreterSelectQuery::executeFetchColumns(`
`1325`	`1325`	`{`
`1326`	`1326`	`max_block_size = std::max(UInt64(1), limit_length + limit_offset);`
`1327`	`1327`	`max_streams = 1;`
	`1328`	`+ pipeline.setMaxThreads(max_streams);`
`1328`	`1329`	`}`
`1329`	`1330`
`1330`	`1331`	`if (!max_block_size)`
Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,8 @@ InterpreterSelectWithUnionQuery::InterpreterSelectWithUnionQuery(`
`34`	`34`	`const Names & required_result_column_names)`
`35`	`35`	`: options(options_),`
`36`	`36`	`query_ptr(query_ptr_),`
`37`		`- context(std::make_shared<Context>(context_))`
	`37`	`+ context(std::make_shared<Context>(context_)),`
	`38`	`+ max_streams(context->getSettingsRef().max_threads)`
`38`	`39`	`{`
`39`	`40`	`const auto & ast = query_ptr->as<ASTSelectWithUnionQuery &>();`
`40`	`41`
`@@ -196,14 +197,23 @@ BlockInputStreams InterpreterSelectWithUnionQuery::executeWithMultipleStreams(Qu`
`196`	`197`	`parent_pipeline.addInterpreterContext(context);`
`197`	`198`	`}`
`198`	`199`
	`200`	`+ /// Update max_streams due to:`
	`201`	`+ /// - max_distributed_connections for Distributed() engine`
	`202`	`+ /// - max_streams_to_max_threads_ratio`
	`203`	`+ ///`
	`204`	`+ /// XXX: res.pipeline.getMaxThreads() cannot be used since it is capped to`
	`205`	`+ /// number of streams, which is empty for non-Processors case.`
	`206`	`+ max_streams = (*std::min_element(nested_interpreters.begin(), nested_interpreters.end(), [](const auto &a, const auto &b)`
	`207`	`+ {`
	`208`	`+ return a->getMaxStreams() < b->getMaxStreams();`
	`209`	`+ }))->getMaxStreams();`
	`210`	`+`
`199`	`211`	`return nested_streams;`
`200`	`212`	`}`
`201`	`213`
`202`	`214`
`203`	`215`	`BlockIO InterpreterSelectWithUnionQuery::execute()`
`204`	`216`	`{`
`205`		`- const Settings & settings = context->getSettingsRef();`
`206`		`-`
`207`	`217`	`BlockIO res;`
`208`	`218`	`BlockInputStreams nested_streams = executeWithMultipleStreams(res.pipeline);`
`209`	`219`	`BlockInputStreamPtr result_stream;`
`@@ -219,7 +229,7 @@ BlockIO InterpreterSelectWithUnionQuery::execute()`
`219`	`229`	`}`
`220`	`230`	`else`
`221`	`231`	`{`
`222`		`- result_stream = std::make_shared<UnionBlockInputStream>(nested_streams, nullptr, settings.max_threads);`
	`232`	`+ result_stream = std::make_shared<UnionBlockInputStream>(nested_streams, nullptr, max_streams);`
`223`	`233`	`nested_streams.clear();`
`224`	`234`	`}`
`225`	`235`
Original file line number	Diff line number	Diff line change
`@@ -751,7 +751,7 @@ void executeQuery(`
`751`	`751`
`752`	`752`	`{`
`753`	`753`	`auto executor = pipeline.execute();`
`754`		`- executor->execute(context.getSettingsRef().max_threads);`
	`754`	`+ executor->execute(pipeline.getNumThreads());`
`755`	`755`	`}`
`756`	`756`	`}`
`757`	`757`	`}`
Original file line number	Diff line number	Diff line change
`@@ -427,7 +427,7 @@ Pipes StorageDistributed::read(`
`427`	`427`	`if (!smaller_cluster)`
`428`	`428`	`{`
`429`	`429`	`LOG_DEBUG(log, "Reading from " << table_id.getNameForLogs() <<`
`430`		`- (has_sharding_key ? "" : "(no sharding key)") << ": "`
	`430`	`+ (has_sharding_key ? "" : " (no sharding key)") << ": "`
`431`	`431`	`"Unable to figure out irrelevant shards from WHERE/PREWHERE clauses - "`
`432`	`432`	`"the query will be sent to all shards of the cluster");`
`433`	`433`