MDEV-12179: Per-engine mysql.gtid_slave_pos table

Intermediate commit.

Implement auto-creation of mysql.gtid_slave_pos* tables with needed engines,
if listed in --gtid-pos-auto-engines.

Uses an asynchronous approach to minimise locking overhead.

The list of available tables is extended with a flag. Extra entries are
added for --gtid-pos-auto-engines tables that do not exist yet, marked as
not existing but ready for auto-creation.

If record_gtid() needs a table marked for auto-creation, it sends a request
to the slave background thread to create the table, and continues to use an
existing table for the current and immediately coming transactions.

As soon as the slave background thread has made the new table available, it
will be used for all subsequent relevant transactions in record_gtid().

This asynchronous approach also avoids a lot of complex issues around trying
to do DDL in the middle of an on-going transaction.

5 files changed, 316 insertions, 17 deletions

diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index ac3c621a5c1..af70f281d8c 100644
--- a/sql/rpl_gtid.cc
+++ b/sql/rpl_gtid.cc
@@ -26,6 +26,7 @@
 #include "key.h"
 #include "rpl_gtid.h"
 #include "rpl_rli.h"
+#include "slave.h"
 
 
 const LEX_STRING rpl_gtid_slave_state_table_name=
@@ -494,8 +495,20 @@ rpl_slave_state::select_gtid_pos_table(THD *thd, LEX_STRING *out_tablename)
     {
       if (table_entry->table_hton == trx_hton)
       {
-        *out_tablename= table_entry->table_name;
-        return;
+        if (likely(table_entry->state == GTID_POS_AVAILABLE))
+        {
+          *out_tablename= table_entry->table_name;
+          return;
+        }
+        /*
+          This engine is marked to automatically create the table.
+          We cannot easily do this here (possibly in the middle of a
+          transaction). But we can request the slave background thread
+          to create it, and in a short while it should become available
+          for following transactions.
+        */
+        slave_background_gtid_pos_create_request(table_entry);
+        break;
       }
       table_entry= table_entry->next;
     }
@@ -1240,7 +1253,8 @@ rpl_slave_state::add_gtid_pos_table(rpl_slave_state::gtid_pos_table *entry)
 
 
 struct rpl_slave_state::gtid_pos_table *
-rpl_slave_state::alloc_gtid_pos_table(LEX_STRING *table_name, void *hton)
+rpl_slave_state::alloc_gtid_pos_table(LEX_STRING *table_name, void *hton,
+                                      rpl_slave_state::gtid_pos_table_state state)
 {
   struct gtid_pos_table *p;
   char *allocated_str;
@@ -1258,6 +1272,7 @@ rpl_slave_state::alloc_gtid_pos_table(LEX_STRING *table_name, void *hton)
   p->table_hton= hton;
   p->table_name.str= allocated_str;
   p->table_name.length= table_name->length;
+  p->state= state;
   return p;
 }
 
diff --git a/sql/rpl_gtid.h b/sql/rpl_gtid.h
index b4f80fe3ddf..e6a3ea1a667 100644
--- a/sql/rpl_gtid.h
+++ b/sql/rpl_gtid.h
@@ -157,6 +157,12 @@ struct rpl_slave_state
   };
 
   /* Descriptor for mysql.gtid_slave_posXXX table in specific engine. */
+  enum gtid_pos_table_state {
+    GTID_POS_AUTO_CREATE,
+    GTID_POS_CREATE_REQUESTED,
+    GTID_POS_CREATE_IN_PROGRESS,
+    GTID_POS_AVAILABLE
+  };
   struct gtid_pos_table {
     struct gtid_pos_table *next;
     /*
@@ -167,6 +173,7 @@ struct rpl_slave_state
     */
     void *table_hton;
     LEX_STRING table_name;
+    uint8 state;
   };
 
   /* Mapping from domain_id to its element. */
@@ -232,7 +239,8 @@ struct rpl_slave_state
   void set_gtid_pos_tables_list(gtid_pos_table *new_list,
                                 gtid_pos_table *default_entry);
   void add_gtid_pos_table(gtid_pos_table *entry);
-  struct gtid_pos_table *alloc_gtid_pos_table(LEX_STRING *table_name, void *hton);
+  struct gtid_pos_table *alloc_gtid_pos_table(LEX_STRING *table_name,
+      void *hton, rpl_slave_state::gtid_pos_table_state state);
   void free_gtid_pos_tables(struct gtid_pos_table *list);
 };
 
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 74e4be524b9..2cda87925d1 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -1690,7 +1690,9 @@ process_gtid_pos_table(THD *thd, LEX_STRING *table_name, void *hton,
     entry= entry->next;
   }
 
-  if (!(p= rpl_global_gtid_slave_state->alloc_gtid_pos_table(table_name, hton)))
+  p= rpl_global_gtid_slave_state->alloc_gtid_pos_table(table_name,
+      hton, rpl_slave_state::GTID_POS_AVAILABLE);
+  if (!p)
     return 1;
   p->next= data->table_list;
   data->table_list= p;
@@ -1700,6 +1702,59 @@ process_gtid_pos_table(THD *thd, LEX_STRING *table_name, void *hton,
 }
 
 
+/*
+  Put tables corresponding to @@gtid_pos_auto_engines at the end of the list,
+  marked to be auto-created if needed.
+*/
+static int
+gtid_pos_auto_create_tables(rpl_slave_state::gtid_pos_table **list_ptr)
+{
+  plugin_ref *auto_engines;
+  int err= 0;
+  mysql_mutex_lock(&LOCK_global_system_variables);
+  for (auto_engines= opt_gtid_pos_auto_plugins;
+       !err && auto_engines && *auto_engines;
+       ++auto_engines)
+  {
+    void *hton= plugin_hton(*auto_engines);
+    char buf[FN_REFLEN+1];
+    LEX_STRING table_name;
+    char *p;
+    rpl_slave_state::gtid_pos_table *entry, **next_ptr;
+
+    /* See if this engine is already in the list. */
+    next_ptr= list_ptr;
+    entry= *list_ptr;
+    while (entry)
+    {
+      if (entry->table_hton == hton)
+        break;
+      next_ptr= &entry->next;
+      entry= entry->next;
+    }
+    if (entry)
+      continue;
+
+    /* Add an auto-create entry for this engine at end of list. */
+    p= strmake(buf, rpl_gtid_slave_state_table_name.str, FN_REFLEN);
+    p= strmake(p, "_", FN_REFLEN - (p - buf));
+    p= strmake(p, plugin_name(*auto_engines)->str, FN_REFLEN - (p - buf));
+    table_name.str= buf;
+    table_name.length= p - buf;
+    entry= rpl_global_gtid_slave_state->alloc_gtid_pos_table
+      (&table_name, hton, rpl_slave_state::GTID_POS_AUTO_CREATE);
+    if (!entry)
+    {
+      err= 1;
+      break;
+    }
+    *next_ptr= entry;
+  }
+  mysql_mutex_unlock(&LOCK_global_system_variables);
+  return err;
+}
+
+
 static int
 load_gtid_state_cb(THD *thd, LEX_STRING *table_name, void *arg)
 {
@@ -1746,6 +1801,18 @@ rpl_load_gtid_slave_state(THD *thd)
   if ((err= scan_all_gtid_slave_pos_table(thd, load_gtid_state_cb, &cb_data)))
     goto end;
 
+  if (!cb_data.default_entry)
+  {
+    /*
+      If the mysql.gtid_slave_pos table does not exist, but at least one other
+      table is available, arbitrarily pick the first in the list to use as
+      default.
+    */
+    cb_data.default_entry= cb_data.table_list;
+  }
+  if ((err= gtid_pos_auto_create_tables(&cb_data.table_list)))
+    goto end;
+
   mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
   if (rpl_global_gtid_slave_state->loaded)
   {
@@ -1757,18 +1824,10 @@ rpl_load_gtid_slave_state(THD *thd)
   {
     my_error(ER_NO_SUCH_TABLE, MYF(0), "mysql",
              rpl_gtid_slave_state_table_name.str);
+    mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
     err= 1;
     goto end;
   }
-  else if (!cb_data.default_entry)
-  {
-    /*
-      If the mysql.gtid_slave_pos table does not exist, but at least one other
-      table is available, arbitrarily pick the first in the list to use as
-      default.
-    */
-    cb_data.default_entry= cb_data.table_list;
-  }
 
   for (i= 0; i < array.elements; ++i)
   {
@@ -1878,7 +1937,7 @@ find_gtid_slave_pos_tables(THD *thd)
     err= 1;
     goto end;
   }
-  else if (!cb_data.default_entry)
+  if (!cb_data.default_entry)
   {
     /*
       If the mysql.gtid_slave_pos table does not exist, but at least one other
@@ -1887,6 +1946,8 @@ find_gtid_slave_pos_tables(THD *thd)
     */
     cb_data.default_entry= cb_data.table_list;
   }
+  if ((err= gtid_pos_auto_create_tables(&cb_data.table_list)))
+    goto end;
 
   any_running= any_slave_sql_running();
   mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
diff --git a/sql/slave.cc b/sql/slave.cc
index fbdb78b5c5d..443e5b7a5f1 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -59,6 +59,7 @@
 #include "rpl_tblmap.h"
 #include "debug_sync.h"
 #include "rpl_parallel.h"
+#include "sql_show.h"
 
 
 #define FLAGSTR(V,F) ((V)&(F)?#F" ":"")
@@ -279,15 +280,178 @@ static void init_slave_psi_keys(void)
 #endif /* HAVE_PSI_INTERFACE */
 
 
+/*
+  Note: This definition needs to be kept in sync with the one in
+  mysql_system_tables.sql which is used by mysql_create_db.
+*/
+static const char gtid_pos_table_definition1[]=
+  "CREATE TABLE ";
+static const char gtid_pos_table_definition2[]=
+  " (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)) CHARSET=latin1 "
+  "COMMENT='Replication slave GTID position' "
+  "ENGINE=";
+
+/*
+  Build a query string
+    CREATE TABLE mysql.gtid_slave_pos_<engine> ... ENGINE=<engine>
+*/
+static bool
+build_gtid_pos_create_query(THD *thd, String *query,
+                            LEX_STRING *table_name,
+                            LEX_STRING *engine_name)
+{
+  bool err= false;
+  err|= query->append(gtid_pos_table_definition1);
+  err|= append_identifier(thd, query, table_name->str, table_name->length);
+  err|= query->append(gtid_pos_table_definition2);
+  err|= append_identifier(thd, query, engine_name->str, engine_name->length);
+  return err;
+}
+
+
+static int
+gtid_pos_table_creation(THD *thd, plugin_ref engine, LEX_STRING *table_name)
+{
+  int err;
+  StringBuffer<sizeof(gtid_pos_table_definition1) +
+               sizeof(gtid_pos_table_definition1) +
+               2*FN_REFLEN> query;
+
+  if (build_gtid_pos_create_query(thd, &query, table_name, plugin_name(engine)))
+  {
+    my_error(ER_OUT_OF_RESOURCES, MYF(0));
+    return 1;
+  }
+
+  thd->set_db("mysql", 5);
+  thd->clear_error();
+  ulonglong thd_saved_option= thd->variables.option_bits;
+  /* This query shuold not be binlogged. */
+  thd->variables.option_bits&= ~(ulonglong)OPTION_BIN_LOG;
+  thd->set_query_and_id(query.c_ptr(), query.length(), thd->charset(),
+                        next_query_id());
+  Parser_state parser_state;
+  err= parser_state.init(thd, thd->query(), thd->query_length());
+  if (err)
+    goto end;
+  mysql_parse(thd, thd->query(), thd->query_length(), &parser_state,
+              FALSE, FALSE);
+  if (thd->is_error())
+    err= 1;
+end:
+  thd->variables.option_bits= thd_saved_option;
+  thd->reset_query();
+  return err;
+}
+
+
+static void
+handle_gtid_pos_auto_create_request(THD *thd, void *hton)
+{
+  int err;
+  plugin_ref engine= NULL, *auto_engines;
+  rpl_slave_state::gtid_pos_table *entry;
+  StringBuffer<FN_REFLEN> loc_table_name;
+  LEX_STRING table_name;
+
+  /*
+    Check that the plugin is still in @@gtid_pos_auto_engines, and lock
+    it.
+  */
+  mysql_mutex_lock(&LOCK_global_system_variables);
+  engine= NULL;
+  for (auto_engines= opt_gtid_pos_auto_plugins;
+       auto_engines && *auto_engines;
+       ++auto_engines)
+  {
+    if (plugin_hton(*auto_engines) == hton)
+    {
+      engine= my_plugin_lock(NULL, *auto_engines);
+      break;
+    }
+  }
+  mysql_mutex_unlock(&LOCK_global_system_variables);
+  if (!engine)
+  {
+    /* The engine is gone from @@gtid_pos_auto_engines, so no action. */
+    goto end;
+  }
+
+  /* Find the entry for the table to auto-create. */
+  mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+  entry= rpl_global_gtid_slave_state->gtid_pos_tables;
+  while (entry)
+  {
+    if (entry->table_hton == hton &&
+        entry->state == rpl_slave_state::GTID_POS_CREATE_REQUESTED)
+      break;
+    entry= entry->next;
+  }
+  if (entry)
+  {
+    entry->state = rpl_slave_state::GTID_POS_CREATE_IN_PROGRESS;
+    err= loc_table_name.append(entry->table_name.str, entry->table_name.length);
+  }
+  mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+  if (!entry)
+    goto end;
+  if (err)
+  {
+    sql_print_error("Out of memory while trying to auto-create GTID position table");
+    goto end;
+  }
+  table_name.str= loc_table_name.c_ptr_safe();
+  table_name.length= loc_table_name.length();
+
+  err= gtid_pos_table_creation(thd, engine, &table_name);
+  if (err)
+  {
+    sql_print_error("Error auto-creating GTID position table `mysql.%s`: %s Error_code: %d",
+                    table_name.str, thd->get_stmt_da()->message(),
+                    thd->get_stmt_da()->sql_errno());
+    thd->clear_error();
+    goto end;
+  }
+
+  /* Now enable the entry for the auto-created table. */
+  mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+  entry= rpl_global_gtid_slave_state->gtid_pos_tables;
+  while (entry)
+  {
+    if (entry->table_hton == hton &&
+        entry->state == rpl_slave_state::GTID_POS_CREATE_IN_PROGRESS)
+    {
+      entry->state= rpl_slave_state::GTID_POS_AVAILABLE;
+      break;
+    }
+    entry= entry->next;
+  }
+  mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+
+end:
+  if (engine)
+    plugin_unlock(NULL, engine);
+}
+
+
 static bool slave_background_thread_running;
 static bool slave_background_thread_stop;
 static bool slave_background_thread_gtid_loaded;
 
-struct slave_background_kill_t {
+static struct slave_background_kill_t {
   slave_background_kill_t *next;
   THD *to_kill;
 } *slave_background_kill_list;
 
+static struct slave_background_gtid_pos_create_t {
+  slave_background_gtid_pos_create_t *next;
+  void *hton;
+} *slave_background_gtid_pos_create_list;
+
 
 pthread_handler_t
 handle_slave_background(void *arg __attribute__((unused)))
@@ -321,6 +485,7 @@ handle_slave_background(void *arg __attribute__((unused)))
   do
   {
     slave_background_kill_t *kill_list;
+    slave_background_gtid_pos_create_t *create_list;
 
     thd->ENTER_COND(&COND_slave_background, &LOCK_slave_background,
                     &stage_slave_background_wait_request,
@@ -329,12 +494,14 @@ handle_slave_background(void *arg __attribute__((unused)))
     {
       stop= abort_loop || thd->killed || slave_background_thread_stop;
       kill_list= slave_background_kill_list;
-      if (stop || kill_list)
+      create_list= slave_background_gtid_pos_create_list;
+      if (stop || kill_list || create_list)
         break;
       mysql_cond_wait(&COND_slave_background, &LOCK_slave_background);
     }
 
     slave_background_kill_list= NULL;
+    slave_background_gtid_pos_create_list= NULL;
     thd->EXIT_COND(&old_stage);
 
     while (kill_list)
@@ -353,6 +520,16 @@ handle_slave_background(void *arg __attribute__((unused)))
       mysql_mutex_unlock(&to_kill->LOCK_wakeup_ready);
       my_free(p);
     }
+
+    while (create_list)
+    {
+      slave_background_gtid_pos_create_t *next= create_list->next;
+      void *hton= create_list->hton;
+      handle_gtid_pos_auto_create_request(thd, hton);
+      my_free(create_list);
+      create_list= next;
+    }
+
     mysql_mutex_lock(&LOCK_slave_background);
   } while (!stop);
 
@@ -392,6 +569,41 @@ slave_background_kill_request(THD *to_kill)
 
 
 /*
+  This function must only be called from a slave SQL thread (or worker thread),
+  to ensure that the table_entry will not go away before we can lock the
+  LOCK_slave_state.
+*/
+void
+slave_background_gtid_pos_create_request(
+        rpl_slave_state::gtid_pos_table *table_entry)
+{
+  slave_background_gtid_pos_create_t *p;
+
+  if (table_entry->state != rpl_slave_state::GTID_POS_AUTO_CREATE)
+    return;
+  p= (slave_background_gtid_pos_create_t *)my_malloc(sizeof(*p), MYF(MY_WME));
+  if (!p)
+    return;
+  mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+  if (table_entry->state != rpl_slave_state::GTID_POS_AUTO_CREATE)
+  {
+    my_free(p);
+    mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+    return;
+  }
+  table_entry->state= rpl_slave_state::GTID_POS_CREATE_REQUESTED;
+  mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+
+  p->hton= table_entry->table_hton;
+  mysql_mutex_lock(&LOCK_slave_background);
+  p->next= slave_background_gtid_pos_create_list;
+  slave_background_gtid_pos_create_list= p;
+  mysql_cond_signal(&COND_slave_background);
+  mysql_mutex_unlock(&LOCK_slave_background);
+}
+
+
+/*
   Start the slave background thread.
 
   This thread is currently used for two purposes:
diff --git a/sql/slave.h b/sql/slave.h
index ded9d76e49d..2afd5277e25 100644
--- a/sql/slave.h
+++ b/sql/slave.h
@@ -48,6 +48,7 @@
 #include "my_list.h"
 #include "rpl_filter.h"
 #include "rpl_tblmap.h"
+#include "rpl_gtid.h"
 
 #define SLAVE_NET_TIMEOUT  60
 
@@ -268,6 +269,8 @@ void slave_output_error_info(rpl_group_info *rgi, THD *thd);
 pthread_handler_t handle_slave_sql(void *arg);
 bool net_request_file(NET* net, const char* fname);
 void slave_background_kill_request(THD *to_kill);
+void slave_background_gtid_pos_create_request
+        (rpl_slave_state::gtid_pos_table *table_entry);
 
 extern bool volatile abort_loop;
 extern Master_info *active_mi; /* active_mi for multi-master */