MDEV-26: Global transaction ID.

Implement START SLAVE UNTIL master_gtid_pos = "<GTID position>".

Add test cases, including a test showing how to use this to promote
a new master among a set of slaves.

1 files changed, 71 insertions, 17 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index aa86fa6ff62..9fbcbb68145 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -6320,6 +6320,7 @@ Gtid_list_log_event::Gtid_list_log_event(const char *buf, uint event_len,
   : Log_event(buf, description_event), count(0), list(0)
 {
   uint32 i;
+  uint32 val;
   uint8 header_size= description_event->common_header_len;
   uint8 post_header_len= description_event->post_header_len[GTID_LIST_EVENT-1];
   if (event_len < header_size + post_header_len ||
@@ -6327,7 +6328,9 @@ Gtid_list_log_event::Gtid_list_log_event(const char *buf, uint event_len,
     return;
 
   buf+= header_size;
-  count= uint4korr(buf) & ((1<<28)-1);
+  val= uint4korr(buf);
+  count= val & ((1<<28)-1);
+  gl_flags= val & ((uint32)0xf << 28);
   buf+= 4;
   if (event_len - (header_size + post_header_len) < count*element_size ||
       (!(list= (rpl_gtid *)my_malloc(count*sizeof(*list) + (count == 0),
@@ -6348,8 +6351,9 @@ Gtid_list_log_event::Gtid_list_log_event(const char *buf, uint event_len,
 
 #ifdef MYSQL_SERVER
 
-Gtid_list_log_event::Gtid_list_log_event(rpl_binlog_state *gtid_set)
-  : count(gtid_set->count()), list(0)
+Gtid_list_log_event::Gtid_list_log_event(rpl_binlog_state *gtid_set,
+                                         uint32 gl_flags_)
+  : count(gtid_set->count()), gl_flags(gl_flags_), list(0)
 {
   cache_type= EVENT_NO_CACHE;
   /* Failure to allocate memory will be caught by is_valid() returning false. */
@@ -6359,32 +6363,70 @@ Gtid_list_log_event::Gtid_list_log_event(rpl_binlog_state *gtid_set)
     gtid_set->get_gtid_list(list, count);
 }
 
+
+#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
 bool
-Gtid_list_log_event::write(IO_CACHE *file)
+Gtid_list_log_event::to_packet(String *packet)
 {
   uint32 i;
-  uchar buf[element_size];
+  uchar *p;
+  uint32 needed_length;
 
   DBUG_ASSERT(count < 1<<28);
 
-  if (write_header(file, get_data_size()))
-    return 1;
-  int4store(buf, count & ((1<<28)-1));
-  if (wrapper_my_b_safe_write(file, buf, GTID_LIST_HEADER_LEN))
-    return 1;
+  needed_length= packet->length() + 4 + count*element_size;
+  if (packet->reserve(needed_length))
+    return true;
+  p= (uchar *)packet->ptr() + packet->length();;
+  packet->length(needed_length);
+  int4store(p, (count & ((1<<28)-1)) | gl_flags);
+  p += 4;
   for (i= 0; i < count; ++i)
   {
-    int4store(buf, list[i].domain_id);
-    int4store(buf+4, list[i].server_id);
-    int8store(buf+8, list[i].seq_no);
-    if (wrapper_my_b_safe_write(file, buf, element_size))
-      return 1;
+    int4store(p, list[i].domain_id);
+    int4store(p+4, list[i].server_id);
+    int8store(p+8, list[i].seq_no);
+    p += 16;
   }
-  return write_footer(file);
+
+  return false;
+}
+
+
+bool
+Gtid_list_log_event::write(IO_CACHE *file)
+{
+  char buf[128];
+  String packet(buf, sizeof(buf), system_charset_info);
+
+  packet.length(0);
+  if (to_packet(&packet))
+    return true;
+  return
+    write_header(file, get_data_size()) ||
+    wrapper_my_b_safe_write(file, (uchar *)packet.ptr(), packet.length()) ||
+    write_footer(file);
+}
+
+
+int
+Gtid_list_log_event::do_apply_event(Relay_log_info const *rli)
+{
+  int ret= Log_event::do_apply_event(rli);
+  if (rli->until_condition == Relay_log_info::UNTIL_GTID &&
+      (gl_flags & FLAG_UNTIL_REACHED))
+  {
+    char str_buf[128];
+    String str(str_buf, sizeof(str_buf), system_charset_info);
+    const_cast<Relay_log_info*>(rli)->until_gtid_pos.to_string(&str);
+    sql_print_information("Slave SQL thread stops because it reached its"
+                          " UNTIL master_gtid_pos %s", str.c_ptr_safe());
+    const_cast<Relay_log_info*>(rli)->abort_slave= true;
+  }
+  return ret;
 }
 
 
-#ifdef HAVE_REPLICATION
 void
 Gtid_list_log_event::pack_info(THD *thd, Protocol *protocol)
 {
@@ -6439,12 +6481,24 @@ Gtid_list_log_event::print(FILE *file, PRINT_EVENT_INFO *print_event_info)
 */
 bool
 Gtid_list_log_event::peek(const char *event_start, uint32 event_len,
+                          uint8 checksum_alg,
                           rpl_gtid **out_gtid_list, uint32 *out_list_len)
 {
   const char *p;
   uint32 count_field, count;
   rpl_gtid *gtid_list;
 
+  if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32)
+  {
+    if (event_len > BINLOG_CHECKSUM_LEN)
+      event_len-= BINLOG_CHECKSUM_LEN;
+    else
+      event_len= 0;
+  }
+  else
+    DBUG_ASSERT(checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF ||
+                checksum_alg == BINLOG_CHECKSUM_ALG_OFF);
+
   if (event_len < LOG_EVENT_HEADER_LEN + GTID_LIST_HEADER_LEN)
     return true;
   p= event_start + LOG_EVENT_HEADER_LEN;