MDEV-4692: mysql.gtid_slave_pos accumulates values for a domain

When we load the slave state from the mysql.gtid_slave_pos at server start, we
need to load all the rows into the in-memory hash, not just the most recent
one in each replication domain. Otherwise we accumulate cruft in the form of
old rows each time the server restarts.

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

@@ -134,5 +134,26 @@ SET sql_log_bin= 0;
 ALTER TABLE mysql.gtid_slave_pos ENGINE=MyISAM;
 SET sql_log_bin= 1;
 include/start_slave.inc
+INSERT INTO t1 VALUES (10);
+include/stop_slave.inc
+CHANGE MASTER TO master_use_gtid= slave_pos;
+include/start_slave.inc
+*** MDEV-4692: mysql.gtid_slave_pos accumulates values for a domain ***
+SELECT domain_id, COUNT(*) FROM mysql.gtid_slave_pos GROUP BY domain_id;
+domain_id	COUNT(*)
+0	2
+1	2
+INSERT INTO t1 VALUES (11);
+SELECT domain_id, COUNT(*) FROM mysql.gtid_slave_pos GROUP BY domain_id;
+domain_id	COUNT(*)
+0	2
+1	2
+include/start_slave.inc
+INSERT INTO t1 VALUES (12);
+INSERT INTO t1 VALUES (13);
+SELECT domain_id, COUNT(*) FROM mysql.gtid_slave_pos GROUP BY domain_id;
+domain_id	COUNT(*)
+0	2
+1	2
 DROP TABLE t1;
 include/rpl_end.inc

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

@@ -229,6 +229,49 @@ ALTER TABLE mysql.gtid_slave_pos ENGINE=MyISAM;
 SET sql_log_bin= 1;
 --source include/start_slave.inc
 
+--connection server_1
+INSERT INTO t1 VALUES (10);
+--save_master_pos
+
+--connection server_2
+--sync_with_master
+--source include/stop_slave.inc
+CHANGE MASTER TO master_use_gtid= slave_pos;
+--source include/start_slave.inc
+
+--echo *** MDEV-4692: mysql.gtid_slave_pos accumulates values for a domain ***
+SELECT domain_id, COUNT(*) FROM mysql.gtid_slave_pos GROUP BY domain_id;
+
+--connection server_1
+INSERT INTO t1 VALUES (11);
+--save_master_pos
+
+--connection server_2
+--sync_with_master
+SELECT domain_id, COUNT(*) FROM mysql.gtid_slave_pos GROUP BY domain_id;
+
+--write_file $MYSQLTEST_VARDIR/tmp/mysqld.2.expect
+wait
+EOF
+--shutdown_server 30
+--source include/wait_until_disconnected.inc
+
+--append_file $MYSQLTEST_VARDIR/tmp/mysqld.2.expect
+restart:  
+EOF
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+--source include/start_slave.inc
+
+--connection server_1
+INSERT INTO t1 VALUES (12);
+INSERT INTO t1 VALUES (13);
+--save_master_pos
+
+--connection server_2
+--sync_with_master
+SELECT domain_id, COUNT(*) FROM mysql.gtid_slave_pos GROUP BY domain_id;
+
 
 --connection server_1
 DROP TABLE t1;

sql/rpl_rli.cc

@@ -1386,9 +1386,11 @@ rpl_load_gtid_slave_state(THD *thd)
   TABLE *table;
   bool table_opened= false;
   bool table_scanned= false;
+  bool array_inited= false;
   struct local_element { uint64 sub_id; rpl_gtid gtid; };
-  struct local_element *entry;
+  struct local_element tmp_entry, *entry;
   HASH hash;
+  DYNAMIC_ARRAY array;
   int err= 0;
   uint32 i;
   DBUG_ENTER("rpl_load_gtid_slave_state");
@@ -1402,6 +1404,9 @@ rpl_load_gtid_slave_state(THD *thd)
   my_hash_init(&hash, &my_charset_bin, 32,
                offsetof(local_element, gtid) + offsetof(rpl_gtid, domain_id),
                sizeof(uint32), NULL, my_free, HASH_UNIQUE);
+  if ((err= my_init_dynamic_array(&array, sizeof(local_element), 0, 0, MYF(0))))
+    goto end;
+  array_inited= true;
 
   mysql_reset_thd_for_next_command(thd, 0);
 
@@ -1450,6 +1455,16 @@ rpl_load_gtid_slave_state(THD *thd)
                         (unsigned)domain_id, (unsigned)server_id,
                         (ulong)seq_no, (ulong)sub_id));
 
+    tmp_entry.sub_id= sub_id;
+    tmp_entry.gtid.domain_id= domain_id;
+    tmp_entry.gtid.server_id= server_id;
+    tmp_entry.gtid.seq_no= seq_no;
+    if ((err= insert_dynamic(&array, (uchar *)&tmp_entry)))
+    {
+      my_error(ER_OUT_OF_RESOURCES, MYF(0));
+      goto end;
+    }
+
     if ((rec= my_hash_search(&hash, (const uchar *)&domain_id, 0)))
     {
       entry= (struct local_element *)rec;
@@ -1488,18 +1503,24 @@ rpl_load_gtid_slave_state(THD *thd)
     rpl_global_gtid_slave_state.unlock();
     goto end;
   }
-  for (i= 0; i < hash.records; ++i)
+
+  for (i= 0; i < array.elements; ++i)
   {
-    entry= (struct local_element *)my_hash_element(&hash, i);
-    if ((err= rpl_global_gtid_slave_state.update(entry->gtid.domain_id,
-                                                 entry->gtid.server_id,
-                                                 entry->sub_id,
-                                                 entry->gtid.seq_no)))
+    get_dynamic(&array, (uchar *)&tmp_entry, i);
+    if ((err= rpl_global_gtid_slave_state.update(tmp_entry.gtid.domain_id,
+                                                 tmp_entry.gtid.server_id,
+                                                 tmp_entry.sub_id,
+                                                 tmp_entry.gtid.seq_no)))
     {
       rpl_global_gtid_slave_state.unlock();
       my_error(ER_OUT_OF_RESOURCES, MYF(0));
       goto end;
     }
+  }
+
+  for (i= 0; i < hash.records; ++i)
+  {
+    entry= (struct local_element *)my_hash_element(&hash, i);
     if (opt_bin_log &&
         mysql_bin_log.bump_seq_no_counter_if_needed(entry->gtid.domain_id,
                                                     entry->gtid.seq_no))
@@ -1509,6 +1530,7 @@ rpl_load_gtid_slave_state(THD *thd)
       goto end;
     }
   }
+
   rpl_global_gtid_slave_state.loaded= true;
   rpl_global_gtid_slave_state.unlock();
 
@@ -1526,6 +1548,8 @@ end:
     close_thread_tables(thd);
     thd->mdl_context.release_transactional_locks();
   }
+  if (array_inited)
+    delete_dynamic(&array);
   my_hash_free(&hash);
   DBUG_RETURN(err);
 }