Merge pull request ClickHouse#11114 from azat/max_threads-simple-query-optimization-fix

KochetovNicolai · web-flow · commit c83014bc7662 · 2020-05-25T14:05:46.000+03:00
Do not reserve extra threads after max threads optimization for simple queries
diff --git a/src/Interpreters/InterpreterSelectQuery.cpp b/src/Interpreters/InterpreterSelectQuery.cpp
@@ -1533,7 +1533,7 @@ void InterpreterSelectQuery::executeFetchColumns(
         if constexpr (pipeline_with_processors)
         {
             if (streams.size() == 1 || pipes.size() == 1)
-                pipeline.setMaxThreads(streams.size());
+                pipeline.setMaxThreads(1);
 
             /// Unify streams. They must have same headers.
             if (streams.size() > 1)
diff --git a/src/Interpreters/InterpreterSelectWithUnionQuery.cpp b/src/Interpreters/InterpreterSelectWithUnionQuery.cpp
@@ -271,6 +271,11 @@ QueryPipeline InterpreterSelectWithUnionQuery::executeWithProcessors()
     {
         auto common_header = getCommonHeaderForUnion(headers);
         main_pipeline.unitePipelines(std::move(pipelines), common_header);
+
+        // nested queries can force 1 thread (due to simplicity)
+        // but in case of union this cannot be done.
+        UInt64 max_threads = context->getSettingsRef().max_threads;
+        main_pipeline.setMaxThreads(std::min<UInt64>(nested_interpreters.size(), max_threads));
     }
 
     main_pipeline.addInterpreterContext(context);
diff --git a/tests/queries/0_stateless/01283_max_threads_simple_query_optimization.reference b/tests/queries/0_stateless/01283_max_threads_simple_query_optimization.reference
diff --git a/tests/queries/0_stateless/01283_max_threads_simple_query_optimization.sql b/tests/queries/0_stateless/01283_max_threads_simple_query_optimization.sql
@@ -0,0 +1,21 @@
+DROP TABLE IF EXISTS data_01283;
+
+CREATE TABLE data_01283 engine=MergeTree()
+ORDER BY key
+PARTITION BY key
+AS SELECT number key FROM numbers(10);
+
+SET log_queries=1;
+SELECT * FROM data_01283 LIMIT 1 FORMAT Null;
+SET log_queries=0;
+SYSTEM FLUSH LOGS;
+
+-- 1 for PullingAsyncPipelineExecutor::pull
+-- 1 for AsynchronousBlockInputStream
+SELECT
+    throwIf(count() != 1, 'no query was logged'),
+    throwIf(length(thread_ids) != 2, 'too many threads used')
+FROM system.query_log
+WHERE type = 'QueryFinish' AND query LIKE '%data_01283 LIMIT 1%'
+GROUP BY thread_ids
+FORMAT Null;

Original file line number	Diff line number	Diff line change
`@@ -1533,7 +1533,7 @@ void InterpreterSelectQuery::executeFetchColumns(`
`1533`	`1533`	`if constexpr (pipeline_with_processors)`
`1534`	`1534`	`{`
`1535`	`1535`	`if (streams.size() == 1 \|\| pipes.size() == 1)`
`1536`		`- pipeline.setMaxThreads(streams.size());`
	`1536`	`+ pipeline.setMaxThreads(1);`
`1537`	`1537`
`1538`	`1538`	`/// Unify streams. They must have same headers.`
`1539`	`1539`	`if (streams.size() > 1)`
Original file line number	Diff line number	Diff line change
`@@ -271,6 +271,11 @@ QueryPipeline InterpreterSelectWithUnionQuery::executeWithProcessors()`
`271`	`271`	`{`
`272`	`272`	`auto common_header = getCommonHeaderForUnion(headers);`
`273`	`273`	`main_pipeline.unitePipelines(std::move(pipelines), common_header);`
	`274`	`+`
	`275`	`+ // nested queries can force 1 thread (due to simplicity)`
	`276`	`+ // but in case of union this cannot be done.`
	`277`	`+ UInt64 max_threads = context->getSettingsRef().max_threads;`
	`278`	`+ main_pipeline.setMaxThreads(std::min<UInt64>(nested_interpreters.size(), max_threads));`
`274`	`279`	`}`
`275`	`280`
`276`	`281`	`main_pipeline.addInterpreterContext(context);`