Merge tag 'mariadb-10.0.23' into 10.0-galera

48 files changed, 830 insertions, 331 deletions

diff --git a/sql/handler.cc b/sql/handler.cc
index 70f1d977c41..a7beb6fe9ed 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -5186,7 +5186,7 @@ bool Discovered_table_list::add_table(const char *tname, size_t tlen)
     custom discover_table_names() method, that calls add_table() directly).
     Note: avoid comparing the same name twice (here and in add_file).
   */
-  if (wild && my_wildcmp(files_charset_info, tname, tname + tlen, wild, wend,
+  if (wild && my_wildcmp(table_alias_charset, tname, tname + tlen, wild, wend,
                          wild_prefix, wild_one, wild_many))
       return 0;
 
diff --git a/sql/init.cc b/sql/init.cc
index 91b4b220bf3..8001e60b65e 100644
--- a/sql/init.cc
+++ b/sql/init.cc
@@ -42,9 +42,6 @@ void unireg_init(ulong options)
 
   current_pid=(ulong) getpid();		/* Save for later ref */
   my_init_time();			/* Init time-functions (read zone) */
-#ifndef EMBEDDED_LIBRARY
-  my_abort_hook=unireg_abort;		/* Abort with close of databases */
-#endif
 
   (void) strmov(reg_ext,".frm");
   reg_ext_length= 4;
diff --git a/sql/item.cc b/sql/item.cc
index 543ea4f0ffb..af6915d7468 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1884,6 +1884,8 @@ void Item::split_sum_func2(THD *thd, Item **ref_pointer_array,
     */
     Item_aggregate_ref *item_ref;
     uint el= fields.elements;
+    DBUG_ASSERT(fields.elements <=
+                thd->lex->current_select->ref_pointer_array_size);
     /*
       If this is an item_ref, get the original item
       This is a safety measure if this is called for things that is
@@ -6716,6 +6718,7 @@ Item *Item_field::update_value_transformer(uchar *select_arg)
   {
     List<Item> *all_fields= &select->join->all_fields;
     Item **ref_pointer_array= select->ref_pointer_array;
+    DBUG_ASSERT(all_fields->elements <= select->ref_pointer_array_size);
     int el= all_fields->elements;
     Item_ref *ref;
 
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 830a12ea48d..c763103a767 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2013, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2013, Monty Program Ab.
+   Copyright (c) 2009, 2015, 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
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index c4933e6d7ed..73f428752f8 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -1,7 +1,7 @@
 #ifndef ITEM_CMPFUNC_INCLUDED
 #define ITEM_CMPFUNC_INCLUDED
-/* Copyright (c) 2000, 2012, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2011, Monty Program Ab.
+/* Copyright (c) 2000, 2015, Oracle and/or its affiliates.
+   Copyright (c) 2009, 2015, 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
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 0e66b71e558..b79be7e9ce4 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2000, 2014, Oracle and/or its affiliates.
+/* Copyright (c) 2000, 2015, Oracle and/or its affiliates.
    Copyright (c) 2009, 2015, MariaDB
 
    This program is free software; you can redistribute it and/or modify
@@ -717,7 +717,7 @@ void Item_func::count_real_length()
 bool Item_func::count_string_result_length(enum_field_types field_type,
                                            Item **items, uint nitems)
 {
-  if (agg_arg_charsets(collation, items, nitems, MY_COLL_ALLOW_CONV, 1))
+  if (agg_arg_charsets_for_string_result(collation, items, nitems, 1))
     return true;
   if (is_temporal_type(field_type))
     count_datetime_length(items, nitems);
@@ -6228,9 +6228,7 @@ bool Item_func_match::fix_fields(THD *thd, Item **ref)
   table= 0;
   for (uint i=1 ; i < arg_count ; i++)
   {
-    item=args[i];
-    if (item->type() == Item::REF_ITEM)
-      args[i]= item= *((Item_ref *)item)->ref;
+    item= args[i]= args[i]->real_item();
     /*
       When running in PS mode, some Item_field's can already be replaced
       to Item_func_conv_charset during PREPARE time. This is possible
@@ -6243,7 +6241,7 @@ bool Item_func_match::fix_fields(THD *thd, Item **ref)
     if (!thd->stmt_arena->is_stmt_execute() &&
         item->type() != Item::FIELD_ITEM)
     {
-      my_error(ER_WRONG_ARGUMENTS, MYF(0), "AGAINST");
+      my_error(ER_WRONG_ARGUMENTS, MYF(0), "MATCH");
       return TRUE;
     }
     /*
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 972ae5afb16..54ab8f6aca1 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -4448,6 +4448,8 @@ bool Item_func_dyncol_create::prepare_arguments(bool force_names_arg)
     case DYN_COL_DYNCOL:
     case DYN_COL_STRING:
       res= args[valpos]->val_str(&tmp);
+      if (res && defs[i].cs)
+        res->set_charset(defs[i].cs);
       if (res &&
           (vals[i].x.string.value.str= sql_strmake(res->ptr(), res->length())))
       {
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index ce01fc7a3b8..7d361263548 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -1,5 +1,5 @@
-/* Copyright (c) 2002, 2012, Oracle and/or its affiliates.
-   Copyright (c) 2010, 2012, Monty Program Ab
+/* Copyright (c) 2002, 2015, Oracle and/or its affiliates.
+   Copyright (c) 2010, 2015, 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
diff --git a/sql/log.cc b/sql/log.cc
index 6432716c1f4..dcc210c7271 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -3424,7 +3424,7 @@ bool MYSQL_BIN_LOG::open(const char *log_name,
               it may be good to consider what actually happens when
               open_purge_index_file succeeds but register or sync fails.
 
-              Perhaps we might need the code below in MYSQL_LOG_BIN::cleanup
+              Perhaps we might need the code below in MYSQL_BIN_LOG::cleanup
               for "real life" purposes as well? 
      */
     DBUG_EXECUTE_IF("fault_injection_registering_index", {
@@ -7917,11 +7917,13 @@ int MYSQL_BIN_LOG::wait_for_update_bin_log(THD* thd,
   int ret= 0;
   DBUG_ENTER("wait_for_update_bin_log");
 
+  thd_wait_begin(thd, THD_WAIT_BINLOG);
   if (!timeout)
     mysql_cond_wait(&update_cond, &LOCK_log);
   else
     ret= mysql_cond_timedwait(&update_cond, &LOCK_log,
                               const_cast<struct timespec *>(timeout));
+  thd_wait_end(thd);
   DBUG_RETURN(ret);
 }
 
diff --git a/sql/log_event.cc b/sql/log_event.cc
index a1dcce7d7c4..3870467e1da 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -4308,7 +4308,8 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
           rgi->gtid_pending= false;
 
           gtid= rgi->current_gtid;
-          if (rpl_global_gtid_slave_state.record_gtid(thd, &gtid, sub_id, true, false))
+          if (rpl_global_gtid_slave_state->record_gtid(thd, &gtid, sub_id,
+                                                       true, false))
           {
             int errcode= thd->get_stmt_da()->sql_errno();
             if (!is_parallel_retry_error(rgi, errcode))
@@ -4527,7 +4528,7 @@ compare_errors:
 
 end:
   if (sub_id && !thd->is_slave_error)
-    rpl_global_gtid_slave_state.update_state_hash(sub_id, &gtid, rgi);
+    rpl_global_gtid_slave_state->update_state_hash(sub_id, &gtid, rgi);
 
   /*
     Probably we have set thd->query, thd->db, thd->catalog to point to places
@@ -6310,7 +6311,7 @@ int Rotate_log_event::do_update_pos(rpl_group_info *rgi)
                         rli->group_master_log_name,
                         (ulong) rli->group_master_log_pos));
     mysql_mutex_unlock(&rli->data_lock);
-    rpl_global_gtid_slave_state.record_and_update_gtid(thd, rgi);
+    rpl_global_gtid_slave_state->record_and_update_gtid(thd, rgi);
     flush_relay_log_info(rli);
     
     /*
@@ -6780,7 +6781,7 @@ Gtid_list_log_event::Gtid_list_log_event(const char *buf, uint event_len,
     for (i= 0; i < count; ++i)
     {
       if (!(sub_id_list[i]=
-            rpl_global_gtid_slave_state.next_sub_id(list[i].domain_id)))
+            rpl_global_gtid_slave_state->next_sub_id(list[i].domain_id)))
       {
         my_free(list);
         my_free(sub_id_list);
@@ -6835,7 +6836,7 @@ Gtid_list_log_event::Gtid_list_log_event(slave_connection_state *gtid_set,
       for (i= 0; i < count; ++i)
       {
         if (!(sub_id_list[i]=
-              rpl_global_gtid_slave_state.next_sub_id(list[i].domain_id)))
+              rpl_global_gtid_slave_state->next_sub_id(list[i].domain_id)))
         {
           my_free(list);
           my_free(sub_id_list);
@@ -6908,11 +6909,11 @@ Gtid_list_log_event::do_apply_event(rpl_group_info *rgi)
     uint32 i;
     for (i= 0; i < count; ++i)
     {
-      if ((ret= rpl_global_gtid_slave_state.record_gtid(thd, &list[i],
+      if ((ret= rpl_global_gtid_slave_state->record_gtid(thd, &list[i],
                                                         sub_id_list[i],
                                                         false, false)))
         return ret;
-      rpl_global_gtid_slave_state.update_state_hash(sub_id_list[i], &list[i],
+      rpl_global_gtid_slave_state->update_state_hash(sub_id_list[i], &list[i],
                                                     NULL);
     }
   }
@@ -7412,7 +7413,8 @@ int Xid_log_event::do_apply_event(rpl_group_info *rgi)
     rgi->gtid_pending= false;
 
     gtid= rgi->current_gtid;
-    err= rpl_global_gtid_slave_state.record_gtid(thd, &gtid, sub_id, true, false);
+    err= rpl_global_gtid_slave_state->record_gtid(thd, &gtid, sub_id, true,
+                                                  false);
     if (err)
     {
       int ec= thd->get_stmt_da()->sql_errno();
@@ -7445,7 +7447,7 @@ int Xid_log_event::do_apply_event(rpl_group_info *rgi)
   thd->mdl_context.release_transactional_locks();
 
   if (!res && sub_id)
-    rpl_global_gtid_slave_state.update_state_hash(sub_id, &gtid, rgi);
+    rpl_global_gtid_slave_state->update_state_hash(sub_id, &gtid, rgi);
 
   /*
     Increment the global status commit count variable
@@ -8212,7 +8214,7 @@ int Stop_log_event::do_update_pos(rpl_group_info *rgi)
     rgi->inc_event_relay_log_pos();
   else if (!rgi->is_parallel_exec)
   {
-    rpl_global_gtid_slave_state.record_and_update_gtid(thd, rgi);
+    rpl_global_gtid_slave_state->record_and_update_gtid(thd, rgi);
     rli->inc_group_relay_log_pos(0, rgi);
     flush_relay_log_info(rli);
   }
diff --git a/sql/mysql_install_db.cc b/sql/mysql_install_db.cc
index 50454f0f66a..9b4f45a9971 100644
--- a/sql/mysql_install_db.cc
+++ b/sql/mysql_install_db.cc
@@ -563,6 +563,10 @@ static int create_db_instance()
   if (!in)
     goto end;
 
+  if (setvbuf(in, NULL, _IONBF, 0))
+  {
+    verbose("WARNING: Cannot disable buffering on mysqld's stdin");
+  }
   if (fwrite("use mysql;\n",11,1, in) != 1)
   {
     verbose("ERROR: Cannot write to mysqld's stdin");
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 323b851ac1f..5b276b57274 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -494,7 +494,7 @@ ulong delay_key_write_options;
 uint protocol_version;
 uint lower_case_table_names;
 ulong tc_heuristic_recover= 0;
-int32 thread_count;
+int32 thread_count, service_thread_count;
 int32 thread_running;
 int32 slave_open_temp_tables;
 ulong thread_created;
@@ -1060,7 +1060,7 @@ PSI_cond_key key_RELAYLOG_update_cond, key_COND_wakeup_ready,
 PSI_cond_key key_RELAYLOG_COND_queue_busy;
 PSI_cond_key key_TC_LOG_MMAP_COND_queue_busy;
 PSI_cond_key key_COND_rpl_thread_queue, key_COND_rpl_thread,
-  key_COND_rpl_thread_pool,
+  key_COND_rpl_thread_stop, key_COND_rpl_thread_pool,
   key_COND_parallel_entry, key_COND_group_commit_orderer,
   key_COND_prepare_ordered, key_COND_slave_init;
 PSI_cond_key key_COND_wait_gtid, key_COND_gtid_ignore_duplicates;
@@ -1115,6 +1115,7 @@ static PSI_cond_info all_server_conds[]=
   { &key_COND_flush_thread_cache, "COND_flush_thread_cache", PSI_FLAG_GLOBAL},
   { &key_COND_rpl_thread, "COND_rpl_thread", 0},
   { &key_COND_rpl_thread_queue, "COND_rpl_thread_queue", 0},
+  { &key_COND_rpl_thread_stop, "COND_rpl_thread_stop", 0},
   { &key_COND_rpl_thread_pool, "COND_rpl_thread_pool", 0},
   { &key_COND_parallel_entry, "COND_parallel_entry", 0},
   { &key_COND_group_commit_orderer, "COND_group_commit_orderer", 0},
@@ -1809,7 +1810,7 @@ static void close_connections(void)
   /* All threads has now been aborted */
   DBUG_PRINT("quit",("Waiting for threads to die (count=%u)",thread_count));
   mysql_mutex_lock(&LOCK_thread_count);
-  while (thread_count)
+  while (thread_count || service_thread_count)
   {
     mysql_cond_wait(&COND_thread_count, &LOCK_thread_count);
     DBUG_PRINT("quit",("One thread died (count=%u)",thread_count));
@@ -2147,14 +2148,9 @@ void clean_up(bool print_message)
   item_func_sleep_free();
   lex_free();				/* Free some memory */
   item_create_cleanup();
-  if (!opt_noacl)
-  {
-#ifdef HAVE_DLOPEN
-    udf_free();
-#endif
-  }
   tdc_start_shutdown();
   plugin_shutdown();
+  udf_free();
   ha_end();
   if (tc_log)
     tc_log->close();
@@ -2889,8 +2885,27 @@ void delete_running_thd(THD *thd)
   delete thd;
   dec_thread_running();
   thread_safe_decrement32(&thread_count, &thread_count_lock);
-  if (!thread_count)
+  signal_thd_deleted();
+}
+
+
+/*
+  Send a signal to unblock close_conneciton() if there is no more
+  threads running with a THD attached
+
+  It's safe to check for thread_count and service_thread_count outside
+  of a mutex as we are only interested to see if they where decremented
+  to 0 by a previous unlink_thd() call.
+
+  We should only signal COND_thread_count if both variables are 0,
+  false positives are ok.
+*/
+
+void signal_thd_deleted()
+{
+  if (!thread_count && ! service_thread_count)
   {
+    /* Signal close_connections() that all THD's are freed */
     mysql_mutex_lock(&LOCK_thread_count);
     mysql_cond_broadcast(&COND_thread_count);
     mysql_mutex_unlock(&LOCK_thread_count);
@@ -3055,19 +3070,7 @@ bool one_thread_per_connection_end(THD *thd, bool put_in_cache)
 #endif /* WITH_WSREP */
     DBUG_RETURN(0);                             // Thread is reused
 
-  /*
-    It's safe to check for thread_count outside of the mutex
-    as we are only interested to see if it was counted to 0 by the
-    above unlink_thd() call. We should only signal COND_thread_count if
-    thread_count is likely to be 0. (false positives are ok)
-  */
-  if (!thread_count)
-  {
-    mysql_mutex_lock(&LOCK_thread_count);
-    DBUG_PRINT("signal", ("Broadcasting COND_thread_count"));
-    mysql_cond_broadcast(&COND_thread_count);
-    mysql_mutex_unlock(&LOCK_thread_count);
-  }
+  signal_thd_deleted();
   DBUG_LEAVE;                                   // Must match DBUG_ENTER()
 #if defined(HAVE_OPENSSL) && !defined(EMBEDDED_LIBRARY)
   ERR_remove_state(0);
@@ -6231,12 +6234,7 @@ int mysqld_main(int argc, char **argv)
   if (!opt_noacl)
     (void) grant_init();
 
-  if (!opt_noacl)
-  {
-#ifdef HAVE_DLOPEN
-    udf_init();
-#endif
-  }
+  udf_init();
 
   if (opt_bootstrap) /* If running with bootstrap, do not start replication. */
     opt_skip_slave_start= 1;
@@ -8994,7 +8992,8 @@ static int mysql_init_variables(void)
   cleanup_done= 0;
   server_id_supplied= 0;
   test_flags= select_errors= dropping_tables= ha_open_options=0;
-  thread_count= thread_running= kill_cached_threads= wake_thread=0;
+  thread_count= thread_running= kill_cached_threads= wake_thread= 0;
+  service_thread_count= 0;
   slave_open_temp_tables= 0;
   cached_thread_count= 0;
   opt_endinfo= using_udf_functions= 0;
@@ -10392,6 +10391,9 @@ PSI_stage_info stage_waiting_for_prior_transaction_to_commit= { 0, "Waiting for
 PSI_stage_info stage_waiting_for_prior_transaction_to_start_commit= { 0, "Waiting for prior transaction to start commit before starting next transaction", 0};
 PSI_stage_info stage_waiting_for_room_in_worker_thread= { 0, "Waiting for room in worker thread event queue", 0};
 PSI_stage_info stage_waiting_for_workers_idle= { 0, "Waiting for worker threads to be idle", 0};
+PSI_stage_info stage_waiting_for_ftwrl= { 0, "Waiting due to global read lock", 0};
+PSI_stage_info stage_waiting_for_ftwrl_threads_to_pause= { 0, "Waiting for worker threads to pause for global read lock", 0};
+PSI_stage_info stage_waiting_for_rpl_thread_pool= { 0, "Waiting while replication worker thread pool is busy", 0};
 PSI_stage_info stage_master_gtid_wait_primary= { 0, "Waiting in MASTER_GTID_WAIT() (primary waiter)", 0};
 PSI_stage_info stage_master_gtid_wait= { 0, "Waiting in MASTER_GTID_WAIT()", 0};
 PSI_stage_info stage_gtid_wait_other_connection= { 0, "Waiting for other master connection to process GTID received on multiple master connections", 0};
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 90204bc6dc6..d549013cd32 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -1,4 +1,5 @@
-/* Copyright (c) 2006, 2013, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2006, 2015, Oracle and/or its affiliates.
+   Copyright (c) 2010, 2015, 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
@@ -59,6 +60,7 @@ void close_connection(THD *thd, uint sql_errno= 0);
 void handle_connection_in_main_thread(THD *thd);
 void create_thread_to_handle_connection(THD *thd);
 void delete_running_thd(THD *thd);
+void signal_thd_deleted();
 void unlink_thd(THD *thd);
 bool one_thread_per_connection_end(THD *thd, bool put_in_cache);
 void flush_thread_cache();
@@ -309,7 +311,7 @@ extern PSI_cond_key key_RELAYLOG_update_cond, key_COND_wakeup_ready,
 extern PSI_cond_key key_RELAYLOG_COND_queue_busy;
 extern PSI_cond_key key_TC_LOG_MMAP_COND_queue_busy;
 extern PSI_cond_key key_COND_rpl_thread, key_COND_rpl_thread_queue,
-  key_COND_rpl_thread_pool,
+  key_COND_rpl_thread_stop, key_COND_rpl_thread_pool,
   key_COND_parallel_entry, key_COND_group_commit_orderer;
 extern PSI_cond_key key_COND_wait_gtid, key_COND_gtid_ignore_duplicates;
 
@@ -458,6 +460,9 @@ extern PSI_stage_info stage_waiting_for_prior_transaction_to_commit;
 extern PSI_stage_info stage_waiting_for_prior_transaction_to_start_commit;
 extern PSI_stage_info stage_waiting_for_room_in_worker_thread;
 extern PSI_stage_info stage_waiting_for_workers_idle;
+extern PSI_stage_info stage_waiting_for_ftwrl;
+extern PSI_stage_info stage_waiting_for_ftwrl_threads_to_pause;
+extern PSI_stage_info stage_waiting_for_rpl_thread_pool;
 extern PSI_stage_info stage_master_gtid_wait_primary;
 extern PSI_stage_info stage_master_gtid_wait;
 extern PSI_stage_info stage_gtid_wait_other_connection;
@@ -532,6 +537,7 @@ extern mysql_mutex_t
        LOCK_slave_init;
 extern MYSQL_PLUGIN_IMPORT mysql_mutex_t LOCK_thread_count;
 #ifdef HAVE_OPENSSL
+extern char* des_key_file;
 extern mysql_mutex_t LOCK_des_key_file;
 #endif
 extern mysql_mutex_t LOCK_server_started;
@@ -542,7 +548,7 @@ extern mysql_cond_t COND_thread_count;
 extern mysql_cond_t COND_manager;
 extern mysql_cond_t COND_slave_init;
 extern int32 thread_running;
-extern int32 thread_count;
+extern int32 thread_count, service_thread_count;
 extern my_atomic_rwlock_t thread_running_lock, thread_count_lock;
 extern my_atomic_rwlock_t slave_executed_entries_lock;
 
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index da18a30ac78..6b72f9c336b 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -3557,9 +3557,9 @@ bool calculate_cond_selectivity_for_table(THD *thd, TABLE *table, Item *cond)
 	    break; 
           bitmap_set_bit(&handled_columns, key_part->fieldnr-1);
         }
-        double selectivity_mult;
         if (i)
         {
+          double UNINIT_VAR(selectivity_mult);
           /* 
             There is at least 1-column prefix of columns whose selectivity has
             not yet been accounted for.
diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index 81f2dd5812a..52cec9f0a85 100644
--- a/sql/rpl_gtid.cc
+++ b/sql/rpl_gtid.cc
@@ -122,7 +122,7 @@ rpl_slave_state::check_duplicate_gtid(rpl_gtid *gtid, rpl_group_info *rgi)
   int res;
   bool did_enter_cond= false;
   PSI_stage_info old_stage;
-  THD *thd;
+  THD *UNINIT_VAR(thd);
   Relay_log_info *rli= rgi->rli;
 
   mysql_mutex_lock(&LOCK_slave_state);
@@ -243,8 +243,10 @@ rpl_slave_state_free_element(void *arg)
 
 
 rpl_slave_state::rpl_slave_state()
-  : last_sub_id(0), inited(false), loaded(false)
+  : last_sub_id(0), loaded(false)
 {
+  mysql_mutex_init(key_LOCK_slave_state, &LOCK_slave_state,
+                   MY_MUTEX_INIT_SLOW);
   my_hash_init(&hash, &my_charset_bin, 32, offsetof(element, domain_id),
                sizeof(uint32), NULL, rpl_slave_state_free_element, HASH_UNIQUE);
 }
@@ -252,15 +254,9 @@ rpl_slave_state::rpl_slave_state()
 
 rpl_slave_state::~rpl_slave_state()
 {
-}
-
-
-void
-rpl_slave_state::init()
-{
-  DBUG_ASSERT(!inited);
-  mysql_mutex_init(key_LOCK_slave_state, &LOCK_slave_state, MY_MUTEX_INIT_SLOW);
-  inited= true;
+  truncate_hash();
+  my_hash_free(&hash);
+  mysql_mutex_destroy(&LOCK_slave_state);
 }
 
 
@@ -285,16 +281,6 @@ rpl_slave_state::truncate_hash()
   my_hash_reset(&hash);
 }
 
-void
-rpl_slave_state::deinit()
-{
-  if (!inited)
-    return;
-  truncate_hash();
-  my_hash_free(&hash);
-  mysql_mutex_destroy(&LOCK_slave_state);
-}
-
 
 int
 rpl_slave_state::update(uint32 domain_id, uint32 server_id, uint64 sub_id,
@@ -2109,16 +2095,16 @@ gtid_waiting::wait_for_gtid(THD *thd, rpl_gtid *wait_gtid,
       uint64 wakeup_seq_no;
       queue_element *cur_waiter;
 
-      mysql_mutex_lock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+      mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
       /*
         The elements in the gtid_slave_state_hash are never re-allocated once
         they enter the hash, so we do not need to re-do the lookup after releasing
         and re-aquiring the lock.
       */
       if (!slave_state_elem &&
-          !(slave_state_elem= rpl_global_gtid_slave_state.get_element(domain_id)))
+          !(slave_state_elem= rpl_global_gtid_slave_state->get_element(domain_id)))
       {
-        mysql_mutex_unlock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+        mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
         remove_from_wait_queue(he, &elem);
         promote_new_waiter(he);
         if (did_enter_cond)
@@ -2135,7 +2121,7 @@ gtid_waiting::wait_for_gtid(THD *thd, rpl_gtid *wait_gtid,
           We do not have to wait. (We will be removed from the wait queue when
           we call process_wait_hash() below.
         */
-        mysql_mutex_unlock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+        mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
       }
       else if ((cur_waiter= slave_state_elem->gtid_waiter) &&
                slave_state_elem->min_wait_seq_no <= seq_no)
@@ -2147,7 +2133,7 @@ gtid_waiting::wait_for_gtid(THD *thd, rpl_gtid *wait_gtid,
           lock).
         */
         elem.do_small_wait= false;
-        mysql_mutex_unlock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+        mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
       }
       else
       {
@@ -2172,7 +2158,7 @@ gtid_waiting::wait_for_gtid(THD *thd, rpl_gtid *wait_gtid,
         else
           mysql_mutex_unlock(&LOCK_gtid_waiting);
         thd->ENTER_COND(&slave_state_elem->COND_wait_gtid,
-                        &rpl_global_gtid_slave_state.LOCK_slave_state,
+                        &rpl_global_gtid_slave_state->LOCK_slave_state,
                         &stage_master_gtid_wait_primary, &old_stage);
         do
         {
@@ -2182,7 +2168,7 @@ gtid_waiting::wait_for_gtid(THD *thd, rpl_gtid *wait_gtid,
           {
             int err=
               mysql_cond_timedwait(&slave_state_elem->COND_wait_gtid,
-                                   &rpl_global_gtid_slave_state.LOCK_slave_state,
+                                   &rpl_global_gtid_slave_state->LOCK_slave_state,
                                    wait_until);
             if (err == ETIMEDOUT || err == ETIME)
             {
@@ -2192,7 +2178,7 @@ gtid_waiting::wait_for_gtid(THD *thd, rpl_gtid *wait_gtid,
           }
           else
             mysql_cond_wait(&slave_state_elem->COND_wait_gtid,
-                            &rpl_global_gtid_slave_state.LOCK_slave_state);
+                            &rpl_global_gtid_slave_state->LOCK_slave_state);
         } while (slave_state_elem->gtid_waiter == &elem);
         wakeup_seq_no= slave_state_elem->highest_seq_no;
         /*
diff --git a/sql/rpl_gtid.h b/sql/rpl_gtid.h
index 997540728a5..d17ddf3451a 100644
--- a/sql/rpl_gtid.h
+++ b/sql/rpl_gtid.h
@@ -161,14 +161,11 @@ struct rpl_slave_state
   mysql_mutex_t LOCK_slave_state;
 
   uint64 last_sub_id;
-  bool inited;
   bool loaded;
 
   rpl_slave_state();
   ~rpl_slave_state();
 
-  void init();
-  void deinit();
   void truncate_hash();
   ulong count() const { return hash.records; }
   int update(uint32 domain_id, uint32 server_id, uint64 sub_id,
diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc
index cc5da77303c..b2e957a3e6e 100644
--- a/sql/rpl_parallel.cc
+++ b/sql/rpl_parallel.cc
@@ -44,6 +44,9 @@ rpt_handle_event(rpl_parallel_thread::queued_event *qev,
   rgi->event_relay_log_pos= qev->event_relay_log_pos;
   rgi->future_event_relay_log_pos= qev->future_event_relay_log_pos;
   strcpy(rgi->future_event_master_log_name, qev->future_event_master_log_name);
+  if (!(ev->is_artificial_event() || ev->is_relay_log_event() ||
+        (ev->when == 0)))
+    rgi->last_master_timestamp= ev->when + (time_t)ev->exec_time;
   mysql_mutex_lock(&rli->data_lock);
   /* Mutex will be released in apply_event_and_update_pos(). */
   err= apply_event_and_update_pos(ev, thd, rgi, rpt);
@@ -272,6 +275,290 @@ register_wait_for_prior_event_group_commit(rpl_group_info *rgi,
 }
 
 
+/*
+  Do not start parallel execution of this event group until all prior groups
+  have reached the commit phase that are not safe to run in parallel with.
+*/
+static bool
+do_gco_wait(rpl_group_info *rgi, group_commit_orderer *gco,
+            bool *did_enter_cond, PSI_stage_info *old_stage)
+{
+  THD *thd= rgi->thd;
+  rpl_parallel_entry *entry= rgi->parallel_entry;
+  uint64 wait_count;
+
+  mysql_mutex_assert_owner(&entry->LOCK_parallel_entry);
+
+  if (!gco->installed)
+  {
+    group_commit_orderer *prev_gco= gco->prev_gco;
+    if (prev_gco)
+    {
+      prev_gco->last_sub_id= gco->prior_sub_id;
+      prev_gco->next_gco= gco;
+    }
+    gco->installed= true;
+  }
+  wait_count= gco->wait_count;
+  if (wait_count > entry->count_committing_event_groups)
+  {
+    DEBUG_SYNC(thd, "rpl_parallel_start_waiting_for_prior");
+    thd->ENTER_COND(&gco->COND_group_commit_orderer,
+                    &entry->LOCK_parallel_entry,
+                    &stage_waiting_for_prior_transaction_to_start_commit,
+                    old_stage);
+    *did_enter_cond= true;
+    do
+    {
+      if (thd->check_killed() && !rgi->worker_error)
+      {
+        DEBUG_SYNC(thd, "rpl_parallel_start_waiting_for_prior_killed");
+        thd->clear_error();
+        thd->get_stmt_da()->reset_diagnostics_area();
+        thd->send_kill_message();
+        slave_output_error_info(rgi, thd);
+        signal_error_to_sql_driver_thread(thd, rgi, 1);
+        /*
+          Even though we were killed, we need to continue waiting for the
+          prior event groups to signal that we can continue. Otherwise we
+          mess up the accounting for ordering. However, now that we have
+          marked the error, events will just be skipped rather than
+          executed, and things will progress quickly towards stop.
+        */
+      }
+      mysql_cond_wait(&gco->COND_group_commit_orderer,
+                      &entry->LOCK_parallel_entry);
+    } while (wait_count > entry->count_committing_event_groups);
+  }
+
+  if (entry->force_abort && wait_count > entry->stop_count)
+  {
+    /*
+      We are stopping (STOP SLAVE), and this event group is beyond the point
+      where we can safely stop. So return a flag that will cause us to skip,
+      rather than execute, the following events.
+    */
+    return true;
+  }
+  else
+    return false;
+}
+
+
+static void
+do_ftwrl_wait(rpl_group_info *rgi,
+              bool *did_enter_cond, PSI_stage_info *old_stage)
+{
+  THD *thd= rgi->thd;
+  rpl_parallel_entry *entry= rgi->parallel_entry;
+  uint64 sub_id= rgi->gtid_sub_id;
+  DBUG_ENTER("do_ftwrl_wait");
+
+  mysql_mutex_assert_owner(&entry->LOCK_parallel_entry);
+
+  /*
+    If a FLUSH TABLES WITH READ LOCK (FTWRL) is pending, check if this
+    transaction is later than transactions that have priority to complete
+    before FTWRL. If so, wait here so that FTWRL can proceed and complete
+    first.
+
+    (entry->pause_sub_id is ULONGLONG_MAX if no FTWRL is pending, which makes
+    this test false as required).
+  */
+  if (unlikely(sub_id > entry->pause_sub_id))
+  {
+    thd->ENTER_COND(&entry->COND_parallel_entry, &entry->LOCK_parallel_entry,
+                    &stage_waiting_for_ftwrl, old_stage);
+    *did_enter_cond= true;
+    do
+    {
+      if (entry->force_abort || rgi->worker_error)
+        break;
+      if (thd->check_killed())
+      {
+        thd->send_kill_message();
+        slave_output_error_info(rgi, thd);
+        signal_error_to_sql_driver_thread(thd, rgi, 1);
+        break;
+      }
+      mysql_cond_wait(&entry->COND_parallel_entry, &entry->LOCK_parallel_entry);
+    } while (sub_id > entry->pause_sub_id);
+
+    /*
+      We do not call EXIT_COND() here, as this will be done later by our
+      caller (since we set *did_enter_cond to true).
+    */
+  }
+
+  if (sub_id > entry->largest_started_sub_id)
+    entry->largest_started_sub_id= sub_id;
+
+  DBUG_VOID_RETURN;
+}
+
+
+static int
+pool_mark_busy(rpl_parallel_thread_pool *pool, THD *thd)
+{
+  PSI_stage_info old_stage;
+  int res= 0;
+
+  /*
+    Wait here while the queue is busy. This is done to make FLUSH TABLES WITH
+    READ LOCK work correctly, without incuring extra locking penalties in
+    normal operation. FLUSH TABLES WITH READ LOCK needs to lock threads in the
+    thread pool, and for this we need to make sure the pool will not go away
+    during the operation. The LOCK_rpl_thread_pool is not suitable for
+    this. It is taken by release_thread() while holding LOCK_rpl_thread; so it
+    must be released before locking any LOCK_rpl_thread lock, or a deadlock
+    can occur.
+
+    So we protect the infrequent operations of FLUSH TABLES WITH READ LOCK and
+    pool size changes with this condition wait.
+  */
+  mysql_mutex_lock(&pool->LOCK_rpl_thread_pool);
+  if (thd)
+    thd->ENTER_COND(&pool->COND_rpl_thread_pool, &pool->LOCK_rpl_thread_pool,
+                    &stage_waiting_for_rpl_thread_pool, &old_stage);
+  while (pool->busy)
+  {
+    if (thd && thd->check_killed())
+    {
+      thd->send_kill_message();
+      res= 1;
+      break;
+    }
+    mysql_cond_wait(&pool->COND_rpl_thread_pool, &pool->LOCK_rpl_thread_pool);
+  }
+  if (!res)
+    pool->busy= true;
+  if (thd)
+    thd->EXIT_COND(&old_stage);
+  else
+    mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool);
+
+  return res;
+}
+
+
+static void
+pool_mark_not_busy(rpl_parallel_thread_pool *pool)
+{
+  mysql_mutex_lock(&pool->LOCK_rpl_thread_pool);
+  DBUG_ASSERT(pool->busy);
+  pool->busy= false;
+  mysql_cond_broadcast(&pool->COND_rpl_thread_pool);
+  mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool);
+}
+
+
+void
+rpl_unpause_after_ftwrl(THD *thd)
+{
+  uint32 i;
+  rpl_parallel_thread_pool *pool= &global_rpl_thread_pool;
+  DBUG_ENTER("rpl_unpause_after_ftwrl");
+
+  DBUG_ASSERT(pool->busy);
+
+  for (i= 0; i < pool->count; ++i)
+  {
+    rpl_parallel_entry *e;
+    rpl_parallel_thread *rpt= pool->threads[i];
+
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+    if (!rpt->current_owner)
+    {
+      mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+      continue;
+    }
+    e= rpt->current_entry;
+    mysql_mutex_lock(&e->LOCK_parallel_entry);
+    rpt->pause_for_ftwrl = false;
+    mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+    e->pause_sub_id= (uint64)ULONGLONG_MAX;
+    mysql_cond_broadcast(&e->COND_parallel_entry);
+    mysql_mutex_unlock(&e->LOCK_parallel_entry);
+  }
+
+  pool_mark_not_busy(pool);
+  DBUG_VOID_RETURN;
+}
+
+
+/*
+  .
+
+  Note: in case of error return, rpl_unpause_after_ftwrl() must _not_ be called.
+*/
+int
+rpl_pause_for_ftwrl(THD *thd)
+{
+  uint32 i;
+  rpl_parallel_thread_pool *pool= &global_rpl_thread_pool;
+  int err;
+  DBUG_ENTER("rpl_pause_for_ftwrl");
+
+  /*
+    While the count_pending_pause_for_ftwrl counter is non-zero, the pool
+    cannot be shutdown/resized, so threads are guaranteed to not disappear.
+
+    This is required to safely be able to access the individual threads below.
+    (We cannot lock an individual thread while holding LOCK_rpl_thread_pool,
+    as this can deadlock against release_thread()).
+  */
+  if ((err= pool_mark_busy(pool, thd)))
+    DBUG_RETURN(err);
+
+  for (i= 0; i < pool->count; ++i)
+  {
+    PSI_stage_info old_stage;
+    rpl_parallel_entry *e;
+    rpl_parallel_thread *rpt= pool->threads[i];
+
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+    if (!rpt->current_owner)
+    {
+      mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+      continue;
+    }
+    e= rpt->current_entry;
+    mysql_mutex_lock(&e->LOCK_parallel_entry);
+    /*
+      Setting the rpt->pause_for_ftwrl flag makes sure that the thread will not
+      de-allocate itself until signalled to do so by rpl_unpause_after_ftwrl().
+    */
+    rpt->pause_for_ftwrl = true;
+    mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+    ++e->need_sub_id_signal;
+    if (e->pause_sub_id == (uint64)ULONGLONG_MAX)
+      e->pause_sub_id= e->largest_started_sub_id;
+    thd->ENTER_COND(&e->COND_parallel_entry, &e->LOCK_parallel_entry,
+                    &stage_waiting_for_ftwrl_threads_to_pause, &old_stage);
+    while (e->pause_sub_id < (uint64)ULONGLONG_MAX &&
+           e->last_committed_sub_id < e->pause_sub_id &&
+           !err)
+    {
+      if (thd->check_killed())
+      {
+        thd->send_kill_message();
+        err= 1;
+        break;
+      }
+      mysql_cond_wait(&e->COND_parallel_entry, &e->LOCK_parallel_entry);
+    };
+    --e->need_sub_id_signal;
+    thd->EXIT_COND(&old_stage);
+    if (err)
+      break;
+  }
+
+  if (err)
+    rpl_unpause_after_ftwrl(thd);
+  DBUG_RETURN(err);
+}
+
+
 #ifndef DBUG_OFF
 static int
 dbug_simulate_tmp_error(rpl_group_info *rgi, THD *thd)
@@ -765,7 +1052,6 @@ handle_rpl_parallel_thread(void *arg)
       {
         bool did_enter_cond= false;
         PSI_stage_info old_stage;
-        uint64 wait_count;
 
         DBUG_EXECUTE_IF("rpl_parallel_scheduled_gtid_0_x_100", {
             if (rgi->current_gtid.domain_id == 0 &&
@@ -803,72 +1089,19 @@ handle_rpl_parallel_thread(void *arg)
         event_gtid_sub_id= rgi->gtid_sub_id;
         rgi->thd= thd;
 
+        mysql_mutex_lock(&entry->LOCK_parallel_entry);
+        skip_event_group= do_gco_wait(rgi, gco, &did_enter_cond, &old_stage);
+
+        if (unlikely(entry->stop_on_error_sub_id <= rgi->wait_commit_sub_id))
+          skip_event_group= true;
+        if (likely(!skip_event_group))
+          do_ftwrl_wait(rgi, &did_enter_cond, &old_stage);
+
         /*
           Register ourself to wait for the previous commit, if we need to do
           such registration _and_ that previous commit has not already
           occured.
-
-          Also do not start parallel execution of this event group until all
-          prior groups have reached the commit phase that are not safe to run
-          in parallel with.
         */
-        mysql_mutex_lock(&entry->LOCK_parallel_entry);
-        if (!gco->installed)
-        {
-          group_commit_orderer *prev_gco= gco->prev_gco;
-          if (prev_gco)
-          {
-            prev_gco->last_sub_id= gco->prior_sub_id;
-            prev_gco->next_gco= gco;
-          }
-          gco->installed= true;
-        }
-        wait_count= gco->wait_count;
-        if (wait_count > entry->count_committing_event_groups)
-        {
-          DEBUG_SYNC(thd, "rpl_parallel_start_waiting_for_prior");
-          thd->ENTER_COND(&gco->COND_group_commit_orderer,
-                          &entry->LOCK_parallel_entry,
-                          &stage_waiting_for_prior_transaction_to_start_commit,
-                          &old_stage);
-          did_enter_cond= true;
-          do
-          {
-            if (thd->check_killed() && !rgi->worker_error)
-            {
-              DEBUG_SYNC(thd, "rpl_parallel_start_waiting_for_prior_killed");
-              thd->clear_error();
-              thd->get_stmt_da()->reset_diagnostics_area();
-              thd->send_kill_message();
-              slave_output_error_info(rgi, thd);
-              signal_error_to_sql_driver_thread(thd, rgi, 1);
-              /*
-                Even though we were killed, we need to continue waiting for the
-                prior event groups to signal that we can continue. Otherwise we
-                mess up the accounting for ordering. However, now that we have
-                marked the error, events will just be skipped rather than
-                executed, and things will progress quickly towards stop.
-              */
-            }
-            mysql_cond_wait(&gco->COND_group_commit_orderer,
-                            &entry->LOCK_parallel_entry);
-          } while (wait_count > entry->count_committing_event_groups);
-        }
-
-        if (entry->force_abort && wait_count > entry->stop_count)
-        {
-          /*
-            We are stopping (STOP SLAVE), and this event group is beyond the
-            point where we can safely stop. So set a flag that will cause us
-            to skip, rather than execute, the following events.
-          */
-          skip_event_group= true;
-        }
-        else
-          skip_event_group= false;
-
-        if (unlikely(entry->stop_on_error_sub_id <= rgi->wait_commit_sub_id))
-          skip_event_group= true;
         register_wait_for_prior_event_group_commit(rgi, entry);
 
         unlock_or_exit_cond(thd, &entry->LOCK_parallel_entry,
@@ -879,7 +1112,7 @@ handle_rpl_parallel_thread(void *arg)
         if (opt_gtid_ignore_duplicates)
         {
           int res=
-            rpl_global_gtid_slave_state.check_duplicate_gtid(&rgi->current_gtid,
+            rpl_global_gtid_slave_state->check_duplicate_gtid(&rgi->current_gtid,
                                                              rgi);
           if (res < 0)
           {
@@ -1020,17 +1253,41 @@ handle_rpl_parallel_thread(void *arg)
     */
     rpt->batch_free();
 
-    if ((events= rpt->event_queue) != NULL)
+    for (;;)
     {
+      if ((events= rpt->event_queue) != NULL)
+      {
+        /*
+          Take next group of events from the replication pool.
+          This is faster than having to wakeup the pool manager thread to give
+          us a new event.
+        */
+        rpt->dequeue1(events);
+        mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
+        goto more_events;
+      }
+      if (!rpt->pause_for_ftwrl ||
+          (in_event_group && !group_rgi->parallel_entry->force_abort))
+        break;
       /*
-        Take next group of events from the replication pool.
-        This is faster than having to wakeup the pool manager thread to give us
-        a new event.
+        We are currently in the delicate process of pausing parallel
+        replication while FLUSH TABLES WITH READ LOCK is starting. We must
+        not de-allocate the thread (setting rpt->current_owner= NULL) until
+        rpl_unpause_after_ftwrl() has woken us up.
       */
-      rpt->dequeue1(events);
+      mysql_mutex_lock(&rpt->current_entry->LOCK_parallel_entry);
       mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
-      goto more_events;
+      if (rpt->pause_for_ftwrl)
+        mysql_cond_wait(&rpt->current_entry->COND_parallel_entry,
+                        &rpt->current_entry->LOCK_parallel_entry);
+      mysql_mutex_unlock(&rpt->current_entry->LOCK_parallel_entry);
+      mysql_mutex_lock(&rpt->LOCK_rpl_thread);
+      /*
+        Now loop to check again for more events available, since we released
+        and re-aquired the LOCK_rpl_thread mutex.
+      */
     }
+
     rpt->inuse_relaylog_refcount_update();
 
     if (in_event_group && group_rgi->parallel_entry->force_abort)
@@ -1059,7 +1316,7 @@ handle_rpl_parallel_thread(void *arg)
       /* Tell wait_for_done() that we are done, if it is waiting. */
       if (likely(rpt->current_entry) &&
           unlikely(rpt->current_entry->force_abort))
-        mysql_cond_broadcast(&rpt->current_entry->COND_parallel_entry);
+        mysql_cond_broadcast(&rpt->COND_rpl_thread_stop);
       rpt->current_entry= NULL;
       if (!rpt->stop)
         rpt->pool->release_thread(rpt);
@@ -1107,6 +1364,10 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool,
   rpl_parallel_thread **new_list= NULL;
   rpl_parallel_thread *new_free_list= NULL;
   rpl_parallel_thread *rpt_array= NULL;
+  int res;
+
+  if ((res= pool_mark_busy(pool, current_thd)))
+    return res;
 
   /*
     Allocate the new list of threads up-front.
@@ -1135,6 +1396,8 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool,
     mysql_cond_init(key_COND_rpl_thread, &new_list[i]->COND_rpl_thread, NULL);
     mysql_cond_init(key_COND_rpl_thread_queue,
                     &new_list[i]->COND_rpl_thread_queue, NULL);
+    mysql_cond_init(key_COND_rpl_thread_stop,
+                    &new_list[i]->COND_rpl_thread_stop, NULL);
     new_list[i]->pool= pool;
     if (mysql_thread_create(key_rpl_parallel_thread, &th, &connection_attrib,
                             handle_rpl_parallel_thread, new_list[i]))
@@ -1155,7 +1418,14 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool,
   */
   for (i= 0; i < pool->count; ++i)
   {
-    rpl_parallel_thread *rpt= pool->get_thread(NULL, NULL);
+    rpl_parallel_thread *rpt;
+
+    mysql_mutex_lock(&pool->LOCK_rpl_thread_pool);
+    while ((rpt= pool->free_list) == NULL)
+      mysql_cond_wait(&pool->COND_rpl_thread_pool, &pool->LOCK_rpl_thread_pool);
+    pool->free_list= rpt->next;
+    mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool);
+    mysql_mutex_lock(&rpt->LOCK_rpl_thread);
     rpt->stop= true;
     mysql_cond_signal(&rpt->COND_rpl_thread);
     mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
@@ -1205,9 +1475,7 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool,
     mysql_mutex_unlock(&pool->threads[i]->LOCK_rpl_thread);
   }
 
-  mysql_mutex_lock(&pool->LOCK_rpl_thread_pool);
-  mysql_cond_broadcast(&pool->COND_rpl_thread_pool);
-  mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool);
+  pool_mark_not_busy(pool);
 
   return 0;
 
@@ -1231,6 +1499,7 @@ err:
     }
     my_free(new_list);
   }
+  pool_mark_not_busy(pool);
   return 1;
 }
 
@@ -1494,7 +1763,7 @@ rpl_parallel_thread::loc_free_gco(group_commit_orderer *gco)
 
 
 rpl_parallel_thread_pool::rpl_parallel_thread_pool()
-  : count(0), threads(0), free_list(0), inited(false)
+  : threads(0), free_list(0), count(0), inited(false), busy(false)
 {
 }
 
@@ -1502,9 +1771,10 @@ rpl_parallel_thread_pool::rpl_parallel_thread_pool()
 int
 rpl_parallel_thread_pool::init(uint32 size)
 {
-  count= 0;
   threads= NULL;
   free_list= NULL;
+  count= 0;
+  busy= false;
 
   mysql_mutex_init(key_LOCK_rpl_thread_pool, &LOCK_rpl_thread_pool,
                    MY_MUTEX_INIT_SLOW);
@@ -1545,7 +1815,7 @@ rpl_parallel_thread_pool::get_thread(rpl_parallel_thread **owner,
   rpl_parallel_thread *rpt;
 
   mysql_mutex_lock(&LOCK_rpl_thread_pool);
-  while ((rpt= free_list) == NULL)
+  while (unlikely(busy) || !(rpt= free_list))
     mysql_cond_wait(&COND_rpl_thread_pool, &LOCK_rpl_thread_pool);
   free_list= rpt->next;
   mysql_mutex_unlock(&LOCK_rpl_thread_pool);
@@ -1756,6 +2026,7 @@ rpl_parallel::find(uint32 domain_id)
     e->rpl_thread_max= count;
     e->domain_id= domain_id;
     e->stop_on_error_sub_id= (uint64)ULONGLONG_MAX;
+    e->pause_sub_id= (uint64)ULONGLONG_MAX;
     if (my_hash_insert(&domain_hash, (uchar *)e))
     {
       my_free(e);
@@ -1837,7 +2108,7 @@ rpl_parallel::wait_for_done(THD *thd, Relay_log_info *rli)
       {
         mysql_mutex_lock(&rpt->LOCK_rpl_thread);
         while (rpt->current_owner == &e->rpl_threads[j])
-          mysql_cond_wait(&e->COND_parallel_entry, &rpt->LOCK_rpl_thread);
+          mysql_cond_wait(&rpt->COND_rpl_thread_stop, &rpt->LOCK_rpl_thread);
         mysql_mutex_unlock(&rpt->LOCK_rpl_thread);
       }
     }
@@ -1957,7 +2228,7 @@ rpl_parallel::wait_for_workers_idle(THD *thd)
 
     e= (struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);
     mysql_mutex_lock(&e->LOCK_parallel_entry);
-    e->need_sub_id_signal= true;
+    ++e->need_sub_id_signal;
     thd->ENTER_COND(&e->COND_parallel_entry, &e->LOCK_parallel_entry,
                     &stage_waiting_for_workers_idle, &old_stage);
     while (e->current_sub_id > e->last_committed_sub_id)
@@ -1970,7 +2241,7 @@ rpl_parallel::wait_for_workers_idle(THD *thd)
       }
       mysql_cond_wait(&e->COND_parallel_entry, &e->LOCK_parallel_entry);
     }
-    e->need_sub_id_signal= false;
+    --e->need_sub_id_signal;
     thd->EXIT_COND(&old_stage);
     if (err)
       return err;
diff --git a/sql/rpl_parallel.h b/sql/rpl_parallel.h
index 0c2e4270646..3012daa8763 100644
--- a/sql/rpl_parallel.h
+++ b/sql/rpl_parallel.h
@@ -70,9 +70,11 @@ struct rpl_parallel_thread {
   bool delay_start;
   bool running;
   bool stop;
+  bool pause_for_ftwrl;
   mysql_mutex_t LOCK_rpl_thread;
   mysql_cond_t COND_rpl_thread;
   mysql_cond_t COND_rpl_thread_queue;
+  mysql_cond_t COND_rpl_thread_stop;
   struct rpl_parallel_thread *next;             /* For free list. */
   struct rpl_parallel_thread_pool *pool;
   THD *thd;
@@ -199,12 +201,18 @@ struct rpl_parallel_thread {
 
 
 struct rpl_parallel_thread_pool {
-  uint32 count;
   struct rpl_parallel_thread **threads;
   struct rpl_parallel_thread *free_list;
   mysql_mutex_t LOCK_rpl_thread_pool;
   mysql_cond_t COND_rpl_thread_pool;
+  uint32 count;
   bool inited;
+  /*
+    While FTWRL runs, this counter is incremented to make SQL thread or
+    STOP/START slave not try to start new activity while that operation
+    is in progress.
+  */
+  bool busy;
 
   rpl_parallel_thread_pool();
   int init(uint32 size);
@@ -219,6 +227,12 @@ struct rpl_parallel_entry {
   mysql_mutex_t LOCK_parallel_entry;
   mysql_cond_t COND_parallel_entry;
   uint32 domain_id;
+  /*
+    Incremented by wait_for_workers_idle() and rpl_pause_for_ftwrl() to show
+    that they are waiting, so that finish_event_group knows to signal them
+    when last_committed_sub_id is increased.
+  */
+  uint32 need_sub_id_signal;
   uint64 last_commit_id;
   bool active;
   /*
@@ -228,12 +242,6 @@ struct rpl_parallel_entry {
   */
   bool force_abort;
   /*
-    Set in wait_for_workers_idle() to show that it is waiting, so that
-    finish_event_group knows to signal it when last_committed_sub_id is
-    increased.
-  */
-  bool need_sub_id_signal;
-  /*
    At STOP SLAVE (force_abort=true), we do not want to process all events in
    the queue (which could unnecessarily delay stop, if a lot of events happen
    to be queued). The stop_count provides a safe point at which to stop, so
@@ -273,6 +281,15 @@ struct rpl_parallel_entry {
     queued for execution by a worker thread.
   */
   uint64 current_sub_id;
+  /*
+    The largest sub_id that has started its transaction. Protected by
+    LOCK_parallel_entry.
+
+    (Transactions can start out-of-order, so this value signifies that no
+    transactions with larger sub_id have started, but not necessarily that all
+    transactions with smaller sub_id have started).
+  */
+  uint64 largest_started_sub_id;
   rpl_group_info *current_group_info;
   /*
     If we get an error in some event group, we set the sub_id of that event
@@ -282,6 +299,12 @@ struct rpl_parallel_entry {
     The value is ULONGLONG_MAX when no error occured.
   */
   uint64 stop_on_error_sub_id;
+  /*
+    During FLUSH TABLES WITH READ LOCK, transactions with sub_id larger than
+    this value must not start, but wait until the global read lock is released.
+    The value is set to ULONGLONG_MAX when no FTWRL is pending.
+  */
+  uint64 pause_sub_id;
   /* Total count of event groups queued so far. */
   uint64 count_queued_event_groups;
   /*
@@ -322,5 +345,7 @@ extern struct rpl_parallel_thread_pool global_rpl_thread_pool;
 extern int rpl_parallel_activate_pool(rpl_parallel_thread_pool *pool);
 extern int rpl_parallel_inactivate_pool(rpl_parallel_thread_pool *pool);
 extern bool process_gtid_for_restart_pos(Relay_log_info *rli, rpl_gtid *gtid);
+extern int rpl_pause_for_ftwrl(THD *thd);
+extern void rpl_unpause_after_ftwrl(THD *thd);
 
 #endif  /* RPL_PARALLEL_H */
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 8a2a55fcde0..8c7724d88a3 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -37,7 +37,7 @@ static int count_relay_log_space(Relay_log_info* rli);
    Current replication state (hash of last GTID executed, per replication
    domain).
 */
-rpl_slave_state rpl_global_gtid_slave_state;
+rpl_slave_state *rpl_global_gtid_slave_state;
 /* Object used for MASTER_GTID_WAIT(). */
 gtid_waiting rpl_global_gtid_waiting;
 
@@ -1001,6 +1001,18 @@ void Relay_log_info::inc_group_relay_log_pos(ulonglong log_pos,
       else if (group_master_log_pos < log_pos)
         group_master_log_pos= log_pos;
     }
+
+    /*
+      In the parallel case, we only update the Seconds_Behind_Master at the
+      end of a transaction. In the non-parallel case, the value is updated as
+      soon as an event is read from the relay log; however this would be too
+      confusing for the user, seeing the slave reported as up-to-date when
+      potentially thousands of events are still queued up for worker threads
+      waiting for execution.
+    */
+    if (rgi->last_master_timestamp &&
+        rgi->last_master_timestamp > last_master_timestamp)
+      last_master_timestamp= rgi->last_master_timestamp;
   }
   else
   {
@@ -1328,7 +1340,7 @@ void Relay_log_info::stmt_done(my_off_t event_master_log_pos, THD *thd,
   else
   {
     inc_group_relay_log_pos(event_master_log_pos, rgi);
-    if (rpl_global_gtid_slave_state.record_and_update_gtid(thd, rgi))
+    if (rpl_global_gtid_slave_state->record_and_update_gtid(thd, rgi))
     {
       report(WARNING_LEVEL, ER_CANNOT_UPDATE_GTID_STATE, rgi->gtid_info(),
              "Failed to update GTID state in %s.%s, slave state may become "
@@ -1454,9 +1466,9 @@ rpl_load_gtid_slave_state(THD *thd)
   uint32 i;
   DBUG_ENTER("rpl_load_gtid_slave_state");
 
-  mysql_mutex_lock(&rpl_global_gtid_slave_state.LOCK_slave_state);
-  bool loaded= rpl_global_gtid_slave_state.loaded;
-  mysql_mutex_unlock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+  mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+  bool loaded= rpl_global_gtid_slave_state->loaded;
+  mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
   if (loaded)
     DBUG_RETURN(0);
 
@@ -1556,23 +1568,23 @@ rpl_load_gtid_slave_state(THD *thd)
     }
   }
 
-  mysql_mutex_lock(&rpl_global_gtid_slave_state.LOCK_slave_state);
-  if (rpl_global_gtid_slave_state.loaded)
+  mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+  if (rpl_global_gtid_slave_state->loaded)
   {
-    mysql_mutex_unlock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+    mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
     goto end;
   }
 
   for (i= 0; i < array.elements; ++i)
   {
     get_dynamic(&array, (uchar *)&tmp_entry, i);
-    if ((err= rpl_global_gtid_slave_state.update(tmp_entry.gtid.domain_id,
+    if ((err= rpl_global_gtid_slave_state->update(tmp_entry.gtid.domain_id,
                                                  tmp_entry.gtid.server_id,
                                                  tmp_entry.sub_id,
                                                  tmp_entry.gtid.seq_no,
                                                  NULL)))
     {
-      mysql_mutex_unlock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+      mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
       my_error(ER_OUT_OF_RESOURCES, MYF(0));
       goto end;
     }
@@ -1585,14 +1597,14 @@ rpl_load_gtid_slave_state(THD *thd)
         mysql_bin_log.bump_seq_no_counter_if_needed(entry->gtid.domain_id,
                                                     entry->gtid.seq_no))
     {
-      mysql_mutex_unlock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+      mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
       my_error(ER_OUT_OF_RESOURCES, MYF(0));
       goto end;
     }
   }
 
-  rpl_global_gtid_slave_state.loaded= true;
-  mysql_mutex_unlock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+  rpl_global_gtid_slave_state->loaded= true;
+  mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
 
   err= 0;                                       /* Clear HA_ERR_END_OF_FILE */
 
@@ -1630,6 +1642,7 @@ rpl_group_info::reinit(Relay_log_info *rli)
   row_stmt_start_timestamp= 0;
   long_find_row_note_printed= false;
   did_mark_start_commit= false;
+  last_master_timestamp = 0;
   gtid_ignore_duplicate_state= GTID_DUPLICATE_NULL;
   commit_orderer.reinit();
 }
@@ -1659,7 +1672,7 @@ rpl_group_info::~rpl_group_info()
 int
 event_group_new_gtid(rpl_group_info *rgi, Gtid_log_event *gev)
 {
-  uint64 sub_id= rpl_global_gtid_slave_state.next_sub_id(gev->domain_id);
+  uint64 sub_id= rpl_global_gtid_slave_state->next_sub_id(gev->domain_id);
   if (!sub_id)
   {
     /* Out of memory caused hash insertion to fail. */
@@ -1774,7 +1787,7 @@ void rpl_group_info::cleanup_context(THD *thd, bool error)
       --gtid-ignore-duplicates.
     */
     if (gtid_ignore_duplicate_state != GTID_DUPLICATE_NULL)
-      rpl_global_gtid_slave_state.release_domain_owner(this);
+      rpl_global_gtid_slave_state->release_domain_owner(this);
   }
 
   /*
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index 2d92f384ef3..efcec83b880 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -669,6 +669,13 @@ struct rpl_group_info
   char gtid_info_buf[5+10+1+10+1+20+1];
 
   /*
+    The timestamp, from the master, of the commit event.
+    Used to do delayed update of rli->last_master_timestamp, for getting
+    reasonable values out of Seconds_Behind_Master in SHOW SLAVE STATUS.
+  */
+  time_t last_master_timestamp;
+
+  /*
     Information to be able to re-try an event group in case of a deadlock or
     other temporary error.
   */
@@ -708,7 +715,7 @@ struct rpl_group_info
   /**
     Save pointer to Annotate_rows event and switch on the
     binlog_annotate_row_events for this sql thread.
-    To be called when sql thread recieves an Annotate_rows event.
+    To be called when sql thread receives an Annotate_rows event.
   */
   inline void set_annotate_event(Annotate_rows_log_event *event)
   {
@@ -836,7 +843,7 @@ public:
 int init_relay_log_info(Relay_log_info* rli, const char* info_fname);
 
 
-extern struct rpl_slave_state rpl_global_gtid_slave_state;
+extern struct rpl_slave_state *rpl_global_gtid_slave_state;
 extern gtid_waiting rpl_global_gtid_waiting;
 
 int rpl_load_gtid_slave_state(THD *thd);
diff --git a/sql/scheduler.cc b/sql/scheduler.cc
index a9b253e478a..bc3166210b5 100644
--- a/sql/scheduler.cc
+++ b/sql/scheduler.cc
@@ -135,7 +135,7 @@ void one_thread_per_connection_scheduler(scheduler_functions *func,
 #endif
 
 /*
-  Initailize scheduler for --thread-handling=no-threads
+  Initialize scheduler for --thread-handling=no-threads
 */
 
 void one_thread_scheduler(scheduler_functions *func)
diff --git a/sql/slave.cc b/sql/slave.cc
index e5bd92bf4fb..a385be1831e 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -297,6 +297,7 @@ handle_slave_init(void *arg __attribute__((unused)))
   thd->thread_id= thread_id++;
   mysql_mutex_unlock(&LOCK_thread_count);
   thd->system_thread = SYSTEM_THREAD_SLAVE_INIT;
+  thread_safe_increment32(&service_thread_count, &thread_count_lock);
   thd->store_globals();
   thd->security_ctx->skip_grants();
   thd->set_command(COM_DAEMON);
@@ -312,6 +313,8 @@ handle_slave_init(void *arg __attribute__((unused)))
   mysql_mutex_lock(&LOCK_thread_count);
   delete thd;
   mysql_mutex_unlock(&LOCK_thread_count);
+  thread_safe_decrement32(&service_thread_count, &thread_count_lock);
+  signal_thd_deleted();
   my_thread_end();
 
   mysql_mutex_lock(&LOCK_slave_init);
@@ -2184,8 +2187,8 @@ after_set_capability:
         (master_row= mysql_fetch_row(master_res)) &&
         (master_row[0] != NULL))
     {
-      rpl_global_gtid_slave_state.load(mi->io_thd, master_row[0],
-                                       strlen(master_row[0]), false, false);
+      rpl_global_gtid_slave_state->load(mi->io_thd, master_row[0],
+                                        strlen(master_row[0]), false, false);
     }
     else if (check_io_slave_killed(mi, NULL))
       goto slave_killed_err;
@@ -2495,7 +2498,7 @@ bool show_master_info(THD *thd, Master_info *mi, bool full)
   DBUG_ENTER("show_master_info");
   String gtid_pos;
 
-  if (full && rpl_global_gtid_slave_state.tostring(&gtid_pos, NULL, 0))
+  if (full && rpl_global_gtid_slave_state->tostring(&gtid_pos, NULL, 0))
     DBUG_RETURN(TRUE);
   if (send_show_master_info_header(thd, full, gtid_pos.length()))
     DBUG_RETURN(TRUE);
@@ -2955,6 +2958,11 @@ static int init_slave_thread(THD* thd, Master_info *mi,
                   simulate_error|= (1 << SLAVE_THD_IO););
   DBUG_EXECUTE_IF("simulate_sql_slave_error_on_init",
                   simulate_error|= (1 << SLAVE_THD_SQL););
+
+  thd->system_thread = (thd_type == SLAVE_THD_SQL) ?
+    SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO;
+  thread_safe_increment32(&service_thread_count, &thread_count_lock);
+
   /* We must call store_globals() before doing my_net_init() */
   if (init_thr_lock() || thd->store_globals() ||
       my_net_init(&thd->net, 0, MYF(MY_THREAD_SPECIFIC)) ||
@@ -2964,8 +2972,6 @@ static int init_slave_thread(THD* thd, Master_info *mi,
     DBUG_RETURN(-1);
   }
 
-  thd->system_thread = (thd_type == SLAVE_THD_SQL) ?
-    SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO;
   thd->security_ctx->skip_grants();
   thd->slave_thread= 1;
   thd->connection_name= mi->connection_name;
@@ -3513,8 +3519,13 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
       If it is an artificial event, or a relay log event (IO thread generated
       event) or ev->when is set to 0, we don't update the
       last_master_timestamp.
+
+      In parallel replication, we might queue a large number of events, and
+      the user might be surprised to see a claim that the slave is up to date
+      long before those queued events are actually executed.
      */
-    if (!(ev->is_artificial_event() || ev->is_relay_log_event() || (ev->when == 0)))
+    if (opt_slave_parallel_threads == 0 &&
+        !(ev->is_artificial_event() || ev->is_relay_log_event() || (ev->when == 0)))
     {
       rli->last_master_timestamp= ev->when + (time_t) ev->exec_time;
       DBUG_ASSERT(rli->last_master_timestamp >= 0);
@@ -3595,7 +3606,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
 
       if (opt_gtid_ignore_duplicates)
       {
-        int res= rpl_global_gtid_slave_state.check_duplicate_gtid
+        int res= rpl_global_gtid_slave_state->check_duplicate_gtid
           (&serial_rgi->current_gtid, serial_rgi);
         if (res < 0)
         {
@@ -4257,11 +4268,14 @@ err_during_init:
   mi->rli.relay_log.description_event_for_queue= 0;
   // TODO: make rpl_status part of Master_info
   change_rpl_status(RPL_ACTIVE_SLAVE,RPL_IDLE_SLAVE);
+
   mysql_mutex_lock(&LOCK_thread_count);
   thd->unlink();
   mysql_mutex_unlock(&LOCK_thread_count);
-  THD_CHECK_SENTRY(thd);
   delete thd;
+  thread_safe_decrement32(&service_thread_count, &thread_count_lock);
+  signal_thd_deleted();
+
   mi->abort_slave= 0;
   mi->slave_running= MYSQL_SLAVE_NOT_RUN;
   mi->io_thd= 0;
@@ -4569,7 +4583,7 @@ pthread_handler_t handle_slave_sql(void *arg)
       It will then be updated as required by GTID and GTID_LIST events found
       while applying events read from relay logs.
     */
-    rli->relay_log_state.load(&rpl_global_gtid_slave_state);
+    rli->relay_log_state.load(rpl_global_gtid_slave_state);
   }
   rli->gtid_skip_flag = GTID_SKIP_NOT;
   if (init_relay_log_pos(rli,
@@ -4822,9 +4836,9 @@ log '%s' at position %s, relay log '%s' position: %s%s", RPL_LOG_NAME,
         To handle this when we restart the SQL thread, mark the current
         per-domain position in the Relay_log_info.
       */
-      mysql_mutex_lock(&rpl_global_gtid_slave_state.LOCK_slave_state);
-      domain_count= rpl_global_gtid_slave_state.count();
-      mysql_mutex_unlock(&rpl_global_gtid_slave_state.LOCK_slave_state);
+      mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
+      domain_count= rpl_global_gtid_slave_state->count();
+      mysql_mutex_unlock(&rpl_global_gtid_slave_state->LOCK_slave_state);
       if (domain_count > 1)
       {
         inuse_relaylog *ir;
@@ -4835,7 +4849,7 @@ log '%s' at position %s, relay log '%s' position: %s%s", RPL_LOG_NAME,
           the relay log back to a known safe place to start (prior to any not
           yet applied transaction in any domain).
         */
-        rli->restart_gtid_pos.load(&rpl_global_gtid_slave_state, NULL, 0);
+        rli->restart_gtid_pos.load(rpl_global_gtid_slave_state, NULL, 0);
         if ((ir= rli->inuse_relaylog_list))
         {
           rpl_gtid *gtid= ir->relay_log_state;
@@ -4932,9 +4946,10 @@ err_during_init:
   mysql_mutex_unlock(&LOCK_active_mi);
 
   mysql_mutex_lock(&LOCK_thread_count);
-  THD_CHECK_SENTRY(thd);
   delete thd;
   mysql_mutex_unlock(&LOCK_thread_count);
+  thread_safe_decrement32(&service_thread_count, &thread_count_lock);
+  signal_thd_deleted();
 
   DBUG_LEAVE;                                   // Must match DBUG_ENTER()
   my_thread_end();
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index bab4659105c..4bef15cef3f 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -10099,6 +10099,8 @@ acl_check_proxy_grant_access(THD *thd, const char *host, const char *user,
     DBUG_RETURN(FALSE);
   }
 
+  mysql_mutex_lock(&acl_cache->lock);
+
   /* check for matching WITH PROXY rights */
   for (uint i=0; i < acl_proxy_users.elements; i++)
   {
@@ -10111,10 +10113,12 @@ acl_check_proxy_grant_access(THD *thd, const char *host, const char *user,
         proxy->get_with_grant())
     {
       DBUG_PRINT("info", ("found"));
+      mysql_mutex_unlock(&acl_cache->lock);
       DBUG_RETURN(FALSE);
     }
   }
 
+  mysql_mutex_unlock(&acl_cache->lock);
   my_error(ER_ACCESS_DENIED_NO_PASSWORD_ERROR, MYF(0),
            thd->security_ctx->user,
            thd->security_ctx->host_or_ip);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 71ef74e1013..434830f1cf0 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -6467,6 +6467,7 @@ find_field_in_table_ref(THD *thd, TABLE_LIST *table_list,
       */
       table_name && table_name[0] &&
       (my_strcasecmp(table_alias_charset, table_list->alias, table_name) ||
+       (db_name && db_name[0] && (!table_list->db || !table_list->db[0])) ||
        (db_name && db_name[0] && table_list->db && table_list->db[0] &&
         (table_list->schema_table ?
          my_strcasecmp(system_charset_info, db_name, table_list->db) :
@@ -6940,7 +6941,10 @@ find_item_in_list(Item *find, List<Item> &items, uint *counter,
 
   for (uint i= 0; (item=li++); i++)
   {
-    if (field_name && item->real_item()->type() == Item::FIELD_ITEM)
+    if (field_name &&
+        (item->real_item()->type() == Item::FIELD_ITEM ||
+         ((item->type() == Item::REF_ITEM) &&
+          (((Item_ref *)item)->ref_type() == Item_ref::VIEW_REF))))
     {
       Item_ident *item_field= (Item_ident*) item;
 
@@ -7066,35 +7070,6 @@ find_item_in_list(Item *find, List<Item> &items, uint *counter,
         break;
       }
     }
-    else if (table_name && item->type() == Item::REF_ITEM &&
-             ((Item_ref *)item)->ref_type() == Item_ref::VIEW_REF)
-    {
-      /*
-        TODO:Here we process prefixed view references only. What we should 
-        really do is process all types of Item_refs. But this will currently 
-        lead to a clash with the way references to outer SELECTs (from the 
-        HAVING clause) are handled in e.g. :
-        SELECT 1 FROM t1 AS t1_o GROUP BY a
-          HAVING (SELECT t1_o.a FROM t1 AS t1_i GROUP BY t1_i.a LIMIT 1).
-        Processing all Item_refs here will cause t1_o.a to resolve to itself.
-        We still need to process the special case of Item_direct_view_ref 
-        because in the context of views they have the same meaning as 
-        Item_field for tables.
-      */
-      Item_ident *item_ref= (Item_ident *) item;
-      if (field_name && item_ref->name && item_ref->table_name &&
-          !my_strcasecmp(system_charset_info, item_ref->name, field_name) &&
-          !my_strcasecmp(table_alias_charset, item_ref->table_name,
-                         table_name) &&
-          (!db_name || (item_ref->db_name && 
-                        !strcmp (item_ref->db_name, db_name))))
-      {
-        found= li.ref();
-        *counter= i;
-        *resolution= RESOLVED_IGNORING_ALIAS;
-        break;
-      }
-    }
   }
   if (!found)
   {
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 83899ecee21..431edc3b38c 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -27,6 +27,7 @@
 #include "log.h"
 #include "rpl_tblmap.h"
 #include "mdl.h"
+#include "field.h"                              // Create_field
 #include "probes_mysql.h"
 #include "sql_locale.h"                         /* my_locale_st */
 #include "sql_profile.h"                   /* PROFILING */
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 4e033f21b25..ef4c4421df0 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2006,7 +2006,7 @@ public:
   */
   MDL_request grl_protection;
 
-  Delayed_insert()
+  Delayed_insert(SELECT_LEX *current_select)
     :locks_in_memory(0), table(0),tables_in_use(0),stacked_inserts(0),
      status(0), handler_thread_initialized(FALSE), group_count(0)
   {
@@ -2016,7 +2016,7 @@ public:
     strmake_buf(thd.security_ctx->priv_user, thd.security_ctx->user);
     thd.current_tablenr=0;
     thd.set_command(COM_DELAYED_INSERT);
-    thd.lex->current_select= 0; 		// for my_message_sql
+    thd.lex->current_select= current_select;
     thd.lex->sql_command= SQLCOM_INSERT;        // For innodb::store_lock()
     /*
       Prevent changes to global.lock_wait_timeout from affecting
@@ -2193,7 +2193,7 @@ bool delayed_get_table(THD *thd, MDL_request *grl_protection_request,
     */
     if (! (di= find_handler(thd, table_list)))
     {
-      if (!(di= new Delayed_insert()))
+      if (!(di= new Delayed_insert(thd->lex->current_select)))
         goto end_create;
 
       thread_safe_increment32(&thread_count, &thread_count_lock);
@@ -2836,6 +2836,16 @@ pthread_handler_t handle_delayed_insert(void *arg)
     if (di->open_and_lock_table())
       goto err;
 
+    /*
+      INSERT DELAYED generally expects thd->lex->current_select to be NULL,
+      since this is not an attribute of the current thread. This can lead to
+      problems if the thread that spawned the current one disconnects.
+      current_select will then point to freed memory. But current_select is
+      required to resolve the partition function. So, after fulfilling that
+      requirement, we set the current_select to 0.
+    */
+    thd->lex->current_select= NULL;
+
     /* Tell client that the thread is initialized */
     mysql_cond_signal(&di->cond_client);
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 03367acab08..802721901a8 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -4686,6 +4686,20 @@ end_with_restore_list:
     }
 #endif /* WITH_WSREP*/
 
+#ifdef HAVE_REPLICATION
+    if (lex->type & REFRESH_READ_LOCK)
+    {
+      /*
+        We need to pause any parallel replication slave workers during FLUSH
+        TABLES WITH READ LOCK. Otherwise we might cause a deadlock, as
+        worker threads eun run in arbitrary order but need to commit in a
+        specific given order.
+      */
+      if (rpl_pause_for_ftwrl(thd))
+        goto error;
+    }
+#endif
+
     /*
       reload_acl_and_cache() will tell us if we are allowed to write to the
       binlog or not.
@@ -4735,6 +4749,10 @@ end_with_restore_list:
       if (!res)
         my_ok(thd);
     } 
+#ifdef HAVE_REPLICATION
+    if (lex->type & REFRESH_READ_LOCK)
+      rpl_unpause_after_ftwrl(thd);
+#endif
     
     break;
   }
@@ -6595,6 +6613,7 @@ bool check_fk_parent_table_access(THD *thd,
         table_name.str= (char *) thd->memdup(fk_key->ref_table.str,
                                              fk_key->ref_table.length+1);
         table_name.length= my_casedn_str(files_charset_info, table_name.str);
+        db_name.length = my_casedn_str(files_charset_info, db_name.str);
       }
 
       parent_table.init_one_table(db_name.str, db_name.length,
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 4b13139014d..3df9ebcd6bc 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1,5 +1,5 @@
-/* Copyright (c) 2002, 2013, Oracle and/or its affiliates.
-   Copyright (c) 2008, 2013, Monty Program Ab
+/* Copyright (c) 2002, 2015, Oracle and/or its affiliates.
+   Copyright (c) 2008, 2015, 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
@@ -1420,7 +1420,8 @@ static int mysql_test_update(Prepared_statement *stmt,
     (SELECT_ACL & ~table_list->table->grant.privilege);
   table_list->register_want_access(SELECT_ACL);
 #endif
-  if (setup_fields(thd, 0, stmt->lex->value_list, MARK_COLUMNS_NONE, 0, 0))
+  if (setup_fields(thd, 0, stmt->lex->value_list, MARK_COLUMNS_NONE, 0, 0) ||
+      check_unique_table(thd, table_list))
     goto error;
   /* TODO: here we should send types of placeholders to the client. */
   DBUG_RETURN(0);
@@ -3842,8 +3843,8 @@ Prepared_statement::swap_prepared_statement(Prepared_statement *copy)
   swap_variables(LEX_STRING, name, copy->name);
   /* Ditto */
   swap_variables(char *, db, copy->db);
+  swap_variables(size_t, db_length, copy->db_length);
 
-  DBUG_ASSERT(db_length == copy->db_length);
   DBUG_ASSERT(param_count == copy->param_count);
   DBUG_ASSERT(thd == copy->thd);
   last_error[0]= '\0';
diff --git a/sql/sql_reload.cc b/sql/sql_reload.cc
index ca99a81b455..63809fae7ee 100644
--- a/sql/sql_reload.cc
+++ b/sql/sql_reload.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2010, 2011, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2010, 2015, Oracle and/or its affiliates. All rights reserved.
 
    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
@@ -27,7 +27,7 @@
 #include "sql_repl.h"    // reset_master, reset_slave
 #include "rpl_mi.h"      // Master_info::data_lock
 #include "debug_sync.h"
-#include "rpl_mi.h"
+#include "des_key_file.h"
 
 static void disable_checkpoints(THD *thd);
 
@@ -345,7 +345,7 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
     }
   }
 #endif
-#ifdef OPENSSL
+#ifdef HAVE_OPENSSL
    if (options & REFRESH_DES_KEY_FILE)
    {
      if (des_key_file && load_des_key_file(des_key_file))
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 6496e1895fb..e0fd7005cd5 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -274,8 +274,9 @@ do_rename(THD *thd, TABLE_LIST *ren_table, char *new_db, char *new_table_name,
         LEX_STRING table_name= { ren_table->table_name,
                                  ren_table->table_name_length };
         LEX_STRING new_table= { (char *) new_alias, strlen(new_alias) };
+        LEX_STRING new_db_name= { (char*)new_db, strlen(new_db)};
         (void) rename_table_in_stat_tables(thd, &db_name, &table_name,
-                                           &db_name, &new_table);
+                                           &new_db_name, &new_table);
         if ((rc= Table_triggers_list::change_table_name(thd, ren_table->db,
                                                         old_alias,
                                                         ren_table->table_name,
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index af02c1ca966..b7f1528e42c 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -986,8 +986,8 @@ check_slave_start_position(binlog_send_info *info, const char **errormsg,
     rpl_gtid master_replication_gtid;
     rpl_gtid start_gtid;
     bool start_at_own_slave_pos=
-      rpl_global_gtid_slave_state.domain_to_gtid(slave_gtid->domain_id,
-                                                 &master_replication_gtid) &&
+      rpl_global_gtid_slave_state->domain_to_gtid(slave_gtid->domain_id,
+                                                  &master_replication_gtid) &&
       slave_gtid->server_id == master_replication_gtid.server_id &&
       slave_gtid->seq_no == master_replication_gtid.seq_no;
 
@@ -4041,14 +4041,14 @@ int log_loaded_block(IO_CACHE* file)
 void
 rpl_init_gtid_slave_state()
 {
-  rpl_global_gtid_slave_state.init();
+  rpl_global_gtid_slave_state= new rpl_slave_state;
 }
 
 
 void
 rpl_deinit_gtid_slave_state()
 {
-  rpl_global_gtid_slave_state.deinit();
+  delete rpl_global_gtid_slave_state;
 }
 
 
@@ -4084,7 +4084,7 @@ rpl_append_gtid_state(String *dest, bool use_binlog)
       (err= mysql_bin_log.get_most_recent_gtid_list(&gtid_list, &num_gtids)))
     return err;
 
-  err= rpl_global_gtid_slave_state.tostring(dest, gtid_list, num_gtids);
+  err= rpl_global_gtid_slave_state->tostring(dest, gtid_list, num_gtids);
   my_free(gtid_list);
 
   return err;
@@ -4109,7 +4109,7 @@ rpl_load_gtid_state(slave_connection_state *state, bool use_binlog)
       (err= mysql_bin_log.get_most_recent_gtid_list(&gtid_list, &num_gtids)))
     return err;
 
-  err= state->load(&rpl_global_gtid_slave_state, gtid_list, num_gtids);
+  err= state->load(rpl_global_gtid_slave_state, gtid_list, num_gtids);
   my_free(gtid_list);
 
   return err;
@@ -4206,7 +4206,7 @@ rpl_gtid_pos_check(THD *thd, char *str, size_t len)
 bool
 rpl_gtid_pos_update(THD *thd, char *str, size_t len)
 {
-  if (rpl_global_gtid_slave_state.load(thd, str, len, true, true))
+  if (rpl_global_gtid_slave_state->load(thd, str, len, true, true))
   {
     my_error(ER_FAILED_GTID_STATE_INIT, MYF(0));
     return true;
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index edfd9f5ebd3..18259a27bc7 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -481,6 +481,7 @@ fix_inner_refs(THD *thd, List<Item> &all_fields, SELECT_LEX *select,
     if (ref_pointer_array && !ref->found_in_select_list)
     {
       int el= all_fields.elements;
+      DBUG_ASSERT(all_fields.elements <= select->ref_pointer_array_size);
       ref_pointer_array[el]= item;
       /* Add the field item to the select list of the current select. */
       all_fields.push_front(item);
@@ -895,6 +896,7 @@ JOIN::prepare(Item ***rref_pointer_array,
       {
         Item_field *field= new Item_field(thd, *(Item_field**)ord->item);
         int el= all_fields.elements;
+        DBUG_ASSERT(all_fields.elements <= select_lex->ref_pointer_array_size);
         ref_pointer_array[el]= field;
         all_fields.push_front(field);
         ord->item= ref_pointer_array + el;
@@ -3247,11 +3249,6 @@ mysql_select(THD *thd, Item ***rref_pointer_array,
     {
       if (select_lex->linkage != GLOBAL_OPTIONS_TYPE)
       {
-	//here is EXPLAIN of subselect or derived table
-	if (join->change_result(result))
-	{
-	  DBUG_RETURN(TRUE);
-	}
         /*
           Original join tabs might be overwritten at first
           subselect execution. So we need to restore them.
@@ -7307,7 +7304,7 @@ double table_multi_eq_cond_selectivity(JOIN *join, uint idx, JOIN_TAB *s,
       {
         double curr_eq_fld_sel;
         fld= fi.get_curr_field();
-        if (!fld->table->map & ~(table_bit | rem_tables))
+        if (!(fld->table->map & ~(table_bit | rem_tables)))
           continue;
         curr_eq_fld_sel= get_column_avg_frequency(fld) /
                          fld->table->stat_records();
@@ -15804,8 +15801,8 @@ create_tmp_table(THD *thd, TMP_TABLE_PARAM *param, List<Item> &fields,
   uint  temp_pool_slot=MY_BIT_NONE;
   uint fieldnr= 0;
   ulong reclength, string_total_length;
-  bool  using_unique_constraint= 0;
-  bool  use_packed_rows= 0;
+  bool  using_unique_constraint= false;
+  bool  use_packed_rows= false;
   bool  not_all_columns= !(select_options & TMP_TABLE_ALL_COLUMNS);
   char  *tmpname,path[FN_REFLEN];
   uchar	*pos, *group_buff, *bitmaps;
@@ -15878,10 +15875,10 @@ create_tmp_table(THD *thd, TMP_TABLE_PARAM *param, List<Item> &fields,
       */
       (*tmp->item)->marker=4;			// Store null in key
       if ((*tmp->item)->too_big_for_varchar())
-	using_unique_constraint=1;
+	using_unique_constraint= true;
     }
     if (param->group_length >= MAX_BLOB_WIDTH)
-      using_unique_constraint=1;
+      using_unique_constraint= true;
     if (group)
       distinct=0;				// Can't use distinct
   }
@@ -16134,12 +16131,14 @@ create_tmp_table(THD *thd, TMP_TABLE_PARAM *param, List<Item> &fields,
         *blob_field++= fieldnr;
 	blob_count++;
       }
+
       if (new_field->real_type() == MYSQL_TYPE_STRING ||
           new_field->real_type() == MYSQL_TYPE_VARCHAR)
       {
         string_count++;
         string_total_length+= new_field->pack_length();
       }
+
       if (item->marker == 4 && item->maybe_null)
       {
 	group_null_items++;
@@ -16192,7 +16191,7 @@ create_tmp_table(THD *thd, TMP_TABLE_PARAM *param, List<Item> &fields,
     if (group &&
 	(param->group_parts > table->file->max_key_parts() ||
 	 param->group_length > table->file->max_key_length()))
-      using_unique_constraint=1;
+      using_unique_constraint= true;
   }
   else
   {
@@ -17105,7 +17104,10 @@ bool create_internal_tmp_table(TABLE *table, KEY *keyinfo,
 		       start_recinfo,
 		       share->uniques, &uniquedef,
 		       &create_info,
-		       HA_CREATE_TMP_TABLE | HA_CREATE_INTERNAL_TABLE)))
+		       HA_CREATE_TMP_TABLE | HA_CREATE_INTERNAL_TABLE |
+                       ((share->db_create_options & HA_OPTION_PACK_RECORD) ?
+                        HA_PACK_RECORD : 0)
+                      )))
   {
     table->file->print_error(error,MYF(0));	/* purecov: inspected */
     table->db_stat=0;
@@ -20011,6 +20013,7 @@ uint find_shortest_key(TABLE *table, const key_map *usable_keys)
           min_cost= cost;
           best=nr;
         }
+        DBUG_ASSERT(best < MAX_KEY);
       }
     }
   }
@@ -21443,6 +21446,8 @@ find_order_in_list(THD *thd, Item **ref_pointer_array, TABLE_LIST *tables,
     return TRUE; /* Wrong field. */
 
   uint el= all_fields.elements;
+  DBUG_ASSERT(all_fields.elements <=
+              thd->lex->current_select->ref_pointer_array_size);
   all_fields.push_front(order_item); /* Add new field to field list. */
   ref_pointer_array[el]= order_item;
   /*
@@ -21702,6 +21707,8 @@ create_distinct_group(THD *thd, Item **ref_pointer_array,
         */
         Item_field *new_item= new Item_field(thd, (Item_field*)item);
         int el= all_fields.elements;
+        DBUG_ASSERT(all_fields.elements <=
+                    thd->lex->current_select->ref_pointer_array_size);
         orig_ref_pointer_array[el]= new_item;
         all_fields.push_front(new_item);
         ord->item= orig_ref_pointer_array + el;
@@ -23473,7 +23480,7 @@ int JOIN::save_explain_data_intern(Explain_query *output, bool need_tmp_table,
                                    bool need_order, bool distinct, 
                                    const char *message)
 {
-  Explain_node *explain_node;
+  Explain_node *UNINIT_VAR(explain_node);
   JOIN *join= this; /* Legacy: this code used to be a non-member function */
   THD *thd=join->thd;
   const CHARSET_INFO *cs= system_charset_info;
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index d1a24345443..66c2682e9a3 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1720,7 +1720,7 @@ int show_create_table(THD *thd, TABLE_LIST *table_list, String *packet,
 	For string types dump collation name only if
 	collation is not primary for the given charset
       */
-      if (!(field->charset()->state & MY_CS_PRIMARY))
+      if (!(field->charset()->state & MY_CS_PRIMARY) && !field->vcol_info)
       {
 	packet->append(STRING_WITH_LEN(" COLLATE "));
 	packet->append(field->charset()->name);
@@ -7790,11 +7790,12 @@ TABLE *create_schema_table(THD *thd, TABLE_LIST *table_list)
   tmp_table_param->field_count= field_count;
   tmp_table_param->schema_table= 1;
   SELECT_LEX *select_lex= thd->lex->current_select;
+  bool keep_row_order= sql_command_flags[thd->lex->sql_command] & CF_STATUS_COMMAND;
   if (!(table= create_tmp_table(thd, tmp_table_param,
                                 field_list, (ORDER*) 0, 0, 0, 
                                 (select_lex->options | thd->variables.option_bits |
-                                 TMP_TABLE_ALL_COLUMNS),
-                                HA_POS_ERROR, table_list->alias)))
+                                 TMP_TABLE_ALL_COLUMNS), HA_POS_ERROR,
+                                table_list->alias, false, keep_row_order)))
     DBUG_RETURN(0);
   my_bitmap_map* bitmaps=
     (my_bitmap_map*) thd->alloc(bitmap_buffer_size(field_count));
diff --git a/sql/sql_statistics.cc b/sql/sql_statistics.cc
index 4ce1f3ec22a..e86c84040b4 100644
--- a/sql/sql_statistics.cc
+++ b/sql/sql_statistics.cc
@@ -592,6 +592,8 @@ public:
     stat_file->extra(HA_EXTRA_FLUSH);
     return FALSE;
   } 
+
+  friend class Stat_table_write_iter;
 };
 
 
@@ -888,7 +890,7 @@ public:
 
     @note
     A value from the field min_value/max_value is always converted
-    into a utf8 string. If the length of the column 'min_value'/'max_value'
+    into a varbinary string. If the length of the column 'min_value'/'max_value'
     is less than the length of the string the string is trimmed to fit the
     length of the column. 
   */    
@@ -896,7 +898,7 @@ public:
   void store_stat_fields()
   {
     char buff[MAX_FIELD_WIDTH];
-    String val(buff, sizeof(buff), &my_charset_utf8_bin);
+    String val(buff, sizeof(buff), &my_charset_bin);
 
     for (uint i= COLUMN_STAT_MIN_VALUE; i <= COLUMN_STAT_HISTOGRAM; i++)
     {  
@@ -913,7 +915,7 @@ public:
           else
           {
             table_field->collected_stats->min_value->val_str(&val);
-            stat_field->store(val.ptr(), val.length(), &my_charset_utf8_bin);
+            stat_field->store(val.ptr(), val.length(), &my_charset_bin);
           }
           break;
         case COLUMN_STAT_MAX_VALUE:
@@ -922,7 +924,7 @@ public:
           else
           {
             table_field->collected_stats->max_value->val_str(&val);
-            stat_field->store(val.ptr(), val.length(), &my_charset_utf8_bin);
+            stat_field->store(val.ptr(), val.length(), &my_charset_bin);
           }
           break;
         case COLUMN_STAT_NULLS_RATIO:
@@ -983,7 +985,7 @@ public:
     if (find_stat())
     {
       char buff[MAX_FIELD_WIDTH];
-      String val(buff, sizeof(buff), &my_charset_utf8_bin);
+      String val(buff, sizeof(buff), &my_charset_bin);
 
       for (uint i= COLUMN_STAT_MIN_VALUE; i <= COLUMN_STAT_HIST_TYPE; i++)
       {  
@@ -1002,12 +1004,12 @@ public:
           case COLUMN_STAT_MIN_VALUE:
             stat_field->val_str(&val);
             table_field->read_stats->min_value->store(val.ptr(), val.length(),
-                                                      &my_charset_utf8_bin);
+                                                      &my_charset_bin);
             break;
           case COLUMN_STAT_MAX_VALUE:
             stat_field->val_str(&val);
             table_field->read_stats->max_value->store(val.ptr(), val.length(),
-                                                      &my_charset_utf8_bin);
+                                                      &my_charset_bin);
             break;
           case COLUMN_STAT_NULLS_RATIO:
             table_field->read_stats->set_nulls_ratio(stat_field->val_real());
@@ -1053,7 +1055,7 @@ public:
     if (find_stat())
     {
       char buff[MAX_FIELD_WIDTH];
-      String val(buff, sizeof(buff), &my_charset_utf8_bin);
+      String val(buff, sizeof(buff), &my_charset_bin);
       uint fldno= COLUMN_STAT_HISTOGRAM;
       Field *stat_field= stat_table->field[fldno];
       table_field->read_stats->set_not_null(fldno);
@@ -1264,6 +1266,117 @@ public:
 
 };
 
+
+/*
+  An iterator to enumerate statistics table rows which allows to modify
+  the rows while reading them.
+
+  Used by RENAME TABLE handling to assign new dbname.tablename to statistic
+  rows.
+*/
+class Stat_table_write_iter
+{
+  Stat_table *owner;
+  IO_CACHE io_cache;
+  uchar *rowid_buf;
+  uint rowid_size;
+
+public:
+  Stat_table_write_iter(Stat_table *stat_table_arg)
+   : owner(stat_table_arg), rowid_buf(NULL),
+     rowid_size(owner->stat_file->ref_length)
+  {
+     my_b_clear(&io_cache);
+  }
+
+  /*
+    Initialize the iterator. It will return rows with n_keyparts matching the
+    curernt values.
+
+    @return  false - OK
+             true  - Error
+  */
+  bool init(uint n_keyparts)
+  {
+    if (!(rowid_buf= (uchar*)my_malloc(rowid_size, MYF(0))))
+      return true;
+
+    if (open_cached_file(&io_cache, mysql_tmpdir, TEMP_PREFIX,
+                         1024, MYF(MY_WME)))
+      return true;
+
+    handler *h= owner->stat_file;
+    uchar key[MAX_KEY_LENGTH];
+    uint prefix_len= 0;
+    for (uint i= 0; i < n_keyparts; i++)
+      prefix_len += owner->stat_key_info->key_part[i].store_length;
+
+    key_copy(key, owner->record[0], owner->stat_key_info,
+             prefix_len);
+    key_part_map prefix_map= (key_part_map) ((1 << n_keyparts) - 1);
+    h->ha_index_init(owner->stat_key_idx, false);
+    int res= h->ha_index_read_map(owner->record[0], key, prefix_map,
+                                  HA_READ_KEY_EXACT);
+    if (res)
+    {
+      reinit_io_cache(&io_cache, READ_CACHE, 0L, 0, 0);
+      /* "Key not found" is not considered an error */
+      return (res == HA_ERR_KEY_NOT_FOUND)? false: true;
+    }
+
+    do {
+      h->position(owner->record[0]);
+      my_b_write(&io_cache, h->ref, rowid_size);
+
+    } while (!h->ha_index_next_same(owner->record[0], key, prefix_len));
+
+    /* Prepare for reading */
+    reinit_io_cache(&io_cache, READ_CACHE, 0L, 0, 0);
+    h->ha_index_or_rnd_end();
+    if (h->ha_rnd_init(false))
+      return true;
+
+    return false;
+  }
+
+  /*
+     Read the next row.
+
+     @return
+        false   OK
+        true    No more rows or error.
+  */
+  bool get_next_row()
+  {
+    if (!my_b_inited(&io_cache) || my_b_read(&io_cache, rowid_buf, rowid_size))
+      return true; /* No more data */
+
+    handler *h= owner->stat_file;
+    /*
+      We should normally be able to find the row that we have rowid for. If we
+      don't, let's consider this an error.
+    */
+    int res= h->ha_rnd_pos(owner->record[0], rowid_buf);
+
+    return (res==0)? false : true;
+  }
+
+  void cleanup()
+  {
+    if (rowid_buf)
+      my_free(rowid_buf);
+    rowid_buf= NULL;
+    owner->stat_file->ha_index_or_rnd_end();
+    close_cached_file(&io_cache);
+    my_b_clear(&io_cache);
+  }
+
+  ~Stat_table_write_iter()
+  {
+    cleanup();
+  }
+};
+
 /*
   Histogram_builder is a helper class that is used to build histograms
   for columns
@@ -3285,25 +3398,34 @@ int rename_table_in_stat_tables(THD *thd, LEX_STRING *db, LEX_STRING *tab,
   stat_table= tables[INDEX_STAT].table;
   Index_stat index_stat(stat_table, db, tab);
   index_stat.set_full_table_name();
-  while (index_stat.find_next_stat_for_prefix(2))
+
+  Stat_table_write_iter index_iter(&index_stat);
+  if (index_iter.init(2))
+    rc= 1;
+  while (!index_iter.get_next_row())
   {
     err= index_stat.update_table_name_key_parts(new_db, new_tab);
     if (err & !rc)
       rc= 1;
     index_stat.set_full_table_name();
   }
+  index_iter.cleanup();
 
   /* Rename table in the statistical table column_stats */
   stat_table= tables[COLUMN_STAT].table;
   Column_stat column_stat(stat_table, db, tab);
   column_stat.set_full_table_name();
-  while (column_stat.find_next_stat_for_prefix(2))
+  Stat_table_write_iter column_iter(&column_stat);
+  if (column_iter.init(2))
+    rc= 1;
+  while (!column_iter.get_next_row())
   {
     err= column_stat.update_table_name_key_parts(new_db, new_tab);
     if (err & !rc)
       rc= 1;
     column_stat.set_full_table_name();
   }
+  column_iter.cleanup();
    
   /* Rename table in the statistical table table_stats */
   stat_table= tables[TABLE_STAT].table;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 65d601968ba..e2ad9237120 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -3183,13 +3183,14 @@ static void check_duplicate_key(THD *thd,
 
     // Report a warning if we have two identical keys.
 
+    DBUG_ASSERT(thd->lex->query_tables->alias);
     if (all_columns_are_identical)
     {
       push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
                           ER_DUP_INDEX, ER(ER_DUP_INDEX),
                           key_info->name,
                           thd->lex->query_tables->db,
-                          thd->lex->query_tables->table_name);
+                          thd->lex->query_tables->alias);
       break;
     }
   }
@@ -3326,9 +3327,10 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
                                                sql_field->interval_list);
         List_iterator<String> int_it(sql_field->interval_list);
         String conv, *tmp;
-        char comma_buf[4]; /* 4 bytes for utf32 */
+        char comma_buf[5]; /* 5 bytes for 'filename' charset */
+        DBUG_ASSERT(sizeof(comma_buf) >= cs->mbmaxlen);
         int comma_length= cs->cset->wc_mb(cs, ',', (uchar*) comma_buf,
-                                          (uchar*) comma_buf + 
+                                          (uchar*) comma_buf +
                                           sizeof(comma_buf));
         DBUG_ASSERT(comma_length > 0);
         for (uint i= 0; (tmp= int_it++); i++)
@@ -4650,8 +4652,8 @@ int create_table_impl(THD *thd,
   bool          frm_only= create_table_mode == C_ALTER_TABLE_FRM_ONLY;
   bool          internal_tmp_table= create_table_mode == C_ALTER_TABLE || frm_only;
   DBUG_ENTER("mysql_create_table_no_lock");
-  DBUG_PRINT("enter", ("db: '%s'  table: '%s'  tmp: %d",
-                       db, table_name, internal_tmp_table));
+  DBUG_PRINT("enter", ("db: '%s'  table: '%s'  tmp: %d  path: %s",
+                       db, table_name, internal_tmp_table, path));
 
   if (thd->variables.sql_mode & MODE_NO_DIR_IN_CREATE)
   {
diff --git a/sql/sql_time.cc b/sql/sql_time.cc
index f2596401c2e..b55b1d76b99 100644
--- a/sql/sql_time.cc
+++ b/sql/sql_time.cc
@@ -1328,7 +1328,7 @@ time_to_datetime_with_warn(THD *thd,
     only in the old mode.
   */
   if (time_to_datetime(thd, from, to) ||
-      ((thd->variables.old_behavior && OLD_MODE_ZERO_DATE_TIME_CAST) &&
+      ((thd->variables.old_behavior & OLD_MODE_ZERO_DATE_TIME_CAST) &&
         check_date(to, fuzzydate, &warn)))
   {
     ErrConvTime str(from);
diff --git a/sql/sql_udf.cc b/sql/sql_udf.cc
index bd5732c3696..0f9043a77df 100644
--- a/sql/sql_udf.cc
+++ b/sql/sql_udf.cc
@@ -143,7 +143,7 @@ void udf_init()
   DBUG_ENTER("ufd_init");
   char db[]= "mysql"; /* A subject to casednstr, can't be constant */
 
-  if (initialized)
+  if (initialized || opt_noacl)
     DBUG_VOID_RETURN;
 
 #ifdef HAVE_PSI_INTERFACE
@@ -268,6 +268,8 @@ void udf_free()
 {
   /* close all shared libraries */
   DBUG_ENTER("udf_free");
+  if (opt_noacl)
+    DBUG_VOID_RETURN;
   for (uint idx=0 ; idx < udf_hash.records ; idx++)
   {
     udf_func *udf=(udf_func*) my_hash_element(&udf_hash,idx);
diff --git a/sql/sql_udf.h b/sql/sql_udf.h
index 4aa055b9858..f10cbc3bbc2 100644
--- a/sql/sql_udf.h
+++ b/sql/sql_udf.h
@@ -143,5 +143,8 @@ udf_func *find_udf(const char *name, uint len=0,bool mark_used=0);
 void free_udf(udf_func *udf);
 int mysql_create_function(THD *thd,udf_func *udf);
 int mysql_drop_function(THD *thd,const LEX_STRING *name);
+#else
+static inline void udf_init(void) { }
+static inline void udf_free(void) { }
 #endif
 #endif /* SQL_UDF_INCLUDED */
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index e64aebcb78c..4c1fdbfa93b 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -364,6 +364,9 @@ int mysql_update(THD *thd,
     DBUG_RETURN(1);				/* purecov: inspected */
   }
 
+  if (check_unique_table(thd, table_list))
+    DBUG_RETURN(TRUE);
+
   /* Apply the IN=>EXISTS transformation to all subqueries and optimize them. */
   if (select_lex->optimize_unflattened_subqueries(false))
     DBUG_RETURN(TRUE);
@@ -1099,19 +1102,30 @@ bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
       setup_ftfuncs(select_lex))
     DBUG_RETURN(TRUE);
 
-  /* Check that we are not using table that we are updating in a sub select */
-  {
-    TABLE_LIST *duplicate;
-    if ((duplicate= unique_table(thd, table_list, table_list->next_global, 0)))
-    {
-      update_non_unique_table_error(table_list, "UPDATE", duplicate);
-      DBUG_RETURN(TRUE);
-    }
-  }
   select_lex->fix_prepare_information(thd, conds, &fake_conds);
   DBUG_RETURN(FALSE);
 }
 
+/**
+  Check that we are not using table that we are updating in a sub select
+
+  @param thd             Thread handle
+  @param table_list      List of table with first to check
+
+  @retval TRUE  Error
+  @retval FALSE OK
+*/
+bool check_unique_table(THD *thd, TABLE_LIST *table_list)
+{
+  TABLE_LIST *duplicate;
+  DBUG_ENTER("check_unique_table");
+  if ((duplicate= unique_table(thd, table_list, table_list->next_global, 0)))
+  {
+    update_non_unique_table_error(table_list, "UPDATE", duplicate);
+    DBUG_RETURN(TRUE);
+  }
+  DBUG_RETURN(FALSE);
+}
 
 /***************************************************************************
   Update multiple tables from join 
diff --git a/sql/sql_update.h b/sql/sql_update.h
index 64029c5d634..4c6f89d8468 100644
--- a/sql/sql_update.h
+++ b/sql/sql_update.h
@@ -27,6 +27,7 @@ typedef class st_select_lex_unit SELECT_LEX_UNIT;
 
 bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
                           Item **conds, uint order_num, ORDER *order);
+bool check_unique_table(THD *thd, TABLE_LIST *table_list);
 int mysql_update(THD *thd,TABLE_LIST *tables,List<Item> &fields,
 		 List<Item> &values,COND *conds,
 		 uint order_num, ORDER *order, ha_rows limit,
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index a63d8a51a86..3814d58ed75 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -1493,6 +1493,11 @@ bool mysql_make_view(THD *thd, File_parser *parser, TABLE_LIST *table,
       */
       lex->sql_command= old_lex->sql_command;
       lex->duplicates= old_lex->duplicates;
+
+      /* Fields in this view can be used in upper select in case of merge.  */
+      if (table->select_lex)
+        table->select_lex->select_n_where_fields+=
+          lex->select_lex.select_n_where_fields;
     }
     /*
       This method has a dependency on the proper lock type being set,
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 13f963e2f4c..ba601aa772e 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -10769,6 +10769,15 @@ table_factor:
               sel->add_joined_table($$);
               lex->pop_context();
               lex->nest_level--;
+              /*
+                Fields in derived table can be used in upper select in
+                case of merge. We do not add HAVING fields because we do
+                not merge such derived. We do not add union because
+                also do not merge them
+              */
+              if (!sel->next_select())
+                $2->select_n_where_fields+=
+                  sel->select_n_where_fields;
             }
             /*else if (($3->select_lex &&
                       $3->select_lex->master_unit()->is_union() &&
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 4d2e35bac87..e971f367fa3 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -1625,7 +1625,7 @@ Sys_var_gtid_slave_pos::global_value_ptr(THD *thd, LEX_STRING *base)
     But if the table is not loaded (eg. missing mysql_upgrade_db or some such),
     then the slave state must be empty anyway.
   */
-  if ((rpl_global_gtid_slave_state.loaded &&
+  if ((rpl_global_gtid_slave_state->loaded &&
        rpl_append_gtid_state(&str, false)) ||
       !(p= thd->strmake(str.ptr(), str.length())))
   {
@@ -3312,7 +3312,7 @@ static Sys_var_charptr Sys_version_compile_machine(
        "version_compile_machine", "version_compile_machine",
        READ_ONLY SHOW_VALUE_IN_HELP
        GLOBAL_VAR(server_version_compile_machine_ptr), NO_CMD_LINE,
-       IN_SYSTEM_CHARSET, DEFAULT(MACHINE_TYPE));
+       IN_SYSTEM_CHARSET, DEFAULT(DEFAULT_MACHINE));
 
 static char *server_version_compile_os_ptr;
 static Sys_var_charptr Sys_version_compile_os(
diff --git a/sql/table.cc b/sql/table.cc
index 90cdc7089b9..5b69ccc23de 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2000, 2014, Oracle and/or its affiliates.
+/* Copyright (c) 2000, 2015, Oracle and/or its affiliates.
    Copyright (c) 2008, 2015, MariaDB
 
    This program is free software; you can redistribute it and/or modify
@@ -2624,21 +2624,6 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
       outparam->record[1]= outparam->record[0];   // Safety
   }
 
-#ifdef HAVE_valgrind
-  /*
-    We need this because when we read var-length rows, we are not updating
-    bytes after end of varchar
-  */
-  if (records > 1)
-  {
-    memcpy(outparam->record[0], share->default_values, share->rec_buff_length);
-    memcpy(outparam->record[1], share->default_values, share->null_bytes);
-    if (records > 2)
-      memcpy(outparam->record[1], share->default_values,
-             share->rec_buff_length);
-  }
-#endif
-
   if (!(field_ptr = (Field **) alloc_root(&outparam->mem_root,
                                           (uint) ((share->fields+1)*
                                                   sizeof(Field*)))))
diff --git a/sql/threadpool.h b/sql/threadpool.h
index c080e5ba343..bcbdca47808 100644
--- a/sql/threadpool.h
+++ b/sql/threadpool.h
@@ -27,6 +27,7 @@ extern uint threadpool_oversubscribe;  /* Maximum active threads in group */
 
 
 /* Common thread pool routines, suitable for different implementations */
+extern void threadpool_cleanup_connection(THD *thd);
 extern void threadpool_remove_connection(THD *thd);
 extern int  threadpool_process_request(THD *thd);
 extern int  threadpool_add_connection(THD *thd);
diff --git a/sql/threadpool_common.cc b/sql/threadpool_common.cc
index 9e0cb07b86c..5bcea767aae 100644
--- a/sql/threadpool_common.cc
+++ b/sql/threadpool_common.cc
@@ -168,22 +168,28 @@ int threadpool_add_connection(THD *thd)
   return retval;
 }
 
+/*
+  threadpool_cleanup_connection() does the bulk of connection shutdown work.
+  Usually called from threadpool_remove_connection(), but rarely it might
+  be called also in the main polling thread if connection initialization fails.
+*/
+void threadpool_cleanup_connection(THD *thd)
+{
+  thd->net.reading_or_writing = 0;
+  end_connection(thd);
+  close_connection(thd, 0);
+  unlink_thd(thd);
+  mysql_cond_broadcast(&COND_thread_count);
+}
+
 
 void threadpool_remove_connection(THD *thd)
 {
-
   Worker_thread_context worker_context;
   worker_context.save();
-
   thread_attach(thd);
-  thd->net.reading_or_writing= 0;
-
-  end_connection(thd);
-  close_connection(thd, 0);
-
-  unlink_thd(thd);
-  mysql_cond_broadcast(&COND_thread_count);
 
+  threadpool_cleanup_connection(thd);
   /*
     Free resources associated with this connection: 
     mysys thread_var and PSI thread.
diff --git a/sql/threadpool_unix.cc b/sql/threadpool_unix.cc
index e720e43498a..89a2036cb10 100644
--- a/sql/threadpool_unix.cc
+++ b/sql/threadpool_unix.cc
@@ -1255,7 +1255,7 @@ void tp_add_connection(THD *thd)
   else
   {
     /* Allocation failed */
-    threadpool_remove_connection(thd);
+    threadpool_cleanup_connection(thd);
   } 
   DBUG_VOID_RETURN;
 }
diff --git a/sql/threadpool_win.cc b/sql/threadpool_win.cc
index 9cef1af272c..4be51f3d6e9 100644
--- a/sql/threadpool_win.cc
+++ b/sql/threadpool_win.cc
@@ -667,7 +667,7 @@ void tp_add_connection(THD *thd)
   if(!con)
   {
     tp_log_warning("Allocation failed", "tp_add_connection");
-    threadpool_remove_connection(thd);
+    threadpool_cleanup_connection(thd);
     return;
   }
 
@@ -685,7 +685,7 @@ void tp_add_connection(THD *thd)
   else
   {
     /* Likely memory pressure */
-    login_callback(NULL, con, NULL); /* deletes connection if something goes wrong */
+    threadpool_cleanup_connection(thd);
   }
 }