Better calculation for recommended number of threads to execute pipeline

dbms/programs/server/TCPHandler.cpp

@@ -546,8 +546,8 @@ void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)
 {
     auto & pipeline = state.io.pipeline;
 
-    if (pipeline.getMaxThreads())
-        num_threads = std::min(num_threads, pipeline.getMaxThreads());
+    /// 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.
     {

dbms/src/Processors/QueryPipeline.cpp

@@ -97,7 +97,7 @@ void QueryPipeline::init(Pipes pipes)
             totals.emplace_back(totals_port);
         }
 
-        streams.emplace_back(&pipe.getPort());
+        streams.addStream(&pipe.getPort());
         auto cur_processors = std::move(pipe).detachProcessors();
         processors.insert(processors.end(), cur_processors.begin(), cur_processors.end());
     }
@@ -225,7 +225,7 @@ void QueryPipeline::addPipe(Processors pipe)
     streams.reserve(last->getOutputs().size());
     for (auto & output : last->getOutputs())
     {
-        streams.emplace_back(&output);
+        streams.addStream(&output);
         if (header)
             assertBlocksHaveEqualStructure(header, output.getHeader(), "QueryPipeline");
         else
@@ -244,7 +244,7 @@ void QueryPipeline::addDelayedStream(ProcessorPtr source)
     assertBlocksHaveEqualStructure(current_header, source->getOutputs().front().getHeader(), "QueryPipeline");
 
     IProcessor::PortNumbers delayed_streams = { streams.size() };
-    streams.emplace_back(&source->getOutputs().front());
+    streams.addStream(&source->getOutputs().front());
     processors.emplace_back(std::move(source));
 
     auto processor = std::make_shared<DelayedPortsProcessor>(current_header, streams.size(), delayed_streams);
@@ -274,7 +274,7 @@ void QueryPipeline::resize(size_t num_streams, bool force, bool strict)
     streams.clear();
     streams.reserve(num_streams);
     for (auto & output : resize->getOutputs())
-        streams.emplace_back(&output);
+        streams.addStream(&output);
 
     processors.emplace_back(std::move(resize));
 }
@@ -302,7 +302,7 @@ void QueryPipeline::addTotalsHavingTransform(ProcessorPtr transform)
 
     auto & outputs = transform->getOutputs();
 
-    streams = { &outputs.front() };
+    streams.assign({ &outputs.front() });
     totals_having_port = &outputs.back();
     current_header = outputs.front().getHeader();
     processors.emplace_back(std::move(transform));
@@ -374,7 +374,7 @@ void QueryPipeline::addExtremesTransform(ProcessorPtr transform)
 
     auto & outputs = transform->getOutputs();
 
-    streams = { &outputs.front() };
+    streams.assign({ &outputs.front() });
     extremes_port = &outputs.back();
     current_header = outputs.front().getHeader();
     processors.emplace_back(std::move(transform));
@@ -394,7 +394,7 @@ void QueryPipeline::addCreatingSetsTransform(ProcessorPtr transform)
     connect(transform->getOutputs().front(), concat->getInputs().front());
     connect(*streams.back(), concat->getInputs().back());
 
-    streams = { &concat->getOutputs().front() };
+    streams.assign({ &concat->getOutputs().front() });
     processors.emplace_back(std::move(transform));
     processors.emplace_back(std::move(concat));
 }
@@ -490,7 +490,7 @@ void QueryPipeline::unitePipelines(
         }
 
         processors.insert(processors.end(), pipeline.processors.begin(), pipeline.processors.end());
-        streams.insert(streams.end(), pipeline.streams.begin(), pipeline.streams.end());
+        streams.addStreams(pipeline.streams);
 
         table_locks.insert(table_locks.end(), std::make_move_iterator(pipeline.table_locks.begin()), std::make_move_iterator(pipeline.table_locks.end()));
         interpreter_context.insert(interpreter_context.end(), pipeline.interpreter_context.begin(), pipeline.interpreter_context.end());

dbms/src/Processors/QueryPipeline.h

@@ -21,6 +21,53 @@ class IOutputFormat;
 
 class QueryPipeline
 {
+private:
+    /// It's a wrapper over std::vector<OutputPort *>
+    /// Is needed to support invariant for max_parallel_streams (see comment below).
+    class Streams
+    {
+    public:
+        auto size() const { return data.size(); }
+        auto begin() { return data.begin(); }
+        auto end() { return data.end(); }
+        auto & front() { return data.front(); }
+        auto & back() { return data.back(); }
+        auto & at(size_t pos) { return data.at(pos); }
+        auto & operator[](size_t pos) { return data[pos]; }
+
+        void clear() { data.clear(); }
+        void reserve(size_t size_) { data.reserve(size_); }
+
+        void addStream(OutputPort * port)
+        {
+            data.push_back(port);
+            max_parallel_streams = std::max<size_t>(max_parallel_streams, data.size());
+        }
+
+        void addStreams(Streams & other)
+        {
+            data.insert(data.end(), other.begin(), other.end());
+            max_parallel_streams = std::max<size_t>(max_parallel_streams, data.size());
+        }
+
+        void assign(std::initializer_list<OutputPort *> list)
+        {
+            data = list;
+            max_parallel_streams = std::max<size_t>(max_parallel_streams, data.size());
+        }
+
+        size_t maxParallelStreams() const { return max_parallel_streams; }
+
+    private:
+        std::vector<OutputPort *> data;
+
+        /// It is the max number of processors which can be executed in parallel for each step.
+        /// Logically, it is the upper limit on the number of threads needed to execute this pipeline.
+        /// Initially, it is the number of sources. It may be increased after resize, aggregation, etc.
+        /// This number is never decreased, and it is calculated as max(streams.size()) over all streams while building.
+        size_t max_parallel_streams = 0;
+    };
+
 public:
     QueryPipeline() = default;
     QueryPipeline(QueryPipeline &&) = default;
@@ -96,8 +143,19 @@ class QueryPipeline
     /// Call after execution.
     void finalize();
 
+    /// Recommend number of threads for pipeline execution.
+    size_t getNumThreads() const
+    {
+        auto num_threads = streams.maxParallelStreams();
+
+        if (max_threads)
+            num_threads = std::min(num_threads, max_threads);
+
+        return std::max<size_t>(1, num_threads);
+    }
+
+    /// Set upper limit for the recommend number of threads
     void setMaxThreads(size_t max_threads_) { max_threads = max_threads_; }
-    size_t getMaxThreads() const { return max_threads; }
 
     /// Convert query pipeline to single pipe.
     Pipe getPipe() &&;
@@ -119,7 +177,7 @@ class QueryPipeline
     Processors processors;
 
     /// Port for each independent "stream".
-    std::vector<OutputPort *> streams;
+    Streams streams;
 
     /// Special ports for extremes and totals having.
     OutputPort * totals_having_port = nullptr;
@@ -130,6 +188,8 @@ class QueryPipeline
 
     IOutputFormat * output_format = nullptr;
 
+    /// Limit on the number of threads. Zero means no limit.
+    /// Sometimes, more streams are created then the number of threads for more optimal execution.
     size_t max_threads = 0;
 
     QueryStatus * process_list_element = nullptr;

dbms/tests/queries/0_stateless/01091_num_threads.reference

@@ -0,0 +1,6 @@
+1
+1
+499999500000
+1
+499999500000
+1

dbms/tests/queries/0_stateless/01091_num_threads.sql

@@ -0,0 +1,47 @@
+set log_queries=1;
+set log_query_threads=1;
+
+SELECT 1;
+SYSTEM FLUSH LOGS;
+
+WITH 
+    (
+        SELECT query_id
+        FROM system.query_log
+        WHERE (query = 'SELECT 1') AND (event_date >= (today() - 1))
+        ORDER BY event_time DESC
+        LIMIT 1
+    ) AS id
+SELECT uniqExact(thread_id)
+FROM system.query_thread_log
+WHERE (event_date >= (today() - 1)) AND (query_id = id) AND (thread_id != master_thread_id);
+
+select sum(number) from numbers(1000000);
+SYSTEM FLUSH LOGS;
+
+WITH 
+    (
+        SELECT query_id
+        FROM system.query_log
+        WHERE (query = 'SELECT sum(number) FROM numbers(1000000)') AND (event_date >= (today() - 1))
+        ORDER BY event_time DESC
+        LIMIT 1
+    ) AS id
+SELECT uniqExact(thread_id)
+FROM system.query_thread_log
+WHERE (event_date >= (today() - 1)) AND (query_id = id) AND (thread_id != master_thread_id);
+
+select sum(number) from numbers_mt(1000000);
+SYSTEM FLUSH LOGS;
+
+WITH 
+    (
+        SELECT query_id
+        FROM system.query_log
+        WHERE (query = 'SELECT sum(number) FROM numbers_mt(1000000)') AND (event_date >= (today() - 1))
+        ORDER BY event_time DESC
+        LIMIT 1
+    ) AS id
+SELECT uniqExact(thread_id) > 2
+FROM system.query_thread_log
+WHERE (event_date >= (today() - 1)) AND (query_id = id) AND (thread_id != master_thread_id);