Rename QueryPipeline to QueryPipelineBuilder.

programs/client/Client.cpp

@@ -33,7 +33,7 @@
 #include <Poco/Util/Application.h>
 #include <Processors/Formats/IInputFormat.h>
 #include <Processors/Executors/PullingAsyncPipelineExecutor.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Columns/ColumnString.h>
 #include <common/find_symbols.h>
 #include <common/LineReader.h>
@@ -2034,7 +2034,7 @@ private:
             });
         }
 
-        QueryPipeline pipeline;
+        QueryPipelineBuilder pipeline;
         pipeline.init(std::move(pipe));
         PullingAsyncPipelineExecutor executor(pipeline);
 

src/DataStreams/BlockIO.h

@@ -4,7 +4,7 @@
 
 #include <functional>
 
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Chain.h>
 
 
@@ -29,10 +29,10 @@ struct BlockIO
     Chain out;
     BlockInputStreamPtr in;
 
-    QueryPipeline pipeline;
+    QueryPipelineBuilder pipeline;
 
     /// Callbacks for query logging could be set here.
-    std::function<void(IBlockInputStream *, QueryPipeline *)>    finish_callback;
+    std::function<void(IBlockInputStream *, QueryPipelineBuilder *)>    finish_callback;
     std::function<void()>                                        exception_callback;
 
     /// When it is true, don't bother sending any non-empty blocks to the out stream
@@ -43,7 +43,7 @@ struct BlockIO
     {
         if (finish_callback)
         {
-            QueryPipeline * pipeline_ptr = nullptr;
+            QueryPipelineBuilder * pipeline_ptr = nullptr;
             if (pipeline.initialized())
                 pipeline_ptr = &pipeline;
 

src/Dictionaries/CacheDictionary.cpp

@@ -14,7 +14,7 @@
 #include <Dictionaries/HierarchyDictionariesUtils.h>
 
 #include <Processors/Executors/PullingPipelineExecutor.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 namespace ProfileEvents
 {

src/Dictionaries/PolygonDictionary.cpp

@@ -165,7 +165,7 @@ void IPolygonDictionary::blockToAttributes(const DB::Block & block)
 
 void IPolygonDictionary::loadData()
 {
-    QueryPipeline pipeline;
+    QueryPipelineBuilder pipeline;
     pipeline.init(source_ptr->loadAll());
 
     PullingPipelineExecutor executor(pipeline);

src/Dictionaries/RangeHashedDictionary.cpp

@@ -303,7 +303,7 @@ void RangeHashedDictionary<dictionary_key_type>::createAttributes()
 template <DictionaryKeyType dictionary_key_type>
 void RangeHashedDictionary<dictionary_key_type>::loadData()
 {
-    QueryPipeline pipeline;
+    QueryPipelineBuilder pipeline;
     pipeline.init(source_ptr->loadAll());
 
     PullingPipelineExecutor executor(pipeline);

src/Interpreters/MutationsInterpreter.h

@@ -14,8 +14,8 @@ namespace DB
 class Context;
 class QueryPlan;
 
-class QueryPipeline;
-using QueryPipelinePtr = std::unique_ptr<QueryPipeline>;
+class QueryPipelineBuilder;
+using QueryPipelineBuilderPtr = std::unique_ptr<QueryPipelineBuilder>;
 
 /// Return false if the data isn't going to be changed by mutations.
 bool isStorageTouchedByMutations(
@@ -84,7 +84,7 @@ private:
     struct Stage;
 
     ASTPtr prepareInterpreterSelectQuery(std::vector<Stage> &prepared_stages, bool dry_run);
-    QueryPipelinePtr addStreamsForLaterStages(const std::vector<Stage> & prepared_stages, QueryPlan & plan) const;
+    QueryPipelineBuilderPtr addStreamsForLaterStages(const std::vector<Stage> & prepared_stages, QueryPlan & plan) const;
 
     std::optional<SortDescription> getStorageSortDescriptionIfPossible(const Block & header) const;
 

src/Interpreters/executeQuery.cpp

@@ -587,7 +587,7 @@ static std::tuple<ASTPtr, BlockIO> executeQueryImpl(
             res = interpreter->execute();
         }
 
-        QueryPipeline & pipeline = res.pipeline;
+        QueryPipelineBuilder & pipeline = res.pipeline;
         bool use_processors = pipeline.initialized();
 
         if (const auto * insert_interpreter = typeid_cast<const InterpreterInsertQuery *>(&*interpreter))
@@ -759,7 +759,7 @@ static std::tuple<ASTPtr, BlockIO> executeQueryImpl(
                  log_queries_min_query_duration_ms = settings.log_queries_min_query_duration_ms.totalMilliseconds(),
                  status_info_to_query_log
             ]
-                (IBlockInputStream * stream_in, QueryPipeline * query_pipeline) mutable
+                (IBlockInputStream * stream_in, QueryPipelineBuilder * query_pipeline) mutable
             {
                 QueryStatus * process_list_elem = context->getProcessListElement();
 

src/Processors/Executors/PipelineExecutingBlockInputStream.cpp

@@ -1,7 +1,7 @@
 #include <Processors/Executors/PipelineExecutingBlockInputStream.h>
 #include <Processors/Executors/PullingAsyncPipelineExecutor.h>
 #include <Processors/Executors/PullingPipelineExecutor.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 namespace DB
 {
@@ -11,8 +11,8 @@ namespace ErrorCodes
     extern const int LOGICAL_ERROR;
 }
 
-PipelineExecutingBlockInputStream::PipelineExecutingBlockInputStream(QueryPipeline pipeline_)
-    : pipeline(std::make_unique<QueryPipeline>(std::move(pipeline_)))
+PipelineExecutingBlockInputStream::PipelineExecutingBlockInputStream(QueryPipelineBuilder pipeline_)
+    : pipeline(std::make_unique<QueryPipelineBuilder>(std::move(pipeline_)))
 {
 }
 

src/Processors/Executors/PipelineExecutingBlockInputStream.h

@@ -4,7 +4,7 @@
 namespace DB
 {
 
-class QueryPipeline;
+class QueryPipelineBuilder;
 class PullingAsyncPipelineExecutor;
 class PullingPipelineExecutor;
 
@@ -13,7 +13,7 @@ class PullingPipelineExecutor;
 class PipelineExecutingBlockInputStream : public IBlockInputStream
 {
 public:
-    explicit PipelineExecutingBlockInputStream(QueryPipeline pipeline_);
+    explicit PipelineExecutingBlockInputStream(QueryPipelineBuilder pipeline_);
     ~PipelineExecutingBlockInputStream() override;
 
     String getName() const override { return "PipelineExecuting"; }
@@ -32,7 +32,7 @@ protected:
     Block readImpl() override;
 
 private:
-    std::unique_ptr<QueryPipeline> pipeline;
+    std::unique_ptr<QueryPipelineBuilder> pipeline;
     /// One of executors is used.
     std::unique_ptr<PullingPipelineExecutor> executor; /// for single thread.
     std::unique_ptr<PullingAsyncPipelineExecutor> async_executor; /// for many threads.

src/Processors/Executors/PullingAsyncPipelineExecutor.cpp

@@ -2,7 +2,7 @@
 #include <Processors/Executors/PipelineExecutor.h>
 #include <Processors/Formats/LazyOutputFormat.h>
 #include <Processors/Transforms/AggregatingTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 #include <Common/setThreadName.h>
 #include <common/scope_guard_safe.h>
@@ -36,7 +36,7 @@ struct PullingAsyncPipelineExecutor::Data
     }
 };
 
-PullingAsyncPipelineExecutor::PullingAsyncPipelineExecutor(QueryPipeline & pipeline_) : pipeline(pipeline_)
+PullingAsyncPipelineExecutor::PullingAsyncPipelineExecutor(QueryPipelineBuilder & pipeline_) : pipeline(pipeline_)
 {
     if (!pipeline.isCompleted())
     {

src/Processors/Executors/PullingAsyncPipelineExecutor.h

@@ -4,7 +4,7 @@
 namespace DB
 {
 
-class QueryPipeline;
+class QueryPipelineBuilder;
 class Block;
 class Chunk;
 class LazyOutputFormat;
@@ -20,7 +20,7 @@ struct BlockStreamProfileInfo;
 class PullingAsyncPipelineExecutor
 {
 public:
-    explicit PullingAsyncPipelineExecutor(QueryPipeline & pipeline_);
+    explicit PullingAsyncPipelineExecutor(QueryPipelineBuilder & pipeline_);
     ~PullingAsyncPipelineExecutor();
 
     /// Get structure of returned block or chunk.
@@ -50,7 +50,7 @@ public:
     struct Data;
 
 private:
-    QueryPipeline & pipeline;
+    QueryPipelineBuilder & pipeline;
     std::shared_ptr<LazyOutputFormat> lazy_format;
     std::unique_ptr<Data> data;
 };

src/Processors/Executors/PullingPipelineExecutor.cpp

@@ -1,12 +1,12 @@
 #include <Processors/Executors/PullingPipelineExecutor.h>
 #include <Processors/Formats/PullingOutputFormat.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/AggregatingTransform.h>
 
 namespace DB
 {
 
-PullingPipelineExecutor::PullingPipelineExecutor(QueryPipeline & pipeline_) : pipeline(pipeline_)
+PullingPipelineExecutor::PullingPipelineExecutor(QueryPipelineBuilder & pipeline_) : pipeline(pipeline_)
 {
     pulling_format = std::make_shared<PullingOutputFormat>(pipeline.getHeader(), has_data_flag);
     pipeline.setOutputFormat(pulling_format);

src/Processors/Executors/PullingPipelineExecutor.h

@@ -7,7 +7,7 @@ namespace DB
 
 class Block;
 class Chunk;
-class QueryPipeline;
+class QueryPipelineBuilder;
 class PipelineExecutor;
 class PullingOutputFormat;
 struct BlockStreamProfileInfo;
@@ -23,7 +23,7 @@ using PipelineExecutorPtr = std::shared_ptr<PipelineExecutor>;
 class PullingPipelineExecutor
 {
 public:
-    explicit PullingPipelineExecutor(QueryPipeline & pipeline_);
+    explicit PullingPipelineExecutor(QueryPipelineBuilder & pipeline_);
     ~PullingPipelineExecutor();
 
     /// Get structure of returned block or chunk.
@@ -50,7 +50,7 @@ public:
 
 private:
     std::atomic_bool has_data_flag = false;
-    QueryPipeline & pipeline;
+    QueryPipelineBuilder & pipeline;
     std::shared_ptr<PullingOutputFormat> pulling_format;
     PipelineExecutorPtr executor;
 };

src/Processors/Pipe.cpp

@@ -777,7 +777,7 @@ void Pipe::setOutputFormat(ProcessorPtr output)
     auto * format = dynamic_cast<IOutputFormat * >(output.get());
 
     if (!format)
-        throw Exception("IOutputFormat processor expected for QueryPipeline::setOutputFormat.",
+        throw Exception("IOutputFormat processor expected for QueryPipelineBuilder::setOutputFormat.",
                         ErrorCodes::LOGICAL_ERROR);
 
     auto & main = format->getPort(IOutputFormat::PortKind::Main);

src/Processors/Pipe.h

@@ -14,8 +14,6 @@ struct StreamLocalLimits;
 class Pipe;
 using Pipes = std::vector<Pipe>;
 
-class QueryPipeline;
-
 using OutputPortRawPtrs = std::vector<OutputPort *>;
 
 /// Pipe is a set of processors which represents the part of pipeline.
@@ -145,7 +143,7 @@ private:
     void setSinks(const Pipe::ProcessorGetterWithStreamKind & getter);
     void setOutputFormat(ProcessorPtr output);
 
-    friend class QueryPipeline;
+    friend class QueryPipelineBuilder;
 };
 
 }

src/Processors/PipelineResourcesHolder.h

@@ -5,7 +5,7 @@
 namespace DB
 {
 
-class QueryPipeline;
+class QueryPipelineBuilder;
 
 class IStorage;
 using StoragePtr = std::shared_ptr<IStorage>;

src/Processors/QueryPipelineBuilder.cpp

@@ -1,4 +1,4 @@
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 #include <Processors/ResizeProcessor.h>
 #include <Processors/LimitTransform.h>
@@ -30,18 +30,18 @@ namespace ErrorCodes
     extern const int LOGICAL_ERROR;
 }
 
-void QueryPipeline::addQueryPlan(std::unique_ptr<QueryPlan> plan)
+void QueryPipelineBuilder::addQueryPlan(std::unique_ptr<QueryPlan> plan)
 {
     pipe.addQueryPlan(std::move(plan));
 }
 
-void QueryPipeline::checkInitialized()
+void QueryPipelineBuilder::checkInitialized()
 {
     if (!initialized())
         throw Exception("QueryPipeline wasn't initialized.", ErrorCodes::LOGICAL_ERROR);
 }
 
-void QueryPipeline::checkInitializedAndNotCompleted()
+void QueryPipelineBuilder::checkInitializedAndNotCompleted()
 {
     checkInitialized();
 
@@ -68,7 +68,7 @@ static void checkSource(const ProcessorPtr & source, bool can_have_totals)
                         toString(source->getOutputs().size()) + " outputs.", ErrorCodes::LOGICAL_ERROR);
 }
 
-void QueryPipeline::init(Pipe pipe_)
+void QueryPipelineBuilder::init(Pipe pipe_)
 {
     if (initialized())
         throw Exception("Pipeline has already been initialized.", ErrorCodes::LOGICAL_ERROR);
@@ -79,43 +79,43 @@ void QueryPipeline::init(Pipe pipe_)
     pipe = std::move(pipe_);
 }
 
-void QueryPipeline::reset()
+void QueryPipelineBuilder::reset()
 {
     Pipe pipe_to_destroy(std::move(pipe));
-    *this = QueryPipeline();
+    *this = QueryPipelineBuilder();
 }
 
-void QueryPipeline::addSimpleTransform(const Pipe::ProcessorGetter & getter)
+void QueryPipelineBuilder::addSimpleTransform(const Pipe::ProcessorGetter & getter)
 {
     checkInitializedAndNotCompleted();
     pipe.addSimpleTransform(getter);
 }
 
-void QueryPipeline::addSimpleTransform(const Pipe::ProcessorGetterWithStreamKind & getter)
+void QueryPipelineBuilder::addSimpleTransform(const Pipe::ProcessorGetterWithStreamKind & getter)
 {
     checkInitializedAndNotCompleted();
     pipe.addSimpleTransform(getter);
 }
 
-void QueryPipeline::addTransform(ProcessorPtr transform)
+void QueryPipelineBuilder::addTransform(ProcessorPtr transform)
 {
     checkInitializedAndNotCompleted();
     pipe.addTransform(std::move(transform));
 }
 
-void QueryPipeline::addTransform(ProcessorPtr transform, InputPort * totals, InputPort * extremes)
+void QueryPipelineBuilder::addTransform(ProcessorPtr transform, InputPort * totals, InputPort * extremes)
 {
     checkInitializedAndNotCompleted();
     pipe.addTransform(std::move(transform), totals, extremes);
 }
 
-void QueryPipeline::addChains(std::vector<Chain> chains)
+void QueryPipelineBuilder::addChains(std::vector<Chain> chains)
 {
     checkInitializedAndNotCompleted();
     pipe.addChains(std::move(chains));
 }
 
-void QueryPipeline::addChain(Chain chain)
+void QueryPipelineBuilder::addChain(Chain chain)
 {
     checkInitializedAndNotCompleted();
     std::vector<Chain> chains;
@@ -124,19 +124,19 @@ void QueryPipeline::addChain(Chain chain)
     pipe.addChains(std::move(chains));
 }
 
-void QueryPipeline::transform(const Transformer & transformer)
+void QueryPipelineBuilder::transform(const Transformer & transformer)
 {
     checkInitializedAndNotCompleted();
     pipe.transform(transformer);
 }
 
-void QueryPipeline::setSinks(const Pipe::ProcessorGetterWithStreamKind & getter)
+void QueryPipelineBuilder::setSinks(const Pipe::ProcessorGetterWithStreamKind & getter)
 {
     checkInitializedAndNotCompleted();
     pipe.setSinks(getter);
 }
 
-void QueryPipeline::addDelayedStream(ProcessorPtr source)
+void QueryPipelineBuilder::addDelayedStream(ProcessorPtr source)
 {
     checkInitializedAndNotCompleted();
 
@@ -150,18 +150,18 @@ void QueryPipeline::addDelayedStream(ProcessorPtr source)
     addTransform(std::move(processor));
 }
 
-void QueryPipeline::addMergingAggregatedMemoryEfficientTransform(AggregatingTransformParamsPtr params, size_t num_merging_processors)
+void QueryPipelineBuilder::addMergingAggregatedMemoryEfficientTransform(AggregatingTransformParamsPtr params, size_t num_merging_processors)
 {
     DB::addMergingAggregatedMemoryEfficientTransform(pipe, std::move(params), num_merging_processors);
 }
 
-void QueryPipeline::resize(size_t num_streams, bool force, bool strict)
+void QueryPipelineBuilder::resize(size_t num_streams, bool force, bool strict)
 {
     checkInitializedAndNotCompleted();
     pipe.resize(num_streams, force, strict);
 }
 
-void QueryPipeline::addTotalsHavingTransform(ProcessorPtr transform)
+void QueryPipelineBuilder::addTotalsHavingTransform(ProcessorPtr transform)
 {
     checkInitializedAndNotCompleted();
 
@@ -178,7 +178,7 @@ void QueryPipeline::addTotalsHavingTransform(ProcessorPtr transform)
     pipe.addTransform(std::move(transform), totals_port, nullptr);
 }
 
-void QueryPipeline::addDefaultTotals()
+void QueryPipelineBuilder::addDefaultTotals()
 {
     checkInitializedAndNotCompleted();
 
@@ -200,13 +200,13 @@ void QueryPipeline::addDefaultTotals()
     pipe.addTotalsSource(std::move(source));
 }
 
-void QueryPipeline::dropTotalsAndExtremes()
+void QueryPipelineBuilder::dropTotalsAndExtremes()
 {
     pipe.dropTotals();
     pipe.dropExtremes();
 }
 
-void QueryPipeline::addExtremesTransform()
+void QueryPipelineBuilder::addExtremesTransform()
 {
     checkInitializedAndNotCompleted();
 
@@ -222,7 +222,7 @@ void QueryPipeline::addExtremesTransform()
     pipe.addTransform(std::move(transform), nullptr, port);
 }
 
-void QueryPipeline::setOutputFormat(ProcessorPtr output)
+void QueryPipelineBuilder::setOutputFormat(ProcessorPtr output)
 {
     checkInitializedAndNotCompleted();
 
@@ -237,8 +237,8 @@ void QueryPipeline::setOutputFormat(ProcessorPtr output)
     initRowsBeforeLimit();
 }
 
-QueryPipeline QueryPipeline::unitePipelines(
-    std::vector<std::unique_ptr<QueryPipeline>> pipelines,
+QueryPipelineBuilder QueryPipelineBuilder::unitePipelines(
+    std::vector<std::unique_ptr<QueryPipelineBuilder>> pipelines,
     size_t max_threads_limit,
     Processors * collected_processors)
 {
@@ -271,7 +271,7 @@ QueryPipeline QueryPipeline::unitePipelines(
             max_threads_limit = pipeline.max_threads;
     }
 
-    QueryPipeline pipeline;
+    QueryPipelineBuilder pipeline;
     pipeline.init(Pipe::unitePipes(std::move(pipes), collected_processors, false));
 
     if (will_limit_max_threads)
@@ -283,9 +283,9 @@ QueryPipeline QueryPipeline::unitePipelines(
     return pipeline;
 }
 
-std::unique_ptr<QueryPipeline> QueryPipeline::joinPipelines(
-    std::unique_ptr<QueryPipeline> left,
-    std::unique_ptr<QueryPipeline> right,
+std::unique_ptr<QueryPipelineBuilder> QueryPipelineBuilder::joinPipelines(
+    std::unique_ptr<QueryPipelineBuilder> left,
+    std::unique_ptr<QueryPipelineBuilder> right,
     JoinPtr join,
     size_t max_block_size,
     Processors * collected_processors)
@@ -375,7 +375,7 @@ std::unique_ptr<QueryPipeline> QueryPipeline::joinPipelines(
     return left;
 }
 
-void QueryPipeline::addCreatingSetsTransform(const Block & res_header, SubqueryForSet subquery_for_set, const SizeLimits & limits, ContextPtr context)
+void QueryPipelineBuilder::addCreatingSetsTransform(const Block & res_header, SubqueryForSet subquery_for_set, const SizeLimits & limits, ContextPtr context)
 {
     resize(1);
 
@@ -394,7 +394,7 @@ void QueryPipeline::addCreatingSetsTransform(const Block & res_header, SubqueryF
     pipe.addTransform(std::move(transform), totals_port, nullptr);
 }
 
-void QueryPipeline::addPipelineBefore(QueryPipeline pipeline)
+void QueryPipelineBuilder::addPipelineBefore(QueryPipelineBuilder pipeline)
 {
     checkInitializedAndNotCompleted();
     if (pipeline.getHeader())
@@ -409,14 +409,14 @@ void QueryPipeline::addPipelineBefore(QueryPipeline pipeline)
 
     Pipes pipes;
     pipes.emplace_back(std::move(pipe));
-    pipes.emplace_back(QueryPipeline::getPipe(std::move(pipeline)));
+    pipes.emplace_back(QueryPipelineBuilder::getPipe(std::move(pipeline)));
     pipe = Pipe::unitePipes(std::move(pipes), collected_processors, true);
 
     auto processor = std::make_shared<DelayedPortsProcessor>(getHeader(), pipe.numOutputPorts(), delayed_streams, true);
     addTransform(std::move(processor));
 }
 
-void QueryPipeline::setProgressCallback(const ProgressCallback & callback)
+void QueryPipelineBuilder::setProgressCallback(const ProgressCallback & callback)
 {
     for (auto & processor : pipe.processors)
     {
@@ -425,7 +425,7 @@ void QueryPipeline::setProgressCallback(const ProgressCallback & callback)
     }
 }
 
-void QueryPipeline::setProcessListElement(QueryStatus * elem)
+void QueryPipelineBuilder::setProcessListElement(QueryStatus * elem)
 {
     process_list_element = elem;
 
@@ -436,7 +436,7 @@ void QueryPipeline::setProcessListElement(QueryStatus * elem)
     }
 }
 
-void QueryPipeline::initRowsBeforeLimit()
+void QueryPipelineBuilder::initRowsBeforeLimit()
 {
     RowsBeforeLimitCounterPtr rows_before_limit_at_least;
 
@@ -530,7 +530,7 @@ void QueryPipeline::initRowsBeforeLimit()
         output_format->setRowsBeforeLimitCounter(rows_before_limit_at_least);
 }
 
-PipelineExecutorPtr QueryPipeline::execute()
+PipelineExecutorPtr QueryPipelineBuilder::execute()
 {
     if (!isCompleted())
         throw Exception("Cannot execute pipeline because it is not completed.", ErrorCodes::LOGICAL_ERROR);
@@ -538,13 +538,13 @@ PipelineExecutorPtr QueryPipeline::execute()
     return std::make_shared<PipelineExecutor>(pipe.processors, process_list_element);
 }
 
-void QueryPipeline::setCollectedProcessors(Processors * processors)
+void QueryPipelineBuilder::setCollectedProcessors(Processors * processors)
 {
     pipe.collected_processors = processors;
 }
 
 
-QueryPipelineProcessorsCollector::QueryPipelineProcessorsCollector(QueryPipeline & pipeline_, IQueryPlanStep * step_)
+QueryPipelineProcessorsCollector::QueryPipelineProcessorsCollector(QueryPipelineBuilder & pipeline_, IQueryPlanStep * step_)
     : pipeline(pipeline_), step(step_)
 {
     pipeline.setCollectedProcessors(&processors);

src/Processors/QueryPipelineBuilder.h

@@ -29,15 +29,15 @@ struct ExpressionActionsSettings;
 class IJoin;
 using JoinPtr = std::shared_ptr<IJoin>;
 
-class QueryPipeline
+class QueryPipelineBuilder
 {
 public:
-    QueryPipeline() = default;
-    ~QueryPipeline() = default;
-    QueryPipeline(QueryPipeline &&) = default;
-    QueryPipeline(const QueryPipeline &) = delete;
-    QueryPipeline & operator= (QueryPipeline && rhs) = default;
-    QueryPipeline & operator= (const QueryPipeline & rhs) = delete;
+    QueryPipelineBuilder() = default;
+    ~QueryPipelineBuilder() = default;
+    QueryPipelineBuilder(QueryPipelineBuilder &&) = default;
+    QueryPipelineBuilder(const QueryPipelineBuilder &) = delete;
+    QueryPipelineBuilder & operator= (QueryPipelineBuilder && rhs) = default;
+    QueryPipelineBuilder & operator= (const QueryPipelineBuilder & rhs) = delete;
 
     /// All pipes must have same header.
     void init(Pipe pipe);
@@ -91,16 +91,16 @@ public:
 
     /// Unite several pipelines together. Result pipeline would have common_header structure.
     /// If collector is used, it will collect only newly-added processors, but not processors from pipelines.
-    static QueryPipeline unitePipelines(
-            std::vector<std::unique_ptr<QueryPipeline>> pipelines,
+    static QueryPipelineBuilder unitePipelines(
+            std::vector<std::unique_ptr<QueryPipelineBuilder>> pipelines,
             size_t max_threads_limit = 0,
             Processors * collected_processors = nullptr);
 
     /// Join two pipelines together using JoinPtr.
     /// If collector is used, it will collect only newly-added processors, but not processors from pipelines.
-    static std::unique_ptr<QueryPipeline> joinPipelines(
-        std::unique_ptr<QueryPipeline> left,
-        std::unique_ptr<QueryPipeline> right,
+    static std::unique_ptr<QueryPipelineBuilder> joinPipelines(
+        std::unique_ptr<QueryPipelineBuilder> left,
+        std::unique_ptr<QueryPipelineBuilder> right,
         JoinPtr join,
         size_t max_block_size,
         Processors * collected_processors = nullptr);
@@ -108,7 +108,7 @@ public:
     /// Add other pipeline and execute it before current one.
     /// Pipeline must have empty header, it should not generate any chunk.
     /// This is used for CreatingSets.
-    void addPipelineBefore(QueryPipeline pipeline);
+    void addPipelineBefore(QueryPipelineBuilder pipeline);
 
     void addCreatingSetsTransform(const Block & res_header, SubqueryForSet subquery_for_set, const SizeLimits & limits, ContextPtr context);
 
@@ -154,7 +154,7 @@ public:
     }
 
     /// Convert query pipeline to pipe.
-    static Pipe getPipe(QueryPipeline pipeline) { return std::move(pipeline.pipe); }
+    static Pipe getPipe(QueryPipelineBuilder pipeline) { return std::move(pipeline.pipe); }
 
 private:
 
@@ -182,13 +182,13 @@ private:
 class QueryPipelineProcessorsCollector
 {
 public:
-    explicit QueryPipelineProcessorsCollector(QueryPipeline & pipeline_, IQueryPlanStep * step_ = nullptr);
+    explicit QueryPipelineProcessorsCollector(QueryPipelineBuilder & pipeline_, IQueryPlanStep * step_ = nullptr);
     ~QueryPipelineProcessorsCollector();
 
     Processors detachProcessors(size_t group = 0);
 
 private:
-    QueryPipeline & pipeline;
+    QueryPipelineBuilder & pipeline;
     IQueryPlanStep * step;
     Processors processors;
 };

src/Processors/QueryPlan/AggregatingStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/AggregatingStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/AggregatingTransform.h>
 #include <Processors/Transforms/AggregatingInOrderTransform.h>
 #include <Processors/Merges/AggregatingSortedTransform.h>
@@ -46,7 +46,7 @@ AggregatingStep::AggregatingStep(
 {
 }
 
-void AggregatingStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void AggregatingStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     QueryPipelineProcessorsCollector collector(pipeline, this);
 

src/Processors/QueryPlan/AggregatingStep.h

@@ -27,7 +27,7 @@ public:
 
     String getName() const override { return "Aggregating"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
 
@@ -57,4 +57,3 @@ private:
 };
 
 }
-

src/Processors/QueryPlan/ArrayJoinStep.cpp

@@ -1,7 +1,7 @@
 #include <Processors/QueryPlan/ArrayJoinStep.h>
 #include <Processors/Transforms/ArrayJoinTransform.h>
 #include <Processors/Transforms/ExpressionTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Interpreters/ArrayJoinAction.h>
 #include <Interpreters/ExpressionActions.h>
 #include <IO/Operators.h>
@@ -46,11 +46,11 @@ void ArrayJoinStep::updateInputStream(DataStream input_stream, Block result_head
     res_header = std::move(result_header);
 }
 
-void ArrayJoinStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings)
+void ArrayJoinStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings)
 {
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type)
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type)
     {
-        bool on_totals = stream_type == QueryPipeline::StreamType::Totals;
+        bool on_totals = stream_type == QueryPipelineBuilder::StreamType::Totals;
         return std::make_shared<ArrayJoinTransform>(header, array_join, on_totals);
     });
 

src/Processors/QueryPlan/ArrayJoinStep.h

@@ -13,7 +13,7 @@ public:
     explicit ArrayJoinStep(const DataStream & input_stream_, ArrayJoinActionPtr array_join_);
     String getName() const override { return "ArrayJoin"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Processors/QueryPlan/CreatingSetsStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/CreatingSetsStep.h>
 #include <Processors/QueryPlan/QueryPlan.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/CreatingSetsTransform.h>
 #include <IO/Operators.h>
 #include <Interpreters/ExpressionActions.h>
@@ -44,7 +44,7 @@ CreatingSetStep::CreatingSetStep(
 {
 }
 
-void CreatingSetStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void CreatingSetStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     pipeline.addCreatingSetsTransform(getOutputStream().header, std::move(subquery_for_set), network_transfer_limits, getContext());
 }
@@ -85,7 +85,7 @@ CreatingSetsStep::CreatingSetsStep(DataStreams input_streams_)
                             input_streams[i].header.dumpStructure());
 }
 
-QueryPipelinePtr CreatingSetsStep::updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings &)
+QueryPipelineBuilderPtr CreatingSetsStep::updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings &)
 {
     if (pipelines.empty())
         throw Exception("CreatingSetsStep cannot be created with no inputs", ErrorCodes::LOGICAL_ERROR);
@@ -96,11 +96,11 @@ QueryPipelinePtr CreatingSetsStep::updatePipeline(QueryPipelines pipelines, cons
 
     pipelines.erase(pipelines.begin());
 
-    QueryPipeline delayed_pipeline;
+    QueryPipelineBuilder delayed_pipeline;
     if (pipelines.size() > 1)
     {
         QueryPipelineProcessorsCollector collector(delayed_pipeline, this);
-        delayed_pipeline = QueryPipeline::unitePipelines(std::move(pipelines));
+        delayed_pipeline = QueryPipelineBuilder::unitePipelines(std::move(pipelines));
         processors = collector.detachProcessors();
     }
     else

src/Processors/QueryPlan/CreatingSetsStep.h

@@ -21,7 +21,7 @@ public:
 
     String getName() const override { return "CreatingSet"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;
@@ -39,7 +39,7 @@ public:
 
     String getName() const override { return "CreatingSets"; }
 
-    QueryPipelinePtr updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings &) override;
+    QueryPipelineBuilderPtr updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings &) override;
 
     void describePipeline(FormatSettings & settings) const override;
 

src/Processors/QueryPlan/CubeStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/CubeStep.h>
 #include <Processors/Transforms/CubeTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 namespace DB
 {
@@ -30,13 +30,13 @@ CubeStep::CubeStep(const DataStream & input_stream_, AggregatingTransformParamsP
         output_stream->distinct_columns.insert(params->params.src_header.getByPosition(key).name);
 }
 
-void CubeStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void CubeStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     pipeline.resize(1);
 
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
     {
-        if (stream_type == QueryPipeline::StreamType::Totals)
+        if (stream_type == QueryPipelineBuilder::StreamType::Totals)
             return nullptr;
 
         return std::make_shared<CubeTransform>(header, std::move(params));

src/Processors/QueryPlan/CubeStep.h

@@ -17,7 +17,7 @@ public:
 
     String getName() const override { return "Cube"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     const Aggregator::Params & getParams() const;
 private:

src/Processors/QueryPlan/DistinctStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/DistinctStep.h>
 #include <Processors/Transforms/DistinctTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <IO/Operators.h>
 #include <Common/JSONBuilder.h>
 
@@ -63,7 +63,7 @@ DistinctStep::DistinctStep(
     }
 }
 
-void DistinctStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void DistinctStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     if (checkColumnsAlreadyDistinct(columns, input_streams.front().distinct_columns))
         return;
@@ -71,9 +71,9 @@ void DistinctStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryP
     if (!pre_distinct)
         pipeline.resize(1);
 
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
     {
-        if (stream_type != QueryPipeline::StreamType::Main)
+        if (stream_type != QueryPipelineBuilder::StreamType::Main)
             return nullptr;
 
         return std::make_shared<DistinctTransform>(header, set_size_limits, limit_hint, columns);

src/Processors/QueryPlan/DistinctStep.h

@@ -18,7 +18,7 @@ public:
 
     String getName() const override { return "Distinct"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Processors/QueryPlan/ExpressionStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/ExpressionStep.h>
 #include <Processors/Transforms/ExpressionTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/JoiningTransform.h>
 #include <Interpreters/ExpressionActions.h>
 #include <IO/Operators.h>
@@ -52,7 +52,7 @@ void ExpressionStep::updateInputStream(DataStream input_stream, bool keep_header
     input_streams.emplace_back(std::move(input_stream));
 }
 
-void ExpressionStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings)
+void ExpressionStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings)
 {
     auto expression = std::make_shared<ExpressionActions>(actions_dag, settings.getActionsSettings());
 

src/Processors/QueryPlan/ExpressionStep.h

@@ -18,7 +18,7 @@ public:
     explicit ExpressionStep(const DataStream & input_stream_, ActionsDAGPtr actions_dag_);
     String getName() const override { return "Expression"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings) override;
 
     void updateInputStream(DataStream input_stream, bool keep_header);
 

src/Processors/QueryPlan/ExtremesStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/ExtremesStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 namespace DB
 {
@@ -25,7 +25,7 @@ ExtremesStep::ExtremesStep(const DataStream & input_stream_)
 {
 }
 
-void ExtremesStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void ExtremesStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     pipeline.addExtremesTransform();
 }

src/Processors/QueryPlan/ExtremesStep.h

@@ -11,7 +11,7 @@ public:
 
     String getName() const override { return "Extremes"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 };
 
 }

src/Processors/QueryPlan/FillingStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/FillingStep.h>
 #include <Processors/Transforms/FillingTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <IO/Operators.h>
 #include <Common/JSONBuilder.h>
 
@@ -36,11 +36,11 @@ FillingStep::FillingStep(const DataStream & input_stream_, SortDescription sort_
         throw Exception("FillingStep expects single input", ErrorCodes::LOGICAL_ERROR);
 }
 
-void FillingStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void FillingStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
     {
-        bool on_totals = stream_type == QueryPipeline::StreamType::Totals;
+        bool on_totals = stream_type == QueryPipelineBuilder::StreamType::Totals;
         return std::make_shared<FillingTransform>(header, sort_description, on_totals);
     });
 }

src/Processors/QueryPlan/FillingStep.h

@@ -13,7 +13,7 @@ public:
 
     String getName() const override { return "Filling"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Processors/QueryPlan/FilterStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/FilterStep.h>
 #include <Processors/Transforms/FilterTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/ExpressionTransform.h>
 #include <Interpreters/ExpressionActions.h>
 #include <IO/Operators.h>
@@ -65,13 +65,13 @@ void FilterStep::updateInputStream(DataStream input_stream, bool keep_header)
     input_streams.emplace_back(std::move(input_stream));
 }
 
-void FilterStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings)
+void FilterStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings)
 {
     auto expression = std::make_shared<ExpressionActions>(actions_dag, settings.getActionsSettings());
 
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type)
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type)
     {
-        bool on_totals = stream_type == QueryPipeline::StreamType::Totals;
+        bool on_totals = stream_type == QueryPipelineBuilder::StreamType::Totals;
         return std::make_shared<FilterTransform>(header, expression, filter_column_name, remove_filter_column, on_totals);
     });
 

src/Processors/QueryPlan/FilterStep.h

@@ -18,7 +18,7 @@ public:
         bool remove_filter_column_);
 
     String getName() const override { return "Filter"; }
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings) override;
 
     void updateInputStream(DataStream input_stream, bool keep_header);
 

src/Processors/QueryPlan/FinishSortingStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/FinishSortingStep.h>
 #include <Processors/Transforms/DistinctTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Merges/MergingSortedTransform.h>
 #include <Processors/Transforms/PartialSortingTransform.h>
 #include <Processors/Transforms/FinishSortingTransform.h>
@@ -54,7 +54,7 @@ void FinishSortingStep::updateLimit(size_t limit_)
     }
 }
 
-void FinishSortingStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void FinishSortingStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     bool need_finish_sorting = (prefix_description.size() < result_description.size());
     if (pipeline.getNumStreams() > 1)
@@ -74,9 +74,9 @@ void FinishSortingStep::transformPipeline(QueryPipeline & pipeline, const BuildQ
 
     if (need_finish_sorting)
     {
-        pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+        pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
         {
-            if (stream_type != QueryPipeline::StreamType::Main)
+            if (stream_type != QueryPipelineBuilder::StreamType::Main)
                 return nullptr;
 
             return std::make_shared<PartialSortingTransform>(header, result_description, limit);

src/Processors/QueryPlan/FinishSortingStep.h

@@ -19,7 +19,7 @@ public:
 
     String getName() const override { return "FinishSorting"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Processors/QueryPlan/IQueryPlanStep.h

@@ -8,9 +8,9 @@ namespace JSONBuilder { class JSONMap; }
 namespace DB
 {
 
-class QueryPipeline;
-using QueryPipelinePtr = std::unique_ptr<QueryPipeline>;
-using QueryPipelines = std::vector<QueryPipelinePtr>;
+class QueryPipelineBuilder;
+using QueryPipelineBuilderPtr = std::unique_ptr<QueryPipelineBuilder>;
+using QueryPipelineBuilders = std::vector<QueryPipelineBuilderPtr>;
 
 class IProcessor;
 using ProcessorPtr = std::shared_ptr<IProcessor>;
@@ -80,7 +80,7 @@ public:
     ///   * header from each pipeline is the same as header from corresponding input_streams
     /// Result pipeline must contain any number of streams with compatible output header is hasOutputStream(),
     ///   or pipeline should be completed otherwise.
-    virtual QueryPipelinePtr updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings & settings) = 0;
+    virtual QueryPipelineBuilderPtr updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings & settings) = 0;
 
     const DataStreams & getInputStreams() const { return input_streams; }
 

src/Processors/QueryPlan/ISourceStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/ISourceStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 namespace DB
 {
@@ -9,9 +9,9 @@ ISourceStep::ISourceStep(DataStream output_stream_)
     output_stream = std::move(output_stream_);
 }
 
-QueryPipelinePtr ISourceStep::updatePipeline(QueryPipelines, const BuildQueryPipelineSettings & settings)
+QueryPipelineBuilderPtr ISourceStep::updatePipeline(QueryPipelineBuilders, const BuildQueryPipelineSettings & settings)
 {
-    auto pipeline = std::make_unique<QueryPipeline>();
+    auto pipeline = std::make_unique<QueryPipelineBuilder>();
     QueryPipelineProcessorsCollector collector(*pipeline, this);
     initializePipeline(*pipeline, settings);
     auto added_processors = collector.detachProcessors();

src/Processors/QueryPlan/ISourceStep.h

@@ -10,9 +10,9 @@ class ISourceStep : public IQueryPlanStep
 public:
     explicit ISourceStep(DataStream output_stream_);
 
-    QueryPipelinePtr updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings & settings) override;
+    QueryPipelineBuilderPtr updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings & settings) override;
 
-    virtual void initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings) = 0;
+    virtual void initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings) = 0;
 
     void describePipeline(FormatSettings & settings) const override;
 

src/Processors/QueryPlan/ITransformingStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/ITransformingStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 namespace DB
 {
@@ -36,7 +36,7 @@ DataStream ITransformingStep::createOutputStream(
 }
 
 
-QueryPipelinePtr ITransformingStep::updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings & settings)
+QueryPipelineBuilderPtr ITransformingStep::updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings & settings)
 {
     if (collect_processors)
     {

src/Processors/QueryPlan/ITransformingStep.h

@@ -48,9 +48,9 @@ public:
 
     ITransformingStep(DataStream input_stream, Block output_header, Traits traits, bool collect_processors_ = true);
 
-    QueryPipelinePtr updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings & settings) override;
+    QueryPipelineBuilderPtr updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings & settings) override;
 
-    virtual void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings) = 0;
+    virtual void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings) = 0;
 
     const TransformTraits & getTransformTraits() const { return transform_traits; }
     const DataStreamTraits & getDataStreamTraits() const { return data_stream_traits; }

src/Processors/QueryPlan/IntersectOrExceptStep.cpp

@@ -2,7 +2,7 @@
 
 #include <Interpreters/Context.h>
 #include <Interpreters/ExpressionActions.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Sources/NullSource.h>
 #include <Processors/Transforms/ExpressionTransform.h>
 #include <Processors/Transforms/IntersectOrExceptTransform.h>
@@ -39,9 +39,9 @@ IntersectOrExceptStep::IntersectOrExceptStep(
     output_stream = DataStream{.header = header};
 }
 
-QueryPipelinePtr IntersectOrExceptStep::updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings &)
+QueryPipelineBuilderPtr IntersectOrExceptStep::updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings &)
 {
-    auto pipeline = std::make_unique<QueryPipeline>();
+    auto pipeline = std::make_unique<QueryPipelineBuilder>();
     QueryPipelineProcessorsCollector collector(*pipeline, this);
 
     if (pipelines.empty())
@@ -72,7 +72,7 @@ QueryPipelinePtr IntersectOrExceptStep::updatePipeline(QueryPipelines pipelines,
         cur_pipeline->addTransform(std::make_shared<ResizeProcessor>(header, cur_pipeline->getNumStreams(), 1));
     }
 
-    *pipeline = QueryPipeline::unitePipelines(std::move(pipelines), max_threads);
+    *pipeline = QueryPipelineBuilder::unitePipelines(std::move(pipelines), max_threads);
     pipeline->addTransform(std::make_shared<IntersectOrExceptTransform>(header, current_operator));
 
     processors = collector.detachProcessors();

src/Processors/QueryPlan/IntersectOrExceptStep.h

@@ -16,7 +16,7 @@ public:
 
     String getName() const override { return "IntersectOrExcept"; }
 
-    QueryPipelinePtr updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings & settings) override;
+    QueryPipelineBuilderPtr updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings & settings) override;
 
     void describePipeline(FormatSettings & settings) const override;
 

src/Processors/QueryPlan/JoinStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/JoinStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/JoiningTransform.h>
 #include <Interpreters/IJoin.h>
 
@@ -26,12 +26,12 @@ JoinStep::JoinStep(
     };
 }
 
-QueryPipelinePtr JoinStep::updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings &)
+QueryPipelineBuilderPtr JoinStep::updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings &)
 {
     if (pipelines.size() != 2)
         throw Exception(ErrorCodes::LOGICAL_ERROR, "JoinStep expect two input steps");
 
-    return QueryPipeline::joinPipelines(std::move(pipelines[0]), std::move(pipelines[1]), join, max_block_size, &processors);
+    return QueryPipelineBuilder::joinPipelines(std::move(pipelines[0]), std::move(pipelines[1]), join, max_block_size, &processors);
 }
 
 void JoinStep::describePipeline(FormatSettings & settings) const
@@ -67,7 +67,7 @@ FilledJoinStep::FilledJoinStep(const DataStream & input_stream_, JoinPtr join_,
         throw Exception(ErrorCodes::LOGICAL_ERROR, "FilledJoinStep expects Join to be filled");
 }
 
-void FilledJoinStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void FilledJoinStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     bool default_totals = false;
     if (!pipeline.hasTotals() && join->getTotals())
@@ -78,9 +78,9 @@ void FilledJoinStep::transformPipeline(QueryPipeline & pipeline, const BuildQuer
 
     auto finish_counter = std::make_shared<JoiningTransform::FinishCounter>(pipeline.getNumStreams());
 
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type)
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type)
     {
-        bool on_totals = stream_type == QueryPipeline::StreamType::Totals;
+        bool on_totals = stream_type == QueryPipelineBuilder::StreamType::Totals;
         auto counter = on_totals ? nullptr : finish_counter;
         return std::make_shared<JoiningTransform>(header, join, max_block_size, on_totals, default_totals, counter);
     });

src/Processors/QueryPlan/JoinStep.h

@@ -20,7 +20,7 @@ public:
 
     String getName() const override { return "Join"; }
 
-    QueryPipelinePtr updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings &) override;
+    QueryPipelineBuilderPtr updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings &) override;
 
     void describePipeline(FormatSettings & settings) const override;
 
@@ -40,7 +40,7 @@ public:
     FilledJoinStep(const DataStream & input_stream_, JoinPtr join_, size_t max_block_size_);
 
     String getName() const override { return "FilledJoin"; }
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
 private:
     JoinPtr join;

src/Processors/QueryPlan/LimitByStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/LimitByStep.h>
 #include <Processors/Transforms/LimitByTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <IO/Operators.h>
 #include <Common/JSONBuilder.h>
 
@@ -34,13 +34,13 @@ LimitByStep::LimitByStep(
 }
 
 
-void LimitByStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void LimitByStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     pipeline.resize(1);
 
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
     {
-        if (stream_type != QueryPipeline::StreamType::Main)
+        if (stream_type != QueryPipelineBuilder::StreamType::Main)
             return nullptr;
 
         return std::make_shared<LimitByTransform>(header, group_length, group_offset, columns);

src/Processors/QueryPlan/LimitByStep.h

@@ -14,7 +14,7 @@ public:
 
     String getName() const override { return "LimitBy"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Processors/QueryPlan/LimitStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/LimitStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/LimitTransform.h>
 #include <IO/Operators.h>
 #include <Common/JSONBuilder.h>
@@ -43,7 +43,7 @@ void LimitStep::updateInputStream(DataStream input_stream)
     output_stream = createOutputStream(input_streams.front(), output_stream->header, getDataStreamTraits());
 }
 
-void LimitStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void LimitStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     auto transform = std::make_shared<LimitTransform>(
         pipeline.getHeader(), limit, offset, pipeline.getNumStreams(), always_read_till_end, with_ties, description);

src/Processors/QueryPlan/LimitStep.h

@@ -18,7 +18,7 @@ public:
 
     String getName() const override { return "Limit"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Processors/QueryPlan/MergeSortingStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/MergeSortingStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/MergeSortingTransform.h>
 #include <IO/Operators.h>
 #include <Common/JSONBuilder.h>
@@ -57,11 +57,11 @@ void MergeSortingStep::updateLimit(size_t limit_)
     }
 }
 
-void MergeSortingStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void MergeSortingStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
     {
-        if (stream_type == QueryPipeline::StreamType::Totals)
+        if (stream_type == QueryPipelineBuilder::StreamType::Totals)
             return nullptr;
 
         return std::make_shared<MergeSortingTransform>(

src/Processors/QueryPlan/MergeSortingStep.h

@@ -24,7 +24,7 @@ public:
 
     String getName() const override { return "MergeSorting"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;
@@ -45,4 +45,3 @@ private:
 };
 
 }
-

src/Processors/QueryPlan/MergingAggregatedStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/MergingAggregatedStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/AggregatingTransform.h>
 #include <Processors/Transforms/MergingAggregatedTransform.h>
 #include <Processors/Transforms/MergingAggregatedMemoryEfficientTransform.h>
@@ -40,7 +40,7 @@ MergingAggregatedStep::MergingAggregatedStep(
         output_stream->distinct_columns.insert(params->params.intermediate_header.getByPosition(key).name);
 }
 
-void MergingAggregatedStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void MergingAggregatedStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     if (!memory_efficient_aggregation)
     {

src/Processors/QueryPlan/MergingAggregatedStep.h

@@ -21,7 +21,7 @@ public:
 
     String getName() const override { return "MergingAggregated"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Processors/QueryPlan/MergingSortedStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/MergingSortedStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Merges/MergingSortedTransform.h>
 #include <IO/Operators.h>
 #include <Common/JSONBuilder.h>
@@ -47,7 +47,7 @@ void MergingSortedStep::updateLimit(size_t limit_)
     }
 }
 
-void MergingSortedStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void MergingSortedStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     /// If there are several streams, then we merge them into one
     if (pipeline.getNumStreams() > 1)

src/Processors/QueryPlan/MergingSortedStep.h

@@ -19,7 +19,7 @@ public:
 
     String getName() const override { return "MergingSorted"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;
@@ -34,5 +34,3 @@ private:
 };
 
 }
-
-

src/Processors/QueryPlan/OffsetStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/OffsetStep.h>
 #include <Processors/OffsetTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <IO/Operators.h>
 #include <Common/JSONBuilder.h>
 
@@ -29,7 +29,7 @@ OffsetStep::OffsetStep(const DataStream & input_stream_, size_t offset_)
 {
 }
 
-void OffsetStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void OffsetStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     auto transform = std::make_shared<OffsetTransform>(
             pipeline.getHeader(), offset, pipeline.getNumStreams());

src/Processors/QueryPlan/OffsetStep.h

@@ -13,7 +13,7 @@ public:
 
     String getName() const override { return "Offset"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Processors/QueryPlan/PartialSortingStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/PartialSortingStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/PartialSortingTransform.h>
 #include <Processors/Transforms/LimitsCheckingTransform.h>
 #include <IO/Operators.h>
@@ -47,11 +47,11 @@ void PartialSortingStep::updateLimit(size_t limit_)
     }
 }
 
-void PartialSortingStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void PartialSortingStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
     {
-        if (stream_type != QueryPipeline::StreamType::Main)
+        if (stream_type != QueryPipelineBuilder::StreamType::Main)
             return nullptr;
 
         return std::make_shared<PartialSortingTransform>(header, sort_description, limit);
@@ -61,9 +61,9 @@ void PartialSortingStep::transformPipeline(QueryPipeline & pipeline, const Build
     limits.mode = LimitsMode::LIMITS_CURRENT; //-V1048
     limits.size_limits = size_limits;
 
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
     {
-        if (stream_type != QueryPipeline::StreamType::Main)
+        if (stream_type != QueryPipelineBuilder::StreamType::Main)
             return nullptr;
 
         auto transform = std::make_shared<LimitsCheckingTransform>(header, limits);

src/Processors/QueryPlan/PartialSortingStep.h

@@ -18,7 +18,7 @@ public:
 
     String getName() const override { return "PartialSorting"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Processors/QueryPlan/QueryPlan.h

@@ -17,8 +17,8 @@ class DataStream;
 class IQueryPlanStep;
 using QueryPlanStepPtr = std::unique_ptr<IQueryPlanStep>;
 
-class QueryPipeline;
-using QueryPipelinePtr = std::unique_ptr<QueryPipeline>;
+class QueryPipelineBuilder;
+using QueryPipelineBuilderPtr = std::unique_ptr<QueryPipelineBuilder>;
 
 class WriteBuffer;
 
@@ -56,7 +56,7 @@ public:
 
     void optimize(const QueryPlanOptimizationSettings & optimization_settings);
 
-    QueryPipelinePtr buildQueryPipeline(
+    QueryPipelineBuilderPtr buildQueryPipeline(
         const QueryPlanOptimizationSettings & optimization_settings,
         const BuildQueryPipelineSettings & build_pipeline_settings);
 

src/Processors/QueryPlan/ReadFromMergeTree.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/ReadFromMergeTree.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/ConcatProcessor.h>
 #include <Processors/Transforms/ReverseTransform.h>
 #include <Processors/Transforms/ExpressionTransform.h>
@@ -927,7 +927,7 @@ ReadFromMergeTree::AnalysisResult ReadFromMergeTree::getAnalysisResult() const
     return std::get<ReadFromMergeTree::AnalysisResult>(result_ptr->result);
 }
 
-void ReadFromMergeTree::initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void ReadFromMergeTree::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     auto result = getAnalysisResult();
     LOG_DEBUG(

src/Processors/QueryPlan/ReadFromMergeTree.h

@@ -102,7 +102,7 @@ public:
 
     String getName() const override { return "ReadFromMergeTree"; }
 
-    void initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(FormatSettings & format_settings) const override;
     void describeIndexes(FormatSettings & format_settings) const override;

src/Processors/QueryPlan/ReadFromPreparedSource.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/ReadFromPreparedSource.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 namespace DB
 {
@@ -11,7 +11,7 @@ ReadFromPreparedSource::ReadFromPreparedSource(Pipe pipe_, std::shared_ptr<const
 {
 }
 
-void ReadFromPreparedSource::initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void ReadFromPreparedSource::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     for (const auto & processor : pipe.getProcessors())
         processors.emplace_back(processor);

src/Processors/QueryPlan/ReadFromPreparedSource.h

@@ -13,7 +13,7 @@ public:
 
     String getName() const override { return "ReadFromPreparedSource"; }
 
-    void initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
 private:
     Pipe pipe;

src/Processors/QueryPlan/ReadFromRemote.cpp

@@ -164,7 +164,7 @@ void ReadFromRemote::addLazyPipe(Pipes & pipes, const ClusterProxy::IStreamFacto
         if (try_results.empty() || local_delay < max_remote_delay)
         {
             auto plan = createLocalPlan(query, header, context, stage, shard_num, shard_count);
-            return QueryPipeline::getPipe(std::move(*plan->buildQueryPipeline(
+            return QueryPipelineBuilder::getPipe(std::move(*plan->buildQueryPipeline(
                 QueryPlanOptimizationSettings::fromContext(context),
                 BuildQueryPipelineSettings::fromContext(context))));
         }
@@ -220,7 +220,7 @@ void ReadFromRemote::addPipe(Pipes & pipes, const ClusterProxy::IStreamFactory::
     addConvertingActions(pipes.back(), output_stream->header);
 }
 
-void ReadFromRemote::initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void ReadFromRemote::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     Pipes pipes;
     for (const auto & shard : shards)

src/Processors/QueryPlan/ReadFromRemote.h

@@ -34,7 +34,7 @@ public:
 
     String getName() const override { return "ReadFromRemote"; }
 
-    void initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
 private:
     ClusterProxy::IStreamFactory::Shards shards;

src/Processors/QueryPlan/ReadNothingStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/ReadNothingStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Sources/NullSource.h>
 
 namespace DB
@@ -10,7 +10,7 @@ ReadNothingStep::ReadNothingStep(Block output_header)
 {
 }
 
-void ReadNothingStep::initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void ReadNothingStep::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     pipeline.init(Pipe(std::make_shared<NullSource>(getOutputStream().header)));
 }

src/Processors/QueryPlan/ReadNothingStep.h

@@ -12,7 +12,7 @@ public:
 
     String getName() const override { return "ReadNothing"; }
 
-    void initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 };
 
 }

src/Processors/QueryPlan/RollupStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/RollupStep.h>
 #include <Processors/Transforms/RollupTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 
 namespace DB
 {
@@ -30,13 +30,13 @@ RollupStep::RollupStep(const DataStream & input_stream_, AggregatingTransformPar
         output_stream->distinct_columns.insert(params->params.src_header.getByPosition(key).name);
 }
 
-void RollupStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void RollupStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     pipeline.resize(1);
 
-    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
     {
-        if (stream_type == QueryPipeline::StreamType::Totals)
+        if (stream_type == QueryPipelineBuilder::StreamType::Totals)
             return nullptr;
 
         return std::make_shared<RollupTransform>(header, std::move(params));

src/Processors/QueryPlan/RollupStep.h

@@ -16,7 +16,7 @@ public:
 
     String getName() const override { return "Rollup"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
 private:
     AggregatingTransformParamsPtr params;

src/Processors/QueryPlan/SettingQuotaAndLimitsStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/SettingQuotaAndLimitsStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Storages/IStorage.h>
 
 namespace DB
@@ -39,7 +39,7 @@ SettingQuotaAndLimitsStep::SettingQuotaAndLimitsStep(
 {
 }
 
-void SettingQuotaAndLimitsStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void SettingQuotaAndLimitsStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     /// Table lock is stored inside pipeline here.
     pipeline.setLimits(limits);

src/Processors/QueryPlan/SettingQuotaAndLimitsStep.h

@@ -32,7 +32,7 @@ public:
 
     String getName() const override { return "SettingQuotaAndLimits"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
 private:
     ContextPtr context;

src/Processors/QueryPlan/TotalsHavingStep.cpp

@@ -1,6 +1,6 @@
 #include <Processors/QueryPlan/TotalsHavingStep.h>
 #include <Processors/Transforms/DistinctTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Transforms/TotalsHavingTransform.h>
 #include <Interpreters/ExpressionActions.h>
 #include <IO/Operators.h>
@@ -49,7 +49,7 @@ TotalsHavingStep::TotalsHavingStep(
 {
 }
 
-void TotalsHavingStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings)
+void TotalsHavingStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings)
 {
     auto expression_actions = actions_dag ? std::make_shared<ExpressionActions>(actions_dag, settings.getActionsSettings()) : nullptr;
 

src/Processors/QueryPlan/TotalsHavingStep.h

@@ -24,7 +24,7 @@ public:
 
     String getName() const override { return "TotalsHaving"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;
@@ -41,4 +41,3 @@ private:
 };
 
 }
-

src/Processors/QueryPlan/UnionStep.cpp

@@ -1,5 +1,5 @@
 #include <Processors/QueryPlan/UnionStep.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Sources/NullSource.h>
 #include <Processors/Transforms/ExpressionTransform.h>
 #include <Interpreters/ExpressionActions.h>
@@ -37,9 +37,9 @@ UnionStep::UnionStep(DataStreams input_streams_, size_t max_threads_)
         output_stream = DataStream{.header = header};
 }
 
-QueryPipelinePtr UnionStep::updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings &)
+QueryPipelineBuilderPtr UnionStep::updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings &)
 {
-    auto pipeline = std::make_unique<QueryPipeline>();
+    auto pipeline = std::make_unique<QueryPipelineBuilder>();
     QueryPipelineProcessorsCollector collector(*pipeline, this);
 
     if (pipelines.empty())
@@ -71,7 +71,7 @@ QueryPipelinePtr UnionStep::updatePipeline(QueryPipelines pipelines, const Build
         }
     }
 
-    *pipeline = QueryPipeline::unitePipelines(std::move(pipelines), max_threads);
+    *pipeline = QueryPipelineBuilder::unitePipelines(std::move(pipelines), max_threads);
 
     processors = collector.detachProcessors();
     return pipeline;

src/Processors/QueryPlan/UnionStep.h

@@ -13,7 +13,7 @@ public:
 
     String getName() const override { return "Union"; }
 
-    QueryPipelinePtr updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings &) override;
+    QueryPipelineBuilderPtr updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings &) override;
 
     void describePipeline(FormatSettings & settings) const override;
 

src/Processors/QueryPlan/WindowStep.cpp

@@ -2,7 +2,7 @@
 
 #include <Processors/Transforms/WindowTransform.h>
 #include <Processors/Transforms/ExpressionTransform.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Interpreters/ExpressionActions.h>
 #include <IO/Operators.h>
 #include <Common/JSONBuilder.h>
@@ -63,7 +63,7 @@ WindowStep::WindowStep(const DataStream & input_stream_,
 
 }
 
-void WindowStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &)
+void WindowStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
 {
     // This resize is needed for cases such as `over ()` when we don't have a
     // sort node, and the input might have multiple streams. The sort node would

src/Processors/QueryPlan/WindowStep.h

@@ -20,7 +20,7 @@ public:
 
     String getName() const override { return "Window"; }
 
-    void transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings &) override;
+    void transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) override;
 
     void describeActions(JSONBuilder::JSONMap & map) const override;
     void describeActions(FormatSettings & settings) const override;

src/Storages/IStorage.h

@@ -7,7 +7,7 @@
 #include <Interpreters/CancellationCode.h>
 #include <Interpreters/Context_fwd.h>
 #include <Interpreters/StorageID.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Storages/CheckResults.h>
 #include <Storages/ColumnDependency.h>
 #include <Storages/IStorage_fwd.h>
@@ -54,8 +54,8 @@ using QueryPlanPtr = std::unique_ptr<QueryPlan>;
 class SinkToStorage;
 using SinkToStoragePtr = std::shared_ptr<SinkToStorage>;
 
-class QueryPipeline;
-using QueryPipelinePtr = std::unique_ptr<QueryPipeline>;
+class QueryPipelineBuilder;
+using QueryPipelineBuilderPtr = std::unique_ptr<QueryPipelineBuilder>;
 
 class IStoragePolicy;
 using StoragePolicyPtr = std::shared_ptr<const IStoragePolicy>;

src/Storages/MergeTree/StorageFromMergeTreeDataPart.h

@@ -6,7 +6,7 @@
 #include <Processors/QueryPlan/QueryPlan.h>
 #include <Processors/QueryPlan/Optimizations/QueryPlanOptimizationSettings.h>
 #include <Processors/QueryPlan/BuildQueryPipelineSettings.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Core/Defines.h>
 
 #include <common/shared_ptr_helper.h>

src/Storages/StorageMerge.cpp

@@ -380,7 +380,7 @@ Pipe StorageMerge::createSources(
 
     if (!storage)
     {
-        pipe = QueryPipeline::getPipe(InterpreterSelectQuery(
+        pipe = QueryPipelineBuilder::getPipe(InterpreterSelectQuery(
             modified_query_info.query, modified_context,
             std::make_shared<OneBlockInputStream>(header),
             SelectQueryOptions(processed_stage).analyze()).execute().pipeline);

src/Storages/StorageS3.h

@@ -82,7 +82,7 @@ private:
 
 
     std::unique_ptr<ReadBuffer> read_buf;
-    std::unique_ptr<QueryPipeline> pipeline;
+    std::unique_ptr<QueryPipelineBuilder> pipeline;
     std::unique_ptr<PullingPipelineExecutor> reader;
     bool initialized = false;
     bool with_file_column = false;

src/Storages/StorageURL.cpp

@@ -20,7 +20,7 @@
 
 #include <Poco/Net/HTTPRequest.h>
 #include <Processors/Sources/SourceWithProgress.h>
-#include <Processors/QueryPipeline.h>
+#include <Processors/QueryPipelineBuilder.h>
 #include <Processors/Executors/PullingPipelineExecutor.h>
 #include <common/logger_useful.h>
 #include <algorithm>
@@ -105,7 +105,7 @@ namespace
                 compression_method);
 
             auto input_format = FormatFactory::instance().getInput(format, *read_buf, sample_block, context, max_block_size, format_settings);
-            pipeline = std::make_unique<QueryPipeline>();
+            pipeline = std::make_unique<QueryPipelineBuilder>();
             pipeline->init(Pipe(input_format));
 
             pipeline->addSimpleTransform([&](const Block & cur_header)
@@ -139,7 +139,7 @@ namespace
     private:
         String name;
         std::unique_ptr<ReadBuffer> read_buf;
-        std::unique_ptr<QueryPipeline> pipeline;
+        std::unique_ptr<QueryPipelineBuilder> pipeline;
         std::unique_ptr<PullingPipelineExecutor> reader;
     };
 }