MDEV-5189: Incorrect parallel apply in parallel replication

Two problems were fixed:

1. When not in GTID mode (master_use_gtid=no), then we must not apply events
   in different domains in parallel (in non-GTID mode we are not capable of
   restarting at different points in different domains).

2. When transactions B and C group commit together, but after and separate
   from A, we can apply B and C in parallel, but both B and C must not start
   until A has committed. Fix sub_id to be globally increasing (not just
   per-domain increasing) so that this wait (which is based on sub_id) can be
   done correctly.

sql/rpl_gtid.cc

@@ -83,7 +83,7 @@ rpl_slave_state::record_and_update_gtid(THD *thd, rpl_group_info *rgi)
 
 
 rpl_slave_state::rpl_slave_state()
-  : inited(false), loaded(false)
+  : last_sub_id(0), inited(false), loaded(false)
 {
   my_hash_init(&hash, &my_charset_bin, 32, offsetof(element, domain_id),
                sizeof(uint32), NULL, my_free, HASH_UNIQUE);
@@ -153,6 +153,9 @@ rpl_slave_state::update(uint32 domain_id, uint32 server_id, uint64 sub_id,
   list_elem->seq_no= seq_no;
 
   elem->add(list_elem);
+  if (last_sub_id < sub_id)
+    last_sub_id= sub_id;
+
   return 0;
 }
 
@@ -169,7 +172,6 @@ rpl_slave_state::get_element(uint32 domain_id)
   if (!(elem= (element *)my_malloc(sizeof(*elem), MYF(MY_WME))))
     return NULL;
   elem->list= NULL;
-  elem->last_sub_id= 0;
   elem->domain_id= domain_id;
   if (my_hash_insert(&hash, (uchar *)elem))
   {
@@ -469,13 +471,10 @@ end:
 uint64
 rpl_slave_state::next_subid(uint32 domain_id)
 {
-  uint32 sub_id= 0;
-  element *elem;
+  uint32 sub_id;
 
   lock();
-  elem= get_element(domain_id);
-  if (elem)
-    sub_id= ++elem->last_sub_id;
+  sub_id= ++last_sub_id;
   unlock();
 
   return sub_id;

sql/rpl_gtid.h

@@ -60,7 +60,6 @@ struct rpl_slave_state
   struct element
   {
     struct list_element *list;
-    uint64 last_sub_id;
     uint32 domain_id;
 
     list_element *grab_list() { list_element *l= list; list= NULL; return l; }
@@ -68,8 +67,6 @@ struct rpl_slave_state
     {
       l->next= list;
       list= l;
-      if (last_sub_id < l->sub_id)
-        last_sub_id= l->sub_id;
     }
   };
 
@@ -78,6 +75,7 @@ struct rpl_slave_state
   /* Mutex protecting access to the state. */
   mysql_mutex_t LOCK_slave_state;
 
+  uint64 last_sub_id;
   bool inited;
   bool loaded;
 

sql/rpl_parallel.cc

@@ -657,8 +657,10 @@ rpl_parallel::do_event(rpl_group_info *serial_rgi, Log_event *ev,
   if (typ == GTID_EVENT)
   {
     Gtid_log_event *gtid_ev= static_cast<Gtid_log_event *>(ev);
+    uint32 domain_id= (rli->mi->using_gtid == Master_info::USE_GTID_NO ?
+                       0 : gtid_ev->domain_id);
 
-    if (!(e= find(gtid_ev->domain_id)) ||
+    if (!(e= find(domain_id)) ||
         !(rgi= new rpl_group_info(rli)) ||
         event_group_new_gtid(rgi, gtid_ev))
     {