Merge 10.5 into 10.6

60 files changed, 614 insertions, 756 deletions

diff --git a/sql/create_options.cc b/sql/create_options.cc
index 04469120db1..60e9b733efc 100644
--- a/sql/create_options.cc
+++ b/sql/create_options.cc
@@ -97,14 +97,13 @@ static bool report_unknown_option(THD *thd, engine_option_value *val,
 {
   DBUG_ENTER("report_unknown_option");
 
-  if (val->parsed || suppress_warning)
+  if (val->parsed || suppress_warning || thd->slave_thread)
   {
     DBUG_PRINT("info", ("parsed => exiting"));
     DBUG_RETURN(FALSE);
   }
 
-  if (!(thd->variables.sql_mode & MODE_IGNORE_BAD_TABLE_OPTIONS) &&
-      !thd->slave_thread)
+  if (!(thd->variables.sql_mode & MODE_IGNORE_BAD_TABLE_OPTIONS))
   {
     my_error(ER_UNKNOWN_OPTION, MYF(0), val->name.str);
     DBUG_RETURN(TRUE);
diff --git a/sql/field.cc b/sql/field.cc
index 45b3a3c703e..52074417046 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -11159,7 +11159,7 @@ key_map Field::get_possible_keys()
 
 bool Field::validate_value_in_record_with_warn(THD *thd, const uchar *record)
 {
-  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->read_set);
+  MY_BITMAP *old_map= dbug_tmp_use_all_columns(table, &table->read_set);
   bool rc;
   if ((rc= validate_value_in_record(thd, record)))
   {
@@ -11171,7 +11171,7 @@ bool Field::validate_value_in_record_with_warn(THD *thd, const uchar *record)
                         ER_THD(thd, ER_INVALID_DEFAULT_VALUE_FOR_FIELD),
                         ErrConvString(&tmp).ptr(), field_name.str);
   }
-  dbug_tmp_restore_column_map(table->read_set, old_map);
+  dbug_tmp_restore_column_map(&table->read_set, old_map);
   return rc;
 }
 
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 7157874a702..7527015682f 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -4418,7 +4418,7 @@ int ha_partition::write_row(const uchar * buf)
   int error;
   longlong func_value;
   bool have_auto_increment= table->next_number_field && buf == table->record[0];
-  my_bitmap_map *old_map;
+  MY_BITMAP *old_map;
   THD *thd= ha_thd();
   Sql_mode_save sms(thd);
   bool saved_auto_inc_field_not_null= table->auto_increment_field_not_null;
@@ -4459,9 +4459,9 @@ int ha_partition::write_row(const uchar * buf)
       thd->variables.sql_mode|= MODE_NO_AUTO_VALUE_ON_ZERO;
     }
   }
-  old_map= dbug_tmp_use_all_columns(table, table->read_set);
+  old_map= dbug_tmp_use_all_columns(table, &table->read_set);
   error= m_part_info->get_partition_id(m_part_info, &part_id, &func_value);
-  dbug_tmp_restore_column_map(table->read_set, old_map);
+  dbug_tmp_restore_column_map(&table->read_set, old_map);
   if (unlikely(error))
   {
     m_part_info->err_value= func_value;
@@ -11425,13 +11425,12 @@ int ha_partition::bulk_update_row(const uchar *old_data, const uchar *new_data,
   int error= 0;
   uint32 part_id;
   longlong func_value;
-  my_bitmap_map *old_map;
   DBUG_ENTER("ha_partition::bulk_update_row");
 
-  old_map= dbug_tmp_use_all_columns(table, table->read_set);
+  MY_BITMAP *old_map= dbug_tmp_use_all_columns(table, &table->read_set);
   error= m_part_info->get_partition_id(m_part_info, &part_id,
                                        &func_value);
-  dbug_tmp_restore_column_map(table->read_set, old_map);
+  dbug_tmp_restore_column_map(&table->read_set, old_map);
   if (unlikely(error))
   {
     m_part_info->err_value= func_value;
diff --git a/sql/handler.cc b/sql/handler.cc
index 228c016e082..751a9cd0a97 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -913,6 +913,7 @@ static my_bool kill_handlerton(THD *thd, plugin_ref plugin,
 {
   handlerton *hton= plugin_hton(plugin);
 
+  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
   if (hton->kill_query && thd_get_ha_data(thd, hton))
     hton->kill_query(hton, thd, *(enum thd_kill_levels *) level);
   return FALSE;
@@ -5515,6 +5516,7 @@ int ha_create_table(THD *thd, const char *path,
   char name_buff[FN_REFLEN];
   const char *name;
   TABLE_SHARE share;
+  Abort_on_warning_instant_set old_abort_on_warning(thd, 0);
   bool temp_table __attribute__((unused)) =
     create_info->options & (HA_LEX_CREATE_TMP_TABLE | HA_CREATE_TMP_ALTER);
   DBUG_ENTER("ha_create_table");
diff --git a/sql/handler.h b/sql/handler.h
index 44bfc3b4edd..777a6b455ef 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1529,7 +1529,7 @@ struct handlerton
    enum handler_create_iterator_result
      (*create_iterator)(handlerton *hton, enum handler_iterator_type type,
                         struct handler_iterator *fill_this_in);
-   int (*abort_transaction)(handlerton *hton, THD *bf_thd,
+   void (*abort_transaction)(handlerton *hton, THD *bf_thd,
 			    THD *victim_thd, my_bool signal);
    int (*set_checkpoint)(handlerton *hton, const XID* xid);
    int (*get_checkpoint)(handlerton *hton, XID* xid);
@@ -3329,7 +3329,7 @@ public:
   {
     cached_table_flags= table_flags();
   }
-  /* ha_ methods: pubilc wrappers for private virtual API */
+  /* ha_ methods: public wrappers for private virtual API */
   
   int ha_open(TABLE *table, const char *name, int mode, uint test_if_locked,
               MEM_ROOT *mem_root= 0, List<String> *partitions_to_open=NULL);
diff --git a/sql/item.cc b/sql/item.cc
index 93dab6a90a7..a21bd490272 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1446,9 +1446,9 @@ int Item::save_in_field_no_warnings(Field *field, bool no_conversions)
   Sql_mode_save sms(thd);
   thd->variables.sql_mode&= ~(MODE_NO_ZERO_IN_DATE | MODE_NO_ZERO_DATE);
   thd->variables.sql_mode|= MODE_INVALID_DATES;
-  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->write_set);
+  MY_BITMAP *old_map= dbug_tmp_use_all_columns(table, &table->write_set);
   res= save_in_field(field, no_conversions);
-  dbug_tmp_restore_column_map(table->write_set, old_map);
+  dbug_tmp_restore_column_map(&table->write_set, old_map);
   return res;
 }
 
@@ -5825,7 +5825,7 @@ bool Item_field::fix_fields(THD *thd, Item **reference)
   Field *from_field= (Field *)not_found_field;
   bool outer_fixed= false;
   SELECT_LEX *select= thd->lex->current_select;
-  
+
   if (select && select->in_tvc)
   {
     my_error(ER_FIELD_REFERENCE_IN_TVC, MYF(0), full_name());
@@ -6728,7 +6728,7 @@ Item *Item_string::make_odbc_literal(THD *thd, const LEX_CSTRING *typestr)
 }
 
 
-static int save_int_value_in_field (Field *field, longlong nr, 
+static int save_int_value_in_field (Field *field, longlong nr,
                                     bool null_value, bool unsigned_flag)
 {
   if (null_value)
@@ -8501,6 +8501,22 @@ bool Item_direct_ref::val_native(THD *thd, Native *to)
 }
 
 
+longlong Item_direct_ref::val_time_packed(THD *thd)
+{
+  longlong tmp = (*ref)->val_time_packed(thd);
+  null_value= (*ref)->null_value;
+  return tmp;
+}
+
+
+longlong Item_direct_ref::val_datetime_packed(THD *thd)
+{
+  longlong tmp = (*ref)->val_datetime_packed(thd);
+  null_value= (*ref)->null_value;
+  return tmp;
+}
+
+
 Item_cache_wrapper::~Item_cache_wrapper()
 {
   DBUG_ASSERT(expr_cache == 0);
diff --git a/sql/item.h b/sql/item.h
index 677112e448c..03c5ad06fa9 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -5580,14 +5580,17 @@ public:
     return Item_ref::fix_fields(thd, it);
   }
   void save_val(Field *to);
+  /* Below we should have all val() methods as in Item_ref */
   double val_real();
   longlong val_int();
-  String *val_str(String* tmp);
-  bool val_native(THD *thd, Native *to);
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
+  String *val_str(String* tmp);
+  bool val_native(THD *thd, Native *to);
   bool is_null();
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
+  longlong val_datetime_packed(THD *);
+  longlong val_time_packed(THD *);
   virtual Ref_Type ref_type() { return DIRECT_REF; }
   Item *get_copy(THD *thd)
   { return get_item_copy<Item_direct_ref>(thd, this); }
@@ -5916,6 +5919,20 @@ public:
     }
     return Item_direct_ref::get_date(thd, ltime, fuzzydate);
   }
+  longlong val_time_packed(THD *thd)
+  {
+    if (check_null_ref())
+      return 0;
+    else
+      return Item_direct_ref::val_time_packed(thd);
+  }
+  longlong val_datetime_packed(THD *thd)
+  {
+    if (check_null_ref())
+      return 0;
+    else
+      return Item_direct_ref::val_datetime_packed(thd);
+  }
   bool send(Protocol *protocol, st_value *buffer);
   void save_org_in_field(Field *field,
                          fast_field_copier data __attribute__ ((__unused__)))
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index c93f4761fc9..ae3b96f8b8a 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -323,13 +323,13 @@ static bool convert_const_to_int(THD *thd, Item_field *field_item,
     TABLE *table= field->table;
     Sql_mode_save sql_mode(thd);
     Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);
-    my_bitmap_map *old_maps[2] = { NULL, NULL };
+    MY_BITMAP *old_maps[2] = { NULL, NULL };
     ulonglong UNINIT_VAR(orig_field_val); /* original field value if valid */
 
     /* table->read_set may not be set if we come here from a CREATE TABLE */
     if (table && table->read_set)
       dbug_tmp_use_all_columns(table, old_maps,
-                               table->read_set, table->write_set);
+                               &table->read_set, &table->write_set);
     /* For comparison purposes allow invalid dates like 2000-01-32 */
     thd->variables.sql_mode= (thd->variables.sql_mode & ~MODE_NO_ZERO_DATE) |
                              MODE_INVALID_DATES;
@@ -370,7 +370,7 @@ static bool convert_const_to_int(THD *thd, Item_field *field_item,
       DBUG_ASSERT(!result);
     }
     if (table && table->read_set)
-      dbug_tmp_restore_column_maps(table->read_set, table->write_set, old_maps);
+      dbug_tmp_restore_column_maps(&table->read_set, &table->write_set, old_maps);
   }
   return result;
 }
@@ -3211,7 +3211,7 @@ bool Item_func_decode_oracle::fix_length_and_dec()
 /*
   Aggregate all THEN and ELSE expression types
   and collations when string result
-  
+
   @param THD       - current thd
   @param start     - an element in args to start aggregating from
 */
diff --git a/sql/key.cc b/sql/key.cc
index 8701d7b8053..53f897de15f 100644
--- a/sql/key.cc
+++ b/sql/key.cc
@@ -245,14 +245,13 @@ void key_restore(uchar *to_record, const uchar *from_key, KEY *key_info,
     else if (key_part->key_part_flag & HA_VAR_LENGTH_PART)
     {
       Field *field= key_part->field;
-      my_bitmap_map *old_map;
       my_ptrdiff_t ptrdiff= to_record - field->table->record[0];
       field->move_field_offset(ptrdiff);
       key_length-= HA_KEY_BLOB_LENGTH;
       length= MY_MIN(key_length, key_part->length);
-      old_map= dbug_tmp_use_all_columns(field->table, field->table->write_set);
+      MY_BITMAP *old_map= dbug_tmp_use_all_columns(field->table, &field->table->write_set);
       field->set_key_image(from_key, length);
-      dbug_tmp_restore_column_map(field->table->write_set, old_map);
+      dbug_tmp_restore_column_map(&field->table->write_set, old_map);
       from_key+= HA_KEY_BLOB_LENGTH;
       field->move_field_offset(-ptrdiff);
     }
@@ -418,7 +417,7 @@ void field_unpack(String *to, Field *field, const uchar *rec, uint max_length,
 
 void key_unpack(String *to, TABLE *table, KEY *key)
 {
-  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->read_set);
+  MY_BITMAP *old_map= dbug_tmp_use_all_columns(table, &table->read_set);
   DBUG_ENTER("key_unpack");
 
   to->length(0);
@@ -442,7 +441,7 @@ void key_unpack(String *to, TABLE *table, KEY *key)
     field_unpack(to, key_part->field, table->record[0], key_part->length,
                  MY_TEST(key_part->key_part_flag & HA_PART_KEY_SEG));
  }
-  dbug_tmp_restore_column_map(table->read_set, old_map);
+  dbug_tmp_restore_column_map(&table->read_set, old_map);
   DBUG_VOID_RETURN;
 }
 
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 3a03ff1dcb7..b35ed1884cb 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3408,7 +3408,6 @@ int Rows_log_event::get_data_size()
                   return (int)(6 + no_bytes_in_map(&m_cols) + (end - buf) +
                   (general_type_code == UPDATE_ROWS_EVENT ? no_bytes_in_map(&m_cols_ai) : 0) +
                   m_rows_cur - m_rows_buf););
-
   int data_size= 0;
   Log_event_type type = get_type_code();
   bool is_v2_event= LOG_EVENT_IS_ROW_V2(type);
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index 5e71f04dbed..55ad50511ec 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -3912,6 +3912,19 @@ int Xid_apply_log_event::do_record_gtid(THD *thd, rpl_group_info *rgi,
   return err;
 }
 
+static bool wsrep_must_replay(THD *thd)
+{
+#ifdef WITH_WSREP
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+  bool res= WSREP(thd) && thd->wsrep_trx().state() == wsrep::transaction::s_must_replay;
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  return res;
+#else
+  return false;
+#endif
+}
+
+
 int Xid_apply_log_event::do_apply_event(rpl_group_info *rgi)
 {
   bool res;
@@ -3970,21 +3983,13 @@ int Xid_apply_log_event::do_apply_event(rpl_group_info *rgi)
       return err;
   }
 
-#ifdef WITH_WSREP
-  if (WSREP(thd)) mysql_mutex_lock(&thd->LOCK_thd_data);
-  if ((!res || (WSREP(thd) && thd->wsrep_trx().state() == wsrep::transaction::s_must_replay )) && sub_id)
-#else
-  if (likely(!res) && sub_id)
-#endif /* WITH_WSREP */
+  if (sub_id && (!res || wsrep_must_replay(thd)))
     rpl_global_gtid_slave_state->update_state_hash(sub_id, &gtid, hton, rgi);
-#ifdef WITH_WSREP
-  if (WSREP(thd)) mysql_mutex_unlock(&thd->LOCK_thd_data);
-#endif /* WITH_WSREP */
   /*
     Increment the global status commit count variable
   */
-  enum enum_sql_command cmd= !thd->transaction->xid_state.is_explicit_XA() ?
-    SQLCOM_COMMIT : SQLCOM_XA_PREPARE;
+  enum enum_sql_command cmd= !thd->transaction->xid_state.is_explicit_XA()
+    ? SQLCOM_COMMIT : SQLCOM_XA_PREPARE;
   status_var_increment(thd->status_var.com_stat[cmd]);
 
   return res;
@@ -7460,11 +7465,11 @@ int Rows_log_event::update_sequence()
     /* This event come from a setval function executed on the master.
        Update the sequence next_number and round, like we do with setval()
     */
-    my_bitmap_map *old_map= dbug_tmp_use_all_columns(table,
-                                                     table->read_set);
+    MY_BITMAP *old_map= dbug_tmp_use_all_columns(table,
+                                                 &table->read_set);
     longlong nextval= table->field[NEXT_FIELD_NO]->val_int();
     longlong round= table->field[ROUND_FIELD_NO]->val_int();
-    dbug_tmp_restore_column_map(table->read_set, old_map);
+    dbug_tmp_restore_column_map(&table->read_set, old_map);
 
     return table->s->sequence->set_value(table, nextval, round, 0) > 0;
   }
diff --git a/sql/mdl.cc b/sql/mdl.cc
index 5f5e5c3ce9a..2b556eb436c 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -1157,7 +1157,7 @@ MDL_wait::timed_wait(MDL_context_owner *owner, struct timespec *abs_timeout,
                    DBUG_ASSERT(!debug_sync_set_action((owner->get_thd()),
                                                       STRING_WITH_LEN(act)));
                  };);
-    if (wsrep_thd_is_BF(owner->get_thd(), false))
+    if (WSREP_ON && wsrep_thd_is_BF(owner->get_thd(), false))
     {
       wait_result= mysql_cond_wait(&m_COND_wait_status, &m_LOCK_wait_status);
     }
@@ -1210,7 +1210,7 @@ void MDL_lock::Ticket_list::add_ticket(MDL_ticket *ticket)
   */
   DBUG_ASSERT(ticket->get_lock());
 #ifdef WITH_WSREP
-  if ((this == &(ticket->get_lock()->m_waiting)) &&
+  if (WSREP_ON && (this == &(ticket->get_lock()->m_waiting)) &&
       wsrep_thd_is_BF(ticket->get_ctx()->get_thd(), false))
   {
     DBUG_ASSERT(WSREP(ticket->get_ctx()->get_thd()));
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 0bf21e02002..02f69e1d512 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -348,7 +348,6 @@ static char compiled_default_collation_name[]= MYSQL_DEFAULT_COLLATION_NAME;
 Thread_cache thread_cache;
 static bool binlog_format_used= false;
 LEX_STRING opt_init_connect, opt_init_slave;
-mysql_cond_t COND_slave_background;
 static DYNAMIC_ARRAY all_options;
 static longlong start_memory_used;
 
@@ -689,7 +688,7 @@ mysql_mutex_t
   LOCK_crypt,
   LOCK_global_system_variables,
   LOCK_user_conn,
-  LOCK_error_messages, LOCK_slave_background;
+  LOCK_error_messages;
 mysql_mutex_t LOCK_stats, LOCK_global_user_client_stats,
               LOCK_global_table_stats, LOCK_global_index_stats;
 
@@ -908,8 +907,7 @@ PSI_mutex_key key_LOCK_stats,
 PSI_mutex_key key_LOCK_gtid_waiting;
 
 PSI_mutex_key key_LOCK_after_binlog_sync;
-PSI_mutex_key key_LOCK_prepare_ordered, key_LOCK_commit_ordered,
-  key_LOCK_slave_background;
+PSI_mutex_key key_LOCK_prepare_ordered, key_LOCK_commit_ordered;
 PSI_mutex_key key_TABLE_SHARE_LOCK_share;
 PSI_mutex_key key_LOCK_ack_receiver;
 
@@ -982,7 +980,6 @@ static PSI_mutex_info all_server_mutexes[]=
   { &key_LOCK_prepare_ordered, "LOCK_prepare_ordered", PSI_FLAG_GLOBAL},
   { &key_LOCK_after_binlog_sync, "LOCK_after_binlog_sync", PSI_FLAG_GLOBAL},
   { &key_LOCK_commit_ordered, "LOCK_commit_ordered", PSI_FLAG_GLOBAL},
-  { &key_LOCK_slave_background, "LOCK_slave_background", PSI_FLAG_GLOBAL},
   { &key_PARTITION_LOCK_auto_inc, "HA_DATA_PARTITION::LOCK_auto_inc", 0},
   { &key_LOCK_slave_state, "LOCK_slave_state", 0},
   { &key_LOCK_start_thread, "LOCK_start_thread", PSI_FLAG_GLOBAL},
@@ -1051,7 +1048,7 @@ PSI_cond_key key_TC_LOG_MMAP_COND_queue_busy;
 PSI_cond_key key_COND_rpl_thread_queue, key_COND_rpl_thread,
   key_COND_rpl_thread_stop, key_COND_rpl_thread_pool,
   key_COND_parallel_entry, key_COND_group_commit_orderer,
-  key_COND_prepare_ordered, key_COND_slave_background;
+  key_COND_prepare_ordered;
 PSI_cond_key key_COND_wait_gtid, key_COND_gtid_ignore_duplicates;
 PSI_cond_key key_COND_ack_receiver;
 
@@ -1097,7 +1094,6 @@ static PSI_cond_info all_server_conds[]=
   { &key_COND_parallel_entry, "COND_parallel_entry", 0},
   { &key_COND_group_commit_orderer, "COND_group_commit_orderer", 0},
   { &key_COND_prepare_ordered, "COND_prepare_ordered", 0},
-  { &key_COND_slave_background, "COND_slave_background", 0},
   { &key_COND_start_thread, "COND_start_thread", PSI_FLAG_GLOBAL},
   { &key_COND_wait_gtid, "COND_wait_gtid", 0},
   { &key_COND_gtid_ignore_duplicates, "COND_gtid_ignore_duplicates", 0},
@@ -1514,31 +1510,9 @@ static void end_ssl();
 /* common callee of two shutdown phases */
 static void kill_thread(THD *thd)
 {
-  if (WSREP(thd)) mysql_mutex_lock(&thd->LOCK_thd_data);
   mysql_mutex_lock(&thd->LOCK_thd_kill);
-  if (thd->mysys_var)
-  {
-    thd->mysys_var->abort= 1;
-    mysql_mutex_lock(&thd->mysys_var->mutex);
-    if (thd->mysys_var->current_cond)
-    {
-      for (uint i= 0; i < 2; i++)
-      {
-        int ret= mysql_mutex_trylock(thd->mysys_var->current_mutex);
-        mysql_cond_broadcast(thd->mysys_var->current_cond);
-        if (!ret)
-        {
-          /* Thread has surely got the signal, unlock and abort */
-          mysql_mutex_unlock(thd->mysys_var->current_mutex);
-          break;
-        }
-        sleep(1);
-      }
-    }
-    mysql_mutex_unlock(&thd->mysys_var->mutex);
-  }
+  thd->abort_current_cond_wait(true);
   mysql_mutex_unlock(&thd->LOCK_thd_kill);
-  if (WSREP(thd)) mysql_mutex_unlock(&thd->LOCK_thd_data);
 }
 
 
@@ -1892,6 +1866,7 @@ extern "C" void unireg_abort(int exit_code)
     wsrep_deinit(true);
     wsrep_deinit_server();
   }
+  wsrep_sst_auth_free();
 #endif // WITH_WSREP
 
   clean_up(!opt_abort && (exit_code || !opt_bootstrap)); /* purecov: inspected */
@@ -2124,8 +2099,6 @@ static void clean_up_mutexes()
   mysql_cond_destroy(&COND_prepare_ordered);
   mysql_mutex_destroy(&LOCK_after_binlog_sync);
   mysql_mutex_destroy(&LOCK_commit_ordered);
-  mysql_mutex_destroy(&LOCK_slave_background);
-  mysql_cond_destroy(&COND_slave_background);
 #ifndef EMBEDDED_LIBRARY
   mysql_mutex_destroy(&LOCK_error_log);
 #endif
@@ -4261,9 +4234,6 @@ static int init_thread_environment()
                    MY_MUTEX_INIT_SLOW);
   mysql_mutex_init(key_LOCK_commit_ordered, &LOCK_commit_ordered,
                    MY_MUTEX_INIT_SLOW);
-  mysql_mutex_init(key_LOCK_slave_background, &LOCK_slave_background,
-                   MY_MUTEX_INIT_SLOW);
-  mysql_cond_init(key_COND_slave_background, &COND_slave_background, NULL);
 
 #ifdef HAVE_OPENSSL
   mysql_mutex_init(key_LOCK_des_key_file,
@@ -5066,6 +5036,10 @@ static int init_server_components()
       that there are unprocessed options.
     */
     my_getopt_skip_unknown= 0;
+#ifdef WITH_WSREP
+    if (wsrep_recovery)
+      my_getopt_skip_unknown= TRUE;
+#endif
 
     if ((ho_error= handle_options(&remaining_argc, &remaining_argv, removed_opts,
                                   mysqld_get_one_option)))
@@ -5075,20 +5049,27 @@ static int init_server_components()
     remaining_argv--;
     my_getopt_skip_unknown= TRUE;
 
-    if (remaining_argc > 1)
+#ifdef WITH_WSREP
+    if (!wsrep_recovery)
     {
-      fprintf(stderr, "%s: Too many arguments (first extra is '%s').\n",
-              my_progname, remaining_argv[1]);
-      unireg_abort(1);
+#endif
+      if (remaining_argc > 1)
+      {
+        fprintf(stderr, "%s: Too many arguments (first extra is '%s').\n",
+                my_progname, remaining_argv[1]);
+        unireg_abort(1);
+      }
+#ifdef WITH_WSREP
     }
+#endif
   }
 
-  if (init_io_cache_encryption())
-    unireg_abort(1);
-
   if (opt_abort)
     unireg_abort(0);
 
+  if (init_io_cache_encryption())
+    unireg_abort(1);
+
   /* if the errmsg.sys is not loaded, terminate to maintain behaviour */
   if (!DEFAULT_ERRMSGS[0][0])
     unireg_abort(1);  
@@ -5615,9 +5596,12 @@ int mysqld_main(int argc, char **argv)
         wsrep_init_startup(false);
       }
       wsrep_new_cluster= false;
-      WSREP_DEBUG("Startup creating %ld applier threads running %lu",
-	      wsrep_slave_threads - 1, wsrep_running_applier_threads);
-      wsrep_create_appliers(wsrep_slave_threads - 1);
+      if (wsrep_cluster_address_exists())
+      {
+        WSREP_DEBUG("Startup creating %ld applier threads running %lu",
+                wsrep_slave_threads - 1, wsrep_running_applier_threads);
+        wsrep_create_appliers(wsrep_slave_threads - 1);
+      }
     }
   }
 #endif /* WITH_WSREP */
diff --git a/sql/mysqld.h b/sql/mysqld.h
index cd05a66fff3..2a73e518ded 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -742,8 +742,7 @@ extern mysql_mutex_t
        LOCK_error_log, LOCK_delayed_insert, LOCK_short_uuid_generator,
        LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone,
        LOCK_active_mi, LOCK_manager, LOCK_user_conn,
-       LOCK_prepared_stmt_count, LOCK_error_messages,
-       LOCK_slave_background;
+       LOCK_prepared_stmt_count, LOCK_error_messages;
 extern MYSQL_PLUGIN_IMPORT mysql_mutex_t LOCK_global_system_variables;
 extern mysql_rwlock_t LOCK_all_status_vars;
 extern mysql_mutex_t LOCK_start_thread;
@@ -758,7 +757,6 @@ extern mysql_rwlock_t LOCK_ssl_refresh;
 extern mysql_prlock_t LOCK_system_variables_hash;
 extern mysql_cond_t COND_start_thread;
 extern mysql_cond_t COND_manager;
-extern mysql_cond_t COND_slave_background;
 extern Atomic_counter<uint32_t> thread_count;
 
 extern my_bool opt_use_ssl;
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 28a8809beec..3684db40242 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -3670,8 +3670,7 @@ bool calculate_cond_selectivity_for_table(THD *thd, TABLE *table, Item **cond)
 
 void store_key_image_to_rec(Field *field, uchar *ptr, uint len)
 {
-  /* Do the same as print_key() does */ 
-  my_bitmap_map *old_map;
+  /* Do the same as print_key() does */
 
   if (field->real_maybe_null())
   {
@@ -3683,10 +3682,10 @@ void store_key_image_to_rec(Field *field, uchar *ptr, uint len)
     field->set_notnull();
     ptr++;
   }    
-  old_map= dbug_tmp_use_all_columns(field->table,
-                                    field->table->write_set);
+  MY_BITMAP *old_map= dbug_tmp_use_all_columns(field->table,
+                                    &field->table->write_set);
   field->set_key_image(ptr, len); 
-  dbug_tmp_restore_column_map(field->table->write_set, old_map);
+  dbug_tmp_restore_column_map(&field->table->write_set, old_map);
 }
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
@@ -3901,7 +3900,7 @@ bool prune_partitions(THD *thd, TABLE *table, Item *pprune_cond)
   PART_PRUNE_PARAM prune_param;
   MEM_ROOT alloc;
   RANGE_OPT_PARAM  *range_par= &prune_param.range_param;
-  my_bitmap_map *old_sets[2];
+  MY_BITMAP *old_sets[2];
 
   prune_param.part_info= part_info;
   init_sql_alloc(key_memory_quick_range_select_root, &alloc,
@@ -3917,7 +3916,7 @@ bool prune_partitions(THD *thd, TABLE *table, Item *pprune_cond)
   }
 
   dbug_tmp_use_all_columns(table, old_sets, 
-                           table->read_set, table->write_set);
+                           &table->read_set, &table->write_set);
   range_par->thd= thd;
   range_par->table= table;
   /* range_par->cond doesn't need initialization */
@@ -4014,7 +4013,7 @@ all_used:
   retval= FALSE; // some partitions are used
   mark_all_partitions_as_used(prune_param.part_info);
 end:
-  dbug_tmp_restore_column_maps(table->read_set, table->write_set, old_sets);
+  dbug_tmp_restore_column_maps(&table->read_set, &table->write_set, old_sets);
   thd->no_errors=0;
   thd->mem_root= range_par->old_root;
   free_root(&alloc,MYF(0));			// Return memory & allocator
@@ -15992,8 +15991,8 @@ static void
 print_sel_arg_key(Field *field, const uchar *key, String *out)
 {
   TABLE *table= field->table;
-  my_bitmap_map *old_sets[2];
-  dbug_tmp_use_all_columns(table, old_sets, table->read_set, table->write_set);
+  MY_BITMAP *old_sets[2];
+  dbug_tmp_use_all_columns(table, old_sets, &table->read_set, &table->write_set);
 
   if (field->real_maybe_null())
   {
@@ -16013,7 +16012,7 @@ print_sel_arg_key(Field *field, const uchar *key, String *out)
     field->val_str(out);
 
 end:
-  dbug_tmp_restore_column_maps(table->read_set, table->write_set, old_sets);
+  dbug_tmp_restore_column_maps(&table->read_set, &table->write_set, old_sets);
 }
 
 
@@ -16108,9 +16107,9 @@ print_key(KEY_PART *key_part, const uchar *key, uint used_length)
   const uchar *key_end= key+used_length;
   uint store_length;
   TABLE *table= key_part->field->table;
-  my_bitmap_map *old_sets[2];
+  MY_BITMAP *old_sets[2];
 
-  dbug_tmp_use_all_columns(table, old_sets, table->read_set, table->write_set);
+  dbug_tmp_use_all_columns(table, old_sets, &table->read_set, &table->write_set);
 
   for (; key < key_end; key+=store_length, key_part++)
   {
@@ -16137,7 +16136,7 @@ print_key(KEY_PART *key_part, const uchar *key, uint used_length)
     if (key+store_length < key_end)
       fputc('/',DBUG_FILE);
   }
-  dbug_tmp_restore_column_maps(table->read_set, table->write_set, old_sets);
+  dbug_tmp_restore_column_maps(&table->read_set, &table->write_set, old_sets);
 }
 
 
@@ -16145,16 +16144,16 @@ static void print_quick(QUICK_SELECT_I *quick, const key_map *needed_reg)
 {
   char buf[MAX_KEY/8+1];
   TABLE *table;
-  my_bitmap_map *old_sets[2];
+  MY_BITMAP *old_sets[2];
   DBUG_ENTER("print_quick");
   if (!quick)
     DBUG_VOID_RETURN;
   DBUG_LOCK_FILE;
 
   table= quick->head;
-  dbug_tmp_use_all_columns(table, old_sets, table->read_set, table->write_set);
+  dbug_tmp_use_all_columns(table, old_sets, &table->read_set, &table->write_set);
   quick->dbug_dump(0, TRUE);
-  dbug_tmp_restore_column_maps(table->read_set, table->write_set, old_sets);
+  dbug_tmp_restore_column_maps(&table->read_set, &table->write_set, old_sets);
 
   fprintf(DBUG_FILE,"other_keys: 0x%s:\n", needed_reg->print(buf));
 
@@ -16377,8 +16376,8 @@ void print_range_for_non_indexed_field(String *out, Field *field,
                                        KEY_MULTI_RANGE *range)
 {
   TABLE *table= field->table;
-  my_bitmap_map *old_sets[2];
-  dbug_tmp_use_all_columns(table, old_sets, table->read_set, table->write_set);
+  MY_BITMAP *old_sets[2];
+  dbug_tmp_use_all_columns(table, old_sets, &table->read_set, &table->write_set);
 
   if (range->start_key.length)
   {
@@ -16393,7 +16392,7 @@ void print_range_for_non_indexed_field(String *out, Field *field,
     print_max_range_operator(out, range->end_key.flag);
     field->print_key_part_value(out, range->end_key.key, field->key_length());
   }
-  dbug_tmp_restore_column_maps(table->read_set, table->write_set, old_sets);
+  dbug_tmp_restore_column_maps(&table->read_set, &table->write_set, old_sets);
 }
 
 
@@ -16460,8 +16459,8 @@ static void print_key_value(String *out, const KEY_PART_INFO *key_part,
   StringBuffer<128> tmp(system_charset_info);
   TABLE *table= field->table;
   uint store_length;
-  my_bitmap_map *old_sets[2];
-  dbug_tmp_use_all_columns(table, old_sets, table->read_set, table->write_set);
+  MY_BITMAP *old_sets[2];
+  dbug_tmp_use_all_columns(table, old_sets, &table->read_set, &table->write_set);
   const uchar *key_end= key+used_length;
 
   for (; key < key_end; key+=store_length, key_part++)
@@ -16474,7 +16473,7 @@ static void print_key_value(String *out, const KEY_PART_INFO *key_part,
     if (key + store_length < key_end)
       out->append(STRING_WITH_LEN(","));
   }
-  dbug_tmp_restore_column_maps(table->read_set, table->write_set, old_sets);
+  dbug_tmp_restore_column_maps(&table->read_set, &table->write_set, old_sets);
   out->append(STRING_WITH_LEN(")"));
 }
 
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index 8ad14ca260c..871411cf6c4 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -1389,13 +1389,13 @@ void partition_info::print_no_partition_found(TABLE *table_arg, myf errflag)
       buf_ptr= (char*)"from column_list";
     else
     {
-      my_bitmap_map *old_map= dbug_tmp_use_all_columns(table_arg, table_arg->read_set);
+      MY_BITMAP *old_map= dbug_tmp_use_all_columns(table_arg, &table_arg->read_set);
       if (part_expr->null_value)
         buf_ptr= (char*)"NULL";
       else
         longlong10_to_str(err_value, buf,
                      part_expr->unsigned_flag ? 10 : -10);
-      dbug_tmp_restore_column_map(table_arg->read_set, old_map);
+      dbug_tmp_restore_column_map(&table_arg->read_set, old_map);
     }
     my_error(ER_NO_PARTITION_FOR_GIVEN_VALUE, errflag, buf_ptr);
   }
diff --git a/sql/protocol.cc b/sql/protocol.cc
index 7d479ab2e3b..f369fa1c66f 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -1580,16 +1580,16 @@ bool Protocol_text::store(Field *field)
     return store_null();
 #ifdef DBUG_ASSERT_EXISTS
   TABLE *table= field->table;
-  my_bitmap_map *old_map= 0;
+  MY_BITMAP *old_map= 0;
   if (table->file)
-    old_map= dbug_tmp_use_all_columns(table, table->read_set);
+    old_map= dbug_tmp_use_all_columns(table, &table->read_set);
 #endif
 
   bool rc= field->send(this);
 
 #ifdef DBUG_ASSERT_EXISTS
   if (old_map)
-    dbug_tmp_restore_column_map(table->read_set, old_map);
+    dbug_tmp_restore_column_map(&table->read_set, old_map);
 #endif
 
   return rc;
diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
index 72347d93ad1..65d5a06a76a 100644
--- a/sql/rpl_parallel.cc
+++ b/sql/rpl_parallel.cc
@@ -4,6 +4,7 @@
 #include "rpl_mi.h"
 #include "sql_parse.h"
 #include "debug_sync.h"
+#include "sql_repl.h"
 #include "wsrep_mysqld.h"
 #ifdef WITH_WSREP
 #include "wsrep_trans_observer.h"
@@ -100,7 +101,7 @@ handle_queued_pos_update(THD *thd, rpl_parallel_thread::queued_event *qev)
     return;
 
   mysql_mutex_lock(&rli->data_lock);
-  cmp= strcmp(rli->group_relay_log_name, qev->event_relay_log_name);
+  cmp= compare_log_name(rli->group_relay_log_name, qev->event_relay_log_name);
   if (cmp < 0)
   {
     rli->group_relay_log_pos= qev->future_event_relay_log_pos;
@@ -109,7 +110,7 @@ handle_queued_pos_update(THD *thd, rpl_parallel_thread::queued_event *qev)
              rli->group_relay_log_pos < qev->future_event_relay_log_pos)
     rli->group_relay_log_pos= qev->future_event_relay_log_pos;
 
-  cmp= strcmp(rli->group_master_log_name, qev->future_event_master_log_name);
+  cmp= compare_log_name(rli->group_master_log_name, qev->future_event_master_log_name);
   if (cmp < 0)
   {
     strcpy(rli->group_master_log_name, qev->future_event_master_log_name);
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 8b61a3a708b..4c04382a5dc 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -993,7 +993,7 @@ void Relay_log_info::inc_group_relay_log_pos(ulonglong log_pos,
   if (rgi->is_parallel_exec)
   {
     /* In case of parallel replication, do not update the position backwards. */
-    int cmp= strcmp(group_relay_log_name, rgi->event_relay_log_name);
+    int cmp= compare_log_name(group_relay_log_name, rgi->event_relay_log_name);
     if (cmp < 0)
     {
       group_relay_log_pos= rgi->future_event_relay_log_pos;
@@ -1005,7 +1005,7 @@ void Relay_log_info::inc_group_relay_log_pos(ulonglong log_pos,
       In the parallel case we need to update the master_log_name here, rather
       than in Rotate_log_event::do_update_pos().
     */
-    cmp= strcmp(group_master_log_name, rgi->future_event_master_log_name);
+    cmp= compare_log_name(group_master_log_name, rgi->future_event_master_log_name);
     if (cmp <= 0)
     {
       if (cmp < 0)
diff --git a/sql/service_wsrep.cc b/sql/service_wsrep.cc
index e7eee5f67a5..d0a96a85a60 100644
--- a/sql/service_wsrep.cc
+++ b/sql/service_wsrep.cc
@@ -37,6 +37,16 @@ extern "C" void wsrep_thd_UNLOCK(const THD *thd)
   mysql_mutex_unlock(&thd->LOCK_thd_data);
 }
 
+extern "C" void wsrep_thd_kill_LOCK(const THD *thd)
+{
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
+}
+
+extern "C" void wsrep_thd_kill_UNLOCK(const THD *thd)
+{
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
+}
+
 extern "C" const char* wsrep_thd_client_state_str(const THD *thd)
 {
   return wsrep::to_c_string(thd->wsrep_cs().state());
@@ -199,16 +209,8 @@ extern "C" void wsrep_handle_SR_rollback(THD *bf_thd,
 extern "C" my_bool wsrep_thd_bf_abort(THD *bf_thd, THD *victim_thd,
                                       my_bool signal)
 {
-  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(bf_thd,
-                                                      STRING_WITH_LEN(act)));
-                 };);
-
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
+  mysql_mutex_assert_not_owner(&victim_thd->LOCK_thd_data);
   my_bool ret= wsrep_bf_abort(bf_thd, victim_thd);
   /*
     Send awake signal if victim was BF aborted or does not
@@ -217,8 +219,6 @@ extern "C" my_bool wsrep_thd_bf_abort(THD *bf_thd, THD *victim_thd,
    */
   if ((ret || !wsrep_on(victim_thd)) && signal)
   {
-    mysql_mutex_assert_not_owner(&victim_thd->LOCK_thd_data);
-    mysql_mutex_assert_not_owner(&victim_thd->LOCK_thd_kill);
     mysql_mutex_lock(&victim_thd->LOCK_thd_data);
 
     if (victim_thd->wsrep_aborter && victim_thd->wsrep_aborter != bf_thd->thread_id)
@@ -229,10 +229,8 @@ extern "C" my_bool wsrep_thd_bf_abort(THD *bf_thd, THD *victim_thd,
       return false;
     }
 
-    mysql_mutex_lock(&victim_thd->LOCK_thd_kill);
     victim_thd->wsrep_aborter= bf_thd->thread_id;
     victim_thd->awake_no_mutex(KILL_QUERY);
-    mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
     mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
   } else {
     WSREP_DEBUG("wsrep_thd_bf_abort skipped awake");
diff --git a/sql/slave.cc b/sql/slave.cc
index 08d27236576..5a5e83b977b 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2017, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2020, MariaDB Corporation.
+   Copyright (c) 2009, 2021, MariaDB Corporation.
 
    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
@@ -67,6 +67,7 @@ Master_info_index *master_info_index;
 #include "rpl_parallel.h"
 #include "sql_show.h"
 #include "semisync_slave.h"
+#include "sql_manager.h"
 
 #define FLAGSTR(V,F) ((V)&(F)?#F" ":"")
 
@@ -363,10 +364,33 @@ end:
   return err;
 }
 
+static THD *new_bg_THD()
+{
+  THD *thd= new THD(next_thread_id());
+  thd->thread_stack= (char*) &thd;
+  thd->store_globals();
+  thd->system_thread = SYSTEM_THREAD_SLAVE_BACKGROUND;
+  thd->security_ctx->skip_grants();
+  thd->set_command(COM_DAEMON);
+  thd->variables.wsrep_on= 0;
+  return thd;
+}
 
-static void
-handle_gtid_pos_auto_create_request(THD *thd, void *hton)
+static void bg_gtid_delete_pending(void *)
 {
+  THD *thd= new_bg_THD();
+
+  rpl_slave_state::list_element *list;
+  list= rpl_global_gtid_slave_state->gtid_grab_pending_delete_list();
+  rpl_global_gtid_slave_state->gtid_delete_pending(thd, &list);
+  if (list)
+    rpl_global_gtid_slave_state->put_back_list(list);
+  delete thd;
+}
+
+static void bg_gtid_pos_auto_create(void *hton)
+{
+  THD *thd= NULL;
   int UNINIT_VAR(err);
   plugin_ref engine= NULL, *auto_engines;
   rpl_slave_state::gtid_pos_table *entry;
@@ -378,7 +402,6 @@ handle_gtid_pos_auto_create_request(THD *thd, void *hton)
     it.
   */
   mysql_mutex_lock(&LOCK_global_system_variables);
-  engine= NULL;
   for (auto_engines= opt_gtid_pos_auto_plugins;
        auto_engines && *auto_engines;
        ++auto_engines)
@@ -423,6 +446,7 @@ handle_gtid_pos_auto_create_request(THD *thd, void *hton)
   table_name.str= loc_table_name.c_ptr_safe();
   table_name.length= loc_table_name.length();
 
+  thd= new_bg_THD();
   err= gtid_pos_table_creation(thd, engine, &table_name);
   if (err)
   {
@@ -450,47 +474,17 @@ handle_gtid_pos_auto_create_request(THD *thd, void *hton)
   mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
 
 end:
+  delete thd;
   if (engine)
     plugin_unlock(NULL, engine);
 }
 
-
-static bool slave_background_thread_running;
-static bool slave_background_thread_stop;
 static bool slave_background_thread_gtid_loaded;
 
-static struct slave_background_kill_t {
-  slave_background_kill_t *next;
-  THD *to_kill;
-} *slave_background_kill_list;
-
-static struct slave_background_gtid_pos_create_t {
-  slave_background_gtid_pos_create_t *next;
-  void *hton;
-} *slave_background_gtid_pos_create_list;
-
-static volatile bool slave_background_gtid_pending_delete_flag;
-
-
-pthread_handler_t
-handle_slave_background(void *arg __attribute__((unused)))
+static void bg_rpl_load_gtid_slave_state(void *)
 {
-  THD *thd;
-  PSI_stage_info old_stage;
-  bool stop;
-
-  my_thread_init();
-  thd= new THD(next_thread_id());
-  thd->thread_stack= (char*) &thd;           /* Set approximate stack start */
-  thd->system_thread = SYSTEM_THREAD_SLAVE_BACKGROUND;
-  thd->store_globals();
-  thd->security_ctx->skip_grants();
-  thd->set_command(COM_DAEMON);
-#ifdef WITH_WSREP
-  thd->variables.wsrep_on= 0;
-#endif
+  THD *thd= new_bg_THD();
   thd->set_psi(PSI_CALL_get_thread());
-
   thd_proc_info(thd, "Loading slave GTID position from table");
   if (rpl_load_gtid_slave_state(thd))
     sql_print_warning("Failed to load slave replication state from table "
@@ -499,207 +493,62 @@ handle_slave_background(void *arg __attribute__((unused)))
                       thd->get_stmt_da()->sql_errno(),
                       thd->get_stmt_da()->message());
 
-  mysql_mutex_lock(&LOCK_slave_background);
+  // hijacking global_rpl_thread_pool cond here - it's only once on startup
+  mysql_mutex_lock(&global_rpl_thread_pool.LOCK_rpl_thread_pool);
   slave_background_thread_gtid_loaded= true;
-  mysql_cond_broadcast(&COND_slave_background);
-
-  THD_STAGE_INFO(thd, stage_slave_background_process_request);
-  do
-  {
-    slave_background_kill_t *kill_list;
-    slave_background_gtid_pos_create_t *create_list;
-    bool pending_deletes;
-
-    thd->ENTER_COND(&COND_slave_background, &LOCK_slave_background,
-                    &stage_slave_background_wait_request,
-                    &old_stage);
-    for (;;)
-    {
-      stop= thd->killed || slave_background_thread_stop;
-      kill_list= slave_background_kill_list;
-      create_list= slave_background_gtid_pos_create_list;
-      pending_deletes= slave_background_gtid_pending_delete_flag;
-      if (stop || kill_list || create_list || pending_deletes)
-        break;
-      mysql_cond_wait(&COND_slave_background, &LOCK_slave_background);
-    }
-
-    slave_background_kill_list= NULL;
-    slave_background_gtid_pos_create_list= NULL;
-    slave_background_gtid_pending_delete_flag= false;
-    thd->EXIT_COND(&old_stage);
-
-    while (kill_list)
-    {
-      slave_background_kill_t *p = kill_list;
-      THD *to_kill= p->to_kill;
-      kill_list= p->next;
-
-      to_kill->awake(KILL_CONNECTION);
-      mysql_mutex_lock(&to_kill->LOCK_wakeup_ready);
-      to_kill->rgi_slave->killed_for_retry=
-        rpl_group_info::RETRY_KILL_KILLED;
-      mysql_cond_broadcast(&to_kill->COND_wakeup_ready);
-      mysql_mutex_unlock(&to_kill->LOCK_wakeup_ready);
-      my_free(p);
-    }
-
-    while (create_list)
-    {
-      slave_background_gtid_pos_create_t *next= create_list->next;
-      void *hton= create_list->hton;
-      handle_gtid_pos_auto_create_request(thd, hton);
-      my_free(create_list);
-      create_list= next;
-    }
-
-    if (pending_deletes)
-    {
-      rpl_slave_state::list_element *list;
-
-      slave_background_gtid_pending_delete_flag= false;
-      list= rpl_global_gtid_slave_state->gtid_grab_pending_delete_list();
-      rpl_global_gtid_slave_state->gtid_delete_pending(thd, &list);
-      if (list)
-        rpl_global_gtid_slave_state->put_back_list(list);
-    }
-
-    mysql_mutex_lock(&LOCK_slave_background);
-  } while (!stop);
-
-  slave_background_thread_running= false;
-  mysql_cond_broadcast(&COND_slave_background);
-  mysql_mutex_unlock(&LOCK_slave_background);
-
+  mysql_cond_signal(&global_rpl_thread_pool.COND_rpl_thread_pool);
+  mysql_mutex_unlock(&global_rpl_thread_pool.LOCK_rpl_thread_pool);
   delete thd;
-
-  my_thread_end();
-  return 0;
 }
 
+static void bg_slave_kill(void *victim)
+{
+  THD *to_kill= (THD *)victim;
+  to_kill->awake(KILL_CONNECTION);
+  mysql_mutex_lock(&to_kill->LOCK_wakeup_ready);
+  to_kill->rgi_slave->killed_for_retry= rpl_group_info::RETRY_KILL_KILLED;
+  mysql_cond_broadcast(&to_kill->COND_wakeup_ready);
+  mysql_mutex_unlock(&to_kill->LOCK_wakeup_ready);
+}
 
-
-void
-slave_background_kill_request(THD *to_kill)
+void slave_background_kill_request(THD *to_kill)
 {
   if (to_kill->rgi_slave->killed_for_retry)
     return;                                     // Already deadlock killed.
-  slave_background_kill_t *p=
-    (slave_background_kill_t *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*p), MYF(MY_WME));
-  if (p)
-  {
-    p->to_kill= to_kill;
-    to_kill->rgi_slave->killed_for_retry=
-      rpl_group_info::RETRY_KILL_PENDING;
-    mysql_mutex_lock(&LOCK_slave_background);
-    p->next= slave_background_kill_list;
-    slave_background_kill_list= p;
-    mysql_cond_signal(&COND_slave_background);
-    mysql_mutex_unlock(&LOCK_slave_background);
-  }
+  to_kill->rgi_slave->killed_for_retry= rpl_group_info::RETRY_KILL_PENDING;
+  mysql_manager_submit(bg_slave_kill, to_kill);
 }
 
-
 /*
   This function must only be called from a slave SQL thread (or worker thread),
   to ensure that the table_entry will not go away before we can lock the
   LOCK_slave_state.
 */
-void
-slave_background_gtid_pos_create_request(
+void slave_background_gtid_pos_create_request(
         rpl_slave_state::gtid_pos_table *table_entry)
 {
-  slave_background_gtid_pos_create_t *p;
-
   if (table_entry->state != rpl_slave_state::GTID_POS_AUTO_CREATE)
     return;
-  p= (slave_background_gtid_pos_create_t *)my_malloc(PSI_INSTRUMENT_ME, sizeof(*p), MYF(MY_WME));
-  if (!p)
-    return;
   mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
   if (table_entry->state != rpl_slave_state::GTID_POS_AUTO_CREATE)
   {
-    my_free(p);
     mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
     return;
   }
   table_entry->state= rpl_slave_state::GTID_POS_CREATE_REQUESTED;
   mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
 
-  p->hton= table_entry->table_hton;
-  mysql_mutex_lock(&LOCK_slave_background);
-  p->next= slave_background_gtid_pos_create_list;
-  slave_background_gtid_pos_create_list= p;
-  mysql_cond_signal(&COND_slave_background);
-  mysql_mutex_unlock(&LOCK_slave_background);
+  mysql_manager_submit(bg_gtid_pos_auto_create, table_entry->table_hton);
 }
 
 
 /*
-  Request the slave background thread to delete no longer used rows from the
+  Request the manager thread to delete no longer used rows from the
   mysql.gtid_slave_pos* tables.
-
-  This is called from time-critical rpl_slave_state::update(), so we avoid
-  taking any locks here. This means we may race with the background thread
-  to occasionally lose a signal. This is not a problem; any pending rows to
-  be deleted will just be deleted a bit later as part of the next batch.
-*/
-void
-slave_background_gtid_pending_delete_request(void)
-{
-  slave_background_gtid_pending_delete_flag= true;
-  mysql_cond_signal(&COND_slave_background);
-}
-
-
-/*
-  Start the slave background thread.
-
-  This thread is currently used for two purposes:
-
-  1. To load the GTID state from mysql.gtid_slave_pos at server start; reading
-     from table requires valid THD, which is otherwise not available during
-     server init.
-
-  2. To kill worker thread transactions during parallel replication, when a
-     storage engine attempts to take an errorneous conflicting lock that would
-     cause a deadlock. Killing is done asynchroneously, as the kill may not
-     be safe within the context of a callback from inside storage engine
-     locking code.
 */
-static int
-start_slave_background_thread()
-{
-  pthread_t th;
-
-  slave_background_thread_running= true;
-  slave_background_thread_stop= false;
-  slave_background_thread_gtid_loaded= false;
-  if (mysql_thread_create(key_thread_slave_background,
-                          &th, &connection_attrib, handle_slave_background,
-                          NULL))
-  {
-    sql_print_error("Failed to create thread while initialising slave");
-    return 1;
-  }
-  mysql_mutex_lock(&LOCK_slave_background);
-  while (!slave_background_thread_gtid_loaded)
-    mysql_cond_wait(&COND_slave_background, &LOCK_slave_background);
-  mysql_mutex_unlock(&LOCK_slave_background);
-
-  return 0;
-}
-
-
-static void
-stop_slave_background_thread()
+void slave_background_gtid_pending_delete_request(void)
 {
-  mysql_mutex_lock(&LOCK_slave_background);
-  slave_background_thread_stop= true;
-  mysql_cond_broadcast(&COND_slave_background);
-  while (slave_background_thread_running)
-    mysql_cond_wait(&COND_slave_background, &LOCK_slave_background);
-  mysql_mutex_unlock(&LOCK_slave_background);
+  mysql_manager_submit(bg_gtid_delete_pending, NULL);
 }
 
 
@@ -714,12 +563,19 @@ int init_slave()
   init_slave_psi_keys();
 #endif
 
-  if (start_slave_background_thread())
-    return 1;
-
   if (global_rpl_thread_pool.init(opt_slave_parallel_threads))
     return 1;
 
+  slave_background_thread_gtid_loaded= false;
+  mysql_manager_submit(bg_rpl_load_gtid_slave_state, NULL);
+
+  // hijacking global_rpl_thread_pool cond here - it's only once on startup
+  mysql_mutex_lock(&global_rpl_thread_pool.LOCK_rpl_thread_pool);
+  while (!slave_background_thread_gtid_loaded)
+    mysql_cond_wait(&global_rpl_thread_pool.COND_rpl_thread_pool,
+                    &global_rpl_thread_pool.LOCK_rpl_thread_pool);
+  mysql_mutex_unlock(&global_rpl_thread_pool.LOCK_rpl_thread_pool);
+
   /*
     This is called when mysqld starts. Before client connections are
     accepted. However bootstrap may conflict with us if it does START SLAVE.
@@ -1215,12 +1071,8 @@ terminate_slave_thread(THD *thd,
     int error __attribute__((unused));
     DBUG_PRINT("loop", ("killing slave thread"));
 
-#ifdef WITH_WSREP
-    /* awake_no_mutex() requires LOCK_thd_data to be locked if wsrep
-       is enabled */
-    if (WSREP(thd)) mysql_mutex_lock(&thd->LOCK_thd_data);
-#endif /* WITH_WSREP */
     mysql_mutex_lock(&thd->LOCK_thd_kill);
+    mysql_mutex_lock(&thd->LOCK_thd_data);
 #ifndef DONT_USE_THR_ALARM
     /*
       Error codes from pthread_kill are:
@@ -1233,9 +1085,7 @@ terminate_slave_thread(THD *thd,
     thd->awake_no_mutex(NOT_KILLED);
 
     mysql_mutex_unlock(&thd->LOCK_thd_kill);
-#ifdef WITH_WSREP
-    if (WSREP(thd)) mysql_mutex_unlock(&thd->LOCK_thd_data);
-#endif /* WITH_WSREP */
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
 
     /*
       There is a small chance that slave thread might miss the first
@@ -1448,7 +1298,6 @@ void slave_prepare_for_shutdown()
   // It's safe to destruct worker pool now when
   // all driver threads are gone.
   global_rpl_thread_pool.deactivate();
-  stop_slave_background_thread();
 }
 
 /*
@@ -1479,8 +1328,6 @@ void end_slave()
   active_mi= 0;
   mysql_mutex_unlock(&LOCK_active_mi);
 
-  stop_slave_background_thread();
-
   global_rpl_thread_pool.destroy();
   free_all_rpl_filters();
   DBUG_VOID_RETURN;
@@ -4782,10 +4629,7 @@ pthread_handler_t handle_slave_io(void *arg)
       goto err;
   }
 
-
-#ifdef WITH_WSREP
   thd->variables.wsrep_on= 0;
-#endif
   if (DBUG_EVALUATE_IF("failed_slave_start", 1, 0)
       || repl_semisync_slave.slave_start(mi))
   {
diff --git a/sql/spatial.cc b/sql/spatial.cc
index 301d50f5d1e..1f93323440c 100644
--- a/sql/spatial.cc
+++ b/sql/spatial.cc
@@ -631,6 +631,7 @@ Geometry *Geometry::create_from_json(Geometry_buffer *buffer,
         if (feature_type_found)
           goto handle_geometry_key;
       }
+      goto err_return;
     }
     else
     {
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index aed0ad02e4d..2e1b74bb3ce 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -3285,6 +3285,12 @@ end:
 
 int acl_check_setrole(THD *thd, const char *rolename, privilege_t *access)
 {
+  if (!initialized)
+  {
+    my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--skip-grant-tables");
+    return 1;
+  }
+
   return check_user_can_set_role(thd, thd->security_ctx->priv_user,
            thd->security_ctx->host, thd->security_ctx->ip, rolename, access);
 }
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 5b1d21bb44b..d23b190b2c5 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -49,9 +49,6 @@
 #include <m_ctype.h>
 #include <sys/stat.h>
 #include <thr_alarm.h>
-#ifdef	__WIN__0
-#include <io.h>
-#endif
 #include <mysys_err.h>
 #include <limits.h>
 
@@ -70,6 +67,8 @@
 #ifdef WITH_WSREP
 #include "wsrep_thd.h"
 #include "wsrep_trans_observer.h"
+#else
+static inline bool wsrep_is_bf_aborted(THD* thd) { return false; }
 #endif /* WITH_WSREP */
 #include "opt_trace.h"
 #include <mysql/psi/mysql_transaction.h>
@@ -486,7 +485,9 @@ void thd_set_ha_data(THD *thd, const struct handlerton *hton,
                      const void *ha_data)
 {
   plugin_ref *lock= &thd->ha_data[hton->slot].lock;
+  mysql_mutex_lock(&thd->LOCK_thd_data);
   thd->ha_data[hton->slot].ha_ptr= const_cast<void*>(ha_data);
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
   if (ha_data && !*lock)
     *lock= ha_lock_engine(NULL, (handlerton*) hton);
   else if (!ha_data && *lock)
@@ -837,6 +838,7 @@ THD::THD(my_thread_id id, bool is_wsrep_applier)
   mysql_mutex_init(key_LOCK_wakeup_ready, &LOCK_wakeup_ready, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_thd_kill, &LOCK_thd_kill, MY_MUTEX_INIT_FAST);
   mysql_cond_init(key_COND_wakeup_ready, &COND_wakeup_ready, 0);
+  mysql_mutex_record_order(&LOCK_thd_kill, &LOCK_thd_data);
 
   /* Variables with default values */
   proc_info="login";
@@ -1532,30 +1534,15 @@ void THD::reset_db(const LEX_CSTRING *new_db)
 
 /* Do operations that may take a long time */
 
-void THD::cleanup(bool have_mutex)
+void THD::cleanup(void)
 {
   DBUG_ENTER("THD::cleanup");
   DBUG_ASSERT(cleanup_done == 0);
 
-  if (have_mutex)
-    set_killed_no_mutex(KILL_CONNECTION,0,0);
-  else
-    set_killed(KILL_CONNECTION);
+  set_killed(KILL_CONNECTION);
 #ifdef WITH_WSREP
   if (wsrep_cs().state() != wsrep::client_state::s_none)
-  {
-    if (have_mutex)
-    {
-      mysql_mutex_assert_owner(static_cast<mysql_mutex_t*>
-                               (m_wsrep_mutex.native()));
-      // Below wsrep-lib function will not acquire any mutexes
-      wsrep::unique_lock<wsrep::mutex> lock(m_wsrep_mutex, std::adopt_lock);
-      wsrep_cs().cleanup(lock);
-      lock.release();
-    }
-    else
-      wsrep_cs().cleanup();
-  }
+    wsrep_cs().cleanup();
   wsrep_client_thread= false;
 #endif /* WITH_WSREP */
 
@@ -1628,28 +1615,6 @@ void THD::cleanup(bool have_mutex)
 void THD::free_connection()
 {
   DBUG_ASSERT(free_connection_done == 0);
-  /* Make sure threads are not available via server_threads.  */
-  assert_not_linked();
-
-  /*
-    Other threads may have a lock on THD::LOCK_thd_data or
-    THD::LOCK_thd_kill to ensure that this THD is not deleted
-    while they access it. The following mutex_lock ensures
-    that no one else is using this THD and it's now safe to
-    continue.
-
-    For example consider KILL-statement execution on
-    sql_parse.cc kill_one_thread() that will use
-    THD::LOCK_thd_data to protect victim thread during
-    THD::awake().
-  */
-  mysql_mutex_lock(&LOCK_thd_data);
-  mysql_mutex_lock(&LOCK_thd_kill);
-
-#ifdef WITH_WSREP
-  delete wsrep_rgi;
-  wsrep_rgi= nullptr;
-#endif /* WITH_WSREP */
   my_free(const_cast<char*>(db.str));
   db= null_clex_str;
 #ifndef EMBEDDED_LIBRARY
@@ -1658,8 +1623,8 @@ void THD::free_connection()
   net.vio= nullptr;
   net_end(&net);
 #endif
-  if (!cleanup_done)
-    cleanup(true); // We have locked THD::LOCK_thd_kill
+ if (!cleanup_done)
+   cleanup();
   ha_close_connection(this);
   plugin_thdvar_cleanup(this);
   mysql_audit_free_thd(this);
@@ -1671,8 +1636,6 @@ void THD::free_connection()
   profiling.restart();                          // Reset profiling
 #endif
   debug_sync_reset_thread(this);
-  mysql_mutex_unlock(&LOCK_thd_kill);
-  mysql_mutex_unlock(&LOCK_thd_data);
 }
 
 /*
@@ -1728,6 +1691,17 @@ THD::~THD()
   if (!status_in_global)
     add_status_to_global();
 
+  /*
+    Other threads may have a lock on LOCK_thd_kill to ensure that this
+    THD is not deleted while they access it. The following mutex_lock
+    ensures that no one else is using this THD and it's now safe to delete
+  */
+  mysql_mutex_lock(&LOCK_thd_kill);
+  mysql_mutex_unlock(&LOCK_thd_kill);
+
+#ifdef WITH_WSREP
+  delete wsrep_rgi;
+#endif
   if (!free_connection_done)
     free_connection();
 
@@ -1918,7 +1892,7 @@ void THD::awake_no_mutex(killed_state state_to_set)
   DBUG_PRINT("enter", ("this: %p current_thd: %p  state: %d",
                        this, current_thd, (int) state_to_set));
   THD_CHECK_SENTRY(this);
-  if (WSREP_NNULL(this)) mysql_mutex_assert_owner(&LOCK_thd_data);
+  mysql_mutex_assert_owner(&LOCK_thd_data);
   mysql_mutex_assert_owner(&LOCK_thd_kill);
 
   print_aborted_warning(3, "KILLED");
@@ -1951,15 +1925,21 @@ void THD::awake_no_mutex(killed_state state_to_set)
   }
 
   /* Interrupt target waiting inside a storage engine. */
-  if (IF_WSREP(state_to_set != NOT_KILLED  && !wsrep_is_bf_aborted(this),
-               state_to_set != NOT_KILLED))
+  if (state_to_set != NOT_KILLED  && !wsrep_is_bf_aborted(this))
     ha_kill_query(this, thd_kill_level(this));
 
-  /* Broadcast a condition to kick the target if it is waiting on it. */
+  abort_current_cond_wait(false);
+  DBUG_VOID_RETURN;
+}
+
+/* Broadcast a condition to kick the target if it is waiting on it. */
+void THD::abort_current_cond_wait(bool force)
+{
+  mysql_mutex_assert_owner(&LOCK_thd_kill);
   if (mysys_var)
   {
     mysql_mutex_lock(&mysys_var->mutex);
-    if (!system_thread)		// Don't abort locks
+    if (!system_thread || force)                 // Don't abort locks
       mysys_var->abort=1;
 
     /*
@@ -2017,7 +1997,6 @@ void THD::awake_no_mutex(killed_state state_to_set)
     }
     mysql_mutex_unlock(&mysys_var->mutex);
   }
-  DBUG_VOID_RETURN;
 }
 
 
@@ -2071,16 +2050,7 @@ bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
     mysql_mutex_lock(&in_use->LOCK_thd_kill);
     if (in_use->killed < KILL_CONNECTION)
       in_use->set_killed_no_mutex(KILL_CONNECTION);
-    if (in_use->mysys_var)
-    {
-      mysql_mutex_lock(&in_use->mysys_var->mutex);
-      if (in_use->mysys_var->current_cond)
-        mysql_cond_broadcast(in_use->mysys_var->current_cond);
-
-      /* Abort if about to wait in thr_upgrade_write_delay_lock */
-      in_use->mysys_var->abort= 1;
-      mysql_mutex_unlock(&in_use->mysys_var->mutex);
-    }
+    in_use->abort_current_cond_wait(true);
     mysql_mutex_unlock(&in_use->LOCK_thd_kill);
     signalled= TRUE;
   }
@@ -4642,7 +4612,13 @@ extern "C" void thd_progress_report(MYSQL_THD thd,
     return;
   if (thd->progress.max_counter != max_progress)        // Simple optimization
   {
-    mysql_mutex_lock(&thd->LOCK_thd_data);
+    /*
+      Better to not wait in the unlikely event that LOCK_thd_data is locked
+      as Galera can potentially have this locked for a long time.
+      Progress counters will fix themselves after the next call.
+    */
+    if (mysql_mutex_trylock(&thd->LOCK_thd_data))
+      return;
     thd->progress.counter= progress;
     thd->progress.max_counter= max_progress;
     mysql_mutex_unlock(&thd->LOCK_thd_data);
@@ -5279,6 +5255,18 @@ thd_need_ordering_with(const MYSQL_THD thd, const MYSQL_THD other_thd)
   DBUG_EXECUTE_IF("disable_thd_need_ordering_with", return 1;);
   if (!thd || !other_thd)
     return 1;
+#ifdef WITH_WSREP
+  /* wsrep applier, replayer and TOI processing threads are ordered
+     by replication provider, relaxed GAP locking protocol can be used
+     between high priority wsrep threads.
+     Note that wsrep_thd_is_BF() doesn't take LOCK_thd_data for either thd,
+     the caller should guarantee that the BF state won't change.
+     (e.g. InnoDB does it by keeping lock_sys.mutex locked)
+  */
+  if (WSREP_ON && wsrep_thd_is_BF(thd, false) &&
+      wsrep_thd_is_BF(other_thd, false))
+    return 0;
+#endif /* WITH_WSREP */
   rgi= thd->rgi_slave;
   other_rgi= other_thd->rgi_slave;
   if (!rgi || !other_rgi)
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 8000939668e..2913a4a8350 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2579,7 +2579,7 @@ public:
     - mysys_var (used by KILL statement and shutdown).
     - Also ensures that THD is not deleted while mutex is hold
   */
-  mysql_mutex_t LOCK_thd_kill;
+  mutable mysql_mutex_t LOCK_thd_kill;
 
   /* all prepared statements and cursors of this connection */
   Statement_map stmt_map;
@@ -3622,7 +3622,7 @@ public:
   void update_all_stats();
   void update_stats(void);
   void change_user(void);
-  void cleanup(bool have_mutex=false);
+  void cleanup(void);
   void cleanup_after_query();
   void free_connection();
   void reset_for_reuse();
@@ -3650,19 +3650,13 @@ public:
   void awake_no_mutex(killed_state state_to_set);
   void awake(killed_state state_to_set)
   {
-    bool wsrep_on_local= variables.wsrep_on;
-    /*
-      mutex locking order (LOCK_thd_data - LOCK_thd_kill)) requires
-      to grab LOCK_thd_data here
-    */
-    if (wsrep_on_local)
-      mysql_mutex_lock(&LOCK_thd_data);
     mysql_mutex_lock(&LOCK_thd_kill);
+    mysql_mutex_lock(&LOCK_thd_data);
     awake_no_mutex(state_to_set);
+    mysql_mutex_unlock(&LOCK_thd_data);
     mysql_mutex_unlock(&LOCK_thd_kill);
-    if (wsrep_on_local)
-      mysql_mutex_unlock(&LOCK_thd_data);
   }
+  void abort_current_cond_wait(bool force);
  
   /** Disconnect the associated communication endpoint. */
   void disconnect();
@@ -4418,8 +4412,7 @@ public:
     mysql_mutex_lock(&LOCK_thd_kill);
     int err= killed_errno();
     if (err)
-      my_message(err, killed_err ? killed_err->msg : ER_THD(this, err),
-                 MYF(0));
+      my_message(err, killed_err ? killed_err->msg : ER_THD(this, err), MYF(0));
     mysql_mutex_unlock(&LOCK_thd_kill);
   }
   /* return TRUE if we will abort query if we make a warning now */
diff --git a/sql/sql_connect.cc b/sql/sql_connect.cc
index 479a310542b..e0119e50eaa 100644
--- a/sql/sql_connect.cc
+++ b/sql/sql_connect.cc
@@ -93,7 +93,6 @@ int get_or_create_user_conn(THD *thd, const char *user,
     uc->host= uc->user + user_len +  1;
     uc->len= (uint)temp_len;
     uc->connections= uc->questions= uc->updates= uc->conn_per_hour= 0;
-    uc->user_resources= *mqh;
     uc->reset_utime= thd->thr_create_utime;
     if (my_hash_insert(&hash_user_connections, (uchar*) uc))
     {
@@ -103,6 +102,7 @@ int get_or_create_user_conn(THD *thd, const char *user,
       goto end;
     }
   }
+  uc->user_resources= *mqh;
   thd->user_connect=uc;
   uc->connections++;
 end:
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index a6f93f6dfee..73bfb8227e1 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -688,7 +688,6 @@ mysql_ha_fix_cond_and_key(SQL_HANDLER *handler,
 
       for (keypart_map= key_len=0 ; (item=it_ke++) ; key_part++)
       {
-        my_bitmap_map *old_map;
 	/* note that 'item' can be changed by fix_fields() call */
         if (item->fix_fields_if_needed_for_scalar(thd, it_ke.ref()))
           return 1;
@@ -700,9 +699,9 @@ mysql_ha_fix_cond_and_key(SQL_HANDLER *handler,
         }
         if (!in_prepare)
         {
-          old_map= dbug_tmp_use_all_columns(table, table->write_set);
+          MY_BITMAP *old_map= dbug_tmp_use_all_columns(table, &table->write_set);
           (void) item->save_in_field(key_part->field, 1);
-          dbug_tmp_restore_column_map(table->write_set, old_map);
+          dbug_tmp_restore_column_map(&table->write_set, old_map);
         }
         key_len+= key_part->store_length;
         keypart_map= (keypart_map << 1) | 1;
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index eb3e1ec7493..f0b84ab126f 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2936,23 +2936,7 @@ void kill_delayed_threads(void)
     mysql_mutex_lock(&di->thd.LOCK_thd_kill);
     if (di->thd.killed < KILL_CONNECTION)
       di->thd.set_killed_no_mutex(KILL_CONNECTION);
-    if (di->thd.mysys_var)
-    {
-      mysql_mutex_lock(&di->thd.mysys_var->mutex);
-      if (di->thd.mysys_var->current_cond)
-      {
-	/*
-	  We need the following test because the main mutex may be locked
-	  in handle_delayed_insert()
-	*/
-	if (&di->mutex != di->thd.mysys_var->current_mutex)
-          mysql_mutex_lock(di->thd.mysys_var->current_mutex);
-        mysql_cond_broadcast(di->thd.mysys_var->current_cond);
-	if (&di->mutex != di->thd.mysys_var->current_mutex)
-          mysql_mutex_unlock(di->thd.mysys_var->current_mutex);
-      }
-      mysql_mutex_unlock(&di->thd.mysys_var->mutex);
-    }
+    di->thd.abort_current_cond_wait(false);
     mysql_mutex_unlock(&di->thd.LOCK_thd_kill);
   }
   mysql_mutex_unlock(&LOCK_delayed_insert); // For unlink from list
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 6871699dc5b..0ebf3907b58 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -5619,6 +5619,9 @@ bool LEX::save_prep_leaf_tables()
 
 bool st_select_lex::save_prep_leaf_tables(THD *thd)
 {
+  if (prep_leaf_list_state == SAVED)
+    return FALSE;
+
   List_iterator_fast<TABLE_LIST> li(leaf_tables);
   TABLE_LIST *table;
 
diff --git a/sql/sql_manager.cc b/sql/sql_manager.cc
index a3d1a7242e4..df6847f2ba1 100644
--- a/sql/sql_manager.cc
+++ b/sql/sql_manager.cc
@@ -26,8 +26,8 @@
 #include "sql_manager.h"
 #include "sql_base.h"                           // flush_tables
 
-static bool volatile manager_thread_in_use;
-static bool abort_manager;
+static bool volatile manager_thread_in_use = 0;
+static bool abort_manager = false;
 
 pthread_t manager_thread;
 mysql_mutex_t LOCK_manager;
@@ -35,32 +35,32 @@ mysql_cond_t COND_manager;
 
 struct handler_cb {
    struct handler_cb *next;
-   void (*action)(void);
+   void (*action)(void *);
+   void *data;
 };
 
-static struct handler_cb * volatile cb_list;
+static struct handler_cb *cb_list; // protected by LOCK_manager
 
-bool mysql_manager_submit(void (*action)())
+bool mysql_manager_submit(void (*action)(void *), void *data)
 {
   bool result= FALSE;
   DBUG_ASSERT(manager_thread_in_use);
-  struct handler_cb * volatile *cb;
+  struct handler_cb **cb;
   mysql_mutex_lock(&LOCK_manager);
   cb= &cb_list;
-  while (*cb && (*cb)->action != action)
+  while (*cb)
     cb= &(*cb)->next;
+  *cb= (struct handler_cb *)my_malloc(PSI_INSTRUMENT_ME,
+                                      sizeof(struct handler_cb), MYF(MY_WME));
   if (!*cb)
+    result= TRUE;
+  else
   {
-    *cb= (struct handler_cb *)my_malloc(PSI_INSTRUMENT_ME,
-                                        sizeof(struct handler_cb), MYF(MY_WME));
-    if (!*cb)
-      result= TRUE;
-    else
-    {
-      (*cb)->next= NULL;
-      (*cb)->action= action;
-    }
+    (*cb)->next= NULL;
+    (*cb)->action= action;
+    (*cb)->data= data;
   }
+  mysql_cond_signal(&COND_manager);
   mysql_mutex_unlock(&LOCK_manager);
   return result;
 }
@@ -70,18 +70,14 @@ pthread_handler_t handle_manager(void *arg __attribute__((unused)))
   int error = 0;
   struct timespec abstime;
   bool reset_flush_time = TRUE;
-  struct handler_cb *cb= NULL;
   my_thread_init();
   DBUG_ENTER("handle_manager");
 
   pthread_detach_this_thread();
   manager_thread = pthread_self();
-  mysql_cond_init(key_COND_manager, &COND_manager,NULL);
-  mysql_mutex_init(key_LOCK_manager, &LOCK_manager, NULL);
-  manager_thread_in_use = 1;
-  for (;;)
+  mysql_mutex_lock(&LOCK_manager);
+  while (!abort_manager)
   {
-    mysql_mutex_lock(&LOCK_manager);
     /* XXX: This will need to be made more general to handle different
      * polling needs. */
     if (flush_time)
@@ -91,40 +87,37 @@ pthread_handler_t handle_manager(void *arg __attribute__((unused)))
 	set_timespec(abstime, flush_time);
         reset_flush_time = FALSE;
       }
-      while ((!error || error == EINTR) && !abort_manager)
+      while ((!error || error == EINTR) && !abort_manager && !cb_list)
         error= mysql_cond_timedwait(&COND_manager, &LOCK_manager, &abstime);
+
+      if (error == ETIMEDOUT || error == ETIME)
+      {
+        tc_purge();
+        error = 0;
+        reset_flush_time = TRUE;
+      }
     }
     else
     {
-      while ((!error || error == EINTR) && !abort_manager)
+      while ((!error || error == EINTR) && !abort_manager && !cb_list)
         error= mysql_cond_wait(&COND_manager, &LOCK_manager);
     }
-    if (cb == NULL)
-    {
-      cb= cb_list;
-      cb_list= NULL;
-    }
-    mysql_mutex_unlock(&LOCK_manager);
 
-    if (abort_manager)
-      break;
-
-    if (error == ETIMEDOUT || error == ETIME)
-    {
-      tc_purge();
-      error = 0;
-      reset_flush_time = TRUE;
-    }
+    struct handler_cb *cb= cb_list;
+    cb_list= NULL;
+    mysql_mutex_unlock(&LOCK_manager);
 
     while (cb)
     {
       struct handler_cb *next= cb->next;
-      cb->action();
+      cb->action(cb->data);
       my_free(cb);
       cb= next;
     }
+    mysql_mutex_lock(&LOCK_manager);
   }
   manager_thread_in_use = 0;
+  mysql_mutex_unlock(&LOCK_manager);
   mysql_mutex_destroy(&LOCK_manager);
   mysql_cond_destroy(&COND_manager);
   DBUG_LEAVE; // Can't use DBUG_RETURN after my_thread_end
@@ -138,15 +131,15 @@ void start_handle_manager()
 {
   DBUG_ENTER("start_handle_manager");
   abort_manager = false;
-  if (flush_time && flush_time != ~(ulong) 0L)
   {
     pthread_t hThread;
-    int error;
-    if ((error= mysql_thread_create(key_thread_handle_manager,
-                                    &hThread, &connection_attrib,
-                                    handle_manager, 0)))
-      sql_print_warning("Can't create handle_manager thread (errno= %d)",
-                        error);
+    int err;
+    manager_thread_in_use = 1;
+    mysql_cond_init(key_COND_manager, &COND_manager,NULL);
+    mysql_mutex_init(key_LOCK_manager, &LOCK_manager, NULL);
+    if ((err= mysql_thread_create(key_thread_handle_manager, &hThread,
+                                  &connection_attrib, handle_manager, 0)))
+      sql_print_warning("Can't create handle_manager thread (errno: %M)", err);
   }
   DBUG_VOID_RETURN;
 }
@@ -156,10 +149,10 @@ void start_handle_manager()
 void stop_handle_manager()
 {
   DBUG_ENTER("stop_handle_manager");
-  abort_manager = true;
   if (manager_thread_in_use)
   {
     mysql_mutex_lock(&LOCK_manager);
+    abort_manager = true;
     DBUG_PRINT("quit", ("initiate shutdown of handle manager thread: %lu",
                         (ulong)manager_thread));
     mysql_cond_signal(&COND_manager);
diff --git a/sql/sql_manager.h b/sql/sql_manager.h
index 9c6c84450ed..f97d4a2cfc5 100644
--- a/sql/sql_manager.h
+++ b/sql/sql_manager.h
@@ -18,6 +18,6 @@
 
 void start_handle_manager();
 void stop_handle_manager();
-bool mysql_manager_submit(void (*action)());
+bool mysql_manager_submit(void (*action)(void *), void *data);
 
 #endif /* SQL_MANAGER_INCLUDED */
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 078d48872b0..bbb880cb117 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2239,6 +2239,7 @@ dispatch_command_return dispatch_command(enum enum_server_command command, THD *
       break;
     general_log_print(thd, command, NullS);
     status_var_increment(thd->status_var.com_stat[SQLCOM_SHOW_STATUS]);
+    *current_global_status_var= global_status_var;
     calc_sum_of_all_status(current_global_status_var);
     if (!(uptime= (ulong) (thd->start_time - server_start_time)))
       queries_per_second1000= 0;
@@ -9090,10 +9091,9 @@ struct find_thread_callback_arg
 };
 
 
-my_bool find_thread_callback(THD *thd, find_thread_callback_arg *arg)
+static my_bool find_thread_callback(THD *thd, find_thread_callback_arg *arg)
 {
-  if (thd->get_command() != COM_DAEMON &&
-      arg->id == (arg->query_id ? thd->query_id : (longlong) thd->thread_id))
+  if (arg->id == (arg->query_id ? thd->query_id : (longlong) thd->thread_id))
   {
     mysql_mutex_lock(&thd->LOCK_thd_kill);    // Lock from delete
     arg->thd= thd;
@@ -9110,27 +9110,6 @@ THD *find_thread_by_id(longlong id, bool query_id)
   return arg.thd;
 }
 
-#ifdef WITH_WSREP
-my_bool find_thread_with_thd_data_lock_callback(THD *thd, find_thread_callback_arg *arg)
-{
-  if (thd->get_command() != COM_DAEMON &&
-      arg->id == (arg->query_id ? thd->query_id : (longlong) thd->thread_id))
-  {
-    if (WSREP(thd)) mysql_mutex_lock(&thd->LOCK_thd_data);
-    mysql_mutex_lock(&thd->LOCK_thd_kill);    // Lock from delete
-    arg->thd= thd;
-    return 1;
-  }
-  return 0;
-}
-THD *find_thread_by_id_with_thd_data_lock(longlong id, bool query_id)
-{
-  find_thread_callback_arg arg(id, query_id);
-  server_threads.iterate(find_thread_with_thd_data_lock_callback, &arg);
-  return arg.thd;
-}
-#endif
-
 /**
   kill one thread.
 
@@ -9147,11 +9126,11 @@ kill_one_thread(THD *thd, longlong id, killed_state kill_signal, killed_type typ
   uint error= (type == KILL_TYPE_QUERY ? ER_NO_SUCH_QUERY : ER_NO_SUCH_THREAD);
   DBUG_ENTER("kill_one_thread");
   DBUG_PRINT("enter", ("id: %lld  signal: %u", id, (uint) kill_signal));
-#ifdef WITH_WSREP
-  if (id && (tmp= find_thread_by_id_with_thd_data_lock(id, type == KILL_TYPE_QUERY)))
-#else
-  if (id && (tmp= find_thread_by_id(id, type == KILL_TYPE_QUERY)))
-#endif
+  tmp= find_thread_by_id(id, type == KILL_TYPE_QUERY);
+  if (!tmp)
+    DBUG_RETURN(error);
+
+  if (tmp->get_command() != COM_DAEMON)
   {
     /*
       If we're SUPER, we can KILL anything, including system-threads.
@@ -9174,6 +9153,7 @@ kill_one_thread(THD *thd, longlong id, killed_state kill_signal, killed_type typ
       faster and do a harder kill than KILL_SYSTEM_THREAD;
     */
 
+    mysql_mutex_lock(&tmp->LOCK_thd_data); // for various wsrep* checks below
 #ifdef WITH_WSREP
     if (((thd->security_ctx->master_access & PRIV_KILL_OTHER_USER_PROCESS) ||
         thd->security_ctx->user_matches(tmp->security_ctx)) &&
@@ -9195,8 +9175,8 @@ kill_one_thread(THD *thd, longlong id, killed_state kill_signal, killed_type typ
       else
 #endif /* WITH_WSREP */
       {
-      WSREP_DEBUG("kill_one_thread %llu, victim: %llu wsrep_aborter %llu by signal %d",
-                  thd->thread_id, id, tmp->wsrep_aborter, kill_signal);
+        WSREP_DEBUG("kill_one_thread %llu, victim: %llu wsrep_aborter %llu by signal %d",
+                    thd->thread_id, id, tmp->wsrep_aborter, kill_signal);
         tmp->awake_no_mutex(kill_signal);
         WSREP_DEBUG("victim: %llu taken care of", id);
         error= 0;
@@ -9205,11 +9185,9 @@ kill_one_thread(THD *thd, longlong id, killed_state kill_signal, killed_type typ
     else
       error= (type == KILL_TYPE_QUERY ? ER_KILL_QUERY_DENIED_ERROR :
                                         ER_KILL_DENIED_ERROR);
-#ifdef WITH_WSREP
-    if (WSREP(tmp)) mysql_mutex_unlock(&tmp->LOCK_thd_data);
-#endif
-    mysql_mutex_unlock(&tmp->LOCK_thd_kill);
+    mysql_mutex_unlock(&tmp->LOCK_thd_data);
   }
+  mysql_mutex_unlock(&tmp->LOCK_thd_kill);
   DBUG_PRINT("exit", ("%d", error));
   DBUG_RETURN(error);
 }
@@ -9256,8 +9234,8 @@ static my_bool kill_threads_callback(THD *thd, kill_threads_callback_arg *arg)
         return 1;
       if (!arg->threads_to_kill.push_back(thd, arg->thd->mem_root))
       {
-        if (WSREP(thd)) mysql_mutex_lock(&thd->LOCK_thd_data);
         mysql_mutex_lock(&thd->LOCK_thd_kill); // Lock from delete
+        mysql_mutex_lock(&thd->LOCK_thd_data);
       }
     }
   }
@@ -9300,7 +9278,7 @@ static uint kill_threads_for_user(THD *thd, LEX_USER *user,
       */
       next_ptr= it2++;
       mysql_mutex_unlock(&ptr->LOCK_thd_kill);
-      if (WSREP(ptr)) mysql_mutex_unlock(&ptr->LOCK_thd_data);
+      mysql_mutex_unlock(&ptr->LOCK_thd_data);
       (*rows)++;
     } while ((ptr= next_ptr));
   }
diff --git a/sql/sql_plugin.h b/sql/sql_plugin.h
index e1360794a3b..eaf0b40f34a 100644
--- a/sql/sql_plugin.h
+++ b/sql/sql_plugin.h
@@ -22,10 +22,11 @@
   that is defined in plugin.h
 */
 #define SHOW_always_last SHOW_KEY_CACHE_LONG, \
-            SHOW_LONG_STATUS, SHOW_DOUBLE_STATUS, \
             SHOW_HAVE, SHOW_MY_BOOL, SHOW_HA_ROWS, SHOW_SYS, \
-            SHOW_LONG_NOFLUSH, SHOW_LONGLONG_STATUS, SHOW_UINT32_STATUS, \
-            SHOW_LEX_STRING, SHOW_ATOMIC_COUNTER_UINT32_T
+            SHOW_LONG_NOFLUSH, SHOW_LEX_STRING, SHOW_ATOMIC_COUNTER_UINT32_T, \
+      /* SHOW_*_STATUS must be at the end, SHOW_LONG_STATUS being first */ \
+            SHOW_LONG_STATUS, SHOW_DOUBLE_STATUS, SHOW_LONGLONG_STATUS, \
+            SHOW_UINT32_STATUS
 #include "mariadb.h"
 #undef SHOW_always_last
 
@@ -203,4 +204,3 @@ extern void sync_dynamic_session_variables(THD* thd, bool global_lock);
 extern void wsrep_plugins_pre_init();
 extern void wsrep_plugins_post_init();
 #endif /* WITH_WSREP */
-
diff --git a/sql/sql_plugin_services.ic b/sql/sql_plugin_services.ic
index 0911170fb74..25fa4b3afd1 100644
--- a/sql/sql_plugin_services.ic
+++ b/sql/sql_plugin_services.ic
@@ -179,7 +179,9 @@ static struct wsrep_service_st wsrep_handler = {
   wsrep_thd_has_ignored_error,
   wsrep_thd_set_ignored_error,
   wsrep_thd_set_wsrep_aborter,
-  wsrep_report_bf_lock_wait
+  wsrep_report_bf_lock_wait,
+  wsrep_thd_kill_LOCK,
+  wsrep_thd_kill_UNLOCK
 };
 
 static struct thd_specifics_service_st thd_specifics_handler=
@@ -255,4 +257,3 @@ static struct st_service_ref list_of_services[]=
   { "json_service",                VERSION_json,                &json_handler },
   { "thd_mdl_service",             VERSION_thd_mdl,             &thd_mdl_handler }
 };
-
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 864aaec180c..f0bb99da2fc 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -121,6 +121,7 @@ When one supplies long data for a placeholder:
 static const uint PARAMETER_FLAG_UNSIGNED= 128U << 8;
 #endif
 #include "lock.h"                               // MYSQL_OPEN_FORCE_SHARED_MDL
+#include "log_event.h"                          // class Log_event
 #include "sql_handler.h"
 #include "transaction.h"                        // trans_rollback_implicit
 #include "mysql/psi/mysql_ps.h"                 // MYSQL_EXECUTE_PS
@@ -2444,6 +2445,16 @@ static bool check_prepared_statement(Prepared_statement *stmt)
       DBUG_RETURN(FALSE);
     }
     break;
+  case SQLCOM_SHOW_BINLOG_EVENTS:
+  case SQLCOM_SHOW_RELAYLOG_EVENTS:
+    {
+      List<Item> field_list;
+      Log_event::init_show_field_list(thd, &field_list);
+
+      if ((res= send_stmt_metadata(thd, stmt, &field_list)) == 2)
+        DBUG_RETURN(FALSE);
+    }
+  break;
 #endif /* EMBEDDED_LIBRARY */
   case SQLCOM_SHOW_CREATE_PROC:
     if ((res= mysql_test_show_create_routine(stmt, &sp_handler_procedure)) == 2)
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index c7efcfd1074..010ea794254 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -3475,8 +3475,8 @@ static my_bool kill_callback(THD *thd, kill_callback_arg *arg)
       thd->variables.server_id == arg->slave_server_id)
   {
     arg->thd= thd;
-    if (WSREP(thd)) mysql_mutex_lock(&thd->LOCK_thd_data);
     mysql_mutex_lock(&thd->LOCK_thd_kill);    // Lock from delete
+    mysql_mutex_lock(&thd->LOCK_thd_data);
     return 1;
   }
   return 0;
@@ -3497,7 +3497,7 @@ void kill_zombie_dump_threads(uint32 slave_server_id)
     */
     arg.thd->awake_no_mutex(KILL_SLAVE_SAME_ID);
     mysql_mutex_unlock(&arg.thd->LOCK_thd_kill);
-    if (WSREP(arg.thd)) mysql_mutex_unlock(&arg.thd->LOCK_thd_data);
+    mysql_mutex_unlock(&arg.thd->LOCK_thd_data);
   }
 }
 
@@ -4683,5 +4683,22 @@ rpl_gtid_pos_update(THD *thd, char *str, size_t len)
     return false;
 }
 
+int compare_log_name(const char *log_1, const char *log_2) {
+  int res= 1;
+  const char *ext1_str= strrchr(log_1, '.');
+  const char *ext2_str= strrchr(log_2, '.');
+  char file_name_1[255], file_name_2[255];
+  strmake(file_name_1, log_1, (ext1_str - log_1));
+  strmake(file_name_2, log_2, (ext2_str - log_2));
+  char *endptr = NULL;
+  res= strcmp(file_name_1, file_name_2);
+  if (!res)
+  {
+    ulong ext1= strtoul(++ext1_str, &endptr, 10);
+    ulong ext2= strtoul(++ext2_str, &endptr, 10);
+    res= (ext1 > ext2 ? 1 : ((ext1 == ext2) ? 0 : -1));
+  }
+  return res;
+}
 
 #endif /* HAVE_REPLICATION */
diff --git a/sql/sql_repl.h b/sql/sql_repl.h
index 18aa7ea3fce..95916e31abf 100644
--- a/sql/sql_repl.h
+++ b/sql/sql_repl.h
@@ -45,6 +45,7 @@ bool show_binlogs(THD* thd);
 extern int init_master_info(Master_info* mi);
 void kill_zombie_dump_threads(uint32 slave_server_id);
 int check_binlog_magic(IO_CACHE* log, const char** errmsg);
+int compare_log_name(const char *log_1, const char *log_2);
 
 struct LOAD_FILE_IO_CACHE : public IO_CACHE
 {
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index dd2f16cd45d..b675e231ba9 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -2075,7 +2075,7 @@ JOIN::optimize_inner()
             join->optimization_state == JOIN::OPTIMIZATION_PHASE_1_DONE &&
             join->with_two_phase_optimization)
           continue;
-        /* 
+        /*
           Do not push conditions from where into materialized inner tables
           of outer joins: this is not valid.
         */
@@ -2277,7 +2277,7 @@ setup_subq_exit:
   if (with_two_phase_optimization)
     optimization_state= JOIN::OPTIMIZATION_PHASE_1_DONE;
   else
-  { 
+  {
     if (optimize_stage2())
       DBUG_RETURN(1);
   }
@@ -2297,7 +2297,7 @@ int JOIN::optimize_stage2()
 
   if (unlikely(thd->check_killed()))
     DBUG_RETURN(1);
-  
+
   /* Generate an execution plan from the found optimal join order. */
   if (get_best_combination())
     DBUG_RETURN(1);
@@ -3957,7 +3957,7 @@ bool JOIN::setup_subquery_caches()
     if (tmp_having)
     {
       DBUG_ASSERT(having == NULL);
-      if (!(tmp_having= 
+      if (!(tmp_having=
             tmp_having->transform(thd,
                                   &Item::expr_cache_insert_transformer,
                                   NULL)))
@@ -6929,7 +6929,7 @@ update_ref_and_keys(THD *thd, DYNAMIC_ARRAY *keyuse,JOIN_TAB *join_tab,
   Special treatment for ft-keys.
 */
 
-bool sort_and_filter_keyuse(THD *thd, DYNAMIC_ARRAY *keyuse, 
+bool sort_and_filter_keyuse(THD *thd, DYNAMIC_ARRAY *keyuse,
                             bool skip_unprefixed_keyparts)
 {
   KEYUSE key_end, *prev, *save_pos, *use;
@@ -8167,7 +8167,7 @@ best_access_path(JOIN      *join,
   pos->use_join_buffer= best_uses_jbuf;
   pos->spl_plan= spl_plan;
   pos->range_rowid_filter_info= best_filter;
-   
+
   loose_scan_opt.save_to_position(s, loose_scan_pos);
 
   if (!best_key &&
@@ -10312,7 +10312,7 @@ bool JOIN::check_two_phase_optimization(THD *thd)
     return true;
   return false;
 }
-  
+
 
 bool JOIN::inject_cond_into_where(Item *injected_cond)
 {
@@ -10343,7 +10343,7 @@ bool JOIN::inject_cond_into_where(Item *injected_cond)
       and_args->push_back(elem, thd->mem_root);
     }
   }
-  
+
   return false;
 
 }
@@ -18488,9 +18488,9 @@ bool Create_tmp_table::add_fields(THD *thd,
       distinct_record_structure= true;
     }
   li.rewind();
-  uint uneven_delta= 0;
   while ((item=li++))
   {
+    uint uneven_delta;
     current_counter= (((param->hidden_field_count < (fieldnr + 1)) &&
                        distinct_record_structure &&
                        (!m_with_cycle ||
@@ -18553,8 +18553,8 @@ bool Create_tmp_table::add_fields(THD *thd,
 
           uneven_delta= m_uneven_bit_length;
           add_field(table, new_field, fieldnr++, param->force_not_null_cols);
-          uneven_delta= m_uneven_bit_length - uneven_delta;
           m_field_count[current_counter]++;
+          m_uneven_bit[current_counter]+= (m_uneven_bit_length - uneven_delta);
 
           if (!(new_field->flags & NOT_NULL_FLAG))
           {
@@ -18635,8 +18635,8 @@ bool Create_tmp_table::add_fields(THD *thd,
 
       uneven_delta= m_uneven_bit_length;
       add_field(table, new_field, fieldnr++, param->force_not_null_cols);
-      uneven_delta= m_uneven_bit_length - uneven_delta;
       m_field_count[current_counter]++;
+      m_uneven_bit[current_counter]+= (m_uneven_bit_length - uneven_delta);
 
       if (item->marker == 4 && item->maybe_null)
       {
@@ -18646,7 +18646,6 @@ bool Create_tmp_table::add_fields(THD *thd,
       if (current_counter == distinct)
         new_field->flags|= FIELD_PART_OF_TMP_UNIQUE;
     }
-    m_uneven_bit[current_counter]+= uneven_delta;
   }
   DBUG_ASSERT(fieldnr == m_field_count[other] + m_field_count[distinct]);
   DBUG_ASSERT(m_blob_count == m_blobs_count[other] + m_blobs_count[distinct]);
@@ -18805,7 +18804,6 @@ bool Create_tmp_table::finalize(THD *thd,
 
     if (!(field->flags & NOT_NULL_FLAG))
     {
-
       recinfo->null_bit= (uint8)1 << (null_counter[current_counter] & 7);
       recinfo->null_pos= (null_pack_base[current_counter] +
                           null_counter[current_counter]/8);
@@ -24297,7 +24295,7 @@ bool
 cp_buffer_from_ref(THD *thd, TABLE *table, TABLE_REF *ref)
 {
   Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);
-  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->write_set);
+  MY_BITMAP *old_map= dbug_tmp_use_all_columns(table, &table->write_set);
   bool result= 0;
 
   for (store_key **copy=ref->key_copy ; *copy ; copy++)
@@ -24308,7 +24306,7 @@ cp_buffer_from_ref(THD *thd, TABLE *table, TABLE_REF *ref)
       break;
     }
   }
-  dbug_tmp_restore_column_map(table->write_set, old_map);
+  dbug_tmp_restore_column_map(&table->write_set, old_map);
   return result;
 }
 
@@ -25980,7 +25978,7 @@ bool JOIN::rollup_init()
   {
     if (!(rollup.null_items[i]= new (thd->mem_root) Item_null_result(thd)))
       return true;
-    
+
     List<Item> *rollup_fields= &rollup.fields[i];
     rollup_fields->empty();
     rollup.ref_pointer_arrays[i]= Ref_ptr_array(ref_array, all_fields.elements);
@@ -26493,7 +26491,7 @@ bool JOIN_TAB::save_explain_data(Explain_table_access *eta,
   {
     JOIN_TAB *ctab= bush_children->start;
     /* table */
-    size_t len= my_snprintf(table_name_buffer, 
+    size_t len= my_snprintf(table_name_buffer,
                          sizeof(table_name_buffer)-1,
                          "<subquery%d>", 
                          ctab->emb_sj_nest->sj_subq_pred->get_identifier());
@@ -27636,7 +27634,7 @@ void TABLE_LIST::print(THD *thd, table_map eliminated_tables, String *str,
 void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
 {
   DBUG_ASSERT(thd);
-  
+
   if (tvc)
   {
     tvc->print(thd, str, query_type);
diff --git a/sql/sql_select.h b/sql/sql_select.h
index f39a70cc3e1..ff9264d8bf7 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -1950,8 +1950,8 @@ class store_key_field: public store_key
   enum store_key_result copy_inner()
   {
     TABLE *table= copy_field.to_field->table;
-    my_bitmap_map *old_map= dbug_tmp_use_all_columns(table,
-                                                     table->write_set);
+    MY_BITMAP *old_map= dbug_tmp_use_all_columns(table,
+                                                 &table->write_set);
 
     /* 
       It looks like the next statement is needed only for a simplified
@@ -1962,7 +1962,7 @@ class store_key_field: public store_key
     bzero(copy_field.to_ptr,copy_field.to_length);
 
     copy_field.do_copy(&copy_field);
-    dbug_tmp_restore_column_map(table->write_set, old_map);
+    dbug_tmp_restore_column_map(&table->write_set, old_map);
     null_key= to_field->is_null();
     return err != 0 ? STORE_KEY_FATAL : STORE_KEY_OK;
   }
@@ -1997,8 +1997,8 @@ public:
   enum store_key_result copy_inner()
   {
     TABLE *table= to_field->table;
-    my_bitmap_map *old_map= dbug_tmp_use_all_columns(table,
-                                                     table->write_set);
+    MY_BITMAP *old_map= dbug_tmp_use_all_columns(table,
+                                                 &table->write_set);
     int res= FALSE;
 
     /* 
@@ -2019,7 +2019,7 @@ public:
     */
     if (!res && table->in_use->is_error())
       res= 1; /* STORE_KEY_FATAL */
-    dbug_tmp_restore_column_map(table->write_set, old_map);
+    dbug_tmp_restore_column_map(&table->write_set, old_map);
     null_key= to_field->is_null() || item->null_value;
     return ((err != 0 || res < 0 || res > 2) ? STORE_KEY_FATAL : 
             (store_key_result) res);
@@ -2055,8 +2055,8 @@ protected:
     {
       inited=1;
       TABLE *table= to_field->table;
-      my_bitmap_map *old_map= dbug_tmp_use_all_columns(table,
-                                                       table->write_set);
+      MY_BITMAP *old_map= dbug_tmp_use_all_columns(table,
+                                                   &table->write_set);
       if ((res= item->save_in_field(to_field, 1)))
       {       
         if (!err)
@@ -2068,7 +2068,7 @@ protected:
         */
       if (!err && to_field->table->in_use->is_error())
         err= 1; /* STORE_KEY_FATAL */
-      dbug_tmp_restore_column_map(table->write_set, old_map);
+      dbug_tmp_restore_column_map(&table->write_set, old_map);
     }
     null_key= to_field->is_null() || item->null_value;
     return (err > 2 ? STORE_KEY_FATAL : (store_key_result) err);
diff --git a/sql/sql_sequence.cc b/sql/sql_sequence.cc
index b9d500d463c..f881f6a8cc6 100644
--- a/sql/sql_sequence.cc
+++ b/sql/sql_sequence.cc
@@ -137,7 +137,7 @@ bool sequence_definition::check_and_adjust(bool set_reserved_until)
 
 void sequence_definition::read_fields(TABLE *table)
 {
-  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->read_set);
+  MY_BITMAP *old_map= dbug_tmp_use_all_columns(table, &table->read_set);
   reserved_until= table->field[0]->val_int();
   min_value=      table->field[1]->val_int();
   max_value=      table->field[2]->val_int();
@@ -146,7 +146,7 @@ void sequence_definition::read_fields(TABLE *table)
   cache=          table->field[5]->val_int();
   cycle=          table->field[6]->val_int();
   round=          table->field[7]->val_int();
-  dbug_tmp_restore_column_map(table->read_set, old_map);
+  dbug_tmp_restore_column_map(&table->read_set, old_map);
   used_fields= ~(uint) 0;
   print_dbug();
 }
@@ -158,7 +158,7 @@ void sequence_definition::read_fields(TABLE *table)
 
 void sequence_definition::store_fields(TABLE *table)
 {
-  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->write_set);
+  MY_BITMAP *old_map= dbug_tmp_use_all_columns(table, &table->write_set);
 
   /* zero possible delete markers & null bits */
   memcpy(table->record[0], table->s->default_values, table->s->null_bytes);
@@ -171,7 +171,7 @@ void sequence_definition::store_fields(TABLE *table)
   table->field[6]->store((longlong) cycle != 0, 0);
   table->field[7]->store((longlong) round, 1);
 
-  dbug_tmp_restore_column_map(table->write_set, old_map);
+  dbug_tmp_restore_column_map(&table->write_set, old_map);
   print_dbug();
 }
 
@@ -526,12 +526,11 @@ int SEQUENCE::read_initial_values(TABLE *table)
 int SEQUENCE::read_stored_values(TABLE *table)
 {
   int error;
-  my_bitmap_map *save_read_set;
   DBUG_ENTER("SEQUENCE::read_stored_values");
 
-  save_read_set= tmp_use_all_columns(table, table->read_set);
+  MY_BITMAP *save_read_set= tmp_use_all_columns(table, &table->read_set);
   error= table->file->ha_read_first_row(table->record[0], MAX_KEY);
-  tmp_restore_column_map(table->read_set, save_read_set);
+  tmp_restore_column_map(&table->read_set, save_read_set);
 
   if (unlikely(error))
   {
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 4c5e877cce8..6fe45fe2de3 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -2051,7 +2051,6 @@ int show_create_table_ex(THD *thd, TABLE_LIST *table_list,
                            !foreign_db_mode;
   bool check_options= !(sql_mode & MODE_IGNORE_BAD_TABLE_OPTIONS) &&
                       !create_info_arg;
-  my_bitmap_map *old_map;
   handlerton *hton;
   int error= 0;
   DBUG_ENTER("show_create_table");
@@ -2132,7 +2131,7 @@ int show_create_table_ex(THD *thd, TABLE_LIST *table_list,
     We have to restore the read_set if we are called from insert in case
     of row based replication.
   */
-  old_map= tmp_use_all_columns(table, table->read_set);
+  MY_BITMAP *old_map= tmp_use_all_columns(table, &table->read_set);
 
   bool not_the_first_field= false;
   for (ptr=table->field ; (field= *ptr); ptr++)
@@ -2171,8 +2170,13 @@ int show_create_table_ex(THD *thd, TABLE_LIST *table_list,
       /*
 	For string types dump collation name only if
 	collation is not primary for the given charset
+
+        For generated fields don't print the COLLATE clause if
+        the collation matches the expression's collation.
       */
-      if (!(field->charset()->state & MY_CS_PRIMARY) && !field->vcol_info)
+      if (!(field->charset()->state & MY_CS_PRIMARY) &&
+          (!field->vcol_info ||
+           field->charset() != field->vcol_info->expr->collation.collation))
       {
 	packet->append(STRING_WITH_LEN(" COLLATE "));
 	packet->append(field->charset()->name);
@@ -2439,7 +2443,7 @@ int show_create_table_ex(THD *thd, TABLE_LIST *table_list,
     }
   }
 #endif
-  tmp_restore_column_map(table->read_set, old_map);
+  tmp_restore_column_map(&table->read_set, old_map);
   DBUG_RETURN(error);
 }
 
@@ -3173,11 +3177,8 @@ int fill_show_explain(THD *thd, TABLE_LIST *table, COND *cond)
     }
     DBUG_RETURN(bres);
   }
-  else
-  {
-    my_error(ER_NO_SUCH_THREAD, MYF(0), (ulong) thread_id);
-    DBUG_RETURN(1);
-  }
+  my_error(ER_NO_SUCH_THREAD, MYF(0), (ulong) thread_id);
+  DBUG_RETURN(1);
 }
 
 
@@ -3771,6 +3772,16 @@ static bool show_status_array(THD *thd, const char *wild,
 
         if (show_type == SHOW_SYS)
           mysql_mutex_lock(&LOCK_global_system_variables);
+        else if (show_type >= SHOW_LONG_STATUS && scope == OPT_GLOBAL &&
+                 !status_var->local_memory_used)
+        {
+          mysql_mutex_lock(&LOCK_status);
+          *status_var= global_status_var;
+          mysql_mutex_unlock(&LOCK_status);
+          calc_sum_of_all_status(status_var);
+          DBUG_ASSERT(status_var->local_memory_used);
+        }
+
         pos= get_one_variable(thd, var, scope, show_type, status_var,
                               &charset, buff, &length);
 
@@ -3828,7 +3839,6 @@ uint calc_sum_of_all_status(STATUS_VAR *to)
   calc_sum_callback_arg arg(to);
   DBUG_ENTER("calc_sum_of_all_status");
 
-  *to= global_status_var;
   to->local_memory_used= 0;
   /* Add to this status from existing threads */
   server_threads.iterate(calc_sum_callback, &arg);
@@ -5237,6 +5247,12 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
                 continue;
             }
 
+            if (thd->killed == ABORT_QUERY)
+            {
+              error= 0;
+              goto err;
+            }
+
             DEBUG_SYNC(thd, "before_open_in_get_all_tables");
             if (fill_schema_table_by_open(thd, &tmp_mem_root, FALSE,
                                           table, schema_table,
@@ -5824,7 +5840,7 @@ static bool print_anchor_data_type(const Spvar_definition *def,
   Let's print it according to the current sql_mode.
   It will make output in line with the value in mysql.proc.param_list,
   so both I_S.XXX.DTD_IDENTIFIER and mysql.proc.param_list use the same notation:
-  default or Oracle, according to the sql_mode at the SP creation time. 
+  default or Oracle, according to the sql_mode at the SP creation time.
   The caller must make sure to set thd->variables.sql_mode to the routine sql_mode.
 */
 static bool print_anchor_dtd_identifier(THD *thd, const Spvar_definition *def,
@@ -7862,10 +7878,7 @@ int fill_status(THD *thd, TABLE_LIST *tables, COND *cond)
   if (partial_cond)
     partial_cond->val_int();
 
-  if (scope == OPT_GLOBAL)
-  {
-    calc_sum_of_all_status(&tmp);
-  }
+  tmp.local_memory_used= 0; // meaning tmp was not populated yet
 
   mysql_rwlock_rdlock(&LOCK_all_status_vars);
   res= show_status_array(thd, wild,
diff --git a/sql/sql_statistics.cc b/sql/sql_statistics.cc
index 7b600bd45c4..72a55bfa86b 100644
--- a/sql/sql_statistics.cc
+++ b/sql/sql_statistics.cc
@@ -1023,9 +1023,8 @@ public:
   {
     char buff[MAX_FIELD_WIDTH];
     String val(buff, sizeof(buff), &my_charset_bin);
-    my_bitmap_map *old_map;
 
-    old_map= dbug_tmp_use_all_columns(stat_table, stat_table->read_set);
+    MY_BITMAP *old_map= dbug_tmp_use_all_columns(stat_table, &stat_table->read_set);
     for (uint i= COLUMN_STAT_MIN_VALUE; i <= COLUMN_STAT_HISTOGRAM; i++)
     {  
       Field *stat_field= stat_table->field[i];
@@ -1087,7 +1086,7 @@ public:
         }
       }
     }
-    dbug_tmp_restore_column_map(stat_table->read_set, old_map);
+    dbug_tmp_restore_column_map(&stat_table->read_set, old_map);
   }
 
 
@@ -2126,6 +2125,10 @@ int alloc_statistics_for_table(THD* thd, TABLE *table)
   ulonglong *idx_avg_frequency= (ulonglong*) alloc_root(&table->mem_root,
                                                sizeof(ulonglong) * key_parts);
 
+  if (table->file->ha_rnd_init(TRUE))
+    DBUG_RETURN(1);
+  table->file->ha_rnd_end();
+
   uint columns= 0;
   for (field_ptr= table->field; *field_ptr; field_ptr++)
   {
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index c9f8c94dffb..c9e21e36b45 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -80,9 +80,8 @@ static int append_system_key_parts(THD *thd, HA_CREATE_INFO *create_info,
 static int mysql_prepare_create_table(THD *, HA_CREATE_INFO *, Alter_info *,
                                       uint *, handler *, KEY **, uint *, int);
 static uint blob_length_by_type(enum_field_types type);
-static bool fix_constraints_names(THD *thd, List<Virtual_column_info>
-                                  *check_constraint_list,
-                                  const HA_CREATE_INFO *create_info);
+static bool fix_constraints_names(THD *, List<Virtual_column_info> *,
+                                  const HA_CREATE_INFO *);
 
 /**
   @brief Helper function for explain_filename
@@ -4250,8 +4249,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
               key_part_length= MY_MIN(max_key_length, file->max_key_part_length());
               /* not a critical problem */
               push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
-                                  ER_TOO_LONG_KEY,
-                                  ER_THD(thd, ER_TOO_LONG_KEY),
+                                  ER_TOO_LONG_KEY, ER_THD(thd, ER_TOO_LONG_KEY),
                                   key_part_length);
               /* Align key length to multibyte char boundary */
               key_part_length-= key_part_length % sql_field->charset->mbmaxlen;
@@ -4295,7 +4293,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         {
           key_part_length= file->max_key_part_length();
           /* not a critical problem */
-          push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+          push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
                               ER_TOO_LONG_KEY, ER_THD(thd, ER_TOO_LONG_KEY),
                               key_part_length);
           /* Align key length to multibyte char boundary */
@@ -4546,8 +4544,6 @@ without_overlaps_err:
         const Virtual_column_info *dup_check;
         while ((dup_check= dup_it++) && dup_check != check)
         {
-          if (!dup_check->name.length || dup_check->automatic_name)
-            continue;
           if (!lex_string_cmp(system_charset_info,
                               &check->name, &dup_check->name))
           {
@@ -4597,7 +4593,8 @@ without_overlaps_err:
                           ER_ILLEGAL_HA_CREATE_OPTION,
                           ER_THD(thd, ER_ILLEGAL_HA_CREATE_OPTION),
                           file->engine_name()->str,
-                          "TRANSACTIONAL=1");
+                          create_info->transactional == HA_CHOICE_YES
+                          ? "TRANSACTIONAL=1" : "TRANSACTIONAL=0");
 
   if (parse_option_list(thd, file->partition_ht(), &create_info->option_struct,
                           &create_info->option_list,
@@ -5561,6 +5558,9 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   if (!opt_explicit_defaults_for_timestamp)
     promote_first_timestamp_column(&alter_info->create_list);
 
+  /* We can abort create table for any table type */
+  thd->abort_on_warning= thd->is_strict_mode();
+
   if (mysql_create_table_no_lock(thd, &create_table->db,
                                  &create_table->table_name, create_info,
                                  alter_info,
@@ -5598,6 +5598,8 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   }
 
 err:
+  thd->abort_on_warning= 0;
+
   /* In RBR or readonly server we don't need to log CREATE TEMPORARY TABLE */
   if (!result && create_info->tmp_table() &&
       (thd->is_current_stmt_binlog_format_row() || (opt_readonly && !thd->slave_thread)))
@@ -9068,37 +9070,28 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
         }
       }
 
-      // NB: `check` is TABLE resident, we must keep it intact.
-      if (keep)
-      {
-        check= check->clone(thd);
-        if (!check)
-        {
-          my_error(ER_OUT_OF_RESOURCES, MYF(0));
-          goto err;
-        }
-      }
-
       if (share->period.constr_name.streq(check->name.str))
       {
-        if (drop_period)
-        {
-          keep= false;
-        }
-        else if(!keep)
+        if (!drop_period && !keep)
         {
           my_error(ER_PERIOD_CONSTRAINT_DROP, MYF(0), check->name.str,
                    share->period.name.str);
           goto err;
         }
-        else
+        keep= keep && !drop_period;
+
+        DBUG_ASSERT(create_info->period_info.constr == NULL || drop_period);
+
+        if (keep)
         {
-          DBUG_ASSERT(create_info->period_info.constr == NULL);
+          Item *expr_copy= check->expr->get_copy(thd);
+          check= new Virtual_column_info();
+          check->name= share->period.constr_name;
+          check->automatic_name= true;
+          check->expr= expr_copy;
           create_info->period_info.constr= check;
-          create_info->period_info.constr->automatic_name= true;
         }
       }
-
       /* see if the constraint depends on *only* on dropped fields */
       if (keep && dropped_fields)
       {
diff --git a/sql/sql_test.cc b/sql/sql_test.cc
index cc3c9badefb..3f1728d334f 100644
--- a/sql/sql_test.cc
+++ b/sql/sql_test.cc
@@ -565,6 +565,7 @@ void mysql_print_status()
   STATUS_VAR tmp;
   uint count;
 
+  tmp= global_status_var;
   count= calc_sum_of_all_status(&tmp);
   printf("\nStatus information:\n\n");
   (void) my_getwd(current_dir, sizeof(current_dir),MYF(0));
diff --git a/sql/sql_tvc.cc b/sql/sql_tvc.cc
index 5f732d474f8..9a7084cdca2 100644
--- a/sql/sql_tvc.cc
+++ b/sql/sql_tvc.cc
@@ -78,7 +78,7 @@ bool fix_fields_for_tvc(THD *thd, List_iterator_fast<List_item> &li)
 
   while ((lst= li++))
   {
-    List_iterator_fast<Item> it(*lst);
+    List_iterator<Item> it(*lst);
     Item *item;
 
     while ((item= it++))
@@ -90,7 +90,7 @@ bool fix_fields_for_tvc(THD *thd, List_iterator_fast<List_item> &li)
         while replacing their values to NAME_CONST()s.
         So fix only those that have not been.
       */
-      if (item->fix_fields_if_needed(thd, 0) ||
+      if (item->fix_fields_if_needed_for_scalar(thd, it.ref()) ||
           item->check_is_evaluable_expression_or_error())
 	DBUG_RETURN(true);
     }
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 4424d7b5572..76a7dc47745 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1720,6 +1720,9 @@ bool Multiupdate_prelocking_strategy::handle_end(THD *thd)
   if (select_lex->handle_derived(thd->lex, DT_MERGE))
     DBUG_RETURN(1);
 
+  if (thd->lex->save_prep_leaf_tables())
+    DBUG_RETURN(1);
+
   List<Item> *fields= &lex->first_select_lex()->item_list;
   if (setup_fields_with_no_wrap(thd, Ref_ptr_array(),
                                 *fields, MARK_COLUMNS_WRITE, 0, 0))
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 5aeb14535e3..69d2aa9f470 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -341,12 +341,12 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 */
 
 /* Start SQL_MODE_DEFAULT_SPECIFIC */
-%expect 46
+%expect 67
 /* End SQL_MODE_DEFAULT_SPECIFIC */
 
 
 /* Start SQL_MODE_ORACLE_SPECIFIC
-%expect 49
+%expect 69
 End SQL_MODE_ORACLE_SPECIFIC */
 
 
@@ -1356,7 +1356,7 @@ End SQL_MODE_ORACLE_SPECIFIC */
 
 %type <sp_handler> sp_handler
 
-%type <Lex_field_type> type_with_opt_collate field_type
+%type <Lex_field_type> field_type field_type_all
         qualified_field_type
         field_type_numeric
         field_type_string
@@ -3002,7 +3002,7 @@ sp_param_name:
         ;
 
 sp_param_name_and_type:
-          sp_param_name type_with_opt_collate
+          sp_param_name field_type
           {
             if (unlikely(Lex->sp_param_fill_definition($$= $1, $2)))
               MYSQL_YYABORT;
@@ -3082,7 +3082,7 @@ optionally_qualified_column_ident:
 
 
 row_field_definition:
-          row_field_name type_with_opt_collate
+          row_field_name field_type
           {
             Lex->last_field->set_attributes(thd, $2, Lex->charset,
                                             COLUMN_DEFINITION_ROUTINE_LOCAL);
@@ -3115,7 +3115,7 @@ sp_decl_idents_init_vars:
 
 sp_decl_variable_list:
           sp_decl_idents_init_vars
-          type_with_opt_collate
+          field_type
           {
             Lex->last_field->set_attributes(thd, $2, Lex->charset,
                                             COLUMN_DEFINITION_ROUTINE_LOCAL);
@@ -6188,19 +6188,25 @@ column_default_expr:
           }
         ;
 
+field_type: field_type_all
+        {
+          Lex->map_data_type(Lex_ident_sys(), &($$= $1));
+        }
+        ;
+
 qualified_field_type:
-          field_type
+          field_type_all
           {
             Lex->map_data_type(Lex_ident_sys(), &($$= $1));
           }
-        | sp_decl_ident '.' field_type
+        | sp_decl_ident '.' field_type_all
           {
             if (Lex->map_data_type($1, &($$= $3)))
               MYSQL_YYABORT;
           }
         ;
 
-field_type:
+field_type_all:
           field_type_numeric
         | field_type_temporal
         | field_type_string
@@ -6663,19 +6669,6 @@ with_or_without_system:
         ;
 
 
-type_with_opt_collate:
-        field_type opt_collate
-        {
-          Lex->map_data_type(Lex_ident_sys(), &($$= $1));
-
-          if ($2)
-          {
-            if (unlikely(!(Lex->charset= merge_charset_and_collation(Lex->charset, $2))))
-              MYSQL_YYABORT;
-          }
-        }
-        ;
-
 charset:
           CHAR_SYM SET { $$= $1; }
         | CHARSET { $$= $1; }
@@ -6749,6 +6742,12 @@ charset_or_alias:
           }
         ;
 
+collate: COLLATE_SYM collation_name_or_default
+         {
+           Lex->charset= $2;
+         }
+       ;
+
 opt_binary:
           /* empty */             { bincmp_collation(NULL, false); }
         | binary {}
@@ -6759,6 +6758,13 @@ binary:
         | charset_or_alias opt_bin_mod { bincmp_collation($1, $2); }
         | BINARY                  { bincmp_collation(NULL, true); }
         | BINARY charset_or_alias { bincmp_collation($2, true); }
+        | charset_or_alias collate
+          {
+            if (!my_charset_same(Lex->charset, $1))
+              my_yyabort_error((ER_COLLATION_CHARSET_MISMATCH, MYF(0),
+                                Lex->charset->name, $1->csname));
+          }
+        | collate { }
         ;
 
 opt_bin_mod:
@@ -14323,7 +14329,7 @@ kill:
             lex->sql_command= SQLCOM_KILL;
             lex->kill_type= KILL_TYPE_ID;
           }
-          kill_type kill_option kill_expr
+          kill_type kill_option
           {
             Lex->kill_signal= (killed_state) ($3 | $4);
           }
@@ -14336,16 +14342,21 @@ kill_type:
         ;
 
 kill_option:
-          /* empty */    { $$= (int) KILL_CONNECTION; }
-        | CONNECTION_SYM { $$= (int) KILL_CONNECTION; }
-        | QUERY_SYM      { $$= (int) KILL_QUERY; }
-        | QUERY_SYM ID_SYM
+          opt_connection kill_expr { $$= (int) KILL_CONNECTION; }
+        | QUERY_SYM      kill_expr { $$= (int) KILL_QUERY; }
+        | QUERY_SYM ID_SYM expr
           {
             $$= (int) KILL_QUERY;
             Lex->kill_type= KILL_TYPE_QUERY;
+            Lex->value_list.push_front($3, thd->mem_root);
           }
         ;
 
+opt_connection:
+          /* empty */    { }
+        | CONNECTION_SYM { }
+        ;
+
 kill_expr:
         expr
         {
@@ -14358,7 +14369,6 @@ kill_expr:
           }
         ;
 
-
 shutdown:
         SHUTDOWN { Lex->sql_command= SQLCOM_SHUTDOWN; }
         shutdown_option {}
@@ -17646,7 +17656,7 @@ sf_return_type:
                                  &empty_clex_str,
                                  thd->variables.collation_database);
           }
-          type_with_opt_collate
+          field_type
           {
             if (unlikely(Lex->sf_return_fill_definition($2)))
               MYSQL_YYABORT;
@@ -18339,7 +18349,7 @@ sp_opt_inout:
         ;
 
 sp_pdparam:
-          sp_param_name sp_opt_inout type_with_opt_collate
+          sp_param_name sp_opt_inout field_type
           {
             $1->mode= $2;
             if (unlikely(Lex->sp_param_fill_definition($1, $3)))
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index a75066271d9..0768d4f89c4 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -5696,13 +5696,12 @@ static Sys_var_charptr Sys_wsrep_cluster_name(
        ON_CHECK(wsrep_cluster_name_check),
        ON_UPDATE(wsrep_cluster_name_update));
 
-static PolyLock_mutex PLock_wsrep_cluster_config(&LOCK_wsrep_cluster_config);
 static Sys_var_charptr Sys_wsrep_cluster_address (
        "wsrep_cluster_address", "Address to initially connect to cluster",
        PREALLOCATED GLOBAL_VAR(wsrep_cluster_address), 
        CMD_LINE(REQUIRED_ARG),
        DEFAULT(""),
-       &PLock_wsrep_cluster_config, NOT_IN_BINLOG,
+       NO_MUTEX_GUARD, NOT_IN_BINLOG,
        ON_CHECK(wsrep_cluster_address_check), 
        ON_UPDATE(wsrep_cluster_address_update));
 
@@ -5733,7 +5732,7 @@ static Sys_var_ulong Sys_wsrep_slave_threads(
        "wsrep_slave_threads", "Number of slave appliers to launch",
        GLOBAL_VAR(wsrep_slave_threads), CMD_LINE(REQUIRED_ARG),
        VALID_RANGE(1, 512), DEFAULT(1), BLOCK_SIZE(1),
-       &PLock_wsrep_cluster_config, NOT_IN_BINLOG,
+       NO_MUTEX_GUARD, NOT_IN_BINLOG,
        ON_CHECK(0),
        ON_UPDATE(wsrep_slave_threads_update));
 
diff --git a/sql/table.h b/sql/table.h
index c48eae94df8..758d2c3d949 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -3041,25 +3041,25 @@ typedef struct st_open_table_list{
 } OPEN_TABLE_LIST;
 
 
-static inline my_bitmap_map *tmp_use_all_columns(TABLE *table,
-                                                 MY_BITMAP *bitmap)
+static inline MY_BITMAP *tmp_use_all_columns(TABLE *table,
+                                             MY_BITMAP **bitmap)
 {
-  my_bitmap_map *old= bitmap->bitmap;
-  bitmap->bitmap= table->s->all_set.bitmap;
+  MY_BITMAP *old= *bitmap;
+  *bitmap= &table->s->all_set;
   return old;
 }
 
 
-static inline void tmp_restore_column_map(MY_BITMAP *bitmap,
-                                          my_bitmap_map *old)
+static inline void tmp_restore_column_map(MY_BITMAP **bitmap,
+                                          MY_BITMAP *old)
 {
-  bitmap->bitmap= old;
+  *bitmap= old;
 }
 
 /* The following is only needed for debugging */
 
-static inline my_bitmap_map *dbug_tmp_use_all_columns(TABLE *table,
-                                                      MY_BITMAP *bitmap)
+static inline MY_BITMAP *dbug_tmp_use_all_columns(TABLE *table,
+                                                      MY_BITMAP **bitmap)
 {
 #ifdef DBUG_ASSERT_EXISTS
   return tmp_use_all_columns(table, bitmap);
@@ -3068,8 +3068,8 @@ static inline my_bitmap_map *dbug_tmp_use_all_columns(TABLE *table,
 #endif
 }
 
-static inline void dbug_tmp_restore_column_map(MY_BITMAP *bitmap,
-                                               my_bitmap_map *old)
+static inline void dbug_tmp_restore_column_map(MY_BITMAP **bitmap,
+                                               MY_BITMAP *old)
 {
 #ifdef DBUG_ASSERT_EXISTS
   tmp_restore_column_map(bitmap, old);
@@ -3082,22 +3082,22 @@ static inline void dbug_tmp_restore_column_map(MY_BITMAP *bitmap,
   Provide for the possiblity of the read set being the same as the write set
 */
 static inline void dbug_tmp_use_all_columns(TABLE *table,
-                                            my_bitmap_map **save,
-                                            MY_BITMAP *read_set,
-                                            MY_BITMAP *write_set)
+                                            MY_BITMAP **save,
+                                            MY_BITMAP **read_set,
+                                            MY_BITMAP **write_set)
 {
 #ifdef DBUG_ASSERT_EXISTS
-  save[0]= read_set->bitmap;
-  save[1]= write_set->bitmap;
+  save[0]= *read_set;
+  save[1]= *write_set;
   (void) tmp_use_all_columns(table, read_set);
   (void) tmp_use_all_columns(table, write_set);
 #endif
 }
 
 
-static inline void dbug_tmp_restore_column_maps(MY_BITMAP *read_set,
-                                                MY_BITMAP *write_set,
-                                                my_bitmap_map **old)
+static inline void dbug_tmp_restore_column_maps(MY_BITMAP **read_set,
+                                                MY_BITMAP **write_set,
+                                                MY_BITMAP **old)
 {
 #ifdef DBUG_ASSERT_EXISTS
   tmp_restore_column_map(read_set, old[0]);
diff --git a/sql/unireg.cc b/sql/unireg.cc
index bffe51b6c49..51c4eeb4a4c 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -57,13 +57,7 @@ static bool make_empty_rec(THD *, uchar *, uint, List<Create_field> &, uint,
 */
 static uchar *extra2_write_len(uchar *pos, size_t len)
 {
-  /* TODO: should be
-     if (len > 0 && len <= 255)
-       *pos++= (uchar)len;
-     ...
-     because extra2_read_len() uses 0 for 2-byte lengths.
-     extra2_str_size() must be fixed too.
-  */
+  DBUG_ASSERT(len);
   if (len <= 255)
     *pos++= (uchar)len;
   else
@@ -1151,6 +1145,8 @@ static bool make_empty_rec(THD *thd, uchar *buff, uint table_options,
   TABLE table;
   TABLE_SHARE share;
   Create_field *field;
+  Check_level_instant_set old_count_cuted_fields(thd, CHECK_FIELD_WARN);
+  Abort_on_warning_instant_set old_abort_on_warning(thd, 0);
   DBUG_ENTER("make_empty_rec");
 
   /* We need a table to generate columns for default values */
@@ -1169,7 +1165,6 @@ static bool make_empty_rec(THD *thd, uchar *buff, uint table_options,
   null_pos= buff;
 
   List_iterator<Create_field> it(create_fields);
-  Check_level_instant_set check_level_save(thd, CHECK_FIELD_WARN);
   while ((field=it++))
   {
     Record_addr addr(buff + field->offset + data_offset,
diff --git a/sql/wsrep_check_opts.cc b/sql/wsrep_check_opts.cc
index 935bacffffc..e5a0dcb2ede 100644
--- a/sql/wsrep_check_opts.cc
+++ b/sql/wsrep_check_opts.cc
@@ -63,7 +63,7 @@ int wsrep_check_opts()
   else
   {
     // non-mysqldump SST requires wsrep_cluster_address on startup
-    if (!wsrep_cluster_address || !wsrep_cluster_address[0])
+    if (!wsrep_cluster_address_exists())
     {
       WSREP_ERROR ("%s SST method requires wsrep_cluster_address to be "
                    "configured on startup.", wsrep_sst_method);
diff --git a/sql/wsrep_client_service.cc b/sql/wsrep_client_service.cc
index 3131e753e60..3eb457a01f3 100644
--- a/sql/wsrep_client_service.cc
+++ b/sql/wsrep_client_service.cc
@@ -68,20 +68,13 @@ bool Wsrep_client_service::interrupted(
   wsrep::unique_lock<wsrep::mutex>& lock WSREP_UNUSED) const
 {
   DBUG_ASSERT(m_thd == current_thd);
-  /* Underlying mutex in lock object points to LOCK_thd_data, which
-     protects m_thd->wsrep_trx(), LOCK_thd_kill protects m_thd->killed.
-     Locking order is:
-     1) LOCK_thd_data
-     2) LOCK_thd_kill */
   mysql_mutex_assert_owner(static_cast<mysql_mutex_t*>(lock.mutex()->native()));
-  mysql_mutex_lock(&m_thd->LOCK_thd_kill);
   bool ret= (m_thd->killed != NOT_KILLED);
   if (ret)
   {
     WSREP_DEBUG("wsrep state is interrupted, THD::killed %d trx state %d",
                 m_thd->killed,  m_thd->wsrep_trx().state());
   }
-  mysql_mutex_unlock(&m_thd->LOCK_thd_kill);
   return ret;
 }
 
diff --git a/sql/wsrep_dummy.cc b/sql/wsrep_dummy.cc
index 8ea9ca79697..5174b6e0b07 100644
--- a/sql/wsrep_dummy.cc
+++ b/sql/wsrep_dummy.cc
@@ -59,6 +59,12 @@ void wsrep_thd_LOCK(const THD *)
 void wsrep_thd_UNLOCK(const THD *)
 { }
 
+void wsrep_thd_kill_LOCK(const THD *)
+{ }
+
+void wsrep_thd_kill_UNLOCK(const THD *)
+{ }
+
 const char *wsrep_thd_conflict_state_str(THD *)
 { return 0; }
 
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 53150bfa7e7..9353aa82540 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -32,6 +32,7 @@
 #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"
@@ -947,13 +948,13 @@ void wsrep_init_startup (bool sst_first)
   if (!strcmp(wsrep_provider, WSREP_NONE)) return;
 
   /* Skip replication start if no cluster address */
-  if (!wsrep_cluster_address || wsrep_cluster_address[0] == 0) return;
+  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()) unireg_abort(1);
+  if (!wsrep_start_replication(wsrep_cluster_address)) unireg_abort(1);
 
   wsrep_create_rollbacker();
   wsrep_create_appliers(1);
@@ -1114,7 +1115,7 @@ void wsrep_shutdown_replication()
   set_current_thd(nullptr);
 }
 
-bool wsrep_start_replication()
+bool wsrep_start_replication(const char *wsrep_cluster_address)
 {
   int rcode;
   WSREP_DEBUG("wsrep_start_replication");
@@ -1129,12 +1130,7 @@ bool wsrep_start_replication()
     return true;
   }
 
-  if (!wsrep_cluster_address || wsrep_cluster_address[0]== 0)
-  {
-    // if provider is non-trivial, but no address is specified, wait for address
-    WSREP_DEBUG("wsrep_start_replication exit due to empty address");
-    return true;
-  }
+  DBUG_ASSERT(wsrep_cluster_address[0]);
 
   bool const bootstrap(TRUE == wsrep_new_cluster);
 
@@ -2041,9 +2037,15 @@ bool wsrep_can_run_in_toi(THD *thd, const char *db, const char *table,
     return true;
     break;
   case SQLCOM_ALTER_TABLE:
-    if (create_info &&
-        !wsrep_should_replicate_ddl(thd, create_info->db_type))
-      return false;
+    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))
@@ -2643,18 +2645,7 @@ static void wsrep_close_thread(THD *thd)
   thd->set_killed(KILL_CONNECTION);
   MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (thd));
   mysql_mutex_lock(&thd->LOCK_thd_kill);
-  if (thd->mysys_var)
-  {
-    thd->mysys_var->abort=1;
-    mysql_mutex_lock(&thd->mysys_var->mutex);
-    if (thd->mysys_var->current_cond)
-    {
-      mysql_mutex_lock(thd->mysys_var->current_mutex);
-      mysql_cond_broadcast(thd->mysys_var->current_cond);
-      mysql_mutex_unlock(thd->mysys_var->current_mutex);
-    }
-    mysql_mutex_unlock(&thd->mysys_var->mutex);
-  }
+  thd->abort_current_cond_wait(true);
   mysql_mutex_unlock(&thd->LOCK_thd_kill);
 }
 
diff --git a/sql/wsrep_mysqld.h b/sql/wsrep_mysqld.h
index ff157fc4981..87991ce33f5 100644
--- a/sql/wsrep_mysqld.h
+++ b/sql/wsrep_mysqld.h
@@ -214,7 +214,7 @@ extern void wsrep_close_applier_threads(int count);
 
 /* new defines */
 extern void wsrep_stop_replication(THD *thd);
-extern bool wsrep_start_replication();
+extern bool wsrep_start_replication(const char *wsrep_cluster_address);
 extern void wsrep_shutdown_replication();
 extern bool wsrep_check_mode (enum_wsrep_mode mask);
 extern bool wsrep_check_mode_after_open_table (THD *thd, legacy_db_type db_type);
@@ -295,6 +295,13 @@ void WSREP_LOG(void (*fun)(const char* fmt, ...), const char* fmt, ...);
 #define WSREP_PROVIDER_EXISTS                                                  \
   (wsrep_provider && strncasecmp(wsrep_provider, WSREP_NONE, FN_REFLEN))
 
+static inline bool wsrep_cluster_address_exists()
+{
+  if (mysqld_server_started)
+    mysql_mutex_assert_owner(&LOCK_global_system_variables);
+  return wsrep_cluster_address && wsrep_cluster_address[0];
+}
+
 #define WSREP_QUERY(thd) (thd->query())
 
 extern my_bool wsrep_ready_get();
@@ -646,6 +653,7 @@ wsrep::key wsrep_prepare_key_for_toi(const char* db, const char* table,
 #define wsrep_init_globals() do {} while(0)
 #define wsrep_create_appliers(X) do {} while(0)
 #define wsrep_should_replicate_ddl(X,Y) (1)
+#define wsrep_cluster_address_exists() (false)
 
 #endif /* WITH_WSREP */
 
diff --git a/sql/wsrep_schema.cc b/sql/wsrep_schema.cc
index 9617a510573..f92700629bb 100644
--- a/sql/wsrep_schema.cc
+++ b/sql/wsrep_schema.cc
@@ -249,6 +249,7 @@ static int open_table(THD* thd,
   tables.init_one_table(schema_name,
                         table_name,
                         NULL, lock_type);
+  thd->lex->query_tables_own_last= 0;
 
   if (!open_n_lock_single_table(thd, &tables, tables.lock_type, flags)) {
     if (thd->is_error()) {
diff --git a/sql/wsrep_server_service.cc b/sql/wsrep_server_service.cc
index b51d7d2197f..ae70f819274 100644
--- a/sql/wsrep_server_service.cc
+++ b/sql/wsrep_server_service.cc
@@ -40,6 +40,7 @@ static void init_service_thd(THD* thd, char* thread_stack)
   thd->prior_thr_create_utime= thd->start_utime= microsecond_interval_timer();
   thd->set_command(COM_SLEEP);
   thd->reset_for_next_command(true);
+  server_threads.insert(thd); // as wsrep_innobase_kill_one_trx() uses find_thread_by_id()
 }
 
 Wsrep_storage_service*
@@ -79,6 +80,7 @@ void Wsrep_server_service::release_storage_service(
     static_cast<Wsrep_storage_service*>(storage_service);
   THD* thd= ss->m_thd;
   wsrep_reset_threadvars(thd);
+  server_threads.erase(thd);
   delete ss;
   delete thd;
 }
@@ -92,7 +94,8 @@ wsrep_create_streaming_applier(THD *orig_thd, const char *ctx)
      streaming transaction is BF aborted and streaming applier
      is created from BF aborter context. */
   Wsrep_threadvars saved_threadvars(wsrep_save_threadvars());
-  wsrep_reset_threadvars(saved_threadvars.cur_thd);
+  if (saved_threadvars.cur_thd)
+    wsrep_reset_threadvars(saved_threadvars.cur_thd);
   THD *thd= 0;
   Wsrep_applier_service *ret= 0;
   if (!wsrep_create_threadvars() &&
@@ -109,7 +112,8 @@ wsrep_create_streaming_applier(THD *orig_thd, const char *ctx)
   }
   /* Restore original thread local storage state before returning. */
   wsrep_restore_threadvars(saved_threadvars);
-  wsrep_store_threadvars(saved_threadvars.cur_thd);
+  if (saved_threadvars.cur_thd)
+    wsrep_store_threadvars(saved_threadvars.cur_thd);
   return ret;
 }
 
@@ -138,6 +142,7 @@ void Wsrep_server_service::release_high_priority_service(wsrep::high_priority_se
   THD* thd= hps->m_thd;
   delete hps;
   wsrep_store_threadvars(thd);
+  server_threads.erase(thd);
   delete thd;
   wsrep_delete_threadvars();
 }
diff --git a/sql/wsrep_sst.cc b/sql/wsrep_sst.cc
index c78a9359704..0fca31bd4ea 100644
--- a/sql/wsrep_sst.cc
+++ b/sql/wsrep_sst.cc
@@ -266,6 +266,12 @@ static bool sst_auth_real_set (const char* value)
       if (wsrep_sst_auth) { my_free((void*) wsrep_sst_auth); }
       wsrep_sst_auth= my_strdup(PSI_INSTRUMENT_ME, WSREP_SST_AUTH_MASK, MYF(0));
     }
+    else
+    {
+      if (wsrep_sst_auth) { my_free((void*) wsrep_sst_auth); }
+      wsrep_sst_auth= NULL;
+    }
+
     return 0;
   }
   return 1;
diff --git a/sql/wsrep_thd.cc b/sql/wsrep_thd.cc
index 21a1c03d33e..816c8ce9fb6 100644
--- a/sql/wsrep_thd.cc
+++ b/sql/wsrep_thd.cc
@@ -22,7 +22,6 @@
 #include "rpl_rli.h"
 #include "log_event.h"
 #include "sql_parse.h"
-#include "sql_base.h" // close_thread_tables()
 #include "mysqld.h"   // start_wsrep_THD();
 #include "wsrep_applier.h"   // start_wsrep_THD();
 #include "mysql/service_wsrep.h"
@@ -126,11 +125,7 @@ bool wsrep_create_appliers(long threads, bool mutex_protected)
     return false;
   }
 
-  if (!wsrep_cluster_address || wsrep_cluster_address[0]== 0)
-  {
-    WSREP_DEBUG("wsrep_create_appliers exit due to empty address");
-    return false;
-  }
+  DBUG_ASSERT(wsrep_cluster_address[0]);
 
   long wsrep_threads=0;
 
@@ -285,16 +280,14 @@ static void wsrep_rollback_process(THD *rollbacker,
 
 void wsrep_create_rollbacker()
 {
-  if (wsrep_cluster_address && wsrep_cluster_address[0] != 0)
-  {
-    Wsrep_thd_args* args(new Wsrep_thd_args(wsrep_rollback_process,
-                                            WSREP_ROLLBACKER_THREAD,
-                                            pthread_self()));
-
-    /* create rollbacker */
-    if (create_wsrep_THD(args, false))
-      WSREP_WARN("Can't create thread to manage wsrep rollback");
-   }
+  DBUG_ASSERT(wsrep_cluster_address[0]);
+  Wsrep_thd_args* args(new Wsrep_thd_args(wsrep_rollback_process,
+                                          WSREP_ROLLBACKER_THREAD,
+                                          pthread_self()));
+
+  /* create rollbacker */
+  if (create_wsrep_THD(args, false))
+    WSREP_WARN("Can't create thread to manage wsrep rollback");
 }
 
 /*
diff --git a/sql/wsrep_var.cc b/sql/wsrep_var.cc
index 7dbe1d5aa60..3c0a7234b15 100644
--- a/sql/wsrep_var.cc
+++ b/sql/wsrep_var.cc
@@ -586,30 +586,31 @@ bool wsrep_cluster_address_update (sys_var *self, THD* thd, enum_var_type type)
   /* stop replication is heavy operation, and includes closing all client
      connections. Closing clients may need to get LOCK_global_system_variables
      at least in MariaDB.
-
-     Note: releasing LOCK_global_system_variables may cause race condition, if
-     there can be several concurrent clients changing wsrep_provider
   */
+  char *tmp= my_strdup(PSI_INSTRUMENT_ME, wsrep_cluster_address, MYF(MY_WME));
   WSREP_DEBUG("wsrep_cluster_address_update: %s", wsrep_cluster_address);
   mysql_mutex_unlock(&LOCK_global_system_variables);
+
+  mysql_mutex_lock(&LOCK_wsrep_cluster_config);
   wsrep_stop_replication(thd);
 
-  if (wsrep_start_replication())
+  if (*tmp && wsrep_start_replication(tmp))
   {
     wsrep_create_rollbacker();
     WSREP_DEBUG("Cluster address update creating %ld applier threads running %lu",
 	    wsrep_slave_threads, wsrep_running_applier_threads);
     wsrep_create_appliers(wsrep_slave_threads);
   }
-  /* locking order to be enforced is:
-     1. LOCK_global_system_variables
-     2. LOCK_wsrep_cluster_config
-     => have to juggle mutexes to comply with this
-  */
-
   mysql_mutex_unlock(&LOCK_wsrep_cluster_config);
+
   mysql_mutex_lock(&LOCK_global_system_variables);
-  mysql_mutex_lock(&LOCK_wsrep_cluster_config);
+  if (strcmp(tmp, wsrep_cluster_address))
+  {
+    my_free((void*)wsrep_cluster_address);
+    wsrep_cluster_address= tmp;
+  }
+  else
+    my_free(tmp);
 
   return false;
 }
@@ -706,7 +707,12 @@ static void wsrep_slave_count_change_update ()
 
 bool wsrep_slave_threads_update (sys_var *self, THD* thd, enum_var_type type)
 {
+  if (!wsrep_cluster_address_exists())
+    return false;
+
+  mysql_mutex_unlock(&LOCK_global_system_variables);
   mysql_mutex_lock(&LOCK_wsrep_slave_threads);
+  mysql_mutex_lock(&LOCK_global_system_variables);
   bool res= false;
 
   wsrep_slave_count_change_update();