Auto Merge from 5.1-rep+2

59 files changed, 2974 insertions, 1150 deletions

diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc
index 0d883e030f5..a6f249d286f 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 354c911e1c0..21082550ba7 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -59,6 +59,8 @@ const char field_separator=',';
 #define ASSERT_COLUMN_MARKED_FOR_READ DBUG_ASSERT(!table || (!table->read_set || bitmap_is_set(table->read_set, field_index)))
 #define ASSERT_COLUMN_MARKED_FOR_WRITE DBUG_ASSERT(!table || (!table->write_set || bitmap_is_set(table->write_set, field_index)))
 
+#define FLAGSTR(S,F) ((S) & (F) ? #F " " : "")
+
 /*
   Rules for merging different types of fields in UNION
 
@@ -997,6 +999,22 @@ test_if_important_data(CHARSET_INFO *cs, const char *str, const char *strend)
 
 
 /**
+   Function to compare two unsigned integers for their relative order.
+   Used below. In an anonymous namespace to not clash with definitions
+   in other files.
+ */
+namespace {
+  int compare(unsigned int a, unsigned int b)
+  {
+    if (a < b)
+      return -1;
+    if (b < a)
+      return 1;
+    return 0;
+}
+}
+
+/**
   Detect Item_result by given field type of UNION merge result.
 
   @param field_type  given field type
@@ -1367,22 +1385,48 @@ bool Field::send_binary(Protocol *protocol)
 /**
    Check to see if field size is compatible with destination.
 
-   This method is used in row-based replication to verify that the slave's
-   field size is less than or equal to the master's field size. The 
-   encoded field metadata (from the master or source) is decoded and compared
-   to the size of this field (the slave or destination). 
+   This method is used in row-based replication to verify that the
+   slave's field size is less than or equal to the master's field
+   size. The encoded field metadata (from the master or source) is
+   decoded and compared to the size of this field (the slave or
+   destination).
+
+   @note
+
+   The comparison is made so that if the source data (from the master)
+   is less than the target data (on the slave), -1 is returned in @c
+   <code>*order_var</code>. This implies that a conversion is
+   necessary, but that it is lossy and can result in truncation of the
+   value.
+
+   If the source data is strictly greater than the target data, 1 is
+   returned in <code>*order_var</code>. This implies that the source
+   type can is contained in the target type and that a conversion is
+   necessary but is non-lossy.
+
+   If no conversion is required to fit the source type in the target
+   type, 0 is returned in <code>*order_var</code>.
 
    @param   field_metadata   Encoded size in field metadata
+   @param   mflags           Flags from the table map event for the table.
+   @param   order_var        Pointer to variable where the order
+                             between the source field and this field
+                             will be returned.
 
-   @retval 0 if this field's size is < the source field's size
-   @retval 1 if this field's size is >= the source field's size
+   @return @c true if this field's size is compatible with the
+   master's field size, @c false otherwise.
 */
-int Field::compatible_field_size(uint field_metadata,
-                                 const Relay_log_info *rli_arg __attribute__((unused)))
+bool Field::compatible_field_size(uint field_metadata,
+                                  Relay_log_info *rli_arg __attribute__((unused)),
+                                  uint16 mflags __attribute__((unused)),
+                                  int *order_var)
 {
   uint const source_size= pack_length_from_metadata(field_metadata);
   uint const destination_size= row_pack_length();
-  return (source_size <= destination_size);
+  DBUG_PRINT("debug", ("real_type: %d, source_size: %u, destination_size: %u",
+                       real_type(), source_size, destination_size));
+  *order_var = compare(source_size, destination_size);
+  return true;
 }
 
 
@@ -2486,6 +2530,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);
@@ -2822,33 +2910,16 @@ uint Field_new_decimal::pack_length_from_metadata(uint field_metadata)
 }
 
 
-/**
-   Check to see if field size is compatible with destination.
-
-   This method is used in row-based replication to verify that the slave's
-   field size is less than or equal to the master's field size. The 
-   encoded field metadata (from the master or source) is decoded and compared
-   to the size of this field (the slave or destination). 
-
-   @param   field_metadata   Encoded size in field metadata
-
-   @retval 0 if this field's size is < the source field's size
-   @retval 1 if this field's size is >= the source field's size
-*/
-int Field_new_decimal::compatible_field_size(uint field_metadata,
-                                             const Relay_log_info * __attribute__((unused)))
+bool Field_new_decimal::compatible_field_size(uint field_metadata,
+                                              Relay_log_info * __attribute__((unused)),
+                                              uint16 mflags __attribute__((unused)),
+                                              int *order_var)
 {
-  int compatible= 0;
   uint const source_precision= (field_metadata >> 8U) & 0x00ff;
   uint const source_decimal= field_metadata & 0x00ff; 
-  uint const source_size= my_decimal_get_binary_size(source_precision, 
-                                                     source_decimal);
-  uint const destination_size= row_pack_length();
-  compatible= (source_size <= destination_size);
-  if (compatible)
-    compatible= (source_precision <= precision) &&
-                (source_decimal <= decimals());
-  return (compatible);
+  int order= compare(source_precision, precision);
+  *order_var= order != 0 ? order : compare(source_decimal, dec);
+  return true;
 }
 
 
@@ -6611,8 +6682,11 @@ check_field_for_37426(const void *param_arg)
 }
 #endif
 
-int Field_string::compatible_field_size(uint field_metadata,
-                                        const Relay_log_info *rli_arg)
+bool
+Field_string::compatible_field_size(uint field_metadata,
+                                    Relay_log_info *rli_arg,
+                                    uint16 mflags __attribute__((unused)),
+                                    int *order_var)
 {
 #ifdef HAVE_REPLICATION
   const Check_field_param check_param = { this };
@@ -6620,7 +6694,7 @@ int Field_string::compatible_field_size(uint field_metadata,
                          check_field_for_37426, &check_param))
     return FALSE;                        // Not compatible field sizes
 #endif
-  return Field::compatible_field_size(field_metadata, rli_arg);
+  return Field::compatible_field_size(field_metadata, rli_arg, mflags, order_var);
 }
 
 
@@ -6682,6 +6756,8 @@ uchar *Field_string::pack(uchar *to, const uchar *from,
 {
   uint length=      min(field_length,max_length);
   uint local_char_length= max_length/field_charset->mbmaxlen;
+  DBUG_PRINT("debug", ("Packing field '%s' - length: %u ", field_name, length));
+
   if (length > local_char_length)
     local_char_length= my_charpos(field_charset, from, from+length,
                                   local_char_length);
@@ -7529,6 +7605,7 @@ Field_blob::Field_blob(uchar *ptr_arg, uchar *null_ptr_arg, uchar null_bit_arg,
                  cs),
    packlength(blob_pack_length)
 {
+  DBUG_ASSERT(blob_pack_length <= 4); // Only pack lengths 1-4 supported currently
   flags|= BLOB_FLAG;
   share->blob_fields++;
   /* TODO: why do not fill table->s->blob_field array here? */
@@ -7939,8 +8016,10 @@ int Field_blob::key_cmp(const uchar *a,const uchar *b)
 */
 int Field_blob::do_save_field_metadata(uchar *metadata_ptr)
 {
+  DBUG_ENTER("Field_blob::do_save_field_metadata");
   *metadata_ptr= pack_length_no_ptr();
-  return 1;
+  DBUG_PRINT("debug", ("metadata: %u (pack_length_no_ptr)", *metadata_ptr));
+  DBUG_RETURN(1);
 }
 
 
@@ -8881,6 +8960,9 @@ Field_bit::Field_bit(uchar *ptr_arg, uint32 len_arg, uchar *null_ptr_arg,
     bit_ptr(bit_ptr_arg), bit_ofs(bit_ofs_arg), bit_len(len_arg & 7),
     bytes_in_rec(len_arg / 8)
 {
+  DBUG_ENTER("Field_bit::Field_bit");
+  DBUG_PRINT("enter", ("ptr_arg: %p, null_ptr_arg: %p, len_arg: %u, bit_len: %u, bytes_in_rec: %u",
+                       ptr_arg, null_ptr_arg, len_arg, bit_len, bytes_in_rec));
   flags|= UNSIGNED_FLAG;
   /*
     Ensure that Field::eq() can distinguish between two different bit fields.
@@ -8888,6 +8970,7 @@ Field_bit::Field_bit(uchar *ptr_arg, uint32 len_arg, uchar *null_ptr_arg,
   */
   if (!null_ptr_arg)
     null_bit= bit_ofs_arg;
+  DBUG_VOID_RETURN;
 }
 
 
@@ -9172,9 +9255,17 @@ uint Field_bit::get_key_image(uchar *buff, uint length, imagetype type_arg)
 */
 int Field_bit::do_save_field_metadata(uchar *metadata_ptr)
 {
-  *metadata_ptr= bit_len;
-  *(metadata_ptr + 1)= bytes_in_rec;
-  return 2;
+  DBUG_ENTER("Field_bit::do_save_field_metadata");
+  DBUG_PRINT("debug", ("bit_len: %d, bytes_in_rec: %d",
+                       bit_len, bytes_in_rec));
+  /*
+    Since this class and Field_bit_as_char have different ideas of
+    what should be stored here, we compute the values of the metadata
+    explicitly using the field_length.
+   */
+  metadata_ptr[0]= field_length % 8;
+  metadata_ptr[1]= field_length / 8;
+  DBUG_RETURN(2);
 }
 
 
@@ -9199,34 +9290,34 @@ uint Field_bit::pack_length_from_metadata(uint field_metadata)
 }
 
 
-/**
-   Check to see if field size is compatible with destination.
-
-   This method is used in row-based replication to verify that the slave's
-   field size is less than or equal to the master's field size. The 
-   encoded field metadata (from the master or source) is decoded and compared
-   to the size of this field (the slave or destination). 
+bool
+Field_bit::compatible_field_size(uint field_metadata,
+                                 Relay_log_info * __attribute__((unused)),
+                                 uint16 mflags,
+                                 int *order_var)
+{
+  DBUG_ENTER("Field_bit::compatible_field_size");
+  DBUG_ASSERT((field_metadata >> 16) == 0);
+  uint from_bit_len=
+    8 * (field_metadata >> 8) + (field_metadata & 0xff);
+  uint to_bit_len= max_display_length();
+  DBUG_PRINT("debug", ("from_bit_len: %u, to_bit_len: %u",
+                       from_bit_len, to_bit_len));
+  /*
+    If the bit length exact flag is clear, we are dealing with an old
+    master, so we allow some less strict behaviour if replicating by
+    moving both bit lengths to an even multiple of 8.
 
-   @param   field_metadata   Encoded size in field metadata
+    We do this by computing the number of bytes to store the field
+    instead, and then compare the result.
+   */
+  if (!(mflags & Table_map_log_event::TM_BIT_LEN_EXACT_F)) {
+    from_bit_len= (from_bit_len + 7) / 8;
+    to_bit_len= (to_bit_len + 7) / 8;
+  }
 
-   @retval 0 if this field's size is < the source field's size
-   @retval 1 if this field's size is >= the source field's size
-*/
-int Field_bit::compatible_field_size(uint field_metadata,
-                                     const Relay_log_info * __attribute__((unused)))
-{
-  int compatible= 0;
-  uint const source_size= pack_length_from_metadata(field_metadata);
-  uint const destination_size= row_pack_length();
-  uint const from_bit_len= field_metadata & 0x00ff;
-  uint const from_len= (field_metadata >> 8U) & 0x00ff;
-  if ((bit_len == 0) || (from_bit_len == 0))
-    compatible= (source_size <= destination_size);
-  else if (from_bit_len > bit_len)
-    compatible= (from_len < bytes_in_rec);
-  else
-    compatible= ((from_bit_len <= bit_len) && (from_len <= bytes_in_rec));
-  return (compatible);
+  *order_var= compare(from_bit_len, to_bit_len);
+  DBUG_RETURN(TRUE);
 }
 
 
@@ -9292,8 +9383,15 @@ const uchar *
 Field_bit::unpack(uchar *to, const uchar *from, uint param_data,
                   bool low_byte_first __attribute__((unused)))
 {
+  DBUG_ENTER("Field_bit::unpack");
+  DBUG_PRINT("enter", ("to: %p, from: %p, param_data: 0x%x",
+                       to, from, param_data));
+  DBUG_PRINT("debug", ("bit_ptr: %p, bit_len: %u, bit_ofs: %u",
+                       bit_ptr, bit_len, bit_ofs));
   uint const from_len= (param_data >> 8U) & 0x00ff;
   uint const from_bit_len= param_data & 0x00ff;
+  DBUG_PRINT("debug", ("from_len: %u, from_bit_len: %u",
+                       from_len, from_bit_len));
   /*
     If the parameter data is zero (i.e., undefined), or if the master
     and slave have the same sizes, then use the old unpack() method.
@@ -9314,7 +9412,7 @@ Field_bit::unpack(uchar *to, const uchar *from, uint param_data,
       from++;
     }
     memcpy(to, from, bytes_in_rec);
-    return from + bytes_in_rec;
+    DBUG_RETURN(from + bytes_in_rec);
   }
 
   /*
@@ -9340,7 +9438,7 @@ Field_bit::unpack(uchar *to, const uchar *from, uint param_data,
   bitmap_set_bit(table->write_set,field_index);
   store(value, new_len, system_charset_info);
   my_afree(value);
-  return from + len;
+  DBUG_RETURN(from + len);
 }
 
 
@@ -9468,8 +9566,11 @@ void Create_field::create_length_to_internal_length(void)
 */
 void Create_field::init_for_tmp_table(enum_field_types sql_type_arg,
                                       uint32 length_arg, uint32 decimals_arg,
-                                      bool maybe_null, bool is_unsigned)
+                                      bool maybe_null, bool is_unsigned,
+                                      uint pack_length)
 {
+  DBUG_ENTER("Create_field::init_for_tmp_table");
+
   field_name= "";
   sql_type= sql_type_arg;
   char_length= length= length_arg;;
@@ -9477,10 +9578,92 @@ void Create_field::init_for_tmp_table(enum_field_types sql_type_arg,
   interval= 0;
   charset= &my_charset_bin;
   geom_type= Field::GEOM_GEOMETRY;
-  pack_flag= (FIELDFLAG_NUMBER |
-              ((decimals_arg & FIELDFLAG_MAX_DEC) << FIELDFLAG_DEC_SHIFT) |
-              (maybe_null ? FIELDFLAG_MAYBE_NULL : 0) |
-              (is_unsigned ? 0 : FIELDFLAG_DECIMAL));
+
+  DBUG_PRINT("enter", ("sql_type: %d, length: %u, pack_length: %u",
+                       sql_type_arg, length_arg, pack_length));
+
+  /*
+    These pack flags are crafted to get it correctly through the
+    branches of make_field().
+   */
+  switch (sql_type_arg)
+  {
+  case MYSQL_TYPE_VARCHAR:
+  case MYSQL_TYPE_VAR_STRING:
+  case MYSQL_TYPE_STRING:
+  case MYSQL_TYPE_SET:
+    pack_flag= 0;
+    break;
+
+  case MYSQL_TYPE_GEOMETRY:
+    pack_flag= FIELDFLAG_GEOM;
+    break;
+
+  case MYSQL_TYPE_ENUM:
+    pack_flag= FIELDFLAG_INTERVAL;
+    break;
+
+  case MYSQL_TYPE_DECIMAL:
+  case MYSQL_TYPE_NEWDECIMAL:
+  case MYSQL_TYPE_FLOAT:
+  case MYSQL_TYPE_DOUBLE:
+    pack_flag= FIELDFLAG_DECIMAL | FIELDFLAG_NUMBER |
+      (decimals_arg & FIELDFLAG_MAX_DEC) << FIELDFLAG_DEC_SHIFT;
+    break;
+
+  case MYSQL_TYPE_TINY_BLOB:
+  case MYSQL_TYPE_MEDIUM_BLOB:
+  case MYSQL_TYPE_LONG_BLOB:
+  case MYSQL_TYPE_BLOB:
+    pack_flag= FIELDFLAG_BLOB;
+    break;
+
+  case MYSQL_TYPE_BIT:
+    pack_flag= FIELDFLAG_NUMBER | FIELDFLAG_TREAT_BIT_AS_CHAR;
+    break;
+
+  default:
+    pack_flag= FIELDFLAG_NUMBER;
+    break;
+  }
+
+  /*
+    Set the pack flag correctly for the blob-like types. This sets the
+    packtype to something that make_field can use. If the pack type is
+    not set correctly, the packlength will be reeeeally wierd (like
+    129 or so).
+   */
+  switch (sql_type_arg)
+  {
+  case MYSQL_TYPE_ENUM:
+  case MYSQL_TYPE_SET:
+  case MYSQL_TYPE_TINY_BLOB:
+  case MYSQL_TYPE_MEDIUM_BLOB:
+  case MYSQL_TYPE_LONG_BLOB:
+  case MYSQL_TYPE_BLOB:
+  case MYSQL_TYPE_GEOMETRY:
+    // If you are going to use the above types, you have to pass a
+    // pack_length as parameter. Assert that is really done.
+    DBUG_ASSERT(pack_length != ~0U);
+    pack_flag|= pack_length_to_packflag(pack_length);
+    break;
+  default:
+    /* Nothing */
+    break;
+  }
+
+  pack_flag|=
+    (maybe_null ? FIELDFLAG_MAYBE_NULL : 0) |
+    (is_unsigned ? 0 : FIELDFLAG_DECIMAL);
+
+  DBUG_PRINT("debug", ("pack_flag: %s%s%s%s%s, pack_type: %d",
+                       FLAGSTR(pack_flag, FIELDFLAG_BINARY),
+                       FLAGSTR(pack_flag, FIELDFLAG_NUMBER),
+                       FLAGSTR(pack_flag, FIELDFLAG_INTERVAL),
+                       FLAGSTR(pack_flag, FIELDFLAG_GEOM),
+                       FLAGSTR(pack_flag, FIELDFLAG_BLOB),
+                       f_packtype(pack_flag)));
+  DBUG_VOID_RETURN;
 }
 
 
@@ -9987,6 +10170,14 @@ Field *make_field(TABLE_SHARE *share, uchar *ptr, uint32 field_length,
   default: break;
   }
 
+  DBUG_PRINT("debug", ("field_type: %d, field_length: %u, interval: %p, pack_flag: %s%s%s%s%s",
+                       field_type, field_length, interval,
+                       FLAGSTR(pack_flag, FIELDFLAG_BINARY),
+                       FLAGSTR(pack_flag, FIELDFLAG_INTERVAL),
+                       FLAGSTR(pack_flag, FIELDFLAG_NUMBER),
+                       FLAGSTR(pack_flag, FIELDFLAG_PACK),
+                       FLAGSTR(pack_flag, FIELDFLAG_BLOB)));
+
   if (f_is_alpha(pack_flag))
   {
     if (!f_is_packed(pack_flag))
diff --git a/sql/field.h b/sql/field.h
index 18b017516a7..041610f5385 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -167,22 +167,13 @@ public:
     table, which is located on disk).
   */
   virtual uint32 pack_length_in_rec() const { return pack_length(); }
-  virtual int compatible_field_size(uint field_metadata,
-                                    const Relay_log_info *);
+  virtual bool compatible_field_size(uint metadata, Relay_log_info *rli,
+                                     uint16 mflags, int *order);
   virtual uint pack_length_from_metadata(uint field_metadata)
-  { return field_metadata; }
-  /*
-    This method is used to return the size of the data in a row-based
-    replication row record. The default implementation of returning 0 is
-    designed to allow fields that do not use metadata to return TRUE (1)
-    from compatible_field_size() which uses this function in the comparison.
-    The default value for field metadata for fields that do not have 
-    metadata is 0. Thus, 0 == 0 means the fields are compatible in size.
-
-    Note: While most classes that override this method return pack_length(),
-    the classes Field_string, Field_varstring, and Field_blob return 
-    field_length + 1, field_length, and pack_length_no_ptr() respectfully.
-  */
+  {
+    DBUG_ENTER("Field::pack_length_from_metadata");
+    DBUG_RETURN(field_metadata);
+  }
   virtual uint row_pack_length() { return 0; }
   virtual int save_field_metadata(uchar *first_byte)
   { return do_save_field_metadata(first_byte); }
@@ -641,6 +632,13 @@ public:
   int store_decimal(const my_decimal *);
   my_decimal *val_decimal(my_decimal *);
   uint is_equal(Create_field *new_field);
+  uint row_pack_length() { return pack_length(); }
+  uint32 pack_length_from_metadata(uint field_metadata) {
+    uint32 length= pack_length();
+    DBUG_PRINT("result", ("pack_length_from_metadata(%d): %u",
+                          field_metadata, length));
+    return length;
+  }
   int check_int(CHARSET_INFO *cs, const char *str, int length,
                 const char *int_end, int error);
   bool get_int(CHARSET_INFO *cs, const char *from, uint len, 
@@ -805,11 +803,12 @@ public:
   uint32 pack_length() const { return (uint32) bin_size; }
   uint pack_length_from_metadata(uint field_metadata);
   uint row_pack_length() { return pack_length(); }
-  int compatible_field_size(uint field_metadata,
-                            const Relay_log_info *rli);
+  bool compatible_field_size(uint field_metadata, Relay_log_info *rli,
+                             uint16 mflags, int *order_var);
   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 *);
 };
 
 
@@ -1500,9 +1499,9 @@ public:
       return row_pack_length();
     return (((field_metadata >> 4) & 0x300) ^ 0x300) + (field_metadata & 0x00ff);
   }
-  int compatible_field_size(uint field_metadata,
-                            const Relay_log_info *rli);
-  uint row_pack_length() { return (field_length + 1); }
+  bool compatible_field_size(uint field_metadata, Relay_log_info *rli,
+                             uint16 mflags, int *order_var);
+  uint row_pack_length() { return field_length; }
   int pack_cmp(const uchar *a,const uchar *b,uint key_length,
                my_bool insert_or_update);
   int pack_cmp(const uchar *b,uint key_length,my_bool insert_or_update);
@@ -1964,8 +1963,8 @@ public:
   uint pack_length_from_metadata(uint field_metadata);
   uint row_pack_length()
   { return (bytes_in_rec + ((bit_len > 0) ? 1 : 0)); }
-  int compatible_field_size(uint field_metadata,
-                            const Relay_log_info *rli);
+  bool compatible_field_size(uint metadata, Relay_log_info *rli,
+                             uint16 mflags, int *order_var);
   void sql_type(String &str) const;
   virtual uchar *pack(uchar *to, const uchar *from,
                       uint max_length, bool low_byte_first);
@@ -2068,7 +2067,8 @@ public:
   /* Init for a tmp table field. To be extended if need be. */
   void init_for_tmp_table(enum_field_types sql_type_arg,
                           uint32 max_length, uint32 decimals,
-                          bool maybe_null, bool is_unsigned);
+                          bool maybe_null, bool is_unsigned,
+                          uint pack_length = ~0U);
 
   bool init(THD *thd, char *field_name, enum_field_types type, char *length,
             char *decimals, uint type_modifier, Item *default_value,
diff --git a/sql/item.cc b/sql/item.cc
index 8f487872f1b..b35a6ae3d6e 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -4899,9 +4899,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,
@@ -6940,9 +6938,24 @@ int stored_field_cmp_to_item(THD *thd, Field *field, Item *item)
 
 Item_cache* Item_cache::get_cache(const Item *item)
 {
-  switch (item->result_type()) {
+  return get_cache(item, item->result_type());
+}
+
+
+/**
+  Get a cache item of given type.
+
+  @param item         value to be cached
+  @param type         required type of cache
+
+  @return cache item
+*/
+
+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:
@@ -6958,6 +6971,13 @@ Item_cache* Item_cache::get_cache(const Item *item)
   }
 }
 
+void Item_cache::store(Item *item)
+{
+  example= item;
+  if (!item)
+    null_value= TRUE;
+  value_cached= FALSE;
+}
 
 void Item_cache::print(String *str, enum_query_type query_type)
 {
@@ -6969,17 +6989,22 @@ void Item_cache::print(String *str, enum_query_type query_type)
   str->append(')');
 }
 
-
-void Item_cache_int::store(Item *item)
+bool  Item_cache_int::cache_value()
 {
-  value= item->val_int_result();
-  null_value= item->null_value;
-  unsigned_flag= item->unsigned_flag;
+  if (!example)
+    return FALSE;
+  value_cached= TRUE;
+  value= example->val_int_result();
+  null_value= example->null_value;
+  unsigned_flag= example->unsigned_flag;
+  return TRUE;
 }
 
 
 void Item_cache_int::store(Item *item, longlong val_arg)
 {
+  /* An explicit values is given, save it. */
+  value_cached= TRUE;
   value= val_arg;
   null_value= item->null_value;
   unsigned_flag= item->unsigned_flag;
@@ -6989,6 +7014,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())
+    return NULL;
   str->set(value, default_charset());
   return str;
 }
@@ -6997,21 +7024,52 @@ 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())
+    return NULL;
   int2my_decimal(E_DEC_FATAL_ERROR, value, unsigned_flag, decimal_val);
   return decimal_val;
 }
 
+double Item_cache_int::val_real()
+{
+  DBUG_ASSERT(fixed == 1);
+  if (!value_cached && !cache_value())
+    return 0.0;
+  return (double) value;
+}
 
-void Item_cache_real::store(Item *item)
+longlong Item_cache_int::val_int()
 {
-  value= item->val_result();
-  null_value= item->null_value;
+  DBUG_ASSERT(fixed == 1);
+  if (!value_cached && !cache_value())
+    return 0;
+  return 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())
+    return 0.0;
+  return value;
+}
+
 longlong Item_cache_real::val_int()
 {
   DBUG_ASSERT(fixed == 1);
+  if (!value_cached && !cache_value())
+    return 0;
   return (longlong) rint(value);
 }
 
@@ -7019,6 +7077,8 @@ longlong Item_cache_real::val_int()
 String* Item_cache_real::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
+  if (!value_cached && !cache_value())
+    return NULL;
   str->set_real(value, decimals, default_charset());
   return str;
 }
@@ -7027,22 +7087,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())
+    return NULL;
   double2my_decimal(E_DEC_FATAL_ERROR, value, decimal_val);
   return decimal_val;
 }
 
 
-void Item_cache_decimal::store(Item *item)
+bool Item_cache_decimal::cache_value()
 {
-  my_decimal *val= item->val_decimal_result(&decimal_value);
-  if (!(null_value= item->null_value) && val != &decimal_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())
+    return NULL;
   my_decimal2double(E_DEC_FATAL_ERROR, &decimal_value, &res);
   return res;
 }
@@ -7051,6 +7119,8 @@ longlong Item_cache_decimal::val_int()
 {
   DBUG_ASSERT(fixed);
   longlong res;
+  if (!value_cached && !cache_value())
+    return 0;
   my_decimal2int(E_DEC_FATAL_ERROR, &decimal_value, unsigned_flag, &res);
   return res;
 }
@@ -7058,6 +7128,8 @@ longlong Item_cache_decimal::val_int()
 String* Item_cache_decimal::val_str(String *str)
 {
   DBUG_ASSERT(fixed);
+  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);
@@ -7067,15 +7139,20 @@ 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())
+    return NULL;
   return &decimal_value;
 }
 
 
-void Item_cache_str::store(Item *item)
+bool Item_cache_str::cache_value()
 {
-  value_buff.set(buffer, sizeof(buffer), item->collation.collation);
-  value= item->str_result(&value_buff);
-  if ((null_value= item->null_value))
+  if (!example)
+    return FALSE;
+  value_cached= TRUE;
+  value_buff.set(buffer, sizeof(buffer), example->collation.collation);
+  value= example->str_result(&value_buff);
+  if ((null_value= example->null_value))
     value= 0;
   else if (value != &value_buff)
   {
@@ -7090,6 +7167,7 @@ void Item_cache_str::store(Item *item)
     value_buff.copy(*value);
     value= &value_buff;
   }
+  return TRUE;
 }
 
 double Item_cache_str::val_real()
@@ -7097,6 +7175,8 @@ double Item_cache_str::val_real()
   DBUG_ASSERT(fixed == 1);
   int err_not_used;
   char *end_not_used;
+  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);
@@ -7108,6 +7188,8 @@ longlong Item_cache_str::val_int()
 {
   DBUG_ASSERT(fixed == 1);
   int err;
+  if (!value_cached && !cache_value())
+    return 0;
   if (value)
     return my_strntoll(value->charset(), value->ptr(),
 		       value->length(), 10, (char**) 0, &err);
@@ -7115,9 +7197,21 @@ longlong Item_cache_str::val_int()
     return (longlong)0;
 }
 
+
+String* Item_cache_str::val_str(String *str)
+{
+  DBUG_ASSERT(fixed == 1);
+  if (!value_cached && !cache_value())
+    return 0;
+  return value;
+}
+
+
 my_decimal *Item_cache_str::val_decimal(my_decimal *decimal_val)
 {
   DBUG_ASSERT(fixed == 1);
+  if (!value_cached && !cache_value())
+    return NULL;
   if (value)
     string2my_decimal(E_DEC_FATAL_ERROR, value, decimal_val);
   else
@@ -7128,6 +7222,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())
+    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;
@@ -7162,13 +7258,30 @@ 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));
+}
+
+
+bool Item_cache_row::cache_value()
+{
+  if (!example)
+    return FALSE;
+  value_cached= TRUE;
   null_value= 0;
-  item->bring_value();
+  example->bring_value();
   for (uint i= 0; i < item_count; i++)
   {
-    values[i]->store(item->element_index(i));
+    values[i]->cache_value();
     null_value|= values[i]->null_value;
   }
+  return TRUE;
 }
 
 
diff --git a/sql/item.h b/sql/item.h
index e83634843be..f37938aedbd 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -1027,7 +1027,11 @@ class sp_head;
 
 class Item_basic_constant :public Item
 {
+  table_map used_table_map;
 public:
+  Item_basic_constant(): Item(), used_table_map(0) {};
+  void set_used_tables(table_map map) { used_table_map= map; }
+  table_map used_tables() const { return used_table_map; }
   /* to prevent drop fixed flag (no need parent cleanup call) */
   void cleanup()
   {
@@ -2137,6 +2141,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;
 };
 
 
@@ -2892,15 +2913,25 @@ protected:
   */  
   Field *cached_field;
   enum enum_field_types cached_field_type;
+  /*
+    TRUE <=> cache holds value of the last stored item (i.e actual value).
+    store() stores item to be cached and sets this flag to FALSE.
+    On the first call of val_xxx function if this flag is set to FALSE the 
+    cache_value() will be called to actually cache value of saved item.
+    cache_value() will set this flag to TRUE.
+  */
+  bool value_cached;
 public:
-  Item_cache(): 
-    example(0), used_table_map(0), cached_field(0), cached_field_type(MYSQL_TYPE_STRING) 
+  Item_cache():
+    example(0), used_table_map(0), cached_field(0), cached_field_type(MYSQL_TYPE_STRING),
+    value_cached(0)
   {
     fixed= 1; 
     null_value= 1;
   }
   Item_cache(enum_field_types field_type_arg):
-    example(0), used_table_map(0), cached_field(0), cached_field_type(field_type_arg)
+    example(0), used_table_map(0), cached_field(0), cached_field_type(field_type_arg),
+    value_cached(0)
   {
     fixed= 1;
     null_value= 1;
@@ -2920,10 +2951,10 @@ public:
       cached_field= ((Item_field *)item)->field;
     return 0;
   };
-  virtual void store(Item *)= 0;
   enum Type type() const { return CACHE_ITEM; }
   enum_field_types field_type() const { return cached_field_type; }
   static Item_cache* get_cache(const Item *item);
+  static Item_cache* get_cache(const Item* item, const Item_result type);
   table_map used_tables() const { return used_table_map; }
   virtual void keep_array() {}
   virtual void print(String *str, enum_query_type query_type);
@@ -2935,6 +2966,8 @@ public:
   {
     return this == item;
   }
+  virtual void store(Item *item);
+  virtual bool cache_value()= 0;
 };
 
 
@@ -2943,18 +2976,19 @@ class Item_cache_int: public Item_cache
 protected:
   longlong value;
 public:
-  Item_cache_int(): Item_cache(), value(0) {}
+  Item_cache_int(): Item_cache(),
+    value(0) {}
   Item_cache_int(enum_field_types field_type_arg):
     Item_cache(field_type_arg), value(0) {}
 
-  void store(Item *item);
   void store(Item *item, longlong val_arg);
-  double val_real() { DBUG_ASSERT(fixed == 1); return (double) value; }
-  longlong val_int() { DBUG_ASSERT(fixed == 1); return value; }
+  double val_real();
+  longlong val_int();
   String* val_str(String *str);
   my_decimal *val_decimal(my_decimal *);
   enum Item_result result_type() const { return INT_RESULT; }
   bool result_as_longlong() { return TRUE; }
+  bool cache_value();
 };
 
 
@@ -2962,14 +2996,15 @@ class Item_cache_real: public Item_cache
 {
   double value;
 public:
-  Item_cache_real(): Item_cache(), value(0) {}
+  Item_cache_real(): Item_cache(),
+    value(0) {}
 
-  void store(Item *item);
-  double val_real() { DBUG_ASSERT(fixed == 1); return value; }
+  double val_real();
   longlong val_int();
   String* val_str(String *str);
   my_decimal *val_decimal(my_decimal *);
   enum Item_result result_type() const { return REAL_RESULT; }
+  bool cache_value();
 };
 
 
@@ -2980,12 +3015,12 @@ protected:
 public:
   Item_cache_decimal(): Item_cache() {}
 
-  void store(Item *item);
   double val_real();
   longlong val_int();
   String* val_str(String *str);
   my_decimal *val_decimal(my_decimal *);
   enum Item_result result_type() const { return DECIMAL_RESULT; }
+  bool cache_value();
 };
 
 
@@ -3003,14 +3038,14 @@ public:
                    MYSQL_TYPE_VARCHAR &&
                  !((const Item_field *) item)->field->has_charset())
   {}
-  void store(Item *item);
   double val_real();
   longlong val_int();
-  String* val_str(String *) { DBUG_ASSERT(fixed == 1); return value; }
+  String* val_str(String *);
   my_decimal *val_decimal(my_decimal *);
   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);
+  bool cache_value();
 };
 
 class Item_cache_row: public Item_cache
@@ -3020,7 +3055,8 @@ class Item_cache_row: public Item_cache
   bool save_array;
 public:
   Item_cache_row()
-    :Item_cache(), values(0), item_count(2), save_array(0) {}
+    :Item_cache(), values(0), item_count(2),
+    save_array(0) {}
   
   /*
     'allocate' used only in row transformer, to preallocate space for row 
@@ -3078,6 +3114,7 @@ public:
       values= 0;
     DBUG_VOID_RETURN;
   }
+  bool cache_value();
 };
 
 
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index c6b88cd8188..e4f26978431 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;
@@ -875,19 +887,21 @@ 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)
 {
   enum enum_date_cmp_type cmp_type;
   ulonglong const_value= (ulonglong)-1;
+  thd= current_thd;
+  owner= owner_arg;
+  set_null= set_null && owner_arg;
   a= a1;
   b= a2;
+  thd= current_thd;
 
   if ((cmp_type= can_compare_as_dates(*a, *b, &const_value)))
   {
-    thd= current_thd;
-    owner= owner_arg;
     a_type= (*a)->field_type();
     b_type= (*b)->field_type();
     a_cache= 0;
@@ -895,6 +909,10 @@ int Arg_comparator::set_cmp_func(Item_bool_func2 *owner_arg,
 
     if (const_value != (ulonglong)-1)
     {
+      /*
+        cache_converted_constant can't be used here because it can't
+        correctly convert a DATETIME value from string to int representation.
+      */
       Item_cache_int *cache= new Item_cache_int();
       /* Mark the cache as non-const to prevent re-caching. */
       cache->set_used_tables(1);
@@ -911,22 +929,22 @@ 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 &&
            (*b)->field_type() == MYSQL_TYPE_TIME)
   {
     /* Compare TIME values as integers. */
-    thd= current_thd;
-    owner= owner_arg;
     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 &&
@@ -935,20 +953,97 @@ 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 (try_year_cmp_func(type))
+    return 0;
 
+  a= cache_converted_constant(thd, a, &a_cache, type);
+  b= cache_converted_constant(thd, b, &b_cache, type);
   return set_compare_func(owner_arg, type);
 }
 
 
-void Arg_comparator::set_datetime_cmp_func(Item **a1, Item **b1)
+/*
+  Helper function to call from Arg_comparator::set_cmp_func()
+*/
+
+bool Arg_comparator::try_year_cmp_func(Item_result type)
+{
+  if (type == ROW_RESULT)
+    return FALSE;
+
+  bool a_is_year= (*a)->field_type() == MYSQL_TYPE_YEAR;
+  bool b_is_year= (*b)->field_type() == MYSQL_TYPE_YEAR;
+
+  if (!a_is_year && !b_is_year)
+    return FALSE;
+
+  if (a_is_year && b_is_year)
+  {
+    get_value_a_func= &get_year_value;
+    get_value_b_func= &get_year_value;
+  }
+  else if (a_is_year && (*b)->is_datetime())
+  {
+    get_value_a_func= &get_year_value;
+    get_value_b_func= &get_datetime_value;
+  }
+  else if (b_is_year && (*a)->is_datetime())
+  {
+    get_value_b_func= &get_year_value;
+    get_value_a_func= &get_datetime_value;
+  }
+  else
+    return FALSE;
+
+  is_nulls_eq= is_owner_equal_func();
+  func= &Arg_comparator::compare_datetime;
+
+  return TRUE;
+}
+
+/**
+  Convert and cache a constant.
+
+  @param value      [in]  An item to cache
+  @param cache_item [out] Placeholder for the cache item
+  @param type       [in]  Comparison type
+
+  @details
+    When given item is a constant and its type differs from comparison type
+    then cache its value to avoid type conversion of this constant on each
+    evaluation. In this case the value is cached and the reference to the cache
+    is returned.
+    Original value is returned otherwise.
+
+  @return cache item or original value.
+*/
+
+Item** Arg_comparator::cache_converted_constant(THD *thd, Item **value,
+                                                Item **cache_item,
+                                                Item_result type)
+{
+  /* Don't need cache if doing context analysis only. */
+  if (!thd->is_context_analysis_only() &&
+      (*value)->const_item() && type != (*value)->result_type())
+  {
+    Item_cache *cache= Item_cache::get_cache(*value, type);
+    cache->setup(*value);
+    *cache_item= cache;
+    return cache_item;
+  }
+  return value;
+}
+
+
+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();
@@ -957,7 +1052,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;
 }
 
 
@@ -1057,6 +1153,56 @@ get_datetime_value(THD *thd, Item ***item_arg, Item **cache_arg,
   return value;
 }
 
+
+/*
+  Retrieves YEAR value of 19XX-00-00 00:00:00 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_datetime() function.
+    Converts year to DATETIME of form YYYY-00-00 00:00:00 for the compatibility
+    with the get_datetime_value function result.
+
+  RETURN
+    obtained value
+*/
+
+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;
+
+  /* Convert year to DATETIME of form YYYY-00-00 00:00:00 (YYYY0000000000). */
+  value*= 10000000000LL;
+
+  return value;
+}
+
+
 /*
   Compare items values as dates.
 
@@ -1089,25 +1235,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. */
@@ -1120,15 +1266,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;
 }
 
@@ -1147,18 +1295,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;
 }
 
@@ -1171,8 +1321,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);
@@ -1182,8 +1332,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);
@@ -1204,13 +1354,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;
 }
 
@@ -1224,11 +1376,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;
 }
 
@@ -1266,7 +1420,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)
@@ -1274,7 +1429,8 @@ int Arg_comparator::compare_real_fixed()
       return 1;
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1297,13 +1453,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;
 }
 
@@ -1320,13 +1478,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;
 }
 
@@ -1343,7 +1503,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)
@@ -1351,7 +1512,8 @@ int Arg_comparator::compare_int_signed_unsigned()
       return 1;
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1368,7 +1530,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)
@@ -1378,7 +1541,8 @@ int Arg_comparator::compare_int_unsigned_signed()
       return 1;
     }
   }
-  owner->null_value= 1;
+  if (set_null)
+    owner->null_value= 1;
   return -1;
 }
 
@@ -1414,10 +1578,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:
@@ -1426,7 +1591,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;
@@ -1582,6 +1747,7 @@ longlong Item_in_optimizer::val_int()
   bool tmp;
   DBUG_ASSERT(fixed == 1);
   cache->store(args[0]);
+  cache->cache_value();
   
   if (cache->null_value)
   {
@@ -1749,8 +1915,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;
@@ -2033,8 +2199,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)
   {
@@ -4362,13 +4528,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;
@@ -4392,7 +4558,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;
@@ -4423,7 +4589,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 && (
@@ -4478,7 +4644,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 ec703d0a9f9..7f0c6d5b106 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,40 @@ 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.
   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);
+  bool try_year_cmp_func(Item_result type);
 public:
   DTCollation cmp_collation;
+  /* Allow owner function to use string buffers. */
+  String value1, value2;
 
-  Arg_comparator(): thd(0), a_cache(0), b_cache(0) {};
+  Arg_comparator(): thd(0), a_cache(0), b_cache(0), set_null(TRUE),
+    get_value_a_func(0), get_value_b_func(0) {};
   Arg_comparator(Item **a1, Item **a2): a(a1), b(a2), thd(0),
-    a_cache(0), b_cache(0) {};
+    a_cache(0), b_cache(0), set_null(TRUE),
+    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()));
@@ -96,8 +106,15 @@ public:
   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;
 };
@@ -327,7 +344,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:
@@ -336,7 +352,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 0d21c50c946..323cdfc6492 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -3526,6 +3526,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(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_get_lock(arg1, arg2);
 }
@@ -3637,6 +3638,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(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_is_free_lock(arg1);
 }
@@ -3647,6 +3649,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(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_is_used_lock(arg1);
 }
@@ -3964,6 +3967,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(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
+
   if (item_list != NULL)
     arg_count= item_list->elements;
 
@@ -4206,6 +4211,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(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_release_lock(arg1);
 }
@@ -4329,6 +4335,7 @@ Create_func_sleep Create_func_sleep::s_singleton;
 Item*
 Create_func_sleep::create(THD *thd, Item *arg1)
 {
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_sleep(arg1);
 }
@@ -4597,6 +4604,7 @@ Create_func_version Create_func_version::s_singleton;
 Item*
 Create_func_version::create(THD *thd)
 {
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   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 ac52f36474a..977a0de39af 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 39c4b8e7033..183a628f8e4 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.cc b/sql/item_subselect.cc
index fa776ea3dca..26d3833f72c 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -480,6 +480,7 @@ Item_singlerow_subselect::select_transformer(JOIN *join)
 void Item_singlerow_subselect::store(uint i, Item *item)
 {
   row[i]->store(item);
+  row[i]->cache_value();
 }
 
 enum Item_result Item_singlerow_subselect::result_type() const
@@ -1831,6 +1832,7 @@ void subselect_engine::set_row(List<Item> &item_list, Item_cache **row)
     if (!(row[i]= Item_cache::get_cache(sel_item)))
       return;
     row[i]->setup(sel_item);
+    row[i]->store(sel_item);
   }
   if (item_list.elements > 1)
     res_type= ROW_RESULT;
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 38251294053..4ab8e75ddf5 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -517,8 +517,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:
@@ -615,35 +614,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)
 {
@@ -663,15 +633,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;
@@ -680,10 +647,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();
@@ -701,6 +668,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)
 {
@@ -1270,8 +1261,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)
@@ -1592,19 +1582,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;
 }
 
@@ -1613,30 +1591,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()
@@ -1644,18 +1599,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();
 }
 
 
@@ -1664,26 +1608,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);
 }
 
 
@@ -1693,25 +1618,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);
 }
 
 
@@ -1720,7 +1627,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.
@@ -1741,63 +1650,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:
+  /* 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(&sum_dec, val) > 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;
 }
@@ -1805,63 +1673,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;
 }
@@ -2226,14 +2053,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 993ec1597b4..da9ad9963c5 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -326,22 +326,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; }
@@ -667,6 +651,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"; }
 };
 
 
@@ -735,6 +720,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"; }
 };
 
 
@@ -810,6 +796,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"; }
 };
 
 /*
@@ -835,14 +822,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;
@@ -850,12 +836,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 b5037c08b3c..b293145cc27 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,9 +876,9 @@ static bool get_interval_info(const char *str,uint length,CHARSET_INFO *cs,
       value= value*LL(10) + (longlong) (*str - '0');
     if (transform_msec && i == count - 1) // microseconds always last
     {
-      long msec_length= 6 - (uint) (str - start);
+      int msec_length= 6 - (int)(str - start);
       if (msec_length > 0)
-	value*= (long) log_10_int[msec_length];
+        value*= (long)log_10_int[msec_length];
     }
     values[i]= value;
     while (str != end && !my_isdigit(cs,*str))
diff --git a/sql/item_xmlfunc.cc b/sql/item_xmlfunc.cc
index 1eff00027f2..3e20b90e68e 100644
--- a/sql/item_xmlfunc.cc
+++ b/sql/item_xmlfunc.cc
@@ -941,14 +941,16 @@ static Item *create_comparator(MY_XPATH *xpath,
      in a loop through all of the nodes in the node set.
     */
 
-    Item *fake= new Item_string("", 0, xpath->cs);
+    Item_string *fake= new Item_string("", 0, xpath->cs);
+    /* Don't cache fake because its value will be changed during comparison.*/
+    fake->set_used_tables(RAND_TABLE_BIT);
     Item_nodeset_func *nodeset;
     Item *scalar, *comp;
     if (a->type() == Item::XPATH_NODESET)
     {
       nodeset= (Item_nodeset_func*) a;
       scalar= b;
-      comp= eq_func(oper, fake, scalar);
+      comp= eq_func(oper, (Item*)fake, scalar);
     }
     else
     {
diff --git a/sql/log.cc b/sql/log.cc
index ba3aed64f21..58a9eabc678 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -4126,11 +4126,8 @@ int THD::binlog_write_table_map(TABLE *table, bool is_transactional)
   DBUG_ASSERT(is_current_stmt_binlog_format_row() && mysql_bin_log.is_open());
   DBUG_ASSERT(table->s->table_map_id != ULONG_MAX);
 
-  Table_map_log_event::flag_set const
-    flags= Table_map_log_event::TM_NO_FLAGS;
-
   Table_map_log_event
-    the_event(this, table, table->s->table_map_id, is_transactional, flags);
+    the_event(this, table, table->s->table_map_id, is_transactional);
 
   if (binlog_table_maps == 0)
     binlog_start_trans_and_stmt();
@@ -6003,9 +6000,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/log_event.cc b/sql/log_event.cc
index c085815b777..a8a08eb5d14 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -29,7 +29,6 @@
 #include "rpl_rli.h"
 #include "rpl_mi.h"
 #include "rpl_filter.h"
-#include "rpl_utility.h"
 #include "rpl_record.h"
 #include <my_dir.h>
 
@@ -37,6 +36,7 @@
 
 #include <base64.h>
 #include <my_bitmap.h>
+#include "rpl_utility.h"
 
 #define log_cs	&my_charset_latin1
 
@@ -1571,37 +1571,14 @@ log_event_print_value(IO_CACHE *file, const uchar *ptr,
         /* a long CHAR() field: see #37426 */
         length= byte1 | (((byte0 & 0x30) ^ 0x30) << 4);
         type= byte0 | 0x30;
-        goto beg;
-      }
-
-      switch (byte0)
-      {
-        case MYSQL_TYPE_SET:
-        case MYSQL_TYPE_ENUM:
-        case MYSQL_TYPE_STRING:
-          type= byte0;
-          length= byte1;
-          break;
-
-        default:
-
-        {
-          char tmp[5];
-          my_snprintf(tmp, sizeof(tmp), "%04X", meta);
-          my_b_printf(file,
-                      "!! Don't know how to handle column type=%d meta=%d (%s)",
-                      type, meta, tmp);
-          return 0;
-        }
       }
+      else
+        length = meta & 0xFF;
     }
     else
       length= meta;
   }
 
-
-beg:
-  
   switch (type) {
   case MYSQL_TYPE_LONG:
     {
@@ -1738,6 +1715,33 @@ beg:
       return 3;
     }
     
+  case MYSQL_TYPE_NEWDATE:
+    {
+      uint32 tmp= uint3korr(ptr);
+      int part;
+      char buf[10];
+      char *pos= &buf[10];
+
+      /* Copied from field.cc */
+      *pos--=0;					// End NULL
+      part=(int) (tmp & 31);
+      *pos--= (char) ('0'+part%10);
+      *pos--= (char) ('0'+part/10);
+      *pos--= ':';
+      part=(int) (tmp >> 5 & 15);
+      *pos--= (char) ('0'+part%10);
+      *pos--= (char) ('0'+part/10);
+      *pos--= ':';
+      part=(int) (tmp >> 9);
+      *pos--= (char) ('0'+part%10); part/=10;
+      *pos--= (char) ('0'+part%10); part/=10;
+      *pos--= (char) ('0'+part%10); part/=10;
+      *pos=   (char) ('0'+part);
+      my_b_printf(file , "'%s'", buf);
+      my_snprintf(typestr, typestr_length, "DATE");
+      return 3;
+    }
+    
   case MYSQL_TYPE_DATE:
     {
       uint i32= uint3korr(ptr);
@@ -1756,7 +1760,7 @@ beg:
     }
   
   case MYSQL_TYPE_ENUM:
-    switch (length) { 
+    switch (meta & 0xFF) {
     case 1:
       my_b_printf(file, "%d", (int) *ptr);
       my_snprintf(typestr, typestr_length, "ENUM(1 byte)");
@@ -1769,15 +1773,15 @@ beg:
         return 2;
       }
     default:
-      my_b_printf(file, "!! Unknown ENUM packlen=%d", length); 
+      my_b_printf(file, "!! Unknown ENUM packlen=%d", meta & 0xFF); 
       return 0;
     }
     break;
     
   case MYSQL_TYPE_SET:
-    my_b_write_bit(file, ptr , length * 8);
-    my_snprintf(typestr, typestr_length, "SET(%d bytes)", length);
-    return length;
+    my_b_write_bit(file, ptr , (meta & 0xFF) * 8);
+    my_snprintf(typestr, typestr_length, "SET(%d bytes)", meta & 0xFF);
+    return meta & 0xFF;
   
   case MYSQL_TYPE_BLOB:
     switch (meta) {
@@ -7457,11 +7461,18 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
     */
 
     {
+      DBUG_PRINT("debug", ("Checking compability of tables to lock - tables_to_lock: %p",
+                           rli->tables_to_lock));
       RPL_TABLE_LIST *ptr= rli->tables_to_lock;
       for ( ; ptr ; ptr= static_cast<RPL_TABLE_LIST*>(ptr->next_global))
       {
-        if (ptr->m_tabledef.compatible_with(rli, ptr->table))
+        TABLE *conv_table;
+        if (!ptr->m_tabledef.compatible_with(thd, const_cast<Relay_log_info*>(rli),
+                                             ptr->table, &conv_table))
         {
+          DBUG_PRINT("debug", ("Table: %s.%s is not compatible with master",
+                               ptr->table->s->db.str,
+                               ptr->table->s->table_name.str));
           /*
             We should not honour --slave-skip-errors at this point as we are
             having severe errors which should not be skiped.
@@ -7472,12 +7483,17 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
           const_cast<Relay_log_info*>(rli)->clear_tables_to_lock();
           DBUG_RETURN(ERR_BAD_TABLE_DEF);
         }
+        DBUG_PRINT("debug", ("Table: %s.%s is compatible with master"
+                             " - conv_table: %p",
+                             ptr->table->s->db.str,
+                             ptr->table->s->table_name.str, conv_table));
+        ptr->m_conv_table= conv_table;
       }
     }
 
     /*
-      ... and then we add all the tables to the table map and remove
-      them from tables to lock.
+      ... and then we add all the tables to the table map and but keep
+      them in the tables to lock list.
 
       We also invalidate the query cache for all the tables, since
       they will now be changed.
@@ -7972,7 +7988,10 @@ int Table_map_log_event::save_field_metadata()
   DBUG_ENTER("Table_map_log_event::save_field_metadata");
   int index= 0;
   for (unsigned int i= 0 ; i < m_table->s->fields ; i++)
+  {
+    DBUG_PRINT("debug", ("field_type: %d", m_coltype[i]));
     index+= m_table->s->field[i]->save_field_metadata(&m_field_metadata[index]);
+  }
   DBUG_RETURN(index);
 }
 #endif /* !defined(MYSQL_CLIENT) */
@@ -7984,7 +8003,7 @@ int Table_map_log_event::save_field_metadata()
  */
 #if !defined(MYSQL_CLIENT)
 Table_map_log_event::Table_map_log_event(THD *thd, TABLE *tbl, ulong tid,
-                                         bool is_transactional, uint16 flags)
+                                         bool is_transactional)
   : Log_event(thd, 0, is_transactional),
     m_table(tbl),
     m_dbnam(tbl->s->db.str),
@@ -7994,7 +8013,7 @@ Table_map_log_event::Table_map_log_event(THD *thd, TABLE *tbl, ulong tid,
     m_colcnt(tbl->s->fields),
     m_memory(NULL),
     m_table_id(tid),
-    m_flags(flags),
+    m_flags(TM_BIT_LEN_EXACT_F),
     m_data_size(0),
     m_field_metadata(0),
     m_field_metadata_size(0),
@@ -8252,8 +8271,10 @@ int Table_map_log_event::do_apply_event(Relay_log_info const *rli)
       inside Relay_log_info::clear_tables_to_lock() by calling the
       table_def destructor explicitly.
     */
-    new (&table_list->m_tabledef) table_def(m_coltype, m_colcnt, 
-         m_field_metadata, m_field_metadata_size, m_null_bits);
+    new (&table_list->m_tabledef)
+      table_def(m_coltype, m_colcnt,
+                m_field_metadata, m_field_metadata_size,
+                m_null_bits, m_flags);
     table_list->m_tabledef_valid= TRUE;
 
     /*
diff --git a/sql/log_event.h b/sql/log_event.h
index 9b91d826da7..4e8edbf79b6 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -3350,16 +3350,14 @@ public:
   /* Special constants representing sets of flags */
   enum 
   {
-    TM_NO_FLAGS = 0U
+    TM_NO_FLAGS = 0U,
+    TM_BIT_LEN_EXACT_F = (1U << 0)
   };
 
-  void set_flags(flag_set flag) { m_flags |= flag; }
-  void clear_flags(flag_set flag) { m_flags &= ~flag; }
   flag_set get_flags(flag_set flag) const { return m_flags & flag; }
 
 #ifndef MYSQL_CLIENT
-  Table_map_log_event(THD *thd, TABLE *tbl, ulong tid, 
-		      bool is_transactional, uint16 flags);
+  Table_map_log_event(THD *thd, TABLE *tbl, ulong tid, bool is_transactional);
 #endif
 #ifdef HAVE_REPLICATION
   Table_map_log_event(const char *buf, uint event_len, 
@@ -3372,7 +3370,7 @@ public:
   table_def *create_table_def()
   {
     return new table_def(m_coltype, m_colcnt, m_field_metadata,
-                         m_field_metadata_size, m_null_bits);
+                         m_field_metadata_size, m_null_bits, m_flags);
   }
 #endif
   ulong get_table_id() const        { return m_table_id; }
diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index ee3d5768370..59a1d2fc50a 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -104,14 +104,24 @@ Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, const Relay_log_info
       RPL_TABLE_LIST *ptr= rli->tables_to_lock;
       for ( ; ptr ; ptr= static_cast<RPL_TABLE_LIST*>(ptr->next_global))
       {
-        if (ptr->m_tabledef.compatible_with(rli, ptr->table))
+        TABLE *conv_table;
+        if (!ptr->m_tabledef.compatible_with(thd, const_cast<Relay_log_info*>(rli),
+                                             ptr->table, &conv_table))
         {
+          DBUG_PRINT("debug", ("Table: %s.%s is not compatible with master",
+                               ptr->table->s->db.str,
+                               ptr->table->s->table_name.str));
           mysql_unlock_tables(thd, thd->lock);
           thd->lock= 0;
           thd->is_slave_error= 1;
           const_cast<Relay_log_info*>(rli)->clear_tables_to_lock();
           DBUG_RETURN(Old_rows_log_event::ERR_BAD_TABLE_DEF);
         }
+        DBUG_PRINT("debug", ("Table: %s.%s is compatible with master"
+                             " - conv_table: %p",
+                             ptr->table->s->db.str,
+                             ptr->table->s->table_name.str, conv_table));
+        ptr->m_conv_table= conv_table;
       }
     }
 
@@ -1547,7 +1557,9 @@ int Old_rows_log_event::do_apply_event(Relay_log_info const *rli)
       RPL_TABLE_LIST *ptr= rli->tables_to_lock;
       for ( ; ptr ; ptr= static_cast<RPL_TABLE_LIST*>(ptr->next_global))
       {
-        if (ptr->m_tabledef.compatible_with(rli, ptr->table))
+        TABLE *conv_table;
+        if (ptr->m_tabledef.compatible_with(thd, const_cast<Relay_log_info*>(rli),
+                                            ptr->table, &conv_table))
         {
           mysql_unlock_tables(thd, thd->lock);
           thd->lock= 0;
@@ -1555,12 +1567,14 @@ int Old_rows_log_event::do_apply_event(Relay_log_info const *rli)
           const_cast<Relay_log_info*>(rli)->clear_tables_to_lock();
           DBUG_RETURN(ERR_BAD_TABLE_DEF);
         }
+        ptr->m_conv_table= conv_table;
       }
     }
 
     /*
-      ... and then we add all the tables to the table map and remove
-      them from tables to lock.
+      ... and then we add all the tables to the table map but keep
+      them in the tables to lock list.
+
 
       We also invalidate the query cache for all the tables, since
       they will now be changed.
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 98c71f53d99..793fb3da4be 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1926,6 +1926,7 @@ extern my_bool opt_sql_bin_update, opt_safe_user_create, opt_no_mix_types;
 extern my_bool opt_safe_show_db, opt_local_infile, opt_myisam_use_mmap;
 extern my_bool opt_slave_compressed_protocol, use_temp_pool;
 extern ulong slave_exec_mode_options;
+extern ulong slave_type_conversions_options;
 extern my_bool opt_readonly, lower_case_file_system;
 extern my_bool opt_enable_named_pipe, opt_sync_frm, opt_allow_suspicious_udfs;
 extern my_bool opt_secure_auth;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 8468e7cf79a..ce471f42aaa 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -527,6 +527,8 @@ ulong open_files_limit, max_binlog_size, max_relay_log_size;
 ulong slave_net_timeout, slave_trans_retries;
 ulong slave_exec_mode_options;
 const char *slave_exec_mode_str= "STRICT";
+ulong slave_type_conversions_options;
+const char *slave_type_conversions_default= "";
 ulong thread_cache_size=0, thread_pool_size= 0;
 ulong binlog_cache_size=0;
 ulonglong  max_binlog_cache_size=0;
@@ -3899,6 +3901,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);
@@ -5205,12 +5228,16 @@ pthread_handler_t handle_connections_sockets(void *arg __attribute__((unused)))
 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)
   {
@@ -5233,7 +5260,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,
@@ -5253,7 +5281,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,
@@ -5696,6 +5725,7 @@ enum options_mysqld
 #endif /* defined(ENABLED_DEBUG_SYNC) */
   OPT_OLD_MODE,
   OPT_SLAVE_EXEC_MODE,
+  OPT_SLAVE_TYPE_CONVERSIONS,
   OPT_GENERAL_LOG_FILE,
   OPT_SLOW_QUERY_LOG_FILE,
   OPT_IGNORE_BUILTIN_INNODB,
@@ -6377,6 +6407,15 @@ replicating a LOAD DATA INFILE command.",
   {"slave-exec-mode", OPT_SLAVE_EXEC_MODE,
    "Modes for how replication events should be executed.  Legal values are STRICT (default) and IDEMPOTENT. In IDEMPOTENT mode, replication will not stop for operations that are idempotent. In STRICT mode, replication will stop on any unexpected difference between the master and the slave.",
    (uchar**) &slave_exec_mode_str, (uchar**) &slave_exec_mode_str, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  {"slave-type-conversions", OPT_SLAVE_TYPE_CONVERSIONS,
+   "Set of slave type conversions that are enabled. Legal values are:"
+   " ALL_LOSSY to enable lossy conversions and"
+   " ALL_NON_LOSSY to enable non-lossy conversions."
+   " If the variable is assigned the empty set, no conversions are"
+   " allowed and it is expected that the types match exactly.",
+   (uchar**) &slave_type_conversions_default,
+   (uchar**) &slave_type_conversions_default,
+   0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
 #endif
   {"slow-query-log", OPT_SLOW_LOG,
    "Enable|disable slow query log", (uchar**) &opt_slow_log,
@@ -7694,6 +7733,11 @@ static int mysql_init_variables(void)
   slave_exec_mode_options= (uint)
     find_bit_type_or_exit(slave_exec_mode_str, &slave_exec_mode_typelib, NULL,
                           &error);
+  /* Slave type conversions */
+  slave_type_conversions_options= 0;
+  slave_type_conversions_options=
+    find_bit_type_or_exit(slave_type_conversions_default, &slave_type_conversions_typelib,
+                          NULL, &error);
   if (error)
     return 1;
   opt_specialflag= SPECIAL_ENGLISH;
@@ -7919,6 +7963,12 @@ mysqld_get_one_option(int optid,
     if (error)
       return 1;
     break;
+  case OPT_SLAVE_TYPE_CONVERSIONS:
+    slave_type_conversions_options= (uint)
+      find_bit_type_or_exit(argument, &slave_type_conversions_typelib, "", &error);
+    if (error)
+      return 1;
+    break;
 #endif
   case OPT_SAFEMALLOC_MEM_LIMIT:
 #if !defined(DBUG_OFF) && defined(SAFEMALLOC)
@@ -8639,14 +8689,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(language,language,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);
@@ -8654,6 +8698,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 5f9bae22c70..94204962345 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -446,9 +446,9 @@ 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->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);
     return res;
@@ -462,9 +462,9 @@ 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->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)
+	!(*range_key_flag & (NO_MAX_RANGE | NEAR_MAX)))
       res+= key_tree->next_key_part->store_max_key(key, range_key,
                                                    range_key_flag);
     return res;
@@ -1700,6 +1700,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
   {
@@ -6671,6 +6672,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
@@ -6699,6 +6701,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);
@@ -6733,25 +6736,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)
         {
@@ -7261,27 +7291,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|-----------------+
@@ -7301,8 +7329,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)
@@ -7557,8 +7585,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)) &&
@@ -7839,8 +7867,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 &&
@@ -9822,7 +9850,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 544063bf337..d73b59b8af3 100644
--- a/sql/repl_failsafe.cc
+++ b/sql/repl_failsafe.cc
@@ -565,7 +565,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_record.cc b/sql/rpl_record.cc
index 8e80620dd2c..7bf7b81aa3a 100644
--- a/sql/rpl_record.cc
+++ b/sql/rpl_record.cc
@@ -104,10 +104,10 @@ pack_row(TABLE *table, MY_BITMAP const* cols,
 #endif
         pack_ptr= field->pack(pack_ptr, field->ptr + offset,
                               field->max_data_length(), TRUE);
-        DBUG_PRINT("debug", ("field: %s; pack_ptr: 0x%lx;"
+        DBUG_PRINT("debug", ("field: %s; real_type: %d, pack_ptr: 0x%lx;"
                              " pack_ptr':0x%lx; bytes: %d",
-                             field->field_name, (ulong) old_pack_ptr,
-                             (ulong) pack_ptr,
+                             field->field_name, field->real_type(),
+                             (ulong) old_pack_ptr, (ulong) pack_ptr,
                              (int) (pack_ptr - old_pack_ptr)));
       }
 
@@ -150,13 +150,20 @@ pack_row(TABLE *table, MY_BITMAP const* cols,
    the various member functions of Field and subclasses expect to
    write.
 
-   The row is assumed to only consist of the fields for which the corresponding
-   bit in bitset @c cols is set; the other parts of the record are left alone.
+   The row is assumed to only consist of the fields for which the
+   corresponding bit in bitset @c cols is set; the other parts of the
+   record are left alone.
 
    At most @c colcnt columns are read: if the table is larger than
    that, the remaining fields are not filled in.
 
-   @param rli     Relay log info
+   @note The relay log information can be NULL, which means that no
+   checking or comparison with the source table is done, simply
+   because it is not used.  This feature is used by MySQL Backup to
+   unpack a row from from the backup image, but can be used for other
+   purposes as well.
+
+   @param rli     Relay log info, which can be NULL
    @param table   Table to unpack into
    @param colcnt  Number of columns to read from record
    @param row_data
@@ -170,10 +177,8 @@ pack_row(TABLE *table, MY_BITMAP const* cols,
 
    @retval 0 No error
 
-   @retval ER_NO_DEFAULT_FOR_FIELD
-   Returned if one of the fields existing on the slave but not on the
-   master does not have a default value (and isn't nullable)
-
+   @retval HA_ERR_GENERIC
+   A generic, internal, error caused the unpacking to fail.
  */
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
 int
@@ -185,6 +190,7 @@ unpack_row(Relay_log_info const *rli,
 {
   DBUG_ENTER("unpack_row");
   DBUG_ASSERT(row_data);
+  DBUG_ASSERT(table);
   size_t const master_null_byte_count= (bitmap_bits_set(cols) + 7) / 8;
   int error= 0;
 
@@ -202,10 +208,37 @@ unpack_row(Relay_log_info const *rli,
   // The "current" null bits
   unsigned int null_bits= *null_ptr++;
   uint i= 0;
-  table_def *tabledef= ((Relay_log_info*)rli)->get_tabledef(table);
+  table_def *tabledef= NULL;
+  TABLE *conv_table= NULL;
+  bool table_found= rli && rli->get_table_data(table, &tabledef, &conv_table);
+  DBUG_PRINT("debug", ("Table data: table_found: %d, tabldef: %p, conv_table: %p",
+                       table_found, tabledef, conv_table));
+  DBUG_ASSERT(table_found);
+
+  /*
+    If rli is NULL it means that there is no source table and that the
+    row shall just be unpacked without doing any checks. This feature
+    is used by MySQL Backup, but can be used for other purposes as
+    well.
+   */
+  if (rli && !table_found)
+    DBUG_RETURN(HA_ERR_GENERIC);
+
   for (field_ptr= begin_ptr ; field_ptr < end_ptr && *field_ptr ; ++field_ptr)
   {
-    Field *const f= *field_ptr;
+    /*
+      If there is a conversion table, we pick up the field pointer to
+      the conversion table.  If the conversion table or the field
+      pointer is NULL, no conversions are necessary.
+     */
+    Field *conv_field=
+      conv_table ? conv_table->field[field_ptr - begin_ptr] : NULL;
+    Field *const f=
+      conv_field ? conv_field : *field_ptr;
+    DBUG_PRINT("debug", ("Conversion %srequired for field '%s' (#%ld)",
+                         conv_field ? "" : "not ",
+                         (*field_ptr)->field_name, field_ptr - begin_ptr));
+    DBUG_ASSERT(f != NULL);
 
     /*
       No need to bother about columns that does not exist: they have
@@ -275,6 +308,39 @@ unpack_row(Relay_log_info const *rli,
                              (int) (pack_ptr - old_pack_ptr)));
       }
 
+      /*
+        If conv_field is set, then we are doing a conversion. In this
+        case, we have unpacked the master data to the conversion
+        table, so we need to copy the value stored in the conversion
+        table into the final table and do the conversion at the same time.
+      */
+      if (conv_field)
+      {
+        Copy_field copy;
+#ifndef DBUG_OFF
+        char source_buf[MAX_FIELD_WIDTH];
+        char value_buf[MAX_FIELD_WIDTH];
+        String source_type(source_buf, sizeof(source_buf), system_charset_info);
+        String value_string(value_buf, sizeof(value_buf), system_charset_info);
+        conv_field->sql_type(source_type);
+        conv_field->val_str(&value_string);
+        DBUG_PRINT("debug", ("Copying field '%s' of type '%s' with value '%s'",
+                             (*field_ptr)->field_name,
+                             source_type.c_ptr_safe(), value_string.c_ptr_safe()));
+#endif
+        copy.set(*field_ptr, f, TRUE);
+        (*copy.do_copy)(&copy);
+#ifndef DBUG_OFF
+        char target_buf[MAX_FIELD_WIDTH];
+        String target_type(target_buf, sizeof(target_buf), system_charset_info);
+        (*field_ptr)->sql_type(target_type);
+        (*field_ptr)->val_str(&value_string);
+        DBUG_PRINT("debug", ("Value of field '%s' of type '%s' is now '%s'",
+                             (*field_ptr)->field_name,
+                             target_type.c_ptr_safe(), value_string.c_ptr_safe()));
+#endif
+      }
+
       null_mask <<= 1;
     }
     i++;
@@ -366,7 +432,6 @@ int prepare_record(TABLE *const table,
   */
   for (Field **field_ptr= table->field+skip; *field_ptr; ++field_ptr)
   {
-    uint32 const mask= NOT_NULL_FLAG | NO_DEFAULT_VALUE_FLAG;
     Field *const f= *field_ptr;
     if ((f->flags &  NO_DEFAULT_VALUE_FLAG) &&
         (f->real_type() != MYSQL_TYPE_ENUM))
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 14b560600d8..e35ca613d23 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_rli.h b/sql/rpl_rli.h
index 1dc7f3ef0d2..7c43f821295 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -333,13 +333,21 @@ public:
   uint tables_to_lock_count;        /* RBR: Count of tables to lock */
   table_mapping m_table_map;      /* RBR: Mapping table-id to table */
 
-  inline table_def *get_tabledef(TABLE *tbl)
+  bool get_table_data(TABLE *table_arg, table_def **tabledef_var, TABLE **conv_table_var) const
   {
-    table_def *td= 0;
-    for (TABLE_LIST *ptr= tables_to_lock; ptr && !td; ptr= ptr->next_global)
-      if (ptr->table == tbl)
-        td= &((RPL_TABLE_LIST *)ptr)->m_tabledef;
-    return (td);
+    DBUG_ASSERT(tabledef_var && conv_table_var);
+    for (TABLE_LIST *ptr= tables_to_lock ; ptr != NULL ; ptr= ptr->next_global)
+      if (ptr->table == table_arg)
+      {
+        *tabledef_var= &static_cast<RPL_TABLE_LIST*>(ptr)->m_tabledef;
+        *conv_table_var= static_cast<RPL_TABLE_LIST*>(ptr)->m_conv_table;
+        DBUG_PRINT("debug", ("Fetching table data for table %s.%s:"
+                             " tabledef: %p, conv_table: %p",
+                             table_arg->s->db.str, table_arg->s->table_name.str,
+                             *tabledef_var, *conv_table_var));
+        return true;
+      }
+    return false;
   }
 
   /*
diff --git a/sql/rpl_tblmap.cc b/sql/rpl_tblmap.cc
index a004c354263..6ef9a8623fe 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/rpl_utility.cc b/sql/rpl_utility.cc
index e34f8561051..feb35527b62 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -14,8 +14,176 @@
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */
 
 #include "rpl_utility.h"
+
+#ifndef MYSQL_CLIENT
 #include "rpl_rli.h"
 
+/**
+   Function to compare two size_t integers for their relative
+   order. Used below.
+ */
+int compare(size_t a, size_t b)
+{
+  if (a < b)
+    return -1;
+  if (b < a)
+    return 1;
+  return 0;
+}
+
+
+/**
+   Max value for an unsigned integer of 'bits' bits.
+
+   The somewhat contorted expression is to avoid overflow.
+ */
+uint32 uint_max(int bits) {
+  return (((1UL << (bits - 1)) - 1) << 1) | 1;
+}
+
+
+/**
+   Compute the maximum display length of a field.
+
+   @param sql_type Type of the field
+   @param metadata The metadata from the master for the field.
+   @return Maximum length of the field in bytes.
+ */
+static uint32
+max_display_length_for_field(enum_field_types sql_type, unsigned int metadata)
+{
+  DBUG_PRINT("debug", ("sql_type: %d, metadata: 0x%x", sql_type, metadata));
+  DBUG_ASSERT(metadata >> 16 == 0);
+
+  switch (sql_type) {
+  case MYSQL_TYPE_NEWDECIMAL:
+    return metadata >> 8;
+
+  case MYSQL_TYPE_FLOAT:
+    return 12;
+
+  case MYSQL_TYPE_DOUBLE:
+    return 22;
+
+  case MYSQL_TYPE_SET:
+  case MYSQL_TYPE_ENUM:
+      return metadata & 0x00ff;
+
+  case MYSQL_TYPE_STRING:
+  {
+    uchar type= metadata >> 8;
+    if (type == MYSQL_TYPE_SET || type == MYSQL_TYPE_ENUM)
+      return metadata & 0xff;
+    else
+      /* This is taken from Field_string::unpack. */
+      return (((metadata >> 4) & 0x300) ^ 0x300) + (metadata & 0x00ff);
+  }
+
+  case MYSQL_TYPE_YEAR:
+  case MYSQL_TYPE_TINY:
+    return 4;
+
+  case MYSQL_TYPE_SHORT:
+    return 6;
+
+  case MYSQL_TYPE_INT24:
+    return 9;
+
+  case MYSQL_TYPE_LONG:
+    return 11;
+
+#ifdef HAVE_LONG_LONG
+  case MYSQL_TYPE_LONGLONG:
+    return 20;
+
+#endif
+  case MYSQL_TYPE_NULL:
+    return 0;
+
+  case MYSQL_TYPE_NEWDATE:
+    return 3;
+
+  case MYSQL_TYPE_DATE:
+  case MYSQL_TYPE_TIME:
+    return 3;
+
+  case MYSQL_TYPE_TIMESTAMP:
+    return 4;
+
+  case MYSQL_TYPE_DATETIME:
+    return 8;
+
+  case MYSQL_TYPE_BIT:
+    /*
+      Decode the size of the bit field from the master.
+    */
+    DBUG_ASSERT((metadata & 0xff) <= 7);
+    return 8 * (metadata >> 8U) + (metadata & 0x00ff);
+
+  case MYSQL_TYPE_VAR_STRING:
+  case MYSQL_TYPE_VARCHAR:
+    return metadata;
+
+    /*
+      The actual length for these types does not really matter since
+      they are used to calc_pack_length, which ignores the given
+      length for these types.
+
+      Since we want this to be accurate for other uses, we return the
+      maximum size in bytes of these BLOBs.
+    */
+
+  case MYSQL_TYPE_TINY_BLOB:
+    return uint_max(1 * 8);
+
+  case MYSQL_TYPE_MEDIUM_BLOB:
+    return uint_max(3 * 8);
+
+  case MYSQL_TYPE_BLOB:
+    /*
+      For the blob type, Field::real_type() lies and say that all
+      blobs are of type MYSQL_TYPE_BLOB. In that case, we have to look
+      at the length instead to decide what the max display size is.
+     */
+    return uint_max(metadata * 8);
+
+  case MYSQL_TYPE_LONG_BLOB:
+  case MYSQL_TYPE_GEOMETRY:
+    return uint_max(4 * 8);
+
+  default:
+    return ~(uint32) 0;
+  }
+}
+
+
+/*
+  Compare the pack lengths of a source field (on the master) and a
+  target field (on the slave).
+
+  @param field    Target field.
+  @param type     Source field type.
+  @param metadata Source field metadata.
+
+  @retval -1 The length of the source field is smaller than the target field.
+  @retval  0 The length of the source and target fields are the same.
+  @retval  1 The length of the source field is greater than the target field.
+ */
+int compare_lengths(Field *field, enum_field_types source_type, uint16 metadata)
+{
+  DBUG_ENTER("compare_lengths");
+  size_t const source_length=
+    max_display_length_for_field(source_type, metadata);
+  size_t const target_length= field->max_display_length();
+  DBUG_PRINT("debug", ("source_length: %lu, source_type: %u,"
+                       " target_length: %lu, target_type: %u",
+                       (unsigned long) source_length, source_type,
+                       (unsigned long) target_length, field->real_type()));
+  int result= compare(source_length, target_length);
+  DBUG_PRINT("result", ("%d", result));
+  DBUG_RETURN(result);
+}
+
 /*********************************************************************
  *                   table_def member definitions                    *
  *********************************************************************/
@@ -169,58 +337,718 @@ uint32 table_def::calc_field_size(uint col, uchar *master_data) const
   return length;
 }
 
-/*
+
+/**
+ */
+void show_sql_type(enum_field_types type, uint16 metadata, String *str)
+{
+  DBUG_ENTER("show_sql_type");
+  DBUG_PRINT("enter", ("type: %d, metadata: 0x%x", type, metadata));
+
+  switch (type)
+  {
+  case MYSQL_TYPE_TINY:
+    str->set_ascii(STRING_WITH_LEN("tinyint"));
+    break;
+
+  case MYSQL_TYPE_SHORT:
+    str->set_ascii(STRING_WITH_LEN("smallint"));
+    break;
+
+  case MYSQL_TYPE_LONG:
+    str->set_ascii(STRING_WITH_LEN("int"));
+    break;
+
+  case MYSQL_TYPE_FLOAT:
+    str->set_ascii(STRING_WITH_LEN("float"));
+    break;
+
+  case MYSQL_TYPE_DOUBLE:
+    str->set_ascii(STRING_WITH_LEN("double"));
+    break;
+
+  case MYSQL_TYPE_NULL:
+    str->set_ascii(STRING_WITH_LEN("null"));
+    break;
+
+  case MYSQL_TYPE_TIMESTAMP:
+    str->set_ascii(STRING_WITH_LEN("timestamp"));
+    break;
+
+  case MYSQL_TYPE_LONGLONG:
+    str->set_ascii(STRING_WITH_LEN("bigint"));
+    break;
+
+  case MYSQL_TYPE_INT24:
+    str->set_ascii(STRING_WITH_LEN("mediumint"));
+    break;
+
+  case MYSQL_TYPE_NEWDATE:
+  case MYSQL_TYPE_DATE:
+    str->set_ascii(STRING_WITH_LEN("date"));
+    break;
+
+  case MYSQL_TYPE_TIME:
+    str->set_ascii(STRING_WITH_LEN("time"));
+    break;
+
+  case MYSQL_TYPE_DATETIME:
+    str->set_ascii(STRING_WITH_LEN("datetime"));
+    break;
+
+  case MYSQL_TYPE_YEAR:
+    str->set_ascii(STRING_WITH_LEN("year"));
+    break;
+
+  case MYSQL_TYPE_VAR_STRING:
+  case MYSQL_TYPE_VARCHAR:
+    {
+      CHARSET_INFO *cs= str->charset();
+      uint32 length=
+        cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
+                           "varchar(%u)", metadata);
+      str->length(length);
+    }
+    break;
+
+  case MYSQL_TYPE_BIT:
+    {
+      CHARSET_INFO *cs= str->charset();
+      int bit_length= 8 * (metadata >> 8) + (metadata & 0xFF);
+      uint32 length=
+        cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
+                           "bit(%d)", bit_length);
+      str->length(length);
+    }
+    break;
+
+  case MYSQL_TYPE_DECIMAL:
+    {
+      CHARSET_INFO *cs= str->charset();
+      uint32 length=
+        cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
+                           "decimal(%d,?)", metadata);
+      str->length(length);
+    }
+    break;
+
+  case MYSQL_TYPE_NEWDECIMAL:
+    {
+      CHARSET_INFO *cs= str->charset();
+      uint32 length=
+        cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
+                           "decimal(%d,%d)", metadata >> 8, metadata & 0xff);
+      str->length(length);
+    }
+    break;
+
+  case MYSQL_TYPE_ENUM:
+    str->set_ascii(STRING_WITH_LEN("enum"));
+    break;
+
+  case MYSQL_TYPE_SET:
+    str->set_ascii(STRING_WITH_LEN("set"));
+    break;
+
+  case MYSQL_TYPE_BLOB:
+    /*
+      Field::real_type() lies regarding the actual type of a BLOB, so
+      it is necessary to check the pack length to figure out what kind
+      of blob it really is.
+     */
+    switch (get_blob_type_from_length(metadata))
+    {
+    case MYSQL_TYPE_TINY_BLOB:
+      str->set_ascii(STRING_WITH_LEN("tinyblob"));
+      break;
+
+    case MYSQL_TYPE_MEDIUM_BLOB:
+      str->set_ascii(STRING_WITH_LEN("mediumblob"));
+      break;
+
+    case MYSQL_TYPE_LONG_BLOB:
+      str->set_ascii(STRING_WITH_LEN("longblob"));
+      break;
+
+    case MYSQL_TYPE_BLOB:
+      str->set_ascii(STRING_WITH_LEN("blob"));
+      break;
+
+    default:
+      DBUG_ASSERT(0);
+      break;
+    }
+    break;
+
+  case MYSQL_TYPE_STRING:
+    {
+      /*
+        This is taken from Field_string::unpack.
+      */
+      CHARSET_INFO *cs= str->charset();
+      uint bytes= (((metadata >> 4) & 0x300) ^ 0x300) + (metadata & 0x00ff);
+      uint32 length=
+        cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
+                           "char(%d)", bytes / cs->mbmaxlen);
+      str->length(length);
+    }
+    break;
+
+  case MYSQL_TYPE_GEOMETRY:
+    str->set_ascii(STRING_WITH_LEN("geometry"));
+    break;
+
+  default:
+    str->set_ascii(STRING_WITH_LEN("<unknown type>"));
+  }
+  DBUG_VOID_RETURN;
+}
+
+
+/**
+   Check the order variable and print errors if the order is not
+   acceptable according to the current settings.
+
+   @param order  The computed order of the conversion needed.
+   @param rli    The relay log info data structure: for error reporting.
+ */
+bool is_conversion_ok(int order, Relay_log_info *rli)
+{
+  DBUG_ENTER("is_conversion_ok");
+  bool allow_non_lossy=
+    bit_is_set(slave_type_conversions_options, SLAVE_TYPE_CONVERSIONS_ALL_NON_LOSSY);
+  bool allow_lossy=
+    bit_is_set(slave_type_conversions_options, SLAVE_TYPE_CONVERSIONS_ALL_LOSSY);
+
+  DBUG_PRINT("enter", ("order: %d, flags:%s%s", order,
+                       allow_non_lossy ? " ALL_NON_LOSSY" : "",
+                       allow_lossy ? " ALL_LOSSY" : ""));
+  if (order < 0 && !allow_non_lossy)
+  {
+    /* !!! Add error message saying that non-lossy conversions need to be allowed. */
+    DBUG_RETURN(false);
+  }
+
+  if (order > 0 && !allow_lossy)
+  {
+    /* !!! Add error message saying that lossy conversions need to be allowed. */
+    DBUG_RETURN(false);
+  }
+
+  DBUG_RETURN(true);
+}
+
+
+/**
+   Can a type potentially be converted to another type?
+
+   This function check if the types are convertible and what
+   conversion is required.
+
+   If conversion is not possible, and error is printed.
+
+   If conversion is possible:
+
+   - *order will be set to -1 if source type is smaller than target
+     type and a non-lossy conversion can be required. This includes
+     the case where the field types are different but types could
+     actually be converted in either direction.
+
+   - *order will be set to 0 if no conversion is required.
+
+   - *order will be set to 1 if the source type is strictly larger
+      than the target type and that conversion is potentially lossy.
+
+   @param[in] field    Target field
+   @param[in] type     Source field type
+   @param[in] metadata Source field metadata
+   @param[in] rli      Relay log info (for error reporting)
+   @param[in] mflags   Flags from the table map event
+   @param[out] order   Order between source field and target field
+
+   @return @c true if conversion is possible according to the current
+   settings, @c false if conversion is not possible according to the
+   current setting.
+ */
+static bool
+can_convert_field_to(Field *field,
+                     enum_field_types source_type, uint16 metadata,
+                     Relay_log_info *rli, uint16 mflags,
+                     int *order_var)
+{
+  DBUG_ENTER("can_convert_field_to");
+#ifndef DBUG_OFF
+  char field_type_buf[MAX_FIELD_WIDTH];
+  String field_type(field_type_buf, sizeof(field_type_buf), field->charset());
+  field->sql_type(field_type);
+  DBUG_PRINT("enter", ("field_type: %s, target_type: %d, source_type: %d, source_metadata: 0x%x",
+                       field_type.c_ptr_safe(), field->real_type(), source_type, metadata));
+#endif
+  /*
+    If the real type is the same, we need to check the metadata to
+    decide if conversions are allowed.
+   */
+  if (field->real_type() == source_type)
+  {
+    if (metadata == 0) // Metadata can only be zero if no metadata was provided
+    {
+      /*
+        If there is no metadata, we either have an old event where no
+        metadata were supplied, or a type that does not require any
+        metadata. In either case, conversion can be done but no
+        conversion table is necessary.
+       */
+      DBUG_PRINT("debug", ("Base types are identical, but there is no metadata"));
+      *order_var= 0;
+      DBUG_RETURN(true);
+    }
+
+    DBUG_PRINT("debug", ("Base types are identical, doing field size comparison"));
+    if (field->compatible_field_size(metadata, rli, mflags, order_var))
+      DBUG_RETURN(is_conversion_ok(*order_var, rli));
+    else
+      DBUG_RETURN(false);
+  }
+  else if (!slave_type_conversions_options)
+    DBUG_RETURN(false);
+
+  /*
+    Here, from and to will always be different. Since the types are
+    different, we cannot use the compatible_field_size() function, but
+    have to rely on hard-coded max-sizes for fields.
+  */
+
+  DBUG_PRINT("debug", ("Base types are different, checking conversion"));
+  switch (source_type)                      // Source type (on master)
+  {
+  case MYSQL_TYPE_DECIMAL:
+  case MYSQL_TYPE_NEWDECIMAL:
+  case MYSQL_TYPE_FLOAT:
+  case MYSQL_TYPE_DOUBLE:
+    switch (field->real_type())
+    {
+    case MYSQL_TYPE_NEWDECIMAL:
+      /*
+        Then the other type is either FLOAT, DOUBLE, or old style
+        DECIMAL, so we require lossy conversion.
+      */
+      *order_var= 1;
+      DBUG_RETURN(is_conversion_ok(*order_var, rli));
+      
+    case MYSQL_TYPE_DECIMAL:
+    case MYSQL_TYPE_FLOAT:
+    case MYSQL_TYPE_DOUBLE:
+    {
+      if (source_type == MYSQL_TYPE_NEWDECIMAL ||
+          source_type == MYSQL_TYPE_DECIMAL)
+        *order_var = 1;                         // Always require lossy conversions
+      else
+        *order_var= compare_lengths(field, source_type, metadata);
+      DBUG_ASSERT(*order_var != 0);
+      DBUG_RETURN(is_conversion_ok(*order_var, rli));
+    }
+
+    default:
+      DBUG_RETURN(false);
+    }
+    break;
+
+  /*
+    The length comparison check will do the correct job of comparing
+    the field lengths (in bytes) of two integer types.
+  */
+  case MYSQL_TYPE_TINY:
+  case MYSQL_TYPE_SHORT:
+  case MYSQL_TYPE_INT24:
+  case MYSQL_TYPE_LONG:
+  case MYSQL_TYPE_LONGLONG:
+    switch (field->real_type())
+    {
+    case MYSQL_TYPE_TINY:
+    case MYSQL_TYPE_SHORT:
+    case MYSQL_TYPE_INT24:
+    case MYSQL_TYPE_LONG:
+    case MYSQL_TYPE_LONGLONG:
+      *order_var= compare_lengths(field, source_type, metadata);
+      DBUG_ASSERT(*order_var != 0);
+      DBUG_RETURN(is_conversion_ok(*order_var, rli));
+
+    default:
+      DBUG_RETURN(false);
+    }
+    break;
+
+  /*
+    Since source and target type is different, and it is not possible
+    to convert bit types to anything else, this will return false.
+   */
+  case MYSQL_TYPE_BIT:
+    DBUG_RETURN(false);
+
+  /*
+    If all conversions are disabled, it is not allowed to convert
+    between these types. Since the TEXT vs. BINARY is distinguished by
+    the charset, and the charset is not replicated, we cannot
+    currently distinguish between , e.g., TEXT and BLOB.
+   */
+  case MYSQL_TYPE_TINY_BLOB:
+  case MYSQL_TYPE_MEDIUM_BLOB:
+  case MYSQL_TYPE_LONG_BLOB:
+  case MYSQL_TYPE_BLOB:
+  case MYSQL_TYPE_STRING:
+  case MYSQL_TYPE_VAR_STRING:
+  case MYSQL_TYPE_VARCHAR:
+    switch (field->real_type())
+    {
+    case MYSQL_TYPE_TINY_BLOB:
+    case MYSQL_TYPE_MEDIUM_BLOB:
+    case MYSQL_TYPE_LONG_BLOB:
+    case MYSQL_TYPE_BLOB:
+    case MYSQL_TYPE_STRING:
+    case MYSQL_TYPE_VAR_STRING:
+    case MYSQL_TYPE_VARCHAR:
+      *order_var= compare_lengths(field, source_type, metadata);
+      /*
+        Here we know that the types are different, so if the order
+        gives that they do not require any conversion, we still need
+        to have non-lossy conversion enabled to allow conversion
+        between different (string) types of the same length.
+       */
+      if (*order_var == 0)
+        *order_var= -1;
+      DBUG_RETURN(is_conversion_ok(*order_var, rli));
+
+    default:
+      DBUG_RETURN(false);
+    }
+    break;
+
+  case MYSQL_TYPE_GEOMETRY:
+  case MYSQL_TYPE_TIMESTAMP:
+  case MYSQL_TYPE_DATE:
+  case MYSQL_TYPE_TIME:
+  case MYSQL_TYPE_DATETIME:
+  case MYSQL_TYPE_YEAR:
+  case MYSQL_TYPE_NEWDATE:
+  case MYSQL_TYPE_NULL:
+  case MYSQL_TYPE_ENUM:
+  case MYSQL_TYPE_SET:
+    DBUG_RETURN(false);
+  }
+  DBUG_RETURN(false);                                 // To keep GCC happy
+}
+
+
+/**
   Is the definition compatible with a table?
 
+  This function will compare the master table with an existing table
+  on the slave and see if they are compatible with respect to the
+  current settings of @c SLAVE_TYPE_CONVERSIONS.
+
+  If the tables are compatible and conversions are required, @c
+  *tmp_table_var will be set to a virtual temporary table with field
+  pointers for the fields that require conversions.  This allow simple
+  checking of whether a conversion are to be applied or not.
+
+  If tables are compatible, but no conversions are necessary, @c
+  *tmp_table_var will be set to NULL.
+
+  @param rli_arg[in]
+  Relay log info, for error reporting.
+
+  @param table[in]
+  Table to compare with
+
+  @param tmp_table_var[out]
+  Virtual temporary table for performing conversions, if necessary.
+
+  @retval true Master table is compatible with slave table.
+  @retval false Master table is not compatible with slave table.
 */
-int
-table_def::compatible_with(Relay_log_info const *rli_arg, TABLE *table)
+bool
+table_def::compatible_with(THD *thd, Relay_log_info *rli,
+                           TABLE *table, TABLE **conv_table_var)
   const
 {
   /*
     We only check the initial columns for the tables.
   */
   uint const cols_to_check= min(table->s->fields, size());
-  int error= 0;
-  Relay_log_info const *rli= const_cast<Relay_log_info*>(rli_arg);
-
-  TABLE_SHARE const *const tsh= table->s;
+  TABLE *tmp_table= NULL;
 
   for (uint col= 0 ; col < cols_to_check ; ++col)
   {
     Field *const field= table->field[col];
-    if (field->type() != type(col))
-    {
-      DBUG_ASSERT(col < size() && col < tsh->fields);
-      DBUG_ASSERT(tsh->db.str && tsh->table_name.str);
-      error= 1;
-      char buf[256];
-      my_snprintf(buf, sizeof(buf), "Column %d type mismatch - "
-                  "received type %d, %s.%s has type %d",
-                  col, type(col), tsh->db.str, tsh->table_name.str,
-                  field->type());
-      rli->report(ERROR_LEVEL, ER_BINLOG_ROW_WRONG_TABLE_DEF,
-                  ER(ER_BINLOG_ROW_WRONG_TABLE_DEF), buf);
+    int order;
+    if (can_convert_field_to(field, type(col), field_metadata(col), rli, m_flags, &order))
+    {
+      DBUG_PRINT("debug", ("Checking column %d -"
+                           " field '%s' can be converted - order: %d",
+                           col, field->field_name, order));
+      DBUG_ASSERT(order >= -1 && order <= 1);
+
+      /*
+        If order is not 0, a conversion is required, so we need to set
+        up the conversion table.
+       */
+      if (order != 0 && tmp_table == NULL)
+      {
+        /*
+          This will create the full table with all fields. This is
+          necessary to ge the correct field lengths for the record.
+        */
+        tmp_table= create_conversion_table(thd, rli, table);
+        if (tmp_table == NULL)
+            return false;
+        /*
+          Clear all fields up to, but not including, this column.
+        */
+        for (unsigned int i= 0; i < col; ++i)
+          tmp_table->field[i]= NULL;
+      }
+
+      if (order == 0 && tmp_table != NULL)
+        tmp_table->field[col]= NULL;
     }
-    /*
-      Check the slave's field size against that of the master.
-    */
-    if (!error &&
-        !field->compatible_field_size(field_metadata(col), rli_arg))
+    else
+    {
+      DBUG_PRINT("debug", ("Checking column %d -"
+                           " field '%s' can not be converted",
+                           col, field->field_name));
+      DBUG_ASSERT(col < size() && col < table->s->fields);
+      DBUG_ASSERT(table->s->db.str && table->s->table_name.str);
+      const char *db_name= table->s->db.str;
+      const char *tbl_name= table->s->table_name.str;
+      char source_buf[MAX_FIELD_WIDTH];
+      char target_buf[MAX_FIELD_WIDTH];
+      String source_type(source_buf, sizeof(source_buf), field->charset());
+      String target_type(target_buf, sizeof(target_buf), field->charset());
+      show_sql_type(type(col), field_metadata(col), &source_type);
+      field->sql_type(target_type);
+      rli->report(ERROR_LEVEL, ER_SLAVE_CONVERSION_FAILED,
+                  ER(ER_SLAVE_CONVERSION_FAILED),
+                  col, db_name, tbl_name,
+                  source_type.c_ptr_safe(), target_type.c_ptr_safe());
+      return false;
+    }
+  }
+
+#ifndef DBUG_OFF
+  if (tmp_table)
+  {
+    for (unsigned int col= 0; col < tmp_table->s->fields; ++col)
+      if (tmp_table->field[col])
+      {
+        char source_buf[MAX_FIELD_WIDTH];
+        char target_buf[MAX_FIELD_WIDTH];
+        String source_type(source_buf, sizeof(source_buf), table->field[col]->charset());
+        String target_type(target_buf, sizeof(target_buf), table->field[col]->charset());
+        tmp_table->field[col]->sql_type(source_type);
+        table->field[col]->sql_type(target_type);
+        DBUG_PRINT("debug", ("Field %s - conversion required."
+                             " Source type: '%s', Target type: '%s'",
+                             tmp_table->field[col]->field_name,
+                             source_type.c_ptr_safe(), target_type.c_ptr_safe()));
+      }
+  }
+#endif
+
+  *conv_table_var= tmp_table;
+  return true;
+}
+
+/**
+  Create a conversion table.
+
+  If the function is unable to create the conversion table, an error
+  will be printed and NULL will be returned.
+
+  @return Pointer to conversion table, or NULL if unable to create
+  conversion table.
+ */
+
+TABLE *table_def::create_conversion_table(THD *thd, Relay_log_info *rli, TABLE *target_table) const
+{
+  DBUG_ENTER("table_def::create_conversion_table");
+
+  List<Create_field> field_list;
+
+  for (uint col= 0 ; col < size() ; ++col)
+  {
+    Create_field *field_def=
+      (Create_field*) alloc_root(thd->mem_root, sizeof(Create_field));
+    if (field_list.push_back(field_def))
+      DBUG_RETURN(NULL);
+
+    uint decimals= 0;
+    TYPELIB* interval= NULL;
+    uint pack_length= 0;
+    uint32 max_length=
+      max_display_length_for_field(type(col), field_metadata(col));
+
+    switch(type(col))
+    {
+      int precision;
+    case MYSQL_TYPE_ENUM:
+    case MYSQL_TYPE_SET:
+      interval= static_cast<Field_enum*>(target_table->field[col])->typelib;
+      pack_length= field_metadata(col) & 0x00ff;
+      break;
+
+    case MYSQL_TYPE_NEWDECIMAL:
+      /*
+        The display length of a DECIMAL type is not the same as the
+        length that should be supplied to make_field, so we correct
+        the length here.
+       */
+      precision= field_metadata(col) >> 8;
+      decimals= field_metadata(col) & 0x00ff;
+      max_length=
+        my_decimal_precision_to_length(precision, decimals, FALSE);
+      break;
+
+    case MYSQL_TYPE_DECIMAL:
+      precision= field_metadata(col);
+      decimals= static_cast<Field_num*>(target_table->field[col])->dec;
+      max_length= field_metadata(col);
+      break;
+
+    case MYSQL_TYPE_TINY_BLOB:
+    case MYSQL_TYPE_MEDIUM_BLOB:
+    case MYSQL_TYPE_LONG_BLOB:
+    case MYSQL_TYPE_BLOB:
+    case MYSQL_TYPE_GEOMETRY:
+      pack_length= field_metadata(col) & 0x00ff;
+      break;
+
+    default:
+      break;
+    }
+
+    DBUG_PRINT("debug", ("sql_type: %d, target_field: '%s', max_length: %d, decimals: %d,"
+                         " maybe_null: %d, unsigned_flag: %d, pack_length: %u",
+                         type(col), target_table->field[col]->field_name,
+                         max_length, decimals, TRUE, FALSE, pack_length));
+    field_def->init_for_tmp_table(type(col),
+                                  max_length,
+                                  decimals,
+                                  TRUE,         // maybe_null
+                                  FALSE,        // unsigned_flag
+                                  pack_length);
+    field_def->charset= target_table->field[col]->charset();
+    field_def->interval= interval;
+  }
+
+  TABLE *conv_table= create_virtual_tmp_table(thd, field_list);
+  if (conv_table == NULL)
+    rli->report(ERROR_LEVEL, ER_SLAVE_CANT_CREATE_CONVERSION,
+                ER(ER_SLAVE_CANT_CREATE_CONVERSION),
+                target_table->s->db.str,
+                target_table->s->table_name.str);
+  DBUG_RETURN(conv_table);
+}
+
+#endif /* MYSQL_CLIENT */
+
+table_def::table_def(unsigned char *types, ulong size,
+                     uchar *field_metadata, int metadata_size,
+                     uchar *null_bitmap, uint16 flags)
+  : m_size(size), m_type(0), m_field_metadata_size(metadata_size),
+    m_field_metadata(0), m_null_bits(0), m_flags(flags),
+    m_memory(NULL)
+{
+  m_memory= (uchar *)my_multi_malloc(MYF(MY_WME),
+                                     &m_type, size,
+                                     &m_field_metadata,
+                                     size * sizeof(uint16),
+                                     &m_null_bits, (size + 7) / 8,
+                                     NULL);
+
+  bzero(m_field_metadata, size * sizeof(uint16));
+
+  if (m_type)
+    memcpy(m_type, types, size);
+  else
+    m_size= 0;
+  /*
+    Extract the data from the table map into the field metadata array
+    iff there is field metadata. The variable metadata_size will be
+    0 if we are replicating from an older version server since no field
+    metadata was written to the table map. This can also happen if 
+    there were no fields in the master that needed extra metadata.
+  */
+  if (m_size && metadata_size)
+  { 
+    int index= 0;
+    for (unsigned int i= 0; i < m_size; i++)
     {
-      error= 1;
-      char buf[256];
-      my_snprintf(buf, sizeof(buf), "Column %d size mismatch - "
-                  "master has size %d, %s.%s on slave has size %d."
-                  " Master's column size should be <= the slave's "
-                  "column size.", col,
-                  field->pack_length_from_metadata(m_field_metadata[col]),
-                  tsh->db.str, tsh->table_name.str,
-                  field->row_pack_length());
-      rli->report(ERROR_LEVEL, ER_BINLOG_ROW_WRONG_TABLE_DEF,
-                  ER(ER_BINLOG_ROW_WRONG_TABLE_DEF), buf);
+      switch (m_type[i]) {
+      case MYSQL_TYPE_TINY_BLOB:
+      case MYSQL_TYPE_BLOB:
+      case MYSQL_TYPE_MEDIUM_BLOB:
+      case MYSQL_TYPE_LONG_BLOB:
+      case MYSQL_TYPE_DOUBLE:
+      case MYSQL_TYPE_FLOAT:
+      case MYSQL_TYPE_GEOMETRY:
+      {
+        /*
+          These types store a single byte.
+        */
+        m_field_metadata[i]= field_metadata[index];
+        index++;
+        break;
+      }
+      case MYSQL_TYPE_SET:
+      case MYSQL_TYPE_ENUM:
+      case MYSQL_TYPE_STRING:
+      {
+        uint16 x= field_metadata[index++] << 8U; // real_type
+        x+= field_metadata[index++];            // pack or field length
+        m_field_metadata[i]= x;
+        break;
+      }
+      case MYSQL_TYPE_BIT:
+      {
+        uint16 x= field_metadata[index++]; 
+        x = x + (field_metadata[index++] << 8U);
+        m_field_metadata[i]= x;
+        break;
+      }
+      case MYSQL_TYPE_VARCHAR:
+      {
+        /*
+          These types store two bytes.
+        */
+        char *ptr= (char *)&field_metadata[index];
+        m_field_metadata[i]= uint2korr(ptr);
+        index= index + 2;
+        break;
+      }
+      case MYSQL_TYPE_NEWDECIMAL:
+      {
+        uint16 x= field_metadata[index++] << 8U; // precision
+        x+= field_metadata[index++];            // decimals
+        m_field_metadata[i]= x;
+        break;
+      }
+      default:
+        m_field_metadata[i]= 0;
+        break;
+      }
     }
   }
+  if (m_size && null_bitmap)
+    memcpy(m_null_bits, null_bitmap, (m_size + 7) / 8);
+}
+
 
-  return error;
+table_def::~table_def()
+{
+  my_free(m_memory, MYF(0));
+#ifndef DBUG_OFF
+  m_type= 0;
+  m_size= 0;
+#endif
 }
+
diff --git a/sql/rpl_utility.h b/sql/rpl_utility.h
index 1f4ca246ff1..4b9bf3be93f 100644
--- a/sql/rpl_utility.h
+++ b/sql/rpl_utility.h
@@ -21,6 +21,7 @@
 #endif
 
 #include "mysql_priv.h"
+#include "mysql_com.h"
 
 class Relay_log_info;
 
@@ -32,127 +33,24 @@ class Relay_log_info;
   - Extract and decode table definition data from the table map event
   - Check if table definition in table map is compatible with table
     definition on slave
-
-  Currently, the only field type data available is an array of the
-  type operators that are present in the table map event.
-
-  @todo Add type operands to this structure to allow detection of
-     difference between, e.g., BIT(5) and BIT(10).
  */
 
 class table_def
 {
 public:
   /**
-    Convenience declaration of the type of the field type data in a
-    table map event.
-  */
-  typedef unsigned char field_type;
-
-  /**
     Constructor.
 
-    @param types Array of types
+    @param types Array of types, each stored as a byte
     @param size  Number of elements in array 'types'
     @param field_metadata Array of extra information about fields
     @param metadata_size Size of the field_metadata array
     @param null_bitmap The bitmap of fields that can be null
    */
-  table_def(field_type *types, ulong size, uchar *field_metadata, 
-      int metadata_size, uchar *null_bitmap)
-    : m_size(size), m_type(0), m_field_metadata_size(metadata_size),
-      m_field_metadata(0), m_null_bits(0), m_memory(NULL)
-  {
-    m_memory= (uchar *)my_multi_malloc(MYF(MY_WME),
-                                       &m_type, size,
-                                       &m_field_metadata,
-                                       size * sizeof(uint16),
-                                       &m_null_bits, (size + 7) / 8,
-                                       NULL);
-
-    bzero(m_field_metadata, size * sizeof(uint16));
-
-    if (m_type)
-      memcpy(m_type, types, size);
-    else
-      m_size= 0;
-    /*
-      Extract the data from the table map into the field metadata array
-      iff there is field metadata. The variable metadata_size will be
-      0 if we are replicating from an older version server since no field
-      metadata was written to the table map. This can also happen if 
-      there were no fields in the master that needed extra metadata.
-    */
-    if (m_size && metadata_size)
-    { 
-      int index= 0;
-      for (unsigned int i= 0; i < m_size; i++)
-      {
-        switch (m_type[i]) {
-        case MYSQL_TYPE_TINY_BLOB:
-        case MYSQL_TYPE_BLOB:
-        case MYSQL_TYPE_MEDIUM_BLOB:
-        case MYSQL_TYPE_LONG_BLOB:
-        case MYSQL_TYPE_DOUBLE:
-        case MYSQL_TYPE_FLOAT:
-        {
-          /*
-            These types store a single byte.
-          */
-          m_field_metadata[i]= field_metadata[index];
-          index++;
-          break;
-        }
-        case MYSQL_TYPE_SET:
-        case MYSQL_TYPE_ENUM:
-        case MYSQL_TYPE_STRING:
-        {
-          uint16 x= field_metadata[index++] << 8U; // real_type
-          x+= field_metadata[index++];            // pack or field length
-          m_field_metadata[i]= x;
-          break;
-        }
-        case MYSQL_TYPE_BIT:
-        {
-          uint16 x= field_metadata[index++]; 
-          x = x + (field_metadata[index++] << 8U);
-          m_field_metadata[i]= x;
-          break;
-        }
-        case MYSQL_TYPE_VARCHAR:
-        {
-          /*
-            These types store two bytes.
-          */
-          char *ptr= (char *)&field_metadata[index];
-          m_field_metadata[i]= uint2korr(ptr);
-          index= index + 2;
-          break;
-        }
-        case MYSQL_TYPE_NEWDECIMAL:
-        {
-          uint16 x= field_metadata[index++] << 8U; // precision
-          x+= field_metadata[index++];            // decimals
-          m_field_metadata[i]= x;
-          break;
-        }
-        default:
-          m_field_metadata[i]= 0;
-          break;
-        }
-      }
-    }
-    if (m_size && null_bitmap)
-       memcpy(m_null_bits, null_bitmap, (m_size + 7) / 8);
-  }
+  table_def(unsigned char *types, ulong size, uchar *field_metadata,
+            int metadata_size, uchar *null_bitmap, uint16 flags);
 
-  ~table_def() {
-    my_free(m_memory, MYF(0));
-#ifndef DBUG_OFF
-    m_type= 0;
-    m_size= 0;
-#endif
-  }
+  ~table_def();
 
   /**
     Return the number of fields there is type data for.
@@ -171,10 +69,40 @@ public:
     <code>index</code>. Currently, only the type identifier is
     returned.
    */
-  field_type type(ulong index) const
+  enum_field_types type(ulong index) const
   {
     DBUG_ASSERT(index < m_size);
-    return m_type[index];
+    /*
+      If the source type is MYSQL_TYPE_STRING, it can in reality be
+      either MYSQL_TYPE_STRING, MYSQL_TYPE_ENUM, or MYSQL_TYPE_SET, so
+      we might need to modify the type to get the real type.
+    */
+    enum_field_types source_type= static_cast<enum_field_types>(m_type[index]);
+    uint16 source_metadata= m_field_metadata[index];
+    switch (source_type)
+    {
+    case MYSQL_TYPE_STRING:
+    {
+      int real_type= source_metadata >> 8;
+      if (real_type == MYSQL_TYPE_ENUM || real_type == MYSQL_TYPE_SET)
+        source_type= static_cast<enum_field_types>(real_type);
+      break;
+    }
+
+    /*
+      This type has not been used since before row-based replication,
+      so we can safely assume that it really is MYSQL_TYPE_NEWDATE.
+    */
+    case MYSQL_TYPE_DATE:
+      source_type= MYSQL_TYPE_NEWDATE;
+      break;
+
+    default:
+      /* Do nothing */
+      break;
+    }
+
+    return source_type;
   }
 
 
@@ -226,26 +154,62 @@ public:
     with it.
 
     A table definition is compatible with a table if:
-      - the columns types of the table definition is a (not
-        necessarily proper) prefix of the column type of the table, or
-      - the other way around
+      - The columns types of the table definition is a (not
+        necessarily proper) prefix of the column type of the table.
+
+      - The other way around.
 
+      - Each column on the master that also exists on the slave can be
+        converted according to the current settings of @c
+        SLAVE_TYPE_CONVERSIONS.
+
+    @param thd
     @param rli   Pointer to relay log info
     @param table Pointer to table to compare with.
 
+    @param[out] tmp_table_var Pointer to temporary table for holding
+    conversion table.
+
     @retval 1  if the table definition is not compatible with @c table
     @retval 0  if the table definition is compatible with @c table
   */
 #ifndef MYSQL_CLIENT
-  int compatible_with(Relay_log_info const *rli, TABLE *table) const;
+  bool compatible_with(THD *thd, Relay_log_info *rli, TABLE *table,
+                      TABLE **conv_table_var) const;
+
+  /**
+   Create a virtual in-memory temporary table structure.
+
+   The table structure has records and field array so that a row can
+   be unpacked into the record for further processing.
+
+   In the virtual table, each field that requires conversion will
+   have a non-NULL value, while fields that do not require
+   conversion will have a NULL value.
+
+   Some information that is missing in the events, such as the
+   character set for string types, are taken from the table that the
+   field is going to be pushed into, so the target table that the data
+   eventually need to be pushed into need to be supplied.
+
+   @param thd Thread to allocate memory from.
+   @param rli Relay log info structure, for error reporting.
+   @param target_table Target table for fields.
+
+   @return A pointer to a temporary table with memory allocated in the
+   thread's memroot, NULL if the table could not be created
+   */
+  TABLE *create_conversion_table(THD *thd, Relay_log_info *rli, TABLE *target_table) const;
 #endif
 
+
 private:
   ulong m_size;           // Number of elements in the types array
-  field_type *m_type;                     // Array of type descriptors
+  unsigned char *m_type;  // Array of type descriptors
   uint m_field_metadata_size;
   uint16 *m_field_metadata;
   uchar *m_null_bits;
+  uint16 m_flags;         // Table flags
   uchar *m_memory;
 };
 
@@ -260,6 +224,7 @@ struct RPL_TABLE_LIST
 {
   bool m_tabledef_valid;
   table_def m_tabledef;
+  TABLE *m_conv_table;
 };
 
 
diff --git a/sql/set_var.cc b/sql/set_var.cc
index b1d1c89cfe6..d9bd14564bf 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -92,6 +92,33 @@ TYPELIB delay_key_write_typelib=
   delay_key_write_type_names, NULL
 };
 
+/**
+  SLAVE_TYPE_CONVERSIONS variable.
+
+  Definition is equivalent to
+  @code
+  SET('ALL_NON_LOSSY', 'ALL_LOSSY')
+  @endcode
+ */
+const char *slave_type_conversions_type_name[]= {
+  "ALL_LOSSY",
+  "ALL_NON_LOSSY",
+  NullS
+};
+
+unsigned int slave_type_conversions_type_length[]= {
+  sizeof("ALL_LOSSY")-1,
+  sizeof("ALL_NON_LOSSY")-1,
+  0
+};
+
+TYPELIB slave_type_conversions_typelib=
+{
+  array_elements(slave_type_conversions_type_name)-1, "",
+  slave_type_conversions_type_name,
+  slave_type_conversions_type_length
+};
+
 const char *slave_exec_mode_names[]=
 { "STRICT", "IDEMPOTENT", NullS };
 static const unsigned int slave_exec_mode_names_len[]=
@@ -577,6 +604,12 @@ static sys_var_set_slave_mode slave_exec_mode(&vars,
                                               &slave_exec_mode_options,
                                               &slave_exec_mode_typelib,
                                               0);
+static sys_var_set slave_type_conversions(&vars,
+                                          "slave_type_conversions",
+                                          &slave_type_conversions_options,
+                                          &slave_type_conversions_typelib,
+                                          0);
+
 static sys_var_long_ptr	sys_slow_launch_time(&vars, "slow_launch_time",
 					     &slow_launch_time);
 static sys_var_thd_ulong	sys_sort_buffer(&vars, "sort_buffer_size",
@@ -1264,7 +1297,6 @@ bool sys_var_thd_binlog_format::check(THD *thd, set_var *var) {
   return result;
 }
 
-
 bool sys_var_thd_binlog_format::is_readonly() const
 {
   /*
diff --git a/sql/set_var.h b/sql/set_var.h
index 9257a6e3081..1bc97916876 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -35,6 +35,7 @@ typedef struct my_locale_st MY_LOCALE;
 
 extern TYPELIB bool_typelib, delay_key_write_typelib, sql_mode_typelib,
   optimizer_switch_typelib, slave_exec_mode_typelib;
+extern TYPELIB slave_type_conversions_typelib;
 
 typedef int (*sys_check_func)(THD *,  set_var *);
 typedef bool (*sys_update_func)(THD *, set_var *);
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 1ce306bb35e..25155c2cb75 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -6249,5 +6249,10 @@ ER_DEBUG_SYNC_TIMEOUT
 ER_DEBUG_SYNC_HIT_LIMIT
   eng "debug sync point hit limit reached"
   ger "Debug Sync Point Hit Limit erreicht"
+
+ER_SLAVE_CONVERSION_FAILED
+  eng "Column %d of table '%-.192s.%-.192s' cannot be converted from type '%-.32s' to type '%-.32s'"
+ER_SLAVE_CANT_CREATE_CONVERSION
+  eng "Can't create conversion table for table '%-.192s.%-.192s'"
 ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_FORMAT
   eng "Cannot modify @@session.binlog_format inside a transaction"
diff --git a/sql/sp.cc b/sql/sp.cc
index 29ee488220b..8df3e7b6509 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("char(64)") },
+    { 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);
 }
 
 
@@ -1506,7 +1674,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;
@@ -1584,16 +1753,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 *)hash_element(src, i);
     if (!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 64898915b7e..b8209a373a2 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 d3c7d70a484..e776eeba16f 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -2088,8 +2088,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");
@@ -2100,7 +2110,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);
 
@@ -2113,7 +2123,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.
@@ -2125,7 +2136,7 @@ sp_head::restore_lex(THD *thd)
     delete sublex;
   }
   thd->lex= oldlex;
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(FALSE);
 }
 
 /**
@@ -3863,7 +3874,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 3de9abd6760..138d0a2f506 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -344,7 +344,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/sp_rcontext.cc b/sql/sp_rcontext.cc
index 9b237b3e7cc..be8f705a53e 100644
--- a/sql/sp_rcontext.cc
+++ b/sql/sp_rcontext.cc
@@ -617,7 +617,7 @@ sp_rcontext::set_case_expr(THD *thd, int case_expr_id, Item **case_expr_item_ptr
   }
 
   m_case_expr_holders[case_expr_id]->store(case_expr_item);
-
+  m_case_expr_holders[case_expr_id]->cache_value();
   return FALSE;
 }
 
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 3711c70494b..0bed0e05455 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
 
@@ -2404,7 +2405,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();
@@ -2438,14 +2444,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*) hash_first(name_hash, (uchar*) helping,
                                             len, &state);
        grant_name ;
@@ -2478,7 +2487,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);
 }
 
 
@@ -2487,7 +2496,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);
 }
 
 
@@ -2609,7 +2618,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;
+      }
     }
   }
 
@@ -3134,12 +3147,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 */
@@ -3343,12 +3356,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,
@@ -3451,6 +3465,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 c0622bd747c..3104c2ff1a7 100644
--- a/sql/sql_acl.h
+++ b/sql/sql_acl.h
@@ -162,8 +162,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 aab82b90da5..ad6048a156d 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2934,7 +2934,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 861bd97928d..f862cbed4f1 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -421,12 +421,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");
@@ -456,7 +460,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);
@@ -1190,8 +1213,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)
     {
@@ -1385,8 +1414,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 777ddd39280..44fc3123b98 100644
--- a/sql/sql_cache.h
+++ b/sql/sql_cache.h
@@ -485,7 +485,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 db47f103e2b..3fe4054c9e5 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -413,6 +413,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)
@@ -422,6 +424,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_class.h b/sql/sql_class.h
index bb895d6f993..13ebb16095f 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -90,6 +90,12 @@ enum enum_delay_key_write { DELAY_KEY_WRITE_NONE, DELAY_KEY_WRITE_ON,
 enum enum_slave_exec_mode { SLAVE_EXEC_MODE_STRICT,
                             SLAVE_EXEC_MODE_IDEMPOTENT,
                             SLAVE_EXEC_MODE_LAST_BIT};
+enum enum_slave_type_conversions {
+  SLAVE_TYPE_CONVERSIONS_ALL_LOSSY,
+  SLAVE_TYPE_CONVERSIONS_ALL_NON_LOSSY,
+  SLAVE_TYPE_CONVERSIONS_COUNT
+};
+
 enum enum_mark_columns
 { MARK_COLUMNS_NONE, MARK_COLUMNS_READ, MARK_COLUMNS_WRITE};
 
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 1aa7d426814..157d5360a71 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -425,7 +425,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
@@ -1079,6 +1080,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));
@@ -1089,6 +1091,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;
     if (!ha_check_storage_engine_flag(table_type, HTON_CAN_RECREATE))
@@ -1152,11 +1155,13 @@ end:
   {
     if (!error)
     {
-      /*
-        TRUNCATE must always be statement-based binlogged (not row-based) so
-        we don't test current_stmt_binlog_format.
-      */
-      write_bin_log(thd, TRUE, thd->query(), thd->query_length());
+      /* In RBR, the statement is not binlogged if the table is temporary. */
+      if (!is_temporary_table || !thd->is_current_stmt_binlog_format_row())
+        /*
+          TRUNCATE must always be statement-based binlogged (not row-based) so
+          we don't test current_stmt_binlog_format.
+        */
+        write_bin_log(thd, TRUE, thd->query(), thd->query_length());
       my_ok(thd);		// This should return record count
     }
     VOID(pthread_mutex_lock(&LOCK_open));
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index ac3e87285c9..dbec84b6d1a 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -500,6 +500,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 6c79f6cfc8d..2f3aff43ae6 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -304,8 +304,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)))
@@ -348,12 +348,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 88ead714ea9..e6e2466e1ef 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -7429,6 +7429,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;
 }
 
 
@@ -7480,6 +7483,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 5a622740638..251e695f79b 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -200,26 +200,27 @@ bool partition_default_handling(TABLE *table, partition_info *part_info,
 {
   DBUG_ENTER("partition_default_handling");
 
-  if (part_info->use_default_no_partitions)
+  if (!is_create_table_ind)
   {
-    if (!is_create_table_ind &&
-        table->file->get_no_parts(normalized_path, &part_info->no_parts))
+    if (part_info->use_default_no_partitions)
     {
-      DBUG_RETURN(TRUE);
+      if (table->file->get_no_parts(normalized_path, &part_info->no_parts))
+      {
+        DBUG_RETURN(TRUE);
+      }
     }
-  }
-  else if (part_info->is_sub_partitioned() &&
-           part_info->use_default_no_subpartitions)
-  {
-    uint no_parts;
-    if (!is_create_table_ind &&
-        (table->file->get_no_parts(normalized_path, &no_parts)))
+    else if (part_info->is_sub_partitioned() &&
+             part_info->use_default_no_subpartitions)
     {
-      DBUG_RETURN(TRUE);
+      uint no_parts;
+      if (table->file->get_no_parts(normalized_path, &no_parts))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      DBUG_ASSERT(part_info->no_parts > 0);
+      DBUG_ASSERT((no_parts % part_info->no_parts) == 0);
+      part_info->no_subparts= no_parts / part_info->no_parts;
     }
-    DBUG_ASSERT(part_info->no_parts > 0);
-    part_info->no_subparts= no_parts / part_info->no_parts;
-    DBUG_ASSERT((no_parts % part_info->no_parts) == 0);
   }
   part_info->set_up_defaults_for_partitioning(table->file,
                                               (ulonglong)0, (uint)0);
@@ -909,6 +910,8 @@ bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
   char* db_name;
   char db_name_string[FN_REFLEN];
   bool save_use_only_table_context;
+  uint8 saved_full_group_by_flag;
+  nesting_map saved_allow_sum_func;
   DBUG_ENTER("fix_fields_part_func");
 
   if (part_info->fixed)
@@ -978,9 +981,19 @@ bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
   save_use_only_table_context= thd->lex->use_only_table_context;
   thd->lex->use_only_table_context= TRUE;
   thd->lex->current_select->cur_pos_in_select_list= UNDEF_POS;
+  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;
+
   thd->lex->use_only_table_context= save_use_only_table_context;
 
   context->table_list= save_table_list;
@@ -1683,7 +1696,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
   if (((part_info->part_type != HASH_PARTITION ||
       part_info->list_of_part_fields == FALSE) &&
       check_part_func_fields(part_info->part_field_array, TRUE)) ||
-      (part_info->list_of_part_fields == FALSE &&
+      (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 cdaebef49f6..aa9a39b8bc4 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -982,14 +982,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))
   {
@@ -1288,11 +1294,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,
@@ -2122,17 +2131,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();
@@ -6478,6 +6483,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)
 {
@@ -6512,45 +6567,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;
@@ -6558,34 +6583,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)
       {
@@ -6665,6 +6676,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 */
@@ -7816,12 +7830,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
       {
@@ -9452,47 +9466,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
@@ -10479,6 +10454,7 @@ TABLE *create_virtual_tmp_table(THD *thd, List<Create_field> &field_list)
   share->blob_field= blob_field;
   share->fields= field_count;
   share->blob_ptr_size= portable_sizeof_char_ptr;
+  share->db_low_byte_first=1;                // True for HEAP and MyISAM
   setup_tmp_table_column_bitmaps(table, bitmaps);
 
   /* Create all fields and calculate the total length of record */
@@ -10543,6 +10519,18 @@ TABLE *create_virtual_tmp_table(THD *thd, List<Create_field> &field_list)
           null_bit= 1;
         }
       }
+      if (cur_field->type() == MYSQL_TYPE_BIT &&
+          cur_field->key_type() == HA_KEYTYPE_BIT)
+      {
+        /* This is a Field_bit since key_type is HA_KEYTYPE_BIT */
+        static_cast<Field_bit*>(cur_field)->set_bit_ptr(null_pos, null_bit);
+        null_bit+= cur_field->field_length & 7;
+        if (null_bit > 7)
+        {
+          null_pos++;
+          null_bit-= 8;
+        }
+      }
       cur_field->reset();
 
       field_pos+= cur_field->pack_length();
@@ -13186,6 +13174,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
 	{
@@ -13991,7 +13981,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_show.cc b/sql/sql_show.cc
index 413733c0148..5a43c807fc9 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -719,7 +719,7 @@ mysqld_show_create(THD *thd, TABLE_LIST *table_list)
     thd->push_internal_handler(&view_error_suppressor);
     bool error= open_normal_and_derived_tables(thd, table_list, 0);
     thd->pop_internal_handler();
-    if (error && thd->main_da.is_error())
+    if (error && (thd->killed || thd->main_da.is_error()))
       DBUG_RETURN(TRUE);
   }
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index df9f0d0aa69..b05afd6776c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -5331,12 +5331,20 @@ binlog:
         }
         VOID(pthread_mutex_unlock(&LOCK_open));
 
-        IF_DBUG(int result=)
-          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)
+        {
+          IF_DBUG(int result=)
+            store_create_info(thd, table, &query,
+                              create_info, FALSE /* show_database */);
 
-        DBUG_ASSERT(result == 0); // store_create_info() always return 0
-        write_bin_log(thd, TRUE, query.ptr(), query.length());
+          DBUG_ASSERT(result == 0); // store_create_info() always return 0
+          write_bin_log(thd, TRUE, query.ptr(), query.length());
+        }
       }
       else                                      // Case 1
         write_bin_log(thd, TRUE, thd->query(), thd->query_length());
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 197870a7296..977b9e622ce 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -389,6 +389,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.
@@ -2398,8 +2530,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;
           }
@@ -2509,7 +2641,8 @@ sp_cursor_stmt:
             }
             lex->sp_lex_in_use= TRUE;
             $$= lex;
-            lex->sphead->restore_lex(YYTHD);
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
         ;
 
@@ -2728,7 +2861,8 @@ sp_proc_stmt_statement:
                     sp->add_instr(i))
                 MYSQL_YYABORT;
             }
-            sp->restore_lex(thd);
+            if (sp->restore_lex(thd))
+              MYSQL_YYABORT;
           }
         ;
 
@@ -2756,7 +2890,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;
           }
         ;
 
@@ -2996,7 +3131,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
           {
@@ -3042,7 +3178,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
@@ -3092,7 +3230,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
@@ -3113,7 +3253,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
@@ -3290,7 +3432,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
           {
@@ -3316,7 +3459,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;
           }
@@ -7562,6 +7706,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(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
             if (global_system_variables.sysdate_is_now == 0)
               $$= new (YYTHD->mem_root) Item_func_sysdate_local();
             else
@@ -11773,7 +11925,8 @@ option_type_value:
                 if (sp->add_instr(i))
                   MYSQL_YYABORT;
               }
-              lex->sphead->restore_lex(thd);
+              if (lex->sphead->restore_lex(thd))
+                MYSQL_YYABORT;
             }
           }
         ;
@@ -11813,98 +11966,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;
             }
           }
@@ -11940,11 +12037,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
           {
@@ -12037,31 +12139,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 d2538eb4d59..752b6ba0bd4 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1315,8 +1315,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
 
@@ -2803,34 +2811,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);
     }
     /*
@@ -2842,7 +2854,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);
@@ -2850,18 +2862,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);
       /*
@@ -2881,47 +2893,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 f9da0637b4b..cc4257c826d 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.
@@ -424,6 +454,18 @@ typedef struct st_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;
 
@@ -1665,17 +1707,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)