set BuildTableState

Signed-off-by: fzhedu <[email protected]>
pingcap · Mar 14, 2022 · a986286 · a986286
1 parent c1724ba
commit a986286
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 17 deletions.
diff --git a/dbms/src/DataStreams/CreatingSetsBlockInputStream.cpp b/dbms/src/DataStreams/CreatingSetsBlockInputStream.cpp
@@ -109,7 +109,7 @@ void CreatingSetsBlockInputStream::createAll()
             for (auto & elem : subqueries_for_sets)
             {
                 if (elem.second.join)
-                    elem.second.join->setBuildTableState(0);
+                    elem.second.join->setBuildTableState(Join::BuildTableState::WAITING);
             }
         }
         Stopwatch watch;
@@ -143,15 +143,14 @@ void CreatingSetsBlockInputStream::createAll()
 
 void CreatingSetsBlockInputStream::createOne(SubqueryForSet & subquery)
 {
-    std::stringstream log_msg;
-    log_msg << std::fixed << std::setprecision(3);
-    log_msg << (subquery.set ? "Creating set. " : "")
-            << (subquery.join ? "Creating join. " : "") << (subquery.table ? "Filling temporary table. " : "") << " for task "
-            << mpp_task_id.toString();
+    auto log_msg = fmt::format("{} for task {}", (subquery.set ? "Creating set. " : subquery.join ? "Creating join. "
+                                                      : subquery.table                            ? "Filling temporary table. "
+                                                                                                  : ""),
+                               mpp_task_id.toString());
     Stopwatch watch;
     try
     {
-        LOG_FMT_DEBUG(log, "{}", log_msg.rdbuf()->str());
+        LOG_FMT_DEBUG(log, "{}", log_msg);
         BlockOutputStreamPtr table_out;
         if (subquery.table)
             table_out = subquery.table->write({}, {});
@@ -217,7 +216,7 @@ void CreatingSetsBlockInputStream::createOne(SubqueryForSet & subquery)
         if (subquery.join)
         {
             FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_mpp_hash_build);
-            subquery.join->setBuildTableState(1);
+            subquery.join->setBuildTableState(Join::BuildTableState::SUCCEED);
         }
 
         if (table_out)
@@ -264,8 +263,8 @@ void CreatingSetsBlockInputStream::createOne(SubqueryForSet & subquery)
         std::unique_lock<std::mutex> lock(exception_mutex);
         exception_from_workers.push_back(std::current_exception());
         if (subquery.join)
-            subquery.join->setBuildTableState(-1);
-        LOG_FMT_ERROR(log, "{} throw exception: {} In {} sec. ", log_msg.rdbuf()->str(), getCurrentExceptionMessage(false, true), watch.elapsedSeconds());
+            subquery.join->setBuildTableState(Join::BuildTableState::FAILED);
+        LOG_FMT_ERROR(log, "{} throw exception: {} In {} sec. ", log_msg, getCurrentExceptionMessage(false, true), watch.elapsedSeconds());
     }
 }
 

diff --git a/dbms/src/Flash/tests/exchange_perftest.cpp b/dbms/src/Flash/tests/exchange_perftest.cpp
@@ -458,7 +458,7 @@ struct ReceiverHelper
     {
         if (join_ptr)
         {
-            join_ptr->setBuildTableState(1);
+            join_ptr->setBuildTableState(Join::BuildTableState::SUCCEED);
             std::cout << fmt::format("Hash table size: {} bytes", join_ptr->getTotalByteCount()) << std::endl;
         }
     }

diff --git a/dbms/src/Interpreters/Join.cpp b/dbms/src/Interpreters/Join.cpp
@@ -70,7 +70,7 @@ Join::Join(const Names & key_names_left_, const Names & key_names_right_, bool u
     , other_condition_ptr(other_condition_ptr_)
     , original_strictness(strictness)
     , max_block_size_for_cross_join(max_block_size_)
-    , build_table_state(1)
+    , build_table_state(BuildTableState::SUCCEED)
     , log(getLogWithPrefix(log_, "Join"))
     , limits(limits)
 {
@@ -96,7 +96,7 @@ Join::Join(const Names & key_names_left_, const Names & key_names_right_, bool u
         throw Exception("Not supported: non right join with right conditions");
 }
 
-void Join::setBuildTableState(int state_)
+void Join::setBuildTableState(BuildTableState state_)
 {
     std::lock_guard<std::mutex> lk(build_table_mutex);
     build_table_state = state_;
@@ -1622,8 +1622,8 @@ void Join::joinBlock(Block & block) const
     {
         std::unique_lock lk(build_table_mutex);
 
-        build_table_cv.wait(lk, [&]() { return build_table_state != 0; });
-        if (build_table_state == -1) /// throw this exception once failed to build the hash table
+        build_table_cv.wait(lk, [&]() { return build_table_state != BuildTableState::WAITING; });
+        if (build_table_state == BuildTableState::FAILED) /// throw this exception once failed to build the hash table
             throw Exception("Build failed before join probe!");
     }
 

diff --git a/dbms/src/Interpreters/Join.h b/dbms/src/Interpreters/Join.h
@@ -142,7 +142,13 @@ class Join
     bool isBuildSetExceeded() const { return build_set_exceeded.load(); }
     size_t getNotJoinedStreamConcurrency() const { return build_concurrency; };
 
-    void setBuildTableState(int);
+    enum BuildTableState
+    {
+        WAITING,
+        FAILED,
+        SUCCEED
+    };
+    void setBuildTableState(BuildTableState state_);
 
     /// Reference to the row in block.
     struct RowRef
@@ -300,7 +306,7 @@ class Join
 
     mutable std::mutex build_table_mutex;
     mutable std::condition_variable build_table_cv;
-    int build_table_state; /// -1 means failure, 0 means waiting , 1 means success
+    BuildTableState build_table_state;
 
     const LogWithPrefixPtr log;