MDEV-31755 Replica's DML event deadlocks wit online alter table

The deadlock was caused by too strong MDL acquired by the start ALTER.

While MDL is shared by the start ALTER wait for its 2nd part
to allow concurrent DML:s to grab the lock.
The fixes uses wait_for_master reentrancy.

mysql-test/suite/rpl/r/rpl_mdev-31755.result

+include/master-slave.inc
+[connection master]
+connection slave;
+include/stop_slave.inc
+set global slave_parallel_threads=3;
+set global slave_parallel_mode= optimistic;
+connection master;
+create table t (id int, a int, b text, primary key (id));
+insert into t values (1,10,''),(2,20,'');
+set @@session.binlog_alter_two_phase=1;
+set debug_sync= 'alter_table_online_progress signal ready wait_for go';
+alter table t force, algorithm=copy, lock=none;
+connect  con1,localhost,root,,;
+set debug_sync= 'now wait_for ready';
+update t set a = 1;
+set debug_sync= 'now signal go';
+connection master;
+include/save_master_gtid.inc
+connection slave;
+include/start_slave.inc
+include/sync_with_master_gtid.inc
+connection master;
+drop table t;
+connection slave;
+include/stop_slave.inc
+include/start_slave.inc
+include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_mdev-31755.test

+source include/have_innodb.inc;
+source include/master-slave.inc;
+
+--connection slave
+source include/stop_slave.inc;
+
+#
+# MDEV-31755 Replica's DML event deadlocks wit online alter table
+#
+# Three threads for SA,U,CA
+--let $slave_parallel_threads=`select @@global.slave_parallel_threads`
+--let $slave_parallel_mode=   `select @@global.slave_parallel_mode`
+set global slave_parallel_threads=3;
+set global slave_parallel_mode= optimistic;
+
+--connection master
+create table t (id int, a int, b text, primary key (id));
+insert into t values (1,10,''),(2,20,'');
+
+set @@session.binlog_alter_two_phase=1;
+set debug_sync= 'alter_table_online_progress signal ready wait_for go';
+send alter table t force, algorithm=copy, lock=none;
+
+connect (con1,localhost,root,,);
+set debug_sync= 'now wait_for ready';
+
+update t set a = 1;
+
+set debug_sync= 'now signal go';
+
+--connection master
+--reap
+--source include/save_master_gtid.inc
+
+--connection slave
+source include/start_slave.inc;
+--source include/sync_with_master_gtid.inc
+
+--connection master
+drop table t;
+
+--connection slave
+source include/stop_slave.inc;
+--exec set global slave_parallel_threads=$slave_parallel_threads
+--exec set global slave_parallel_mode=   $slave_parallel_mode
+source include/start_slave.inc;
+
+--source include/rpl_end.inc

sql/sql_table.cc

@@ -86,7 +86,7 @@ static int copy_data_between_tables(THD *, TABLE *,TABLE *,
                                     List<Create_field> &, bool, uint, ORDER *,
                                     ha_rows *, ha_rows *,
                                     Alter_info::enum_enable_or_disable,
-                                    Alter_table_ctx *, bool);
+                                    Alter_table_ctx *, bool, uint64);
 static int append_system_key_parts(THD *thd, HA_CREATE_INFO *create_info,
                                    Key *key);
 static int mysql_prepare_create_table(THD *, HA_CREATE_INFO *, Alter_info *,
@@ -11138,7 +11138,7 @@ do_continue:;
                                  alter_info->create_list, ignore,
                                  order_num, order, &copied, &deleted,
                                  alter_info->keys_onoff,
-                                 &alter_ctx, online))
+                                 &alter_ctx, online, start_alter_id))
       goto err_new_table_cleanup;
   }
   else
@@ -11698,7 +11698,8 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
                          uint order_num, ORDER *order,
                          ha_rows *copied, ha_rows *deleted,
                          Alter_info::enum_enable_or_disable keys_onoff,
-                         Alter_table_ctx *alter_ctx, bool online)
+                         Alter_table_ctx *alter_ctx, bool online,
+                         uint64 start_alter_id)
 {
   int error= 1;
   Copy_field *copy= NULL, *copy_end;
@@ -12109,6 +12110,13 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     thd->lock= NULL;
 
     error= online_alter_read_from_binlog(thd, &rgi, binlog);
+    if (start_alter_id)
+    {
+      DBUG_ASSERT(thd->slave_thread);
+
+      if (wait_for_master(thd))
+        error= 1;
+    }
 
     // flip() makes reinit_io_cache, so it should be here
     DBUG_EXECUTE_IF("online_alter_small_cache_2",