From d9898c9a71c2d2bad99b622050846104586fc6fc Mon Sep 17 00:00:00 2001 From: Sachin Date: Sun, 21 Feb 2021 09:30:20 +0000 Subject: [PATCH] MDEV-7409 On RBR, extend the PROCESSLIST info to include at least the name of the recently used table 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. --- mysql-test/suite/rpl/r/rpl_rbr_monitor.result | 43 +++++++++++ mysql-test/suite/rpl/t/rpl_rbr_monitor.test | 76 +++++++++++++++++++ sql/log_event_server.cc | 74 +++++++++++++----- 3 files changed, 174 insertions(+), 19 deletions(-) create mode 100644 mysql-test/suite/rpl/r/rpl_rbr_monitor.result create mode 100644 mysql-test/suite/rpl/t/rpl_rbr_monitor.test 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; }