MDEV-29621: Replica stopped by locks on sequencebb-10.4-andrei

When using binlog_row_image=FULL with sequence table inserts, a
replica can deadlock because it treats full inserts in a sequence as DDL
statements by getting an exclusive lock on the sequence table. It
has been observed that with parallel replication, this exclusive
lock on the sequence table can lead to a deadlock where one
transaction has the exclusive lock and is waiting on a prior
transaction to commit, whereas this prior transaction is waiting on
the MDL lock.

This fix for this is on the master side, to raise FL_DDL
flag on the GTID of a full binlog_row_image INSERT of a sequence table.
This forces the slave to execute the statement serially so a deadlock
cannot happen.

A test verifies the deadlock also to prove it happen on the pre-fixes
slave.

OLD (buggy) -> NEW (fixed) replication is provided.
As the pre-fixes master's full row-image may represent both
SELECT NEXT VALUE and INSERT, in case of a buggy master,
the fixed parallel slave pessimistically waits in either case for
the prior transaction to commit before to take on the
critical part of the event execution.
The waiting exploits a parallel slave's retry mechanism which is
controlled by `@@global.slave_transaction_retries`.

Note that in order to avoid any persistent 'Deadlock found' 2013 error
in OLD -> NEW, `slave_transaction_retries` may need to be set to a
higher than the default value.
START-SLAVE is an effective work-around if this still happens.

8 files changed, 190 insertions, 13 deletions

diff --git a/mysql-test/std_data/rpl/master-bin-seq_10.3.36.000001 b/mysql-test/std_data/rpl/master-bin-seq_10.3.36.000001
new file mode 100644
index 00000000000..0fa163d0484
--- /dev/null
+++ b/mysql-test/std_data/rpl/master-bin-seq_10.3.36.000001
diff --git a/mysql-test/suite/rpl/r/rpl_parallel_seq.result b/mysql-test/suite/rpl/r/rpl_parallel_seq.result
new file mode 100644
index 00000000000..d3512d6cc53
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_parallel_seq.result
@@ -0,0 +1,46 @@
+include/master-slave.inc
+[connection master]
+connection slave;
+include/stop_slave.inc
+ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
+# MDEV-29621 the sequence engine binlog_row_image-full events
+#            MDL-deadlock on the parallel slave.
+connection master;
+CREATE SEQUENCE s1;
+SET @@session.binlog_row_image=FULL;
+SET @@session.debug_dbug="+d,binlog_force_commit_id";
+SET @commit_id=7;
+SET @@gtid_seq_no=100;
+SELECT NEXT VALUE FOR s1;
+NEXT VALUE FOR s1
+1
+INSERT INTO s1 VALUES(2, 1, 10, 1, 2, 1, 1, 0);
+SET @@session.debug_dbug="";
+connection slave;
+SET @@global.slave_parallel_threads=2;
+SET @@global.slave_parallel_mode=optimistic;
+SET @@global.debug_dbug="+d,hold_worker_on_schedule";
+include/start_slave.inc
+SET DEBUG_SYNC = 'now SIGNAL continue_worker';
+connection master;
+DROP SEQUENCE s1;
+connection slave;
+include/stop_slave.inc
+# Simulate buggy 10.3.36 master to prove the parallel applier
+# does not deadlock now at replaying the above master load.
+connection master;
+include/rpl_stop_server.inc [server_number=1]
+include/rpl_start_server.inc [server_number=1]
+connection slave;
+RESET MASTER;
+SET @@global.gtid_slave_pos="";
+CHANGE MASTER TO master_host='127.0.0.1', master_port=SERVER_MYPORT_1, master_user='root', master_use_gtid=slave_pos;
+START SLAVE UNTIL MASTER_GTID_POS='0-1-102';
+SET DEBUG_SYNC = 'now SIGNAL continue_worker';
+include/wait_for_slave_to_stop.inc
+SET debug_sync = RESET;
+SET @@global.slave_parallel_threads= 0;
+SET @@global.slave_parallel_mode= conservative;
+SET @@global.debug_dbug = "";
+include/start_slave.inc
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_parallel_seq.test b/mysql-test/suite/rpl/t/rpl_parallel_seq.test
new file mode 100644
index 00000000000..e92c0f61746
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_parallel_seq.test
@@ -0,0 +1,81 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+
+--connection slave
+--source include/stop_slave.inc
+ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
+
+--echo # MDEV-29621 the sequence engine binlog_row_image-full events
+--echo #            MDL-deadlock on the parallel slave.
+--connection master
+CREATE SEQUENCE s1;
+SET @@session.binlog_row_image=FULL;
+SET @@session.debug_dbug="+d,binlog_force_commit_id";
+SET @commit_id=7;
+SET @@gtid_seq_no=100;
+SELECT NEXT VALUE FOR s1;
+INSERT INTO s1 VALUES(2, 1, 10, 1, 2, 1, 1, 0);
+SET @@session.debug_dbug="";
+
+--connection slave
+--let $slave_parallel_threads=`select @@global.slave_parallel_threads`
+--let $slave_parallel_mode=`select @@global.slave_parallel_mode`
+SET @@global.slave_parallel_threads=2;
+SET @@global.slave_parallel_mode=optimistic;
+SET @@global.debug_dbug="+d,hold_worker_on_schedule";
+--source include/start_slave.inc
+
+--let $wait_condition= SELECT count(*) = 1 FROM information_schema.processlist WHERE state LIKE "Waiting for prior transaction to start commit before starting%"
+--source include/wait_condition.inc
+SET DEBUG_SYNC = 'now SIGNAL continue_worker';
+
+--connection master
+DROP SEQUENCE s1;
+--sync_slave_with_master
+--source include/stop_slave.inc
+
+--echo # Simulate buggy 10.3.36 master to prove the parallel applier
+--echo # does not deadlock now at replaying the above master load.
+--connection master
+--let $datadir= `SELECT @@datadir`
+
+--let $rpl_server_number= 1
+--source include/rpl_stop_server.inc
+
+--remove_file $datadir/master-bin.000001
+--copy_file $MYSQL_TEST_DIR/std_data/rpl/master-bin-seq_10.3.36.000001 $datadir/master-bin.000001
+
+--let $rpl_server_number= 1
+--source include/rpl_start_server.inc
+
+--source include/wait_until_connected_again.inc
+--save_master_pos
+
+--connection slave
+RESET MASTER;
+SET @@global.gtid_slave_pos="";
+
+--replace_result $SERVER_MYPORT_1 SERVER_MYPORT_1
+eval CHANGE MASTER TO master_host='127.0.0.1', master_port=$SERVER_MYPORT_1, master_user='root', master_use_gtid=slave_pos;
+
+START SLAVE UNTIL MASTER_GTID_POS='0-1-102';
+
+--let $wait_condition= SELECT count(*) = 1 FROM information_schema.processlist WHERE state LIKE "Waiting for prior transaction to commit"
+--source include/wait_condition.inc
+SET DEBUG_SYNC = 'now SIGNAL continue_worker';
+
+#
+# MDEV-29621 clean up.
+#
+--source include/wait_for_slave_to_stop.inc
+SET debug_sync = RESET;
+--eval SET @@global.slave_parallel_threads= $slave_parallel_threads
+--eval SET @@global.slave_parallel_mode= $slave_parallel_mode
+       SET @@global.debug_dbug = "";
+--source include/start_slave.inc
+
+
+--source include/rpl_end.inc
diff --git a/sql/ha_sequence.cc b/sql/ha_sequence.cc
index 1331fea74d1..03aee6a43dc 100644
--- a/sql/ha_sequence.cc
+++ b/sql/ha_sequence.cc
@@ -240,6 +240,8 @@ int ha_sequence::write_row(const uchar *buf)
         on master and slaves
       - Check that the new row is an accurate SEQUENCE object
     */
+    /* mark a full binlog image insert to force non-parallel slave */
+    thd->transaction.stmt.mark_trans_did_ddl();
     if (table->s->tmp_table == NO_TMP_TABLE &&
         thd->mdl_context.upgrade_shared_lock(table->mdl_ticket,
                                              MDL_EXCLUSIVE,
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 74b39dcadef..db682bd0835 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -13762,8 +13762,14 @@ Rows_log_event::write_row(rpl_group_info *rgi,
 int Rows_log_event::update_sequence()
 {
   TABLE *table= m_table;  // pointer to event's table
+  bool old_master= false;
+  int err= 0;
 
-  if (!bitmap_is_set(table->rpl_write_set, MIN_VALUE_FIELD_NO))
+  if (!bitmap_is_set(table->rpl_write_set, MIN_VALUE_FIELD_NO) ||
+      (!(table->in_use->rgi_slave->gtid_ev_flags2 & Gtid_log_event::FL_DDL) &&
+       !(old_master=
+         rpl_master_has_bug(thd->rgi_slave->rli,
+                            29621, FALSE, FALSE, FALSE, TRUE))))
   {
     /* This event come from a setval function executed on the master.
        Update the sequence next_number and round, like we do with setval()
@@ -13776,12 +13782,27 @@ int Rows_log_event::update_sequence()
 
     return table->s->sequence->set_value(table, nextval, round, 0) > 0;
   }
-
+  if (thd->rgi_slave->is_parallel_exec && old_master)
+  {
+    DBUG_ASSERT(thd->rgi_slave->parallel_entry);
+    /*
+      With parallel replication enabled, we can't execute alongside any other
+      transaction in which we may depend, so we force retry to release
+      the server layer table lock for possible prior in binlog order
+      same table transactions.
+    */
+    if (thd->rgi_slave->parallel_entry->last_committed_sub_id <
+        thd->rgi_slave->wait_commit_sub_id)
+    {
+      err= ER_LOCK_DEADLOCK;
+      my_error(err, MYF(0));
+    }
+  }
   /*
     Update all fields in table and update the active sequence, like with
     ALTER SEQUENCE
   */
-  return table->file->ha_write_row(table->record[0]);
+  return err == 0 ? table->file->ha_write_row(table->record[0]) : err;
 }
 
 
diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
index 746c923ba44..b550315d69f 100644
--- a/sql/rpl_parallel.cc
+++ b/sql/rpl_parallel.cc
@@ -2847,7 +2847,12 @@ rpl_parallel::do_event(rpl_group_info *serial_rgi, Log_event *ev,
 
       if (mode <= SLAVE_PARALLEL_MINIMAL ||
           !(gtid_flags & Gtid_log_event::FL_GROUP_COMMIT_ID) ||
-          e->last_commit_id != gtid_ev->commit_id)
+          e->last_commit_id != gtid_ev->commit_id ||
+          /*
+            MULTI_BATCH is also set when the current gtid even being a member
+            of a commit group is flagged as DDL which disallows parallel.
+          */
+          (gtid_flags & Gtid_log_event::FL_DDL))
         flags|= group_commit_orderer::MULTI_BATCH;
       /* Make sure we do not attempt to run DDL in parallel speculatively. */
       if (gtid_flags & Gtid_log_event::FL_DDL)
diff --git a/sql/slave.cc b/sql/slave.cc
index 42c293dfce4..3f5e1b43979 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -8086,14 +8086,15 @@ end:
    @return TRUE if master has the bug, FALSE if it does not.
 */
 bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
-                        bool (*pred)(const void *), const void *param)
+                        bool (*pred)(const void *), const void *param,
+                        bool maria_master)
 {
   struct st_version_range_for_one_bug {
     uint        bug_id;
     Version introduced_in; // first version with bug
     Version fixed_in;      // first version with fix
   };
-  static struct st_version_range_for_one_bug versions_for_all_bugs[]=
+  static struct st_version_range_for_one_bug versions_for_their_bugs[]=
   {
     {24432, { 5, 0, 24 }, { 5, 0, 38 } },
     {24432, { 5, 1, 12 }, { 5, 1, 17 } },
@@ -8101,11 +8102,27 @@ bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
     {33029, { 5, 1,  0 }, { 5, 1, 12 } },
     {37426, { 5, 1,  0 }, { 5, 1, 26 } },
   };
+  static struct st_version_range_for_one_bug versions_for_our_bugs[]=
+  {
+    {29621, { 10, 3, 36 }, { 10, 3, 39 } },
+    {29621, { 10, 4, 26 }, { 10, 4, 29 } },
+    {29621, { 10, 5, 17 }, { 10, 5, 20 } },
+    {29621, { 10, 6, 9  }, { 10, 6, 13 } },
+    {29621, { 10, 7, 5  }, { 10, 7, 9 } },
+    {29621, { 10, 8, 4  }, { 10, 8, 8  } },
+    {29621, { 10, 9, 2  }, { 10, 9, 6  } },
+    {29621, { 10, 10,1  }, { 10, 10,4  } },
+    {29621, { 10, 11,1  }, { 10, 11,3  } },
+  };
   const Version &master_ver=
     rli->relay_log.description_event_for_exec->server_version_split;
-
-  for (uint i= 0;
-       i < sizeof(versions_for_all_bugs)/sizeof(*versions_for_all_bugs);i++)
+  struct st_version_range_for_one_bug* versions_for_all_bugs= maria_master ?
+    versions_for_our_bugs : versions_for_their_bugs;
+  uint all_size= maria_master ? 
+    sizeof(versions_for_our_bugs)/sizeof(*versions_for_our_bugs) :
+    sizeof(versions_for_their_bugs)/sizeof(*versions_for_their_bugs);
+    
+  for (uint i= 0; i < all_size; i++)
   {
     const Version &introduced_in= versions_for_all_bugs[i].introduced_in;
     const Version &fixed_in= versions_for_all_bugs[i].fixed_in;
@@ -8114,18 +8131,21 @@ bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
         fixed_in > master_ver &&
         (pred == NULL || (*pred)(param)))
     {
+      const char *bug_source= maria_master ?
+        "https://jira.mariadb.org/browse/MDEV-" :
+        "http://bugs.mysql.com/bug.php?id=";
       if (!report)
 	return TRUE;
       // a short message for SHOW SLAVE STATUS (message length constraints)
       my_printf_error(ER_UNKNOWN_ERROR, "master may suffer from"
-                      " http://bugs.mysql.com/bug.php?id=%u"
+                      " %s%u"
                       " so slave stops; check error log on slave"
-                      " for more info", MYF(0), bug_id);
+                      " for more info", MYF(0), bug_source, bug_id);
       // a verbose message for the error log
       rli->report(ERROR_LEVEL, ER_UNKNOWN_ERROR, NULL,
                   "According to the master's version ('%s'),"
                   " it is probable that master suffers from this bug:"
-                      " http://bugs.mysql.com/bug.php?id=%u"
+                      " %s%u"
                       " and thus replicating the current binary log event"
                       " may make the slave's data become different from the"
                       " master's data."
@@ -8139,6 +8159,7 @@ bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
                       " equal to '%d.%d.%d'. Then replication can be"
                       " restarted.",
                       rli->relay_log.description_event_for_exec->server_version,
+                      bug_source,
                       bug_id,
                       fixed_in[0], fixed_in[1], fixed_in[2]);
       return TRUE;
diff --git a/sql/slave.h b/sql/slave.h
index e2bd5cec1b9..02de9135c2a 100644
--- a/sql/slave.h
+++ b/sql/slave.h
@@ -231,7 +231,8 @@ bool show_all_master_info(THD* thd);
 void show_binlog_info_get_fields(THD *thd, List<Item> *field_list);
 bool show_binlog_info(THD* thd);
 bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
-                        bool (*pred)(const void *), const void *param);
+                        bool (*pred)(const void *), const void *param,
+                        bool maria_master= false);
 bool rpl_master_erroneous_autoinc(THD* thd);
 
 const char *print_slave_db_safe(const char *db);