merge

58 files changed, 1280 insertions, 736 deletions

diff --git a/sql/Makefile.am b/sql/Makefile.am
index b7c4bf44111..16161798078 100644
--- a/sql/Makefile.am
+++ b/sql/Makefile.am
@@ -81,7 +81,6 @@ mysqld_DEPENDENCIES=	@mysql_plugin_libs@ $(SUPPORTING_LIBS) libndb.la
 LDADD = $(SUPPORTING_LIBS) @ZLIB_LIBS@ @NDB_SCI_LIBS@
 mysqld_LDADD =		libndb.la \
 			@MYSQLD_EXTRA_LDFLAGS@ \
-			@pstack_libs@ \
 			@mysql_plugin_libs@ \
 			$(LDADD)  $(CXXLDFLAGS) $(WRAPLIBS) @LIBDL@ \
 			$(yassl_libs) $(openssl_libs) @MYSQLD_EXTRA_LIBS@
diff --git a/sql/derror.cc b/sql/derror.cc
index bf8c589a65f..db9cd3c0c58 100644
--- a/sql/derror.cc
+++ b/sql/derror.cc
@@ -35,6 +35,8 @@ static void init_myfunc_errs(void);
 C_MODE_START
 static const char **get_server_errmsgs()
 {
+  if (!current_thd)
+    return DEFAULT_ERRMSGS;
   return CURRENT_THD_ERRMSGS;
 }
 C_MODE_END
diff --git a/sql/events.cc b/sql/events.cc
index 10a7535425f..e7e47801586 100644
--- a/sql/events.cc
+++ b/sql/events.cc
@@ -485,10 +485,7 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
       ret= TRUE;                                // OOM
     else if ((ret= db_repository->load_named_event(thd, dbname, name,
                                                    new_element)))
-    {
-      DBUG_ASSERT(ret == OP_LOAD_ERROR);
       delete new_element;
-    }
     else
     {
       /*
diff --git a/sql/events.h b/sql/events.h
index 1482e736d29..f3ebc6da4ad 100644
--- a/sql/events.h
+++ b/sql/events.h
@@ -44,19 +44,6 @@ class THD;
 typedef class Item COND;
 typedef struct charset_info_st CHARSET_INFO;
 
-/* Return codes */
-enum enum_events_error_code
-{
-  OP_OK= 0,
-  OP_NOT_RUNNING,
-  OP_CANT_KILL,
-  OP_CANT_INIT,
-  OP_DISABLED_EVENT,
-  OP_LOAD_ERROR,
-  OP_ALREADY_EXISTS
-};
-
-
 int
 sortcmp_lex_string(LEX_STRING s, LEX_STRING t, CHARSET_INFO *cs);
 
diff --git a/sql/field.cc b/sql/field.cc
index 0e55b624633..be7441f6bfd 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1564,7 +1564,7 @@ void Field::make_field(Send_field *field)
   }
   else
     field->org_table_name= field->db_name= "";
-  if (orig_table)
+  if (orig_table && orig_table->alias)
   {
     field->table_name= orig_table->alias;
     field->org_col_name= field_name;
@@ -9115,7 +9115,7 @@ void Create_field::init_for_tmp_table(enum_field_types sql_type_arg,
   case MYSQL_TYPE_NEWDECIMAL:
   case MYSQL_TYPE_FLOAT:
   case MYSQL_TYPE_DOUBLE:
-    pack_flag= FIELDFLAG_DECIMAL | FIELDFLAG_NUMBER |
+    pack_flag= FIELDFLAG_NUMBER |
       (decimals_arg & FIELDFLAG_MAX_DEC) << FIELDFLAG_DEC_SHIFT;
     break;
 
@@ -9164,12 +9164,13 @@ void Create_field::init_for_tmp_table(enum_field_types sql_type_arg,
     (maybe_null ? FIELDFLAG_MAYBE_NULL : 0) |
     (is_unsigned ? 0 : FIELDFLAG_DECIMAL);
 
-  DBUG_PRINT("debug", ("pack_flag: %s%s%s%s%s, pack_type: %d",
+  DBUG_PRINT("debug", ("pack_flag: %s%s%s%s%s%s, pack_type: %d",
                        FLAGSTR(pack_flag, FIELDFLAG_BINARY),
                        FLAGSTR(pack_flag, FIELDFLAG_NUMBER),
                        FLAGSTR(pack_flag, FIELDFLAG_INTERVAL),
                        FLAGSTR(pack_flag, FIELDFLAG_GEOM),
                        FLAGSTR(pack_flag, FIELDFLAG_BLOB),
+                       FLAGSTR(pack_flag, FIELDFLAG_DECIMAL),
                        f_packtype(pack_flag)));
   DBUG_VOID_RETURN;
 }
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index e6915b45a97..8bf35f79ba9 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -231,6 +231,8 @@ void ha_partition::init_handler_variables()
   m_innodb= FALSE;
   m_extra_cache= FALSE;
   m_extra_cache_size= 0;
+  m_extra_prepare_for_update= FALSE;
+  m_extra_cache_part_id= NO_CURRENT_PART_ID;
   m_handler_status= handler_not_initialized;
   m_low_byte_first= 1;
   m_part_field_array= NULL;
@@ -5515,9 +5517,6 @@ void ha_partition::get_dynamic_partition_info(PARTITION_STATS *stat_info,
     when performing the sequential scan we will check this recorded value
     and call extra_opt whenever we start scanning a new partition.
 
-    monty: Neads to be fixed so that it's passed to all handlers when we
-    move to another partition during table scan.
-
   HA_EXTRA_NO_CACHE:
     When performing a UNION SELECT HA_EXTRA_NO_CACHE is called from the
     flush method in the select_union class.
@@ -5529,7 +5528,7 @@ void ha_partition::get_dynamic_partition_info(PARTITION_STATS *stat_info,
     for. If no cache is in use they will quickly return after finding
     this out. And we also ensure that all caches are disabled and no one
     is left by mistake.
-    In the future this call will probably be deleted an we will instead call
+    In the future this call will probably be deleted and we will instead call
     ::reset();
 
   HA_EXTRA_WRITE_CACHE:
@@ -5541,8 +5540,9 @@ void ha_partition::get_dynamic_partition_info(PARTITION_STATS *stat_info,
     This is called as part of a multi-table update. When the table to be
     updated is also scanned then this informs MyISAM handler to drop any
     caches if dynamic records are used (fixed size records do not care
-    about this call). We pass this along to all underlying MyISAM handlers
-    and ignore it for the rest.
+    about this call). We pass this along to the first partition to scan, and
+    flag that it is to be called after HA_EXTRA_CACHE when moving to the next
+    partition to scan.
 
   HA_EXTRA_PREPARE_FOR_DROP:
     Only used by MyISAM, called in preparation for a DROP TABLE.
@@ -5697,9 +5697,22 @@ int ha_partition::extra(enum ha_extra_function operation)
   case HA_EXTRA_PREPARE_FOR_RENAME:
     DBUG_RETURN(prepare_for_rename());
     break;
+  case HA_EXTRA_PREPARE_FOR_UPDATE:
+    /*
+      Needs to be run on the first partition in the range now, and 
+      later in late_extra_cache, when switching to a new partition to scan.
+    */
+    m_extra_prepare_for_update= TRUE;
+    if (m_part_spec.start_part != NO_CURRENT_PART_ID)
+    {
+      if (!m_extra_cache)
+        m_extra_cache_part_id= m_part_spec.start_part;
+      DBUG_ASSERT(m_extra_cache_part_id == m_part_spec.start_part);
+      (void) m_file[m_part_spec.start_part]->extra(HA_EXTRA_PREPARE_FOR_UPDATE);
+    }
+    break;
   case HA_EXTRA_NORMAL:
   case HA_EXTRA_QUICK:
-  case HA_EXTRA_PREPARE_FOR_UPDATE:
   case HA_EXTRA_FORCE_REOPEN:
   case HA_EXTRA_PREPARE_FOR_DROP:
   case HA_EXTRA_FLUSH_CACHE:
@@ -5722,10 +5735,22 @@ int ha_partition::extra(enum ha_extra_function operation)
     break;
   }
   case HA_EXTRA_NO_CACHE:
+  {
+    int ret= 0;
+    if (m_extra_cache_part_id != NO_CURRENT_PART_ID)
+      ret= m_file[m_extra_cache_part_id]->extra(HA_EXTRA_NO_CACHE);
+    m_extra_cache= FALSE;
+    m_extra_cache_size= 0;
+    m_extra_prepare_for_update= FALSE;
+    m_extra_cache_part_id= NO_CURRENT_PART_ID;
+    DBUG_RETURN(ret);
+  }
   case HA_EXTRA_WRITE_CACHE:
   {
     m_extra_cache= FALSE;
     m_extra_cache_size= 0;
+    m_extra_prepare_for_update= FALSE;
+    m_extra_cache_part_id= NO_CURRENT_PART_ID;
     DBUG_RETURN(loop_extra(operation));
   }
   case HA_EXTRA_IGNORE_NO_KEY:
@@ -5862,6 +5887,7 @@ int ha_partition::extra_opt(enum ha_extra_function operation, ulong cachesize)
 void ha_partition::prepare_extra_cache(uint cachesize)
 {
   DBUG_ENTER("ha_partition::prepare_extra_cache()");
+  DBUG_PRINT("info", ("cachesize %u", cachesize));
 
   m_extra_cache= TRUE;
   m_extra_cache_size= cachesize;
@@ -5920,16 +5946,18 @@ int ha_partition::loop_extra(enum ha_extra_function operation)
 {
   int result= 0, tmp;
   handler **file;
+  bool is_select;
   DBUG_ENTER("ha_partition::loop_extra()");
   
-  /* 
-    TODO, 5.2: this is where you could possibly add optimisations to add the bitmap
-    _if_ a SELECT.
-  */
+  is_select= (thd_sql_command(ha_thd()) == SQLCOM_SELECT);
   for (file= m_file; *file; file++)
   {
-    if ((tmp= (*file)->extra(operation)))
-      result= tmp;
+    if (!is_select ||
+        bitmap_is_set(&(m_part_info->used_partitions), file - m_file))
+    {
+      if ((tmp= (*file)->extra(operation)))
+        result= tmp;
+    }
   }
   DBUG_RETURN(result);
 }
@@ -5950,14 +5978,25 @@ void ha_partition::late_extra_cache(uint partition_id)
 {
   handler *file;
   DBUG_ENTER("ha_partition::late_extra_cache");
+  DBUG_PRINT("info", ("extra_cache %u prepare %u partid %u size %u",
+                      m_extra_cache, m_extra_prepare_for_update,
+                      partition_id, m_extra_cache_size));
 
-  if (!m_extra_cache)
+  if (!m_extra_cache && !m_extra_prepare_for_update)
     DBUG_VOID_RETURN;
   file= m_file[partition_id];
-  if (m_extra_cache_size == 0)
-    (void) file->extra(HA_EXTRA_CACHE);
-  else
-    (void) file->extra_opt(HA_EXTRA_CACHE, m_extra_cache_size);
+  if (m_extra_cache)
+  {
+    if (m_extra_cache_size == 0)
+      (void) file->extra(HA_EXTRA_CACHE);
+    else
+      (void) file->extra_opt(HA_EXTRA_CACHE, m_extra_cache_size);
+  }
+  if (m_extra_prepare_for_update)
+  {
+    (void) file->extra(HA_EXTRA_PREPARE_FOR_UPDATE);
+  }
+  m_extra_cache_part_id= partition_id;
   DBUG_VOID_RETURN;
 }
 
@@ -5978,10 +6017,12 @@ void ha_partition::late_extra_no_cache(uint partition_id)
   handler *file;
   DBUG_ENTER("ha_partition::late_extra_no_cache");
 
-  if (!m_extra_cache)
+  if (!m_extra_cache && !m_extra_prepare_for_update)
     DBUG_VOID_RETURN;
   file= m_file[partition_id];
   (void) file->extra(HA_EXTRA_NO_CACHE);
+  DBUG_ASSERT(partition_id == m_extra_cache_part_id);
+  m_extra_cache_part_id= NO_CURRENT_PART_ID;
   DBUG_VOID_RETURN;
 }
 
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index cfab2dde394..353e0d17159 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -137,6 +137,10 @@ private:
   */
   bool m_extra_cache;
   uint m_extra_cache_size;
+  /* The same goes for HA_EXTRA_PREPARE_FOR_UPDATE */
+  bool m_extra_prepare_for_update;
+  /* Which partition has active cache */
+  uint m_extra_cache_part_id;
 
   void init_handler_variables();
   /*
diff --git a/sql/item.cc b/sql/item.cc
index 427146e11dc..e782e90b874 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -7287,14 +7287,16 @@ int stored_field_cmp_to_item(THD *thd, Field *field, Item *item)
 
     enum_field_types field_type= field->type();
 
-    if (field_type == MYSQL_TYPE_DATE || field_type == MYSQL_TYPE_DATETIME)
+    if (field_type == MYSQL_TYPE_DATE || field_type == MYSQL_TYPE_DATETIME ||
+        field_type == MYSQL_TYPE_TIMESTAMP)
     {
       enum_mysql_timestamp_type type= MYSQL_TIMESTAMP_ERROR;
 
       if (field_type == MYSQL_TYPE_DATE)
         type= MYSQL_TIMESTAMP_DATE;
 
-      if (field_type == MYSQL_TYPE_DATETIME)
+      if (field_type == MYSQL_TYPE_DATETIME ||
+          field_type == MYSQL_TYPE_TIMESTAMP)
         type= MYSQL_TIMESTAMP_DATETIME;
         
       const char *field_name= field->field_name;
@@ -7850,9 +7852,12 @@ bool Item_cache_row::null_inside()
 
 void Item_cache_row::bring_value()
 {
+  if (!example)
+    return;
+  example->bring_value();
+  null_value= example->null_value;
   for (uint i= 0; i < item_count; i++)
     values[i]->bring_value();
-  return;
 }
 
 
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 8c0f22b0947..b7aad733e67 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -1595,6 +1595,13 @@ int Arg_comparator::compare_row()
   bool was_null= 0;
   (*a)->bring_value();
   (*b)->bring_value();
+
+  if ((*a)->null_value || (*b)->null_value)
+  {
+    owner->null_value= 1;
+    return -1;
+  }
+
   uint n= (*a)->cols();
   for (uint i= 0; i<n; i++)
   {
@@ -1763,6 +1770,76 @@ bool Item_in_optimizer::fix_fields(THD *thd, Item **ref)
 }
 
 
+/**
+   The implementation of optimized \<outer expression\> [NOT] IN \<subquery\>
+   predicates. The implementation works as follows.
+
+   For the current value of the outer expression
+   
+   - If it contains only NULL values, the original (before rewrite by the
+     Item_in_subselect rewrite methods) inner subquery is non-correlated and
+     was previously executed, there is no need to re-execute it, and the
+     previous return value is returned.
+
+   - If it contains NULL values, check if there is a partial match for the
+     inner query block by evaluating it. For clarity we repeat here the
+     transformation previously performed on the sub-query. The expression
+
+     <tt>
+     ( oc_1, ..., oc_n ) 
+     \<in predicate\>
+     ( SELECT ic_1, ..., ic_n
+       FROM \<table\>
+       WHERE \<inner where\> 
+     )
+     </tt>
+
+     was transformed into
+     
+     <tt>
+     ( oc_1, ..., oc_n ) 
+     \<in predicate\>
+     ( SELECT ic_1, ..., ic_n 
+       FROM \<table\> 
+       WHERE \<inner where\> AND ... ( ic_k = oc_k OR ic_k IS NULL ) 
+       HAVING ... NOT ic_k IS NULL
+     )
+     </tt>
+
+     The evaluation will now proceed according to special rules set up
+     elsewhere. These rules include:
+
+     - The HAVING NOT \<inner column\> IS NULL conditions added by the
+       aforementioned rewrite methods will detect whether they evaluated (and
+       rejected) a NULL value and if so, will cause the subquery to evaluate
+       to NULL. 
+
+     - The added WHERE and HAVING conditions are present only for those inner
+       columns that correspond to outer column that are not NULL at the moment.
+     
+     - If there is an eligible index for executing the subquery, the special
+       access method "Full scan on NULL key" is employed which ensures that
+       the inner query will detect if there are NULL values resulting from the
+       inner query. This access method will quietly resort to table scan if it
+       needs to find NULL values as well.
+
+     - Under these conditions, the sub-query need only be evaluated in order to
+       find out whether it produced any rows.
+     
+       - If it did, we know that there was a partial match since there are
+         NULL values in the outer row expression.
+
+       - If it did not, the result is FALSE or UNKNOWN. If at least one of the
+         HAVING sub-predicates rejected a NULL value corresponding to an outer
+         non-NULL, and hence the inner query block returns UNKNOWN upon
+         evaluation, there was a partial match and the result is UNKNOWN.
+
+   - If it contains no NULL values, the call is forwarded to the inner query
+     block.
+
+     @see Item_in_subselect::val_bool()
+     @see Item_is_not_null_test::val_int()
+ */
 longlong Item_in_optimizer::val_int()
 {
   bool tmp;
@@ -1816,7 +1893,7 @@ longlong Item_in_optimizer::val_int()
           all_left_cols_null= false;
       }
 
-      if (!((Item_in_subselect*)args[1])->is_correlated && 
+      if (!item_subs->is_correlated && 
           all_left_cols_null && result_for_null_param != UNKNOWN)
       {
         /* 
@@ -1830,8 +1907,11 @@ longlong Item_in_optimizer::val_int()
       else 
       {
         /* The subquery has to be evaluated */
-        (void) args[1]->val_bool_result();
-        null_value= !item_subs->engine->no_rows();
+        (void) item_subs->val_bool_result();
+        if (item_subs->engine->no_rows())
+          null_value= item_subs->null_value;
+        else
+          null_value= TRUE;
         if (all_left_cols_null)
           result_for_null_param= null_value;
       }
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 6699b2820e6..7906ed71bc7 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -2553,7 +2553,7 @@ void Item_func_min_max::fix_length_and_dec()
     stored to the value pointer, if latter is provided.
 
   RETURN
-   0	If one of arguments is NULL
+   0	If one of arguments is NULL or there was a execution error
    #	index of the least/greatest argument
 */
 
@@ -2567,6 +2567,14 @@ uint Item_func_min_max::cmp_datetimes(ulonglong *value)
     Item **arg= args + i;
     bool is_null;
     longlong res= get_datetime_value(thd, &arg, 0, datetime_item, &is_null);
+
+    /* Check if we need to stop (because of error or KILL)  and stop the loop */
+    if (thd->is_error())
+    {
+      null_value= 1;
+      return 0;
+    }
+
     if ((null_value= args[i]->null_value))
       return 0;
     if (i == 0 || (res < min_max ? cmp_sign : -cmp_sign) > 0)
@@ -2595,6 +2603,12 @@ String *Item_func_min_max::val_str(String *str)
     if (null_value)
       return 0;
     str_res= args[min_max_idx]->val_str(str);
+    if (args[min_max_idx]->null_value)
+    {
+      // check if the call to val_str() above returns a NULL value
+      null_value= 1;
+      return NULL;
+    }
     str_res->set_charset(collation.collation);
     return str_res;
   }
@@ -4623,6 +4637,14 @@ longlong Item_func_set_user_var::val_int_result()
   return entry->val_int(&null_value);
 }
 
+bool Item_func_set_user_var::val_bool_result()
+{
+  DBUG_ASSERT(fixed == 1);
+  check(TRUE);
+  update();					// Store expression
+  return entry->val_int(&null_value) != 0;
+}
+
 String *Item_func_set_user_var::str_result(String *str)
 {
   DBUG_ASSERT(fixed == 1);
@@ -5001,7 +5023,7 @@ void Item_func_get_user_var::fix_length_and_dec()
       decimals=0;
       break;
     case STRING_RESULT:
-      max_length= MAX_BLOB_WIDTH;
+      max_length= MAX_BLOB_WIDTH - 1;
       break;
     case DECIMAL_RESULT:
       fix_char_length(DECIMAL_MAX_STR_LENGTH);
diff --git a/sql/item_func.h b/sql/item_func.h
index 834ecd60e21..15730516c71 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -1416,6 +1416,7 @@ public:
   my_decimal *val_decimal(my_decimal *);
   double val_result();
   longlong val_int_result();
+  bool val_bool_result();
   String *str_result(String *str);
   my_decimal *val_decimal_result(my_decimal *);
   bool is_null_result();
diff --git a/sql/item_geofunc.h b/sql/item_geofunc.h
index 25755de1e2c..84034841ad5 100644
--- a/sql/item_geofunc.h
+++ b/sql/item_geofunc.h
@@ -179,6 +179,21 @@ public:
     item_type=it;
   }
   String *val_str(String *);
+  void fix_length_and_dec()
+  {
+    for (unsigned int i= 0; i < arg_count; ++i)
+    {
+      if (args[i]->fixed && args[i]->field_type() != MYSQL_TYPE_GEOMETRY)
+      {
+        String str;
+        args[i]->print(&str, QT_ORDINARY);
+        str.append('\0');
+        my_error(ER_ILLEGAL_VALUE_FOR_TYPE, MYF(0), "non geometric",
+                 str.ptr());
+      }
+    }
+  }
+ 
   const char *func_name() const { return "multipoint"; }
 };
 
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 809926338af..56bd8c75a20 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -55,7 +55,7 @@ Item_subselect::Item_subselect():
     item value is NULL if select_subselect not changed this value
     (i.e. some rows will be found returned)
   */
-  null_value= 1;
+  null_value= TRUE;
 }
 
 
@@ -435,9 +435,9 @@ void Item_maxmin_subselect::print(String *str, enum_query_type query_type)
 
 void Item_singlerow_subselect::reset()
 {
-  null_value= 1;
+  null_value= TRUE;
   if (value)
-    value->null_value= 1;
+    value->null_value= TRUE;
 }
 
 
@@ -574,7 +574,10 @@ bool Item_singlerow_subselect::null_inside()
 
 void Item_singlerow_subselect::bring_value()
 {
-  exec();
+  if (!exec() && assigned())
+    null_value= 0;
+  else
+    reset();
 }
 
 double Item_singlerow_subselect::val_real()
@@ -582,7 +585,7 @@ double Item_singlerow_subselect::val_real()
   DBUG_ASSERT(fixed == 1);
   if (!exec() && !value->null_value)
   {
-    null_value= 0;
+    null_value= FALSE;
     return value->val_real();
   }
   else
@@ -597,7 +600,7 @@ longlong Item_singlerow_subselect::val_int()
   DBUG_ASSERT(fixed == 1);
   if (!exec() && !value->null_value)
   {
-    null_value= 0;
+    null_value= FALSE;
     return value->val_int();
   }
   else
@@ -611,7 +614,7 @@ String *Item_singlerow_subselect::val_str(String *str)
 {
   if (!exec() && !value->null_value)
   {
-    null_value= 0;
+    null_value= FALSE;
     return value->val_str(str);
   }
   else
@@ -626,7 +629,7 @@ my_decimal *Item_singlerow_subselect::val_decimal(my_decimal *decimal_value)
 {
   if (!exec() && !value->null_value)
   {
-    null_value= 0;
+    null_value= FALSE;
     return value->val_decimal(decimal_value);
   }
   else
@@ -641,7 +644,7 @@ bool Item_singlerow_subselect::val_bool()
 {
   if (!exec() && !value->null_value)
   {
-    null_value= 0;
+    null_value= FALSE;
     return value->val_bool();
   }
   else
@@ -659,7 +662,7 @@ Item_exists_subselect::Item_exists_subselect(st_select_lex *select_lex):
   bool val_bool();
   init(select_lex, new select_exists_subselect(this));
   max_columns= UINT_MAX;
-  null_value= 0; //can't be NULL
+  null_value= FALSE; //can't be NULL
   maybe_null= 0; //can't be NULL
   value= 0;
   DBUG_VOID_RETURN;
@@ -822,15 +825,14 @@ double Item_in_subselect::val_real()
   */
   DBUG_ASSERT(0);
   DBUG_ASSERT(fixed == 1);
-  null_value= 0;
+  null_value= was_null= FALSE;
   if (exec())
   {
     reset();
-    null_value= 1;
     return 0;
   }
   if (was_null && !value)
-    null_value= 1;
+    null_value= TRUE;
   return (double) value;
 }
 
@@ -843,15 +845,14 @@ longlong Item_in_subselect::val_int()
   */
   DBUG_ASSERT(0);
   DBUG_ASSERT(fixed == 1);
-  null_value= 0;
+  null_value= was_null= FALSE;
   if (exec())
   {
     reset();
-    null_value= 1;
     return 0;
   }
   if (was_null && !value)
-    null_value= 1;
+    null_value= TRUE;
   return value;
 }
 
@@ -864,16 +865,15 @@ String *Item_in_subselect::val_str(String *str)
   */
   DBUG_ASSERT(0);
   DBUG_ASSERT(fixed == 1);
-  null_value= 0;
+  null_value= was_null= FALSE;
   if (exec())
   {
     reset();
-    null_value= 1;
     return 0;
   }
   if (was_null && !value)
   {
-    null_value= 1;
+    null_value= TRUE;
     return 0;
   }
   str->set((ulonglong)value, &my_charset_bin);
@@ -884,20 +884,14 @@ String *Item_in_subselect::val_str(String *str)
 bool Item_in_subselect::val_bool()
 {
   DBUG_ASSERT(fixed == 1);
-  null_value= 0;
+  null_value= was_null= FALSE;
   if (exec())
   {
     reset();
-    /* 
-      Must mark the IN predicate as NULL so as to make sure an enclosing NOT
-      predicate will return FALSE. See the comments in 
-      subselect_uniquesubquery_engine::copy_ref_key for further details.
-    */
-    null_value= 1;
     return 0;
   }
   if (was_null && !value)
-    null_value= 1;
+    null_value= TRUE;
   return value;
 }
 
@@ -908,16 +902,15 @@ my_decimal *Item_in_subselect::val_decimal(my_decimal *decimal_value)
     method should not be used
   */
   DBUG_ASSERT(0);
-  null_value= 0;
+  null_value= was_null= FALSE;
   DBUG_ASSERT(fixed == 1);
   if (exec())
   {
     reset();
-    null_value= 1;
     return 0;
   }
   if (was_null && !value)
-    null_value= 1;
+    null_value= TRUE;
   int2my_decimal(E_DEC_FATAL_ERROR, value, 0, decimal_value);
   return decimal_value;
 }
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index 1c2c1cfe512..24cf4da0a95 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -2338,8 +2338,6 @@ void Item_extract::print(String *str, enum_query_type query_type)
 
 void Item_extract::fix_length_and_dec()
 {
-  value.alloc(32);				// alloc buffer
-
   maybe_null=1;					// If wrong date
   switch (int_type) {
   case INTERVAL_YEAR:		max_length=4; date_value=1; break;
@@ -2382,6 +2380,8 @@ longlong Item_extract::val_int()
   }
   else
   {
+    char buf[40];
+    String value(buf, sizeof(buf), &my_charset_bin);;
     String *res= args[0]->val_str(&value);
     if (!res ||
         str_to_time_with_warn(res->charset(), res->ptr(), res->length(),
@@ -3368,6 +3368,8 @@ void Item_func_str_to_date::fix_length_and_dec()
   cached_field_type= MYSQL_TYPE_DATETIME;
   max_length= MAX_DATETIME_FULL_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
   cached_timestamp_type= MYSQL_TIMESTAMP_NONE;
+  sql_mode= (current_thd->variables.sql_mode &
+             (MODE_NO_ZERO_IN_DATE | MODE_NO_ZERO_DATE));
   if ((const_item= args[1]->const_item()))
   {
     char format_buff[64];
@@ -3433,6 +3435,14 @@ bool Item_func_str_to_date::get_date(MYSQL_TIME *ltime, uint fuzzy_date)
   return 0;
 
 null_date:
+  if (fuzzy_date & TIME_NO_ZERO_DATE)
+  {
+    char buff[128];
+    strmake(buff, val->ptr(), min(val->length(), sizeof(buff)-1));
+    push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                        ER_WRONG_VALUE_FOR_TYPE, ER(ER_WRONG_VALUE_FOR_TYPE),
+                        "datetime", buff, "str_to_date");
+  }
   return (null_value=1);
 }
 
@@ -3442,7 +3452,7 @@ String *Item_func_str_to_date::val_str(String *str)
   DBUG_ASSERT(fixed == 1);
   MYSQL_TIME ltime;
 
-  if (Item_func_str_to_date::get_date(&ltime, TIME_FUZZY_DATE))
+  if (Item_func_str_to_date::get_date(&ltime, TIME_FUZZY_DATE | sql_mode))
     return 0;
 
   if (!make_datetime((const_item ? cached_format_type :
@@ -3453,6 +3463,29 @@ String *Item_func_str_to_date::val_str(String *str)
 }
 
 
+longlong Item_func_str_to_date::val_int()
+{
+  DBUG_ASSERT(fixed == 1);
+  MYSQL_TIME ltime;
+
+  if (Item_func_str_to_date::get_date(&ltime, TIME_FUZZY_DATE | sql_mode))
+    return 0;
+
+  if (const_item)
+  {
+    switch (cached_field_type) {
+    case MYSQL_TYPE_DATE:
+      return TIME_to_ulonglong_date(&ltime);
+    case MYSQL_TYPE_TIME:
+      return TIME_to_ulonglong_time(&ltime);
+    default:
+      return TIME_to_ulonglong_datetime(&ltime);
+    }
+  }
+  return TIME_to_ulonglong_datetime(&ltime);
+}
+
+
 bool Item_func_last_day::get_date(MYSQL_TIME *ltime, uint fuzzy_date)
 {
   if (get_arg0_date(ltime, fuzzy_date & ~TIME_FUZZY_DATE) ||
diff --git a/sql/item_timefunc.h b/sql/item_timefunc.h
index 2c08853fda4..27dfd548f73 100644
--- a/sql/item_timefunc.h
+++ b/sql/item_timefunc.h
@@ -733,7 +733,6 @@ public:
 
 class Item_extract :public Item_int_func
 {
-  String value;
   bool date_value;
  public:
   const interval_type int_type; // keep it public
@@ -1039,6 +1038,7 @@ class Item_func_str_to_date :public Item_str_func
   date_time_format_types cached_format_type;
   timestamp_type cached_timestamp_type;
   bool const_item;
+  ulonglong sql_mode;
 public:
   Item_func_str_to_date(Item *a, Item *b)
     :Item_str_func(a, b), const_item(false)
@@ -1052,6 +1052,8 @@ public:
   {
     return tmp_table_field_from_field_type(table, 1);
   }
+  longlong val_int();
+  bool result_as_longlong() { return TRUE; }
 };
 
 
diff --git a/sql/log.cc b/sql/log.cc
index 88352706318..0e15c3b8e79 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1907,7 +1907,9 @@ static int binlog_savepoint_set(handlerton *hton, THD *thd, void *sv)
 
   String log_query;
   if (log_query.append(STRING_WITH_LEN("SAVEPOINT ")) ||
-      log_query.append(thd->lex->ident.str, thd->lex->ident.length))
+      log_query.append("`") ||
+      log_query.append(thd->lex->ident.str, thd->lex->ident.length) ||
+      log_query.append("`"))
     DBUG_RETURN(1);
   int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
   Query_log_event qinfo(thd, log_query.c_ptr_safe(), log_query.length(),
@@ -1929,7 +1931,9 @@ static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
   {
     String log_query;
     if (log_query.append(STRING_WITH_LEN("ROLLBACK TO ")) ||
-        log_query.append(thd->lex->ident.str, thd->lex->ident.length))
+        log_query.append("`") ||
+        log_query.append(thd->lex->ident.str, thd->lex->ident.length) ||
+        log_query.append("`"))
       DBUG_RETURN(1);
     int errcode= query_error_code(thd, thd->killed == THD::NOT_KILLED);
     Query_log_event qinfo(thd, log_query.c_ptr_safe(), log_query.length(),
@@ -5284,11 +5288,8 @@ int MYSQL_BIN_LOG::wait_for_update_bin_log(THD* thd,
                                            const struct timespec *timeout)
 {
   int ret= 0;
-  const char* old_msg = thd->proc_info;
   DBUG_ENTER("wait_for_update_bin_log");
-  old_msg= thd->enter_cond(&update_cond, &LOCK_log,
-                           "Master has sent all binlog to slave; "
-                           "waiting for binlog to be updated");
+
   if (!timeout)
     mysql_cond_wait(&update_cond, &LOCK_log);
   else
@@ -5452,10 +5453,32 @@ extern "C" my_bool reopen_fstreams(const char *filename,
                                    FILE *outstream, FILE *errstream)
 {
   int handle_fd;
-  int stream_fd;
+  int err_fd, out_fd;
   HANDLE osfh;
 
-  DBUG_ASSERT(filename && (outstream || errstream));
+  DBUG_ASSERT(filename && errstream);
+
+  // Services don't have stdout/stderr on Windows, so _fileno returns -1.
+  err_fd= _fileno(errstream);
+  if (err_fd < 0)
+  {
+    if (!freopen(filename, "a+", errstream))
+      return TRUE;
+
+    setbuf(errstream, NULL);
+    err_fd= _fileno(errstream);
+  }
+
+  if (outstream)
+  {
+    out_fd= _fileno(outstream);
+    if (out_fd < 0)
+    {
+      if (!freopen(filename, "a+", outstream))
+        return TRUE;
+      out_fd= _fileno(outstream);
+    }
+  }
 
   if ((osfh= CreateFile(filename, GENERIC_READ | GENERIC_WRITE,
                         FILE_SHARE_READ | FILE_SHARE_WRITE |
@@ -5471,24 +5494,16 @@ extern "C" my_bool reopen_fstreams(const char *filename,
     return TRUE;
   }
 
-  if (outstream)
+  if (_dup2(handle_fd, err_fd) < 0)
   {
-    stream_fd= _fileno(outstream);
-    if (_dup2(handle_fd, stream_fd) < 0)
-    {
-      CloseHandle(osfh);
-      return TRUE;
-    }
+    CloseHandle(osfh);
+    return TRUE;
   }
 
-  if (errstream)
+  if (outstream && _dup2(handle_fd, out_fd) < 0)
   {
-    stream_fd= _fileno(errstream);
-    if (_dup2(handle_fd, stream_fd) < 0)
-    {
-      CloseHandle(osfh);
-      return TRUE;
-    }
+    CloseHandle(osfh);
+    return TRUE;
   }
 
   _close(handle_fd);
diff --git a/sql/log.h b/sql/log.h
index ebb23deda81..89b3594cd1e 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -396,8 +396,8 @@ public:
 
   bool write(Log_event* event_info); // binary log write
   bool write(THD *thd, IO_CACHE *cache, Log_event *commit_event, bool incident);
-
   bool write_incident(THD *thd, bool lock);
+
   int  write_cache(IO_CACHE *cache, bool lock_log, bool flush_and_sync);
   void set_write_error(THD *thd);
   bool check_write_error(THD *thd);
@@ -460,6 +460,7 @@ public:
   inline char* get_log_fname() { return log_file_name; }
   inline char* get_name() { return name; }
   inline mysql_mutex_t* get_log_lock() { return &LOCK_log; }
+  inline mysql_cond_t* get_log_cond() { return &update_cond; }
   inline IO_CACHE* get_log_file() { return &log_file; }
 
   inline void lock_index() { mysql_mutex_lock(&LOCK_index);}
diff --git a/sql/mdl.cc b/sql/mdl.cc
index aa7c2a4b7f2..d53ddcee0c8 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -124,6 +124,7 @@ public:
   Deadlock_detection_visitor(MDL_context *start_node_arg)
     : m_start_node(start_node_arg),
       m_victim(NULL),
+      m_current_search_depth(0),
       m_found_deadlock(FALSE)
   {}
   virtual bool enter_node(MDL_context *node);
@@ -132,8 +133,6 @@ public:
   virtual bool inspect_edge(MDL_context *dest);
 
   MDL_context *get_victim() const { return m_victim; }
-
-  void abort_traversal(MDL_context *node);
 private:
   /**
     Change the deadlock victim to a new one if it has lower deadlock
@@ -148,6 +147,13 @@ private:
   MDL_context *m_start_node;
   /** If a deadlock is found, the context that identifies the victim. */
   MDL_context *m_victim;
+  /** Set to the 0 at start. Increased whenever
+    we descend into another MDL context (aka traverse to the next
+    wait-for graph node). When MAX_SEARCH_DEPTH is reached, we
+    assume that a deadlock is found, even if we have not found a
+    loop.
+  */
+  uint m_current_search_depth;
   /** TRUE if we found a deadlock. */
   bool m_found_deadlock;
   /**
@@ -181,7 +187,7 @@ private:
 
 bool Deadlock_detection_visitor::enter_node(MDL_context *node)
 {
-  m_found_deadlock= m_current_search_depth >= MAX_SEARCH_DEPTH;
+  m_found_deadlock= ++m_current_search_depth >= MAX_SEARCH_DEPTH;
   if (m_found_deadlock)
   {
     DBUG_ASSERT(! m_victim);
@@ -201,6 +207,7 @@ bool Deadlock_detection_visitor::enter_node(MDL_context *node)
 
 void Deadlock_detection_visitor::leave_node(MDL_context *node)
 {
+  --m_current_search_depth;
   if (m_found_deadlock)
     opt_change_victim_to(node);
 }
@@ -245,21 +252,6 @@ Deadlock_detection_visitor::opt_change_victim_to(MDL_context *new_victim)
 
 
 /**
-  Abort traversal of a wait-for graph and report a deadlock.
-
-  @param node Node which we were about to visit when abort
-              was initiated.
-*/
-
-void Deadlock_detection_visitor::abort_traversal(MDL_context *node)
-{
-  DBUG_ASSERT(! m_victim);
-  m_found_deadlock= TRUE;
-  opt_change_victim_to(node);
-}
-
-
-/**
   Get a bit corresponding to enum_mdl_type value in a granted/waiting bitmaps
   and compatibility matrices.
 */
@@ -2064,13 +2056,8 @@ bool MDL_lock::visit_subgraph(MDL_ticket *waiting_ticket,
     are visiting it but this is OK: in the worst case we might do some
     extra work and one more context might be chosen as a victim.
   */
-  ++gvisitor->m_current_search_depth;
-
   if (gvisitor->enter_node(src_ctx))
-  {
-    --gvisitor->m_current_search_depth;
     goto end;
-  }
 
   /*
     We do a breadth-first search first -- that is, inspect all
@@ -2127,7 +2114,6 @@ bool MDL_lock::visit_subgraph(MDL_ticket *waiting_ticket,
 
 end_leave_node:
   gvisitor->leave_node(src_ctx);
-  --gvisitor->m_current_search_depth;
 
 end:
   mysql_prlock_unlock(&m_rwlock);
diff --git a/sql/mdl.h b/sql/mdl.h
index e1d4cf74dd6..7938d833eac 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -385,10 +385,7 @@ public:
 
   virtual bool inspect_edge(MDL_context *dest) = 0;
   virtual ~MDL_wait_for_graph_visitor();
-  MDL_wait_for_graph_visitor() :m_lock_open_count(0),
-                                m_current_search_depth(0)
-  { }
-  virtual void abort_traversal(MDL_context *node) = 0;
+  MDL_wait_for_graph_visitor() :m_lock_open_count(0) {}
 public:
   /**
    XXX, hack: During deadlock search, we may need to
@@ -399,17 +396,6 @@ public:
    LOCK_open since it has significant performance impacts.
   */
   uint m_lock_open_count;
-  /**
-    Set to the 0 at start. Increased whenever
-    we descend into another MDL context (aka traverse to the next
-    wait-for graph node). When MAX_SEARCH_DEPTH is reached, we
-    assume that a deadlock is found, even if we have not found a
-    loop.
-
-    XXX: This member belongs to this class only temporarily until
-         bug #56405 is fixed.
-  */
-  uint m_current_search_depth;
 };
 
 /**
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index baeeed20ffc..1ab335eb106 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -98,13 +98,6 @@
 
 #define mysqld_charset &my_charset_latin1
 
-/* stack traces are only supported on linux intel */
-#if defined(__linux__)  && defined(__i386__) && defined(USE_PSTACK)
-#define	HAVE_STACK_TRACE_ON_SEGV
-#include "../pstack/pstack.h"
-char pstack_file_name[80];
-#endif /* __linux__ */
-
 /* We have HAVE_purify below as this speeds up the shutdown of MySQL */
 
 #if defined(HAVE_DEC_3_2_THREADS) || defined(SIGNALS_DONT_BREAK_READ) || defined(HAVE_purify) && defined(__linux__)
@@ -651,9 +644,6 @@ char *opt_logname, *opt_slow_logname;
 /* Static variables */
 
 static bool kill_in_progress, segfaulted;
-#ifdef HAVE_STACK_TRACE_ON_SEGV
-static my_bool opt_do_pstack;
-#endif /* HAVE_STACK_TRACE_ON_SEGV */
 static my_bool opt_bootstrap, opt_myisam_log;
 static int cleanup_done;
 static ulong opt_specialflag;
@@ -1396,6 +1386,12 @@ static void mysqld_exit(int exit_code)
   mysql_audit_finalize();
   clean_up_mutexes();
   clean_up_error_log_mutex();
+#ifdef WITH_PERFSCHEMA_STORAGE_ENGINE
+  /*
+    Bug#56666 needs to be fixed before calling:
+    shutdown_performance_schema();
+  */
+#endif
   my_end(opt_endinfo ? MY_CHECK_ERROR | MY_GIVE_INFO : 0);
   exit(exit_code); /* purecov: inspected */
 }
@@ -1488,6 +1484,7 @@ void clean_up(bool print_message)
   MYSQL_CALLBACK(thread_scheduler, end, ());
   mysql_client_plugin_deinit();
   finish_client_errs();
+  (void) my_error_unregister(ER_ERROR_FIRST, ER_ERROR_LAST); // finish server errs
   DBUG_PRINT("quit", ("Error messages freed"));
   /* Tell main we are ready */
   logger.cleanup_end();
@@ -2681,14 +2678,6 @@ pthread_handler_t signal_hand(void *arg __attribute__((unused)))
   if (!opt_bootstrap)
     create_pid_file();
 
-#ifdef HAVE_STACK_TRACE_ON_SEGV
-  if (opt_do_pstack)
-  {
-    sprintf(pstack_file_name,"mysqld-%lu-%%d-%%d.backtrace", (ulong)getpid());
-    pstack_install_segv_action(pstack_file_name);
-  }
-#endif /* HAVE_STACK_TRACE_ON_SEGV */
-
   /*
     signal to start_signal_handler that we are ready
     This works by waiting for start_signal_handler to free mutex,
@@ -2734,6 +2723,11 @@ pthread_handler_t signal_hand(void *arg __attribute__((unused)))
       if (!abort_loop)
       {
 	abort_loop=1;				// mark abort for threads
+#ifdef HAVE_PSI_INTERFACE
+        /* Delete the instrumentation for the signal thread */
+        if (likely(PSI_server != NULL))
+          PSI_server->delete_current_thread();
+#endif
 #ifdef USE_ONE_SIGNAL_HAND
 	pthread_t tmp;
         if (mysql_thread_create(0, /* Not instrumented */
@@ -4590,6 +4584,15 @@ int mysqld_main(int argc, char **argv)
 #endif
 #endif /* __WIN__ */
 
+#ifdef HAVE_PSI_INTERFACE
+  /*
+    Disable the main thread instrumentation,
+    to avoid recording events during the shutdown.
+  */
+  if (PSI_server)
+    PSI_server->delete_current_thread();
+#endif
+
   /* Wait until cleanup is done */
   mysql_mutex_lock(&LOCK_thread_count);
   while (!ready_to_exit)
@@ -4607,18 +4610,6 @@ int mysqld_main(int argc, char **argv)
   }
 #endif
   clean_up(1);
-#ifdef HAVE_PSI_INTERFACE
-  /*
-    Disable the instrumentation, to avoid recording events
-    during the shutdown.
-  */
-  if (PSI_server)
-  {
-    PSI_server->delete_current_thread();
-    PSI_server= NULL;
-  }
-  shutdown_performance_schema();
-#endif
   mysqld_exit(0);
 }
 
@@ -5734,11 +5725,6 @@ struct my_option my_long_options[]=
    &disconnect_slave_event_count, &disconnect_slave_event_count,
    0, GET_INT, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
 #endif /* HAVE_REPLICATION */
-#ifdef HAVE_STACK_TRACE_ON_SEGV
-  {"enable-pstack", 0, "Print a symbolic stack trace on failure.",
-   &opt_do_pstack, &opt_do_pstack, 0, GET_BOOL, NO_ARG, 0, 0,
-   0, 0, 0, 0},
-#endif /* HAVE_STACK_TRACE_ON_SEGV */
   {"exit-info", 'T', "Used for debugging. Use at your own risk.", 0, 0, 0,
    GET_LONG, OPT_ARG, 0, 0, 0, 0, 0, 0},
 
diff --git a/sql/net_serv.cc b/sql/net_serv.cc
index 6f0b7a817ca..488e27296b6 100644
--- a/sql/net_serv.cc
+++ b/sql/net_serv.cc
@@ -164,7 +164,17 @@ my_bool net_realloc(NET *net, size_t length)
   DBUG_ENTER("net_realloc");
   DBUG_PRINT("enter",("length: %lu", (ulong) length));
 
-  if (length >= net->max_packet_size)
+  /*
+    When compression is off, net->where_b is always 0.
+    With compression turned on, net->where_b may indicate
+    that we still have a piece of the previous logical
+    packet in the buffer, unprocessed. Take it into account
+    when checking that max_allowed_packet is not exceeded.
+    This ensures that the client treats max_allowed_packet
+    limit identically, regardless of compression being on
+    or off.
+  */
+  if (length >= (net->max_packet_size + net->where_b))
   {
     DBUG_PRINT("error", ("Packet too large. Max size: %lu",
                          net->max_packet_size));
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index b9819def54d..ffff9e3f6c8 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -8550,9 +8550,14 @@ int QUICK_RANGE_SELECT::reset()
   in_range= FALSE;
   cur_range= (QUICK_RANGE**) ranges.buffer;
 
-  if (file->inited == handler::NONE && (error= file->ha_index_init(index,1)))
-    DBUG_RETURN(error);
- 
+  if (file->inited == handler::NONE)
+  {
+    if (in_ror_merged_scan)
+      head->column_bitmaps_set_no_signal(&column_bitmap, &column_bitmap);
+    if ((error= file->ha_index_init(index,1)))
+        DBUG_RETURN(error);
+  }
+
   /* Do not allocate the buffers twice. */
   if (multi_range_length)
   {
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index caf28fdd83e..b54339db354 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -742,7 +742,6 @@ bool partition_info::check_range_constants(THD *thd)
     longlong part_range_value;
     bool signed_flag= !part_expr->unsigned_flag;
 
-    part_result_type= INT_RESULT;
     range_int_array= (longlong*)sql_alloc(num_parts * sizeof(longlong));
     if (unlikely(range_int_array == NULL))
     {
@@ -917,7 +916,6 @@ bool partition_info::check_list_constants(THD *thd)
   List_iterator<partition_element> list_func_it(partitions);
   DBUG_ENTER("partition_info::check_list_constants");
 
-  part_result_type= INT_RESULT;
   num_list_values= 0;
   /*
     We begin by calculating the number of list values that have been
@@ -1608,6 +1606,52 @@ id_err:
   return 1;
 }
 
+
+/**
+  Check what kind of error to report
+
+  @param use_subpart_expr Use the subpart_expr instead of part_expr
+  @param part_str         Name of partition to report error (or NULL)
+*/
+void partition_info::report_part_expr_error(bool use_subpart_expr)
+{
+  Item *expr= part_expr;
+  DBUG_ENTER("partition_info::report_part_expr_error");
+  if (use_subpart_expr)
+    expr= subpart_expr;
+
+  if (expr->type() == Item::FIELD_ITEM)
+  {
+    partition_type type= part_type;
+    bool list_of_fields= list_of_part_fields;
+    Item_field *item_field= (Item_field*) expr;
+    /*
+      The expression consists of a single field.
+      It must be of integer type unless KEY or COLUMNS partitioning.
+    */
+    if (use_subpart_expr)
+    {
+      type= subpart_type;
+      list_of_fields= list_of_subpart_fields;
+    }
+    if (!column_list &&
+        item_field->field &&
+        item_field->field->result_type() != INT_RESULT &&
+        !(type == HASH_PARTITION && list_of_fields))
+    {
+      my_error(ER_FIELD_TYPE_NOT_ALLOWED_AS_PARTITION_FIELD, MYF(0),
+               item_field->name);
+      DBUG_VOID_RETURN;
+    }
+  }
+  if (use_subpart_expr)
+    my_error(ER_PARTITION_FUNC_NOT_ALLOWED_ERROR, MYF(0), "SUBPARTITION");
+  else
+    my_error(ER_PARTITION_FUNC_NOT_ALLOWED_ERROR, MYF(0), "PARTITION");
+  DBUG_VOID_RETURN;
+}
+ 
+
 /*
   Create a new column value in current list with maxvalue
   Called from parser
@@ -1891,7 +1935,7 @@ int partition_info::reorganize_into_single_field_col_val()
   code.
 
   SYNOPSIS
-  fix_func_partition()
+  fix_partition_values()
   thd                             Thread object
   col_val                         Array of one value
   part_elem                       The partition instance
@@ -1901,13 +1945,13 @@ int partition_info::reorganize_into_single_field_col_val()
     TRUE                     Failure
     FALSE                    Success
 */
-int partition_info::fix_func_partition(THD *thd,
-                                       part_elem_value *val,
-                                       partition_element *part_elem,
-                                       uint part_id)
+int partition_info::fix_partition_values(THD *thd,
+                                         part_elem_value *val,
+                                         partition_element *part_elem,
+                                         uint part_id)
 {
   part_column_list_val *col_val= val->col_val_array;
-  DBUG_ENTER("partition_info::fix_func_partition");
+  DBUG_ENTER("partition_info::fix_partition_values");
 
   if (col_val->fixed)
   {
@@ -1953,7 +1997,8 @@ int partition_info::fix_func_partition(THD *thd,
     }
     else if (item_expr->result_type() != INT_RESULT)
     {
-      my_error(ER_INCONSISTENT_TYPE_OF_FUNCTIONS_ERROR, MYF(0));
+      my_error(ER_VALUES_IS_NOT_INT_TYPE_ERROR, MYF(0),
+               part_elem->partition_name);
       DBUG_RETURN(TRUE);
     }
     if (part_type == RANGE_PARTITION)
@@ -2168,7 +2213,7 @@ int partition_info::fix_parser_data(THD *thd)
       }
       else
       {
-        if (fix_func_partition(thd, val, part_elem, i))
+        if (fix_partition_values(thd, val, part_elem, i))
         {
           DBUG_RETURN(TRUE);
         }
diff --git a/sql/partition_info.h b/sql/partition_info.h
index b196d0b59a2..6ae210d9574 100644
--- a/sql/partition_info.h
+++ b/sql/partition_info.h
@@ -166,7 +166,6 @@ public:
   key_map some_fields_in_PF;
 
   handlerton *default_engine_type;
-  Item_result part_result_type;
   partition_type part_type;
   partition_type subpart_type;
 
@@ -226,7 +225,6 @@ public:
     curr_part_elem(NULL), current_partition(NULL),
     curr_list_object(0), num_columns(0),
     default_engine_type(NULL),
-    part_result_type(INT_RESULT),
     part_type(NOT_A_PARTITION), subpart_type(NOT_A_PARTITION),
     part_info_len(0),
     part_func_len(0), subpart_func_len(0),
@@ -279,10 +277,10 @@ public:
   void print_no_partition_found(TABLE *table);
   void print_debug(const char *str, uint*);
   Item* get_column_item(Item *item, Field *field);
-  int fix_func_partition(THD *thd,
-                         part_elem_value *val,
-                         partition_element *part_elem,
-                         uint part_id);
+  int fix_partition_values(THD *thd,
+                           part_elem_value *val,
+                           partition_element *part_elem,
+                           uint part_id);
   bool fix_column_value_functions(THD *thd,
                                   part_elem_value *val,
                                   uint part_id);
@@ -299,6 +297,7 @@ public:
   bool init_column_part();
   bool add_column_list_value(THD *thd, Item *item);
   void set_show_version_string(String *packet);
+  void report_part_expr_error(bool use_subpart_expr);
 private:
   static int list_part_cmp(const void* a, const void* b);
   bool set_up_default_partitions(handler *file, HA_CREATE_INFO *info,
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 19fff76c1da..4313dcce917 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -108,7 +108,7 @@ void sys_var_end()
   my_hash_free(&system_variable_hash);
 
   for (sys_var *var=all_sys_vars.first; var; var= var->next)
-    var->~sys_var();
+    var->cleanup();
 
   DBUG_VOID_RETURN;
 }
diff --git a/sql/set_var.h b/sql/set_var.h
index 5f1f889c4ce..2972b430e48 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -91,11 +91,13 @@ public:
           longlong def_val, PolyLock *lock, enum binlog_status_enum binlog_status_arg,
           on_check_function on_check_func, on_update_function on_update_func,
           uint deprecated_version, const char *substitute, int parse_flag);
+
+  virtual ~sys_var() {}
+
   /**
-    The instance should only be destroyed on shutdown, as it doesn't unlink
-    itself from the chain.
+    All the cleanup procedures should be performed here
   */
-  virtual ~sys_var() {}
+  virtual void cleanup() {}
   /**
     downcast for sys_var_pluginvar. Returns this if it's an instance
     of sys_var_pluginvar, and 0 otherwise.
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index f3a4ffbece9..07f5589135d 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6347,6 +6347,10 @@ ER_FAILED_READ_FROM_PAR_FILE
   eng "Failed to read from the .par file"
   swe "Misslyckades läsa från .par filen"
 
+ER_VALUES_IS_NOT_INT_TYPE_ERROR
+  eng "VALUES value for partition '%-.64s' must have type INT"
+  swe "Värden i VALUES för partition '%-.64s' måste ha typen INT"
+
 ER_ACCESS_DENIED_NO_PASSWORD_ERROR 28000 
         cze "P-Břístup pro uživatele '%-.48s'@'%-.64s'"
         dan "Adgang nægtet bruger: '%-.48s'@'%-.64s'"
diff --git a/sql/spatial.cc b/sql/spatial.cc
index f1dd0e6cebd..b02feb722f7 100644
--- a/sql/spatial.cc
+++ b/sql/spatial.cc
@@ -511,7 +511,7 @@ uint Gis_line_string::init_from_wkb(const char *wkb, uint len,
   n_points= wkb_get_uint(wkb, bo);
   proper_length= 4 + n_points * POINT_DATA_SIZE;
 
-  if (len < proper_length || res->reserve(proper_length))
+  if (!n_points || len < proper_length || res->reserve(proper_length))
     return 0;
 
   res->q_append(n_points);
@@ -729,7 +729,9 @@ uint Gis_polygon::init_from_wkb(const char *wkb, uint len, wkbByteOrder bo,
   if (len < 4)
     return 0;
 
-  n_linear_rings= wkb_get_uint(wkb, bo);
+  if (!(n_linear_rings= wkb_get_uint(wkb, bo)))
+    return 0;
+
   if (res->reserve(4, 512))
     return 0;
   wkb+= 4;
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 4f00d89c4c0..571a7890116 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -4490,7 +4490,8 @@ end:
 bool check_grant(THD *thd, ulong want_access, TABLE_LIST *tables,
                  bool any_combination_will_do, uint number, bool no_errors)
 {
-  TABLE_LIST *table, *first_not_own_table= thd->lex->first_not_own_table();
+  TABLE_LIST *tl;
+  TABLE_LIST *first_not_own_table= thd->lex->first_not_own_table();
   Security_context *sctx= thd->security_ctx;
   uint i;
   ulong orig_want_access= want_access;
@@ -4507,34 +4508,32 @@ bool check_grant(THD *thd, ulong want_access, TABLE_LIST *tables,
     the given table list refers to the list for prelocking (contains tables
     of other queries). For simple queries first_not_own_table is 0.
   */
-  for (i= 0, table= tables;
-       i < number  && table != first_not_own_table;
-       table= table->next_global, i++)
+  for (i= 0, tl= tables;
+       i < number  && tl != first_not_own_table;
+       tl= tl->next_global, i++)
   {
     /*
       Save a copy of the privileges without the SHOW_VIEW_ACL attribute.
       It will be checked during making view.
     */
-    table->grant.orig_want_privilege= (want_access & ~SHOW_VIEW_ACL);
+    tl->grant.orig_want_privilege= (want_access & ~SHOW_VIEW_ACL);
   }
 
   mysql_rwlock_rdlock(&LOCK_grant);
-  for (table= tables;
-       table && number-- && table != first_not_own_table;
-       table= table->next_global)
+  for (tl= tables;
+       tl && number-- && tl != first_not_own_table;
+       tl= tl->next_global)
   {
-    GRANT_TABLE *grant_table;
-    sctx = test(table->security_ctx) ?
-      table->security_ctx : thd->security_ctx;
+    sctx = test(tl->security_ctx) ? tl->security_ctx : thd->security_ctx;
 
-    const ACL_internal_table_access *access;
-    access= get_cached_table_access(&table->grant.m_internal,
-                                    table->get_db_name(),
-                                    table->get_table_name());
+    const ACL_internal_table_access *access=
+      get_cached_table_access(&tl->grant.m_internal,
+                              tl->get_db_name(),
+                              tl->get_table_name());
 
     if (access)
     {
-      switch(access->check(orig_want_access, &table->grant.privilege))
+      switch(access->check(orig_want_access, &tl->grant.privilege))
       {
       case ACL_INTERNAL_ACCESS_GRANTED:
         /*
@@ -4558,29 +4557,33 @@ bool check_grant(THD *thd, ulong want_access, TABLE_LIST *tables,
     if (!want_access)
       continue;                                 // ok
 
-    if (!(~table->grant.privilege & want_access) ||
-        table->is_anonymous_derived_table() || table->schema_table)
+    if (!(~tl->grant.privilege & want_access) ||
+        tl->is_anonymous_derived_table() || tl->schema_table)
     {
       /*
-        It is subquery in the FROM clause. VIEW set table->derived after
+        It is subquery in the FROM clause. VIEW set tl->derived after
         table opening, but this function always called before table opening.
       */
-      if (!table->referencing_view)
+      if (!tl->referencing_view)
       {
         /*
           If it's a temporary table created for a subquery in the FROM
           clause, or an INFORMATION_SCHEMA table, drop the request for
           a privilege.
         */
-        table->grant.want_privilege= 0;
+        tl->grant.want_privilege= 0;
       }
       continue;
     }
-    if (!(grant_table= table_hash_search(sctx->host, sctx->ip,
-                                         table->get_db_name(), sctx->priv_user,
-                                         table->get_table_name(), FALSE)))
+    GRANT_TABLE *grant_table= table_hash_search(sctx->host, sctx->ip,
+                                                tl->get_db_name(),
+                                                sctx->priv_user,
+                                                tl->get_table_name(),
+                                                FALSE);
+
+    if (!grant_table)
     {
-      want_access &= ~table->grant.privilege;
+      want_access &= ~tl->grant.privilege;
       goto err;					// No grants
     }
 
@@ -4591,18 +4594,17 @@ bool check_grant(THD *thd, ulong want_access, TABLE_LIST *tables,
     if (any_combination_will_do)
       continue;
 
-    table->grant.grant_table=grant_table;	// Remember for column test
-    table->grant.version=grant_version;
-    table->grant.privilege|= grant_table->privs;
-    table->grant.want_privilege= ((want_access & COL_ACLS)
-				  & ~table->grant.privilege);
+    tl->grant.grant_table= grant_table; // Remember for column test
+    tl->grant.version= grant_version;
+    tl->grant.privilege|= grant_table->privs;
+    tl->grant.want_privilege= ((want_access & COL_ACLS) & ~tl->grant.privilege);
 
-    if (!(~table->grant.privilege & want_access))
+    if (!(~tl->grant.privilege & want_access))
       continue;
 
-    if (want_access & ~(grant_table->cols | table->grant.privilege))
+    if (want_access & ~(grant_table->cols | tl->grant.privilege))
     {
-      want_access &= ~(grant_table->cols | table->grant.privilege);
+      want_access &= ~(grant_table->cols | tl->grant.privilege);
       goto err;					// impossible
     }
   }
@@ -4619,7 +4621,7 @@ err:
              command,
              sctx->priv_user,
              sctx->host_or_ip,
-             table ? table->get_table_name() : "unknown");
+             tl ? tl->get_table_name() : "unknown");
   }
   DBUG_RETURN(TRUE);
 }
diff --git a/sql/sql_admin.cc b/sql/sql_admin.cc
index 1f96f1cf0e4..21a05a5baca 100644
--- a/sql/sql_admin.cc
+++ b/sql/sql_admin.cc
@@ -111,9 +111,6 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
     table= &tmp_table;
   }
 
-  /* A MERGE table must not come here. */
-  DBUG_ASSERT(table->file->ht->db_type != DB_TYPE_MRG_MYISAM);
-
   /*
     REPAIR TABLE ... USE_FRM for temporary tables makes little sense.
   */
@@ -151,6 +148,9 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
   if (!ext[0] || !ext[1])
     goto end;					// No data file
 
+  /* A MERGE table must not come here. */
+  DBUG_ASSERT(table->file->ht->db_type != DB_TYPE_MRG_MYISAM);
+
   // Name of data file
   strxmov(from, table->s->normalized_path.str, ext[1], NullS);
   if (!mysql_file_stat(key_file_misc, from, &stat_info, MYF(0)))
@@ -231,6 +231,26 @@ end:
 }
 
 
+/**
+  Check if a given error is something that could occur during
+  open_and_lock_tables() that does not indicate table corruption.
+
+  @param  sql_errno  Error number to check.
+
+  @retval TRUE       Error does not indicate table corruption.
+  @retval FALSE      Error could indicate table corruption.
+*/
+
+static inline bool table_not_corrupt_error(uint sql_errno)
+{
+  return (sql_errno == ER_NO_SUCH_TABLE ||
+          sql_errno == ER_FILE_NOT_FOUND ||
+          sql_errno == ER_LOCK_WAIT_TIMEOUT ||
+          sql_errno == ER_LOCK_DEADLOCK ||
+          sql_errno == ER_CANT_LOCK_LOG_TABLE ||
+          sql_errno == ER_OPEN_AS_READONLY);
+}
+
 
 /*
   RETURN VALUES
@@ -311,7 +331,13 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       lex->query_tables= table;
       lex->query_tables_last= &table->next_global;
       lex->query_tables_own_last= 0;
-      thd->no_warnings_for_error= no_warnings_for_error;
+      /*
+        Under locked tables, we know that the table can be opened,
+        so any errors opening the table are logical errors.
+        In these cases it makes sense to report them.
+      */
+      if (!thd->locked_tables_mode)
+        thd->no_warnings_for_error= no_warnings_for_error;
       if (view_operator_func == NULL)
         table->required_type=FRMTYPE_TABLE;
 
@@ -320,6 +346,14 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       table->next_global= save_next_global;
       table->next_local= save_next_local;
       thd->open_options&= ~extra_open_options;
+
+      /*
+        If open_and_lock_tables() failed, close_thread_tables() will close
+        the table and table->table can therefore be invalid.
+      */
+      if (open_error)
+        table->table= NULL;
+
       /*
         Under locked tables, we know that the table can be opened,
         so any errors opening the table are logical errors.
@@ -418,9 +452,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
         push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                      ER_VIEW_CHECKSUM, ER(ER_VIEW_CHECKSUM));
       if (thd->stmt_da->is_error() &&
-          (thd->stmt_da->sql_errno() == ER_NO_SUCH_TABLE ||
-           thd->stmt_da->sql_errno() == ER_FILE_NOT_FOUND))
-        /* A missing table is just issued as a failed command */
+          table_not_corrupt_error(thd->stmt_da->sql_errno()))
         result_code= HA_ADMIN_FAILED;
       else
         /* Default failure code is corrupt table */
diff --git a/sql/sql_alter.cc b/sql/sql_alter.cc
index 046e09b20a3..5af01523aa7 100644
--- a/sql/sql_alter.cc
+++ b/sql/sql_alter.cc
@@ -13,8 +13,7 @@
    along with this program; if not, write to the Free Software
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
 
-#include "sql_parse.h"                       // check_access,
-                                             // check_merge_table_access
+#include "sql_parse.h"                       // check_access
 #include "sql_table.h"                       // mysql_alter_table,
                                              // mysql_exchange_partition
 #include "sql_alter.h"
@@ -60,11 +59,15 @@ bool Alter_table_statement::execute(THD *thd)
       check_access(thd, INSERT_ACL | CREATE_ACL, select_lex->db,
                    &priv,
                    NULL, /* Don't use first_tab->grant with sel_lex->db */
-                   0, 0) ||
-      check_merge_table_access(thd, first_table->db,
-                               create_info.merge_list.first))
+                   0, 0))
     DBUG_RETURN(TRUE);                  /* purecov: inspected */
 
+  /* If it is a merge table, check privileges for merge children. */
+  if (create_info.merge_list.first &&
+      check_table_access(thd, SELECT_ACL | UPDATE_ACL | DELETE_ACL,
+                         create_info.merge_list.first, FALSE, UINT_MAX, FALSE))
+    DBUG_RETURN(TRUE);
+
   if (check_grant(thd, priv_needed, first_table, FALSE, UINT_MAX, FALSE))
     DBUG_RETURN(TRUE);                  /* purecov: inspected */
 
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index b168f561954..8305283cd17 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -100,8 +100,6 @@ bool No_such_table_error_handler::safely_trapped_errors()
   TABLE_SHAREs, refresh_version and the table id counter.
 */
 mysql_mutex_t LOCK_open;
-mysql_mutex_t LOCK_dd_owns_lock_open;
-uint dd_owns_lock_open= 0;
 
 #ifdef HAVE_PSI_INTERFACE
 static PSI_mutex_key key_LOCK_open;
@@ -249,7 +247,8 @@ static void check_unused(void)
     Length of key
 */
 
-uint create_table_def_key(THD *thd, char *key, TABLE_LIST *table_list,
+uint create_table_def_key(THD *thd, char *key,
+                          const TABLE_LIST *table_list,
                           bool tmp_table)
 {
   uint key_length= (uint) (strmov(strmov(key, table_list->db)+1,
@@ -300,7 +299,6 @@ bool table_def_init(void)
   init_tdc_psi_keys();
 #endif
   mysql_mutex_init(key_LOCK_open, &LOCK_open, MY_MUTEX_INIT_FAST);
-  mysql_mutex_init(NULL, &LOCK_dd_owns_lock_open, MY_MUTEX_INIT_FAST);
   oldest_unused_share= &end_of_unused_share;
   end_of_unused_share.prev= &oldest_unused_share;
 
@@ -344,7 +342,6 @@ void table_def_free(void)
     table_def_inited= 0;
     /* Free table definitions. */
     my_hash_free(&table_def_cache);
-    mysql_mutex_destroy(&LOCK_dd_owns_lock_open);
     mysql_mutex_destroy(&LOCK_open);
   }
   DBUG_VOID_RETURN;
@@ -1991,39 +1988,60 @@ void update_non_unique_table_error(TABLE_LIST *update,
 }
 
 
+/**
+  Find temporary table specified by database and table names in the
+  THD::temporary_tables list.
+
+  @return TABLE instance if a temporary table has been found; NULL otherwise.
+*/
+
 TABLE *find_temporary_table(THD *thd, const char *db, const char *table_name)
 {
-  TABLE_LIST table_list;
+  TABLE_LIST tl;
 
-  table_list.db= (char*) db;
-  table_list.table_name= (char*) table_name;
-  return find_temporary_table(thd, &table_list);
+  tl.db= (char*) db;
+  tl.table_name= (char*) table_name;
+
+  return find_temporary_table(thd, &tl);
 }
 
 
-TABLE *find_temporary_table(THD *thd, TABLE_LIST *table_list)
+/**
+  Find a temporary table specified by TABLE_LIST instance in the
+  THD::temporary_tables list.
+
+  @return TABLE instance if a temporary table has been found; NULL otherwise.
+*/
+
+TABLE *find_temporary_table(THD *thd, const TABLE_LIST *tl)
 {
-  char	key[MAX_DBKEY_LENGTH];
-  uint	key_length;
-  TABLE *table;
-  DBUG_ENTER("find_temporary_table");
-  DBUG_PRINT("enter", ("table: '%s'.'%s'",
-                       table_list->db, table_list->table_name));
+  char key[MAX_DBKEY_LENGTH];
+  uint key_length= create_table_def_key(thd, key, tl, 1);
+
+  return find_temporary_table(thd, key, key_length);
+}
 
-  key_length= create_table_def_key(thd, key, table_list, 1);
-  for (table=thd->temporary_tables ; table ; table= table->next)
+
+/**
+  Find a temporary table specified by a key in the THD::temporary_tables list.
+
+  @return TABLE instance if a temporary table has been found; NULL otherwise.
+*/
+
+TABLE *find_temporary_table(THD *thd,
+                            const char *table_key,
+                            uint table_key_length)
+{
+  for (TABLE *table= thd->temporary_tables; table; table= table->next)
   {
-    if (table->s->table_cache_key.length == key_length &&
-	!memcmp(table->s->table_cache_key.str, key, key_length))
+    if (table->s->table_cache_key.length == table_key_length &&
+        !memcmp(table->s->table_cache_key.str, table_key, table_key_length))
     {
-      DBUG_PRINT("info",
-                 ("Found table. server_id: %u  pseudo_thread_id: %lu",
-                  (uint) thd->server_id,
-                  (ulong) thd->variables.pseudo_thread_id));
-      DBUG_RETURN(table);
+      return table;
     }
   }
-  DBUG_RETURN(0);                               // Not a temporary table
+
+  return NULL;
 }
 
 
@@ -3027,41 +3045,11 @@ retry_share:
   table->reginfo.lock_type=TL_READ;		/* Assume read */
 
  reset:
-  DBUG_ASSERT(table->s->ref_count > 0 || table->s->tmp_table != NO_TMP_TABLE);
-
-  if (thd->lex->need_correct_ident())
-    table->alias_name_used= my_strcasecmp(table_alias_charset,
-                                          table->s->table_name.str, alias);
-  /* Fix alias if table name changes */
-  if (strcmp(table->alias, alias))
-  {
-    uint length=(uint) strlen(alias)+1;
-    table->alias= (char*) my_realloc((char*) table->alias, length,
-                                     MYF(MY_WME));
-    memcpy((char*) table->alias, alias, length);
-  }
-  table->tablenr=thd->current_tablenr++;
-  table->used_fields=0;
-  table->const_table=0;
-  table->null_row= table->maybe_null= 0;
-  table->force_index= table->force_index_order= table->force_index_group= 0;
-  table->status=STATUS_NO_RECORD;
-  table->insert_values= 0;
-  table->fulltext_searched= 0;
-  table->file->ft_handler= 0;
-  table->reginfo.impossible_range= 0;
-  /* Catch wrong handling of the auto_increment_field_not_null. */
-  DBUG_ASSERT(!table->auto_increment_field_not_null);
-  table->auto_increment_field_not_null= FALSE;
-  if (table->timestamp_field)
-    table->timestamp_field_type= table->timestamp_field->get_auto_set_type();
-  table->pos_in_table_list= table_list;
   table_list->updatable= 1; // It is not derived table nor non-updatable VIEW
-  table->clear_column_bitmaps();
   table_list->table= table;
-  DBUG_ASSERT(table->key_read == 0);
-  /* Tables may be reused in a sub statement. */
-  DBUG_ASSERT(! table->file->extra(HA_EXTRA_IS_ATTACHED_CHILDREN));
+
+  table->init(thd, table_list);
+
   DBUG_RETURN(FALSE);
 
 err_lock:
@@ -4534,13 +4522,15 @@ lock_table_names(THD *thd,
            ! (flags & MYSQL_OPEN_SKIP_TEMPORARY) &&
            find_temporary_table(thd, table))))
     {
-      if (schema_set.insert(table))
+      if (! (flags & MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK) &&
+          schema_set.insert(table))
         return TRUE;
       mdl_requests.push_front(&table->mdl_request);
     }
   }
 
-  if (! mdl_requests.is_empty())
+  if (! (flags & MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK) &&
+      ! mdl_requests.is_empty())
   {
     /*
       Scoped locks: Take intention exclusive locks on all involved
@@ -5175,6 +5165,8 @@ static bool check_lock_and_start_stmt(THD *thd,
   @param[in]    lock_type       lock to use for table
   @param[in]    flags           options to be used while opening and locking
                                 table (see open_table(), mysql_lock_tables())
+  @param[in]    prelocking_strategy  Strategy which specifies how prelocking
+                                     algorithm should work for this statement.
 
   @return       table
     @retval     != NULL         OK, opened table returned
@@ -5200,7 +5192,8 @@ static bool check_lock_and_start_stmt(THD *thd,
 */
 
 TABLE *open_n_lock_single_table(THD *thd, TABLE_LIST *table_l,
-                                thr_lock_type lock_type, uint flags)
+                                thr_lock_type lock_type, uint flags,
+                                Prelocking_strategy *prelocking_strategy)
 {
   TABLE_LIST *save_next_global;
   DBUG_ENTER("open_n_lock_single_table");
@@ -5216,7 +5209,8 @@ TABLE *open_n_lock_single_table(THD *thd, TABLE_LIST *table_l,
   table_l->required_type= FRMTYPE_TABLE;
 
   /* Open the table. */
-  if (open_and_lock_tables(thd, table_l, FALSE, flags))
+  if (open_and_lock_tables(thd, table_l, FALSE, flags,
+                           prelocking_strategy))
     table_l->table= NULL; /* Just to be sure. */
 
   /* Restore list. */
@@ -5708,35 +5702,37 @@ void close_tables_for_reopen(THD *thd, TABLE_LIST **tables,
 }
 
 
-/*
-  Open a single table without table caching and don't set it in open_list
-
-  SYNPOSIS
-    open_temporary_table()
-    thd		  Thread object
-    path	  Path (without .frm)
-    db		  database
-    table_name	  Table name
-    link_in_list  1 if table should be linked into thd->temporary_tables
-
- NOTES:
-    Used by alter_table to open a temporary table and when creating
-    a temporary table with CREATE TEMPORARY ...
-
- RETURN
-   0  Error
-   #  TABLE object
+/**
+  Open a single table without table caching and don't add it to
+  THD::open_tables. Depending on the 'add_to_temporary_tables_list' value,
+  the opened TABLE instance will be addded to THD::temporary_tables list.
+
+  @param thd                          Thread context.
+  @param path                         Path (without .frm)
+  @param db                           Database name.
+  @param table_name                   Table name.
+  @param add_to_temporary_tables_list Specifies if the opened TABLE
+                                      instance should be linked into
+                                      THD::temporary_tables list.
+
+  @note This function is used:
+    - by alter_table() to open a temporary table;
+    - when creating a temporary table with CREATE TEMPORARY TABLE.
+
+  @return TABLE instance for opened table.
+  @retval NULL on error.
 */
 
-TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
-			    const char *table_name, bool link_in_list)
+TABLE *open_table_uncached(THD *thd, const char *path, const char *db,
+                           const char *table_name,
+                           bool add_to_temporary_tables_list)
 {
   TABLE *tmp_table;
   TABLE_SHARE *share;
   char cache_key[MAX_DBKEY_LENGTH], *saved_cache_key, *tmp_path;
   uint key_length;
   TABLE_LIST table_list;
-  DBUG_ENTER("open_temporary_table");
+  DBUG_ENTER("open_table_uncached");
   DBUG_PRINT("enter",
              ("table: '%s'.'%s'  path: '%s'  server_id: %u  "
               "pseudo_thread_id: %lu",
@@ -5779,7 +5775,7 @@ TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
   share->tmp_table= (tmp_table->file->has_transactions() ? 
                      TRANSACTIONAL_TMP_TABLE : NON_TRANSACTIONAL_TMP_TABLE);
 
-  if (link_in_list)
+  if (add_to_temporary_tables_list)
   {
     /* growing temp list at the head */
     tmp_table->next= thd->temporary_tables;
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 2e4554313e5..7ae3971942b 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -17,7 +17,6 @@
 #define SQL_BASE_INCLUDED
 
 #include "unireg.h"                    // REQUIRED: for other includes
-#include "table.h"                              /* open_table_mode */
 #include "sql_trigger.h"                        /* trg_event_type */
 #include "sql_class.h"                          /* enum_mark_columns */
 #include "mysqld.h"                             /* key_map */
@@ -71,8 +70,6 @@ enum enum_tdc_remove_table_type {TDC_RT_REMOVE_ALL, TDC_RT_REMOVE_NOT_OWN,
 
 bool check_dup(const char *db, const char *name, TABLE_LIST *tables);
 extern mysql_mutex_t LOCK_open;
-extern mysql_mutex_t LOCK_dd_owns_lock_open;
-extern uint dd_owns_lock_open;
 bool table_cache_init(void);
 void table_cache_free(void);
 bool table_def_init(void);
@@ -81,7 +78,8 @@ void table_def_start_shutdown(void);
 void assign_new_table_id(TABLE_SHARE *share);
 uint cached_open_tables(void);
 uint cached_table_definitions(void);
-uint create_table_def_key(THD *thd, char *key, TABLE_LIST *table_list,
+uint create_table_def_key(THD *thd, char *key,
+                          const TABLE_LIST *table_list,
                           bool tmp_table);
 TABLE_SHARE *get_table_share(THD *thd, TABLE_LIST *table_list, char *key,
                              uint key_length, uint db_flags, int *error,
@@ -124,6 +122,11 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type update,
   (LONG_TIMEOUT = 1 year) rather than the user-supplied timeout value.
 */
 #define MYSQL_LOCK_IGNORE_TIMEOUT               0x0800
+/**
+  When acquiring "strong" (SNW, SNRW, X) metadata locks on tables to
+  be open do not acquire global and schema-scope IX locks.
+*/
+#define MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK         0x1000
 
 /** Please refer to the internals manual. */
 #define MYSQL_OPEN_REOPEN  (MYSQL_OPEN_IGNORE_FLUSH |\
@@ -143,8 +146,9 @@ bool open_new_frm(THD *thd, TABLE_SHARE *share, const char *alias,
 
 bool get_key_map_from_key_list(key_map *map, TABLE *table,
                                List<String> *index_list);
-TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
-			    const char *table_name, bool link_in_list);
+TABLE *open_table_uncached(THD *thd, const char *path, const char *db,
+			   const char *table_name,
+                           bool add_to_temporary_tables_list);
 TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name);
 TABLE *find_write_locked_table(TABLE *list, const char *db,
                                const char *table_name);
@@ -161,7 +165,9 @@ TABLE_LIST *find_table_in_list(TABLE_LIST *table,
                                const char *db_name,
                                const char *table_name);
 TABLE *find_temporary_table(THD *thd, const char *db, const char *table_name);
-TABLE *find_temporary_table(THD *thd, TABLE_LIST *table_list);
+TABLE *find_temporary_table(THD *thd, const TABLE_LIST *tl);
+TABLE *find_temporary_table(THD *thd, const char *table_key,
+                            uint table_key_length);
 void close_thread_tables(THD *thd);
 bool fill_record_n_invoke_before_triggers(THD *thd, List<Item> &fields,
                                           List<Item> &values,
@@ -240,7 +246,8 @@ bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
 int open_and_lock_tables_derived(THD *thd, TABLE_LIST *tables, bool derived);
 /* simple open_and_lock_tables without derived handling for single table */
 TABLE *open_n_lock_single_table(THD *thd, TABLE_LIST *table_l,
-                                thr_lock_type lock_type, uint flags);
+                                thr_lock_type lock_type, uint flags,
+                                Prelocking_strategy *prelocking_strategy);
 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 decide_logging_format(THD *thd, TABLE_LIST *tables);
@@ -449,6 +456,16 @@ open_tables(THD *thd, TABLE_LIST **tables, uint *counter, uint flags)
 }
 
 
+inline TABLE *open_n_lock_single_table(THD *thd, TABLE_LIST *table_l,
+                                       thr_lock_type lock_type, uint flags)
+{
+  DML_prelocking_strategy prelocking_strategy;
+
+  return open_n_lock_single_table(thd, table_l, lock_type, flags,
+                                  &prelocking_strategy);
+}
+
+
 /* open_and_lock_tables with derived handling */
 inline bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
                                  bool derived, uint flags)
diff --git a/sql/sql_cache.cc b/sql/sql_cache.cc
index d6dc9edc502..b57c851edf4 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -1346,6 +1346,57 @@ end:
 }
 
 
+#ifndef EMBEDDED_LIBRARY
+/**
+  Send a single memory block from the query cache.
+
+  Respects the client/server protocol limits for the
+  size of the network packet, and splits a large block
+  in pieces to ensure that individual piece doesn't exceed
+  the maximal allowed size of the network packet (16M).
+
+  @param[in] net NET handler
+  @param[in] packet packet to send
+  @param[in] len packet length
+
+  @return Operation status
+    @retval FALSE On success
+    @retval TRUE On error
+*/
+static bool
+send_data_in_chunks(NET *net, const uchar *packet, ulong len)
+{
+  /*
+    On the client we may require more memory than max_allowed_packet
+    to keep, both, the truncated last logical packet, and the
+    compressed next packet.  This never (or in practice never)
+    happens without compression, since without compression it's very
+    unlikely that a) a truncated logical packet would remain on the
+    client when it's time to read the next packet b) a subsequent
+    logical packet that is being read would be so large that
+    size-of-new-packet + size-of-old-packet-tail >
+    max_allowed_packet.  To remedy this issue, we send data in 1MB
+    sized packets, that's below the current client default of 16MB
+    for max_allowed_packet, but large enough to ensure there is no
+    unnecessary overhead from too many syscalls per result set.
+  */
+  static const ulong MAX_CHUNK_LENGTH= 1024*1024;
+
+  while (len > MAX_CHUNK_LENGTH)
+  {
+    if (net_real_write(net, packet, MAX_CHUNK_LENGTH))
+      return TRUE;
+    packet+= MAX_CHUNK_LENGTH;
+    len-= MAX_CHUNK_LENGTH;
+  }
+  if (len && net_real_write(net, packet, len))
+    return TRUE;
+
+  return FALSE;
+}
+#endif
+
+
 /*
   Check if the query is in the cache. If it was cached, send it
   to the user.
@@ -1655,11 +1706,11 @@ def_week_frmt: %lu, in_trans: %d, autocommit: %d",
                                    ALIGN_SIZE(sizeof(Query_cache_result)))));
     
     Query_cache_result *result = result_block->result();
-    if (net_real_write(&thd->net, result->data(),
-		       result_block->used -
-		       result_block->headers_len() -
-		       ALIGN_SIZE(sizeof(Query_cache_result))))
-      break;					// Client aborted
+    if (send_data_in_chunks(&thd->net, result->data(),
+                            result_block->used -
+                            result_block->headers_len() -
+                            ALIGN_SIZE(sizeof(Query_cache_result))))
+      break;                                    // Client aborted
     result_block = result_block->next;
     thd->net.pkt_nr= query->last_pkt_nr; // Keep packet number updated
   } while (result_block != first_result_block);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 7d2a6ad864e..3ac6cf28c90 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -621,6 +621,30 @@ bool open_and_lock_for_insert_delayed(THD *thd, TABLE_LIST *table_list)
 
 
 /**
+  Create a new query string for removing DELAYED keyword for
+  multi INSERT DEALAYED statement.
+
+  @param[in] thd                 Thread handler
+  @param[in] buf                 Query string
+
+  @return
+             0           ok
+             1           error
+*/
+static int
+create_insert_stmt_from_insert_delayed(THD *thd, String *buf)
+{
+  /* Make a copy of thd->query() and then remove the "DELAYED" keyword */
+  if (buf->append(thd->query()) ||
+      buf->replace(thd->lex->keyword_delayed_begin_offset,
+                   thd->lex->keyword_delayed_end_offset -
+                   thd->lex->keyword_delayed_begin_offset, 0))
+    return 1;
+  return 0;
+}
+
+
+/**
   INSERT statement implementation
 
   @note Like implementations of other DDL/DML in MySQL, this function
@@ -999,13 +1023,28 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
 	such case the flag is ignored for constructing binlog event.
 	*/
 	DBUG_ASSERT(thd->killed != THD::KILL_BAD_DATA || error > 0);
-	if (thd->binlog_query(THD::ROW_QUERY_TYPE,
-			      thd->query(), thd->query_length(),
-			      transactional_table, FALSE, FALSE,
-                              errcode))
+        if (was_insert_delayed && table_list->lock_type ==  TL_WRITE)
         {
+          /* Binlog multi INSERT DELAYED as INSERT without DELAYED. */
+          String log_query;
+          if (create_insert_stmt_from_insert_delayed(thd, &log_query))
+          {
+            sql_print_error("Event Error: An error occurred while creating query string"
+                            "for INSERT DELAYED stmt, before writing it into binary log.");
+
+            error= 1;
+          }
+          else if (thd->binlog_query(THD::ROW_QUERY_TYPE,
+                                     log_query.c_ptr(), log_query.length(),
+                                     transactional_table, FALSE, FALSE,
+                                     errcode))
+            error= 1;
+        }
+        else if (thd->binlog_query(THD::ROW_QUERY_TYPE,
+			           thd->query(), thd->query_length(),
+			           transactional_table, FALSE, FALSE,
+                                   errcode))
 	  error= 1;
-	}
       }
     }
     DBUG_ASSERT(transactional_table || !changed || 
@@ -2457,6 +2496,65 @@ void kill_delayed_threads(void)
 
 
 /**
+  A strategy for the prelocking algorithm which prevents the
+  delayed insert thread from opening tables with engines which
+  do not support delayed inserts.
+
+  Particularly it allows to abort open_tables() as soon as we
+  discover that we have opened a MERGE table, without acquiring
+  metadata locks on underlying tables.
+*/
+
+class Delayed_prelocking_strategy : public Prelocking_strategy
+{
+public:
+  virtual bool handle_routine(THD *thd, Query_tables_list *prelocking_ctx,
+                              Sroutine_hash_entry *rt, sp_head *sp,
+                              bool *need_prelocking);
+  virtual bool handle_table(THD *thd, Query_tables_list *prelocking_ctx,
+                            TABLE_LIST *table_list, bool *need_prelocking);
+  virtual bool handle_view(THD *thd, Query_tables_list *prelocking_ctx,
+                           TABLE_LIST *table_list, bool *need_prelocking);
+};
+
+
+bool Delayed_prelocking_strategy::
+handle_table(THD *thd, Query_tables_list *prelocking_ctx,
+             TABLE_LIST *table_list, bool *need_prelocking)
+{
+  DBUG_ASSERT(table_list->lock_type == TL_WRITE_DELAYED);
+
+  if (!(table_list->table->file->ha_table_flags() & HA_CAN_INSERT_DELAYED))
+  {
+    my_error(ER_DELAYED_NOT_SUPPORTED, MYF(0), table_list->table_name);
+    return TRUE;
+  }
+  return FALSE;
+}
+
+
+bool Delayed_prelocking_strategy::
+handle_routine(THD *thd, Query_tables_list *prelocking_ctx,
+               Sroutine_hash_entry *rt, sp_head *sp,
+               bool *need_prelocking)
+{
+  /* LEX used by the delayed insert thread has no routines. */
+  DBUG_ASSERT(0);
+  return FALSE;
+}
+
+
+bool Delayed_prelocking_strategy::
+handle_view(THD *thd, Query_tables_list *prelocking_ctx,
+            TABLE_LIST *table_list, bool *need_prelocking)
+{
+  /* We don't open views in the delayed insert thread. */
+  DBUG_ASSERT(0);
+  return FALSE;
+}
+
+
+/**
    Open and lock table for use by delayed thread and check that
    this table is suitable for delayed inserts.
 
@@ -2466,21 +2564,21 @@ void kill_delayed_threads(void)
 
 bool Delayed_insert::open_and_lock_table()
 {
+  Delayed_prelocking_strategy prelocking_strategy;
+
+  /*
+    Use special prelocking strategy to get ER_DELAYED_NOT_SUPPORTED
+    error for tables with engines which don't support delayed inserts.
+  */
   if (!(table= open_n_lock_single_table(&thd, &table_list,
                                         TL_WRITE_DELAYED,
-                                        MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK)))
+                                        MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK,
+                                        &prelocking_strategy)))
   {
     thd.fatal_error();				// Abort waiting inserts
     return TRUE;
   }
-  if (!(table->file->ha_table_flags() & HA_CAN_INSERT_DELAYED))
-  {
-    /* To rollback InnoDB statement transaction. */
-    trans_rollback_stmt(&thd);
-    my_error(ER_DELAYED_NOT_SUPPORTED, MYF(ME_FATALERROR),
-             table_list.table_name);
-    return TRUE;
-  }
+
   if (table->triggers)
   {
     /*
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index c6662501cf6..3a559433e2d 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2357,10 +2357,20 @@ struct LEX: public Query_tables_list
 
     This pointer is required to add possibly omitted DEFINER-clause to the
     DDL-statement before dumping it to the binlog.
+
+    keyword_delayed_begin_offset is the offset to the beginning of the DELAYED
+    keyword in INSERT DELAYED statement. keyword_delayed_end_offset is the
+    offset to the character right after the DELAYED keyword.
   */
-  const char *stmt_definition_begin;
+  union {
+    const char *stmt_definition_begin;
+    uint keyword_delayed_begin_offset;
+  };
 
-  const char *stmt_definition_end;
+  union {
+    const char *stmt_definition_end;
+    uint keyword_delayed_end_offset;
+  };
 
   /**
     During name resolution search only in the table list given by 
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 15efa488173..ca6e0d818e2 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -1345,6 +1345,7 @@ READ_INFO::READ_INFO(File file_par, uint tot_length, CHARSET_INFO *cs,
 		      MYF(MY_WME)))
     {
       my_free(buffer); /* purecov: inspected */
+      buffer= NULL;
       error=1;
     }
     else
@@ -1371,13 +1372,11 @@ READ_INFO::READ_INFO(File file_par, uint tot_length, CHARSET_INFO *cs,
 
 READ_INFO::~READ_INFO()
 {
-  if (!error)
-  {
-    if (need_end_io_cache)
-      ::end_io_cache(&cache);
+  if (!error && need_end_io_cache)
+    ::end_io_cache(&cache);
+
+  if (buffer != NULL)
     my_free(buffer);
-    error=1;
-  }
   List_iterator<XML_TAG> xmlit(taglist);
   XML_TAG *t;
   while ((t= xmlit++))
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 5d669de3110..e44d2563483 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -177,8 +177,7 @@ static bool some_non_temp_table_to_be_updated(THD *thd, TABLE_LIST *tables)
   for (TABLE_LIST *table= tables; table; table= table->next_global)
   {
     DBUG_ASSERT(table->db && table->table_name);
-    if (table->updating &&
-        !find_temporary_table(thd, table->db, table->table_name))
+    if (table->updating && !find_temporary_table(thd, table))
       return 1;
   }
   return 0;
@@ -402,6 +401,7 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_REPAIR]=    CF_WRITE_LOGS_COMMAND | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_OPTIMIZE]|= CF_WRITE_LOGS_COMMAND | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ANALYZE]=   CF_WRITE_LOGS_COMMAND | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_CHECK]=     CF_WRITE_LOGS_COMMAND | CF_AUTO_COMMIT_TRANS;
 
   sql_command_flags[SQLCOM_CREATE_USER]|=       CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_USER]|=         CF_AUTO_COMMIT_TRANS;
@@ -415,7 +415,6 @@ void init_update_queries(void)
 
   sql_command_flags[SQLCOM_FLUSH]=              CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_RESET]=              CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_CHECK]=              CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_CREATE_SERVER]=      CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ALTER_SERVER]=       CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_SERVER]=        CF_AUTO_COMMIT_TRANS;
@@ -642,45 +641,6 @@ end:
 }
 
 
-/**
-  @brief Check access privs for a MERGE table and fix children lock types.
-
-  @param[in]        thd         thread handle
-  @param[in]        db          database name
-  @param[in,out]    table_list  list of child tables (merge_list)
-                                lock_type and optionally db set per table
-
-  @return           status
-    @retval         0           OK
-    @retval         != 0        Error
-
-  @detail
-    This function is used for write access to MERGE tables only
-    (CREATE TABLE, ALTER TABLE ... UNION=(...)). Set TL_WRITE for
-    every child. Set 'db' for every child if not present.
-*/
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-bool check_merge_table_access(THD *thd, char *db, TABLE_LIST *table_list)
-{
-  int error= 0;
-
-  if (table_list)
-  {
-    /* Check that all tables use the current database */
-    TABLE_LIST *tlist;
-
-    for (tlist= table_list; tlist; tlist= tlist->next_local)
-    {
-      if (!tlist->db || !tlist->db[0])
-        tlist->db= db; /* purecov: inspected */
-    }
-    error= check_table_access(thd, SELECT_ACL | UPDATE_ACL | DELETE_ACL,
-                              table_list, FALSE, UINT_MAX, FALSE);
-  }
-  return error;
-}
-#endif
-
 /* This works because items are allocated with sql_alloc() */
 
 void free_items(Item *item)
@@ -6910,10 +6870,16 @@ bool create_table_precheck(THD *thd, TABLE_LIST *tables,
   if (check_access(thd, want_priv, create_table->db,
                    &create_table->grant.privilege,
                    &create_table->grant.m_internal,
-                   0, 0) ||
-      check_merge_table_access(thd, create_table->db,
-                               lex->create_info.merge_list.first))
+                   0, 0))
+    goto err;
+
+  /* If it is a merge table, check privileges for merge children. */
+  if (lex->create_info.merge_list.first &&
+      check_table_access(thd, SELECT_ACL | UPDATE_ACL | DELETE_ACL,
+                         lex->create_info.merge_list.first,
+                         FALSE, UINT_MAX, FALSE))
     goto err;
+
   if (want_priv != CREATE_TMP_ACL &&
       check_grant(thd, want_priv, create_table, FALSE, 1, FALSE))
     goto err;
diff --git a/sql/sql_parse.h b/sql/sql_parse.h
index ad620544d29..c9b0f981d19 100644
--- a/sql/sql_parse.h
+++ b/sql/sql_parse.h
@@ -153,7 +153,6 @@ bool check_single_table_access(THD *thd, ulong privilege,
 bool check_routine_access(THD *thd,ulong want_access,char *db,char *name,
 			  bool is_proc, bool no_errors);
 bool check_some_access(THD *thd, ulong want_access, TABLE_LIST *table);
-bool check_merge_table_access(THD *thd, char *db, TABLE_LIST *table_list);
 bool check_some_routine_access(THD *thd, const char *db, const char *name, bool is_proc);
 bool check_access(THD *thd, ulong want_access, const char *db, ulong *save_priv,
                   GRANT_INTERNAL_INFO *grant_internal_info,
@@ -176,8 +175,6 @@ inline bool check_some_access(THD *thd, ulong want_access, TABLE_LIST *table)
   table->grant.privilege= want_access;
   return false;
 }
-inline bool check_merge_table_access(THD *thd, char *db, TABLE_LIST *table_list)
-{ return false; }
 inline bool check_some_routine_access(THD *thd, const char *db,
                                       const char *name, bool is_proc)
 { return false; }
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index b72816f8ce3..0629d9f3737 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -70,6 +70,7 @@
 #ifdef WITH_PARTITION_STORAGE_ENGINE
 #include "ha_partition.h"
 
+/* TODO: Change abort() to DBUG_SUICIDE() when bug#52002 is pushed */
 #define ERROR_INJECT_CRASH(code) \
   DBUG_EVALUATE_IF(code, (abort(), 0), 0)
 #define ERROR_INJECT_ERROR(code) \
@@ -89,7 +90,6 @@ const LEX_STRING partition_keywords[]=
   { C_STRING_WITH_LEN(" COLUMNS") }
 };
 static const char *part_str= "PARTITION";
-static const char *subpart_str= "SUBPARTITION";
 static const char *sub_str= "SUB";
 static const char *by_str= "BY";
 static const char *space_str= " ";
@@ -1762,8 +1762,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
         goto end;
       if (unlikely(part_info->subpart_expr->result_type() != INT_RESULT))
       {
-        my_error(ER_PARTITION_FUNCTION_IS_NOT_ALLOWED, MYF(0),
-                 subpart_str);
+        part_info->report_part_expr_error(TRUE);
         goto end;
       }
     }
@@ -1790,10 +1789,9 @@ bool fix_partition_func(THD *thd, TABLE *table,
         goto end;
       if (unlikely(part_info->part_expr->result_type() != INT_RESULT))
       {
-        my_error(ER_PARTITION_FUNCTION_IS_NOT_ALLOWED, MYF(0), part_str);
+        part_info->report_part_expr_error(FALSE);
         goto end;
       }
-      part_info->part_result_type= INT_RESULT;
     }
     part_info->fixed= TRUE;
   }
@@ -1839,18 +1837,22 @@ bool fix_partition_func(THD *thd, TABLE *table,
     if (unlikely(!part_info->column_list &&
                   part_info->part_expr->result_type() != INT_RESULT))
     {
-      my_error(ER_PARTITION_FUNC_NOT_ALLOWED_ERROR, MYF(0), part_str);
+      part_info->report_part_expr_error(FALSE);
       goto end;
     }
   }
   if (((part_info->part_type != HASH_PARTITION ||
-      part_info->list_of_part_fields == FALSE) &&
-      (!part_info->column_list &&
-      check_part_func_fields(part_info->part_field_array, TRUE))) ||
+        part_info->list_of_part_fields == FALSE) &&
+       !part_info->column_list &&
+       check_part_func_fields(part_info->part_field_array, TRUE)) ||
       (part_info->list_of_subpart_fields == FALSE &&
        part_info->is_sub_partitioned() &&
        check_part_func_fields(part_info->subpart_field_array, TRUE)))
   {
+    /*
+      Range/List/HASH (but not KEY) and not COLUMNS or HASH subpartitioning
+      with columns in the partitioning expression using unallowed charset.
+    */
     my_error(ER_PARTITION_FUNCTION_IS_NOT_ALLOWED, MYF(0));
     goto end;
   }
@@ -4353,43 +4355,31 @@ set_engine_all_partitions(partition_info *part_info,
     }
   } while (++i < part_info->num_parts);
 }
-/*
-  SYNOPSIS
-    fast_end_partition()
-    thd                           Thread object
-    out:copied                    Number of records copied
-    out:deleted                   Number of records deleted
-    table_list                    Table list with the one table in it
-    empty                         Has nothing been done
-    lpt                           Struct to be used by error handler
 
-  RETURN VALUES
-    FALSE                         Success
-    TRUE                          Failure
 
-  DESCRIPTION
-    Support routine to handle the successful cases for partition
-    management.
+/**
+  Support routine to handle the successful cases for partition management.
+
+  @param thd               Thread object
+  @param copied            Number of records copied
+  @param deleted           Number of records deleted
+  @param table_list        Table list with the one table in it
+
+  @return Operation status
+    @retval FALSE          Success
+    @retval TRUE           Failure
 */
 
 static int fast_end_partition(THD *thd, ulonglong copied,
                               ulonglong deleted,
-                              TABLE_LIST *table_list, bool is_empty,
-                              ALTER_PARTITION_PARAM_TYPE *lpt,
-                              bool written_bin_log)
+                              TABLE_LIST *table_list)
 {
   char tmp_name[80];
   DBUG_ENTER("fast_end_partition");
 
   thd->proc_info="end";
 
-  if (!is_empty)
-    query_cache_invalidate3(thd, table_list, 0);
-
-  if ((!is_empty) && (!written_bin_log) &&
-      (!thd->lex->no_write_to_binlog) &&
-    write_bin_log(thd, FALSE, thd->query(), thd->query_length()))
-    DBUG_RETURN(TRUE);
+  query_cache_invalidate3(thd, table_list, 0);
 
   my_snprintf(tmp_name, sizeof(tmp_name), ER(ER_INSERT_INFO),
               (ulong) (copied + deleted),
@@ -4511,29 +4501,25 @@ uint set_part_state(Alter_info *alter_info, partition_info *tab_part_info,
 }
 
 
-/*
+/**
   Prepare for ALTER TABLE of partition structure
 
-  SYNOPSIS
-    prep_alter_part_table()
-    thd                        Thread object
-    table                      Table object
-    inout:alter_info           Alter information
-    inout:create_info          Create info for CREATE TABLE
-    old_db_type                Old engine type
-    out:partition_changed      Boolean indicating whether partition changed
-    out:fast_alter_partition   Boolean indicating whether fast partition
-                               change is requested
+  @param[in] thd                 Thread object
+  @param[in] table               Table object
+  @param[in,out] alter_info      Alter information
+  @param[in,out] create_info     Create info for CREATE TABLE
+  @param[in] old_db_type         Old engine type
+  @param[out] partition_changed  Boolean indicating whether partition changed
+  @param[out] fast_alter_table   Internal temporary table allowing fast
+                                 partition change or NULL if not possible
 
-  RETURN VALUES
-    TRUE                       Error
-    FALSE                      Success
-    partition_changed
-    fast_alter_partition
+  @return Operation status
+    @retval TRUE                 Error
+    @retval FALSE                Success
 
-  DESCRIPTION
+  @note 
     This method handles all preparations for ALTER TABLE for partitioned
-    tables
+    tables.
     We need to handle both partition management command such as Add Partition
     and others here as well as an ALTER TABLE that completely changes the
     partitioning and yet others that don't change anything at all. We start
@@ -4545,8 +4531,12 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
                            HA_CREATE_INFO *create_info,
                            handlerton *old_db_type,
                            bool *partition_changed,
-                           uint *fast_alter_partition)
+                           char *db,
+                           const char *table_name,
+                           const char *path,
+                           TABLE **fast_alter_table)
 {
+  TABLE *new_table= NULL;
   DBUG_ENTER("prep_alter_part_table");
 
   /* Foreign keys on partitioned tables are not supported, waits for WL#148 */
@@ -4555,15 +4545,9 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
     my_error(ER_FOREIGN_KEY_ON_PARTITIONED, MYF(0));
     DBUG_RETURN(TRUE);
   }
-  /*
-    We are going to manipulate the partition info on the table object
-    so we need to ensure that the table instance is removed from the
-    table cache.
-  */
-  if (table->part_info)
-    table->m_needs_reopen= TRUE;
 
   thd->work_part_info= thd->lex->part_info;
+
   if (thd->work_part_info &&
       !(thd->work_part_info= thd->lex->part_info->get_clone()))
     DBUG_RETURN(TRUE);
@@ -4576,7 +4560,7 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
        ALTER_COALESCE_PARTITION | ALTER_REORGANIZE_PARTITION |
        ALTER_TABLE_REORG | ALTER_REBUILD_PARTITION))
   {
-    partition_info *tab_part_info= table->part_info;
+    partition_info *tab_part_info;
     partition_info *alt_part_info= thd->work_part_info;
     uint flags= 0;
     bool is_last_partition_reorged= FALSE;
@@ -4584,11 +4568,32 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
     part_elem_value *alt_max_elem_val= NULL;
     longlong tab_max_range= 0, alt_max_range= 0;
 
-    if (!tab_part_info)
+    if (!table->part_info)
     {
       my_error(ER_PARTITION_MGMT_ON_NONPARTITIONED, MYF(0));
       DBUG_RETURN(TRUE);
     }
+
+    /*
+      Open our intermediate table, we will operate on a temporary instance
+      of the original table, to be able to skip copying all partitions.
+      Open it as a copy of the original table, and modify its partition_info
+      object to allow fast_alter_partition_table to perform the changes.
+    */
+    DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, db, table_name,
+                                               MDL_INTENTION_EXCLUSIVE));
+    new_table= open_table_uncached(thd, path, db, table_name, 0);
+    if (!new_table)
+      DBUG_RETURN(TRUE);
+
+    /*
+      This table may be used for copy rows between partitions
+      and also read/write columns when fixing the partition_info struct.
+    */
+    new_table->use_all_columns();
+ 
+    tab_part_info= new_table->part_info;
+
     if (alter_info->flags & ALTER_TABLE_REORG)
     {
       uint new_part_no, curr_part_no;
@@ -4596,9 +4601,9 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
           tab_part_info->use_default_num_partitions)
       {
         my_error(ER_REORG_NO_PARAM_ERROR, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
-      new_part_no= table->file->get_default_no_partitions(create_info);
+      new_part_no= new_table->file->get_default_no_partitions(create_info);
       curr_part_no= tab_part_info->num_parts;
       if (new_part_no == curr_part_no)
       {
@@ -4607,7 +4612,8 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
           after the change as before. Thus we can reply ok immediately
           without any changes at all.
         */
-        *fast_alter_partition= TRUE;
+        *fast_alter_table= new_table;
+        thd->work_part_info= tab_part_info;
         DBUG_RETURN(FALSE);
       }
       else if (new_part_no > curr_part_no)
@@ -4629,15 +4635,15 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
         alter_info->num_parts= curr_part_no - new_part_no;
       }
     }
-    if (!(flags= table->file->alter_table_flags(alter_info->flags)))
+    if (!(flags= new_table->file->alter_table_flags(alter_info->flags)))
     {
       my_error(ER_PARTITION_FUNCTION_FAILURE, MYF(0));
-      DBUG_RETURN(1);
+      goto err;
     }
-    *fast_alter_partition=
-      ((flags & (HA_FAST_CHANGE_PARTITION | HA_PARTITION_ONE_PHASE)) != 0);
-    DBUG_PRINT("info", ("*fast_alter_partition: %d  flags: 0x%x",
-                        *fast_alter_partition, flags));
+    if ((flags & (HA_FAST_CHANGE_PARTITION | HA_PARTITION_ONE_PHASE)) != 0)
+      *fast_alter_table= new_table;
+    DBUG_PRINT("info", ("*fast_alter_table: %p  flags: 0x%x",
+                        *fast_alter_table, flags));
     if ((alter_info->flags & ALTER_ADD_PARTITION) ||
          (alter_info->flags & ALTER_REORGANIZE_PARTITION))
     {
@@ -4648,12 +4654,12 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
           if (tab_part_info->part_type == RANGE_PARTITION)
           {
             my_error(ER_PARTITIONS_MUST_BE_DEFINED_ERROR, MYF(0), "RANGE");
-            DBUG_RETURN(TRUE);
+            goto err;
           }
           else if (tab_part_info->part_type == LIST_PARTITION)
           {
             my_error(ER_PARTITIONS_MUST_BE_DEFINED_ERROR, MYF(0), "LIST");
-            DBUG_RETURN(TRUE);
+            goto err;
           }
           /*
             Hash partitions can be altered without parser finds out about
@@ -4684,7 +4690,7 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
             my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
                      "LIST", "IN");
           }
-          DBUG_RETURN(TRUE);
+          goto err;
         }
       }
       if ((tab_part_info->column_list &&
@@ -4698,12 +4704,12 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
             alt_part_info->num_columns != 0))
       {
         my_error(ER_PARTITION_COLUMN_LIST_ERROR, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       alt_part_info->column_list= tab_part_info->column_list;
       if (alt_part_info->fix_parser_data(thd))
       {
-        DBUG_RETURN(TRUE);
+        goto err;
       }
     }
     if (alter_info->flags & ALTER_ADD_PARTITION)
@@ -4726,17 +4732,17 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
           tab_part_info->part_type != HASH_PARTITION)
       {
         my_error(ER_NO_BINLOG_ERROR, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       if (tab_part_info->defined_max_value)
       {
         my_error(ER_PARTITION_MAXVALUE_ERROR, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       if (num_new_partitions == 0)
       {
         my_error(ER_ADD_PARTITION_NO_NEW_PARTITION, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       if (tab_part_info->is_sub_partitioned())
       {
@@ -4745,7 +4751,7 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
         else if (alt_part_info->num_subparts != tab_part_info->num_subparts)
         {
           my_error(ER_ADD_PARTITION_SUBPART_ERROR, MYF(0));
-          DBUG_RETURN(TRUE);
+          goto err;
         }
         check_total_partitions= new_total_partitions*
                                 alt_part_info->num_subparts;
@@ -4753,15 +4759,15 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
       if (check_total_partitions > MAX_PARTITIONS)
       {
         my_error(ER_TOO_MANY_PARTITIONS_ERROR, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       alt_part_info->part_type= tab_part_info->part_type;
       alt_part_info->subpart_type= tab_part_info->subpart_type;
-      if (alt_part_info->set_up_defaults_for_partitioning(table->file,
+      if (alt_part_info->set_up_defaults_for_partitioning(new_table->file,
                                                     ULL(0), 
                                                     tab_part_info->num_parts))
       {
-        DBUG_RETURN(TRUE);
+        goto err;
       }
 /*
 Handling of on-line cases:
@@ -4825,7 +4831,7 @@ adding and copying partitions, the second after completing the adding
 and copying and finally the third line after also dropping the partitions
 that are reorganised.
 */
-      if (*fast_alter_partition &&
+      if (*fast_alter_table &&
           tab_part_info->part_type == HASH_PARTITION)
       {
         uint part_no= 0, start_part= 1, start_sec_part= 1;
@@ -4930,12 +4936,12 @@ that are reorganised.
         do
         {
           partition_element *part_elem= alt_it++;
-          if (*fast_alter_partition)
+          if (*fast_alter_table)
             part_elem->part_state= PART_TO_BE_ADDED;
           if (tab_part_info->partitions.push_back(part_elem))
           {
             mem_alloc_error(1);
-            DBUG_RETURN(TRUE);
+            goto err;
           }
         } while (++part_count < num_new_partitions);
         tab_part_info->num_parts+= num_new_partitions;
@@ -4976,12 +4982,12 @@ that are reorganised.
             tab_part_info->part_type == LIST_PARTITION))
       {
         my_error(ER_ONLY_ON_RANGE_LIST_PARTITION, MYF(0), "DROP");
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       if (num_parts_dropped >= tab_part_info->num_parts)
       {
         my_error(ER_DROP_LAST_PARTITION, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       do
       {
@@ -4999,12 +5005,12 @@ that are reorganised.
       if (num_parts_found != num_parts_dropped)
       {
         my_error(ER_DROP_PARTITION_NON_EXISTENT, MYF(0), "DROP");
-        DBUG_RETURN(TRUE);
+        goto err;
       }
-      if (table->file->is_fk_defined_on_table_or_index(MAX_KEY))
+      if (new_table->file->is_fk_defined_on_table_or_index(MAX_KEY))
       {
         my_error(ER_ROW_IS_REFERENCED, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       tab_part_info->num_parts-= num_parts_dropped;
     }
@@ -5017,12 +5023,12 @@ that are reorganised.
           (!(alter_info->flags & ALTER_ALL_PARTITION)))
       {
         my_error(ER_DROP_PARTITION_NON_EXISTENT, MYF(0), "REBUILD");
-        DBUG_RETURN(TRUE);
+        goto err;
       }
-      if (!(*fast_alter_partition))
+      if (!(*fast_alter_table))
       {
-        table->file->print_error(HA_ERR_WRONG_COMMAND, MYF(0));
-        DBUG_RETURN(TRUE);
+        new_table->file->print_error(HA_ERR_WRONG_COMMAND, MYF(0));
+        goto err;
       }
     }
     else if (alter_info->flags & ALTER_COALESCE_PARTITION)
@@ -5033,17 +5039,17 @@ that are reorganised.
       if (tab_part_info->part_type != HASH_PARTITION)
       {
         my_error(ER_COALESCE_ONLY_ON_HASH_PARTITION, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       if (num_parts_coalesced == 0)
       {
         my_error(ER_COALESCE_PARTITION_NO_PARTITION, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       if (num_parts_coalesced >= tab_part_info->num_parts)
       {
         my_error(ER_DROP_LAST_PARTITION, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
 /*
 Online handling:
@@ -5082,7 +5088,7 @@ state of p1.
         uint part_count= 0, start_part= 1, start_sec_part= 1;
         uint end_part= 0, end_sec_part= 0;
         bool all_parts= TRUE;
-        if (*fast_alter_partition &&
+        if (*fast_alter_table &&
             tab_part_info->linear_hash_ind)
         {
           uint upper_2n= tab_part_info->linear_hash_mask + 1;
@@ -5108,14 +5114,14 @@ state of p1.
         do
         {
           partition_element *p_elem= part_it++;
-          if (*fast_alter_partition &&
+          if (*fast_alter_table &&
               (all_parts ||
               (part_count >= start_part && part_count <= end_part) ||
               (part_count >= start_sec_part && part_count <= end_sec_part)))
             p_elem->part_state= PART_CHANGED;
           if (++part_count > num_parts_remain)
           {
-            if (*fast_alter_partition)
+            if (*fast_alter_table)
               p_elem->part_state= PART_REORGED_DROPPED;
             else
               part_it.remove();
@@ -5150,38 +5156,38 @@ state of p1.
       if (num_parts_reorged > tab_part_info->num_parts)
       {
         my_error(ER_REORG_PARTITION_NOT_EXIST, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       if (!(tab_part_info->part_type == RANGE_PARTITION ||
             tab_part_info->part_type == LIST_PARTITION) &&
            (num_parts_new != num_parts_reorged))
       {
         my_error(ER_REORG_HASH_ONLY_ON_SAME_NO, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       if (tab_part_info->is_sub_partitioned() &&
           alt_part_info->num_subparts &&
           alt_part_info->num_subparts != tab_part_info->num_subparts)
       {
         my_error(ER_PARTITION_WRONG_NO_SUBPART_ERROR, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       check_total_partitions= tab_part_info->num_parts + num_parts_new;
       check_total_partitions-= num_parts_reorged;
       if (check_total_partitions > MAX_PARTITIONS)
       {
         my_error(ER_TOO_MANY_PARTITIONS_ERROR, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       alt_part_info->part_type= tab_part_info->part_type;
       alt_part_info->subpart_type= tab_part_info->subpart_type;
       alt_part_info->num_subparts= tab_part_info->num_subparts;
       DBUG_ASSERT(!alt_part_info->use_default_partitions);
-      if (alt_part_info->set_up_defaults_for_partitioning(table->file,
+      if (alt_part_info->set_up_defaults_for_partitioning(new_table->file,
                                                           ULL(0), 
                                                           0))
       {
-        DBUG_RETURN(TRUE);
+        goto err;
       }
 /*
 Online handling:
@@ -5240,13 +5246,13 @@ the generated partition syntax in a correct manner.
             }
             else
               tab_max_range= part_elem->range_value;
-            if (*fast_alter_partition &&
+            if (*fast_alter_table &&
                 tab_part_info->temp_partitions.push_back(part_elem))
             {
               mem_alloc_error(1);
-              DBUG_RETURN(TRUE);
+              goto err;
             }
-            if (*fast_alter_partition)
+            if (*fast_alter_table)
               part_elem->part_state= PART_TO_BE_REORGED;
             if (!found_first)
             {
@@ -5266,7 +5272,7 @@ the generated partition syntax in a correct manner.
                 else
                   alt_max_range= alt_part_elem->range_value;
 
-                if (*fast_alter_partition)
+                if (*fast_alter_table)
                   alt_part_elem->part_state= PART_TO_BE_ADDED;
                 if (alt_part_count == 0)
                   tab_it.replace(alt_part_elem);
@@ -5277,7 +5283,7 @@ the generated partition syntax in a correct manner.
             else if (found_last)
             {
               my_error(ER_CONSECUTIVE_REORG_PARTITIONS, MYF(0));
-              DBUG_RETURN(TRUE);
+              goto err;
             }
             else
               tab_it.remove();
@@ -5291,7 +5297,7 @@ the generated partition syntax in a correct manner.
         if (drop_count != num_parts_reorged)
         {
           my_error(ER_DROP_PARTITION_NON_EXISTENT, MYF(0), "REORGANIZE");
-          DBUG_RETURN(TRUE);
+          goto err;
         }
         tab_part_info->num_parts= check_total_partitions;
       }
@@ -5312,9 +5318,9 @@ the generated partition syntax in a correct manner.
         tab_part_info->use_default_num_subpartitions= FALSE;
       }
       if (tab_part_info->check_partition_info(thd, (handlerton**)NULL,
-                                              table->file, ULL(0), TRUE))
+                                              new_table->file, ULL(0), TRUE))
       {
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       /*
         The check below needs to be performed after check_partition_info
@@ -5346,7 +5352,7 @@ the generated partition syntax in a correct manner.
           to create "holes".
         */
         my_error(ER_REORG_OUTSIDE_RANGE, MYF(0));
-        DBUG_RETURN(TRUE);
+        goto err;
       }
     }
   }
@@ -5463,7 +5469,7 @@ the generated partition syntax in a correct manner.
         *partition_changed= TRUE;
         if (thd->work_part_info->fix_parser_data(thd))
         {
-          DBUG_RETURN(TRUE);
+          goto err;
         }
       }
       /*
@@ -5484,7 +5490,7 @@ the generated partition syntax in a correct manner.
       if (check_native_partitioned(create_info, &is_native_partitioned,
                                    part_info, thd))
       {
-        DBUG_RETURN(TRUE);
+        goto err;
       }
       if (!is_native_partitioned)
       {
@@ -5494,6 +5500,17 @@ the generated partition syntax in a correct manner.
     }
   }
   DBUG_RETURN(FALSE);
+err:
+  if (new_table)
+  {
+    /*
+      Only remove the intermediate table object and its share object,
+      do not remove the .frm file, since it is the original one.
+    */
+    close_temporary(new_table, 1, 0);
+  }
+  *fast_alter_table= NULL;
+  DBUG_RETURN(TRUE);
 }
 
 
@@ -5536,6 +5553,11 @@ static bool mysql_change_partitions(ALTER_PARTITION_PARAM_TYPE *lpt)
   if(mysql_trans_prepare_alter_copy_data(thd))
     DBUG_RETURN(TRUE);
 
+  if (file->ha_external_lock(thd, F_WRLCK))
+    DBUG_RETURN(TRUE);
+
+  /* TODO: test if bulk_insert would increase the performance */
+
   if ((error= file->ha_change_partitions(lpt->create_info, path, &lpt->copied,
                                          &lpt->deleted, lpt->pack_frm_data,
                                          lpt->pack_frm_len)))
@@ -5544,7 +5566,10 @@ static bool mysql_change_partitions(ALTER_PARTITION_PARAM_TYPE *lpt)
   }
 
   if (mysql_trans_commit_alter_copy_data(thd))
-    DBUG_RETURN(TRUE);                      /* The error has been reported */
+    error= 1;                                /* The error has been reported */
+
+  if (file->ha_external_lock(thd, F_UNLCK))
+    error= 1;
 
   DBUG_RETURN(test(error));
 }
@@ -6294,8 +6319,18 @@ static void alter_partition_lock_handling(ALTER_PARTITION_PARAM_TYPE *lpt)
 {
   THD *thd= lpt->thd;
 
-  close_all_tables_for_name(thd, lpt->table->s, FALSE);
+  if (lpt->old_table)
+    close_all_tables_for_name(thd, lpt->old_table->s, FALSE);
+  if (lpt->table)
+  {
+    /*
+      Only remove the intermediate table object and its share object,
+      do not remove the .frm file, since it is the original one.
+    */
+    close_temporary(lpt->table, 1, 0);
+  }
   lpt->table= 0;
+  lpt->old_table= 0;
   lpt->table_list->table= 0;
   if (thd->locked_tables_list.reopen_tables(thd))
     sql_print_warning("We failed to reacquire LOCKs in ALTER TABLE");
@@ -6307,62 +6342,40 @@ static void alter_partition_lock_handling(ALTER_PARTITION_PARAM_TYPE *lpt)
   SYNOPSIS
     alter_close_tables()
     lpt                        Struct carrying parameters
+    close_old                  Close original table too
   RETURN VALUES
     0
 */
 
-static int alter_close_tables(ALTER_PARTITION_PARAM_TYPE *lpt)
+static int alter_close_tables(ALTER_PARTITION_PARAM_TYPE *lpt, bool close_old)
 {
-  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 (lpt->table->db_stat)
   {
-    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, TRUE);
-    }
+    lpt->table->file->close();
+    lpt->table->db_stat= 0;                        // Mark file closed
+  }
+  if (close_old && lpt->old_table)
+  {
+    close_all_tables_for_name(lpt->thd, lpt->old_table->s, FALSE);
+    lpt->old_table= 0;
   }
-  mysql_mutex_unlock(&LOCK_open);
   DBUG_RETURN(0);
 }
 
 
-/*
-  Handle errors for ALTER TABLE for partitioning
-  SYNOPSIS
-    handle_alter_part_error()
-    lpt                        Struct carrying parameters
-    not_completed              Was request in complete phase when error occurred
-  RETURN VALUES
-    NONE
+/**
+  Handle errors for ALTER TABLE for partitioning.
+
+  @param lpt                Struct carrying parameters
+  @param action_completed   The action must be completed, NOT reverted
+  @param drop_partition     Partitions has not been dropped yet
+  @param frm_install        The shadow frm-file has not yet been installed
+  @param close_table        Table is still open, close it before reverting
 */
 
 void handle_alter_part_error(ALTER_PARTITION_PARAM_TYPE *lpt,
-                             bool not_completed,
+                             bool action_completed,
                              bool drop_partition,
                              bool frm_install,
                              bool close_table)
@@ -6379,7 +6392,7 @@ void handle_alter_part_error(ALTER_PARTITION_PARAM_TYPE *lpt,
       so we clone it first to have a copy.
     */
     part_info= lpt->part_info->get_clone();
-    alter_close_tables(lpt);
+    alter_close_tables(lpt, action_completed);
   }
 
   if (part_info->first_log_entry &&
@@ -6392,7 +6405,7 @@ void handle_alter_part_error(ALTER_PARTITION_PARAM_TYPE *lpt,
     */
     write_log_completed(lpt, FALSE);
     release_log_entries(part_info);
-    if (not_completed)
+    if (!action_completed)
     {
       if (drop_partition)
       {
@@ -6457,7 +6470,7 @@ void handle_alter_part_error(ALTER_PARTITION_PARAM_TYPE *lpt,
   else
   {
     release_log_entries(part_info);
-    if (not_completed)
+    if (!action_completed)
     {
       /*
         We hit an error before things were completed but managed
@@ -6500,25 +6513,24 @@ static void downgrade_mdl_if_lock_tables_mode(THD *thd, MDL_ticket *ticket,
 }
 
 
-/*
+/**
   Actually perform the change requested by ALTER TABLE of partitions
   previously prepared.
 
-  SYNOPSIS
-    fast_alter_partition_table()
-    thd                           Thread object
-    table                         Table object
-    alter_info                    ALTER TABLE info
-    create_info                   Create info for CREATE TABLE
-    table_list                    List of the table involved
-    db                            Database name of new table
-    table_name                    Table name of new table
+  @param thd                           Thread object
+  @param table                         Original table object
+  @param alter_info                    ALTER TABLE info
+  @param create_info                   Create info for CREATE TABLE
+  @param table_list                    List of the table involved
+  @param db                            Database name of new table
+  @param table_name                    Table name of new table
+  @param fast_alter_table              Prepared table object
 
-  RETURN VALUES
-    TRUE                          Error
-    FALSE                         Success
+  @return Operation status
+    @retval TRUE                          Error
+    @retval FALSE                         Success
 
-  DESCRIPTION
+  @note
     Perform all ALTER TABLE operations for partitioned tables that can be
     performed fast without a full copy of the original table.
 */
@@ -6529,19 +6541,20 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
                                 TABLE_LIST *table_list,
                                 char *db,
                                 const char *table_name,
-                                uint fast_alter_partition)
+                                TABLE *fast_alter_table)
 {
   /* Set-up struct used to write frm files */
-  partition_info *part_info= table->part_info;
+  partition_info *part_info;
   ALTER_PARTITION_PARAM_TYPE lpt_obj;
   ALTER_PARTITION_PARAM_TYPE *lpt= &lpt_obj;
-  bool written_bin_log= TRUE;
-  bool not_completed= TRUE;
+  bool action_completed= FALSE;
   bool close_table_on_failure= FALSE;
   bool frm_install= FALSE;
   MDL_ticket *mdl_ticket= table->mdl_ticket;
+  DBUG_ASSERT(fast_alter_table);
   DBUG_ENTER("fast_alter_partition_table");
 
+  part_info= fast_alter_table->part_info;
   lpt->thd= thd;
   lpt->table_list= table_list;
   lpt->part_info= part_info;
@@ -6550,7 +6563,8 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
   lpt->db_options= create_info->table_options;
   if (create_info->row_type == ROW_TYPE_DYNAMIC)
     lpt->db_options|= HA_OPTION_PACK_RECORD;
-  lpt->table= table;
+  lpt->table= fast_alter_table;
+  lpt->old_table= table;
   lpt->key_info_buffer= 0;
   lpt->key_count= 0;
   lpt->db= db;
@@ -6559,12 +6573,12 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
   lpt->deleted= 0;
   lpt->pack_frm_data= NULL;
   lpt->pack_frm_len= 0;
-  thd->work_part_info= part_info;
 
   /* Never update timestamp columns when alter */
-  table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
+  lpt->table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
 
-  if (fast_alter_partition & HA_PARTITION_ONE_PHASE)
+  if (table->file->alter_table_flags(alter_info->flags) &
+        HA_PARTITION_ONE_PHASE)
   {
     /*
       In the case where the engine supports one phase online partition
@@ -6681,11 +6695,11 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT_ERROR("fail_drop_partition_3") ||
         (close_table_on_failure= TRUE, FALSE) ||
         write_log_drop_partition(lpt) ||
+        (action_completed= TRUE, FALSE) ||
         ERROR_INJECT_CRASH("crash_drop_partition_4") ||
         ERROR_INJECT_ERROR("fail_drop_partition_4") ||
+        alter_close_tables(lpt, action_completed) ||
         (close_table_on_failure= FALSE, FALSE) ||
-        (not_completed= FALSE, FALSE) ||
-        alter_close_tables(lpt) ||
         ERROR_INJECT_CRASH("crash_drop_partition_5") ||
         ERROR_INJECT_ERROR("fail_drop_partition_5") ||
         ((!thd->lex->no_write_to_binlog) &&
@@ -6706,7 +6720,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT_ERROR("fail_drop_partition_9") ||
         (alter_partition_lock_handling(lpt), FALSE)) 
     {
-      handle_alter_part_error(lpt, not_completed, TRUE, frm_install,
+      handle_alter_part_error(lpt, action_completed, TRUE, frm_install,
                               close_table_on_failure);
       goto err;
     }
@@ -6761,7 +6775,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT_CRASH("crash_add_partition_5") ||
         ERROR_INJECT_ERROR("fail_add_partition_5") ||
         (close_table_on_failure= FALSE, FALSE) ||
-        alter_close_tables(lpt) ||
+        alter_close_tables(lpt, action_completed) ||
         ERROR_INJECT_CRASH("crash_add_partition_6") ||
         ERROR_INJECT_ERROR("fail_add_partition_6") ||
         ((!thd->lex->no_write_to_binlog) &&
@@ -6770,11 +6784,12 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT_CRASH("crash_add_partition_7") ||
         ERROR_INJECT_ERROR("fail_add_partition_7") ||
         write_log_rename_frm(lpt) ||
-        (frm_install= TRUE, FALSE) ||
-        (not_completed= FALSE, FALSE) ||
+        (action_completed= TRUE, FALSE) ||
         ERROR_INJECT_CRASH("crash_add_partition_8") ||
         ERROR_INJECT_ERROR("fail_add_partition_8") ||
+        (frm_install= TRUE, FALSE) ||
         mysql_write_frm(lpt, WFRM_INSTALL_SHADOW) ||
+        (frm_install= FALSE, FALSE) ||
         ERROR_INJECT_CRASH("crash_add_partition_9") ||
         ERROR_INJECT_ERROR("fail_add_partition_9") ||
         (write_log_completed(lpt, FALSE), FALSE) ||
@@ -6782,7 +6797,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT_ERROR("fail_add_partition_10") ||
         (alter_partition_lock_handling(lpt), FALSE))
     {
-      handle_alter_part_error(lpt, not_completed, FALSE, frm_install,
+      handle_alter_part_error(lpt, action_completed, FALSE, frm_install,
                               close_table_on_failure);
       goto err;
     }
@@ -6831,10 +6846,10 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
          can release all other locks on the table and since no one can open
          the table, there can be no new threads accessing the table. They
          will be hanging on this exclusive lock.
-      5) Close all instances of the table and remove them from the table cache.
-      6) Log that operation is completed and log all complete actions
+      5) Log that operation is completed and log all complete actions
          needed to complete operation from here
-      7) Write bin log
+      6) Write bin log
+      7) Close all instances of the table and remove them from the table cache.
       8) Prepare handlers for rename and delete of partitions
       9) Rename and drop the reorged partitions such that they are no
          longer used and rename those added to their real new names.
@@ -6858,27 +6873,28 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         wait_while_table_is_used(thd, table, HA_EXTRA_FORCE_REOPEN) ||
         ERROR_INJECT_CRASH("crash_change_partition_5") ||
         ERROR_INJECT_ERROR("fail_change_partition_5") ||
-        alter_close_tables(lpt) ||
-        (close_table_on_failure= FALSE) ||
+        write_log_final_change_partition(lpt) ||
+        (action_completed= TRUE, FALSE) ||
         ERROR_INJECT_CRASH("crash_change_partition_6") ||
         ERROR_INJECT_ERROR("fail_change_partition_6") ||
-        write_log_final_change_partition(lpt) ||
-        (not_completed= FALSE) ||
-        ERROR_INJECT_CRASH("crash_change_partition_7") ||
-        ERROR_INJECT_ERROR("fail_change_partition_7") ||
         ((!thd->lex->no_write_to_binlog) &&
          (write_bin_log(thd, FALSE,
                         thd->query(), thd->query_length()), FALSE)) ||
+        ERROR_INJECT_CRASH("crash_change_partition_7") ||
+        ERROR_INJECT_ERROR("fail_change_partition_7") ||
+        ((frm_install= TRUE), FALSE) ||
+        mysql_write_frm(lpt, WFRM_INSTALL_SHADOW) ||
+        (frm_install= FALSE, FALSE) ||
         ERROR_INJECT_CRASH("crash_change_partition_8") ||
         ERROR_INJECT_ERROR("fail_change_partition_8") ||
-        mysql_write_frm(lpt, WFRM_INSTALL_SHADOW) ||
+        alter_close_tables(lpt, action_completed) ||
+        (close_table_on_failure= FALSE, FALSE) ||
         ERROR_INJECT_CRASH("crash_change_partition_9") ||
         ERROR_INJECT_ERROR("fail_change_partition_9") ||
         mysql_drop_partitions(lpt) ||
         ERROR_INJECT_CRASH("crash_change_partition_10") ||
         ERROR_INJECT_ERROR("fail_change_partition_10") ||
         mysql_rename_partitions(lpt) ||
-        ((frm_install= TRUE), FALSE) ||
         ERROR_INJECT_CRASH("crash_change_partition_11") ||
         ERROR_INJECT_ERROR("fail_change_partition_11") ||
         (write_log_completed(lpt, FALSE), FALSE) ||
@@ -6886,7 +6902,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT_ERROR("fail_change_partition_12") ||
         (alter_partition_lock_handling(lpt), FALSE))
     {
-      handle_alter_part_error(lpt, not_completed, FALSE, frm_install,
+      handle_alter_part_error(lpt, action_completed, FALSE, frm_install,
                               close_table_on_failure);
       goto err;
     }
@@ -6896,12 +6912,25 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
     A final step is to write the query to the binlog and send ok to the
     user
   */
-  DBUG_RETURN(fast_end_partition(thd, lpt->copied, lpt->deleted,
-                                 table_list, FALSE, NULL,
-                                 written_bin_log));
+  DBUG_RETURN(fast_end_partition(thd, lpt->copied, lpt->deleted, table_list));
 err:
+  if (action_completed)
+  {
+    /*
+      Although error occurred, the action was forced to retry for completion.
+      Therefore we must close+reopen all instances of the table.
+    */
+    (void) alter_partition_lock_handling(lpt);
+  }
+  else
+  {
+    /*
+      The failed action was reverted, leave the original table as is and
+      close/destroy the intermediate table object and its share.
+    */
+    close_temporary(lpt->table, 1, 0);
+  }
   downgrade_mdl_if_lock_tables_mode(thd, mdl_ticket, MDL_SHARED_NO_READ_WRITE);
-  table->m_needs_reopen= TRUE;
   DBUG_RETURN(TRUE);
 }
 #endif
diff --git a/sql/sql_partition.h b/sql/sql_partition.h
index c644e63794c..9a9a0bd56fa 100644
--- a/sql/sql_partition.h
+++ b/sql/sql_partition.h
@@ -54,6 +54,7 @@ typedef struct st_lock_param_type
   HA_CREATE_INFO *create_info;
   Alter_info *alter_info;
   TABLE *table;
+  TABLE *old_table;
   KEY *key_info_buffer;
   const char *db;
   const char *table_name;
@@ -252,14 +253,17 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
                                 TABLE_LIST *table_list,
                                 char *db,
                                 const char *table_name,
-                                uint fast_alter_partition);
+                                TABLE  *fast_alter_table);
 uint set_part_state(Alter_info *alter_info, partition_info *tab_part_info,
                     enum partition_state part_state);
 uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
                            HA_CREATE_INFO *create_info,
                            handlerton *old_db_type,
                            bool *partition_changed,
-                           uint *fast_alter_partition);
+                           char *db,
+                           const char *table_name,
+                           const char *path,
+                           TABLE **fast_alter_table);
 char *generate_partition_syntax(partition_info *part_info,
                                 uint *buf_length, bool use_sql_alloc,
                                 bool show_partition_options,
diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc
index 05ff973cba0..f88282487d9 100644
--- a/sql/sql_plugin.cc
+++ b/sql/sql_plugin.cc
@@ -523,7 +523,7 @@ static st_plugin_dl *plugin_dl_add(const LEX_STRING *dl, int report)
       /* no op */;
 
     cur= (struct st_mysql_plugin*)
-          my_malloc(i*sizeof(struct st_mysql_plugin), MYF(MY_ZEROFILL|MY_WME));
+      my_malloc((i+1)*sizeof(struct st_mysql_plugin), MYF(MY_ZEROFILL|MY_WME));
     if (!cur)
     {
       free_plugin_mem(&plugin_dl);
diff --git a/sql/sql_reload.cc b/sql/sql_reload.cc
index bf38af78536..35f27408247 100644
--- a/sql/sql_reload.cc
+++ b/sql/sql_reload.cc
@@ -328,7 +328,6 @@ bool reload_acl_and_cache(THD *thd, unsigned long options,
   -------------------------------------
   - you can't flush WITH READ LOCK a non-existent table
   - you can't flush WITH READ LOCK under LOCK TABLES
-  - currently incompatible with the GRL (@todo: fix)
 
   Effect on views and temporary tables.
   ------------------------------------
@@ -338,6 +337,13 @@ bool reload_acl_and_cache(THD *thd, unsigned long options,
   if there is a base table, it's used, otherwise ER_NO_SUCH_TABLE
   is returned.
 
+  Handling of MERGE tables
+  ------------------------
+  For MERGE table this statement will open and lock child tables
+  for read (it is impossible to lock parent table without it).
+  Child tables won't be flushed unless they are explicitly present
+  in the statement's table list.
+
   Implicit commit
   ---------------
   This statement causes an implicit commit before and
@@ -354,7 +360,6 @@ bool flush_tables_with_read_lock(THD *thd, TABLE_LIST *all_tables)
 {
   Lock_tables_prelocking_strategy lock_tables_prelocking_strategy;
   TABLE_LIST *table_list;
-  MDL_request_list mdl_requests;
 
   /*
     This is called from SQLCOM_FLUSH, the transaction has
@@ -368,17 +373,13 @@ bool flush_tables_with_read_lock(THD *thd, TABLE_LIST *all_tables)
   }
 
   /*
-    Acquire SNW locks on tables to be flushed. We can't use
-    lock_table_names() here as this call will also acquire global IX
-    and database-scope IX locks on the tables, and this will make
+    Acquire SNW locks on tables to be flushed. Don't acquire global
+    IX and database-scope IX locks on the tables as this will make
     this statement incompatible with FLUSH TABLES WITH READ LOCK.
   */
-  for (table_list= all_tables; table_list;
-       table_list= table_list->next_global)
-    mdl_requests.push_front(&table_list->mdl_request);
-
-  if (thd->mdl_context.acquire_locks(&mdl_requests,
-                                     thd->variables.lock_wait_timeout))
+  if (lock_table_names(thd, all_tables, NULL,
+                       thd->variables.lock_wait_timeout,
+                       MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK))
     goto error;
 
   DEBUG_SYNC(thd,"flush_tables_with_read_lock_after_acquire_locks");
@@ -390,21 +391,24 @@ bool flush_tables_with_read_lock(THD *thd, TABLE_LIST *all_tables)
     tdc_remove_table(thd, TDC_RT_REMOVE_UNUSED,
                      table_list->db,
                      table_list->table_name, FALSE);
-
-    /* Skip views and temporary tables. */
-    table_list->required_type= FRMTYPE_TABLE; /* Don't try to flush views. */
-    table_list->open_type= OT_BASE_ONLY;      /* Ignore temporary tables. */
+    /* Reset ticket to satisfy asserts in open_tables(). */
+    table_list->mdl_request.ticket= NULL;
   }
 
   /*
     Before opening and locking tables the below call also waits
     for old shares to go away, so the fact that we don't pass
     MYSQL_LOCK_IGNORE_FLUSH flag to it is important.
+    Also we don't pass MYSQL_OPEN_HAS_MDL_LOCK flag as we want
+    to open underlying tables if merge table is flushed.
+    For underlying tables of the merge the below call has to
+    acquire SNW locks to ensure that they can be locked for
+    read without further waiting.
   */
-  if  (open_and_lock_tables(thd, all_tables, FALSE,
-                            MYSQL_OPEN_HAS_MDL_LOCK,
-                            &lock_tables_prelocking_strategy) ||
-       thd->locked_tables_list.init_locked_tables(thd))
+  if (open_and_lock_tables(thd, all_tables, FALSE,
+                           MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK,
+                           &lock_tables_prelocking_strategy) ||
+      thd->locked_tables_list.init_locked_tables(thd))
   {
     goto error;
   }
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 97f8e46d052..ac15239b040 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -285,6 +285,7 @@ do_rename(THD *thd, TABLE_LIST *ren_table, char *new_db, char *new_table_name,
         {
           if ((rc= Table_triggers_list::change_table_name(thd, ren_table->db,
                                                           old_alias,
+                                                          ren_table->table_name,
                                                           new_db,
                                                           new_alias)))
           {
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index ab6c6e738b2..7bca41f1265 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -448,6 +448,8 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
   const char *errmsg = "Unknown error";
   NET* net = &thd->net;
   mysql_mutex_t *log_lock;
+  mysql_cond_t *log_cond;
+
   bool binlog_can_be_corrupted= FALSE;
 #ifndef DBUG_OFF
   int left_events = max_binlog_dump_events;
@@ -591,7 +593,8 @@ impossible position";
     mysql_bin_log, and it's already inited, and it will be destroyed
     only at shutdown).
   */
-  log_lock = mysql_bin_log.get_log_lock();
+  log_lock= mysql_bin_log.get_log_lock();
+  log_cond= mysql_bin_log.get_log_cond();
   if (pos > BIN_LOG_HEADER_SIZE)
   {
     /* reset transmit packet for the event read from binary log
@@ -826,6 +829,7 @@ impossible position";
 #ifndef DBUG_OFF
           ulong hb_info_counter= 0;
 #endif
+          const char* old_msg= thd->proc_info;
           signal_cnt= mysql_bin_log.signal_cnt;
           do 
           {
@@ -834,6 +838,9 @@ impossible position";
               DBUG_ASSERT(heartbeat_ts && heartbeat_period != 0);
               set_timespec_nsec(*heartbeat_ts, heartbeat_period);
             }
+            thd->enter_cond(log_cond, log_lock,
+                            "Master has sent all binlog to slave; "
+                            "waiting for binlog to be updated");
             ret= mysql_bin_log.wait_for_update_bin_log(thd, heartbeat_ts);
             DBUG_ASSERT(ret == 0 || (heartbeat_period != 0 && coord != NULL));
             if (ret == ETIMEDOUT || ret == ETIME)
@@ -849,12 +856,15 @@ impossible position";
 #endif
               /* reset transmit packet for the heartbeat event */
               if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
+              {
+                thd->exit_cond(old_msg);
                 goto err;
+              }
               if (send_heartbeat_event(net, packet, coord))
               {
                 errmsg = "Failed on my_net_write()";
                 my_errno= ER_UNKNOWN_ERROR;
-                mysql_mutex_unlock(log_lock);
+                thd->exit_cond(old_msg);
                 goto err;
               }
             }
@@ -863,7 +873,7 @@ impossible position";
               DBUG_PRINT("wait",("binary log received update or a broadcast signal caught"));
             }
           } while (signal_cnt == mysql_bin_log.signal_cnt && !thd->killed);
-          mysql_mutex_unlock(log_lock);
+          thd->exit_cond(old_msg);
         }
         break;
             
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 29a78cb31dc..2a2fe3eb36f 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1523,6 +1523,15 @@ JOIN::optimize()
     if (order)
     {
       /*
+        Do we need a temporary table due to the ORDER BY not being equal to
+        the GROUP BY? The call to test_if_skip_sort_order above tests for the
+        GROUP BY clause only and hence is not valid in this case. So the
+        estimated number of rows to be read from the first table is not valid.
+        We clear it here so that it doesn't show up in EXPLAIN.
+       */
+      if (need_tmp && (select_options & SELECT_DESCRIBE) != 0)
+        join_tab[const_tables].limit= 0;
+      /*
         Force using of tmp table if sorting by a SP or UDF function due to
         their expensive and probably non-deterministic nature.
       */
@@ -2402,13 +2411,8 @@ JOIN::destroy()
 
   cleanup(1);
  /* Cleanup items referencing temporary table columns */
-  if (!tmp_all_fields3.is_empty())
-  {
-    List_iterator_fast<Item> it(tmp_all_fields3);
-    Item *item;
-    while ((item= it++))
-      item->cleanup();
-  }
+  cleanup_item_list(tmp_all_fields1);
+  cleanup_item_list(tmp_all_fields3);
   if (exec_tmp_table1)
     free_tmp_table(thd, exec_tmp_table1);
   if (exec_tmp_table2)
@@ -2419,6 +2423,19 @@ JOIN::destroy()
   DBUG_RETURN(error);
 }
 
+
+void JOIN::cleanup_item_list(List<Item> &items) const
+{
+  if (!items.is_empty())
+  {
+    List_iterator_fast<Item> it(items);
+    Item *item;
+    while ((item= it++))
+      item->cleanup();
+  }
+}
+
+
 /**
   An entry point to single-unit select (a select without UNION).
 
@@ -9008,10 +9025,10 @@ simplify_joins(JOIN *join, List<TABLE_LIST> *join_list, COND *conds, bool top)
     
   /* Flatten nested joins that can be flattened. */
   TABLE_LIST *right_neighbor= NULL;
-  bool fix_name_res= FALSE;
   li.rewind();
   while ((table= li++))
   {
+    bool fix_name_res= FALSE;
     nested_join= table->nested_join;
     if (nested_join && !table->on_expr)
     {
@@ -16696,7 +16713,15 @@ static void select_describe(JOIN *join, bool need_tmp_table, bool need_order,
         if (tab->select && tab->select->quick)
           examined_rows= tab->select->quick->records;
         else if (tab->type == JT_NEXT || tab->type == JT_ALL)
-          examined_rows= tab->limit ? tab->limit : tab->table->file->records();
+        {
+          if (tab->limit)
+            examined_rows= tab->limit;
+          else
+          {
+            tab->table->file->info(HA_STATUS_VARIABLE);
+            examined_rows= tab->table->file->stats.records;
+          }
+        }
         else
           examined_rows=(ha_rows)join->best_positions[i].records_read; 
  
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 77fff4ee24c..93885e23f76 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -583,6 +583,7 @@ private:
   */
   bool implicit_grouping; 
   bool make_simple_join(JOIN *join, TABLE *tmp_table);
+  void cleanup_item_list(List<Item> &items) const;
 };
 
 
diff --git a/sql/sql_string.h b/sql/sql_string.h
index 0ce67108423..d21b5353b76 100644
--- a/sql/sql_string.h
+++ b/sql/sql_string.h
@@ -104,7 +104,7 @@ public:
   inline uint32 alloced_length() const { return Alloced_length;}
   inline char& operator [] (uint32 i) const { return Ptr[i]; }
   inline void length(uint32 len) { str_length=len ; }
-  inline bool is_empty() { return (str_length == 0); }
+  inline bool is_empty() const { return (str_length == 0); }
   inline void mark_as_const() { Alloced_length= 0;}
   inline const char *ptr() const { return Ptr; }
   inline char *c_ptr()
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 3f0a0326c84..1c6ca4a48d9 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -22,7 +22,7 @@
 #include "sql_rename.h" // do_rename
 #include "sql_parse.h"                        // test_if_data_home_dir
 #include "sql_cache.h"                          // query_cache_*
-#include "sql_base.h"   // open_temporary_table, lock_table_names
+#include "sql_base.h"   // open_table_uncached, lock_table_names
 #include "lock.h"       // wait_if_global_read_lock
                         // start_waiting_global_read_lock,
                         // mysql_unlock_tables
@@ -1725,8 +1725,6 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
                                                   CHF_DELETE_FLAG, NULL) ||
         deactivate_ddl_log_entry(part_info->frm_log_entry->entry_pos) ||
         (sync_ddl_log(), FALSE) ||
-#endif
-#ifdef WITH_PARTITION_STORAGE_ENGINE
         mysql_file_rename(key_file_frm,
                           shadow_frm_name, frm_name, MYF(MY_WME)) ||
         lpt->table->file->ha_create_handler_files(path, shadow_path,
@@ -2008,7 +2006,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     {
       for (table= tables; table; table= table->next_local)
         if (table->open_type != OT_BASE_ONLY &&
-	    find_temporary_table(thd, table->db, table->table_name))
+	    find_temporary_table(thd, table))
         {
           /*
             A temporary table.
@@ -3536,6 +3534,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
       key_part_info->length=(uint16) length;
       /* Use packed keys for long strings on the first column */
       if (!((*db_options) & HA_OPTION_NO_PACK_KEYS) &&
+          !((create_info->table_options & HA_OPTION_NO_PACK_KEYS)) &&
 	  (length >= KEY_DEFAULT_PACK_LENGTH &&
 	   (sql_field->sql_type == MYSQL_TYPE_STRING ||
 	    sql_field->sql_type == MYSQL_TYPE_VARCHAR ||
@@ -4224,9 +4223,14 @@ bool mysql_create_table_no_lock(THD *thd,
 
   if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
   {
-    TABLE *table= NULL;
-    /* Open table and put in temporary table list */
-    if (!(table= open_temporary_table(thd, path, db, table_name, 1)))
+    /*
+      Open a table (skipping table cache) and add it into
+      THD::temporary_tables list.
+    */
+
+    TABLE *table= open_table_uncached(thd, path, db, table_name, TRUE);
+
+    if (!table)
     {
       (void) rm_temporary_table(create_info->db_type, path);
       goto err;
@@ -5599,7 +5603,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   handlerton *old_db_type, *new_db_type, *save_old_db_type;
   enum_alter_table_change_level need_copy_table= ALTER_TABLE_METADATA_ONLY;
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-  uint fast_alter_partition= 0;
+  TABLE *table_for_fast_alter_partition= NULL;
   bool partition_changed= FALSE;
 #endif
   bool need_lock_for_indexes= TRUE;
@@ -5913,7 +5917,8 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
         *fn_ext(new_name)=0;
         if (mysql_rename_table(old_db_type,db,table_name,new_db,new_alias, 0))
           error= -1;
-        else if (Table_triggers_list::change_table_name(thd, db, table_name,
+        else if (Table_triggers_list::change_table_name(thd, db,
+                                                        alias, table_name,
                                                         new_db, new_alias))
         {
           (void) mysql_rename_table(old_db_type, new_db, new_alias, db,
@@ -5968,7 +5973,9 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   if (prep_alter_part_table(thd, table, alter_info, create_info, old_db_type,
-                            &partition_changed, &fast_alter_partition))
+                            &partition_changed,
+                            db, table_name, path,
+                            &table_for_fast_alter_partition))
     goto err;
 #endif
   /*
@@ -6201,12 +6208,12 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     create_info->frm_only= 1;
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-  if (fast_alter_partition)
+  if (table_for_fast_alter_partition)
   {
     DBUG_RETURN(fast_alter_partition_table(thd, table, alter_info,
                                            create_info, table_list,
                                            db, table_name,
-                                           fast_alter_partition));
+                                           table_for_fast_alter_partition));
   }
 #endif
 
@@ -6301,8 +6308,8 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       /* table is a normal table: Create temporary table in same directory */
       build_table_filename(path, sizeof(path) - 1, new_db, tmp_name, "",
                            FN_IS_TMP);
-      /* Open our intermediate table */
-      new_table= open_temporary_table(thd, path, new_db, tmp_name, 1);
+      /* Open our intermediate table. */
+      new_table= open_table_uncached(thd, path, new_db, tmp_name, TRUE);
     }
     if (!new_table)
       goto err_new_table_cleanup;
@@ -6550,7 +6557,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
            (need_copy_table != ALTER_TABLE_METADATA_ONLY ||
             mysql_rename_table(save_old_db_type, db, table_name, new_db,
                                new_alias, NO_FRM_RENAME)) &&
-           Table_triggers_list::change_table_name(thd, db, table_name,
+           Table_triggers_list::change_table_name(thd, db, alias, table_name,
                                                   new_db, new_alias)))
   {
     /* Try to get everything back. */
@@ -6642,7 +6649,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     char path[FN_REFLEN];
     TABLE *t_table;
     build_table_filename(path + 1, sizeof(path) - 1, new_db, table_name, "", 0);
-    t_table= open_temporary_table(thd, path, new_db, tmp_name, 0);
+    t_table= open_table_uncached(thd, path, new_db, tmp_name, FALSE);
     if (t_table)
     {
       intern_close_table(t_table);
@@ -6686,6 +6693,11 @@ err_new_table_cleanup:
                           create_info->frm_only ? FN_IS_TMP | FRM_ONLY : FN_IS_TMP);
 
 err:
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  /* If prep_alter_part_table created an intermediate table, destroy it. */
+  if (table_for_fast_alter_partition)
+    close_temporary(table_for_fast_alter_partition, 1, 0);
+#endif /* WITH_PARTITION_STORAGE_ENGINE */
   /*
     No default value was provided for a DATE/DATETIME field, the
     current sql_mode doesn't allow the '0000-00-00' value and
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index b81461e8371..a9b52eee9fc 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -458,7 +458,7 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
   DBUG_ASSERT(tables->next_global == 0);
 
   /* We do not allow creation of triggers on temporary tables. */
-  if (create && find_temporary_table(thd, tables->db, tables->table_name))
+  if (create && find_temporary_table(thd, tables))
   {
     my_error(ER_TRG_ON_VIEW_OR_TEMP_TABLE, MYF(0), tables->alias);
     goto end;
@@ -1874,6 +1874,7 @@ Table_triggers_list::change_table_name_in_trignames(const char *old_db_name,
 
   @param[in,out] thd Thread context
   @param[in] db Old database of subject table
+  @param[in] old_alias Old alias of subject table
   @param[in] old_table Old name of subject table
   @param[in] new_db New database for subject table
   @param[in] new_table New name of subject table
@@ -1890,6 +1891,7 @@ Table_triggers_list::change_table_name_in_trignames(const char *old_db_name,
 */
 
 bool Table_triggers_list::change_table_name(THD *thd, const char *db,
+                                            const char *old_alias,
                                             const char *old_table,
                                             const char *new_db,
                                             const char *new_table)
@@ -1911,7 +1913,7 @@ bool Table_triggers_list::change_table_name(THD *thd, const char *db,
                                              MDL_EXCLUSIVE));
 
   DBUG_ASSERT(my_strcasecmp(table_alias_charset, db, new_db) ||
-              my_strcasecmp(table_alias_charset, old_table, new_table));
+              my_strcasecmp(table_alias_charset, old_alias, new_table));
 
   if (Table_triggers_list::check_n_load(thd, db, old_table, &table, TRUE))
   {
@@ -1920,7 +1922,7 @@ bool Table_triggers_list::change_table_name(THD *thd, const char *db,
   }
   if (table.triggers)
   {
-    LEX_STRING old_table_name= { (char *) old_table, strlen(old_table) };
+    LEX_STRING old_table_name= { (char *) old_alias, strlen(old_alias) };
     LEX_STRING new_table_name= { (char *) new_table, strlen(new_table) };
     /*
       Since triggers should be in the same schema as their subject tables
@@ -2006,6 +2008,7 @@ bool Table_triggers_list::process_triggers(THD *thd,
   bool err_status;
   Sub_statement_state statement_state;
   sp_head *sp_trigger= bodies[event][time_type];
+  SELECT_LEX *save_current_select;
 
   if (sp_trigger == NULL)
     return FALSE;
@@ -2029,11 +2032,19 @@ bool Table_triggers_list::process_triggers(THD *thd,
 
   thd->reset_sub_statement_state(&statement_state, SUB_STMT_TRIGGER);
 
+  /*
+    Reset current_select before call execute_trigger() and
+    restore it after return from one. This way error is set
+    in case of failure during trigger execution.
+  */
+  save_current_select= thd->lex->current_select;
+  thd->lex->current_select= NULL;
   err_status=
     sp_trigger->execute_trigger(thd,
                                 &trigger_table->s->db,
                                 &trigger_table->s->table_name,
                                 &subject_table_grants[event][time_type]);
+  thd->lex->current_select= save_current_select;
 
   thd->restore_sub_statement_state(&statement_state);
 
diff --git a/sql/sql_trigger.h b/sql/sql_trigger.h
index 28bf0a60877..dd954957f07 100644
--- a/sql/sql_trigger.h
+++ b/sql/sql_trigger.h
@@ -157,6 +157,7 @@ public:
                            TABLE *table, bool names_only);
   static bool drop_all_triggers(THD *thd, char *db, char *table_name);
   static bool change_table_name(THD *thd, const char *db,
+                                const char *old_alias,
                                 const char *old_table,
                                 const char *new_db,
                                 const char *new_table);
diff --git a/sql/sql_truncate.cc b/sql/sql_truncate.cc
index a61abdafe3e..c0bc726a188 100644
--- a/sql/sql_truncate.cc
+++ b/sql/sql_truncate.cc
@@ -208,8 +208,8 @@ static bool recreate_temporary_table(THD *thd, TABLE *table)
   ha_create_table(thd, share->normalized_path.str, share->db.str,
                   share->table_name.str, &create_info, 1);
 
-  if (open_temporary_table(thd, share->path.str, share->db.str,
-                           share->table_name.str, 1))
+  if (open_table_uncached(thd, share->path.str, share->db.str,
+                          share->table_name.str, TRUE))
   {
     error= FALSE;
     thd->thread_specific_used= TRUE;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 731e54ef6d8..7c24f18aa6a 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -4815,7 +4815,7 @@ part_value_expr_item:
 
             if (!lex->safe_to_cache_query)
             {
-              my_error(ER_PARTITION_FUNCTION_IS_NOT_ALLOWED, MYF(0));
+              my_parse_error(ER(ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR));
               MYSQL_YYABORT;
             }
             if (part_info->add_column_list_value(YYTHD, part_expr))
@@ -10448,8 +10448,11 @@ insert_lock_option:
         | LOW_PRIORITY  { $$= TL_WRITE_LOW_PRIORITY; }
         | DELAYED_SYM
         {
+          Lex->keyword_delayed_begin_offset= (uint)(YYLIP->get_tok_start() -
+                                                    YYTHD->query());
+          Lex->keyword_delayed_end_offset= Lex->keyword_delayed_begin_offset +
+                                           YYLIP->yyLength() + 1;
           $$= TL_WRITE_DELAYED;
-          Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_DELAYED);
         }
         | HIGH_PRIORITY { $$= TL_WRITE; }
         ;
@@ -10458,8 +10461,11 @@ replace_lock_option:
           opt_low_priority { $$= $1; }
         | DELAYED_SYM
         {
+          Lex->keyword_delayed_begin_offset= (uint)(YYLIP->get_tok_start() -
+                                                    YYTHD->query());
+          Lex->keyword_delayed_end_offset= Lex->keyword_delayed_begin_offset +
+                                           YYLIP->yyLength() + 1;
           $$= TL_WRITE_DELAYED;
-          Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_DELAYED);
         }
         ;
 
@@ -11279,7 +11285,11 @@ opt_with_read_lock:
             TABLE_LIST *tables= Lex->query_tables;
             Lex->type|= REFRESH_READ_LOCK;
             for (; tables; tables= tables->next_global)
+            {
               tables->mdl_request.set_type(MDL_SHARED_NO_WRITE);
+              tables->required_type= FRMTYPE_TABLE; /* Don't try to flush views. */
+              tables->open_type= OT_BASE_ONLY;      /* Ignore temporary tables. */
+            }
           }
         ;
 
@@ -13249,6 +13259,13 @@ handler:
           handler_read_or_scan where_clause opt_limit_clause
           {
             Lex->expr_allows_subselect= TRUE;
+            /* Stored functions are not supported for HANDLER READ. */
+            if (Lex->uses_stored_routines())
+            {
+              my_error(ER_NOT_SUPPORTED_YET, MYF(0),
+                       "stored functions in HANDLER ... READ");
+              MYSQL_YYABORT;
+            }
           }
         ;
 
diff --git a/sql/sys_vars.h b/sql/sys_vars.h
index 3db89b3b4fa..89e2bdfdf60 100644
--- a/sql/sys_vars.h
+++ b/sql/sys_vars.h
@@ -385,7 +385,7 @@ public:
     DBUG_ASSERT(scope() == GLOBAL);
     DBUG_ASSERT(size == sizeof(char *));
   }
-  ~Sys_var_charptr()
+  void cleanup()
   {
     if (flags & ALLOCATED)
       my_free(global_var(char*));
diff --git a/sql/table.cc b/sql/table.cc
index 7d16ad7974e..f55095d6e82 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -426,6 +426,18 @@ void TABLE_SHARE::destroy()
       info_it->flags= 0;
     }
   }
+  if (ha_data_destroy)
+  {
+    ha_data_destroy(ha_data);
+    ha_data_destroy= NULL;
+  }
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  if (ha_part_data_destroy)
+  {
+    ha_part_data_destroy(ha_part_data);
+    ha_part_data_destroy= NULL;
+  }
+#endif /* WITH_PARTITION_STORAGE_ENGINE */
   /*
     Make a copy since the share is allocated in its own root,
     and free_root() updates its argument after freeing the memory.
@@ -1704,11 +1716,17 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   delete handler_file;
   my_hash_free(&share->name_hash);
   if (share->ha_data_destroy)
+  {
     share->ha_data_destroy(share->ha_data);
+    share->ha_data_destroy= NULL;
+  }
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   if (share->ha_part_data_destroy)
+  {
     share->ha_part_data_destroy(share->ha_part_data);
-#endif
+    share->ha_data_destroy= NULL;
+  }
+#endif /* WITH_PARTITION_STORAGE_ENGINE */
 
   open_table_error(share, error, share->open_errno, errarg);
   DBUG_RETURN(error);
@@ -3085,30 +3103,7 @@ bool TABLE_SHARE::visit_subgraph(Wait_for_flush *wait_for_flush,
     holding a write-lock on MDL_lock::m_rwlock.
   */
   if (gvisitor->m_lock_open_count++ == 0)
-  {
-    /*
-      To circumvent bug #56405 "Deadlock in the MDL deadlock detector"
-      we don't try to lock LOCK_open mutex if some thread doing
-      deadlock detection already owns it and current search depth is
-      greater than 0. Instead we report a deadlock.
-
-      TODO/FIXME: The proper fix for this bug is to use rwlocks for
-                  protection of table shares/instead of LOCK_open.
-                  Unfortunately it requires more effort/has significant
-                  performance effect.
-    */
-    mysql_mutex_lock(&LOCK_dd_owns_lock_open);
-    if (gvisitor->m_current_search_depth > 0 && dd_owns_lock_open > 0)
-    {
-      mysql_mutex_unlock(&LOCK_dd_owns_lock_open);
-      --gvisitor->m_lock_open_count;
-      gvisitor->abort_traversal(src_ctx);
-      return TRUE;
-    }
-    ++dd_owns_lock_open;
-    mysql_mutex_unlock(&LOCK_dd_owns_lock_open);
     mysql_mutex_lock(&LOCK_open);
-  }
 
   I_P_List_iterator <TABLE, TABLE_share> tables_it(used_tables);
 
@@ -3123,12 +3118,8 @@ bool TABLE_SHARE::visit_subgraph(Wait_for_flush *wait_for_flush,
     goto end;
   }
 
-  ++gvisitor->m_current_search_depth;
   if (gvisitor->enter_node(src_ctx))
-  {
-    --gvisitor->m_current_search_depth;
     goto end;
-  }
 
   while ((table= tables_it++))
   {
@@ -3151,16 +3142,10 @@ bool TABLE_SHARE::visit_subgraph(Wait_for_flush *wait_for_flush,
 
 end_leave_node:
   gvisitor->leave_node(src_ctx);
-  --gvisitor->m_current_search_depth;
 
 end:
   if (gvisitor->m_lock_open_count-- == 1)
-  {
     mysql_mutex_unlock(&LOCK_open);
-    mysql_mutex_lock(&LOCK_dd_owns_lock_open);
-    --dd_owns_lock_open;
-    mysql_mutex_unlock(&LOCK_dd_owns_lock_open);
-  }
 
   return result;
 }
@@ -3256,6 +3241,65 @@ bool TABLE_SHARE::wait_for_old_version(THD *thd, struct timespec *abstime,
 }
 
 
+/**
+  Initialize TABLE instance (newly created, or coming either from table
+  cache or THD::temporary_tables list) and prepare it for further use
+  during statement execution. Set the 'alias' attribute from the specified
+  TABLE_LIST element. Remember the TABLE_LIST element in the
+  TABLE::pos_in_table_list member.
+
+  @param thd  Thread context.
+  @param tl   TABLE_LIST element.
+*/
+
+void TABLE::init(THD *thd, TABLE_LIST *tl)
+{
+  DBUG_ASSERT(s->ref_count > 0 || s->tmp_table != NO_TMP_TABLE);
+
+  if (thd->lex->need_correct_ident())
+    alias_name_used= my_strcasecmp(table_alias_charset,
+                                   s->table_name.str,
+                                   tl->alias);
+  /* Fix alias if table name changes. */
+  if (strcmp(alias, tl->alias))
+  {
+    uint length= (uint) strlen(tl->alias)+1;
+    alias= (char*) my_realloc((char*) alias, length, MYF(MY_WME));
+    memcpy((char*) alias, tl->alias, length);
+  }
+
+  tablenr= thd->current_tablenr++;
+  used_fields= 0;
+  const_table= 0;
+  null_row= 0;
+  maybe_null= 0;
+  force_index= 0;
+  force_index_order= 0;
+  force_index_group= 0;
+  status= STATUS_NO_RECORD;
+  insert_values= 0;
+  fulltext_searched= 0;
+  file->ft_handler= 0;
+  reginfo.impossible_range= 0;
+
+  /* Catch wrong handling of the auto_increment_field_not_null. */
+  DBUG_ASSERT(!auto_increment_field_not_null);
+  auto_increment_field_not_null= FALSE;
+
+  if (timestamp_field)
+    timestamp_field_type= timestamp_field->get_auto_set_type();
+
+  pos_in_table_list= tl;
+
+  clear_column_bitmaps();
+
+  DBUG_ASSERT(key_read == 0);
+
+  /* Tables may be reused in a sub statement. */
+  DBUG_ASSERT(!file->extra(HA_EXTRA_IS_ATTACHED_CHILDREN));
+}
+
+
 /*
   Create Item_field for each column in the table.
 
diff --git a/sql/table.h b/sql/table.h
index 45b9aa3e699..6723293c1ec 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -83,18 +83,6 @@ enum enum_table_ref_type
 };
 
 
-/**
-  Opening modes for open_temporary_table and open_table_from_share
-*/
-
-enum open_table_mode
-{
-  OTM_OPEN= 0,
-  OTM_CREATE= 1,
-  OTM_ALTER= 2
-};
-
-
 /*************************************************************************/
 
 /**
@@ -1096,6 +1084,7 @@ public:
 #endif
   MDL_ticket *mdl_ticket;
 
+  void init(THD *thd, TABLE_LIST *tl);
   bool fill_item_list(List<Item> *item_list) const;
   void reset_item_list(List<Item> *item_list) const;
   void clear_column_bitmaps(void);
diff --git a/sql/transaction.cc b/sql/transaction.cc
index a28fba8805d..d3e3ba142b9 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -565,7 +565,8 @@ bool trans_xa_end(THD *thd)
   else if (!xa_trans_rolled_back(&thd->transaction.xid_state))
     thd->transaction.xid_state.xa_state= XA_IDLE;
 
-  DBUG_RETURN(thd->transaction.xid_state.xa_state != XA_IDLE);
+  DBUG_RETURN(thd->is_error() ||
+              thd->transaction.xid_state.xa_state != XA_IDLE);
 }
 
 
@@ -596,7 +597,8 @@ bool trans_xa_prepare(THD *thd)
   else
     thd->transaction.xid_state.xa_state= XA_PREPARED;
 
-  DBUG_RETURN(thd->transaction.xid_state.xa_state != XA_PREPARED);
+  DBUG_RETURN(thd->is_error() ||
+              thd->transaction.xid_state.xa_state != XA_PREPARED);
 }
 
 
diff --git a/sql/tztime.cc b/sql/tztime.cc
index 43d43123158..a2f319d4307 100644
--- a/sql/tztime.cc
+++ b/sql/tztime.cc
@@ -2300,7 +2300,7 @@ my_tz_find(THD *thd, const String *name)
   DBUG_PRINT("enter", ("time zone name='%s'",
                        name ? ((String *)name)->c_ptr_safe() : "NULL"));
 
-  if (!name)
+  if (!name || name->is_empty())
     DBUG_RETURN(0);
 
   mysql_mutex_lock(&tz_LOCK);