MPP: update the state of building a hash table when createOnce throw exceptions (#4202)

dbms/src/Common/FailPoint.cpp

@@ -47,7 +47,8 @@ std::unordered_map<String, std::shared_ptr<FailPointChannel>> FailPointHelper::f
     M(segment_merge_after_ingest_packs)                           \
     M(force_formal_page_file_not_exists)                          \
     M(force_legacy_or_checkpoint_page_file_exists)                \
-    M(exception_in_creating_set_input_stream)
+    M(exception_in_creating_set_input_stream)                     \
+    M(exception_mpp_hash_build)
 
 #define APPLY_FOR_FAILPOINTS(M)                              \
     M(force_set_page_file_write_errno)                       \

dbms/src/DataStreams/CreatingSetsBlockInputStream.cpp

@@ -16,7 +16,8 @@ namespace DB
 namespace FailPoints
 {
 extern const char exception_in_creating_set_input_stream[];
-}
+extern const char exception_mpp_hash_build[];
+} // namespace FailPoints
 namespace ErrorCodes
 {
 extern const int SET_SIZE_LIMIT_EXCEEDED;
@@ -94,9 +95,10 @@ void CreatingSetsBlockInputStream::createAll()
             for (auto &elem : subqueries_for_sets)
             {
                 if (elem.second.join)
-                    elem.second.join->setFinishBuildTable(false);
+                    elem.second.join->setBuildTableState(Join::BuildTableState::WAITING);
             }
         }
+        Stopwatch watch;
         for (auto & subqueries_for_sets : subqueries_for_sets_list)
         {
             for (auto & elem : subqueries_for_sets)
@@ -116,27 +118,33 @@ void CreatingSetsBlockInputStream::createAll()
         }
 
         if (!exception_from_workers.empty())
+        {
+            LOG_ERROR(log,
+                "Creating all tasks of " << std::to_string(mpp_task_id) << " takes " << std::to_string(watch.elapsedSeconds())
+                                         << " sec with exception and rethrow the first of total " << exception_from_workers.size()
+                                         << " exceptions.");
             std::rethrow_exception(exception_from_workers.front());
-
+        }
+        LOG_DEBUG(
+            log, "Creating all tasks of " << std::to_string(mpp_task_id) << " takes " << std::to_string(watch.elapsedSeconds()) << "sec. ");
         created = true;
     }
 }
 
 void CreatingSetsBlockInputStream::createOne(SubqueryForSet & subquery)
 {
+    Stopwatch watch;
     try
     {
         LOG_DEBUG(log,
             (subquery.set ? "Creating set. " : "")
                 << (subquery.join ? "Creating join. " : "") << (subquery.table ? "Filling temporary table. " : "") << " for task "
                 << std::to_string(mpp_task_id));
-        Stopwatch watch;
 
         BlockOutputStreamPtr table_out;
         if (subquery.table)
             table_out = subquery.table->write({}, {});
 
-
         bool done_with_set = !subquery.set;
         bool done_with_join = !subquery.join;
         bool done_with_table = !subquery.table;
@@ -193,7 +201,10 @@ void CreatingSetsBlockInputStream::createOne(SubqueryForSet & subquery)
 
 
         if (subquery.join)
-            subquery.join->setFinishBuildTable(true);
+        {
+            FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_mpp_hash_build);
+            subquery.join->setBuildTableState(Join::BuildTableState::SUCCEED);
+        }
 
         if (table_out)
             table_out->writeSuffix();
@@ -231,13 +242,18 @@ void CreatingSetsBlockInputStream::createOne(SubqueryForSet & subquery)
         }
         else
         {
-            LOG_DEBUG(log, "Subquery has empty result for task " << std::to_string(mpp_task_id) << ".");
+            LOG_DEBUG(log, "Subquery has empty result for task" << std::to_string(mpp_task_id));
         }
     }
-    catch (std::exception & e)
+    catch (...)
     {
         std::unique_lock<std::mutex> lock(exception_mutex);
         exception_from_workers.push_back(std::current_exception());
+        if (subquery.join)
+            subquery.join->setBuildTableState(Join::BuildTableState::FAILED);
+        LOG_ERROR(log,
+            "task" << std::to_string(mpp_task_id) << " throw exception: " << getCurrentExceptionMessage(false, true) << " In "
+                   << std::to_string(watch.elapsedSeconds()) << " sec. ");
     }
 }