MDEV-20220: Merge 5.7 P_S replication table 'replication_applier_status_by_worker

Step1:
=====
Backport 'replication_applier_status_by_worker' from upstream.

Iterate through rpl_parallel_thread_pool and display slave worker thread
specific information as part of 'replication_applier_status_by_worker'
table.

---------------------------------------------------------------------------
|Column Name:           |        Description:                             |
|-------------------------------------------------------------------------|
|                       |                                                 |
|CHANNEL_NAME           | Name of replication channel through which the   |
|                       | transaction is received.                        |
|                       |                                                 |
|THREAD_ID              | Thread_Id as displayed in 'performance_schema.  |
|                       | threads' table for thread with name             |
|                       | 'thread/sql/rpl_parallel_thread'                |
|                       |                                                 |
|                       | THREAD_ID will be NULL when worker threads are  |
|                       | stopped due to an error/force stop              |
|                       |                                                 |
|SERVICE_STATE          | Thread is running or not                        |
|                       |                                                 |
|LAST_SEEN_TRANSACTION  | Last GTID executed by worker                    |
|                       |                                                 |
|LAST_ERROR_NUMBER      | Last Error that occured on a particular worker  |
|                       |                                                 |
|LAST_ERROR_MESSAGE     | Last error specific message                     |
|                       |                                                 |
|LAST_ERROR_TIMESTAMP   | Time stamp of last error                        |
|                       |                                                 |
---------------------------------------------------------------------------

CHANNEL_NAME will be empty when the worker has not processed any
transaction. Channel_name points to valid source channel_name when it is
processing a transaction/event group.

1 files changed, 17 insertions, 0 deletions

diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
index 65d5a06a76a..d1ca1c1bf6f 100644
--- a/sql/rpl_parallel.cc
+++ b/sql/rpl_parallel.cc
@@ -1099,6 +1099,11 @@ handle_rpl_parallel_thread(void *arg)
 
   mysql_mutex_lock(&rpt->LOCK_rpl_thread);
   rpt->thd= thd;
+  PSI_thread *psi= PSI_CALL_get_thread();
+  PSI_CALL_set_thread_os_id(psi);
+  PSI_CALL_set_thread_THD(psi, thd);
+  PSI_CALL_set_thread_id(psi, thd->thread_id);
+  rpt->thd->set_psi(psi);
 
   while (rpt->delay_start)
     mysql_cond_wait(&rpt->COND_rpl_thread, &rpt->LOCK_rpl_thread);
@@ -1188,6 +1193,12 @@ handle_rpl_parallel_thread(void *arg)
       /* Handle a new event group, which will be initiated by a GTID event. */
       if ((event_type= qev->ev->get_type_code()) == GTID_EVENT)
       {
+        rpt->last_seen_gtid= rgi->current_gtid;
+        rpt->channel_name_length= (uint)rgi->rli->mi->connection_name.length;
+        if (rpt->channel_name_length)
+          memcpy(rpt->channel_name, rgi->rli->mi->connection_name.str,
+                 rgi->rli->mi->connection_name.length);
+
         bool did_enter_cond= false;
         PSI_stage_info old_stage;
 
@@ -2003,6 +2014,12 @@ rpl_parallel_thread::loc_free_gco(group_commit_orderer *gco)
 }
 
 
+rpl_parallel_thread::rpl_parallel_thread()
+  : channel_name_length(0), last_error_number(0), last_error_timestamp(0)
+{
+}
+
+
 rpl_parallel_thread_pool::rpl_parallel_thread_pool()
   : threads(0), free_list(0), count(0), inited(false), busy(false)
 {