Change copy of shared pointer to const reference of shared pointer

src/Processors/IProcessor.h

@@ -238,7 +238,7 @@ public:
     virtual bool isPartialResultProcessor() const { return false; }
     virtual bool supportPartialResultProcessor() const { return false; }
 
-    virtual ProcessorPtr getPartialResultProcessor(ProcessorPtr /*current_processor*/, UInt64 /*partial_result_limit*/, UInt64 /*partial_result_duration_ms*/)
+    virtual ProcessorPtr getPartialResultProcessor(const ProcessorPtr & /*current_processor*/, UInt64 /*partial_result_limit*/, UInt64 /*partial_result_duration_ms*/)
     {
         throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Method 'getPartialResultProcessor' is not implemented for {} processor", getName());
     }

src/Processors/LimitTransform.cpp

@@ -366,7 +366,7 @@ bool LimitTransform::sortColumnsEqualAt(const ColumnRawPtrs & current_chunk_sort
     return true;
 }
 
-ProcessorPtr LimitTransform::getPartialResultProcessor(ProcessorPtr /*current_processor*/, UInt64 partial_result_limit, UInt64 partial_result_duration_ms)
+ProcessorPtr LimitTransform::getPartialResultProcessor(const ProcessorPtr & /*current_processor*/, UInt64 partial_result_limit, UInt64 partial_result_duration_ms)
 {
     const auto & header = inputs.front().getHeader();
     return std::make_shared<LimitPartialResultTransform>(header, partial_result_limit, partial_result_duration_ms, limit, offset);

src/Processors/LimitTransform.h

@@ -56,7 +56,7 @@ private:
     bool sortColumnsEqualAt(const ColumnRawPtrs & current_chunk_sort_columns, UInt64 current_chunk_row_num) const;
 
     bool supportPartialResultProcessor() const override { return true; }
-    ProcessorPtr getPartialResultProcessor(ProcessorPtr current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms) override;
+    ProcessorPtr getPartialResultProcessor(const ProcessorPtr & current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms) override;
 
 public:
     LimitTransform(

src/Processors/Transforms/ExpressionTransform.cpp

@@ -25,7 +25,7 @@ void ExpressionTransform::transform(Chunk & chunk)
     chunk.setColumns(block.getColumns(), num_rows);
 }
 
-ProcessorPtr ExpressionTransform::getPartialResultProcessor(ProcessorPtr /*current_processor*/, UInt64 /*partial_result_limit*/, UInt64 /*partial_result_duration_ms*/)
+ProcessorPtr ExpressionTransform::getPartialResultProcessor(const ProcessorPtr & /*current_processor*/, UInt64 /*partial_result_limit*/, UInt64 /*partial_result_duration_ms*/)
 {
     const auto & header = getInputPort().getHeader();
     return std::make_shared<ExpressionTransform>(header, expression);

src/Processors/Transforms/ExpressionTransform.h

@@ -30,7 +30,7 @@ protected:
     void transform(Chunk & chunk) override;
 
     bool supportPartialResultProcessor() const override { return true; }
-    ProcessorPtr getPartialResultProcessor(ProcessorPtr current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms) override;
+    ProcessorPtr getPartialResultProcessor(const ProcessorPtr & current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms) override;
 
 private:
     ExpressionActionsPtr expression;

src/Processors/Transforms/LimitsCheckingTransform.cpp

@@ -75,7 +75,7 @@ void LimitsCheckingTransform::checkQuota(Chunk & chunk)
     }
 }
 
-ProcessorPtr LimitsCheckingTransform::getPartialResultProcessor(ProcessorPtr /*current_processor*/, UInt64 partial_result_limit, UInt64 partial_result_duration_ms)
+ProcessorPtr LimitsCheckingTransform::getPartialResultProcessor(const ProcessorPtr & /*current_processor*/, UInt64 partial_result_limit, UInt64 partial_result_duration_ms)
 {
     const auto & header = inputs.front().getHeader();
     return std::make_shared<PartialResultTransform>(header, partial_result_limit, partial_result_duration_ms);

src/Processors/Transforms/LimitsCheckingTransform.h

@@ -37,7 +37,7 @@ protected:
     void transform(Chunk & chunk) override;
 
     bool supportPartialResultProcessor() const override { return true; }
-    ProcessorPtr getPartialResultProcessor(ProcessorPtr current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms) override;
+    ProcessorPtr getPartialResultProcessor(const ProcessorPtr & current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms) override;
 
 private:
     StreamLocalLimits limits;

src/Processors/Transforms/MergeSortingTransform.cpp

@@ -333,7 +333,7 @@ void MergeSortingTransform::remerge()
     sum_bytes_in_blocks = new_sum_bytes_in_blocks;
 }
 
-ProcessorPtr MergeSortingTransform::getPartialResultProcessor(ProcessorPtr current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms)
+ProcessorPtr MergeSortingTransform::getPartialResultProcessor(const ProcessorPtr & current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms)
 {
     if (getName() != current_processor->getName() || current_processor.get() != this)
         throw Exception(
@@ -347,7 +347,7 @@ ProcessorPtr MergeSortingTransform::getPartialResultProcessor(ProcessorPtr curre
             static_cast<void*>(current_processor.get()));
 
     const auto & header = inputs.front().getHeader();
-    const auto & merge_sorting_processor = std::static_pointer_cast<MergeSortingTransform>(current_processor);
+    auto merge_sorting_processor = std::dynamic_pointer_cast<MergeSortingTransform>(current_processor);
     return std::make_shared<MergeSortingPartialResultTransform>(header, std::move(merge_sorting_processor), partial_result_limit, partial_result_duration_ms);
 }
 

src/Processors/Transforms/MergeSortingTransform.h

@@ -42,7 +42,7 @@ protected:
     Processors expandPipeline() override;
 
     bool supportPartialResultProcessor() const override { return true; }
-    ProcessorPtr getPartialResultProcessor(ProcessorPtr current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms) override;
+    ProcessorPtr getPartialResultProcessor(const ProcessorPtr & current_processor, UInt64 partial_result_limit, UInt64 partial_result_duration_ms) override;
 
 private:
     size_t max_bytes_before_remerge;

src/QueryPipeline/Pipe.cpp

@@ -628,7 +628,7 @@ void Pipe::addTransform(ProcessorPtr transform, InputPort * totals, InputPort *
     max_parallel_streams = std::max<size_t>(max_parallel_streams, output_ports.size());
 }
 
-void Pipe::addPartialResultSimpleTransform(ProcessorPtr transform, size_t partial_result_port_id)
+void Pipe::addPartialResultSimpleTransform(const ProcessorPtr & transform, size_t partial_result_port_id)
 {
     if (isPartialResultActive())
     {
@@ -640,7 +640,7 @@ void Pipe::addPartialResultSimpleTransform(ProcessorPtr transform, size_t partia
             return;
         }
 
-        auto partial_result_transform = transform->getPartialResultProcessor(std::move(transform), partial_result_limit, partial_result_duration_ms);
+        auto partial_result_transform = transform->getPartialResultProcessor(transform, partial_result_limit, partial_result_duration_ms);
 
         connectPartialResultPort(partial_result_port, partial_result_transform->getInputs().front());
 
@@ -650,7 +650,7 @@ void Pipe::addPartialResultSimpleTransform(ProcessorPtr transform, size_t partia
     }
 }
 
-void Pipe::addPartialResultTransform(ProcessorPtr transform)
+void Pipe::addPartialResultTransform(const ProcessorPtr & transform)
 {
     if (isPartialResultActive())
     {
@@ -665,7 +665,7 @@ void Pipe::addPartialResultTransform(ProcessorPtr transform)
             return;
         }
 
-        auto partial_result_transform = transform->getPartialResultProcessor(std::move(transform), partial_result_limit, partial_result_duration_ms);
+        auto partial_result_transform = transform->getPartialResultProcessor(transform, partial_result_limit, partial_result_duration_ms);
         auto & inputs = partial_result_transform->getInputs();
 
         if (inputs.size() != partial_result_ports.size())

src/QueryPipeline/Pipe.h

@@ -76,8 +76,8 @@ public:
     void addTransform(ProcessorPtr transform, OutputPort * totals, OutputPort * extremes);
     void addTransform(ProcessorPtr transform, InputPort * totals, InputPort * extremes);
 
-    void addPartialResultTransform(ProcessorPtr transform);
-    void addPartialResultSimpleTransform(ProcessorPtr transform, size_t partial_result_port_id);
+    void addPartialResultTransform(const ProcessorPtr & transform);
+    void addPartialResultSimpleTransform(const ProcessorPtr & transform, size_t partial_result_port_id);
     void connectPartialResultPort(OutputPort * partial_result_port, InputPort & partial_result_transform_port);
 
     enum class StreamType