Merge 10.6 into 10.7

17 files changed, 439 insertions, 248 deletions

diff --git a/sql/field.cc b/sql/field.cc
index a5cd595f05b..63d39c77d95 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1276,6 +1276,21 @@ bool Field::can_be_substituted_to_equal_item(const Context &ctx,
 }
 
 
+bool Field::cmp_is_done_using_type_handler_of_this(const Item_bool_func *cond,
+                                                   const Item *item) const
+{
+  /*
+    We could eventually take comparison_type_handler() from cond,
+    instead of calculating it again. But only some descendants of
+    Item_bool_func has this method. So this needs some hierarchy changes.
+    Another option is to pass "class Context" to this method.
+  */
+  Type_handler_hybrid_field_type cmp(type_handler_for_comparison());
+  return !cmp.aggregate_for_comparison(item->type_handler_for_comparison()) &&
+         cmp.type_handler() == type_handler_for_comparison();
+}
+
+
 /*
   This handles all numeric and BIT data types.
 */ 
@@ -7368,7 +7383,7 @@ bool
 Field_longstr::cmp_to_string_with_same_collation(const Item_bool_func *cond,
                                                  const Item *item) const
 {
-  return item->cmp_type() == STRING_RESULT &&
+  return cmp_is_done_using_type_handler_of_this(cond, item) &&
          charset() == cond->compare_collation();
 }
 
@@ -7377,7 +7392,7 @@ bool
 Field_longstr::cmp_to_string_with_stricter_collation(const Item_bool_func *cond,
                                                      const Item *item) const
 {
-  return item->cmp_type() == STRING_RESULT &&
+  return cmp_is_done_using_type_handler_of_this(cond, item) &&
          (charset() == cond->compare_collation() ||
           cond->compare_collation()->state & MY_CS_BINSORT);
 }
diff --git a/sql/field.h b/sql/field.h
index 0408ae90b3f..0b8f317b5b8 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -1651,6 +1651,8 @@ protected:
   }
   int warn_if_overflow(int op_result);
   Copy_func *get_identical_copy_func() const;
+  bool cmp_is_done_using_type_handler_of_this(const Item_bool_func *cond,
+                                              const Item *item) const;
   bool can_optimize_scalar_range(const RANGE_OPT_PARAM *param,
                                  const KEY_PART *key_part,
                                  const Item_bool_func *cond,
diff --git a/sql/item.cc b/sql/item.cc
index 82c09f595fd..ea88fb1a572 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -3846,8 +3846,48 @@ void Item_string::print(String *str, enum_query_type query_type)
   }
   else
   {
-    // Caller wants a result in the charset of str_value.
-    str_value.print(str);
+    /*
+      We're restoring a parse-able statement from an Item tree.
+      Make sure to revert character set conversions that previously
+      happened in the parser when Item_string was created.
+    */
+    if (print_introducer)
+    {
+      /*
+        Print the string as is, without conversion:
+        Strings with introducers are not converted in the parser.
+      */
+      str_value.print(str);
+    }
+    else
+    {
+      /*
+        Print the string with conversion.
+        Strings without introducers are converted in the parser,
+        from character_set_client to character_set_connection.
+
+        When restoring a CREATE VIEW statement,
+        - str_value.charsets() contains parse time character_set_connection
+        - str->charset() contains parse time character_set_client
+        So we convert the string back from parse-time character_set_connection
+        to parse time character_set_client.
+
+        In some cases, e.g. SHOW PROCEDURE CODE, it's also possible
+        that str->charset() is "utf8mb3" instead of parse time
+        character_set_client. In these cases we convert
+        here from the parse-time character_set_connection to utf8mb3.
+
+        QQ: perhaps the code behind SHOW PROCEDURE CODE should
+        also request the result in the parse-time character_set_client
+        (like the code restoring CREATE VIEW statements does),
+        rather than in utf8mb3:
+        - utf8mb3 does not work well with non-BMP characters (e.g. emoji).
+        - Simply changing utf8mb3 to utf8mb4 will not fully help:
+          some character sets have unassigned characters,
+          they get lost during during cs->utf8mb4->cs round trip.
+      */
+      str_value.print_with_conversion(str, str->charset());
+    }
   }
 
   str->append('\'');
diff --git a/sql/item_jsonfunc.cc b/sql/item_jsonfunc.cc
index 4d966a7d79b..fefb2a1c662 100644
--- a/sql/item_jsonfunc.cc
+++ b/sql/item_jsonfunc.cc
@@ -3640,6 +3640,7 @@ LEX_CSTRING Item_func_json_format::func_name_cstring() const
 bool Item_func_json_format::fix_length_and_dec()
 {
   decimals= 0;
+  collation.set(args[0]->collation);
   max_length= args[0]->max_length;
   set_maybe_null();
   return FALSE;
diff --git a/sql/log.h b/sql/log.h
index 000d22854e9..b55ed0f31f6 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -951,6 +951,20 @@ public:
   void unlock_binlog_end_pos() { mysql_mutex_unlock(&LOCK_binlog_end_pos); }
   mysql_mutex_t* get_binlog_end_pos_lock() { return &LOCK_binlog_end_pos; }
 
+  /*
+    Ensures the log's state is either LOG_OPEN or LOG_CLOSED. If something
+    failed along the desired path and left the log in invalid state, i.e.
+    LOG_TO_BE_OPENED, forces the state to be LOG_CLOSED.
+  */
+  void try_fix_log_state()
+  {
+    mysql_mutex_lock(get_log_lock());
+    /* Only change the log state if it is LOG_TO_BE_OPENED */
+    if (log_state == LOG_TO_BE_OPENED)
+      log_state= LOG_CLOSED;
+    mysql_mutex_unlock(get_log_lock());
+  }
+
   int wait_for_update_binlog_end_pos(THD* thd, struct timespec * timeout);
 
   /*
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index e797511fb9c..d1df2d7721d 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -5621,20 +5621,14 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
       }
     }
 
-#ifdef HAVE_QUERY_CACHE
-#ifdef WITH_WSREP
+#if defined(WITH_WSREP) && defined(HAVE_QUERY_CACHE)
     /*
       Moved invalidation right before the call to rows_event_stmt_cleanup(),
       to avoid query cache being polluted with stale entries,
     */
-    if (! (WSREP(thd) && wsrep_thd_is_applying(thd)))
-    {
-#endif /* WITH_WSREP */
-    query_cache.invalidate_locked_for_write(thd, rgi->tables_to_lock);
-#ifdef WITH_WSREP
-    }
-#endif /* WITH_WSREP */
-#endif
+    if (WSREP(thd) && wsrep_thd_is_applying(thd))
+      query_cache.invalidate_locked_for_write(thd, rgi->tables_to_lock);
+#endif /* WITH_WSREP && HAVE_QUERY_CACHE */
   }
 
   table= m_table= rgi->m_table_map.get_table(m_table_id);
@@ -5835,19 +5829,20 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
   restore_empty_query_table_list(thd->lex);
 
 #if defined(WITH_WSREP) && defined(HAVE_QUERY_CACHE)
-    if (WSREP(thd) && wsrep_thd_is_applying(thd))
-    {
-      query_cache.invalidate_locked_for_write(thd, rgi->tables_to_lock);
-    }
+  if (WSREP(thd) && wsrep_thd_is_applying(thd))
+    query_cache.invalidate_locked_for_write(thd, rgi->tables_to_lock);
 #endif /* WITH_WSREP && HAVE_QUERY_CACHE */
 
-    if (unlikely(get_flags(STMT_END_F) &&
-                 (error= rows_event_stmt_cleanup(rgi, thd))))
-    slave_rows_error_report(ERROR_LEVEL,
-                            thd->is_error() ? 0 : error,
-                            rgi, thd, table,
-                            get_type_str(),
-                            RPL_LOG_NAME, log_pos);
+  if (get_flags(STMT_END_F))
+  {
+    if (unlikely((error= rows_event_stmt_cleanup(rgi, thd))))
+      slave_rows_error_report(ERROR_LEVEL, thd->is_error() ? 0 : error,
+                              rgi, thd, table, get_type_str(),
+                              RPL_LOG_NAME, log_pos);
+    if (thd->slave_thread)
+      free_root(thd->mem_root, MYF(MY_KEEP_PREALLOC));
+  }
+
   DBUG_RETURN(error);
 
 err:
diff --git a/sql/my_json_writer.cc b/sql/my_json_writer.cc
index a70ae164f18..55e71bc1b52 100644
--- a/sql/my_json_writer.cc
+++ b/sql/my_json_writer.cc
@@ -260,6 +260,10 @@ void Json_writer::add_str(const String &str)
   add_str(str.ptr(), str.length());
 }
 
+#ifdef ENABLED_JSON_WRITER_CONSISTENCY_CHECKS
+thread_local std::vector<bool> Json_writer_struct::named_items_expectation;
+#endif
+
 Json_writer_temp_disable::Json_writer_temp_disable(THD *thd_arg)
 {
   thd= thd_arg;
diff --git a/sql/my_json_writer.h b/sql/my_json_writer.h
index 27aec74d08d..d82313f996f 100644
--- a/sql/my_json_writer.h
+++ b/sql/my_json_writer.h
@@ -15,8 +15,12 @@
 
 #ifndef JSON_WRITER_INCLUDED
 #define JSON_WRITER_INCLUDED
+
 #include "my_base.h"
 #include "sql_select.h"
+
+#include <vector>
+
 class Opt_trace_stmt;
 class Opt_trace_context;
 class Json_writer;
@@ -308,6 +312,9 @@ public:
 /* A common base for Json_writer_object and Json_writer_array */
 class Json_writer_struct
 {
+#ifdef ENABLED_JSON_WRITER_CONSISTENCY_CHECKS
+  static thread_local std::vector<bool> named_items_expectation;
+#endif
 protected:
   Json_writer* my_writer;
   Json_value_helper context;
@@ -317,16 +324,35 @@ protected:
   bool closed;
 
 public:
-  explicit Json_writer_struct(THD *thd)
+  explicit Json_writer_struct(THD *thd, bool expect_named_children)
   {
     my_writer= thd->opt_trace.get_current_json();
     context.init(my_writer);
     closed= false;
+#ifdef ENABLED_JSON_WRITER_CONSISTENCY_CHECKS
+    named_items_expectation.push_back(expect_named_children);
+#endif
+  }
+
+  virtual ~Json_writer_struct()
+  {
+#ifdef ENABLED_JSON_WRITER_CONSISTENCY_CHECKS
+    named_items_expectation.pop_back();
+#endif
   }
-  bool trace_started()
+
+  bool trace_started() const
   {
     return my_writer != 0;
   }
+
+#ifdef ENABLED_JSON_WRITER_CONSISTENCY_CHECKS
+  bool named_item_expected() const
+  {
+    return named_items_expectation.size() > 1
+        && *(named_items_expectation.rbegin() + 1);
+  }
+#endif
 };
 
 
@@ -347,15 +373,21 @@ private:
   }
 public:
   explicit Json_writer_object(THD *thd)
-  : Json_writer_struct(thd)
+  : Json_writer_struct(thd, true)
   {
+#ifdef ENABLED_JSON_WRITER_CONSISTENCY_CHECKS
+    DBUG_ASSERT(!named_item_expected());
+#endif
     if (unlikely(my_writer))
       my_writer->start_object();
   }
 
   explicit Json_writer_object(THD* thd, const char *str)
-    : Json_writer_struct(thd)
+  : Json_writer_struct(thd, true)
   {
+#ifdef ENABLED_JSON_WRITER_CONSISTENCY_CHECKS
+    DBUG_ASSERT(named_item_expected());
+#endif
     if (unlikely(my_writer))
       my_writer->add_member(str).start_object();
   }
@@ -519,14 +551,22 @@ public:
 class Json_writer_array : public Json_writer_struct
 {
 public:
-  Json_writer_array(THD *thd): Json_writer_struct(thd)
+  Json_writer_array(THD *thd)
+  : Json_writer_struct(thd, false)
   {
+#ifdef ENABLED_JSON_WRITER_CONSISTENCY_CHECKS
+    DBUG_ASSERT(!named_item_expected());
+#endif
     if (unlikely(my_writer))
       my_writer->start_array();
   }
 
-  Json_writer_array(THD *thd, const char *str) : Json_writer_struct(thd)
+  Json_writer_array(THD *thd, const char *str)
+  : Json_writer_struct(thd, false)
   {
+#ifdef ENABLED_JSON_WRITER_CONSISTENCY_CHECKS
+    DBUG_ASSERT(named_item_expected());
+#endif
     if (unlikely(my_writer))
       my_writer->add_member(str).start_array();
   }
diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc
index 68753647366..4fd36891a4b 100644
--- a/sql/rpl_mi.cc
+++ b/sql/rpl_mi.cc
@@ -170,6 +170,8 @@ void Master_info::clear_in_memory_info(bool all)
   {
     port= MYSQL_PORT;
     host[0] = 0; user[0] = 0; password[0] = 0;
+    domain_id_filter.clear_ids();
+    reset_dynamic(&ignore_server_ids);
   }
 }
 
@@ -1816,6 +1818,12 @@ void Domain_id_filter::reset_filter()
   m_filter= false;
 }
 
+void Domain_id_filter::clear_ids()
+{
+  reset_dynamic(&m_domain_ids[DO_DOMAIN_IDS]);
+  reset_dynamic(&m_domain_ids[IGNORE_DOMAIN_IDS]);
+}
+
 /**
   Update the do/ignore domain id filter lists.
 
diff --git a/sql/rpl_mi.h b/sql/rpl_mi.h
index d2232ac2664..a4a06d42a5c 100644
--- a/sql/rpl_mi.h
+++ b/sql/rpl_mi.h
@@ -79,6 +79,11 @@ public:
   void reset_filter();
 
   /*
+    Clear do_ids and ignore_ids to disable domain id filtering
+  */
+  void clear_ids();
+
+  /*
     Update the do/ignore domain id filter lists.
 
     @param do_ids     [IN]            domain ids to be kept
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 5327691ce89..675019fb88e 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -3362,24 +3362,24 @@ static int acl_user_update(THD *thd, ACL_USER *acl_user, uint nauth,
                            const Account_options &options,
                            const privilege_t privileges)
 {
+  ACL_USER_PARAM::AUTH *work_copy= NULL;
   if (nauth)
   {
-    if (acl_user->nauth >= nauth)
-      acl_user->nauth= nauth;
-    else
-      acl_user->alloc_auth(&acl_memroot, nauth);
+    if (!(work_copy= (ACL_USER_PARAM::AUTH*)
+            alloc_root(thd->mem_root, nauth * sizeof(ACL_USER_PARAM::AUTH))))
+      return 1;
 
     USER_AUTH *auth= combo.auth;
     for (uint i= 0; i < nauth; i++, auth= auth->next)
     {
-      acl_user->auth[i].plugin= auth->plugin;
-      acl_user->auth[i].auth_string= safe_lexcstrdup_root(&acl_memroot, auth->auth_str);
-      if (fix_user_plugin_ptr(acl_user->auth + i))
-        acl_user->auth[i].plugin= safe_lexcstrdup_root(&acl_memroot, auth->plugin);
-      LEX_CSTRING host= { acl_user->host.hostname ,
-                          acl_user->hostname_length};
-      if (set_user_auth(thd, acl_user->user, host,
-                        acl_user->auth + i, auth->pwtext))
+      work_copy[i].plugin= auth->plugin;
+      work_copy[i].auth_string= safe_lexcstrdup_root(&acl_memroot,
+                                                     auth->auth_str);
+      if (fix_user_plugin_ptr(work_copy + i))
+        work_copy[i].plugin= safe_lexcstrdup_root(&acl_memroot, auth->plugin);
+      if (set_user_auth(thd, acl_user->user,
+                        {acl_user->host.hostname, acl_user->hostname_length},
+                        work_copy + i, auth->pwtext))
         return 1;
     }
   }
@@ -3407,11 +3407,34 @@ static int acl_user_update(THD *thd, ACL_USER *acl_user, uint nauth,
   if (options.account_locked != ACCOUNTLOCK_UNSPECIFIED)
     acl_user->account_locked= options.account_locked == ACCOUNTLOCK_LOCKED;
 
-  /* Unexpire the user password */
+  if (thd->is_error())
+  {
+    // If something went wrong (including OOM) we will not spoil acl cache
+    return 1;
+  }
+  /* Unexpire the user password and copy AUTH (when no more errors possible)*/
   if (nauth)
   {
     acl_user->password_expired= false;
-    acl_user->password_last_changed= thd->query_start();;
+    acl_user->password_last_changed= thd->query_start();
+
+    if (acl_user->nauth >= nauth)
+    {
+      acl_user->nauth= nauth;
+    }
+    else
+    {
+      if (acl_user->alloc_auth(&acl_memroot, nauth))
+      {
+        /*
+          acl_user is a copy, so NULL assigned in case of an error do not
+          change the acl cache
+        */
+        return 1;
+      }
+    }
+    DBUG_ASSERT(work_copy); // allocated under the same condinition
+    memcpy(acl_user->auth, work_copy,  nauth * sizeof(ACL_USER_PARAM::AUTH));
   }
 
   switch (options.password_expire) {
@@ -10208,8 +10231,8 @@ static int handle_grant_struct(enum enum_acl_lists struct_no, bool drop,
                                LEX_USER *user_from, LEX_USER *user_to)
 {
   int result= 0;
-  int idx;
   int elements;
+  bool restart;
   const char *UNINIT_VAR(user);
   const char *UNINIT_VAR(host);
   ACL_USER *acl_user= NULL;
@@ -10317,82 +10340,98 @@ static int handle_grant_struct(enum enum_acl_lists struct_no, bool drop,
     DBUG_RETURN(-1);
   }
 
+
 #ifdef EXTRA_DEBUG
     DBUG_PRINT("loop",("scan struct: %u  search    user: '%s'  host: '%s'",
                        struct_no, user_from->user.str, user_from->host.str));
 #endif
-  /* Loop over all elements *backwards* (see the comment below). */
-  for (idx= elements - 1; idx >= 0; idx--)
-  {
-    /*
-      Get a pointer to the element.
-    */
-    switch (struct_no) {
-    case USER_ACL:
-      acl_user= dynamic_element(&acl_users, idx, ACL_USER*);
-      user= acl_user->user.str;
-      host= acl_user->host.hostname;
-    break;
+  /* Loop over elements backwards as it may reduce the number of mem-moves
+     for dynamic arrays.
 
-    case DB_ACL:
-      acl_db= &acl_dbs.at(idx);
-      user= acl_db->user;
-      host= acl_db->host.hostname;
+     We restart the loop, if we deleted or updated anything in a hash table
+     because calling my_hash_delete or my_hash_update shuffles elements indices
+     and we can miss some if we do only one scan.
+  */
+  do {
+    restart= false;
+    for (int idx= elements - 1; idx >= 0; idx--)
+    {
+      /*
+        Get a pointer to the element.
+      */
+      switch (struct_no) {
+      case USER_ACL:
+        acl_user= dynamic_element(&acl_users, idx, ACL_USER*);
+        user= acl_user->user.str;
+        host= acl_user->host.hostname;
       break;
 
-    case COLUMN_PRIVILEGES_HASH:
-    case PROC_PRIVILEGES_HASH:
-    case FUNC_PRIVILEGES_HASH:
-    case PACKAGE_SPEC_PRIVILEGES_HASH:
-    case PACKAGE_BODY_PRIVILEGES_HASH:
-      grant_name= (GRANT_NAME*) my_hash_element(grant_name_hash, idx);
-      user= grant_name->user;
-      host= grant_name->host.hostname;
-      break;
+      case DB_ACL:
+        acl_db= &acl_dbs.at(idx);
+        user= acl_db->user;
+        host= acl_db->host.hostname;
+        break;
 
-    case PROXY_USERS_ACL:
-      acl_proxy_user= dynamic_element(&acl_proxy_users, idx, ACL_PROXY_USER*);
-      user= acl_proxy_user->get_user();
-      host= acl_proxy_user->get_host();
-      break;
+      case COLUMN_PRIVILEGES_HASH:
+      case PROC_PRIVILEGES_HASH:
+      case FUNC_PRIVILEGES_HASH:
+      case PACKAGE_SPEC_PRIVILEGES_HASH:
+      case PACKAGE_BODY_PRIVILEGES_HASH:
+        grant_name= (GRANT_NAME*) my_hash_element(grant_name_hash, idx);
+        user= grant_name->user;
+        host= grant_name->host.hostname;
+        break;
 
-    case ROLES_MAPPINGS_HASH:
-      role_grant_pair= (ROLE_GRANT_PAIR *) my_hash_element(roles_mappings_hash, idx);
-      user= role_grant_pair->u_uname;
-      host= role_grant_pair->u_hname;
-      break;
+      case PROXY_USERS_ACL:
+        acl_proxy_user= dynamic_element(&acl_proxy_users, idx, ACL_PROXY_USER*);
+        user= acl_proxy_user->get_user();
+        host= acl_proxy_user->get_host();
+        break;
 
-    default:
-      DBUG_ASSERT(0);
-    }
-    if (! host)
-      host= "";
+      case ROLES_MAPPINGS_HASH:
+        role_grant_pair= (ROLE_GRANT_PAIR *) my_hash_element(roles_mappings_hash, idx);
+        user= role_grant_pair->u_uname;
+        host= role_grant_pair->u_hname;
+        break;
+
+      default:
+        DBUG_ASSERT(0);
+      }
+      if (! host)
+        host= "";
 
 #ifdef EXTRA_DEBUG
-    DBUG_PRINT("loop",("scan struct: %u  index: %u  user: '%s'  host: '%s'",
-                       struct_no, idx, user, host));
+      DBUG_PRINT("loop",("scan struct: %u  index: %u  user: '%s'  host: '%s'",
+                         struct_no, idx, user, host));
 #endif
 
-    if (struct_no == ROLES_MAPPINGS_HASH)
-    {
-      const char* role= role_grant_pair->r_uname? role_grant_pair->r_uname: "";
-      if (user_from->is_role())
+      if (struct_no == ROLES_MAPPINGS_HASH)
       {
-        /* When searching for roles within the ROLES_MAPPINGS_HASH, we have
-           to check both the user field as well as the role field for a match.
+        const char* role= role_grant_pair->r_uname? role_grant_pair->r_uname: "";
+        if (user_from->is_role())
+        {
+          /* When searching for roles within the ROLES_MAPPINGS_HASH, we have
+             to check both the user field as well as the role field for a match.
 
-           It is possible to have a role granted to a role. If we are going
-           to modify the mapping entry, it needs to be done on either on the
-           "user" end (here represented by a role) or the "role" end. At least
-           one part must match.
+             It is possible to have a role granted to a role. If we are going
+             to modify the mapping entry, it needs to be done on either on the
+             "user" end (here represented by a role) or the "role" end. At least
+             one part must match.
 
-           If the "user" end has a not-empty host string, it can never match
-           as we are searching for a role here. A role always has an empty host
-           string.
-        */
-        if ((*host || strcmp(user_from->user.str, user)) &&
-            strcmp(user_from->user.str, role))
-          continue;
+             If the "user" end has a not-empty host string, it can never match
+             as we are searching for a role here. A role always has an empty host
+             string.
+          */
+          if ((*host || strcmp(user_from->user.str, user)) &&
+              strcmp(user_from->user.str, role))
+            continue;
+        }
+        else
+        {
+          if (strcmp(user_from->user.str, user) ||
+              my_strcasecmp(system_charset_info, user_from->host.str, host))
+            continue;
+        }
       }
       else
       {
@@ -10400,157 +10439,134 @@ static int handle_grant_struct(enum enum_acl_lists struct_no, bool drop,
             my_strcasecmp(system_charset_info, user_from->host.str, host))
           continue;
       }
-    }
-    else
-    {
-      if (strcmp(user_from->user.str, user) ||
-          my_strcasecmp(system_charset_info, user_from->host.str, host))
-        continue;
-    }
 
-    result= 1; /* At least one element found. */
-    if ( drop )
-    {
-      elements--;
-      switch ( struct_no ) {
-      case USER_ACL:
-        free_acl_user(dynamic_element(&acl_users, idx, ACL_USER*));
-        delete_dynamic_element(&acl_users, idx);
-        break;
+      result= 1; /* At least one element found. */
+      if ( drop )
+      {
+        elements--;
+        switch ( struct_no ) {
+        case USER_ACL:
+          free_acl_user(dynamic_element(&acl_users, idx, ACL_USER*));
+          delete_dynamic_element(&acl_users, idx);
+          break;
 
-      case DB_ACL:
-        acl_dbs.del(idx);
-        break;
+        case DB_ACL:
+          acl_dbs.del(idx);
+          break;
 
-      case COLUMN_PRIVILEGES_HASH:
-      case PROC_PRIVILEGES_HASH:
-      case FUNC_PRIVILEGES_HASH:
-      case PACKAGE_SPEC_PRIVILEGES_HASH:
-      case PACKAGE_BODY_PRIVILEGES_HASH:
-        my_hash_delete(grant_name_hash, (uchar*) grant_name);
-        /*
-          In our HASH implementation on deletion one elements
-          is moved into a place where a deleted element was,
-          and the last element is moved into the empty space.
-          Thus we need to re-examine the current element, but
-          we don't have to restart the search from the beginning.
-        */
-        if (idx != elements)
-          idx++;
-	break;
+        case COLUMN_PRIVILEGES_HASH:
+        case PROC_PRIVILEGES_HASH:
+        case FUNC_PRIVILEGES_HASH:
+        case PACKAGE_SPEC_PRIVILEGES_HASH:
+        case PACKAGE_BODY_PRIVILEGES_HASH:
+          my_hash_delete(grant_name_hash, (uchar*) grant_name);
+          restart= true;
+          break;
 
-      case PROXY_USERS_ACL:
-        delete_dynamic_element(&acl_proxy_users, idx);
-        break;
+        case PROXY_USERS_ACL:
+          delete_dynamic_element(&acl_proxy_users, idx);
+          break;
 
-      case ROLES_MAPPINGS_HASH:
-        my_hash_delete(roles_mappings_hash, (uchar*) role_grant_pair);
-        if (idx != elements)
-          idx++;
-        break;
+        case ROLES_MAPPINGS_HASH:
+          my_hash_delete(roles_mappings_hash, (uchar*) role_grant_pair);
+          restart= true;
+          break;
 
-      default:
-        DBUG_ASSERT(0);
-        break;
+        default:
+          DBUG_ASSERT(0);
+          break;
+        }
       }
-    }
-    else if ( user_to )
-    {
-      switch ( struct_no ) {
-      case USER_ACL:
-        acl_user->user= safe_lexcstrdup_root(&acl_memroot, user_to->user);
-        update_hostname(&acl_user->host, strdup_root(&acl_memroot, user_to->host.str));
-        acl_user->hostname_length= strlen(acl_user->host.hostname);
-        break;
-
-      case DB_ACL:
-        acl_db->user= strdup_root(&acl_memroot, user_to->user.str);
-        update_hostname(&acl_db->host, strdup_root(&acl_memroot, user_to->host.str));
-        break;
+      else if ( user_to )
+      {
+        switch ( struct_no ) {
+        case USER_ACL:
+          acl_user->user= safe_lexcstrdup_root(&acl_memroot, user_to->user);
+          update_hostname(&acl_user->host, strdup_root(&acl_memroot, user_to->host.str));
+          acl_user->hostname_length= strlen(acl_user->host.hostname);
+          break;
 
-      case COLUMN_PRIVILEGES_HASH:
-      case PROC_PRIVILEGES_HASH:
-      case FUNC_PRIVILEGES_HASH:
-      case PACKAGE_SPEC_PRIVILEGES_HASH:
-      case PACKAGE_BODY_PRIVILEGES_HASH:
-        {
-          /*
-            Save old hash key and its length to be able to properly update
-            element position in hash.
-          */
-          char *old_key= grant_name->hash_key;
-          size_t old_key_length= grant_name->key_length;
+        case DB_ACL:
+          acl_db->user= strdup_root(&acl_memroot, user_to->user.str);
+          update_hostname(&acl_db->host, strdup_root(&acl_memroot, user_to->host.str));
+          break;
 
-          /*
-            Update the grant structure with the new user name and host name.
-          */
-          grant_name->set_user_details(user_to->host.str, grant_name->db,
-                                       user_to->user.str, grant_name->tname,
-                                       TRUE);
+        case COLUMN_PRIVILEGES_HASH:
+        case PROC_PRIVILEGES_HASH:
+        case FUNC_PRIVILEGES_HASH:
+        case PACKAGE_SPEC_PRIVILEGES_HASH:
+        case PACKAGE_BODY_PRIVILEGES_HASH:
+          {
+            /*
+              Save old hash key and its length to be able to properly update
+              element position in hash.
+            */
+            char *old_key= grant_name->hash_key;
+            size_t old_key_length= grant_name->key_length;
+
+            /*
+              Update the grant structure with the new user name and host name.
+            */
+            grant_name->set_user_details(user_to->host.str, grant_name->db,
+                                         user_to->user.str, grant_name->tname,
+                                         TRUE);
+
+            /*
+              Since username is part of the hash key, when the user name
+              is renamed, the hash key is changed. Update the hash to
+              ensure that the position matches the new hash key value
+            */
+            my_hash_update(grant_name_hash, (uchar*) grant_name, (uchar*) old_key,
+                           old_key_length);
+            restart= true;
+            break;
+          }
 
-          /*
-            Since username is part of the hash key, when the user name
-            is renamed, the hash key is changed. Update the hash to
-            ensure that the position matches the new hash key value
-          */
-          my_hash_update(grant_name_hash, (uchar*) grant_name, (uchar*) old_key,
-                         old_key_length);
-          /*
-            hash_update() operation could have moved element from the tail or
-            the head of the hash to the current position.  But it can never
-            move an element from the head to the tail or from the tail to the
-            head over the current element.
-            So we need to examine the current element once again, but
-            we don't need to restart the search from the beginning.
-          */
-          idx++;
+        case PROXY_USERS_ACL:
+          acl_proxy_user->set_user (&acl_memroot, user_to->user.str);
+          acl_proxy_user->set_host (&acl_memroot, user_to->host.str);
           break;
-        }
 
-      case PROXY_USERS_ACL:
-        acl_proxy_user->set_user (&acl_memroot, user_to->user.str);
-        acl_proxy_user->set_host (&acl_memroot, user_to->host.str);
-        break;
+        case ROLES_MAPPINGS_HASH:
+          {
+            /*
+              Save old hash key and its length to be able to properly update
+              element position in hash.
+            */
+            char *old_key= role_grant_pair->hashkey.str;
+            size_t old_key_length= role_grant_pair->hashkey.length;
+            bool oom;
+
+            if (user_to->is_role())
+              oom= role_grant_pair->init(&acl_memroot, role_grant_pair->u_uname,
+                                         role_grant_pair->u_hname,
+                                         user_to->user.str, false);
+            else
+              oom= role_grant_pair->init(&acl_memroot, user_to->user.str,
+                                         user_to->host.str,
+                                         role_grant_pair->r_uname, false);
+            if (oom)
+              DBUG_RETURN(-1);
+
+            my_hash_update(roles_mappings_hash, (uchar*) role_grant_pair,
+                           (uchar*) old_key, old_key_length);
+            restart= true;
+            break;
+          }
 
-      case ROLES_MAPPINGS_HASH:
-        {
-          /*
-            Save old hash key and its length to be able to properly update
-            element position in hash.
-          */
-          char *old_key= role_grant_pair->hashkey.str;
-          size_t old_key_length= role_grant_pair->hashkey.length;
-          bool oom;
-
-          if (user_to->is_role())
-            oom= role_grant_pair->init(&acl_memroot, role_grant_pair->u_uname,
-                                       role_grant_pair->u_hname,
-                                       user_to->user.str, false);
-          else
-            oom= role_grant_pair->init(&acl_memroot, user_to->user.str,
-                                       user_to->host.str,
-                                       role_grant_pair->r_uname, false);
-          if (oom)
-            DBUG_RETURN(-1);
-
-          my_hash_update(roles_mappings_hash, (uchar*) role_grant_pair,
-                         (uchar*) old_key, old_key_length);
-          idx++; // see the comment above
+        default:
+          DBUG_ASSERT(0);
           break;
         }
 
-      default:
-        DBUG_ASSERT(0);
+      }
+      else
+      {
+        /* If search is requested, we do not need to search further. */
         break;
       }
-
     }
-    else
-    {
-      /* If search is requested, we do not need to search further. */
-      break;
-    }
-  }
+  } while (restart);
 #ifdef EXTRA_DEBUG
   DBUG_PRINT("loop",("scan struct: %u  result %d", struct_no, result));
 #endif
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 1a1eead9fdc..833492d3d9f 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -796,6 +796,9 @@ retry:
   if (!(handler= mysql_ha_find_handler(thd, &tables->alias)))
     goto err0;
 
+  if (thd->transaction->xid_state.check_has_uncommitted_xa())
+    goto err0;
+
   table= handler->table;
   tables->table= table;                         // This is used by fix_fields
   table->pos_in_table_list= tables;
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 7bcff12a735..90fdce1b56f 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -3959,6 +3959,16 @@ err:
   mi->unlock_slave_threads();
   if (ret == FALSE)
     my_ok(thd);
+  else
+  {
+    /*
+      Depending on where CHANGE MASTER failed, the logs may be waiting to be
+      reopened. This would break future log updates and CHANGE MASTER calls.
+      `try_fix_log_state()` allows the relay log to fix its state to no longer
+      expect to be reopened.
+    */
+    mi->rli.relay_log.try_fix_log_state();
+  }
   DBUG_RETURN(ret);
 }
 
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 3aa0ed629e9..a7f0a0dd52a 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -28244,6 +28244,11 @@ void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
 
   //Item List
   bool first= 1;
+  /*
+    outer_select() can not be used here because it is for name resolution
+    and will return NULL at any end of name resolution chain (view/derived)
+  */
+  bool top_level= (get_master()->get_master() == 0);
   List_iterator_fast<Item> it(item_list);
   Item *item;
   while ((item= it++))
@@ -28253,7 +28258,8 @@ void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
     else
       str->append(',');
 
-    if (is_subquery_function() && !item->is_explicit_name())
+    if ((is_subquery_function() && !item->is_explicit_name()) ||
+        !item->name.str)
     {
       /*
         Do not print auto-generated aliases in subqueries. It has no purpose
@@ -28262,7 +28268,20 @@ void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
       item->print(str, query_type);
     }
     else
-      item->print_item_w_name(str, query_type);
+    {
+      /*
+        Do not print illegal names (if it is not top level SELECT).
+        Top level view checked (and correct name are assigned),
+        other cases of top level SELECT are not important, because
+        it is not "table field".
+      */
+      if (top_level ||
+          item->is_explicit_name() ||
+          !check_column_name(item->name.str))
+        item->print_item_w_name(str, query_type);
+      else
+        item->print(str, query_type);
+    }
   }
 
   /*
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 546de590ec1..dad9e34b87d 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -4479,7 +4479,9 @@ make_table_name_list(THD *thd, Dynamic_array<LEX_CSTRING*> *table_names,
   if (!lookup_field_vals->wild_table_value &&
       lookup_field_vals->table_value.str)
   {
-    if (lookup_field_vals->table_value.length > NAME_LEN)
+    if (check_table_name(lookup_field_vals->table_value.str,
+                         lookup_field_vals->table_value.length,
+                         false))
     {
       /*
         Impossible value for a table name,
@@ -4516,6 +4518,9 @@ make_table_name_list(THD *thd, Dynamic_array<LEX_CSTRING*> *table_names,
     return (schema_tables_add(thd, table_names,
                               lookup_field_vals->table_value.str));
 
+  if (check_db_name((LEX_STRING*)db_name))
+    return 0; // Impossible TABLE_SCHEMA name
+
   find_files_result res= find_files(thd, table_names, db_name, path,
                                     &lookup_field_vals->table_value);
   if (res != FIND_FILES_OK)
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 882da2b6705..efce4172f62 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -3246,8 +3246,6 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
               /* Align key length to multibyte char boundary */
               key_part_length-= key_part_length % sql_field->charset->mbmaxlen;
             }
-            else
-              is_hash_field_needed= true;
           }
         }
         // Catch invalid use of partial keys 
@@ -3293,11 +3291,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         }
         else
         {
-          if (key->type == Key::UNIQUE)
-          {
-            is_hash_field_needed= true;
-          }
-          else
+          if (key->type != Key::UNIQUE)
           {
             key_part_length= MY_MIN(max_key_length, file->max_key_part_length());
             my_error(ER_TOO_LONG_KEY, MYF(0), key_part_length);
@@ -3306,6 +3300,11 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         }
       }
 
+      if (key->type == Key::UNIQUE
+          && key_part_length > MY_MIN(max_key_length,
+                                      file->max_key_part_length()))
+        is_hash_field_needed= true;
+
       /* We can not store key_part_length more then 2^16 - 1 in frm */
       if (is_hash_field_needed && column->length > UINT_MAX16)
       {
@@ -3340,7 +3339,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
       unique_key=1;
     key_info->key_length=(uint16) key_length;
     if (key_info->key_length > max_key_length && key->type == Key::UNIQUE)
-      is_hash_field_needed= true;
+      is_hash_field_needed= true;  // for case "a BLOB UNIQUE"
     if (key_length > max_key_length && key->type != Key::FULLTEXT &&
         !is_hash_field_needed)
     {
diff --git a/sql/table.cc b/sql/table.cc
index 9d891ce413a..df94cac6475 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -5040,6 +5040,21 @@ bool check_table_name(const char *name, size_t length, bool check_for_path_chars
     if (check_for_path_chars &&
         (*name == '/' || *name == '\\' || *name == '~' || *name == FN_EXTCHAR))
       return 1;
+    /*
+      We don't allow zero byte in table/schema names:
+      - Some code still uses NULL-terminated strings.
+        Zero bytes will confuse this code.
+      - There is a little practical use of zero bytes in names anyway.
+      Note, if the string passed as "name" comes here
+      from the parser as an identifier, it does not contain zero bytes,
+      as the parser rejects zero bytes in identifiers.
+      But "name" can also come here from queries like this:
+        SELECT * FROM I_S.TABLES WHERE TABLE_NAME='str';
+      In this case "name" is a general string expression
+      and it can have any arbitrary bytes, including zero bytes.
+    */
+    if (*name == 0x00)
+      return 1;
     name++;
     name_length++;
   }