Merge 10.4 into 10.5

18 files changed, 333 insertions, 102 deletions

diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index a08928d964b..80cb6dd4322 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -3953,7 +3953,12 @@ int ha_partition::external_lock(THD *thd, int lock_type)
   {
     if (m_part_info->part_expr)
       m_part_info->part_expr->walk(&Item::register_field_in_read_map, 1, 0);
-    if (m_part_info->part_type == VERSIONING_PARTITION)
+    if (m_part_info->part_type == VERSIONING_PARTITION &&
+      /* TODO: MDEV-20345 exclude more inapproriate commands like INSERT
+         These commands may be excluded because working history partition is needed
+         only for versioned DML. */
+      thd->lex->sql_command != SQLCOM_SELECT &&
+      thd->lex->sql_command != SQLCOM_INSERT_SELECT)
       m_part_info->vers_set_hist_part(thd);
   }
   DBUG_RETURN(0);
@@ -4095,8 +4100,24 @@ int ha_partition::start_stmt(THD *thd, thr_lock_type lock_type)
     /* Add partition to be called in reset(). */
     bitmap_set_bit(&m_partitions_to_reset, i);
   }
-  if (lock_type == F_WRLCK && m_part_info->part_expr)
-    m_part_info->part_expr->walk(&Item::register_field_in_read_map, 1, 0);
+  switch (lock_type)
+  {
+  case TL_WRITE_ALLOW_WRITE:
+  case TL_WRITE_CONCURRENT_INSERT:
+  case TL_WRITE_DELAYED:
+  case TL_WRITE_DEFAULT:
+  case TL_WRITE_LOW_PRIORITY:
+  case TL_WRITE:
+  case TL_WRITE_ONLY:
+    if (m_part_info->part_expr)
+      m_part_info->part_expr->walk(&Item::register_field_in_read_map, 1, 0);
+    if (m_part_info->part_type == VERSIONING_PARTITION &&
+      // TODO: MDEV-20345 (see above)
+      thd->lex->sql_command != SQLCOM_SELECT &&
+      thd->lex->sql_command != SQLCOM_INSERT_SELECT)
+      m_part_info->vers_set_hist_part(thd);
+  default:;
+  }
   DBUG_RETURN(error);
 }
 
diff --git a/sql/item.h b/sql/item.h
index 49852d38e1a..d220f171aed 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -2,7 +2,7 @@
 #define SQL_ITEM_INCLUDED
 
 /* Copyright (c) 2000, 2017, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2018, MariaDB Corporation
+   Copyright (c) 2009, 2019, 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
@@ -719,7 +719,6 @@ public:
 class Item: public Value_source,
             public Type_all_attributes
 {
-  void operator=(Item &);
   /**
     The index in the JOIN::join_tab array of the JOIN_TAB this Item is attached
     to. Items are attached (or 'pushed') to JOIN_TABs during optimization by the
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index b0eb8f4d158..234d36b36fe 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -3176,7 +3176,7 @@ bool Item_func_pad::fix_length_and_dec()
   DBUG_ASSERT(collation.collation->mbmaxlen > 0);
   if (args[1]->const_item() && !args[1]->is_expensive())
   {
-    fix_char_length(Repeat_count(args[1]).count());
+    fix_char_length_ulonglong(Repeat_count(args[1]).count());
     return false;
   }
   max_length= MAX_BLOB_WIDTH;
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index f4bd76e4695..9e6b786d11f 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2015, Oracle and/or its affiliates.
-   Copyright (c) 2008, 2015, MariaDB
+   Copyright (c) 2008, 2019, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -2692,6 +2692,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
     KEY_PART *key_parts;
     KEY *key_info;
     PARAM param;
+    bool force_group_by = false;
 
     if (check_stack_overrun(thd, 2*STACK_MIN_SIZE + sizeof(PARAM), buff))
       DBUG_RETURN(0);                           // Fatal error flag is set
@@ -2856,6 +2857,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
       Try to construct a QUICK_GROUP_MIN_MAX_SELECT.
       Notice that it can be constructed no matter if there is a range tree.
     */
+    DBUG_EXECUTE_IF("force_group_by", force_group_by = true; );
     if (!only_single_index_range_scan)
       group_trp= get_best_group_min_max(&param, tree, best_read_time);
     if (group_trp)
@@ -2867,11 +2869,15 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
       if (unlikely(thd->trace_started()))
         group_trp->trace_basic_info(&param, &grp_summary);
 
-      if (group_trp->read_cost < best_read_time)
+      if (group_trp->read_cost < best_read_time || force_group_by)
       {
         grp_summary.add("chosen", true);
         best_trp= group_trp;
         best_read_time= best_trp->read_cost;
+        if (force_group_by)
+        {
+          goto force_plan;
+        }
       }
       else
         grp_summary.add("chosen", false).add("cause", "cost");
@@ -2977,6 +2983,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
       }
     }
 
+force_plan:
     thd->mem_root= param.old_root;
 
     /* If we got a read plan, create a quick select from it. */
@@ -12173,13 +12180,28 @@ int QUICK_RANGE_SELECT::get_next_prefix(uint prefix_length,
       DBUG_ASSERT(cur_prefix != NULL);
       result= file->ha_index_read_map(record, cur_prefix, keypart_map,
                                       HA_READ_AFTER_KEY);
-      if (result || last_range->max_keypart_map == 0)
-        DBUG_RETURN(result);
-
-      key_range previous_endpoint;
-      last_range->make_max_endpoint(&previous_endpoint, prefix_length, keypart_map);
-      if (file->compare_key(&previous_endpoint) <= 0)
-        DBUG_RETURN(0);
+      if (result || last_range->max_keypart_map == 0) {
+        /*
+          Only return if actual failure occurred. For HA_ERR_KEY_NOT_FOUND
+          or HA_ERR_END_OF_FILE, we just want to continue to reach the next
+          set of ranges. It is possible for the storage engine to return
+          HA_ERR_KEY_NOT_FOUND/HA_ERR_END_OF_FILE even when there are more
+          keys if it respects the end range set by the read_range_first call
+          below.
+        */
+        if (result != HA_ERR_KEY_NOT_FOUND && result != HA_ERR_END_OF_FILE)
+          DBUG_RETURN(result);
+      } else {
+        /*
+          For storage engines that don't respect end range, check if we've
+          moved past the current range.
+        */
+        key_range previous_endpoint;
+        last_range->make_max_endpoint(&previous_endpoint, prefix_length,
+                                      keypart_map);
+        if (file->compare_key(&previous_endpoint) <= 0)
+          DBUG_RETURN(0);
+      }
     }
 
     uint count= ranges.elements - (uint)(cur_range - (QUICK_RANGE**) ranges.buffer);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 5f39f514b59..7ac0dcad596 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2608,6 +2608,7 @@ unlink_all_closed_tables(THD *thd, MYSQL_LOCK *lock, size_t reopen_count)
       DBUG_ASSERT(thd->open_tables == m_reopen_array[reopen_count]);
 
       thd->open_tables->pos_in_locked_tables->table= NULL;
+      thd->open_tables->pos_in_locked_tables= NULL;
 
       close_thread_table(thd, &thd->open_tables);
     }
@@ -8461,8 +8462,8 @@ fill_record(THD *thd, TABLE *table_arg, List<Item> &fields, List<Item> &values,
         rfield->field_index ==  table->next_number_field->field_index)
       table->auto_increment_field_not_null= TRUE;
     Item::Type type= value->type();
-    bool vers_sys_field= table->versioned() && rfield->vers_sys_field();
-    if ((rfield->vcol_info || vers_sys_field) &&
+    const bool skip_sys_field= rfield->vers_sys_field(); // TODO: && !thd->vers_modify_history() [MDEV-16546]
+    if ((rfield->vcol_info || skip_sys_field) &&
         type != Item::DEFAULT_VALUE_ITEM &&
         type != Item::NULL_ITEM &&
         table->s->table_category != TABLE_CATEGORY_TEMPORARY)
@@ -8471,15 +8472,14 @@ fill_record(THD *thd, TABLE *table_arg, List<Item> &fields, List<Item> &values,
                           ER_WARNING_NON_DEFAULT_VALUE_FOR_GENERATED_COLUMN,
                           ER_THD(thd, ER_WARNING_NON_DEFAULT_VALUE_FOR_GENERATED_COLUMN),
                           rfield->field_name.str, table->s->table_name.str);
-      if (vers_sys_field)
-        continue;
     }
     if (only_unvers_fields && !rfield->vers_update_unversioned())
       only_unvers_fields= false;
 
     if (rfield->stored_in_db())
     {
-      if (unlikely(value->save_in_field(rfield, 0) < 0) && !ignore_errors)
+      if (!skip_sys_field &&
+          unlikely(value->save_in_field(rfield, 0) < 0) && !ignore_errors)
       {
         my_message(ER_UNKNOWN_ERROR, ER_THD(thd, ER_UNKNOWN_ERROR), MYF(0));
         goto err;
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 9078396a575..4eab241232b 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -4906,12 +4906,6 @@ extern "C" int thd_slave_thread(const MYSQL_THD thd)
 }
 
 
-extern "C" int thd_rpl_stmt_based(const MYSQL_THD thd)
-{
-  return thd &&
-    !thd->is_current_stmt_binlog_format_row() &&
-    !thd->is_current_stmt_binlog_disabled();
-}
 
 
 /* Returns high resolution timestamp for the start
@@ -6251,6 +6245,48 @@ int THD::decide_logging_format(TABLE_LIST *tables)
   DBUG_RETURN(0);
 }
 
+int THD::decide_logging_format_low(TABLE *table)
+{
+  /*
+   INSERT...ON DUPLICATE KEY UPDATE on a table with more than one unique keys
+   can be unsafe.
+   */
+  if(wsrep_binlog_format() <= BINLOG_FORMAT_STMT &&
+       !is_current_stmt_binlog_format_row() &&
+       !lex->is_stmt_unsafe() &&
+       lex->sql_command == SQLCOM_INSERT &&
+       lex->duplicates == DUP_UPDATE)
+  {
+    uint unique_keys= 0;
+    uint keys= table->s->keys, i= 0;
+    Field *field;
+    for (KEY* keyinfo= table->s->key_info;
+             i < keys && unique_keys <= 1; i++, keyinfo++)
+      if (keyinfo->flags & HA_NOSAME &&
+         !(keyinfo->key_part->field->flags & AUTO_INCREMENT_FLAG &&
+             //User given auto inc can be unsafe
+             !keyinfo->key_part->field->val_int()))
+      {
+        for (uint j= 0; j < keyinfo->user_defined_key_parts; j++)
+        {
+          field= keyinfo->key_part[j].field;
+          if(!bitmap_is_set(table->write_set,field->field_index))
+            goto exit;
+        }
+        unique_keys++;
+exit:;
+      }
+
+    if (unique_keys > 1)
+    {
+      lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_TWO_KEYS);
+      binlog_unsafe_warning_flags|= lex->get_stmt_unsafe_flags();
+      set_current_stmt_binlog_format_row_if_mixed();
+      return 1;
+    }
+  }
+  return 0;
+}
 
 /*
   Implementation of interface to write rows to the binary log through the
diff --git a/sql/sql_class.h b/sql/sql_class.h
index e8abcb3af34..152bf0617cd 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2448,6 +2448,20 @@ public:
   /* container for handler's private per-connection data */
   Ha_data ha_data[MAX_HA];
 
+  /**
+    Bit field for the state of binlog warnings.
+
+    The first Lex::BINLOG_STMT_UNSAFE_COUNT bits list all types of
+    unsafeness that the current statement has.
+
+    This must be a member of THD and not of LEX, because warnings are
+    detected and issued in different places (@c
+    decide_logging_format() and @c binlog_query(), respectively).
+    Between these calls, the THD->lex object may change; e.g., if a
+    stored routine is invoked.  Only THD persists between the calls.
+  */
+  uint32 binlog_unsafe_warning_flags;
+
 #ifndef MYSQL_CLIENT
   binlog_cache_mngr *  binlog_setup_trx_data();
 
@@ -2557,20 +2571,6 @@ private:
   */
   enum_binlog_format current_stmt_binlog_format;
 
-  /**
-    Bit field for the state of binlog warnings.
-
-    The first Lex::BINLOG_STMT_UNSAFE_COUNT bits list all types of
-    unsafeness that the current statement has.
-
-    This must be a member of THD and not of LEX, because warnings are
-    detected and issued in different places (@c
-    decide_logging_format() and @c binlog_query(), respectively).
-    Between these calls, the THD->lex object may change; e.g., if a
-    stored routine is invoked.  Only THD persists between the calls.
-  */
-  uint32 binlog_unsafe_warning_flags;
-
   /*
     Number of outstanding table maps, i.e., table maps in the
     transaction cache.
@@ -4558,6 +4558,18 @@ public:
   }
   void leave_locked_tables_mode();
   int decide_logging_format(TABLE_LIST *tables);
+  /*
+   In Some cases when decide_logging_format is called it does not have all
+   information to decide the logging format. So that cases we call decide_logging_format_2
+   at later stages in execution.
+   One example would be binlog format for IODKU but column with unique key is not inserted.
+   We dont have inserted columns info when we call decide_logging_format so on later stage we call
+   decide_logging_format_low
+
+   @returns 0 if no format is changed
+            1 if there is change in binlog format
+  */
+  int decide_logging_format_low(TABLE *table);
 
   enum need_invoker { INVOKER_NONE=0, INVOKER_USER, INVOKER_ROLE};
   void binlog_invoker(bool role) { m_binlog_invoker= role ? INVOKER_ROLE : INVOKER_USER; }
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 7d399464e4c..e8e320e8439 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1061,6 +1061,7 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
         break;
       }
 
+      thd->decide_logging_format_low(table);
 #ifndef EMBEDDED_LIBRARY
       if (lock_type == TL_WRITE_DELAYED)
       {
diff --git a/sql/sql_list.h b/sql/sql_list.h
index f5d8ed98b02..3585adbd714 100644
--- a/sql/sql_list.h
+++ b/sql/sql_list.h
@@ -1,6 +1,7 @@
 #ifndef INCLUDES_MYSQL_SQL_LIST_H
 #define INCLUDES_MYSQL_SQL_LIST_H
 /* Copyright (c) 2000, 2012, Oracle and/or its affiliates.
+   Copyright (c) 2019, 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
@@ -48,6 +49,14 @@ public:
     next= elements ? tmp.next : &first;
   }
 
+  SQL_I_List& operator=(const SQL_I_List &tmp)
+  {
+    elements= tmp.elements;
+    first= tmp.first;
+    next= tmp.next;
+    return *this;
+  }
+
   inline void empty()
   {
     elements= 0;
@@ -488,7 +497,6 @@ template <class T> class List :public base_list
 {
 public:
   inline List() :base_list() {}
-  inline List(const List<T> &tmp) :base_list(tmp) {}
   inline List(const List<T> &tmp, MEM_ROOT *mem_root) :
     base_list(tmp, mem_root) {}
   inline bool push_back(T *a) { return base_list::push_back(a); }
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index a15fb860578..0e8f331f2cf 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -868,7 +868,6 @@ Item* period_get_condition(THD *thd, TABLE_LIST *table, SELECT_LEX *select,
     cond1= and_items(thd, cond3, cond1);
   }
   return cond1;
-#undef newx
 }
 
 static
@@ -917,6 +916,19 @@ Item* SELECT_LEX::period_setup_conds(THD *thd, TABLE_LIST *tables, Item *where)
   DBUG_RETURN(result);
 }
 
+/**
+  Setup System Versioning conditions
+
+  Add WHERE condition according to FOR SYSTEM_TIME clause.
+
+  If the table is partitioned by SYSTEM_TIME and there is no FOR SYSTEM_TIME
+  clause, then select now-partition instead of modifying WHERE condition.
+
+  @retval
+    -1    on error
+  @retval
+    0     on success
+*/
 int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
 {
   DBUG_ENTER("SELECT_LEX::vers_setup_conds");
@@ -974,12 +986,13 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
     vers_select_conds_t &vers_conditions= table->vers_conditions;
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-      /*
-        if the history is stored in partitions, then partitions
-        themselves are not versioned
-      */
-      if (table->partition_names && table->table->part_info->vers_info)
+    Vers_part_info *vers_info;
+    if (table->table->part_info && (vers_info= table->table->part_info->vers_info))
+    {
+      if (table->partition_names)
       {
+        /* If the history is stored in partitions, then partitions
+            themselves are not versioned. */
         if (vers_conditions.is_set())
         {
           my_error(ER_VERS_QUERY_IN_PARTITION, MYF(0), table->alias.str);
@@ -988,6 +1001,19 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
         else
           vers_conditions.init(SYSTEM_TIME_ALL);
       }
+      else if (!vers_conditions.is_set() &&
+               /* We cannot optimize REPLACE .. SELECT because it may need
+                  to call vers_set_hist_part() to update history. */
+               thd->lex->sql_command != SQLCOM_REPLACE_SELECT)
+      {
+        table->partition_names= newx List<String>;
+        String *s= newx String(vers_info->now_part->partition_name,
+                               system_charset_info);
+        table->partition_names->push_back(s);
+        table->table->file->change_partitions_to_open(table->partition_names);
+        vers_conditions.init(SYSTEM_TIME_ALL);
+      }
+    }
 #endif
 
     if (outer_table && !vers_conditions.is_set())
@@ -1042,6 +1068,7 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
 
   DBUG_RETURN(0);
 }
+#undef newx
 
 /*****************************************************************************
   Check fields, find best join, do the select and output fields.
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index b7781d3389e..9bb1d98152b 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2392,35 +2392,6 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       /* remove .frm file and engine files */
       path_length= build_table_filename(path, sizeof(path) - 1, db.str, alias.str,
                                         reg_ext, 0);
-
-      /*
-        This handles the case where a "DROP" was executed and a regular
-        table "may be" dropped as drop_temporary is FALSE and error is
-        TRUE. If the error was FALSE a temporary table was dropped and
-        regardless of the status of drop_temporary a "DROP TEMPORARY"
-        must be used.
-      */
-      if (!dont_log_query)
-      {
-        /*
-          Note that unless if_exists is TRUE or a temporary table was deleted, 
-          there is no means to know if the statement should be written to the
-          binary log. See further information on this variable in what follows.
-        */
-        non_tmp_table_deleted= (if_exists ? TRUE : non_tmp_table_deleted);
-        /*
-          Don't write the database name if it is the current one (or if
-          thd->db is NULL).
-        */
-        if (thd->db.str == NULL || cmp(&db, &thd->db) != 0)
-        {
-          append_identifier(thd, &built_query, &db);
-          built_query.append(".");
-        }
-
-        append_identifier(thd, &built_query, &table->table_name);
-        built_query.append(",");
-      }
     }
     DEBUG_SYNC(thd, "rm_table_no_locks_before_delete_table");
     error= 0;
@@ -2500,9 +2471,16 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       // Remove extension for delete
       *(end= path + path_length - reg_ext_length)= '\0';
 
-      error= ha_delete_table(thd, table_type, path, &db, &table->table_name,
-                             !dont_log_query);
-      if (!error)
+      if ((error= ha_delete_table(thd, table_type, path, &db, &table->table_name,
+                                  !dont_log_query)))
+      {
+        if (thd->is_killed())
+        {
+          error= -1;
+          goto err;
+        }
+      }
+      else
       {
         /* Delete the table definition file */
         strmov(end,reg_ext);
@@ -2546,7 +2524,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
     if (error)
     {
       if (wrong_tables.length())
-	wrong_tables.append(',');
+        wrong_tables.append(',');
       wrong_tables.append(&db);
       wrong_tables.append('.');
       wrong_tables.append(&table->table_name);
@@ -2559,6 +2537,22 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       mysql_audit_drop_table(thd, table);
     }
 
+    if (!dont_log_query && !drop_temporary)
+    {
+      non_tmp_table_deleted= (if_exists ? TRUE : non_tmp_table_deleted);
+      /*
+         Don't write the database name if it is the current one (or if
+         thd->db is NULL).
+       */
+      if (thd->db.str == NULL || cmp(&db, &thd->db) != 0)
+      {
+        append_identifier(thd, &built_query, &db);
+        built_query.append(".");
+      }
+
+      append_identifier(thd, &built_query, &table->table_name);
+      built_query.append(",");
+    }
     DBUG_PRINT("table", ("table: %p  s: %p", table->table,
                          table->table ?  table->table->s :  NULL));
   }
@@ -5293,7 +5287,7 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
 
 err:
   /* In RBR we don't need to log CREATE TEMPORARY TABLE */
-  if (thd->is_current_stmt_binlog_format_row() && create_info->tmp_table())
+  if (!result && thd->is_current_stmt_binlog_format_row() && create_info->tmp_table())
     DBUG_RETURN(result);
 
   if (create_info->tmp_table())
@@ -7961,9 +7955,11 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
   KEY *key_info=table->key_info;
   bool rc= TRUE;
   bool modified_primary_key= FALSE;
+  bool vers_system_invisible= false;
   Create_field *def;
   Field **f_ptr,*field;
   MY_BITMAP *dropped_fields= NULL; // if it's NULL - no dropped fields
+  bool save_reopen= table->m_needs_reopen;
   bool drop_period= false;
   DBUG_ENTER("mysql_prepare_alter_table");
 
@@ -8070,7 +8066,11 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
       bitmap_set_bit(dropped_fields, field->field_index);
       continue;
     }
-
+    if (field->invisible == INVISIBLE_SYSTEM &&
+        field->flags & VERS_SYSTEM_FIELD)
+    {
+      vers_system_invisible= true;
+    }
     /* invisible versioning column is dropped automatically on DROP SYSTEM VERSIONING */
     if (!drop && field->invisible >= INVISIBLE_SYSTEM &&
         field->flags & VERS_SYSTEM_FIELD &&
@@ -8188,7 +8188,8 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
   dropped_sys_vers_fields &= VERS_SYSTEM_FIELD;
   if ((dropped_sys_vers_fields ||
        alter_info->flags & ALTER_DROP_PERIOD) &&
-      dropped_sys_vers_fields != VERS_SYSTEM_FIELD)
+      dropped_sys_vers_fields != VERS_SYSTEM_FIELD &&
+      !vers_system_invisible)
   {
     StringBuffer<NAME_LEN*3> tmp;
     append_drop_column(thd, dropped_sys_vers_fields & VERS_SYS_START_FLAG,
@@ -8198,6 +8199,11 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
     my_error(ER_MISSING, MYF(0), table->s->table_name.str, tmp.c_ptr());
     goto err;
   }
+  else if (alter_info->flags & ALTER_DROP_PERIOD && vers_system_invisible)
+  {
+    my_error(ER_CANT_DROP_FIELD_OR_KEY, MYF(0), "PERIOD FOR SYSTEM_TIME on", table->s->table_name.str);
+    goto err;
+  }
   alter_info->flags &= ~(ALTER_DROP_PERIOD | ALTER_ADD_PERIOD);
   def_it.rewind();
   while ((def=def_it++))			// Add new columns
@@ -8701,7 +8707,9 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
   alter_info->create_list.swap(new_create_list);
   alter_info->key_list.swap(new_key_list);
   alter_info->check_constraint_list.swap(new_constraint_list);
+  DBUG_RETURN(rc);
 err:
+  table->m_needs_reopen= save_reopen;
   DBUG_RETURN(rc);
 }
 
@@ -10412,6 +10420,7 @@ end_temporary:
 	      (ulong) (copied + deleted), (ulong) deleted,
 	      (ulong) thd->get_stmt_da()->current_statement_warn_count());
   my_ok(thd, copied + deleted, 0L, alter_ctx.tmp_buff);
+  DEBUG_SYNC(thd, "alter_table_inplace_trans_commit");
   DBUG_RETURN(false);
 
 err_new_table_cleanup:
@@ -10462,7 +10471,8 @@ err_with_mdl:
     tables and release the exclusive metadata lock.
   */
   thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
-  thd->mdl_context.release_all_locks_for_name(mdl_ticket);
+  if (!table_list->table)
+    thd->mdl_context.release_all_locks_for_name(mdl_ticket);
   DBUG_RETURN(true);
 }
 
@@ -10495,12 +10505,14 @@ bool mysql_trans_commit_alter_copy_data(THD *thd)
   uint save_unsafe_rollback_flags;
   DBUG_ENTER("mysql_trans_commit_alter_copy_data");
 
-  /* Save flags as transcommit_implicit_are_deleting_them */
+  /* Save flags as trans_commit_implicit are deleting them */
   save_unsafe_rollback_flags= thd->transaction.stmt.m_unsafe_rollback_flags;
 
+  DEBUG_SYNC(thd, "alter_table_copy_trans_commit");
+
   if (ha_enable_transaction(thd, TRUE))
     DBUG_RETURN(TRUE);
-  
+
   /*
     Ensure that the new table is saved properly to disk before installing
     the new .frm.
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index c518bf7fccc..edf6b761e48 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7274,6 +7274,11 @@ serial_attribute:
           {
             Lex->last_field->versioning= $1;
             Lex->create_info.options|= HA_VERSIONED_TABLE;
+            if (Lex->alter_info.flags & ALTER_DROP_SYSTEM_VERSIONING)
+            {
+              my_yyabort_error((ER_VERS_NOT_VERSIONED, MYF(0),
+                       Lex->create_last_non_select_table->table_name.str));
+            }
           }
         ;
 
@@ -8485,6 +8490,7 @@ alter_list_item:
         | DROP SYSTEM VERSIONING_SYM
           {
             Lex->alter_info.flags|= ALTER_DROP_SYSTEM_VERSIONING;
+            Lex->create_info.options&= ~HA_VERSIONED_TABLE;
           }
         | DROP PERIOD_SYM FOR_SYSTEM_TIME_SYM
           {
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index bdefb76ad81..f938bcfd9d5 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -7352,6 +7352,11 @@ serial_attribute:
           {
             Lex->last_field->versioning= $1;
             Lex->create_info.options|= HA_VERSIONED_TABLE;
+            if (Lex->alter_info.flags & ALTER_DROP_SYSTEM_VERSIONING)
+            {
+              my_yyabort_error((ER_VERS_NOT_VERSIONED, MYF(0),
+                       Lex->create_last_non_select_table->table_name.str));
+            }
           }
         ;
 
@@ -8576,6 +8581,7 @@ alter_list_item:
         | DROP SYSTEM VERSIONING_SYM
           {
             Lex->alter_info.flags|= ALTER_DROP_SYSTEM_VERSIONING;
+            Lex->create_info.options&= ~HA_VERSIONED_TABLE;
           }
         | DROP PERIOD_SYM FOR_SYSTEM_TIME_SYM
           {
diff --git a/sql/tztime.cc b/sql/tztime.cc
index 09d775e7e51..c45a6598d69 100644
--- a/sql/tztime.cc
+++ b/sql/tztime.cc
@@ -147,6 +147,7 @@ typedef struct st_time_zone_info
 
 static my_bool prepare_tz_info(TIME_ZONE_INFO *sp, MEM_ROOT *storage);
 
+my_bool opt_leap, opt_verbose, opt_skip_write_binlog;
 
 #if defined(TZINFO2SQL) || defined(TESTTIME)
 
@@ -2428,6 +2429,14 @@ print_tz_leaps_as_sql(const TIME_ZONE_INFO *sp)
     We are assuming that there are only one list of leap seconds
     For all timezones.
   */
+  if (!opt_skip_write_binlog)
+      printf("\\d |\n"
+        "IF (select count(*) from information_schema.global_variables where\n"
+        "variable_name='wsrep_on') = 1 THEN\n"
+        "ALTER TABLE time_zone_leap_second ENGINE=InnoDB;\n"
+        "END IF|\n"
+        "\\d ;\n");
+
   printf("TRUNCATE TABLE time_zone_leap_second;\n");
 
   if (sp->leapcnt)
@@ -2440,6 +2449,14 @@ print_tz_leaps_as_sql(const TIME_ZONE_INFO *sp)
     printf(";\n");
   }
 
+  if (!opt_skip_write_binlog)
+      printf("\\d |\n"
+        "IF (select count(*) from information_schema.global_variables where\n"
+        "variable_name='wsrep_on') = 1 THEN\n"
+        "ALTER TABLE time_zone_leap_second ENGINE=Aria;\n"
+        "END IF|\n"
+        "\\d ;\n");
+
   printf("ALTER TABLE time_zone_leap_second ORDER BY Transition_time;\n");
 }
 
@@ -2597,8 +2614,6 @@ scan_tz_dir(char * name_end, uint symlink_recursion_level, uint verbose)
 }
 
 
-my_bool opt_leap, opt_verbose;
-
 static const char *load_default_groups[]=
 { "mysql_tzinfo_to_sql", 0};
 
@@ -2619,6 +2634,8 @@ static struct my_option my_long_options[] =
    &opt_verbose, &opt_verbose, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"version", 'V', "Output version information and exit.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"skip-write-binlog", 'S', "Do not replicate changes to time zone tables to other nodes in a Galera cluster",
+   &opt_skip_write_binlog,&opt_skip_write_binlog, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 
@@ -2687,11 +2704,14 @@ main(int argc, char **argv)
     return 1;
   }
 
-  // Replicate MyISAM DDL for this session, cf. lp:1161432
-  // timezone info unfixable in XtraDB Cluster
-  printf("set @prep=if((select count(*) from information_schema.global_variables where variable_name='wsrep_on'), 'SET GLOBAL wsrep_replicate_myisam=?', 'do ?');\n"
-         "prepare set_wsrep_myisam from @prep;\n"
-         "set @toggle=1; execute set_wsrep_myisam using @toggle;\n");
+  if (opt_skip_write_binlog)
+    /* If skip_write_binlog is set and wsrep is compiled in we disable
+       sql_log_bin and wsrep_on to avoid Galera replicating below
+       truncate table clauses. This will allow user to set different
+       time zones to nodes in Galera cluster. */
+    printf("set @prep1=if((select count(*) from information_schema.global_variables where variable_name='wsrep_on'), 'SET SESSION SQL_LOG_BIN=?, WSREP_ON=OFF;', 'do ?');\n"
+           "prepare set_wsrep_write_binlog from @prep1;\n"
+           "set @toggle=0; execute set_wsrep_write_binlog using @toggle;\n");
 
   if (argc == 1 && !opt_leap)
   {
@@ -2699,6 +2719,21 @@ main(int argc, char **argv)
 
     root_name_end= strmake_buf(fullname, argv[0]);
 
+    if(!opt_skip_write_binlog)
+    {
+      // Alter time zone tables to InnoDB if wsrep_on is enabled
+      // to allow changes to them to replicate with Galera
+      printf("\\d |\n"
+        "IF (select count(*) from information_schema.global_variables where\n"
+        "variable_name='wsrep_on') = 1 THEN\n"
+        "ALTER TABLE time_zone ENGINE=InnoDB;\n"
+        "ALTER TABLE time_zone_name ENGINE=InnoDB;\n"
+        "ALTER TABLE time_zone_transition ENGINE=InnoDB;\n"
+        "ALTER TABLE time_zone_transition_type ENGINE=InnoDB;\n"
+        "END IF|\n"
+        "\\d ;\n");
+    }
+
     printf("TRUNCATE TABLE time_zone;\n");
     printf("TRUNCATE TABLE time_zone_name;\n");
     printf("TRUNCATE TABLE time_zone_transition;\n");
@@ -2740,8 +2775,19 @@ main(int argc, char **argv)
     free_root(&tz_storage, MYF(0));
   }
 
-  // Reset wsrep_replicate_myisam. lp:1161432
-  printf("set @toggle=0; execute set_wsrep_myisam using @toggle;\n");
+  if(!opt_skip_write_binlog)
+  {
+      // Fall back to Aria
+      printf("\\d |\n"
+        "IF (select count(*) from information_schema.global_variables where\n"
+        "variable_name='wsrep_on') = 1 THEN\n"
+        "ALTER TABLE time_zone ENGINE=Aria;\n"
+        "ALTER TABLE time_zone_name ENGINE=Aria;\n"
+        "ALTER TABLE time_zone_transition ENGINE=Aria;\n"
+        "ALTER TABLE time_zone_transition_type ENGINE=Aria;\n"
+        "END IF|\n"
+        "\\d ;\n");
+  }
 
   free_defaults(default_argv);
   my_end(0);
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 6ea9eab96aa..128d80ad226 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -212,7 +212,19 @@ static PSI_file_info wsrep_files[]=
 {
   { &key_file_wsrep_gra_log, "wsrep_gra_log", 0}
 };
-#endif
+
+PSI_thread_key key_wsrep_sst_joiner, key_wsrep_sst_donor,
+  key_wsrep_rollbacker, key_wsrep_applier;
+
+static PSI_thread_info wsrep_threads[]=
+{
+ {&key_wsrep_sst_joiner, "wsrep_sst_joiner_thread", PSI_FLAG_GLOBAL},
+ {&key_wsrep_sst_donor, "wsrep_sst_donor_thread", PSI_FLAG_GLOBAL},
+ {&key_wsrep_rollbacker, "wsrep_rollbacker_thread", PSI_FLAG_GLOBAL},
+ {&key_wsrep_applier, "wsrep_applier_thread", PSI_FLAG_GLOBAL}
+};
+
+#endif /* HAVE_PSI_INTERFACE */
 
 my_bool wsrep_inited= 0; // initialized ?
 
@@ -759,6 +771,7 @@ void wsrep_thr_init()
   mysql_mutex_register("sql", wsrep_mutexes, array_elements(wsrep_mutexes));
   mysql_cond_register("sql", wsrep_conds, array_elements(wsrep_conds));
   mysql_file_register("sql", wsrep_files, array_elements(wsrep_files));
+  mysql_thread_register("sql", wsrep_threads, array_elements(wsrep_threads));
 #endif
 
   mysql_mutex_init(key_LOCK_wsrep_ready, &LOCK_wsrep_ready, MY_MUTEX_INIT_FAST);
diff --git a/sql/wsrep_mysqld.h b/sql/wsrep_mysqld.h
index 844437bab95..8714753ba76 100644
--- a/sql/wsrep_mysqld.h
+++ b/sql/wsrep_mysqld.h
@@ -341,7 +341,14 @@ extern PSI_mutex_key key_LOCK_wsrep_thd_queue;
 extern PSI_cond_key  key_COND_wsrep_thd_queue;
 
 extern PSI_file_key key_file_wsrep_gra_log;
+
+extern PSI_thread_key key_wsrep_sst_joiner;
+extern PSI_thread_key key_wsrep_sst_donor;
+extern PSI_thread_key key_wsrep_rollbacker;
+extern PSI_thread_key key_wsrep_applier;
 #endif /* HAVE_PSI_INTERFACE */
+
+
 struct TABLE_LIST;
 class Alter_info;
 int wsrep_to_isolation_begin(THD *thd, const char *db_, const char *table_,
diff --git a/sql/wsrep_sst.cc b/sql/wsrep_sst.cc
index 3181415dad1..85d5aca342d 100644
--- a/sql/wsrep_sst.cc
+++ b/sql/wsrep_sst.cc
@@ -663,10 +663,10 @@ static ssize_t sst_prepare_other (const char*  method,
   pthread_t tmp;
   sst_thread_arg arg(cmd_str(), env());
   mysql_mutex_lock (&arg.lock);
-  ret= pthread_create (&tmp, NULL, sst_joiner_thread, &arg);
+  ret = mysql_thread_create (key_wsrep_sst_joiner, &tmp, NULL, sst_joiner_thread, &arg);
   if (ret)
   {
-    WSREP_ERROR("sst_prepare_other(): pthread_create() failed: %d (%s)",
+    WSREP_ERROR("sst_prepare_other(): mysql_thread_create() failed: %d (%s)",
                 ret, strerror(ret));
     return -ret;
   }
@@ -1350,10 +1350,10 @@ static int sst_donate_other (const char*        method,
   pthread_t tmp;
   sst_thread_arg arg(cmd_str(), env);
   mysql_mutex_lock (&arg.lock);
-  ret= pthread_create (&tmp, NULL, sst_donor_thread, &arg);
+  ret = mysql_thread_create (key_wsrep_sst_donor, &tmp, NULL, sst_donor_thread, &arg);
   if (ret)
   {
-    WSREP_ERROR("sst_donate_other(): pthread_create() failed: %d (%s)",
+    WSREP_ERROR("sst_donate_other(): mysql_thread_create() failed: %d (%s)",
                 ret, strerror(ret));
     return ret;
   }
diff --git a/sql/wsrep_thd.cc b/sql/wsrep_thd.cc
index c62132b16a2..5907d495ee9 100644
--- a/sql/wsrep_thd.cc
+++ b/sql/wsrep_thd.cc
@@ -87,9 +87,24 @@ static bool create_wsrep_THD(Wsrep_thd_args* args)
 {
   ulong old_wsrep_running_threads= wsrep_running_threads;
   pthread_t unused;
+#ifdef HAVE_PSI_THREAD_INTERFACE
+  PSI_thread_key key;
 
-  bool res= pthread_create(&unused, &connection_attrib, start_wsrep_THD,
-                           args);
+  switch (args->thread_type())
+  {
+    case WSREP_APPLIER_THREAD:
+      key= key_wsrep_applier;
+      break;
+    case WSREP_ROLLBACKER_THREAD:
+      key= key_wsrep_rollbacker;
+      break;
+    default:
+      assert(0);
+      break;
+  }
+#endif
+  bool res= mysql_thread_create(key, &unused, &connection_attrib,
+                                start_wsrep_THD, (void*)args);
   /*
     if starting a thread on server startup, wait until the this thread's THD
     is fully initialized (otherwise a THD initialization code might