Merge with MySQL 5.1.53

Open issues:
- A better fix for #57688; Igor is working on this
- Test failure in index_merge_innodb.test ; Igor promised to look at this
- Some Innodb tests fails (need to merge with latest xtradb) ; Kristian promised to look at this.
 - Failing tests: innodb_plugin.innodb_bug56143 innodb_plugin.innodb_bug56632 innodb_plugin.innodb_bug56680 innodb_plugin.innodb_bug57255 
- Werror is disabled;  Should be enabled after merge with xtradb.

32 files changed, 336 insertions, 133 deletions

diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index ceb60d127cf..131946c2494 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -6269,8 +6269,7 @@ void ha_ndbcluster::get_auto_increment(ulonglong offset, ulonglong increment,
     Ndb_tuple_id_range_guard g(m_share);
     if ((m_skip_auto_increment &&
          ndb->readAutoIncrementValue(m_table, g.range, auto_value)) ||
-        ndb->getAutoIncrementValue(m_table, g.range, auto_value, cache_size,
-                                    increment, offset))
+        ndb->getAutoIncrementValue(m_table, g.range, auto_value, cache_size, increment, offset))
     {
       if (--retries &&
           ndb->getNdbError().status == NdbError::TemporaryError)
@@ -9919,10 +9918,8 @@ bool ha_ndbcluster::check_if_incompatible_data(HA_CREATE_INFO *create_info,
   {
     Field *field= table->field[i];
     const NDBCOL *col= tab->getColumn(i);
-    if ((col->getStorageType() == NDB_STORAGETYPE_MEMORY &&
-         create_info->storage_media != HA_SM_MEMORY) ||
-        (col->getStorageType() == NDB_STORAGETYPE_DISK &&
-         create_info->storage_media != HA_SM_DISK))
+    if ((col->getStorageType() == NDB_STORAGETYPE_MEMORY && create_info->storage_media != HA_SM_MEMORY) ||
+        (col->getStorageType() == NDB_STORAGETYPE_DISK && create_info->storage_media != HA_SM_DISK))
     {
       DBUG_PRINT("info", ("Column storage media is changed"));
       DBUG_RETURN(COMPATIBLE_DATA_NO);
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index b28e270a61c..a08e0c93865 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -4266,8 +4266,12 @@ int ha_partition::index_read_idx_map(uchar *buf, uint index,
 
     get_partition_set(table, buf, index, &m_start_key, &m_part_spec);
 
-    /* How can it be more than one partition with the current use? */
-    DBUG_ASSERT(m_part_spec.start_part == m_part_spec.end_part);
+    /* 
+      We have either found exactly 1 partition
+      (in which case start_part == end_part)
+      or no matching partitions (start_part > end_part)
+    */
+    DBUG_ASSERT(m_part_spec.start_part >= m_part_spec.end_part);
 
     for (part= m_part_spec.start_part; part <= m_part_spec.end_part; part++)
     {
@@ -4502,6 +4506,7 @@ int ha_partition::partition_scan_set_up(uchar * buf, bool idx_read_flag)
       key not found.
     */
     DBUG_PRINT("info", ("scan with no partition to scan"));
+    table->status= STATUS_NOT_FOUND;
     DBUG_RETURN(HA_ERR_END_OF_FILE);
   }
   if (m_part_spec.start_part == m_part_spec.end_part)
@@ -4526,6 +4531,7 @@ int ha_partition::partition_scan_set_up(uchar * buf, bool idx_read_flag)
     if (start_part == MY_BIT_NONE)
     {
       DBUG_PRINT("info", ("scan with no partition to scan"));
+      table->status= STATUS_NOT_FOUND;
       DBUG_RETURN(HA_ERR_END_OF_FILE);
     }
     if (start_part > m_part_spec.start_part)
diff --git a/sql/handler.cc b/sql/handler.cc
index d45692e8465..5bba6034a4d 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1134,7 +1134,7 @@ int ha_commit_trans(THD *thd, bool all)
     uint rw_ha_count;
     bool rw_trans;
 
-    DBUG_EXECUTE_IF("crash_commit_before", abort(););
+    DBUG_EXECUTE_IF("crash_commit_before", DBUG_SUICIDE(););
 
     /* Close all cursors that can not survive COMMIT */
     if (is_real_trans)                          /* not a statement commit */
@@ -1186,7 +1186,7 @@ int ha_commit_trans(THD *thd, bool all)
         }
         status_var_increment(thd->status_var.ha_prepare_count);
       }
-      DBUG_EXECUTE_IF("crash_commit_after_prepare", DBUG_ABORT(););
+      DBUG_EXECUTE_IF("crash_commit_after_prepare", DBUG_SUICIDE(););
       if (error || (is_real_trans && xid &&
                     (error= !(cookie= tc_log->log_xid(thd, xid)))))
       {
@@ -1194,13 +1194,13 @@ int ha_commit_trans(THD *thd, bool all)
         error= 1;
         goto end;
       }
-      DBUG_EXECUTE_IF("crash_commit_after_log", DBUG_ABORT(););
+      DBUG_EXECUTE_IF("crash_commit_after_log", DBUG_SUICIDE(););
     }
     error=ha_commit_one_phase(thd, all) ? (cookie ? 2 : 1) : 0;
-    DBUG_EXECUTE_IF("crash_commit_before_unlog", DBUG_ABORT(););
+    DBUG_EXECUTE_IF("crash_commit_before_unlog", DBUG_SUICIDE(););
     if (cookie)
       tc_log->unlog(cookie, xid);
-    DBUG_EXECUTE_IF("crash_commit_after", DBUG_ABORT(););
+    DBUG_EXECUTE_IF("crash_commit_after", DBUG_SUICIDE(););
 end:
     if (rw_trans)
       start_waiting_global_read_lock(thd);
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 03a517ec33d..c35db7ad1cb 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -1355,9 +1355,13 @@ longlong Item_func_int_div::val_int()
     signal_divide_by_null();
     return 0;
   }
-  return (unsigned_flag ?
-	  (ulonglong) value / (ulonglong) val2 :
-	  value / val2);
+
+  if (unsigned_flag)
+    return  ((ulonglong) value / (ulonglong) val2);
+  else if (value == LONGLONG_MIN && val2 == -1)
+    return LONGLONG_MIN;
+  else
+    return value / val2;
 }
 
 
@@ -1391,9 +1395,9 @@ longlong Item_func_mod::int_op()
   if (args[0]->unsigned_flag)
     result= args[1]->unsigned_flag ? 
       ((ulonglong) value) % ((ulonglong) val2) : ((ulonglong) value) % val2;
-  else
-    result= args[1]->unsigned_flag ?
-      value % ((ulonglong) val2) : value % val2;
+  else result= args[1]->unsigned_flag ?
+         value % ((ulonglong) val2) :
+         (val2 == -1) ? 0 : value % val2;
 
   return result;
 }
@@ -5300,7 +5304,17 @@ void Item_func_match::init_search(bool no_order)
 
   /* Check if init_search() has been called before */
   if (ft_handler)
+  {
+    /*
+      We should reset ft_handler as it is cleaned up
+      on destruction of FT_SELECT object
+      (necessary in case of re-execution of subquery).
+      TODO: FT_SELECT should not clean up ft_handler.
+    */
+    if (join_key)
+      table->file->ft_handler= ft_handler;
     DBUG_VOID_RETURN;
+  }
 
   if (key == NO_SUCH_KEY)
   {
diff --git a/sql/item_func.h b/sql/item_func.h
index bdaa217d555..e45638472e4 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -1545,7 +1545,7 @@ public:
        join_key(0), ft_handler(0), table(0), master(0), concat_ws(0) { }
   void cleanup()
   {
-    DBUG_ENTER("Item_func_match");
+    DBUG_ENTER("Item_func_match::cleanup");
     Item_real_func::cleanup();
     if (!master && ft_handler)
       ft_handler->please->close_search(ft_handler);
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 4c90eceec99..686b5d5fea3 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -1014,6 +1014,20 @@ String *Item_func_insert::val_str(String *str)
   if ((length < 0) || (length > res->length()))
     length= res->length();
 
+  /*
+    There is one exception not handled (intentionaly) by the character set
+    aggregation code. If one string is strong side and is binary, and
+    another one is weak side and is a multi-byte character string,
+    then we need to operate on the second string in terms on bytes when
+    calling ::numchars() and ::charpos(), rather than in terms of characters.
+    Lets substitute its character set to binary.
+  */
+  if (collation.collation == &my_charset_bin)
+  {
+    res->set_charset(&my_charset_bin);
+    res2->set_charset(&my_charset_bin);
+  }
+
   /* start and length are now sufficiently valid to pass to charpos function */
    start= res->charpos((int) start);
    length= res->charpos((int) length, (uint32) start);
@@ -2515,6 +2529,20 @@ String *Item_func_rpad::val_str(String *str)
   /* Set here so that rest of code sees out-of-bound value as such. */
   if ((ulonglong) count > INT_MAX32)
     count= INT_MAX32;
+  /*
+    There is one exception not handled (intentionaly) by the character set
+    aggregation code. If one string is strong side and is binary, and
+    another one is weak side and is a multi-byte character string,
+    then we need to operate on the second string in terms on bytes when
+    calling ::numchars() and ::charpos(), rather than in terms of characters.
+    Lets substitute its character set to binary.
+  */
+  if (collation.collation == &my_charset_bin)
+  {
+    res->set_charset(&my_charset_bin);
+    rpad->set_charset(&my_charset_bin);
+  }
+
   if (count <= (res_char_length= res->numchars()))
   {						// String to pad is big enough
     res->length(res->charpos((int) count));	// Shorten result if longer
@@ -2617,6 +2645,20 @@ String *Item_func_lpad::val_str(String *str)
   if ((ulonglong) count > INT_MAX32)
     count= INT_MAX32;
 
+  /*
+    There is one exception not handled (intentionaly) by the character set
+    aggregation code. If one string is strong side and is binary, and
+    another one is weak side and is a multi-byte character string,
+    then we need to operate on the second string in terms on bytes when
+    calling ::numchars() and ::charpos(), rather than in terms of characters.
+    Lets substitute its character set to binary.
+  */
+  if (collation.collation == &my_charset_bin)
+  {
+    res->set_charset(&my_charset_bin);
+    pad->set_charset(&my_charset_bin);
+  }
+
   res_char_length= res->numchars();
 
   if (count <= res_char_length)
@@ -3094,6 +3136,7 @@ String* Item_func_inet_ntoa::val_str(String* str)
   if ((null_value= (args[0]->null_value || n > (ulonglong) LL(4294967295))))
     return 0;					// Null value
 
+  str->set_charset(collation.collation);
   str->length(0);
   int4store(buf,n);
 
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index cd854f012a8..2164d5ae070 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -1935,21 +1935,26 @@ int subselect_single_select_engine::exec()
       DBUG_RETURN(join->error ? join->error : 1);
     }
     if (!select_lex->uncacheable && thd->lex->describe && 
-        !(join->select_options & SELECT_DESCRIBE) && 
-        join->need_tmp)
+        !(join->select_options & SELECT_DESCRIBE))
     {
       item->update_used_tables();
       if (item->const_item())
       {
         /*
+          It's necessary to keep original JOIN table because
+          create_sort_index() function may overwrite original
+          JOIN_TAB::type and wrong optimization method can be
+          selected on re-execution.
+        */
+        select_lex->uncacheable|= UNCACHEABLE_EXPLAIN;
+        select_lex->master_unit()->uncacheable|= UNCACHEABLE_EXPLAIN;
+        /*
           Force join->join_tmp creation, because this subquery will be replaced
           by a simple select from the materialization temp table by optimize()
           called by EXPLAIN and we need to preserve the initial query structure
           so we can display it.
         */
-        select_lex->uncacheable|= UNCACHEABLE_EXPLAIN;
-        select_lex->master_unit()->uncacheable|= UNCACHEABLE_EXPLAIN;
-        if (join->init_save_join_tab())
+        if (join->need_tmp && join->init_save_join_tab())
           DBUG_RETURN(1);                        /* purecov: inspected */
       }
     }
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 0a51a0de5d7..9de1014afc3 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -3409,8 +3409,6 @@ String* Item_func_group_concat::val_str(String* str)
 
 void Item_func_group_concat::print(String *str, enum_query_type query_type)
 {
-  /* orig_args is not filled with valid values until fix_fields() */
-  Item **pargs= fixed ? orig_args : args;
   str->append(STRING_WITH_LEN("group_concat("));
   if (distinct)
     str->append(STRING_WITH_LEN("distinct "));
@@ -3418,7 +3416,7 @@ void Item_func_group_concat::print(String *str, enum_query_type query_type)
   {
     if (i)
       str->append(',');
-    pargs[i]->print(str, query_type);
+    orig_args[i]->print(str, query_type);
   }
   if (arg_count_order)
   {
@@ -3427,7 +3425,7 @@ void Item_func_group_concat::print(String *str, enum_query_type query_type)
     {
       if (i)
         str->append(',');
-      pargs[i + arg_count_field]->print(str, query_type);
+      orig_args[i + arg_count_field]->print(str, query_type);
       if (order[i]->asc)
         str->append(STRING_WITH_LEN(" ASC"));
       else
diff --git a/sql/lock.cc b/sql/lock.cc
index 566275c5ea2..e5ea85e7ce7 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -736,7 +736,6 @@ TABLE_LIST *mysql_lock_have_duplicate(THD *thd, TABLE_LIST *needle,
                                       TABLE_LIST *haystack)
 {
   MYSQL_LOCK            *mylock;
-  TABLE                 **lock_tables;
   TABLE                 *table;
   TABLE                 *table2;
   THR_LOCK_DATA         **lock_locks;
@@ -765,12 +764,11 @@ TABLE_LIST *mysql_lock_have_duplicate(THD *thd, TABLE_LIST *needle,
   if (mylock->table_count < 2)
     goto end;
 
-  lock_locks=  mylock->locks;
-  lock_tables= mylock->table;
+  lock_locks= mylock->locks;
 
   /* Prepare table related variables that don't change in loop. */
   DBUG_ASSERT((table->lock_position < mylock->table_count) &&
-              (table == lock_tables[table->lock_position]));
+              (table == mylock->table[table->lock_position]));
   table_lock_data= lock_locks + table->lock_data_start;
   end_data= table_lock_data + table->lock_count;
 
@@ -784,7 +782,7 @@ TABLE_LIST *mysql_lock_have_duplicate(THD *thd, TABLE_LIST *needle,
 
     /* All tables in list must be in lock. */
     DBUG_ASSERT((table2->lock_position < mylock->table_count) &&
-                (table2 == lock_tables[table2->lock_position]));
+                (table2 == mylock->table[table2->lock_position]));
 
     for (lock_data2=  lock_locks + table2->lock_data_start,
            end_data2= lock_data2 + table2->lock_count;
diff --git a/sql/log.cc b/sql/log.cc
index de5814d2b07..ce18c3e5edf 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1230,7 +1230,7 @@ void LOGGER::deactivate_log_handler(THD *thd, uint log_type)
     file_log= file_log_handler->get_mysql_log();
     break;
   default:
-    assert(0);                                  // Impossible
+    MY_ASSERT_UNREACHABLE();
   }
 
   if (!(*tmp_opt))
@@ -2639,7 +2639,7 @@ bool MYSQL_BIN_LOG::open(const char *log_name,
     sql_print_error("MSYQL_BIN_LOG::open failed to sync the index file.");
     DBUG_RETURN(1);
   }
-  DBUG_EXECUTE_IF("crash_create_non_critical_before_update_index", abort(););
+  DBUG_EXECUTE_IF("crash_create_non_critical_before_update_index", DBUG_SUICIDE(););
 #endif
 
   write_error= 0;
@@ -2736,7 +2736,7 @@ bool MYSQL_BIN_LOG::open(const char *log_name,
     if (write_file_name_to_index_file)
     {
 #ifdef HAVE_REPLICATION
-      DBUG_EXECUTE_IF("crash_create_critical_before_update_index", abort(););
+      DBUG_EXECUTE_IF("crash_create_critical_before_update_index", DBUG_SUICIDE(););
 #endif
 
       DBUG_ASSERT(my_b_inited(&index_file) != 0);
@@ -2755,7 +2755,7 @@ bool MYSQL_BIN_LOG::open(const char *log_name,
         goto err;
 
 #ifdef HAVE_REPLICATION
-      DBUG_EXECUTE_IF("crash_create_after_update_index", abort(););
+      DBUG_EXECUTE_IF("crash_create_after_update_index", DBUG_SUICIDE(););
 #endif
     }
   }
@@ -3207,7 +3207,7 @@ int MYSQL_BIN_LOG::purge_first_log(Relay_log_info* rli, bool included)
   /* Store where we are in the new file for the execution thread */
   flush_relay_log_info(rli);
 
-  DBUG_EXECUTE_IF("crash_before_purge_logs", abort(););
+  DBUG_EXECUTE_IF("crash_before_purge_logs", DBUG_SUICIDE(););
 
   pthread_mutex_lock(&rli->log_space_lock);
   rli->relay_log.purge_logs(to_purge_if_included, included,
@@ -3335,7 +3335,7 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
       break;
   }
 
-  DBUG_EXECUTE_IF("crash_purge_before_update_index", abort(););
+  DBUG_EXECUTE_IF("crash_purge_before_update_index", DBUG_SUICIDE(););
 
   if ((error= sync_purge_index_file()))
   {
@@ -3350,7 +3350,7 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
     goto err;
   }
 
-  DBUG_EXECUTE_IF("crash_purge_critical_after_update_index", abort(););
+  DBUG_EXECUTE_IF("crash_purge_critical_after_update_index", DBUG_SUICIDE(););
 
 err:
   /* Read each entry from purge_index_file and delete the file. */
@@ -3360,7 +3360,7 @@ err:
                     " that would be purged.");
   close_purge_index_file();
 
-  DBUG_EXECUTE_IF("crash_purge_non_critical_after_update_index", abort(););
+  DBUG_EXECUTE_IF("crash_purge_non_critical_after_update_index", DBUG_SUICIDE(););
 
   if (need_mutex)
     pthread_mutex_unlock(&LOCK_index);
@@ -4871,7 +4871,7 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event,
                           DBUG_PRINT("info", ("error writing binlog cache: %d",
                                                write_error));
                         DBUG_PRINT("info", ("crashing before writing xid"));
-                        abort();
+                        DBUG_SUICIDE();
                       });
 
       if ((write_error= write_cache(cache, false, false)))
@@ -4885,7 +4885,7 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event,
 
       if (flush_and_sync())
         goto err;
-      DBUG_EXECUTE_IF("half_binlogged_transaction", DBUG_ABORT(););
+      DBUG_EXECUTE_IF("half_binlogged_transaction", DBUG_SUICIDE(););
       if (cache->error)				// Error on read
       {
         sql_print_error(ER(ER_ERROR_ON_READ), cache->file_name, errno);
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 262e8cae705..3969b6b6f4d 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -144,16 +144,21 @@ static void inline slave_rows_error_report(enum loglevel level, int ha_error,
     len= my_snprintf(slider, buff_end - slider,
                      " %s, Error_code: %d;", err->msg, err->code);
   }
-  
-  rli->report(level, thd->is_error()? thd->main_da.sql_errno() : 0,
-              "Could not execute %s event on table %s.%s;"
-              "%s handler error %s; "
-              "the event's master log %s, end_log_pos %lu",
-              type, table->s->db.str,
-              table->s->table_name.str,
-              buff,
-              handler_error == NULL? "<unknown>" : handler_error,
-              log_name, pos);
+
+  if (ha_error != 0)
+    rli->report(level, thd->is_error() ? thd->main_da.sql_errno() : 0,
+                "Could not execute %s event on table %s.%s;"
+                "%s handler error %s; "
+                "the event's master log %s, end_log_pos %lu",
+                type, table->s->db.str, table->s->table_name.str,
+                buff, handler_error == NULL ? "<unknown>" : handler_error,
+                log_name, pos);
+  else
+    rli->report(level, thd->is_error() ? thd->main_da.sql_errno() : 0,
+                "Could not execute %s event on table %s.%s;"
+                "%s the event's master log %s, end_log_pos %lu",
+                type, table->s->db.str, table->s->table_name.str,
+                buff, log_name, pos);
 }
 #endif
 
@@ -1224,7 +1229,7 @@ Log_event* Log_event::read_log_event(const char* buf, uint event_len,
       break;
 #ifdef HAVE_REPLICATION
     case SLAVE_EVENT: /* can never happen (unused event) */
-      ev = new Slave_log_event(buf, event_len);
+      ev = new Slave_log_event(buf, event_len, description_event);
       break;
 #endif /* HAVE_REPLICATION */
     case CREATE_FILE_EVENT:
@@ -1312,8 +1317,10 @@ Log_event* Log_event::read_log_event(const char* buf, uint event_len,
     (because constructor is "void") ; so instead we leave the pointer we
     wanted to allocate (e.g. 'query') to 0 and we test it in is_valid().
     Same for Format_description_log_event, member 'post_header_len'.
+
+    SLAVE_EVENT is never used, so it should not be read ever.
   */
-  if (!ev || !ev->is_valid())
+  if (!ev || !ev->is_valid() || (event_type == SLAVE_EVENT))
   {
     DBUG_PRINT("error",("Found invalid event in binary log"));
 
@@ -2306,7 +2313,7 @@ bool Query_log_event::write(IO_CACHE* file)
     start+= 4;
   }
 
-  if (thd && thd->is_current_user_used())
+  if (thd && thd->need_binlog_invoker())
   {
     LEX_STRING user;
     LEX_STRING host;
@@ -5981,8 +5988,12 @@ void Slave_log_event::init_from_mem_pool(int data_size)
 
 
 /** This code is not used, so has not been updated to be format-tolerant. */
-Slave_log_event::Slave_log_event(const char* buf, uint event_len)
-  :Log_event(buf,0) /*unused event*/ ,mem_pool(0),master_host(0)
+/* We are using description_event so that slave does not crash on Log_event
+  constructor */
+Slave_log_event::Slave_log_event(const char* buf, 
+                                 uint event_len,
+                                 const Format_description_log_event* description_event)
+  :Log_event(buf,description_event),mem_pool(0),master_host(0)
 {
   if (event_len < LOG_EVENT_HEADER_LEN)
     return;
@@ -7653,7 +7664,6 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
                              get_type_str(),
                              RPL_LOG_NAME, (ulong) log_pos);
     thd->reset_current_stmt_binlog_row_based();
-    const_cast<Relay_log_info*>(rli)->cleanup_context(thd, error);
     thd->is_slave_error= 1;
     DBUG_RETURN(error);
   }
@@ -7684,14 +7694,12 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
     const_cast<Relay_log_info*>(rli)->last_event_start_time= my_time(0);
   }
 
-  if (get_flags(STMT_END_F))
-    if ((error= rows_event_stmt_cleanup(rli, thd)))
-      rli->report(ERROR_LEVEL, error,
-                  "Error in %s event: commit of row events failed, "
-                  "table `%s`.`%s`",
-                  get_type_str(), m_table->s->db.str,
-                  m_table->s->table_name.str);
-
+  if (get_flags(STMT_END_F) && (error= rows_event_stmt_cleanup(rli, thd)))
+    slave_rows_error_report(ERROR_LEVEL,
+                            thd->is_error() ? 0 : error,
+                            rli, thd, table,
+                            get_type_str(),
+                            RPL_LOG_NAME, (ulong) log_pos);
   DBUG_RETURN(error);
 }
 
@@ -8254,6 +8262,7 @@ int Table_map_log_event::do_apply_event(Relay_log_info const *rli)
                 m_field_metadata, m_field_metadata_size,
                 m_null_bits, m_flags);
     table_list->m_tabledef_valid= TRUE;
+    table_list->skip_temporary= 1;
 
     /*
       We record in the slave's information that the table should be
@@ -8358,7 +8367,7 @@ void Table_map_log_event::pack_info(Protocol *protocol)
 
 
 #ifdef MYSQL_CLIENT
-void Table_map_log_event::print(FILE *file, PRINT_EVENT_INFO *print_event_info)
+void Table_map_log_event::print(FILE *, PRINT_EVENT_INFO *print_event_info)
 {
   if (!print_event_info->short_form)
   {
diff --git a/sql/log_event.h b/sql/log_event.h
index 2a2b1807781..6df2507b9aa 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -979,9 +979,9 @@ public:
     return (void*) my_malloc((uint)size, MYF(MY_WME|MY_FAE));
   }
 
-  static void operator delete(void *ptr, size_t size)
+  static void operator delete(void *ptr, size_t)
   {
-    my_free((uchar*) ptr, MYF(MY_WME|MY_ALLOW_ZERO_PTR));
+    my_free(ptr, MYF(MY_WME|MY_ALLOW_ZERO_PTR));
   }
 
   /* Placement version of the above operators */
@@ -1782,7 +1782,9 @@ public:
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
 #endif
 
-  Slave_log_event(const char* buf, uint event_len);
+  Slave_log_event(const char* buf,
+                  uint event_len,
+                  const Format_description_log_event *description_event);
   ~Slave_log_event();
   int get_data_size();
   bool is_valid() const { return master_host != 0; }
diff --git a/sql/my_decimal.h b/sql/my_decimal.h
index d736bad9a4b..86f5db29ed7 100644
--- a/sql/my_decimal.h
+++ b/sql/my_decimal.h
@@ -308,7 +308,7 @@ int my_decimal2int(uint mask, const my_decimal *d, my_bool unsigned_flag,
 
 
 inline
-int my_decimal2double(uint mask, const my_decimal *d, double *result)
+int my_decimal2double(uint, const my_decimal *d, double *result)
 {
   /* No need to call check_result as this will always succeed */
   return decimal2double((decimal_t*) d, result);
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 5dedbfc3007..7905a0f52e5 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1064,7 +1064,8 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 		      char* packet, uint packet_length);
 void log_slow_statement(THD *thd);
 bool check_dup(const char *db, const char *name, TABLE_LIST *tables);
-bool compare_record(TABLE *table);
+bool records_are_comparable(const TABLE *table);
+bool compare_record(const TABLE *table);
 bool append_file_to_dir(THD *thd, const char **filename_ptr, 
                         const char *table_name);
 void wait_while_table_is_used(THD *thd, TABLE *table,
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index 2c1f581e156..5cafcf47086 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -94,6 +94,16 @@ public:
   */
   MYSQL_BIN_LOG relay_log;
   LOG_INFO linfo;
+
+  /*
+   cur_log
+     Pointer that either points at relay_log.get_log_file() or
+     &rli->cache_buf, depending on whether the log is hot or there was
+     the need to open a cold relay_log.
+
+   cache_buf 
+     IO_CACHE used when opening cold relay logs.
+   */
   IO_CACHE cache_buf,*cur_log;
 
   /* The following variables are safe to read any time */
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 9e5cf7ab7dd..3270976e6a0 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -2662,7 +2662,7 @@ bool update_sys_var_str_path(THD *thd, sys_var_str *var_str,
     file_log= logger.get_log_file_handler();
     break;
   default:
-    assert(0);                                  // Impossible
+    MY_ASSERT_UNREACHABLE();
   }
 
   if (!old_value)
diff --git a/sql/slave.cc b/sql/slave.cc
index b9138961c25..da0d54597dc 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -740,8 +740,17 @@ static bool sql_slave_killed(THD* thd, Relay_log_info* rli)
   DBUG_ASSERT(rli->slave_running == 1);// tracking buffer overrun
   if (abort_loop || thd->killed || rli->abort_slave)
   {
+    /*
+      The transaction should always be binlogged if OPTION_KEEP_LOG is set
+      (it implies that something can not be rolled back). And such case
+      should be regarded similarly as modifing a non-transactional table
+      because retrying of the transaction will lead to an error or inconsistency
+      as well.
+      Example: OPTION_KEEP_LOG is set if a temporary table is created or dropped.
+    */
     if (rli->abort_slave && rli->is_in_group() &&
-        thd->transaction.all.modified_non_trans_table)
+        (thd->transaction.all.modified_non_trans_table ||
+         (thd->options & OPTION_KEEP_LOG)))
       DBUG_RETURN(0);
     /*
       If we are in an unsafe situation (stopping could corrupt replication),
@@ -2304,7 +2313,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
           else
           {
             exec_res= 0;
-            end_trans(thd, ROLLBACK);
+            rli->cleanup_context(thd, 1);
             /* chance for concurrent connection to get more locks */
             safe_sleep(thd, min(rli->trans_retries, MAX_SLAVE_RETRY_PAUSE),
                        (CHECK_KILLED_FUNC)sql_slave_killed, (void*)rli);
@@ -3120,6 +3129,7 @@ the slave SQL thread with \"SLAVE START\". We stopped at log \
     request is detected only by the present function, not by events), so we
     must "proactively" clear playgrounds:
   */
+  thd->clear_error();
   rli->cleanup_context(thd, 1);
   /*
     Some extra safety, which should not been needed (normally, event deletion
@@ -4312,12 +4322,66 @@ static Log_event* next_event(Relay_log_info* rli)
         DBUG_ASSERT(rli->cur_log_fd == -1);
 
         /*
-          Read pointer has to be at the start since we are the only
-          reader.
-          We must keep the LOCK_log to read the 4 first bytes, as this is a hot
-          log (same as when we call read_log_event() above: for a hot log we
-          take the mutex).
+           When the SQL thread is [stopped and] (re)started the
+           following may happen:
+
+           1. Log was hot at stop time and remains hot at restart
+
+              SQL thread reads again from hot_log (SQL thread was
+              reading from the active log when it was stopped and the
+              very same log is still active on SQL thread restart).
+
+              In this case, my_b_seek is performed on cur_log, while
+              cur_log points to relay_log.get_log_file();
+
+           2. Log was hot at stop time but got cold before restart
+
+              The log was hot when SQL thread stopped, but it is not
+              anymore when the SQL thread restarts.
+
+              In this case, the SQL thread reopens the log, using
+              cache_buf, ie, cur_log points to &cache_buf, and thence
+              its coordinates are reset.
+
+           3. Log was already cold at stop time
+
+              The log was not hot when the SQL thread stopped, and, of
+              course, it will not be hot when it restarts.
+
+              In this case, the SQL thread opens the cold log again,
+              using cache_buf, ie, cur_log points to &cache_buf, and
+              thence its coordinates are reset.
+
+           4. Log was hot at stop time, DBA changes to previous cold
+              log and restarts SQL thread
+
+              The log was hot when the SQL thread was stopped, but the
+              user changed the coordinates of the SQL thread to
+              restart from a previous cold log.
+
+              In this case, at start time, cur_log points to a cold
+              log, opened using &cache_buf as cache, and coordinates
+              are reset. However, as it moves on to the next logs, it
+              will eventually reach the hot log. If the hot log is the
+              same at the time the SQL thread was stopped, then
+              coordinates were not reset - the cur_log will point to
+              relay_log.get_log_file(), and not a freshly opened
+              IO_CACHE through cache_buf. For this reason we need to
+              deploy a my_b_seek before calling check_binlog_magic at
+              this point of the code (see: BUG#55263 for more
+              details).
+          
+          NOTES: 
+            - We must keep the LOCK_log to read the 4 first bytes, as
+              this is a hot log (same as when we call read_log_event()
+              above: for a hot log we take the mutex).
+
+            - Because of scenario #4 above, we need to have a
+              my_b_seek here. Otherwise, we might hit the assertion
+              inside check_binlog_magic.
         */
+
+        my_b_seek(cur_log, (my_off_t) 0);
         if (check_binlog_magic(cur_log,&errmsg))
         {
           if (!hot_log) pthread_mutex_unlock(log_lock);
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 24f590f893a..dde81f1926f 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -1594,6 +1594,7 @@ bool change_password(THD *thd, const char *host, const char *user,
   /* Buffer should be extended when password length is extended. */
   char buff[512];
   ulong query_length;
+  bool save_binlog_row_based;
   uint new_password_len= (uint) strlen(new_password);
   bool result= 1;
   DBUG_ENTER("change_password");
@@ -1629,6 +1630,14 @@ bool change_password(THD *thd, const char *host, const char *user,
   if (!(table= open_ltable(thd, &tables, TL_WRITE, 0)))
     DBUG_RETURN(1);
 
+  /*
+    This statement will be replicated as a statement, even when using
+    row-based replication.  The flag will be reset at the end of the
+    statement.
+  */
+  if ((save_binlog_row_based= thd->current_stmt_binlog_row_based))
+    thd->clear_current_stmt_binlog_row_based();
+
   VOID(pthread_mutex_lock(&acl_cache->lock));
   ACL_USER *acl_user;
   if (!(acl_user= find_acl_user(host, user, TRUE)))
@@ -1666,6 +1675,12 @@ bool change_password(THD *thd, const char *host, const char *user,
   }
 end:
   close_thread_tables(thd);
+
+  /* Restore the state of binlog format */
+  DBUG_ASSERT(!thd->current_stmt_binlog_row_based);
+  if (save_binlog_row_based)
+    thd->set_current_stmt_binlog_row_based();
+
   DBUG_RETURN(result);
 }
 
@@ -5469,7 +5484,7 @@ static int handle_grant_struct(uint struct_no, bool drop,
       host= grant_name->host.hostname;
       break;
     default:
-      assert(0);
+      MY_ASSERT_UNREACHABLE();
     }
     if (! user)
       user= "";
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 2be9dcfb777..1bf287a97d6 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -3099,8 +3099,6 @@ bool reopen_table(TABLE *table)
   tmp.maybe_null=	table->maybe_null;
   tmp.status=		table->status;
 
-  tmp.s->table_map_id=  table->s->table_map_id;
-
   /* Get state */
   tmp.in_use=    	thd;
   tmp.reginfo.lock_type=table->reginfo.lock_type;
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index cad6e920b6c..fcb927ce7fe 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -749,7 +749,7 @@ THD::THD()
   thr_lock_owner_init(&main_lock_id, &lock_info);
 
   m_internal_handler= NULL;
-  current_user_used= FALSE;
+  m_binlog_invoker= FALSE;
   memset(&invoker_user, 0, sizeof(invoker_user));
   memset(&invoker_host, 0, sizeof(invoker_host));
 }
@@ -1314,7 +1314,7 @@ void THD::cleanup_after_query()
   where= THD::DEFAULT_WHERE;
   /* reset table map for multi-table update */
   table_map_for_update= 0;
-  clean_current_user_used();
+  m_binlog_invoker= FALSE;
 }
 
 
@@ -3350,7 +3350,7 @@ void THD::set_query(char *query_arg, uint32 query_length_arg)
 
 void THD::get_definer(LEX_USER *definer)
 {
-  set_current_user_used();
+  binlog_invoker();
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
   if (slave_thread && has_invoker())
   {
diff --git a/sql/sql_class.h b/sql/sql_class.h
index fd47de29a63..71dcd505a62 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2371,9 +2371,8 @@ public:
     Protected with LOCK_thd_data mutex.
   */
   void set_query(char *query_arg, uint32 query_length_arg);
-  void set_current_user_used() { current_user_used= TRUE; }
-  bool is_current_user_used() { return current_user_used; }
-  void clean_current_user_used() { current_user_used= FALSE; }
+  void binlog_invoker() { m_binlog_invoker= TRUE; }
+  bool need_binlog_invoker() { return m_binlog_invoker; }
   void get_definer(LEX_USER *definer);
   void set_invoker(const LEX_STRING *user, const LEX_STRING *host)
   {
@@ -2408,10 +2407,10 @@ private:
     statements or default definer is set in CREATE/ALTER SP, SF, Event,
     TRIGGER or VIEW statements.
 
-    Current user will be binlogged into Query_log_event if current_user_used
+    Current user will be binlogged into Query_log_event if m_binlog_invoker
     is TRUE; It will be stored into invoker_host and invoker_user by SQL thread.
    */
-  bool current_user_used;
+  bool m_binlog_invoker;
 
   /**
     It points to the invoker in the Query_log_event.
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 1dd659283ac..7aa48524b20 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -1197,6 +1197,12 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
       VOID(filename_to_tablename(file->name, table_list->table_name,
                                  MYSQL50_TABLE_NAME_PREFIX_LENGTH +
                                  strlen(file->name) + 1));
+
+      /* To be able to correctly look up the table in the table cache. */
+      if (lower_case_table_names)
+        table_list->table_name_length= my_casedn_str(files_charset_info,
+                                                     table_list->table_name);
+
       table_list->alias= table_list->table_name;	// If lower_case_table_names=2
       table_list->internal_tmp_table= is_prefix(file->name, tmp_file_prefix);
       /* Link into list */
diff --git a/sql/sql_help.cc b/sql/sql_help.cc
index 7eff6a5e0fa..795b8a3f464 100644
--- a/sql/sql_help.cc
+++ b/sql/sql_help.cc
@@ -687,7 +687,7 @@ bool mysqld_help(THD *thd, const char *mask)
 
   if (count_topics == 0)
   {
-    int key_id;
+    int UNINIT_VAR(key_id);
     if (!(select=
           prepare_select_for_name(thd,mask,mlen,tables,tables[3].table,
                                   used_fields[help_keyword_name].field,
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 37cc20a07e7..83735fd2f41 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1509,9 +1509,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
           table->file->adjust_next_insert_id_after_explicit_value(
             table->next_number_field->val_int());
         info->touched++;
-        if (((table->file->ha_table_flags() & HA_PARTIAL_COLUMN_READ) &&
-             !bitmap_is_subset(table->write_set, table->read_set)) ||
-            compare_record(table))
+        if (!records_are_comparable(table) || compare_record(table))
         {
           if ((error=table->file->ha_update_row(table->record[1],
                                                 table->record[0])) &&
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index db96b6fc9a2..57ee5686bed 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -27,6 +27,7 @@
 #include "sp_cache.h"
 #include "events.h"
 #include "sql_trigger.h"
+#include "debug_sync.h"
 
 #ifdef WITH_MARIA_STORAGE_ENGINE
 #include "../storage/maria/ha_maria.h"
@@ -3283,6 +3284,15 @@ end_with_restore_list:
       thd->first_successful_insert_id_in_cur_stmt=
         thd->first_successful_insert_id_in_prev_stmt;
 
+    DBUG_EXECUTE_IF("after_mysql_insert",
+                    {
+                      const char act[]=
+                        "now "
+                        "wait_for signal.continue";
+                      DBUG_ASSERT(opt_debug_sync_timeout > 0);
+                      DBUG_ASSERT(!debug_sync_set_action(current_thd,
+                                                         STRING_WITH_LEN(act)));
+                    };);
     break;
   }
   case SQLCOM_REPLACE_SELECT:
@@ -3938,6 +3948,10 @@ end_with_restore_list:
     if (check_access(thd, UPDATE_ACL, "mysql", 0, 1, 1, 0) &&
         check_global_access(thd,CREATE_USER_ACL))
       break;
+
+    /* Replicate current user as grantor */
+    thd->binlog_invoker();
+
     /* Conditionally writes to binlog */
     if (!(res = mysql_revoke_all(thd, lex->users_list)))
       my_ok(thd);
@@ -3958,6 +3972,9 @@ end_with_restore_list:
                      is_schema_db(select_lex->db) : 0))
       goto error;
 
+    /* Replicate current user as grantor */
+    thd->binlog_invoker();
+
     if (thd->security_ctx->user)              // If not replication
     {
       LEX_USER *user, *tmp_user;
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 4ff23713daa..71e5185012c 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -6786,8 +6786,8 @@ int get_part_iter_for_interval_via_mapping(partition_info *part_info,
     }
   }
   else
-    assert(0);
-  
+    MY_ASSERT_UNREACHABLE();
+
   can_match_multiple_values= (flags || !min_value || !max_value ||
                               memcmp(min_value, max_value, field_len));
   if (can_match_multiple_values &&
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 65c76c0792f..972f4f7610f 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1703,11 +1703,14 @@ JOIN::reinit()
     /* Reset effect of possible no_rows_in_result() */
     List_iterator_fast<Item> it(fields_list);
     Item *item;
-
     no_rows_in_result_called= 0;
     while ((item= it++))
       item->restore_to_before_no_rows_in_result();
-  }  
+  }
+
+  if (!(select_options & SELECT_DESCRIBE))
+    init_ftfuncs(thd, select_lex, test(order));
+
   DBUG_RETURN(0);
 }
 
@@ -2494,6 +2497,13 @@ mysql_select(THD *thd, Item ***rref_pointer_array,
 	{
 	  DBUG_RETURN(TRUE);
 	}
+        /*
+          Original join tabs might be overwritten at first
+          subselect execution. So we need to restore them.
+        */
+        Item_subselect *subselect= select_lex->master_unit()->item;
+        if (subselect && subselect->is_uncacheable() && join->reinit())
+          DBUG_RETURN(TRUE);
       }
       else
       {
@@ -13498,6 +13508,8 @@ static bool
 list_contains_unique_index(TABLE *table,
                           bool (*find_func) (Field *, void *), void *data)
 {
+  if (table->pos_in_table_list->outer_join)
+    return 0;
   for (uint keynr= 0; keynr < table->s->keys; keynr++)
   {
     if (keynr == table->s->primary_key ||
@@ -13511,7 +13523,7 @@ list_contains_unique_index(TABLE *table,
            key_part < key_part_end;
            key_part++)
       {
-        if (key_part->field->maybe_null() || 
+        if (key_part->field->real_maybe_null() || 
             !find_func(key_part->field, data))
           break;
       }
diff --git a/sql/sql_string.h b/sql/sql_string.h
index 6eb0d74fde9..d5d31c8db65 100644
--- a/sql/sql_string.h
+++ b/sql/sql_string.h
@@ -89,9 +89,13 @@ public:
   }
   static void *operator new(size_t size, MEM_ROOT *mem_root) throw ()
   { return (void*) alloc_root(mem_root, (uint) size); }
-  static void operator delete(void *ptr_arg,size_t size)
-  { TRASH(ptr_arg, size); }
-  static void operator delete(void *ptr_arg, MEM_ROOT *mem_root)
+  static void operator delete(void *ptr_arg, size_t size)
+  {
+    (void) ptr_arg;
+    (void) size;
+    TRASH(ptr_arg, size);
+  }
+  static void operator delete(void *, MEM_ROOT *)
   { /* never called */ }
   ~String() { free(); }
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 3760afc2b1f..8446ca46e3c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -3907,7 +3907,7 @@ bool mysql_create_table_no_lock(THD *thd,
       Then she could create the table. This case is pretty obscure and
       therefore we don't introduce a new error message only for it.
     */
-    if (get_cached_table_share(db, alias))
+    if (get_cached_table_share(db, table_name))
     {
       my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
       goto unlock_and_end;
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index 3b2ef92ab48..2cedce497b6 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -173,7 +173,6 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
   SELECT_LEX *sl, *first_sl= first_select();
   select_result *tmp_result;
   bool is_union_select;
-  TABLE *empty_table= 0;
   DBUG_ENTER("st_select_lex_unit::prepare");
 
   describe= test(additional_options & SELECT_DESCRIBE);
@@ -275,14 +274,6 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
       types= first_sl->item_list;
     else if (sl == first_sl)
     {
-      /*
-        We need to create an empty table object. It is used
-        to create tmp_table fields in Item_type_holder.
-        The main reason of this is that we can't create
-        field object without table.
-      */
-      DBUG_ASSERT(!empty_table);
-      empty_table= (TABLE*) thd->calloc(sizeof(TABLE));
       types.empty();
       List_iterator_fast<Item> it(sl->item_list);
       Item *item_tmp;
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 33834df3ca6..c569cda64c2 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -25,9 +25,26 @@
 #include "sql_trigger.h"
 #include "debug_sync.h"
 
-/* Return 0 if row hasn't changed */
 
-bool compare_record(TABLE *table)
+/**
+   True if the table's input and output record buffers are comparable using
+   compare_record(TABLE*).
+ */
+bool records_are_comparable(const TABLE *table) {
+  return ((table->file->ha_table_flags() & HA_PARTIAL_COLUMN_READ) == 0) ||
+    bitmap_is_subset(table->write_set, table->read_set);
+}
+
+
+/**
+   Compares the input and outbut record buffers of the table to see if a row
+   has changed.
+
+   @return true if row has changed.
+   @return false otherwise.
+*/
+
+bool compare_record(const TABLE *table)
 {
   if (table->s->can_cmp_whole_record)
     return cmp_record(table,record[1]);
@@ -578,9 +595,7 @@ int mysql_update(THD *thd,
     the table handler is returning all columns OR if
     if all updated columns are read
   */
-  can_compare_record= (!(table->file->ha_table_flags() &
-                         HA_PARTIAL_COLUMN_READ) ||
-                       bitmap_is_subset(table->write_set, table->read_set));
+  can_compare_record= records_are_comparable(table);
 
   while (!(error=info.read_record(&info)) && !thd->killed)
   {
@@ -1691,18 +1706,16 @@ bool multi_update::send_data(List<Item> &not_used_values)
     if (table->status & (STATUS_NULL_ROW | STATUS_UPDATED))
       continue;
 
-    /*
-      We can use compare_record() to optimize away updates if
-      the table handler is returning all columns OR if
-      if all updated columns are read
-    */
     if (table == table_to_update)
     {
+      /*
+        We can use compare_record() to optimize away updates if
+        the table handler is returning all columns OR if
+        if all updated columns are read
+      */
       bool can_compare_record;
-      can_compare_record= (!(table->file->ha_table_flags() &
-                             HA_PARTIAL_COLUMN_READ) ||
-                           bitmap_is_subset(table->write_set,
-                                            table->read_set));
+      can_compare_record= records_are_comparable(table);
+
       table->status|= STATUS_UPDATED;
       store_record(table,record[1]);
       if (fill_record_n_invoke_before_triggers(thd, *fields_for_table[offset],
@@ -1943,10 +1956,7 @@ int multi_update::do_updates()
     if ((local_error = tmp_table->file->ha_rnd_init(1)))
       goto err;
 
-    can_compare_record= (!(table->file->ha_table_flags() &
-                           HA_PARTIAL_COLUMN_READ) ||
-                         bitmap_is_subset(table->write_set,
-                                          table->read_set));
+    can_compare_record= records_are_comparable(table);
 
     for (;;)
     {
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index f7f1e30c907..77b5e92c051 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -11580,6 +11580,12 @@ user:
                                          system_charset_info, 0) ||
                 check_host_name(&$$->host))
               MYSQL_YYABORT;
+            /*
+              Convert hostname part of username to lowercase.
+              It's OK to use in-place lowercase as long as
+              the character set is utf8.
+            */
+            my_casedn_str(system_charset_info, $$->host.str);
           }
         | CURRENT_USER optional_braces
           {