Auto-merge from mysql-trunk.

57 files changed, 1051 insertions, 588 deletions

diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index 357575c5b47..a0e157b7806 100755
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -92,7 +92,7 @@ TARGET_LINK_LIBRARIES(sql ${MYSQLD_STATIC_PLUGIN_LIBS}
  
 
 IF(WIN32)
-  SET(MYSQLD_SOURCE main.cc nt_servc.cc nt_servc.h  message.rc)
+  SET(MYSQLD_SOURCE main.cc nt_servc.cc nt_servc.h message.rc)
 ELSE()
   SET(MYSQLD_SOURCE main.cc ${DTRACE_PROBES_ALL})
 ENDIF()
diff --git a/sql/authors.h b/sql/authors.h
index 555fe2ae43a..5de2659f098 100644
--- a/sql/authors.h
+++ b/sql/authors.h
@@ -80,6 +80,7 @@ struct show_table_authors_st show_table_authors[]= {
   { "Eric Herman", "Amsterdam, Netherlands", "Bug fixing - federated" },
   { "Andrey Hristov", "Walldorf, Germany", "Event scheduler (5.1)" },
   { "Alexander (Alexi) Ivanov", "St. Petersburg, Russia", "Replication" },
+  { "Mattias Jonsson", "Uppsala, Sweden", "Partitioning" },
   { "Alexander (Salle) Keremidarski", "Sofia, Bulgaria",
     "Bug fixing" },
   { "Mats Kindahl", "Storvreta, Sweden", "Replication" },
diff --git a/sql/events.cc b/sql/events.cc
index 96b86e6798f..a548bda53e2 100644
--- a/sql/events.cc
+++ b/sql/events.cc
@@ -276,7 +276,9 @@ create_query_string(THD *thd, String *buf)
   /* Append definer */
   append_definer(thd, buf, &(thd->lex->definer->user), &(thd->lex->definer->host));
   /* Append the left part of thd->query after "DEFINER" part */
-  if (buf->append(thd->lex->stmt_definition_begin))
+  if (buf->append(thd->lex->stmt_definition_begin,
+                  thd->lex->stmt_definition_end -
+                  thd->lex->stmt_definition_begin))
     return 1;
  
   return 0;
diff --git a/sql/field.cc b/sql/field.cc
index bd091f7eb57..fcabaeaa74d 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -8464,14 +8464,20 @@ bool Field_num::eq_def(Field *field)
 }
 
 
+/**
+  Check whether two numeric fields can be considered 'equal' for table
+  alteration purposes. Fields are equal if they are of the same type
+  and retain the same pack length.
+*/
+
 uint Field_num::is_equal(Create_field *new_field)
 {
   return ((new_field->sql_type == real_type()) &&
-	  ((new_field->flags & UNSIGNED_FLAG) == (uint) (flags &
-							 UNSIGNED_FLAG)) &&
+          ((new_field->flags & UNSIGNED_FLAG) == 
+           (uint) (flags & UNSIGNED_FLAG)) &&
 	  ((new_field->flags & AUTO_INCREMENT_FLAG) ==
 	   (uint) (flags & AUTO_INCREMENT_FLAG)) &&
-	  (new_field->length <= max_display_length()));
+          (new_field->pack_length == pack_length()));
 }
 
 
diff --git a/sql/field.h b/sql/field.h
index d1fcc40ff5d..66b13d02b89 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -685,6 +685,10 @@ public:
   int  store_decimal(const my_decimal *);
   int  store(const char *to,uint length,CHARSET_INFO *cs)=0;
   uint size_of() const { return sizeof(*this); }
+  uint repertoire(void) const
+  {
+    return my_charset_repertoire(field_charset);
+  }
   CHARSET_INFO *charset(void) const { return field_charset; }
   void set_charset(CHARSET_INFO *charset_arg) { field_charset= charset_arg; }
   enum Derivation derivation(void) const { return field_derivation; }
diff --git a/sql/field_conv.cc b/sql/field_conv.cc
index e3be33c0f3c..299865e6114 100644
--- a/sql/field_conv.cc
+++ b/sql/field_conv.cc
@@ -123,13 +123,18 @@ set_field_to_null(Field *field)
     return 0;
   }
   field->reset();
-  if (field->table->in_use->count_cuted_fields == CHECK_FIELD_WARN)
-  {
+  switch (field->table->in_use->count_cuted_fields) {
+  case CHECK_FIELD_WARN:
     field->set_warning(MYSQL_ERROR::WARN_LEVEL_WARN, WARN_DATA_TRUNCATED, 1);
+    /* fall through */
+  case CHECK_FIELD_IGNORE:
     return 0;
+  case CHECK_FIELD_ERROR_FOR_NULL:
+    if (!field->table->in_use->no_errors)
+      my_error(ER_BAD_NULL_ERROR, MYF(0), field->field_name);
+    return -1;
   }
-  if (!field->table->in_use->no_errors)
-    my_error(ER_BAD_NULL_ERROR, MYF(0), field->field_name);
+  DBUG_ASSERT(0); // impossible
   return -1;
 }
 
@@ -179,13 +184,18 @@ set_field_to_null_with_conversions(Field *field, bool no_conversions)
     field->table->auto_increment_field_not_null= FALSE;
     return 0;				  // field is set in fill_record()
   }
-  if (field->table->in_use->count_cuted_fields == CHECK_FIELD_WARN)
-  {
+  switch (field->table->in_use->count_cuted_fields) {
+  case CHECK_FIELD_WARN:
     field->set_warning(MYSQL_ERROR::WARN_LEVEL_WARN, ER_BAD_NULL_ERROR, 1);
+    /* fall through */
+  case CHECK_FIELD_IGNORE:
     return 0;
+  case CHECK_FIELD_ERROR_FOR_NULL:
+    if (!field->table->in_use->no_errors)
+      my_error(ER_BAD_NULL_ERROR, MYF(0), field->field_name);
+    return -1;
   }
-  if (!field->table->in_use->no_errors)
-    my_error(ER_BAD_NULL_ERROR, MYF(0), field->field_name);
+  DBUG_ASSERT(0); // impossible
   return -1;
 }
 
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 86332f8c79c..2506e2fc8b3 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -61,6 +61,8 @@
 #include "sql_plugin.h"
 #include "table.h"                           /* HA_DATA_PARTITION */
 
+#include "debug_sync.h"
+
 static const char *ha_par_ext= ".par";
 #ifdef NOT_USED
 static int free_share(PARTITION_SHARE * share);
@@ -693,6 +695,7 @@ int ha_partition::rename_partitions(const char *path)
   DBUG_ASSERT(!strcmp(path, get_canonical_filename(m_file[0], path,
                                                    norm_name_buff)));
 
+  DEBUG_SYNC(ha_thd(), "before_rename_partitions");
   if (temp_partitions)
   {
     /*
@@ -2686,6 +2689,7 @@ int ha_partition::open(const char *name, int mode, uint test_if_locked)
   DBUG_RETURN(0);
 
 err_handler:
+  DEBUG_SYNC(ha_thd(), "partition_open_error");
   while (file-- != m_file)
     (*file)->close();
   bitmap_free(&m_bulk_insert_started);
diff --git a/sql/handler.cc b/sql/handler.cc
index 8b56921a3a1..844c7305825 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -174,7 +174,7 @@ redo:
 }
 
 
-plugin_ref ha_lock_engine(THD *thd, handlerton *hton)
+plugin_ref ha_lock_engine(THD *thd, const handlerton *hton)
 {
   if (hton)
   {
@@ -642,9 +642,13 @@ static my_bool closecon_handlerton(THD *thd, plugin_ref plugin,
     there's no need to rollback here as all transactions must
     be rolled back already
   */
-  if (hton->state == SHOW_OPTION_YES && hton->close_connection &&
-      thd_get_ha_data(thd, hton))
-    hton->close_connection(hton, thd);
+  if (hton->state == SHOW_OPTION_YES && thd_get_ha_data(thd, hton))
+  {
+    if (hton->close_connection)
+      hton->close_connection(hton, thd);
+    /* make sure ha_data is reset and ha_data_lock is released */
+    thd_set_ha_data(thd, hton, NULL);
+  }
   return FALSE;
 }
 
diff --git a/sql/handler.h b/sql/handler.h
index 1b5d5f8cb1f..ad26534d91d 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -2066,7 +2066,7 @@ extern ulong total_ha, total_ha_2pc;
 /* lookups */
 handlerton *ha_default_handlerton(THD *thd);
 plugin_ref ha_resolve_by_name(THD *thd, const LEX_STRING *name);
-plugin_ref ha_lock_engine(THD *thd, handlerton *hton);
+plugin_ref ha_lock_engine(THD *thd, const handlerton *hton);
 handlerton *ha_resolve_by_legacy_type(THD *thd, enum legacy_db_type db_type);
 handler *get_new_handler(TABLE_SHARE *share, MEM_ROOT *alloc,
                          handlerton *db_type);
diff --git a/sql/item.cc b/sql/item.cc
index 5fea7b9c075..5905c3ee090 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1816,7 +1816,16 @@ bool agg_item_set_converter(DTCollation &coll, const char *fname,
 
     if (!(conv= (*arg)->safe_charset_converter(coll.collation)) &&
         ((*arg)->collation.repertoire == MY_REPERTOIRE_ASCII))
-      conv= new Item_func_conv_charset(*arg, coll.collation, 1);
+    {
+      /*
+        We should disable const subselect item evaluation because
+        subselect transformation does not happen in view_prepare_mode
+        and thus val_...() methods can not be called for const items.
+      */
+      bool resolve_const= ((*arg)->type() == Item::SUBSELECT_ITEM &&
+                           thd->lex->view_prepare_mode) ? FALSE : TRUE;
+      conv= new Item_func_conv_charset(*arg, coll.collation, resolve_const);
+    }
 
     if (!conv)
     {
@@ -3785,7 +3794,7 @@ void Item_copy_decimal::copy()
 {
   my_decimal *nr= item->val_decimal(&cached_value);
   if (nr && nr != &cached_value)
-    memcpy (&cached_value, nr, sizeof (my_decimal)); 
+    my_decimal2decimal (nr, &cached_value);
   null_value= item->null_value;
 }
 
@@ -5330,14 +5339,22 @@ int Item_field::save_in_field(Field *to, bool no_conversions)
   if (result_field->is_null())
   {
     null_value=1;
-    res= set_field_to_null_with_conversions(to, no_conversions);
+    return set_field_to_null_with_conversions(to, no_conversions);
   }
-  else
+  to->set_notnull();
+
+  /*
+    If we're setting the same field as the one we're reading from there's 
+    nothing to do. This can happen in 'SET x = x' type of scenarios.
+  */  
+  if (to == result_field)
   {
-    to->set_notnull();
-    res= field_conv(to,result_field);
     null_value=0;
+    return 0;
   }
+
+  res= field_conv(to,result_field);
+  null_value=0;
   return res;
 }
 
@@ -5623,13 +5640,25 @@ inline uint char_val(char X)
 		 X-'a'+10);
 }
 
+Item_hex_string::Item_hex_string()
+{
+  hex_string_init("", 0);
+}
 
 Item_hex_string::Item_hex_string(const char *str, uint str_length)
 {
+  hex_string_init(str, str_length);
+}
+
+void Item_hex_string::hex_string_init(const char *str, uint str_length)
+{
   max_length=(str_length+1)/2;
   char *ptr=(char*) sql_alloc(max_length+1);
   if (!ptr)
+  {
+    str_value.set("", 0, &my_charset_bin);
     return;
+  }
   str_value.set(ptr,max_length,&my_charset_bin);
   char *end=ptr+max_length;
   if (max_length*2 != str_length)
diff --git a/sql/item.h b/sql/item.h
index 11b4199da2c..5f4f96f97d3 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -2302,7 +2302,7 @@ public:
 class Item_hex_string: public Item_basic_constant
 {
 public:
-  Item_hex_string() {}
+  Item_hex_string();
   Item_hex_string(const char *str,uint str_length);
   enum Type type() const { return VARBIN_ITEM; }
   double val_real()
@@ -2322,6 +2322,8 @@ public:
   bool eq(const Item *item, bool binary_cmp) const;
   virtual Item *safe_charset_converter(CHARSET_INFO *tocs);
   bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
+private:
+  void hex_string_init(const char *str, uint str_length);
 };
 
 
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index afd25688e79..a0b3f2c29a1 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -60,9 +60,9 @@ public:
   /* Allow owner function to use string buffers. */
   String value1, value2;
 
-  Arg_comparator(): thd(0), a_cache(0), b_cache(0), set_null(TRUE),
+  Arg_comparator(): comparators(0), thd(0), a_cache(0), b_cache(0), set_null(TRUE),
     get_value_a_func(0), get_value_b_func(0) {};
-  Arg_comparator(Item **a1, Item **a2): a(a1), b(a2), thd(0),
+  Arg_comparator(Item **a1, Item **a2): a(a1), b(a2), comparators(0), thd(0),
     a_cache(0), b_cache(0), set_null(TRUE),
     get_value_a_func(0), get_value_b_func(0) {};
 
@@ -118,6 +118,11 @@ public:
     return (owner->type() == Item::FUNC_ITEM &&
            ((Item_func*)owner)->functype() == Item_func::EQUAL_FUNC);
   }
+  void cleanup()
+  {
+    delete [] comparators;
+    comparators= 0;
+  }
 
   friend class Item_func;
 };
@@ -371,6 +376,11 @@ public:
   CHARSET_INFO *compare_collation() { return cmp.cmp_collation.collation; }
   uint decimal_precision() const { return 1; }
   void top_level_item() { abort_on_null= TRUE; }
+  void cleanup()
+  {
+    Item_int_func::cleanup();
+    cmp.cleanup();
+  }
 
   friend class  Arg_comparator;
 };
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index f2b34fe59ed..61febb01e93 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -512,7 +512,7 @@ String *Item_func_concat::val_str(String *str)
       }
       else if (str->alloced_length() >= res->length()+res2->length())
       {
-	if (str == res2)
+	if (str->ptr() == res2->ptr())
 	  str->replace(0,0,*res);
 	else
 	{
@@ -3192,8 +3192,7 @@ String *Item_load_file::val_str(String *str)
 		   MY_RELATIVE_PATH | MY_UNPACK_FILENAME);
 
   /* Read only allowed from within dir specified by secure_file_priv */
-  if (opt_secure_file_priv &&
-      strncmp(opt_secure_file_priv, path, strlen(opt_secure_file_priv)))
+  if (!is_secure_file_path(path))
     goto err;
 
   if (!mysql_file_stat(key_file_loadfile, path, &stat_info, MYF(0)))
diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h
index 7b7f8d3308a..2a34babae87 100644
--- a/sql/item_strfunc.h
+++ b/sql/item_strfunc.h
@@ -735,8 +735,9 @@ public:
   String *val_str(String *);
   void fix_length_and_dec()
   {
+    ulonglong max_result_length= (ulonglong) args[0]->max_length * 2 + 2;
+    max_length= (uint32) min(max_result_length, MAX_BLOB_WIDTH);
     collation.set(args[0]->collation);
-    max_length= args[0]->max_length * 2 + 2;
   }
 };
 
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index c341dd97460..10ef992594e 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -130,6 +130,21 @@ void Item_subselect::cleanup()
   DBUG_VOID_RETURN;
 }
 
+
+/*
+   We cannot use generic Item::safe_charset_converter() because
+   Subselect transformation does not happen in view_prepare_mode
+   and thus we can not evaluate val_...() for const items.
+*/
+
+Item *Item_subselect::safe_charset_converter(CHARSET_INFO *tocs)
+{
+  Item_func_conv_charset *conv=
+    new Item_func_conv_charset(this, tocs, thd->lex->view_prepare_mode ? 0 : 1);
+  return conv->safe ? conv : NULL;
+}
+
+
 void Item_singlerow_subselect::cleanup()
 {
   DBUG_ENTER("Item_singlerow_subselect::cleanup");
diff --git a/sql/item_subselect.h b/sql/item_subselect.h
index 04daab19adc..34b09ca6fdc 100644
--- a/sql/item_subselect.h
+++ b/sql/item_subselect.h
@@ -138,6 +138,7 @@ public:
   virtual void reset_value_registration() {}
   enum_parsing_place place() { return parsing_place; }
   bool walk(Item_processor processor, bool walk_subquery, uchar *arg);
+  Item *safe_charset_converter(CHARSET_INFO *tocs);
 
   /**
     Get the SELECT_LEX structure associated with this Item.
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 4746a6057c6..917acb0e908 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -3411,7 +3411,7 @@ void Item_func_group_concat::print(String *str, enum_query_type query_type)
     {
       if (i)
         str->append(',');
-      (*order[i]->item)->print(str, query_type);
+      pargs[i + arg_count_field]->print(str, query_type);
       if (order[i]->asc)
         str->append(STRING_WITH_LEN(" ASC"));
       else
diff --git a/sql/log.cc b/sql/log.cc
index 6c0ab33e87a..fd17e04b212 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1528,11 +1528,6 @@ binlog_flush_trx_cache(THD *thd, binlog_cache_mngr *cache_mngr,
                              cache_mngr->trx_cache.has_incident());
   cache_mngr->reset_cache(&cache_mngr->trx_cache);
 
-  /*
-    We need to step the table map version after writing the
-    transaction cache to disk.
-  */
-  mysql_bin_log.update_table_map_version();
   statistic_increment(binlog_cache_use, &LOCK_status);
   if (cache_log->disk_writes != 0)
   {
@@ -1592,13 +1587,6 @@ binlog_truncate_trx_cache(THD *thd, binlog_cache_mngr *cache_mngr, bool all)
   else
     cache_mngr->trx_cache.restore_prev_position();
 
-  /*
-    We need to step the table map version on a rollback to ensure that a new
-    table map event is generated instead of the one that was written to the
-    thrown-away transaction cache.
-  */
-  mysql_bin_log.update_table_map_version();
-
   DBUG_ASSERT(thd->binlog_get_pending_rows_event(is_transactional) == NULL);
   DBUG_RETURN(error);
 }
@@ -1650,11 +1638,6 @@ binlog_flush_stmt_cache(THD *thd, binlog_cache_mngr *cache_mngr)
     DBUG_RETURN(error);
   cache_mngr->reset_cache(&cache_mngr->stmt_cache);
 
-  /*
-    We need to step the table map version after writing the
-    transaction cache to disk.
-  */
-  mysql_bin_log.update_table_map_version();
   statistic_increment(binlog_cache_use, &LOCK_status);
   if (cache_log->disk_writes != 0)
   {
@@ -1890,12 +1873,14 @@ static int binlog_savepoint_set(handlerton *hton, THD *thd, void *sv)
   binlog_trans_log_savepos(thd, (my_off_t*) sv);
   /* Write it to the binary log */
 
+  String log_query;
+  if (log_query.append(STRING_WITH_LEN("SAVEPOINT ")) ||
+      log_query.append(thd->lex->ident.str, thd->lex->ident.length))
+    DBUG_RETURN(1);
   int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
-  int const error=
-    thd->binlog_query(THD::STMT_QUERY_TYPE,
-                      thd->query(), thd->query_length(), TRUE, FALSE, FALSE,
-                      errcode);
-  DBUG_RETURN(error);
+  Query_log_event qinfo(thd, log_query.c_ptr_safe(), log_query.length(),
+                        TRUE, FALSE, TRUE, errcode);
+  DBUG_RETURN(mysql_bin_log.write(&qinfo));
 }
 
 static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
@@ -1910,12 +1895,14 @@ static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
   if (unlikely(trans_has_updated_non_trans_table(thd) ||
                (thd->variables.option_bits & OPTION_KEEP_LOG)))
   {
+    String log_query;
+    if (log_query.append(STRING_WITH_LEN("ROLLBACK TO ")) ||
+        log_query.append(thd->lex->ident.str, thd->lex->ident.length))
+      DBUG_RETURN(1);
     int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
-    int error=
-      thd->binlog_query(THD::STMT_QUERY_TYPE,
-                        thd->query(), thd->query_length(), TRUE, FALSE, FALSE,
-                        errcode);
-    DBUG_RETURN(error);
+    Query_log_event qinfo(thd, log_query.c_ptr_safe(), log_query.length(),
+                          TRUE, FALSE, TRUE, errcode);
+    DBUG_RETURN(mysql_bin_log.write(&qinfo));
   }
   binlog_trans_log_truncate(thd, *(my_off_t*)sv);
   DBUG_RETURN(0);
@@ -2658,7 +2645,7 @@ const char *MYSQL_LOG::generate_name(const char *log_name,
 
 MYSQL_BIN_LOG::MYSQL_BIN_LOG(uint *sync_period)
   :bytes_written(0), prepared_xids(0), file_id(1), open_count(1),
-   need_start_event(TRUE), m_table_map_version(0),
+   need_start_event(TRUE),
    sync_period_ptr(sync_period),
    is_relay_log(0), signal_cnt(0),
    description_event_for_exec(0), description_event_for_queue(0)
@@ -4443,7 +4430,6 @@ int THD::binlog_write_table_map(TABLE *table, bool is_transactional)
     DBUG_RETURN(error);
 
   binlog_table_maps++;
-  table->s->table_map_version= mysql_bin_log.table_map_version();
   DBUG_RETURN(0);
 }
 
@@ -4585,21 +4571,6 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
       DBUG_RETURN(1);
     }
 
-    /*
-      We step the table map version if we are writing an event
-      representing the end of a statement.
-
-      In an ideal world, we could avoid stepping the table map version,
-      since we could then reuse the table map that was written earlier
-      in the cache. This does not work since STMT_END_F implies closing
-      all table mappings on the slave side.
-    
-      TODO: Find a solution so that table maps does not have to be
-      written several times within a transaction.
-    */
-    if (pending->get_flags(Rows_log_event::STMT_END_F))
-      ++m_table_map_version;
-
     delete pending;
   }
 
@@ -4657,7 +4628,9 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info)
     */
     const char *local_db= event_info->get_db();
     if ((thd && !(thd->variables.option_bits & OPTION_BIN_LOG)) ||
-	!binlog_filter->db_ok(local_db))
+	(thd->lex->sql_command != SQLCOM_ROLLBACK_TO_SAVEPOINT &&
+         thd->lex->sql_command != SQLCOM_SAVEPOINT &&
+         !binlog_filter->db_ok(local_db)))
       DBUG_RETURN(0);
 #endif /* HAVE_REPLICATION */
 
@@ -4786,9 +4759,6 @@ unlock:
     }
   }
 
-  if (event_info->flags & LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F)
-    ++m_table_map_version;
-
   DBUG_RETURN(error);
 }
 
diff --git a/sql/log.h b/sql/log.h
index 12e02969485..cd3faace598 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -299,8 +299,6 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
   */
   bool no_auto_events;
 
-  ulonglong m_table_map_version;
-
   /* pointer to the sync period variable, for binlog this will be
      sync_binlog_period, for relay log this will be
      sync_relay_log_period
@@ -353,13 +351,6 @@ public:
   void unlog(ulong cookie, my_xid xid);
   int recover(IO_CACHE *log, Format_description_log_event *fdle);
 #if !defined(MYSQL_CLIENT)
-  bool is_table_mapped(TABLE *table) const
-  {
-    return table->s->table_map_version == table_map_version();
-  }
-
-  ulonglong table_map_version() const { return m_table_map_version; }
-  void update_table_map_version() { ++m_table_map_version; }
 
   int flush_and_set_pending_rows_event(THD *thd, Rows_log_event* event,
                                        bool is_transactional);
diff --git a/sql/log_event.cc b/sql/log_event.cc
index d5ae1c954ff..db49cde6f03 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3189,10 +3189,7 @@ int Query_log_event::do_apply_event(Relay_log_info const *rli,
             ::do_apply_event(), then the companion SET also have so
             we don't need to reset_one_shot_variables().
   */
-  if (!strncmp(query_arg, "BEGIN", q_len_arg) ||
-      !strncmp(query_arg, "COMMIT", q_len_arg) ||
-      !strncmp(query_arg, "ROLLBACK", q_len_arg) ||
-      rpl_filter->db_ok(thd->db))
+  if (is_trans_keyword() || rpl_filter->db_ok(thd->db))
   {
     thd->set_time((time_t)when);
     thd->set_query_and_id((char*)query_arg, q_len_arg, next_query_id());
@@ -8941,11 +8938,28 @@ static bool record_compare(TABLE *table)
   {
     for (int i = 0 ; i < 2 ; ++i)
     {
-      saved_x[i]= table->record[i][0];
-      saved_filler[i]= table->record[i][table->s->null_bytes - 1];
-      table->record[i][0]|= 1U;
-      table->record[i][table->s->null_bytes - 1]|=
-        256U - (1U << table->s->last_null_bit_pos);
+      /* 
+        If we have an X bit then we need to take care of it.
+      */
+      if (!(table->s->db_options_in_use & HA_OPTION_PACK_RECORD))
+      {
+        saved_x[i]= table->record[i][0];
+        table->record[i][0]|= 1U;
+      }
+
+      /*
+         If (last_null_bit_pos == 0 && null_bytes > 1), then:
+
+         X bit (if any) + N nullable fields + M Field_bit fields = 8 bits 
+
+         Ie, the entire byte is used.
+      */
+      if (table->s->last_null_bit_pos > 0)
+      {
+        saved_filler[i]= table->record[i][table->s->null_bytes - 1];
+        table->record[i][table->s->null_bytes - 1]|=
+          256U - (1U << table->s->last_null_bit_pos);
+      }
     }
   }
 
@@ -8985,8 +8999,11 @@ record_compare_exit:
   {
     for (int i = 0 ; i < 2 ; ++i)
     {
-      table->record[i][0]= saved_x[i];
-      table->record[i][table->s->null_bytes - 1]= saved_filler[i];
+      if (!(table->s->db_options_in_use & HA_OPTION_PACK_RECORD))
+        table->record[i][0]= saved_x[i];
+
+      if (table->s->last_null_bit_pos)
+        table->record[i][table->s->null_bytes - 1]= saved_filler[i];
     }
   }
 
diff --git a/sql/log_event.h b/sql/log_event.h
index e281fd6e206..bd95c74b6c5 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -469,10 +469,10 @@ struct sql_ex_info
 #define LOG_EVENT_SUPPRESS_USE_F    0x8
 
 /*
-  The table map version internal to the log should be increased after
-  the event has been written to the binary log.
+  Note: this is a place holder for the flag
+  LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F (0x10), which is not used any
+  more, please do not reused this value for other flags.
  */
-#define LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F 0x10
 
 /**
    @def LOG_EVENT_ARTIFICIAL_F
@@ -1746,6 +1746,28 @@ public:        /* !!! Public in this patch to allow old usage */
                        const char *query_arg,
                        uint32 q_len_arg);
 #endif /* HAVE_REPLICATION */
+  /*
+    If true, the event always be applied by slave SQL thread or be printed by
+    mysqlbinlog
+   */
+  bool is_trans_keyword()
+  {
+    /*
+      Before the patch for bug#50407, The 'SAVEPOINT and ROLLBACK TO'
+      queries input by user was written into log events directly.
+      So the keywords can be written in both upper case and lower case
+      together, strncasecmp is used to check both cases. they also could be
+      binlogged with comments in the front of these keywords. for examples:
+        / * bla bla * / SAVEPOINT a;
+        / * bla bla * / ROLLBACK TO a;
+      but we don't handle these cases and after the patch, both quiries are
+      binlogged in upper case with no comments.
+     */
+    return !strncmp(query, "BEGIN", q_len) ||
+      !strncmp(query, "COMMIT", q_len) ||
+      !strncasecmp(query, "SAVEPOINT", 9) ||
+      !strncasecmp(query, "ROLLBACK", 8);
+  }
 };
 
 
diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index f6c5b5d1023..76eda43aa48 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -314,12 +314,29 @@ static bool record_compare(TABLE *table)
   if (table->s->null_bytes > 0)
   {
     for (int i = 0 ; i < 2 ; ++i)
-    {
-      saved_x[i]= table->record[i][0];
-      saved_filler[i]= table->record[i][table->s->null_bytes - 1];
-      table->record[i][0]|= 1U;
-      table->record[i][table->s->null_bytes - 1]|=
-        256U - (1U << table->s->last_null_bit_pos);
+    { 
+      /* 
+        If we have an X bit then we need to take care of it.
+      */
+      if (!(table->s->db_options_in_use & HA_OPTION_PACK_RECORD))
+      {
+        saved_x[i]= table->record[i][0];
+        table->record[i][0]|= 1U;
+      }
+      
+      /*
+         If (last_null_bit_pos == 0 && null_bytes > 1), then:
+
+         X bit (if any) + N nullable fields + M Field_bit fields = 8 bits 
+
+         Ie, the entire byte is used.
+      */
+      if (table->s->last_null_bit_pos > 0)
+      {
+        saved_filler[i]= table->record[i][table->s->null_bytes - 1];
+        table->record[i][table->s->null_bytes - 1]|=
+          256U - (1U << table->s->last_null_bit_pos);
+      }
     }
   }
 
@@ -359,8 +376,11 @@ record_compare_exit:
   {
     for (int i = 0 ; i < 2 ; ++i)
     {
-      table->record[i][0]= saved_x[i];
-      table->record[i][table->s->null_bytes - 1]= saved_filler[i];
+      if (!(table->s->db_options_in_use & HA_OPTION_PACK_RECORD))
+        table->record[i][0]= saved_x[i];
+
+      if (table->s->last_null_bit_pos > 0)
+        table->record[i][table->s->null_bytes - 1]= saved_filler[i];
     }
   }
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 89b601d07f9..f6a7ca08465 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -400,6 +400,7 @@ ulonglong log_output_options;
 my_bool opt_log_queries_not_using_indexes= 0;
 bool opt_error_log= IF_WIN(1,0);
 bool opt_disable_networking=0, opt_skip_show_db=0;
+bool opt_skip_name_resolve=0;
 my_bool opt_character_set_client_handshake= 1;
 bool server_id_supplied = 0;
 bool opt_endinfo, using_udf_functions;
@@ -6209,9 +6210,6 @@ Can't be set to 1 if --log-slave-updates is used.",
 #endif
   {"skip-host-cache", OPT_SKIP_HOST_CACHE, "Don't cache host names.", 0, 0, 0,
    GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
-  {"skip-name-resolve", OPT_SKIP_RESOLVE,
-   "Don't resolve hostnames. All hostnames are IP's or 'localhost'.",
-   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"skip-new", OPT_SKIP_NEW, "Don't use new, possibly wrong routines.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"skip-slave-start", 0,
@@ -6963,6 +6961,7 @@ static int mysql_init_variables(void)
   opt_log= opt_slow_log= 0;
   opt_bin_log= 0;
   opt_disable_networking= opt_skip_show_db=0;
+  opt_skip_name_resolve= 0;
   opt_ignore_builtin_innodb= 0;
   opt_logname= opt_update_logname= opt_binlog_index_name= opt_slow_logname= 0;
   opt_tc_log_file= (char *)"tc.log";      // no hostname in tc_log file name !
@@ -7332,6 +7331,7 @@ mysqld_get_one_option(int optid,
     opt_specialflag|= SPECIAL_NO_HOST_CACHE;
     break;
   case (int) OPT_SKIP_RESOLVE:
+    opt_skip_name_resolve= 1;
     opt_specialflag|=SPECIAL_NO_RESOLVE;
     break;
   case (int) OPT_WANT_CORE:
@@ -7709,6 +7709,48 @@ fn_format_relative_to_data_home(char * to, const char *name,
 }
 
 
+/**
+  Test a file path to determine if the path is compatible with the secure file
+  path restriction.
+ 
+  @param path null terminated character string
+
+  @return
+    @retval TRUE The path is secure
+    @retval FALSE The path isn't secure
+*/
+
+bool is_secure_file_path(char *path)
+{
+  char buff1[FN_REFLEN], buff2[FN_REFLEN];
+  /*
+    All paths are secure if opt_secure_file_path is 0
+  */
+  if (!opt_secure_file_priv)
+    return TRUE;
+
+  if (strlen(path) >= FN_REFLEN)
+    return FALSE;
+
+  if (my_realpath(buff1, path, 0))
+  {
+    /*
+      The supplied file path might have been a file and not a directory.
+    */
+    int length= (int)dirname_length(path);
+    if (length >= FN_REFLEN)
+      return FALSE;
+    memcpy(buff2, path, length);
+    buff2[length]= '\0';
+    if (length == 0 || my_realpath(buff1, buff2, 0))
+      return FALSE;
+  }
+  convert_dirname(buff2, buff1, NullS);
+  if (strncmp(opt_secure_file_priv, buff2, strlen(opt_secure_file_priv)))
+    return FALSE;
+  return TRUE;
+}
+
 static int fix_paths(void)
 {
   char buff[FN_REFLEN],*pos;
@@ -7769,10 +7811,26 @@ static int fix_paths(void)
    */
   if (opt_secure_file_priv)
   {
-    convert_dirname(buff, opt_secure_file_priv, NullS);
-    x_free(opt_secure_file_priv);
-    opt_secure_file_priv= my_strdup(buff, MYF(MY_FAE));
+    if (*opt_secure_file_priv == 0)
+    {
+      opt_secure_file_priv= 0;
+    }
+    else
+    {
+      if (strlen(opt_secure_file_priv) >= FN_REFLEN)
+        opt_secure_file_priv[FN_REFLEN-1]= '\0';
+      if (my_realpath(buff, opt_secure_file_priv, 0))
+      {
+        sql_print_warning("Failed to normalize the argument for --secure-file-priv.");
+        return 1;
+      }
+      char *secure_file_real_path= (char *)my_malloc(FN_REFLEN, MYF(MY_FAE));
+      convert_dirname(secure_file_real_path, buff, NullS);
+      my_free(opt_secure_file_priv, MYF(0));
+      opt_secure_file_priv= secure_file_real_path;
+    }
   }
+  
   return 0;
 }
 
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 842d74d063a..e14cd15ceb8 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -71,6 +71,7 @@ void unlink_thd(THD *thd);
 bool one_thread_per_connection_end(THD *thd, bool put_in_cache);
 void flush_thread_cache();
 void refresh_status(THD *thd);
+bool is_secure_file_path(char *path);
 
 extern MYSQL_PLUGIN_IMPORT CHARSET_INFO *system_charset_info;
 extern MYSQL_PLUGIN_IMPORT CHARSET_INFO *files_charset_info ;
@@ -97,6 +98,7 @@ extern ulonglong log_output_options;
 extern ulong log_backup_output_options;
 extern my_bool opt_log_queries_not_using_indexes;
 extern bool opt_disable_networking, opt_skip_show_db;
+extern bool opt_skip_name_resolve;
 extern bool opt_ignore_builtin_innodb;
 extern my_bool opt_character_set_client_handshake;
 extern bool volatile abort_loop;
diff --git a/sql/net_serv.cc b/sql/net_serv.cc
index fc8655ea2e7..28ed4cbdbaf 100644
--- a/sql/net_serv.cc
+++ b/sql/net_serv.cc
@@ -133,6 +133,9 @@ my_bool my_net_init(NET *net, Vio* vio)
   net->where_b = net->remain_in_buf=0;
   net->last_errno=0;
   net->unused= 0;
+#if defined(MYSQL_SERVER) && !defined(EMBEDDED_LIBRARY)
+  net->skip_big_packet= FALSE;
+#endif
 
   if (vio != 0)					/* If real connection */
   {
@@ -967,6 +970,7 @@ my_real_read(NET *net, size_t *complen)
 	  {
 #if defined(MYSQL_SERVER) && !defined(NO_ALARM)
 	    if (!net->compress &&
+                net->skip_big_packet &&
 		!my_net_skip_rest(net, (uint32) len, &alarmed, &alarm_buff))
 	      net->error= 3;		/* Successfully skiped packet */
 #endif
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index a46bfcf2d0e..5e985625c78 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -8717,8 +8717,6 @@ int QUICK_RANGE_SELECT::get_next()
 {
   int             result;
   KEY_MULTI_RANGE *mrange;
-  key_range       *start_key;
-  key_range       *end_key;
   DBUG_ENTER("QUICK_RANGE_SELECT::get_next");
   DBUG_ASSERT(multi_range_length && multi_range &&
               (cur_range >= (QUICK_RANGE**) ranges.buffer) &&
@@ -8758,26 +8756,9 @@ int QUICK_RANGE_SELECT::get_next()
          mrange_slot < mrange_end;
          mrange_slot++)
     {
-      start_key= &mrange_slot->start_key;
-      end_key= &mrange_slot->end_key;
       last_range= *(cur_range++);
-
-      start_key->key=    (const uchar*) last_range->min_key;
-      start_key->length= last_range->min_length;
-      start_key->flag=   ((last_range->flag & NEAR_MIN) ? HA_READ_AFTER_KEY :
-                          (last_range->flag & EQ_RANGE) ?
-                          HA_READ_KEY_EXACT : HA_READ_KEY_OR_NEXT);
-      start_key->keypart_map= last_range->min_keypart_map;
-      end_key->key=      (const uchar*) last_range->max_key;
-      end_key->length=   last_range->max_length;
-      /*
-        We use HA_READ_AFTER_KEY here because if we are reading on a key
-        prefix. We want to find all keys with this prefix.
-      */
-      end_key->flag=     (last_range->flag & NEAR_MAX ? HA_READ_BEFORE_KEY :
-                          HA_READ_AFTER_KEY);
-      end_key->keypart_map= last_range->max_keypart_map;
-
+      last_range->make_min_endpoint(&mrange_slot->start_key);
+      last_range->make_max_endpoint(&mrange_slot->end_key);
       mrange_slot->range_flag= last_range->flag;
     }
 
@@ -8801,49 +8782,52 @@ end:
 /*
   Get the next record with a different prefix.
 
-  SYNOPSIS
-    QUICK_RANGE_SELECT::get_next_prefix()
-    prefix_length  length of cur_prefix
-    cur_prefix     prefix of a key to be searched for
+  @param prefix_length   length of cur_prefix
+  @param group_key_parts The number of key parts in the group prefix
+  @param cur_prefix      prefix of a key to be searched for
 
-  DESCRIPTION
-    Each subsequent call to the method retrieves the first record that has a
-    prefix with length prefix_length different from cur_prefix, such that the
-    record with the new prefix is within the ranges described by
-    this->ranges. The record found is stored into the buffer pointed by
-    this->record.
-    The method is useful for GROUP-BY queries with range conditions to
-    discover the prefix of the next group that satisfies the range conditions.
+  Each subsequent call to the method retrieves the first record that has a
+  prefix with length prefix_length and which is different from cur_prefix,
+  such that the record with the new prefix is within the ranges described by
+  this->ranges. The record found is stored into the buffer pointed by
+  this->record. The method is useful for GROUP-BY queries with range
+  conditions to discover the prefix of the next group that satisfies the range
+  conditions.
+
+  @todo
 
-  TODO
     This method is a modified copy of QUICK_RANGE_SELECT::get_next(), so both
     methods should be unified into a more general one to reduce code
     duplication.
 
-  RETURN
-    0                  on success
-    HA_ERR_END_OF_FILE if returned all keys
-    other              if some error occurred
+  @retval 0                  on success
+  @retval HA_ERR_END_OF_FILE if returned all keys
+  @retval other              if some error occurred
 */
 
 int QUICK_RANGE_SELECT::get_next_prefix(uint prefix_length,
-                                        key_part_map keypart_map,
+                                        uint group_key_parts,
                                         uchar *cur_prefix)
 {
   DBUG_ENTER("QUICK_RANGE_SELECT::get_next_prefix");
+  const key_part_map keypart_map= make_prev_keypart_map(group_key_parts);
 
   for (;;)
   {
     int result;
-    key_range start_key, end_key;
     if (last_range)
     {
       /* Read the next record in the same range with prefix after cur_prefix. */
-      DBUG_ASSERT(cur_prefix != 0);
+      DBUG_ASSERT(cur_prefix != NULL);
       result= file->index_read_map(record, cur_prefix, keypart_map,
                                    HA_READ_AFTER_KEY);
-      if (result || (file->compare_key(file->end_range) <= 0))
+      if (result || last_range->max_keypart_map == 0)
         DBUG_RETURN(result);
+
+      key_range previous_endpoint;
+      last_range->make_max_endpoint(&previous_endpoint, prefix_length, keypart_map);
+      if (file->compare_key(&previous_endpoint) <= 0)
+        DBUG_RETURN(0);
     }
 
     uint count= ranges.elements - (cur_range - (QUICK_RANGE**) ranges.buffer);
@@ -8855,21 +8839,9 @@ int QUICK_RANGE_SELECT::get_next_prefix(uint prefix_length,
     }
     last_range= *(cur_range++);
 
-    start_key.key=    (const uchar*) last_range->min_key;
-    start_key.length= min(last_range->min_length, prefix_length);
-    start_key.keypart_map= last_range->min_keypart_map & keypart_map;
-    start_key.flag=   ((last_range->flag & NEAR_MIN) ? HA_READ_AFTER_KEY :
-		       (last_range->flag & EQ_RANGE) ?
-		       HA_READ_KEY_EXACT : HA_READ_KEY_OR_NEXT);
-    end_key.key=      (const uchar*) last_range->max_key;
-    end_key.length=   min(last_range->max_length, prefix_length);
-    end_key.keypart_map= last_range->max_keypart_map & keypart_map;
-    /*
-      We use READ_AFTER_KEY here because if we are reading on a key
-      prefix we want to find all keys with this prefix
-    */
-    end_key.flag=     (last_range->flag & NEAR_MAX ? HA_READ_BEFORE_KEY :
-		       HA_READ_AFTER_KEY);
+    key_range start_key, end_key;
+    last_range->make_min_endpoint(&start_key, prefix_length, keypart_map);
+    last_range->make_max_endpoint(&end_key, prefix_length, keypart_map);
 
     result= file->read_range_first(last_range->min_keypart_map ? &start_key : 0,
 				   last_range->max_keypart_map ? &end_key : 0,
@@ -8964,9 +8936,9 @@ bool QUICK_RANGE_SELECT::row_in_ranges()
 }
 
 /*
-  This is a hack: we inherit from QUICK_SELECT so that we can use the
+  This is a hack: we inherit from QUICK_RANGE_SELECT so that we can use the
   get_next() interface, but we have to hold a pointer to the original
-  QUICK_SELECT because its data are used all over the place.  What
+  QUICK_RANGE_SELECT because its data are used all over the place. What
   should be done is to factor out the data that is needed into a base
   class (QUICK_SELECT), and then have two subclasses (_ASC and _DESC)
   which handle the ranges and implement the get_next() function.  But
@@ -11159,7 +11131,8 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_prefix()
   {
     uchar *cur_prefix= seen_first_key ? group_prefix : NULL;
     if ((result= quick_prefix_select->get_next_prefix(group_prefix_len,
-                         make_prev_keypart_map(group_key_parts), cur_prefix)))
+                                                      group_key_parts, 
+                                                      cur_prefix)))
       DBUG_RETURN(result);
     seen_first_key= TRUE;
   }
diff --git a/sql/opt_range.h b/sql/opt_range.h
index 3b3b36f7689..85d59671b42 100644
--- a/sql/opt_range.h
+++ b/sql/opt_range.h
@@ -79,6 +79,85 @@ class QUICK_RANGE :public Sql_alloc {
       dummy=0;
 #endif
     }
+
+  /**
+     Initalizes a key_range object for communication with storage engine. 
+
+     This function facilitates communication with the Storage Engine API by
+     translating the minimum endpoint of the interval represented by this
+     QUICK_RANGE into an index range endpoint specifier for the engine.
+
+     @param Pointer to an uninitialized key_range C struct.
+
+     @param prefix_length The length of the search key prefix to be used for
+     lookup.
+     
+     @param keypart_map A set (bitmap) of keyparts to be used.
+  */
+  void make_min_endpoint(key_range *kr, uint prefix_length, 
+                         key_part_map keypart_map) {
+    make_min_endpoint(kr);
+    kr->length= min(kr->length, prefix_length);
+    kr->keypart_map&= keypart_map;
+  }
+  
+  /**
+     Initalizes a key_range object for communication with storage engine. 
+
+     This function facilitates communication with the Storage Engine API by
+     translating the minimum endpoint of the interval represented by this
+     QUICK_RANGE into an index range endpoint specifier for the engine.
+
+     @param Pointer to an uninitialized key_range C struct.
+  */
+  void make_min_endpoint(key_range *kr) {
+    kr->key= (const uchar*)min_key;
+    kr->length= min_length;
+    kr->keypart_map= min_keypart_map;
+    kr->flag= ((flag & NEAR_MIN) ? HA_READ_AFTER_KEY :
+               (flag & EQ_RANGE) ? HA_READ_KEY_EXACT : HA_READ_KEY_OR_NEXT);
+  }
+
+  /**
+     Initalizes a key_range object for communication with storage engine. 
+
+     This function facilitates communication with the Storage Engine API by
+     translating the maximum endpoint of the interval represented by this
+     QUICK_RANGE into an index range endpoint specifier for the engine.
+
+     @param Pointer to an uninitialized key_range C struct.
+
+     @param prefix_length The length of the search key prefix to be used for
+     lookup.
+     
+     @param keypart_map A set (bitmap) of keyparts to be used.
+  */
+  void make_max_endpoint(key_range *kr, uint prefix_length, 
+                         key_part_map keypart_map) {
+    make_max_endpoint(kr);
+    kr->length= min(kr->length, prefix_length);
+    kr->keypart_map&= keypart_map;
+  }
+
+  /**
+     Initalizes a key_range object for communication with storage engine. 
+
+     This function facilitates communication with the Storage Engine API by
+     translating the maximum endpoint of the interval represented by this
+     QUICK_RANGE into an index range endpoint specifier for the engine.
+
+     @param Pointer to an uninitialized key_range C struct.
+  */
+  void make_max_endpoint(key_range *kr) {
+    kr->key= (const uchar*)max_key;
+    kr->length= max_length;
+    kr->keypart_map= max_keypart_map;
+    /*
+      We use READ_AFTER_KEY here because if we are reading on a key
+      prefix we want to find all keys with this prefix
+    */
+    kr->flag= (flag & NEAR_MAX ? HA_READ_BEFORE_KEY : HA_READ_AFTER_KEY);
+  }
 };
 
 
@@ -345,7 +424,7 @@ public:
   int reset(void);
   int get_next();
   void range_end();
-  int get_next_prefix(uint prefix_length, key_part_map keypart_map,
+  int get_next_prefix(uint prefix_length, uint group_key_parts, 
                       uchar *cur_prefix);
   bool reverse_sorted() { return 0; }
   bool unique_key_range();
@@ -625,7 +704,7 @@ private:
   uchar *record;          /* Buffer where the next record is returned. */
   uchar *tmp_record;      /* Temporary storage for next_min(), next_max(). */
   uchar *group_prefix;    /* Key prefix consisting of the GROUP fields. */
-  uint group_prefix_len; /* Length of the group prefix. */
+  const uint group_prefix_len; /* Length of the group prefix. */
   uint group_key_parts;  /* A number of keyparts in the group prefix */
   uchar *last_prefix;     /* Prefix of the last group for detecting EOF. */
   bool have_min;         /* Specify whether we are computing */
diff --git a/sql/opt_sum.cc b/sql/opt_sum.cc
index fd2040a4979..0c79c8dc797 100644
--- a/sql/opt_sum.cc
+++ b/sql/opt_sum.cc
@@ -90,6 +90,123 @@ static ulonglong get_exact_record_count(TABLE_LIST *tables)
 
 
 /**
+  Use index to read MIN(field) value.
+  
+  @param table      Table object
+  @param ref        Reference to the structure where we store the key value
+  @item_field       Field used in MIN()
+  @range_fl         Whether range endpoint is strict less than
+  @prefix_len       Length of common key part for the range
+  
+  @retval
+    0               No errors
+    HA_ERR_...      Otherwise
+*/
+
+static int get_index_min_value(TABLE *table, TABLE_REF *ref,
+                               Item_field *item_field, uint range_fl,
+                               uint prefix_len)
+{
+  int error;
+  
+  if (!ref->key_length)
+    error= table->file->index_first(table->record[0]);
+  else 
+  {
+    /*
+      Use index to replace MIN/MAX functions with their values
+      according to the following rules:
+
+      1) Insert the minimum non-null values where the WHERE clause still
+         matches, or
+      2) a NULL value if there are only NULL values for key_part_k.
+      3) Fail, producing a row of nulls
+
+      Implementation: Read the smallest value using the search key. If
+      the interval is open, read the next value after the search
+      key. If read fails, and we're looking for a MIN() value for a
+      nullable column, test if there is an exact match for the key.
+    */
+    if (!(range_fl & NEAR_MIN))
+      /* 
+         Closed interval: Either The MIN argument is non-nullable, or
+         we have a >= predicate for the MIN argument.
+      */
+      error= table->file->index_read_map(table->record[0],
+                                         ref->key_buff,
+                                         make_prev_keypart_map(ref->key_parts),
+                                         HA_READ_KEY_OR_NEXT);
+    else
+    {
+      /*
+        Open interval: There are two cases:
+        1) We have only MIN() and the argument column is nullable, or
+        2) there is a > predicate on it, nullability is irrelevant.
+        We need to scan the next bigger record first.
+      */
+      error= table->file->index_read_map(table->record[0],
+                                         ref->key_buff,
+                                         make_prev_keypart_map(ref->key_parts),
+                                         HA_READ_AFTER_KEY);
+      /* 
+         If the found record is outside the group formed by the search
+         prefix, or there is no such record at all, check if all
+         records in that group have NULL in the MIN argument
+         column. If that is the case return that NULL.
+
+         Check if case 1 from above holds. If it does, we should read
+         the skipped tuple.
+      */
+      if (item_field->field->real_maybe_null() &&
+          ref->key_buff[prefix_len] == 1 &&
+          /*
+            Last keypart (i.e. the argument to MIN) is set to NULL by
+            find_key_for_maxmin only if all other keyparts are bound
+            to constants in a conjunction of equalities. Hence, we
+            can detect this by checking only if the last keypart is
+            NULL.
+          */
+          (error == HA_ERR_KEY_NOT_FOUND ||
+           key_cmp_if_same(table, ref->key_buff, ref->key, prefix_len)))
+      {
+        DBUG_ASSERT(item_field->field->real_maybe_null());
+        error= table->file->index_read_map(table->record[0],
+                                           ref->key_buff,
+                                           make_prev_keypart_map(ref->key_parts),
+                                           HA_READ_KEY_EXACT);
+      }
+    }
+  }
+  return error;
+}
+
+
+/**
+  Use index to read MAX(field) value.
+  
+  @param table      Table object
+  @param ref        Reference to the structure where we store the key value
+  @range_fl         Whether range endpoint is strict greater than
+  
+  @retval
+    0               No errors
+    HA_ERR_...      Otherwise
+*/
+
+static int get_index_max_value(TABLE *table, TABLE_REF *ref, uint range_fl)
+{
+  return (ref->key_length ?
+          table->file->index_read_map(table->record[0], ref->key_buff,
+                                      make_prev_keypart_map(ref->key_parts),
+                                      range_fl & NEAR_MAX ?
+                                      HA_READ_BEFORE_KEY : 
+                                      HA_READ_PREFIX_LAST_OR_PREV) :
+          table->file->index_last(table->record[0]));
+}
+
+
+
+/**
   Substitutes constants for some COUNT(), MIN() and MAX() functions.
 
   @param tables                list of leaves of join table tree
@@ -220,9 +337,11 @@ int opt_sum_query(TABLE_LIST *tables, List<Item> &all_fields,COND *conds)
           const_result= 0;
         break;
       case Item_sum::MIN_FUNC:
+      case Item_sum::MAX_FUNC:
       {
+        int is_max= test(item_sum->sum_func() == Item_sum::MAX_FUNC);
         /*
-          If MIN(expr) is the first part of a key or if all previous
+          If MIN/MAX(expr) is the first part of a key or if all previous
           parts of the key is found in the COND, then we can use
           indexes to find the key.
         */
@@ -241,89 +360,26 @@ int opt_sum_query(TABLE_LIST *tables, List<Item> &all_fields,COND *conds)
             Look for a partial key that can be used for optimization.
             If we succeed, ref.key_length will contain the length of
             this key, while prefix_len will contain the length of 
-            the beginning of this key without field used in MIN(). 
+            the beginning of this key without field used in MIN/MAX(). 
             Type of range for the key part for this field will be
             returned in range_fl.
           */
           if (table->file->inited || (outer_tables & table->map) ||
-              !find_key_for_maxmin(0, &ref, item_field->field, conds,
+              !find_key_for_maxmin(is_max, &ref, item_field->field, conds,
                                    &range_fl, &prefix_len))
           {
             const_result= 0;
             break;
           }
-          error= table->file->ha_index_init((uint) ref.key, 1);
+          table->file->ha_index_init((uint) ref.key, 1);
+
+          error= is_max ? 
+                 get_index_max_value(table, &ref, range_fl) :
+                 get_index_min_value(table, &ref, item_field, range_fl,
+                                     prefix_len);
 
-          if (!ref.key_length)
-            error= table->file->index_first(table->record[0]);
-          else 
-          {
-            /*
-              Use index to replace MIN/MAX functions with their values
-              according to the following rules:
-           
-              1) Insert the minimum non-null values where the WHERE clause still
-                 matches, or
-              2) a NULL value if there are only NULL values for key_part_k.
-              3) Fail, producing a row of nulls
-
-              Implementation: Read the smallest value using the search key. If
-              the interval is open, read the next value after the search
-              key. If read fails, and we're looking for a MIN() value for a
-              nullable column, test if there is an exact match for the key.
-            */
-            if (!(range_fl & NEAR_MIN))
-              /* 
-                 Closed interval: Either The MIN argument is non-nullable, or
-                 we have a >= predicate for the MIN argument.
-              */
-              error= table->file->index_read_map(table->record[0],
-                                                 ref.key_buff,
-                                                 make_prev_keypart_map(ref.key_parts),
-                                                 HA_READ_KEY_OR_NEXT);
-            else
-            {
-              /*
-                Open interval: There are two cases:
-                1) We have only MIN() and the argument column is nullable, or
-                2) there is a > predicate on it, nullability is irrelevant.
-                We need to scan the next bigger record first.
-              */
-              error= table->file->index_read_map(table->record[0],
-                                                 ref.key_buff, 
-                                                 make_prev_keypart_map(ref.key_parts),
-                                                 HA_READ_AFTER_KEY);
-              /* 
-                 If the found record is outside the group formed by the search
-                 prefix, or there is no such record at all, check if all
-                 records in that group have NULL in the MIN argument
-                 column. If that is the case return that NULL.
-
-                 Check if case 1 from above holds. If it does, we should read
-                 the skipped tuple.
-              */
-              if (item_field->field->real_maybe_null() &&
-                  ref.key_buff[prefix_len] == 1 &&
-                  /*
-                     Last keypart (i.e. the argument to MIN) is set to NULL by
-                     find_key_for_maxmin only if all other keyparts are bound
-                     to constants in a conjunction of equalities. Hence, we
-                     can detect this by checking only if the last keypart is
-                     NULL.
-                  */
-                  (error == HA_ERR_KEY_NOT_FOUND ||
-                   key_cmp_if_same(table, ref.key_buff, ref.key, prefix_len)))
-              {
-                DBUG_ASSERT(item_field->field->real_maybe_null());
-                error= table->file->index_read_map(table->record[0],
-                                                   ref.key_buff,
-                                                   make_prev_keypart_map(ref.key_parts),
-                                                   HA_READ_KEY_EXACT);
-              }
-            }
-          }
           /* Verify that the read tuple indeed matches the search key */
-	  if (!error && reckey_in_range(0, &ref, item_field->field, 
+	  if (!error && reckey_in_range(is_max, &ref, item_field->field, 
 			                conds, range_fl, prefix_len))
 	    error= HA_ERR_KEY_NOT_FOUND;
           table->set_keyread(FALSE);
@@ -355,100 +411,18 @@ int opt_sum_query(TABLE_LIST *tables, List<Item> &all_fields,COND *conds)
         item_sum->set_aggregator(item_sum->has_with_distinct() ? 
                                  Aggregator::DISTINCT_AGGREGATOR :
                                  Aggregator::SIMPLE_AGGREGATOR);
-        if (!count)
-        {
-          /* If count == 0, then we know that is_exact_count == TRUE. */
-          ((Item_sum_min*) item_sum)->aggregator_clear(); /* Set to NULL. */
-        }
-        else
-          ((Item_sum_min*) item_sum)->reset(); /* Set to the constant value. */
-        ((Item_sum_min*) item_sum)->make_const();
-        recalc_const_item= 1;
-        break;
-      }
-      case Item_sum::MAX_FUNC:
-      {
         /*
-          If MAX(expr) is the first part of a key or if all previous
-          parts of the key is found in the COND, then we can use
-          indexes to find the key.
+          If count == 0 (so is_exact_count == TRUE) and
+          there're no outer joins, set to NULL,
+          otherwise set to the constant value.
         */
-        Item *expr=item_sum->get_arg(0);
-        if (expr->real_item()->type() == Item::FIELD_ITEM)
-        {
-          uchar key_buff[MAX_KEY_LENGTH];
-          TABLE_REF ref;
-          uint range_fl, prefix_len;
-
-          ref.key_buff= key_buff;
-          Item_field *item_field= (Item_field*) (expr->real_item());
-          TABLE *table= item_field->field->table;
-
-          /* 
-            Look for a partial key that can be used for optimization.
-            If we succeed, ref.key_length will contain the length of
-            this key, while prefix_len will contain the length of 
-            the beginning of this key without field used in MAX().
-            Type of range for the key part for this field will be
-            returned in range_fl.
-          */
-          if (table->file->inited || (outer_tables & table->map) ||
-	          !find_key_for_maxmin(1, &ref, item_field->field, conds,
-				                   &range_fl, &prefix_len))
-          {
-            const_result= 0;
-            break;
-          }
-          error= table->file->ha_index_init((uint) ref.key, 1);
-
-          if (!ref.key_length)
-            error= table->file->index_last(table->record[0]);
-          else
-	    error= table->file->index_read_map(table->record[0], key_buff,
-                                               make_prev_keypart_map(ref.key_parts),
-                                               range_fl & NEAR_MAX ?
-                                               HA_READ_BEFORE_KEY :
-                                               HA_READ_PREFIX_LAST_OR_PREV);
-	  if (!error && reckey_in_range(1, &ref, item_field->field,
-			                conds, range_fl, prefix_len))
-	    error= HA_ERR_KEY_NOT_FOUND;
-          table->set_keyread(FALSE);
-          table->file->ha_index_end();
-          if (error)
-          {
-	    if (error == HA_ERR_KEY_NOT_FOUND || error == HA_ERR_END_OF_FILE)
-	      return HA_ERR_KEY_NOT_FOUND;	     // No rows matching WHERE
-	    /* HA_ERR_LOCK_DEADLOCK or some other error */
-            table->file->print_error(error, MYF(ME_FATALERROR));
-            return(error);
-	  }
-          removed_tables|= table->map;
-        }
-        else if (!expr->const_item() || !is_exact_count)
-        {
-          /*
-            The optimization is not applicable in both cases:
-            (a) 'expr' is a non-constant expression. Then we can't
-            replace 'expr' by a constant.
-            (b) 'expr' is a costant. According to ANSI, MIN/MAX must return
-            NULL if the query does not return any rows. Thus, if we are not
-            able to determine if the query returns any rows, we can't apply
-            the optimization and replace MIN/MAX with a constant.
-          */
-          const_result= 0;
-          break;
-        }
-        item_sum->set_aggregator(item_sum->has_with_distinct() ? 
-                                 Aggregator::DISTINCT_AGGREGATOR :
-                                 Aggregator::SIMPLE_AGGREGATOR);
-        if (!count)
+        if (!count && !outer_tables)
         {
-          /* If count != 1, then we know that is_exact_count == TRUE. */
-          ((Item_sum_max*) item_sum)->aggregator_clear(); /* Set to NULL. */
+          item_sum->aggregator_clear();
         }
         else
-          ((Item_sum_max*) item_sum)->reset(); /* Set to the constant value. */
-        ((Item_sum_max*) item_sum)->make_const();
+          item_sum->reset();
+        item_sum->make_const();
         recalc_const_item= 1;
         break;
       }
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index e505630d08f..7be10ecde2c 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -1199,7 +1199,7 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
           part_elem->engine_type= default_engine_type;
         }
         if (check_table_name(part_elem->partition_name,
-                             strlen(part_elem->partition_name)))
+                             strlen(part_elem->partition_name), FALSE))
         {
           my_error(ER_WRONG_PARTITION_NAME, MYF(0));
           goto end;
@@ -1218,7 +1218,7 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
           sub_elem= sub_it++;
           warn_if_dir_in_part_elem(thd, sub_elem);
           if (check_table_name(sub_elem->partition_name,
-                               strlen(sub_elem->partition_name)))
+                               strlen(sub_elem->partition_name), FALSE))
           {
             my_error(ER_WRONG_PARTITION_NAME, MYF(0));
             goto end;
diff --git a/sql/protocol.cc b/sql/protocol.cc
index fd01e4a8885..eeb248012ab 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -1014,8 +1014,8 @@ bool Protocol_text::store(const char *from, size_t length,
 {
   CHARSET_INFO *tocs= this->thd->variables.character_set_results;
 #ifndef DBUG_OFF
-  DBUG_PRINT("info", ("Protocol_text::store field %u (%u): %*.s",
-                      field_pos, field_count, (int) length, from));
+  DBUG_PRINT("info", ("Protocol_text::store field %u (%u): %s", field_pos,
+                      field_count, (length == 0? "" : from)));
   DBUG_ASSERT(field_pos < field_count);
   DBUG_ASSERT(field_types == 0 ||
 	      field_types[field_pos] == MYSQL_TYPE_DECIMAL ||
diff --git a/sql/slave.cc b/sql/slave.cc
index e23763696df..6ebdea4a42a 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -3124,6 +3124,11 @@ pthread_handler_t handle_slave_sql(void *arg)
 {
   THD *thd;                     /* needs to be first for thread_stack */
   char llbuff[22],llbuff1[22];
+  char saved_log_name[FN_REFLEN];
+  char saved_master_log_name[FN_REFLEN];
+  my_off_t saved_log_pos;
+  my_off_t saved_master_log_pos;
+  my_off_t saved_skip= 0;
 
   Relay_log_info* rli = &((Master_info*)arg)->rli;
   const char *errmsg;
@@ -3269,6 +3274,17 @@ log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
     do not want to wait for next event in this case.
   */
   mysql_mutex_lock(&rli->data_lock);
+  if (rli->slave_skip_counter)
+  {
+    char *pos;
+    pos= strmake(saved_log_name, rli->group_relay_log_name, FN_REFLEN - 1);
+    pos= '\0';
+    pos= strmake(saved_master_log_name, rli->group_master_log_name, FN_REFLEN - 1);
+    pos= '\0';
+    saved_log_pos= rli->group_relay_log_pos;
+    saved_master_log_pos= rli->group_master_log_pos;
+    saved_skip= rli->slave_skip_counter;
+  }
   if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
       rli->is_until_satisfied(thd, NULL))
   {
@@ -3287,6 +3303,21 @@ log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
     thd_proc_info(thd, "Reading event from the relay log");
     DBUG_ASSERT(rli->sql_thd == thd);
     THD_CHECK_SENTRY(thd);
+
+    if (saved_skip && rli->slave_skip_counter == 0)
+    {
+      sql_print_information("'SQL_SLAVE_SKIP_COUNTER=%ld' executed at "
+        "relay_log_file='%s', relay_log_pos='%ld', master_log_name='%s', "
+        "master_log_pos='%ld' and new position at "
+        "relay_log_file='%s', relay_log_pos='%ld', master_log_name='%s', "
+        "master_log_pos='%ld' ",
+        (ulong) saved_skip, saved_log_name, (ulong) saved_log_pos,
+        saved_master_log_name, (ulong) saved_master_log_pos,
+        rli->group_relay_log_name, (ulong) rli->group_relay_log_pos,
+        rli->group_master_log_name, (ulong) rli->group_master_log_pos);
+      saved_skip= 0;
+    }
+    
     if (exec_relay_log_event(thd,rli))
     {
       DBUG_PRINT("info", ("exec_relay_log_event() failed"));
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 2e66aec91e5..06918e42e47 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -747,21 +747,12 @@ create_typelib(MEM_ROOT *mem_root, Create_field *field_def, List<String> *src)
 
 sp_head::~sp_head()
 {
+  LEX *lex;
+  sp_instr *i;
   DBUG_ENTER("sp_head::~sp_head");
-  destroy();
-  delete m_next_cached_sp;
-  if (m_thd)
-    restore_thd_mem_root(m_thd);
-  DBUG_VOID_RETURN;
-}
 
-void
-sp_head::destroy()
-{
-  sp_instr *i;
-  LEX *lex;
-  DBUG_ENTER("sp_head::destroy");
-  DBUG_PRINT("info", ("name: %s", m_name.str));
+  /* sp_head::restore_thd_mem_root() must already have been called. */
+  DBUG_ASSERT(m_thd == NULL);
 
   for (uint ip = 0 ; (i = get_instr(ip)) ; ip++)
     delete i;
@@ -772,21 +763,22 @@ sp_head::destroy()
   /*
     If we have non-empty LEX stack then we just came out of parser with
     error. Now we should delete all auxilary LEXes and restore original
-    THD::lex (In this case sp_head::restore_thd_mem_root() was not called
-    too, so m_thd points to the current thread context).
-    It is safe to not update LEX::ptr because further query string parsing
-    and execution will be stopped anyway.
+    THD::lex. It is safe to not update LEX::ptr because further query
+    string parsing and execution will be stopped anyway.
   */
-  DBUG_ASSERT(m_lex.is_empty() || m_thd);
   while ((lex= (LEX *)m_lex.pop()))
   {
-    lex_end(m_thd->lex);
-    delete m_thd->lex;
-    m_thd->lex= lex;
+    THD *thd= lex->thd;
+    lex_end(thd->lex);
+    delete thd->lex;
+    thd->lex= lex;
   }
 
   my_hash_free(&m_sptabs);
   my_hash_free(&m_sroutines);
+
+  delete m_next_cached_sp;
+
   DBUG_VOID_RETURN;
 }
 
@@ -3051,6 +3043,7 @@ int
 sp_instr_set_trigger_field::execute(THD *thd, uint *nextp)
 {
   DBUG_ENTER("sp_instr_set_trigger_field::execute");
+  thd->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;
   DBUG_RETURN(m_lex_keeper.reset_lex_and_exec_core(thd, nextp, TRUE, this));
 }
 
diff --git a/sql/sp_head.h b/sql/sp_head.h
index 165f88321a9..539a2da5f8c 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -305,10 +305,6 @@ public:
 
   virtual ~sp_head();
 
-  /// Free memory
-  void
-  destroy();
-
   bool
   execute_trigger(THD *thd,
                   const LEX_STRING *db_name,
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 55d83f49245..ec25e4cb68b 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -6261,21 +6261,21 @@ bool mysql_revoke_all(THD *thd,  List <LEX_USER> &list)
 
   mysql_mutex_unlock(&acl_cache->lock);
 
-  int binlog_error=
+  if (result)
+    my_message(ER_REVOKE_GRANTS, ER(ER_REVOKE_GRANTS), MYF(0));
+
+  result= result |
     write_bin_log(thd, FALSE, thd->query(), thd->query_length());
 
   mysql_rwlock_unlock(&LOCK_grant);
   close_thread_tables(thd);
 
-  /* error for writing binary log has already been reported */
-  if (result && !binlog_error)
-    my_message(ER_REVOKE_GRANTS, ER(ER_REVOKE_GRANTS), MYF(0));
   /* Restore the state of binlog format */
   DBUG_ASSERT(!thd->is_current_stmt_binlog_format_row());
   if (save_binlog_row_based)
     thd->set_current_stmt_binlog_format_row();
 
-  DBUG_RETURN(result || binlog_error);
+  DBUG_RETURN(result);
 }
 
 
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 28633365e28..1e537112d01 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1418,6 +1418,12 @@ void close_thread_tables(THD *thd)
                           table->s->table_name.str, (long) table));
 #endif
 
+#if defined(ENABLED_DEBUG_SYNC)
+  /* debug_sync may not be initialized for some slave threads */
+  if (thd->debug_sync_control)
+    DEBUG_SYNC(thd, "before_close_thread_tables");
+#endif
+
   /* Detach MERGE children after every statement. Even under LOCK TABLES. */
   for (table= thd->open_tables; table; table= table->next)
   {
@@ -5281,8 +5287,8 @@ bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
     thd		- thread handler
     tables	- list of tables for open
     flags       - bitmap of flags to modify how the tables will be open:
-                  MYSQL_OPEN_IGNORE_FLUSH - open table even if someone has
-                  done a flush or namelock on it.
+                  MYSQL_LOCK_IGNORE_FLUSH - open table even if someone has
+                  done a flush on it.
 
   RETURN
     FALSE - ok
@@ -8821,8 +8827,57 @@ bool is_equal(const LEX_STRING *a, const LEX_STRING *b)
 
 
 /*
+  Unlock and close table before renaming and dropping partitions
+  SYNOPSIS
+    alter_close_tables()
+    lpt                        Struct carrying parameters
+  RETURN VALUES
+    0
+*/
+
+static int alter_close_tables(ALTER_PARTITION_PARAM_TYPE *lpt)
+{
+  TABLE_SHARE *share= lpt->table->s;
+  THD *thd= lpt->thd;
+  TABLE *table;
+  DBUG_ENTER("alter_close_tables");
+  /*
+    We must keep LOCK_open while manipulating with thd->open_tables.
+    Another thread may be working on it.
+  */
+  mysql_mutex_lock(&LOCK_open);
+  /*
+    We can safely remove locks for all tables with the same name:
+    later they will all be closed anyway in
+    alter_partition_lock_handling().
+  */
+  for (table= thd->open_tables; table ; table= table->next)
+  {
+    if (!strcmp(table->s->table_name.str, share->table_name.str) &&
+	!strcmp(table->s->db.str, share->db.str))
+    {
+      mysql_lock_remove(thd, thd->lock, table);
+      table->file->close();
+      table->db_stat= 0;                        // Mark file closed
+      /*
+        Ensure that we won't end up with a crippled table instance
+        in the table cache if an error occurs before we reach
+        alter_partition_lock_handling() and the table is closed
+        by close_thread_tables() instead.
+      */
+      tdc_remove_table(thd, TDC_RT_REMOVE_UNUSED,
+                       table->s->db.str,
+                       table->s->table_name.str);
+    }
+  }
+  mysql_mutex_unlock(&LOCK_open);
+  DBUG_RETURN(0);
+}
+
+
+/*
   SYNOPSIS
-    abort_and_upgrade_lock()
+    abort_and_upgrade_lock_and_close_table()
     lpt                           Parameter passing struct
     All parameters passed through the ALTER_PARTITION_PARAM_TYPE object
   RETURN VALUE
@@ -8831,7 +8886,7 @@ bool is_equal(const LEX_STRING *a, const LEX_STRING *b)
     Remember old lock level (for possible downgrade later on), abort all
     waiting threads and ensure that all keeping locks currently are
     completed such that we own the lock exclusively and no other interaction
-    is ongoing.
+    is ongoing. Close the table and hold the name lock.
 
     thd                           Thread object
     table                         Table object
@@ -8840,12 +8895,14 @@ bool is_equal(const LEX_STRING *a, const LEX_STRING *b)
     old_lock_level                Old lock level
 */
 
-int abort_and_upgrade_lock(ALTER_PARTITION_PARAM_TYPE *lpt)
+int abort_and_upgrade_lock_and_close_table(ALTER_PARTITION_PARAM_TYPE *lpt)
 {
-  DBUG_ENTER("abort_and_upgrade_lock");
+  DBUG_ENTER("abort_and_upgrade_lock_and_close_table");
 
   if (wait_while_table_is_used(lpt->thd, lpt->table, HA_EXTRA_FORCE_REOPEN))
     DBUG_RETURN(1);
+  if (alter_close_tables(lpt))
+    DBUG_RETURN(1);
   DBUG_RETURN(0);
 }
 
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 0c16151e43a..0fe70e4bc9d 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -217,7 +217,7 @@ TABLE *open_n_lock_single_table(THD *thd, TABLE_LIST *table_l,
                                 thr_lock_type lock_type, uint flags);
 bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables, uint flags);
 bool lock_tables(THD *thd, TABLE_LIST *tables, uint counter, uint flags);
-int abort_and_upgrade_lock(ALTER_PARTITION_PARAM_TYPE *lpt);
+int abort_and_upgrade_lock_and_close_table(ALTER_PARTITION_PARAM_TYPE *lpt);
 int decide_logging_format(THD *thd, TABLE_LIST *tables);
 void free_io_cache(TABLE *entry);
 void intern_close_table(TABLE *entry);
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 789b01443f7..814ef27e574 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -305,6 +305,37 @@ void **thd_ha_data(const THD *thd, const struct handlerton *hton)
   return (void **) &thd->ha_data[hton->slot].ha_ptr;
 }
 
+
+/**
+  Provide a handler data getter to simplify coding
+*/
+extern "C"
+void *thd_get_ha_data(const THD *thd, const struct handlerton *hton)
+{
+  return *thd_ha_data(thd, hton);
+}
+
+
+/**
+  Provide a handler data setter to simplify coding
+  @see thd_set_ha_data() definition in plugin.h
+*/
+extern "C"
+void thd_set_ha_data(THD *thd, const struct handlerton *hton,
+                     const void *ha_data)
+{
+  plugin_ref *lock= &thd->ha_data[hton->slot].lock;
+  if (ha_data && !*lock)
+    *lock= ha_lock_engine(NULL, (handlerton*) hton);
+  else if (!ha_data && *lock)
+  {
+    plugin_unlock(NULL, *lock);
+    *lock= NULL;
+  }
+  *thd_ha_data(thd, hton)= (void*) ha_data;
+}
+
+
 extern "C"
 long long thd_test_options(const THD *thd, long long test_options)
 {
@@ -1884,8 +1915,7 @@ static File create_file(THD *thd, char *path, sql_exchange *exchange,
   else
     (void) fn_format(path, exchange->file_name, mysql_real_data_home, "", option);
 
-  if (opt_secure_file_priv &&
-      strncmp(opt_secure_file_priv, path, strlen(opt_secure_file_priv)))
+  if (!is_secure_file_path(path))
   {
     /* Write only allowed to dir or subdir specified by secure_file_priv */
     my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--secure-file-priv");
@@ -2054,9 +2084,21 @@ bool select_export::send_data(List<Item> &items)
       const char *from_end_pos;
       const char *error_pos;
       uint32 bytes;
-      bytes= well_formed_copy_nchars(write_cs, cvt_buff, sizeof(cvt_buff),
+      uint64 estimated_bytes=
+        ((uint64) res->length() / res->charset()->mbminlen + 1) *
+        write_cs->mbmaxlen + 1;
+      set_if_smaller(estimated_bytes, UINT_MAX32);
+      if (cvt_str.realloc((uint32) estimated_bytes))
+      {
+        my_error(ER_OUTOFMEMORY, MYF(0), (uint32) estimated_bytes);
+        goto err;
+      }
+
+      bytes= well_formed_copy_nchars(write_cs, (char *) cvt_str.ptr(),
+                                     cvt_str.alloced_length(),
                                      res->charset(), res->ptr(), res->length(),
-                                     sizeof(cvt_buff),
+                                     UINT_MAX32, // copy all input chars,
+                                                 // i.e. ignore nchars parameter
                                      &well_formed_error_pos,
                                      &cannot_convert_error_pos,
                                      &from_end_pos);
@@ -2074,6 +2116,15 @@ bool select_export::send_data(List<Item> &items)
                             "string", printable_buff,
                             item->name, row_count);
       }
+      else if (from_end_pos < res->ptr() + res->length())
+      { 
+        /*
+          result is longer than UINT_MAX32 and doesn't fit into String
+        */
+        push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                            WARN_DATA_TRUNCATED, ER(WARN_DATA_TRUNCATED),
+                            item->full_name(), row_count);
+      }
       cvt_str.length(bytes);
       res= &cvt_str;
     }
@@ -3185,6 +3236,7 @@ void THD::reset_sub_statement_state(Sub_statement_state *backup,
 #endif
   
   backup->option_bits=     variables.option_bits;
+  backup->count_cuted_fields= count_cuted_fields;
   backup->in_sub_stmt=     in_sub_stmt;
   backup->enable_slow_log= enable_slow_log;
   backup->limit_found_rows= limit_found_rows;
@@ -3222,6 +3274,7 @@ void THD::reset_sub_statement_state(Sub_statement_state *backup,
 
 void THD::restore_sub_statement_state(Sub_statement_state *backup)
 {
+  DBUG_ENTER("THD::restore_sub_statement_state");
 #ifndef EMBEDDED_LIBRARY
   /* BUG#33029, if we are replicating from a buggy master, restore
      auto_inc_intervals_forced so that the top statement can use the
@@ -3248,6 +3301,7 @@ void THD::restore_sub_statement_state(Sub_statement_state *backup)
     /* ha_release_savepoint() never returns error. */
     (void)ha_release_savepoint(this, sv);
   }
+  count_cuted_fields= backup->count_cuted_fields;
   transaction.savepoints= backup->savepoints;
   variables.option_bits= backup->option_bits;
   in_sub_stmt=      backup->in_sub_stmt;
@@ -3277,6 +3331,7 @@ void THD::restore_sub_statement_state(Sub_statement_state *backup)
   */
   examined_row_count+= backup->examined_row_count;
   cuted_fields+=       backup->cuted_fields;
+  DBUG_VOID_RETURN;
 }
 
 
@@ -4406,7 +4461,6 @@ int THD::binlog_flush_pending_rows_event(bool stmt_end, bool is_transactional)
     if (stmt_end)
     {
       pending->set_flags(Rows_log_event::STMT_END_F);
-      pending->flags|= LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F;
       binlog_table_maps= 0;
     }
 
@@ -4592,7 +4646,6 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
     {
       Query_log_event qinfo(this, query_arg, query_len, is_trans, direct,
                             suppress_use, errcode);
-      qinfo.flags|= LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F;
       /*
         Binlog table maps will be irrelevant after a Query_log_event
         (they are just removed on the slave side) so after the query
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 916b79f8353..fb28d3b09c8 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1094,6 +1094,7 @@ public:
   bool enable_slow_log;
   bool last_insert_id_used;
   SAVEPOINT *savepoints;
+  enum enum_check_fields count_cuted_fields;
 };
 
 
@@ -1469,7 +1470,11 @@ struct Ha_data
     @sa trans_register_ha()
   */
   Ha_trx_info ha_info[2];
-
+  /**
+    NULL: engine is not bound to this thread
+    non-NULL: engine is bound to this thread, engine shutdown forbidden
+  */
+  plugin_ref lock;
   Ha_data() :ha_ptr(NULL) {}
 };
 
@@ -2046,8 +2051,15 @@ public:
   */
   ha_rows    sent_row_count;
 
-  /*
-    number of rows we read, sent or not, including in create_sort_index()
+  /**
+    Number of rows read and/or evaluated for a statement. Used for
+    slow log reporting.
+
+    An examined row is defined as a row that is read and/or evaluated
+    according to a statement condition, including in
+    create_sort_index(). Rows may be counted more than once, e.g., a
+    statement including ORDER BY could possibly evaluate the row in
+    filesort() before reading it for e.g. update.
   */
   ha_rows    examined_row_count;
 
diff --git a/sql/sql_connect.cc b/sql/sql_connect.cc
index e2d0977def7..c0081c13366 100644
--- a/sql/sql_connect.cc
+++ b/sql/sql_connect.cc
@@ -496,6 +496,13 @@ check_user(THD *thd, enum enum_server_command command,
       }
       my_ok(thd);
       thd->password= test(passwd_len);          // remember for error messages 
+      /*
+        Allow the network layer to skip big packets. Although a malicious
+        authenticated session might use this to trick the server to read
+        big packets indefinitely, this is a previously established behavior
+        that needs to be preserved as to not break backwards compatibility.
+      */
+      thd->net.skip_big_packet= TRUE;
       /* Ready to handle queries */
       DBUG_RETURN(0);
     }
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 10bdb8a22a6..25e470f56ea 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -262,6 +262,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
         free_underlaid_joins(thd, &thd->lex->select_lex);
         DBUG_RETURN(TRUE);
       }
+      thd->examined_row_count+= examined_rows;
       /*
         Filesort has already found and selected the rows we want to delete,
         so we don't need the where clause
@@ -279,7 +280,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     free_underlaid_joins(thd, select_lex);
     DBUG_RETURN(TRUE);
   }
-  if (usable_index==MAX_KEY)
+  if (usable_index==MAX_KEY || (select && select->quick))
     init_read_record(&info, thd, table, select, 1, 1, FALSE);
   else
     init_read_record_idx(&info, thd, table, 1, usable_index);
@@ -318,6 +319,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   while (!(error=info.read_record(&info)) && !thd->killed &&
 	 ! thd->is_error())
   {
+    thd->examined_row_count++;
     // thd->is_error() is tested to disallow delete row on error
     if (!(select && select->skip_record())&& ! thd->is_error() )
     {
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 1f9d69a798e..d40f0dcb410 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3256,7 +3256,7 @@ bool select_insert::send_data(List<Item> &values)
 
   thd->count_cuted_fields= CHECK_FIELD_WARN;	// Calculate cuted fields
   store_values(values);
-  thd->count_cuted_fields= CHECK_FIELD_IGNORE;
+  thd->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;
   if (thd->is_error())
   {
     table->auto_increment_field_not_null= FALSE;
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 1795bc272f1..88cba22d115 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -2218,6 +2218,7 @@ void LEX::cleanup_lex_after_parse_error(THD *thd)
   */
   if (thd->lex->sphead)
   {
+    thd->lex->sphead->restore_thd_mem_root(thd);
     delete thd->lex->sphead;
     thd->lex->sphead= NULL;
   }
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 2ce6bdeed42..40bd3875793 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2040,6 +2040,7 @@ struct LEX: public Query_tables_list
       - CREATE TRIGGER (points to "TRIGGER");
       - CREATE PROCEDURE (points to "PROCEDURE");
       - CREATE FUNCTION (points to "FUNCTION" or "AGGREGATE");
+      - CREATE EVENT (points to "EVENT")
 
     This pointer is required to add possibly omitted DEFINER-clause to the
     DDL-statement before dumping it to the binlog.
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index ff9c16d229b..6054bb0ac23 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -397,16 +397,11 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
         DBUG_ASSERT(FALSE); 
 #endif
       }
-      else if (opt_secure_file_priv)
+      else if (!is_secure_file_path(name))
       {
-        char secure_file_real_path[FN_REFLEN];
-        (void) my_realpath(secure_file_real_path, opt_secure_file_priv, 0);
-        if (strncmp(secure_file_real_path, name, strlen(secure_file_real_path)))
-        {
-          /* Read only allowed from within dir specified by secure_file_priv */
-          my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--secure-file-priv");
-          DBUG_RETURN(TRUE);
-        }
+        /* Read only allowed from within dir specified by secure_file_priv */
+        my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--secure-file-priv");
+        DBUG_RETURN(TRUE);
       }
 
     }
@@ -572,7 +567,6 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
 	  else
 	  {
 	    Delete_file_log_event d(thd, db, transactional_table);
-            d.flags|= LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F;
 	    (void) mysql_bin_log.write(&d);
 	  }
 	}
@@ -748,16 +742,13 @@ static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
   strcpy(end, p);
   end += pl;
 
-  thd->set_query_inner(load_data_query, end - load_data_query);
-
   Execute_load_query_log_event
-    e(thd, thd->query(), thd->query_length(),
-      (uint) ((char*) fname_start - (char*) thd->query() - 1),
-      (uint) ((char*) fname_end - (char*) thd->query()),
+    e(thd, load_data_query, end-load_data_query,
+      (uint) ((char*) fname_start - load_data_query - 1),
+      (uint) ((char*) fname_end - load_data_query),
       (duplicates == DUP_REPLACE) ? LOAD_DUP_REPLACE :
       (ignore ? LOAD_DUP_IGNORE : LOAD_DUP_ERROR),
       transactional_table, FALSE, FALSE, errcode);
-  e.flags|= LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F;
   return mysql_bin_log.write(&e);
 }
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 2e47ce1e02a..ed4390a23cb 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1160,8 +1160,23 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
       We have name + wildcard in packet, separated by endzero
     */
     arg_end= strend(packet);
+    uint arg_length= arg_end - packet;
+    
+    /* Check given table name length. */
+    if (arg_length >= packet_length || arg_length > NAME_LEN)
+    {
+      my_message(ER_UNKNOWN_COM_ERROR, ER(ER_UNKNOWN_COM_ERROR), MYF(0));
+      break;
+    }
     thd->convert_string(&conv_name, system_charset_info,
-			packet, (uint) (arg_end - packet), thd->charset());
+			packet, arg_length, thd->charset());
+    if (check_table_name(conv_name.str, conv_name.length, FALSE))
+    {
+      /* this is OK due to convert_string() null-terminating the string */
+      my_error(ER_WRONG_TABLE_NAME, MYF(0), conv_name.str);
+      break;
+    }
+
     table_list.alias= table_list.table_name= conv_name.str;
     packet= arg_end + 1;
 
@@ -3238,7 +3253,7 @@ end_with_restore_list:
           TODO: this is workaround. right way will be move invalidating in
           the unlock procedure.
         */
-        if (first_table->lock_type ==  TL_WRITE_CONCURRENT_INSERT &&
+        if (!res && first_table->lock_type ==  TL_WRITE_CONCURRENT_INSERT &&
             thd->lock)
         {
           /* INSERT ... SELECT should invalidate only the very first table */
@@ -6042,7 +6057,7 @@ TABLE_LIST *st_select_lex::add_table_to_list(THD *thd,
     DBUG_RETURN(0);				// End of memory
   alias_str= alias ? alias->str : table->table.str;
   if (!test(table_options & TL_OPTION_ALIAS) && 
-      check_table_name(table->table.str, table->table.length))
+      check_table_name(table->table.str, table->table.length, FALSE))
   {
     my_error(ER_WRONG_TABLE_NAME, MYF(0), table->table.str);
     DBUG_RETURN(0);
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index e6f9a10801b..fa9c698622b 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -6280,54 +6280,6 @@ static void alter_partition_lock_handling(ALTER_PARTITION_PARAM_TYPE *lpt)
     sql_print_warning("We failed to reacquire LOCKs in ALTER TABLE");
 }
 
-/*
-  Unlock and close table before renaming and dropping partitions
-  SYNOPSIS
-    alter_close_tables()
-    lpt                        Struct carrying parameters
-  RETURN VALUES
-    0
-*/
-
-static int alter_close_tables(ALTER_PARTITION_PARAM_TYPE *lpt)
-{
-  TABLE_SHARE *share= lpt->table->s;
-  THD *thd= lpt->thd;
-  TABLE *table;
-  DBUG_ENTER("alter_close_tables");
-  /*
-    We must keep LOCK_open while manipulating with thd->open_tables.
-    Another thread may be working on it.
-  */
-  mysql_mutex_lock(&LOCK_open);
-  /*
-    We can safely remove locks for all tables with the same name:
-    later they will all be closed anyway in
-    alter_partition_lock_handling().
-  */
-  for (table= thd->open_tables; table ; table= table->next)
-  {
-    if (!strcmp(table->s->table_name.str, share->table_name.str) &&
-	!strcmp(table->s->db.str, share->db.str))
-    {
-      mysql_lock_remove(thd, thd->lock, table);
-      table->file->close();
-      table->db_stat= 0;                        // Mark file closed
-      /*
-        Ensure that we won't end up with a crippled table instance
-        in the table cache if an error occurs before we reach
-        alter_partition_lock_handling() and the table is closed
-        by close_thread_tables() instead.
-      */
-      tdc_remove_table(thd, TDC_RT_REMOVE_UNUSED,
-                       table->s->db.str,
-                       table->s->table_name.str);
-    }
-  }
-  mysql_mutex_unlock(&LOCK_open);
-  DBUG_RETURN(0);
-}
-
 
 /*
   Handle errors for ALTER TABLE for partitioning
@@ -6626,9 +6578,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         write_log_drop_partition(lpt) ||
         ERROR_INJECT_CRASH("crash_drop_partition_3") ||
         (not_completed= FALSE) ||
-        abort_and_upgrade_lock(lpt) ||
-        ERROR_INJECT_CRASH("crash_drop_partition_4") ||
-        alter_close_tables(lpt) ||
+        abort_and_upgrade_lock_and_close_table(lpt) ||
         ERROR_INJECT_CRASH("crash_drop_partition_5") ||
         ((!thd->lex->no_write_to_binlog) &&
          (write_bin_log(thd, FALSE,
@@ -6694,9 +6644,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT_CRASH("crash_add_partition_2") ||
         mysql_change_partitions(lpt) ||
         ERROR_INJECT_CRASH("crash_add_partition_3") ||
-        abort_and_upgrade_lock(lpt) ||
-        ERROR_INJECT_CRASH("crash_add_partition_4") ||
-        alter_close_tables(lpt) ||
+        abort_and_upgrade_lock_and_close_table(lpt) ||
         ERROR_INJECT_CRASH("crash_add_partition_5") ||
         ((!thd->lex->no_write_to_binlog) &&
          (write_bin_log(thd, FALSE,
@@ -6779,9 +6727,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         write_log_final_change_partition(lpt) ||
         ERROR_INJECT_CRASH("crash_change_partition_4") ||
         (not_completed= FALSE) ||
-        abort_and_upgrade_lock(lpt) ||
-        ERROR_INJECT_CRASH("crash_change_partition_5") ||
-        alter_close_tables(lpt) ||
+        abort_and_upgrade_lock_and_close_table(lpt) ||
         ERROR_INJECT_CRASH("crash_change_partition_6") ||
         ((!thd->lex->no_write_to_binlog) &&
          (write_bin_log(thd, FALSE,
diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc
index 31e0cced207..0282053ce7e 100644
--- a/sql/sql_plugin.cc
+++ b/sql/sql_plugin.cc
@@ -1755,6 +1755,12 @@ bool mysql_install_plugin(THD *thd, const LEX_STRING *name, const LEX_STRING *dl
   struct st_plugin_int *tmp;
   DBUG_ENTER("mysql_install_plugin");
 
+  if (opt_noacl)
+  {
+    my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--skip-grant-tables");
+    DBUG_RETURN(TRUE);
+  }
+
   tables.init_one_table("mysql", 5, "plugin", 6, "plugin", TL_WRITE);
   if (check_table_access(thd, INSERT_ACL, &tables, FALSE, 1, FALSE))
     DBUG_RETURN(TRUE);
@@ -1829,6 +1835,12 @@ bool mysql_uninstall_plugin(THD *thd, const LEX_STRING *name)
   struct st_plugin_int *plugin;
   DBUG_ENTER("mysql_uninstall_plugin");
 
+  if (opt_noacl)
+  {
+    my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--skip-grant-tables");
+    DBUG_RETURN(TRUE);
+  }
+
   tables.init_one_table("mysql", 5, "plugin", 6, "plugin", TL_WRITE);
 
   if (check_table_access(thd, DELETE_ACL, &tables, FALSE, 1, FALSE))
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 1ee770f70ef..787f9dcae2c 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -1347,6 +1347,10 @@ bool change_master(THD* thd, Master_info* mi)
   const char* errmsg= 0;
   bool need_relay_log_purge= 1;
   bool ret= FALSE;
+  char saved_host[HOSTNAME_LENGTH + 1];
+  uint saved_port;
+  char saved_log_name[FN_REFLEN];
+  my_off_t saved_log_pos;
   DBUG_ENTER("change_master");
 
   lock_slave_threads(mi);
@@ -1389,6 +1393,17 @@ bool change_master(THD* thd, Master_info* mi)
   */
 
   /*
+    Before processing the command, save the previous state.
+  */
+  char *pos;
+  pos= strmake(saved_host, mi->host, HOSTNAME_LENGTH);
+  pos= '\0';
+  saved_port= mi->port;
+  pos= strmake(saved_log_name, mi->master_log_name, FN_REFLEN - 1);
+  pos= '\0';
+  saved_log_pos= mi->master_log_pos;
+
+  /*
     If the user specified host or port without binlog or position,
     reset binlog's name to FIRST and position to 4.
   */
@@ -1586,6 +1601,15 @@ bool change_master(THD* thd, Master_info* mi)
   /* Clear the errors, for a clean start */
   mi->rli.clear_error();
   mi->rli.clear_until_condition();
+
+  sql_print_information("'CHANGE MASTER TO executed'. "
+    "Previous state master_host='%s', master_port='%u', master_log_file='%s', "
+    "master_log_pos='%ld'. "
+    "New state master_host='%s', master_port='%u', master_log_file='%s', "
+    "master_log_pos='%ld'.", saved_host, saved_port, saved_log_name,
+    (ulong) saved_log_pos, mi->host, mi->port, mi->master_log_name,
+    (ulong) mi->master_log_pos);
+
   /*
     If we don't write new coordinates to disk now, then old will remain in
     relay-log.info until START SLAVE is issued; but if mysqld is shutdown
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index d126d0e4ec6..8112bbba267 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1126,9 +1126,8 @@ JOIN::optimize()
     }
   }
 
-  if (conds &&!outer_join && const_table_map != found_const_table_map && 
-      (select_options & SELECT_DESCRIBE) &&
-      select_lex->master_unit() == &thd->lex->unit) // upper level SELECT
+  if (conds && const_table_map != found_const_table_map &&
+      (select_options & SELECT_DESCRIBE))
   {
     conds=new Item_int((longlong) 0,1);	// Always false
   }
@@ -1145,13 +1144,13 @@ JOIN::optimize()
     elements may be lost during further having
     condition transformation in JOIN::exec.
   */
-  if (having && !having->with_sum_func)
+  if (having && const_table_map)
   {
-    COND *const_cond= make_cond_for_table(having, const_table_map, 0);
-    DBUG_EXECUTE("where", print_where(const_cond, "const_having_cond",
-                                      QT_ORDINARY););
-    if (const_cond && !const_cond->val_int())
+    having->update_used_tables();
+    having= remove_eq_conds(thd, having, &having_value);
+    if (having_value == Item::COND_FALSE)
     {
+      having= new Item_int((longlong) 0,1);
       zero_result_cause= "Impossible HAVING noticed after reading const tables";
       error= 0;
       DBUG_RETURN(0);
@@ -3006,8 +3005,7 @@ make_join_statistics(JOIN *join, TABLE_LIST *tables_arg, COND *conds,
       s->quick=select->quick;
       s->needed_reg=select->needed_reg;
       select->quick=0;
-      if (records == 0 && s->table->reginfo.impossible_range &&
-          (s->table->file->ha_table_flags() & HA_STATS_RECORDS_IS_EXACT))
+      if (records == 0 && s->table->reginfo.impossible_range)
       {
 	/*
 	  Impossible WHERE or ON expression
@@ -5162,6 +5160,11 @@ greedy_search(JOIN      *join,
     if (best_extension_by_limited_search(join, remaining_tables, idx, record_count,
                                          read_time, search_depth, prune_level))
       DBUG_RETURN(TRUE);
+    /*
+      'best_read < DBL_MAX' means that optimizer managed to find
+      some plan and updated 'best_positions' array accordingly.
+    */
+    DBUG_ASSERT(join->best_read < DBL_MAX); 
 
     if (size_remain <= search_depth)
     {
@@ -8942,8 +8945,14 @@ simplify_joins(JOIN *join, List<TABLE_LIST> *join_list, COND *conds, bool top)
           we still make the inner tables dependent on the outer tables.
           It would be enough to set dependency only on one outer table
           for them. Yet this is really a rare case.
+          Note:
+          RAND_TABLE_BIT mask should not be counted as it
+          prevents update of inner table dependences.
+          For example it might happen if RAND() function
+          is used in JOIN ON clause.
 	*/  
-        if (!(prev_table->on_expr->used_tables() & ~prev_used_tables))
+        if (!((prev_table->on_expr->used_tables() & ~RAND_TABLE_BIT) &
+              ~prev_used_tables))
           prev_table->dep_tables|= used_tables;
       }
     }
@@ -9199,6 +9208,46 @@ static bool check_interleaving_with_nj(JOIN_TAB *next_tab)
 /**
   Nested joins perspective: Remove the last table from the join order.
 
+  The algorithm is the reciprocal of check_interleaving_with_nj(), hence
+  parent join nest nodes are updated only when the last table in its child
+  node is removed. The ASCII graphic below will clarify.
+
+  %A table nesting such as <tt> t1 x [ ( t2 x t3 ) x ( t4 x t5 ) ] </tt>is
+  represented by the below join nest tree.
+
+  @verbatim
+                     NJ1
+                  _/ /  \
+                _/  /    NJ2
+              _/   /     / \ 
+             /    /     /   \
+   t1 x [ (t2 x t3) x (t4 x t5) ]
+  @endverbatim
+
+  At the point in time when check_interleaving_with_nj() adds the table t5 to
+  the query execution plan, QEP, it also directs the node named NJ2 to mark
+  the table as covered. NJ2 does so by incrementing its @c counter
+  member. Since all of NJ2's tables are now covered by the QEP, the algorithm
+  proceeds up the tree to NJ1, incrementing its counter as well. All join
+  nests are now completely covered by the QEP.
+
+  restore_prev_nj_state() does the above in reverse. As seen above, the node
+  NJ1 contains the nodes t2, t3, and NJ2. Its counter being equal to 3 means
+  that the plan covers t2, t3, and NJ2, @e and that the sub-plan (t4 x t5)
+  completely covers NJ2. The removal of t5 from the partial plan will first
+  decrement NJ2's counter to 1. It will then detect that NJ2 went from being
+  completely to partially covered, and hence the algorithm must continue
+  upwards to NJ1 and decrement its counter to 2. %A subsequent removal of t4
+  will however not influence NJ1 since it did not un-cover the last table in
+  NJ2.
+
+  SYNOPSIS
+    restore_prev_nj_state()
+      last  join table to remove, it is assumed to be the last in current 
+            partial join order.
+     
+  DESCRIPTION
+
     Remove the last table from the partial join order and update the nested
     joins counters and join->cur_embedding_map. It is ok to call this 
     function for the first table in join order (for which 
@@ -9212,19 +9261,20 @@ static void restore_prev_nj_state(JOIN_TAB *last)
 {
   TABLE_LIST *last_emb= last->table->pos_in_table_list->embedding;
   JOIN *join= last->join;
-  while (last_emb)
+  for (;last_emb != NULL; last_emb= last_emb->embedding)
   {
-    if (!(--last_emb->nested_join->counter))
-      join->cur_embedding_map&= ~last_emb->nested_join->nj_map;
-    else if (last_emb->nested_join->join_list.elements-1 ==
-             last_emb->nested_join->counter) 
-    {
-      join->cur_embedding_map|= last_emb->nested_join->nj_map;
-      break;
-    }
-    else
+    NESTED_JOIN *nest= last_emb->nested_join;
+    DBUG_ASSERT(nest->counter > 0);
+    
+    bool was_fully_covered= nest->is_fully_covered();
+    
+    if (--nest->counter == 0)
+      join->cur_embedding_map&= ~nest->nj_map;
+    
+    if (!was_fully_covered)
       break;
-    last_emb= last_emb->embedding;
+    
+    join->cur_embedding_map|= nest->nj_map;
   }
 }
 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 07470ff4f42..dda434a557a 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -50,7 +50,8 @@
 #include "event_data_objects.h"
 #endif
 #include <my_dir.h>
-#include "lock.h"                           // MYSQL_LOCK_IGNORE_FLUSH
+#include "debug_sync.h"
+#include "lock.h"                           // MYSQL_OPEN_IGNORE_FLUSH
 
 #define STR_OR_NIL(S) ((S) ? (S) : "<nil>")
 
@@ -1915,6 +1916,8 @@ int fill_schema_processlist(THD* thd, TABLE_LIST* tables, COND* cond)
         mysql_mutex_unlock(&mysys_var->mutex);
 
       /* INFO */
+      /* Lock THD mutex that protects its data when looking at it. */
+      mysql_mutex_lock(&tmp->LOCK_thd_data);
       if (tmp->query())
       {
         table->field[7]->store(tmp->query(),
@@ -1922,6 +1925,7 @@ int fill_schema_processlist(THD* thd, TABLE_LIST* tables, COND* cond)
                                    tmp->query_length()), cs);
         table->field[7]->set_notnull();
       }
+      mysql_mutex_unlock(&tmp->LOCK_thd_data);
 
       if (schema_table_store_record(thd, table))
       {
@@ -3546,6 +3550,7 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
             lex->sql_command= SQLCOM_SHOW_FIELDS;
             show_table_list->i_s_requested_object=
               schema_table->i_s_requested_object;
+            DEBUG_SYNC(thd, "before_open_in_get_all_tables");
             res= open_normal_and_derived_tables(thd, show_table_list,
                    (MYSQL_OPEN_IGNORE_FLUSH |
                     MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL |
@@ -4324,6 +4329,15 @@ int fill_schema_coll_charset_app(THD *thd, TABLE_LIST *tables, COND *cond)
 }
 
 
+static inline void copy_field_as_string(Field *to_field, Field *from_field)
+{
+  char buff[MAX_FIELD_WIDTH];
+  String tmp_str(buff, sizeof(buff), system_charset_info);
+  from_field->val_str(&tmp_str);
+  to_field->store(tmp_str.ptr(), tmp_str.length(), system_charset_info);
+}
+
+
 /**
   @brief Store record into I_S.PARAMETERS table
 
@@ -4490,18 +4504,26 @@ bool store_schema_params(THD *thd, TABLE *table, TABLE *proc_table,
 bool store_schema_proc(THD *thd, TABLE *table, TABLE *proc_table,
                        const char *wild, bool full_access, const char *sp_user)
 {
-  String tmp_string;
-  String sp_db, sp_name, definer;
   MYSQL_TIME time;
   LEX *lex= thd->lex;
   CHARSET_INFO *cs= system_charset_info;
-  get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_DB], &sp_db);
-  get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_NAME], &sp_name);
-  get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_DEFINER],&definer);
+  char sp_db_buff[NAME_LEN + 1], sp_name_buff[NAME_LEN + 1],
+    definer_buff[USERNAME_LENGTH + HOSTNAME_LENGTH + 2],
+    returns_buff[MAX_FIELD_WIDTH];
+
+  String sp_db(sp_db_buff, sizeof(sp_db_buff), cs);
+  String sp_name(sp_name_buff, sizeof(sp_name_buff), cs);
+  String definer(definer_buff, sizeof(definer_buff), cs);
+  String returns(returns_buff, sizeof(returns_buff), cs);
+
+  proc_table->field[MYSQL_PROC_FIELD_DB]->val_str(&sp_db);
+  proc_table->field[MYSQL_PROC_FIELD_NAME]->val_str(&sp_name);
+  proc_table->field[MYSQL_PROC_FIELD_DEFINER]->val_str(&definer);
+
   if (!full_access)
-    full_access= !strcmp(sp_user, definer.ptr());
-  if (!full_access && 
-      check_some_routine_access(thd, sp_db.ptr(), sp_name.ptr(),
+    full_access= !strcmp(sp_user, definer.c_ptr_safe());
+  if (!full_access &&
+      check_some_routine_access(thd, sp_db.c_ptr_safe(), sp_name.c_ptr_safe(),
                                 proc_table->field[MYSQL_PROC_MYSQL_TYPE]->
                                 val_int() == TYPE_ENUM_PROCEDURE))
     return 0;
@@ -4515,32 +4537,30 @@ bool store_schema_proc(THD *thd, TABLE *table, TABLE *proc_table,
       (sql_command_flags[lex->sql_command] & CF_STATUS_COMMAND) == 0)
   {
     restore_record(table, s->default_values);
-    if (!wild || !wild[0] || !wild_compare(sp_name.ptr(), wild, 0))
+    if (!wild || !wild[0] || !wild_compare(sp_name.c_ptr_safe(), wild, 0))
     {
       int enum_idx= (int) proc_table->field[MYSQL_PROC_FIELD_ACCESS]->val_int();
       table->field[3]->store(sp_name.ptr(), sp_name.length(), cs);
-      get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_SPECIFIC_NAME],
-                &tmp_string);
-      table->field[0]->store(tmp_string.ptr(), tmp_string.length(), cs);
+
+      copy_field_as_string(table->field[0],
+                           proc_table->field[MYSQL_PROC_FIELD_SPECIFIC_NAME]);
       table->field[1]->store(STRING_WITH_LEN("def"), cs);
       table->field[2]->store(sp_db.ptr(), sp_db.length(), cs);
-      get_field(thd->mem_root, proc_table->field[MYSQL_PROC_MYSQL_TYPE],
-                &tmp_string);
-      table->field[4]->store(tmp_string.ptr(), tmp_string.length(), cs);
+      copy_field_as_string(table->field[4],
+                           proc_table->field[MYSQL_PROC_MYSQL_TYPE]);
+
       if (proc_table->field[MYSQL_PROC_MYSQL_TYPE]->val_int() ==
           TYPE_ENUM_FUNCTION)
       {
         sp_head *sp;
         bool free_sp_head;
-        get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_RETURNS],
-                  &tmp_string);
-
+        proc_table->field[MYSQL_PROC_FIELD_RETURNS]->val_str(&returns);
         sp= sp_load_for_information_schema(thd, proc_table, &sp_db, &sp_name,
                                            (ulong) proc_table->
                                            field[MYSQL_PROC_FIELD_SQL_MODE]->
                                            val_int(),
                                            TYPE_ENUM_FUNCTION,
-                                           tmp_string.c_ptr_safe(),
+                                           returns.c_ptr_safe(),
                                            "", &free_sp_head);
 
         if (sp)
@@ -4571,24 +4591,19 @@ bool store_schema_proc(THD *thd, TABLE *table, TABLE *proc_table,
 
       if (full_access)
       {
-        get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_BODY_UTF8],
-                  &tmp_string);
-        table->field[14]->store(tmp_string.ptr(), tmp_string.length(), cs);
+        copy_field_as_string(table->field[14],
+                             proc_table->field[MYSQL_PROC_FIELD_BODY_UTF8]);
         table->field[14]->set_notnull();
       }
       table->field[13]->store(STRING_WITH_LEN("SQL"), cs);
       table->field[17]->store(STRING_WITH_LEN("SQL"), cs);
-
-
-      get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_DETERMINISTIC],
-                &tmp_string);
-      table->field[18]->store(tmp_string.ptr(), tmp_string.length(), cs);
+      copy_field_as_string(table->field[18],
+                           proc_table->field[MYSQL_PROC_FIELD_DETERMINISTIC]);
       table->field[19]->store(sp_data_access_name[enum_idx].str, 
                               sp_data_access_name[enum_idx].length , cs);
+      copy_field_as_string(table->field[21],
+                           proc_table->field[MYSQL_PROC_FIELD_SECURITY_TYPE]);
 
-      get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_SECURITY_TYPE],
-                &tmp_string);
-      table->field[21]->store(tmp_string.ptr(), tmp_string.length(), cs);
       bzero((char *)&time, sizeof(time));
       ((Field_timestamp *) proc_table->field[MYSQL_PROC_FIELD_CREATED])->
         get_time(&time);
@@ -4597,29 +4612,20 @@ bool store_schema_proc(THD *thd, TABLE *table, TABLE *proc_table,
       ((Field_timestamp *) proc_table->field[MYSQL_PROC_FIELD_MODIFIED])->
         get_time(&time);
       table->field[23]->store_time(&time, MYSQL_TIMESTAMP_DATETIME);
+      copy_field_as_string(table->field[24],
+                           proc_table->field[MYSQL_PROC_FIELD_SQL_MODE]);
+      copy_field_as_string(table->field[25],
+                           proc_table->field[MYSQL_PROC_FIELD_COMMENT]);
 
-      get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_SQL_MODE],
-                &tmp_string);
-      table->field[24]->store(tmp_string.ptr(), tmp_string.length(), cs);
-
-      get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_COMMENT],
-                &tmp_string);
-      table->field[25]->store(tmp_string.ptr(), tmp_string.length(), cs);
       table->field[26]->store(definer.ptr(), definer.length(), cs);
-
-      get_field(thd->mem_root,
-                proc_table->field[MYSQL_PROC_FIELD_CHARACTER_SET_CLIENT],
-                &tmp_string);
-      table->field[27]->store(tmp_string.ptr(), tmp_string.length(), cs);
-
-      get_field(thd->mem_root,
-                proc_table->field[ MYSQL_PROC_FIELD_COLLATION_CONNECTION],
-                &tmp_string);
-      table->field[28]->store(tmp_string.ptr(), tmp_string.length(), cs);
-
-      get_field(thd->mem_root, proc_table->field[MYSQL_PROC_FIELD_DB_COLLATION],
-                &tmp_string);
-      table->field[29]->store(tmp_string.ptr(), tmp_string.length(), cs);
+      copy_field_as_string(table->field[27],
+                           proc_table->
+                           field[MYSQL_PROC_FIELD_CHARACTER_SET_CLIENT]);
+      copy_field_as_string(table->field[28],
+                           proc_table->
+                           field[MYSQL_PROC_FIELD_COLLATION_CONNECTION]);
+      copy_field_as_string(table->field[29],
+			   proc_table->field[MYSQL_PROC_FIELD_DB_COLLATION]);
 
       return schema_table_store_record(thd, table);
     }
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index f8d99d87228..5052e29ac30 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -426,6 +426,25 @@ uint filename_to_tablename(const char *from, char *to, uint to_length)
 
 
 /**
+  Check if given string begins with "#mysql50#" prefix
+  
+  @param   name          string to check cut 
+  
+  @retval
+    FALSE  no prefix found
+  @retval
+    TRUE   prefix found
+*/
+
+bool check_mysql50_prefix(const char *name)
+{
+  return (name[0] == '#' && 
+         !strncmp(name, MYSQL50_TABLE_NAME_PREFIX,
+                  MYSQL50_TABLE_NAME_PREFIX_LENGTH));
+}
+
+
+/**
   Check if given string begins with "#mysql50#" prefix, cut it if so.
   
   @param   from          string to check and cut 
@@ -440,9 +459,7 @@ uint filename_to_tablename(const char *from, char *to, uint to_length)
 
 uint check_n_cut_mysql50_prefix(const char *from, char *to, uint to_length)
 {
-  if (from[0] == '#' && 
-      !strncmp(from, MYSQL50_TABLE_NAME_PREFIX,
-               MYSQL50_TABLE_NAME_PREFIX_LENGTH))
+  if (check_mysql50_prefix(from))
     return (uint) (strmake(to, from + MYSQL50_TABLE_NAME_PREFIX_LENGTH,
                            to_length - 1) - to);
   return 0;
@@ -469,7 +486,21 @@ uint tablename_to_filename(const char *from, char *to, uint to_length)
   DBUG_PRINT("enter", ("from '%s'", from));
 
   if ((length= check_n_cut_mysql50_prefix(from, to, to_length)))
+  {
+    /*
+      Check if the name supplied is a valid mysql 5.0 name and 
+      make the name a zero length string if it's not.
+      Note that just returning zero length is not enough : 
+      a lot of places don't check the return value and expect 
+      a zero terminated string.
+    */  
+    if (check_table_name(to, length, TRUE))
+    {
+      to[0]= 0;
+      length= 0;
+    }
     DBUG_RETURN(length);
+  }
   length= strconvert(system_charset_info, from,
                      &my_charset_filename, to, to_length, &errors);
   if (check_if_legal_tablename(to) &&
@@ -6939,7 +6970,14 @@ view_err:
                        &index_add_buffer, &index_add_count,
                        &candidate_key_count))
       goto err;
-
+   
+    DBUG_EXECUTE_IF("alter_table_only_metadata_change", {
+      if (need_copy_table_res != ALTER_TABLE_METADATA_ONLY)
+        goto err; });
+    DBUG_EXECUTE_IF("alter_table_only_index_change", {
+      if (need_copy_table_res != ALTER_TABLE_INDEX_CHANGED)
+        goto err; });
+   
     if (need_copy_table == ALTER_TABLE_METADATA_ONLY)
       need_copy_table= need_copy_table_res;
   }
diff --git a/sql/sql_table.h b/sql/sql_table.h
index 3114876f5ed..40b24605bd6 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -126,6 +126,7 @@ enum enum_explain_filename_mode
 uint filename_to_tablename(const char *from, char *to, uint to_length);
 uint tablename_to_filename(const char *from, char *to, uint to_length);
 uint check_n_cut_mysql50_prefix(const char *from, char *to, uint to_length);
+bool check_mysql50_prefix(const char *name);
 uint build_table_filename(char *buff, size_t bufflen, const char *db,
                           const char *table, const char *ext, uint flags);
 uint build_table_shadow_filename(char *buff, size_t bufflen,
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 300ca1098fb..9adfe896c73 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -436,6 +436,7 @@ int mysql_update(THD *thd,
       {
 	goto err;
       }
+      thd->examined_row_count+= examined_rows;
       /*
 	Filesort has already found and selected the rows we want to update,
 	so we don't need the where clause
@@ -482,6 +483,7 @@ int mysql_update(THD *thd,
 
       while (!(error=info.read_record(&info)) && !thd->killed)
       {
+        thd->examined_row_count++;
 	if (!(select && select->skip_record()))
 	{
           if (table->file->was_semi_consistent_read())
@@ -588,6 +590,7 @@ int mysql_update(THD *thd,
 
   while (!(error=info.read_record(&info)) && !thd->killed)
   {
+    thd->examined_row_count++;
     if (!(select && select->skip_record()))
     {
       if (table->file->was_semi_consistent_read())
@@ -1336,6 +1339,16 @@ int multi_update::prepare(List<Item> &not_used_values,
     {
       table->read_set= &table->def_read_set;
       bitmap_union(table->read_set, &table->tmp_set);
+      /*
+        If a timestamp field settable on UPDATE is present then to avoid wrong
+        update force the table handler to retrieve write-only fields to be able
+        to compare records and detect data change.
+        */
+      if (table->file->ha_table_flags() & HA_PARTIAL_COLUMN_READ &&
+          table->timestamp_field &&
+          (table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_UPDATE ||
+           table->timestamp_field_type == TIMESTAMP_AUTO_SET_ON_BOTH))
+        bitmap_union(table->read_set, table->write_set);
     }
   }
   
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index aa336f3c072..9f20a4ccf71 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -6670,7 +6670,7 @@ alter_list_item:
             {
               MYSQL_YYABORT;
             }
-            if (check_table_name($3->table.str,$3->table.length) ||
+            if (check_table_name($3->table.str,$3->table.length, FALSE) ||
                 ($3->db.str && check_db_name(&$3->db)))
             {
               my_error(ER_WRONG_TABLE_NAME, MYF(0), $3->table.str);
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index aeb9dda3a8b..5c1a44ba844 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -1595,6 +1595,13 @@ static Sys_var_mybool Sys_skip_networking(
        READ_ONLY GLOBAL_VAR(opt_disable_networking), CMD_LINE(OPT_ARG),
        DEFAULT(FALSE));
 
+static Sys_var_mybool Sys_skip_name_resolve(
+       "skip_name_resolve",
+       "Don't resolve hostnames. All hostnames are IP's or 'localhost'.",
+       READ_ONLY GLOBAL_VAR(opt_skip_name_resolve),
+       CMD_LINE(OPT_ARG, OPT_SKIP_RESOLVE),
+       DEFAULT(FALSE));
+
 static Sys_var_mybool Sys_skip_show_database(
        "skip_show_database", "Don't allow 'SHOW DATABASE' commands",
        READ_ONLY GLOBAL_VAR(opt_skip_show_db), CMD_LINE(OPT_ARG),
diff --git a/sql/table.cc b/sql/table.cc
index 5ffe1cf17ae..b104c212593 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -311,13 +311,6 @@ TABLE_SHARE *alloc_table_share(TABLE_LIST *table_list, char *key,
     share->version=       refresh_version;
 
     /*
-      This constant is used to mark that no table map version has been
-      assigned.  No arithmetic is done on the value: it will be
-      overwritten with a value taken from MYSQL_BIN_LOG.
-    */
-    share->table_map_version= ~(ulonglong)0;
-
-    /*
       Since alloc_table_share() can be called without any locking (for
       example, ha_create_table... functions), we do not assign a table
       map id here.  Instead we assign a value that is not used
@@ -383,11 +376,6 @@ void init_tmp_table_share(THD *thd, TABLE_SHARE *share, const char *key,
   share->path.length= share->normalized_path.length= strlen(path);
   share->frm_version= 		 FRM_VER_TRUE_VARCHAR;
 
-  /*
-    Temporary tables are not replicated, but we set up these fields
-    anyway to be able to catch errors.
-   */
-  share->table_map_version= ~(ulonglong)0;
   share->cached_row_logging_check= -1;
 
   /*
@@ -500,6 +488,26 @@ inline bool is_system_table_name(const char *name, uint length)
 }
 
 
+/**
+  Check if a string contains path elements
+*/  
+
+static inline bool has_disabled_path_chars(const char *str)
+{
+  for (; *str; str++)
+    switch (*str)
+    {
+      case FN_EXTCHAR:
+      case '/':
+      case '\\':
+      case '~':
+      case '@':
+        return TRUE;
+    }
+  return FALSE;
+}
+
+
 /*
   Read table definition from a binary / text based .frm file
   
@@ -556,7 +564,8 @@ int open_table_def(THD *thd, TABLE_SHARE *share, uint db_flags)
         This kind of tables must have been opened only by the
         mysql_file_open() above.
     */
-    if (strchr(share->table_name.str, '@') ||
+    if (has_disabled_path_chars(share->table_name.str) ||
+        has_disabled_path_chars(share->db.str) ||
         !strncmp(share->db.str, MYSQL50_TABLE_NAME_PREFIX,
                  MYSQL50_TABLE_NAME_PREFIX_LENGTH) ||
         !strncmp(share->table_name.str, MYSQL50_TABLE_NAME_PREFIX,
@@ -2740,35 +2749,21 @@ bool check_db_name(LEX_STRING *org_name)
 {
   char *name= org_name->str;
   uint name_length= org_name->length;
+  bool check_for_path_chars;
 
   if (!name_length || name_length > NAME_LEN)
     return 1;
 
+  if ((check_for_path_chars= check_mysql50_prefix(name)))
+  {
+    name+= MYSQL50_TABLE_NAME_PREFIX_LENGTH;
+    name_length-= MYSQL50_TABLE_NAME_PREFIX_LENGTH;
+  }
+
   if (lower_case_table_names && name != any_db)
     my_casedn_str(files_charset_info, name);
 
-#if defined(USE_MB) && defined(USE_MB_IDENT)
-  if (use_mb(system_charset_info))
-  {
-    name_length= 0;
-    bool last_char_is_space= TRUE;
-    char *end= name + org_name->length;
-    while (name < end)
-    {
-      int len;
-      last_char_is_space= my_isspace(system_charset_info, *name);
-      len= my_ismbchar(system_charset_info, name, end);
-      if (!len)
-        len= 1;
-      name+= len;
-      name_length++;
-    }
-    return (last_char_is_space || name_length > NAME_CHAR_LEN);
-  }
-  else
-#endif
-    return ((org_name->str[org_name->length - 1] != ' ') ||
-            (name_length > NAME_CHAR_LEN)); /* purecov: inspected */
+  return check_table_name(name, name_length, check_for_path_chars);
 }
 
 
@@ -2778,8 +2773,7 @@ bool check_db_name(LEX_STRING *org_name)
   returns 1 on error
 */
 
-
-bool check_table_name(const char *name, uint length)
+bool check_table_name(const char *name, uint length, bool check_for_path_chars)
 {
   uint name_length= 0;  // name length in symbols
   const char *end= name+length;
@@ -2807,6 +2801,9 @@ bool check_table_name(const char *name, uint length)
       }
     }
 #endif
+    if (check_for_path_chars &&
+        (*name == '/' || *name == '\\' || *name == '~' || *name == FN_EXTCHAR))
+      return 1;
     name++;
     name_length++;
   }
diff --git a/sql/table.h b/sql/table.h
index af357807add..87044ac769b 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -630,7 +630,6 @@ struct TABLE_SHARE
   bool crashed;
   bool is_view;
   ulong table_map_id;                   /* for row-based replication */
-  ulonglong table_map_version;
 
   /*
     Cache for row-based replication table share checks that does not
@@ -1913,7 +1912,11 @@ typedef struct st_nested_join
   List<TABLE_LIST>  join_list;       /* list of elements in the nested join */
   table_map         used_tables;     /* bitmap of tables in the nested join */
   table_map         not_null_tables; /* tables that rejects nulls           */
-  struct st_join_table *first_nested;/* the first nested table in the plan  */
+  /**
+    Used for pointing out the first table in the plan being covered by this
+    join nest. It is used exclusively within make_outerjoin_info().
+   */
+  struct st_join_table *first_nested;
   /* 
     Used to count tables in the nested join in 2 isolated places:
     1. In make_outerjoin_info(). 
@@ -1923,6 +1926,15 @@ typedef struct st_nested_join
   */
   uint              counter;
   nested_join_map   nj_map;          /* Bit used to identify this nested join*/
+  /**
+     True if this join nest node is completely covered by the query execution
+     plan. This means two things.
+
+     1. All tables on its @c join_list are covered by the plan.
+
+     2. All child join nest nodes are fully covered.
+   */
+  bool is_fully_covered() const { return join_list.elements == counter; }
 } NESTED_JOIN;
 
 
@@ -2026,7 +2038,7 @@ void update_create_info_from_table(HA_CREATE_INFO *info, TABLE *form);
 bool check_and_convert_db_name(LEX_STRING *db, bool preserve_lettercase);
 bool check_db_name(LEX_STRING *db);
 bool check_column_name(const char *name);
-bool check_table_name(const char *name, uint length);
+bool check_table_name(const char *name, uint length, bool check_for_path_chars);
 int rename_file_ext(const char * from,const char * to,const char * ext);
 char *get_field(MEM_ROOT *mem, Field *field);
 bool get_field(MEM_ROOT *mem, Field *field, class String *res);