MDEV-12179: Per-engine mysql.gtid_slave_pos table

Intermediate commit.

Implement a --gtid-pos-auto-engines system variable. The variable is a list
of engines for which mysql.gtid_slave_pos_ENGINE should be auto-created if
needed.

This commit only implements the option variable. It is not yet used to
actually auto-create any tables, nor is there a corresponding command-line
version of the option yet.

7 files changed, 329 insertions, 0 deletions

diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 6108ed6ffd4..3c3acdf6030 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -374,6 +374,8 @@ char *my_bind_addr_str;
 static char *default_collation_name;
 char *default_storage_engine, *default_tmp_storage_engine;
 char *enforced_storage_engine=NULL;
+char *gtid_pos_auto_engines;
+plugin_ref *opt_gtid_pos_auto_plugins;
 static char compiled_default_collation_name[]= MYSQL_DEFAULT_COLLATION_NAME;
 static I_List<CONNECT> thread_cache;
 static bool binlog_format_used= false;
@@ -4234,6 +4236,7 @@ static int init_common_variables()
   default_storage_engine= const_cast<char *>("MyISAM");
 #endif
   default_tmp_storage_engine= NULL;
+  gtid_pos_auto_engines= const_cast<char *>("");
 
   /*
     Add server status variables to the dynamic list of
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 613b57b133d..be8b97c80c4 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -19,6 +19,7 @@
 
 #include <my_global.h> /* MYSQL_PLUGIN_IMPORT, FN_REFLEN, FN_EXTLEN */
 #include "sql_basic_types.h"			/* query_id_t */
+#include "sql_plugin.h"
 #include "sql_bitmap.h"                         /* Bitmap */
 #include "my_decimal.h"                         /* my_decimal */
 #include "mysql_com.h"                     /* SERVER_VERSION_LENGTH */
@@ -153,6 +154,8 @@ extern char *default_tz_name;
 extern Time_zone *default_tz;
 extern char *default_storage_engine, *default_tmp_storage_engine;
 extern char *enforced_storage_engine;
+extern char *gtid_pos_auto_engines;
+extern plugin_ref *opt_gtid_pos_auto_plugins;
 extern bool opt_endinfo, using_udf_functions;
 extern my_bool locked_in_memory;
 extern bool opt_using_transactions;
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 07395e3e708..7b190f6244e 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -1278,3 +1278,177 @@ enum sys_var::where get_sys_var_value_origin(void *ptr)
   return sys_var::CONFIG;
 }
 
+
+/*
+  Find the next item in string of comma-separated items.
+  END_POS points at the end of the string.
+  ITEM_START and ITEM_END return the limits of the next item.
+  Returns true while items are available, false at the end.
+*/
+static bool
+engine_list_next_item(const char **pos, const char *end_pos,
+                      const char **item_start, const char **item_end)
+{
+  if (*pos >= end_pos)
+    return false;
+  *item_start= *pos;
+  while (*pos < end_pos && **pos != ',')
+    ++*pos;
+  *item_end= *pos;
+  ++*pos;
+  return true;
+}
+
+
+static bool
+resolve_engine_list_item(plugin_ref *list, uint32 *idx,
+                         const char *pos, const char *pos_end)
+{
+  LEX_STRING item_str;
+  plugin_ref ref;
+  uint32_t i;
+
+  item_str.str= const_cast<char*>(pos);
+  item_str.length= pos_end-pos;
+  ref= ha_resolve_by_name(NULL, &item_str, false);
+  if (!ref)
+  {
+    ErrConvString err(pos, pos_end-pos, system_charset_info);
+    my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), err.ptr());
+    return true;
+  }
+  /* Ignore duplicates, like --plugin-load does. */
+  for (i= 0; i < *idx; ++i)
+  {
+    if (plugin_hton(list[i]) == plugin_hton(ref))
+    {
+      plugin_unlock(NULL, ref);
+      return false;
+    }
+  }
+  list[*idx]= ref;
+  ++*idx;
+  return false;
+}
+
+
+/*
+  Helper for class Sys_var_pluginlist.
+  Resolve a comma-separated list of storage engine names to a null-terminated
+  array of plugin_ref.
+*/
+plugin_ref *
+resolve_engine_list(const char *str_arg, size_t str_arg_len)
+{
+  uint32 count, idx;
+  const char *pos, *item_start, *item_end;
+  const char *str_arg_end= str_arg + str_arg_len;
+  plugin_ref *res;
+
+  count= 0;
+  pos= str_arg;
+  for (;;)
+  {
+    if (!engine_list_next_item(&pos, str_arg_end, &item_start, &item_end))
+      break;
+    ++count;
+  }
+
+  res= (plugin_ref *)my_malloc((count+1)*sizeof(*res), MYF(MY_ZEROFILL|MY_WME));
+  if (!res)
+  {
+    my_error(ER_OUTOFMEMORY, MYF(0), (int)((count+1)*sizeof(*res)));
+    goto err;
+  }
+
+  idx= 0;
+  pos= str_arg;
+  for (;;)
+  {
+    if (!engine_list_next_item(&pos, str_arg_end, &item_start, &item_end))
+      break;
+    DBUG_ASSERT(idx < count);
+    if (idx >= count)
+      break;
+    if (resolve_engine_list_item(res, &idx, item_start, item_end))
+      goto err;
+  }
+
+  return res;
+
+err:
+  free_engine_list(res);
+  return NULL;
+}
+
+
+void
+free_engine_list(plugin_ref *list)
+{
+  plugin_ref *p;
+
+  if (!list)
+    return;
+  for (p= list; *p; ++p)
+    plugin_unlock(NULL, *p);
+  my_free(list);
+}
+
+
+plugin_ref *
+copy_engine_list(plugin_ref *list)
+{
+  plugin_ref *p;
+  uint32 count, i;
+
+  for (p= list, count= 0; *p; ++p, ++count)
+    ;
+  p= (plugin_ref *)my_malloc((count+1)*sizeof(*p), MYF(0));
+  if (!p)
+  {
+    my_error(ER_OUTOFMEMORY, MYF(0), (int)((count+1)*sizeof(*p)));
+    return NULL;
+  }
+  for (i= 0; i < count; ++i)
+    p[i]= my_plugin_lock(NULL, list[i]);
+  return p;
+}
+
+
+char *
+pretty_print_engine_list(THD *thd, plugin_ref *list)
+{
+  plugin_ref *p;
+  size_t size;
+  char *buf, *pos;
+
+  if (!list)
+    return thd->strmake("", 0);
+
+  size= 0;
+  for (p= list; *p; ++p)
+    size+= plugin_name(*p)->length + 1;
+  buf= static_cast<char *>(thd->alloc(size));
+  if (!buf)
+    return NULL;
+  pos= buf;
+  for (p= list; *p; ++p)
+  {
+    LEX_STRING *name;
+    size_t remain;
+
+    remain= buf + size - pos;
+    DBUG_ASSERT(remain > 0);
+    if (remain <= 1)
+      break;
+    if (pos != buf)
+    {
+      pos= strmake(pos, ",", remain-1);
+      --remain;
+    }
+    name= plugin_name(*p);
+    pos= strmake(pos, name->str, MY_MIN(name->length, remain-1));
+  }
+  *pos= '\0';
+  return buf;
+}
diff --git a/sql/set_var.h b/sql/set_var.h
index 97dc3b5ba51..a6f0cb02fe8 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -286,6 +286,7 @@ public:
     longlong longlong_value;            ///< for signed integer
     double double_value;                ///< for Sys_var_double
     plugin_ref plugin;                  ///< for Sys_var_plugin
+    plugin_ref *plugins;                ///< for Sys_var_pluginlist
     Time_zone *time_zone;               ///< for Sys_var_tz
     LEX_STRING string_value;            ///< for Sys_var_charptr and others
     const void *ptr;                    ///< for Sys_var_struct
@@ -422,6 +423,10 @@ int sys_var_init();
 uint sys_var_elements();
 int sys_var_add_options(DYNAMIC_ARRAY *long_options, int parse_flags);
 void sys_var_end(void);
+plugin_ref *resolve_engine_list(const char *str_arg, size_t str_arg_len);
+void free_engine_list(plugin_ref *list);
+plugin_ref *copy_engine_list(plugin_ref *list);
+char *pretty_print_engine_list(THD *thd, plugin_ref *list);
 
 #endif
 
diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc
index 101ea3fd3c7..45d78c525fe 100644
--- a/sql/sql_plugin.cc
+++ b/sql/sql_plugin.cc
@@ -1922,6 +1922,12 @@ void plugin_shutdown(void)
 
   if (initialized)
   {
+    if (opt_gtid_pos_auto_plugins)
+    {
+      free_engine_list(opt_gtid_pos_auto_plugins);
+      opt_gtid_pos_auto_plugins= NULL;
+    }
+
     mysql_mutex_lock(&LOCK_plugin);
 
     reap_needed= true;
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index a3bc3f6d9cf..8f0a081acc7 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -3521,6 +3521,46 @@ static Sys_var_plugin Sys_enforce_storage_engine(
        NO_CMD_LINE, MYSQL_STORAGE_ENGINE_PLUGIN,
        DEFAULT(&enforced_storage_engine), NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(check_has_super));
 
+/*
+  Check
+   1. Value for gtid_pos_auto_engines is not NULL.
+   2. No slave SQL thread is running.
+*/
+static bool
+check_gtid_pos_auto_engines(sys_var *self, THD *thd, set_var *var)
+{
+  bool running;
+  bool err= false;
+
+  DBUG_ASSERT(var->type == OPT_GLOBAL);
+  if (var->value && var->value->is_null())
+    err= true;
+  else
+  {
+    running= give_error_if_slave_running(false);
+    if (running)
+      err= true;
+  }
+  if (err && var->save_result.plugins)
+  {
+    free_engine_list(var->save_result.plugins);
+    var->save_result.plugins= NULL;
+  }
+  return err;
+}
+
+
+static Sys_var_pluginlist Sys_gtid_pos_auto_engines(
+       "gtid_pos_auto_engines",
+       "List of engines for which to automatically create a "
+       "mysql.gtid_slave_pos_ENGINE table, if a transaction using that engine "
+       "is replicated. This can be used to avoid introducing cross-engine "
+       "transactions, if engines are used different from that used by table "
+       "mysql.gtid_slave_pos",
+       GLOBAL_VAR(opt_gtid_pos_auto_plugins), NO_CMD_LINE,
+       DEFAULT(&gtid_pos_auto_engines),
+       NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(check_gtid_pos_auto_engines));
+
 #if defined(ENABLED_DEBUG_SYNC)
 /*
   Variable can be set for the session only.
diff --git a/sql/sys_vars.ic b/sql/sys_vars.ic
index 780450b484b..e76f60b56b0 100644
--- a/sql/sys_vars.ic
+++ b/sql/sys_vars.ic
@@ -1533,6 +1533,104 @@ public:
   { return valptr(thd, get_default(thd)); }
 };
 
+/**
+  Class for variables that containg a list of plugins.
+  Currently this is used only for @@gtid_pos_auto_create_engines
+
+  Backing store: plugin_ref
+
+  @note
+  Currently this is only used for storage engine type plugins, and thus only
+  storage engine type plugin is implemented. It could be extended to other
+  plugin types later if needed, similar to Sys_var_plugin.
+
+  These variables don't support command-line equivalents, any such
+  command-line options should be added manually to my_long_options in mysqld.cc
+*/
+class Sys_var_pluginlist: public sys_var
+{
+  int plugin_type;
+public:
+  Sys_var_pluginlist(const char *name_arg,
+          const char *comment, int flag_args, ptrdiff_t off, size_t size,
+          CMD_LINE getopt,
+          char **def_val, PolyLock *lock=0,
+          enum binlog_status_enum binlog_status_arg=VARIABLE_NOT_IN_BINLOG,
+          on_check_function on_check_func=0,
+          on_update_function on_update_func=0,
+          const char *substitute=0)
+    : sys_var(&all_sys_vars, name_arg, comment, flag_args, off, getopt.id,
+              getopt.arg_type, SHOW_CHAR, (intptr)def_val,
+              lock, binlog_status_arg, on_check_func, on_update_func,
+              substitute)
+  {
+    option.var_type|= GET_STR;
+    SYSVAR_ASSERT(size == sizeof(plugin_ref));
+    SYSVAR_ASSERT(getopt.id < 0); // force NO_CMD_LINE
+  }
+  bool do_check(THD *thd, set_var *var)
+  {
+    char buff[STRING_BUFFER_USUAL_SIZE];
+    String str(buff,sizeof(buff), system_charset_info), *res;
+    plugin_ref *plugins;
+
+    if (!(res=var->value->val_str(&str)))
+      plugins= resolve_engine_list("", 0);
+    else
+      plugins= resolve_engine_list(res->ptr(), res->length());
+    if (!plugins)
+      return true;
+    var->save_result.plugins= plugins;
+    return false;
+  }
+  void do_update(plugin_ref **valptr, plugin_ref* newval)
+  {
+    plugin_ref *oldval= *valptr;
+    *valptr= newval;
+    free_engine_list(oldval);
+  }
+  bool session_update(THD *thd, set_var *var)
+  {
+    do_update((plugin_ref**)session_var_ptr(thd),
+              var->save_result.plugins);
+    return false;
+  }
+  bool global_update(THD *thd, set_var *var)
+  {
+    do_update((plugin_ref**)global_var_ptr(),
+              var->save_result.plugins);
+    return false;
+  }
+  void session_save_default(THD *thd, set_var *var)
+  {
+    plugin_ref* plugins= global_var(plugin_ref *);
+    var->save_result.plugins= plugins ? copy_engine_list(plugins) : 0;
+  }
+  plugin_ref *get_default(THD *thd)
+  {
+    char *default_value= *reinterpret_cast<char**>(option.def_value);
+    if (!default_value)
+      return 0;
+    return resolve_engine_list(default_value, strlen(default_value));
+  }
+
+  void global_save_default(THD *thd, set_var *var)
+  {
+    var->save_result.plugins= get_default(thd);
+  }
+
+  uchar *valptr(THD *thd, plugin_ref *plugins)
+  {
+    return (uchar*)pretty_print_engine_list(thd, plugins);
+  }
+  uchar *session_value_ptr(THD *thd, const LEX_STRING *base)
+  { return valptr(thd, session_var(thd, plugin_ref*)); }
+  uchar *global_value_ptr(THD *thd, const LEX_STRING *base)
+  { return valptr(thd, global_var(plugin_ref*)); }
+  uchar *default_value_ptr(THD *thd)
+  { return valptr(thd, get_default(thd)); }
+};
+
 #if defined(ENABLED_DEBUG_SYNC)
 
 #include "debug_sync.h"