Refine merge agg stream (#6793)

ref #5900
pingcap · Feb 14, 2023 · 34fc70f · 34fc70f
1 parent f5f17bb
commit 34fc70f
Show file tree

Hide file tree

Showing 9 changed files with 280 additions and 841 deletions.
diff --git a/dbms/src/DataStreams/AggregatingBlockInputStream.cpp b/dbms/src/DataStreams/AggregatingBlockInputStream.cpp
@@ -14,6 +14,8 @@
 
 #include <DataStreams/AggregatingBlockInputStream.h>
 #include <DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.h>
+#include <DataStreams/MergingAndConvertingBlockInputStream.h>
+#include <DataStreams/NullBlockInputStream.h>
 
 namespace DB
 {
@@ -41,7 +43,16 @@ Block AggregatingBlockInputStream::readImpl()
         if (!aggregator.hasSpilledData())
         {
             ManyAggregatedDataVariants many_data{data_variants};
-            impl = aggregator.mergeAndConvertToBlocks(many_data, final, 1);
+            auto merging_buckets = aggregator.mergeAndConvertToBlocks(many_data, final, 1);
+            if (!merging_buckets)
+            {
+                impl = std::make_unique<NullBlockInputStream>(aggregator.getHeader(final));
+            }
+            else
+            {
+                RUNTIME_CHECK(1 == merging_buckets->getConcurrency());
+                impl = std::make_unique<MergingAndConvertingBlockInputStream>(merging_buckets, 0, log->identifier());
+            }
         }
         else
         {

diff --git a/dbms/src/DataStreams/MergingAggregatedBlockInputStream.cpp b/dbms/src/DataStreams/MergingAggregatedBlockInputStream.cpp
diff --git a/dbms/src/DataStreams/MergingAggregatedBlockInputStream.h b/dbms/src/DataStreams/MergingAggregatedBlockInputStream.h