Manual merge from mysql-trunk-merge.

Conflicts:
  - extra/comp_err.c
  - mysql-test/collections/default.experimental
  - mysql-test/r/archive.result
  - mysql-test/r/select.result
  - mysql-test/suite/binlog/r/binlog_unsafe.result
  - mysql-test/suite/binlog/t/binlog_unsafe.test
  - mysql-test/suite/rpl/t/disabled.def
  - mysql-test/t/archive.test
  - mysql-test/t/select.test
  - sql/item.cc
  - sql/item.h
  - sql/item_timefunc.cc
  - sql/sql_base.cc
  - sql/sql_delete.cc
  - sql/sql_load.cc
  - sql/sql_partition.cc
  - sql/sql_table.cc
  - storage/innobase/handler/ha_innodb.cc
  - vio/vio.c

43 files changed, 1448 insertions, 844 deletions

diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc
index 8faab5023da..9f3863eb2b0 100644
--- a/sql/event_db_repository.cc
+++ b/sql/event_db_repository.cc
@@ -26,7 +26,7 @@
 */
 
 static
-const TABLE_FIELD_W_TYPE event_table_fields[ET_FIELD_COUNT] =
+const TABLE_FIELD_TYPE event_table_fields[ET_FIELD_COUNT] =
 {
   {
     { C_STRING_WITH_LEN("db") },
@@ -151,6 +151,24 @@ const TABLE_FIELD_W_TYPE event_table_fields[ET_FIELD_COUNT] =
   }
 };
 
+static const TABLE_FIELD_DEF
+  event_table_def= {ET_FIELD_COUNT, event_table_fields};
+
+class Event_db_intact : public Table_check_intact
+{
+protected:
+  void report_error(uint, const char *fmt, ...)
+  {
+    va_list args;
+    va_start(args, fmt);
+    error_log_print(ERROR_LEVEL, fmt, args);
+    va_end(args);
+  }
+};
+
+/** In case of an error, a message is printed to the error log. */
+static Event_db_intact table_intact;
+
 
 /**
   Puts some data common to CREATE and ALTER EVENT into a row.
@@ -1117,10 +1135,8 @@ Event_db_repository::check_system_tables(THD *thd)
   }
   else
   {
-    if (table_check_intact(tables.table, MYSQL_DB_FIELD_COUNT,
-                           mysql_db_table_fields))
+    if (table_intact.check(tables.table, &mysql_db_table_def))
       ret= 1;
-    /* in case of an error, the message is printed inside table_check_intact */
 
     close_thread_tables(thd);
   }
@@ -1154,9 +1170,8 @@ Event_db_repository::check_system_tables(THD *thd)
   }
   else
   {
-    if (table_check_intact(tables.table, ET_FIELD_COUNT, event_table_fields))
+    if (table_intact.check(tables.table, &event_table_def))
       ret= 1;
-    /* in case of an error, the message is printed inside table_check_intact */
     close_thread_tables(thd);
   }
 
diff --git a/sql/field.cc b/sql/field.cc
index 8077336583e..0934bb04ccd 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -2519,6 +2519,50 @@ Field_new_decimal::Field_new_decimal(uint32 len_arg,
 }
 
 
+Field *Field_new_decimal::create_from_item (Item *item)
+{
+  uint8 dec= item->decimals;
+  uint8 intg= item->decimal_precision() - dec;
+  uint32 len= item->max_length;
+
+  DBUG_ASSERT (item->result_type() == DECIMAL_RESULT);
+
+  /*
+    Trying to put too many digits overall in a DECIMAL(prec,dec)
+    will always throw a warning. We must limit dec to
+    DECIMAL_MAX_SCALE however to prevent an assert() later.
+  */
+
+  if (dec > 0)
+  {
+    signed int overflow;
+
+    dec= min(dec, DECIMAL_MAX_SCALE);
+
+    /*
+      If the value still overflows the field with the corrected dec,
+      we'll throw out decimals rather than integers. This is still
+      bad and of course throws a truncation warning.
+      +1: for decimal point
+      */
+
+    const int required_length=
+      my_decimal_precision_to_length(intg + dec, dec,
+                                     item->unsigned_flag);
+
+    overflow= required_length - len;
+
+    if (overflow > 0)
+      dec= max(0, dec - overflow);            // too long, discard fract
+    else
+      /* Corrected value fits. */
+      len= required_length;
+  }
+  return new Field_new_decimal(len, item->maybe_null, item->name,
+                               dec, item->unsigned_flag);
+}
+
+
 int Field_new_decimal::reset(void)
 {
   store_value(&decimal_zero);
diff --git a/sql/field.h b/sql/field.h
index fb6ca34e88e..bb3636c654e 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -788,6 +788,7 @@ public:
   uint is_equal(Create_field *new_field);
   virtual const uchar *unpack(uchar* to, const uchar *from,
                               uint param_data, bool low_byte_first);
+  static Field *create_from_item (Item *);
 };
 
 
diff --git a/sql/item.cc b/sql/item.cc
index 7de32423927..f6ce117a41f 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -5063,9 +5063,7 @@ Field *Item::tmp_table_field_from_field_type(TABLE *table, bool fixed_length)
   switch (field_type()) {
   case MYSQL_TYPE_DECIMAL:
   case MYSQL_TYPE_NEWDECIMAL:
-    field= new Field_new_decimal((uchar*) 0, max_length, null_ptr, 0,
-                                 Field::NONE, name, decimals, 0,
-                                 unsigned_flag);
+    field= Field_new_decimal::create_from_item(this);
     break;
   case MYSQL_TYPE_TINY:
     field= new Field_tiny((uchar*) 0, max_length, null_ptr, 0, Field::NONE,
@@ -7182,7 +7180,7 @@ Item_cache* Item_cache::get_cache(const Item *item, const Item_result type)
 {
   switch (type) {
   case INT_RESULT:
-    return new Item_cache_int();
+    return new Item_cache_int(item->field_type());
   case REAL_RESULT:
     return new Item_cache_real();
   case DECIMAL_RESULT:
@@ -7204,8 +7202,9 @@ Item_cache* Item_cache::get_cache(const Item *item, const Item_result type)
 
 void Item_cache::store(Item *item)
 {
-  if (item)
-    example= item;
+  example= item;
+  if (!item)
+    null_value= TRUE;
   value_cached= FALSE;
 }
 
@@ -7219,12 +7218,15 @@ void Item_cache::print(String *str, enum_query_type query_type)
   str->append(')');
 }
 
-void  Item_cache_int::cache_value()
+bool  Item_cache_int::cache_value()
 {
+  if (!example)
+    return FALSE;
   value_cached= TRUE;
   value= example->val_int_result();
   null_value= example->null_value;
   unsigned_flag= example->unsigned_flag;
+  return TRUE;
 }
 
 
@@ -7241,8 +7243,8 @@ void Item_cache_int::store(Item *item, longlong val_arg)
 String *Item_cache_int::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return NULL;
   str->set(value, default_charset());
   return str;
 }
@@ -7251,8 +7253,8 @@ String *Item_cache_int::val_str(String *str)
 my_decimal *Item_cache_int::val_decimal(my_decimal *decimal_val)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return NULL;
   int2my_decimal(E_DEC_FATAL_ERROR, value, unsigned_flag, decimal_val);
   return decimal_val;
 }
@@ -7260,21 +7262,24 @@ my_decimal *Item_cache_int::val_decimal(my_decimal *decimal_val)
 double Item_cache_int::val_real()
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return 0.0;
   return (double) value;
 }
 
 longlong Item_cache_int::val_int()
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return 0;
   return value;
 }
 
 void  Item_cache_datetime::cache_value_int()
 {
+  if (!example)
+    return;
+
   value_cached= TRUE;
   /* Assume here that the underlying item will do correct conversion.*/
   int_value= example->val_int_result();
@@ -7283,8 +7288,10 @@ void  Item_cache_datetime::cache_value_int()
 }
 
 
-void  Item_cache_datetime::cache_value()
+bool  Item_cache_datetime::cache_value()
 {
+  if (!example)
+    return FALSE;
   str_value_cached= TRUE;
   /* Assume here that the underlying item will do correct conversion.*/
   String *res= example->str_result(&str_value);
@@ -7292,6 +7299,7 @@ void  Item_cache_datetime::cache_value()
     str_value.copy(*res);
   null_value= example->null_value;
   unsigned_flag= example->unsigned_flag;
+  return TRUE;
 }
 
 
@@ -7339,27 +7347,30 @@ longlong Item_cache_datetime::val_int()
   return int_value;
 }
 
-void Item_cache_real::cache_value()
+bool Item_cache_real::cache_value()
 {
+  if (!example)
+    return FALSE;
   value_cached= TRUE;
   value= example->val_result();
   null_value= example->null_value;
+  return TRUE;
 }
 
 
 double Item_cache_real::val_real()
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return 0.0;
   return value;
 }
 
 longlong Item_cache_real::val_int()
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return 0;
   return (longlong) rint(value);
 }
 
@@ -7367,8 +7378,8 @@ longlong Item_cache_real::val_int()
 String* Item_cache_real::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return NULL;
   str->set_real(value, decimals, default_charset());
   return str;
 }
@@ -7377,27 +7388,30 @@ String* Item_cache_real::val_str(String *str)
 my_decimal *Item_cache_real::val_decimal(my_decimal *decimal_val)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return NULL;
   double2my_decimal(E_DEC_FATAL_ERROR, value, decimal_val);
   return decimal_val;
 }
 
 
-void Item_cache_decimal::cache_value()
+bool Item_cache_decimal::cache_value()
 {
+  if (!example)
+    return FALSE;
   value_cached= TRUE;
   my_decimal *val= example->val_decimal_result(&decimal_value);
   if (!(null_value= example->null_value) && val != &decimal_value)
     my_decimal2decimal(val, &decimal_value);
+  return TRUE;
 }
 
 double Item_cache_decimal::val_real()
 {
   DBUG_ASSERT(fixed);
   double res;
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return NULL;
   my_decimal2double(E_DEC_FATAL_ERROR, &decimal_value, &res);
   return res;
 }
@@ -7406,8 +7420,8 @@ longlong Item_cache_decimal::val_int()
 {
   DBUG_ASSERT(fixed);
   longlong res;
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return 0;
   my_decimal2int(E_DEC_FATAL_ERROR, &decimal_value, unsigned_flag, &res);
   return res;
 }
@@ -7415,8 +7429,8 @@ longlong Item_cache_decimal::val_int()
 String* Item_cache_decimal::val_str(String *str)
 {
   DBUG_ASSERT(fixed);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return NULL;
   my_decimal_round(E_DEC_FATAL_ERROR, &decimal_value, decimals, FALSE,
                    &decimal_value);
   my_decimal2string(E_DEC_FATAL_ERROR, &decimal_value, 0, 0, 0, str);
@@ -7426,14 +7440,16 @@ String* Item_cache_decimal::val_str(String *str)
 my_decimal *Item_cache_decimal::val_decimal(my_decimal *val)
 {
   DBUG_ASSERT(fixed);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return NULL;
   return &decimal_value;
 }
 
 
-void Item_cache_str::cache_value()
+bool Item_cache_str::cache_value()
 {
+  if (!example)
+    return FALSE;
   value_cached= TRUE;
   value_buff.set(buffer, sizeof(buffer), example->collation.collation);
   value= example->str_result(&value_buff);
@@ -7452,6 +7468,7 @@ void Item_cache_str::cache_value()
     value_buff.copy(*value);
     value= &value_buff;
   }
+  return TRUE;
 }
 
 double Item_cache_str::val_real()
@@ -7459,8 +7476,8 @@ double Item_cache_str::val_real()
   DBUG_ASSERT(fixed == 1);
   int err_not_used;
   char *end_not_used;
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return 0.0;
   if (value)
     return my_strntod(value->charset(), (char*) value->ptr(),
 		      value->length(), &end_not_used, &err_not_used);
@@ -7472,8 +7489,8 @@ longlong Item_cache_str::val_int()
 {
   DBUG_ASSERT(fixed == 1);
   int err;
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return 0;
   if (value)
     return my_strntoll(value->charset(), value->ptr(),
 		       value->length(), 10, (char**) 0, &err);
@@ -7485,8 +7502,8 @@ longlong Item_cache_str::val_int()
 String* Item_cache_str::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return 0;
   return value;
 }
 
@@ -7494,8 +7511,8 @@ String* Item_cache_str::val_str(String *str)
 my_decimal *Item_cache_str::val_decimal(my_decimal *decimal_val)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return NULL;
   if (value)
     string2my_decimal(E_DEC_FATAL_ERROR, value, decimal_val);
   else
@@ -7506,8 +7523,8 @@ my_decimal *Item_cache_str::val_decimal(my_decimal *decimal_val)
 
 int Item_cache_str::save_in_field(Field *field, bool no_conversions)
 {
-  if (!value_cached)
-    cache_value();
+  if (!value_cached && !cache_value())
+    return 0;
   int res= Item_cache::save_in_field(field, no_conversions);
   return (is_varbinary && field->type() == MYSQL_TYPE_STRING &&
           value->length() < field->field_length) ? 1 : res;
@@ -7542,13 +7559,21 @@ bool Item_cache_row::setup(Item * item)
 
 void Item_cache_row::store(Item * item)
 {
+  example= item;
+  if (!item)
+  {
+    null_value= TRUE;
+    return;
+  }
   for (uint i= 0; i < item_count; i++)
     values[i]->store(item->element_index(i));
 }
 
 
-void Item_cache_row::cache_value()
+bool Item_cache_row::cache_value()
 {
+  if (!example)
+    return FALSE;
   value_cached= TRUE;
   null_value= 0;
   example->bring_value();
@@ -7557,6 +7582,7 @@ void Item_cache_row::cache_value()
     values[i]->cache_value();
     null_value|= values[i]->null_value;
   }
+  return TRUE;
 }
 
 
diff --git a/sql/item.h b/sql/item.h
index a963206a2dc..699cc8db35a 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -2170,6 +2170,23 @@ public:
     save_in_field(result_field, no_conversions);
   }
   void cleanup();
+  /*
+    This method is used for debug purposes to print the name of an
+    item to the debug log. The second use of this method is as
+    a helper function of print() and error messages, where it is
+    applicable. To suit both goals it should return a meaningful,
+    distinguishable and sintactically correct string. This method
+    should not be used for runtime type identification, use enum
+    {Sum}Functype and Item_func::functype()/Item_sum::sum_func()
+    instead.
+    Added here, to the parent class of both Item_func and Item_sum_func.
+
+    NOTE: for Items inherited from Item_sum, func_name() return part of
+    function name till first argument (including '(') to make difference in
+    names for functions with 'distinct' clause and without 'distinct' and
+    also to make printing of items inherited from Item_sum uniform.
+  */
+  virtual const char *func_name() const= 0;
 };
 
 
@@ -2980,7 +2997,7 @@ public:
     return this == item;
   }
   virtual void store(Item *item);
-  virtual void cache_value()= 0;
+  virtual bool cache_value()= 0;
   bool basic_const_item() const
   { return test(example && example->basic_const_item());}
 };
@@ -3003,7 +3020,7 @@ public:
   my_decimal *val_decimal(my_decimal *);
   enum Item_result result_type() const { return INT_RESULT; }
   bool result_as_longlong() { return TRUE; }
-  void cache_value();
+  bool cache_value();
 };
 
 
@@ -3019,7 +3036,7 @@ public:
   String* val_str(String *str);
   my_decimal *val_decimal(my_decimal *);
   enum Item_result result_type() const { return REAL_RESULT; }
-  void cache_value();
+  bool cache_value();
 };
 
 
@@ -3035,7 +3052,7 @@ public:
   String* val_str(String *str);
   my_decimal *val_decimal(my_decimal *);
   enum Item_result result_type() const { return DECIMAL_RESULT; }
-  void cache_value();
+  bool cache_value();
 };
 
 
@@ -3060,7 +3077,7 @@ public:
   enum Item_result result_type() const { return STRING_RESULT; }
   CHARSET_INFO *charset() const { return value->charset(); };
   int save_in_field(Field *field, bool no_conversions);
-  void cache_value();
+  bool cache_value();
 };
 
 class Item_cache_row: public Item_cache
@@ -3129,7 +3146,7 @@ public:
       values= 0;
     DBUG_VOID_RETURN;
   }
-  void cache_value();
+  bool cache_value();
 };
 
 
@@ -3161,7 +3178,7 @@ public:
     correct conversion.
   */
   void cache_value_int();
-  void cache_value();
+  bool cache_value();
 };
 
 
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index fabb9a64b0d..4a1077179a9 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -30,6 +30,9 @@
 #include "sql_select.h"
 
 static bool convert_constant_item(THD *, Item_field *, Item **);
+static longlong
+get_year_value(THD *thd, Item ***item_arg, Item **cache_arg,
+               Item *warn_item, bool *is_null);
 
 static Item_result item_store_type(Item_result a, Item *item,
                                    my_bool unsigned_flag)
@@ -533,11 +536,12 @@ void Item_bool_func2::fix_length_and_dec()
 }
 
 
-int Arg_comparator::set_compare_func(Item_bool_func2 *item, Item_result type)
+int Arg_comparator::set_compare_func(Item_result_field *item, Item_result type)
 {
   owner= item;
   func= comparator_matrix[type]
-                         [test(owner->functype() == Item_func::EQUAL_FUNC)];
+                         [is_owner_equal_func()];
+
   switch (type) {
   case ROW_RESULT:
   {
@@ -557,7 +561,8 @@ int Arg_comparator::set_compare_func(Item_bool_func2 *item, Item_result type)
 	my_error(ER_OPERAND_COLUMNS, MYF(0), (*a)->element_index(i)->cols());
 	return 1;
       }
-      if (comparators[i].set_cmp_func(owner, (*a)->addr(i), (*b)->addr(i)))
+      if (comparators[i].set_cmp_func(owner, (*a)->addr(i), (*b)->addr(i),
+                                      set_null))
         return 1;
     }
     break;
@@ -571,7 +576,8 @@ int Arg_comparator::set_compare_func(Item_bool_func2 *item, Item_result type)
     if (cmp_collation.set((*a)->collation, (*b)->collation) || 
 	cmp_collation.derivation == DERIVATION_NONE)
     {
-      my_coll_agg_error((*a)->collation, (*b)->collation, owner->func_name());
+      my_coll_agg_error((*a)->collation, (*b)->collation,
+                        owner->func_name());
       return 1;
     }
     if (cmp_collation.collation == &my_charset_bin)
@@ -785,15 +791,21 @@ Arg_comparator::can_compare_as_dates(Item *a, Item *b, ulonglong *const_value)
 
   if (cmp_type != CMP_DATE_DFLT)
   {
+    THD *thd= current_thd;
     /*
       Do not cache GET_USER_VAR() function as its const_item() may return TRUE
       for the current thread but it still may change during the execution.
+      Don't use cache while in the context analysis mode only (i.e. for 
+      EXPLAIN/CREATE VIEW and similar queries). Cache is useless in such 
+      cases and can cause problems. For example evaluating subqueries can 
+      confuse storage engines since in context analysis mode tables 
+      aren't locked.
     */
-    if (cmp_type != CMP_DATE_WITH_DATE && str_arg->const_item() &&
+    if (!thd->is_context_analysis_only() &&
+        cmp_type != CMP_DATE_WITH_DATE && str_arg->const_item() &&
         (str_arg->type() != Item::FUNC_ITEM ||
         ((Item_func*)str_arg)->functype() != Item_func::GUSERVAR_FUNC))
     {
-      THD *thd= current_thd;
       ulonglong value;
       bool error;
       String tmp, *str_val= 0;
@@ -876,7 +888,7 @@ get_time_value(THD *thd, Item ***item_arg, Item **cache_arg,
 }
 
 
-int Arg_comparator::set_cmp_func(Item_bool_func2 *owner_arg,
+int Arg_comparator::set_cmp_func(Item_result_field *owner_arg,
                                         Item **a1, Item **a2,
                                         Item_result type)
 {
@@ -886,6 +898,8 @@ int Arg_comparator::set_cmp_func(Item_bool_func2 *owner_arg,
   owner= owner_arg;
   a= a1;
   b= a2;
+  owner= owner_arg;
+  thd= current_thd;
 
   if ((cmp_type= can_compare_as_dates(*a, *b, &const_value)))
   {
@@ -916,9 +930,10 @@ int Arg_comparator::set_cmp_func(Item_bool_func2 *owner_arg,
         b= (Item **)&b_cache;
       }
     }
-    is_nulls_eq= test(owner && owner->functype() == Item_func::EQUAL_FUNC);
+    is_nulls_eq= is_owner_equal_func();
     func= &Arg_comparator::compare_datetime;
-    get_value_func= &get_datetime_value;
+    get_value_a_func= &get_datetime_value;
+    get_value_b_func= &get_datetime_value;
     return 0;
   }
   else if (type == STRING_RESULT && (*a)->field_type() == MYSQL_TYPE_TIME &&
@@ -927,9 +942,10 @@ int Arg_comparator::set_cmp_func(Item_bool_func2 *owner_arg,
     /* Compare TIME values as integers. */
     a_cache= 0;
     b_cache= 0;
-    is_nulls_eq= test(owner && owner->functype() == Item_func::EQUAL_FUNC);
+    is_nulls_eq= is_owner_equal_func();
     func= &Arg_comparator::compare_datetime;
-    get_value_func= &get_time_value;
+    get_value_a_func= &get_time_value;
+    get_value_b_func= &get_time_value;
     return 0;
   }
   else if (type == STRING_RESULT &&
@@ -938,9 +954,42 @@ int Arg_comparator::set_cmp_func(Item_bool_func2 *owner_arg,
   {
     DTCollation coll;
     coll.set((*a)->collation.collation);
-    if (agg_item_set_converter(coll, owner_arg->func_name(),
+    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;
   }
 
   a= cache_converted_constant(thd, a, &a_cache, type);
@@ -983,11 +1032,11 @@ Item** Arg_comparator::cache_converted_constant(THD *thd, Item **value,
 }
 
 
-void Arg_comparator::set_datetime_cmp_func(Item **a1, Item **b1)
+void Arg_comparator::set_datetime_cmp_func(Item_result_field *owner_arg,
+                                           Item **a1, Item **b1)
 {
   thd= current_thd;
-  /* A caller will handle null values by itself. */
-  owner= NULL;
+  owner= owner_arg;
   a= a1;
   b= b1;
   a_type= (*a)->field_type();
@@ -996,7 +1045,8 @@ void Arg_comparator::set_datetime_cmp_func(Item **a1, Item **b1)
   b_cache= 0;
   is_nulls_eq= FALSE;
   func= &Arg_comparator::compare_datetime;
-  get_value_func= &get_datetime_value;
+  get_value_a_func= &get_datetime_value;
+  get_value_b_func= &get_datetime_value;
 }
 
 
@@ -1096,6 +1146,51 @@ get_datetime_value(THD *thd, Item ***item_arg, Item **cache_arg,
   return value;
 }
 
+
+/*
+  Retrieves YEAR value of 19XX form from given item.
+
+  SYNOPSIS
+    get_year_value()
+    thd                 thread handle
+    item_arg   [in/out] item to retrieve YEAR value from
+    cache_arg  [in/out] pointer to place to store the caching item to
+    warn_item  [in]     item for issuing the conversion warning
+    is_null    [out]    TRUE <=> the item_arg is null
+
+  DESCRIPTION
+    Retrieves the YEAR value of 19XX form from given item for comparison by the
+    compare_year() function.
+
+  RETURN
+    obtained value
+*/
+
+static longlong
+get_year_value(THD *thd, Item ***item_arg, Item **cache_arg,
+               Item *warn_item, bool *is_null)
+{
+  longlong value= 0;
+  Item *item= **item_arg;
+
+  value= item->val_int();
+  *is_null= item->null_value;
+  if (*is_null)
+    return ~(ulonglong) 0;
+
+  /*
+    Coerce value to the 19XX form in order to correctly compare
+    YEAR(2) & YEAR(4) types.
+  */
+  if (value < 70)
+    value+= 100;
+  if (value <= 1900)
+    value+= 1900;
+
+  return value;
+}
+
+
 /*
   Compare items values as dates.
 
@@ -1128,25 +1223,25 @@ int Arg_comparator::compare_datetime()
   longlong a_value, b_value;
 
   /* Get DATE/DATETIME/TIME value of the 'a' item. */
-  a_value= (*get_value_func)(thd, &a, &a_cache, *b, &a_is_null);
+  a_value= (*get_value_a_func)(thd, &a, &a_cache, *b, &a_is_null);
   if (!is_nulls_eq && a_is_null)
   {
-    if (owner)
+    if (set_null)
       owner->null_value= 1;
     return -1;
   }
 
   /* Get DATE/DATETIME/TIME value of the 'b' item. */
-  b_value= (*get_value_func)(thd, &b, &b_cache, *a, &b_is_null);
+  b_value= (*get_value_b_func)(thd, &b, &b_cache, *a, &b_is_null);
   if (a_is_null || b_is_null)
   {
-    if (owner)
+    if (set_null)
       owner->null_value= is_nulls_eq ? 0 : 1;
     return is_nulls_eq ? (a_is_null == b_is_null) : -1;
   }
 
   /* Here we have two not-NULL values. */
-  if (owner)
+  if (set_null)
     owner->null_value= 0;
 
   /* Compare values. */
@@ -1159,15 +1254,17 @@ int Arg_comparator::compare_datetime()
 int Arg_comparator::compare_string()
 {
   String *res1,*res2;
-  if ((res1= (*a)->val_str(&owner->tmp_value1)))
+  if ((res1= (*a)->val_str(&value1)))
   {
-    if ((res2= (*b)->val_str(&owner->tmp_value2)))
+    if ((res2= (*b)->val_str(&value2)))
     {
-      owner->null_value= 0;
+      if (set_null)
+        owner->null_value= 0;
       return sortcmp(res1,res2,cmp_collation.collation);
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1186,18 +1283,20 @@ int Arg_comparator::compare_string()
 int Arg_comparator::compare_binary_string()
 {
   String *res1,*res2;
-  if ((res1= (*a)->val_str(&owner->tmp_value1)))
+  if ((res1= (*a)->val_str(&value1)))
   {
-    if ((res2= (*b)->val_str(&owner->tmp_value2)))
+    if ((res2= (*b)->val_str(&value2)))
     {
-      owner->null_value= 0;
+      if (set_null)
+        owner->null_value= 0;
       uint res1_length= res1->length();
       uint res2_length= res2->length();
       int cmp= memcmp(res1->ptr(), res2->ptr(), min(res1_length,res2_length));
       return cmp ? cmp : (int) (res1_length - res2_length);
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1210,8 +1309,8 @@ int Arg_comparator::compare_binary_string()
 int Arg_comparator::compare_e_string()
 {
   String *res1,*res2;
-  res1= (*a)->val_str(&owner->tmp_value1);
-  res2= (*b)->val_str(&owner->tmp_value2);
+  res1= (*a)->val_str(&value1);
+  res2= (*b)->val_str(&value2);
   if (!res1 || !res2)
     return test(res1 == res2);
   return test(sortcmp(res1, res2, cmp_collation.collation) == 0);
@@ -1221,8 +1320,8 @@ int Arg_comparator::compare_e_string()
 int Arg_comparator::compare_e_binary_string()
 {
   String *res1,*res2;
-  res1= (*a)->val_str(&owner->tmp_value1);
-  res2= (*b)->val_str(&owner->tmp_value2);
+  res1= (*a)->val_str(&value1);
+  res2= (*b)->val_str(&value2);
   if (!res1 || !res2)
     return test(res1 == res2);
   return test(stringcmp(res1, res2) == 0);
@@ -1243,13 +1342,15 @@ int Arg_comparator::compare_real()
     val2= (*b)->val_real();
     if (!(*b)->null_value)
     {
-      owner->null_value= 0;
+      if (set_null)
+        owner->null_value= 0;
       if (val1 < val2)	return -1;
       if (val1 == val2) return 0;
       return 1;
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1263,11 +1364,13 @@ int Arg_comparator::compare_decimal()
     my_decimal *val2= (*b)->val_decimal(&value2);
     if (!(*b)->null_value)
     {
-      owner->null_value= 0;
+      if (set_null)
+        owner->null_value= 0;
       return my_decimal_cmp(val1, val2);
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1305,7 +1408,8 @@ int Arg_comparator::compare_real_fixed()
     val2= (*b)->val_real();
     if (!(*b)->null_value)
     {
-      owner->null_value= 0;
+      if (set_null)
+        owner->null_value= 0;
       if (val1 == val2 || fabs(val1 - val2) < precision)
         return 0;
       if (val1 < val2)
@@ -1313,7 +1417,8 @@ int Arg_comparator::compare_real_fixed()
       return 1;
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1336,13 +1441,15 @@ int Arg_comparator::compare_int_signed()
     longlong val2= (*b)->val_int();
     if (!(*b)->null_value)
     {
-      owner->null_value= 0;
+      if (set_null)
+        owner->null_value= 0;
       if (val1 < val2)	return -1;
       if (val1 == val2)   return 0;
       return 1;
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1359,13 +1466,15 @@ int Arg_comparator::compare_int_unsigned()
     ulonglong val2= (*b)->val_int();
     if (!(*b)->null_value)
     {
-      owner->null_value= 0;
+      if (set_null)
+        owner->null_value= 0;
       if (val1 < val2)	return -1;
       if (val1 == val2)   return 0;
       return 1;
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1382,7 +1491,8 @@ int Arg_comparator::compare_int_signed_unsigned()
     ulonglong uval2= (ulonglong)(*b)->val_int();
     if (!(*b)->null_value)
     {
-      owner->null_value= 0;
+      if (set_null)
+        owner->null_value= 0;
       if (sval1 < 0 || (ulonglong)sval1 < uval2)
         return -1;
       if ((ulonglong)sval1 == uval2)
@@ -1390,7 +1500,8 @@ int Arg_comparator::compare_int_signed_unsigned()
       return 1;
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1407,7 +1518,8 @@ int Arg_comparator::compare_int_unsigned_signed()
     longlong sval2= (*b)->val_int();
     if (!(*b)->null_value)
     {
-      owner->null_value= 0;
+      if (set_null)
+        owner->null_value= 0;
       if (sval2 < 0)
         return 1;
       if (uval1 < (ulonglong)sval2)
@@ -1417,7 +1529,8 @@ int Arg_comparator::compare_int_unsigned_signed()
       return 1;
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1453,10 +1566,11 @@ int Arg_comparator::compare_row()
   for (uint i= 0; i<n; i++)
   {
     res= comparators[i].compare();
-    if (owner->null_value)
+    /* Aggregate functions don't need special null handling. */
+    if (owner->null_value && owner->type() == Item::FUNC_ITEM)
     {
       // NULL was compared
-      switch (owner->functype()) {
+      switch (((Item_func*)owner)->functype()) {
       case Item_func::NE_FUNC:
         break; // NE never aborts on NULL even if abort_on_null is set
       case Item_func::LT_FUNC:
@@ -1465,7 +1579,7 @@ int Arg_comparator::compare_row()
       case Item_func::GE_FUNC:
         return -1; // <, <=, > and >= always fail on NULL
       default: // EQ_FUNC
-        if (owner->abort_on_null)
+        if (((Item_bool_func2*)owner)->abort_on_null)
           return -1; // We do not need correct NULL returning
       }
       was_null= 1;
@@ -1502,6 +1616,67 @@ 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;
@@ -1789,8 +1964,8 @@ longlong Item_func_lt::val_int()
 longlong Item_func_strcmp::val_int()
 {
   DBUG_ASSERT(fixed == 1);
-  String *a=args[0]->val_str(&tmp_value1);
-  String *b=args[1]->val_str(&tmp_value2);
+  String *a=args[0]->val_str(&cmp.value1);
+  String *b=args[1]->val_str(&cmp.value2);
   if (!a || !b)
   {
     null_value=1;
@@ -2073,8 +2248,8 @@ void Item_func_between::fix_length_and_dec()
 
   if (compare_as_dates)
   {
-    ge_cmp.set_datetime_cmp_func(args, args + 1);
-    le_cmp.set_datetime_cmp_func(args, args + 2);
+    ge_cmp.set_datetime_cmp_func(this, args, args + 1);
+    le_cmp.set_datetime_cmp_func(this, args, args + 2);
   }
   else if (time_items_found == 3)
   {
@@ -4402,13 +4577,13 @@ void Item_func_isnotnull::print(String *str, enum_query_type query_type)
 longlong Item_func_like::val_int()
 {
   DBUG_ASSERT(fixed == 1);
-  String* res = args[0]->val_str(&tmp_value1);
+  String* res = args[0]->val_str(&cmp.value1);
   if (args[0]->null_value)
   {
     null_value=1;
     return 0;
   }
-  String* res2 = args[1]->val_str(&tmp_value2);
+  String* res2 = args[1]->val_str(&cmp.value2);
   if (args[1]->null_value)
   {
     null_value=1;
@@ -4432,7 +4607,7 @@ Item_func::optimize_type Item_func_like::select_optimize() const
 {
   if (args[1]->const_item())
   {
-    String* res2= args[1]->val_str((String *)&tmp_value2);
+    String* res2= args[1]->val_str((String *)&cmp.value2);
 
     if (!res2)
       return OPTIMIZE_NONE;
@@ -4463,7 +4638,7 @@ bool Item_func_like::fix_fields(THD *thd, Item **ref)
   if (escape_item->const_item())
   {
     /* If we are on execution stage */
-    String *escape_str= escape_item->val_str(&tmp_value1);
+    String *escape_str= escape_item->val_str(&cmp.value1);
     if (escape_str)
     {
       if (escape_used_in_parsing && (
@@ -4518,7 +4693,7 @@ bool Item_func_like::fix_fields(THD *thd, Item **ref)
     if (args[1]->const_item() && !use_strnxfrm(collation.collation) &&
        !(specialflag & SPECIAL_NO_NEW_FUNC))
     {
-      String* res2 = args[1]->val_str(&tmp_value2);
+      String* res2 = args[1]->val_str(&cmp.value2);
       if (!res2)
         return FALSE;				// Null argument
       
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index 488909ae761..7805411dfa2 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -35,7 +35,7 @@ class Arg_comparator: public Sql_alloc
 {
   Item **a, **b;
   arg_cmp_func func;
-  Item_bool_func2 *owner;
+  Item_result_field *owner;
   Arg_comparator *comparators;   // used only for compare_row()
   double precision;
   /* Fields used in DATE/DATETIME comparison. */
@@ -43,30 +43,42 @@ class Arg_comparator: public Sql_alloc
   enum_field_types a_type, b_type; // Types of a and b items
   Item *a_cache, *b_cache;         // Cached values of a and b items
   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_func)(THD *thd, Item ***item_arg, Item **cache_arg,
-                             Item *warn_item, bool *is_null);
+  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);
 public:
   DTCollation cmp_collation;
+  /* Allow owner function to use string buffers. */
+  String value1, value2;
 
-  Arg_comparator(): thd(0), a_cache(0), b_cache(0) {};
+  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) {};
   Arg_comparator(Item **a1, Item **a2): a(a1), b(a2), thd(0),
-    a_cache(0), b_cache(0) {};
+    a_cache(0), b_cache(0), set_null(0), year_as_datetime(0),
+    get_value_a_func(0), get_value_b_func(0) {};
 
-  int set_compare_func(Item_bool_func2 *owner, Item_result type);
-  inline int set_compare_func(Item_bool_func2 *owner_arg)
+  int set_compare_func(Item_result_field *owner, Item_result type);
+  inline int set_compare_func(Item_result_field *owner_arg)
   {
     return set_compare_func(owner_arg, item_cmp_type((*a)->result_type(),
                                                      (*b)->result_type()));
   }
-  int set_cmp_func(Item_bool_func2 *owner_arg,
+  int set_cmp_func(Item_result_field *owner_arg,
 			  Item **a1, Item **a2,
 			  Item_result type);
 
-  inline int set_cmp_func(Item_bool_func2 *owner_arg,
-			  Item **a1, Item **a2)
+  inline int set_cmp_func(Item_result_field *owner_arg,
+			  Item **a1, Item **a2, bool set_null_arg)
   {
+    set_null= set_null_arg;
     return set_cmp_func(owner_arg, a1, a2,
                         item_cmp_type((*a1)->result_type(),
                                       (*a2)->result_type()));
@@ -92,14 +104,20 @@ 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);
 
-  void set_datetime_cmp_func(Item **a1, Item **b1);
   Item** cache_converted_constant(THD *thd, Item **value, Item **cache,
                                   Item_result type);
+  void set_datetime_cmp_func(Item_result_field *owner_arg, Item **a1, Item **b1);
   static arg_cmp_func comparator_matrix [5][2];
+  inline bool is_owner_equal_func()
+  {
+    return (owner->type() == Item::FUNC_ITEM &&
+           ((Item_func*)owner)->functype() == Item_func::EQUAL_FUNC);
+  }
 
   friend class Item_func;
 };
@@ -329,7 +347,6 @@ class Item_bool_func2 :public Item_int_func
 {						/* Bool with 2 string args */
 protected:
   Arg_comparator cmp;
-  String tmp_value1,tmp_value2;
   bool abort_on_null;
 
 public:
@@ -338,7 +355,7 @@ public:
   void fix_length_and_dec();
   void set_cmp_func()
   {
-    cmp.set_cmp_func(this, tmp_arg, tmp_arg+1);
+    cmp.set_cmp_func(this, tmp_arg, tmp_arg+1, TRUE);
   }
   optimize_type select_optimize() const { return OPTIMIZE_OP; }
   virtual enum Functype rev_functype() const { return UNKNOWN_FUNC; }
diff --git a/sql/item_create.cc b/sql/item_create.cc
index 070ab5263dd..c00b5ec1701 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -3561,6 +3561,7 @@ Create_func_get_lock Create_func_get_lock::s_singleton;
 Item*
 Create_func_get_lock::create(THD *thd, Item *arg1, Item *arg2)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_get_lock(arg1, arg2);
 }
@@ -3672,6 +3673,7 @@ Create_func_is_free_lock Create_func_is_free_lock::s_singleton;
 Item*
 Create_func_is_free_lock::create(THD *thd, Item *arg1)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_is_free_lock(arg1);
 }
@@ -3682,6 +3684,7 @@ Create_func_is_used_lock Create_func_is_used_lock::s_singleton;
 Item*
 Create_func_is_used_lock::create(THD *thd, Item *arg1)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_is_used_lock(arg1);
 }
@@ -3998,6 +4001,8 @@ Create_func_master_pos_wait::create_native(THD *thd, LEX_STRING name,
   Item *func= NULL;
   int arg_count= 0;
 
+  thd->lex->set_stmt_unsafe();
+
   if (item_list != NULL)
     arg_count= item_list->elements;
 
@@ -4240,6 +4245,7 @@ Create_func_release_lock Create_func_release_lock::s_singleton;
 Item*
 Create_func_release_lock::create(THD *thd, Item *arg1)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_release_lock(arg1);
 }
@@ -4362,6 +4368,7 @@ Create_func_sleep Create_func_sleep::s_singleton;
 Item*
 Create_func_sleep::create(THD *thd, Item *arg1)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_sleep(arg1);
 }
@@ -4637,6 +4644,7 @@ Create_func_version Create_func_version::s_singleton;
 Item*
 Create_func_version::create(THD *thd)
 {
+  thd->lex->set_stmt_unsafe();
   return new (thd->mem_root) Item_static_string_func("version()",
                                                      server_version,
                                                      (uint) strlen(server_version),
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 883476369f2..691af708d51 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -451,45 +451,8 @@ Field *Item_func::tmp_table_field(TABLE *table)
     return make_string_field(table);
     break;
   case DECIMAL_RESULT:
-  {
-    uint8 dec= decimals;
-    uint8 intg= decimal_precision() - dec;
-    uint32 len= max_length;
-
-    /*
-      Trying to put too many digits overall in a DECIMAL(prec,dec)
-      will always throw a warning. We must limit dec to
-      DECIMAL_MAX_SCALE however to prevent an assert() later.
-    */
-
-    if (dec > 0)
-    {
-      int overflow;
-
-      dec= min(dec, DECIMAL_MAX_SCALE);
-
-      /*
-        If the value still overflows the field with the corrected dec,
-        we'll throw out decimals rather than integers. This is still
-        bad and of course throws a truncation warning.
-      */
-
-      const int required_length=
-        my_decimal_precision_to_length(intg + dec, dec,
-                                                     unsigned_flag);
-
-      overflow= required_length - len;
-
-      if (overflow > 0)
-        dec= max(0, dec - overflow);            // too long, discard fract
-      else
-        /* Corrected value fits. */
-        len= required_length;
-    }
-
-    field= new Field_new_decimal(len, maybe_null, name, dec, unsigned_flag);
+    field= Field_new_decimal::create_from_item(this);
     break;
-  }
   case ROW_RESULT:
   default:
     // This case should never be chosen
diff --git a/sql/item_func.h b/sql/item_func.h
index 6f2b3f0bfc2..f22bc0c301c 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -127,17 +127,6 @@ public:
   virtual optimize_type select_optimize() const { return OPTIMIZE_NONE; }
   virtual bool have_rev_func() const { return 0; }
   virtual Item *key_item() const { return args[0]; }
-  /*
-    This method is used for debug purposes to print the name of an
-    item to the debug log. The second use of this method is as
-    a helper function of print(), where it is applicable.
-    To suit both goals it should return a meaningful,
-    distinguishable and sintactically correct string.  This method
-    should not be used for runtime type identification, use enum
-    {Sum}Functype and Item_func::functype()/Item_sum::sum_func()
-    instead.
-  */
-  virtual const char *func_name() const= 0;
   virtual bool const_item() const { return const_item_cache; }
   inline Item **arguments() const { return args; }
   void set_arguments(List<Item> &list);
diff --git a/sql/item_geofunc.cc b/sql/item_geofunc.cc
index 3c5990eb359..8c38cb2a859 100644
--- a/sql/item_geofunc.cc
+++ b/sql/item_geofunc.cc
@@ -511,8 +511,8 @@ err:
 longlong Item_func_spatial_rel::val_int()
 {
   DBUG_ASSERT(fixed == 1);
-  String *res1= args[0]->val_str(&tmp_value1);
-  String *res2= args[1]->val_str(&tmp_value2);
+  String *res1= args[0]->val_str(&cmp.value1);
+  String *res2= args[1]->val_str(&cmp.value2);
   Geometry_buffer buffer1, buffer2;
   Geometry *g1, *g2;
   MBR mbr1, mbr2;
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 923b943c950..95e28e791ec 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -1827,8 +1827,9 @@ String *Item_func_database::val_str(String *str)
 
 
 /**
-  @todo
-  make USER() replicate properly (currently it is replicated to "")
+  @note USER() is replicated correctly if binlog_format=ROW or (as of
+  BUG#28086) binlog_format=MIXED, but is incorrectly replicated to ''
+  if binlog_format=STATEMENT.
 */
 bool Item_func_user::init(const char *user, const char *host)
 {
diff --git a/sql/item_subselect.h b/sql/item_subselect.h
index ea59521aab1..3503d42edc0 100644
--- a/sql/item_subselect.h
+++ b/sql/item_subselect.h
@@ -135,6 +135,7 @@ public:
     @return the SELECT_LEX structure associated with this Item
   */
   st_select_lex* get_select_lex();
+  const char *func_name() const { DBUG_ASSERT(0); return "subselect"; }
 
   friend class select_subselect;
   friend class Item_in_optimizer;
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index f72b64fc66e..d925c68e66c 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -534,8 +534,7 @@ Field *Item_sum::create_tmp_field(bool group, TABLE *table,
                                name, table->s, collation.collation);
     break;
   case DECIMAL_RESULT:
-    field= new Field_new_decimal(max_length, maybe_null, name,
-                                 decimals, unsigned_flag);
+    field= Field_new_decimal::create_from_item(this);
     break;
   case ROW_RESULT:
   default:
@@ -1137,35 +1136,6 @@ Item_sum_num::fix_fields(THD *thd, Item **ref)
 }
 
 
-Item_sum_hybrid::Item_sum_hybrid(THD *thd, Item_sum_hybrid *item)
-  :Item_sum(thd, item), value(item->value), hybrid_type(item->hybrid_type),
-  hybrid_field_type(item->hybrid_field_type), cmp_sign(item->cmp_sign),
-  was_values(item->was_values)
-{
-  /* copy results from old value */
-  switch (hybrid_type) {
-  case INT_RESULT:
-    sum_int= item->sum_int;
-    break;
-  case DECIMAL_RESULT:
-    my_decimal2decimal(&item->sum_dec, &sum_dec);
-    break;
-  case REAL_RESULT:
-    sum= item->sum;
-    break;
-  case STRING_RESULT:
-    /*
-      This can happen with ROLLUP. Note that the value is already
-      copied at function call.
-    */
-    break;
-  case ROW_RESULT:
-  default:
-    DBUG_ASSERT(0);
-  }
-  collation.set(item->collation);
-}
-
 bool
 Item_sum_hybrid::fix_fields(THD *thd, Item **ref)
 {
@@ -1185,15 +1155,12 @@ Item_sum_hybrid::fix_fields(THD *thd, Item **ref)
   switch (hybrid_type= item->result_type()) {
   case INT_RESULT:
     max_length= 20;
-    sum_int= 0;
     break;
   case DECIMAL_RESULT:
     max_length= item->max_length;
-    my_decimal_set_zero(&sum_dec);
     break;
   case REAL_RESULT:
     max_length= float_length(decimals);
-    sum= 0.0;
     break;
   case STRING_RESULT:
     max_length= item->max_length;
@@ -1202,10 +1169,10 @@ Item_sum_hybrid::fix_fields(THD *thd, Item **ref)
   default:
     DBUG_ASSERT(0);
   };
+  setup(args[0], NULL);
   /* MIN/MAX can return NULL for empty set indepedent of the used column */
   maybe_null= 1;
   unsigned_flag=item->unsigned_flag;
-  collation.set(item->collation);
   result_field=0;
   null_value=1;
   fix_length_and_dec();
@@ -1223,6 +1190,30 @@ Item_sum_hybrid::fix_fields(THD *thd, Item **ref)
   return FALSE;
 }
 
+
+/**
+  MIN/MAX function setup.
+
+  @param item       argument of MIN/MAX function
+  @param value_arg  calculated value of MIN/MAX function
+
+  @details
+    Setup cache/comparator of MIN/MAX functions. When called by the
+    copy_or_same function value_arg parameter contains calculated value
+    of the original MIN/MAX object and it is saved in this object's cache.
+*/
+
+void Item_sum_hybrid::setup(Item *item, Item *value_arg)
+{
+  value= Item_cache::get_cache(item);
+  value->setup(item);
+  value->store(value_arg);
+  cmp= new Arg_comparator();
+  cmp->set_cmp_func(this, args, (Item**)&value, FALSE);
+  collation.set(item->collation);
+}
+
+
 Field *Item_sum_hybrid::create_tmp_field(bool group, TABLE *table,
 					 uint convert_blob_length)
 {
@@ -1577,8 +1568,7 @@ Field *Item_sum_avg::create_tmp_field(bool group, TABLE *table,
                             0, name, &my_charset_bin);
   }
   else if (hybrid_type == DECIMAL_RESULT)
-    field= new Field_new_decimal(max_length, maybe_null, name,
-                                 decimals, unsigned_flag);
+    field= Field_new_decimal::create_from_item(this);
   else
     field= new Field_double(max_length, maybe_null, name, decimals, TRUE);
   if (field)
@@ -1905,19 +1895,7 @@ void Item_sum_variance::update_field()
 
 void Item_sum_hybrid::clear()
 {
-  switch (hybrid_type) {
-  case INT_RESULT:
-    sum_int= 0;
-    break;
-  case DECIMAL_RESULT:
-    my_decimal_set_zero(&sum_dec);
-    break;
-  case REAL_RESULT:
-    sum= 0.0;
-    break;
-  default:
-    value.length(0);
-  }
+  value->null_value= 1;
   null_value= 1;
 }
 
@@ -1926,30 +1904,7 @@ double Item_sum_hybrid::val_real()
   DBUG_ASSERT(fixed == 1);
   if (null_value)
     return 0.0;
-  switch (hybrid_type) {
-  case STRING_RESULT:
-  {
-    char *end_not_used;
-    int err_not_used;
-    String *res;  res=val_str(&str_value);
-    return (res ? my_strntod(res->charset(), (char*) res->ptr(), res->length(),
-			     &end_not_used, &err_not_used) : 0.0);
-  }
-  case INT_RESULT:
-    if (unsigned_flag)
-      return ulonglong2double(sum_int);
-    return (double) sum_int;
-  case DECIMAL_RESULT:
-    my_decimal2double(E_DEC_FATAL_ERROR, &sum_dec, &sum);
-    return sum;
-  case REAL_RESULT:
-    return sum;
-  case ROW_RESULT:
-  default:
-    // This case should never be choosen
-    DBUG_ASSERT(0);
-    return 0;
-  }
+  return value->val_real();
 }
 
 longlong Item_sum_hybrid::val_int()
@@ -1957,18 +1912,7 @@ longlong Item_sum_hybrid::val_int()
   DBUG_ASSERT(fixed == 1);
   if (null_value)
     return 0;
-  switch (hybrid_type) {
-  case INT_RESULT:
-    return sum_int;
-  case DECIMAL_RESULT:
-  {
-    longlong result;
-    my_decimal2int(E_DEC_FATAL_ERROR, &sum_dec, unsigned_flag, &result);
-    return sum_int;
-  }
-  default:
-    return (longlong) rint(Item_sum_hybrid::val_real());
-  }
+  return value->val_int();
 }
 
 
@@ -1977,26 +1921,7 @@ my_decimal *Item_sum_hybrid::val_decimal(my_decimal *val)
   DBUG_ASSERT(fixed == 1);
   if (null_value)
     return 0;
-  switch (hybrid_type) {
-  case STRING_RESULT:
-    string2my_decimal(E_DEC_FATAL_ERROR, &value, val);
-    break;
-  case REAL_RESULT:
-    double2my_decimal(E_DEC_FATAL_ERROR, sum, val);
-    break;
-  case DECIMAL_RESULT:
-    val= &sum_dec;
-    break;
-  case INT_RESULT:
-    int2my_decimal(E_DEC_FATAL_ERROR, sum_int, unsigned_flag, val);
-    break;
-  case ROW_RESULT:
-  default:
-    // This case should never be choosen
-    DBUG_ASSERT(0);
-    break;
-  }
-  return val;					// Keep compiler happy
+  return value->val_decimal(val);
 }
 
 
@@ -2006,25 +1931,7 @@ Item_sum_hybrid::val_str(String *str)
   DBUG_ASSERT(fixed == 1);
   if (null_value)
     return 0;
-  switch (hybrid_type) {
-  case STRING_RESULT:
-    return &value;
-  case REAL_RESULT:
-    str->set_real(sum,decimals, &my_charset_bin);
-    break;
-  case DECIMAL_RESULT:
-    my_decimal2string(E_DEC_FATAL_ERROR, &sum_dec, 0, 0, 0, str);
-    return str;
-  case INT_RESULT:
-    str->set_int(sum_int, unsigned_flag, &my_charset_bin);
-    break;
-  case ROW_RESULT:
-  default:
-    // This case should never be choosen
-    DBUG_ASSERT(0);
-    break;
-  }
-  return str;					// Keep compiler happy
+  return value->val_str(str);
 }
 
 
@@ -2033,7 +1940,9 @@ void Item_sum_hybrid::cleanup()
   DBUG_ENTER("Item_sum_hybrid::cleanup");
   Item_sum::cleanup();
   forced_const= FALSE;
-
+  if (cmp)
+    delete cmp;
+  cmp= 0;
   /*
     by default it is TRUE to avoid TRUE reporting by
     Item_func_not_all/Item_func_nop_all if this item was never called.
@@ -2054,63 +1963,22 @@ void Item_sum_hybrid::no_rows_in_result()
 
 Item *Item_sum_min::copy_or_same(THD* thd)
 {
-  return new (thd->mem_root) Item_sum_min(thd, this);
+  Item_sum_min *item= new (thd->mem_root) Item_sum_min(thd, this);
+  item->setup(args[0], value);
+  return item;
 }
 
 
 bool Item_sum_min::add()
 {
-  switch (hybrid_type) {
-  case STRING_RESULT:
-  {
-    String *result=args[0]->val_str(&tmp_value);
-    if (!args[0]->null_value &&
-	(null_value || sortcmp(&value,result,collation.collation) > 0))
-    {
-      value.copy(*result);
-      null_value=0;
-    }
-  }
-  break;
-  case INT_RESULT:
-  {
-    longlong nr=args[0]->val_int();
-    if (!args[0]->null_value && (null_value ||
-				 (unsigned_flag && 
-				  (ulonglong) nr < (ulonglong) sum_int) ||
-				 (!unsigned_flag && nr < sum_int)))
-    {
-      sum_int=nr;
-      null_value=0;
-    }
-  }
-  break;
-  case DECIMAL_RESULT:
-  {
-    my_decimal value_buff, *val= args[0]->val_decimal(&value_buff);
-    if (!args[0]->null_value &&
-        (null_value || (my_decimal_cmp(&sum_dec, val) > 0)))
-    {
-      my_decimal2decimal(val, &sum_dec);
-      null_value= 0;
-    }
-  }
-  break;
-  case REAL_RESULT:
+  /* args[0] < value */
+  int res= cmp->compare();
+  if (!args[0]->null_value &&
+      (null_value || res < 0))
   {
-    double nr= args[0]->val_real();
-    if (!args[0]->null_value && (null_value || nr < sum))
-    {
-      sum=nr;
-      null_value=0;
-    }
-  }
-  break;
-  case ROW_RESULT:
-  default:
-    // This case should never be choosen
-    DBUG_ASSERT(0);
-    break;
+    value->store(args[0]);
+    value->cache_value();
+    null_value= 0;
   }
   return 0;
 }
@@ -2118,63 +1986,22 @@ bool Item_sum_min::add()
 
 Item *Item_sum_max::copy_or_same(THD* thd)
 {
-  return new (thd->mem_root) Item_sum_max(thd, this);
+  Item_sum_max *item= new (thd->mem_root) Item_sum_max(thd, this);
+  item->setup(args[0], value);
+  return item;
 }
 
 
 bool Item_sum_max::add()
 {
-  switch (hybrid_type) {
-  case STRING_RESULT:
+  /* args[0] > value */
+  int res= cmp->compare();
+  if (!args[0]->null_value &&
+      (null_value || res > 0))
   {
-    String *result=args[0]->val_str(&tmp_value);
-    if (!args[0]->null_value &&
-	(null_value || sortcmp(&value,result,collation.collation) < 0))
-    {
-      value.copy(*result);
-      null_value=0;
-    }
-  }
-  break;
-  case INT_RESULT:
-  {
-    longlong nr=args[0]->val_int();
-    if (!args[0]->null_value && (null_value ||
-				 (unsigned_flag && 
-				  (ulonglong) nr > (ulonglong) sum_int) ||
-				 (!unsigned_flag && nr > sum_int)))
-    {
-      sum_int=nr;
-      null_value=0;
-    }
-  }
-  break;
-  case DECIMAL_RESULT:
-  {
-    my_decimal value_buff, *val= args[0]->val_decimal(&value_buff);
-    if (!args[0]->null_value &&
-        (null_value || (my_decimal_cmp(val, &sum_dec) > 0)))
-    {
-      my_decimal2decimal(val, &sum_dec);
-      null_value= 0;
-    }
-  }
-  break;
-  case REAL_RESULT:
-  {
-    double nr= args[0]->val_real();
-    if (!args[0]->null_value && (null_value || nr > sum))
-    {
-      sum=nr;
-      null_value=0;
-    }
-  }
-  break;
-  case ROW_RESULT:
-  default:
-    // This case should never be choosen
-    DBUG_ASSERT(0);
-    break;
+    value->store(args[0]);
+    value->cache_value();
+    null_value= 0;
   }
   return 0;
 }
@@ -2546,14 +2373,15 @@ void Item_sum_hybrid::update_field()
 void
 Item_sum_hybrid::min_max_update_str_field()
 {
-  String *res_str=args[0]->val_str(&value);
+  DBUG_ASSERT(cmp);
+  String *res_str=args[0]->val_str(&cmp->value1);
 
   if (!args[0]->null_value)
   {
-    result_field->val_str(&tmp_value);
+    result_field->val_str(&cmp->value2);
 
     if (result_field->is_null() ||
-	(cmp_sign * sortcmp(res_str,&tmp_value,collation.collation)) < 0)
+	(cmp_sign * sortcmp(res_str,&cmp->value2,collation.collation)) < 0)
       result_field->store(res_str->ptr(),res_str->length(),res_str->charset());
     result_field->set_notnull();
   }
diff --git a/sql/item_sum.h b/sql/item_sum.h
index 9e062ce1d16..26f0a08096e 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -413,22 +413,6 @@ public:
   virtual void update_field()=0;
   virtual bool keep_field_type(void) const { return 0; }
   virtual void fix_length_and_dec() { maybe_null=1; null_value=1; }
-  /*
-    This method is used for debug purposes to print the name of an
-    item to the debug log. The second use of this method is as
-    a helper function of print(), where it is applicable.
-    To suit both goals it should return a meaningful,
-    distinguishable and sintactically correct string.  This method
-    should not be used for runtime type identification, use enum
-    {Sum}Functype and Item_func::functype()/Item_sum::sum_func()
-    instead.
-
-    NOTE: for Items inherited from Item_sum, func_name() return part of
-    function name till first argument (including '(') to make difference in
-    names for functions with 'distinct' clause and without 'distinct' and
-    also to make printing of items inherited from Item_sum uniform.
-  */
-  virtual const char *func_name() const= 0;
   virtual Item *result_item(Field *field)
     { return new Item_field(field); }
   table_map used_tables() const { return used_tables_cache; }
@@ -800,6 +784,7 @@ public:
   }
   void fix_length_and_dec() {}
   enum Item_result result_type () const { return hybrid_type; }
+  const char *func_name() const { DBUG_ASSERT(0); return "avg_field"; }
 };
 
 
@@ -876,6 +861,7 @@ public:
   }
   void fix_length_and_dec() {}
   enum Item_result result_type () const { return hybrid_type; }
+  const char *func_name() const { DBUG_ASSERT(0); return "variance_field"; }
 };
 
 
@@ -951,6 +937,7 @@ public:
   my_decimal *val_decimal(my_decimal *);
   enum Item_result result_type () const { return REAL_RESULT; }
   enum_field_types field_type() const { return MYSQL_TYPE_DOUBLE;}
+  const char *func_name() const { DBUG_ASSERT(0); return "std_field"; }
 };
 
 /*
@@ -976,14 +963,13 @@ class Item_sum_std :public Item_sum_variance
 };
 
 // This class is a string or number function depending on num_func
-
+class Arg_comparator;
+class Item_cache;
 class Item_sum_hybrid :public Item_sum
 {
 protected:
-  String value,tmp_value;
-  double sum;
-  longlong sum_int;
-  my_decimal sum_dec;
+  Item_cache *value;
+  Arg_comparator *cmp;
   Item_result hybrid_type;
   enum_field_types hybrid_field_type;
   int cmp_sign;
@@ -991,12 +977,17 @@ protected:
 
   public:
   Item_sum_hybrid(Item *item_par,int sign)
-    :Item_sum(item_par), sum(0.0), sum_int(0),
+    :Item_sum(item_par), value(0), cmp(0),
     hybrid_type(INT_RESULT), hybrid_field_type(MYSQL_TYPE_LONGLONG),
     cmp_sign(sign), was_values(TRUE)
   { collation.set(&my_charset_bin); }
-  Item_sum_hybrid(THD *thd, Item_sum_hybrid *item);
+  Item_sum_hybrid(THD *thd, Item_sum_hybrid *item)
+    :Item_sum(thd, item), value(item->value), hybrid_type(item->hybrid_type),
+    hybrid_field_type(item->hybrid_field_type), cmp_sign(item->cmp_sign),
+    was_values(item->was_values)
+  { }
   bool fix_fields(THD *, Item **);
+  void setup(Item *item, Item *value_arg);
   void clear();
   double val_real();
   longlong val_int();
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index 3009c48cac7..e4644aa6436 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -386,7 +386,7 @@ static bool extract_date_time(DATE_TIME_FORMAT *format,
 	if (tmp - val > 6)
 	  tmp= (char*) val + 6;
 	l_time->second_part= (int) my_strtoll10(val, &tmp, &error);
-	frac_part= 6 - (uint) (tmp - val);
+	frac_part= 6 - (int) (tmp - val);
 	if (frac_part > 0)
 	  l_time->second_part*= (ulong) log_10_int[frac_part];
 	val= tmp;
@@ -876,7 +876,12 @@ static bool get_interval_info(const char *str,uint length,CHARSET_INFO *cs,
     const char *start= str;
     for (value=0; str != end && my_isdigit(cs,*str) ; str++)
       value= value*LL(10) + (longlong) (*str - '0');
-    msec_length= 6 - (str - start);
+    if (transform_msec && i == count - 1) // microseconds always last
+    {
+      int msec_length= 6 - (str - start);
+      if (msec_length > 0)
+        value*= (long)log_10_int[msec_length];
+    }
     values[i]= value;
     while (str != end && !my_isdigit(cs,*str))
       str++;
diff --git a/sql/log.cc b/sql/log.cc
index f488dda1e56..9f479a128fc 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -5802,9 +5802,8 @@ int TC_LOG_BINLOG::recover(IO_CACHE *log, Format_description_log_event *fdle)
       Xid_log_event *xev=(Xid_log_event *)ev;
       uchar *x= (uchar *) memdup_root(&mem_root, (uchar*) &xev->xid,
                                       sizeof(xev->xid));
-      if (! x)
+      if (!x || my_hash_insert(&xids, x))
         goto err2;
-      my_hash_insert(&xids, x);
     }
     delete ev;
   }
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index c72e81d6ba1..2bc2fcfd384 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -3947,6 +3947,27 @@ will be ignored as the --log-bin option is not defined.");
 
   if (opt_bin_log)
   {
+    /* Reports an error and aborts, if the --log-bin's path 
+       is a directory.*/
+    if (opt_bin_logname && 
+        opt_bin_logname[strlen(opt_bin_logname) - 1] == FN_LIBCHAR)
+    {
+      sql_print_error("Path '%s' is a directory name, please specify \
+a file name for --log-bin option", opt_bin_logname);
+      unireg_abort(1);
+    }
+
+    /* Reports an error and aborts, if the --log-bin-index's path 
+       is a directory.*/
+    if (opt_binlog_index_name && 
+        opt_binlog_index_name[strlen(opt_binlog_index_name) - 1] 
+        == FN_LIBCHAR)
+    {
+      sql_print_error("Path '%s' is a directory name, please specify \
+a file name for --log-bin-index option", opt_binlog_index_name);
+      unireg_abort(1);
+    }
+
     char buf[FN_REFLEN];
     const char *ln;
     ln= mysql_bin_log.generate_name(opt_bin_logname, "-bin", 1, buf);
@@ -5242,12 +5263,16 @@ pthread_handler_t handle_connections_sockets_thread(void *arg)
 pthread_handler_t handle_connections_namedpipes(void *arg)
 {
   HANDLE hConnectedPipe;
-  OVERLAPPED connectOverlapped = {0};
+  OVERLAPPED connectOverlapped= {0};
   THD *thd;
   my_thread_init();
   DBUG_ENTER("handle_connections_namedpipes");
-  connectOverlapped.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
-
+  connectOverlapped.hEvent= CreateEvent(NULL, TRUE, FALSE, NULL);
+  if (!connectOverlapped.hEvent)
+  {
+    sql_print_error("Can't create event, last error=%u", GetLastError());
+    unireg_abort(1);
+  }
   DBUG_PRINT("general",("Waiting for named pipe connections."));
   while (!abort_loop)
   {
@@ -5270,7 +5295,8 @@ pthread_handler_t handle_connections_namedpipes(void *arg)
     {
       CloseHandle(hPipe);
       if ((hPipe= CreateNamedPipe(pipe_name,
-                                  PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED,
+                                  PIPE_ACCESS_DUPLEX |
+                                  FILE_FLAG_OVERLAPPED,
                                   PIPE_TYPE_BYTE |
                                   PIPE_READMODE_BYTE |
                                   PIPE_WAIT,
@@ -5290,7 +5316,8 @@ pthread_handler_t handle_connections_namedpipes(void *arg)
     hConnectedPipe = hPipe;
     /* create new pipe for new connection */
     if ((hPipe = CreateNamedPipe(pipe_name,
-				 PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED,
+                 PIPE_ACCESS_DUPLEX |
+                 FILE_FLAG_OVERLAPPED,
 				 PIPE_TYPE_BYTE |
 				 PIPE_READMODE_BYTE |
 				 PIPE_WAIT,
@@ -8679,14 +8706,8 @@ static int fix_paths(void)
     pos[0]= FN_LIBCHAR;
     pos[1]= 0;
   }
-  convert_dirname(mysql_real_data_home,mysql_real_data_home,NullS);
-  my_realpath(mysql_unpacked_real_data_home, mysql_real_data_home, MYF(0));
-  mysql_unpacked_real_data_home_len= strlen(mysql_unpacked_real_data_home);
-  if (mysql_unpacked_real_data_home[mysql_unpacked_real_data_home_len-1] == FN_LIBCHAR)
-    --mysql_unpacked_real_data_home_len;
-
-
   convert_dirname(lc_messages_dir, lc_messages_dir, NullS);
+  convert_dirname(mysql_real_data_home,mysql_real_data_home,NullS);
   (void) my_load_path(mysql_home,mysql_home,""); // Resolve current dir
   (void) my_load_path(mysql_real_data_home,mysql_real_data_home,mysql_home);
   (void) my_load_path(pidfile_name,pidfile_name,mysql_real_data_home);
@@ -8694,6 +8715,12 @@ static int fix_paths(void)
                                       get_relative_path(PLUGINDIR), mysql_home);
   opt_plugin_dir_ptr= opt_plugin_dir;
 
+  my_realpath(mysql_unpacked_real_data_home, mysql_real_data_home, MYF(0));
+  mysql_unpacked_real_data_home_len= 
+    (int) strlen(mysql_unpacked_real_data_home);
+  if (mysql_unpacked_real_data_home[mysql_unpacked_real_data_home_len-1] == FN_LIBCHAR)
+    --mysql_unpacked_real_data_home_len;
+
   char *sharedir=get_relative_path(SHAREDIR);
   if (test_if_hard_path(sharedir))
     strmake(buff,sharedir,sizeof(buff)-1);		/* purecov: tested */
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 731e9872aa5..ac5b1f575de 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -457,10 +457,10 @@ public:
                                   range_key, *range_key_flag);
     *range_key_flag|= key_tree->min_flag;
     if (key_tree->next_key_part &&
+	key_tree->next_key_part->type == SEL_ARG::KEY_RANGE &&
         key_tree->part != last_part &&
 	key_tree->next_key_part->part == key_tree->part+1 &&
-	!(*range_key_flag & (NO_MIN_RANGE | NEAR_MIN)) &&
-	key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
+	!(*range_key_flag & (NO_MIN_RANGE | NEAR_MIN)))
       res+= key_tree->next_key_part->store_min_key(key,
                                                    range_key,
                                                    range_key_flag,
@@ -479,10 +479,10 @@ public:
                                  range_key, *range_key_flag);
     (*range_key_flag)|= key_tree->max_flag;
     if (key_tree->next_key_part &&
+	key_tree->next_key_part->type == SEL_ARG::KEY_RANGE &&
         key_tree->part != last_part &&
-        key_tree->next_key_part->part == key_tree->part+1 &&
-        !(*range_key_flag & (NO_MAX_RANGE | NEAR_MAX)) &&
-        key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
+	key_tree->next_key_part->part == key_tree->part+1 &&
+	!(*range_key_flag & (NO_MAX_RANGE | NEAR_MAX)))
       res+= key_tree->next_key_part->store_max_key(key,
                                                    range_key,
                                                    range_key_flag,
@@ -1727,6 +1727,7 @@ SEL_ARG *SEL_ARG::clone(RANGE_OPT_PARAM *param, SEL_ARG *new_parent,
     tmp->prev= *next_arg;			// Link into next/prev chain
     (*next_arg)->next=tmp;
     (*next_arg)= tmp;
+    tmp->part= this->part;
   }
   else
   {
@@ -6848,6 +6849,7 @@ key_or(RANGE_OPT_PARAM *param, SEL_ARG *key1,SEL_ARG *key2)
     else if ((cmp=tmp->cmp_max_to_min(key2)) < 0)
     {						// Found tmp.max < key2.min
       SEL_ARG *next=tmp->next;
+      /* key1 on the left of key2 non-overlapping */
       if (cmp == -2 && eq_tree(tmp->next_key_part,key2->next_key_part))
       {
 	// Join near ranges like tmp.max < 0 and key2.min >= 0
@@ -6876,6 +6878,7 @@ key_or(RANGE_OPT_PARAM *param, SEL_ARG *key1,SEL_ARG *key2)
       int tmp_cmp;
       if ((tmp_cmp=tmp->cmp_min_to_max(key2)) > 0) // if tmp.min > key2.max
       {
+        /* tmp is on the right of key2 non-overlapping */
 	if (tmp_cmp == 2 && eq_tree(tmp->next_key_part,key2->next_key_part))
 	{					// ranges are connected
 	  tmp->copy_min_to_min(key2);
@@ -6910,25 +6913,52 @@ key_or(RANGE_OPT_PARAM *param, SEL_ARG *key1,SEL_ARG *key2)
       }
     }
 
-    // tmp.max >= key2.min && tmp.min <= key.max  (overlapping ranges)
+    /* 
+      tmp.min >= key2.min && tmp.min <= key.max  (overlapping ranges)
+      key2.min <= tmp.min <= key2.max 
+    */  
     if (eq_tree(tmp->next_key_part,key2->next_key_part))
     {
       if (tmp->is_same(key2))
       {
+        /* 
+          Found exact match of key2 inside key1. 
+          Use the relevant range in key1.
+        */
 	tmp->merge_flags(key2);			// Copy maybe flags
 	key2->increment_use_count(-1);		// Free not used tree
       }
       else
       {
 	SEL_ARG *last=tmp;
+        SEL_ARG *first=tmp;
+        /* 
+          Find the last range in tmp that overlaps key2 and has the same 
+          condition on the rest of the keyparts.
+        */
 	while (last->next && last->next->cmp_min_to_max(key2) <= 0 &&
 	       eq_tree(last->next->next_key_part,key2->next_key_part))
 	{
+          /*
+            We've found the last overlapping key1 range in last.
+            This means that the ranges between (and including) the 
+            first overlapping range (tmp) and the last overlapping range
+            (last) are fully nested into the current range of key2 
+            and can safely be discarded. We just need the minimum endpoint
+            of the first overlapping range (tmp) so we can compare it with
+            the minimum endpoint of the enclosing key2 range.
+          */
 	  SEL_ARG *save=last;
 	  last=last->next;
 	  key1=key1->tree_delete(save);
 	}
-        last->copy_min(tmp);
+        /*
+          The tmp range (the first overlapping range) could have been discarded
+          by the previous loop. We should re-direct tmp to the new united range 
+          that's taking its place.
+        */
+        tmp= last;
+        last->copy_min(first);
         bool full_range= last->copy_min(key2);
         if (!full_range)
         {
@@ -7438,27 +7468,25 @@ int test_rb_tree(SEL_ARG *element,SEL_ARG *parent)
 }
 
 
-/*
-  Count how many times SEL_ARG graph "root" refers to its part "key"
+/**
+  Count how many times SEL_ARG graph "root" refers to its part "key" via
+  transitive closure.
   
-  SYNOPSIS
-    count_key_part_usage()
-      root  An RB-Root node in a SEL_ARG graph.
-      key   Another RB-Root node in that SEL_ARG graph.
-
-  DESCRIPTION
-    The passed "root" node may refer to "key" node via root->next_key_part,
-    root->next->n
-
-    This function counts how many times the node "key" is referred (via
-    SEL_ARG::next_key_part) by 
-     - intervals of RB-tree pointed by "root", 
-     - intervals of RB-trees that are pointed by SEL_ARG::next_key_part from 
-       intervals of RB-tree pointed by "root",
-     - and so on.
+  @param root  An RB-Root node in a SEL_ARG graph.
+  @param key   Another RB-Root node in that SEL_ARG graph.
+
+  The passed "root" node may refer to "key" node via root->next_key_part,
+  root->next->n
+
+  This function counts how many times the node "key" is referred (via
+  SEL_ARG::next_key_part) by 
+  - intervals of RB-tree pointed by "root", 
+  - intervals of RB-trees that are pointed by SEL_ARG::next_key_part from 
+  intervals of RB-tree pointed by "root",
+  - and so on.
     
-    Here is an example (horizontal links represent next_key_part pointers, 
-    vertical links - next/prev prev pointers):  
+  Here is an example (horizontal links represent next_key_part pointers, 
+  vertical links - next/prev prev pointers):  
     
          +----+               $
          |root|-----------------+
@@ -7478,8 +7506,8 @@ int test_rb_tree(SEL_ARG *element,SEL_ARG *parent)
           ...     +---+       $    |
                   |   |------------+
                   +---+       $
-  RETURN 
-    Number of links to "key" from nodes reachable from "root".
+  @return 
+  Number of links to "key" from nodes reachable from "root".
 */
 
 static ulong count_key_part_usage(SEL_ARG *root, SEL_ARG *key)
@@ -7734,8 +7762,8 @@ check_quick_keys(PARAM *param, uint idx, SEL_ARG *key_tree,
     param->first_null_comp= key_tree->part+1;
 
   if (key_tree->next_key_part &&
-      key_tree->next_key_part->part == key_tree->part+1 &&
-      key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
+      key_tree->next_key_part->type == SEL_ARG::KEY_RANGE &&
+      key_tree->next_key_part->part == key_tree->part+1)
   {						// const key as prefix
     if (min_key_length == max_key_length &&
 	!memcmp(min_key, max_key, (uint) (tmp_max_key - max_key)) &&
@@ -8020,8 +8048,8 @@ get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key,
                                  &tmp_max_key,max_key_flag);
 
   if (key_tree->next_key_part &&
-      key_tree->next_key_part->part == key_tree->part+1 &&
-      key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
+      key_tree->next_key_part->type == SEL_ARG::KEY_RANGE &&
+      key_tree->next_key_part->part == key_tree->part+1)
   {						  // const key as prefix
     if ((tmp_min_key - min_key) == (tmp_max_key - max_key) &&
          memcmp(min_key, max_key, (uint)(tmp_max_key - max_key))==0 &&
@@ -10032,7 +10060,11 @@ check_group_min_max_predicates(COND *cond, Item_field *min_max_arg_item,
     }
     else if (cur_arg->const_item())
     {
-      DBUG_RETURN(TRUE);
+      /*
+        For predicates of the form "const OP expr" we also have to check 'expr'
+        to make a decision.
+      */
+      continue;
     }
     else
       DBUG_RETURN(FALSE);
diff --git a/sql/repl_failsafe.cc b/sql/repl_failsafe.cc
index e3893ce2500..8392b2c1c3d 100644
--- a/sql/repl_failsafe.cc
+++ b/sql/repl_failsafe.cc
@@ -559,7 +559,12 @@ HOSTS";
 	goto err;
       }
       si->server_id = log_server_id;
-      my_hash_insert(&slave_list, (uchar*)si);
+      if (my_hash_insert(&slave_list, (uchar*)si))
+      {
+        error= "the slave is out of memory";
+        pthread_mutex_unlock(&LOCK_slave_list);
+        goto err;
+      }
     }
     strmake(si->host, row[1], sizeof(si->host)-1);
     si->port = atoi(row[port_ind]);
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index ae3cbf789a5..a1403d2ff71 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -106,7 +106,8 @@ int init_relay_log_info(Relay_log_info* rli,
   rli->tables_to_lock_count= 0;
 
   char pattern[FN_REFLEN];
-  if (fn_format(pattern, PREFIX_SQL_LOAD, slave_load_tmpdir, "",
+  (void) my_realpath(pattern, slave_load_tmpdir, 0);
+  if (fn_format(pattern, PREFIX_SQL_LOAD, pattern, "",
             MY_SAFE_PATH | MY_RETURN_REAL_PATH) == NullS)
   {
     pthread_mutex_unlock(&rli->data_lock);
@@ -133,6 +134,29 @@ int init_relay_log_info(Relay_log_info* rli,
     rli->relay_log.max_size (and mysql_bin_log.max_size).
   */
   {
+    /* Reports an error and returns, if the --relay-log's path 
+       is a directory.*/
+    if (opt_relay_logname && 
+        opt_relay_logname[strlen(opt_relay_logname) - 1] == FN_LIBCHAR)
+    {
+      pthread_mutex_unlock(&rli->data_lock);
+      sql_print_error("Path '%s' is a directory name, please specify \
+a file name for --relay-log option", opt_relay_logname);
+      DBUG_RETURN(1);
+    }
+
+    /* Reports an error and returns, if the --relay-log-index's path 
+       is a directory.*/
+    if (opt_relaylog_index_name && 
+        opt_relaylog_index_name[strlen(opt_relaylog_index_name) - 1] 
+        == FN_LIBCHAR)
+    {
+      pthread_mutex_unlock(&rli->data_lock);
+      sql_print_error("Path '%s' is a directory name, please specify \
+a file name for --relay-log-index option", opt_relaylog_index_name);
+      DBUG_RETURN(1);
+    }
+
     char buf[FN_REFLEN];
     const char *ln;
     static bool name_warning_sent= 0;
diff --git a/sql/rpl_tblmap.cc b/sql/rpl_tblmap.cc
index 8b7fea3b40e..b04a3120a86 100644
--- a/sql/rpl_tblmap.cc
+++ b/sql/rpl_tblmap.cc
@@ -119,7 +119,13 @@ int table_mapping::set_table(ulong table_id, TABLE* table)
   }
   e->table_id= table_id;
   e->table= table;
-  my_hash_insert(&m_table_ids,(uchar *)e);
+  if (my_hash_insert(&m_table_ids,(uchar *)e))
+  {
+    /* we add this entry to the chain of free (free for use) entries */
+    e->next= m_free;
+    m_free= e;
+    DBUG_RETURN(ERR_MEMORY_ALLOCATION);
+  }
 
   DBUG_PRINT("info", ("tid %lu -> table 0x%lx (%s)", 
 		      table_id, (long) e->table,
diff --git a/sql/sp.cc b/sql/sp.cc
index 6e98dfdf4bc..1edacee8aad 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -70,6 +70,122 @@ enum
   MYSQL_PROC_FIELD_COUNT
 };
 
+static const
+TABLE_FIELD_TYPE proc_table_fields[MYSQL_PROC_FIELD_COUNT] =
+{
+  {
+    { C_STRING_WITH_LEN("db") },
+    { C_STRING_WITH_LEN("char(64)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("name") },
+    { C_STRING_WITH_LEN("char(64)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("type") },
+    { C_STRING_WITH_LEN("enum('FUNCTION','PROCEDURE')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("specific_name") },
+    { C_STRING_WITH_LEN("char(64)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("language") },
+    { C_STRING_WITH_LEN("enum('SQL')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("sql_data_access") },
+    { C_STRING_WITH_LEN("enum('CONTAINS_SQL','NO_SQL','READS_SQL_DATA','MODIFIES_SQL_DATA')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("is_deterministic") },
+    { C_STRING_WITH_LEN("enum('YES','NO')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("security_type") },
+    { C_STRING_WITH_LEN("enum('INVOKER','DEFINER')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("param_list") },
+    { C_STRING_WITH_LEN("blob") },
+    { NULL, 0 }
+  },
+
+  {
+    { C_STRING_WITH_LEN("returns") },
+    { C_STRING_WITH_LEN("longblob") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("body") },
+    { C_STRING_WITH_LEN("longblob") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("definer") },
+    { C_STRING_WITH_LEN("char(77)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("created") },
+    { C_STRING_WITH_LEN("timestamp") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("modified") },
+    { C_STRING_WITH_LEN("timestamp") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("sql_mode") },
+    { C_STRING_WITH_LEN("set('REAL_AS_FLOAT','PIPES_AS_CONCAT','ANSI_QUOTES',"
+    "'IGNORE_SPACE','NOT_USED','ONLY_FULL_GROUP_BY','NO_UNSIGNED_SUBTRACTION',"
+    "'NO_DIR_IN_CREATE','POSTGRESQL','ORACLE','MSSQL','DB2','MAXDB',"
+    "'NO_KEY_OPTIONS','NO_TABLE_OPTIONS','NO_FIELD_OPTIONS','MYSQL323','MYSQL40',"
+    "'ANSI','NO_AUTO_VALUE_ON_ZERO','NO_BACKSLASH_ESCAPES','STRICT_TRANS_TABLES',"
+    "'STRICT_ALL_TABLES','NO_ZERO_IN_DATE','NO_ZERO_DATE','INVALID_DATES',"
+    "'ERROR_FOR_DIVISION_BY_ZERO','TRADITIONAL','NO_AUTO_CREATE_USER',"
+    "'HIGH_NOT_PRECEDENCE','NO_ENGINE_SUBSTITUTION','PAD_CHAR_TO_FULL_LENGTH')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("comment") },
+    { C_STRING_WITH_LEN("text") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("character_set_client") },
+    { C_STRING_WITH_LEN("char(32)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("collation_connection") },
+    { C_STRING_WITH_LEN("char(32)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("db_collation") },
+    { C_STRING_WITH_LEN("char(32)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("body_utf8") },
+    { C_STRING_WITH_LEN("longblob") },
+    { NULL, 0 }
+  }
+};
+
+static const TABLE_FIELD_DEF
+  proc_table_def= {MYSQL_PROC_FIELD_COUNT, proc_table_fields};
+
 /*************************************************************************/
 
 /**
@@ -247,6 +363,50 @@ Stored_routine_creation_ctx::load_from_db(THD *thd,
 
 /*************************************************************************/
 
+class Proc_table_intact : public Table_check_intact
+{
+private:
+  bool m_print_once;
+
+public:
+  Proc_table_intact() : m_print_once(TRUE) {}
+
+protected:
+  void report_error(uint code, const char *fmt, ...);
+};
+
+
+/**
+  Report failure to validate the mysql.proc table definition.
+  Print a message to the error log only once.
+*/
+
+void Proc_table_intact::report_error(uint code, const char *fmt, ...)
+{
+  va_list args;
+  char buf[512];
+
+  va_start(args, fmt);
+  my_vsnprintf(buf, sizeof(buf), fmt, args);
+  va_end(args);
+
+  if (code)
+    my_message(code, buf, MYF(0));
+  else
+    my_error(ER_CANNOT_LOAD_FROM_TABLE, MYF(0), "proc");
+
+  if (m_print_once)
+  {
+    m_print_once= FALSE;
+    sql_print_error("%s", buf);
+  }
+};
+
+
+/** Single instance used to control printing to the error log. */
+static Proc_table_intact proc_table_intact;
+
+
 /**
   Open the mysql.proc table for read.
 
@@ -266,15 +426,17 @@ TABLE *open_proc_table_for_read(THD *thd, Open_tables_state *backup)
   DBUG_ENTER("open_proc_table_for_read");
 
   TABLE_LIST table;
-  bzero((char*) &table, sizeof(table));
-  table.db= (char*) "mysql";
-  table.table_name= table.alias= (char*)"proc";
-  table.lock_type= TL_READ;
+  table.init_one_table("mysql", "proc", TL_READ);
 
-  if (!open_system_tables_for_read(thd, &table, backup))
+  if (open_system_tables_for_read(thd, &table, backup))
+    DBUG_RETURN(NULL);
+
+  if (!proc_table_intact.check(table.table, &proc_table_def))
     DBUG_RETURN(table.table);
-  else
-    DBUG_RETURN(0);
+
+  close_system_tables(thd, backup);
+
+  DBUG_RETURN(NULL);
 }
 
 
@@ -296,13 +458,19 @@ static TABLE *open_proc_table_for_update(THD *thd)
 {
   DBUG_ENTER("open_proc_table_for_update");
 
-  TABLE_LIST table;
-  bzero((char*) &table, sizeof(table));
-  table.db= (char*) "mysql";
-  table.table_name= table.alias= (char*)"proc";
-  table.lock_type= TL_WRITE;
+  TABLE *table;
+  TABLE_LIST table_list;
+  table_list.init_one_table("mysql", "proc", TL_WRITE);
+
+  if (!(table= open_system_table_for_update(thd, &table_list)))
+    DBUG_RETURN(NULL);
 
-  DBUG_RETURN(open_system_table_for_update(thd, &table));
+  if (!proc_table_intact.check(table, &proc_table_def))
+    DBUG_RETURN(table);
+
+  close_thread_tables(thd);
+
+  DBUG_RETURN(NULL);
 }
 
 
@@ -1522,7 +1690,8 @@ static bool add_used_routine(LEX *lex, Query_arena *arena,
     rn->key.length= key->length;
     rn->key.str= (char *)rn + sizeof(Sroutine_hash_entry);
     memcpy(rn->key.str, key->str, key->length + 1);
-    my_hash_insert(&lex->sroutines, (uchar *)rn);
+    if (my_hash_insert(&lex->sroutines, (uchar *)rn))
+      return FALSE;
     lex->sroutines_list.link_in_list((uchar *)rn, (uchar **)&rn->next);
     rn->belong_to_view= belong_to_view;
     return TRUE;
@@ -1600,16 +1769,24 @@ void sp_remove_not_own_routines(LEX *lex)
     dependant on time of life of elements from source hash. It also
     won't touch lists linking elements in source and destination
     hashes.
+
+  @returns
+    @return TRUE Failure
+    @return FALSE Success
 */
 
-void sp_update_sp_used_routines(HASH *dst, HASH *src)
+bool sp_update_sp_used_routines(HASH *dst, HASH *src)
 {
   for (uint i=0 ; i < src->records ; i++)
   {
     Sroutine_hash_entry *rt= (Sroutine_hash_entry *)my_hash_element(src, i);
     if (!my_hash_search(dst, (uchar *)rt->key.str, rt->key.length))
-      my_hash_insert(dst, (uchar *)rt);
+    {
+      if (my_hash_insert(dst, (uchar *)rt))
+        return TRUE;
+    }
   }
+  return FALSE;
 }
 
 
diff --git a/sql/sp.h b/sql/sp.h
index 5a190c5480e..876287d9704 100644
--- a/sql/sp.h
+++ b/sql/sp.h
@@ -69,7 +69,7 @@ void sp_get_prelocking_info(THD *thd, bool *need_prelocking,
 void sp_add_used_routine(LEX *lex, Query_arena *arena,
                          sp_name *rt, char rt_type);
 void sp_remove_not_own_routines(LEX *lex);
-void sp_update_sp_used_routines(HASH *dst, HASH *src);
+bool sp_update_sp_used_routines(HASH *dst, HASH *src);
 int sp_cache_routines_and_add_tables(THD *thd, LEX *lex,
                                      bool first_no_prelock);
 int sp_cache_routines_and_add_tables_for_view(THD *thd, LEX *lex,
diff --git a/sql/sp_cache.cc b/sql/sp_cache.cc
index d9a23d2be4e..0da5e44b846 100644
--- a/sql/sp_cache.cc
+++ b/sql/sp_cache.cc
@@ -36,10 +36,16 @@ public:
   sp_cache();
   ~sp_cache();
 
-  inline void insert(sp_head *sp)
+  /**
+   Inserts a sp_head object into a hash table.
+
+   @returns Success status
+     @return TRUE Failure
+     @return FALSE Success
+  */
+  inline bool insert(sp_head *sp)
   {
-    /* TODO: why don't we check return value? */
-    my_hash_insert(&m_hashtable, (const uchar *)sp);
+    return my_hash_insert(&m_hashtable, (const uchar *)sp);
   }
 
   inline sp_head *lookup(char *name, uint namelen)
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 99b269d7a12..542db282700 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -2118,8 +2118,18 @@ sp_head::reset_lex(THD *thd)
   DBUG_RETURN(FALSE);
 }
 
-/// Restore lex during parsing, after we have parsed a sub statement.
-void
+
+/**
+  Restore lex during parsing, after we have parsed a sub statement.
+
+  @param thd Thread handle
+
+  @return
+    @retval TRUE failure
+    @retval FALSE success
+*/
+
+bool
 sp_head::restore_lex(THD *thd)
 {
   DBUG_ENTER("sp_head::restore_lex");
@@ -2130,7 +2140,7 @@ sp_head::restore_lex(THD *thd)
 
   oldlex= (LEX *)m_lex.pop();
   if (! oldlex)
-    return;			// Nothing to restore
+    DBUG_RETURN(FALSE);			// Nothing to restore
 
   oldlex->trg_table_fields.push_back(&sublex->trg_table_fields);
 
@@ -2146,7 +2156,8 @@ sp_head::restore_lex(THD *thd)
     Add routines which are used by statement to respective set for
     this routine.
   */
-  sp_update_sp_used_routines(&m_sroutines, &sublex->sroutines);
+  if (sp_update_sp_used_routines(&m_sroutines, &sublex->sroutines))
+    DBUG_RETURN(TRUE);
   /*
     Merge tables used by this statement (but not by its functions or
     procedures) to multiset of tables used by this routine.
@@ -2158,7 +2169,7 @@ sp_head::restore_lex(THD *thd)
     delete sublex;
   }
   thd->lex= oldlex;
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(FALSE);
 }
 
 /**
@@ -3905,7 +3916,8 @@ sp_head::merge_table_list(THD *thd, TABLE_LIST *table, LEX *lex_for_tmp_check)
         tab->lock_type= table->lock_type;
         tab->lock_count= tab->query_lock_count= 1;
         tab->trg_event_map= table->trg_event_map;
-	my_hash_insert(&m_sptabs, (uchar *)tab);
+	if (my_hash_insert(&m_sptabs, (uchar *)tab))
+          return FALSE;
       }
     }
   return TRUE;
diff --git a/sql/sp_head.h b/sql/sp_head.h
index dd11f8693ac..00c96d44f70 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -340,7 +340,7 @@ public:
 
     @todo Conflicting comment in sp_head.cc
   */
-  void
+  bool
   restore_lex(THD *thd);
 
   /// Put the instruction on the backpatch list, associated with the label.
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 4288c6ee2f6..b91dc9a7e25 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -31,9 +31,8 @@
 #include "sp_head.h"
 #include "sp.h"
 
-time_t mysql_db_table_last_check= 0L;
-
-TABLE_FIELD_W_TYPE mysql_db_table_fields[MYSQL_DB_FIELD_COUNT] = {
+static const
+TABLE_FIELD_TYPE mysql_db_table_fields[MYSQL_DB_FIELD_COUNT] = {
   {
     { C_STRING_WITH_LEN("Host") },            
     { C_STRING_WITH_LEN("char(60)") },
@@ -146,6 +145,8 @@ TABLE_FIELD_W_TYPE mysql_db_table_fields[MYSQL_DB_FIELD_COUNT] = {
   }
 };
 
+const TABLE_FIELD_DEF
+  mysql_db_table_def= {MYSQL_DB_FIELD_COUNT, mysql_db_table_fields};
 
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
 
@@ -2464,7 +2465,12 @@ GRANT_TABLE::GRANT_TABLE(TABLE *form, TABLE *col_privs)
         privs = cols = 0;			/* purecov: deadcode */
         return;				/* purecov: deadcode */
       }
-      my_hash_insert(&hash_columns, (uchar *) mem_check);
+      if (my_hash_insert(&hash_columns, (uchar *) mem_check))
+      {
+        /* Invalidate this entry */
+        privs= cols= 0;
+        return;
+      }
     } while (!col_privs->file->index_next(col_privs->record[0]) &&
              !key_cmp_if_same(col_privs,key,0,key_prefix_len));
     col_privs->file->ha_index_end();
@@ -2498,14 +2504,17 @@ static GRANT_NAME *name_hash_search(HASH *name_hash,
                                     const char *host,const char* ip,
                                     const char *db,
                                     const char *user, const char *tname,
-                                    bool exact)
+                                    bool exact, bool name_tolower)
 {
-  char helping [NAME_LEN*2+USERNAME_LENGTH+3];
+  char helping [NAME_LEN*2+USERNAME_LENGTH+3], *name_ptr;
   uint len;
   GRANT_NAME *grant_name,*found=0;
   HASH_SEARCH_STATE state;
 
-  len  = (uint) (strmov(strmov(strmov(helping,user)+1,db)+1,tname)-helping)+ 1;
+  name_ptr= strmov(strmov(helping, user) + 1, db) + 1;
+  len  = (uint) (strmov(name_ptr, tname) - helping) + 1;
+  if (name_tolower)
+    my_casedn_str(files_charset_info, name_ptr);
   for (grant_name= (GRANT_NAME*) my_hash_first(name_hash, (uchar*) helping,
                                                len, &state);
        grant_name ;
@@ -2538,7 +2547,7 @@ routine_hash_search(const char *host, const char *ip, const char *db,
 {
   return (GRANT_TABLE*)
     name_hash_search(proc ? &proc_priv_hash : &func_priv_hash,
-		     host, ip, db, user, tname, exact);
+		     host, ip, db, user, tname, exact, TRUE);
 }
 
 
@@ -2547,7 +2556,7 @@ table_hash_search(const char *host, const char *ip, const char *db,
 		  const char *user, const char *tname, bool exact)
 {
   return (GRANT_TABLE*) name_hash_search(&column_priv_hash, host, ip, db,
-					 user, tname, exact);
+					 user, tname, exact, FALSE);
 }
 
 
@@ -2670,7 +2679,11 @@ static int replace_column_table(GRANT_TABLE *g_t,
 	goto end;				/* purecov: inspected */
       }
       grant_column= new GRANT_COLUMN(column->column,privileges);
-      my_hash_insert(&g_t->hash_columns,(uchar*) grant_column);
+      if (my_hash_insert(&g_t->hash_columns,(uchar*) grant_column))
+      {
+        result= -1;
+        goto end;
+      }
     }
   }
 
@@ -3196,12 +3209,12 @@ int mysql_table_grant(THD *thd, TABLE_LIST *table_list,
 				     Str->user.str, table_name,
 				     rights,
 				     column_priv);
-      if (!grant_table)				// end of memory
+      if (!grant_table ||
+        my_hash_insert(&column_priv_hash,(uchar*) grant_table))
       {
 	result= TRUE;				/* purecov: deadcode */
 	continue;				/* purecov: deadcode */
       }
-      my_hash_insert(&column_priv_hash,(uchar*) grant_table);
     }
 
     /* If revoke_grant, calculate the new column privilege for tables_priv */
@@ -3405,12 +3418,13 @@ bool mysql_routine_grant(THD *thd, TABLE_LIST *table_list, bool is_proc,
       grant_name= new GRANT_NAME(Str->host.str, db_name,
 				 Str->user.str, table_name,
 				 rights, TRUE);
-      if (!grant_name)
+      if (!grant_name ||
+        my_hash_insert(is_proc ?
+                       &proc_priv_hash : &func_priv_hash,(uchar*) grant_name))
       {
         result= TRUE;
 	continue;
       }
-      my_hash_insert(is_proc ? &proc_priv_hash : &func_priv_hash,(uchar*) grant_name);
     }
 
     if (replace_routine_table(thd, grant_name, tables[1].table, *Str,
@@ -3514,6 +3528,13 @@ bool mysql_grant(THD *thd, const char *db, List <LEX_USER> &list,
       result= TRUE;
       continue;
     }
+    /*
+      No User, but a password?
+      They did GRANT ... TO CURRENT_USER() IDENTIFIED BY ... !
+      Get the current user, and shallow-copy the new password to them!
+    */
+    if (!tmp_Str->user.str && tmp_Str->password.str)
+      Str->password= tmp_Str->password;
     if (replace_user_table(thd, tables[0].table, *Str,
                            (!db ? rights : 0), revoke_grant, create_new_users,
                            test(thd->variables.sql_mode &
diff --git a/sql/sql_acl.h b/sql/sql_acl.h
index 5c1f34b0b34..25a4766e58c 100644
--- a/sql/sql_acl.h
+++ b/sql/sql_acl.h
@@ -168,8 +168,7 @@ enum mysql_db_table_field
   MYSQL_DB_FIELD_COUNT
 };
 
-extern TABLE_FIELD_W_TYPE mysql_db_table_fields[];
-extern time_t mysql_db_table_last_check;
+extern const TABLE_FIELD_DEF mysql_db_table_def;
 
 /* Classes */
 
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 9c7c3e14436..c67c3266e59 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2966,7 +2966,12 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     DBUG_PRINT("info", ("inserting table '%s'.'%s' 0x%lx into the cache",
                         table->s->db.str, table->s->table_name.str,
                         (long) table));
-    (void) my_hash_insert(&open_cache,(uchar*) table);
+    if (my_hash_insert(&open_cache,(uchar*) table))
+    {
+      my_free(table, MYF(0));
+      (void) pthread_mutex_unlock(&LOCK_open);
+      DBUG_RETURN(NULL);
+    }
   }
 
   check_unused();				// Debugging call
diff --git a/sql/sql_cache.cc b/sql/sql_cache.cc
index fb370cbd16a..40316299366 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -423,12 +423,16 @@ TYPELIB query_cache_type_typelib=
   effect by another thread. This enables a quick path in execution to skip waits
   when the outcome is known.
 
+  @param use_timeout TRUE if the lock can abort because of a timeout.
+
+  @note use_timeout is optional and default value is FALSE.
+
   @return
    @retval FALSE An exclusive lock was taken
    @retval TRUE The locking attempt failed
 */
 
-bool Query_cache::try_lock(void)
+bool Query_cache::try_lock(bool use_timeout)
 {
   bool interrupt= FALSE;
   DBUG_ENTER("Query_cache::try_lock");
@@ -458,7 +462,26 @@ bool Query_cache::try_lock(void)
     else
     {
       DBUG_ASSERT(m_cache_lock_status == Query_cache::LOCKED);
-      pthread_cond_wait(&COND_cache_status_changed, &structure_guard_mutex);
+      /*
+        To prevent send_result_to_client() and query_cache_insert() from
+        blocking execution for too long a timeout is put on the lock.
+      */
+      if (use_timeout)
+      {
+        struct timespec waittime;
+        set_timespec_nsec(waittime,(ulong)(50000000L));  /* Wait for 50 msec */
+        int res= pthread_cond_timedwait(&COND_cache_status_changed,
+                                        &structure_guard_mutex,&waittime);
+        if (res == ETIMEDOUT)
+        {
+          interrupt= TRUE;
+          break;
+        }
+      }
+      else
+      {
+        pthread_cond_wait(&COND_cache_status_changed, &structure_guard_mutex);
+      }
     }
   }
   pthread_mutex_unlock(&structure_guard_mutex);
@@ -1213,8 +1236,14 @@ def_week_frmt: %lu, in_trans: %d, autocommit: %d",
       A table- or a full flush operation can potentially take a long time to
       finish. We choose not to wait for them and skip caching statements
       instead.
+
+      In case the wait time can't be determined there is an upper limit which
+      causes try_lock() to abort with a time out.
+
+      The 'TRUE' parameter indicate that the lock is allowed to timeout
+
     */
-    if (try_lock())
+    if (try_lock(TRUE))
       DBUG_VOID_RETURN;
     if (query_cache_size == 0)
     {
@@ -1415,8 +1444,10 @@ Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
     Try to obtain an exclusive lock on the query cache. If the cache is
     disabled or if a full cache flush is in progress, the attempt to
     get the lock is aborted.
+
+    The 'TRUE' parameter indicate that the lock is allowed to timeout
   */
-  if (try_lock())
+  if (try_lock(TRUE))
     goto err;
 
   if (query_cache_size == 0)
diff --git a/sql/sql_cache.h b/sql/sql_cache.h
index 6b187ebd6c2..695d6fb4db3 100644
--- a/sql/sql_cache.h
+++ b/sql/sql_cache.h
@@ -496,7 +496,7 @@ protected:
 			const char *name);
   my_bool in_blocks(Query_cache_block * point);
 
-  bool try_lock(void);
+  bool try_lock(bool use_timeout= FALSE);
   void lock(void);
   void lock_and_suspend(void);
   void unlock(void);
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index ee5c27c3a00..94329bbcd28 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -385,6 +385,8 @@ char *thd_security_context(THD *thd, char *buffer, unsigned int length,
     str.append(proc_info);
   }
 
+  pthread_mutex_lock(&thd->LOCK_thd_data);
+
   if (thd->query())
   {
     if (max_query_len < 1)
@@ -394,6 +396,9 @@ char *thd_security_context(THD *thd, char *buffer, unsigned int length,
     str.append('\n');
     str.append(thd->query(), len);
   }
+
+  pthread_mutex_unlock(&thd->LOCK_thd_data);
+
   if (str.c_ptr_safe() == buffer)
     return buffer;
 
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index abdf545ccb9..8ffc7bd1bc6 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -426,7 +426,8 @@ cleanup:
   }
   DBUG_ASSERT(transactional_table || !deleted || thd->transaction.stmt.modified_non_trans_table);
   free_underlaid_joins(thd, select_lex);
-  if (error < 0 || (thd->lex->ignore && !thd->is_fatal_error))
+  if (error < 0 || 
+      (thd->lex->ignore && !thd->is_error() && !thd->is_fatal_error))
   {
     /*
       If a TRUNCATE TABLE was issued, the number of rows should be reported as
@@ -1090,6 +1091,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   TABLE *table;
   bool error;
   uint path_length;
+  bool is_temporary_table= false;
   DBUG_ENTER("mysql_truncate");
 
   bzero((char*) &create_info,sizeof(create_info));
@@ -1100,6 +1102,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   /* If it is a temporary table, close and regenerate it */
   if (!dont_send_ok && (table= find_temporary_table(thd, table_list)))
   {
+    is_temporary_table= true;
     handlerton *table_type= table->s->db_type();
     TABLE_SHARE *share= table->s;
     /* Note that a temporary table cannot be partitioned */
@@ -1178,13 +1181,13 @@ end:
   {
     if (!error)
     {
-      /*
-        TRUNCATE must always be statement-based binlogged (not row-based) so
-        we don't test current_stmt_binlog_row_based.
-      */
-      error= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
-      if (!error)
-        my_ok(thd);		// This should return record count
+      /* In RBR, the statement is not binlogged if the table is temporary. */
+      if (!is_temporary_table || !thd->current_stmt_binlog_row_based)
+      {
+        error= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
+        if (!error)
+          my_ok(thd);		// This should return record count
+      }
     }
     pthread_mutex_lock(&LOCK_open);
     unlock_table_name(thd, table_list);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 0cf346f9cb3..9dc25f8dc51 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -518,6 +518,22 @@ bool open_and_lock_for_insert_delayed(THD *thd, TABLE_LIST *table_list)
   DBUG_ENTER("open_and_lock_for_insert_delayed");
 
 #ifndef EMBEDDED_LIBRARY
+  if (thd->locked_tables && thd->global_read_lock)
+  {
+    /*
+      If this connection has the global read lock, the handler thread
+      will not be able to lock the table. It will wait for the global
+      read lock to go away, but this will never happen since the
+      connection thread will be stuck waiting for the handler thread
+      to open and lock the table.
+      If we are not in locked tables mode, INSERT will seek protection
+      against the global read lock (and fail), thus we will only get
+      to this point in locked tables mode.
+    */
+    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
+    DBUG_RETURN(TRUE);
+  }
+
   if (delayed_get_table(thd, table_list))
     DBUG_RETURN(TRUE);
 
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index b444e3db1f3..9368053c658 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -337,8 +337,8 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
     else
     {
       (void) fn_format(name, ex->file_name, mysql_real_data_home, "",
-		       MY_RELATIVE_PATH | MY_UNPACK_FILENAME);
-
+                       MY_RELATIVE_PATH | MY_UNPACK_FILENAME |
+                       MY_RETURN_REAL_PATH);
 #if !defined(__WIN__) && ! defined(__NETWARE__)
       MY_STAT stat_info;
       if (!my_stat(name,&stat_info,MYF(MY_WME)))
@@ -381,12 +381,16 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
         DBUG_ASSERT(FALSE); 
 #endif
       }
-      else if (opt_secure_file_priv &&
-               strncmp(opt_secure_file_priv, name, strlen(opt_secure_file_priv)))
+      else if (opt_secure_file_priv)
       {
-        /* Read only allowed from within dir specified by secure_file_priv */
-        my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--secure-file-priv");
-        DBUG_RETURN(TRUE);
+        char secure_file_real_path[FN_REFLEN];
+        (void) my_realpath(secure_file_real_path, opt_secure_file_priv, 0);
+        if (strncmp(secure_file_real_path, name, strlen(secure_file_real_path)))
+        {
+          /* Read only allowed from within dir specified by secure_file_priv */
+          my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--secure-file-priv");
+          DBUG_RETURN(TRUE);
+        }
       }
 
     }
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index fe31b880d1c..501234e9ba9 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -7636,6 +7636,9 @@ void get_default_definer(THD *thd, LEX_USER *definer)
 
   definer->host.str= (char *) sctx->priv_host;
   definer->host.length= strlen(definer->host.str);
+
+  definer->password.str= NULL;
+  definer->password.length= 0;
 }
 
 
@@ -7687,6 +7690,8 @@ LEX_USER *create_definer(THD *thd, LEX_STRING *user_name, LEX_STRING *host_name)
 
   definer->user= *user_name;
   definer->host= *host_name;
+  definer->password.str= NULL;
+  definer->password.length= 0;
 
   return definer;
 }
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index c0eb3e67c84..062d2fb701c 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -238,26 +238,27 @@ bool partition_default_handling(TABLE *table, partition_info *part_info,
 {
   DBUG_ENTER("partition_default_handling");
 
-  if (part_info->use_default_num_partitions)
+  if (!is_create_table_ind)
   {
-    if (!is_create_table_ind &&
-        table->file->get_no_parts(normalized_path, &part_info->num_parts))
+    if (part_info->use_default_num_partitions)
     {
-      DBUG_RETURN(TRUE);
+      if (table->file->get_no_parts(normalized_path, &part_info->num_parts))
+      {
+        DBUG_RETURN(TRUE);
+      }
     }
-  }
-  else if (part_info->is_sub_partitioned() &&
-           part_info->use_default_num_subpartitions)
-  {
-    uint num_parts;
-    if (!is_create_table_ind &&
-        (table->file->get_no_parts(normalized_path, &num_parts)))
+    else if (part_info->is_sub_partitioned() &&
+             part_info->use_default_num_subpartitions)
     {
-      DBUG_RETURN(TRUE);
+      uint num_parts;
+      if (table->file->get_no_parts(normalized_path, &num_parts))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      DBUG_ASSERT(part_info->num_parts > 0);
+      DBUG_ASSERT((num_parts % part_info->num_parts) == 0);
+      part_info->num_subparts= num_parts / part_info->num_parts;
     }
-    DBUG_ASSERT(part_info->num_parts > 0);
-    part_info->num_subparts= num_parts / part_info->num_parts;
-    DBUG_ASSERT((num_parts % part_info->num_parts) == 0);
   }
   part_info->set_up_defaults_for_partitioning(table->file,
                                               (ulonglong)0, (uint)0);
@@ -1056,6 +1057,8 @@ static bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
   const char *save_where;
   LEX *old_lex= thd->lex;
   LEX lex;
+  uint8 saved_full_group_by_flag;
+  nesting_map saved_allow_sum_func;
   DBUG_ENTER("fix_fields_part_func");
 
   if (init_lex_with_single_table(thd, table, &lex))
@@ -1081,8 +1084,19 @@ static bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
     This is a tricky call to prepare for since it can have a large number
     of interesting side effects, both desirable and undesirable.
   */
+  saved_full_group_by_flag= thd->lex->current_select->full_group_by_flag;
+  saved_allow_sum_func= thd->lex->allow_sum_func;
+  thd->lex->allow_sum_func= 0;
+
   error= func_expr->fix_fields(thd, (Item**)&func_expr);
 
+  /*
+    Restore full_group_by_flag and allow_sum_func,
+    fix_fields should not affect mysql_select later, see Bug#46923.
+  */
+  thd->lex->current_select->full_group_by_flag= saved_full_group_by_flag;
+  thd->lex->allow_sum_func= saved_allow_sum_func;
+
   if (unlikely(error))
   {
     DBUG_PRINT("info", ("Field in partition function not part of table"));
@@ -1793,8 +1807,8 @@ bool fix_partition_func(THD *thd, TABLE *table,
   if (((part_info->part_type != HASH_PARTITION ||
       part_info->list_of_part_fields == FALSE) &&
       (!part_info->column_list &&
-       check_part_func_fields(part_info->part_field_array, TRUE))) ||
-      (part_info->list_of_part_fields == FALSE &&
+      check_part_func_fields(part_info->part_field_array, TRUE))) ||
+      (part_info->list_of_subpart_fields == FALSE &&
        part_info->is_sub_partitioned() &&
        check_part_func_fields(part_info->subpart_field_array, TRUE)))
   {
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 65ff44622a8..3c3d3a62ef5 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -986,14 +986,20 @@ JOIN::optimize()
     DBUG_RETURN(1);
   }
 
-  if (select_lex->olap == ROLLUP_TYPE && rollup_process_const_fields())
+  if (rollup.state != ROLLUP::STATE_NONE)
   {
-    DBUG_PRINT("error", ("Error: rollup_process_fields() failed"));
-    DBUG_RETURN(1);
+    if (rollup_process_const_fields())
+    {
+      DBUG_PRINT("error", ("Error: rollup_process_fields() failed"));
+      DBUG_RETURN(1);
+    }
+  }
+  else
+  {
+    /* Remove distinct if only const tables */
+    select_distinct= select_distinct && (const_tables != tables);
   }
 
-  /* Remove distinct if only const tables */
-  select_distinct= select_distinct && (const_tables != tables);
   thd_proc_info(thd, "preparing");
   if (result->initialize_tables(this))
   {
@@ -1300,11 +1306,14 @@ JOIN::optimize()
     - We are using an ORDER BY or GROUP BY on fields not in the first table
     - We are using different ORDER BY and GROUP BY orders
     - The user wants us to buffer the result.
+    When the WITH ROLLUP modifier is present, we cannot skip temporary table
+    creation for the DISTINCT clause just because there are only const tables.
   */
-  need_tmp= (const_tables != tables &&
+  need_tmp= ((const_tables != tables &&
 	     ((select_distinct || !simple_order || !simple_group) ||
 	      (group_list && order) ||
-	      test(select_options & OPTION_BUFFER_RESULT)));
+	      test(select_options & OPTION_BUFFER_RESULT))) ||
+             (rollup.state != ROLLUP::STATE_NONE && select_distinct));
 
   // No cache for MATCH
   make_join_readinfo(this,
@@ -2150,17 +2159,13 @@ JOIN::exec()
 	    DBUG_VOID_RETURN;
 	if (!curr_table->select->cond)
 	  curr_table->select->cond= sort_table_cond;
-	else					// This should never happen
+	else
 	{
 	  if (!(curr_table->select->cond=
 		new Item_cond_and(curr_table->select->cond,
 				  sort_table_cond)))
 	    DBUG_VOID_RETURN;
-	  /*
-	    Item_cond_and do not need fix_fields for execution, its parameters
-	    are fixed or do not need fix_fields, too
-	  */
-	  curr_table->select->cond->quick_fix_field();
+	  curr_table->select->cond->fix_fields(thd, 0);
 	}
 	curr_table->select_cond= curr_table->select->cond;
 	curr_table->select_cond->top_level_item();
@@ -6590,6 +6595,56 @@ void rr_unlock_row(st_join_table *tab)
 
 
 
+/**
+  Pick the appropriate access method functions
+
+  Sets the functions for the selected table access method
+
+  @param      tab               Table reference to put access method
+*/
+
+static void
+pick_table_access_method(JOIN_TAB *tab)
+{
+  switch (tab->type) 
+  {
+  case JT_REF:
+    tab->read_first_record= join_read_always_key;
+    tab->read_record.read_record= join_read_next_same;
+    break;
+
+  case JT_REF_OR_NULL:
+    tab->read_first_record= join_read_always_key_or_null;
+    tab->read_record.read_record= join_read_next_same_or_null;
+    break;
+
+  case JT_CONST:
+    tab->read_first_record= join_read_const;
+    tab->read_record.read_record= join_no_more_records;
+    break;
+
+  case JT_EQ_REF:
+    tab->read_first_record= join_read_key;
+    tab->read_record.read_record= join_no_more_records;
+    break;
+
+  case JT_FT:
+    tab->read_first_record= join_ft_read_first;
+    tab->read_record.read_record= join_ft_read_next;
+    break;
+
+  case JT_SYSTEM:
+    tab->read_first_record= join_read_system;
+    tab->read_record.read_record= join_no_more_records;
+    break;
+
+  /* keep gcc happy */  
+  default:
+    break;  
+  }
+}
+
+
 static void
 make_join_readinfo(JOIN *join, ulonglong options)
 {
@@ -6624,45 +6679,15 @@ make_join_readinfo(JOIN *join, ulonglong options)
 
     tab->sorted= sorted;
     sorted= 0;                                  // only first must be sorted
+    table->status=STATUS_NO_RECORD;
+    pick_table_access_method (tab);
+
     switch (tab->type) {
-    case JT_SYSTEM:				// Only happens with left join
-      table->status=STATUS_NO_RECORD;
-      tab->read_first_record= join_read_system;
-      tab->read_record.read_record= join_no_more_records;
-      break;
-    case JT_CONST:				// Only happens with left join
-      table->status=STATUS_NO_RECORD;
-      tab->read_first_record= join_read_const;
-      tab->read_record.read_record= join_no_more_records;
-      if (table->covering_keys.is_set(tab->ref.key) &&
-          !table->no_keyread)
-      {
-        table->key_read=1;
-        table->file->extra(HA_EXTRA_KEYREAD);
-      }
-      break;
     case JT_EQ_REF:
-      table->status=STATUS_NO_RECORD;
-      if (tab->select)
-      {
-	delete tab->select->quick;
-	tab->select->quick=0;
-      }
-      delete tab->quick;
-      tab->quick=0;
-      tab->read_first_record= join_read_key;
       tab->read_record.unlock_row= join_read_key_unlock_row;
-      tab->read_record.read_record= join_no_more_records;
-      if (table->covering_keys.is_set(tab->ref.key) &&
-	  !table->no_keyread)
-      {
-	table->key_read=1;
-	table->file->extra(HA_EXTRA_KEYREAD);
-      }
-      break;
+      /* fall through */
     case JT_REF_OR_NULL:
     case JT_REF:
-      table->status=STATUS_NO_RECORD;
       if (tab->select)
       {
 	delete tab->select->quick;
@@ -6670,34 +6695,20 @@ make_join_readinfo(JOIN *join, ulonglong options)
       }
       delete tab->quick;
       tab->quick=0;
+      /* fall through */
+    case JT_CONST:				// Only happens with left join
       if (table->covering_keys.is_set(tab->ref.key) &&
 	  !table->no_keyread)
       {
 	table->key_read=1;
 	table->file->extra(HA_EXTRA_KEYREAD);
       }
-      if (tab->type == JT_REF)
-      {
-	tab->read_first_record= join_read_always_key;
-	tab->read_record.read_record= join_read_next_same;
-      }
-      else
-      {
-	tab->read_first_record= join_read_always_key_or_null;
-	tab->read_record.read_record= join_read_next_same_or_null;
-      }
-      break;
-    case JT_FT:
-      table->status=STATUS_NO_RECORD;
-      tab->read_first_record= join_ft_read_first;
-      tab->read_record.read_record= join_ft_read_next;
       break;
     case JT_ALL:
       /*
 	If previous table use cache
         If the incoming data set is already sorted don't use cache.
       */
-      table->status=STATUS_NO_RECORD;
       if (i != join->const_tables && !(options & SELECT_NO_JOIN_CACHE) &&
           tab->use_quick != 2 && !tab->first_inner && !ordered_set)
       {
@@ -6777,6 +6788,9 @@ make_join_readinfo(JOIN *join, ulonglong options)
 	}
       }
       break;
+    case JT_FT:
+    case JT_SYSTEM: 
+      break;
     default:
       DBUG_PRINT("error",("Table type %d found",tab->type)); /* purecov: deadcode */
       break;					/* purecov: deadcode */
@@ -7927,12 +7941,12 @@ static COND *build_equal_items_for_cond(THD *thd, COND *cond,
         {
           item_equal->fix_length_and_dec();
           item_equal->update_used_tables();
+          set_if_bigger(thd->lex->current_select->max_equal_elems,
+                        item_equal->members());  
+          return item_equal;
 	}
-        else
-          item_equal= (Item_equal *) eq_list.pop();
-        set_if_bigger(thd->lex->current_select->max_equal_elems,
-                      item_equal->members());  
-        return item_equal;
+
+        return eq_list.pop();
       }
       else
       {
@@ -9615,47 +9629,8 @@ static Field *create_tmp_field_from_item(THD *thd, Item *item, TABLE *table,
     new_field->set_derivation(item->collation.derivation);
     break;
   case DECIMAL_RESULT:
-  {
-    uint8 dec= item->decimals;
-    uint8 intg= ((Item_decimal *) item)->decimal_precision() - dec;
-    uint32 len= item->max_length;
-
-    /*
-      Trying to put too many digits overall in a DECIMAL(prec,dec)
-      will always throw a warning. We must limit dec to
-      DECIMAL_MAX_SCALE however to prevent an assert() later.
-    */
-
-    if (dec > 0)
-    {
-      signed int overflow;
-
-      dec= min(dec, DECIMAL_MAX_SCALE);
-
-      /*
-        If the value still overflows the field with the corrected dec,
-        we'll throw out decimals rather than integers. This is still
-        bad and of course throws a truncation warning.
-        +1: for decimal point
-      */
-
-      const int required_length=
-        my_decimal_precision_to_length(intg + dec, dec,
-                                                     item->unsigned_flag);
-
-      overflow= required_length - len;
-
-      if (overflow > 0)
-        dec= max(0, dec - overflow);            // too long, discard fract
-      else
-        /* Corrected value fits. */
-        len= required_length;
-    }
-
-    new_field= new Field_new_decimal(len, maybe_null, item->name,
-                                     dec, item->unsigned_flag);
+    new_field= Field_new_decimal::create_from_item(item);
     break;
-  }
   case ROW_RESULT:
   default:
     // This case should never be choosen
@@ -13355,6 +13330,8 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
           if (create_ref_for_key(tab->join, tab, keyuse, 
                                  tab->join->const_table_map))
             DBUG_RETURN(0);
+
+          pick_table_access_method(tab);
 	}
 	else
 	{
@@ -14160,7 +14137,10 @@ static int remove_dup_with_hash_index(THD *thd, TABLE *table,
 	goto err;
     }
     else
-      (void) my_hash_insert(&hash, org_key_pos);
+    {
+      if (my_hash_insert(&hash, org_key_pos))
+        goto err;
+    }
     key_pos+=extra_length;
   }
   my_free((char*) key_buffer,MYF(0));
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index efeca20caf4..e06659a549c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -5401,13 +5401,21 @@ binlog:
         }
         pthread_mutex_unlock(&LOCK_open);
 
-        int result __attribute__((unused))=
-          store_create_info(thd, table, &query,
-                            create_info, FALSE /* show_database */);
+       /*
+         The condition avoids a crash as described in BUG#48506. Other
+         binlogging problems related to CREATE TABLE IF NOT EXISTS LIKE
+         when the existing object is a view will be solved by BUG 47442.
+       */
+        if (!table->view)
+        {
+          int result __attribute__((unused))=
+            store_create_info(thd, table, &query,
+                              create_info, FALSE /* show_database */);
 
-        DBUG_ASSERT(result == 0); // store_create_info() always return 0
-        if (write_bin_log(thd, TRUE, query.ptr(), query.length()))
-          goto err;
+          DBUG_ASSERT(result == 0); // store_create_info() always return 0
+          if (write_bin_log(thd, TRUE, query.ptr(), query.length()))
+            goto err;
+        }
       }
       else                                      // Case 1
         if (write_bin_log(thd, TRUE, thd->query(), thd->query_length()))
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 891131e54f9..a57897f4eac 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -391,6 +391,138 @@ void case_stmt_action_end_case(LEX *lex, bool simple)
   lex->sphead->do_cont_backpatch();
 }
 
+
+static bool
+find_sys_var_null_base(THD *thd, struct sys_var_with_base *tmp)
+{
+  tmp->var= find_sys_var(thd, tmp->base_name.str, tmp->base_name.length);
+
+  if (tmp->var == NULL)
+    my_error(ER_UNKNOWN_SYSTEM_VARIABLE, MYF(0), tmp->base_name.str);
+  else
+    tmp->base_name= null_lex_str;
+
+  return thd->is_error();
+}
+
+
+/**
+  Helper action for a SET statement.
+  Used to push a system variable into the assignment list.
+
+  @param thd      the current thread
+  @param tmp      the system variable with base name
+  @param var_type the scope of the variable
+  @param val      the value being assigned to the variable
+
+  @return TRUE if error, FALSE otherwise.
+*/
+
+static bool
+set_system_variable(THD *thd, struct sys_var_with_base *tmp,
+                    enum enum_var_type var_type, Item *val)
+{
+  set_var *var;
+  LEX *lex= thd->lex;
+
+  /* No AUTOCOMMIT from a stored function or trigger. */
+  if (lex->spcont && tmp->var == &sys_autocommit)
+    lex->sphead->m_flags|= sp_head::HAS_SET_AUTOCOMMIT_STMT;
+
+  if (! (var= new set_var(var_type, tmp->var, &tmp->base_name, val)))
+    return TRUE;
+
+  return lex->var_list.push_back(var);
+}
+
+
+/**
+  Helper action for a SET statement.
+  Used to push a SP local variable into the assignment list.
+
+  @param thd      the current thread
+  @param var_type the SP local variable
+  @param val      the value being assigned to the variable
+
+  @return TRUE if error, FALSE otherwise.
+*/
+
+static bool
+set_local_variable(THD *thd, sp_variable_t *spv, Item *val)
+{
+  Item *it;
+  LEX *lex= thd->lex;
+  sp_instr_set *sp_set;
+
+  if (val)
+    it= val;
+  else if (spv->dflt)
+    it= spv->dflt;
+  else
+  {
+    it= new (thd->mem_root) Item_null();
+    if (it == NULL)
+      return TRUE;
+  }
+
+  sp_set= new sp_instr_set(lex->sphead->instructions(), lex->spcont,
+                           spv->offset, it, spv->type, lex, TRUE);
+
+  return (sp_set == NULL || lex->sphead->add_instr(sp_set));
+}
+
+
+/**
+  Helper action for a SET statement.
+  Used to SET a field of NEW row.
+
+  @param thd      the current thread
+  @param name     the field name
+  @param val      the value being assigned to the row
+
+  @return TRUE if error, FALSE otherwise.
+*/
+
+static bool
+set_trigger_new_row(THD *thd, LEX_STRING *name, Item *val)
+{
+  LEX *lex= thd->lex;
+  Item_trigger_field *trg_fld;
+  sp_instr_set_trigger_field *sp_fld;
+
+  /* QQ: Shouldn't this be field's default value ? */
+  if (! val)
+    val= new Item_null();
+
+  DBUG_ASSERT(lex->trg_chistics.action_time == TRG_ACTION_BEFORE &&
+              (lex->trg_chistics.event == TRG_EVENT_INSERT ||
+               lex->trg_chistics.event == TRG_EVENT_UPDATE));
+
+  trg_fld= new (thd->mem_root)
+            Item_trigger_field(lex->current_context(),
+                               Item_trigger_field::NEW_ROW,
+                               name->str, UPDATE_ACL, FALSE);
+
+  if (trg_fld == NULL)
+    return TRUE;
+
+  sp_fld= new sp_instr_set_trigger_field(lex->sphead->instructions(),
+                                         lex->spcont, trg_fld, val, lex);
+
+  if (sp_fld == NULL)
+    return TRUE;
+
+  /*
+    Let us add this item to list of all Item_trigger_field
+    objects in trigger.
+  */
+  lex->trg_table_fields.link_in_list((uchar *) trg_fld,
+                                     (uchar **) &trg_fld->next_trg_field);
+
+  return lex->sphead->add_instr(sp_fld);
+}
+
+
 /**
   Helper to resolve the SQL:2003 Syntax exception 1) in <in predicate>.
   See SQL:2003, Part 2, section 8.4 <in predicate>, Note 184, page 383.
@@ -2508,8 +2640,8 @@ sp_decl:
             }
 
             pctx->declare_var_boundary(0);
-            lex->sphead->restore_lex(YYTHD);
-
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
             $$.vars= $2;
             $$.conds= $$.hndlrs= $$.curs= 0;
           }
@@ -2619,7 +2751,8 @@ sp_cursor_stmt:
             }
             lex->sp_lex_in_use= TRUE;
             $$= lex;
-            lex->sphead->restore_lex(YYTHD);
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
         ;
 
@@ -3001,7 +3134,8 @@ sp_proc_stmt_statement:
                     sp->add_instr(i))
                 MYSQL_YYABORT;
             }
-            sp->restore_lex(thd);
+            if (sp->restore_lex(thd))
+              MYSQL_YYABORT;
           }
         ;
 
@@ -3029,7 +3163,8 @@ sp_proc_stmt_return:
                 MYSQL_YYABORT;
               sp->m_flags|= sp_head::HAS_RETURN;
             }
-            sp->restore_lex(YYTHD);
+            if (sp->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
         ;
 
@@ -3269,7 +3404,8 @@ sp_if:
                 sp->add_cont_backpatch(i) ||
                 sp->add_instr(i))
               MYSQL_YYABORT;
-            sp->restore_lex(YYTHD);
+            if (sp->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           sp_proc_stmts1
           {
@@ -3315,7 +3451,9 @@ simple_case_stmt:
             if (case_stmt_action_expr(lex, $3))
               MYSQL_YYABORT;
 
-            lex->sphead->restore_lex(YYTHD); /* For expr $3 */
+            /* For expr $3 */
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           simple_when_clause_list
           else_clause_opt
@@ -3365,7 +3503,9 @@ simple_when_clause:
             LEX *lex= Lex;
             if (case_stmt_action_when(lex, $3, true))
               MYSQL_YYABORT;
-            lex->sphead->restore_lex(YYTHD); /* For expr $3 */
+            /* For expr $3 */
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           THEN_SYM
           sp_proc_stmts1
@@ -3386,7 +3526,9 @@ searched_when_clause:
             LEX *lex= Lex;
             if (case_stmt_action_when(lex, $3, false))
               MYSQL_YYABORT;
-            lex->sphead->restore_lex(YYTHD); /* For expr $3 */
+            /* For expr $3 */
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           THEN_SYM
           sp_proc_stmts1
@@ -3563,7 +3705,8 @@ sp_unlabeled_control:
                 sp->new_cont_backpatch(i) ||
                 sp->add_instr(i))
               MYSQL_YYABORT;
-            sp->restore_lex(YYTHD);
+            if (sp->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           sp_proc_stmts1 END WHILE_SYM
           {
@@ -3589,7 +3732,8 @@ sp_unlabeled_control:
             if (i == NULL ||
                 lex->sphead->add_instr(i))
               MYSQL_YYABORT;
-            lex->sphead->restore_lex(YYTHD);
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
             /* We can shortcut the cont_backpatch here */
             i->m_cont_dest= ip+1;
           }
@@ -7980,6 +8124,14 @@ function_call_nonkeyword:
           }
         | SYSDATE optional_braces
           {
+            /*
+              Unlike other time-related functions, SYSDATE() is
+              replication-unsafe because it is not affected by the
+              TIMESTAMP variable.  It is unsafe even if
+              sysdate_is_now=1, because the slave may have
+              sysdate_is_now=0.
+            */
+            Lex->set_stmt_unsafe();
             if (global_system_variables.sysdate_is_now == 0)
               $$= new (YYTHD->mem_root) Item_func_sysdate_local();
             else
@@ -12279,7 +12431,8 @@ option_type_value:
                 if (sp->add_instr(i))
                   MYSQL_YYABORT;
               }
-              lex->sphead->restore_lex(thd);
+              if (lex->sphead->restore_lex(thd))
+                MYSQL_YYABORT;
             }
           }
         ;
@@ -12319,98 +12472,42 @@ sys_option_value:
           option_type internal_variable_name equal set_expr_or_default
           {
             THD *thd= YYTHD;
-            LEX *lex=Lex;
+            LEX *lex= Lex;
+            LEX_STRING *name= &$2.base_name;
 
             if ($2.var == trg_new_row_fake_var)
             {
               /* We are in trigger and assigning value to field of new row */
-              Item *it;
-              Item_trigger_field *trg_fld;
-              sp_instr_set_trigger_field *sp_fld;
-              LINT_INIT(sp_fld);
               if ($1)
               {
                 my_parse_error(ER(ER_SYNTAX_ERROR));
                 MYSQL_YYABORT;
               }
-              if ($4)
-                it= $4;
-              else
-              {
-                /* QQ: Shouldn't this be field's default value ? */
-                it= new Item_null();
-              }
-
-              DBUG_ASSERT(lex->trg_chistics.action_time == TRG_ACTION_BEFORE &&
-                          (lex->trg_chistics.event == TRG_EVENT_INSERT ||
-                           lex->trg_chistics.event == TRG_EVENT_UPDATE));
-
-              trg_fld= new (thd->mem_root)
-                         Item_trigger_field(Lex->current_context(),
-                                            Item_trigger_field::NEW_ROW,
-                                            $2.base_name.str,
-                                            UPDATE_ACL, FALSE);
-              if (trg_fld == NULL)
-                MYSQL_YYABORT;
-
-              sp_fld= new sp_instr_set_trigger_field(lex->sphead->
-                                                     instructions(),
-                                                     lex->spcont,
-                                                     trg_fld,
-                                                     it, lex);
-              if (sp_fld == NULL)
-                MYSQL_YYABORT;
-
-              /*
-                Let us add this item to list of all Item_trigger_field
-                objects in trigger.
-              */
-              lex->trg_table_fields.link_in_list((uchar *)trg_fld,
-                                                 (uchar **) &trg_fld->
-                                                   next_trg_field);
-
-              if (lex->sphead->add_instr(sp_fld))
+              if (set_trigger_new_row(YYTHD, name, $4))
                 MYSQL_YYABORT;
             }
             else if ($2.var)
-            { /* System variable */
+            {
               if ($1)
                 lex->option_type= $1;
-              set_var *var= new set_var(lex->option_type, $2.var,
-                                        &$2.base_name, $4);
-              if (var == NULL)
+
+              /* It is a system variable. */
+              if (set_system_variable(thd, &$2, lex->option_type, $4))
                 MYSQL_YYABORT;
-              lex->var_list.push_back(var);
             }
             else
             {
-              /* An SP local variable */
-              sp_pcontext *ctx= lex->spcont;
-              sp_variable_t *spv;
-              sp_instr_set *sp_set;
-              Item *it;
+              sp_pcontext *spc= lex->spcont;
+              sp_variable_t *spv= spc->find_variable(name);
+
               if ($1)
               {
                 my_parse_error(ER(ER_SYNTAX_ERROR));
                 MYSQL_YYABORT;
               }
 
-              spv= ctx->find_variable(&$2.base_name);
-
-              if ($4)
-                it= $4;
-              else if (spv->dflt)
-                it= spv->dflt;
-              else
-              {
-                it= new (thd->mem_root) Item_null();
-                if (it == NULL)
-                  MYSQL_YYABORT;
-              }
-              sp_set= new sp_instr_set(lex->sphead->instructions(), ctx,
-                                       spv->offset, it, spv->type, lex, TRUE);
-              if (sp_set == NULL ||
-                  lex->sphead->add_instr(sp_set))
+              /* It is a local variable. */
+              if (set_local_variable(thd, spv, $4))
                 MYSQL_YYABORT;
             }
           }
@@ -12446,11 +12543,16 @@ option_value:
           }
         | '@' '@' opt_var_ident_type internal_variable_name equal set_expr_or_default
           {
-            LEX *lex=Lex;
-            set_var *var= new set_var($3, $4.var, &$4.base_name, $6);
-            if (var == NULL)
+            THD *thd= YYTHD;
+            struct sys_var_with_base tmp= $4;
+            /* Lookup if necessary: must be a system variable. */
+            if (tmp.var == NULL)
+            {
+              if (find_sys_var_null_base(thd, &tmp))
+                MYSQL_YYABORT;
+            }
+            if (set_system_variable(thd, &tmp, $3, $6))
               MYSQL_YYABORT;
-            lex->var_list.push_back(var);
           }
         | charset old_or_new_charset_name_or_default
           {
@@ -12543,31 +12645,26 @@ internal_variable_name:
           ident
           {
             THD *thd= YYTHD;
-            LEX *lex= thd->lex;
-            sp_pcontext *spc= lex->spcont;
+            sp_pcontext *spc= thd->lex->spcont;
             sp_variable_t *spv;
 
-            /* We have to lookup here since local vars can shadow sysvars */
+            /* Best effort lookup for system variable. */
             if (!spc || !(spv = spc->find_variable(&$1)))
             {
+              struct sys_var_with_base tmp= {NULL, $1};
+
               /* Not an SP local variable */
-              sys_var *tmp=find_sys_var(thd, $1.str, $1.length);
-              if (!tmp)
+              if (find_sys_var_null_base(thd, &tmp))
                 MYSQL_YYABORT;
-              $$.var= tmp;
-              $$.base_name= null_lex_str;
-              if (spc && tmp == &sys_autocommit)
-              {
-                /*
-                  We don't allow setting AUTOCOMMIT from a stored function
-                  or trigger.
-                */
-                lex->sphead->m_flags|= sp_head::HAS_SET_AUTOCOMMIT_STMT;
-              }
+
+              $$= tmp;
             }
             else
             {
-              /* An SP local variable */
+              /*
+                Possibly an SP local variable (or a shadowed sysvar).
+                Will depend on the context of the SET statement.
+              */
               $$.var= NULL;
               $$.base_name= $1;
             }
diff --git a/sql/table.cc b/sql/table.cc
index 736b89821c4..0f68bb7c29e 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1318,8 +1318,16 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
       share->timestamp_field_offset= i;
 
     if (use_hash)
-      (void) my_hash_insert(&share->name_hash,
-                            (uchar*) field_ptr); // never fail
+      if (my_hash_insert(&share->name_hash, (uchar*) field_ptr) )
+      {
+        /*
+          Set return code 8 here to indicate that an error has
+          occurred but that the error message already has been
+          sent (OOM).
+        */
+        error= 8; 
+        goto err;
+      }
   }
   *field_ptr=0;					// End marker
 
@@ -2805,34 +2813,38 @@ bool check_column_name(const char *name)
                   and such errors never reach the user.
 */
 
-my_bool
-table_check_intact(TABLE *table, const uint table_f_count,
-                   const TABLE_FIELD_W_TYPE *table_def)
+bool
+Table_check_intact::check(TABLE *table, const TABLE_FIELD_DEF *table_def)
 {
   uint i;
   my_bool error= FALSE;
-  my_bool fields_diff_count;
+  const TABLE_FIELD_TYPE *field_def= table_def->field;
   DBUG_ENTER("table_check_intact");
   DBUG_PRINT("info",("table: %s  expected_count: %d",
-                     table->alias, table_f_count));
+                     table->alias, table_def->count));
+
+  /* Whether the table definition has already been validated. */
+  if (table->s->table_field_def_cache == table_def)
+    DBUG_RETURN(FALSE);
 
-  fields_diff_count= (table->s->fields != table_f_count);
-  if (fields_diff_count)
+  if (table->s->fields != table_def->count)
   {
     DBUG_PRINT("info", ("Column count has changed, checking the definition"));
 
     /* previous MySQL version */
     if (MYSQL_VERSION_ID > table->s->mysql_version)
     {
-      sql_print_error(ER(ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE),
-                      table->alias, table_f_count, table->s->fields,
-                      table->s->mysql_version, MYSQL_VERSION_ID);
+      report_error(ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE,
+                   ER(ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE),
+                   table->alias, table_def->count, table->s->fields,
+                   table->s->mysql_version, MYSQL_VERSION_ID);
       DBUG_RETURN(TRUE);
     }
     else if (MYSQL_VERSION_ID == table->s->mysql_version)
     {
-      sql_print_error(ER(ER_COL_COUNT_DOESNT_MATCH_CORRUPTED), table->alias,
-                      table_f_count, table->s->fields);
+      report_error(ER_COL_COUNT_DOESNT_MATCH_CORRUPTED,
+                   ER(ER_COL_COUNT_DOESNT_MATCH_CORRUPTED), table->alias,
+                   table_def->count, table->s->fields);
       DBUG_RETURN(TRUE);
     }
     /*
@@ -2844,7 +2856,7 @@ table_check_intact(TABLE *table, const uint table_f_count,
     */
   }
   char buffer[STRING_BUFFER_USUAL_SIZE];
-  for (i=0 ; i < table_f_count; i++, table_def++)
+  for (i=0 ; i < table_def->count; i++, field_def++)
   {
     String sql_type(buffer, sizeof(buffer), system_charset_info);
     sql_type.length(0);
@@ -2852,18 +2864,18 @@ table_check_intact(TABLE *table, const uint table_f_count,
     {
       Field *field= table->field[i];
 
-      if (strncmp(field->field_name, table_def->name.str,
-                  table_def->name.length))
+      if (strncmp(field->field_name, field_def->name.str,
+                  field_def->name.length))
       {
         /*
           Name changes are not fatal, we use ordinal numbers to access columns.
           Still this can be a sign of a tampered table, output an error
           to the error log.
         */
-        sql_print_error("Incorrect definition of table %s.%s: "
-                        "expected column '%s' at position %d, found '%s'.",
-                        table->s->db.str, table->alias, table_def->name.str, i,
-                        field->field_name);
+        report_error(0, "Incorrect definition of table %s.%s: "
+                     "expected column '%s' at position %d, found '%s'.",
+                     table->s->db.str, table->alias, field_def->name.str, i,
+                     field->field_name);
       }
       field->sql_type(sql_type);
       /*
@@ -2883,47 +2895,51 @@ table_check_intact(TABLE *table, const uint table_f_count,
         the new table definition is backward compatible with the
         original one.
        */
-      if (strncmp(sql_type.c_ptr_safe(), table_def->type.str,
-                  table_def->type.length - 1))
+      if (strncmp(sql_type.c_ptr_safe(), field_def->type.str,
+                  field_def->type.length - 1))
       {
-        sql_print_error("Incorrect definition of table %s.%s: "
-                        "expected column '%s' at position %d to have type "
-                        "%s, found type %s.", table->s->db.str, table->alias,
-                        table_def->name.str, i, table_def->type.str,
-                        sql_type.c_ptr_safe());
+        report_error(0, "Incorrect definition of table %s.%s: "
+                     "expected column '%s' at position %d to have type "
+                     "%s, found type %s.", table->s->db.str, table->alias,
+                     field_def->name.str, i, field_def->type.str,
+                     sql_type.c_ptr_safe());
         error= TRUE;
       }
-      else if (table_def->cset.str && !field->has_charset())
+      else if (field_def->cset.str && !field->has_charset())
       {
-        sql_print_error("Incorrect definition of table %s.%s: "
-                        "expected the type of column '%s' at position %d "
-                        "to have character set '%s' but the type has no "
-                        "character set.", table->s->db.str, table->alias,
-                        table_def->name.str, i, table_def->cset.str);
+        report_error(0, "Incorrect definition of table %s.%s: "
+                     "expected the type of column '%s' at position %d "
+                     "to have character set '%s' but the type has no "
+                     "character set.", table->s->db.str, table->alias,
+                     field_def->name.str, i, field_def->cset.str);
         error= TRUE;
       }
-      else if (table_def->cset.str &&
-               strcmp(field->charset()->csname, table_def->cset.str))
+      else if (field_def->cset.str &&
+               strcmp(field->charset()->csname, field_def->cset.str))
       {
-        sql_print_error("Incorrect definition of table %s.%s: "
-                        "expected the type of column '%s' at position %d "
-                        "to have character set '%s' but found "
-                        "character set '%s'.", table->s->db.str, table->alias,
-                        table_def->name.str, i, table_def->cset.str,
-                        field->charset()->csname);
+        report_error(0, "Incorrect definition of table %s.%s: "
+                     "expected the type of column '%s' at position %d "
+                     "to have character set '%s' but found "
+                     "character set '%s'.", table->s->db.str, table->alias,
+                     field_def->name.str, i, field_def->cset.str,
+                     field->charset()->csname);
         error= TRUE;
       }
     }
     else
     {
-      sql_print_error("Incorrect definition of table %s.%s: "
-                      "expected column '%s' at position %d to have type %s "
-                      " but the column is not found.",
-                      table->s->db.str, table->alias,
-                      table_def->name.str, i, table_def->type.str);
+      report_error(0, "Incorrect definition of table %s.%s: "
+                   "expected column '%s' at position %d to have type %s "
+                   " but the column is not found.",
+                   table->s->db.str, table->alias,
+                   field_def->name.str, i, field_def->type.str);
       error= TRUE;
     }
   }
+
+  if (! error)
+    table->s->table_field_def_cache= table_def;
+
   DBUG_RETURN(error);
 }
 
diff --git a/sql/table.h b/sql/table.h
index 03220487684..e2f47c9c588 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -288,6 +288,36 @@ typedef enum enum_table_category TABLE_CATEGORY;
 TABLE_CATEGORY get_table_category(const LEX_STRING *db,
                                   const LEX_STRING *name);
 
+
+typedef struct st_table_field_type
+{
+  LEX_STRING name;
+  LEX_STRING type;
+  LEX_STRING cset;
+} TABLE_FIELD_TYPE;
+
+
+typedef struct st_table_field_def
+{
+  uint count;
+  const TABLE_FIELD_TYPE *field;
+} TABLE_FIELD_DEF;
+
+
+class Table_check_intact
+{
+protected:
+  virtual void report_error(uint code, const char *fmt, ...)= 0;
+
+public:
+  Table_check_intact() {}
+  virtual ~Table_check_intact() {}
+
+  /** Checks whether a table is intact. */
+  bool check(TABLE *table, const TABLE_FIELD_DEF *table_def);
+};
+
+
 /*
   This structure is shared between different table objects. There is one
   instance of table share per one table in the database.
@@ -420,6 +450,18 @@ struct TABLE_SHARE
   handlerton *default_part_db_type;
 #endif
 
+  /**
+    Cache the checked structure of this table.
+
+    The pointer data is used to describe the structure that
+    a instance of the table must have. Each element of the
+    array specifies a field that must exist on the table.
+
+    The pointer is cached in order to perform the check only
+    once -- when the table is loaded from the disk.
+  */
+  const TABLE_FIELD_DEF *table_field_def_cache;
+
   /** place to store storage engine specific data */
   void *ha_data;
   void (*ha_data_destroy)(void *); /* An optional destructor for ha_data */
@@ -1626,17 +1668,6 @@ typedef struct st_open_table_list{
   uint32 in_use,locked;
 } OPEN_TABLE_LIST;
 
-typedef struct st_table_field_w_type
-{
-  LEX_STRING name;
-  LEX_STRING type;
-  LEX_STRING cset;
-} TABLE_FIELD_W_TYPE;
-
-
-my_bool
-table_check_intact(TABLE *table, const uint table_f_count,
-                   const TABLE_FIELD_W_TYPE *table_def);
 
 static inline my_bitmap_map *tmp_use_all_columns(TABLE *table,
                                                  MY_BITMAP *bitmap)