MDEV-26: Global transaction commit. Intermediate commit.

Now slave records GTID in mysql.rpl_slave_state when applying XID log event.

sql/log.cc

@@ -120,6 +120,8 @@ static MYSQL_BIN_LOG::xid_count_per_binlog *
 static bool start_binlog_background_thread();
 
 
+rpl_binlog_state rpl_global_gtid_binlog_state;
+
 /**
    purge logs, master and slave sides both, related error code
    convertor.
@@ -5334,7 +5336,7 @@ MYSQL_BIN_LOG::write_gtid_event(THD *thd, bool standalone,
 
   /* Update the replication state (last GTID in each replication domain). */
   mysql_mutex_lock(&LOCK_rpl_gtid_state);
-  global_rpl_gtid_state.update(&gtid);
+  rpl_global_gtid_binlog_state.update(&gtid);
   mysql_mutex_unlock(&LOCK_rpl_gtid_state);
   return false;
 }

sql/log_event.h

@@ -2953,18 +2953,92 @@ struct rpl_gtid
 };
 
 
-struct rpl_state
+/*
+  Replication slave state.
+
+  For every independent replication stream (identified by domain_id), this
+  remembers the last gtid applied on the slave within this domain.
+
+  Since events are always committed in-order within a single domain, this is
+  sufficient to maintain the state of the replication slave.
+*/
+struct rpl_slave_state
 {
+  /* Elements in the list of GTIDs kept for each domain_id. */
+  struct list_element
+  {
+    struct list_element *next;
+    uint64 sub_id;
+    uint64 seq_no;
+    uint32 server_id;
+  };
+
+  /* Elements in the HASH that hold the state for one domain_id. */
+  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; }
+    void add (list_element *l)
+    {
+      l->next= list;
+      list= l;
+      if (last_sub_id < l->sub_id)
+        last_sub_id= l->sub_id;
+    }
+  };
+
+  /* Mapping from domain_id to its element. */
   HASH hash;
+  /* Mutex protecting access to the state. */
+  mysql_mutex_t LOCK_slave_state;
+
+  bool inited;
+  bool loaded;
 
-  rpl_state();
-  ~rpl_state();
+  rpl_slave_state();
+  ~rpl_slave_state();
 
+  void init();
+  void deinit();
   ulong count() const { return hash.records; }
-  int update(const struct rpl_gtid *gtid);
+  int update(uint32 domain_id, uint32 server_id, uint64 sub_id, uint64 seq_no);
+  int record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
+                      bool in_transaction);
+  uint64 next_subid(uint32 domain_id);
+
+  void lock() { DBUG_ASSERT(inited); mysql_mutex_lock(&LOCK_slave_state); }
+  void unlock() { DBUG_ASSERT(inited); mysql_mutex_unlock(&LOCK_slave_state); }
+
+  element *get_element(uint32 domain_id);
 };
 
-extern rpl_state global_rpl_gtid_state;
+
+/*
+  Binlog state.
+  This keeps the last GTID written to the binlog for every distinct
+  (domain_id, server_id) pair.
+  This will be logged at the start of the next binlog file as a
+  Gtid_list_log_event; this way, it is easy to find the binlog file
+  containing a gigen GTID, by simply scanning backwards from the newest
+  one until a lower seq_no is found in the Gtid_list_log_event at the
+  start of a binlog for the given domain_id and server_id.
+*/
+struct rpl_binlog_state
+{
+  /* Mapping from (domain_id,server_id) to its GTID. */
+  HASH hash;
+  /* Mutex protecting access to the state. */
+  mysql_mutex_t LOCK_binlog_state;
+
+  rpl_binlog_state();
+  ~rpl_binlog_state();
+
+  ulong count() const { return hash.records; }
+  int update(const struct rpl_gtid *gtid);
+};
 
 /**
   @class Gtid_log_event
@@ -3129,7 +3203,7 @@ public:
   static const uint element_size= 4+4+8;
 
 #ifdef MYSQL_SERVER
-  Gtid_list_log_event(rpl_state *gtid_set);
+  Gtid_list_log_event(rpl_binlog_state *gtid_set);
 #ifdef HAVE_REPLICATION
   void pack_info(THD *thd, Protocol *protocol);
 #endif

sql/mysqld.cc

@@ -766,6 +766,7 @@ PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_xid_list,
   key_LOCK_error_messages, key_LOG_INFO_lock, key_LOCK_thread_count,
   key_PARTITION_LOCK_auto_inc;
 PSI_mutex_key key_RELAYLOG_LOCK_index;
+PSI_mutex_key key_LOCK_slave_state, key_LOCK_binlog_state;
 
 PSI_mutex_key key_LOCK_stats,
   key_LOCK_global_user_client_stats, key_LOCK_global_table_stats,
@@ -838,7 +839,9 @@ static PSI_mutex_info all_server_mutexes[]=
   { &key_LOCK_commit_ordered, "LOCK_commit_ordered", PSI_FLAG_GLOBAL},
   { &key_LOG_INFO_lock, "LOG_INFO::lock", 0},
   { &key_LOCK_thread_count, "LOCK_thread_count", PSI_FLAG_GLOBAL},
-  { &key_PARTITION_LOCK_auto_inc, "HA_DATA_PARTITION::LOCK_auto_inc", 0}
+  { &key_PARTITION_LOCK_auto_inc, "HA_DATA_PARTITION::LOCK_auto_inc", 0},
+  { &key_LOCK_slave_state, "key_LOCK_slave_state", 0},
+  { &key_LOCK_binlog_state, "key_LOCK_binlog_state", 0}
 };
 
 PSI_rwlock_key key_rwlock_LOCK_grant, key_rwlock_LOCK_logger,
@@ -1783,6 +1786,7 @@ static void mysqld_exit(int exit_code)
     but if a kill -15 signal was sent, the signal thread did
     spawn the kill_server_thread thread, which is running concurrently.
   */
+  rpl_deinit_gtid_slave_state();
   wait_for_signal_thread_to_end();
   mysql_audit_finalize();
   clean_up_mutexes();
@@ -4064,6 +4068,10 @@ static int init_thread_environment()
 				     PTHREAD_CREATE_DETACHED);
   pthread_attr_setscope(&connection_attrib, PTHREAD_SCOPE_SYSTEM);
 
+#ifdef HAVE_REPLICATION
+  rpl_init_gtid_slave_state();
+#endif
+
   DBUG_RETURN(0);
 }
 

sql/mysqld.h

@@ -247,6 +247,7 @@ extern PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_xid_list,
   key_structure_guard_mutex, key_TABLE_SHARE_LOCK_ha_data,
   key_LOCK_error_messages, key_LOCK_thread_count, key_PARTITION_LOCK_auto_inc;
 extern PSI_mutex_key key_RELAYLOG_LOCK_index;
+extern PSI_mutex_key key_LOCK_slave_state, key_LOCK_binlog_state;
 
 extern PSI_mutex_key key_LOCK_stats,
   key_LOCK_global_user_client_stats, key_LOCK_global_table_stats,

sql/rpl_rli.cc

@@ -31,6 +31,16 @@
 
 static int count_relay_log_space(Relay_log_info* rli);
 
+/**
+   Current replication state (hash of last GTID executed, per replication
+   domain).
+*/
+rpl_slave_state rpl_global_gtid_slave_state;
+
+const LEX_STRING rpl_gtid_slave_state_table_name=
+  { STRING_WITH_LEN("rpl_slave_state") };
+
+
 // Defined in slave.cc
 int init_intvar_from_file(int* var, IO_CACHE* f, int default_val);
 int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
@@ -51,7 +61,7 @@ Relay_log_info::Relay_log_info(bool is_slave_recovery)
    abort_pos_wait(0), slave_run_id(0), sql_thd(0),
    inited(0), abort_slave(0), slave_running(0), until_condition(UNTIL_NONE),
    until_log_pos(0), retried_trans(0), executed_entries(0),
-   tables_to_lock(0), tables_to_lock_count(0),
+   gtid_sub_id(0), tables_to_lock(0), tables_to_lock_count(0),
    last_event_start_time(0), deferred_events(NULL),m_flags(0),
    row_stmt_start_timestamp(0), long_find_row_note_printed(false),
    m_annotate_event(0)

sql/rpl_rli.h

@@ -307,6 +307,14 @@ public:
   char slave_patternload_file[FN_REFLEN]; 
   size_t slave_patternload_file_size;  
 
+  /*
+    Current GTID being processed.
+    The sub_id gives the binlog order within one domain_id. A zero sub_id
+    means that there is no active GTID.
+  */
+  uint64 gtid_sub_id;
+  rpl_gtid current_gtid;
+
   Relay_log_info(bool is_slave_recovery);
   ~Relay_log_info();
 
@@ -584,4 +592,8 @@ private:
 int init_relay_log_info(Relay_log_info* rli, const char* info_fname);
 
 
+extern const LEX_STRING rpl_gtid_slave_state_table_name;
+extern struct rpl_slave_state rpl_global_gtid_slave_state;
+
+
 #endif /* RPL_RLI_H */

sql/slave.cc

@@ -3724,6 +3724,15 @@ log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
     goto err;
   }
 
+  /* Load the set of seen GTIDs, if we did not already. */
+  if (rpl_load_gtid_slave_state(thd))
+  {
+    rli->report(ERROR_LEVEL, thd->stmt_da->sql_errno(), 
+                "Unable to load replication GTID slave state from mysql.%s: %s",
+                rpl_gtid_slave_state_table_name.str, thd->stmt_da->message());
+    goto err;
+  }
+
   /* execute init_slave variable */
   if (opt_init_slave.length)
   {
@@ -5189,6 +5198,27 @@ static Log_event* next_event(Relay_log_info* rli)
         inc_event_relay_log_pos()
       */
       rli->future_event_relay_log_pos= my_b_tell(cur_log);
+      /*
+        For GTID, allocate a new sub_id for the given domain_id.
+        The sub_id must be allocated in increasing order of binlog order.
+      */
+      if (ev->get_type_code() == GTID_EVENT)
+      {
+        Gtid_log_event *gev= static_cast<Gtid_log_event *>(ev);
+        uint64 sub_id= rpl_global_gtid_slave_state.next_subid(gev->domain_id);
+        if (!sub_id)
+        {
+          errmsg = "slave SQL thread aborted because of out-of-memory error";
+          if (hot_log)
+            mysql_mutex_unlock(log_lock);
+          goto err;
+        }
+        rli->gtid_sub_id= sub_id;
+        rli->current_gtid.server_id= gev->server_id;
+        rli->current_gtid.domain_id= gev->domain_id;
+        rli->current_gtid.seq_no= gev->seq_no;
+      }
+
       if (hot_log)
         mysql_mutex_unlock(log_lock);
       DBUG_RETURN(ev);

sql/sql_repl.cc

@@ -16,10 +16,12 @@
 
 #include "sql_priv.h"
 #include "unireg.h"
+#include "sql_base.h"
 #include "sql_parse.h"                          // check_access
 #ifdef HAVE_REPLICATION
 
 #include "rpl_mi.h"
+#include "rpl_rli.h"
 #include "sql_repl.h"
 #include "sql_acl.h"                            // SUPER_ACL
 #include "log_event.h"
@@ -748,7 +750,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
   mariadb_slave_capability= get_mariadb_slave_capability(thd);
   if (global_system_variables.log_warnings > 1)
     sql_print_information("Start binlog_dump to slave_server(%d), pos(%s, %lu)",
-                        thd->variables.server_id, log_ident, (ulong)pos);
+                          (int)thd->variables.server_id, log_ident, (ulong)pos);
   if (RUN_HOOK(binlog_transmit, transmit_start, (thd, flags, log_ident, pos)))
   {
     errmsg= "Failed to run hook 'transmit_start'";
@@ -2442,4 +2444,118 @@ int log_loaded_block(IO_CACHE* file)
   DBUG_RETURN(0);
 }
 
+
+/**
+   Initialise the slave replication state from the mysql.rpl_slave_state table.
+
+   This is called each time an SQL thread starts, but the data is only actually
+   loaded on the first call.
+
+   The slave state is the last GTID applied on the slave within each
+   replication domain.
+
+   To avoid row lock contention, there are multiple rows for each domain_id.
+   The one containing the current slave state is the one with the maximal
+   sub_id value, within each domain_id.
+
+    CREATE TABLE mysql.rpl_slave_state (
+      domain_id INT UNSIGNED NOT NULL,
+      sub_id BIGINT UNSIGNED NOT NULL,
+      server_id INT UNSIGNED NOT NULL,
+      seq_no BIGINT UNSIGNED NOT NULL,
+      PRIMARY KEY (domain_id, sub_id))
+*/
+
+void
+rpl_init_gtid_slave_state()
+{
+  rpl_global_gtid_slave_state.init();
+}
+
+
+void
+rpl_deinit_gtid_slave_state()
+{
+  rpl_global_gtid_slave_state.deinit();
+}
+
+
+int
+rpl_load_gtid_slave_state(THD *thd)
+{
+  TABLE_LIST tlist;
+  TABLE *table;
+  bool table_opened= false;
+  bool table_scanned= false;
+  DBUG_ENTER("rpl_load_gtid_slave_state");
+
+  int err= 0;
+  rpl_global_gtid_slave_state.lock();
+  if (rpl_global_gtid_slave_state.loaded)
+    goto end;
+
+  mysql_reset_thd_for_next_command(thd, 0);
+
+  tlist.init_one_table(STRING_WITH_LEN("mysql"),
+                       rpl_gtid_slave_state_table_name.str,
+                       rpl_gtid_slave_state_table_name.length,
+                       NULL, TL_READ);
+  if ((err= open_and_lock_tables(thd, &tlist, FALSE, 0)))
+    goto end;
+  table_opened= true;
+  table= tlist.table;
+
+  /*
+    ToDo: Check the table definition, error if not as expected.
+    We need the correct first 4 columns with correct type, and the primary key.
+  */
+
+  bitmap_set_bit(table->read_set, table->field[0]->field_index);
+  bitmap_set_bit(table->read_set, table->field[1]->field_index);
+  bitmap_set_bit(table->read_set, table->field[2]->field_index);
+  bitmap_set_bit(table->read_set, table->field[3]->field_index);
+  if ((err= table->file->ha_rnd_init_with_error(1)))
+    goto end;
+  table_scanned= true;
+  for (;;)
+  {
+    uint32 domain_id, server_id;
+    uint64 sub_id, seq_no;
+    if ((err= table->file->ha_rnd_next(table->record[0])))
+    {
+      if (err == HA_ERR_RECORD_DELETED)
+        continue;
+      else if (err == HA_ERR_END_OF_FILE)
+        break;
+      else
+        goto end;
+    }
+    domain_id= (ulonglong)table->field[0]->val_int();
+    sub_id= (ulonglong)table->field[1]->val_int();
+    server_id= (ulonglong)table->field[2]->val_int();
+    seq_no= (ulonglong)table->field[3]->val_int();
+    DBUG_PRINT("info", ("Read slave state row: %u:%u-%lu sub_id=%lu\n",
+                        (unsigned)domain_id, (unsigned)server_id,
+                        (ulong)seq_no, (ulong)sub_id));
+    if ((err= rpl_global_gtid_slave_state.update(domain_id, server_id,
+                                                 sub_id, seq_no)))
+      goto end;
+  }
+  err= 0;                                       /* Clear HA_ERR_END_OF_FILE */
+
+  rpl_global_gtid_slave_state.loaded= true;
+
+end:
+  if (table_scanned)
+  {
+    table->file->ha_index_or_rnd_end();
+    ha_commit_trans(thd, FALSE);
+    ha_commit_trans(thd, TRUE);
+  }
+  if (table_opened)
+    close_thread_tables(thd);
+  rpl_global_gtid_slave_state.unlock();
+  DBUG_RETURN(err);
+}
+
 #endif /* HAVE_REPLICATION */

sql/sql_repl.h

@@ -65,6 +65,11 @@ int log_loaded_block(IO_CACHE* file);
 int init_replication_sys_vars();
 void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos, ushort flags);
 
+extern PSI_mutex_key key_LOCK_slave_state, key_LOCK_binlog_state;
+void rpl_init_gtid_slave_state();
+void rpl_deinit_gtid_slave_state();
+int rpl_load_gtid_slave_state(THD *thd);
+
 #endif /* HAVE_REPLICATION */
 
 #endif /* SQL_REPL_INCLUDED */