Adding upstream version 1:10.11.6.upstream/1%10.11.6upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 3913 insertions, 0 deletions

diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
new file mode 100644
index 00000000..0a615228
--- /dev/null
+++ b/sql/wsrep_mysqld.cc
@@ -0,0 +1,3913 @@
+/* Copyright (c) 2008, 2023 Codership Oy <http://www.codership.com>
+   Copyright (c) 2020, 2022, MariaDB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.x1
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA */
+
+#include "sql_plugin.h"                         /* wsrep_plugins_pre_init() */
+#include "my_global.h"
+#include "wsrep_server_state.h"
+#include "wsrep_status.h"
+
+#include "mariadb.h"
+#include <mysqld.h>
+#include <transaction.h>
+#include <sql_class.h>
+#include <sql_parse.h>
+#include <sql_base.h> /* find_temporary_table() */
+#include <sql_statistics.h> /* is_stat_table() */
+#include "slave.h"
+#include "rpl_mi.h"
+#include "sql_repl.h"
+#include "rpl_filter.h"
+#include "sql_callback.h"
+#include "sp_head.h"
+#include "sql_show.h"
+#include "sp.h"
+#include "handler.h"
+#include "wsrep_priv.h"
+#include "wsrep_thd.h"
+#include "wsrep_sst.h"
+#include "wsrep_utils.h"
+#include "wsrep_var.h"
+#include "wsrep_binlog.h"
+#include "wsrep_applier.h"
+#include "wsrep_schema.h"
+#include "wsrep_xid.h"
+#include "wsrep_trans_observer.h"
+#include "mysql/service_wsrep.h"
+#include <cstdio>
+#include <cstdlib>
+#include <string>
+#include "log_event.h"
+#include "sql_connect.h"
+#include "thread_cache.h"
+#include "debug_sync.h"
+
+#include <sstream>
+
+/* wsrep-lib */
+Wsrep_server_state* Wsrep_server_state::m_instance;
+
+my_bool wsrep_emulate_bin_log= FALSE; // activating parts of binlog interface
+my_bool wsrep_preordered_opt= FALSE;
+
+/* Streaming Replication */
+const char *wsrep_fragment_units[]= { "bytes", "rows", "statements", NullS };
+const char *wsrep_SR_store_types[]= { "none", "table", NullS };
+
+/*
+ * Begin configuration options
+ */
+
+extern my_bool plugins_are_initialized;
+
+/* System variables. */
+const char *wsrep_provider;
+const char *wsrep_provider_options;
+const char *wsrep_cluster_address;
+const char *wsrep_cluster_name;
+const char *wsrep_node_name;
+const char *wsrep_node_address;
+const char *wsrep_node_incoming_address;
+const char *wsrep_start_position;
+const char *wsrep_data_home_dir;
+const char *wsrep_dbug_option;
+const char *wsrep_notify_cmd;
+const char *wsrep_status_file;
+const char *wsrep_allowlist;
+
+ulong   wsrep_debug;                            // Debug level logging
+my_bool wsrep_convert_LOCK_to_trx;              // Convert locking sessions to trx
+my_bool wsrep_auto_increment_control;           // Control auto increment variables
+my_bool wsrep_drupal_282555_workaround;         // Retry autoinc insert after dupkey
+my_bool wsrep_certify_nonPK;                    // Certify, even when no primary key
+ulong   wsrep_certification_rules      = WSREP_CERTIFICATION_RULES_STRICT;
+my_bool wsrep_recovery;                         // Recovery
+my_bool wsrep_log_conflicts;
+my_bool wsrep_load_data_splitting= 0;           // Commit load data every 10K intervals
+my_bool wsrep_slave_UK_checks;                  // Slave thread does UK checks
+my_bool wsrep_slave_FK_checks;                  // Slave thread does FK checks
+my_bool wsrep_restart_slave;                    // Should mysql slave thread be
+                                                // restarted, when node joins back?
+my_bool wsrep_desync;                           // De(re)synchronize the node from the
+                                                // cluster
+ulonglong wsrep_mode;
+bool wsrep_service_started;                     // If Galera was initialized
+long wsrep_slave_threads;                       // No. of slave appliers threads
+ulong wsrep_retry_autocommit;                   // Retry aborted autocommit trx
+ulong wsrep_max_ws_size;                        // Max allowed ws (RBR buffer) size
+ulong wsrep_max_ws_rows;                        // Max number of rows in ws
+ulong wsrep_forced_binlog_format= BINLOG_FORMAT_UNSPEC;
+ulong wsrep_mysql_replication_bundle;
+
+bool              wsrep_gtid_mode;              // Enable WSREP native GTID support
+Wsrep_gtid_server wsrep_gtid_server;
+uint  wsrep_gtid_domain_id=0;                   // Domain id on above structure
+
+/* Other configuration variables and their default values. */
+my_bool wsrep_incremental_data_collection= 0;   // Incremental data collection
+my_bool wsrep_restart_slave_activated= 0;       // Node has dropped, and slave
+                                                // restart will be needed
+bool wsrep_new_cluster= false;                  // Bootstrap the cluster?
+int wsrep_slave_count_change= 0;                // No. of appliers to stop/start
+int wsrep_to_isolation= 0;                      // No. of active TO isolation threads
+long wsrep_max_protocol_version= 4;             // Maximum protocol version to use
+long int  wsrep_protocol_version= wsrep_max_protocol_version;
+ulong wsrep_trx_fragment_unit= WSREP_FRAG_BYTES;
+                                                // unit for fragment size
+ulong wsrep_SR_store_type= WSREP_SR_STORE_TABLE;
+uint  wsrep_ignore_apply_errors= 0;
+
+std::atomic <bool> wsrep_thread_create_failed;
+
+/*
+ * End configuration options
+ */
+
+/*
+ * Cached variables
+ */
+
+// Whether the Galera write-set replication provider is set
+// wsrep_provider && strcmp(wsrep_provider, WSREP_NONE)
+bool WSREP_PROVIDER_EXISTS_;
+
+// Whether the Galera write-set replication is enabled
+// global_system_variables.wsrep_on && WSREP_PROVIDER_EXISTS_
+bool WSREP_ON_;
+
+/*
+ * Other wsrep global variables.
+ */
+
+mysql_mutex_t LOCK_wsrep_ready;
+mysql_cond_t  COND_wsrep_ready;
+mysql_mutex_t LOCK_wsrep_sst;
+mysql_cond_t  COND_wsrep_sst;
+mysql_mutex_t LOCK_wsrep_sst_init;
+mysql_cond_t  COND_wsrep_sst_init;
+mysql_mutex_t LOCK_wsrep_replaying;
+mysql_cond_t  COND_wsrep_replaying;
+mysql_mutex_t LOCK_wsrep_slave_threads;
+mysql_cond_t  COND_wsrep_slave_threads;
+mysql_mutex_t LOCK_wsrep_gtid_wait_upto;
+mysql_mutex_t LOCK_wsrep_cluster_config;
+mysql_mutex_t LOCK_wsrep_desync;
+mysql_mutex_t LOCK_wsrep_config_state;
+mysql_mutex_t LOCK_wsrep_group_commit;
+mysql_mutex_t LOCK_wsrep_SR_pool;
+mysql_mutex_t LOCK_wsrep_SR_store;
+mysql_mutex_t LOCK_wsrep_joiner_monitor;
+mysql_mutex_t LOCK_wsrep_donor_monitor;
+mysql_cond_t  COND_wsrep_joiner_monitor;
+mysql_cond_t  COND_wsrep_donor_monitor;
+
+int wsrep_replaying= 0;
+ulong  wsrep_running_threads = 0; // # of currently running wsrep
+				  // # threads
+ulong  wsrep_running_applier_threads = 0; // # of running applier threads
+ulong  wsrep_running_rollbacker_threads = 0; // # of running
+					     // # rollbacker threads
+ulong  my_bind_addr;
+
+#ifdef HAVE_PSI_INTERFACE
+PSI_mutex_key
+  key_LOCK_wsrep_replaying, key_LOCK_wsrep_ready, key_LOCK_wsrep_sst,
+  key_LOCK_wsrep_sst_thread, key_LOCK_wsrep_sst_init,
+  key_LOCK_wsrep_slave_threads, key_LOCK_wsrep_gtid_wait_upto,
+  key_LOCK_wsrep_desync,
+  key_LOCK_wsrep_config_state, key_LOCK_wsrep_cluster_config,
+  key_LOCK_wsrep_group_commit,
+  key_LOCK_wsrep_SR_pool,
+  key_LOCK_wsrep_SR_store,
+  key_LOCK_wsrep_thd_queue,
+  key_LOCK_wsrep_joiner_monitor,
+  key_LOCK_wsrep_donor_monitor;
+
+PSI_cond_key key_COND_wsrep_thd,
+  key_COND_wsrep_replaying, key_COND_wsrep_ready, key_COND_wsrep_sst,
+  key_COND_wsrep_sst_init, key_COND_wsrep_sst_thread,
+  key_COND_wsrep_thd_queue, key_COND_wsrep_slave_threads, key_COND_wsrep_gtid_wait_upto,
+  key_COND_wsrep_joiner_monitor, key_COND_wsrep_donor_monitor;
+
+PSI_file_key key_file_wsrep_gra_log;
+
+static PSI_mutex_info wsrep_mutexes[]=
+{
+  { &key_LOCK_wsrep_ready, "LOCK_wsrep_ready", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_sst, "LOCK_wsrep_sst", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_sst_thread, "wsrep_sst_thread", 0},
+  { &key_LOCK_wsrep_sst_init, "LOCK_wsrep_sst_init", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_sst, "LOCK_wsrep_sst", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_replaying, "LOCK_wsrep_replaying", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_slave_threads, "LOCK_wsrep_slave_threads", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_gtid_wait_upto, "LOCK_wsrep_gtid_wait_upto", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_cluster_config, "LOCK_wsrep_cluster_config", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_desync, "LOCK_wsrep_desync", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_config_state, "LOCK_wsrep_config_state", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_group_commit, "LOCK_wsrep_group_commit", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_SR_pool, "LOCK_wsrep_SR_pool", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_SR_store, "LOCK_wsrep_SR_store", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_joiner_monitor, "LOCK_wsrep_joiner_monitor", PSI_FLAG_GLOBAL},
+  { &key_LOCK_wsrep_donor_monitor, "LOCK_wsrep_donor_monitor", PSI_FLAG_GLOBAL}
+};
+
+static PSI_cond_info wsrep_conds[]=
+{
+  { &key_COND_wsrep_ready, "COND_wsrep_ready", PSI_FLAG_GLOBAL},
+  { &key_COND_wsrep_sst, "COND_wsrep_sst", PSI_FLAG_GLOBAL},
+  { &key_COND_wsrep_sst_init, "COND_wsrep_sst_init", PSI_FLAG_GLOBAL},
+  { &key_COND_wsrep_sst_thread, "wsrep_sst_thread", 0},
+  { &key_COND_wsrep_thd, "THD::COND_wsrep_thd", 0},
+  { &key_COND_wsrep_replaying, "COND_wsrep_replaying", PSI_FLAG_GLOBAL},
+  { &key_COND_wsrep_slave_threads, "COND_wsrep_wsrep_slave_threads", PSI_FLAG_GLOBAL},
+  { &key_COND_wsrep_gtid_wait_upto, "COND_wsrep_gtid_wait_upto", PSI_FLAG_GLOBAL},
+  { &key_COND_wsrep_joiner_monitor, "COND_wsrep_joiner_monitor", PSI_FLAG_GLOBAL},
+  { &key_COND_wsrep_donor_monitor, "COND_wsrep_donor_monitor", PSI_FLAG_GLOBAL}
+};
+
+static PSI_file_info wsrep_files[]=
+{
+  { &key_file_wsrep_gra_log, "wsrep_gra_log", 0}
+};
+
+PSI_thread_key key_wsrep_sst_joiner, key_wsrep_sst_donor,
+  key_wsrep_rollbacker, key_wsrep_applier,
+  key_wsrep_sst_joiner_monitor, key_wsrep_sst_donor_monitor;
+
+static PSI_thread_info wsrep_threads[]=
+{
+ {&key_wsrep_sst_joiner, "wsrep_sst_joiner_thread", PSI_FLAG_GLOBAL},
+ {&key_wsrep_sst_donor, "wsrep_sst_donor_thread", PSI_FLAG_GLOBAL},
+ {&key_wsrep_rollbacker, "wsrep_rollbacker_thread", PSI_FLAG_GLOBAL},
+ {&key_wsrep_applier, "wsrep_applier_thread", PSI_FLAG_GLOBAL},
+ {&key_wsrep_sst_joiner_monitor, "wsrep_sst_joiner_monitor", PSI_FLAG_GLOBAL},
+ {&key_wsrep_sst_donor_monitor, "wsrep_sst_donor_monitor", PSI_FLAG_GLOBAL}
+};
+
+#endif /* HAVE_PSI_INTERFACE */
+
+my_bool wsrep_inited= 0; // initialized ?
+
+static wsrep_uuid_t node_uuid= WSREP_UUID_UNDEFINED;
+static char         cluster_uuid_str[40]= { 0, };
+
+static char provider_name[256]= { 0, };
+static char provider_version[256]= { 0, };
+static char provider_vendor[256]= { 0, };
+
+/*
+ * Wsrep status variables. LOCK_status must be locked When modifying
+ * these variables,
+ */
+my_bool     wsrep_connected         = FALSE;
+my_bool     wsrep_ready             = FALSE;
+const char* wsrep_cluster_state_uuid= cluster_uuid_str;
+long long   wsrep_cluster_conf_id   = WSREP_SEQNO_UNDEFINED;
+const char* wsrep_cluster_status    = "Disconnected";
+long        wsrep_cluster_size      = 0;
+long        wsrep_local_index       = -1;
+long long   wsrep_local_bf_aborts   = 0;
+const char* wsrep_provider_name     = provider_name;
+const char* wsrep_provider_version  = provider_version;
+const char* wsrep_provider_vendor   = provider_vendor;
+char* wsrep_provider_capabilities   = NULL;
+char* wsrep_cluster_capabilities    = NULL;
+/* End wsrep status variables */
+
+wsp::Config_state *wsrep_config_state;
+
+void WSREP_LOG(void (*fun)(const char* fmt, ...), const char* fmt, ...)
+{
+  /* Allocate short buffer from stack. If the vsnprintf() return value
+     indicates that the message was truncated, a new buffer will be allocated
+     dynamically and the message will be reprinted. */
+  char msg[128] = {'\0'};
+  va_list arglist;
+  va_start(arglist, fmt);
+  int n= vsnprintf(msg, sizeof(msg), fmt, arglist);
+  va_end(arglist);
+  if (n < 0)
+  {
+    sql_print_warning("WSREP: Printing message failed");
+  }
+  else if (n < (int)sizeof(msg))
+  {
+    fun("WSREP: %s", msg);
+  }
+  else
+  {
+    size_t dynbuf_size= std::max(n, 4096);
+    char* dynbuf= (char*) my_malloc(PSI_NOT_INSTRUMENTED, dynbuf_size, MYF(0));
+    if (dynbuf)
+    {
+      va_start(arglist, fmt);
+      (void)vsnprintf(&dynbuf[0], dynbuf_size - 1, fmt, arglist);
+      va_end(arglist);
+      dynbuf[dynbuf_size - 1] = '\0';
+      fun("WSREP: %s", &dynbuf[0]);
+      my_free(dynbuf);
+    }
+    else
+    {
+      /* Memory allocation for vector failed, print truncated message. */
+      fun("WSREP: %s", msg);
+    }
+  }
+}
+
+
+wsrep_uuid_t               local_uuid       = WSREP_UUID_UNDEFINED;
+wsrep_seqno_t              local_seqno      = WSREP_SEQNO_UNDEFINED;
+
+/*
+ */
+Wsrep_schema *wsrep_schema= 0;
+
+static void wsrep_log_cb(wsrep::log::level level,
+                         const char*, const char *msg)
+{
+  /*
+    Silence all wsrep related logging from lib and provider if
+    wsrep is not enabled.
+  */
+  if (!WSREP_ON) return;
+
+  switch (level) {
+  case wsrep::log::info:
+    WSREP_INFO("%s", msg);
+    break;
+  case wsrep::log::warning:
+    WSREP_WARN("%s", msg);
+    break;
+  case wsrep::log::error:
+    WSREP_ERROR("%s", msg);
+    break;
+  case wsrep::log::debug:
+    WSREP_DEBUG("%s", msg);
+    break;
+  case wsrep::log::unknown:
+    WSREP_UNKNOWN("%s", msg);
+    break;
+  }
+}
+
+void wsrep_init_gtid()
+{
+  wsrep_server_gtid_t stored_gtid= wsrep_get_SE_checkpoint<wsrep_server_gtid_t>();
+  // Domain id may have changed, use the one
+  // received during state transfer.
+  stored_gtid.domain_id= wsrep_gtid_server.domain_id;
+  if (stored_gtid.server_id == 0)
+  {
+    rpl_gtid wsrep_last_gtid;
+    if (mysql_bin_log.is_open() &&
+        mysql_bin_log.lookup_domain_in_binlog_state(stored_gtid.domain_id,
+                                                    &wsrep_last_gtid))
+    {
+      stored_gtid.server_id= wsrep_last_gtid.server_id;
+      stored_gtid.seqno= wsrep_last_gtid.seq_no;
+    }
+    else
+    {
+      stored_gtid.server_id= global_system_variables.server_id;
+      stored_gtid.seqno= 0;
+    }
+  }
+  wsrep_gtid_server.gtid(stored_gtid);
+}
+
+bool wsrep_get_binlog_gtid_seqno(wsrep_server_gtid_t& gtid)
+{
+  rpl_gtid binlog_gtid;
+  int ret= 0;
+  if (mysql_bin_log.is_open() &&
+      mysql_bin_log.find_in_binlog_state(gtid.domain_id,
+                                         gtid.server_id,
+                                         &binlog_gtid))
+  {
+    gtid.domain_id= binlog_gtid.domain_id;
+    gtid.server_id= binlog_gtid.server_id;
+    gtid.seqno=     binlog_gtid.seq_no;
+    ret= 1;
+  }
+  return ret;
+}
+
+bool wsrep_check_gtid_seqno(const uint32& domain, const uint32& server,
+                            uint64& seqno)
+{
+  if (domain == wsrep_gtid_server.domain_id && 
+      server == wsrep_gtid_server.server_id)
+  {
+    if (wsrep_gtid_server.seqno_committed() < seqno) return 1;
+    return 0;
+  }
+  return 0;
+}
+
+void wsrep_init_sidno(const wsrep::id& uuid)
+{
+  /*
+    Protocol versions starting from 4 use group gtid as it is.
+    For lesser protocol versions generate new Sid map entry from inverted
+    uuid.
+  */
+  rpl_gtid sid;
+  if (wsrep_protocol_version >= 4)
+  {
+    memcpy((void*)&sid, (const uchar*)uuid.data(),16);
+  }
+  else
+  {
+    wsrep_uuid_t ltid_uuid;
+    for (size_t i= 0; i < sizeof(ltid_uuid.data); ++i)
+    {
+      ltid_uuid.data[i]= ~((const uchar*)uuid.data())[i];
+    }
+    memcpy((void*)&sid, (const uchar*)ltid_uuid.data,16);
+  }
+#ifdef GTID_SUPPORT
+  global_sid_lock->wrlock();
+  wsrep_sidno= global_sid_map->add_sid(sid);
+  WSREP_INFO("Initialized wsrep sidno %d", wsrep_sidno);
+  global_sid_lock->unlock();
+#endif
+}
+
+void wsrep_init_schema()
+{
+  DBUG_ASSERT(!wsrep_schema);
+
+  WSREP_INFO("wsrep_init_schema_and_SR %p", wsrep_schema);
+  if (!wsrep_schema)
+  {
+    wsrep_schema= new Wsrep_schema();
+    if (wsrep_schema->init())
+    {
+      WSREP_ERROR("Failed to init wsrep schema");
+      unireg_abort(1);
+    }
+    // If we are bootstraping new cluster we should 
+    // clear allowlist table and populate it from variable
+    if (wsrep_new_cluster) 
+    {
+      wsrep_schema->clear_allowlist();
+      std::vector<std::string> ip_allowlist;
+      if (wsrep_split_allowlist(ip_allowlist))
+      {
+        wsrep_schema->store_allowlist(ip_allowlist);
+      }
+    }
+  }
+}
+
+void wsrep_deinit_schema()
+{
+  delete wsrep_schema;
+  wsrep_schema= 0;
+}
+
+void wsrep_recover_sr_from_storage(THD *orig_thd)
+{
+  switch (wsrep_SR_store_type)
+  {
+  case WSREP_SR_STORE_TABLE:
+    if (!wsrep_schema)
+    {
+      WSREP_ERROR("Wsrep schema not initialized when trying to recover "
+                  "streaming transactions: wsrep_on %d", WSREP_ON);
+      trans_commit(orig_thd);
+    }
+    if (wsrep_schema->recover_sr_transactions(orig_thd))
+    {
+      WSREP_ERROR("Failed to recover SR transactions from schema: wsrep_on : %d", WSREP_ON);
+      trans_commit(orig_thd);
+    }
+    break;
+  default:
+    /* */
+    WSREP_ERROR("Unsupported wsrep SR store type: %lu wsrep_on: %d",
+                wsrep_SR_store_type, WSREP_ON);
+    trans_commit(orig_thd);
+    break;
+  }
+}
+
+/** Export the WSREP provider's capabilities as a human readable string.
+ * The result is saved in a dynamically allocated string of the form:
+ * :cap1:cap2:cap3:
+ */
+static void wsrep_capabilities_export(wsrep_cap_t const cap, char** str)
+{
+  static const char* names[] =
+  {
+    /* Keep in sync with wsrep/wsrep_api.h WSREP_CAP_* macros. */
+    "MULTI_MASTER",
+    "CERTIFICATION",
+    "PARALLEL_APPLYING",
+    "TRX_REPLAY",
+    "ISOLATION",
+    "PAUSE",
+    "CAUSAL_READS",
+    "CAUSAL_TRX",
+    "INCREMENTAL_WRITESET",
+    "SESSION_LOCKS",
+    "DISTRIBUTED_LOCKS",
+    "CONSISTENCY_CHECK",
+    "UNORDERED",
+    "ANNOTATION",
+    "PREORDERED",
+    "STREAMING",
+    "SNAPSHOT",
+    "NBO",
+  };
+
+  std::string s;
+  for (size_t i= 0; i < sizeof(names) / sizeof(names[0]); ++i)
+  {
+    if (cap & (1ULL << i))
+    {
+      if (s.empty())
+      {
+        s= ":";
+      }
+      s += names[i];
+      s += ":";
+    }
+  }
+
+  /* A read from the string pointed to by *str may be started at any time,
+   * so it must never point to free(3)d memory or non '\0' terminated string. */
+
+  char* const previous= *str;
+
+  *str= strdup(s.c_str());
+
+  if (previous != NULL)
+  {
+    free(previous);
+  }
+}
+
+/* Verifies that SE position is consistent with the group position
+ * and initializes other variables */
+void wsrep_verify_SE_checkpoint(const wsrep_uuid_t& uuid,
+                                wsrep_seqno_t const seqno)
+{
+}
+
+/*
+  Wsrep is considered ready if
+  1) Provider is not loaded (native mode)
+  2) Server has reached synced state
+  3) Server is in joiner mode and mysqldump SST method has been
+     specified
+  See Wsrep_server_service::log_state_change() for further details.
+ */
+my_bool wsrep_ready_get (void)
+{
+  if (mysql_mutex_lock (&LOCK_wsrep_ready)) abort();
+  my_bool ret= wsrep_ready;
+  mysql_mutex_unlock (&LOCK_wsrep_ready);
+  return ret;
+}
+
+int wsrep_show_ready(THD *thd, SHOW_VAR *var, char *buff)
+{
+  var->type= SHOW_MY_BOOL;
+  var->value= buff;
+  *((my_bool *)buff)= wsrep_ready_get();
+  return 0;
+}
+
+void wsrep_update_cluster_state_uuid(const char* uuid)
+{
+  strncpy(cluster_uuid_str, uuid, sizeof(cluster_uuid_str) - 1);
+}
+
+static void wsrep_init_position()
+{
+}
+
+/****************************************************************************
+                         Helpers for wsrep_init()
+ ****************************************************************************/
+static std::string wsrep_server_name()
+{
+  std::string ret(wsrep_node_name ? wsrep_node_name : "");
+  return ret;
+}
+
+static std::string wsrep_server_id()
+{
+  /* using empty server_id, which enables view change handler to
+     set final server_id later on
+  */
+  std::string ret("");
+  return ret;
+}
+
+static std::string wsrep_server_node_address()
+{
+
+  std::string ret;
+  if (!wsrep_data_home_dir || strlen(wsrep_data_home_dir) == 0)
+    wsrep_data_home_dir= mysql_real_data_home;
+
+  /* Initialize node address */
+  if (!wsrep_node_address || !strcmp(wsrep_node_address, ""))
+  {
+    char node_addr[512]= {0, };
+    const size_t node_addr_max= sizeof(node_addr) - 1;
+    size_t guess_ip_ret= wsrep_guess_ip(node_addr, node_addr_max);
+    if (!(guess_ip_ret > 0 && guess_ip_ret < node_addr_max))
+    {
+      WSREP_WARN("Failed to guess base node address. Set it explicitly via "
+                 "wsrep_node_address.");
+    }
+    else
+    {
+      ret= node_addr;
+    }
+  }
+  else
+  {
+    ret= wsrep_node_address;
+  }
+  return ret;
+}
+
+static std::string wsrep_server_incoming_address()
+{
+  std::string ret;
+  const std::string node_addr(wsrep_server_node_address());
+  char inc_addr[512]= { 0, };
+  size_t const inc_addr_max= sizeof (inc_addr);
+
+  /*
+    In case wsrep_node_incoming_address is either not set or set to AUTO,
+    we need to use mysqld's my_bind_addr_str:mysqld_port, lastly fallback
+    to wsrep_node_address' value if mysqld's bind-address is not set either.
+  */
+  if ((!wsrep_node_incoming_address ||
+       !strcmp (wsrep_node_incoming_address, WSREP_NODE_INCOMING_AUTO)))
+  {
+    bool is_ipv6= false;
+    unsigned int my_bind_ip= INADDR_ANY; // default if not set
+
+    if (my_bind_addr_str && strlen(my_bind_addr_str) &&
+        strcmp(my_bind_addr_str, "*") != 0)
+    {
+      my_bind_ip= wsrep_check_ip(my_bind_addr_str, &is_ipv6);
+    }
+
+    if (INADDR_ANY != my_bind_ip)
+    {
+      /*
+        If its a not a valid address, leave inc_addr as empty string. mysqld
+        is not listening for client connections on network interfaces.
+      */
+      if (INADDR_NONE != my_bind_ip && INADDR_LOOPBACK != my_bind_ip)
+      {
+        const char *fmt= (is_ipv6) ? "[%s]:%u" : "%s:%u";
+        snprintf(inc_addr, inc_addr_max, fmt, my_bind_addr_str, mysqld_port);
+      }
+    }
+    else /* mysqld binds to 0.0.0.0, try taking IP from wsrep_node_address. */
+    {
+      if (node_addr.size())
+      {
+        size_t const ip_len_mdb= wsrep_host_len(node_addr.c_str(), node_addr.size());
+        if (ip_len_mdb + 7 /* :55555\0 */ < inc_addr_max)
+        {
+          memcpy (inc_addr, node_addr.c_str(), ip_len_mdb);
+          snprintf(inc_addr + ip_len_mdb, inc_addr_max - ip_len_mdb, ":%u",
+                   (int)mysqld_port);
+        }
+        else
+        {
+          WSREP_WARN("Guessing address for incoming client connections: "
+                     "address too long.");
+          inc_addr[0]= '\0';
+        }
+      }
+
+      if (!strlen(inc_addr))
+      {
+        WSREP_WARN("Guessing address for incoming client connections failed. "
+                   "Try setting wsrep_node_incoming_address explicitly.");
+        WSREP_INFO("Node addr: %s", node_addr.c_str());
+      }
+    }
+  }
+  else
+  {
+    wsp::Address addr(wsrep_node_incoming_address);
+
+    if (!addr.is_valid())
+    {
+      WSREP_WARN("Could not parse wsrep_node_incoming_address : %s",
+                 wsrep_node_incoming_address);
+      goto done;
+    }
+
+    /*
+      In case port is not specified in wsrep_node_incoming_address, we use
+      mysqld_port.
+      Note that we might get here before we execute set_ports().
+    */
+    int local_port= (addr.get_port() > 0) ? addr.get_port() : (int) mysqld_port;
+    if (!local_port)
+      local_port= MYSQL_PORT;
+    const char *fmt= (addr.is_ipv6()) ? "[%s]:%u" : "%s:%u";
+
+    snprintf(inc_addr, inc_addr_max, fmt, addr.get_address(), local_port);
+  }
+
+ done:
+  if (!strlen(inc_addr))
+    ret= wsrep_node_incoming_address;
+  else
+    ret= inc_addr;
+  WSREP_DEBUG("wsrep_incoming_address = %s", ret.c_str());
+  return ret;
+}
+
+static std::string wsrep_server_working_dir()
+{
+  std::string ret;
+  if (!wsrep_data_home_dir || strlen(wsrep_data_home_dir) == 0)
+  {
+    ret= mysql_real_data_home;
+  }
+  else
+  {
+    ret= wsrep_data_home_dir;
+  }
+  return ret;
+}
+
+static wsrep::gtid wsrep_server_initial_position()
+{
+  wsrep::gtid ret;
+  WSREP_DEBUG("Server initial position: %s", wsrep_start_position);
+  std::istringstream is(wsrep_start_position);
+  is >> ret;
+  return ret;
+}
+
+/*
+  Intitialize provider specific status variables
+ */
+static void wsrep_init_provider_status_variables()
+{
+  wsrep_inited= 1;
+  const wsrep::provider& provider=
+    Wsrep_server_state::instance().provider();
+  strncpy(provider_name,
+          provider.name().c_str(),    sizeof(provider_name) - 1);
+  strncpy(provider_version,
+          provider.version().c_str(), sizeof(provider_version) - 1);
+  strncpy(provider_vendor,
+          provider.vendor().c_str(),  sizeof(provider_vendor) - 1);
+}
+
+int wsrep_init_server()
+{
+  wsrep::log::logger_fn(wsrep_log_cb);
+  try
+  {
+    Wsrep_status::init_once(wsrep_status_file);
+
+    std::string server_name;
+    std::string server_id;
+    std::string node_address;
+    std::string incoming_address;
+    std::string working_dir;
+    wsrep::gtid initial_position;
+
+    server_name= wsrep_server_name();
+    server_id= wsrep_server_id();
+    node_address= wsrep_server_node_address();
+    incoming_address= wsrep_server_incoming_address();
+    working_dir= wsrep_server_working_dir();
+    initial_position= wsrep_server_initial_position();
+
+    Wsrep_server_state::init_once(server_name,
+                                  incoming_address,
+                                  node_address,
+                                  working_dir,
+                                  initial_position,
+                                  wsrep_max_protocol_version);
+    Wsrep_server_state::instance().debug_log_level(wsrep_debug);
+  }
+  catch (const wsrep::runtime_error& e)
+  {
+    WSREP_ERROR("Failed to init wsrep server %s", e.what());
+    return 1;
+  }
+  catch (const std::exception& e)
+  {
+    WSREP_ERROR("Failed to init wsrep server %s", e.what());
+  }
+  return 0;
+}
+
+void wsrep_init_globals()
+{
+  wsrep_init_sidno(Wsrep_server_state::instance().connected_gtid().id());
+  /* Recover last written wsrep gtid */
+  wsrep_init_gtid();
+  if (wsrep_new_cluster)
+  {
+    /* Start with provided domain_id & server_id found in configuration */
+    wsrep_server_gtid_t new_gtid;
+    new_gtid.domain_id= wsrep_gtid_domain_id;
+    new_gtid.server_id= global_system_variables.server_id;
+    new_gtid.seqno= 0;
+    /* Try to search for domain_id and server_id combination in binlog if found continue from last seqno */
+    wsrep_get_binlog_gtid_seqno(new_gtid);
+    wsrep_gtid_server.gtid(new_gtid);
+  }
+  else
+  {
+    if (wsrep_gtid_mode && wsrep_gtid_server.server_id != global_system_variables.server_id)
+    {
+      WSREP_WARN("Ignoring server id for non bootstrap node.");
+    }
+  }
+  wsrep_init_schema();
+  if (WSREP_ON)
+  {
+    Wsrep_server_state::instance().initialized();
+  }
+}
+
+void wsrep_deinit_server()
+{
+  wsrep_deinit_schema();
+  Wsrep_server_state::destroy();
+  Wsrep_status::destroy();
+  wsrep_free_status_vars();
+}
+
+int wsrep_init()
+{
+  assert(wsrep_provider);
+
+  wsrep_init_position();
+  wsrep_sst_auth_init();
+
+  if (strlen(wsrep_provider)== 0 ||
+      !strcmp(wsrep_provider, WSREP_NONE))
+  {
+    // enable normal operation in case no provider is specified
+    global_system_variables.wsrep_on= 0;
+    int err= Wsrep_server_state::instance().load_provider(wsrep_provider, wsrep_provider_options ? wsrep_provider_options : "");
+    if (err)
+    {
+      DBUG_PRINT("wsrep",("wsrep::init() failed: %d", err));
+      WSREP_ERROR("wsrep::init() failed: %d, must shutdown", err);
+    }
+    else
+      wsrep_init_provider_status_variables();
+    return err;
+  }
+
+  if (wsrep_gtid_mode && opt_bin_log && !opt_log_slave_updates)
+  {
+    WSREP_ERROR("Option --log-slave-updates is required if "
+                "binlog is enabled, GTID mode is on and wsrep provider "
+                "is specified");
+    return 1;
+  }
+
+  if (!wsrep_data_home_dir || strlen(wsrep_data_home_dir) == 0)
+    wsrep_data_home_dir= mysql_real_data_home;
+
+  Wsrep_server_state::init_provider_services();
+  if (Wsrep_server_state::instance().load_provider(
+      wsrep_provider,
+      wsrep_provider_options,
+      Wsrep_server_state::instance().provider_services()))
+  {
+    WSREP_ERROR("Failed to load provider");
+    Wsrep_server_state::deinit_provider_services();
+    return 1;
+  }
+
+  if (!wsrep_provider_is_SR_capable() &&
+      global_system_variables.wsrep_trx_fragment_size > 0)
+  {
+    WSREP_ERROR("The WSREP provider (%s) does not support streaming "
+                "replication but wsrep_trx_fragment_size is set to a "
+                "value other than 0 (%llu). Cannot continue. Either set "
+                "wsrep_trx_fragment_size to 0 or use wsrep_provider that "
+                "supports streaming replication.",
+                wsrep_provider, global_system_variables.wsrep_trx_fragment_size);
+    Wsrep_server_state::instance().unload_provider();
+    Wsrep_server_state::deinit_provider_services();
+    return 1;
+  }
+
+  /* Now WSREP is fully initialized */
+  global_system_variables.wsrep_on= 1;
+  WSREP_ON_= wsrep_provider && strcmp(wsrep_provider, WSREP_NONE);
+  wsrep_service_started= 1;
+
+  wsrep_init_provider_status_variables();
+  wsrep_capabilities_export(Wsrep_server_state::instance().provider().capabilities(),
+                            &wsrep_provider_capabilities);
+
+  WSREP_DEBUG("SR storage init for: %s",
+              (wsrep_SR_store_type == WSREP_SR_STORE_TABLE) ? "table" : "void");
+
+  return 0;
+}
+
+/* Initialize wsrep thread LOCKs and CONDs */
+void wsrep_thr_init()
+{
+  DBUG_ENTER("wsrep_thr_init");
+  wsrep_config_state= new wsp::Config_state;
+#ifdef HAVE_PSI_INTERFACE
+  mysql_mutex_register("sql", wsrep_mutexes, array_elements(wsrep_mutexes));
+  mysql_cond_register("sql", wsrep_conds, array_elements(wsrep_conds));
+  mysql_file_register("sql", wsrep_files, array_elements(wsrep_files));
+  mysql_thread_register("sql", wsrep_threads, array_elements(wsrep_threads));
+#endif
+
+  mysql_mutex_init(key_LOCK_wsrep_ready, &LOCK_wsrep_ready, MY_MUTEX_INIT_FAST);
+  mysql_cond_init(key_COND_wsrep_ready, &COND_wsrep_ready, NULL);
+  mysql_mutex_init(key_LOCK_wsrep_sst, &LOCK_wsrep_sst, MY_MUTEX_INIT_FAST);
+  mysql_cond_init(key_COND_wsrep_sst, &COND_wsrep_sst, NULL);
+  mysql_mutex_init(key_LOCK_wsrep_sst_init, &LOCK_wsrep_sst_init, MY_MUTEX_INIT_FAST);
+  mysql_cond_init(key_COND_wsrep_sst_init, &COND_wsrep_sst_init, NULL);
+  mysql_mutex_init(key_LOCK_wsrep_replaying, &LOCK_wsrep_replaying, MY_MUTEX_INIT_FAST);
+  mysql_cond_init(key_COND_wsrep_replaying, &COND_wsrep_replaying, NULL);
+  mysql_mutex_init(key_LOCK_wsrep_slave_threads, &LOCK_wsrep_slave_threads, MY_MUTEX_INIT_FAST);
+  mysql_cond_init(key_COND_wsrep_slave_threads, &COND_wsrep_slave_threads, NULL);
+  mysql_mutex_init(key_LOCK_wsrep_gtid_wait_upto, &LOCK_wsrep_gtid_wait_upto, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_wsrep_cluster_config, &LOCK_wsrep_cluster_config, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_wsrep_desync, &LOCK_wsrep_desync, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_wsrep_config_state, &LOCK_wsrep_config_state, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_wsrep_group_commit, &LOCK_wsrep_group_commit, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_wsrep_SR_pool,
+                   &LOCK_wsrep_SR_pool, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_wsrep_SR_store,
+                   &LOCK_wsrep_SR_store, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_wsrep_joiner_monitor,
+                   &LOCK_wsrep_joiner_monitor, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_wsrep_donor_monitor,
+                   &LOCK_wsrep_donor_monitor, MY_MUTEX_INIT_FAST);
+  mysql_cond_init(key_COND_wsrep_joiner_monitor, &COND_wsrep_joiner_monitor, NULL);
+  mysql_cond_init(key_COND_wsrep_donor_monitor, &COND_wsrep_donor_monitor, NULL);
+
+  DBUG_VOID_RETURN;
+}
+
+void wsrep_init_startup (bool sst_first)
+{
+  if (wsrep_init()) unireg_abort(1);
+
+  /*
+    Pre-initialize global_system_variables.table_plugin with a dummy engine
+    (placeholder) required during the initialization of wsrep threads (THDs).
+    (see: plugin_thdvar_init())
+    Note: This only needs to be done for rsync & mariabackup based SST methods.
+    In case of mysqldump SST method, the wsrep threads are created after the
+    server plugins & global system variables are initialized.
+  */
+  if (wsrep_before_SE())
+    wsrep_plugins_pre_init();
+
+  /* Skip replication start if dummy wsrep provider is loaded */
+  if (!strcmp(wsrep_provider, WSREP_NONE)) return;
+
+  /* Skip replication start if no cluster address */
+  if (!wsrep_cluster_address_exists()) return;
+
+  /*
+    Read value of wsrep_new_cluster before wsrep_start_replication(),
+    the value is reset to FALSE inside wsrep_start_replication.
+  */
+  if (!wsrep_start_replication(wsrep_cluster_address)) unireg_abort(1);
+
+  wsrep_create_rollbacker();
+  wsrep_create_appliers(1);
+
+  Wsrep_server_state& server_state= Wsrep_server_state::instance();
+  /*
+    If the SST happens before server initialization, wait until the server
+    state reaches initializing. This indicates that
+    either SST was not necessary or SST has been delivered.
+
+    With mysqldump SST (!sst_first) wait until the server reaches
+    joiner state and procedd to accepting connections.
+  */
+  int err= 0;
+  if (sst_first)
+  {
+    err= server_state.wait_until_state(Wsrep_server_state::s_initializing);
+  }
+  else
+  {
+    err= server_state.wait_until_state(Wsrep_server_state::s_joiner);
+  }
+  if (err)
+  {
+    WSREP_ERROR("Wsrep startup was interrupted");
+    unireg_abort(1);
+  }
+}
+
+
+void wsrep_deinit(bool free_options)
+{
+  DBUG_ASSERT(wsrep_inited == 1);
+  WSREP_DEBUG("wsrep_deinit");
+
+  Wsrep_server_state::instance().unload_provider();
+  Wsrep_server_state::deinit_provider_services();
+
+  provider_name[0]=    '\0';
+  provider_version[0]= '\0';
+  provider_vendor[0]=  '\0';
+
+  wsrep_inited= 0;
+
+  if (wsrep_provider_capabilities != NULL)
+  {
+    char* p= wsrep_provider_capabilities;
+    wsrep_provider_capabilities= NULL;
+    free(p);
+  }
+
+  if (free_options)
+  {
+    wsrep_sst_auth_free();
+  }
+}
+
+/* Destroy wsrep thread LOCKs and CONDs */
+void wsrep_thr_deinit()
+{
+  if (!wsrep_config_state)
+    return;                                     // Never initialized
+  WSREP_DEBUG("wsrep_thr_deinit");
+  mysql_mutex_destroy(&LOCK_wsrep_ready);
+  mysql_cond_destroy(&COND_wsrep_ready);
+  mysql_mutex_destroy(&LOCK_wsrep_sst);
+  mysql_cond_destroy(&COND_wsrep_sst);
+  mysql_mutex_destroy(&LOCK_wsrep_sst_init);
+  mysql_cond_destroy(&COND_wsrep_sst_init);
+  mysql_mutex_destroy(&LOCK_wsrep_replaying);
+  mysql_cond_destroy(&COND_wsrep_replaying);
+  mysql_mutex_destroy(&LOCK_wsrep_gtid_wait_upto);
+  mysql_mutex_destroy(&LOCK_wsrep_slave_threads);
+  mysql_cond_destroy(&COND_wsrep_slave_threads);
+  mysql_mutex_destroy(&LOCK_wsrep_cluster_config);
+  mysql_mutex_destroy(&LOCK_wsrep_desync);
+  mysql_mutex_destroy(&LOCK_wsrep_config_state);
+  mysql_mutex_destroy(&LOCK_wsrep_group_commit);
+  mysql_mutex_destroy(&LOCK_wsrep_SR_pool);
+  mysql_mutex_destroy(&LOCK_wsrep_SR_store);
+  mysql_mutex_destroy(&LOCK_wsrep_joiner_monitor);
+  mysql_mutex_destroy(&LOCK_wsrep_donor_monitor);
+  mysql_cond_destroy(&COND_wsrep_joiner_monitor);
+  mysql_cond_destroy(&COND_wsrep_donor_monitor);
+
+  delete wsrep_config_state;
+  wsrep_config_state= 0;                        // Safety
+
+  if (wsrep_cluster_capabilities != NULL)
+  {
+    char* p= wsrep_cluster_capabilities;
+    wsrep_cluster_capabilities= NULL;
+    free(p);
+  }
+}
+
+void wsrep_recover()
+{
+  char uuid_str[40];
+
+  if (wsrep_uuid_compare(&local_uuid, &WSREP_UUID_UNDEFINED) == 0 &&
+      local_seqno == -2)
+  {
+    wsrep_uuid_print(&local_uuid, uuid_str, sizeof(uuid_str));
+    WSREP_INFO("Position %s:%lld given at startup, skipping position recovery",
+               uuid_str, (long long)local_seqno);
+    return;
+  }
+  wsrep::gtid gtid= wsrep_get_SE_checkpoint<wsrep::gtid>();
+  std::ostringstream oss;
+  oss << gtid;
+  if (wsrep_gtid_mode)
+  {
+    wsrep_server_gtid_t server_gtid=  wsrep_get_SE_checkpoint<wsrep_server_gtid_t>();
+    WSREP_INFO("Recovered position: %s,%d-%d-%llu", oss.str().c_str(), server_gtid.domain_id,
+                server_gtid.server_id, server_gtid.seqno);
+  }
+  else
+  {
+    WSREP_INFO("Recovered position: %s", oss.str().c_str());
+  }
+}
+
+
+void wsrep_stop_replication(THD *thd)
+{
+  WSREP_INFO("Stop replication by %llu", (thd) ? thd->thread_id : 0);
+  if (Wsrep_server_state::instance().state() !=
+      Wsrep_server_state::s_disconnected)
+  {
+    WSREP_DEBUG("Disconnect provider");
+    Wsrep_server_state::instance().disconnect();
+    if (Wsrep_server_state::instance().wait_until_state(
+            Wsrep_server_state::s_disconnected))
+    {
+      WSREP_WARN("Wsrep interrupted while waiting for disconnected state");
+    }
+  }
+
+  /* my connection, should not terminate with wsrep_close_client_connection(),
+     make transaction to rollback
+  */
+  if (thd && !thd->wsrep_applier) trans_rollback(thd);
+  wsrep_close_client_connections(TRUE, thd);
+
+  /* wait until appliers have stopped */
+  wsrep_wait_appliers_close(thd);
+
+  node_uuid= WSREP_UUID_UNDEFINED;
+}
+
+void wsrep_shutdown_replication()
+{
+  WSREP_INFO("Shutdown replication");
+  if (Wsrep_server_state::instance().state() != wsrep::server_state::s_disconnected)
+  {
+    WSREP_DEBUG("Disconnect provider");
+    Wsrep_server_state::instance().disconnect();
+    if (Wsrep_server_state::instance().wait_until_state(
+            Wsrep_server_state::s_disconnected))
+    {
+      WSREP_WARN("Wsrep interrupted while waiting for disconnected state");
+    }
+  }
+
+  wsrep_close_client_connections(TRUE);
+
+  /* wait until appliers have stopped */
+  wsrep_wait_appliers_close(NULL);
+  node_uuid= WSREP_UUID_UNDEFINED;
+
+  /* Undocking the thread specific data. */
+  set_current_thd(nullptr);
+}
+
+bool wsrep_start_replication(const char *wsrep_cluster_address)
+{
+  int rcode;
+  WSREP_DEBUG("wsrep_start_replication");
+
+  /*
+    if provider is trivial, don't even try to connect,
+    but resume local node operation
+  */
+  if (!WSREP_PROVIDER_EXISTS)
+  {
+    // enable normal operation in case no provider is specified
+    return true;
+  }
+
+  DBUG_ASSERT(wsrep_cluster_address[0]);
+
+  // --wsrep-new-cluster flag is not used, checking wsrep_cluster_address
+  // it should match gcomm:// only to be considered as bootstrap node.
+  // This logic is used in galera.
+  if (!wsrep_new_cluster && 
+      (strlen(wsrep_cluster_address) == 8) &&
+      !strncmp(wsrep_cluster_address, "gcomm://", 8)) 
+  {
+    wsrep_new_cluster= true;
+  }
+
+  bool const bootstrap(TRUE == wsrep_new_cluster);
+
+  WSREP_INFO("Start replication");
+
+  if ((rcode= Wsrep_server_state::instance().connect(
+        wsrep_cluster_name,
+        wsrep_cluster_address,
+        wsrep_sst_donor,
+        bootstrap)))
+  {
+    DBUG_PRINT("wsrep",("wsrep_ptr->connect(%s) failed: %d",
+                        wsrep_cluster_address, rcode));
+    WSREP_ERROR("wsrep::connect(%s) failed: %d",
+                wsrep_cluster_address, rcode);
+    return false;
+  }
+  else
+  {
+    try
+    {
+      std::string opts= Wsrep_server_state::instance().provider().options();
+      wsrep_provider_options_init(opts.c_str());
+    }
+    catch (const wsrep::runtime_error&)
+    {
+      WSREP_WARN("Failed to get wsrep options");
+    }
+  }
+
+  return true;
+}
+
+bool wsrep_check_mode (enum_wsrep_mode mask)
+{
+  return wsrep_mode & mask;
+}
+
+//seconds after which the limit warnings suppression will be activated
+#define WSREP_WARNING_ACTIVATION_TIMEOUT 5*60
+//number of limit warnings after which the suppression will be activated
+#define WSREP_WARNING_ACTIVATION_THRESHOLD 10
+
+enum wsrep_warning_type {
+  WSREP_DISABLED = 0,
+  WSREP_REQUIRE_PRIMARY_KEY= 1,
+  WSREP_REQUIRE_INNODB= 2,
+  WSREP_REQUIRE_MAX=3,
+};
+
+static ulonglong wsrep_warning_start_time=0;
+static bool wsrep_warning_active[WSREP_REQUIRE_MAX+1];
+static ulonglong wsrep_warning_count[WSREP_REQUIRE_MAX+1];
+static ulonglong wsrep_total_warnings_count=0;
+
+/**
+  Auxiliary function to reset the limit of wsrep warnings.
+  This is done without mutex protection, but this should be good
+  enough as it doesn't matter if we loose a couple of suppressed
+  messages or if this is called multiple times.
+*/
+
+static void wsrep_reset_warnings(ulonglong now)
+{
+  uint i;
+
+  wsrep_warning_start_time= now;
+  wsrep_total_warnings_count= 0;
+
+  for (i= 0 ; i < WSREP_REQUIRE_MAX ; i++)
+  {
+    wsrep_warning_active[i]= false;
+    wsrep_warning_count[i]= 0;
+  }
+}
+
+static const char* wsrep_warning_name(const enum wsrep_warning_type type)
+{
+  switch(type)
+  {
+  case WSREP_REQUIRE_PRIMARY_KEY:
+    return "WSREP_REQUIRE_PRIMARY_KEY"; break;
+  case WSREP_REQUIRE_INNODB:
+    return "WSREP_REQUIRE_INNODB"; break;
+  default: assert(0); return " "; break; // for compiler
+  }
+}
+/**
+  Auxiliary function to check if the warning statements should be
+  thrown or suppressed.
+
+  Logic is:
+  - If we get more than WSREP_WARNING_ACTIVATION_THRESHOLD errors
+    of one type, that type of errors will be suppressed for
+    WSREP_WARNING_ACTIVATION_TIMEOUT.
+  - When the time limit has been reached, all suppressions are reset.
+
+  This means that if one gets many different types of errors, some of them
+  may be reset less than WSREP_WARNING_ACTIVATION_TIMEOUT. However at
+  least one error is disabled for this time.
+
+  SYNOPSIS:
+  @params
+   warning_type - The type of warning.
+
+  RETURN:
+    0   0k to log
+    1   Message suppressed
+*/
+
+static bool wsrep_protect_against_warning_flood(
+              enum wsrep_warning_type warning_type)
+{
+  ulonglong count;
+  ulonglong now= my_interval_timer()/1000000000ULL;
+
+  count= ++wsrep_warning_count[warning_type];
+  wsrep_total_warnings_count++;
+
+  /*
+    INITIALIZING:
+    If this is the first time this function is called with log warning
+    enabled, the monitoring the warnings should start.
+  */
+  if (wsrep_warning_start_time == 0)
+  {
+    wsrep_reset_warnings(now);
+    return false;
+  }
+
+  /*
+    The following is true if we got too many errors or if the error was
+    already suppressed
+  */
+  if (count >= WSREP_WARNING_ACTIVATION_THRESHOLD)
+  {
+    ulonglong diff_time= (now - wsrep_warning_start_time);
+
+    if (!wsrep_warning_active[warning_type])
+    {
+      /*
+        ACTIVATION:
+        We got WSREP_WARNING_ACTIVATION_THRESHOLD warnings in
+        less than WSREP_WARNING_ACTIVATION_TIMEOUT we activate the
+        suppression.
+      */
+      if (diff_time <= WSREP_WARNING_ACTIVATION_TIMEOUT)
+      {
+        wsrep_warning_active[warning_type]= true;
+        WSREP_INFO("Suppressing warnings of type '%s' for up to %d seconds because of flooding",
+                   wsrep_warning_name(warning_type),
+                   WSREP_WARNING_ACTIVATION_TIMEOUT);
+      }
+      else
+      {
+        /*
+          There is no flooding till now, therefore we restart the monitoring
+        */
+        wsrep_reset_warnings(now);
+      }
+    }
+    else
+    {
+      /* This type of warnings was suppressed */
+      if (diff_time > WSREP_WARNING_ACTIVATION_TIMEOUT)
+      {
+        ulonglong save_count= wsrep_total_warnings_count;
+        /* Print a suppression note and remove the suppression */
+        wsrep_reset_warnings(now);
+        WSREP_INFO("Suppressed %lu unsafe warnings during "
+                              "the last %d seconds",
+                              save_count, (int) diff_time);
+      }
+    }
+  }
+
+  return wsrep_warning_active[warning_type];
+}
+
+/**
+  Auxiliary function to push warning to client and to the error log
+*/
+static void wsrep_push_warning(THD *thd,
+                               enum wsrep_warning_type type,
+                               const handlerton *hton,
+                               const TABLE_LIST *tables)
+{
+  switch(type)
+  {
+  case WSREP_REQUIRE_PRIMARY_KEY:
+    push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                        ER_OPTION_PREVENTS_STATEMENT,
+                        "WSREP: wsrep_mode = REQUIRED_PRIMARY_KEY enabled. "
+                        "Table '%s'.'%s' should have PRIMARY KEY defined.",
+                        tables->db.str, tables->table_name.str);
+    if (global_system_variables.log_warnings > 1 &&
+	!wsrep_protect_against_warning_flood(type))
+      WSREP_WARN("wsrep_mode = REQUIRED_PRIMARY_KEY enabled. "
+                 "Table '%s'.'%s' should have PRIMARY KEY defined",
+                 tables->db.str, tables->table_name.str);
+    break;
+  case WSREP_REQUIRE_INNODB:
+    push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                        ER_OPTION_PREVENTS_STATEMENT,
+                        "WSREP: wsrep_mode = STRICT_REPLICATION enabled. "
+                        "Storage engine %s for table '%s'.'%s' is "
+                        "not supported in Galera",
+                        ha_resolve_storage_engine_name(hton),
+                        tables->db.str, tables->table_name.str);
+    if (global_system_variables.log_warnings > 1 &&
+	!wsrep_protect_against_warning_flood(type))
+      WSREP_WARN("wsrep_mode = STRICT_REPLICATION enabled. "
+                 "Storage engine %s for table '%s'.'%s' is "
+                 "not supported in Galera",
+                 ha_resolve_storage_engine_name(hton),
+                 tables->db.str, tables->table_name.str);
+    break;
+
+  default: assert(0); break;
+  }
+}
+
+bool wsrep_check_mode_after_open_table (THD *thd,
+	const handlerton *hton,
+	TABLE_LIST *tables)
+{
+  enum_sql_command sql_command= thd->lex->sql_command;
+  bool is_dml_stmt= thd->get_command() != COM_STMT_PREPARE &&
+                    (sql_command == SQLCOM_INSERT ||
+                     sql_command == SQLCOM_INSERT_SELECT ||
+                     sql_command == SQLCOM_REPLACE ||
+                     sql_command == SQLCOM_REPLACE_SELECT ||
+                     sql_command == SQLCOM_UPDATE ||
+                     sql_command == SQLCOM_UPDATE_MULTI ||
+                     sql_command == SQLCOM_LOAD ||
+                     sql_command == SQLCOM_DELETE);
+
+  if (!is_dml_stmt)
+    return true;
+
+  const legacy_db_type db_type= hton->db_type;
+  bool replicate= ((db_type == DB_TYPE_MYISAM && wsrep_check_mode(WSREP_MODE_REPLICATE_MYISAM)) ||
+                   (db_type == DB_TYPE_ARIA && wsrep_check_mode(WSREP_MODE_REPLICATE_ARIA)));
+  TABLE *tbl= tables->table;
+
+  if (replicate)
+  {
+    /* It is not recommended to replicate MyISAM as it lacks rollback feature
+    but if user demands then actions are replicated using TOI.
+    Following code will kick-start the TOI but this has to be done only once
+    per statement.
+    Note: kick-start will take-care of creating isolation key for all tables
+    involved in the list (provided all of them are MYISAM or Aria tables). */
+    if (!is_stat_table(&tables->db, &tables->alias))
+    {
+      if (tbl->s->primary_key == MAX_KEY &&
+          wsrep_check_mode(WSREP_MODE_REQUIRED_PRIMARY_KEY))
+      {
+        /* Other replicated table doesn't have explicit primary-key defined. */
+        wsrep_push_warning(thd, WSREP_REQUIRE_PRIMARY_KEY, hton, tables);
+      }
+
+      wsrep_before_rollback(thd, true);
+      wsrep_after_rollback(thd, true);
+      wsrep_after_statement(thd);
+      WSREP_TO_ISOLATION_BEGIN(NULL, NULL, (tables));
+    }
+  } else if (db_type != DB_TYPE_UNKNOWN &&
+             db_type != DB_TYPE_PERFORMANCE_SCHEMA)
+  {
+    bool is_system_db= (tbl &&
+                       ((strcmp(tbl->s->db.str, "mysql") == 0) ||
+                        (strcmp(tbl->s->db.str, "information_schema") == 0)));
+
+    if (!is_system_db &&
+	!is_temporary_table(tables))
+    {
+
+      if (db_type != DB_TYPE_INNODB &&
+	  wsrep_check_mode(WSREP_MODE_STRICT_REPLICATION))
+      {
+        /* Table is not an InnoDB table and strict replication is requested*/
+        wsrep_push_warning(thd, WSREP_REQUIRE_INNODB, hton, tables);
+      }
+
+      if (tbl->s->primary_key == MAX_KEY &&
+          db_type == DB_TYPE_INNODB &&
+          wsrep_check_mode(WSREP_MODE_REQUIRED_PRIMARY_KEY))
+      {
+        /* InnoDB table doesn't have explicit primary-key defined. */
+        wsrep_push_warning(thd, WSREP_REQUIRE_PRIMARY_KEY, hton, tables);
+      }
+
+      if (db_type != DB_TYPE_INNODB &&
+	  thd->variables.sql_log_bin == 1 &&
+	  wsrep_check_mode(WSREP_MODE_DISALLOW_LOCAL_GTID))
+      {
+        /* Table is not an InnoDB table and local GTIDs are disallowed */
+        my_error(ER_GALERA_REPLICATION_NOT_SUPPORTED, MYF(0));
+        push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                            ER_OPTION_PREVENTS_STATEMENT,
+                            "You can't execute statements that would generate local "
+                            "GTIDs when wsrep_mode = DISALLOW_LOCAL_GTID is set. "
+                            "Try disabling binary logging with SET sql_log_bin=0 "
+                            "to execute this statement.");
+        goto wsrep_error_label;
+      }
+    }
+  }
+
+  return true;
+
+wsrep_error_label:
+  return false;
+}
+
+bool wsrep_check_mode_before_cmd_execute (THD *thd)
+{
+  bool ret= true;
+  if (wsrep_check_mode(WSREP_MODE_BINLOG_ROW_FORMAT_ONLY) &&
+      !thd->is_current_stmt_binlog_format_row() && is_update_query(thd->lex->sql_command))
+  {
+    my_error(ER_GALERA_REPLICATION_NOT_SUPPORTED, MYF(0));
+    push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                        ER_OPTION_PREVENTS_STATEMENT,
+                        "WSREP: wsrep_mode = BINLOG_ROW_FORMAT_ONLY enabled. Only ROW binlog format is supported.");
+    ret= false;
+  }
+  if (wsrep_check_mode(WSREP_MODE_REQUIRED_PRIMARY_KEY) &&
+      thd->lex->sql_command == SQLCOM_CREATE_TABLE)
+  {
+    Key *key;
+    List_iterator<Key> key_iterator(thd->lex->alter_info.key_list);
+    bool primary_key_found= false;
+    while ((key= key_iterator++))
+    {
+      if (key->type == Key::PRIMARY)
+      {
+        primary_key_found= true;
+        break;
+      }
+    }
+    if (!primary_key_found)
+    {
+      my_error(ER_GALERA_REPLICATION_NOT_SUPPORTED, MYF(0));
+      push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                          ER_OPTION_PREVENTS_STATEMENT,
+                          "WSREP: wsrep_mode = REQUIRED_PRIMARY_KEY enabled. Table should have PRIMARY KEY defined.");
+      ret= false;
+    }
+  }
+  return ret;
+}
+
+bool wsrep_must_sync_wait (THD* thd, uint mask)
+{
+  bool ret= 0;
+  if (thd->variables.wsrep_on)
+  {
+    mysql_mutex_lock(&thd->LOCK_thd_data);
+    ret= (thd->variables.wsrep_sync_wait & mask) &&
+      thd->wsrep_client_thread &&
+      WSREP_ON &&
+      !(thd->variables.wsrep_dirty_reads &&
+        !is_update_query(thd->lex->sql_command)) &&
+      !thd->in_active_multi_stmt_transaction() &&
+      thd->wsrep_trx().state() !=
+      wsrep::transaction::s_replaying &&
+      thd->wsrep_cs().sync_wait_gtid().is_undefined();
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
+  }
+  return ret;
+}
+
+bool wsrep_sync_wait (THD* thd, uint mask)
+{
+  if (wsrep_must_sync_wait(thd, mask))
+  {
+    WSREP_DEBUG("wsrep_sync_wait: thd->variables.wsrep_sync_wait= %u, "
+                "mask= %u, thd->variables.wsrep_on= %d",
+                thd->variables.wsrep_sync_wait, mask,
+                thd->variables.wsrep_on);
+    /*
+      This allows autocommit SELECTs and a first SELECT after SET AUTOCOMMIT=0
+      TODO: modify to check if thd has locked any rows.
+    */
+    if (thd->wsrep_cs().sync_wait(-1))
+    {
+      const char* msg;
+      int err;
+
+      /*
+        Possibly relevant error codes:
+        ER_CHECKREAD, ER_ERROR_ON_READ, ER_INVALID_DEFAULT, ER_EMPTY_QUERY,
+        ER_FUNCTION_NOT_DEFINED, ER_NOT_ALLOWED_COMMAND, ER_NOT_SUPPORTED_YET,
+        ER_FEATURE_DISABLED, ER_QUERY_INTERRUPTED
+      */
+
+      switch (thd->wsrep_cs().current_error())
+      {
+      case wsrep::e_not_supported_error:
+        msg= "synchronous reads by wsrep backend. "
+          "Please unset wsrep_causal_reads variable.";
+        err= ER_NOT_SUPPORTED_YET;
+        break;
+      default:
+        msg= "Synchronous wait failed.";
+        err= ER_LOCK_WAIT_TIMEOUT; // NOTE: the above msg won't be displayed
+                                   //       with ER_LOCK_WAIT_TIMEOUT
+      }
+
+      my_error(err, MYF(0), msg);
+
+      return true;
+    }
+  }
+
+  return false;
+}
+
+enum wsrep::provider::status
+wsrep_sync_wait_upto (THD*          thd,
+                      wsrep_gtid_t* upto,
+                      int           timeout)
+{
+  DBUG_ASSERT(upto);
+  enum wsrep::provider::status ret;
+  if (upto)
+  {
+    wsrep::gtid upto_gtid(wsrep::id(upto->uuid.data, sizeof(upto->uuid.data)),
+                          wsrep::seqno(upto->seqno));
+    ret= Wsrep_server_state::instance().wait_for_gtid(upto_gtid, timeout);
+  }
+  else
+  {
+    ret= Wsrep_server_state::instance().causal_read(timeout).second;
+  }
+  WSREP_DEBUG("wsrep_sync_wait_upto: %d", ret);
+  return ret;
+}
+
+bool wsrep_is_show_query(enum enum_sql_command command)
+{
+  DBUG_ASSERT(command >= 0 && command <= SQLCOM_END);
+  return (sql_command_flags[command] & CF_STATUS_COMMAND) != 0;
+}
+
+static bool wsrep_is_diagnostic_query(enum enum_sql_command command)
+{
+  assert(command >= 0 && command <= SQLCOM_END);
+  return (sql_command_flags[command] & CF_DIAGNOSTIC_STMT) != 0;
+}
+
+static enum enum_wsrep_sync_wait
+wsrep_sync_wait_mask_for_command(enum enum_sql_command command)
+{
+  switch (command)
+  {
+  case SQLCOM_SELECT:
+  case SQLCOM_CHECKSUM:
+    return WSREP_SYNC_WAIT_BEFORE_READ;
+  case SQLCOM_DELETE:
+  case SQLCOM_DELETE_MULTI:
+  case SQLCOM_UPDATE:
+  case SQLCOM_UPDATE_MULTI:
+    return WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE;
+  case SQLCOM_REPLACE:
+  case SQLCOM_INSERT:
+  case SQLCOM_REPLACE_SELECT:
+  case SQLCOM_INSERT_SELECT:
+    return WSREP_SYNC_WAIT_BEFORE_INSERT_REPLACE;
+  default:
+    if (wsrep_is_diagnostic_query(command))
+    {
+      return WSREP_SYNC_WAIT_NONE;
+    }
+    if (wsrep_is_show_query(command))
+    {
+      switch (command)
+      {
+      case SQLCOM_SHOW_PROFILE:
+      case SQLCOM_SHOW_PROFILES:
+      case SQLCOM_SHOW_SLAVE_HOSTS:
+      case SQLCOM_SHOW_RELAYLOG_EVENTS:
+      case SQLCOM_SHOW_SLAVE_STAT:
+      case SQLCOM_SHOW_BINLOG_STAT:
+      case SQLCOM_SHOW_ENGINE_STATUS:
+      case SQLCOM_SHOW_ENGINE_MUTEX:
+      case SQLCOM_SHOW_ENGINE_LOGS:
+      case SQLCOM_SHOW_PROCESSLIST:
+      case SQLCOM_SHOW_PRIVILEGES:
+        return WSREP_SYNC_WAIT_NONE;
+      default:
+        return WSREP_SYNC_WAIT_BEFORE_SHOW;
+      }
+    }
+  }
+  return WSREP_SYNC_WAIT_NONE;
+}
+
+bool wsrep_sync_wait(THD* thd, enum enum_sql_command command)
+{
+  bool res = false;
+  if (WSREP_CLIENT(thd) && thd->variables.wsrep_sync_wait)
+    res = wsrep_sync_wait(thd, wsrep_sync_wait_mask_for_command(command));
+  return res;
+}
+
+void wsrep_keys_free(wsrep_key_arr_t* key_arr)
+{
+    for (size_t i= 0; i < key_arr->keys_len; ++i)
+    {
+        my_free((void*)key_arr->keys[i].key_parts);
+    }
+    my_free(key_arr->keys);
+    key_arr->keys= 0;
+    key_arr->keys_len= 0;
+}
+
+/*!
+ * @param thd    thread
+ * @param tables list of tables
+ * @param keys   prepared keys
+
+ * @return true if parent table append was successfull, otherwise false.
+*/
+bool
+wsrep_append_fk_parent_table(THD* thd, TABLE_LIST* tables, wsrep::key_array* keys)
+{
+    bool fail= false;
+    TABLE_LIST *table;
+    TABLE_LIST *table_last_in_list;
+
+    for (table= tables; table; table= table->next_local)
+    {
+      if (is_temporary_table(table))
+      {
+        WSREP_DEBUG("Temporary table %s.%s already opened query=%s", table->db.str,
+                    table->table_name.str, wsrep_thd_query(thd));
+	return false;
+      }
+    }
+
+    thd->release_transactional_locks();
+    uint counter;
+    MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
+
+    for (table_last_in_list= tables;;table_last_in_list= table_last_in_list->next_local)    {
+      if (!table_last_in_list->next_local) {
+        break;
+      }
+    }
+
+    if (open_tables(thd, &tables, &counter, MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL))
+    {
+      WSREP_DEBUG("Unable to open table for FK checks for %s", wsrep_thd_query(thd));
+      fail= true;
+      goto exit;
+    }
+
+    for (table= tables; table; table= table->next_local)
+    {
+      if (!is_temporary_table(table) && table->table)
+      {
+        FOREIGN_KEY_INFO *f_key_info;
+        List<FOREIGN_KEY_INFO> f_key_list;
+
+        table->table->file->get_foreign_key_list(thd, &f_key_list);
+        List_iterator_fast<FOREIGN_KEY_INFO> it(f_key_list);
+        while ((f_key_info=it++))
+        {
+          WSREP_DEBUG("appended fkey %s", f_key_info->referenced_table->str);
+          keys->push_back(wsrep_prepare_key_for_toi(f_key_info->referenced_db->str,
+                                                    f_key_info->referenced_table->str,
+                                                    wsrep::key::shared));
+        }
+      }
+    }
+
+exit:
+    /* close the table and release MDL locks */
+    close_thread_tables(thd);
+    thd->mdl_context.rollback_to_savepoint(mdl_savepoint);
+    bool invalidate_next_global= false;
+    for (table= tables; table; table= table->next_local)
+    {
+      table->table= NULL;
+      table->mdl_request.ticket= NULL;
+      // We should invalidate `next_global` only for entries that are added 
+      // in this function
+      if (table == table_last_in_list) {
+        invalidate_next_global= true;
+      }
+      if (invalidate_next_global) {
+        table->next_global= NULL;
+      }
+    }
+
+    return fail;
+}
+
+bool wsrep_reload_ssl()
+{
+  try
+  {
+    std::string opts= Wsrep_server_state::instance().provider().options();
+    if (opts.find("socket.ssl_reload") == std::string::npos)
+    {
+      WSREP_DEBUG("Option `socket.ssl_reload` not found in parameters.");
+      return false;
+    }
+    const std::string reload_ssl_param("socket.ssl_reload=1");
+    enum wsrep::provider::status ret= Wsrep_server_state::instance().provider().options(reload_ssl_param);
+    if (ret)
+    {
+      WSREP_ERROR("Set options returned %d", ret);
+      return true;
+    }
+    return false;
+  }
+  catch (...)
+  {
+    WSREP_ERROR("Failed to get provider options");
+    return true;
+  }
+}
+
+bool wsrep_split_allowlist(std::vector<std::string>& allowlist)
+{
+  if (!wsrep_allowlist || 0 == strlen(wsrep_allowlist))
+  {
+    return false;
+  }
+  std::istringstream ss{wsrep_allowlist};
+  std::string token;
+  while (std::getline(ss, token, ',')) 
+  {
+    if (!token.empty()) 
+    {
+      struct sockaddr_in sa_4;
+      struct sockaddr_in6 sa_6;
+      if ((inet_pton(AF_INET, token.c_str(), &(sa_4.sin_addr)) != 0) ||
+          (inet_pton(AF_INET6, token.c_str(), &(sa_6.sin6_addr)) != 0))
+      {
+        allowlist.push_back(token);
+      }
+      else
+      {
+        WSREP_WARN("Invalid IP address %s provided in `wsrep_allowlist` variable", token.c_str());
+      }
+    }
+  }
+  return allowlist.size();
+}
+
+/*!
+ * @param db      Database string
+ * @param table   Table string
+ * @param key     Array of wsrep_key_t
+ * @param key_len In: number of elements in key array, Out: number of
+ *                elements populated
+ *
+ * @return true if preparation was successful, otherwise false.
+ */
+
+static bool wsrep_prepare_key_for_isolation(const char* db,
+                                           const char* table,
+                                           wsrep_buf_t* key,
+                                           size_t* key_len)
+{
+  if (*key_len < 2) return false;
+
+  switch (wsrep_protocol_version)
+  {
+  case 0:
+    *key_len= 0;
+    break;
+  case 1:
+  case 2:
+  case 3:
+  case 4:
+  {
+    *key_len= 0;
+    if (db)
+    {
+      key[*key_len].ptr= db;
+      key[*key_len].len= strlen(db);
+      ++(*key_len);
+      if (table)
+      {
+        key[*key_len].ptr= table;
+        key[*key_len].len= strlen(table);
+        ++(*key_len);
+      }
+    }
+    break;
+  }
+  default:
+    assert(0);
+    WSREP_ERROR("Unsupported protocol version: %ld", wsrep_protocol_version);
+    unireg_abort(1);
+    return false;
+  }
+
+    return true;
+}
+
+static bool wsrep_prepare_key_for_isolation(const char* db,
+                                            const char* table,
+                                            wsrep_key_arr_t* ka)
+{
+  wsrep_key_t* tmp;
+  tmp= (wsrep_key_t*)my_realloc(PSI_INSTRUMENT_ME, ka->keys,
+                                (ka->keys_len + 1) * sizeof(wsrep_key_t),
+                                MYF(MY_ALLOW_ZERO_PTR));
+  if (!tmp)
+  {
+    WSREP_ERROR("Can't allocate memory for key_array");
+    return false;
+  }
+  ka->keys= tmp;
+  if (!(ka->keys[ka->keys_len].key_parts= (wsrep_buf_t*)
+        my_malloc(PSI_INSTRUMENT_ME, sizeof(wsrep_buf_t)*2, MYF(0))))
+  {
+    WSREP_ERROR("Can't allocate memory for key_parts");
+    return false;
+  }
+  ka->keys[ka->keys_len].key_parts_num= 2;
+  ++ka->keys_len;
+  if (!wsrep_prepare_key_for_isolation(db, table,
+                                       (wsrep_buf_t*)ka->keys[ka->keys_len - 1].key_parts,
+                                       &ka->keys[ka->keys_len - 1].key_parts_num))
+  {
+    WSREP_ERROR("Preparing keys for isolation failed");
+    return false;
+  }
+
+  return true;
+}
+
+static bool wsrep_prepare_keys_for_alter_add_fk(const char* child_table_db,
+                                                const Alter_info* alter_info,
+                                                wsrep_key_arr_t* ka)
+{
+  Key *key;
+  List_iterator<Key> key_iterator(const_cast<Alter_info*>(alter_info)->key_list);
+  while ((key= key_iterator++))
+  {
+    if (key->type == Key::FOREIGN_KEY)
+    {
+      Foreign_key *fk_key= (Foreign_key *)key;
+      const char *db_name= fk_key->ref_db.str;
+      const char *table_name= fk_key->ref_table.str;
+      if (!db_name)
+      {
+        db_name= child_table_db;
+      }
+      if (!wsrep_prepare_key_for_isolation(db_name, table_name, ka))
+      {
+        return false;
+      }
+    }
+  }
+  return true;
+}
+
+static bool wsrep_prepare_keys_for_isolation(THD*              thd,
+                                             const char*       db,
+                                             const char*       table,
+                                             const TABLE_LIST* table_list,
+                                             Alter_info*       alter_info,
+                                             wsrep_key_arr_t*  ka)
+{
+  ka->keys= 0;
+  ka->keys_len= 0;
+
+  if (db || table)
+  {
+    if (!wsrep_prepare_key_for_isolation(db, table, ka))
+      goto err;
+  }
+
+  for (const TABLE_LIST* table= table_list; table; table= table->next_global)
+  {
+    if (!wsrep_prepare_key_for_isolation(table->db.str, table->table_name.str, ka))
+      goto err;
+  }
+
+  if (alter_info)
+  {
+    if (!wsrep_prepare_keys_for_alter_add_fk(table_list->db.str, alter_info, ka))
+      goto err;
+  }
+  return false;
+
+err:
+    wsrep_keys_free(ka);
+    return true;
+}
+
+/*
+ * Prepare key list from db/table and table_list
+ *
+ * Return zero in case of success, 1 in case of failure.
+ */
+
+bool wsrep_prepare_keys_for_isolation(THD*              thd,
+                                      const char*       db,
+                                      const char*       table,
+                                      const TABLE_LIST* table_list,
+                                      wsrep_key_arr_t*  ka)
+{
+  return wsrep_prepare_keys_for_isolation(thd, db, table, table_list, NULL, ka);
+}
+
+bool wsrep_prepare_key(const uchar* cache_key, size_t cache_key_len,
+                       const uchar* row_id, size_t row_id_len,
+                       wsrep_buf_t* key, size_t* key_len)
+{
+    if (*key_len < 3) return false;
+
+    *key_len= 0;
+    switch (wsrep_protocol_version)
+    {
+    case 0:
+    {
+        key[0].ptr= cache_key;
+        key[0].len= cache_key_len;
+
+        *key_len= 1;
+        break;
+    }
+    case 1:
+    case 2:
+    case 3:
+    case 4:
+    {
+        key[0].ptr= cache_key;
+        key[0].len= strlen( (char*)cache_key );
+
+        key[1].ptr= cache_key + strlen( (char*)cache_key ) + 1;
+        key[1].len= strlen( (char*)(key[1].ptr) );
+
+        *key_len= 2;
+        break;
+    }
+    default:
+        return false;
+    }
+
+    key[*key_len].ptr= row_id;
+    key[*key_len].len= row_id_len;
+    ++(*key_len);
+
+    return true;
+}
+
+bool wsrep_prepare_key_for_innodb(THD* thd,
+                                  const uchar* cache_key,
+                                  size_t cache_key_len,
+                                  const uchar* row_id,
+                                  size_t row_id_len,
+                                  wsrep_buf_t* key,
+                                  size_t* key_len)
+{
+
+  return wsrep_prepare_key(cache_key, cache_key_len, row_id, row_id_len, key, key_len);
+}
+
+wsrep::key wsrep_prepare_key_for_toi(const char* db, const char* table,
+                                     enum wsrep::key::type type)
+{
+  wsrep::key ret(type);
+  DBUG_ASSERT(db);
+  ret.append_key_part(db, strlen(db));
+  if (table) ret.append_key_part(table, strlen(table));
+  return ret;
+}
+
+wsrep::key_array
+wsrep_prepare_keys_for_alter_add_fk(const char* child_table_db,
+                                    const Alter_info* alter_info)
+
+{
+  wsrep::key_array ret;
+  Key *key;
+  List_iterator<Key> key_iterator(const_cast<Alter_info*>(alter_info)->key_list);
+  while ((key= key_iterator++))
+  {
+    if (key->type == Key::FOREIGN_KEY)
+    {
+      Foreign_key *fk_key= (Foreign_key *)key;
+      const char *db_name= fk_key->ref_db.str;
+      const char *table_name= fk_key->ref_table.str;
+      if (!db_name)
+      {
+        db_name= child_table_db;
+      }
+      ret.push_back(wsrep_prepare_key_for_toi(db_name, table_name,
+                                              wsrep::key::exclusive));
+    }
+  }
+  return ret;
+}
+
+wsrep::key_array wsrep_prepare_keys_for_toi(const char *db,
+                                            const char *table,
+                                            const TABLE_LIST *table_list,
+                                            const Alter_info *alter_info,
+                                            const wsrep::key_array *fk_tables)
+{
+  wsrep::key_array ret;
+  if (db || table)
+  {
+    ret.push_back(wsrep_prepare_key_for_toi(db, table, wsrep::key::exclusive));
+  }
+  for (const TABLE_LIST* table= table_list; table; table= table->next_global)
+  {
+    ret.push_back(wsrep_prepare_key_for_toi(table->db.str, table->table_name.str,
+                                            wsrep::key::exclusive));
+  }
+  if (alter_info)
+  {
+    wsrep::key_array fk(wsrep_prepare_keys_for_alter_add_fk(table_list->db.str, alter_info));
+    if (!fk.empty())
+    {
+      ret.insert(ret.end(), fk.begin(), fk.end());
+    }
+  }
+  if (fk_tables && !fk_tables->empty())
+  {
+    ret.insert(ret.end(), fk_tables->begin(), fk_tables->end());
+  }
+  return ret;
+}
+
+/*
+ * Construct Query_log_Event from thd query and serialize it
+ * into buffer.
+ *
+ * Return 0 in case of success, 1 in case of error.
+ */
+int wsrep_to_buf_helper(
+    THD* thd, const char *query, uint query_len, uchar** buf, size_t* buf_len)
+{
+  IO_CACHE tmp_io_cache;
+  Log_event_writer writer(&tmp_io_cache, 0);
+  if (open_cached_file(&tmp_io_cache, mysql_tmpdir, TEMP_PREFIX,
+                       65536, MYF(MY_WME)))
+    return 1;
+  int ret(0);
+  enum enum_binlog_checksum_alg current_binlog_check_alg=
+    (enum_binlog_checksum_alg) binlog_checksum_options;
+
+  Format_description_log_event *tmp_fd= new Format_description_log_event(4);
+  tmp_fd->checksum_alg= current_binlog_check_alg;
+  writer.write(tmp_fd);
+  delete tmp_fd;
+
+#ifdef GTID_SUPPORT
+  if (thd->variables.gtid_next.type == GTID_GROUP)
+  {
+      Gtid_log_event gtid_ev(thd, FALSE, &thd->variables.gtid_next);
+      if (!gtid_ev.is_valid()) ret= 0;
+      if (!ret && writer.write(&gtid_ev)) ret= 1;
+  }
+#endif /* GTID_SUPPORT */
+  /*
+   * Check if this is applier thread, slave_thread or
+   * we have set manually WSREP GTID seqno. Add GTID event.
+   */
+  if (thd->slave_thread || wsrep_thd_is_applying(thd) || 
+      thd->variables.wsrep_gtid_seq_no)
+  {
+    uint64 seqno= thd->variables.gtid_seq_no;
+    uint32 domain_id= thd->variables.gtid_domain_id;
+    uint32 server_id= thd->variables.server_id;
+    if (!thd->variables.gtid_seq_no && thd->variables.wsrep_gtid_seq_no)
+    {
+      seqno= thd->variables.wsrep_gtid_seq_no;
+      domain_id= wsrep_gtid_server.domain_id;
+      server_id= wsrep_gtid_server.server_id;
+    }
+    Gtid_log_event gtid_event(thd, seqno, domain_id, true,
+                              LOG_EVENT_SUPPRESS_USE_F, true, 0);
+    gtid_event.server_id= server_id;
+    if (!gtid_event.is_valid()) ret= 0;
+    ret= writer.write(&gtid_event);
+  }
+  /*
+    It's local DDL so in case of possible gtid seqno (SET gtid_seq_no=X)
+    manipulation, seqno value will be ignored.
+   */
+  else
+  {
+    thd->variables.gtid_seq_no= 0;
+  }
+
+  /* if there is prepare query, add event for it */
+  if (!ret && thd->wsrep_TOI_pre_query)
+  {
+    Query_log_event ev(thd, thd->wsrep_TOI_pre_query,
+		       thd->wsrep_TOI_pre_query_len,
+		       FALSE, FALSE, FALSE, 0);
+    ev.checksum_alg= current_binlog_check_alg;
+    if (writer.write(&ev)) ret= 1;
+  }
+
+  /* continue to append the actual query */
+  Query_log_event ev(thd, query, query_len, FALSE, FALSE, FALSE, 0);
+  /* WSREP GTID mode, we need to change server_id */
+  if (wsrep_gtid_mode && !thd->variables.gtid_seq_no)
+    ev.server_id= wsrep_gtid_server.server_id;
+  ev.checksum_alg= current_binlog_check_alg;
+  if (!ret && writer.write(&ev)) ret= 1;
+  if (!ret && wsrep_write_cache_buf(&tmp_io_cache, buf, buf_len)) ret= 1;
+  close_cached_file(&tmp_io_cache);
+  return ret;
+}
+
+static int
+wsrep_alter_query_string(THD *thd, String *buf)
+{
+  /* Append the "ALTER" part of the query */
+  if (buf->append(STRING_WITH_LEN("ALTER ")))
+    return 1;
+  /* Append definer */
+  append_definer(thd, buf, &(thd->lex->definer->user), &(thd->lex->definer->host));
+  /* Append the left part of thd->query after event name part */
+  if (buf->append(thd->lex->stmt_definition_begin,
+                  thd->lex->stmt_definition_end -
+                  thd->lex->stmt_definition_begin))
+    return 1;
+
+  return 0;
+}
+
+static int wsrep_alter_event_query(THD *thd, uchar** buf, size_t* buf_len)
+{
+  String log_query;
+
+  if (wsrep_alter_query_string(thd, &log_query))
+  {
+    WSREP_WARN("events alter string failed: schema: %s, query: %s",
+               thd->get_db(), thd->query());
+    return 1;
+  }
+  return wsrep_to_buf_helper(thd, log_query.ptr(), log_query.length(), buf, buf_len);
+}
+
+#include "sql_show.h"
+static int
+create_view_query(THD *thd, uchar** buf, size_t* buf_len)
+{
+    LEX *lex= thd->lex;
+    SELECT_LEX *select_lex= lex->first_select_lex();
+    TABLE_LIST *first_table= select_lex->table_list.first;
+    TABLE_LIST *views= first_table;
+    LEX_USER *definer;
+    String buff;
+    const LEX_CSTRING command[3]=
+      {{ STRING_WITH_LEN("CREATE ") },
+       { STRING_WITH_LEN("ALTER ") },
+       { STRING_WITH_LEN("CREATE OR REPLACE ") }};
+
+    buff.append(&command[thd->lex->create_view->mode]);
+
+    if (lex->definer)
+      definer= get_current_user(thd, lex->definer);
+    else
+    {
+      /*
+        DEFINER-clause is missing; we have to create default definer in
+        persistent arena to be PS/SP friendly.
+        If this is an ALTER VIEW then the current user should be set as
+        the definer.
+      */
+      definer= create_default_definer(thd, false);
+    }
+
+    if (definer)
+    {
+      views->definer.user= definer->user;
+      views->definer.host= definer->host;
+    } else {
+      WSREP_ERROR("Failed to get DEFINER for VIEW.");
+      return 1;
+    }
+
+    views->algorithm   = lex->create_view->algorithm;
+    views->view_suid   = lex->create_view->suid;
+    views->with_check  = lex->create_view->check;
+
+    view_store_options(thd, views, &buff);
+    buff.append(STRING_WITH_LEN("VIEW "));
+    /* Test if user supplied a db (ie: we did not use thd->db) */
+    if (views->db.str && views->db.str[0] &&
+        (thd->db.str == NULL || cmp(&views->db, &thd->db)))
+    {
+      append_identifier(thd, &buff, &views->db);
+      buff.append('.');
+    }
+    append_identifier(thd, &buff, &views->table_name);
+    if (lex->view_list.elements)
+    {
+      List_iterator_fast<LEX_CSTRING> names(lex->view_list);
+      LEX_CSTRING *name;
+      int i;
+
+      buff.append('(');
+      for (i= 0; (name= names++); i++)
+      {
+        append_identifier(thd, &buff, name);
+        buff.append(", ", 2);
+      }
+      if (i)
+        buff.length(buff.length()-2);
+      buff.append(')');
+    }
+    buff.append(STRING_WITH_LEN(" AS "));
+    buff.append(thd->lex->create_view->select.str,
+                thd->lex->create_view->select.length);
+    return wsrep_to_buf_helper(thd, buff.ptr(), buff.length(), buf, buf_len);
+}
+
+/*
+  Rewrite DROP TABLE for TOI. Temporary tables are eliminated from
+  the query as they are visible only to client connection.
+
+  TODO: See comments for sql_base.cc:drop_temporary_table() and refine
+  the function to deal with transactional locked tables.
+ */
+static int wsrep_drop_table_query(THD* thd, uchar** buf, size_t* buf_len)
+{
+
+  LEX* lex= thd->lex;
+  SELECT_LEX* select_lex= lex->first_select_lex();
+  TABLE_LIST* first_table= select_lex->table_list.first;
+  String buff;
+
+  DBUG_ASSERT(!lex->create_info.tmp_table());
+
+  bool found_temp_table= false;
+  for (TABLE_LIST* table= first_table; table; table= table->next_global)
+  {
+    if (thd->find_temporary_table(table->db.str, table->table_name.str))
+    {
+      found_temp_table= true;
+      break;
+    }
+  }
+
+  if (found_temp_table)
+  {
+    buff.append(STRING_WITH_LEN("DROP TABLE "));
+    if (lex->check_exists)
+      buff.append(STRING_WITH_LEN("IF EXISTS "));
+
+    for (TABLE_LIST* table= first_table; table; table= table->next_global)
+    {
+      if (!thd->find_temporary_table(table->db.str, table->table_name.str))
+      {
+        append_identifier(thd, &buff, table->db.str, table->db.length);
+        buff.append('.');
+        append_identifier(thd, &buff,
+                          table->table_name.str, table->table_name.length);
+        buff.append(',');
+      }
+    }
+
+    /* Chop the last comma */
+    buff.chop();
+    buff.append(STRING_WITH_LEN(" /* generated by wsrep */"));
+
+    WSREP_DEBUG("Rewrote '%s' as '%s'", thd->query(), buff.ptr());
+
+    return wsrep_to_buf_helper(thd, buff.ptr(), buff.length(), buf, buf_len);
+  }
+  else
+  {
+    return wsrep_to_buf_helper(thd, thd->query(), thd->query_length(),
+                               buf, buf_len);
+  }
+}
+
+
+/* Forward declarations. */
+int wsrep_create_trigger_query(THD *thd, uchar** buf, size_t* buf_len);
+
+bool wsrep_should_replicate_ddl_iterate(THD* thd, const TABLE_LIST* table_list)
+{
+  if (WSREP(thd))
+  {
+    for (const TABLE_LIST* it= table_list; it; it= it->next_global)
+    {
+      if (it->table &&
+          !wsrep_should_replicate_ddl(thd, it->table->s->db_type()))
+        return false;
+    }
+  }
+  return true;
+}
+
+bool wsrep_should_replicate_ddl(THD* thd, const handlerton *hton)
+{
+  if (!wsrep_check_mode(WSREP_MODE_STRICT_REPLICATION))
+    return true;
+
+  if (!hton)
+    return true;
+
+  switch (hton->db_type)
+  {
+    case DB_TYPE_INNODB:
+      return true;
+      break;
+    case DB_TYPE_MYISAM:
+      if (wsrep_check_mode(WSREP_MODE_REPLICATE_MYISAM))
+        return true;
+      else
+        WSREP_DEBUG("wsrep OSU failed for %s", wsrep_thd_query(thd));
+      break;
+    case DB_TYPE_ARIA:
+      if (wsrep_check_mode(WSREP_MODE_REPLICATE_ARIA))
+	return true;
+      else
+        WSREP_DEBUG("wsrep OSU failed for %s", wsrep_thd_query(thd));
+      break;
+    default:
+      WSREP_DEBUG("wsrep OSU failed for %s", wsrep_thd_query(thd));
+      break;
+  }
+
+  /* wsrep_mode = STRICT_REPLICATION, treat as error */
+  my_error(ER_GALERA_REPLICATION_NOT_SUPPORTED, MYF(0));
+  push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                      ER_ILLEGAL_HA,
+                      "WSREP: wsrep_mode = STRICT_REPLICATION enabled. "
+                      "Storage engine %s not supported.",
+                      ha_resolve_storage_engine_name(hton));
+  return false;
+}
+/*
+  Decide if statement should run in TOI.
+
+  Look if table or table_list contain temporary tables. If the
+  statement affects only temporary tables,   statement should not run
+  in TOI. If the table list contains mix of regular and temporary tables
+  (DROP TABLE, OPTIMIZE, ANALYZE), statement should be run in TOI but
+  should be rewritten at later time for replication to contain only
+  non-temporary tables.
+ */
+bool wsrep_can_run_in_toi(THD *thd, const char *db, const char *table,
+                                 const TABLE_LIST *table_list,
+                                 const HA_CREATE_INFO* create_info)
+{
+  DBUG_ASSERT(!table || db);
+  DBUG_ASSERT(table_list || db);
+
+  LEX* lex= thd->lex;
+  SELECT_LEX* select_lex= lex->first_select_lex();
+  const TABLE_LIST* first_table= select_lex->table_list.first;
+
+  switch (lex->sql_command)
+  {
+  case SQLCOM_CREATE_TABLE:
+    if (thd->lex->create_info.options & HA_LEX_CREATE_TMP_TABLE)
+    {
+      return false;
+    }
+    if (!wsrep_should_replicate_ddl(thd, create_info->db_type))
+    {
+      return false;
+    }
+    /*
+      If mariadb master has replicated a CTAS, we should not replicate the create table
+      part separately as TOI, but to replicate both create table and following inserts
+      as one write set.
+      Howver, if CTAS creates empty table, we should replicate the create table alone
+      as TOI. We have to do relay log event lookup to see if row events follow the
+      create table event.
+    */
+    if (thd->slave_thread &&
+	!(thd->rgi_slave->gtid_ev_flags2 & Gtid_log_event::FL_STANDALONE))
+    {
+      /* this is CTAS, either empty or populated table */
+      ulonglong event_size = 0;
+      enum Log_event_type ev_type= wsrep_peak_event(thd->rgi_slave, &event_size);
+      switch (ev_type)
+      {
+      case QUERY_EVENT:
+        /* CTAS with empty table, we replicate create table as TOI */
+        break;
+
+      case TABLE_MAP_EVENT:
+        WSREP_DEBUG("replicating CTAS of empty table as TOI");
+        // fall through
+      case WRITE_ROWS_EVENT:
+        /* CTAS with populated table, we replicate later at commit time */
+        WSREP_DEBUG("skipping create table of CTAS replication");
+        return false;
+
+      default:
+        WSREP_WARN("unexpected async replication event: %d", ev_type);
+      }
+      return true;
+    }
+    /* no next async replication event */
+    return true;
+    break;
+  case SQLCOM_CREATE_VIEW:
+
+    DBUG_ASSERT(!table_list);
+    DBUG_ASSERT(first_table); /* First table is view name */
+    /*
+      If any of the remaining tables refer to temporary table error
+      is returned to client, so TOI can be skipped
+    */
+    for (const TABLE_LIST* it= first_table->next_global; it; it= it->next_global)
+    {
+      if (thd->find_temporary_table(it))
+      {
+        return false;
+      }
+    }
+    return true;
+    break;
+  case SQLCOM_CREATE_TRIGGER:
+
+    DBUG_ASSERT(first_table);
+
+    if (thd->find_temporary_table(first_table))
+    {
+      return false;
+    }
+    return true;
+    break;
+  case SQLCOM_DROP_TRIGGER:
+    DBUG_ASSERT(table_list);
+    if (thd->find_temporary_table(table_list))
+    {
+      return false;
+    }
+    return true;
+    break;
+  case SQLCOM_ALTER_TABLE:
+    if (create_info)
+    {
+      const handlerton *hton= create_info->db_type;
+
+      if (!hton)
+	hton= ha_default_handlerton(thd);
+      if (!wsrep_should_replicate_ddl(thd, hton))
+        return false;
+    }
+    /* fallthrough */
+  default:
+    if (table && !thd->find_temporary_table(db, table))
+    {
+      return true;
+    }
+
+    if (table_list)
+    {
+      for (const TABLE_LIST* table= first_table; table; table= table->next_global)
+      {
+        if (!thd->find_temporary_table(table->db.str, table->table_name.str))
+        {
+          return true;
+        }
+      }
+    }
+
+    return !(table || table_list);
+    break;
+  case SQLCOM_CREATE_SEQUENCE:
+    /* No TOI for temporary sequences as they are
+       not replicated */
+    if (thd->lex->tmp_table())
+    {
+      return false;
+    }
+    return true;
+
+  }
+}
+
+static int wsrep_create_sp(THD *thd, uchar** buf, size_t* buf_len)
+{
+  String log_query;
+  sp_head *sp= thd->lex->sphead;
+  sql_mode_t saved_mode= thd->variables.sql_mode;
+  String retstr(64);
+  LEX_CSTRING returns= empty_clex_str;
+  retstr.set_charset(system_charset_info);
+
+  log_query.set_charset(system_charset_info);
+
+  if (sp->m_handler->type() == SP_TYPE_FUNCTION)
+  {
+    sp_returns_type(thd, retstr, sp);
+    retstr.get_value(&returns);
+  }
+  if (sp->m_handler->
+      show_create_sp(thd, &log_query,
+                     sp->m_explicit_name ? sp->m_db : null_clex_str,
+                     sp->m_name, sp->m_params, returns,
+                     sp->m_body, sp->chistics(),
+                     thd->lex->definer[0],
+                     thd->lex->create_info,
+                     saved_mode))
+  {
+    WSREP_WARN("SP create string failed: schema: %s, query: %s",
+               thd->get_db(), thd->query());
+    return 1;
+  }
+
+  return wsrep_to_buf_helper(thd, log_query.ptr(), log_query.length(), buf, buf_len);
+}
+
+static int wsrep_TOI_event_buf(THD* thd, uchar** buf, size_t* buf_len)
+{
+  int err;
+  switch (thd->lex->sql_command)
+  {
+  case SQLCOM_CREATE_VIEW:
+    err= create_view_query(thd, buf, buf_len);
+    break;
+  case SQLCOM_CREATE_PROCEDURE:
+  case SQLCOM_CREATE_SPFUNCTION:
+    err= wsrep_create_sp(thd, buf, buf_len);
+    break;
+  case SQLCOM_CREATE_TRIGGER:
+    err= wsrep_create_trigger_query(thd, buf, buf_len);
+    break;
+  case SQLCOM_CREATE_EVENT:
+    err= wsrep_create_event_query(thd, buf, buf_len);
+    break;
+  case SQLCOM_ALTER_EVENT:
+    err= wsrep_alter_event_query(thd, buf, buf_len);
+    break;
+  case SQLCOM_DROP_TABLE:
+    err= wsrep_drop_table_query(thd, buf, buf_len);
+    break;
+  case SQLCOM_CREATE_ROLE:
+    if (sp_process_definer(thd))
+    {
+      WSREP_WARN("Failed to set CREATE ROLE definer for TOI.");
+    }
+    /* fallthrough */
+  default:
+    err= wsrep_to_buf_helper(thd, thd->query(), thd->query_length(), buf,
+                             buf_len);
+    break;
+  }
+
+  return err;
+}
+
+static void wsrep_TOI_begin_failed(THD* thd, const wsrep_buf_t* /* const err */)
+{
+  if (wsrep_thd_trx_seqno(thd) > 0)
+  {
+    /* GTID was granted and TO acquired - need to log event and release TO */
+    if (wsrep_emulate_bin_log) wsrep_thd_binlog_trx_reset(thd);
+    if (wsrep_write_dummy_event(thd, "TOI begin failed")) { goto fail; }
+    wsrep::client_state& cs(thd->wsrep_cs());
+    std::string const err(wsrep::to_c_string(cs.current_error()));
+    wsrep::mutable_buffer err_buf;
+    err_buf.push_back(err);
+    int const ret= cs.leave_toi_local(err_buf);
+    if (ret)
+    {
+      WSREP_ERROR("Leaving critical section for failed TOI failed: thd: %lld, "
+                  "schema: %s, SQL: %s, rcode: %d wsrep_error: %s",
+                  (long long)thd->real_id, thd->db.str,
+                  thd->query(), ret, err.c_str());
+      goto fail;
+    }
+  }
+  return;
+fail:
+  WSREP_ERROR("Failed to release TOI resources. Need to abort.");
+  unireg_abort(1);
+}
+
+
+/*
+  returns:
+   0: statement was replicated as TOI
+   1: TOI replication was skipped
+  -1: TOI replication failed
+ */
+static int wsrep_TOI_begin(THD *thd, const char *db, const char *table,
+                           const TABLE_LIST *table_list,
+                           const Alter_info *alter_info,
+                           const wsrep::key_array *fk_tables,
+                           const HA_CREATE_INFO *create_info)
+{
+  DBUG_ASSERT(wsrep_OSU_method_get(thd) == WSREP_OSU_TOI);
+
+  WSREP_DEBUG("TOI Begin: %s", wsrep_thd_query(thd));
+
+  if (wsrep_can_run_in_toi(thd, db, table, table_list, create_info) == false)
+  {
+    WSREP_DEBUG("No TOI for %s", wsrep_thd_query(thd));
+    return 1;
+  }
+
+  uchar* buf= 0;
+  size_t buf_len(0);
+  int buf_err;
+  int rc;
+
+  buf_err= wsrep_TOI_event_buf(thd, &buf, &buf_len);
+
+  if (buf_err) {
+    WSREP_ERROR("Failed to create TOI event buf: %d", buf_err);
+    my_message(ER_UNKNOWN_ERROR,
+               "WSREP replication failed to prepare TOI event buffer. "
+               "Check your query.",
+               MYF(0));
+    return -1;
+  }
+
+  struct wsrep_buf buff= { buf, buf_len };
+
+  wsrep::key_array key_array=
+    wsrep_prepare_keys_for_toi(db, table, table_list, alter_info, fk_tables);
+
+  if (thd->has_read_only_protection())
+  {
+    /* non replicated DDL, affecting temporary tables only */
+    WSREP_DEBUG("TO isolation skipped, sql: %s."
+                "Only temporary tables affected.",
+                wsrep_thd_query(thd));
+    if (buf) my_free(buf);
+    return -1;
+  }
+
+  thd_proc_info(thd, "acquiring total order isolation");
+  WSREP_DEBUG("wsrep_TOI_begin for %s", wsrep_thd_query(thd));
+  THD_STAGE_INFO(thd, stage_waiting_isolation);
+
+  wsrep::client_state& cs(thd->wsrep_cs());
+
+  int ret= cs.enter_toi_local(key_array,
+                              wsrep::const_buffer(buff.ptr, buff.len));
+
+  if (ret)
+  {
+    DBUG_ASSERT(cs.current_error());
+    WSREP_DEBUG("to_execute_start() failed for %llu: %s, seqno: %lld",
+                thd->thread_id, wsrep_thd_query(thd),
+                (long long)wsrep_thd_trx_seqno(thd));
+
+    /* jump to error handler in mysql_execute_command() */
+    switch (cs.current_error())
+    {
+    case wsrep::e_size_exceeded_error:
+      WSREP_WARN("TO isolation failed for: %d, schema: %s, sql: %s. "
+                 "Maximum size exceeded.",
+                 ret,
+                 (thd->db.str ? thd->db.str : "(null)"),
+                 wsrep_thd_query(thd));
+      my_error(ER_UNKNOWN_ERROR, MYF(0), "Maximum writeset size exceeded");
+      break;
+    case wsrep::e_deadlock_error:
+      WSREP_WARN("TO isolation failed for: %d, schema: %s, sql: %s. "
+                 "Deadlock error.",
+                 ret,
+                 (thd->db.str ? thd->db.str : "(null)"),
+                 wsrep_thd_query(thd));
+      my_error(ER_LOCK_DEADLOCK, MYF(0));
+      break;
+    case wsrep::e_timeout_error:
+      WSREP_WARN("TO isolation failed for: %d, schema: %s, sql: %s. "
+                 "Operation timed out.",
+                 ret,
+                 (thd->db.str ? thd->db.str : "(null)"),
+                 wsrep_thd_query(thd));
+      my_error(ER_LOCK_WAIT_TIMEOUT, MYF(0));
+      break;
+    default:
+      WSREP_WARN("TO isolation failed for: %d, schema: %s, sql: %s. "
+                 "Check your wsrep connection state and retry the query.",
+                 ret,
+                 (thd->db.str ? thd->db.str : "(null)"),
+                 wsrep_thd_query(thd));
+
+      if (!thd->is_error())
+      {
+        my_error(ER_LOCK_DEADLOCK, MYF(0), "WSREP replication failed. Check "
+                 "your wsrep connection state and retry the query.");
+      }
+    }
+    rc= -1;
+  }
+  else {
+    if (!thd->variables.gtid_seq_no)
+    {
+    uint64 seqno= 0;
+      if (thd->variables.wsrep_gtid_seq_no &&
+          thd->variables.wsrep_gtid_seq_no > wsrep_gtid_server.seqno())
+      {
+        seqno= thd->variables.wsrep_gtid_seq_no;
+        wsrep_gtid_server.seqno(thd->variables.wsrep_gtid_seq_no);
+      }
+      else
+      {
+        seqno= wsrep_gtid_server.seqno_inc();
+      }
+      thd->variables.wsrep_gtid_seq_no= 0;
+      thd->wsrep_current_gtid_seqno= seqno;
+      if (mysql_bin_log.is_open() && wsrep_gtid_mode)
+      {
+        thd->variables.gtid_seq_no= seqno;
+        thd->variables.gtid_domain_id= wsrep_gtid_server.domain_id;
+        thd->variables.server_id= wsrep_gtid_server.server_id;
+      }
+    }
+    ++wsrep_to_isolation;
+    rc= 0;
+  }
+
+  if (buf) my_free(buf);
+
+  if (rc) wsrep_TOI_begin_failed(thd, NULL);
+
+  return rc;
+}
+
+static void wsrep_TOI_end(THD *thd) {
+  wsrep_to_isolation--;
+  wsrep::client_state& client_state(thd->wsrep_cs());
+  DBUG_ASSERT(wsrep_thd_is_local_toi(thd));
+  WSREP_DEBUG("TO END: %lld: %s", client_state.toi_meta().seqno().get(),
+              wsrep_thd_query(thd));
+
+  wsrep_gtid_server.signal_waiters(thd->wsrep_current_gtid_seqno, false);
+
+  if (wsrep_thd_is_local_toi(thd))
+  {
+    wsrep::mutable_buffer err;
+
+    thd->wsrep_last_written_gtid_seqno= thd->wsrep_current_gtid_seqno;
+    wsrep_set_SE_checkpoint(client_state.toi_meta().gtid(), wsrep_gtid_server.gtid());
+
+    if (thd->is_error() && !wsrep_must_ignore_error(thd))
+    {
+      wsrep_store_error(thd, err);
+    }
+
+    int const ret= client_state.leave_toi_local(err);
+
+    if (!ret)
+    {
+      WSREP_DEBUG("TO END: %lld", client_state.toi_meta().seqno().get());
+    }
+    else
+    {
+      WSREP_WARN("TO isolation end failed for: %d, schema: %s, sql: %s",
+                 ret, (thd->db.str ? thd->db.str : "(null)"), wsrep_thd_query(thd));
+    }
+  }
+}
+
+static int wsrep_RSU_begin(THD *thd, const char *db_, const char *table_)
+{
+  WSREP_DEBUG("RSU BEGIN: %lld, : %s", wsrep_thd_trx_seqno(thd),
+              wsrep_thd_query(thd));
+
+  /* For CREATE TEMPORARY SEQUENCE we do not start RSU because
+     object is local only and actually CREATE TABLE + INSERT
+  */
+  if (thd->lex->sql_command == SQLCOM_CREATE_SEQUENCE &&
+      thd->lex->tmp_table())
+    return 1;
+
+  if (thd->variables.wsrep_OSU_method == WSREP_OSU_RSU &&
+      thd->variables.sql_log_bin == 1 &&
+      wsrep_check_mode(WSREP_MODE_DISALLOW_LOCAL_GTID))
+  {
+    /* wsrep_mode = WSREP_MODE_DISALLOW_LOCAL_GTID, treat as error */
+    my_error(ER_GALERA_REPLICATION_NOT_SUPPORTED, MYF(0));
+    push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                        ER_OPTION_PREVENTS_STATEMENT,
+                        "You can't execute statements that would generate local "
+                        "GTIDs when wsrep_mode = DISALLOW_LOCAL_GTID is set. "
+                        "Try disabling binary logging with SET sql_log_bin=0 "
+                        "to execute this statement.");
+
+    return -1;
+  }
+
+  if (thd->wsrep_cs().begin_rsu(5000))
+  {
+    WSREP_WARN("RSU begin failed");
+  }
+  else
+  {
+    thd->variables.wsrep_on= 0;
+  }
+  return 0;
+}
+
+static void wsrep_RSU_end(THD *thd)
+{
+  WSREP_DEBUG("RSU END: %lld : %s", wsrep_thd_trx_seqno(thd),
+              wsrep_thd_query(thd));
+  if (thd->wsrep_cs().end_rsu())
+  {
+    WSREP_WARN("Failed to end RSU, server may need to be restarted");
+  }
+  thd->variables.wsrep_on= 1;
+}
+
+static inline bool is_replaying_connection(THD *thd)
+{
+  bool ret;
+
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+  ret=  (thd->wsrep_trx().state() == wsrep::transaction::s_replaying) ? true : false;
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+
+  return ret;
+}
+
+int wsrep_to_isolation_begin(THD *thd, const char *db_, const char *table_,
+                             const TABLE_LIST* table_list,
+                             const Alter_info *alter_info,
+                             const wsrep::key_array *fk_tables,
+                             const HA_CREATE_INFO *create_info)
+{
+  /*
+    No isolation for applier or replaying threads.
+   */
+  if (!wsrep_thd_is_local(thd))
+  {
+    if (wsrep_OSU_method_get(thd) == WSREP_OSU_TOI)
+      WSREP_DEBUG("%s TOI Begin: %s",
+                  is_replaying_connection(thd) ? "Replay" : "Apply",
+                  wsrep_thd_query(thd));
+
+    return 0;
+  }
+
+  if (thd->wsrep_parallel_slave_wait_for_prior_commit())
+  {
+    WSREP_WARN("TOI: wait_for_prior_commit() returned error.");
+    return -1;
+  }
+
+  int ret= 0;
+
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+
+  if (thd->wsrep_trx().state() == wsrep::transaction::s_must_abort)
+  {
+    WSREP_INFO("thread: %lld  schema: %s  query: %s has been aborted due to multi-master conflict",
+               (longlong) thd->thread_id, thd->get_db(), thd->query());
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
+    return WSREP_TRX_FAIL;
+  }
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+
+  DBUG_ASSERT(wsrep_thd_is_local(thd));
+  DBUG_ASSERT(thd->wsrep_trx().ws_meta().seqno().is_undefined());
+
+  if (Wsrep_server_state::instance().desynced_on_pause())
+  {
+    my_message(ER_UNKNOWN_COM_ERROR,
+               "Aborting TOI: Replication paused on node for FTWRL/BACKUP STAGE.", MYF(0));
+    WSREP_DEBUG("Aborting TOI: Replication paused on node for FTWRL/BACKUP STAGE.: %s %llu",
+                wsrep_thd_query(thd), thd->thread_id);
+    return -1;
+  }
+
+  /* If we are inside LOCK TABLE we release it and give warning. */
+  if (thd->variables.option_bits & OPTION_TABLE_LOCK &&
+      thd->lex->sql_command == SQLCOM_CREATE_SEQUENCE)
+  {
+    thd->locked_tables_list.unlock_locked_tables(thd);
+    thd->variables.option_bits&= ~(OPTION_TABLE_LOCK);
+    push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+			HA_ERR_UNSUPPORTED,
+			"Galera cluster does not support LOCK TABLE on "
+			"SEQUENCES. Lock is released.");
+  }
+  if (wsrep_debug && thd->mdl_context.has_locks())
+  {
+    WSREP_DEBUG("thread holds MDL locks at TO begin: %s %llu",
+                wsrep_thd_query(thd), thd->thread_id);
+  }
+
+  /*
+    It makes sense to set auto_increment_* to defaults in TOI operations.
+    Must be done before wsrep_TOI_begin() since Query_log_event encapsulating
+    TOI statement and auto inc variables for wsrep replication is constructed
+    there. Variables are reset back in THD::reset_for_next_command() before
+    processing of next command.
+   */
+  if (wsrep_auto_increment_control)
+  {
+    thd->variables.auto_increment_offset= 1;
+    thd->variables.auto_increment_increment= 1;
+  }
+
+  if (thd->variables.wsrep_on && wsrep_thd_is_local(thd))
+  {
+    switch (wsrep_OSU_method_get(thd)) {
+    case WSREP_OSU_TOI:
+      ret= wsrep_TOI_begin(thd, db_, table_, table_list, alter_info, fk_tables,
+                           create_info);
+      break;
+    case WSREP_OSU_RSU:
+      ret= wsrep_RSU_begin(thd, db_, table_);
+      break;
+    default:
+      WSREP_ERROR("Unsupported OSU method: %lu",
+                  wsrep_OSU_method_get(thd));
+      ret= -1;
+      break;
+    }
+
+    switch (ret) {
+    case 0: /* wsrep_TOI_begin should set toi mode */
+      if (thd->variables.wsrep_OSU_method == WSREP_OSU_TOI)
+      {
+        /*
+          TOI operations ignore the provided lock_wait_timeout once replicated,
+          and restore it after operation is done.
+         */
+        thd->variables.saved_lock_wait_timeout= thd->variables.lock_wait_timeout;
+        thd->variables.lock_wait_timeout= LONG_TIMEOUT;
+      }
+      break;
+    case 1:
+        /* TOI replication skipped, treat as success */
+        ret= 0;
+      break;
+    case -1:
+      /* TOI replication failed, treat as error */
+      break;
+    }
+  }
+
+  return ret;
+}
+
+void wsrep_to_isolation_end(THD *thd)
+{
+  if (wsrep_thd_is_local_toi(thd))
+  {
+    thd->variables.lock_wait_timeout= thd->variables.saved_lock_wait_timeout;
+    DBUG_ASSERT(wsrep_OSU_method_get(thd) == WSREP_OSU_TOI);
+    wsrep_TOI_end(thd);
+  }
+  else if (wsrep_thd_is_in_rsu(thd))
+  {
+    thd->variables.lock_wait_timeout= thd->variables.saved_lock_wait_timeout;
+    DBUG_ASSERT(wsrep_OSU_method_get(thd) == WSREP_OSU_RSU);
+    wsrep_RSU_end(thd);
+  }
+  else
+  {
+   /* Applier or replaying threads just output TO END */
+   if (wsrep_debug)
+   {
+     wsrep::client_state& client_state(thd->wsrep_cs());
+     WSREP_DEBUG("%s TO END: %lld: %s",
+                 is_replaying_connection(thd) ? "Replay" : "Apply",
+                 client_state.toi_meta().seqno().get(),
+                 wsrep_thd_query(thd));
+   }
+   return;
+  }
+
+  if (wsrep_emulate_bin_log) wsrep_thd_binlog_trx_reset(thd);
+}
+
+#define WSREP_MDL_LOG(severity, msg, schema, schema_len, req, gra)             \
+    WSREP_##severity(                                                          \
+      "%s\n"                                                                   \
+      "schema:  %.*s\n"                                                        \
+      "request: (%llu \tseqno %lld \twsrep (%s, %s, %s) cmd %d %d \t%s)\n"     \
+      "granted: (%llu \tseqno %lld \twsrep (%s, %s, %s) cmd %d %d \t%s)",      \
+      msg, schema_len, schema,                                                 \
+      req->thread_id, (long long)wsrep_thd_trx_seqno(req),                     \
+      wsrep_thd_client_mode_str(req), wsrep_thd_client_state_str(req), wsrep_thd_transaction_state_str(req), \
+      req->get_command(), req->lex->sql_command, req->query(),                 \
+      gra->thread_id, (long long)wsrep_thd_trx_seqno(gra),                     \
+      wsrep_thd_client_mode_str(gra), wsrep_thd_client_state_str(gra), wsrep_thd_transaction_state_str(gra), \
+      gra->get_command(), gra->lex->sql_command, gra->query());
+
+/**
+  Check if request for the metadata lock should be granted to the requester.
+
+  @param  requestor_ctx        The MDL context of the requestor
+  @param  ticket               MDL ticket for the requested lock
+
+  @retval TRUE   Lock request can be granted
+  @retval FALSE  Lock request cannot be granted
+*/
+
+void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
+                               const MDL_ticket *ticket,
+                               const MDL_key *key)
+{
+  THD *request_thd= requestor_ctx->get_thd();
+  THD *granted_thd= ticket->get_ctx()->get_thd();
+
+  /* Fallback to the non-wsrep behaviour */
+  if (!WSREP(request_thd)) return;
+
+  const char* schema= key->db_name();
+  int schema_len= key->db_name_length();
+
+  mysql_mutex_lock(&request_thd->LOCK_thd_data);
+
+  if (wsrep_thd_is_toi(request_thd) ||
+      wsrep_thd_is_applying(request_thd))
+  {
+    WSREP_DEBUG("wsrep_handle_mdl_conflict request TOI/APPLY for %s",
+                wsrep_thd_query(request_thd));
+    THD_STAGE_INFO(request_thd, stage_waiting_isolation);
+    mysql_mutex_unlock(&request_thd->LOCK_thd_data);
+    WSREP_MDL_LOG(DEBUG, "MDL conflict ", schema, schema_len,
+                  request_thd, granted_thd);
+    ticket->wsrep_report(wsrep_debug);
+
+    DEBUG_SYNC(request_thd, "before_wsrep_thd_abort");
+    DBUG_EXECUTE_IF("sync.before_wsrep_thd_abort", {
+      const char act[]= "now "
+                        "SIGNAL sync.before_wsrep_thd_abort_reached "
+                        "WAIT_FOR signal.before_wsrep_thd_abort";
+      DBUG_ASSERT(!debug_sync_set_action(request_thd, STRING_WITH_LEN(act)));
+    };);
+
+    /* Here we will call wsrep_abort_transaction so we should hold
+    THD::LOCK_thd_data to protect victim from concurrent usage
+    and THD::LOCK_thd_kill to protect from disconnect or delete.
+
+    */
+    mysql_mutex_lock(&granted_thd->LOCK_thd_kill);
+    mysql_mutex_lock(&granted_thd->LOCK_thd_data);
+
+    if (wsrep_thd_is_toi(granted_thd) ||
+        wsrep_thd_is_applying(granted_thd))
+    {
+      if (wsrep_thd_is_aborting(granted_thd))
+      {
+        WSREP_DEBUG("BF thread waiting for SR in aborting state for %s",
+                    wsrep_thd_query(request_thd));
+        THD_STAGE_INFO(request_thd, stage_waiting_isolation);
+        ticket->wsrep_report(wsrep_debug);
+      }
+      else if (wsrep_thd_is_SR(granted_thd) && !wsrep_thd_is_SR(request_thd))
+      {
+        WSREP_MDL_LOG(INFO, "MDL conflict, DDL vs SR",
+                      schema, schema_len, request_thd, granted_thd);
+        WSREP_DEBUG("wsrep_handle_mdl_conflict DDL vs SR for %s",
+                    wsrep_thd_query(request_thd));
+        THD_STAGE_INFO(request_thd, stage_waiting_isolation);
+        wsrep_abort_thd(request_thd, granted_thd, 1);
+      }
+      else
+      {
+        WSREP_MDL_LOG(INFO, "MDL BF-BF conflict", schema, schema_len,
+                      request_thd, granted_thd);
+        ticket->wsrep_report(true);
+        mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+        mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
+        unireg_abort(1);
+      }
+    }
+    else if (granted_thd->lex->sql_command == SQLCOM_FLUSH ||
+             granted_thd->mdl_context.has_explicit_locks())
+    {
+      WSREP_DEBUG("BF thread waiting for FLUSH for %s",
+                  wsrep_thd_query(request_thd));
+      THD_STAGE_INFO(request_thd, stage_waiting_ddl);
+      ticket->wsrep_report(wsrep_debug);
+      if (granted_thd->current_backup_stage != BACKUP_FINISHED &&
+	  wsrep_check_mode(WSREP_MODE_BF_MARIABACKUP))
+      {
+	wsrep_abort_thd(request_thd, granted_thd, 1);
+      }
+    }
+    else if (request_thd->lex->sql_command == SQLCOM_DROP_TABLE)
+    {
+      WSREP_DEBUG("DROP caused BF abort, conf %s for %s",
+                  wsrep_thd_transaction_state_str(granted_thd),
+                  wsrep_thd_query(request_thd));
+      THD_STAGE_INFO(request_thd, stage_waiting_isolation);
+      ticket->wsrep_report(wsrep_debug);
+      wsrep_abort_thd(request_thd, granted_thd, 1);
+    }
+    else
+    {
+      WSREP_MDL_LOG(DEBUG, "MDL conflict-> BF abort", schema, schema_len,
+                    request_thd, granted_thd);
+      WSREP_DEBUG("wsrep_handle_mdl_conflict -> BF abort for %s",
+                  wsrep_thd_query(request_thd));
+      THD_STAGE_INFO(request_thd, stage_waiting_isolation);
+      ticket->wsrep_report(wsrep_debug);
+
+      if (granted_thd->wsrep_trx().active())
+      {
+        wsrep_abort_thd(request_thd, granted_thd, 1);
+      }
+      else
+      {
+        /*
+          Granted_thd is likely executing with wsrep_on=0. If the requesting
+          thd is BF, BF abort and wait.
+        */
+        if (wsrep_thd_is_BF(request_thd, FALSE))
+        {
+          granted_thd->awake_no_mutex(KILL_QUERY_HARD);
+          ha_abort_transaction(request_thd, granted_thd, TRUE);
+        }
+        else
+        {
+	  WSREP_MDL_LOG(INFO, "MDL unknown BF-BF conflict",
+                        schema, schema_len,
+                        request_thd, granted_thd);
+	  ticket->wsrep_report(true);
+          mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+          mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
+	  unireg_abort(1);
+        }
+      }
+    }
+    mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+    mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
+  }
+  else
+  {
+    mysql_mutex_unlock(&request_thd->LOCK_thd_data);
+  }
+}
+
+/**/
+static bool abort_replicated(THD *thd)
+{
+  bool ret_code= false;
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+  if (thd->wsrep_trx().state() == wsrep::transaction::s_committing)
+  {
+    WSREP_DEBUG("aborting replicated trx: %llu", (ulonglong)(thd->real_id));
+
+    wsrep_abort_thd(thd, thd, TRUE);
+    ret_code= true;
+  }
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
+  return ret_code;
+}
+
+/**/
+static inline bool is_client_connection(THD *thd)
+{
+  return (thd->wsrep_client_thread && thd->variables.wsrep_on);
+}
+
+static inline bool is_committing_connection(THD *thd)
+{
+  bool ret;
+
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+  ret=  (thd->wsrep_trx().state() == wsrep::transaction::s_committing) ? true : false;
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+
+  return ret;
+}
+
+static my_bool have_client_connections(THD *thd, void*)
+{
+  DBUG_PRINT("quit",("Informing thread %lld that it's time to die",
+                     (longlong) thd->thread_id));
+  if (is_client_connection(thd) &&
+      (thd->killed == KILL_CONNECTION ||
+       thd->killed == KILL_CONNECTION_HARD))
+  {
+    (void)abort_replicated(thd);
+    return 1;
+  }
+  return 0;
+}
+
+static void wsrep_close_thread(THD *thd)
+{
+  thd->set_killed(KILL_CONNECTION_HARD);
+  MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (thd));
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
+  thd->abort_current_cond_wait(true);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
+}
+
+static my_bool have_committing_connections(THD *thd, void *)
+{
+  return is_client_connection(thd) && is_committing_connection(thd) ? 1 : 0;
+}
+
+int wsrep_wait_committing_connections_close(int wait_time)
+{
+  int sleep_time= 100;
+
+  WSREP_DEBUG("wait for committing transaction to close: %d sleep: %d", wait_time, sleep_time);
+  while (server_threads.iterate(have_committing_connections) && wait_time > 0)
+  {
+    WSREP_DEBUG("wait for committing transaction to close: %d", wait_time);
+    my_sleep(sleep_time);
+    wait_time -= sleep_time;
+  }
+  return server_threads.iterate(have_committing_connections);
+}
+
+static my_bool kill_all_threads(THD *thd, THD *caller_thd)
+{
+  DBUG_PRINT("quit", ("Informing thread %lld that it's time to die",
+                      (longlong) thd->thread_id));
+  /* We skip slave threads & scheduler on this first loop through. */
+  if (is_client_connection(thd) && thd != caller_thd)
+  {
+    if (is_replaying_connection(thd))
+      thd->set_killed(KILL_CONNECTION_HARD);
+    else if (!abort_replicated(thd))
+    {
+      /* replicated transactions must be skipped */
+      WSREP_DEBUG("closing connection %lld", (longlong) thd->thread_id);
+      /* instead of wsrep_close_thread() we do now  soft kill by THD::awake */
+      thd->awake(KILL_CONNECTION_HARD);
+    }
+  }
+  return 0;
+}
+
+static my_bool kill_remaining_threads(THD *thd, THD *caller_thd)
+{
+#ifndef __bsdi__				// Bug in BSDI kernel
+  if (is_client_connection(thd) &&
+      !abort_replicated(thd)    &&
+      !is_replaying_connection(thd) &&
+      thd_is_connection_alive(thd) &&
+      thd != caller_thd)
+  {
+
+    WSREP_INFO("killing local connection: %lld", (longlong) thd->thread_id);
+    close_connection(thd);
+  }
+#endif
+  return 0;
+}
+
+void wsrep_close_client_connections(my_bool wait_to_end, THD* except_caller_thd)
+{
+  /* Clear thread cache */
+  thread_cache.final_flush();
+
+  /*
+    First signal all threads that it's time to die
+  */
+  server_threads.iterate(kill_all_threads, except_caller_thd);
+
+  /*
+    Force remaining threads to die by closing the connection to the client
+  */
+  server_threads.iterate(kill_remaining_threads, except_caller_thd);
+
+  DBUG_PRINT("quit", ("Waiting for threads to die (count=%u)", THD_count::value()));
+  WSREP_DEBUG("waiting for client connections to close: %u", THD_count::value());
+
+  while (wait_to_end && server_threads.iterate(have_client_connections))
+  {
+    sleep(1);
+    DBUG_PRINT("quit",("One thread died (count=%u)", THD_count::value()));
+  }
+
+  /* All client connection threads have now been aborted */
+}
+
+
+void wsrep_close_applier(THD *thd)
+{
+  WSREP_DEBUG("closing applier %lld", (longlong) thd->thread_id);
+  wsrep_close_thread(thd);
+}
+
+static my_bool wsrep_close_threads_callback(THD *thd, THD *caller_thd)
+{
+  DBUG_PRINT("quit",("Informing thread %lld that it's time to die",
+                     (longlong) thd->thread_id));
+  /* We skip slave threads & scheduler on this first loop through. */
+  if (thd->wsrep_applier && thd != caller_thd)
+  {
+    WSREP_DEBUG("closing wsrep thread %lld", (longlong) thd->thread_id);
+    wsrep_close_thread(thd);
+  }
+  return 0;
+}
+
+void wsrep_close_threads(THD *thd)
+{
+  server_threads.iterate(wsrep_close_threads_callback, thd);
+}
+
+void wsrep_wait_appliers_close(THD *thd)
+{
+  /* Wait for wsrep appliers to gracefully exit */
+  mysql_mutex_lock(&LOCK_wsrep_slave_threads);
+  while (wsrep_running_threads > 2)
+    /*
+      2 is for rollbacker thread which needs to be killed explicitly.
+      This gotta be fixed in a more elegant manner if we gonna have arbitrary
+      number of non-applier wsrep threads.
+    */
+  {
+    mysql_cond_wait(&COND_wsrep_slave_threads, &LOCK_wsrep_slave_threads);
+  }
+  mysql_mutex_unlock(&LOCK_wsrep_slave_threads);
+  DBUG_PRINT("quit",("applier threads have died (count=%u)",
+                     uint32_t(wsrep_running_threads)));
+
+  /* Now kill remaining wsrep threads: rollbacker */
+  wsrep_close_threads (thd);
+  /* and wait for them to die */
+  mysql_mutex_lock(&LOCK_wsrep_slave_threads);
+  while (wsrep_running_threads > 0)
+  {
+    mysql_cond_wait(&COND_wsrep_slave_threads, &LOCK_wsrep_slave_threads);
+  }
+  mysql_mutex_unlock(&LOCK_wsrep_slave_threads);
+  DBUG_PRINT("quit",("all wsrep system threads have died"));
+
+  /* All wsrep applier threads have now been aborted. However, if this thread
+     is also applier, we are still running...
+  */
+}
+int wsrep_must_ignore_error(THD* thd)
+{
+  const int error= thd->get_stmt_da()->sql_errno();
+  const uint flags= sql_command_flags[thd->lex->sql_command];
+
+  DBUG_ASSERT(error);
+  DBUG_ASSERT(wsrep_thd_is_toi(thd));
+
+  if ((wsrep_ignore_apply_errors & WSREP_IGNORE_ERRORS_ON_DDL))
+    goto ignore_error;
+
+  if ((flags & CF_WSREP_MAY_IGNORE_ERRORS) &&
+      (wsrep_ignore_apply_errors & WSREP_IGNORE_ERRORS_ON_RECONCILING_DDL))
+  {
+    switch (error)
+    {
+    case ER_DB_DROP_EXISTS:
+    case ER_BAD_TABLE_ERROR:
+    case ER_CANT_DROP_FIELD_OR_KEY:
+      goto ignore_error;
+    }
+  }
+
+  return 0;
+
+ignore_error:
+  WSREP_WARN("Ignoring error '%s' on query. "
+             "Default database: '%s'. Query: '%s', Error_code: %d",
+             thd->get_stmt_da()->message(),
+             print_slave_db_safe(thd->db.str),
+             thd->query(),
+             error);
+  return 1;
+}
+
+int wsrep_ignored_error_code(Log_event* ev, int error)
+{
+  const THD* thd= ev->thd;
+
+  DBUG_ASSERT(error);
+  /* Note that binlog events can be executed on master also with
+     BINLOG '....'; */
+  DBUG_ASSERT(!wsrep_thd_is_local_toi(thd));
+
+  if ((wsrep_ignore_apply_errors & WSREP_IGNORE_ERRORS_ON_RECONCILING_DML))
+  {
+    const int ev_type= ev->get_type_code();
+    if ((ev_type == DELETE_ROWS_EVENT || ev_type == DELETE_ROWS_EVENT_V1)
+        && error == ER_KEY_NOT_FOUND)
+      goto ignore_error;
+  }
+
+  return 0;
+
+ignore_error:
+  WSREP_WARN("Ignoring error '%s' on %s event. Error_code: %d",
+             thd->get_stmt_da()->message(),
+             ev->get_type_str(),
+             error);
+  return 1;
+}
+
+bool wsrep_provider_is_SR_capable()
+{
+  return Wsrep_server_state::has_capability(wsrep::provider::capability::streaming);
+}
+
+int wsrep_thd_retry_counter(const THD *thd)
+{
+  return thd->wsrep_retry_counter;
+}
+
+extern  bool wsrep_thd_ignore_table(THD *thd)
+{
+  return thd->wsrep_ignore_table;
+}
+
+bool wsrep_create_like_table(THD* thd, TABLE_LIST* table,
+                             TABLE_LIST* src_table,
+                             HA_CREATE_INFO *create_info)
+{
+  if (create_info->tmp_table())
+  {
+    /* CREATE TEMPORARY TABLE LIKE must be skipped from replication */
+    WSREP_DEBUG("CREATE TEMPORARY TABLE LIKE... skipped replication\n %s",
+                thd->query());
+  }
+  else if (!(thd->find_temporary_table(src_table)))
+  {
+    /* this is straight CREATE TABLE LIKE... with no tmp tables */
+    WSREP_TO_ISOLATION_BEGIN_CREATE(table->db.str, table->table_name.str, table, create_info);
+  }
+  else
+  {
+    /* Non-MERGE tables ignore this call. */
+    if (src_table->table->file->extra(HA_EXTRA_ADD_CHILDREN_LIST))
+      return (true);
+
+    char buf[2048];
+    String query(buf, sizeof(buf), system_charset_info);
+    query.length(0);  // Have to zero it since constructor doesn't
+
+    int result __attribute__((unused))=
+      show_create_table(thd, src_table, &query, NULL, WITH_DB_NAME);
+    WSREP_DEBUG("TMP TABLE: %s ret_code %d", query.ptr(), result);
+
+    thd->wsrep_TOI_pre_query=     query.ptr();
+    thd->wsrep_TOI_pre_query_len= query.length();
+
+    WSREP_TO_ISOLATION_BEGIN_CREATE(table->db.str, table->table_name.str, table, create_info);
+
+    thd->wsrep_TOI_pre_query=      NULL;
+    thd->wsrep_TOI_pre_query_len= 0;
+
+    /* Non-MERGE tables ignore this call. */
+    src_table->table->file->extra(HA_EXTRA_DETACH_CHILDREN);
+  }
+
+  return(false);
+#ifdef WITH_WSREP
+wsrep_error_label:
+  thd->wsrep_TOI_pre_query= NULL;
+  return (true);
+#endif
+}
+
+int wsrep_create_trigger_query(THD *thd, uchar** buf, size_t* buf_len)
+{
+  LEX *lex= thd->lex;
+  String stmt_query;
+
+  LEX_CSTRING definer_user;
+  LEX_CSTRING definer_host;
+
+  if (!lex->definer)
+  {
+    if (!thd->slave_thread)
+    {
+      if (!(lex->definer= create_default_definer(thd, false)))
+        return 1;
+    }
+  }
+
+  if (lex->definer)
+  {
+    /* SUID trigger. */
+    LEX_USER *d= get_current_user(thd, lex->definer);
+
+    if (!d)
+      return 1;
+
+    definer_user= d->user;
+    definer_host= d->host;
+  }
+  else
+  {
+    /* non-SUID trigger. */
+
+    definer_user.str= 0;
+    definer_user.length= 0;
+
+    definer_host.str= 0;
+    definer_host.length= 0;
+  }
+
+  const LEX_CSTRING command[2]=
+      {{ C_STRING_WITH_LEN("CREATE ") },
+       { C_STRING_WITH_LEN("CREATE OR REPLACE ") }};
+
+  if (thd->lex->create_info.or_replace())
+    stmt_query.append(command[1]);
+  else
+    stmt_query.append(command[0]);
+
+  append_definer(thd, &stmt_query, &definer_user, &definer_host);
+
+  LEX_CSTRING stmt_definition;
+  stmt_definition.str= (char*) thd->lex->stmt_definition_begin;
+  stmt_definition.length= thd->lex->stmt_definition_end
+    - thd->lex->stmt_definition_begin;
+  trim_whitespace(thd->charset(), &stmt_definition);
+
+  stmt_query.append(stmt_definition.str, stmt_definition.length);
+
+  return wsrep_to_buf_helper(thd, stmt_query.c_ptr(), stmt_query.length(),
+                             buf, buf_len);
+}
+
+void* start_wsrep_THD(void *arg)
+{
+  THD *thd= NULL;
+
+  Wsrep_thd_args* thd_args= (Wsrep_thd_args*) arg;
+
+  if (my_thread_init() || (!(thd= new THD(next_thread_id(), true))))
+  {
+    goto error;
+  }
+
+  statistic_increment(thread_created, &LOCK_status);
+
+  thd->real_id=pthread_self(); // Keep purify happy
+
+  my_net_init(&thd->net,(st_vio*) 0, thd, MYF(0));
+
+  DBUG_PRINT("wsrep",(("creating thread %lld"), (long long)thd->thread_id));
+  thd->prior_thr_create_utime= thd->start_utime= microsecond_interval_timer();
+
+  server_threads.insert(thd);
+
+  /* from bootstrap()... */
+  thd->bootstrap=1;
+  thd->max_client_packet_length= thd->net.max_packet;
+  thd->security_ctx->master_access= ALL_KNOWN_ACL;
+
+  /* from handle_one_connection... */
+  pthread_detach_this_thread();
+
+  mysql_thread_set_psi_id(thd->thread_id);
+  thd->thr_create_utime=  microsecond_interval_timer();
+
+  DBUG_EXECUTE_IF("wsrep_simulate_failed_connection_1", goto error; );
+// </5.1.17>
+  /*
+    handle_one_connection() is normally the only way a thread would
+    start and would always be on the very high end of the stack ,
+    therefore, the thread stack always starts at the address of the
+    first local variable of handle_one_connection, which is thd. We
+    need to know the start of the stack so that we could check for
+    stack overruns.
+  */
+  DBUG_PRINT("wsrep", ("handle_one_connection called by thread %lld",
+                       (long long)thd->thread_id));
+  /* now that we've called my_thread_init(), it is safe to call DBUG_* */
+
+  thd->thread_stack= (char*) &thd;
+  wsrep_assign_from_threadvars(thd);
+  wsrep_store_threadvars(thd);
+
+  thd->system_thread= SYSTEM_THREAD_SLAVE_SQL;
+  thd->security_ctx->skip_grants();
+
+  /* handle_one_connection() again... */
+  thd->proc_info= 0;
+  thd->set_command(COM_SLEEP);
+  thd->init_for_queries();
+  mysql_mutex_lock(&LOCK_wsrep_slave_threads);
+
+  wsrep_running_threads++;
+
+  switch (thd_args->thread_type()) {
+    case WSREP_APPLIER_THREAD:
+      wsrep_running_applier_threads++;
+      break;
+    case WSREP_ROLLBACKER_THREAD:
+      wsrep_running_rollbacker_threads++;
+      break;
+    default:
+      WSREP_ERROR("Incorrect wsrep thread type: %d", thd_args->thread_type());
+      break;
+  }
+
+  mysql_cond_broadcast(&COND_wsrep_slave_threads);
+  mysql_mutex_unlock(&LOCK_wsrep_slave_threads);
+
+  WSREP_DEBUG("wsrep system thread %llu, %p starting",
+              thd->thread_id, thd);
+  thd_args->fun()(thd, static_cast<void *>(thd_args));
+
+  WSREP_DEBUG("wsrep system thread: %llu, %p closing",
+              thd->thread_id, thd);
+
+  /* Wsrep may reset globals during thread context switches, store globals
+     before cleanup. */
+  wsrep_store_threadvars(thd);
+
+  close_connection(thd, 0);
+
+  mysql_mutex_lock(&LOCK_wsrep_slave_threads);
+  DBUG_ASSERT(wsrep_running_threads > 0);
+  wsrep_running_threads--;
+
+  switch (thd_args->thread_type()) {
+    case WSREP_APPLIER_THREAD:
+      DBUG_ASSERT(wsrep_running_applier_threads > 0);
+      wsrep_running_applier_threads--;
+      break;
+    case WSREP_ROLLBACKER_THREAD:
+      DBUG_ASSERT(wsrep_running_rollbacker_threads > 0);
+      wsrep_running_rollbacker_threads--;
+      break;
+    default:
+      WSREP_ERROR("Incorrect wsrep thread type: %d", thd_args->thread_type());
+      break;
+  }
+
+  delete thd_args;
+  WSREP_DEBUG("wsrep running threads now: %lu", wsrep_running_threads);
+  mysql_cond_broadcast(&COND_wsrep_slave_threads);
+  mysql_mutex_unlock(&LOCK_wsrep_slave_threads);
+  /*
+    Note: We can't call THD destructor without crashing
+    if plugins have not been initialized. However, in most of the
+    cases this means that pre SE initialization SST failed and
+    we are going to exit anyway.
+  */
+  if (plugins_are_initialized)
+  {
+    net_end(&thd->net);
+    unlink_thd(thd);
+  }
+  else
+  {
+    /*
+      TODO: lightweight cleanup to get rid of:
+      'Error in my_thread_global_end(): 2 threads didn't exit'
+      at server shutdown
+    */
+    server_threads.erase(thd);
+  }
+
+  delete thd;
+  my_thread_end();
+  return(NULL);
+
+error:
+  WSREP_ERROR("Failed to create/initialize system thread");
+
+  if (thd)
+  {
+    close_connection(thd, ER_OUT_OF_RESOURCES);
+    statistic_increment(aborted_connects, &LOCK_status);
+    server_threads.erase(thd);
+    delete thd;
+    my_thread_end();
+  }
+  delete thd_args;
+  // This will signal error to wsrep_slave_threads_update
+  wsrep_thread_create_failed.store(true, std::memory_order_relaxed);
+
+  /* Abort if its the first applier/rollbacker thread. */
+  if (!mysqld_server_initialized)
+    unireg_abort(1);
+  else
+    return NULL;
+}
+
+enum wsrep::streaming_context::fragment_unit wsrep_fragment_unit(ulong unit)
+{
+  switch (unit)
+  {
+  case WSREP_FRAG_BYTES: return wsrep::streaming_context::bytes;
+  case WSREP_FRAG_ROWS: return wsrep::streaming_context::row;
+  case WSREP_FRAG_STATEMENTS: return wsrep::streaming_context::statement;
+  default:
+    DBUG_ASSERT(0);
+    return wsrep::streaming_context::bytes;
+  }
+}
+
+bool THD::wsrep_parallel_slave_wait_for_prior_commit()
+{
+  if (rgi_slave && rgi_slave->is_parallel_exec && wait_for_prior_commit())
+  {
+    return 1;
+  }
+  return 0;
+}
+
+/***** callbacks for wsrep service ************/
+
+my_bool get_wsrep_recovery()
+{
+  return wsrep_recovery;
+}
+
+bool wsrep_consistency_check(THD *thd)
+{
+  return thd->wsrep_consistency_check == CONSISTENCY_CHECK_RUNNING;
+}
+
+// Wait until wsrep has reached ready state
+void wsrep_wait_ready(THD *thd)
+{
+  mysql_mutex_lock(&LOCK_wsrep_ready);
+  while(!wsrep_ready)
+  {
+    WSREP_INFO("Waiting to reach ready state");
+    mysql_cond_wait(&COND_wsrep_ready, &LOCK_wsrep_ready);
+  }
+  WSREP_INFO("ready state reached");
+  mysql_mutex_unlock(&LOCK_wsrep_ready);
+}
+
+void wsrep_ready_set(bool ready_value)
+{
+  WSREP_DEBUG("Setting wsrep_ready to %d", ready_value);
+  mysql_mutex_lock(&LOCK_wsrep_ready);
+  wsrep_ready= ready_value;
+  // Signal if we have reached ready state
+  if (wsrep_ready)
+    mysql_cond_signal(&COND_wsrep_ready);
+  mysql_mutex_unlock(&LOCK_wsrep_ready);
+}
+
+
+/*
+  Commit an empty transaction.
+
+  If the transaction is real and the wsrep transaction is still active,
+  the transaction did not generate any rows or keys and is committed
+  as empty. Here the wsrep transaction is rolled back and after statement
+  step is performed to leave the wsrep transaction in the state as it
+  never existed.
+
+  This should not be an inline functions as it requires a lot of stack space
+  because of WSREP_DBUG() usage.  It's also not a function that is
+  frequently called.
+*/
+
+void wsrep_commit_empty(THD* thd, bool all)
+{
+  DBUG_ENTER("wsrep_commit_empty");
+  WSREP_DEBUG("wsrep_commit_empty for %llu client_state %s client_mode"
+              " %s trans_state %s sql %s",
+              thd_get_thread_id(thd),
+              wsrep::to_c_string(thd->wsrep_cs().state()),
+              wsrep::to_c_string(thd->wsrep_cs().mode()),
+              wsrep::to_c_string(thd->wsrep_cs().transaction().state()),
+              wsrep_thd_query(thd));
+
+  if (wsrep_is_real(thd, all) &&
+      wsrep_thd_is_local(thd) &&
+      thd->wsrep_trx().active() &&
+      !thd->internal_transaction() &&
+      thd->wsrep_trx().state() != wsrep::transaction::s_committed)
+  {
+    /* Here transaction is either empty (i.e. no changes) or
+       it was CREATE TABLE with no row binlog format or
+       we have already aborted transaction e.g. because max writeset size
+       has been reached. */
+    DBUG_ASSERT(!wsrep_has_changes(thd) ||
+                (thd->lex->sql_command == SQLCOM_CREATE_TABLE &&
+                 !thd->is_current_stmt_binlog_format_row()) ||
+                thd->wsrep_cs().transaction().state() == wsrep::transaction::s_aborted);
+    bool have_error= wsrep_current_error(thd);
+    int ret= wsrep_before_rollback(thd, all) ||
+      wsrep_after_rollback(thd, all) ||
+      wsrep_after_statement(thd);
+    /* The committing transaction was empty but it held some locks and
+       got BF aborted. As there were no certified changes in the
+       data, we ignore the deadlock error and rely on error reporting
+       by storage engine/server. */
+    if (!ret && !have_error && wsrep_current_error(thd))
+    {
+      DBUG_ASSERT(wsrep_current_error(thd) == wsrep::e_deadlock_error);
+      thd->wsrep_cs().reset_error();
+    }
+    if (ret)
+    {
+      WSREP_DEBUG("wsrep_commit_empty failed: %d", wsrep_current_error(thd));
+    }
+  }
+  DBUG_VOID_RETURN;
+}