fix tiflash crash when setting join_concurrent_build = 0

dbms/src/DataStreams/HashJoinBuildBlockInputStream.cpp

@@ -16,13 +16,12 @@
 #include <DataStreams/HashJoinBuildBlockInputStream.h>
 namespace DB
 {
-
 Block HashJoinBuildBlockInputStream::readImpl()
 {
     Block block = children.back()->read();
     if (!block)
         return block;
-    join->insertFromBlock(block, stream_index);
+    join->insertFromBlock(block, concurrency_build_index);
     return block;
 }
 

dbms/src/DataStreams/HashJoinBuildBlockInputStream.h

@@ -28,9 +28,9 @@ class HashJoinBuildBlockInputStream : public IProfilingBlockInputStream
     HashJoinBuildBlockInputStream(
         const BlockInputStreamPtr & input,
         JoinPtr join_,
-        size_t stream_index_,
+        size_t concurrency_build_index_,
         const String & req_id)
-        : stream_index(stream_index_)
+        : concurrency_build_index(concurrency_build_index_)
         , log(Logger::get(NAME, req_id))
     {
         children.push_back(input);
@@ -44,7 +44,7 @@ class HashJoinBuildBlockInputStream : public IProfilingBlockInputStream
 
 private:
     JoinPtr join;
-    size_t stream_index;
+    size_t concurrency_build_index;
     const LoggerPtr log;
 };
 

dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.cpp

@@ -697,9 +697,13 @@ void DAGQueryBlockInterpreter::handleJoin(const tipb::Join & join, DAGPipeline &
     recordJoinExecuteInfo(swap_join_side ? 0 : 1, join_ptr);
 
     // add a HashJoinBuildBlockInputStream to build a shared hash table
-    size_t stream_index = 0;
-    right_pipeline.transform(
-        [&](auto & stream) { stream = std::make_shared<HashJoinBuildBlockInputStream>(stream, join_ptr, stream_index++, log->identifier()); });
+    size_t concurrency_build_index = 0;
+    auto get_concurrency_build_index = [&concurrency_build_index, &join_build_concurrency]() {
+        return (concurrency_build_index++) % join_build_concurrency;
+    };
+    right_pipeline.transform([&](auto & stream) {
+        stream = std::make_shared<HashJoinBuildBlockInputStream>(stream, join_ptr, get_concurrency_build_index(), log->identifier());
+    });
     executeUnion(right_pipeline, max_streams, log, /*ignore_block=*/true);
 
     right_query.source = right_pipeline.firstStream();

dbms/src/Interpreters/Join.cpp

@@ -772,6 +772,8 @@ bool Join::insertFromBlock(const Block & block)
 /// the block should be valid.
 void Join::insertFromBlock(const Block & block, size_t stream_index)
 {
+    assert(stream_index < build_concurrency);
+
     if (empty())
         throw Exception("Logical error: Join was not initialized", ErrorCodes::LOGICAL_ERROR);
     std::shared_lock lock(rwlock);