MDEV-31804 Assertion `thd->m_transaction_psi == __null' fails

... upon replicating online ALTER

When an online event is applied and slave_exec_mode is idempotent,
Write_rows_log_event::do_before_row_operations had reset
thd->lex->sql_command to SQLCOM_REPLACE.

This led to that a statement was detected as a row-type during binlogging,
and was logged as not standalone.

So the corresponding Gtid_log_event, when applied on replica, did not exit
early and created a new PSI transaction. Hence the difference with
non-online ALTER.

mysql-test/suite/rpl/r/rpl_alter_online_debug.result

@@ -4,6 +4,29 @@ connection master;
 call mtr.add_suppression("Unsafe statement written to the binary log using statement format");
 connection slave;
 call mtr.add_suppression("Unsafe statement written to the binary log using statement format");
+# MDEV-31804 Assertion `thd->m_transaction_psi == __null' fails upon
+# replicating online ALTER
+connection master;
+create table t (a char(8)) engine=myisam;
+insert into t values ('foo'),('bar');
+set debug_sync= 'alter_table_online_progress signal go_dml wait_for go_alter';
+set @old_slave_exec_mode= @@global.slave_exec_mode;
+set @@global.slave_exec_mode= idempotent;
+alter table t force;
+connection master1;
+set debug_sync= 'now wait_for go_dml';
+insert into t (a) values ('qux');
+set debug_sync= 'now signal go_alter';
+connection master;
+connection slave;
+connection master;
+drop table t;
+set global slave_exec_mode= @old_slave_exec_mode;
+set debug_sync= reset;
+#
+# End of 11.2 tests (Single-phase alter)
+#
+connection slave;
 include/stop_slave.inc
 set global slave_parallel_threads=3;
 set global slave_parallel_mode= optimistic;
@@ -49,6 +72,9 @@ connection master;
 drop table t;
 connection slave;
 connection master;
+#
+# End of 11.2 tests (Two-phase alter)
+#
 connection slave;
 include/stop_slave.inc
 set global binlog_row_image=FULL;

mysql-test/suite/rpl/t/rpl_alter_online_debug.test

@@ -8,6 +8,46 @@ call mtr.add_suppression("Unsafe statement written to the binary log using state
 --connection slave
 call mtr.add_suppression("Unsafe statement written to the binary log using statement format");
 
+#
+# Single-phase alter
+#
+
+--echo # MDEV-31804 Assertion `thd->m_transaction_psi == __null' fails upon
+--echo # replicating online ALTER
+--connection master
+create table t (a char(8)) engine=myisam;
+insert into t values ('foo'),('bar');
+
+set debug_sync= 'alter_table_online_progress signal go_dml wait_for go_alter';
+set @old_slave_exec_mode= @@global.slave_exec_mode;
+set @@global.slave_exec_mode= idempotent;
+send alter table t force;
+
+--connection master1
+set debug_sync= 'now wait_for go_dml';
+insert into t (a) values ('qux');
+set debug_sync= 'now signal go_alter';
+
+--connection master
+--reap
+--sync_slave_with_master
+
+# Cleanup
+--connection master
+drop table t;
+set global slave_exec_mode= @old_slave_exec_mode;
+set debug_sync= reset;
+
+
+--echo #
+--echo # End of 11.2 tests (Single-phase alter)
+--echo #
+
+
+#
+# Two-phase alter
+#
+--connection slave
 source include/stop_slave.inc;
 --let $slave_parallel_threads=`select @@global.slave_parallel_threads`
 --let $slave_parallel_mode=   `select @@global.slave_parallel_mode`
@@ -75,7 +115,7 @@ drop table t;
 --connection master
 
 --echo #
---echo # End of 11.2 tests
+--echo # End of 11.2 tests (Two-phase alter)
 --echo #
 
 --connection slave

sql/log_event.cc

@@ -719,6 +719,9 @@ const char* Log_event::get_type_str()
 Log_event::Log_event(const uchar *buf,
                      const Format_description_log_event* description_event)
   :temp_buf(0), exec_time(0), cache_type(Log_event::EVENT_INVALID_CACHE),
+#ifndef MYSQL_CLIENT
+    slave_exec_mode(SLAVE_EXEC_MODE_STRICT),
+#endif
     checksum_alg(BINLOG_CHECKSUM_ALG_UNDEF)
 {
 #ifndef MYSQL_CLIENT

sql/log_event_server.cc

@@ -545,6 +545,7 @@ int append_query_string(CHARSET_INFO *csinfo, String *to,
 
 Log_event::Log_event(THD* thd_arg, uint16 flags_arg, bool using_trans)
   :log_pos(0), temp_buf(0), exec_time(0),
+   slave_exec_mode(SLAVE_EXEC_MODE_STRICT),
    checksum_alg(BINLOG_CHECKSUM_ALG_UNDEF), thd(thd_arg)
 {
   server_id=	thd->variables.server_id;
@@ -569,6 +570,7 @@ Log_event::Log_event(THD* thd_arg, uint16 flags_arg, bool using_trans)
 
 Log_event::Log_event()
   :temp_buf(0), exec_time(0), flags(0), cache_type(EVENT_INVALID_CACHE),
+   slave_exec_mode(SLAVE_EXEC_MODE_STRICT),
    checksum_alg(BINLOG_CHECKSUM_ALG_UNDEF), thd(0)
 {
   server_id=	global_system_variables.server_id;
@@ -5099,7 +5101,8 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
       bitmap_set_bit(table->write_set, table->s->vers.end_fieldno);
     }
 
-    this->slave_exec_mode= (enum_slave_exec_mode)slave_exec_mode_options;
+    if (!rpl_data.is_online_alter())
+      this->slave_exec_mode= (enum_slave_exec_mode)slave_exec_mode_options;
 
     // Do event specific preparations 
     error= do_before_row_operations(rgi);

sql/sql_table.cc

@@ -12116,6 +12116,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     MEM_UNDEFINED(from->record[0], from->s->rec_buff_length * 2);
     MEM_UNDEFINED(to->record[0], to->s->rec_buff_length * 2);
     thd_progress_next_stage(thd);
+    enum_sql_command saved_sql_command= thd->lex->sql_command;
     Table_map_log_event table_event(thd, from, from->s->table_map_id,
                                     from->file->has_transactions());
     Relay_log_info rli(false);
@@ -12171,6 +12172,8 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     if (error)
       from->s->tdc->flush_unused(1); // to free the binlog
     to->pos_in_table_list= NULL; // Safety
+    DBUG_ASSERT(thd->lex->sql_command == saved_sql_command);
+    thd->lex->sql_command= saved_sql_command; // Just in case
   }
   else if (online) // error was on copy stage
   {