Fix mpp hang error if some error happens during compile of mpp plan in TiFlash

dbms/src/Common/FailPoint.cpp

@@ -8,26 +8,35 @@ namespace DB
 {
 std::unordered_map<String, std::shared_ptr<FailPointChannel>> FailPointHelper::fail_point_wait_channels;
 
-#define APPLY_FOR_FAILPOINTS(M)                             \
-    M(exception_between_drop_meta_and_data)                 \
-    M(exception_between_alter_data_and_meta)                \
-    M(exception_drop_table_during_remove_meta)              \
-    M(exception_between_rename_table_data_and_metadata)     \
-    M(exception_between_create_database_meta_and_directory) \
-    M(exception_before_rename_table_old_meta_removed)       \
-    M(exception_after_step_1_in_exchange_partition)         \
-    M(exception_before_step_2_rename_in_exchange_partition) \
-    M(exception_after_step_2_in_exchange_partition)         \
-    M(exception_before_step_3_rename_in_exchange_partition) \
-    M(exception_after_step_3_in_exchange_partition)         \
-    M(region_exception_after_read_from_storage_some_error)  \
-    M(region_exception_after_read_from_storage_all_error)   \
-    M(exception_before_dmfile_remove_encryption)            \
-    M(exception_before_dmfile_remove_from_disk)             \
-    M(force_enable_region_persister_compatible_mode)        \
-    M(force_disable_region_persister_compatible_mode)       \
-    M(force_triggle_background_merge_delta)                 \
-    M(force_triggle_foreground_flush)
+#define APPLY_FOR_FAILPOINTS(M)                                   \
+    M(exception_between_drop_meta_and_data)                       \
+    M(exception_between_alter_data_and_meta)                      \
+    M(exception_drop_table_during_remove_meta)                    \
+    M(exception_between_rename_table_data_and_metadata)           \
+    M(exception_between_create_database_meta_and_directory)       \
+    M(exception_before_rename_table_old_meta_removed)             \
+    M(exception_after_step_1_in_exchange_partition)               \
+    M(exception_before_step_2_rename_in_exchange_partition)       \
+    M(exception_after_step_2_in_exchange_partition)               \
+    M(exception_before_step_3_rename_in_exchange_partition)       \
+    M(exception_after_step_3_in_exchange_partition)               \
+    M(region_exception_after_read_from_storage_some_error)        \
+    M(region_exception_after_read_from_storage_all_error)         \
+    M(exception_before_dmfile_remove_encryption)                  \
+    M(exception_before_dmfile_remove_from_disk)                   \
+    M(force_enable_region_persister_compatible_mode)              \
+    M(force_disable_region_persister_compatible_mode)             \
+    M(force_triggle_background_merge_delta)                       \
+    M(force_triggle_foreground_flush)                             \
+    M(exception_before_mpp_register_non_root_mpp_task)            \
+    M(exception_before_mpp_register_tunnel_for_non_root_mpp_task) \
+    M(exception_during_mpp_register_tunnel_for_non_root_mpp_task) \
+    M(exception_before_mpp_non_root_task_run)                     \
+    M(exception_during_mpp_non_root_task_run)                     \
+    M(exception_before_mpp_register_root_mpp_task)                \
+    M(exception_before_mpp_register_tunnel_for_root_mpp_task)     \
+    M(exception_before_mpp_root_task_run)                         \
+    M(exception_during_mpp_root_task_run)
 
 #define APPLY_FOR_FAILPOINTS_WITH_CHANNEL(M)  \
     M(pause_after_learner_read)               \

dbms/src/Flash/Mpp/MPPHandler.cpp

@@ -16,6 +16,15 @@ namespace DB
 namespace FailPoints
 {
 extern const char hang_in_execution[];
+extern const char exception_before_mpp_register_non_root_mpp_task[];
+extern const char exception_before_mpp_register_root_mpp_task[];
+extern const char exception_before_mpp_register_tunnel_for_non_root_mpp_task[];
+extern const char exception_before_mpp_register_tunnel_for_root_mpp_task[];
+extern const char exception_during_mpp_register_tunnel_for_non_root_mpp_task[];
+extern const char exception_before_mpp_non_root_task_run[];
+extern const char exception_before_mpp_root_task_run[];
+extern const char exception_during_mpp_non_root_task_run[];
+extern const char exception_during_mpp_root_task_run[];
 } // namespace FailPoints
 
 bool MPPTaskProgress::isTaskHanging(const Context & context)
@@ -108,21 +117,37 @@ void MPPTask::prepare(const mpp::DispatchTaskRequest & task_request)
     context.getTimezoneInfo().resetByDAGRequest(*dag_req);
     context.setProgressCallback([this](const Progress & progress) { this->updateProgress(progress); });
 
+    dag_context = std::make_unique<DAGContext>(*dag_req, task_request.meta());
+    context.setDAGContext(dag_context.get());
+
     // register task.
     TMTContext & tmt_context = context.getTMTContext();
     auto task_manager = tmt_context.getMPPTaskManager();
     LOG_DEBUG(log, "begin to register the task " << id.toString());
+
+    if (dag_context->isRootMPPTask())
+    {
+        FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_before_mpp_register_root_mpp_task);
+    }
+    else
+    {
+        FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_before_mpp_register_non_root_mpp_task);
+    }
     if (!task_manager->registerTask(shared_from_this()))
     {
         throw TiFlashException(std::string(__PRETTY_FUNCTION__) + ": Failed to register MPP Task", Errors::Coprocessor::BadRequest);
     }
 
-
-    dag_context = std::make_unique<DAGContext>(*dag_req, task_request.meta());
-    context.setDAGContext(dag_context.get());
-
     DAGQuerySource dag(context, regions, *dag_req, true);
 
+    if (dag_context->isRootMPPTask())
+    {
+        FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_before_mpp_register_tunnel_for_root_mpp_task);
+    }
+    else
+    {
+        FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_before_mpp_register_tunnel_for_non_root_mpp_task);
+    }
     // register tunnels
     MPPTunnelSetPtr tunnel_set = std::make_shared<MPPTunnelSet>();
     const auto & exchangeSender = dag_req->root_executor().exchange_sender();
@@ -136,6 +161,10 @@ void MPPTask::prepare(const mpp::DispatchTaskRequest & task_request)
         LOG_DEBUG(log, "begin to register the tunnel " << tunnel->tunnel_id);
         registerTunnel(MPPTaskId{task_meta.start_ts(), task_meta.task_id()}, tunnel);
         tunnel_set->tunnels.emplace_back(tunnel);
+        if (!dag_context->isRootMPPTask())
+        {
+            FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_during_mpp_register_tunnel_for_non_root_mpp_task);
+        }
     }
     // read index , this may take a long time.
     io = executeQuery(dag, context, false, QueryProcessingStage::Complete);
@@ -202,6 +231,14 @@ void MPPTask::runImpl()
             count += block.rows();
             to->write(block);
             FAIL_POINT_PAUSE(FailPoints::hang_in_execution);
+            if (dag_context->isRootMPPTask())
+            {
+                FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_during_mpp_root_task_run);
+            }
+            else
+            {
+                FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_during_mpp_non_root_task_run);
+            }
         }
 
         /// For outputting additional information in some formats.
@@ -278,14 +315,45 @@ void MPPTask::cancel()
     LOG_WARNING(log, "Finish cancel task: " + id.toString());
 }
 
+void MPPHandler::handleError(MPPTaskPtr task, String error)
+{
+    try
+    {
+        if (task != nullptr)
+        {
+            /// for root task, the tunnel is only connected after DispatchMPPTask
+            /// finishes without error, for non-root task, tunnel can be connected
+            /// even if the DispatchMPPTask fails, so for non-root task, we write
+            /// error to all tunnels, while for root task, we just close the tunnel.
+            if (!task->dag_context->isRootMPPTask())
+                task->writeErrToAllTunnel(error);
+            else
+                task->closeAllTunnel();
+            task->unregisterTask();
+        }
+    }
+    catch (...)
+    {
+        tryLogCurrentException(log, "Fail to handle error and clean task");
+    }
+}
 // execute is responsible for making plan , register tasks and tunnels and start the running thread.
 grpc::Status MPPHandler::execute(Context & context, mpp::DispatchTaskResponse * response)
 {
+    MPPTaskPtr task = nullptr;
     try
     {
         Stopwatch stopwatch;
-        MPPTaskPtr task = std::make_shared<MPPTask>(task_request.meta(), context);
+        task = std::make_shared<MPPTask>(task_request.meta(), context);
         task->prepare(task_request);
+        if (task->dag_context->isRootMPPTask())
+        {
+            FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_before_mpp_root_task_run);
+        }
+        else
+        {
+            FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_before_mpp_non_root_task_run);
+        }
         task->memory_tracker = current_memory_tracker;
         task->run();
         LOG_INFO(log, "processing dispatch is over; the time cost is " << std::to_string(stopwatch.elapsedMilliseconds()) << " ms");
@@ -295,18 +363,21 @@ grpc::Status MPPHandler::execute(Context & context, mpp::DispatchTaskResponse *
         LOG_ERROR(log, "dispatch task meet error : " << e.displayText());
         auto * err = response->mutable_error();
         err->set_msg(e.displayText());
+        handleError(task, e.displayText());
     }
     catch (std::exception & e)
     {
         LOG_ERROR(log, "dispatch task meet error : " << e.what());
         auto * err = response->mutable_error();
         err->set_msg(e.what());
+        handleError(task, e.what());
     }
     catch (...)
     {
         LOG_ERROR(log, "dispatch task meet fatal error");
         auto * err = response->mutable_error();
         err->set_msg("fatal error");
+        handleError(task, "fatal error");
     }
     return grpc::Status::OK;
 }

dbms/src/Flash/Mpp/MPPHandler.h

@@ -67,7 +67,7 @@ struct MPPTunnel
         }
         catch (...)
         {
-            LOG_WARNING(log, "Error in destructor function of MPPTunnel");
+            tryLogCurrentException(log, "Error in destructor function of MPPTunnel");
         }
     }
 
@@ -126,6 +126,18 @@ struct MPPTunnel
         cv_for_finished.notify_all();
     }
 
+    /// close() finishes the tunnel without checking the connect status, this function
+    /// should only be used when handling error if DispatchMPPTask fails for
+    /// root task. Because for root task, if DispatchMPPTask fails, TiDB does
+    /// not sending establish MPP connection request at all, it is meaningless
+    /// to check the connect status in this case, just finish the tunnel.
+    void close()
+    {
+        std::unique_lock<std::mutex> lk(mu);
+        finished = true;
+        cv_for_finished.notify_all();
+    }
+
     // a MPPConn request has arrived. it will build connection by this tunnel;
     void connect(::grpc::ServerWriter<::mpp::MPPDataPacket> * writer_)
     {
@@ -256,7 +268,7 @@ struct MPPTask : std::enable_shared_from_this<MPPTask>, private boost::noncopyab
     // which targeted task we should send data by which tunnel.
     std::map<MPPTaskId, MPPTunnelPtr> tunnel_map;
 
-    MPPTaskManager * manager;
+    MPPTaskManager * manager = nullptr;
 
     Logger * log;
 
@@ -279,6 +291,20 @@ struct MPPTask : std::enable_shared_from_this<MPPTask>, private boost::noncopyab
 
     void cancel();
 
+    void closeAllTunnel()
+    {
+        try
+        {
+            for (auto & it : tunnel_map)
+            {
+                it.second->close();
+            }
+        }
+        catch (...)
+        {
+            tryLogCurrentException(log, "Failed to close all tunnels");
+        }
+    }
     void writeErrToAllTunnel(const String & e)
     {
         try
@@ -294,7 +320,7 @@ struct MPPTask : std::enable_shared_from_this<MPPTask>, private boost::noncopyab
         }
         catch (...)
         {
-            LOG_WARNING(log, "Failed to write error " + e + " to all tunnel");
+            tryLogCurrentException(log, "Failed to write error " + e + " to all tunnels");
         }
     }
 
@@ -516,6 +542,7 @@ class MPPHandler
 public:
     MPPHandler(const mpp::DispatchTaskRequest & task_request_) : task_request(task_request_), log(&Logger::get("MPPHandler")) {}
     grpc::Status execute(Context & context, mpp::DispatchTaskResponse * response);
+    void handleError(MPPTaskPtr task, String error);
 };
 
 } // namespace DB

tests/fullstack-test/mpp/mpp_fail.test

@@ -0,0 +1,69 @@
+# Preparation.
+=> DBGInvoke __init_fail_point()
+
+mysql> drop table if exists test.t
+mysql> create table test.t (id int, value varchar(64))
+mysql> insert into test.t values(1,'a'),(2,'b'),(3,'c')
+mysql> alter table test.t set tiflash replica 1
+
+func> wait_table test t
+
+
+# Data.
+
+## exception before mpp register non root mpp task
+=> DBGInvoke __enable_fail_point(exception_before_mpp_register_non_root_mpp_task)
+mysql> use test; set @@tidb_isolation_read_engines='tiflash'; set @@tidb_allow_mpp=1; select count(value), id from t group by id;
+ERROR 1105 (HY000) at line 1: DB::Exception: Fail point FailPoints::exception_before_mpp_register_non_root_mpp_task is triggered.
+=> DBGInvoke __disable_fail_point(exception_before_mpp_register_non_root_mpp_task)
+
+## exception before mpp register root mpp task
+=> DBGInvoke __enable_fail_point(exception_before_mpp_register_root_mpp_task)
+mysql> use test; set @@tidb_isolation_read_engines='tiflash'; set @@tidb_allow_mpp=1; select count(value), id from t group by id;
+ERROR 1105 (HY000) at line 1: DB::Exception: Fail point FailPoints::exception_before_mpp_register_root_mpp_task is triggered.
+=> DBGInvoke __disable_fail_point(exception_before_mpp_register_root_mpp_task)
+
+## exception before mpp register tunnel for non root mpp task
+=> DBGInvoke __enable_fail_point(exception_before_mpp_register_tunnel_for_non_root_mpp_task)
+mysql> use test; set @@tidb_isolation_read_engines='tiflash'; set @@tidb_allow_mpp=1; select count(value), id from t group by id;
+ERROR 1105 (HY000) at line 1: DB::Exception: Fail point FailPoints::exception_before_mpp_register_tunnel_for_non_root_mpp_task is triggered.
+=> DBGInvoke __disable_fail_point(exception_before_mpp_register_tunnel_for_non_root_mpp_task)
+
+## exception before mpp register tunnel for root mpp task
+=> DBGInvoke __enable_fail_point(exception_before_mpp_register_tunnel_for_root_mpp_task)
+mysql> use test; set @@tidb_isolation_read_engines='tiflash'; set @@tidb_allow_mpp=1; select count(value), id from t group by id;
+ERROR 1105 (HY000) at line 1: DB::Exception: Fail point FailPoints::exception_before_mpp_register_tunnel_for_root_mpp_task is triggered.
+=> DBGInvoke __disable_fail_point(exception_before_mpp_register_tunnel_for_root_mpp_task)
+
+## exception during mpp register tunnel for non root mpp task
+=> DBGInvoke __enable_fail_point(exception_during_mpp_register_tunnel_for_non_root_mpp_task)
+mysql> use test; set @@tidb_isolation_read_engines='tiflash'; set @@tidb_allow_mpp=1; select count(value), id from t group by id;
+ERROR 1105 (HY000) at line 1: DB::Exception: Fail point FailPoints::exception_during_mpp_register_tunnel_for_non_root_mpp_task is triggered.
+=> DBGInvoke __disable_fail_point(exception_during_mpp_register_tunnel_for_non_root_mpp_task)
+
+## exception before mpp run non root task
+=> DBGInvoke __enable_fail_point(exception_before_mpp_non_root_task_run)
+mysql> use test; set @@tidb_isolation_read_engines='tiflash'; set @@tidb_allow_mpp=1; select count(value), id from t group by id;
+ERROR 1105 (HY000) at line 1: DB::Exception: Fail point FailPoints::exception_before_mpp_non_root_task_run is triggered.
+=> DBGInvoke __disable_fail_point(exception_before_mpp_non_root_task_run)
+
+## exception before mpp run root task
+=> DBGInvoke __enable_fail_point(exception_before_mpp_root_task_run)
+mysql> use test; set @@tidb_isolation_read_engines='tiflash'; set @@tidb_allow_mpp=1; select count(value), id from t group by id;
+ERROR 1105 (HY000) at line 1: DB::Exception: Fail point FailPoints::exception_before_mpp_root_task_run is triggered.
+=> DBGInvoke __disable_fail_point(exception_before_mpp_root_task_run)
+
+## exception during mpp run non root task
+=> DBGInvoke __enable_fail_point(exception_during_mpp_non_root_task_run)
+mysql> use test; set @@tidb_isolation_read_engines='tiflash'; set @@tidb_allow_mpp=1; select count(value), id from t group by id;
+ERROR 1105 (HY000) at line 1: other error for mpp stream: DB::Exception: exchange receiver meet error : DB::Exception: Fail point FailPoints::exception_during_mpp_non_root_task_run is triggered.
+=> DBGInvoke __disable_fail_point(exception_during_mpp_non_root_task_run)
+
+## exception during mpp run root task
+=> DBGInvoke __enable_fail_point(exception_during_mpp_root_task_run)
+mysql> use test; set @@tidb_isolation_read_engines='tiflash'; set @@tidb_allow_mpp=1; select count(value), id from t group by id;
+ERROR 1105 (HY000) at line 1: other error for mpp stream: DB::Exception: Fail point FailPoints::exception_during_mpp_root_task_run is triggered.
+=> DBGInvoke __disable_fail_point(exception_during_mpp_root_task_run)
+
+# Clean up.
+mysql> drop table if exists test.t