Merge 10.4 into 10.5

46 files changed, 481 insertions, 207 deletions

diff --git a/sql/derror.cc b/sql/derror.cc
index a1150596ee3..14b60ee4db4 100644
--- a/sql/derror.cc
+++ b/sql/derror.cc
@@ -69,6 +69,9 @@ bool init_errmessage(void)
 {
   const char **errmsgs;
   bool error= FALSE;
+  const char *lang= my_default_lc_messages->errmsgs->language;
+  my_bool use_english;
+
   DBUG_ENTER("init_errmessage");
 
   free_error_messages();
@@ -77,35 +80,63 @@ bool init_errmessage(void)
 
   error_message_charset_info= system_charset_info;
 
-  /* Read messages from file. */
-  if (read_texts(ERRMSG_FILE, my_default_lc_messages->errmsgs->language,
-                 &original_error_messages))
+  use_english= !strcmp(lang, "english");
+  if (!use_english)
   {
-    /*
-      No error messages.  Create a temporary empty error message so
-      that we don't get a crash if some code wrongly tries to access
-      a non existing error message.
-    */
+    /* Read messages from file. */
+    use_english= !read_texts(ERRMSG_FILE,lang, &original_error_messages);
+    error= TRUE;
+  }
+
+  if (use_english)
+  {
+    static const struct
+    {
+      const char* name;
+      uint id;
+      const char* fmt;
+    }
+    english_msgs[]=
+    {
+      #include <mysqld_ername.h>
+    };
+
+    memset(errors_per_range, 0, sizeof(errors_per_range));
+    /* Calculate nr of messages per range. */
+    for (size_t i= 0; i < array_elements(english_msgs); i++)
+    {
+      uint id= english_msgs[i].id;
+
+      // We rely on the fact the array is sorted by id.
+      DBUG_ASSERT(i == 0 || english_msgs[i-1].id < id);
+
+      errors_per_range[id/ERRORS_PER_RANGE-1]= id%ERRORS_PER_RANGE + 1;
+    }
+
+    size_t all_errors= 0;
+    for (size_t i= 0; i < MAX_ERROR_RANGES; i++)
+      all_errors+= errors_per_range[i];
+
     if (!(original_error_messages= (const char***)
-          my_malloc(MAX_ERROR_RANGES * sizeof(char**) +
-                    (ERRORS_PER_RANGE * sizeof(char*)),
-                     MYF(0))))
+          my_malloc((all_errors + MAX_ERROR_RANGES)* sizeof(void*),
+                     MYF(MY_ZEROFILL))))
       DBUG_RETURN(TRUE);
-    errmsgs= (const char**) (original_error_messages + MAX_ERROR_RANGES);
 
-    for (uint i=0 ; i < MAX_ERROR_RANGES ; i++)
+    errmsgs= (const char**)(original_error_messages + MAX_ERROR_RANGES);
+
+    original_error_messages[0]= errmsgs;
+    for (uint i= 1; i < MAX_ERROR_RANGES; i++)
     {
-      original_error_messages[i]= errmsgs;
-      errors_per_range[i]= ERRORS_PER_RANGE;
+      original_error_messages[i]=
+        original_error_messages[i-1] + errors_per_range[i-1];
     }
-    errors_per_range[2]= 0;                     // MYSYS error messages
-
-    for (const char **ptr= errmsgs;
-         ptr < errmsgs + ERRORS_PER_RANGE ;
-         ptr++)
-      *ptr= "";
 
-    error= TRUE;
+    for (uint i= 0; i < array_elements(english_msgs); i++)
+    {
+      uint id= english_msgs[i].id;
+      original_error_messages[id/ERRORS_PER_RANGE-1][id%ERRORS_PER_RANGE]=
+         english_msgs[i].fmt;
+    }
   }
 
   /* Register messages for use with my_error(). */
diff --git a/sql/filesort.cc b/sql/filesort.cc
index bd6ac9d7374..df6e1eb9104 100644
--- a/sql/filesort.cc
+++ b/sql/filesort.cc
@@ -309,6 +309,7 @@ SORT_INFO *filesort(THD *thd, TABLE *table, Filesort *filesort,
     param.max_keys_per_buffer=((param.max_keys_per_buffer *
                                 (param.rec_length + sizeof(char*))) /
                                param.rec_length - 1);
+    set_if_bigger(param.max_keys_per_buffer, 1);
     maxbuffer--;				// Offset from 0
     if (merge_many_buff(&param,
                         (uchar*) sort->get_sort_keys(),
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 77a53668638..182621e6de4 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -564,7 +564,7 @@ bool ha_partition::initialize_partition(MEM_ROOT *mem_root)
     point.
 
     If you do not implement this, the default delete_table() is called from
-    handler.cc and it will delete all files with the file extentions returned
+    handler.cc and it will delete all files with the file extensions returned
     by bas_ext().
 
     Called from handler.cc by delete_table and  ha_create_table(). Only used
@@ -596,7 +596,7 @@ int ha_partition::delete_table(const char *name)
     Renames a table from one name to another from alter table call.
 
     If you do not implement this, the default rename_table() is called from
-    handler.cc and it will rename all files with the file extentions returned
+    handler.cc and it will rename all files with the file extensions returned
     by bas_ext().
 
     Called from sql_table.cc by mysql_rename_table().
@@ -9264,7 +9264,6 @@ void ha_partition::late_extra_cache(uint partition_id)
   }
   if (m_extra_prepare_for_update)
   {
-    DBUG_ASSERT(m_extra_cache);
     (void) file->extra(HA_EXTRA_PREPARE_FOR_UPDATE);
   }
   m_extra_cache_part_id= partition_id;
diff --git a/sql/ha_sequence.cc b/sql/ha_sequence.cc
index 87d8339881d..6cb9937ebb4 100644
--- a/sql/ha_sequence.cc
+++ b/sql/ha_sequence.cc
@@ -108,7 +108,7 @@ int ha_sequence::open(const char *name, int mode, uint flags)
       MY_TEST(flags & HA_OPEN_INTERNAL_TABLE);
     reset_statistics();
 
-    /* Don't try to read the inital row the call is part of create code */
+    /* Don't try to read the initial row the call is part of create code */
     if (!(flags & (HA_OPEN_FOR_CREATE | HA_OPEN_FOR_REPAIR)))
     {
       if (unlikely((error= table->s->sequence->read_initial_values(table))))
diff --git a/sql/handler.h b/sql/handler.h
index e22361b3449..0a561ec8b3f 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1517,9 +1517,9 @@ struct handlerton
      Used by open_table_error(), by the default rename_table and delete_table
      handler methods, and by the default discovery implementation.
   
-     For engines that have more than one file name extentions (separate
+     For engines that have more than one file name extensions (separate
      metadata, index, and/or data files), the order of elements is relevant.
-     First element of engine file name extentions array should be metadata
+     First element of engine file name extensions array should be metadata
      file extention. This is implied by the open_table_error()
      and the default discovery implementation.
      
diff --git a/sql/item.cc b/sql/item.cc
index 7d2647e9a78..1ed178f1194 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -2571,7 +2571,7 @@ bool Type_std_attributes::agg_item_set_converter(const DTCollation &coll,
       
    @retval
      clone of the item
-     0 if an error occured
+     0 if an error occurred
 */ 
 
 Item* Item_func_or_sum::build_clone(THD *thd)
@@ -2871,7 +2871,7 @@ Item_sp::init_result_field(THD *thd, uint max_length, uint maybe_null,
       
    @retval
      clone of the item
-     0 if an error occured
+     0 if an error occurred
 */ 
 
 Item* Item_ref::build_clone(THD *thd)
@@ -10117,6 +10117,8 @@ bool Item_cache_str::cache_value()
     value_buff.copy(*value);
     value= &value_buff;
   }
+  else
+    value_buff.copy();
   return TRUE;
 }
 
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index aaf6baf87f6..e9571381dc8 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -5305,7 +5305,7 @@ void Item_cond::neg_arguments(THD *thd)
       
    @retval
      clone of the item
-     0 if an error occured
+     0 if an error occurred
 */ 
 
 Item *Item_cond::build_clone(THD *thd)
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index ffe17f8a0bc..cb1a467c357 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -2476,6 +2476,7 @@ public:
   bool to_be_transformed_into_in_subq(THD *thd);
   bool create_value_list_for_tvc(THD *thd, List< List<Item> > *values);
   Item *in_predicate_to_in_subs_transformer(THD *thd, uchar *arg);
+  uint32 max_length_of_left_expr();
 };
 
 class cmp_item_row :public cmp_item
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index a9ee0f83b0a..5342a9080b6 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -2398,7 +2398,7 @@ String *Item_func_sqlerrm::val_str(String *str)
               system_charset_info);
     return str;
   }
-  str->copy(STRING_WITH_LEN("normal, successful completition"),
+  str->copy(STRING_WITH_LEN("normal, successful completion"),
             system_charset_info);
   return str;
 }
diff --git a/sql/lex.h b/sql/lex.h
index bb998d5b70f..f3fc1513369 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -23,7 +23,7 @@
 #include "lex_symbol.h"
 
 SYM_GROUP sym_group_common= {"", ""};
-SYM_GROUP sym_group_geom= {"Spatial extentions", "HAVE_SPATIAL"};
+SYM_GROUP sym_group_geom= {"Spatial extensions", "HAVE_SPATIAL"};
 SYM_GROUP sym_group_rtree= {"RTree keys", "HAVE_RTREE_KEYS"};
 
 /* We don't want to include sql_yacc.h into gen_lex_hash */
diff --git a/sql/log_event.h b/sql/log_event.h
index 88a6e06c506..a787f4e3a72 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -3626,7 +3626,7 @@ public:
   bool write_data_header();
   bool write_data_body();
   /*
-    Cut out Create_file extentions and
+    Cut out Create_file extensions and
     write it as Load event - used on the slave
   */
   bool write_base();
@@ -5541,6 +5541,9 @@ bool event_that_should_be_ignored(const char *buf);
 bool event_checksum_test(uchar *buf, ulong event_len, enum_binlog_checksum_alg alg);
 enum enum_binlog_checksum_alg get_checksum_alg(const char* buf, ulong len);
 extern TYPELIB binlog_checksum_typelib;
+#ifdef WITH_WSREP
+enum Log_event_type wsrep_peak_event(rpl_group_info *rgi, ulonglong* event_size);
+#endif /* WITH_WSREP */
 
 /**
   @} (end of group Replication)
diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index 829a278f215..e01488abbb3 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -835,7 +835,7 @@ int Write_rows_log_event_old::do_after_row_operations(TABLE *table, int error)
   table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
   table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
   /*
-    reseting the extra with 
+    resetting the extra with 
     table->file->extra(HA_EXTRA_NO_IGNORE_NO_KEY); 
     fires bug#27077
     todo: explain or fix
@@ -2459,7 +2459,7 @@ Write_rows_log_event_old::do_after_row_operations(const Slave_reporting_capabili
   m_table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
   m_table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
   /*
-    reseting the extra with 
+    resetting the extra with 
     table->file->extra(HA_EXTRA_NO_IGNORE_NO_KEY); 
     fires bug#27077
     todo: explain or fix
diff --git a/sql/mysql_upgrade_service.cc b/sql/mysql_upgrade_service.cc
index dc4b2d0923c..60fd8fc7efa 100644
--- a/sql/mysql_upgrade_service.cc
+++ b/sql/mysql_upgrade_service.cc
@@ -132,7 +132,7 @@ static void die(const char *fmt, ...)
   }
 
   /*
-    Stop service that we started, if it was not initally running at
+    Stop service that we started, if it was not initially running at
     program start.
   */
   if (initial_service_state != UINT_MAX && initial_service_state != SERVICE_RUNNING)
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 59c37da2de8..8c2b63e27fe 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -7740,8 +7740,8 @@ static void usage(void)
          "\nbecause execution stopped before plugins were initialized.");
   }
 
-  puts("\nTo see what values a running MySQL server is using, type"
-       "\n'mysqladmin variables' instead of 'mysqld --verbose --help'.");
+    puts("\nTo see what variables a running MySQL server is using, type"
+         "\n'mysqladmin variables' instead of 'mysqld --verbose --help'.");
   }
   DBUG_VOID_RETURN;
 }
diff --git a/sql/opt_subselect.cc b/sql/opt_subselect.cc
index d706afbbe10..c7d95fdadec 100644
--- a/sql/opt_subselect.cc
+++ b/sql/opt_subselect.cc
@@ -861,7 +861,12 @@ bool subquery_types_allow_materialization(THD* thd, Item_in_subselect *in_subs)
   OPT_TRACE_TRANSFORM(thd, trace_wrapper, trace_transform,
                      in_subs->get_select_lex()->select_number,
                       "IN (SELECT)", "materialization");
-  
+
+  /*
+    The checks here must be kept in sync with the one in
+    Item_func_in::in_predicate_to_in_subs_transformer().
+  */
+
   bool all_are_fields= TRUE;
   uint32 total_key_length = 0;
   for (uint i= 0; i < elements; i++)
@@ -4610,7 +4615,7 @@ SJ_TMP_TABLE::create_sj_weedout_tmp_table(THD *thd)
     field->reset();
     /*
       Test if there is a default field value. The test for ->ptr is to skip
-      'offset' fields generated by initalize_tables
+      'offset' fields generated by initialize_tables
     */
     // Initialize the table field:
     bzero(field->ptr, field->pack_length());
diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc
index 732edcd5bc6..43a02147496 100644
--- a/sql/rpl_mi.cc
+++ b/sql/rpl_mi.cc
@@ -1173,7 +1173,7 @@ bool Master_info_index::init_all_master_info()
     }
     else
     {
-      /* Initialization of Master_info succeded. Add it to HASH */
+      /* Initialization of Master_info succeeded. Add it to HASH */
       if (global_system_variables.log_warnings > 1)
         sql_print_information("Initialized Master_info from '%s'",
                               buf_master_info_file);
diff --git a/sql/service_wsrep.cc b/sql/service_wsrep.cc
index b24f3cb511a..36ea311f59f 100644
--- a/sql/service_wsrep.cc
+++ b/sql/service_wsrep.cc
@@ -196,18 +196,10 @@ extern "C" void wsrep_handle_SR_rollback(THD *bf_thd,
   }
 }
 
-extern "C" my_bool wsrep_thd_bf_abort(const THD *bf_thd, THD *victim_thd,
+extern "C" my_bool wsrep_thd_bf_abort(THD *bf_thd, THD *victim_thd,
                                       my_bool signal)
 {
-  /* Note: do not store/reset globals before wsrep_bf_abort() call
-     to avoid losing BF thd context. */
-  if (WSREP(victim_thd) && !victim_thd->wsrep_trx().active())
-  {
-    WSREP_DEBUG("BF abort for non active transaction");
-    wsrep_start_transaction(victim_thd, victim_thd->wsrep_next_trx_id());
-  }
   my_bool ret= wsrep_bf_abort(bf_thd, victim_thd);
-  wsrep_store_threadvars((THD*)bf_thd);
   /*
     Send awake signal if victim was BF aborted or does not
     have wsrep on. Note that this should never interrupt RSU
diff --git a/sql/session_tracker.cc b/sql/session_tracker.cc
index dc22d6dc9a3..33ee6141658 100644
--- a/sql/session_tracker.cc
+++ b/sql/session_tracker.cc
@@ -814,7 +814,7 @@ bool Transaction_state_tracker::store(THD *thd, String *buf)
         statement even for a transaction that isn't the first in an
         ongoing chain. Consider
 
-          SET TRANSACTION ISOLATION LEVEL READ UNCOMMITED;
+          SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
           START TRANSACTION READ ONLY, WITH CONSISTENT SNAPSHOT;
           # work
           COMMIT AND CHAIN;
@@ -822,7 +822,7 @@ bool Transaction_state_tracker::store(THD *thd, String *buf)
         If we switch away at this point, the replay in the new session
         needs to be
 
-          SET TRANSACTION ISOLATION LEVEL READ UNCOMMITED;
+          SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
           START TRANSACTION READ ONLY;
 
         When a transaction ends (COMMIT/ROLLBACK sans CHAIN), all
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 36188e58624..68b0b6bd586 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7457,7 +7457,7 @@ ER_GIS_UNSUPPORTED_ARGUMENT
   eng "Calling geometry function %s with unsupported types of arguments."
 
 ER_GIS_UNKNOWN_ERROR
-  eng "Unknown GIS error occured in function %s."
+  eng "Unknown GIS error occurred in function %s."
 
 ER_GIS_UNKNOWN_EXCEPTION
   eng "Unknown exception caught in GIS function %s."
@@ -7871,7 +7871,7 @@ ER_DROP_VERSIONING_SYSTEM_TIME_PARTITION
         eng "Can not DROP SYSTEM VERSIONING for table %`s partitioned BY SYSTEM_TIME"
 
 ER_VERS_DB_NOT_SUPPORTED
-      eng "System-versioned tables in the %`s database are not suported"
+      eng "System-versioned tables in the %`s database are not supported"
 
 ER_VERS_TRT_IS_DISABLED
         eng "Transaction registry is disabled"
diff --git a/sql/slave.cc b/sql/slave.cc
index da87ab44b67..9b1f0b96bce 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -485,6 +485,9 @@ handle_slave_background(void *arg __attribute__((unused)))
   thd->store_globals();
   thd->security_ctx->skip_grants();
   thd->set_command(COM_DAEMON);
+#ifdef WITH_WSREP
+  thd->variables.wsrep_on= 0;
+#endif
 
   thd_proc_info(thd, "Loading slave GTID position from table");
   if (rpl_load_gtid_slave_state(thd))
@@ -4761,6 +4764,9 @@ pthread_handler_t handle_slave_io(void *arg)
   }
 
 
+#ifdef WITH_WSREP
+  thd->variables.wsrep_on= 0;
+#endif
   if (DBUG_EVALUATE_IF("failed_slave_start", 1, 0)
       || repl_semisync_slave.slave_start(mi))
   {
@@ -7956,7 +7962,39 @@ err:
     sql_print_error("Error reading relay log event: %s", errmsg);
   DBUG_RETURN(0);
 }
+#ifdef WITH_WSREP
+enum Log_event_type wsrep_peak_event(rpl_group_info *rgi, ulonglong* event_size)
+{
+  enum Log_event_type ev_type;
+
+  mysql_mutex_lock(&rgi->rli->data_lock);
+
+  unsigned long long event_pos= rgi->event_relay_log_pos;
+  unsigned long long orig_future_pos= rgi->future_event_relay_log_pos;
+  unsigned long long future_pos= rgi->future_event_relay_log_pos;
+
+  /* scan the log to read next event and we skip
+     annotate events. */
+  do {
+    my_b_seek(rgi->rli->cur_log, future_pos);
+    rgi->rli->event_relay_log_pos= future_pos;
+    rgi->event_relay_log_pos= future_pos;
+    Log_event* ev= next_event(rgi, event_size);
+    ev_type= (ev) ? ev->get_type_code() : UNKNOWN_EVENT;
+    delete ev;
+    future_pos+= *event_size;
+  } while (ev_type == ANNOTATE_ROWS_EVENT || ev_type == XID_EVENT);
+
+  /* scan the log back and re-set the positions to original values */
+  rgi->rli->event_relay_log_pos= event_pos;
+  rgi->event_relay_log_pos= event_pos;
+  my_b_seek(rgi->rli->cur_log, orig_future_pos);
+
+  mysql_mutex_unlock(&rgi->rli->data_lock);
 
+  return ev_type;
+}
+#endif /* WITH_WSREP */
 /*
   Rotate a relay log (this is used only by FLUSH LOGS; the automatic rotation
   because of size is simpler because when we do it we already have all relevant
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 33590646ef7..3830fe60b4e 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -2625,7 +2625,7 @@ sp_head::backpatch_goto(THD *thd, sp_label *lab,sp_label *lab_begin_block)
       }
       if (bp->instr_type == CPOP)
       {
-        uint n= lab->ctx->diff_cursors(lab_begin_block->ctx, true);
+        uint n= bp->instr->m_ctx->diff_cursors(lab_begin_block->ctx, true);
         if (n == 0)
         {
           // Remove cpop instr
@@ -2642,7 +2642,7 @@ sp_head::backpatch_goto(THD *thd, sp_label *lab,sp_label *lab_begin_block)
       }
       if (bp->instr_type == HPOP)
       {
-        uint n= lab->ctx->diff_handlers(lab_begin_block->ctx, true);
+        uint n= bp->instr->m_ctx->diff_handlers(lab_begin_block->ctx, true);
         if (n == 0)
         {
           // Remove hpop instr
@@ -3147,6 +3147,8 @@ void sp_head::optimize()
   sp_instr *i;
   uint src, dst;
 
+  DBUG_EXECUTE_IF("sp_head_optimize_disable", return; );
+
   opt_mark();
 
   bp.empty();
@@ -3451,11 +3453,7 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
     Update the state of the active arena if no errors on
     open_tables stage.
   */
-  if (likely(!res) || likely(!thd->is_error()) ||
-      (thd->get_stmt_da()->sql_errno() != ER_CANT_REOPEN_TABLE &&
-       thd->get_stmt_da()->sql_errno() != ER_NO_SUCH_TABLE &&
-       thd->get_stmt_da()->sql_errno() != ER_NO_SUCH_TABLE_IN_ENGINE &&
-       thd->get_stmt_da()->sql_errno() != ER_UPDATE_TABLE_USED))
+  if (likely(!res) || likely(!thd->is_error()))
     thd->stmt_arena->state= Query_arena::STMT_EXECUTED;
 
   /*
diff --git a/sql/sql_admin.cc b/sql/sql_admin.cc
index 3d1f4c3b505..e5347f2fda4 100644
--- a/sql/sql_admin.cc
+++ b/sql/sql_admin.cc
@@ -175,7 +175,7 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
   /*
     Check if this is a table type that stores index and data separately,
     like ISAM or MyISAM. We assume fixed order of engine file name
-    extentions array. First element of engine file name extentions array
+    extensions array. First element of engine file name extensions array
     is meta/index file extention. Second element - data file extention. 
   */
   ext= table->file->bas_ext();
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index f070ff1222f..38dba2c11fd 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -51,7 +51,7 @@
 #define MAX_DROP_TABLE_Q_LEN      1024
 
 const char *del_exts[]= {".BAK", ".opt", NullS};
-static TYPELIB deletable_extentions=
+static TYPELIB deletable_extensions=
 {array_elements(del_exts)-1,"del_exts", del_exts, NULL};
 
 static bool find_db_tables_and_rm_known_files(THD *, MY_DIR *, const char *,
@@ -1189,7 +1189,7 @@ static bool find_db_tables_and_rm_known_files(THD *thd, MY_DIR *dirp,
     }
     if (!(extension= strrchr(file->name, '.')))
       extension= strend(file->name);
-    if (find_type(extension, &deletable_extentions, FIND_TYPE_NO_PREFIX) > 0)
+    if (find_type(extension, &deletable_extensions, FIND_TYPE_NO_PREFIX) > 0)
     {
       strxmov(filePath, path, "/", file->name, NullS);
       /*
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 29d6904f7f9..6a4ce266af2 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -1006,7 +1006,7 @@ int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list, Item **conds,
 
   DBUG_ASSERT(table_list->table);
   // conds could be cached from previous SP call
-  DBUG_ASSERT(!table_list->vers_conditions.is_set() ||
+  DBUG_ASSERT(!table_list->vers_conditions.need_setup() ||
               !*conds || thd->stmt_arena->is_stmt_execute());
   if (select_lex->vers_setup_conds(thd, table_list))
     DBUG_RETURN(TRUE);
diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc
index fc6bf44bbb1..c713f1322dc 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -741,8 +741,15 @@ bool mysql_derived_prepare(THD *thd, LEX *lex, TABLE_LIST *derived)
       case SQLCOM_DELETE_MULTI:
       case SQLCOM_UPDATE:
       case SQLCOM_UPDATE_MULTI:
-        if ((res= unit->prepare(derived, derived->derived_result, 0)))
+        if ((res= first_select->vers_setup_conds(thd,
+                                                 derived->merge_underlying_list)))
           goto exit;
+        if (derived->merge_underlying_list->where)
+        {
+          Query_arena_stmt on_stmt_arena(thd);
+          derived->where= and_items(thd, derived->where,
+                                    derived->merge_underlying_list->where);
+        }
       default:
         break;
       }
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 63ce6ef30cf..0f267fff17b 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2147,11 +2147,15 @@ public:
   ulong auto_increment_offset;
   LEX_STRING query;
   Time_zone *time_zone;
+  char *user, *host, *ip;
+  query_id_t query_id;
+  my_thread_id thread_id;
 
   delayed_row(LEX_STRING const query_arg, enum_duplicates dup_arg,
               bool ignore_arg, bool log_query_arg)
     : record(0), dup(dup_arg), ignore(ignore_arg), log_query(log_query_arg),
-      forced_insert_id(0), query(query_arg), time_zone(0)
+      forced_insert_id(0), query(query_arg), time_zone(0),
+      user(0), host(0), ip(0)
     {}
   ~delayed_row()
   {
@@ -2199,6 +2203,27 @@ public:
     passed from connection thread to the handler thread.
   */
   MDL_request grl_protection;
+  my_thread_id orig_thread_id;
+  void set_default_user()
+  {
+    thd.security_ctx->user=(char*) delayed_user;
+    thd.security_ctx->host=(char*) my_localhost;
+    thd.security_ctx->ip= NULL;
+    thd.query_id= 0;
+    thd.thread_id= orig_thread_id;
+  }
+
+  void set_user_from_row(const delayed_row *r)
+  {
+    if (r)
+    {
+      thd.security_ctx->user= r->user;
+      thd.security_ctx->host= r->host;
+      thd.security_ctx->ip= r->ip;
+      thd.query_id= r->query_id;
+      thd.thread_id= r->thread_id;
+    }
+  }
 
   Delayed_insert(SELECT_LEX *current_select)
     :locks_in_memory(0), thd(next_thread_id()),
@@ -2206,8 +2231,8 @@ public:
      status(0), retry(0), handler_thread_initialized(FALSE), group_count(0)
   {
     DBUG_ENTER("Delayed_insert constructor");
-    thd.security_ctx->user=(char*) delayed_user;
-    thd.security_ctx->host=(char*) my_localhost;
+    orig_thread_id= thd.thread_id;
+    set_default_user();
     strmake_buf(thd.security_ctx->priv_user, thd.security_ctx->user);
     thd.current_tablenr=0;
     thd.set_command(COM_DELAYED_INSERT);
@@ -2703,6 +2728,7 @@ int write_delayed(THD *thd, TABLE *table, enum_duplicates duplic,
   delayed_row *row= 0;
   Delayed_insert *di=thd->di;
   const Discrete_interval *forced_auto_inc;
+  size_t user_len, host_len, ip_len;
   DBUG_ENTER("write_delayed");
   DBUG_PRINT("enter", ("query = '%s' length %lu", query.str,
                        (ulong) query.length));
@@ -2736,11 +2762,45 @@ int write_delayed(THD *thd, TABLE *table, enum_duplicates duplic,
     goto err;
   }
 
+  user_len= host_len= ip_len= 0;
+  row->user= row->host= row->ip= NULL;
+  if (thd->security_ctx)
+  {
+    if (thd->security_ctx->user)
+      user_len= strlen(thd->security_ctx->user) + 1;
+    if (thd->security_ctx->host)
+      host_len= strlen(thd->security_ctx->host) + 1;
+    if (thd->security_ctx->ip)
+      ip_len= strlen(thd->security_ctx->ip) + 1;
+  }
   /* This can't be THREAD_SPECIFIC as it's freed in delayed thread */
-  if (!(row->record= (char*) my_malloc(table->s->reclength,
+  if (!(row->record= (char*) my_malloc(table->s->reclength +
+                                       user_len + host_len + ip_len,
                                        MYF(MY_WME))))
     goto err;
   memcpy(row->record, table->record[0], table->s->reclength);
+
+  if (thd->security_ctx)
+  {
+    if (thd->security_ctx->user)
+    {
+      row->user= row->record + table->s->reclength;
+      memcpy(row->user, thd->security_ctx->user, user_len);
+    }
+    if (thd->security_ctx->host)
+    {
+      row->host= row->record + table->s->reclength + user_len;
+      memcpy(row->host, thd->security_ctx->host, host_len);
+    }
+    if (thd->security_ctx->ip)
+    {
+      row->ip= row->record + table->s->reclength + user_len + host_len;
+      memcpy(row->ip, thd->security_ctx->ip, ip_len);
+    }
+  }
+  row->query_id= thd->query_id;
+  row->thread_id= thd->thread_id;
+
   row->start_time=                thd->start_time;
   row->start_time_sec_part=       thd->start_time_sec_part;
   row->query_start_sec_part_used= thd->query_start_sec_part_used;
@@ -3158,6 +3218,7 @@ pthread_handler_t handle_delayed_insert(void *arg)
       if (di->tables_in_use && ! thd->lock &&
           (!thd->killed || di->stacked_inserts))
       {
+        di->set_user_from_row(di->rows.head());
         /*
           Request for new delayed insert.
           Lock the table, but avoid to be blocked by a global read lock.
@@ -3177,6 +3238,18 @@ pthread_handler_t handle_delayed_insert(void *arg)
       }
       if (di->stacked_inserts)
       {
+        delayed_row *row;
+        I_List_iterator<delayed_row> it(di->rows);
+        while ((row= it++))
+        {
+          if (di->thd.thread_id != row->thread_id)
+          {
+            di->set_user_from_row(row);
+            mysql_audit_external_lock(&di->thd, di->table->s, F_WRLCK);
+          }
+        }
+        di->set_default_user();
+
         if (di->handle_inserts())
         {
           /* Some fatal error */
diff --git a/sql/sql_join_cache.cc b/sql/sql_join_cache.cc
index 3b051259ad6..3a509b3d750 100644
--- a/sql/sql_join_cache.cc
+++ b/sql/sql_join_cache.cc
@@ -2141,7 +2141,7 @@ enum_nested_loop_state JOIN_CACHE::join_records(bool skip_last)
     DBUG_ASSERT(!is_key_access());
     /*
        Restore the last record from the join buffer to generate
-       all extentions for it.
+       all extensions for it.
     */
     get_record();		               
   }
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 39322b18a4f..73ffeb7e9e3 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -3344,7 +3344,7 @@ int reset_slave(THD *thd, Master_info* mi)
   char fname[FN_REFLEN];
   int thread_mask= 0, error= 0;
   uint sql_errno=ER_UNKNOWN_ERROR;
-  const char* errmsg= "Unknown error occurred while reseting slave";
+  const char* errmsg= "Unknown error occurred while resetting slave";
   char master_info_file_tmp[FN_REFLEN];
   char relay_log_info_file_tmp[FN_REFLEN];
   DBUG_ENTER("reset_slave");
@@ -3951,7 +3951,7 @@ int reset_master(THD* thd, rpl_gtid *init_state, uint32 init_state_len,
   }
 #endif /* WITH_WSREP */
   bool ret= 0;
-  /* Temporarily disable master semisync before reseting master. */
+  /* Temporarily disable master semisync before resetting master. */
   repl_semisync_master.before_reset_master();
   ret= mysql_bin_log.reset_logs(thd, 1, init_state, init_state_len,
                                 next_log_number);
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 220abd31187..3619c603697 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1986,7 +1986,20 @@ JOIN::optimize_inner()
     }
   }
 
-  conds= optimize_cond(this, conds, join_list, FALSE,
+  bool ignore_on_expr= false;
+  /*
+    PS/SP note: on_expr of versioned table can not be reallocated
+    (see build_equal_items() below) because it can be not rebuilt
+    at second invocation.
+  */
+  if (!thd->stmt_arena->is_conventional() && thd->mem_root != thd->stmt_arena->mem_root)
+    for (TABLE_LIST *tbl= tables_list; tbl; tbl= tbl->next_local)
+      if (tbl->table && tbl->on_expr && tbl->table->versioned())
+      {
+        ignore_on_expr= true;
+        break;
+      }
+  conds= optimize_cond(this, conds, join_list, ignore_on_expr,
                        &cond_value, &cond_equal, OPT_LINK_EQUAL_FIELDS);
 
   if (thd->is_error())
@@ -18520,7 +18533,7 @@ bool Create_tmp_table::finalize(THD *thd,
 
     /*
       Test if there is a default field value. The test for ->ptr is to skip
-      'offset' fields generated by initalize_tables
+      'offset' fields generated by initialize_tables
     */
     if (m_default_field[i] && m_default_field[i]->ptr)
     {
@@ -21373,7 +21386,7 @@ join_read_last(JOIN_TAB *tab)
 {
   TABLE *table=tab->table;
   int error= 0;
-  DBUG_ENTER("join_read_first");
+  DBUG_ENTER("join_read_last");
 
   DBUG_ASSERT(table->no_keyread ||
               !table->covering_keys.is_set(tab->index) ||
diff --git a/sql/sql_sequence.cc b/sql/sql_sequence.cc
index 544c9b7f436..8b2f49a5910 100644
--- a/sql/sql_sequence.cc
+++ b/sql/sql_sequence.cc
@@ -487,9 +487,12 @@ int SEQUENCE::read_initial_values(TABLE *table)
       Doing mysql_lock_tables() may have started a read only transaction.
       If that happend, it's better that we commit it now, as a lot of
       code assumes that there is no active stmt transaction directly after
-      open_tables()
+      open_tables().
+      But we also don't want to commit the stmt transaction while in a
+      substatement, see MDEV-15977.
     */
-    if (!has_active_transaction && !thd->transaction.stmt.is_empty())
+    if (!has_active_transaction && !thd->transaction.stmt.is_empty() &&
+        !thd->in_sub_stmt)
       trans_commit_stmt(thd);
   }
   write_unlock(table);
diff --git a/sql/sql_servers.cc b/sql/sql_servers.cc
index 8f0f08da075..7913a7d2b9f 100644
--- a/sql/sql_servers.cc
+++ b/sql/sql_servers.cc
@@ -431,7 +431,7 @@ insert_server_record_into_cache(FOREIGN_SERVER *server)
   int error=0;
   DBUG_ENTER("insert_server_record_into_cache");
   /*
-    We succeded in insertion of the server to the table, now insert
+    We succeeded in insertion of the server to the table, now insert
     the server to the cache
   */
   DBUG_PRINT("info", ("inserting server %s at %p, length %zd",
@@ -686,7 +686,7 @@ delete_server_record_in_cache(LEX_SERVER_OPTIONS *server_options)
     goto end;
   }
   /*
-    We succeded in deletion of the server to the table, now delete
+    We succeeded in deletion of the server to the table, now delete
     the server from the cache
   */
   DBUG_PRINT("info",("deleting server %s length %zd",
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 43914a0003d..9ba31a0ff83 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1190,7 +1190,8 @@ mysqld_show_create_get_fields(THD *thd, TABLE_LIST *table_list,
     goto exit;
   }
   else if (lex->table_type == TABLE_TYPE_SEQUENCE &&
-           table_list->table->s->table_type != TABLE_TYPE_SEQUENCE)
+           (!table_list->table ||
+            table_list->table->s->table_type != TABLE_TYPE_SEQUENCE))
   {
     my_error(ER_NOT_SEQUENCE, MYF(0),
              table_list->db.str, table_list->table_name.str);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 440cc0dfba7..bc955fb17ba 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -4315,10 +4315,10 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         continue;
 
       {
-        /* Check that there's no repeating constraint names. */
+        /* Check that there's no repeating table CHECK constraint names. */
         List_iterator_fast<Virtual_column_info>
           dup_it(alter_info->check_constraint_list);
-        Virtual_column_info *dup_check;
+        const Virtual_column_info *dup_check;
         while ((dup_check= dup_it++) && dup_check != check)
         {
           if (!lex_string_cmp(system_charset_info,
@@ -4330,6 +4330,27 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         }
       }
 
+      /* Check that there's no repeating key constraint names. */
+      List_iterator_fast<Key> key_it(alter_info->key_list);
+      while (const Key *key= key_it++)
+      {
+        /*
+          Not all keys considered to be the CONSTRAINT
+          Noly Primary Key UNIQUE and Foreign keys.
+        */
+        if (key->type != Key::PRIMARY && key->type != Key::UNIQUE &&
+            key->type != Key::FOREIGN_KEY)
+          continue;
+
+        if (check->name.length == key->name.length &&
+            my_strcasecmp(system_charset_info,
+              check->name.str, key->name.str) == 0)
+        {
+          my_error(ER_DUP_CONSTRAINT_NAME, MYF(0), "CHECK", check->name.str);
+          DBUG_RETURN(TRUE);
+        }
+      }
+
       if (check_string_char_length(&check->name, 0, NAME_CHAR_LEN,
                                    system_charset_info, 1))
       {
@@ -8666,6 +8687,35 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
       }
     }
   }
+
+  if (!alter_info->check_constraint_list.is_empty())
+  {
+    /* Check the table FOREIGN KEYs for name duplications. */
+    List <FOREIGN_KEY_INFO> fk_child_key_list;
+    FOREIGN_KEY_INFO *f_key;
+    table->file->get_foreign_key_list(thd, &fk_child_key_list);
+    List_iterator<FOREIGN_KEY_INFO> fk_key_it(fk_child_key_list);
+    while ((f_key= fk_key_it++))
+    {
+      List_iterator_fast<Virtual_column_info>
+        c_it(alter_info->check_constraint_list);
+      Virtual_column_info *check;
+      while ((check= c_it++))
+      {
+        if (!check->name.length || check->automatic_name)
+          continue;
+
+        if (check->name.length == f_key->foreign_id->length &&
+            my_strcasecmp(system_charset_info, f_key->foreign_id->str,
+                          check->name.str) == 0)
+        {
+          my_error(ER_DUP_CONSTRAINT_NAME, MYF(0), "CHECK", check->name.str);
+          goto err;
+        }
+      }
+    }
+  }
+
   /* Add new constraints */
   new_constraint_list.append(&alter_info->check_constraint_list);
 
diff --git a/sql/sql_tvc.cc b/sql/sql_tvc.cc
index d53ff1f6f30..678dd81709e 100644
--- a/sql/sql_tvc.cc
+++ b/sql/sql_tvc.cc
@@ -797,6 +797,38 @@ bool Item_subselect::wrap_tvc_into_select(THD *thd, st_select_lex *tvc_sl)
 }
 
 
+/*
+  @brief
+   Check whether the items are of comparable type or not
+
+  @details
+    This check are done because materialization is not performed
+    if the left expr and right expr are of the same types.
+    @see subquery_types_allow_materialization()
+
+  @retval
+   0 comparable
+   1 not comparable
+*/
+
+static bool cmp_row_types(Item* item1, Item* item2)
+{
+  uint n= item1->cols();
+  if (item2->check_cols(n))
+    return true;
+
+  for (uint i=0; i < n; i++)
+  {
+    Item *inner= item1->element_index(i);
+    Item *outer= item2->element_index(i);
+    if (!inner->type_handler()->subquery_type_allows_materialization(inner,
+                                                                     outer))
+      return true;
+  }
+  return false;
+}
+
+
 /**
   @brief
     Transform IN predicate into IN subquery
@@ -841,10 +873,22 @@ Item *Item_func_in::in_predicate_to_in_subs_transformer(THD *thd,
   /* SELECT_LEX object where the transformation is performed */
   SELECT_LEX *parent_select= lex->current_select;
   uint8 save_derived_tables= lex->derived_tables;
+
+  /*
+    Make sure that create_tmp_table will not fail due to too long keys.
+    Here the strategy would mainly use materialization, so we need to make
+    sure that the materialized table can be created.
+
+    The checks here are the same as in subquery_type_allows_materialization()
+  */
+  uint32 length= max_length_of_left_expr();
+  if (!length  || length > tmp_table_max_key_length() ||
+      args[0]->cols() > tmp_table_max_key_parts())
+    return this;
   
   for (uint i=1; i < arg_count; i++)
   {
-    if (!args[i]->const_item())
+    if (!args[i]->const_item() || cmp_row_types(args[0], args[i]))
       return this;
   }
 
@@ -949,6 +993,16 @@ err:
 }
 
 
+uint32 Item_func_in::max_length_of_left_expr()
+{
+  uint n= args[0]->cols();
+  uint32 length= 0;
+  for (uint i=0; i < n; i++)
+    length+= args[0]->element_index(i)->max_length;
+  return length;
+}
+
+
 /**
   @brief
     Check if this IN-predicate can be transformed in IN-subquery
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 76cbb257fea..55dece1ec5c 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1386,7 +1386,7 @@ bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
 
   DBUG_ASSERT(table_list->table);
   // conds could be cached from previous SP call
-  DBUG_ASSERT(!table_list->vers_conditions.is_set() ||
+  DBUG_ASSERT(!table_list->vers_conditions.need_setup() ||
               !*conds || thd->stmt_arena->is_stmt_execute());
   if (select_lex->vers_setup_conds(thd, table_list))
     DBUG_RETURN(TRUE);
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index ff92b042ccc..ab53f339b2b 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -6341,7 +6341,7 @@ static Sys_var_enum Sys_session_track_transaction_info(
        "Track changes to the transaction attributes. OFF to disable; "
        "STATE to track just transaction state (Is there an active transaction? "
        "Does it have any data? etc.); CHARACTERISTICS to track transaction "
-       "state and report all statements needed to start a transaction with"
+       "state and report all statements needed to start a transaction with "
        "the same characteristics (isolation level, read only/read write,"
        "snapshot - but not any work done / data modified within the "
        "transaction).",
diff --git a/sql/table.h b/sql/table.h
index 5391177f625..acaa8bcdafe 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1923,6 +1923,10 @@ struct vers_select_conds_t
   {
     return orig_type != SYSTEM_TIME_UNSPECIFIED;
   }
+  bool need_setup() const
+  {
+    return type != SYSTEM_TIME_UNSPECIFIED && type != SYSTEM_TIME_ALL;
+  }
   bool eq(const vers_select_conds_t &conds) const;
 };
 
@@ -3138,7 +3142,7 @@ public:
 
      @param[in] timestamp
      @param[in] true if we search for a lesser timestamp, false if greater
-     @retval true if exists, false it not exists or an error occured
+     @retval true if exists, false it not exists or an error occurred
    */
   bool query(MYSQL_TIME &commit_time, bool backwards);
   /**
diff --git a/sql/temporary_tables.cc b/sql/temporary_tables.cc
index b39423e9131..2e0f0a4918e 100644
--- a/sql/temporary_tables.cc
+++ b/sql/temporary_tables.cc
@@ -346,6 +346,13 @@ bool THD::open_temporary_table(TABLE_LIST *tl)
     DBUG_RETURN(false);
   }
 
+  if (!tl->db.str)
+  {
+    DBUG_PRINT("info",
+               ("Table reference to a temporary table must have database set"));
+    DBUG_RETURN(false);
+  }
+
   /*
     Temporary tables are not safe for parallel replication. They were
     designed to be visible to one thread only, so have no table locking.
diff --git a/sql/threadpool_common.cc b/sql/threadpool_common.cc
index 6ed80c8de76..0588562ae61 100644
--- a/sql/threadpool_common.cc
+++ b/sql/threadpool_common.cc
@@ -195,7 +195,7 @@ void tp_callback(TP_connection *c)
   }
   else if (threadpool_process_request(thd))
   {
-    /* QUIT or an error occured. */
+    /* QUIT or an error occurred. */
     goto error;
   }
 
diff --git a/sql/wsrep_dummy.cc b/sql/wsrep_dummy.cc
index b5adef17442..b87e807c7c9 100644
--- a/sql/wsrep_dummy.cc
+++ b/sql/wsrep_dummy.cc
@@ -115,7 +115,7 @@ const char* wsrep_thd_transaction_state_str(const THD*)
 query_id_t wsrep_thd_transaction_id(const THD *)
 { return 0; }
 
-my_bool wsrep_thd_bf_abort(const THD *, THD *, my_bool)
+my_bool wsrep_thd_bf_abort(THD *, THD *, my_bool)
 { return 0; }
 
 my_bool wsrep_thd_order_before(const THD*, const THD *)
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 19d6f88478b..1ec80acf510 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -46,7 +46,6 @@
 #include <cstdlib>
 #include <string>
 #include "log_event.h"
-#include <slave.h>
 
 #include <sstream>
 
@@ -1646,6 +1645,39 @@ static bool wsrep_can_run_in_toi(THD *thd, const char *db, const char *table,
     {
       return false;
     }
+    /*
+      If mariadb master has replicated a CTAS, we should not replicate the create table
+      part separately as TOI, but to replicate both create table and following inserts
+      as one write set.
+      Howver, if CTAS creates empty table, we should replicate the create table alone
+      as TOI. We have to do relay log event lookup to see if row events follow the
+      create table event.
+    */
+    if (thd->slave_thread && !(thd->rgi_slave->gtid_ev_flags2 & Gtid_log_event::FL_STANDALONE))
+    {
+      /* this is CTAS, either empty or populated table */
+      ulonglong event_size = 0;
+      enum Log_event_type ev_type= wsrep_peak_event(thd->rgi_slave, &event_size);
+      switch (ev_type)
+      {
+      case QUERY_EVENT:
+        /* CTAS with empty table, we replicate create table as TOI */
+        break;
+
+      case TABLE_MAP_EVENT:
+        WSREP_DEBUG("replicating CTAS of empty table as TOI");
+        // fall through
+      case WRITE_ROWS_EVENT:
+        /* CTAS with populated table, we replicate later at commit time */
+        WSREP_DEBUG("skipping create table of CTAS replication");
+        return false;
+
+      default:
+        WSREP_WARN("unexpected async replication event: %d", ev_type);
+      }
+      return true;
+    }
+    /* no next async replication event */
     return true;
 
   case SQLCOM_CREATE_VIEW:
@@ -2109,12 +2141,18 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
     if (wsrep_thd_is_toi(granted_thd) ||
         wsrep_thd_is_applying(granted_thd))
     {
-      if (wsrep_thd_is_SR(granted_thd) && !wsrep_thd_is_SR(request_thd))
+      if (wsrep_thd_is_aborting(granted_thd))
+      {
+        WSREP_DEBUG("BF thread waiting for SR in aborting state");
+        ticket->wsrep_report(wsrep_debug);
+        mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+      }
+      else if (wsrep_thd_is_SR(granted_thd) && !wsrep_thd_is_SR(request_thd))
       {
         WSREP_MDL_LOG(INFO, "MDL conflict, DDL vs SR", 
                       schema, schema_len, request_thd, granted_thd);
         mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
-        wsrep_abort_thd((void*)request_thd, (void*)granted_thd, 1);
+        wsrep_abort_thd(request_thd, granted_thd, 1);
       }
       else
       {
@@ -2138,7 +2176,7 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
                   wsrep_thd_transaction_state_str(granted_thd));
       ticket->wsrep_report(wsrep_debug);
       mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
-      wsrep_abort_thd((void*)request_thd, (void*)granted_thd, 1);
+      wsrep_abort_thd(request_thd, granted_thd, 1);
     }
     else
     {
diff --git a/sql/wsrep_schema.cc b/sql/wsrep_schema.cc
index c7ea378d4fb..95c788ba806 100644
--- a/sql/wsrep_schema.cc
+++ b/sql/wsrep_schema.cc
@@ -1049,37 +1049,23 @@ int Wsrep_schema::remove_fragments(THD* thd,
   Wsrep_schema_impl::wsrep_off  wsrep_off(thd);
   Wsrep_schema_impl::binlog_off binlog_off(thd);
 
-  /*
-    Open SR table for write.
-    Adopted from Rpl_info_table_access::open_table()
-  */
-  uint flags= (MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK |
-               MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY |
-               MYSQL_OPEN_IGNORE_FLUSH |
-               MYSQL_LOCK_IGNORE_TIMEOUT);
   Query_tables_list query_tables_list_backup;
   Open_tables_backup open_tables_backup;
   thd->lex->reset_n_backup_query_tables_list(&query_tables_list_backup);
   thd->reset_n_backup_open_tables_state(&open_tables_backup);
-  TABLE_LIST tables;
-  LEX_CSTRING schema_str= { wsrep_schema_str.c_str(), wsrep_schema_str.length() };
-  LEX_CSTRING table_str= { sr_table_str.c_str(), sr_table_str.length() };
-  tables.init_one_table(&schema_str,
-                        &table_str, 0, TL_WRITE);
 
-  if (!open_n_lock_single_table(thd, &tables, tables.lock_type, flags))
+  TABLE* frag_table= 0;
+  if (Wsrep_schema_impl::open_for_write(thd, sr_table_str.c_str(), &frag_table))
   {
-    WSREP_DEBUG("Failed to open SR table for access");
     ret= 1;
   }
   else
   {
-    tables.table->use_all_columns();
     for (std::vector<wsrep::seqno>::const_iterator i= fragments.begin();
          i != fragments.end(); ++i)
     {
       if (remove_fragment(thd,
-                          tables.table,
+                          frag_table,
                           server_id,
                           transaction_id, *i))
       {
diff --git a/sql/wsrep_server_service.cc b/sql/wsrep_server_service.cc
index bfb85e3d0ab..aa288e67420 100644
--- a/sql/wsrep_server_service.cc
+++ b/sql/wsrep_server_service.cc
@@ -42,15 +42,13 @@ static void init_service_thd(THD* thd, char* thread_stack)
   thd->reset_for_next_command(true);
 }
 
-wsrep::storage_service* Wsrep_server_service::storage_service(
-  wsrep::client_service& client_service)
+Wsrep_storage_service*
+wsrep_create_storage_service(THD* orig_THD, const char* ctx)
 {
-  Wsrep_client_service& cs=
-    static_cast<Wsrep_client_service&>(client_service);
-  THD* thd= new THD(next_thread_id(), true);
-  init_service_thd(thd, cs.m_thd->thread_stack);
-  WSREP_DEBUG("Created storage service with thread id %llu",
-              thd->thread_id);
+  THD* thd= new THD(true, true);
+  init_service_thd(thd, orig_THD->thread_stack);
+  WSREP_DEBUG("Created storage service in %s context with thread id %llu",
+              ctx, thd->thread_id);
   /* Use variables from the current thd attached to client_service.
      This is because we need to be able to BF abort storage access
      operations. */
@@ -59,16 +57,19 @@ wsrep::storage_service* Wsrep_server_service::storage_service(
 }
 
 wsrep::storage_service* Wsrep_server_service::storage_service(
+  wsrep::client_service& client_service)
+{
+  Wsrep_client_service& cs=
+    static_cast<Wsrep_client_service&>(client_service);
+  return wsrep_create_storage_service(cs.m_thd, "local");
+}
+
+wsrep::storage_service* Wsrep_server_service::storage_service(
   wsrep::high_priority_service& high_priority_service)
 {
   Wsrep_high_priority_service& hps=
     static_cast<Wsrep_high_priority_service&>(high_priority_service);
-  THD* thd= new THD(next_thread_id(), true);
-  init_service_thd(thd, hps.m_thd->thread_stack);
-  WSREP_DEBUG("Created high priority storage service with thread id %llu",
-              thd->thread_id);
-  wsrep_assign_from_threadvars(thd);
-  return new Wsrep_storage_service(thd);
+  return wsrep_create_storage_service(hps.m_thd, "high priority");
 }
 
 void Wsrep_server_service::release_storage_service(
diff --git a/sql/wsrep_server_service.h b/sql/wsrep_server_service.h
index 6336fe2c473..4017c9b2d58 100644
--- a/sql/wsrep_server_service.h
+++ b/sql/wsrep_server_service.h
@@ -87,4 +87,14 @@ class Wsrep_applier_service;
 Wsrep_applier_service*
 wsrep_create_streaming_applier(THD *orig_thd, const char *ctx);
 
+/**
+   Helper method to create new storage service.
+
+   @param orig_thd Original thd context to copy operation context from.
+   @param ctx Context string for debug logging.
+*/
+class Wsrep_storage_service;
+Wsrep_storage_service*
+wsrep_create_storage_service(THD *orig_thd, const char *ctx);
+
 #endif /* WSREP_SERVER_SERVICE */
diff --git a/sql/wsrep_thd.cc b/sql/wsrep_thd.cc
index 9d70875c027..7f1818def73 100644
--- a/sql/wsrep_thd.cc
+++ b/sql/wsrep_thd.cc
@@ -136,99 +136,60 @@ void wsrep_create_appliers(long threads)
   }
 }
 
-static void wsrep_rollback_streaming_aborted_by_toi(THD *thd)
+static void wsrep_remove_streaming_fragments(THD* thd, const char* ctx)
 {
-  WSREP_INFO("wsrep_rollback_streaming_aborted_by_toi");
-  /* Set thd->event_scheduler.data temporarily to NULL to avoid
-     callbacks to threadpool wait_begin() during rollback. */
-  auto saved_esd= thd->event_scheduler.data;
-  thd->event_scheduler.data= 0;
-  if (thd->wsrep_cs().mode() == wsrep::client_state::m_high_priority)
-  {
-    DBUG_ASSERT(!saved_esd);
-    DBUG_ASSERT(thd->wsrep_applier_service);
-    thd->wsrep_applier_service->rollback(wsrep::ws_handle(),
-                                         wsrep::ws_meta());
-    thd->wsrep_applier_service->after_apply();
-    /* Will free THD */
-    Wsrep_server_state::instance().server_service().
-      release_high_priority_service(thd->wsrep_applier_service);
-  }
-  else
-  {
-    mysql_mutex_lock(&thd->LOCK_thd_data);
-    /* prepare THD for rollback processing */
-    thd->reset_for_next_command(true);
-    thd->lex->sql_command= SQLCOM_ROLLBACK;
-    mysql_mutex_unlock(&thd->LOCK_thd_data);
-    /* Perform a client rollback, restore globals and signal
-       the victim only when all the resources have been
-       released */
-    thd->wsrep_cs().client_service().bf_rollback();
-    wsrep_reset_threadvars(thd);
-    /* Assign saved event_scheduler.data back before letting
-       client to continue. */
-    thd->event_scheduler.data= saved_esd;
-    thd->wsrep_cs().sync_rollback_complete();
-  }
+  wsrep::transaction_id transaction_id(thd->wsrep_trx().id());
+  Wsrep_storage_service* storage_service= wsrep_create_storage_service(thd, ctx);
+  storage_service->store_globals();
+  storage_service->adopt_transaction(thd->wsrep_trx());
+  storage_service->remove_fragments();
+  storage_service->commit(wsrep::ws_handle(transaction_id, 0),
+                          wsrep::ws_meta());
+  Wsrep_server_state::instance().server_service()
+    .release_storage_service(storage_service);
+  wsrep_store_threadvars(thd);
 }
 
-static void wsrep_rollback_high_priority(THD *thd)
+static void wsrep_rollback_high_priority(THD *thd, THD *rollbacker)
 {
-  WSREP_INFO("rollbacker aborting SR thd: (%lld %llu)",
-             thd->thread_id, (long long)thd->real_id);
+  WSREP_DEBUG("Rollbacker aborting SR applier thd (%llu %lu)",
+              thd->thread_id, thd->real_id);
+  char* orig_thread_stack= thd->thread_stack;
+  thd->thread_stack= rollbacker->thread_stack;
   DBUG_ASSERT(thd->wsrep_cs().mode() == Wsrep_client_state::m_high_priority);
   /* Must be streaming and must have been removed from the
      server state streaming appliers map. */
   DBUG_ASSERT(thd->wsrep_trx().is_streaming());
   DBUG_ASSERT(!Wsrep_server_state::instance().find_streaming_applier(
-    thd->wsrep_trx().server_id(),
-    thd->wsrep_trx().id()));
+                thd->wsrep_trx().server_id(),
+                thd->wsrep_trx().id()));
   DBUG_ASSERT(thd->wsrep_applier_service);
 
   /* Fragment removal should happen before rollback to make
      the transaction non-observable in SR table after the rollback
      completes. For correctness the order does not matter here,
      but currently it is mandated by checks in some MTR tests. */
-  wsrep::transaction_id transaction_id(thd->wsrep_trx().id());
-  Wsrep_storage_service* storage_service=
-    static_cast<Wsrep_storage_service*>(
-          Wsrep_server_state::instance().server_service().storage_service(
-            *thd->wsrep_applier_service));
-  storage_service->store_globals();
-  storage_service->adopt_transaction(thd->wsrep_trx());
-  storage_service->remove_fragments();
-  storage_service->commit(wsrep::ws_handle(transaction_id, 0),
-                          wsrep::ws_meta());
-  Wsrep_server_state::instance().server_service().release_storage_service(storage_service);
-  wsrep_store_threadvars(thd);
+  wsrep_remove_streaming_fragments(thd, "high priority");
   thd->wsrep_applier_service->rollback(wsrep::ws_handle(),
                                        wsrep::ws_meta());
   thd->wsrep_applier_service->after_apply();
+  thd->thread_stack= orig_thread_stack;
+  WSREP_DEBUG("rollbacker aborted thd: (%llu %lu)",
+              thd->thread_id, thd->real_id);
   /* Will free THD */
   Wsrep_server_state::instance().server_service()
     .release_high_priority_service(thd->wsrep_applier_service);
 }
 
-static void wsrep_rollback_local(THD *thd)
+static void wsrep_rollback_local(THD *thd, THD *rollbacker)
 {
-  WSREP_INFO("Wsrep_rollback_local");
+  WSREP_DEBUG("Rollbacker aborting local thd (%llu %lu)",
+              thd->thread_id, thd->real_id);
+  char* orig_thread_stack= thd->thread_stack;
+  thd->thread_stack= rollbacker->thread_stack;
   if (thd->wsrep_trx().is_streaming())
   {
-    wsrep::transaction_id transaction_id(thd->wsrep_trx().id());
-    Wsrep_storage_service* storage_service=
-      static_cast<Wsrep_storage_service*>(
-            Wsrep_server_state::instance().server_service().
-            storage_service(thd->wsrep_cs().client_service()));
-
-    storage_service->store_globals();
-    storage_service->adopt_transaction(thd->wsrep_trx());
-    storage_service->remove_fragments();
-    storage_service->commit(wsrep::ws_handle(transaction_id, 0),
-                            wsrep::ws_meta());
-    Wsrep_server_state::instance().server_service().
-      release_storage_service(storage_service);
-    wsrep_store_threadvars(thd);
+    wsrep_remove_streaming_fragments(thd, "local");
   }
   /* Set thd->event_scheduler.data temporarily to NULL to avoid
      callbacks to threadpool wait_begin() during rollback. */
@@ -247,9 +208,10 @@ static void wsrep_rollback_local(THD *thd)
   /* Assign saved event_scheduler.data back before letting
      client to continue. */
   thd->event_scheduler.data= saved_esd;
+  thd->thread_stack= orig_thread_stack;
   thd->wsrep_cs().sync_rollback_complete();
-  WSREP_DEBUG("rollbacker aborted thd: (%llu %llu)",
-              thd->thread_id, (long long)thd->real_id);
+  WSREP_DEBUG("rollbacker aborted thd: (%llu %lu)",
+              thd->thread_id, thd->real_id);
 }
 
 static void wsrep_rollback_process(THD *rollbacker,
@@ -286,18 +248,13 @@ static void wsrep_rollback_process(THD *rollbacker,
 
     /* Rollback methods below may free thd pointer. Do not try
        to access it after method returns. */
-    if (thd->wsrep_trx().is_streaming() &&
-        thd->wsrep_trx().bf_aborted_in_total_order())
-    {
-      wsrep_rollback_streaming_aborted_by_toi(thd);
-    }
-    else if (wsrep_thd_is_applying(thd))
+    if (wsrep_thd_is_applying(thd))
     {
-      wsrep_rollback_high_priority(thd);
+      wsrep_rollback_high_priority(thd, rollbacker);
     }
     else
     {
-      wsrep_rollback_local(thd);
+      wsrep_rollback_local(thd, rollbacker);
     }
     wsrep_store_threadvars(rollbacker);
     thd_proc_info(rollbacker, "wsrep aborter idle");
@@ -345,7 +302,7 @@ void wsrep_fire_rollbacker(THD *thd)
 }
 
 
-int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
+int wsrep_abort_thd(THD *bf_thd_ptr, THD *victim_thd_ptr, my_bool signal)
 {
   DBUG_ENTER("wsrep_abort_thd");
   THD *victim_thd= (THD *) victim_thd_ptr;
@@ -373,7 +330,7 @@ int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
 
 bool wsrep_bf_abort(const THD* bf_thd, THD* victim_thd)
 {
-  WSREP_LOG_THD((THD*)bf_thd, "BF aborter before");
+  WSREP_LOG_THD(bf_thd, "BF aborter before");
   WSREP_LOG_THD(victim_thd, "victim before");
   wsrep::seqno bf_seqno(bf_thd->wsrep_trx().ws_meta().seqno());
 
diff --git a/sql/wsrep_thd.h b/sql/wsrep_thd.h
index 872570cd028..f98c15a5a85 100644
--- a/sql/wsrep_thd.h
+++ b/sql/wsrep_thd.h
@@ -88,8 +88,8 @@ void wsrep_create_appliers(long threads);
 void wsrep_create_rollbacker();
 
 bool wsrep_bf_abort(const THD*, THD*);
-int  wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr,
-                                my_bool signal);
+int  wsrep_abort_thd(THD *bf_thd_ptr, THD *victim_thd_ptr, my_bool signal);
+
 extern void  wsrep_thd_set_PA_safe(void *thd_ptr, my_bool safe);
 
 /*
@@ -262,7 +262,7 @@ static inline void wsrep_override_error(THD* thd,
    @param message Optional message
    @param function Function where the call was made from
  */
-static inline void wsrep_log_thd(THD *thd,
+static inline void wsrep_log_thd(const THD *thd,
                                  const char *message,
                                  const char *function)
 {