diff --git a/dbms/include/DB/DataStreams/ParallelAggregatingBlockInputStream.h b/dbms/include/DB/DataStreams/ParallelAggregatingBlockInputStream.h
index d2625f7da3c..dcfe7ba7d20 100644
--- a/dbms/include/DB/DataStreams/ParallelAggregatingBlockInputStream.h
+++ b/dbms/include/DB/DataStreams/ParallelAggregatingBlockInputStream.h
@@ -124,7 +124,7 @@ protected:
 			<< " in " << elapsed_seconds << " sec."
 			<< " (" << total_src_rows / elapsed_seconds << " rows/sec., " << total_src_bytes / elapsed_seconds / 1048576.0 << " MiB/sec.)");
 
-		AggregatedDataVariantsPtr res = aggregator->merge(many_data);
+		AggregatedDataVariantsPtr res = aggregator->merge(many_data, max_threads);
 
 		if (isCancelled())
 			return Block();
diff --git a/dbms/include/DB/Interpreters/Aggregator.h b/dbms/include/DB/Interpreters/Aggregator.h
index 87fd71a17e0..1d732c1e049 100644
--- a/dbms/include/DB/Interpreters/Aggregator.h
+++ b/dbms/include/DB/Interpreters/Aggregator.h
@@ -584,7 +584,7 @@ public:
 	  * После объединения, все стркутуры агрегации (а не только те, в которую они будут слиты) должны жить, пока не будет вызвана функция convertToBlock.
 	  * Это нужно, так как в слитом результате могут остаться указатели на память в пуле, которым владеют другие структуры агрегации.
 	  */
-	AggregatedDataVariantsPtr merge(ManyAggregatedDataVariants & data_variants);
+	AggregatedDataVariantsPtr merge(ManyAggregatedDataVariants & data_variants, size_t max_threads);
 
 	/** Объединить несколько агрегированных блоков в одну структуру данных.
 	  * (Доагрегировать несколько блоков, которые представляют собой результат независимых агрегаций с удалённых серверов.)
diff --git a/dbms/src/Interpreters/Aggregator.cpp b/dbms/src/Interpreters/Aggregator.cpp
index f56126cd851..d6e33986a3a 100644
--- a/dbms/src/Interpreters/Aggregator.cpp
+++ b/dbms/src/Interpreters/Aggregator.cpp
@@ -766,7 +766,7 @@ Block Aggregator::convertToBlock(AggregatedDataVariants & data_variants, bool fi
 }
 
 
-AggregatedDataVariantsPtr Aggregator::merge(ManyAggregatedDataVariants & data_variants)
+AggregatedDataVariantsPtr Aggregator::merge(ManyAggregatedDataVariants & data_variants, size_t max_threads)
 {
 	if (data_variants.empty())
  		throw Exception("Empty data passed to Aggregator::merge().", ErrorCodes::EMPTY_DATA_PASSED);
@@ -805,22 +805,32 @@ AggregatedDataVariantsPtr Aggregator::merge(ManyAggregatedDataVariants & data_va
 	if (res->type == AggregatedDataVariants::WITHOUT_KEY || overflow_row)
 		mergeWithoutKeyDataImpl(non_empty_data);
 
+	boost::threadpool::pool * thread_pool = nullptr;
+	if (max_threads > 1 && rows > 100000	/// TODO Сделать настраиваемый порог.
+		&& (   res->type == AggregatedDataVariants::KEY_32
+			|| res->type == AggregatedDataVariants::KEY_64
+			|| res->type == AggregatedDataVariants::KEY_STRING
+			|| res->type == AggregatedDataVariants::KEY_FIXED_STRING
+			|| res->type == AggregatedDataVariants::KEYS_128
+			|| res->type == AggregatedDataVariants::HASHED))
+		thread_pool = new boost::threadpool::pool(max_threads);
+
 	if (res->type == AggregatedDataVariants::KEY_8)
 		mergeSingleLevelDataImpl<AggregationMethodOneNumber<UInt8>>(non_empty_data);
 	else if (res->type == AggregatedDataVariants::KEY_16)
 		mergeSingleLevelDataImpl<AggregationMethodOneNumber<UInt16>>(non_empty_data);
 	else if (res->type == AggregatedDataVariants::KEY_32)
-		mergeTwoLevelDataImpl<AggregationMethodOneNumber<UInt32>>(non_empty_data, nullptr);
+		mergeTwoLevelDataImpl<AggregationMethodOneNumber<UInt32>>(non_empty_data, thread_pool);
 	else if (res->type == AggregatedDataVariants::KEY_64)
-		mergeTwoLevelDataImpl<AggregationMethodOneNumber<UInt64>>(non_empty_data, nullptr);
+		mergeTwoLevelDataImpl<AggregationMethodOneNumber<UInt64>>(non_empty_data, thread_pool);
 	else if (res->type == AggregatedDataVariants::KEY_STRING)
-		mergeTwoLevelDataImpl<AggregationMethodString>(non_empty_data, nullptr);
+		mergeTwoLevelDataImpl<AggregationMethodString>(non_empty_data, thread_pool);
 	else if (res->type == AggregatedDataVariants::KEY_FIXED_STRING)
-		mergeTwoLevelDataImpl<AggregationMethodFixedString>(non_empty_data, nullptr);
+		mergeTwoLevelDataImpl<AggregationMethodFixedString>(non_empty_data, thread_pool);
 	else if (res->type == AggregatedDataVariants::KEYS_128)
-		mergeTwoLevelDataImpl<AggregationMethodKeys128>(non_empty_data, nullptr);
+		mergeTwoLevelDataImpl<AggregationMethodKeys128>(non_empty_data, thread_pool);
 	else if (res->type == AggregatedDataVariants::HASHED)
-		mergeTwoLevelDataImpl<AggregationMethodHashed>(non_empty_data, nullptr);
+		mergeTwoLevelDataImpl<AggregationMethodHashed>(non_empty_data, thread_pool);
 	else if (res->type != AggregatedDataVariants::WITHOUT_KEY)
 		throw Exception("Unknown aggregated data variant.", ErrorCodes::UNKNOWN_AGGREGATED_DATA_VARIANT);