Manual merge from mysql-trunk-merge.

Conflicts:
  - include/my_no_pthread.h
  - mysql-test/r/sp-ucs2.result
  - sql/log.cc
  - sql/sql_acl.cc
  - sql/sql_yacc.yy

32 files changed, 760 insertions, 398 deletions

diff --git a/sql/item.h b/sql/item.h
index c6ef5bd349a..44892deb2c2 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -968,6 +968,15 @@ public:
   virtual Item *equal_fields_propagator(uchar * arg) { return this; }
   virtual bool set_no_const_sub(uchar *arg) { return FALSE; }
   virtual Item *replace_equal_field(uchar * arg) { return this; }
+  /*
+    Check if an expression value depends on the current timezone. Used by
+    partitioning code to reject timezone-dependent expressions in a
+    (sub)partitioning function.
+  */
+  virtual bool is_timezone_dependent_processor(uchar *bool_arg)
+  {
+    return FALSE;
+  }
 
   /*
     For SP local variable returns pointer to Item representing its
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 4a1077179a9..40e236a37f8 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -957,40 +957,9 @@ int Arg_comparator::set_cmp_func(Item_result_field *owner_arg,
     if (agg_item_set_converter(coll, owner->func_name(),
                                b, 1, MY_COLL_CMP_CONV, 1))
       return 1;
-  } else if (type != ROW_RESULT && ((*a)->field_type() == MYSQL_TYPE_YEAR ||
-             (*b)->field_type() == MYSQL_TYPE_YEAR))
-  {
-    is_nulls_eq= is_owner_equal_func();
-    year_as_datetime= FALSE;
-
-    if ((*a)->is_datetime())
-    {
-      year_as_datetime= TRUE;
-      get_value_a_func= &get_datetime_value;
-    } else if ((*a)->field_type() == MYSQL_TYPE_YEAR)
-      get_value_a_func= &get_year_value;
-    else
-    {
-      /*
-        Because convert_constant_item is called only for EXECUTE in PS mode
-        the value of get_value_x_func set in PREPARE might be not
-        valid for EXECUTE.
-      */
-      get_value_a_func= NULL;
-    }
-
-    if ((*b)->is_datetime())
-    {
-      year_as_datetime= TRUE;
-      get_value_b_func= &get_datetime_value;
-    } else if ((*b)->field_type() == MYSQL_TYPE_YEAR)
-      get_value_b_func= &get_year_value;
-    else
-      get_value_b_func= NULL;
-
-    func= &Arg_comparator::compare_year;
-    return 0;
   }
+  else if (try_year_cmp_func(type))
+    return 0;
 
   a= cache_converted_constant(thd, a, &a_cache, type);
   b= cache_converted_constant(thd, b, &b_cache, type);
@@ -998,6 +967,45 @@ int Arg_comparator::set_cmp_func(Item_result_field *owner_arg,
 }
 
 
+/*
+  Helper function to call from Arg_comparator::set_cmp_func()
+*/
+
+bool Arg_comparator::try_year_cmp_func(Item_result type)
+{
+  if (type == ROW_RESULT)
+    return FALSE;
+
+  bool a_is_year= (*a)->field_type() == MYSQL_TYPE_YEAR;
+  bool b_is_year= (*b)->field_type() == MYSQL_TYPE_YEAR;
+
+  if (!a_is_year && !b_is_year)
+    return FALSE;
+
+  if (a_is_year && b_is_year)
+  {
+    get_value_a_func= &get_year_value;
+    get_value_b_func= &get_year_value;
+  }
+  else if (a_is_year && (*b)->is_datetime())
+  {
+    get_value_a_func= &get_year_value;
+    get_value_b_func= &get_datetime_value;
+  }
+  else if (b_is_year && (*a)->is_datetime())
+  {
+    get_value_b_func= &get_year_value;
+    get_value_a_func= &get_datetime_value;
+  }
+  else
+    return FALSE;
+
+  is_nulls_eq= is_owner_equal_func();
+  func= &Arg_comparator::compare_datetime;
+
+  return TRUE;
+}
+
 /**
   Convert and cache a constant.
 
@@ -1024,7 +1032,7 @@ Item** Arg_comparator::cache_converted_constant(THD *thd, Item **value,
       (*value)->const_item() && type != (*value)->result_type())
   {
     Item_cache *cache= Item_cache::get_cache(*value, type);
-    cache->store(*value);
+    cache->setup(*value);
     *cache_item= cache;
     return cache_item;
   }
@@ -1148,7 +1156,7 @@ get_datetime_value(THD *thd, Item ***item_arg, Item **cache_arg,
 
 
 /*
-  Retrieves YEAR value of 19XX form from given item.
+  Retrieves YEAR value of 19XX-00-00 00:00:00 form from given item.
 
   SYNOPSIS
     get_year_value()
@@ -1160,7 +1168,9 @@ get_datetime_value(THD *thd, Item ***item_arg, Item **cache_arg,
 
   DESCRIPTION
     Retrieves the YEAR value of 19XX form from given item for comparison by the
-    compare_year() function.
+    compare_datetime() function.
+    Converts year to DATETIME of form YYYY-00-00 00:00:00 for the compatibility
+    with the get_datetime_value function result.
 
   RETURN
     obtained value
@@ -1187,6 +1197,9 @@ get_year_value(THD *thd, Item ***item_arg, Item **cache_arg,
   if (value <= 1900)
     value+= 1900;
 
+  /* Convert year to DATETIME of form YYYY-00-00 00:00:00 (YYYY0000000000). */
+  value*= 10000000000LL;
+
   return value;
 }
 
@@ -1616,67 +1629,6 @@ int Arg_comparator::compare_e_row()
 }
 
 
-/**
-  Compare values as YEAR.
-
-  @details
-    Compare items as YEAR for EQUAL_FUNC and for other comparison functions.
-    The YEAR values of form 19XX are obtained with help of the get_year_value()
-    function.
-    If one of arguments is of DATE/DATETIME type its value is obtained
-    with help of the get_datetime_value function. In this case YEAR values
-    prior to comparison are converted to the ulonglong YYYY-00-00 00:00:00
-    DATETIME form.
-    If an argument type neither YEAR nor DATE/DATEIME then val_int function
-    is used to obtain value for comparison.
-
-  RETURN
-    If is_nulls_eq is TRUE:
-       1    if items are equal or both are null
-       0    otherwise
-    If is_nulls_eq is FALSE:
-      -1   a < b
-       0   a == b or at least one of items is null
-       1   a > b
-    See the table:
-    is_nulls_eq | 1 | 1 | 1 | 1 | 0 | 0 | 0 | 0 |
-    a_is_null   | 1 | 0 | 1 | 0 | 1 | 0 | 1 | 0 |
-    b_is_null   | 1 | 1 | 0 | 0 | 1 | 1 | 0 | 0 |
-    result      | 1 | 0 | 0 |0/1| 0 | 0 | 0 |-1/0/1|
-*/
-
-int Arg_comparator::compare_year()
-{
-  bool a_is_null, b_is_null;
-  ulonglong val1= get_value_a_func ?
-                    (*get_value_a_func)(thd, &a, &a_cache, *b, &a_is_null) :
-                    (*a)->val_int();
-  ulonglong val2= get_value_b_func ?
-                    (*get_value_b_func)(thd, &b, &b_cache, *a, &b_is_null) :
-                    (*b)->val_int();
-  if (!(*a)->null_value)
-  {
-    if (!(*b)->null_value)
-    {
-      if (set_null)
-        owner->null_value= 0;
-      /* Convert year to DATETIME of form YYYY-00-00 00:00:00 when necessary. */
-      if((*a)->field_type() == MYSQL_TYPE_YEAR && year_as_datetime)
-        val1*= 10000000000LL;
-      if((*b)->field_type() == MYSQL_TYPE_YEAR && year_as_datetime)
-        val2*= 10000000000LL;
-
-      if (val1 < val2)	return is_nulls_eq ? 0 : -1;
-      if (val1 == val2)   return is_nulls_eq ? 1 : 0;
-      return is_nulls_eq ? 0 : 1;
-    }
-  }
-  if (set_null)
-    owner->null_value= is_nulls_eq ? 0 : 1;
-  return (is_nulls_eq && (*a)->null_value == (*b)->null_value) ? 1 : 0;
-}
-
-
 void Item_func_truth::fix_length_and_dec()
 {
   maybe_null= 0;
@@ -4292,7 +4244,7 @@ Item *Item_cond::compile(Item_analyzer analyzer, uchar **arg_p,
     uchar *arg_v= *arg_p;
     Item *new_item= item->compile(analyzer, &arg_v, transformer, arg_t);
     if (new_item && new_item != item)
-      li.replace(new_item);
+      current_thd->change_item_tree(li.ref(), new_item);
   }
   return Item_func::transform(transformer, arg_t);
 }
@@ -5293,7 +5245,8 @@ Item *Item_bool_rowready_func2::negated_item()
 }
 
 Item_equal::Item_equal(Item_field *f1, Item_field *f2)
-  : Item_bool_func(), const_item(0), eval_item(0), cond_false(0)
+  : Item_bool_func(), const_item(0), eval_item(0), cond_false(0),
+    compare_as_dates(FALSE)
 {
   const_item_cache= 0;
   fields.push_back(f1);
@@ -5306,6 +5259,7 @@ Item_equal::Item_equal(Item *c, Item_field *f)
   const_item_cache= 0;
   fields.push_back(f);
   const_item= c;
+  compare_as_dates= f->is_datetime();
 }
 
 
@@ -5320,9 +5274,45 @@ Item_equal::Item_equal(Item_equal *item_equal)
     fields.push_back(item);
   }
   const_item= item_equal->const_item;
+  compare_as_dates= item_equal->compare_as_dates;
   cond_false= item_equal->cond_false;
 }
 
+
+void Item_equal::compare_const(Item *c)
+{
+  if (compare_as_dates)
+  {
+    cmp.set_datetime_cmp_func(this, &c, &const_item);
+    cond_false= cmp.compare();
+  }
+  else
+  {
+    Item_func_eq *func= new Item_func_eq(c, const_item);
+    func->set_cmp_func();
+    func->quick_fix_field();
+    cond_false= !func->val_int();
+  }
+  if (cond_false)
+    const_item_cache= 1;
+}
+
+
+void Item_equal::add(Item *c, Item_field *f)
+{
+  if (cond_false)
+    return;
+  if (!const_item)
+  {
+    DBUG_ASSERT(f);
+    const_item= c;
+    compare_as_dates= f->is_datetime();
+    return;
+  }
+  compare_const(c);
+}
+
+
 void Item_equal::add(Item *c)
 {
   if (cond_false)
@@ -5332,11 +5322,7 @@ void Item_equal::add(Item *c)
     const_item= c;
     return;
   }
-  Item_func_eq *func= new Item_func_eq(c, const_item);
-  func->set_cmp_func();
-  func->quick_fix_field();
-  if ((cond_false= !func->val_int()))
-    const_item_cache= 1;
+  compare_const(c);
 }
 
 void Item_equal::add(Item_field *f)
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index 7805411dfa2..0e23ba64843 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -45,24 +45,22 @@ class Arg_comparator: public Sql_alloc
   bool is_nulls_eq;                // TRUE <=> compare for the EQUAL_FUNC
   bool set_null;                   // TRUE <=> set owner->null_value
                                    //   when one of arguments is NULL.
-  bool year_as_datetime;           // TRUE <=> convert YEAR value to
-                                   // the YYYY-00-00 00:00:00 DATETIME
-                                   // format. See compare_year.
   enum enum_date_cmp_type { CMP_DATE_DFLT= 0, CMP_DATE_WITH_DATE,
                             CMP_DATE_WITH_STR, CMP_STR_WITH_DATE };
   longlong (*get_value_a_func)(THD *thd, Item ***item_arg, Item **cache_arg,
                                Item *warn_item, bool *is_null);
   longlong (*get_value_b_func)(THD *thd, Item ***item_arg, Item **cache_arg,
                                Item *warn_item, bool *is_null);
+  bool try_year_cmp_func(Item_result type);
 public:
   DTCollation cmp_collation;
   /* Allow owner function to use string buffers. */
   String value1, value2;
 
   Arg_comparator(): thd(0), a_cache(0), b_cache(0), set_null(0),
-    year_as_datetime(0), get_value_a_func(0), get_value_b_func(0) {};
+    get_value_a_func(0), get_value_b_func(0) {};
   Arg_comparator(Item **a1, Item **a2): a(a1), b(a2), thd(0),
-    a_cache(0), b_cache(0), set_null(0), year_as_datetime(0),
+    a_cache(0), b_cache(0), set_null(0),
     get_value_a_func(0), get_value_b_func(0) {};
 
   int set_compare_func(Item_result_field *owner, Item_result type);
@@ -104,7 +102,6 @@ public:
   int compare_real_fixed();
   int compare_e_real_fixed();
   int compare_datetime();        // compare args[0] & args[1] as DATETIMEs
-  int compare_year();
 
   static enum enum_date_cmp_type can_compare_as_dates(Item *a, Item *b,
                                                       ulonglong *const_val_arg);
@@ -1586,7 +1583,9 @@ class Item_equal: public Item_bool_func
   List<Item_field> fields; /* list of equal field items                    */
   Item *const_item;        /* optional constant item equal to fields items */
   cmp_item *eval_item;
+  Arg_comparator cmp;
   bool cond_false;
+  bool compare_as_dates;
 public:
   inline Item_equal()
     : Item_bool_func(), const_item(0), eval_item(0), cond_false(0)
@@ -1595,6 +1594,8 @@ public:
   Item_equal(Item *c, Item_field *f);
   Item_equal(Item_equal *item_equal);
   inline Item* get_const() { return const_item; }
+  void compare_const(Item *c);
+  void add(Item *c, Item_field *f);
   void add(Item *c);
   void add(Item_field *f);
   uint members();
diff --git a/sql/item_func.h b/sql/item_func.h
index f22bc0c301c..e3690232904 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -192,6 +192,29 @@ public:
     null_value=1;
     return 0.0;
   }
+  bool has_timestamp_args()
+  {
+    DBUG_ASSERT(fixed == TRUE);
+    for (uint i= 0; i < arg_count; i++)
+    {
+      if (args[i]->type() == Item::FIELD_ITEM &&
+          args[i]->field_type() == MYSQL_TYPE_TIMESTAMP)
+        return TRUE;
+    }
+    return FALSE;
+  }
+  /*
+    We assume the result of any function that has a TIMESTAMP argument to be
+    timezone-dependent, since a TIMESTAMP value in both numeric and string
+    contexts is interpreted according to the current timezone.
+    The only exception is UNIX_TIMESTAMP() which returns the internal
+    representation of a TIMESTAMP argument verbatim, and thus does not depend on
+    the timezone.
+   */
+  virtual bool is_timezone_dependent_processor(uchar *bool_arg)
+  {
+    return has_timestamp_args();
+  }
 };
 
 
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 95e28e791ec..d63226f2dbb 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -1720,68 +1720,65 @@ String *Item_func_encrypt::val_str(String *str)
 #endif	/* HAVE_CRYPT */
 }
 
+bool Item_func_encode::seed()
+{
+  char buf[80];
+  ulong rand_nr[2];
+  String *key, tmp(buf, sizeof(buf), system_charset_info);
+
+  if (!(key= args[1]->val_str(&tmp)))
+    return TRUE;
+
+  hash_password(rand_nr, key->ptr(), key->length());
+  sql_crypt.init(rand_nr);
+
+  return FALSE;
+}
+
 void Item_func_encode::fix_length_and_dec()
 {
   max_length=args[0]->max_length;
   maybe_null=args[0]->maybe_null || args[1]->maybe_null;
   collation.set(&my_charset_bin);
+  /* Precompute the seed state if the item is constant. */
+  seeded= args[1]->const_item() &&
+          (args[1]->result_type() == STRING_RESULT) && !seed();
 }
 
 String *Item_func_encode::val_str(String *str)
 {
   String *res;
-  char pw_buff[80];
-  String tmp_pw_value(pw_buff, sizeof(pw_buff), system_charset_info);
-  String *password;
   DBUG_ASSERT(fixed == 1);
 
   if (!(res=args[0]->val_str(str)))
   {
-    null_value=1; /* purecov: inspected */
-    return 0; /* purecov: inspected */
+    null_value= 1;
+    return NULL;
   }
 
-  if (!(password=args[1]->val_str(& tmp_pw_value)))
+  if (!seeded && seed())
   {
-    null_value=1;
-    return 0;
+    null_value= 1;
+    return NULL;
   }
 
-  null_value=0;
-  res=copy_if_not_alloced(str,res,res->length());
-  SQL_CRYPT sql_crypt(password->ptr(), password->length());
-  sql_crypt.init();
-  sql_crypt.encode((char*) res->ptr(),res->length());
-  res->set_charset(&my_charset_bin);
+  null_value= 0;
+  res= copy_if_not_alloced(str, res, res->length());
+  transform(res);
+  sql_crypt.reinit();
+
   return res;
 }
 
-String *Item_func_decode::val_str(String *str)
+void Item_func_encode::transform(String *res)
 {
-  String *res;
-  char pw_buff[80];
-  String tmp_pw_value(pw_buff, sizeof(pw_buff), system_charset_info);
-  String *password;
-  DBUG_ASSERT(fixed == 1);
-
-  if (!(res=args[0]->val_str(str)))
-  {
-    null_value=1; /* purecov: inspected */
-    return 0; /* purecov: inspected */
-  }
-
-  if (!(password=args[1]->val_str(& tmp_pw_value)))
-  {
-    null_value=1;
-    return 0;
-  }
+  sql_crypt.encode((char*) res->ptr(),res->length());
+  res->set_charset(&my_charset_bin);
+}
 
-  null_value=0;
-  res=copy_if_not_alloced(str,res,res->length());
-  SQL_CRYPT sql_crypt(password->ptr(), password->length());
-  sql_crypt.init();
+void Item_func_decode::transform(String *res)
+{
   sql_crypt.decode((char*) res->ptr(),res->length());
-  return res;
 }
 
 
diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h
index 2233679e40c..ced164554e4 100644
--- a/sql/item_strfunc.h
+++ b/sql/item_strfunc.h
@@ -354,12 +354,22 @@ public:
 
 class Item_func_encode :public Item_str_func
 {
+private:
+  /** Whether the PRNG has already been seeded. */
+  bool seeded;
+protected:
+  SQL_CRYPT sql_crypt;
 public:
   Item_func_encode(Item *a, Item *seed):
     Item_str_func(a, seed) {}
   String *val_str(String *);
   void fix_length_and_dec();
   const char *func_name() const { return "encode"; }
+protected:
+  virtual void transform(String *);
+private:
+  /** Provide a seed for the PRNG sequence. */
+  bool seed();
 };
 
 
@@ -367,8 +377,9 @@ class Item_func_decode :public Item_func_encode
 {
 public:
   Item_func_decode(Item *a, Item *seed): Item_func_encode(a, seed) {}
-  String *val_str(String *);
   const char *func_name() const { return "decode"; }
+protected:
+  void transform(String *);
 };
 
 
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index 56c46efb1a8..392bc936f17 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -2609,9 +2609,9 @@ void Item_char_typecast::fix_length_and_dec()
                        from_cs != &my_charset_bin &&
                        cast_cs != &my_charset_bin);
   collation.set(cast_cs, DERIVATION_IMPLICIT);
-  char_length= (cast_length >= 0) ?
-                 cast_length :
-                 args[0]->max_length / args[0]->collation.collation->mbmaxlen;
+  char_length= (cast_length >= 0) ? cast_length :
+                args[0]->max_length /
+                (cast_cs == &my_charset_bin ? 1 : args[0]->collation.collation->mbmaxlen);
   max_length= char_length * cast_cs->mbmaxlen;
 }
 
diff --git a/sql/item_timefunc.h b/sql/item_timefunc.h
index cdd74c8c601..b25812299f0 100644
--- a/sql/item_timefunc.h
+++ b/sql/item_timefunc.h
@@ -326,6 +326,16 @@ public:
   Item_func_unix_timestamp(Item *a) :Item_int_func(a) {}
   longlong val_int();
   const char *func_name() const { return "unix_timestamp"; }
+  bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
+  /*
+    UNIX_TIMESTAMP() depends on the current timezone
+    (and thus may not be used as a partitioning function)
+    when its argument is NOT of the TIMESTAMP type.
+  */
+  bool is_timezone_dependent_processor(uchar *int_arg)
+  {
+    return !has_timestamp_args();
+  }
   void fix_length_and_dec()
   {
     decimals=0;
diff --git a/sql/log.cc b/sql/log.cc
index 9f479a128fc..d390f9caba9 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1979,6 +1979,22 @@ void MYSQL_LOG::init(enum_log_type log_type_arg,
 }
 
 
+bool MYSQL_LOG::init_and_set_log_file_name(const char *log_name,
+                                           const char *new_name,
+                                           enum_log_type log_type_arg,
+                                           enum cache_type io_cache_type_arg)
+{
+  init(log_type_arg, io_cache_type_arg);
+
+  if (new_name && !strmov(log_file_name, new_name))
+    return TRUE;
+  else if (!new_name && generate_new_name(log_file_name, log_name))
+    return TRUE;
+
+  return FALSE;
+}
+
+
 /*
   Open a (new) log file.
 
@@ -2011,17 +2027,14 @@ bool MYSQL_LOG::open(const char *log_name, enum_log_type log_type_arg,
 
   write_error= 0;
 
-  init(log_type_arg, io_cache_type_arg);
-
   if (!(name= my_strdup(log_name, MYF(MY_WME))))
   {
     name= (char *)log_name; // for the error message
     goto err;
   }
 
-  if (new_name)
-    strmov(log_file_name, new_name);
-  else if (generate_new_name(log_file_name, name))
+  if (init_and_set_log_file_name(name, new_name,
+                                 log_type_arg, io_cache_type_arg))
     goto err;
 
   if (io_cache_type == SEQ_READ_APPEND)
@@ -2519,7 +2532,7 @@ MYSQL_BIN_LOG::MYSQL_BIN_LOG(uint *sync_period)
   */
   index_file_name[0] = 0;
   bzero((char*) &index_file, sizeof(index_file));
-  bzero((char*) &purge_temp, sizeof(purge_temp));
+  bzero((char*) &purge_index_file, sizeof(purge_index_file));
 }
 
 /* this is called only once */
@@ -2563,7 +2576,7 @@ void MYSQL_BIN_LOG::init_pthread_objects()
 
 
 bool MYSQL_BIN_LOG::open_index_file(const char *index_file_name_arg,
-                                const char *log_name)
+                                    const char *log_name, bool need_mutex)
 {
   File index_file_nr= -1;
   DBUG_ASSERT(!my_b_inited(&index_file));
@@ -2588,7 +2601,8 @@ bool MYSQL_BIN_LOG::open_index_file(const char *index_file_name_arg,
        init_io_cache(&index_file, index_file_nr,
                      IO_SIZE, WRITE_CACHE,
                      my_seek(index_file_nr,0L,MY_SEEK_END,MYF(0)),
-			0, MYF(MY_WME | MY_WAIT_IF_FULL)))
+			0, MYF(MY_WME | MY_WAIT_IF_FULL)) ||
+      DBUG_EVALUATE_IF("fault_injection_openning_index", 1, 0))
   {
     /*
       TODO: all operations creating/deleting the index file or a log, should
@@ -2599,6 +2613,28 @@ bool MYSQL_BIN_LOG::open_index_file(const char *index_file_name_arg,
       my_close(index_file_nr,MYF(0));
     return TRUE;
   }
+
+#ifdef HAVE_REPLICATION
+  /*
+    Sync the index by purging any binary log file that is not registered.
+    In other words, either purge binary log files that were removed from
+    the index but not purged from the file system due to a crash or purge
+    any binary log file that was created but not register in the index
+    due to a crash.
+  */
+
+  if (set_purge_index_file_name(index_file_name_arg) ||
+      open_purge_index_file(FALSE) ||
+      purge_index_entry(NULL, NULL, need_mutex) ||
+      close_purge_index_file() ||
+      DBUG_EVALUATE_IF("fault_injection_recovering_index", 1, 0))
+  {
+    sql_print_error("MYSQL_BIN_LOG::open_index_file failed to sync the index "
+                    "file.");
+    return TRUE;
+  }
+#endif
+
   return FALSE;
 }
 
@@ -2623,17 +2659,44 @@ bool MYSQL_BIN_LOG::open(const char *log_name,
                          enum cache_type io_cache_type_arg,
                          bool no_auto_events_arg,
                          ulong max_size_arg,
-                         bool null_created_arg)
+                         bool null_created_arg,
+                         bool need_mutex)
 {
   File file= -1;
+
   DBUG_ENTER("MYSQL_BIN_LOG::open");
   DBUG_PRINT("enter",("log_type: %d",(int) log_type_arg));
 
-  write_error=0;
+  if (init_and_set_log_file_name(log_name, new_name, log_type_arg,
+                                 io_cache_type_arg))
+  {
+    sql_print_error("MSYQL_BIN_LOG::open failed to generate new file name.");
+    DBUG_RETURN(1);
+  }
+
+#ifdef HAVE_REPLICATION
+  if (open_purge_index_file(TRUE) ||
+      register_create_index_entry(log_file_name) ||
+      sync_purge_index_file() ||
+      DBUG_EVALUATE_IF("fault_injection_registering_index", 1, 0))
+  {
+    sql_print_error("MSYQL_BIN_LOG::open failed to sync the index file.");
+    DBUG_RETURN(1);
+  }
+  DBUG_EXECUTE_IF("crash_create_non_critical_before_update_index", abort(););
+#endif
+
+  write_error= 0;
 
   /* open the main log file */
-  if (MYSQL_LOG::open(log_name, log_type_arg, new_name, io_cache_type_arg))
+  if (MYSQL_LOG::open(log_name, log_type_arg, new_name,
+                      io_cache_type_arg))
+  {
+#ifdef HAVE_REPLICATION
+    close_purge_index_file();
+#endif
     DBUG_RETURN(1);                            /* all warnings issued */
+  }
 
   init(no_auto_events_arg, max_size_arg);
 
@@ -2659,9 +2722,6 @@ bool MYSQL_BIN_LOG::open(const char *log_name,
       write_file_name_to_index_file= 1;
     }
 
-    DBUG_ASSERT(my_b_inited(&index_file) != 0);
-    reinit_io_cache(&index_file, WRITE_CACHE,
-                    my_b_filelength(&index_file), 0, 0);
     if (need_start_event && !no_auto_events)
     {
       /*
@@ -2719,23 +2779,44 @@ bool MYSQL_BIN_LOG::open(const char *log_name,
 
     if (write_file_name_to_index_file)
     {
+#ifdef HAVE_REPLICATION
+      DBUG_EXECUTE_IF("crash_create_critical_before_update_index", abort(););
+#endif
+
+      DBUG_ASSERT(my_b_inited(&index_file) != 0);
+      reinit_io_cache(&index_file, WRITE_CACHE,
+                      my_b_filelength(&index_file), 0, 0);
       /*
         As this is a new log file, we write the file name to the index
         file. As every time we write to the index file, we sync it.
       */
-      if (my_b_write(&index_file, (uchar*) log_file_name,
-		     strlen(log_file_name)) ||
-	  my_b_write(&index_file, (uchar*) "\n", 1) ||
-	  flush_io_cache(&index_file) ||
+      if (DBUG_EVALUATE_IF("fault_injection_updating_index", 1, 0) ||
+          my_b_write(&index_file, (uchar*) log_file_name,
+                     strlen(log_file_name)) ||
+          my_b_write(&index_file, (uchar*) "\n", 1) ||
+          flush_io_cache(&index_file) ||
           my_sync(index_file.file, MYF(MY_WME)))
-	goto err;
+        goto err;
+
+#ifdef HAVE_REPLICATION
+      DBUG_EXECUTE_IF("crash_create_after_update_index", abort(););
+#endif
     }
   }
   log_state= LOG_OPENED;
 
+#ifdef HAVE_REPLICATION
+  close_purge_index_file();
+#endif
+
   DBUG_RETURN(0);
 
 err:
+#ifdef HAVE_REPLICATION
+  if (is_inited_purge_index_file())
+    purge_index_entry(NULL, NULL, need_mutex);
+  close_purge_index_file();
+#endif
   sql_print_error("Could not use %s for logging (error %d). \
 Turning logging off for the whole duration of the MySQL server process. \
 To turn it on again: fix the cause, \
@@ -2995,7 +3076,15 @@ bool MYSQL_BIN_LOG::reset_logs(THD* thd)
   name=0;					// Protect against free
   close(LOG_CLOSE_TO_BE_OPENED);
 
-  /* First delete all old log files */
+  /*
+    First delete all old log files and then update the index file.
+    As we first delete the log files and do not use sort of logging,
+    a crash may lead to an inconsistent state where the index has
+    references to non-existent files.
+
+    We need to invert the steps and use the purge_index_file methods
+    in order to make the operation safe.
+  */
 
   if ((err= find_log_pos(&linfo, NullS, 0)) != 0)
   {
@@ -3066,8 +3155,8 @@ bool MYSQL_BIN_LOG::reset_logs(THD* thd)
   }
   if (!thd->slave_thread)
     need_start_event=1;
-  if (!open_index_file(index_file_name, 0))
-    open(save_name, log_type, 0, io_cache_type, no_auto_events, max_size, 0);
+  if (!open_index_file(index_file_name, 0, FALSE))
+    open(save_name, log_type, 0, io_cache_type, no_auto_events, max_size, 0, FALSE);
   my_free((uchar*) save_name, MYF(0));
 
 err:
@@ -3256,7 +3345,7 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
                           bool need_update_threads, 
                           ulonglong *decrease_log_space)
 {
-  int error;
+  int error= 0;
   bool exit_loop= 0;
   LOG_INFO log_info;
   THD *thd= current_thd;
@@ -3267,33 +3356,15 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
     pthread_mutex_lock(&LOCK_index);
   if ((error=find_log_pos(&log_info, to_log, 0 /*no mutex*/))) 
   {
-    sql_print_error("MYSQL_LOG::purge_logs was called with file %s not "
+    sql_print_error("MYSQL_BIN_LOG::purge_logs was called with file %s not "
                     "listed in the index.", to_log);
     goto err;
   }
 
-  /*
-    For crash recovery reasons the index needs to be updated before
-    any files are deleted. Move files to be deleted into a temp file
-    to be processed after the index is updated.
-  */
-  if (!my_b_inited(&purge_temp))
-  {
-    if ((error=open_cached_file(&purge_temp, mysql_tmpdir, TEMP_PREFIX,
-                                DISK_BUFFER_SIZE, MYF(MY_WME))))
-    {
-      sql_print_error("MYSQL_LOG::purge_logs failed to open purge_temp");
-      goto err;
-    }
-  }
-  else
+  if ((error= open_purge_index_file(TRUE)))
   {
-    if ((error=reinit_io_cache(&purge_temp, WRITE_CACHE, 0, 0, 1)))
-    {
-      sql_print_error("MYSQL_LOG::purge_logs failed to reinit purge_temp "
-                      "for write");
-      goto err;
-    }
+    sql_print_error("MYSQL_BIN_LOG::purge_logs failed to sync the index file.");
+    goto err;
   }
 
   /*
@@ -3303,51 +3374,177 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
   if ((error=find_log_pos(&log_info, NullS, 0 /*no mutex*/)))
     goto err;
   while ((strcmp(to_log,log_info.log_file_name) || (exit_loop=included)) &&
+         !is_active(log_info.log_file_name) &&
          !log_in_use(log_info.log_file_name))
   {
-    if ((error=my_b_write(&purge_temp, (const uchar*)log_info.log_file_name,
-                          strlen(log_info.log_file_name))) ||
-        (error=my_b_write(&purge_temp, (const uchar*)"\n", 1)))
+    if ((error= register_purge_index_entry(log_info.log_file_name)))
     {
-      sql_print_error("MYSQL_LOG::purge_logs failed to copy %s to purge_temp",
+      sql_print_error("MYSQL_BIN_LOG::purge_logs failed to copy %s to register file.",
                       log_info.log_file_name);
       goto err;
     }
 
     if (find_next_log(&log_info, 0) || exit_loop)
       break;
- }
+  }
+
+  DBUG_EXECUTE_IF("crash_purge_before_update_index", abort(););
+
+  if ((error= sync_purge_index_file()))
+  {
+    sql_print_error("MSYQL_BIN_LOG::purge_logs failed to flush register file.");
+    goto err;
+  }
 
   /* We know how many files to delete. Update index file. */
   if ((error=update_log_index(&log_info, need_update_threads)))
   {
-    sql_print_error("MSYQL_LOG::purge_logs failed to update the index file");
+    sql_print_error("MSYQL_BIN_LOG::purge_logs failed to update the index file");
     goto err;
   }
 
-  DBUG_EXECUTE_IF("crash_after_update_index", abort(););
+  DBUG_EXECUTE_IF("crash_purge_critical_after_update_index", abort(););
+
+err:
+  /* Read each entry from purge_index_file and delete the file. */
+  if (is_inited_purge_index_file() &&
+      (error= purge_index_entry(thd, decrease_log_space, FALSE)))
+    sql_print_error("MSYQL_BIN_LOG::purge_logs failed to process registered files"
+                    " that would be purged.");
+  close_purge_index_file();
+
+  DBUG_EXECUTE_IF("crash_purge_non_critical_after_update_index", abort(););
+
+  if (need_mutex)
+    pthread_mutex_unlock(&LOCK_index);
+  DBUG_RETURN(error);
+}
+
+int MYSQL_BIN_LOG::set_purge_index_file_name(const char *base_file_name)
+{
+  int error= 0;
+  DBUG_ENTER("MYSQL_BIN_LOG::set_purge_index_file_name");
+  if (fn_format(purge_index_file_name, base_file_name, mysql_data_home,
+                ".~rec~", MYF(MY_UNPACK_FILENAME | MY_SAFE_PATH |
+                              MY_REPLACE_EXT)) == NULL)
+  {
+    error= 1;
+    sql_print_error("MYSQL_BIN_LOG::set_purge_index_file_name failed to set "
+                      "file name.");
+  }
+  DBUG_RETURN(error);
+}
+
+int MYSQL_BIN_LOG::open_purge_index_file(bool destroy)
+{
+  int error= 0;
+  File file= -1;
 
-  /* Switch purge_temp for read. */
-  if ((error=reinit_io_cache(&purge_temp, READ_CACHE, 0, 0, 0)))
+  DBUG_ENTER("MYSQL_BIN_LOG::open_purge_index_file");
+
+  if (destroy)
+    close_purge_index_file();
+
+  if (!my_b_inited(&purge_index_file))
   {
-    sql_print_error("MSYQL_LOG::purge_logs failed to reinit purge_temp "
+    if ((file= my_open(purge_index_file_name, O_RDWR | O_CREAT | O_BINARY,
+                       MYF(MY_WME | ME_WAITTANG))) < 0  ||
+        init_io_cache(&purge_index_file, file, IO_SIZE,
+                      (destroy ? WRITE_CACHE : READ_CACHE),
+                      0, 0, MYF(MY_WME | MY_NABP | MY_WAIT_IF_FULL)))
+    {
+      error= 1;
+      sql_print_error("MYSQL_BIN_LOG::open_purge_index_file failed to open register "
+                      " file.");
+    }
+  }
+  DBUG_RETURN(error);
+}
+
+int MYSQL_BIN_LOG::close_purge_index_file()
+{
+  int error= 0;
+
+  DBUG_ENTER("MYSQL_BIN_LOG::close_purge_index_file");
+
+  if (my_b_inited(&purge_index_file))
+  {
+    end_io_cache(&purge_index_file);
+    error= my_close(purge_index_file.file, MYF(0));
+  }
+  my_delete(purge_index_file_name, MYF(0));
+  bzero((char*) &purge_index_file, sizeof(purge_index_file));
+
+  DBUG_RETURN(error);
+}
+
+bool MYSQL_BIN_LOG::is_inited_purge_index_file()
+{
+  DBUG_ENTER("MYSQL_BIN_LOG::is_inited_purge_index_file");
+  DBUG_RETURN (my_b_inited(&purge_index_file));
+}
+
+int MYSQL_BIN_LOG::sync_purge_index_file()
+{
+  int error= 0;
+  DBUG_ENTER("MYSQL_BIN_LOG::sync_purge_index_file");
+
+  if ((error= flush_io_cache(&purge_index_file)) ||
+      (error= my_sync(purge_index_file.file, MYF(MY_WME))))
+    DBUG_RETURN(error);
+
+  DBUG_RETURN(error);
+}
+
+int MYSQL_BIN_LOG::register_purge_index_entry(const char *entry)
+{
+  int error= 0;
+  DBUG_ENTER("MYSQL_BIN_LOG::register_purge_index_entry");
+
+  if ((error=my_b_write(&purge_index_file, (const uchar*)entry, strlen(entry))) ||
+      (error=my_b_write(&purge_index_file, (const uchar*)"\n", 1)))
+    DBUG_RETURN (error);
+
+  DBUG_RETURN(error);
+}
+
+int MYSQL_BIN_LOG::register_create_index_entry(const char *entry)
+{
+  DBUG_ENTER("MYSQL_BIN_LOG::register_create_index_entry");
+  DBUG_RETURN(register_purge_index_entry(entry));
+}
+
+int MYSQL_BIN_LOG::purge_index_entry(THD *thd, ulonglong *decrease_log_space,
+                                     bool need_mutex)
+{
+  MY_STAT s;
+  int error= 0;
+  LOG_INFO log_info;
+  LOG_INFO check_log_info;
+
+  DBUG_ENTER("MYSQL_BIN_LOG:purge_index_entry");
+
+  DBUG_ASSERT(my_b_inited(&purge_index_file));
+
+  if ((error=reinit_io_cache(&purge_index_file, READ_CACHE, 0, 0, 0)))
+  {
+    sql_print_error("MSYQL_BIN_LOG::purge_index_entry failed to reinit register file "
                     "for read");
     goto err;
   }
 
-  /* Read each entry from purge_temp and delete the file. */
   for (;;)
   {
     uint length;
 
-    if ((length=my_b_gets(&purge_temp, log_info.log_file_name,
+    if ((length=my_b_gets(&purge_index_file, log_info.log_file_name,
                           FN_REFLEN)) <= 1)
     {
-      if (purge_temp.error)
+      if (purge_index_file.error)
       {
-        error= purge_temp.error;
-        sql_print_error("MSYQL_LOG::purge_logs error %d reading from "
-                        "purge_temp", error);
+        error= purge_index_file.error;
+        sql_print_error("MSYQL_BIN_LOG::purge_index_entry error %d reading from "
+                        "register file.", error);
         goto err;
       }
 
@@ -3358,9 +3555,6 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
     /* Get rid of the trailing '\n' */
     log_info.log_file_name[length-1]= 0;
 
-    ha_binlog_index_purge_file(current_thd, log_info.log_file_name);
-
-    MY_STAT s;
     if (!my_stat(log_info.log_file_name, &s, MYF(0)))
     {
       if (my_errno == ENOENT) 
@@ -3408,64 +3602,92 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
     }
     else
     {
-      DBUG_PRINT("info",("purging %s",log_info.log_file_name));
-      if (!my_delete(log_info.log_file_name, MYF(0)))
-      {
-        if (decrease_log_space)
-          *decrease_log_space-= s.st_size;
-      }
-      else
+      if ((error= find_log_pos(&check_log_info, log_info.log_file_name, need_mutex)))
       {
-        if (my_errno == ENOENT) 
+        if (error != LOG_INFO_EOF)
         {
           if (thd)
           {
             push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                                ER_LOG_PURGE_NO_FILE, ER(ER_LOG_PURGE_NO_FILE),
+                                ER_BINLOG_PURGE_FATAL_ERR,
+                                "a problem with deleting %s and "
+                                "reading the binlog index file",
                                 log_info.log_file_name);
           }
-          sql_print_information("Failed to delete file '%s'",
-                                log_info.log_file_name);
-          my_errno= 0;
+          else
+          {
+            sql_print_information("Failed to delete file '%s' and "
+                                  "read the binlog index file",
+                                  log_info.log_file_name);
+          }
+          goto err;
+        }
+           
+        error= 0;
+        if (!need_mutex)
+        {
+          /*
+            This is to avoid triggering an error in NDB.
+          */
+          ha_binlog_index_purge_file(current_thd, log_info.log_file_name);
+        }
+
+        DBUG_PRINT("info",("purging %s",log_info.log_file_name));
+        if (!my_delete(log_info.log_file_name, MYF(0)))
+        {
+          if (decrease_log_space)
+            *decrease_log_space-= s.st_size;
         }
         else
         {
-          if (thd)
+          if (my_errno == ENOENT)
           {
-            push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                                ER_BINLOG_PURGE_FATAL_ERR,
-                                "a problem with deleting %s; "
-                                "consider examining correspondence "
-                                "of your binlog index file "
-                                "to the actual binlog files",
-                                log_info.log_file_name);
+            if (thd)
+            {
+              push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                                  ER_LOG_PURGE_NO_FILE, ER(ER_LOG_PURGE_NO_FILE),
+                                  log_info.log_file_name);
+            }
+            sql_print_information("Failed to delete file '%s'",
+                                  log_info.log_file_name);
+            my_errno= 0;
           }
           else
           {
-            sql_print_information("Failed to delete file '%s'; "
+            if (thd)
+            {
+              push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
+                                  ER_BINLOG_PURGE_FATAL_ERR,
+                                  "a problem with deleting %s; "
                                   "consider examining correspondence "
                                   "of your binlog index file "
                                   "to the actual binlog files",
                                   log_info.log_file_name);
-          }
-          if (my_errno == EMFILE)
-          {
-            DBUG_PRINT("info",
-                       ("my_errno: %d, set ret = LOG_INFO_EMFILE", my_errno));
-            error= LOG_INFO_EMFILE;
+            }
+            else
+            {
+              sql_print_information("Failed to delete file '%s'; "
+                                    "consider examining correspondence "
+                                    "of your binlog index file "
+                                    "to the actual binlog files",
+                                    log_info.log_file_name);
+            }
+            if (my_errno == EMFILE)
+            {
+              DBUG_PRINT("info",
+                         ("my_errno: %d, set ret = LOG_INFO_EMFILE", my_errno));
+              error= LOG_INFO_EMFILE;
+              goto err;
+            }
+            error= LOG_INFO_FATAL;
             goto err;
           }
-          error= LOG_INFO_FATAL;
-          goto err;
         }
       }
     }
   }
 
 err:
-  close_cached_file(&purge_temp);
-  if (need_mutex)
-    pthread_mutex_unlock(&LOCK_index);
   DBUG_RETURN(error);
 }
 
@@ -3505,7 +3727,8 @@ int MYSQL_BIN_LOG::purge_logs_before_date(time_t purge_time)
     goto err;
 
   while (strcmp(log_file_name, log_info.log_file_name) &&
-	 !log_in_use(log_info.log_file_name))
+	 !is_active(log_info.log_file_name) &&
+         !log_in_use(log_info.log_file_name))
   {
     if (!my_stat(log_info.log_file_name, &stat_area, MYF(0)))
     {
@@ -3514,14 +3737,6 @@ int MYSQL_BIN_LOG::purge_logs_before_date(time_t purge_time)
         /*
           It's not fatal if we can't stat a log file that does not exist.
         */
-        if (thd)
-        {
-          push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                              ER_LOG_PURGE_NO_FILE, ER(ER_LOG_PURGE_NO_FILE),
-                              log_info.log_file_name);
-        }
-        sql_print_information("Failed to execute my_stat on file '%s'",
-                              log_info.log_file_name);
         my_errno= 0;
       }
       else
@@ -3716,9 +3931,9 @@ void MYSQL_BIN_LOG::new_file_impl(bool need_lock)
   */
 
   /* reopen index binlog file, BUG#34582 */
-  if (!open_index_file(index_file_name, 0))
-    open(old_name, log_type, new_name_ptr, 
-         io_cache_type, no_auto_events, max_size, 1);
+  if (!open_index_file(index_file_name, 0, FALSE))
+    open(old_name, log_type, new_name_ptr,
+         io_cache_type, no_auto_events, max_size, 1, FALSE);
   my_free(old_name,MYF(0));
 
 end:
@@ -4380,6 +4595,9 @@ bool general_log_write(THD *thd, enum enum_server_command command,
 
 void MYSQL_BIN_LOG::rotate_and_purge(uint flags)
 {
+#ifdef HAVE_REPLICATION
+  bool check_purge= false;
+#endif
   if (!(flags & RP_LOCK_LOG_IS_ALREADY_LOCKED))
     pthread_mutex_lock(&LOCK_log);
   if ((flags & RP_FORCE_ROTATE) ||
@@ -4387,16 +4605,24 @@ void MYSQL_BIN_LOG::rotate_and_purge(uint flags)
   {
     new_file_without_locking();
 #ifdef HAVE_REPLICATION
-    if (expire_logs_days)
-    {
-      time_t purge_time= my_time(0) - expire_logs_days*24*60*60;
-      if (purge_time >= 0)
-        purge_logs_before_date(purge_time);
-    }
+    check_purge= true;
 #endif
   }
   if (!(flags & RP_LOCK_LOG_IS_ALREADY_LOCKED))
     pthread_mutex_unlock(&LOCK_log);
+
+#ifdef HAVE_REPLICATION
+  /*
+    NOTE: Run purge_logs wo/ holding LOCK_log
+          as it otherwise will deadlock in ndbcluster_binlog_index_purge_file
+  */
+  if (check_purge && expire_logs_days)
+  {
+    time_t purge_time= my_time(0) - expire_logs_days*24*60*60;
+    if (purge_time >= 0)
+      purge_logs_before_date(purge_time);
+  }
+#endif
 }
 
 uint MYSQL_BIN_LOG::next_file_id()
@@ -5674,7 +5900,7 @@ int TC_LOG_BINLOG::open(const char *opt_name)
   if (using_heuristic_recover())
   {
     /* generate a new binlog to mask a corrupted one */
-    open(opt_name, LOG_BIN, 0, WRITE_CACHE, 0, max_binlog_size, 0);
+    open(opt_name, LOG_BIN, 0, WRITE_CACHE, 0, max_binlog_size, 0, TRUE);
     cleanup();
     return 1;
   }
diff --git a/sql/log.h b/sql/log.h
index 9be86b797a7..fa24a2c5803 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -185,6 +185,10 @@ public:
             enum_log_type log_type,
             const char *new_name,
             enum cache_type io_cache_type_arg);
+  bool init_and_set_log_file_name(const char *log_name,
+                                  const char *new_name,
+                                  enum_log_type log_type_arg,
+                                  enum cache_type io_cache_type_arg);
   void init(enum_log_type log_type_arg,
             enum cache_type io_cache_type_arg);
   void close(uint exiting);
@@ -246,14 +250,15 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
   pthread_cond_t update_cond;
   ulonglong bytes_written;
   IO_CACHE index_file;
+  char index_file_name[FN_REFLEN];
   /*
-    purge_temp is a temp file used in purge_logs so that the index file
+    purge_file is a temp file used in purge_logs so that the index file
     can be updated before deleting files from disk, yielding better crash
     recovery. It is created on demand the first time purge_logs is called
     and then reused for subsequent calls. It is cleaned up in cleanup().
   */
-  IO_CACHE purge_temp;
-  char index_file_name[FN_REFLEN];
+  IO_CACHE purge_index_file;
+  char purge_index_file_name[FN_REFLEN];
   /*
      The max size before rotation (usable only if log_type == LOG_BIN: binary
      logs and relay logs).
@@ -375,9 +380,10 @@ public:
             const char *new_name,
 	    enum cache_type io_cache_type_arg,
 	    bool no_auto_events_arg, ulong max_size,
-            bool null_created);
+            bool null_created,
+            bool need_mutex);
   bool open_index_file(const char *index_file_name_arg,
-                       const char *log_name);
+                       const char *log_name, bool need_mutex);
   /* Use this to start writing a new log file */
   void new_file();
 
@@ -423,6 +429,16 @@ public:
                  ulonglong *decrease_log_space);
   int purge_logs_before_date(time_t purge_time);
   int purge_first_log(Relay_log_info* rli, bool included);
+  int set_purge_index_file_name(const char *base_file_name);
+  int open_purge_index_file(bool destroy);
+  bool is_inited_purge_index_file();
+  int close_purge_index_file();
+  int clean_purge_index_file();
+  int sync_purge_index_file();
+  int register_purge_index_entry(const char* entry);
+  int register_create_index_entry(const char* entry);
+  int purge_index_entry(THD *thd, ulonglong *decrease_log_space,
+                        bool need_mutex);
   bool reset_logs(THD* thd);
   void close(uint exiting);
 
diff --git a/sql/log_event.cc b/sql/log_event.cc
index cc2b4667115..85b8c44703d 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -2139,8 +2139,8 @@ void Query_log_event::pack_info(Protocol *protocol)
 /**
   Utility function for the next method (Query_log_event::write()) .
 */
-static void write_str_with_code_and_len(char **dst, const char *src,
-                                        int len, uint code)
+static void write_str_with_code_and_len(uchar **dst, const char *src,
+                                        uint len, uint code)
 {
   /*
     only 1 byte to store the length of catalog, so it should not
@@ -2235,7 +2235,7 @@ bool Query_log_event::write(IO_CACHE* file)
   }
   if (catalog_len) // i.e. this var is inited (false for 4.0 events)
   {
-    write_str_with_code_and_len((char **)(&start),
+    write_str_with_code_and_len(&start,
                                 catalog, catalog_len, Q_CATALOG_NZ_CODE);
     /*
       In 5.0.x where x<4 masters we used to store the end zero here. This was
@@ -2273,7 +2273,7 @@ bool Query_log_event::write(IO_CACHE* file)
   {
     /* In the TZ sys table, column Name is of length 64 so this should be ok */
     DBUG_ASSERT(time_zone_len <= MAX_TIME_ZONE_NAME_LENGTH);
-    write_str_with_code_and_len((char **)(&start),
+    write_str_with_code_and_len(&start,
                                 time_zone_str, time_zone_len, Q_TIME_ZONE_CODE);
   }
   if (lc_time_names_number)
@@ -4038,6 +4038,7 @@ uint Load_log_event::get_query_buffer_length()
   return
     5 + db_len + 3 +                        // "use DB; "
     18 + fname_len + 2 +                    // "LOAD DATA INFILE 'file''"
+    11 +                                    // "CONCURRENT "
     7 +					    // LOCAL
     9 +                                     // " REPLACE or IGNORE "
     13 + table_name_len*2 +                 // "INTO TABLE `table`"
@@ -4065,6 +4066,9 @@ void Load_log_event::print_query(bool need_db, const char *cs, char *buf,
 
   pos= strmov(pos, "LOAD DATA ");
 
+  if (thd->lex->lock_option == TL_WRITE_CONCURRENT_INSERT)
+    pos= strmov(pos, "CONCURRENT ");
+
   if (fn_start)
     *fn_start= pos;
 
@@ -4542,6 +4546,7 @@ int Load_log_event::do_apply_event(NET* net, Relay_log_info const *rli,
     as the present method does not call mysql_parse().
   */
   lex_start(thd);
+  thd->lex->local_file= local_fname;
   mysql_reset_thd_for_next_command(thd);
 
   if (!use_rli_only_for_errors)
diff --git a/sql/log_event.h b/sql/log_event.h
index cd5e659c910..1fdd7a05968 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -1787,7 +1787,7 @@ private:
 
   @verbatim
    (1)    USE db;
-   (2)    LOAD DATA [LOCAL] INFILE 'file_name'
+   (2)    LOAD DATA [CONCURRENT] [LOCAL] INFILE 'file_name'
    (3)    [REPLACE | IGNORE]
    (4)    INTO TABLE 'table_name'
    (5)    [FIELDS
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 2bc2fcfd384..4d4814ec6ef 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -1301,7 +1301,6 @@ void clean_up(bool print_message)
   lex_free();				/* Free some memory */
   item_create_cleanup();
   set_var_free();
-  free_charsets();
   if (!opt_noacl)
   {
 #ifdef HAVE_DLOPEN
@@ -3991,7 +3990,7 @@ a file name for --log-bin-index option", opt_binlog_index_name);
       my_free(opt_bin_logname, MYF(MY_ALLOW_ZERO_PTR));
       opt_bin_logname=my_strdup(buf, MYF(0));
     }
-    if (mysql_bin_log.open_index_file(opt_binlog_index_name, ln))
+    if (mysql_bin_log.open_index_file(opt_binlog_index_name, ln, TRUE))
     {
       unireg_abort(1);
     }
@@ -4155,7 +4154,7 @@ a file name for --log-bin-index option", opt_binlog_index_name);
   }
 
   if (opt_bin_log && mysql_bin_log.open(opt_bin_logname, LOG_BIN, 0,
-                                        WRITE_CACHE, 0, max_binlog_size, 0))
+                                        WRITE_CACHE, 0, max_binlog_size, 0, TRUE))
     unireg_abort(1);
 
 #ifdef HAVE_REPLICATION
@@ -5987,7 +5986,8 @@ Disable with --skip-super-large-pages.",
 #endif
   {"init-rpl-role", OPT_INIT_RPL_ROLE, "Set the replication role.", 0, 0, 0,
    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
-  {"init-slave", OPT_INIT_SLAVE, "Command(s) that are executed when a slave connects to this master",
+  {"init-slave", OPT_INIT_SLAVE, "Command(s) that are executed by a slave server \
+each time the SQL thread starts.",
    (uchar**) &opt_init_slave, (uchar**) &opt_init_slave, 0, GET_STR_ALLOC,
    REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"language", 'L',
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index a1403d2ff71..4bbafa0253a 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -183,10 +183,10 @@ a file name for --relay-log-index option", opt_relaylog_index_name);
       note, that if open() fails, we'll still have index file open
       but a destructor will take care of that
     */
-    if (rli->relay_log.open_index_file(opt_relaylog_index_name, ln) ||
+    if (rli->relay_log.open_index_file(opt_relaylog_index_name, ln, TRUE) ||
         rli->relay_log.open(ln, LOG_BIN, 0, SEQ_READ_APPEND, 0,
                             (max_relay_log_size ? max_relay_log_size :
-                            max_binlog_size), 1))
+                            max_binlog_size), 1, TRUE))
     {
       pthread_mutex_unlock(&rli->data_lock);
       sql_print_error("Failed in open_log() called from init_relay_log_info()");
@@ -1029,7 +1029,7 @@ err:
      false - condition not met
 */
 
-bool Relay_log_info::is_until_satisfied(my_off_t master_beg_pos)
+bool Relay_log_info::is_until_satisfied(THD *thd, Log_event *ev)
 {
   const char *log_name;
   ulonglong log_pos;
@@ -1039,8 +1039,12 @@ bool Relay_log_info::is_until_satisfied(my_off_t master_beg_pos)
 
   if (until_condition == UNTIL_MASTER_POS)
   {
+    if (ev && ev->server_id == (uint32) ::server_id && !replicate_same_server_id)
+      DBUG_RETURN(FALSE);
     log_name= group_master_log_name;
-    log_pos= master_beg_pos;
+    log_pos= (!ev)? group_master_log_pos :
+      ((thd->options & OPTION_BEGIN || !ev->log_pos) ?
+       group_master_log_pos : ev->log_pos - ev->data_written);
   }
   else
   { /* until_condition == UNTIL_RELAY_POS */
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index 1dc7f3ef0d2..ec314e50cc7 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -322,7 +322,7 @@ public:
   void close_temporary_tables();
 
   /* Check if UNTIL condition is satisfied. See slave.cc for more. */
-  bool is_until_satisfied(my_off_t master_beg_pos);
+  bool is_until_satisfied(THD *thd, Log_event *ev);
   inline ulonglong until_pos()
   {
     return ((until_condition == UNTIL_MASTER_POS) ? group_master_log_pos :
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 4260efdeb56..03d7ff30f7e 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -5690,8 +5690,8 @@ ER_PARTITION_WRONG_NO_SUBPART_ERROR
         eng "Wrong number of subpartitions defined, mismatch with previous setting"
         ger "Falsche Anzahl von Unterpartitionen definiert, stimmt nicht mit vorherigen Einstellungen überein"
         swe "Antal subpartitioner definierade och antal subpartitioner är inte lika"
-ER_CONST_EXPR_IN_PARTITION_FUNC_ERROR
-        eng "Constant/Random expression in (sub)partitioning function is not allowed"
+ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR
+        eng "Constant, random or timezone-dependent expressions in (sub)partitioning function are not allowed"
         ger "Konstante oder Random-Ausdrücke in (Unter-)Partitionsfunktionen sind nicht erlaubt"
         swe "Konstanta uttryck eller slumpmässiga uttryck är inte tillåtna (sub)partitioneringsfunktioner"
 ER_NO_CONST_EXPR_IN_RANGE_OR_LIST_ERROR
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 93385292c24..ab81c122497 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5691,8 +5691,8 @@ ER_PARTITION_WRONG_NO_SUBPART_ERROR
         eng "Wrong number of subpartitions defined, mismatch with previous setting"
         ger "Falsche Anzahl von Unterpartitionen definiert, stimmt nicht mit vorherigen Einstellungen überein"
         swe "Antal subpartitioner definierade och antal subpartitioner är inte lika"
-ER_CONST_EXPR_IN_PARTITION_FUNC_ERROR
-        eng "Constant/Random expression in (sub)partitioning function is not allowed"
+ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR
+        eng "Constant, random or timezone-dependent expressions in (sub)partitioning function are not allowed"
         ger "Konstante oder Random-Ausdrücke in (Unter-)Partitionsfunktionen sind nicht erlaubt"
         swe "Konstanta uttryck eller slumpmässiga uttryck är inte tillåtna (sub)partitioneringsfunktioner"
 ER_NO_CONST_EXPR_IN_RANGE_OR_LIST_ERROR
diff --git a/sql/slave.cc b/sql/slave.cc
index 6d78d44f6f2..89e2fd8c302 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -2339,9 +2339,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
       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))
+        rli->is_until_satisfied(thd, ev))
     {
       char buf[22];
       sql_print_information("Slave SQL thread stopped because it reached its"
@@ -3139,7 +3137,7 @@ log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
   */
   pthread_mutex_lock(&rli->data_lock);
   if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
-      rli->is_until_satisfied(rli->group_master_log_pos))
+      rli->is_until_satisfied(thd, NULL))
   {
     char buf[22];
     sql_print_information("Slave SQL thread stopped because it reached its"
diff --git a/sql/sp.cc b/sql/sp.cc
index 1edacee8aad..0ed4855e5d3 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -913,6 +913,7 @@ sp_create_routine(THD *thd, int type, sp_head *sp)
   DBUG_PRINT("enter", ("type: %d  name: %.*s",type, (int) sp->m_name.length,
                        sp->m_name.str));
   String retstr(64);
+  retstr.set_charset(system_charset_info);
 
   DBUG_ASSERT(type == TYPE_ENUM_PROCEDURE ||
               type == TYPE_ENUM_FUNCTION);
@@ -1419,6 +1420,7 @@ sp_find_routine(THD *thd, int type, sp_name *name, sp_cache **cp,
       64 -- size of "returns" column of mysql.proc.
     */
     String retstr(64);
+    retstr.set_charset(sp->get_creation_ctx()->get_client_cs());
 
     DBUG_PRINT("info", ("found: 0x%lx", (ulong)sp));
     if (sp->m_first_free_instance)
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index b91dc9a7e25..05e8cee9151 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -3440,7 +3440,7 @@ bool mysql_routine_grant(THD *thd, TABLE_LIST *table_list, bool is_proc,
 
   if (write_to_binlog)
   {
-    if (write_bin_log(thd, TRUE, thd->query(), thd->query_length()))
+    if (write_bin_log(thd, FALSE, thd->query(), thd->query_length()))
       result= TRUE;
   }
 
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index c67c3266e59..8a8cd4659e4 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -5718,7 +5718,8 @@ find_field_in_view(THD *thd, TABLE_LIST *table_list,
     if (!my_strcasecmp(system_charset_info, field_it.name(), name))
     {
       // in PS use own arena or data will be freed after prepare
-      if (register_tree_change && thd->stmt_arena->is_stmt_prepare_or_first_sp_execute())
+      if (register_tree_change &&
+          thd->stmt_arena->is_stmt_prepare_or_first_stmt_execute())
         arena= thd->activate_stmt_arena_if_needed(&backup);
       /*
         create_item() may, or may not create a new Item, depending on
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 55bb3e754de..13fed9e3bce 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -581,6 +581,8 @@ public:
   { return state == INITIALIZED_FOR_SP; }
   inline bool is_stmt_prepare_or_first_sp_execute() const
   { return (int)state < (int)PREPARED; }
+  inline bool is_stmt_prepare_or_first_stmt_execute() const
+  { return (int)state <= (int)PREPARED; }
   inline bool is_first_stmt_execute() const { return state == PREPARED; }
   inline bool is_stmt_execute() const
   { return state == PREPARED || state == EXECUTED; }
diff --git a/sql/sql_crypt.cc b/sql/sql_crypt.cc
index c4f93cc2a33..3d7d248782b 100644
--- a/sql/sql_crypt.cc
+++ b/sql/sql_crypt.cc
@@ -28,14 +28,7 @@
 
 #include "mysql_priv.h"
 
-SQL_CRYPT::SQL_CRYPT(const char *password, uint length)
-{
-  ulong rand_nr[2];
-  hash_password(rand_nr,password, length);
-  crypt_init(rand_nr);
-}
-
-void SQL_CRYPT::crypt_init(ulong *rand_nr)
+void SQL_CRYPT::init(ulong *rand_nr)
 {
   uint i;
   randominit(&rand,rand_nr[0],rand_nr[1]);
diff --git a/sql/sql_crypt.h b/sql/sql_crypt.h
index 8d5a761cbdf..dd1cba9ee0c 100644
--- a/sql/sql_crypt.h
+++ b/sql/sql_crypt.h
@@ -26,15 +26,15 @@ class SQL_CRYPT :public Sql_alloc
   struct rand_struct rand,org_rand;
   char decode_buff[256],encode_buff[256];
   uint shift;
-  void crypt_init(ulong *seed);
  public:
-  SQL_CRYPT(const char *seed, uint length);
+  SQL_CRYPT() {}
   SQL_CRYPT(ulong *seed)
   {
-    crypt_init(seed);
+    init(seed);
   }
   ~SQL_CRYPT() {}
-  void init() { shift=0; rand=org_rand; }
+  void init(ulong *seed);
+  void reinit() { shift=0; rand=org_rand; }
   void encode(char *str, uint length);
   void decode(char *str, uint length);
 };
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 51098987e81..2ae876985b7 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -696,6 +696,7 @@ int mysql_create_db(THD *thd, char *db, HA_CREATE_INFO *create_info,
       file.  In this case it's best to just continue as if nothing has
       happened.  (This is a very unlikely senario)
     */
+    thd->clear_error();
   }
 
 not_silent:
@@ -815,9 +816,9 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info)
 
   if (mysql_bin_log.is_open())
   {
-    int errcode= query_error_code(thd, TRUE);
+    thd->clear_error();
     Query_log_event qinfo(thd, thd->query(), thd->query_length(), 0,
-			  /* suppress_use */ TRUE, errcode);
+			  /* suppress_use */ TRUE, 0);
 
     /*
       Write should use the database being created as the "current
@@ -827,7 +828,6 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info)
     qinfo.db     = db;
     qinfo.db_len = strlen(db);
 
-    thd->clear_error();
     /* These DDL methods and logging protected with LOCK_mysql_create_db */
     if (error= mysql_bin_log.write(&qinfo))
       goto exit;
@@ -968,9 +968,9 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
     }
     if (mysql_bin_log.is_open())
     {
-      int errcode= query_error_code(thd, TRUE);
+      thd->clear_error();
       Query_log_event qinfo(thd, query, query_length, 0, 
-			    /* suppress_use */ TRUE, errcode);
+			    /* suppress_use */ TRUE, 0);
       /*
         Write should use the database being created as the "current
         database" and not the threads current database, which is the
@@ -979,7 +979,6 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       qinfo.db     = db;
       qinfo.db_len = strlen(db);
 
-      thd->clear_error();
       /* These DDL methods and logging protected with LOCK_mysql_create_db */
       if (mysql_bin_log.write(&qinfo))
       {
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 062d2fb701c..4586879dec4 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -1026,6 +1026,8 @@ end_lex_with_single_table(THD *thd, TABLE *table, LEX *old_lex)
     table                The table object
     part_info            Reference to partitioning data structure
     is_sub_part          Is the table subpartitioned as well
+    is_create_table_ind  Indicator of whether openfrm was called as part of
+                         CREATE or ALTER TABLE
 
   RETURN VALUE
     TRUE                 An error occurred, something was wrong with the
@@ -1049,7 +1051,7 @@ end_lex_with_single_table(THD *thd, TABLE *table, LEX *old_lex)
 */
 
 static bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
-                                 bool is_sub_part)
+                          bool is_sub_part, bool is_create_table_ind)
 {
   partition_info *part_info= table->part_info;
   bool result= TRUE;
@@ -1105,10 +1107,31 @@ static bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
   }
   if (unlikely(func_expr->const_item()))
   {
-    my_error(ER_CONST_EXPR_IN_PARTITION_FUNC_ERROR, MYF(0));
+    my_error(ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR, MYF(0));
     clear_field_flag(table);
     goto end;
   }
+
+  /*
+    We don't allow creating partitions with timezone-dependent expressions as
+    a (sub)partitioning function, but we want to allow such expressions when
+    opening existing tables for easier maintenance. This exception should be
+    deprecated at some point in future so that we always throw an error.
+  */
+  if (func_expr->walk(&Item::is_timezone_dependent_processor,
+                      0, NULL))
+  {
+    if (is_create_table_ind)
+    {
+      my_error(ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR, MYF(0));
+      goto end;
+    }
+    else
+      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                   ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR,
+                   ER(ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR));
+  }
+
   if ((!is_sub_part) && (error= check_signed_flag(part_info)))
     goto end;
   result= set_up_field_array(table, is_sub_part);
@@ -1719,7 +1742,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
     else
     {
       if (unlikely(fix_fields_part_func(thd, part_info->subpart_expr,
-                                        table, TRUE)))
+                                        table, TRUE, is_create_table_ind)))
         goto end;
       if (unlikely(part_info->subpart_expr->result_type() != INT_RESULT))
       {
@@ -1747,7 +1770,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
     else
     {
       if (unlikely(fix_fields_part_func(thd, part_info->part_expr,
-                                        table, FALSE)))
+                                        table, FALSE, is_create_table_ind)))
         goto end;
       if (unlikely(part_info->part_expr->result_type() != INT_RESULT))
       {
@@ -1770,7 +1793,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
     else
     {
       if (unlikely(fix_fields_part_func(thd, part_info->part_expr,
-                                        table, FALSE)))
+                                        table, FALSE, is_create_table_ind)))
         goto end;
     }
     part_info->fixed= TRUE;
diff --git a/sql/sql_partition.h b/sql/sql_partition.h
index 6e1bf8b5728..7ac1415c158 100644
--- a/sql/sql_partition.h
+++ b/sql/sql_partition.h
@@ -98,9 +98,6 @@ uint32 get_list_array_idx_for_endpoint(partition_info *part_info,
 uint32 get_partition_id_range_for_endpoint(partition_info *part_info,
                                            bool left_endpoint,
                                            bool include_endpoint);
-bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
-                          bool is_sub_part, bool is_field_to_be_setup);
-
 bool check_part_func_fields(Field **ptr, bool ok_with_charsets);
 bool field_is_partition_charset(Field *field);
 Item* convert_charset_partition_constant(Item *item, CHARSET_INFO *cs);
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 3c3d3a62ef5..334efea1b63 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1644,6 +1644,11 @@ JOIN::reinit()
   if (join_tab_save)
     memcpy(join_tab, join_tab_save, sizeof(JOIN_TAB) * tables);
 
+  /* need to reset ref access state (see join_read_key) */
+  if (join_tab)
+    for (uint i= 0; i < tables; i++)
+      join_tab[i].ref.key_err= TRUE;
+
   if (tmp_join)
     restore_tmp();
 
@@ -5971,6 +5976,7 @@ inline void add_cond_and_fix(Item **e1, Item *e2)
     {
       *e1= res;
       res->quick_fix_field();
+      res->update_used_tables();
     }
   }
   else
@@ -7637,7 +7643,7 @@ static bool check_simple_equality(Item *left_item, Item *right_item,
           already contains a constant and its value is  not equal to
           the value of const_item.
         */
-        item_equal->add(const_item);
+        item_equal->add(const_item, field_item);
       }
       else
       {
@@ -13655,7 +13661,7 @@ check_reverse_order:
 	select->quick=tmp;
       }
     }
-    else if (tab->type != JT_NEXT && 
+    else if (tab->type != JT_NEXT && tab->type != JT_REF_OR_NULL &&
              tab->ref.key >= 0 && tab->ref.key_parts <= used_key_parts)
     {
       /*
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 8c8d0166b38..7ba62c62259 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -641,7 +641,7 @@ mysqld_show_create(THD *thd, TABLE_LIST *table_list)
     thd->push_internal_handler(&view_error_suppressor);
     bool error= open_normal_and_derived_tables(thd, table_list, 0);
     thd->pop_internal_handler();
-    if (error && thd->is_error())
+    if (error && (thd->killed || thd->is_error()))
       DBUG_RETURN(TRUE);
   }
 
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 2d768609220..cc92d224cd9 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -831,7 +831,7 @@ int mysql_update(THD *thd,
 
   if (error < 0)
   {
-    char buff[STRING_BUFFER_USUAL_SIZE];
+    char buff[MYSQL_ERRMSG_SIZE];
     my_snprintf(buff, sizeof(buff), ER(ER_UPDATE_INFO), (ulong) found,
                 (ulong) updated,
                 (ulong) thd->warning_info->statement_warn_count());
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index f65c5dc54c1..c911f4fe7b9 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -1775,7 +1775,7 @@ bool check_key_in_view(THD *thd, TABLE_LIST *view)
       if (!fld->item->fixed && fld->item->fix_fields(thd, &fld->item))
       {
         thd->mark_used_columns= save_mark_used_columns;
-        return TRUE;
+        DBUG_RETURN(TRUE);
       }
     }
     thd->mark_used_columns= save_mark_used_columns;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index a57897f4eac..0b0c4913476 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -682,6 +682,34 @@ bool setup_select_in_parentheses(LEX *lex)
   return FALSE;
 }
 
+static bool add_create_index_prepare (LEX *lex, Table_ident *table)
+{
+  lex->sql_command= SQLCOM_CREATE_INDEX;
+  if (!lex->current_select->add_table_to_list(lex->thd, table, NULL,
+                                              TL_OPTION_UPDATING))
+    return TRUE;
+  lex->alter_info.reset();
+  lex->alter_info.flags= ALTER_ADD_INDEX;
+  lex->col_list.empty();
+  lex->change= NullS;
+  return FALSE;
+}
+
+static bool add_create_index (LEX *lex, Key::Keytype type,
+                              const LEX_STRING &name,
+                              KEY_CREATE_INFO *info= NULL, bool generated= 0)
+{
+  Key *key;
+  key= new Key(type, name, info ? info : &lex->key_create_info, generated, 
+               lex->col_list);
+  if (key == NULL)
+    return TRUE;
+
+  lex->alter_info.key_list.push_back(key);
+  lex->col_list.empty();
+  return FALSE;
+}
+
 %}
 %union {
   int  num;
@@ -1435,7 +1463,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         option_type opt_var_type opt_var_ident_type
 
 %type <key_type>
-        key_type opt_unique_or_fulltext constraint_key_type
+        normal_key_type opt_unique constraint_key_type fulltext spatial
 
 %type <key_alg>
         btree_or_rtree
@@ -1538,7 +1566,10 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         view_suid view_tail view_list_opt view_list view_select
         view_check_option trigger_tail sp_tail sf_tail udf_tail event_tail
         install uninstall partition_entry binlog_base64_event
-        init_key_options key_options key_opts key_opt key_using_alg
+        init_key_options normal_key_options normal_key_opts all_key_opt 
+        spatial_key_options fulltext_key_options normal_key_opt 
+        fulltext_key_opt spatial_key_opt fulltext_key_opts spatial_key_opts
+        key_using_alg
         part_column_list
         server_def server_options_list server_option
         definer_opt no_definer definer
@@ -2001,35 +2032,37 @@ create:
                                   $5->table.str);
             }
           }
-        | CREATE opt_unique_or_fulltext INDEX_SYM ident key_alg ON
+        | CREATE opt_unique INDEX_SYM ident key_alg ON table_ident
+          {
+            if (add_create_index_prepare(Lex, $7))
+              MYSQL_YYABORT;
+          }
+          '(' key_list ')' normal_key_options
+          {
+            if (add_create_index(Lex, $2, $4))
+              MYSQL_YYABORT;
+          }
+        | CREATE fulltext INDEX_SYM ident init_key_options ON
           table_ident
           {
-            LEX *lex=Lex;
-            lex->sql_command= SQLCOM_CREATE_INDEX;
-            if (!lex->current_select->add_table_to_list(lex->thd, $7,
-                                                        NULL,
-                                                        TL_OPTION_UPDATING))
+            if (add_create_index_prepare(Lex, $7))
               MYSQL_YYABORT;
-            lex->alter_info.reset();
-            lex->alter_info.flags= ALTER_ADD_INDEX;
-            lex->col_list.empty();
-            lex->change=NullS;
           }
-          '(' key_list ')' key_options
+          '(' key_list ')' fulltext_key_options
           {
-            LEX *lex=Lex;
-            Key *key;
-            if ($2 != Key::FULLTEXT && lex->key_create_info.parser_name.str)
-            {
-              my_parse_error(ER(ER_SYNTAX_ERROR));
+            if (add_create_index(Lex, $2, $4))
               MYSQL_YYABORT;
-            }
-            key= new Key($2, $4, &lex->key_create_info, 0,
-                         lex->col_list);
-            if (key == NULL)
+          }
+        | CREATE spatial INDEX_SYM ident init_key_options ON
+          table_ident
+          {
+            if (add_create_index_prepare(Lex, $7))
+              MYSQL_YYABORT;
+          }
+          '(' key_list ')' spatial_key_options
+          {
+            if (add_create_index(Lex, $2, $4))
               MYSQL_YYABORT;
-            lex->alter_info.key_list.push_back(key);
-            lex->col_list.empty();
           }
         | CREATE DATABASE opt_if_not_exists ident
           {
@@ -4440,7 +4473,7 @@ part_func_expr:
             lex->safe_to_cache_query= 1;
             if (not_corr_func)
             {
-              my_parse_error(ER(ER_CONST_EXPR_IN_PARTITION_FUNC_ERROR));
+              my_parse_error(ER(ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR));
               MYSQL_YYABORT;
             }
             $$=$1;
@@ -5204,31 +5237,28 @@ column_def:
         ;
 
 key_def:
-          key_type opt_ident key_alg '(' key_list ')' key_options
+          normal_key_type opt_ident key_alg '(' key_list ')' normal_key_options
           {
-            LEX *lex=Lex;
-            if ($1 != Key::FULLTEXT && lex->key_create_info.parser_name.str)
-            {
-              my_parse_error(ER(ER_SYNTAX_ERROR));
+            if (add_create_index (Lex, $1, $2))
               MYSQL_YYABORT;
-            }
-            Key *key= new Key($1, $2, &lex->key_create_info, 0,
-                              lex->col_list);
-            if (key == NULL)
+          }
+        | fulltext opt_key_or_index opt_ident init_key_options 
+            '(' key_list ')' fulltext_key_options
+          {
+            if (add_create_index (Lex, $1, $3))
+              MYSQL_YYABORT;
+          }
+        | spatial opt_key_or_index opt_ident init_key_options 
+            '(' key_list ')' spatial_key_options
+          {
+            if (add_create_index (Lex, $1, $3))
               MYSQL_YYABORT;
-            lex->alter_info.key_list.push_back(key);
-            lex->col_list.empty(); /* Alloced by sql_alloc */
           }
         | opt_constraint constraint_key_type opt_ident key_alg
-          '(' key_list ')' key_options
+          '(' key_list ')' normal_key_options
           {
-            LEX *lex=Lex;
-            Key *key= new Key($2, $3.str ? $3 : $1, &lex->key_create_info, 0,
-                              lex->col_list);
-            if (key == NULL)
+            if (add_create_index (Lex, $2, $3.str ? $3 : $1))
               MYSQL_YYABORT;
-            lex->alter_info.key_list.push_back(key);
-            lex->col_list.empty(); /* Alloced by sql_alloc */
           }
         | opt_constraint FOREIGN KEY_SYM opt_ident '(' key_list ')' references
           {
@@ -5242,13 +5272,9 @@ key_def:
             if (key == NULL)
               MYSQL_YYABORT;
             lex->alter_info.key_list.push_back(key);
-            key= new Key(Key::MULTIPLE, $1.str ? $1 : $4,
-                         &default_key_create_info, 1,
-                         lex->col_list);
-            if (key == NULL)
+            if (add_create_index (lex, Key::MULTIPLE, $1.str ? $1 : $4,
+                                  &default_key_create_info, 1))
               MYSQL_YYABORT;
-            lex->alter_info.key_list.push_back(key);
-            lex->col_list.empty(); /* Alloced by sql_alloc */
             /* Only used for ALTER TABLE. Ignored otherwise. */
             lex->alter_info.flags|= ALTER_FOREIGN_KEY;
           }
@@ -5867,19 +5893,8 @@ delete_option:
         | SET DEFAULT   { $$= (int) Foreign_key::FK_OPTION_DEFAULT;  }
         ;
 
-key_type:
+normal_key_type:
           key_or_index { $$= Key::MULTIPLE; }
-        | FULLTEXT_SYM opt_key_or_index { $$= Key::FULLTEXT; }
-        | SPATIAL_SYM opt_key_or_index
-          {
-#ifdef HAVE_SPATIAL
-            $$= Key::SPATIAL;
-#else
-            my_error(ER_FEATURE_DISABLED, MYF(0),
-                     sym_group_geom.name, sym_group_geom.needed_define);
-            MYSQL_YYABORT;
-#endif
-          }
         ;
 
 constraint_key_type:
@@ -5903,11 +5918,17 @@ keys_or_index:
         | INDEXES {}
         ;
 
-opt_unique_or_fulltext:
+opt_unique:
           /* empty */  { $$= Key::MULTIPLE; }
         | UNIQUE_SYM   { $$= Key::UNIQUE; }
-        | FULLTEXT_SYM { $$= Key::FULLTEXT;}
-        | SPATIAL_SYM
+        ;
+
+fulltext:
+          FULLTEXT_SYM { $$= Key::FULLTEXT;}
+        ;
+
+spatial:
+          SPATIAL_SYM
           {
 #ifdef HAVE_SPATIAL
             $$= Key::SPATIAL;
@@ -5936,14 +5957,34 @@ key_alg:
         | init_key_options key_using_alg
         ;
 
-key_options:
+normal_key_options:
+          /* empty */ {}
+        | normal_key_opts
+        ;
+
+fulltext_key_options:
           /* empty */ {}
-        | key_opts
+        | fulltext_key_opts
+        ;
+
+spatial_key_options:
+          /* empty */ {}
+        | spatial_key_opts
+        ;
+
+normal_key_opts:
+          normal_key_opt
+        | normal_key_opts normal_key_opt
+        ;
+
+spatial_key_opts:
+          spatial_key_opt
+        | spatial_key_opts spatial_key_opt
         ;
 
-key_opts:
-          key_opt
-        | key_opts key_opt
+fulltext_key_opts:
+          fulltext_key_opt
+        | fulltext_key_opts fulltext_key_opt
         ;
 
 key_using_alg:
@@ -5951,10 +5992,22 @@ key_using_alg:
         | TYPE_SYM btree_or_rtree  { Lex->key_create_info.algorithm= $2; }
         ;
 
-key_opt:
-          key_using_alg
-        | KEY_BLOCK_SIZE opt_equal ulong_num
+all_key_opt:
+          KEY_BLOCK_SIZE opt_equal ulong_num
           { Lex->key_create_info.block_size= $3; }
+        ;
+
+normal_key_opt:
+          all_key_opt
+        | key_using_alg
+        ;
+
+spatial_key_opt:
+          all_key_opt
+        ;
+
+fulltext_key_opt:
+          all_key_opt
         | WITH PARSER_SYM IDENT_sys
           {
             if (plugin_is_ready(&$3, MYSQL_FTPARSER_PLUGIN))