Merge stella.local:/home2/mydev/mysql-5.1-amain

into  stella.local:/home2/mydev/mysql-5.1-axmrg


mysql-test/r/ctype_big5.result:
  Auto merged
mysql-test/r/ctype_euckr.result:
  Auto merged
mysql-test/r/ctype_gb2312.result:
  Auto merged
mysql-test/r/ctype_gbk.result:
  Auto merged
mysql-test/r/ctype_uca.result:
  Auto merged
mysql-test/r/ctype_ucs.result:
  Auto merged
mysql-test/suite/binlog/r/binlog_row_ctype_cp932.result:
  Auto merged
mysql-test/suite/binlog/r/binlog_stm_ctype_cp932.result:
  Auto merged
mysql-test/t/ctype_ucs.test:
  Auto merged
sql/item_func.cc:
  Auto merged
sql/item_func.h:
  Auto merged
sql/log.cc:
  Auto merged
sql/log_event.cc:
  Auto merged
sql/rpl_rli.cc:
  Auto merged
sql/slave.cc:
  Auto merged
sql/sql_yacc.yy:
  Auto merged
mysql-test/lib/mtr_report.pl:
  SCCS merged

27 files changed, 746 insertions, 350 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 4b07da2c1c1..32bf5855d8b 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -7636,6 +7636,17 @@ int Field_blob::store(const char *from,uint length,CHARSET_INFO *cs)
   if (value.alloc(new_length))
     goto oom_error;
 
+
+  if (f_is_hex_escape(flags))
+  {
+    copy_length= my_copy_with_hex_escaping(field_charset,
+                                           (char*) value.ptr(), new_length,
+                                            from, length);
+    Field_blob::store_length(copy_length);
+    tmp= value.ptr();
+    bmove(ptr + packlength, (uchar*) &tmp, sizeof(char*));
+    return 0;
+  }
   /*
     "length" is OK as "nchars" argument to well_formed_copy_nchars as this
     is never used to limit the length of the data. The cut of long data
@@ -8779,6 +8790,23 @@ Field_bit::Field_bit(uchar *ptr_arg, uint32 len_arg, uchar *null_ptr_arg,
 }
 
 
+void Field_bit::hash(ulong *nr, ulong *nr2)
+{
+  if (is_null())
+  {
+    *nr^= (*nr << 1) | 1;
+  }
+  else
+  {
+    CHARSET_INFO *cs= &my_charset_bin;
+    longlong value= Field_bit::val_int();
+    uchar tmp[8];
+    mi_int8store(tmp,value);
+    cs->coll->hash_sort(cs, tmp, 8, nr, nr2);
+  }
+}
+
+
 size_t
 Field_bit::do_last_null_byte() const
 {
diff --git a/sql/field.h b/sql/field.h
index 9855ec563d5..5457b7284e9 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -1906,6 +1906,7 @@ public:
     Field::move_field_offset(ptr_diff);
     bit_ptr= ADD_TO_PTR(bit_ptr, ptr_diff, uchar*);
   }
+  void hash(ulong *nr, ulong *nr2);
 
 private:
   virtual size_t do_last_null_byte() const;
@@ -2070,6 +2071,7 @@ int set_field_to_null_with_conversions(Field *field, bool no_conversions);
 #define FIELDFLAG_NO_DEFAULT		16384   /* sql */
 #define FIELDFLAG_SUM			((uint) 32768)// predit: +#fieldflag
 #define FIELDFLAG_MAYBE_NULL		((uint) 32768)// sql
+#define FIELDFLAG_HEX_ESCAPE		((uint) 0x10000)
 #define FIELDFLAG_PACK_SHIFT		3
 #define FIELDFLAG_DEC_SHIFT		8
 #define FIELDFLAG_MAX_DEC		31
@@ -2095,3 +2097,4 @@ int set_field_to_null_with_conversions(Field *field, bool no_conversions);
 #define f_maybe_null(x)		(x & FIELDFLAG_MAYBE_NULL)
 #define f_no_default(x)		(x & FIELDFLAG_NO_DEFAULT)
 #define f_bit_as_char(x)        ((x) & FIELDFLAG_TREAT_BIT_AS_CHAR)
+#define f_is_hex_escape(x)      ((x) & FIELDFLAG_HEX_ESCAPE)
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 51070a525c5..ae7e2215517 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -1028,6 +1028,7 @@ int ha_partition::repair_partitions(THD *thd)
     0                         Success
 */
 
+#ifdef WL4176_IS_DONE
 static int handle_opt_part(THD *thd, HA_CHECK_OPT *check_opt,
                            handler *file, uint flag)
 {
@@ -1035,6 +1036,12 @@ static int handle_opt_part(THD *thd, HA_CHECK_OPT *check_opt,
   DBUG_ENTER("handle_opt_part");
   DBUG_PRINT("enter", ("flag = %u", flag));
 
+  /*
+    TODO:
+    Rewrite the code for ANALYZE/CHECK/OPTIMIZE/REPAIR PARTITION WL4176
+  */
+  DBUG_RETURN(HA_ADMIN_NOT_IMPLEMENTED);
+
   if (flag == OPTIMIZE_PARTS)
     error= file->ha_optimize(thd, check_opt);
   else if (flag == ANALYZE_PARTS)
@@ -1052,6 +1059,7 @@ static int handle_opt_part(THD *thd, HA_CHECK_OPT *check_opt,
     error= 0;
   DBUG_RETURN(error);
 }
+#endif
 
 
 /*
@@ -1072,14 +1080,22 @@ static int handle_opt_part(THD *thd, HA_CHECK_OPT *check_opt,
 int ha_partition::handle_opt_partitions(THD *thd, HA_CHECK_OPT *check_opt,
                                         uint flag, bool all_parts)
 {
+#ifdef WL4176_IS_DONE
   List_iterator<partition_element> part_it(m_part_info->partitions);
   uint no_parts= m_part_info->no_parts;
   uint no_subparts= m_part_info->no_subparts;
   uint i= 0;
   int error;
+#endif
   DBUG_ENTER("ha_partition::handle_opt_partitions");
   DBUG_PRINT("enter", ("all_parts %u, flag= %u", all_parts, flag));
 
+  /*
+    TODO:
+    Rewrite the code for ANALYZE/CHECK/OPTIMIZE/REPAIR PARTITION WL4176
+  */
+  DBUG_RETURN(HA_ADMIN_NOT_IMPLEMENTED);
+#ifdef WL4176_IS_DONE
   do
   {
     partition_element *part_elem= part_it++;
@@ -1110,6 +1126,7 @@ int ha_partition::handle_opt_partitions(THD *thd, HA_CHECK_OPT *check_opt,
     }
   } while (++i < no_parts);
   DBUG_RETURN(FALSE);
+#endif
 }
 
 /*
@@ -1629,6 +1646,15 @@ void ha_partition::change_table_ptr(TABLE *table_arg, TABLE_SHARE *share)
   {
     (*file_array)->change_table_ptr(table_arg, share);
   } while (*(++file_array));
+  if (m_added_file && m_added_file[0])
+  {
+    /* if in middle of a drop/rename etc */
+    file_array= m_added_file;
+    do
+    {
+      (*file_array)->change_table_ptr(table_arg, share);
+    } while (*(++file_array));
+  }
 }
 
 /*
diff --git a/sql/item.cc b/sql/item.cc
index cfb8f336a91..0a1db55761f 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1296,6 +1296,7 @@ bool Item_name_const::fix_fields(THD *thd, Item **ref)
     return TRUE;
   }
   set_name(item_name->ptr(), (uint) item_name->length(), system_charset_info);
+  collation.set(value_item->collation.collation, DERIVATION_IMPLICIT);
   max_length= value_item->max_length;
   decimals= value_item->decimals;
   fixed= 1;
diff --git a/sql/item_create.cc b/sql/item_create.cc
index 40578bef5f8..49cc33b95a7 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -4354,12 +4354,12 @@ Create_func_space::create(THD *thd, Item *arg1)
   if (cs->mbminlen > 1)
   {
     uint dummy_errors;
-    sp= new (thd->mem_root) Item_string("", 0, cs);
+    sp= new (thd->mem_root) Item_string("", 0, cs, DERIVATION_COERCIBLE, MY_REPERTOIRE_ASCII);
     sp->str_value.copy(" ", 1, &my_charset_latin1, cs, &dummy_errors);
   }
   else
   {
-    sp= new (thd->mem_root) Item_string(" ", 1, cs);
+    sp= new (thd->mem_root) Item_string(" ", 1, cs, DERIVATION_COERCIBLE, MY_REPERTOIRE_ASCII);
   }
 
   return new (thd->mem_root) Item_func_repeat(sp, arg1);
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 02b15461c22..c938bf5ebe1 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -3715,11 +3715,24 @@ longlong Item_func_sleep::val_int()
   DBUG_ASSERT(fixed == 1);
 
   double time= args[0]->val_real();
+  /*
+    On 64-bit OSX pthread_cond_timedwait() waits forever
+    if passed abstime time has already been exceeded by 
+    the system time.
+    When given a very short timeout (< 10 mcs) just return 
+    immediately.
+    We assume that the lines between this test and the call 
+    to pthread_cond_timedwait() will be executed in less than 0.00001 sec.
+  */
+  if (time < 0.00001)
+    return 0;
+    
   set_timespec_nsec(abstime, (ulonglong)(time * ULL(1000000000)));
 
   pthread_cond_init(&cond, NULL);
   pthread_mutex_lock(&LOCK_user_locks);
 
+  thd_proc_info(thd, "User sleep");
   thd->mysys_var->current_mutex= &LOCK_user_locks;
   thd->mysys_var->current_cond=  &cond;
 
@@ -3731,6 +3744,7 @@ longlong Item_func_sleep::val_int()
       break;
     error= 0;
   }
+  thd_proc_info(thd, 0);
   pthread_mutex_unlock(&LOCK_user_locks);
   pthread_mutex_lock(&thd->mysys_var->mutex);
   thd->mysys_var->current_mutex= 0;
diff --git a/sql/item_func.h b/sql/item_func.h
index b6ee0a3de5e..02631d7643d 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -309,7 +309,6 @@ class Item_num_op :public Item_func_numhybrid
 
   void find_num_type();
   String *str_op(String *str) { DBUG_ASSERT(0); return 0; }
-  bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
 };
 
 
@@ -395,6 +394,7 @@ class Item_func_additive_op :public Item_num_op
 public:
   Item_func_additive_op(Item *a,Item *b) :Item_num_op(a,b) {}
   void result_precision();
+  bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
 };
 
 
@@ -429,6 +429,7 @@ public:
   double real_op();
   my_decimal *decimal_op(my_decimal *);
   void result_precision();
+  bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
 };
 
 
@@ -474,6 +475,7 @@ public:
   const char *func_name() const { return "%"; }
   void result_precision();
   void fix_length_and_dec();
+  bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
 };
 
 
diff --git a/sql/key.cc b/sql/key.cc
index 7f075674ab6..47e5c5ebdd7 100644
--- a/sql/key.cc
+++ b/sql/key.cc
@@ -168,6 +168,7 @@ void key_restore(uchar *to_record, uchar *from_key, KEY *key_info,
   }
   for (key_part= key_info->key_part ; (int) key_length > 0 ; key_part++)
   {
+    uchar used_uneven_bits= 0;
     if (key_part->null_bit)
     {
       if (*from_key++)
@@ -186,6 +187,8 @@ void key_restore(uchar *to_record, uchar *from_key, KEY *key_info,
         set_rec_bits(bits, to_record + key_part->null_offset +
                      (key_part->null_bit == 128),
                      field->bit_ofs, field->bit_len);
+        /* we have now used the byte with 'uneven' bits */
+        used_uneven_bits= 1;
       }
     }
     if (key_part->key_part_flag & HA_BLOB_PART)
@@ -222,7 +225,9 @@ void key_restore(uchar *to_record, uchar *from_key, KEY *key_info,
     else
     {
       length= min(key_length, key_part->length);
-      memcpy(to_record + key_part->offset, from_key, (size_t) length);
+      /* skip the byte with 'uneven' bits, if used */
+      memcpy(to_record + key_part->offset, from_key + used_uneven_bits
+             , (size_t) length - used_uneven_bits);
     }
     from_key+= length;
     key_length-= length;
diff --git a/sql/log.cc b/sql/log.cc
index 103a7ccdf41..2591e326156 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -418,6 +418,7 @@ bool Log_to_csv_event_handler::
     A positive return value in store() means truncation.
     Still logging a message in the log in this case.
   */
+  table->field[5]->flags|= FIELDFLAG_HEX_ESCAPE;
   if (table->field[5]->store(sql_text, sql_text_len, client_cs) < 0)
     goto err;
 
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 15f1a957149..1f5351c1b3f 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -609,6 +609,7 @@ Log_event::Log_event(const char* buf,
 #endif
   when = uint4korr(buf);
   server_id = uint4korr(buf + SERVER_ID_OFFSET);
+  data_written= uint4korr(buf + EVENT_LEN_OFFSET);
   if (description_event->binlog_version==1)
   {
     log_pos= 0;
@@ -641,7 +642,7 @@ Log_event::Log_event(const char* buf,
         binlog, so which will cause problems if the user uses this value
         in CHANGE MASTER).
       */
-    log_pos+= uint4korr(buf + EVENT_LEN_OFFSET);
+    log_pos+= data_written; /* purecov: inspected */
   }
   DBUG_PRINT("info", ("log_pos: %lu", (ulong) log_pos));
 
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 9bdf6e2a1d5..0a51ae529ef 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1652,6 +1652,7 @@ extern pthread_mutex_t LOCK_gdl;
 #define WFRM_WRITE_SHADOW 1
 #define WFRM_INSTALL_SHADOW 2
 #define WFRM_PACK_FRM 4
+#define WFRM_KEEP_SHARE 8
 bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags);
 int abort_and_upgrade_lock(ALTER_PARTITION_PARAM_TYPE *lpt);
 void close_open_tables_and_downgrade(ALTER_PARTITION_PARAM_TYPE *lpt);
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index 3130f84bd73..8feac884c77 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -419,41 +419,167 @@ char *partition_info::has_unique_names()
 
 
 /*
-  Check that all partitions use the same storage engine.
-  This is currently a limitation in this version.
+  Check that the partition/subpartition is setup to use the correct
+  storage engine
+  SYNOPSIS
+    check_engine_condition()
+    p_elem                   Partition element
+    table_engine_set         Have user specified engine on table level
+    inout::engine_type       Current engine used
+    inout::first             Is it first partition
+  RETURN VALUE
+    TRUE                     Failed check
+    FALSE                    Ok
+  DESCRIPTION
+    Specified engine for table and partitions p0 and pn
+    Must be correct both on CREATE and ALTER commands
+    table p0 pn res (0 - OK, 1 - FAIL)
+        -  -  - 0
+        -  -  x 1
+        -  x  - 1
+        -  x  x 0
+        x  -  - 0
+        x  -  x 0
+        x  x  - 0
+        x  x  x 0
+    i.e:
+    - All subpartitions must use the same engine
+      AND it must be the same as the partition.
+    - All partitions must use the same engine
+      AND it must be the same as the table.
+    - if one does NOT specify an engine on the table level
+      then one must either NOT specify any engine on any
+      partition/subpartition OR for ALL partitions/subpartitions
+    Note:
+    When ALTER a table, the engines are already set for all levels
+    (table, all partitions and subpartitions). So if one want to
+    change the storage engine, one must specify it on the table level
 
+*/
+
+static bool check_engine_condition(partition_element *p_elem,
+                                   bool table_engine_set,
+                                   handlerton **engine_type,
+                                   bool *first)
+{
+  DBUG_ENTER("check_engine_condition");
+
+  DBUG_PRINT("enter", ("p_eng %s t_eng %s t_eng_set %u first %u state %u",
+                       ha_resolve_storage_engine_name(p_elem->engine_type),
+                       ha_resolve_storage_engine_name(*engine_type),
+                       table_engine_set, *first, p_elem->part_state));
+  if (*first && !table_engine_set)
+  {
+    *engine_type= p_elem->engine_type;
+    DBUG_PRINT("info", ("setting table_engine = %s",
+                         ha_resolve_storage_engine_name(*engine_type)));
+  }
+  *first= FALSE;
+  if ((table_engine_set &&
+      (p_elem->engine_type != (*engine_type) &&
+       p_elem->engine_type)) ||
+      (!table_engine_set &&
+       p_elem->engine_type != (*engine_type)))
+  {
+    DBUG_RETURN(TRUE);
+  }
+  else
+  {
+    DBUG_RETURN(FALSE);
+  }
+}
+
+
+/*
+  Check engine mix that it is correct
+  Current limitation is that all partitions and subpartitions
+  must use the same storage engine.
   SYNOPSIS
     check_engine_mix()
-    engine_array           An array of engine identifiers
-    no_parts               Total number of partitions
-
+    inout::engine_type       Current engine used
+    table_engine_set         Have user specified engine on table level
   RETURN VALUE
-    TRUE                   Error, mixed engines
-    FALSE                  Ok, no mixed engines
+    TRUE                     Error, mixed engines
+    FALSE                    Ok, no mixed engines
   DESCRIPTION
     Current check verifies only that all handlers are the same.
     Later this check will be more sophisticated.
+    (specified partition handler ) specified table handler
+    (NDB, NDB) NDB           OK
+    (MYISAM, MYISAM) -       OK
+    (MYISAM, -)      -       NOT OK
+    (MYISAM, -)    MYISAM    OK
+    (- , MYISAM)   -         NOT OK
+    (- , -)        MYISAM    OK
+    (-,-)          -         OK
+    (NDB, MYISAM) *          NOT OK
 */
 
-bool partition_info::check_engine_mix(handlerton **engine_array, uint no_parts)
+bool partition_info::check_engine_mix(handlerton *engine_type,
+                                      bool table_engine_set)
 {
-  uint i= 0;
+  handlerton *old_engine_type= engine_type;
+  bool first= TRUE;
+  uint no_parts= partitions.elements;
   DBUG_ENTER("partition_info::check_engine_mix");
-
-  do
+  DBUG_PRINT("info", ("in: engine_type = %s, table_engine_set = %u",
+                       ha_resolve_storage_engine_name(engine_type),
+                       table_engine_set));
+  if (no_parts)
   {
-    if (engine_array[i] != engine_array[0])
+    List_iterator<partition_element> part_it(partitions);
+    uint i= 0;
+    do
     {
-      my_error(ER_MIX_HANDLER_ERROR, MYF(0));
-      DBUG_RETURN(TRUE);
-    }
-  } while (++i < no_parts);
-  if (engine_array[0]->flags & HTON_NO_PARTITION)
+      partition_element *part_elem= part_it++;
+      DBUG_PRINT("info", ("part = %d engine = %s table_engine_set %u",
+                 i, ha_resolve_storage_engine_name(part_elem->engine_type),
+                 table_engine_set));
+      if (is_sub_partitioned() &&
+          part_elem->subpartitions.elements)
+      {
+        uint no_subparts= part_elem->subpartitions.elements;
+        uint j= 0;
+        List_iterator<partition_element> sub_it(part_elem->subpartitions);
+        do
+        {
+          partition_element *sub_elem= sub_it++;
+          DBUG_PRINT("info", ("sub = %d engine = %s table_engie_set %u",
+                     j, ha_resolve_storage_engine_name(sub_elem->engine_type),
+                     table_engine_set));
+          if (check_engine_condition(sub_elem, table_engine_set,
+                                     &engine_type, &first))
+            goto error;
+        } while (++j < no_subparts);
+        /* ensure that the partition also has correct engine */
+        if (check_engine_condition(part_elem, table_engine_set,
+                                   &engine_type, &first))
+          goto error;
+      }
+      else if (check_engine_condition(part_elem, table_engine_set,
+                                      &engine_type, &first))
+        goto error;
+    } while (++i < no_parts);
+  }
+  DBUG_PRINT("info", ("engine_type = %s",
+                       ha_resolve_storage_engine_name(engine_type)));
+  if (!engine_type)
+    engine_type= old_engine_type;
+  if (engine_type->flags & HTON_NO_PARTITION)
   {
     my_error(ER_PARTITION_MERGE_ERROR, MYF(0));
     DBUG_RETURN(TRUE);
   }
+  DBUG_PRINT("info", ("out: engine_type = %s",
+                       ha_resolve_storage_engine_name(engine_type)));
+  DBUG_ASSERT(engine_type != partition_hton);
   DBUG_RETURN(FALSE);
+error:
+  /*
+    Mixed engines not yet supported but when supported it will need
+    the partition handler
+  */
+  DBUG_RETURN(TRUE);
 }
 
 
@@ -726,13 +852,15 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
                                           handler *file, HA_CREATE_INFO *info,
                                           bool check_partition_function)
 {
-  handlerton **engine_array= NULL;
-  uint part_count= 0;
+  handlerton *table_engine= default_engine_type;
   uint i, tot_partitions;
-  bool result= TRUE;
+  bool result= TRUE, table_engine_set;
   char *same_name;
   DBUG_ENTER("partition_info::check_partition_info");
+  DBUG_ASSERT(default_engine_type != partition_hton);
 
+  DBUG_PRINT("info", ("default table_engine = %s",
+                      ha_resolve_storage_engine_name(table_engine)));
   if (check_partition_function)
   {
     int err= 0;
@@ -777,44 +905,89 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
     my_error(ER_TOO_MANY_PARTITIONS_ERROR, MYF(0));
     goto end;
   }
+  /*
+    if NOT specified ENGINE = <engine>:
+      If Create, always use create_info->db_type
+      else, use previous tables db_type 
+      either ALL or NONE partition should be set to
+      default_engine_type when not table_engine_set
+      Note: after a table is created its storage engines for
+      the table and all partitions/subpartitions are set.
+      So when ALTER it is already set on table level
+  */
+  if (info && info->used_fields & HA_CREATE_USED_ENGINE)
+  {
+    table_engine_set= TRUE;
+    table_engine= info->db_type;
+    /* if partition_hton, use thd->lex->create_info */
+    if (table_engine == partition_hton)
+      table_engine= thd->lex->create_info.db_type;
+    DBUG_ASSERT(table_engine != partition_hton);
+    DBUG_PRINT("info", ("Using table_engine = %s",
+                        ha_resolve_storage_engine_name(table_engine)));
+  }
+  else
+  {
+    table_engine_set= FALSE;
+    if (thd->lex->sql_command != SQLCOM_CREATE_TABLE)
+    {
+      table_engine_set= TRUE;
+      DBUG_PRINT("info", ("No create, table_engine = %s",
+                          ha_resolve_storage_engine_name(table_engine)));
+      DBUG_ASSERT(table_engine && table_engine != partition_hton);
+    }
+  }
+
   if ((same_name= has_unique_names()))
   {
     my_error(ER_SAME_NAME_PARTITION, MYF(0), same_name);
     goto end;
   }
-  engine_array= (handlerton**)my_malloc(tot_partitions * sizeof(handlerton *), 
-                                        MYF(MY_WME));
-  if (unlikely(!engine_array))
-    goto end;
   i= 0;
   {
     List_iterator<partition_element> part_it(partitions);
+    uint no_parts_not_set= 0;
+    uint prev_no_subparts_not_set= no_subparts + 1;
     do
     {
       partition_element *part_elem= part_it++;
-      if (part_elem->engine_type == NULL)
-        part_elem->engine_type= default_engine_type;
-      if (thd->variables.sql_mode & MODE_NO_DIR_IN_CREATE)
-        part_elem->data_file_name= part_elem->index_file_name= 0;
+#ifdef HAVE_READLINK
+      if (!my_use_symdir || (thd->variables.sql_mode & MODE_NO_DIR_IN_CREATE))
+#endif
+      {
+        if (part_elem->data_file_name)
+          push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0,
+                       "DATA DIRECTORY option ignored");
+        if (part_elem->index_file_name)
+          push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0,
+                       "INDEX DIRECTORY option ignored");
+        part_elem->data_file_name= part_elem->index_file_name= NULL;
+      }
       if (!is_sub_partitioned())
       {
+        if (part_elem->engine_type == NULL)
+        {
+          no_parts_not_set++;
+          part_elem->engine_type= default_engine_type;
+        }
         if (check_table_name(part_elem->partition_name,
                              strlen(part_elem->partition_name)))
         {
           my_error(ER_WRONG_PARTITION_NAME, MYF(0));
           goto end;
         }
-        DBUG_PRINT("info", ("engine = %d",
-                   ha_legacy_type(part_elem->engine_type)));
-        engine_array[part_count++]= part_elem->engine_type;
+        DBUG_PRINT("info", ("part = %d engine = %s",
+                   i, ha_resolve_storage_engine_name(part_elem->engine_type)));
       }
       else
       {
         uint j= 0;
+        uint no_subparts_not_set= 0;
         List_iterator<partition_element> sub_it(part_elem->subpartitions);
+        partition_element *sub_elem;
         do
         {
-          partition_element *sub_elem= sub_it++;
+          sub_elem= sub_it++;
           if (check_table_name(sub_elem->partition_name,
                                strlen(sub_elem->partition_name)))
           {
@@ -822,19 +995,65 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
             goto end;
           }
           if (sub_elem->engine_type == NULL)
-            sub_elem->engine_type= default_engine_type;
-          DBUG_PRINT("info", ("engine = %u",
-                     ha_legacy_type(sub_elem->engine_type)));
-          engine_array[part_count++]= sub_elem->engine_type;
+          {
+            if (part_elem->engine_type != NULL)
+              sub_elem->engine_type= part_elem->engine_type;
+            else
+            {
+              sub_elem->engine_type= default_engine_type;
+              no_subparts_not_set++;
+            }
+          }
+          DBUG_PRINT("info", ("part = %d sub = %d engine = %s", i, j,
+                     ha_resolve_storage_engine_name(sub_elem->engine_type)));
         } while (++j < no_subparts);
+
+        if (prev_no_subparts_not_set == (no_subparts + 1) &&
+            (no_subparts_not_set == 0 || no_subparts_not_set == no_subparts))
+          prev_no_subparts_not_set= no_subparts_not_set;
+
+        if (!table_engine_set &&
+            prev_no_subparts_not_set != no_subparts_not_set)
+        {
+          DBUG_PRINT("info", ("no_subparts_not_set = %u no_subparts = %u",
+                     no_subparts_not_set, no_subparts));
+          my_error(ER_MIX_HANDLER_ERROR, MYF(0));
+          goto end;
+        }
+
+        if (part_elem->engine_type == NULL)
+        {
+          if (no_subparts_not_set == 0)
+            part_elem->engine_type= sub_elem->engine_type;
+          else
+          {
+            no_parts_not_set++;
+            part_elem->engine_type= default_engine_type;
+          }
+        }
       }
     } while (++i < no_parts);
+    if (!table_engine_set &&
+        no_parts_not_set != 0 &&
+        no_parts_not_set != no_parts)
+    {
+      DBUG_PRINT("info", ("no_parts_not_set = %u no_parts = %u",
+                 no_parts_not_set, no_subparts));
+      my_error(ER_MIX_HANDLER_ERROR, MYF(0));
+      goto end;
+    }
   }
-  if (unlikely(partition_info::check_engine_mix(engine_array, part_count)))
+  if (unlikely(check_engine_mix(table_engine, table_engine_set)))
+  {
+    my_error(ER_MIX_HANDLER_ERROR, MYF(0));
     goto end;
+  }
 
+  DBUG_ASSERT(table_engine != partition_hton &&
+              default_engine_type == table_engine);
   if (eng_type)
-    *eng_type= (handlerton*)engine_array[0];
+    *eng_type= table_engine;
+
 
   /*
     We need to check all constant expressions that they are of the correct
@@ -850,7 +1069,6 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
   }
   result= FALSE;
 end:
-  my_free((char*)engine_array,MYF(MY_ALLOW_ZERO_PTR));
   DBUG_RETURN(result);
 }
 
@@ -1094,11 +1312,11 @@ bool check_partition_dirs(partition_info *part_info)
   return 0;
 
 dd_err:
-  my_error(ER_WRONG_ARGUMENTS,MYF(0),"DATA DIRECORY");
+  my_error(ER_WRONG_ARGUMENTS,MYF(0),"DATA DIRECTORY");
   return 1;
 
 id_err:
-  my_error(ER_WRONG_ARGUMENTS,MYF(0),"INDEX DIRECORY");
+  my_error(ER_WRONG_ARGUMENTS,MYF(0),"INDEX DIRECTORY");
   return 1;
 }
 
diff --git a/sql/partition_info.h b/sql/partition_info.h
index 2859aaec3be..2af7fa1717c 100644
--- a/sql/partition_info.h
+++ b/sql/partition_info.h
@@ -269,7 +269,7 @@ public:
   bool set_up_defaults_for_partitioning(handler *file, HA_CREATE_INFO *info,
                                         uint start_no);
   char *has_unique_names();
-  static bool check_engine_mix(handlerton **engine_array, uint no_parts);
+  bool check_engine_mix(handlerton *engine_type, bool default_engine);
   bool check_range_constants();
   bool check_list_constants();
   bool check_partition_info(THD *thd, handlerton **eng_type,
diff --git a/sql/repl_failsafe.cc b/sql/repl_failsafe.cc
index a8953217ce1..d7e783f534f 100644
--- a/sql/repl_failsafe.cc
+++ b/sql/repl_failsafe.cc
@@ -118,11 +118,14 @@ void change_rpl_status(RPL_STATUS from_status, RPL_STATUS to_status)
 }
 
 
-#define get_object(p, obj) \
+#define get_object(p, obj, msg) \
 {\
   uint len = (uint)*p++;  \
   if (p + len > p_end || len >= sizeof(obj)) \
+  {\
+    errmsg= msg;\
     goto err; \
+  }\
   strmake(obj,(char*) p,len); \
   p+= len; \
 }\
@@ -168,6 +171,7 @@ int register_slave(THD* thd, uchar* packet, uint packet_length)
   int res;
   SLAVE_INFO *si;
   uchar *p= packet, *p_end= packet + packet_length;
+  const char *errmsg= "Wrong parameters to function register_slave";
 
   if (check_access(thd, REPL_SLAVE_ACL, any_db,0,0,0,0))
     return 1;
@@ -176,9 +180,9 @@ int register_slave(THD* thd, uchar* packet, uint packet_length)
 
   thd->server_id= si->server_id= uint4korr(p);
   p+= 4;
-  get_object(p,si->host);
-  get_object(p,si->user);
-  get_object(p,si->password);
+  get_object(p,si->host, "Failed to register slave: too long 'report-host'");
+  get_object(p,si->user, "Failed to register slave: too long 'report-user'");
+  get_object(p,si->password, "Failed to register slave; too long 'report-password'");
   if (p+10 > p_end)
     goto err;
   si->port= uint2korr(p);
@@ -197,8 +201,7 @@ int register_slave(THD* thd, uchar* packet, uint packet_length)
 
 err:
   my_free(si, MYF(MY_WME));
-  my_message(ER_UNKNOWN_ERROR, "Wrong parameters to function register_slave",
-	     MYF(0));
+  my_message(ER_UNKNOWN_ERROR, errmsg, MYF(0)); /* purecov: inspected */
 err2:
   return 1;
 }
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 6f30031d948..b6dfd1a1dc8 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -955,6 +955,11 @@ err:
      Check if condition stated in UNTIL clause of START SLAVE is reached.
    SYNOPSYS
      Relay_log_info::is_until_satisfied()
+     master_beg_pos    position of the beginning of to be executed event
+                       (not log_pos member of the event that points to the
+                        beginning of the following event)
+
+
    DESCRIPTION
      Checks if UNTIL condition is reached. Uses caching result of last
      comparison of current log file name and target log file name. So cached
@@ -979,7 +984,7 @@ err:
      false - condition not met
 */
 
-bool Relay_log_info::is_until_satisfied()
+bool Relay_log_info::is_until_satisfied(my_off_t master_beg_pos)
 {
   const char *log_name;
   ulonglong log_pos;
@@ -990,7 +995,7 @@ bool Relay_log_info::is_until_satisfied()
   if (until_condition == UNTIL_MASTER_POS)
   {
     log_name= group_master_log_name;
-    log_pos= group_master_log_pos;
+    log_pos= master_beg_pos;
   }
   else
   { /* until_condition == UNTIL_RELAY_POS */
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index 36daffae1af..e13ea93842c 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -296,7 +296,7 @@ public:
   void close_temporary_tables();
 
   /* Check if UNTIL condition is satisfied. See slave.cc for more. */
-  bool is_until_satisfied();
+  bool is_until_satisfied(my_off_t master_beg_pos);
   inline ulonglong until_pos()
   {
     return ((until_condition == UNTIL_MASTER_POS) ? group_master_log_pos :
diff --git a/sql/slave.cc b/sql/slave.cc
index cce139a9238..837ae89294b 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -2019,28 +2019,6 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
      wait for something for example inside of next_event().
    */
   pthread_mutex_lock(&rli->data_lock);
-  /*
-    This tests if the position of the end of the last previous executed event
-    hits the UNTIL barrier.
-    We would prefer to test if the position of the start (or possibly) end of
-    the to-be-read event hits the UNTIL barrier, this is different if there
-    was an event ignored by the I/O thread just before (BUG#13861 to be
-    fixed).
-  */
-  if (rli->until_condition!=Relay_log_info::UNTIL_NONE &&
-      rli->is_until_satisfied())
-  {
-    char buf[22];
-    sql_print_information("Slave SQL thread stopped because it reached its"
-                    " UNTIL position %s", llstr(rli->until_pos(), buf));
-    /*
-      Setting abort_slave flag because we do not want additional message about
-      error in query execution to be printed.
-    */
-    rli->abort_slave= 1;
-    pthread_mutex_unlock(&rli->data_lock);
-    DBUG_RETURN(1);
-  }
 
   Log_event * ev = next_event(rli);
 
@@ -2054,7 +2032,30 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
   }
   if (ev)
   {
-    int exec_res= apply_event_and_update_pos(ev, thd, rli, TRUE);
+    int exec_res;
+
+    /*
+      This tests if the position of the beginning of the current event
+      hits the UNTIL barrier.
+    */
+    if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
+        rli->is_until_satisfied((rli->is_in_group() || !ev->log_pos) ?
+                                rli->group_master_log_pos :
+                                ev->log_pos - ev->data_written))
+    {
+      char buf[22];
+      sql_print_information("Slave SQL thread stopped because it reached its"
+                            " UNTIL position %s", llstr(rli->until_pos(), buf));
+      /*
+        Setting abort_slave flag because we do not want additional message about
+        error in query execution to be printed.
+      */
+      rli->abort_slave= 1;
+      pthread_mutex_unlock(&rli->data_lock);
+      delete ev;
+      DBUG_RETURN(1);
+    }
+    exec_res= apply_event_and_update_pos(ev, thd, rli, TRUE);
 
     /*
       Format_description_log_event should not be deleted because it will be
@@ -2690,6 +2691,22 @@ Slave SQL thread aborted. Can't execute init_slave query");
     }
   }
 
+  /*
+    First check until condition - probably there is nothing to execute. We
+    do not want to wait for next event in this case.
+  */
+  pthread_mutex_lock(&rli->data_lock);
+  if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
+      rli->is_until_satisfied(rli->group_master_log_pos))
+  {
+    char buf[22];
+    sql_print_information("Slave SQL thread stopped because it reached its"
+                          " UNTIL position %s", llstr(rli->until_pos(), buf));
+    pthread_mutex_unlock(&rli->data_lock);
+    goto err;
+  }
+  pthread_mutex_unlock(&rli->data_lock);
+
   /* Read queries from the IO/THREAD until this thread is killed */
 
   while (!sql_slave_killed(thd,rli))
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 01e0009ad0d..ba4f695dcb8 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -747,6 +747,7 @@ void close_handle_and_leave_table_as_lock(TABLE *table)
   table->db_stat= 0;                            // Mark file closed
   release_table_share(table->s, RELEASE_NORMAL);
   table->s= share;
+  table->file->change_table_ptr(table, table->s);
 
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 1125127c852..2f6994c1244 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2351,37 +2351,7 @@ mysql_execute_command(THD *thd)
     /* Might have been updated in create_table_precheck */
     create_info.alias= create_table->alias;
 
-#ifndef HAVE_READLINK
-    if (create_info.data_file_name)
-      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0,
-                   "DATA DIRECTORY option ignored");
-    if (create_info.index_file_name)
-      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0,
-                   "INDEX DIRECTORY option ignored");
-    create_info.data_file_name= create_info.index_file_name= NULL;
-#else
-
-    if (test_if_data_home_dir(lex->create_info.data_file_name))
-    {
-      my_error(ER_WRONG_ARGUMENTS,MYF(0),"DATA DIRECORY");
-      res= -1;
-      break;
-    }
-    if (test_if_data_home_dir(lex->create_info.index_file_name))
-    {
-      my_error(ER_WRONG_ARGUMENTS,MYF(0),"INDEX DIRECORY");
-      res= -1;
-      break;
-    }
-
-#ifdef WITH_PARTITION_STORAGE_ENGINE
-    if (check_partition_dirs(thd->lex->part_info))
-    {
-      res= -1;
-      break;
-    }
-#endif
-
+#ifdef HAVE_READLINK
     /* Fix names if symlinked tables */
     if (append_file_to_dir(thd, &create_info.data_file_name,
 			   create_table->table_name) ||
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 45a82782de0..037da87be7f 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -3819,9 +3819,9 @@ bool mysql_unpack_partition(THD *thd,
 
   DBUG_PRINT("info", ("Successful parse"));
   part_info= lex.part_info;
-  DBUG_PRINT("info", ("default engine = %d, default_db_type = %d",
-             ha_legacy_type(part_info->default_engine_type),
-             ha_legacy_type(default_db_type)));
+  DBUG_PRINT("info", ("default engine = %s, default_db_type = %s",
+             ha_resolve_storage_engine_name(part_info->default_engine_type),
+             ha_resolve_storage_engine_name(default_db_type)));
   if (is_create_table_ind && old_lex->sql_command == SQLCOM_CREATE_TABLE)
   {
     if (old_lex->create_info.options & HA_LEX_CREATE_TABLE_LIKE)
@@ -3863,6 +3863,8 @@ bool mysql_unpack_partition(THD *thd,
   if (!part_info->default_engine_type)
     part_info->default_engine_type= default_db_type;
   DBUG_ASSERT(part_info->default_engine_type == default_db_type);
+  DBUG_ASSERT(part_info->default_engine_type->db_type != DB_TYPE_UNKNOWN);
+  DBUG_ASSERT(part_info->default_engine_type != partition_hton);
 
   {
   /*
@@ -4001,56 +4003,6 @@ static int fast_end_partition(THD *thd, ulonglong copied,
 
 
 /*
-  Check engine mix that it is correct
-  SYNOPSIS
-    check_engine_condition()
-    p_elem                   Partition element
-    default_engine           Have user specified engine on table level
-    inout::engine_type       Current engine used
-    inout::first             Is it first partition
-  RETURN VALUE
-    TRUE                     Failed check
-    FALSE                    Ok
-  DESCRIPTION
-    (specified partition handler ) specified table handler
-    (NDB, NDB) NDB           OK
-    (MYISAM, MYISAM) -       OK
-    (MYISAM, -)      -       NOT OK
-    (MYISAM, -)    MYISAM    OK
-    (- , MYISAM)   -         NOT OK
-    (- , -)        MYISAM    OK
-    (-,-)          -         OK
-    (NDB, MYISAM) *          NOT OK
-*/
-
-static bool check_engine_condition(partition_element *p_elem,
-                                   bool default_engine,
-                                   handlerton **engine_type,
-                                   bool *first)
-{
-  DBUG_ENTER("check_engine_condition");
-
-  DBUG_PRINT("enter", ("def_eng = %u, first = %u", default_engine, *first));
-  if (*first && default_engine)
-  {
-    *engine_type= p_elem->engine_type;
-  }
-  *first= FALSE;
-  if ((!default_engine &&
-      (p_elem->engine_type != (*engine_type) &&
-       p_elem->engine_type)) ||
-      (default_engine &&
-       p_elem->engine_type != (*engine_type)))
-  {
-    DBUG_RETURN(TRUE);
-  }
-  else
-  {
-    DBUG_RETURN(FALSE);
-  }
-}
-
-/*
   We need to check if engine used by all partitions can handle
   partitioning natively.
 
@@ -4074,52 +4026,30 @@ static bool check_engine_condition(partition_element *p_elem,
 static bool check_native_partitioned(HA_CREATE_INFO *create_info,bool *ret_val,
                                      partition_info *part_info, THD *thd)
 {
-  List_iterator<partition_element> part_it(part_info->partitions);
-  bool first= TRUE;
-  bool default_engine;
-  handlerton *engine_type= create_info->db_type;
+  bool table_engine_set;
+  handlerton *engine_type= part_info->default_engine_type;
   handlerton *old_engine_type= engine_type;
-  uint i= 0;
-  uint no_parts= part_info->partitions.elements;
   DBUG_ENTER("check_native_partitioned");
 
-  default_engine= (create_info->used_fields & HA_CREATE_USED_ENGINE) ?
-                   FALSE : TRUE;
-  DBUG_PRINT("info", ("engine_type = %u, default = %u",
-                       ha_legacy_type(engine_type),
-                       default_engine));
-  if (no_parts)
+  if (create_info->used_fields & HA_CREATE_USED_ENGINE)
   {
-    do
+    table_engine_set= TRUE;
+    engine_type= create_info->db_type;
+  }
+  else
+  {
+    table_engine_set= FALSE;
+    if (thd->lex->sql_command != SQLCOM_CREATE_TABLE)
     {
-      partition_element *part_elem= part_it++;
-      if (part_info->is_sub_partitioned() &&
-          part_elem->subpartitions.elements)
-      {
-        uint no_subparts= part_elem->subpartitions.elements;
-        uint j= 0;
-        List_iterator<partition_element> sub_it(part_elem->subpartitions);
-        do
-        {
-          partition_element *sub_elem= sub_it++;
-          if (check_engine_condition(sub_elem, default_engine,
-                                     &engine_type, &first))
-            goto error;
-        } while (++j < no_subparts);
-        /*
-          In case of subpartitioning and defaults we allow that only
-          subparts have specified engines, as long as the parts haven't
-          specified the wrong engine it's ok.
-        */
-        if (check_engine_condition(part_elem, FALSE,
-                                   &engine_type, &first))
-          goto error;
-      }
-      else if (check_engine_condition(part_elem, default_engine,
-                                      &engine_type, &first))
-        goto error;
-    } while (++i < no_parts);
+      table_engine_set= TRUE;
+      DBUG_ASSERT(engine_type && engine_type != partition_hton);
+    }
   }
+  DBUG_PRINT("info", ("engine_type = %s, table_engine_set = %u",
+                       ha_resolve_storage_engine_name(engine_type),
+                       table_engine_set));
+  if (part_info->check_engine_mix(engine_type, table_engine_set))
+    goto error;
 
   /*
     All engines are of the same type. Check if this engine supports
@@ -4216,7 +4146,7 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
       my_error(ER_PARTITION_MGMT_ON_NONPARTITIONED, MYF(0));
       DBUG_RETURN(TRUE);
     }
-    if (alter_info->flags == ALTER_TABLE_REORG)
+    if (alter_info->flags & ALTER_TABLE_REORG)
     {
       uint new_part_no, curr_part_no;
       if (tab_part_info->part_type != HASH_PARTITION ||
@@ -4317,7 +4247,12 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
       {
         my_error(ER_NO_BINLOG_ERROR, MYF(0));
         DBUG_RETURN(TRUE);
-      } 
+      }
+      if (tab_part_info->defined_max_value)
+      {
+        my_error(ER_PARTITION_MAXVALUE_ERROR, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
       if (no_new_partitions == 0)
       {
         my_error(ER_ADD_PARTITION_NO_NEW_PARTITION, MYF(0));
@@ -4542,7 +4477,7 @@ that are reorganised.
         tab_part_info->is_auto_partitioned= FALSE;
       }
     }
-    else if (alter_info->flags == ALTER_DROP_PARTITION)
+    else if (alter_info->flags & ALTER_DROP_PARTITION)
     {
       /*
         Drop a partition from a range partition and list partitioning is
@@ -4746,7 +4681,7 @@ state of p1.
         tab_part_info->is_auto_partitioned= FALSE;
       }
     }
-    else if (alter_info->flags == ALTER_REORGANIZE_PARTITION)
+    else if (alter_info->flags & ALTER_REORGANIZE_PARTITION)
     {
       /*
         Reorganise partitions takes a number of partitions that are next
@@ -4927,8 +4862,8 @@ the generated partition syntax in a correct manner.
     }
     *partition_changed= TRUE;
     thd->work_part_info= tab_part_info;
-    if (alter_info->flags == ALTER_ADD_PARTITION ||
-        alter_info->flags == ALTER_REORGANIZE_PARTITION)
+    if (alter_info->flags & ALTER_ADD_PARTITION ||
+        alter_info->flags & ALTER_REORGANIZE_PARTITION)
     {
       if (tab_part_info->use_default_subpartitions &&
           !alt_part_info->use_default_subpartitions)
@@ -5055,13 +4990,21 @@ the generated partition syntax in a correct manner.
         DBUG_PRINT("info", ("partition changed"));
         *partition_changed= TRUE;
       }
-      if (create_info->db_type == partition_hton)
+      /*
+        Set up partition default_engine_type either from the create_info
+        or from the previus table
+      */
+      if (create_info->used_fields & HA_CREATE_USED_ENGINE)
+        part_info->default_engine_type= create_info->db_type;
+      else
       {
-        if (!part_info->default_engine_type)
+        if (table->part_info)
           part_info->default_engine_type= table->part_info->default_engine_type;
+        else
+          part_info->default_engine_type= create_info->db_type;
       }
-      else
-        part_info->default_engine_type= create_info->db_type;
+      DBUG_ASSERT(part_info->default_engine_type &&
+                  part_info->default_engine_type != partition_hton);
       if (check_native_partitioned(create_info, &is_native_partitioned,
                                    part_info, thd))
       {
@@ -5842,32 +5785,42 @@ static void release_log_entries(partition_info *part_info)
 
 
 /*
-  Get a lock on table name to avoid that anyone can open the table in
-  a critical part of the ALTER TABLE.
-  SYNOPSIS
-    get_name_lock()
+  Final part of partition changes to handle things when under
+  LOCK TABLES.
+  SYNPOSIS
+    alter_partition_lock_handling()
     lpt                        Struct carrying parameters
   RETURN VALUES
-    FALSE                      Success
-    TRUE                       Failure
+    NONE
 */
-
-static int get_name_lock(ALTER_PARTITION_PARAM_TYPE *lpt)
+static void alter_partition_lock_handling(ALTER_PARTITION_PARAM_TYPE *lpt)
 {
-  int error= 0;
-  DBUG_ENTER("get_name_lock");
-
-  bzero(&lpt->table_list, sizeof(lpt->table_list));
-  lpt->table_list.db= (char*)lpt->db;
-  lpt->table_list.table= lpt->table;
-  lpt->table_list.table_name= (char*)lpt->table_name;
-  pthread_mutex_lock(&LOCK_open);
-  error= lock_table_name(lpt->thd, &lpt->table_list, FALSE);
-  pthread_mutex_unlock(&LOCK_open);
-  DBUG_RETURN(error);
+  int err;
+  if (lpt->thd->locked_tables)
+  {
+    /*
+      When we have the table locked, it is necessary to reopen the table
+      since all table objects were closed and removed as part of the
+      ALTER TABLE of partitioning structure.
+    */
+    pthread_mutex_lock(&LOCK_open);
+    lpt->thd->in_lock_tables= 1;
+    err= reopen_tables(lpt->thd, 1, 1);
+    lpt->thd->in_lock_tables= 0;
+    if (err)
+    {
+      /*
+       Issue a warning since we weren't able to regain the lock again.
+       We also need to unlink table from thread's open list and from
+       table_cache
+     */
+      unlink_open_table(lpt->thd, lpt->table, FALSE);
+      sql_print_warning("We failed to reacquire LOCKs in ALTER TABLE");
+    }
+    pthread_mutex_unlock(&LOCK_open);
+  }
 }
 
-
 /*
   Unlock and close table before renaming and dropping partitions
   SYNOPSIS
@@ -5880,35 +5833,16 @@ static int get_name_lock(ALTER_PARTITION_PARAM_TYPE *lpt)
 static int alter_close_tables(ALTER_PARTITION_PARAM_TYPE *lpt)
 {
   THD *thd= lpt->thd;
-  TABLE *table= lpt->table;
+  const char *db= lpt->db;
+  const char *table_name= lpt->table_name;
   DBUG_ENTER("alter_close_tables");
   /*
     We need to also unlock tables and close all handlers.
     We set lock to zero to ensure we don't do this twice
     and we set db_stat to zero to ensure we don't close twice.
   */
-  mysql_unlock_tables(thd, thd->lock);
-  thd->lock= 0;
-  table->file->close();
-  table->db_stat= 0;
-  DBUG_RETURN(0);
-}
-
-
-/*
-  Release a lock name
-  SYNOPSIS
-    release_name_lock()
-    lpt
-  RETURN VALUES
-    0
-*/
-
-static int release_name_lock(ALTER_PARTITION_PARAM_TYPE *lpt)
-{
-  DBUG_ENTER("release_name_lock");
   pthread_mutex_lock(&LOCK_open);
-  unlock_table_name(lpt->thd, &lpt->table_list);
+  close_data_files_and_morph_locks(thd, db, table_name);
   pthread_mutex_unlock(&LOCK_open);
   DBUG_RETURN(0);
 }
@@ -6117,7 +6051,19 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ((alter_info->flags & ALTER_REPAIR_PARTITION) &&
          (error= table->file->ha_repair_partitions(thd))))
     {
-      table->file->print_error(error, MYF(0));
+      if (error == HA_ADMIN_NOT_IMPLEMENTED) {
+        if (alter_info->flags & ALTER_OPTIMIZE_PARTITION)
+          my_error(ER_CHECK_NOT_IMPLEMENTED, MYF(0), "optimize partition");
+        else if (alter_info->flags & ALTER_ANALYZE_PARTITION)
+          my_error(ER_CHECK_NOT_IMPLEMENTED, MYF(0), "analyze partition");
+        else if (alter_info->flags & ALTER_CHECK_PARTITION)
+          my_error(ER_CHECK_NOT_IMPLEMENTED, MYF(0), "check partition");
+        else if (alter_info->flags & ALTER_REPAIR_PARTITION)
+          my_error(ER_CHECK_NOT_IMPLEMENTED, MYF(0), "repair partition");
+        else
+          table->file->print_error(error, MYF(0));
+      } else
+        table->file->print_error(error, MYF(0));
       goto err;
     }
   }
@@ -6168,7 +6114,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
       goto err;
     }
   }
-  else if (alter_info->flags == ALTER_DROP_PARTITION)
+  else if (alter_info->flags & ALTER_DROP_PARTITION)
   {
     /*
       Now after all checks and setting state on dropped partitions we can
@@ -6206,7 +6152,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
          name lock.
       5) Close all tables that have already been opened but didn't stumble on
          the abort locked previously. This is done as part of the
-         get_name_lock call.
+         close_data_files_and_morph_locks call.
       6) We are now ready to release all locks we got in this thread.
       7) Write the bin log
          Unfortunately the writing of the binlog is not synchronised with
@@ -6223,8 +6169,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
       9) Prepare handlers for drop of partitions
       10) Drop the partitions
       11) Remove entries from ddl log
-      12) Release name lock so that all other threads can access the table
-          again.
+      12) Reopen table if under lock tables
       13) Complete query
 
       We insert Error injections at all places where it could be interesting
@@ -6239,23 +6184,21 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         (not_completed= FALSE) ||
         abort_and_upgrade_lock(lpt) || /* Always returns 0 */
         ERROR_INJECT_CRASH("crash_drop_partition_4") ||
-        get_name_lock(lpt) ||
-        ERROR_INJECT_CRASH("crash_drop_partition_5") ||
         alter_close_tables(lpt) ||
-        ERROR_INJECT_CRASH("crash_drop_partition_6") ||
+        ERROR_INJECT_CRASH("crash_drop_partition_5") ||
         ((!thd->lex->no_write_to_binlog) &&
          (write_bin_log(thd, FALSE,
                         thd->query, thd->query_length), FALSE)) ||
-        ERROR_INJECT_CRASH("crash_drop_partition_7") ||
+        ERROR_INJECT_CRASH("crash_drop_partition_6") ||
         ((frm_install= TRUE), FALSE) ||
         mysql_write_frm(lpt, WFRM_INSTALL_SHADOW) ||
         ((frm_install= FALSE), FALSE) ||
-        ERROR_INJECT_CRASH("crash_drop_partition_8") ||
+        ERROR_INJECT_CRASH("crash_drop_partition_7") ||
         mysql_drop_partitions(lpt) ||
-        ERROR_INJECT_CRASH("crash_drop_partition_9") ||
+        ERROR_INJECT_CRASH("crash_drop_partition_8") ||
         (write_log_completed(lpt, FALSE), FALSE) ||
-        ERROR_INJECT_CRASH("crash_drop_partition_10") ||
-        (release_name_lock(lpt), FALSE)) 
+        ERROR_INJECT_CRASH("crash_drop_partition_9") ||
+        (alter_partition_lock_handling(lpt), FALSE)) 
     {
       handle_alter_part_error(lpt, not_completed, TRUE, frm_install);
       goto err;
@@ -6287,7 +6230,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
          name lock.
       5) Close all tables that have already been opened but didn't stumble on
          the abort locked previously. This is done as part of the
-         get_name_lock call.
+         close_data_files_and_morph_locks call.
       6) Close all table handlers and unlock all handlers but retain name lock
       7) Write binlog
       8) Now the change is completed except for the installation of the
@@ -6297,7 +6240,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
          added to the table.
       10)Wait until all accesses using the old frm file has completed
       11)Remove entries from ddl log
-      12)Release name lock
+      12)Reopen tables if under lock tables
       13)Complete query
     */
     if (write_log_add_change_partition(lpt) ||
@@ -6307,8 +6250,6 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         mysql_change_partitions(lpt) ||
         ERROR_INJECT_CRASH("crash_add_partition_3") ||
         abort_and_upgrade_lock(lpt) || /* Always returns 0 */
-        ERROR_INJECT_CRASH("crash_add_partition_3") ||
-        get_name_lock(lpt) ||
         ERROR_INJECT_CRASH("crash_add_partition_4") ||
         alter_close_tables(lpt) ||
         ERROR_INJECT_CRASH("crash_add_partition_5") ||
@@ -6324,7 +6265,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT_CRASH("crash_add_partition_8") ||
         (write_log_completed(lpt, FALSE), FALSE) ||
         ERROR_INJECT_CRASH("crash_add_partition_9") ||
-        (release_name_lock(lpt), FALSE)) 
+        (alter_partition_lock_handling(lpt), FALSE)) 
     {
       handle_alter_part_error(lpt, not_completed, FALSE, frm_install);
       goto err;
@@ -6378,15 +6319,15 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
       7) Close all tables opened but not yet locked, after this call we are
          certain that no other thread is in the lock wait queue or has
          opened the table. The name lock will ensure that they are blocked
-         on the open call. This is achieved also by get_name_lock call.
+         on the open call.
+         This is achieved also by close_data_files_and_morph_locks call.
       8) Close all partitions opened by this thread, but retain name lock.
       9) Write bin log
       10) Prepare handlers for rename and delete of partitions
       11) Rename and drop the reorged partitions such that they are no
           longer used and rename those added to their real new names.
       12) Install the shadow frm file
-      13) Release the name lock to enable other threads to start using the
-          table again.
+      13) Reopen the table if under lock tables
       14) Complete query
     */
     if (write_log_add_change_partition(lpt) ||
@@ -6400,24 +6341,22 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         (not_completed= FALSE) ||
         abort_and_upgrade_lock(lpt) || /* Always returns 0 */
         ERROR_INJECT_CRASH("crash_change_partition_5") ||
-        get_name_lock(lpt) ||
-        ERROR_INJECT_CRASH("crash_change_partition_6") ||
         alter_close_tables(lpt) ||
-        ERROR_INJECT_CRASH("crash_change_partition_7") ||
+        ERROR_INJECT_CRASH("crash_change_partition_6") ||
         ((!thd->lex->no_write_to_binlog) &&
          (write_bin_log(thd, FALSE,
                         thd->query, thd->query_length), FALSE)) ||
-        ERROR_INJECT_CRASH("crash_change_partition_8") ||
+        ERROR_INJECT_CRASH("crash_change_partition_7") ||
         mysql_write_frm(lpt, WFRM_INSTALL_SHADOW) ||
-        ERROR_INJECT_CRASH("crash_change_partition_9") ||
+        ERROR_INJECT_CRASH("crash_change_partition_8") ||
         mysql_drop_partitions(lpt) ||
-        ERROR_INJECT_CRASH("crash_change_partition_10") ||
+        ERROR_INJECT_CRASH("crash_change_partition_9") ||
         mysql_rename_partitions(lpt) ||
         ((frm_install= TRUE), FALSE) ||
-        ERROR_INJECT_CRASH("crash_change_partition_11") ||
+        ERROR_INJECT_CRASH("crash_change_partition_10") ||
         (write_log_completed(lpt, FALSE), FALSE) ||
-        ERROR_INJECT_CRASH("crash_change_partition_12") ||
-        (release_name_lock(lpt), FALSE))
+        ERROR_INJECT_CRASH("crash_change_partition_11") ||
+        (alter_partition_lock_handling(lpt), FALSE))
     {
       handle_alter_part_error(lpt, not_completed, FALSE, frm_install);
       goto err;
diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc
index 4da4f1373e2..e2449613458 100644
--- a/sql/sql_plugin.cc
+++ b/sql/sql_plugin.cc
@@ -1183,9 +1183,8 @@ int plugin_init(int *argc, char **argv, int flags)
   /* Register all dynamic plugins */
   if (!(flags & PLUGIN_INIT_SKIP_DYNAMIC_LOADING))
   {
-    if (opt_plugin_load &&
-        plugin_load_list(&tmp_root, argc, argv, opt_plugin_load))
-      goto err;
+    if (opt_plugin_load)
+      plugin_load_list(&tmp_root, argc, argv, opt_plugin_load);
     if (!(flags & PLUGIN_INIT_SKIP_PLUGIN_TABLE))
       plugin_load(&tmp_root, argc, argv);
   }
@@ -1414,7 +1413,11 @@ static bool plugin_load_list(MEM_ROOT *tmp_root, int *argc, char **argv,
   while (list)
   {
     if (p == buffer + sizeof(buffer) - 1)
-      break;
+    {
+      sql_print_error("plugin-load parameter too long");
+      DBUG_RETURN(TRUE);
+    }
+
     switch ((*(p++)= *(list++))) {
     case '\0':
       list= NULL; /* terminate the loop */
@@ -1423,10 +1426,17 @@ static bool plugin_load_list(MEM_ROOT *tmp_root, int *argc, char **argv,
     case ':':     /* can't use this as delimiter as it may be drive letter */
 #endif
     case ';':
-      name.str[name.length]= '\0';
-      if (str != &dl)  // load all plugins in named module
+      str->str[str->length]= '\0';
+      if (str == &name)  // load all plugins in named module
       {
+        if (!name.length)
+        {
+          p--;    /* reset pointer */
+          continue;
+        }
+
         dl= name;
+        pthread_mutex_lock(&LOCK_plugin);
         if ((plugin_dl= plugin_dl_add(&dl, REPORT_TO_LOG)))
         {
           for (plugin= plugin_dl->plugins; plugin->info; plugin++)
@@ -1436,7 +1446,10 @@ static bool plugin_load_list(MEM_ROOT *tmp_root, int *argc, char **argv,
 
             free_root(tmp_root, MYF(MY_MARK_BLOCKS_FREE));
             if (plugin_add(tmp_root, &name, &dl, argc, argv, REPORT_TO_LOG))
+            {
+              pthread_mutex_unlock(&LOCK_plugin);
               goto error;
+            }
           }
           plugin_dl_del(&dl); // reduce ref count
         }
@@ -1444,9 +1457,14 @@ static bool plugin_load_list(MEM_ROOT *tmp_root, int *argc, char **argv,
       else
       {
         free_root(tmp_root, MYF(MY_MARK_BLOCKS_FREE));
+        pthread_mutex_lock(&LOCK_plugin);
         if (plugin_add(tmp_root, &name, &dl, argc, argv, REPORT_TO_LOG))
+        {
+          pthread_mutex_unlock(&LOCK_plugin);
           goto error;
+        }
       }
+      pthread_mutex_unlock(&LOCK_plugin);
       name.length= dl.length= 0;
       dl.str= NULL; name.str= p= buffer;
       str= &name;
@@ -1455,6 +1473,7 @@ static bool plugin_load_list(MEM_ROOT *tmp_root, int *argc, char **argv,
     case '#':
       if (str == &name)
       {
+        name.str[name.length]= '\0';
         str= &dl;
         str->str= p;
         continue;
@@ -2046,35 +2065,35 @@ err:
 
 
 static void update_func_bool(THD *thd, struct st_mysql_sys_var *var,
-                             void *tgt, void *save)
+                             void *tgt, const void *save)
 {
   *(my_bool *) tgt= *(int *) save ? 1 : 0;
 }
 
 
 static void update_func_int(THD *thd, struct st_mysql_sys_var *var,
-                             void *tgt, void *save)
+                             void *tgt, const void *save)
 {
   *(int *)tgt= *(int *) save;
 }
 
 
 static void update_func_long(THD *thd, struct st_mysql_sys_var *var,
-                             void *tgt, void *save)
+                             void *tgt, const void *save)
 {
   *(long *)tgt= *(long *) save;
 }
 
 
 static void update_func_longlong(THD *thd, struct st_mysql_sys_var *var,
-                             void *tgt, void *save)
+                             void *tgt, const void *save)
 {
   *(longlong *)tgt= *(ulonglong *) save;
 }
 
 
 static void update_func_str(THD *thd, struct st_mysql_sys_var *var,
-                             void *tgt, void *save)
+                             void *tgt, const void *save)
 {
   char *old= *(char **) tgt;
   *(char **)tgt= *(char **) save;
@@ -2631,7 +2650,8 @@ bool sys_var_pluginvar::check(THD *thd, set_var *var)
 
 void sys_var_pluginvar::set_default(THD *thd, enum_var_type type)
 {
-  void *tgt, *src;
+  const void *src;
+  void *tgt;
 
   DBUG_ASSERT(is_readonly() || plugin_var->update);
 
@@ -2644,9 +2664,34 @@ void sys_var_pluginvar::set_default(THD *thd, enum_var_type type)
 
   if (plugin_var->flags & PLUGIN_VAR_THDLOCAL)
   {
-    src= ((int*) (plugin_var + 1) + 1);
     if (type != OPT_GLOBAL)
       src= real_value_ptr(thd, OPT_GLOBAL);
+    else
+    switch (plugin_var->flags & PLUGIN_VAR_TYPEMASK) {
+	case PLUGIN_VAR_INT:
+	  src= &((thdvar_uint_t*) plugin_var)->def_val;
+	  break;
+	case PLUGIN_VAR_LONG:
+	  src= &((thdvar_ulong_t*) plugin_var)->def_val;
+	  break;
+	case PLUGIN_VAR_LONGLONG:
+	  src= &((thdvar_ulonglong_t*) plugin_var)->def_val;
+	  break;
+	case PLUGIN_VAR_ENUM:
+	  src= &((thdvar_enum_t*) plugin_var)->def_val;
+	  break;
+	case PLUGIN_VAR_SET:
+	  src= &((thdvar_set_t*) plugin_var)->def_val;
+	  break;
+	case PLUGIN_VAR_BOOL:
+	  src= &((thdvar_bool_t*) plugin_var)->def_val;
+	  break;
+	case PLUGIN_VAR_STR:
+	  src= &((thdvar_str_t*) plugin_var)->def_val;
+	  break;
+	default:
+	  DBUG_ASSERT(0);
+	}
   }
 
   /* thd must equal current_thd if PLUGIN_VAR_THDLOCAL flag is set */
@@ -2734,25 +2779,25 @@ static void plugin_opt_set_limits(struct my_option *options,
   case PLUGIN_VAR_ENUM:
     options->var_type= GET_ENUM;
     options->typelib= ((sysvar_enum_t*) opt)->typelib;
-    options->def_value= *(ulong*) ((int*) (opt + 1) + 1);
+    options->def_value= ((sysvar_enum_t*) opt)->def_val;
     options->min_value= options->block_size= 0;
     options->max_value= options->typelib->count - 1;
     break;
   case PLUGIN_VAR_SET:
     options->var_type= GET_SET;
     options->typelib= ((sysvar_set_t*) opt)->typelib;
-    options->def_value= *(ulonglong*) ((int*) (opt + 1) + 1);
+    options->def_value= ((sysvar_set_t*) opt)->def_val;
     options->min_value= options->block_size= 0;
     options->max_value= (ULL(1) << options->typelib->count) - 1;
     break;
   case PLUGIN_VAR_BOOL:
     options->var_type= GET_BOOL;
-    options->def_value= *(my_bool*) ((void**)(opt + 1) + 1);
+    options->def_value= ((sysvar_bool_t*) opt)->def_val;
     break;
   case PLUGIN_VAR_STR:
     options->var_type= ((opt->flags & PLUGIN_VAR_MEMALLOC) ?
                         GET_STR_ALLOC : GET_STR);
-    options->def_value= (ulonglong)(intptr) *((char**) ((void**) (opt + 1) + 1));
+    options->def_value= (intptr) ((sysvar_str_t*) opt)->def_val;
     break;
   /* threadlocal variables */
   case PLUGIN_VAR_INT | PLUGIN_VAR_THDLOCAL:
@@ -2776,25 +2821,25 @@ static void plugin_opt_set_limits(struct my_option *options,
   case PLUGIN_VAR_ENUM | PLUGIN_VAR_THDLOCAL:
     options->var_type= GET_ENUM;
     options->typelib= ((thdvar_enum_t*) opt)->typelib;
-    options->def_value= *(ulong*) ((int*) (opt + 1) + 1);
+    options->def_value= ((thdvar_enum_t*) opt)->def_val;
     options->min_value= options->block_size= 0;
     options->max_value= options->typelib->count - 1;
     break;
   case PLUGIN_VAR_SET | PLUGIN_VAR_THDLOCAL:
     options->var_type= GET_SET;
     options->typelib= ((thdvar_set_t*) opt)->typelib;
-    options->def_value= *(ulonglong*) ((int*) (opt + 1) + 1);
+    options->def_value= ((thdvar_set_t*) opt)->def_val;
     options->min_value= options->block_size= 0;
     options->max_value= (ULL(1) << options->typelib->count) - 1;
     break;
   case PLUGIN_VAR_BOOL | PLUGIN_VAR_THDLOCAL:
     options->var_type= GET_BOOL;
-    options->def_value= *(my_bool*) ((int*) (opt + 1) + 1);
+    options->def_value= ((thdvar_bool_t*) opt)->def_val;
     break;
   case PLUGIN_VAR_STR | PLUGIN_VAR_THDLOCAL:
     options->var_type= ((opt->flags & PLUGIN_VAR_MEMALLOC) ?
                         GET_STR_ALLOC : GET_STR);
-    options->def_value= (intptr) *((char**) ((void**) (opt + 1) + 1));
+    options->def_value= (intptr) ((thdvar_str_t*) opt)->def_val;
     break;
   default:
     DBUG_ASSERT(0);
@@ -2976,7 +3021,8 @@ static int construct_options(MEM_ROOT *mem_root, struct st_plugin_int *tmp,
       DBUG_RETURN(-1);
     }
 
-    if (opt->flags & PLUGIN_VAR_NOCMDOPT)
+    if ((opt->flags & (PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_THDLOCAL))
+                    == PLUGIN_VAR_NOCMDOPT)
       continue;
 
     if (!opt->name)
@@ -2986,7 +3032,7 @@ static int construct_options(MEM_ROOT *mem_root, struct st_plugin_int *tmp,
       DBUG_RETURN(-1);
     }
 
-    if (!(v= find_bookmark(name, opt->name, opt->flags)))
+    if (!(opt->flags & PLUGIN_VAR_THDLOCAL))
     {
       optnamelen= strlen(opt->name);
       optname= (char*) alloc_root(mem_root, namelen + optnamelen + 2);
@@ -2994,7 +3040,23 @@ static int construct_options(MEM_ROOT *mem_root, struct st_plugin_int *tmp,
       optnamelen= namelen + optnamelen + 1;
     }
     else
-      optname= (char*) memdup_root(mem_root, v->key + 1, (optnamelen= v->name_len) + 1);
+    {
+      /* this should not fail because register_var should create entry */
+      if (!(v= find_bookmark(name, opt->name, opt->flags)))
+      {
+        sql_print_error("Thread local variable '%s' not allocated "
+                        "in plugin '%s'.", opt->name, plugin_name);
+        DBUG_RETURN(-1);
+      }
+
+      *(int*)(opt + 1)= offset= v->offset;
+
+      if (opt->flags & PLUGIN_VAR_NOCMDOPT)
+        continue;
+
+      optname= (char*) memdup_root(mem_root, v->key + 1, 
+                                   (optnamelen= v->name_len) + 1);
+    }
 
     /* convert '_' to '-' */
     for (p= optname; *p; p++)
@@ -3006,20 +3068,13 @@ static int construct_options(MEM_ROOT *mem_root, struct st_plugin_int *tmp,
     options->app_type= opt;
     options->id= (options-1)->id + 1;
 
-    if (opt->flags & PLUGIN_VAR_THDLOCAL)
-      *(int*)(opt + 1)= offset= v->offset;
-
     plugin_opt_set_limits(options, opt);
 
-    if ((opt->flags & PLUGIN_VAR_TYPEMASK) != PLUGIN_VAR_ENUM &&
-        (opt->flags & PLUGIN_VAR_TYPEMASK) != PLUGIN_VAR_SET)
-    {
-      if (opt->flags & PLUGIN_VAR_THDLOCAL)
-        options->value= options->u_max_value= (uchar**)
-          (global_system_variables.dynamic_variables_ptr + offset);
-      else
-        options->value= options->u_max_value= *(uchar***) (opt + 1);
-    }
+    if (opt->flags & PLUGIN_VAR_THDLOCAL)
+      options->value= options->u_max_value= (uchar**)
+        (global_system_variables.dynamic_variables_ptr + offset);
+    else
+      options->value= options->u_max_value= *(uchar***) (opt + 1);
 
     options[1]= options[0];
     options[1].name= p= (char*) alloc_root(mem_root, optnamelen + 8);
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 6e61ee6da5f..aaababc330a 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -3430,7 +3430,7 @@ static int get_schema_tables_record(THD *thd, TABLE_LIST *tables,
     /*
       there was errors during opening tables
     */
-    const char *error= thd->main_da.message();
+    const char *error= thd->is_error() ? thd->main_da.message() : "";
     if (tables->view)
       table->field[3]->store(STRING_WITH_LEN("VIEW"), cs);
     else if (tables->schema_table)
@@ -3631,8 +3631,9 @@ static int get_schema_column_record(THD *thd, TABLE_LIST *tables,
         I.e. we are in SELECT FROM INFORMATION_SCHEMA.COLUMS
         rather than in SHOW COLUMNS
       */ 
-      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                   thd->main_da.sql_errno(), thd->main_da.message());
+      if (thd->is_error())
+        push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                     thd->main_da.sql_errno(), thd->main_da.message());
       thd->clear_error();
       res= 0;
     }
diff --git a/sql/sql_string.cc b/sql/sql_string.cc
index 7fa3786c382..34b310931d6 100644
--- a/sql/sql_string.cc
+++ b/sql/sql_string.cc
@@ -840,6 +840,68 @@ outp:
 }
 
 
+/**
+  Copy string with HEX-encoding of "bad" characters.
+
+  @details This functions copies the string pointed by "src"
+  to the string pointed by "dst". Not more than "srclen" bytes
+  are read from "src". Any sequences of bytes representing
+  a not-well-formed substring (according to cs) are hex-encoded,
+  and all well-formed substrings (according to cs) are copied as is.
+  Not more than "dstlen" bytes are written to "dst". The number 
+  of bytes written to "dst" is returned.
+  
+   @param      cs       character set pointer of the destination string
+   @param[out] dst      destination string
+   @param      dstlen   size of dst
+   @param      src      source string
+   @param      srclen   length of src
+
+   @retval     result length
+*/
+
+size_t
+my_copy_with_hex_escaping(CHARSET_INFO *cs,
+                          char *dst, size_t dstlen,
+                          const char *src, size_t srclen)
+{
+  const char *srcend= src + srclen;
+  char *dst0= dst;
+
+  for ( ; src < srcend ; )
+  {
+    size_t chlen;
+    if ((chlen= my_ismbchar(cs, src, srcend)))
+    {
+      if (dstlen < chlen)
+        break; /* purecov: inspected */
+      memcpy(dst, src, chlen);
+      src+= chlen;
+      dst+= chlen;
+      dstlen-= chlen;
+    }
+    else if (*src & 0x80)
+    {
+      if (dstlen < 4)
+        break; /* purecov: inspected */
+      *dst++= '\\';
+      *dst++= 'x';
+      *dst++= _dig_vec_upper[((unsigned char) *src) >> 4];
+      *dst++= _dig_vec_upper[((unsigned char) *src) & 15];
+      src++;
+      dstlen-= 4;
+    }
+    else
+    {
+      if (dstlen < 1)
+        break; /* purecov: inspected */
+      *dst++= *src++;
+      dstlen--;
+    }
+  }
+  return dst - dst0;
+}
+
 /*
   copy a string,
   with optional character set conversion,
diff --git a/sql/sql_string.h b/sql/sql_string.h
index 128ed749b5f..b4d76a1779a 100644
--- a/sql/sql_string.h
+++ b/sql/sql_string.h
@@ -37,6 +37,9 @@ uint32 well_formed_copy_nchars(CHARSET_INFO *to_cs,
                                const char **well_formed_error_pos,
                                const char **cannot_convert_error_pos,
                                const char **from_end_pos);
+size_t my_copy_with_hex_escaping(CHARSET_INFO *cs,
+                                 char *dst, size_t dstlen,
+                                 const char *src, size_t srclen);
 
 class String
 {
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index c62d0545fda..b42045446d3 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1226,8 +1226,12 @@ uint build_table_shadow_filename(char *buff, size_t bufflen,
     flags                  Flags as defined below
       WFRM_INITIAL_WRITE        If set we need to prepare table before
                                 creating the frm file
-      WFRM_CREATE_HANDLER_FILES If set we need to create the handler file as
-                                part of the creation of the frm file
+      WFRM_INSTALL_SHADOW       If set we should install the new frm
+      WFRM_KEEP_SHARE           If set we know that the share is to be
+                                retained and thus we should ensure share
+                                object is correct, if not set we don't
+                                set the new partition syntax string since
+                                we know the share object is destroyed.
       WFRM_PACK_FRM             If set we should pack the frm file and delete
                                 the frm file
 
@@ -1370,7 +1374,7 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
       goto err;
     }
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-    if (part_info)
+    if (part_info && (flags & WFRM_KEEP_SHARE))
     {
       TABLE_SHARE *share= lpt->table->s;
       char *tmp_part_syntax_str;
@@ -3322,8 +3326,9 @@ bool mysql_create_table_no_lock(THD *thd,
         }
       }
     }
-    DBUG_PRINT("info", ("db_type = %d",
-                         ha_legacy_type(part_info->default_engine_type)));
+    DBUG_PRINT("info", ("db_type = %s create_info->db_type = %s",
+             ha_resolve_storage_engine_name(part_info->default_engine_type),
+             ha_resolve_storage_engine_name(create_info->db_type)));
     if (part_info->check_partition_info(thd, &engine_type, file,
                                         create_info, TRUE))
       goto err;
@@ -3347,8 +3352,8 @@ bool mysql_create_table_no_lock(THD *thd,
         The handler assigned to the table cannot handle partitioning.
         Assign the partition handler as the handler of the table.
       */
-      DBUG_PRINT("info", ("db_type: %d",
-                          ha_legacy_type(create_info->db_type)));
+      DBUG_PRINT("info", ("db_type: %s",
+                        ha_resolve_storage_engine_name(create_info->db_type)));
       delete file;
       create_info->db_type= partition_hton;
       if (!(file= get_ha_partition(part_info)))
@@ -3511,8 +3516,36 @@ bool mysql_create_table_no_lock(THD *thd,
   thd_proc_info(thd, "creating table");
   create_info->table_existed= 0;		// Mark that table is created
 
-  if (thd->variables.sql_mode & MODE_NO_DIR_IN_CREATE)
+#ifdef HAVE_READLINK
+  if (test_if_data_home_dir(create_info->data_file_name))
+  {
+    my_error(ER_WRONG_ARGUMENTS, MYF(0), "DATA DIRECTORY");
+    goto unlock_and_end;
+  }
+  if (test_if_data_home_dir(create_info->index_file_name))
+  {
+    my_error(ER_WRONG_ARGUMENTS, MYF(0), "INDEX DIRECTORY");
+    goto unlock_and_end;
+  }
+
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  if (check_partition_dirs(thd->lex->part_info))
+  {
+    goto unlock_and_end;
+  }
+#endif /* WITH_PARTITION_STORAGE_ENGINE */
+
+  if (!my_use_symdir || (thd->variables.sql_mode & MODE_NO_DIR_IN_CREATE))
+#endif /* HAVE_READLINK */
+  {
+    if (create_info->data_file_name)
+      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0,
+                   "DATA DIRECTORY option ignored");
+    if (create_info->index_file_name)
+      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0,
+                   "INDEX DIRECTORY option ignored");
     create_info->data_file_name= create_info->index_file_name= 0;
+  }
   create_info->table_options=db_options;
 
   path[path_length - reg_ext_length]= '\0'; // Remove .frm extension
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 46a0f502aeb..5e28bdb45e8 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -4357,7 +4357,7 @@ create_table_option:
             Lex->create_info.row_type= $3;
             Lex->create_info.used_fields|= HA_CREATE_USED_ROW_FORMAT;
           }
-        | UNION_SYM opt_equal '(' table_list ')'
+        | UNION_SYM opt_equal '(' opt_table_list ')'
           {
             /* Move the union list to the merge_list */
             LEX *lex=Lex;
diff --git a/sql/table.cc b/sql/table.cc
index cacb3a94582..100821f28d5 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -892,26 +892,31 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
                             ha_legacy_type(share->db_type())));
       }
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-      else
+      else if (str_db_type_length == 9 &&
+               !strncmp((char *) next_chunk + 2, "partition", 9))
       {
-        LEX_STRING pname= { C_STRING_WITH_LEN( "partition" ) };
-        if (str_db_type_length == pname.length &&
-            !strncmp((char *) next_chunk + 2, pname.str, pname.length))
-        {
-          /*
-            Use partition handler
-            tmp_plugin is locked with a local lock.
-            we unlock the old value of share->db_plugin before
-            replacing it with a globally locked version of tmp_plugin
-          */
-          plugin_unlock(NULL, share->db_plugin);
-          share->db_plugin= ha_lock_engine(NULL, partition_hton);
-          DBUG_PRINT("info", ("setting dbtype to '%.*s' (%d)",
-                              str_db_type_length, next_chunk + 2,
-                              ha_legacy_type(share->db_type())));
-        }
+        /*
+          Use partition handler
+          tmp_plugin is locked with a local lock.
+          we unlock the old value of share->db_plugin before
+          replacing it with a globally locked version of tmp_plugin
+        */
+        plugin_unlock(NULL, share->db_plugin);
+        share->db_plugin= ha_lock_engine(NULL, partition_hton);
+        DBUG_PRINT("info", ("setting dbtype to '%.*s' (%d)",
+                            str_db_type_length, next_chunk + 2,
+                            ha_legacy_type(share->db_type())));
       }
 #endif
+      else if (!tmp_plugin)
+      {
+        /* purecov: begin inspected */
+        error= 8;
+        my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), name.str);
+        my_free(buff, MYF(0));
+        goto err;
+        /* purecov: end */
+      }
       next_chunk+= str_db_type_length + 2;
     }
     if (next_chunk + 5 < buff_end)
@@ -2200,6 +2205,8 @@ void open_table_error(TABLE_SHARE *share, int error, int db_errno, int errarg)
                     "of MySQL and cannot be read", 
                     MYF(0), buff);
     break;
+  case 8:
+    break;
   default:				/* Better wrong error than none */
   case 4:
     strxmov(buff, share->normalized_path.str, reg_ext, NullS);