MDEV-31838 Assertion fails upon replication online alter with MINIMAL row

Replica honors its own binlog_row_image value when it sets up read_set.
When a slave thread is applying replicated row events in parallel with the
running online alter, we need all columns to be read from the table
(for the online alter logged row event) not only those that were present in
the pre-image for the replicated row event.

Avoid shrinking the set when online alter is running, i.e. leave it all set

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

@@ -1,9 +1,13 @@
 include/master-slave.inc
 [connection master]
+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");
 include/stop_slave.inc
 set global slave_parallel_threads=3;
 set global slave_parallel_mode= optimistic;
+set global binlog_row_image=MINIMAL;
 connection master;
 #
 # MDEV-31755 Replica's DML event deadlocks wit online alter table
@@ -29,10 +33,25 @@ id	c	a
 2	qqq	1
 connection master;
 drop table t;
+# MDEV-31838 Assertion fails on replica upon parallel
+# replication with two-phase alter and MINIMAL row image
+create table t (id int, a int, primary key(id)) engine=aria;
+insert into t values (1,0);
+set @@session.binlog_alter_two_phase=1;
+set debug_sync= 'alter_table_online_progress signal go_dml wait_for go_alter';
+alter table t force, algorithm=copy, lock=none;
+connection master1;
+set binlog_row_image=MINIMAL;
+set debug_sync= 'now wait_for go_dml';
+update ignore t set a = 1;
+set debug_sync= 'now signal go_alter';
+connection master;
+drop table t;
 connection slave;
 connection master;
 connection slave;
 include/stop_slave.inc
+set global binlog_row_image=FULL;
 set global slave_parallel_threads=0;
 set global slave_parallel_mode=   optimistic;
 include/start_slave.inc

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

@@ -2,13 +2,19 @@ source include/have_debug_sync.inc;
 source include/have_innodb.inc;
 source include/master-slave.inc;
 
+--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");
+
 source include/stop_slave.inc;
 --let $slave_parallel_threads=`select @@global.slave_parallel_threads`
 --let $slave_parallel_mode=   `select @@global.slave_parallel_mode`
+--let $binlog_row_image= `select @@global.binlog_row_image`
 set global slave_parallel_threads=3;
 set global slave_parallel_mode= optimistic;
+set global binlog_row_image=MINIMAL;
 --connection master
 
 
@@ -43,6 +49,28 @@ select * from t;
 # Cleanup
 --connection master
 drop table t;
+
+--echo # MDEV-31838 Assertion fails on replica upon parallel
+--echo # replication with two-phase alter and MINIMAL row image
+create table t (id int, a int, primary key(id)) engine=aria;
+insert into t values (1,0);
+
+set @@session.binlog_alter_two_phase=1;
+set debug_sync= 'alter_table_online_progress signal go_dml wait_for go_alter';
+send alter table t force, algorithm=copy, lock=none;
+
+--connection master1
+set binlog_row_image=MINIMAL;
+set debug_sync= 'now wait_for go_dml';
+--disable_warnings
+update ignore t set a = 1;
+--enable_warnings
+set debug_sync= 'now signal go_alter';
+
+# Cleanup
+--connection master
+--reap
+drop table t;
 --sync_slave_with_master
 --connection master
 
@@ -52,6 +80,7 @@ drop table t;
 
 --connection slave
 source include/stop_slave.inc;
+--eval set global binlog_row_image=$binlog_row_image
 --eval set global slave_parallel_threads=$slave_parallel_threads
 --eval set global slave_parallel_mode=   $slave_parallel_mode
 source include/start_slave.inc;

sql/log_event_server.cc

@@ -5082,7 +5082,8 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
     else
     if (get_general_type_code() == WRITE_ROWS_EVENT)
       bitmap_copy(table->write_set, &m_cols); // for sequences
-    else
+    else // If online alter, leave all columns set (i.e. skip intersects)
+    if (!thd->slave_thread || !table->s->online_alter_binlog)
     {
       bitmap_intersect(table->read_set,&m_cols);
       if (get_general_type_code() == UPDATE_ROWS_EVENT)