From 12f4cfb2c7f272ba26972f075ca6194d43e292b6 Mon Sep 17 00:00:00 2001
From: Nikolai Kochetov <nik-kochetov@yandex-team.ru>
Date: Tue, 14 Apr 2020 13:04:49 +0300
Subject: [PATCH] Use processors for final.

---
 .../MergeTree/MergeTreeDataSelectExecutor.cpp | 41 ++++++++++---------
 1 file changed, 21 insertions(+), 20 deletions(-)

diff --git a/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp b/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
index 4a7a411f45a..7c621513cab 100644
--- a/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
+++ b/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
@@ -38,18 +38,9 @@ namespace std
 }
 #endif
 
-#include <DataStreams/ExpressionBlockInputStream.h>
-#include <DataStreams/FilterBlockInputStream.h>
 #include <DataStreams/CollapsingFinalBlockInputStream.h>
-#include <DataStreams/AddingConstColumnBlockInputStream.h>
 #include <DataStreams/CreatingSetsBlockInputStream.h>
-#include <DataStreams/MergingSortedBlockInputStream.h>
-#include <DataStreams/NullBlockInputStream.h>
-#include <DataStreams/SummingSortedBlockInputStream.h>
-#include <DataStreams/ReplacingSortedBlockInputStream.h>
 #include <DataStreams/ReverseBlockInputStream.h>
-#include <DataStreams/AggregatingSortedBlockInputStream.h>
-#include <DataStreams/VersionedCollapsingSortedBlockInputStream.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <DataTypes/DataTypeDate.h>
 #include <DataTypes/DataTypeEnum.h>
@@ -59,6 +50,10 @@ namespace std
 #include <Processors/Transforms/ExpressionTransform.h>
 #include <Processors/Transforms/ReverseTransform.h>
 #include <Processors/Merges/MergingSortedTransform.h>
+#include <Processors/Merges/SummingSortedTransform.h>
+#include <Processors/Merges/AggregatingSortedTransform.h>
+#include <Processors/Merges/ReplacingSortedTransform.h>
+#include <Processors/Merges/VersionedCollapsingTransform.h>
 #include <Processors/Executors/TreeExecutorBlockInputStream.h>
 #include <Processors/Sources/SourceFromInputStream.h>
 #include <Processors/ConcatProcessor.h>
@@ -1096,16 +1091,14 @@ Pipes MergeTreeDataSelectExecutor::spreadMarkRangesAmongStreamsFinal(
     };
 
     BlockInputStreamPtr merged;
+    ProcessorPtr merged_processor;
     switch (data.merging_params.mode)
     {
         case MergeTreeData::MergingParams::Ordinary:
         {
-            auto merged_processor =
-                    std::make_shared<MergingSortedTransform>(header, pipes.size(), sort_description, max_block_size);
-            Pipe pipe(std::move(pipes), std::move(merged_processor));
-            pipes = Pipes();
-            pipes.emplace_back(std::move(pipe));
-            return pipes;
+            merged_processor = std::make_shared<MergingSortedTransform>(header, pipes.size(),
+                    sort_description, max_block_size);
+            break;
         }
 
         case MergeTreeData::MergingParams::Collapsing:
@@ -1114,28 +1107,36 @@ Pipes MergeTreeDataSelectExecutor::spreadMarkRangesAmongStreamsFinal(
             break;
 
         case MergeTreeData::MergingParams::Summing:
-            merged = std::make_shared<SummingSortedBlockInputStream>(streams_to_merge(),
+            merged_processor = std::make_shared<SummingSortedTransform>(header, pipes.size(),
                     sort_description, data.merging_params.columns_to_sum, max_block_size);
             break;
 
         case MergeTreeData::MergingParams::Aggregating:
-            merged = std::make_shared<AggregatingSortedBlockInputStream>(streams_to_merge(), sort_description, max_block_size);
+            merged_processor = std::make_shared<AggregatingSortedTransform>(header, pipes.size(),
+                    sort_description, max_block_size);
             break;
 
         case MergeTreeData::MergingParams::Replacing:    /// TODO Make ReplacingFinalBlockInputStream
-            merged = std::make_shared<ReplacingSortedBlockInputStream>(streams_to_merge(),
+            merged_processor = std::make_shared<ReplacingSortedTransform>(header, pipes.size(),
                     sort_description, data.merging_params.version_column, max_block_size);
             break;
 
         case MergeTreeData::MergingParams::VersionedCollapsing: /// TODO Make VersionedCollapsingFinalBlockInputStream
-            merged = std::make_shared<VersionedCollapsingSortedBlockInputStream>(
-                    streams_to_merge(), sort_description, data.merging_params.sign_column, max_block_size);
+            merged_processor = std::make_shared<VersionedCollapsingTransform>(header, pipes.size(),
+                    sort_description, data.merging_params.sign_column, max_block_size);
             break;
 
         case MergeTreeData::MergingParams::Graphite:
             throw Exception("GraphiteMergeTree doesn't support FINAL", ErrorCodes::LOGICAL_ERROR);
     }
 
+    if (merged_processor)
+    {
+        Pipe pipe(std::move(pipes), std::move(merged_processor));
+        pipes = Pipes();
+        pipes.emplace_back(std::move(pipe));
+    }
+
     if (merged)
         pipes.emplace_back(std::make_shared<SourceFromInputStream>(merged));