Merge pull request #12085 from ClickHouse/fix-11937

KochetovNicolai · web-flow · commit 17c89f3fe0f8 · 2020-07-03T10:35:04.000+03:00
Fix limiting the number of threads for VIEW.
diff --git a/src/Processors/QueryPipeline.h b/src/Processors/QueryPipeline.h
@@ -167,6 +167,13 @@ class QueryPipeline
     /// Set upper limit for the recommend number of threads
     void setMaxThreads(size_t max_threads_) { max_threads = max_threads_; }
 
+    /// Update upper limit for the recommend number of threads
+    void limitMaxThreads(size_t max_threads_)
+    {
+        if (max_threads == 0 || max_threads_ < max_threads)
+            max_threads = max_threads_;
+    }
+
     /// Convert query pipeline to single or several pipes.
     Pipe getPipe() &&;
     Pipes getPipes() &&;
diff --git a/src/Processors/QueryPlan/QueryPlan.cpp b/src/Processors/QueryPlan/QueryPlan.cpp
@@ -153,8 +153,8 @@ QueryPipelinePtr QueryPlan::buildQueryPipeline()
             bool limit_max_threads = frame.pipelines.empty();
             last_pipeline = frame.node->step->updatePipeline(std::move(frame.pipelines));
 
-            if (limit_max_threads)
-                last_pipeline->setMaxThreads(max_threads);
+            if (limit_max_threads && max_threads)
+                last_pipeline->limitMaxThreads(max_threads);
 
             stack.pop();
         }
diff --git a/src/Processors/QueryPlan/ReadFromStorageStep.cpp b/src/Processors/QueryPlan/ReadFromStorageStep.cpp
@@ -113,7 +113,7 @@ ReadFromStorageStep::ReadFromStorageStep(
         }
     }
 
-    if (pipes.size() == 1)
+    if (pipes.size() == 1 && !storage->isView())
         pipeline->setMaxThreads(1);
 
     for (auto & pipe : pipes)
diff --git a/tests/queries/0_stateless/01356_view_threads.reference b/tests/queries/0_stateless/01356_view_threads.reference
@@ -0,0 +1,3 @@
+0	249999500000
+1	250000000000
+1
diff --git a/tests/queries/0_stateless/01356_view_threads.sql b/tests/queries/0_stateless/01356_view_threads.sql
@@ -0,0 +1,12 @@
+drop table if exists table_01356_view_threads;
+
+create view table_01356_view_threads as select number % 10 as g, sum(number) as s from numbers_mt(1000000) group by g;
+
+set log_queries = 1;
+set max_threads = 16;
+select g % 2 as gg, sum(s) from table_01356_view_threads group by gg order by gg;
+
+system flush logs;
+select length(thread_ids) >= 16 from system.query_log where event_date >= today() - 1 and lower(query) like '%select g % 2 as gg, sum(s) from table_01356_view_threads group by gg order by gg%' and type = 'QueryFinish' order by query_start_time desc limit 1;
+
+drop table if exists table_01356_view_threads;

Original file line number	Diff line number	Diff line change
`@@ -153,8 +153,8 @@ QueryPipelinePtr QueryPlan::buildQueryPipeline()`
`153`	`153`	`bool limit_max_threads = frame.pipelines.empty();`
`154`	`154`	`last_pipeline = frame.node->step->updatePipeline(std::move(frame.pipelines));`
`155`	`155`
`156`		`- if (limit_max_threads)`
`157`		`- last_pipeline->setMaxThreads(max_threads);`
	`156`	`+ if (limit_max_threads && max_threads)`
	`157`	`+ last_pipeline->limitMaxThreads(max_threads);`
`158`	`158`
`159`	`159`	`stack.pop();`
`160`	`160`	`}`
Original file line number	Diff line number	Diff line change
`@@ -113,7 +113,7 @@ ReadFromStorageStep::ReadFromStorageStep(`
`113`	`113`	`}`
`114`	`114`	`}`
`115`	`115`
`116`		`- if (pipes.size() == 1)`
	`116`	`+ if (pipes.size() == 1 && !storage->isView())`
`117`	`117`	`pipeline->setMaxThreads(1);`
`118`	`118`
`119`	`119`	`for (auto & pipe : pipes)`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+0 249999500000`
	`2`	`+1 250000000000`
	`3`	`+1`