Refine merge agg stream

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
         {