Fix StorageBuffer/

dbms/src/Storages/StorageBuffer.cpp

@@ -30,6 +30,7 @@
 #include <Processors/Transforms/ConvertingTransform.h>
 #include <Processors/Transforms/FilterTransform.h>
 #include <Processors/Transforms/ExpressionTransform.h>
+#include <Processors/Sources/SourceFromInputStream.h>
 
 
 namespace ProfileEvents
@@ -162,6 +163,26 @@ Pipes StorageBuffer::readWithProcessors(
     size_t max_block_size,
     unsigned num_streams)
 {
+    auto read_as_pipes = [](const StoragePtr & storage,
+                            const Names & column_names,
+                            const SelectQueryInfo & query_info,
+                            const Context & context,
+                            QueryProcessingStage::Enum processed_stage,
+                            size_t max_block_size,
+                            unsigned num_streams)
+    {
+        if (storage->supportProcessorsPipeline())
+            return storage->readWithProcessors(column_names, query_info, context, processed_stage, max_block_size, num_streams);
+
+        auto streams = storage->read(column_names, query_info, context, processed_stage, max_block_size, num_streams);
+
+        Pipes pipes;
+        for (auto & stream : streams)
+            pipes.emplace_back(std::make_shared<SourceFromInputStream>(stream));
+
+        return pipes;
+    };
+
     Pipes pipes_from_dst;
 
     if (!no_destination)
@@ -185,7 +206,7 @@ Pipes StorageBuffer::readWithProcessors(
                 query_info.input_sorting_info = query_info.order_by_optimizer->getInputOrder(destination);
 
             /// The destination table has the same structure of the requested columns and we can simply read blocks from there.
-            pipes_from_dst = destination->readWithProcessors(column_names, query_info, context, processed_stage, max_block_size, num_streams);
+            pipes_from_dst = read_as_pipes(destination, column_names, query_info, context, processed_stage, max_block_size, num_streams));
         }
         else
         {
@@ -220,7 +241,7 @@ Pipes StorageBuffer::readWithProcessors(
             }
             else
             {
-                pipes_from_dst = destination->readWithProcessors(columns_intersection, query_info, context, processed_stage, max_block_size, num_streams);
+                pipes_from_dst = read_as_pipes(destination, columns_intersection, query_info, context, processed_stage, max_block_size, num_streams);
                 for (auto & pipe : pipes_from_dst)
                 {
                     pipe.addSimpleTransform(std::make_shared<AddingMissedTransform>(