Try fix tests.

src/Processors/QueryPlan/QueryPlan.cpp

@@ -185,6 +185,17 @@ QueryPipelinePtr QueryPlan::buildQueryPipeline()
     return last_pipeline;
 }
 
+Pipe QueryPlan::convertToPipe()
+{
+    if (!isInitialized())
+        return {};
+
+    if (isCompleted())
+        throw Exception("Cannot convert completed QueryPlan to Pipe", ErrorCodes::LOGICAL_ERROR);
+
+    return QueryPipeline::getPipe(std::move(*buildQueryPipeline()));
+}
+
 void QueryPlan::addInterpreterContext(std::shared_ptr<Context> context)
 {
     interpreter_context.emplace_back(std::move(context));

src/Processors/QueryPlan/QueryPlan.h

@@ -20,6 +20,8 @@ class WriteBuffer;
 class QueryPlan;
 using QueryPlanPtr = std::unique_ptr<QueryPlan>;
 
+class Pipe;
+
 /// A tree of query steps.
 /// The goal of QueryPlan is to build QueryPipeline.
 /// QueryPlan let delay pipeline creation which is helpful for pipeline-level optimisations.
@@ -42,6 +44,9 @@ public:
 
     QueryPipelinePtr buildQueryPipeline();
 
+    /// If initialized, build pipeline and convert to pipe. Otherwise, return empty pipe.
+    Pipe convertToPipe();
+
     struct ExplainPlanOptions
     {
         /// Add output header to step.

src/Storages/MergeTree/StorageFromMergeTreeDataPart.h

@@ -33,10 +33,7 @@ public:
             std::move(*MergeTreeDataSelectExecutor(part->storage)
                       .readFromParts({part}, column_names, metadata_snapshot, query_info, context, max_block_size, num_streams));
 
-        if (!query_plan.isInitialized())
-            return {};
-
-        return QueryPipeline::getPipe(std::move(*query_plan.buildQueryPipeline()));
+        return query_plan.convertToPipe();
     }
 
 

src/Storages/StorageBuffer.cpp

@@ -160,7 +160,7 @@ Pipe StorageBuffer::read(
 {
     QueryPlan plan;
     read(plan, column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
-    return QueryPipeline::getPipe(std::move(*plan.buildQueryPipeline()));
+    return plan.convertToPipe();
 }
 
 void StorageBuffer::read(

src/Storages/StorageDistributed.cpp

@@ -505,7 +505,7 @@ Pipe StorageDistributed::read(
 {
     QueryPlan plan;
     read(plan, column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
-    return QueryPipeline::getPipe(std::move(*plan.buildQueryPipeline()));
+    return plan.convertToPipe();
 }
 
 void StorageDistributed::read(

src/Storages/StorageMaterializedView.cpp

@@ -117,7 +117,7 @@ Pipe StorageMaterializedView::read(
 {
     QueryPlan plan;
     read(plan, column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
-    return QueryPipeline::getPipe(std::move(*plan.buildQueryPipeline()));
+    return plan.convertToPipe();
 }
 
 void StorageMaterializedView::read(

src/Storages/StorageMergeTree.cpp

@@ -195,7 +195,7 @@ Pipe StorageMergeTree::read(
 {
     QueryPlan plan;
     read(plan, column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
-    return QueryPipeline::getPipe(std::move(*plan.buildQueryPipeline()));
+    return plan.convertToPipe();
 }
 
 std::optional<UInt64> StorageMergeTree::totalRows() const

src/Storages/StorageReplicatedMergeTree.cpp

@@ -3551,7 +3551,7 @@ Pipe StorageReplicatedMergeTree::read(
 {
     QueryPlan plan;
     read(plan, column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
-    return QueryPipeline::getPipe(std::move(*plan.buildQueryPipeline()));
+    return plan.convertToPipe();
 }
 
 

src/Storages/StorageView.cpp

@@ -61,7 +61,7 @@ Pipe StorageView::read(
 {
     QueryPlan plan;
     read(plan, column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
-    return QueryPipeline::getPipe(std::move(*plan.buildQueryPipeline()));
+    return plan.convertToPipe();
 }
 
 void StorageView::read(