separate different streaming join

src/Core/DataBlockWithShard.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include <Core/Block.h>
+#include <Core/LightChunk.h>
+
+namespace DB
+{
+template <typename DataBlock>
+struct DataBlockWithShard
+{
+    DataBlock block;
+    int32_t shard;
+
+    DataBlockWithShard(DataBlock && block_, int32_t shard_) : block(std::move(block_)), shard(shard_) { }
+};
+
+using BlockWithShard = DataBlockWithShard<Block>;
+using BlocksWithShard = std::vector<BlockWithShard>;
+
+using LightChunkWithShard = DataBlockWithShard<LightChunk>;
+using LightChunksWithShard = std::vector<LightChunkWithShard>;
+}
+

src/Core/LightChunk.h

@@ -27,6 +27,9 @@ struct LightChunk
     void concat(const LightChunk & other)
     {
         auto added_rows = other.rows();
+        if (added_rows <= 0)
+            return;
+
         assert(columns() == other.columns());
         for (size_t c = 0; auto & col : data)
         {
@@ -35,9 +38,21 @@ struct LightChunk
         }
     }
 
+    LightChunk cloneEmpty() const
+    {
+        LightChunk res;
+        res.data.reserve(data.size());
+
+        for (const auto & elem : data)
+            res.data.emplace_back(elem->cloneEmpty());
+
+        return res;
+    }
+
     size_t rows() const noexcept { return data.empty() ? 0 : data[0]->size(); }
     size_t columns() const noexcept { return data.size(); }
 
+    Columns & getColumns() noexcept { return data; }
     const Columns & getColumns() const noexcept { return data; }
     Columns detachColumns() noexcept { return std::move(data); }
 
@@ -88,7 +103,9 @@ struct LightChunkWithTimestamp
     LightChunkWithTimestamp() = default;
     LightChunkWithTimestamp(Columns && data_) : chunk(std::move(data_)) { }
     LightChunkWithTimestamp(Chunk && chunk_, Int64 min_ts, Int64 max_ts)
-        : chunk(std::move(chunk_)), min_timestamp(min_ts), max_timestamp(max_ts) { }
+        : chunk(std::move(chunk_)), min_timestamp(min_ts), max_timestamp(max_ts)
+    {
+    }
     LightChunkWithTimestamp(const Block & block)
         : chunk(block), min_timestamp(block.minTimestamp()), max_timestamp(block.maxTimestamp()) { }
 
@@ -122,4 +139,4 @@ struct LightChunkWithTimestamp
     Int64 maxTimestamp() const noexcept { return max_timestamp; }
 };
 
-}
+}

src/Interpreters/ExpressionAnalyzer.cpp

@@ -2331,7 +2331,7 @@ std::shared_ptr<IJoin> SelectQueryExpressionAnalyzer::chooseJoinAlgorithmStreami
         return std::make_shared<Streaming::ConcurrentHashJoin>(
             analyzed_join, max_threads, std::move(left_join_stream_desc), std::move(right_join_stream_desc));
     else
-        return std::make_shared<Streaming::HashJoin>(analyzed_join, std::move(left_join_stream_desc), std::move(right_join_stream_desc));
+        return Streaming::HashJoin::create(analyzed_join, std::move(left_join_stream_desc), std::move(right_join_stream_desc));
 }
 /// proton : ends
 

src/Interpreters/Streaming/AsofHashJoin.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <Interpreters/Streaming/HashJoin.h>
+
+namespace DB
+{
+namespace Streaming
+{
+class AsofHashJoin final : public HashJoin
+{
+public:
+    using HashJoin::HashJoin;
+    HashJoinType type() const override { return HashJoinType::Asof; }
+};
+
+}
+}

src/Interpreters/Streaming/BidirectionalAllHashJoin.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <Interpreters/Streaming/HashJoin.h>
+
+namespace DB
+{
+namespace Streaming
+{
+class BidirectionalAllHashJoin final : public HashJoin
+{
+public:
+    using HashJoin::HashJoin;
+    HashJoinType type() const override { return HashJoinType::BidirectionalAll; }
+};
+
+}
+}

src/Interpreters/Streaming/BidirectionalChangelogHashJoin.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <Interpreters/Streaming/HashJoin.h>
+
+namespace DB
+{
+namespace Streaming
+{
+class BidirectionalChangelogHashJoin final : public HashJoin
+{
+public:
+    using HashJoin::HashJoin;
+    HashJoinType type() const override { return HashJoinType::BidirectionalChangelog; }
+};
+
+}
+}

src/Interpreters/Streaming/BidirectionalRangeHashJoin.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <Interpreters/Streaming/HashJoin.h>
+
+namespace DB
+{
+namespace Streaming
+{
+class BidirectionalRangeHashJoin final : public HashJoin
+{
+public:
+    using HashJoin::HashJoin;
+    HashJoinType type() const override { return HashJoinType::BidirectionalRange; }
+};
+
+}
+}

src/Interpreters/Streaming/ChangelogHashJoin.h

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <Interpreters/Streaming/HashJoin.h>
+
+namespace DB
+{
+namespace Streaming
+{
+class ChangelogHashJoin final : public HashJoin
+{
+public:
+    using HashJoin::HashJoin;
+    HashJoinType type() const override { return HashJoinType::Changelog; }
+};
+
+}
+}