From 6b25a368161760a89726cbba16b49df31faae5b0 Mon Sep 17 00:00:00 2001
From: Sergey Skvortsov <mail@bigredeye.me>
Date: Mon, 20 Jun 2022 03:13:01 +0300
Subject: [PATCH] fix: Add signal ports in JoiningTransform conditionally

---
 src/Processors/Transforms/JoiningTransform.cpp | 15 +++++++++------
 src/Processors/Transforms/JoiningTransform.h   |  2 +-
 src/QueryPipeline/QueryPipelineBuilder.cpp     |  2 +-
 3 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/src/Processors/Transforms/JoiningTransform.cpp b/src/Processors/Transforms/JoiningTransform.cpp
index 537d94b9b3d..b3c314dc8ce 100644
--- a/src/Processors/Transforms/JoiningTransform.cpp
+++ b/src/Processors/Transforms/JoiningTransform.cpp
@@ -29,7 +29,7 @@ JoiningTransform::JoiningTransform(
     bool on_totals_,
     bool default_totals_,
     FinishCounterPtr finish_counter_)
-    : IProcessor({input_header}, {transformHeader(input_header, join_), Block()})
+    : IProcessor({input_header}, {transformHeader(input_header, join_)})
     , join(std::move(join_))
     , on_totals(on_totals_)
     , default_totals(default_totals_)
@@ -37,15 +37,19 @@ JoiningTransform::JoiningTransform(
     , max_block_size(max_block_size_)
 {
     if (!join->isFilled())
-        inputs.emplace_back(Block(), this);
+    {
+        inputs.emplace_back(Block(), this); // Wait for FillingRightJoinSideTransform
+        if (!on_totals)
+            outputs.emplace_back(Block(), this); // Signal for DelayedJoinedBlocksTransform
+    }
 }
 
 JoiningTransform::~JoiningTransform() = default;
 
-OutputPort* JoiningTransform::getFinishedSignal()
+OutputPort & JoiningTransform::getFinishedSignal()
 {
     assert(outputs.size() == 2);
-    return &outputs.back();
+    return outputs.back();
 }
 
 IProcessor::Status JoiningTransform::prepare()
@@ -310,8 +314,7 @@ void FillingRightJoinSideTransform::work()
 
 
 DelayedJoinedBlocksTransform::DelayedJoinedBlocksTransform(Block input_header, JoinPtr join_)
-    : IProcessor({Block()}, {JoiningTransform::transformHeader(input_header, join_)})
-    , join{std::move(join_)}
+    : IProcessor({Block()}, {JoiningTransform::transformHeader(input_header, join_)}), join{std::move(join_)}
 {
 }
 
diff --git a/src/Processors/Transforms/JoiningTransform.h b/src/Processors/Transforms/JoiningTransform.h
index d5d56163c78..53024356566 100644
--- a/src/Processors/Transforms/JoiningTransform.h
+++ b/src/Processors/Transforms/JoiningTransform.h
@@ -53,7 +53,7 @@ public:
 
     static Block transformHeader(Block header, const JoinPtr & join);
 
-    OutputPort* getFinishedSignal();
+    OutputPort & getFinishedSignal();
 
     Status prepare() override;
     void work() override;
diff --git a/src/QueryPipeline/QueryPipelineBuilder.cpp b/src/QueryPipeline/QueryPipelineBuilder.cpp
index 1bb8e857b91..caa059a0a3f 100644
--- a/src/QueryPipeline/QueryPipelineBuilder.cpp
+++ b/src/QueryPipeline/QueryPipelineBuilder.cpp
@@ -401,7 +401,7 @@ std::unique_ptr<QueryPipelineBuilder> QueryPipelineBuilder::joinPipelines(
         auto joining = std::make_shared<JoiningTransform>(left->getHeader(), join, max_block_size, false, default_totals, finish_counter);
         connect(**lit, joining->getInputs().front());
         connect(**rit, joining->getInputs().back());
-        connect(*joining->getFinishedSignal(), barrier->addInputPort());
+        connect(joining->getFinishedSignal(), barrier->addInputPort());
 
         // Process delayed joined blocks when all JoiningTransform are finished.
         auto delayed = std::make_shared<DelayedJoinedBlocksTransform>(left->getHeader(), join);