MDEV-7409 On RBR, extend the PROCESSLIST info to include at least the name of the recently used tablebb-10.6-sachin

When RBR is used, add the db name to db Field and table name to Status
Field  of the "SHOW FULL PROCESSLIST" command for SQL thread.

3 files changed, 174 insertions, 19 deletions

diff --git a/mysql-test/suite/rpl/r/rpl_rbr_monitor.result b/mysql-test/suite/rpl/r/rpl_rbr_monitor.result
new file mode 100644
index 00000000000..e8ac6277233
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_rbr_monitor.result
@@ -0,0 +1,43 @@
+include/master-slave.inc
+[connection master]
+connection master;
+create table t1(a int primary key) engine=innodb;
+connection slave;
+connection slave1;
+begin;
+insert into t1(a) values(1);
+connection master;
+select * from t1;
+a
+insert into t1(a) values(1);
+#monitoring write rows
+connection slave;
+#monitoring update rows
+connection slave1;
+rollback;
+begin;
+select a from t1 for update;
+a
+1
+connection master;
+update t1 set a = a + 1 ;
+connection slave;
+#monitoring delete rows
+connection slave1;
+rollback;
+begin;
+select * from t1 for update;
+a
+2
+connection master;
+delete from t1;
+connection slave;
+select * from t1;
+a
+2
+connection slave1;
+rollback;
+connection master;
+drop table t1;
+connection slave;
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_rbr_monitor.test b/mysql-test/suite/rpl/t/rpl_rbr_monitor.test
new file mode 100644
index 00000000000..2bc1f9cd482
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_rbr_monitor.test
@@ -0,0 +1,76 @@
+#
+# Mdev-7409 On RBR, extend the PROCESSLIST info to include at least the name of
+#  the recently used table
+# This testcase create Write_rows_log_event , Update_rows_log_event and
+#  Delete_rows_log_event which is blocked on slave and we will check whether
+#  whether processinfo includes table name or not.
+--source include/have_innodb.inc
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+--enable_connect_log
+
+--connection master
+create table t1(a int primary key) engine=innodb;
+
+--sync_slave_with_master
+--connection slave1
+begin;
+insert into t1(a) values(1);
+--connection master
+select * from t1;
+
+insert into t1(a) values(1);
+--save_master_pos
+
+--echo #monitoring write rows
+--connection slave
+
+
+let $wait_condition= SELECT COUNT(*) = 1 FROM  INFORMATION_SCHEMA.PROCESSLIST
+      WHERE DB = 'test' AND STATE LIKE  "Write_rows_log_event::write_row(%) on table %";
+--source include/wait_condition.inc
+
+
+--echo #monitoring update rows
+--connection slave1
+rollback;
+--sync_with_master
+begin;
+select a from t1 for update;
+
+--connection master
+update t1 set a = a + 1 ;
+--save_master_pos
+
+--connection slave
+let $wait_condition= SELECT COUNT(*) = 1 FROM  INFORMATION_SCHEMA.PROCESSLIST
+  WHERE DB = 'test' AND STATE LIKE  "Update_rows_log_event::find_row(%) on table %";
+--source include/wait_condition.inc
+
+--echo #monitoring delete rows
+--connection slave1
+rollback;
+--sync_with_master
+begin;
+select * from t1 for update;
+
+--connection master
+delete from t1;
+--save_master_pos
+
+--connection slave
+select * from t1;
+let $wait_condition= SELECT COUNT(*) = 1 FROM  INFORMATION_SCHEMA.PROCESSLIST
+  WHERE DB = 'test' AND STATE LIKE  "Delete_rows_log_event::find_row(%) on table %";
+--source include/wait_condition.inc
+
+#CleanUp
+--connection slave1
+rollback;
+--sync_with_master
+
+--connection master
+drop table t1;
+--sync_slave_with_master
+
+--source include/rpl_end.inc
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index 55ad50511ec..91e789cd639 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -7492,13 +7492,21 @@ Write_rows_log_event::do_exec_row(rpl_group_info *rgi)
 {
   DBUG_ASSERT(m_table != NULL);
   const char *tmp= thd->get_proc_info();
-  const char *message= "Write_rows_log_event::write_row()";
+  LEX_CSTRING tmp_db= thd->db;
+  char *message, msg[128];
+  const char *table_name= m_table->s->table_name.str;
+  char quote_char= get_quote_char_for_identifier(thd, STRING_WITH_LEN(table_name));
+  my_snprintf(msg, sizeof(msg),"Write_rows_log_event::write_row() on table %c%s%c",
+                   quote_char, table_name, quote_char);
+  thd->reset_db(&m_table->s->db);
+  message= msg;
   int error;
 
 #ifdef WSREP_PROC_INFO
   my_snprintf(thd->wsrep_info, sizeof(thd->wsrep_info) - 1,
-              "Write_rows_log_event::write_row(%lld)",
-              (long long) wsrep_thd_trx_seqno(thd));
+              "Write_rows_log_event::write_row(%lld) on table %c%s%c",
+              (long long) wsrep_thd_trx_seqno(thd), quote_char, table_name,
+              quote_char);
   message= thd->wsrep_info;
 #endif /* WSREP_PROC_INFO */
 
@@ -7512,6 +7520,7 @@ Write_rows_log_event::do_exec_row(rpl_group_info *rgi)
     my_error(ER_UNKNOWN_ERROR, MYF(0));
   }
 
+  thd->reset_db(&tmp_db);
   return error;
 }
 
@@ -8108,14 +8117,22 @@ int Delete_rows_log_event::do_exec_row(rpl_group_info *rgi)
 {
   int error;
   const char *tmp= thd->get_proc_info();
-  const char *message= "Delete_rows_log_event::find_row()";
+  LEX_CSTRING tmp_db= thd->db;
+  char *message, msg[128];
+  const char *table_name= m_table->s->table_name.str;
+  char quote_char= get_quote_char_for_identifier(thd, STRING_WITH_LEN(table_name));
+  my_snprintf(msg, sizeof(msg),"Delete_rows_log_event::find_row() on table %c%s%c",
+                   quote_char, table_name, quote_char);
+  thd->reset_db(&m_table->s->db);
+  message= msg;
   const bool invoke_triggers= (m_table->triggers && do_invoke_trigger());
   DBUG_ASSERT(m_table != NULL);
 
 #ifdef WSREP_PROC_INFO
   my_snprintf(thd->wsrep_info, sizeof(thd->wsrep_info) - 1,
-              "Delete_rows_log_event::find_row(%lld)",
-              (long long) wsrep_thd_trx_seqno(thd));
+              "Delete_rows_log_event::find_row(%lld) on table %c%s%c",
+              (long long) wsrep_thd_trx_seqno(thd), quote_char, table_name,
+              quote_char);
   message= thd->wsrep_info;
 #endif /* WSREP_PROC_INFO */
 
@@ -8125,11 +8142,14 @@ int Delete_rows_log_event::do_exec_row(rpl_group_info *rgi)
     /*
       Delete the record found, located in record[0]
     */
-    message= "Delete_rows_log_event::ha_delete_row()";
+    my_snprintf(msg, sizeof(msg),"Delete_rows_log_event::ha_delete_row() on table %c%s%c",
+                   quote_char, table_name, quote_char);
+    message= msg;
 #ifdef WSREP_PROC_INFO
     snprintf(thd->wsrep_info, sizeof(thd->wsrep_info) - 1,
-             "Delete_rows_log_event::ha_delete_row(%lld)",
-             (long long) wsrep_thd_trx_seqno(thd));
+             "Delete_rows_log_event::ha_delete_row(%lld) on table %c%s%c",
+              (long long) wsrep_thd_trx_seqno(thd), quote_char, table_name,
+              quote_char);
     message= thd->wsrep_info;
 #endif
     thd_proc_info(thd, message);
@@ -8160,6 +8180,7 @@ int Delete_rows_log_event::do_exec_row(rpl_group_info *rgi)
       error= HA_ERR_GENERIC; // in case if error is not set yet
     m_table->file->ha_index_or_rnd_end();
   }
+  thd->reset_db(&tmp_db);
   thd_proc_info(thd, tmp);
   return error;
 }
@@ -8259,13 +8280,21 @@ Update_rows_log_event::do_exec_row(rpl_group_info *rgi)
 {
   const bool invoke_triggers= (m_table->triggers && do_invoke_trigger());
   const char *tmp= thd->get_proc_info();
-  const char *message= "Update_rows_log_event::find_row()";
   DBUG_ASSERT(m_table != NULL);
+  LEX_CSTRING tmp_db= thd->db;
+  char *message, msg[128];
+  const char *table_name= m_table->s->table_name.str;
+  char quote_char= get_quote_char_for_identifier(thd, STRING_WITH_LEN(table_name));
+  my_snprintf(msg, sizeof(msg),"Update_rows_log_event::find_row() on table %c%s%c",
+                   quote_char, table_name, quote_char);
+  thd->reset_db(&m_table->s->db);
+  message= msg;
 
 #ifdef WSREP_PROC_INFO
   my_snprintf(thd->wsrep_info, sizeof(thd->wsrep_info) - 1,
-              "Update_rows_log_event::find_row(%lld)",
-              (long long) wsrep_thd_trx_seqno(thd));
+              "Update_rows_log_event::find_row(%lld) on table %c%s%c",
+              (long long) wsrep_thd_trx_seqno(thd), quote_char, table_name,
+              quote_char);
   message= thd->wsrep_info;
 #endif /* WSREP_PROC_INFO */
 
@@ -8286,6 +8315,7 @@ Update_rows_log_event::do_exec_row(rpl_group_info *rgi)
     if ((m_curr_row= m_curr_row_end))
       unpack_current_row(rgi, &m_cols_ai);
     thd_proc_info(thd, tmp);
+    thd->reset_db(&tmp_db);
     return error;
   }
 
@@ -8303,11 +8333,14 @@ Update_rows_log_event::do_exec_row(rpl_group_info *rgi)
   store_record(m_table,record[1]);
 
   m_curr_row= m_curr_row_end;
-  message= "Update_rows_log_event::unpack_current_row()";
+  my_snprintf(msg, sizeof(msg),"Update_rows_log_event::unpack_current_row() on table %c%s%c",
+                   quote_char, table_name, quote_char);
+  message= msg;
 #ifdef WSREP_PROC_INFO
   my_snprintf(thd->wsrep_info, sizeof(thd->wsrep_info) - 1,
-              "Update_rows_log_event::unpack_current_row(%lld)",
-              (long long) wsrep_thd_trx_seqno(thd));
+              "Update_rows_log_event::unpack_current_row(%lld) on table %c%s%c",
+              (long long) wsrep_thd_trx_seqno(thd), quote_char, table_name,
+              quote_char);
   message= thd->wsrep_info;
 #endif /* WSREP_PROC_INFO */
 
@@ -8330,11 +8363,13 @@ Update_rows_log_event::do_exec_row(rpl_group_info *rgi)
   DBUG_DUMP("new values", m_table->record[0], m_table->s->reclength);
 #endif
 
-  message= "Update_rows_log_event::ha_update_row()";
+  my_snprintf(msg, sizeof(msg),"Update_rows_log_event::ha_update_row() on table %c%s%c",
+              quote_char, table_name, quote_char);
+  message= msg;
 #ifdef WSREP_PROC_INFO
   my_snprintf(thd->wsrep_info, sizeof(thd->wsrep_info) - 1,
-              "Update_rows_log_event::ha_update_row(%lld)",
-              (long long) wsrep_thd_trx_seqno(thd));
+              "Update_rows_log_event::ha_update_row(%lld) on table %c%s%c",
+              (long long) wsrep_thd_trx_seqno(thd), quote_char, table_name, quote_char);
   message= thd->wsrep_info;
 #endif /* WSREP_PROC_INFO */
 
@@ -8363,9 +8398,10 @@ Update_rows_log_event::do_exec_row(rpl_group_info *rgi)
       unlikely(process_triggers(TRG_EVENT_UPDATE, TRG_ACTION_AFTER, TRUE)))
     error= HA_ERR_GENERIC; // in case if error is not set yet
 
-  thd_proc_info(thd, tmp);
 
 err:
+  thd_proc_info(thd, tmp);
+  thd->reset_db(&tmp_db);
   m_table->file->ha_index_or_rnd_end();
   return error;
 }