MDEV-31792 Assertion fails in MDL_context::acquire_lock upon parallel replication of CREATE SEQUENCE

The assert's reason was in missed FL_DDL flagging of CREATE-or-REPLACE
Query event.
MDEV-27365 fixes covered only the non-pre-existing table execution branch so
did not see a possibility of implicit commit in
the middle of execution in a rollback branch when the being CREATEd
sequence table is actually replaced.
The pre-existing table branch cleared the DDL modification
flag so the query lost FL_DDL in binlog and its parallel execution
on slave may have ended up with the assert to indicate the query
is raced by a following in binlog order event.

Fixed with applying the MDEV-27365 pattern.
An mtr test is added to cover the rollback situation.
The description test [ pass ] with a generous number of mtr parallel
reties.

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

@@ -82,6 +82,37 @@ SELECT @@global.gtid_binlog_state, @@global.gtid_slave_pos as "all through 101 h
 @@global.gtid_binlog_state	all through 101 have been committed
 0-1-101	0-1-101
 connection slave;
+flush tables with read lock;
+connection master;
+CREATE OR REPLACE SEQUENCE s3 ENGINE=innodb;
+SELECT NEXT VALUE FOR s3 into @tmpvar;
+include/save_master_gtid.inc
+connection slave;
+unlock tables;
+include/sync_with_master_gtid.inc
+connection slave;
+flush tables with read lock;
+connection master;
+CREATE OR REPLACE SEQUENCE s3 ENGINE=innodb;
+SELECT NEXT VALUE FOR s3 into @tmpvar;
+include/save_master_gtid.inc
+connection slave;
+unlock tables;
+include/sync_with_master_gtid.inc
+connection slave;
+BEGIN /* slave local Trx */;
+select count(*) from s3;
+count(*)
+1
+connection master;
+CREATE OR REPLACE SEQUENCE s3 ENGINE=innodb;
+SELECT NEXT VALUE FOR s3 into @tmpvar;
+include/save_master_gtid.inc
+connection slave;
+connection slave;
+rollback /* Trx */;
+include/sync_with_master_gtid.inc
+connection slave;
 include/stop_slave.inc
 SET debug_sync = RESET;
 SET @@global.slave_parallel_threads= 0;
@@ -90,7 +121,7 @@ SET @@global.debug_dbug = "";
 SET @@global.gtid_strict_mode=0;
 include/start_slave.inc
 connection master;
-DROP SEQUENCE s2;
+DROP SEQUENCE s2,s3;
 DROP TABLE ti;
 connection slave;
 include/rpl_end.inc

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

@@ -128,8 +128,51 @@ SET DEBUG_SYNC = 'now SIGNAL continue_worker';
 
 SELECT @@global.gtid_binlog_state, @@global.gtid_slave_pos as "all through 101 have been committed";
 
+# MDEV-31792 Assertion in MDL_context::acquire_lock upon parallel replication of CREATE SEQUENCE
+
+--let $iter = 3
+while ($iter)
+{
+--connection slave
+if (`select $iter > 1`)
+{
+  flush tables with read lock;
+}
+if (`select $iter = 1`)
+{
+  BEGIN /* slave local Trx */;
+   select count(*) from s3;
+}
+
+--connection master
+CREATE OR REPLACE SEQUENCE s3 ENGINE=innodb;
+# select may return non-deterministically, don't print its result
+SELECT NEXT VALUE FOR s3 into @tmpvar;
+--source include/save_master_gtid.inc
+
+--connection slave
+--let $wait_condition= SELECT count(*) = 1 FROM information_schema.processlist WHERE state LIKE "Waiting for prior transaction to start commit%"
+--source include/wait_condition.inc
+
+if (`select $iter > 1`)
+{
+  unlock tables;
+}
+if (`select $iter = 1`)
+{
+--connection slave
+rollback /* Trx */;
+}
+
+
+--source include/sync_with_master_gtid.inc
+
+--dec $iter
+}
+
+
 #
-# MDEV-29621/MDEV-31077 clean up.
+# MDEV-29621/MDEV-31077/MDEV-31792 clean up.
 #
 --connection slave
 --source include/stop_slave.inc
@@ -142,7 +185,7 @@ SET debug_sync = RESET;
 --source include/start_slave.inc
 
 --connection master
-DROP SEQUENCE s2;
+DROP SEQUENCE s2,s3;
 DROP TABLE ti;
 
 --sync_slave_with_master

sql/log_event.cc

@@ -8023,6 +8023,9 @@ Gtid_log_event::Gtid_log_event(THD *thd_arg, uint64 seq_no_arg,
   /* Preserve any DDL or WAITED flag in the slave's binlog. */
   if (thd_arg->rgi_slave)
     flags2|= (thd_arg->rgi_slave->gtid_ev_flags2 & (FL_DDL|FL_WAITED));
+
+  DBUG_ASSERT(thd_arg->lex->sql_command != SQLCOM_CREATE_SEQUENCE ||
+              (flags2 & FL_DDL));
 }
 
 

sql/sql_table.cc

@@ -5101,7 +5101,13 @@ int create_table_impl(THD *thd, const LEX_CSTRING &orig_db,
           Rollback the empty transaction started in mysql_create_table()
           call to open_and_lock_tables() when we are using LOCK TABLES.
         */
-        (void) trans_rollback_stmt(thd);
+        {
+          uint save_unsafe_rollback_flags=
+            thd->transaction.stmt.m_unsafe_rollback_flags;
+          (void) trans_rollback_stmt(thd);
+          thd->transaction.stmt.m_unsafe_rollback_flags=
+            save_unsafe_rollback_flags;
+        }
         /* Remove normal table without logging. Keep tables locked */
         if (mysql_rm_table_no_locks(thd, &table_list, 0, 0, 0, 0, 1, 1))
           goto err;