Manual merge from mysql-next-mr.

Conflicts:
  - sql/sql_base.cc

44 files changed, 1327 insertions, 584 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 51bb527fc85..51235d9f0e8 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1779,11 +1779,10 @@ uint Field::fill_cache_field(CACHE_FIELD *copy)
   uint store_length;
   copy->str=ptr;
   copy->length=pack_length();
-  copy->blob_field=0;
+  copy->field= this;
   if (flags & BLOB_FLAG)
   {
-    copy->blob_field=(Field_blob*) this;
-    copy->strip=0;
+    copy->type= CACHE_BLOB;
     copy->length-= table->s->blob_ptr_size;
     return copy->length;
   }
@@ -1791,15 +1790,15 @@ uint Field::fill_cache_field(CACHE_FIELD *copy)
            (type() == MYSQL_TYPE_STRING && copy->length >= 4 &&
             copy->length < 256))
   {
-    copy->strip=1;				/* Remove end space */
+    copy->type= CACHE_STRIPPED;
     store_length= 2;
   }
   else
   {
-    copy->strip=0;
+    copy->type= 0;
     store_length= 0;
   }
-  return copy->length+ store_length;
+  return copy->length + store_length;
 }
 
 
@@ -1808,7 +1807,7 @@ bool Field::get_date(MYSQL_TIME *ltime,uint fuzzydate)
   char buff[40];
   String tmp(buff,sizeof(buff),&my_charset_bin),*res;
   if (!(res=val_str(&tmp)) ||
-      str_to_datetime_with_warn(res->ptr(), res->length(),
+      str_to_datetime_with_warn(res->charset(), res->ptr(), res->length(),
                                 ltime, fuzzydate) <= MYSQL_TIMESTAMP_ERROR)
     return 1;
   return 0;
@@ -1819,7 +1818,7 @@ bool Field::get_time(MYSQL_TIME *ltime)
   char buff[40];
   String tmp(buff,sizeof(buff),&my_charset_bin),*res;
   if (!(res=val_str(&tmp)) ||
-      str_to_time_with_warn(res->ptr(), res->length(), ltime))
+      str_to_time_with_warn(res->charset(), res->ptr(), res->length(), ltime))
     return 1;
   return 0;
 }
@@ -1836,7 +1835,9 @@ int Field::store_time(MYSQL_TIME *ltime, timestamp_type type_arg)
   ASSERT_COLUMN_MARKED_FOR_WRITE;
   char buff[MAX_DATE_STRING_REP_LENGTH];
   uint length= (uint) my_TIME_to_str(ltime, buff);
-  return store(buff, length, &my_charset_bin);
+  /* Avoid conversion when field character set is ASCII compatible */
+  return store(buff, length, (charset()->state & MY_CS_NONASCII) ?
+                              &my_charset_latin1 : charset());
 }
 
 
@@ -2428,7 +2429,7 @@ String *Field_decimal::val_str(String *val_buffer __attribute__((unused)),
   size_t tmp_length;
 
   for (str=ptr ; *str == ' ' ; str++) ;
-  val_ptr->set_charset(&my_charset_bin);
+  val_ptr->set_charset(&my_charset_numeric);
   tmp_length= (size_t) (str-ptr);
   if (field_length < tmp_length)		// Error in data
     val_ptr->length(0);
@@ -2558,7 +2559,7 @@ Field *Field_new_decimal::create_from_item (Item *item)
 {
   uint8 dec= item->decimals;
   uint8 intg= item->decimal_precision() - dec;
-  uint32 len= item->max_length;
+  uint32 len= item->max_char_length();
 
   DBUG_ASSERT (item->result_type() == DECIMAL_RESULT);
 
@@ -2850,6 +2851,7 @@ String *Field_new_decimal::val_str(String *val_buffer,
   uint fixed_precision= zerofill ? precision : 0;
   my_decimal2string(E_DEC_FATAL_ERROR, val_decimal(&decimal_value),
                     fixed_precision, dec, '0', val_buffer);
+  val_buffer->set_charset(&my_charset_numeric);
   return val_buffer;
 }
 
@@ -3121,7 +3123,7 @@ String *Field_tiny::val_str(String *val_buffer,
 			    String *val_ptr __attribute__((unused)))
 {
   ASSERT_COLUMN_MARKED_FOR_READ;
-  CHARSET_INFO *cs= &my_charset_bin;
+  CHARSET_INFO *cs= &my_charset_numeric;
   uint length;
   uint mlength=max(field_length+1,5*cs->mbmaxlen);
   val_buffer->alloc(mlength);
@@ -3137,6 +3139,7 @@ String *Field_tiny::val_str(String *val_buffer,
   val_buffer->length(length);
   if (zerofill)
     prepend_zeros(val_buffer);
+  val_buffer->set_charset(cs);
   return val_buffer;
 }
 
@@ -3333,7 +3336,7 @@ String *Field_short::val_str(String *val_buffer,
 			     String *val_ptr __attribute__((unused)))
 {
   ASSERT_COLUMN_MARKED_FOR_READ;
-  CHARSET_INFO *cs= &my_charset_bin;
+  CHARSET_INFO *cs= &my_charset_numeric;
   uint length;
   uint mlength=max(field_length+1,7*cs->mbmaxlen);
   val_buffer->alloc(mlength);
@@ -3354,6 +3357,7 @@ String *Field_short::val_str(String *val_buffer,
   val_buffer->length(length);
   if (zerofill)
     prepend_zeros(val_buffer);
+  val_buffer->set_charset(cs);
   return val_buffer;
 }
 
@@ -3550,7 +3554,7 @@ String *Field_medium::val_str(String *val_buffer,
 			      String *val_ptr __attribute__((unused)))
 {
   ASSERT_COLUMN_MARKED_FOR_READ;
-  CHARSET_INFO *cs= &my_charset_bin;
+  CHARSET_INFO *cs= &my_charset_numeric;
   uint length;
   uint mlength=max(field_length+1,10*cs->mbmaxlen);
   val_buffer->alloc(mlength);
@@ -3561,6 +3565,7 @@ String *Field_medium::val_str(String *val_buffer,
   val_buffer->length(length);
   if (zerofill)
     prepend_zeros(val_buffer); /* purecov: inspected */
+  val_buffer->set_charset(cs);
   return val_buffer;
 }
 
@@ -3769,7 +3774,7 @@ String *Field_long::val_str(String *val_buffer,
 			    String *val_ptr __attribute__((unused)))
 {
   ASSERT_COLUMN_MARKED_FOR_READ;
-  CHARSET_INFO *cs= &my_charset_bin;
+  CHARSET_INFO *cs= &my_charset_numeric;
   uint length;
   uint mlength=max(field_length+1,12*cs->mbmaxlen);
   val_buffer->alloc(mlength);
@@ -3789,6 +3794,7 @@ String *Field_long::val_str(String *val_buffer,
   val_buffer->length(length);
   if (zerofill)
     prepend_zeros(val_buffer);
+  val_buffer->set_charset(cs);
   return val_buffer;
 }
 
@@ -4010,7 +4016,7 @@ longlong Field_longlong::val_int(void)
 String *Field_longlong::val_str(String *val_buffer,
 				String *val_ptr __attribute__((unused)))
 {
-  CHARSET_INFO *cs= &my_charset_bin;
+  CHARSET_INFO *cs= &my_charset_numeric;
   uint length;
   uint mlength=max(field_length+1,22*cs->mbmaxlen);
   val_buffer->alloc(mlength);
@@ -4028,6 +4034,7 @@ String *Field_longlong::val_str(String *val_buffer,
   val_buffer->length(length);
   if (zerofill)
     prepend_zeros(val_buffer);
+  val_buffer->set_charset(cs);
   return val_buffer;
 }
 
@@ -4254,6 +4261,7 @@ String *Field_float::val_str(String *val_buffer,
   val_buffer->length((uint) len);
   if (zerofill)
     prepend_zeros(val_buffer);
+  val_buffer->set_charset(&my_charset_numeric);
   return val_buffer;
 }
 
@@ -4564,6 +4572,7 @@ String *Field_double::val_str(String *val_buffer,
   val_buffer->length((uint) len);
   if (zerofill)
     prepend_zeros(val_buffer);
+  val_buffer->set_charset(&my_charset_numeric);
   return val_buffer;
 }
 
@@ -4698,7 +4707,7 @@ Field_timestamp::Field_timestamp(uchar *ptr_arg, uint32 len_arg,
 	     unireg_check_arg, field_name_arg, cs)
 {
   /* For 4.0 MYD and 4.0 InnoDB compatibility */
-  flags|= ZEROFILL_FLAG | UNSIGNED_FLAG;
+  flags|= ZEROFILL_FLAG | UNSIGNED_FLAG | BINARY_FLAG;
   if (!share->timestamp_field && unireg_check != NONE)
   {
     /* This timestamp has auto-update */
@@ -4718,7 +4727,7 @@ Field_timestamp::Field_timestamp(bool maybe_null_arg,
 	     NONE, field_name_arg, cs)
 {
   /* For 4.0 MYD and 4.0 InnoDB compatibility */
-  flags|= ZEROFILL_FLAG | UNSIGNED_FLAG;
+  flags|= ZEROFILL_FLAG | UNSIGNED_FLAG | BINARY_FLAG;
     if (unireg_check != TIMESTAMP_DN_FIELD)
       flags|= ON_UPDATE_NOW_FLAG;
 }
@@ -4770,7 +4779,7 @@ int Field_timestamp::store(const char *from,uint len,CHARSET_INFO *cs)
   THD *thd= table ? table->in_use : current_thd;
 
   /* We don't want to store invalid or fuzzy datetime values in TIMESTAMP */
-  have_smth_to_conv= (str_to_datetime(from, len, &l_time,
+  have_smth_to_conv= (str_to_datetime(cs, from, len, &l_time,
                                       (thd->variables.sql_mode &
                                        MODE_NO_ZERO_DATE) |
                                       MODE_NO_ZERO_IN_DATE, &error) >
@@ -4919,10 +4928,10 @@ String *Field_timestamp::val_str(String *val_buffer, String *val_ptr)
 
   if (temp == 0L)
   {				      /* Zero time is "000000" */
-    val_ptr->set(STRING_WITH_LEN("0000-00-00 00:00:00"), &my_charset_bin);
+    val_ptr->set(STRING_WITH_LEN("0000-00-00 00:00:00"), &my_charset_numeric);
     return val_ptr;
   }
-  val_buffer->set_charset(&my_charset_bin);	// Safety
+  val_buffer->set_charset(&my_charset_numeric);	// Safety
   
   thd->variables.time_zone->gmt_sec_to_TIME(&time_tmp,(my_time_t)temp);
 
@@ -4966,6 +4975,7 @@ String *Field_timestamp::val_str(String *val_buffer, String *val_ptr)
   *to++= (char) ('0'+(char) (temp2));
   *to++= (char) ('0'+(char) (temp));
   *to= 0;
+  val_buffer->set_charset(&my_charset_numeric);
   return val_buffer;
 }
 
@@ -5076,7 +5086,7 @@ int Field_time::store(const char *from,uint len,CHARSET_INFO *cs)
   int error= 0;
   int warning;
 
-  if (str_to_time(from, len, &ltime, &warning))
+  if (str_to_time(cs, from, len, &ltime, &warning))
   {
     tmp=0L;
     error= 2;
@@ -5235,6 +5245,7 @@ String *Field_time::val_str(String *val_buffer,
   ltime.minute= (uint) (tmp/100 % 100);
   ltime.second= (uint) (tmp % 100);
   make_time((DATE_TIME_FORMAT*) 0, &ltime, val_buffer);
+  val_buffer->set_charset(&my_charset_numeric);
   return val_buffer;
 }
 
@@ -5432,6 +5443,7 @@ String *Field_year::val_str(String *val_buffer,
   val_buffer->length(field_length);
   char *to=(char*) val_buffer->ptr();
   sprintf(to,field_length == 2 ? "%02d" : "%04d",(int) Field_year::val_int());
+  val_buffer->set_charset(&my_charset_numeric);
   return val_buffer;
 }
 
@@ -5459,7 +5471,7 @@ int Field_date::store(const char *from, uint len,CHARSET_INFO *cs)
   int error;
   THD *thd= table ? table->in_use : current_thd;
 
-  if (str_to_datetime(from, len, &l_time, TIME_FUZZY_DATE |
+  if (str_to_datetime(cs, from, len, &l_time, TIME_FUZZY_DATE |
                       (thd->variables.sql_mode &
                        (MODE_NO_ZERO_IN_DATE | MODE_NO_ZERO_DATE |
                         MODE_INVALID_DATES)),
@@ -5606,6 +5618,7 @@ String *Field_date::val_str(String *val_buffer,
   ltime.month= (int) ((uint32) tmp/100 % 100);
   ltime.day= (int) ((uint32) tmp % 100);
   make_date((DATE_TIME_FORMAT *) 0, &ltime, val_buffer);
+  val_buffer->set_charset(&my_charset_numeric);
   return val_buffer;
 }
 
@@ -5695,7 +5708,7 @@ int Field_newdate::store(const char *from,uint len,CHARSET_INFO *cs)
   int error;
   THD *thd= table ? table->in_use : current_thd;
   enum enum_mysql_timestamp_type ret;
-  if ((ret= str_to_datetime(from, len, &l_time,
+  if ((ret= str_to_datetime(cs, from, len, &l_time,
                             (TIME_FUZZY_DATE |
                              (thd->variables.sql_mode &
                               (MODE_NO_ZERO_IN_DATE | MODE_NO_ZERO_DATE |
@@ -5867,6 +5880,7 @@ String *Field_newdate::val_str(String *val_buffer,
   *pos--= (char) ('0'+part%10); part/=10;
   *pos--= (char) ('0'+part%10); part/=10;
   *pos=   (char) ('0'+part);
+  val_buffer->set_charset(&my_charset_numeric);
   return val_buffer;
 }
 
@@ -5929,7 +5943,7 @@ int Field_datetime::store(const char *from,uint len,CHARSET_INFO *cs)
   enum enum_mysql_timestamp_type func_res;
   THD *thd= table ? table->in_use : current_thd;
 
-  func_res= str_to_datetime(from, len, &time_tmp,
+  func_res= str_to_datetime(cs, from, len, &time_tmp,
                             (TIME_FUZZY_DATE |
                              (thd->variables.sql_mode &
                               (MODE_NO_ZERO_IN_DATE | MODE_NO_ZERO_DATE |
@@ -6129,6 +6143,7 @@ String *Field_datetime::val_str(String *val_buffer,
   *pos--= (char) ('0'+(char) (part3%10)); part3/=10;
   *pos--= (char) ('0'+(char) (part3%10)); part3/=10;
   *pos=(char) ('0'+(char) part3);
+  val_buffer->set_charset(&my_charset_numeric);
   return val_buffer;
 }
 
@@ -6370,7 +6385,7 @@ int Field_str::store(double nr)
     else
       set_warning(MYSQL_ERROR::WARN_LEVEL_WARN, WARN_DATA_TRUNCATED, 1);
   }
-  return store(buff, length, charset());
+  return store(buff, length, &my_charset_numeric);
 }
 
 
@@ -6405,7 +6420,7 @@ int Field_string::store(longlong nr, bool unsigned_val)
 int Field_longstr::store_decimal(const my_decimal *d)
 {
   char buff[DECIMAL_MAX_STR_LENGTH+1];
-  String str(buff, sizeof(buff), &my_charset_bin);
+  String str(buff, sizeof(buff), &my_charset_numeric);
   my_decimal2string(E_DEC_FATAL_ERROR, d, 0, 0, 0, &str);
   return store(str.ptr(), str.length(), str.charset());
 }
@@ -9565,8 +9580,8 @@ bool Create_field::init(THD *thd, char *fld_name, enum_field_types fld_type,
   case MYSQL_TYPE_TIME:
   case MYSQL_TYPE_DATETIME:
   case MYSQL_TYPE_TIMESTAMP:
-    charset= &my_charset_bin;
-    flags|= BINCMP_FLAG;
+    charset= &my_charset_numeric;
+    flags|= BINARY_FLAG;
   default: break;
   }
 
@@ -9682,7 +9697,7 @@ Field *make_field(TABLE_SHARE *share, uchar *ptr, uint32 field_length,
   case MYSQL_TYPE_TIME:
   case MYSQL_TYPE_DATETIME:
   case MYSQL_TYPE_TIMESTAMP:
-    field_charset= &my_charset_bin;
+    field_charset= &my_charset_numeric;
   default: break;
   }
 
diff --git a/sql/field.h b/sql/field.h
index 7903e8daa0b..fb985ab9930 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -414,11 +414,14 @@ public:
   virtual bool get_date(MYSQL_TIME *ltime,uint fuzzydate);
   virtual bool get_time(MYSQL_TIME *ltime);
   virtual CHARSET_INFO *charset(void) const { return &my_charset_bin; }
+  virtual CHARSET_INFO *charset_for_protocol(void) const
+  { return binary() ? &my_charset_bin : charset(); }
   virtual CHARSET_INFO *sort_charset(void) const { return charset(); }
   virtual bool has_charset(void) const { return FALSE; }
   virtual void set_charset(CHARSET_INFO *charset_arg) { }
   virtual enum Derivation derivation(void) const
   { return DERIVATION_IMPLICIT; }
+  virtual uint repertoire(void) const { return MY_REPERTOIRE_UNICODE30; }
   virtual void set_derivation(enum Derivation derivation_arg) { }
   bool set_warning(MYSQL_ERROR::enum_warning_level, unsigned int code,
                    int cuted_increment);
@@ -599,6 +602,9 @@ public:
 	    const char *field_name_arg,
             uint8 dec_arg, bool zero_arg, bool unsigned_arg);
   Item_result result_type () const { return REAL_RESULT; }
+  enum Derivation derivation(void) const { return DERIVATION_NUMERIC; }
+  uint repertoire(void) const { return MY_REPERTOIRE_NUMERIC; }
+  CHARSET_INFO *charset(void) const { return &my_charset_numeric; }
   void prepend_zeros(String *value);
   void add_zerofill_and_unsigned(String &res) const;
   friend class Create_field;
@@ -1169,6 +1175,10 @@ public:
   enum_field_types type() const { return MYSQL_TYPE_TIMESTAMP;}
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_ULONG_INT; }
   enum Item_result cmp_type () const { return INT_RESULT; }
+  enum Derivation derivation(void) const { return DERIVATION_NUMERIC; }
+  uint repertoire(void) const { return MY_REPERTOIRE_NUMERIC; }
+  CHARSET_INFO *charset(void) const { return &my_charset_numeric; }
+  bool binary() const { return 1; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(double nr);
   int  store(longlong nr, bool unsigned_val);
@@ -1261,14 +1271,18 @@ public:
 	     CHARSET_INFO *cs)
     :Field_str(ptr_arg, MAX_DATE_WIDTH, null_ptr_arg, null_bit_arg,
 	       unireg_check_arg, field_name_arg, cs)
-    {}
+    { flags|= BINARY_FLAG; }
   Field_date(bool maybe_null_arg, const char *field_name_arg,
              CHARSET_INFO *cs)
     :Field_str((uchar*) 0, MAX_DATE_WIDTH, maybe_null_arg ? (uchar*) "": 0,0,
-	       NONE, field_name_arg, cs) {}
+	       NONE, field_name_arg, cs) { flags|= BINARY_FLAG; }
   enum_field_types type() const { return MYSQL_TYPE_DATE;}
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_ULONG_INT; }
   enum Item_result cmp_type () const { return INT_RESULT; }
+  enum Derivation derivation(void) const { return DERIVATION_NUMERIC; }
+  uint repertoire(void) const { return MY_REPERTOIRE_NUMERIC; }
+  CHARSET_INFO *charset(void) const { return &my_charset_numeric; }
+  bool binary() const { return 1; }
   int store(const char *to,uint length,CHARSET_INFO *charset);
   int store(double nr);
   int store(longlong nr, bool unsigned_val);
@@ -1305,15 +1319,19 @@ public:
 		CHARSET_INFO *cs)
     :Field_str(ptr_arg, 10, null_ptr_arg, null_bit_arg,
 	       unireg_check_arg, field_name_arg, cs)
-    {}
+    { flags|= BINARY_FLAG; }
   Field_newdate(bool maybe_null_arg, const char *field_name_arg,
                 CHARSET_INFO *cs)
     :Field_str((uchar*) 0,10, maybe_null_arg ? (uchar*) "": 0,0,
-               NONE, field_name_arg, cs) {}
+               NONE, field_name_arg, cs) { flags|= BINARY_FLAG; }
   enum_field_types type() const { return MYSQL_TYPE_DATE;}
   enum_field_types real_type() const { return MYSQL_TYPE_NEWDATE; }
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_UINT24; }
   enum Item_result cmp_type () const { return INT_RESULT; }
+  enum Derivation derivation(void) const { return DERIVATION_NUMERIC; }
+  uint repertoire(void) const { return MY_REPERTOIRE_NUMERIC; }
+  CHARSET_INFO *charset(void) const { return &my_charset_numeric; }
+  bool binary() const { return 1; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(double nr);
   int  store(longlong nr, bool unsigned_val);
@@ -1341,14 +1359,18 @@ public:
 	     CHARSET_INFO *cs)
     :Field_str(ptr_arg, 8, null_ptr_arg, null_bit_arg,
 	       unireg_check_arg, field_name_arg, cs)
-    {}
+    { flags|= BINARY_FLAG; }
   Field_time(bool maybe_null_arg, const char *field_name_arg,
              CHARSET_INFO *cs)
     :Field_str((uchar*) 0,8, maybe_null_arg ? (uchar*) "": 0,0,
-	       NONE, field_name_arg, cs) {}
+	       NONE, field_name_arg, cs) { flags|= BINARY_FLAG; }
   enum_field_types type() const { return MYSQL_TYPE_TIME;}
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_INT24; }
   enum Item_result cmp_type () const { return INT_RESULT; }
+  enum Derivation derivation(void) const { return DERIVATION_NUMERIC; }
+  uint repertoire(void) const { return MY_REPERTOIRE_NUMERIC; }
+  CHARSET_INFO *charset(void) const { return &my_charset_numeric; }
+  bool binary() const { return 1; }
   int store_time(MYSQL_TIME *ltime, timestamp_type type);
   int store(const char *to,uint length,CHARSET_INFO *charset);
   int store(double nr);
@@ -1376,16 +1398,20 @@ public:
 		 CHARSET_INFO *cs)
     :Field_str(ptr_arg, MAX_DATETIME_WIDTH, null_ptr_arg, null_bit_arg,
 	       unireg_check_arg, field_name_arg, cs)
-    {}
+    { flags|= BINARY_FLAG; }
   Field_datetime(bool maybe_null_arg, const char *field_name_arg,
 		 CHARSET_INFO *cs)
     :Field_str((uchar*) 0, MAX_DATETIME_WIDTH, maybe_null_arg ? (uchar*) "": 0,0,
-	       NONE, field_name_arg, cs) {}
+	       NONE, field_name_arg, cs) { flags|= BINARY_FLAG; }
   enum_field_types type() const { return MYSQL_TYPE_DATETIME;}
 #ifdef HAVE_LONG_LONG
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_ULONGLONG; }
 #endif
   enum Item_result cmp_type () const { return INT_RESULT; }
+  enum Derivation derivation(void) const { return DERIVATION_NUMERIC; }
+  uint repertoire(void) const { return MY_REPERTOIRE_NUMERIC; }
+  CHARSET_INFO *charset(void) const { return &my_charset_numeric; }
+  bool binary() const { return 1; }
   uint decimals() const { return DATETIME_DEC; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(double nr);
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 3289b529edf..877cf070abd 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -1803,13 +1803,23 @@ void ha_partition::update_create_info(HA_CREATE_INFO *create_info)
 
 void ha_partition::change_table_ptr(TABLE *table_arg, TABLE_SHARE *share)
 {
-  handler **file_array= m_file;
+  handler **file_array;
   table= table_arg;
   table_share= share;
-  do
+  /*
+    m_file can be NULL when using an old cached table in DROP TABLE, when the
+    table just has REMOVED PARTITIONING, see Bug#42438
+  */
+  if (m_file)
   {
-    (*file_array)->change_table_ptr(table_arg, share);
-  } while (*(++file_array));
+    file_array= m_file;
+    DBUG_ASSERT(*file_array);
+    do
+    {
+      (*file_array)->change_table_ptr(table_arg, share);
+    } while (*(++file_array));
+  }
+
   if (m_added_file && m_added_file[0])
   {
     /* if in middle of a drop/rename etc */
@@ -6226,7 +6236,13 @@ void ha_partition::print_error(int error, myf errflag)
       thd->lex->sql_command != SQLCOM_TRUNCATE)
     m_part_info->print_no_partition_found(table);
   else
-    m_file[m_last_part]->print_error(error, errflag);
+  {
+    /* In case m_file has not been initialized, like in bug#42438 */
+    if (m_file)
+      m_file[m_last_part]->print_error(error, errflag);
+    else
+      handler::print_error(error, errflag);
+  }
   DBUG_VOID_RETURN;
 }
 
@@ -6236,7 +6252,12 @@ bool ha_partition::get_error_message(int error, String *buf)
   DBUG_ENTER("ha_partition::get_error_message");
 
   /* Should probably look for my own errors first */
-  DBUG_RETURN(m_file[m_last_part]->get_error_message(error, buf));
+
+  /* In case m_file has not been initialized, like in bug#42438 */
+  if (m_file)
+    DBUG_RETURN(m_file[m_last_part]->get_error_message(error, buf));
+  DBUG_RETURN(handler::get_error_message(error, buf));
+
 }
 
 
diff --git a/sql/item.cc b/sql/item.cc
index e785f0addde..e269f6399e9 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -201,6 +201,37 @@ bool Item::val_bool()
 }
 
 
+/*
+  For the items which don't have its own fast val_str_ascii()
+  implementation we provide a generic slower version,
+  which converts from the Item character set to ASCII.
+  For better performance conversion happens only in 
+  case of a "tricky" Item character set (e.g. UCS2).
+  Normally conversion does not happen.
+*/
+String *Item::val_str_ascii(String *str)
+{
+  DBUG_ASSERT(fixed == 1);
+
+  if (!(collation.collation->state & MY_CS_NONASCII))
+    return val_str(str);
+  
+  DBUG_ASSERT(str != &str_value);
+  
+  uint errors;
+  String *res= val_str(&str_value);
+  if (!res)
+    return 0;
+  
+  if ((null_value= str->copy(res->ptr(), res->length(),
+                             collation.collation, &my_charset_latin1,
+                             &errors)))
+    return 0;
+  
+  return str;
+}
+
+
 String *Item::val_string_from_real(String *str)
 {
   double nr= val_real();
@@ -443,10 +474,11 @@ uint Item::decimal_precision() const
   if ((restype == DECIMAL_RESULT) || (restype == INT_RESULT))
   {
     uint prec= 
-      my_decimal_length_to_precision(max_length, decimals, unsigned_flag);
+      my_decimal_length_to_precision(max_char_length(), decimals,
+                                     unsigned_flag);
     return min(prec, DECIMAL_MAX_PRECISION);
   }
-  return min(max_length, DECIMAL_MAX_PRECISION);
+  return min(max_char_length(), DECIMAL_MAX_PRECISION);
 }
 
 
@@ -783,15 +815,40 @@ Item *Item::safe_charset_converter(CHARSET_INFO *tocs)
 */
 Item *Item_num::safe_charset_converter(CHARSET_INFO *tocs)
 {
+  /*
+    Item_num returns pure ASCII result,
+    so conversion is needed only in case of "tricky" character
+    sets like UCS2. If tocs is not "tricky", return the item itself.
+  */
+  if (!(tocs->state & MY_CS_NONASCII))
+    return this;
+  
   Item_string *conv;
-  char buf[64];
-  String *s, tmp(buf, sizeof(buf), &my_charset_bin);
-  s= val_str(&tmp);
-  if ((conv= new Item_string(s->ptr(), s->length(), s->charset())))
+  uint conv_errors;
+  char buf[64], buf2[64];
+  String tmp(buf, sizeof(buf), &my_charset_bin);
+  String cstr(buf2, sizeof(buf2), &my_charset_bin);
+  String *ostr= val_str(&tmp);
+  char *ptr;
+  cstr.copy(ostr->ptr(), ostr->length(), ostr->charset(), tocs, &conv_errors);
+  if (conv_errors || !(conv= new Item_string(cstr.ptr(), cstr.length(),
+                                             cstr.charset(),
+                                             collation.derivation)))
   {
-    conv->str_value.copy();
-    conv->str_value.mark_as_const();
+    /*
+      Safe conversion is not possible (or EOM).
+      We could not convert a string into the requested character set
+      without data loss. The target charset does not cover all the
+      characters from the string. Operation cannot be done correctly.
+    */
+    return NULL;
   }
+  if (!(ptr= current_thd->strmake(cstr.ptr(), cstr.length())))
+    return NULL;
+  conv->str_value.set(ptr, cstr.length(), cstr.charset());
+  /* Ensure that no one is going to change the result string */
+  conv->str_value.mark_as_const();
+  conv->fix_char_length(max_char_length());
   return conv;
 }
 
@@ -854,7 +911,7 @@ Item *Item_param::safe_charset_converter(CHARSET_INFO *tocs)
     cnvitem->max_length= cnvitem->str_value.numchars() * tocs->mbmaxlen;
     return cnvitem;
   }
-  return NULL;
+  return Item::safe_charset_converter(tocs);
 }
 
 
@@ -910,7 +967,7 @@ bool Item::get_date(MYSQL_TIME *ltime,uint fuzzydate)
     char buff[40];
     String tmp(buff,sizeof(buff), &my_charset_bin),*res;
     if (!(res=val_str(&tmp)) ||
-        str_to_datetime_with_warn(res->ptr(), res->length(),
+        str_to_datetime_with_warn(res->charset(), res->ptr(), res->length(),
                                   ltime, fuzzydate) <= MYSQL_TIMESTAMP_ERROR)
       goto err;
   }
@@ -945,8 +1002,8 @@ bool Item::get_time(MYSQL_TIME *ltime)
 {
   char buff[40];
   String tmp(buff,sizeof(buff),&my_charset_bin),*res;
-  if (!(res=val_str(&tmp)) ||
-      str_to_time_with_warn(res->ptr(), res->length(), ltime))
+  if (!(res=val_str_ascii(&tmp)) ||
+      str_to_time_with_warn(res->charset(), res->ptr(), res->length(), ltime))
   {
     bzero((char*) ltime,sizeof(*ltime));
     return 1;
@@ -1436,7 +1493,12 @@ left_is_superset(DTCollation *left, DTCollation *right)
   if (left->collation->state & MY_CS_UNICODE &&
       (left->derivation < right->derivation ||
        (left->derivation == right->derivation &&
-        !(right->collation->state & MY_CS_UNICODE))))
+        (!(right->collation->state & MY_CS_UNICODE) ||
+         /* The code below makes 4-byte utf8 a superset over 3-byte utf8 */
+         (left->collation->state & MY_CS_UNICODE_SUPPLEMENT &&
+          !(right->collation->state & MY_CS_UNICODE_SUPPLEMENT) &&
+          left->collation->mbmaxlen > right->collation->mbmaxlen &&
+          left->collation->mbminlen == right->collation->mbminlen)))))
     return TRUE;
   /* Allow convert from ASCII */
   if (right->repertoire == MY_REPERTOIRE_ASCII &&
@@ -1650,6 +1712,11 @@ bool agg_item_collations(DTCollation &c, const char *fname,
     my_coll_agg_error(av, count, fname, item_sep);
     return TRUE;
   }
+  
+  /* If all arguments where numbers, reset to @@collation_connection */
+  if (c.derivation == DERIVATION_NUMERIC)
+    c.set(Item::default_charset(), DERIVATION_COERCIBLE, MY_REPERTOIRE_NUMERIC);
+
   return FALSE;
 }
 
@@ -1695,7 +1762,7 @@ bool agg_item_set_converter(DTCollation &coll, const char *fname,
   {
     Item* conv;
     uint32 dummy_offset;
-    if (!String::needs_conversion(0, (*arg)->collation.collation,
+    if (!String::needs_conversion(1, (*arg)->collation.collation,
                                   coll.collation,
                                   &dummy_offset))
       continue;
@@ -1895,13 +1962,14 @@ void Item_field::set_field(Field *field_par)
   field=result_field=field_par;			// for easy coding with fields
   maybe_null=field->maybe_null();
   decimals= field->decimals();
-  max_length= field_par->max_display_length();
   table_name= *field_par->table_name;
   field_name= field_par->field_name;
   db_name= field_par->table->s->db.str;
   alias_name_used= field_par->table->alias_name_used;
   unsigned_flag=test(field_par->flags & UNSIGNED_FLAG);
-  collation.set(field_par->charset(), field_par->derivation());
+  collation.set(field_par->charset(), field_par->derivation(),
+                field_par->repertoire());
+  fix_char_length(field_par->char_length());
   fixed= 1;
   if (field->table->s->tmp_table == SYSTEM_TMP_TABLE)
     any_privileges= 0;
@@ -2210,14 +2278,14 @@ String *Item_int::val_str(String *str)
 {
   // following assert is redundant, because fixed=1 assigned in constructor
   DBUG_ASSERT(fixed == 1);
-  str->set(value, &my_charset_bin);
+  str->set_int(value, unsigned_flag, collation.collation);
   return str;
 }
 
 void Item_int::print(String *str, enum_query_type query_type)
 {
   // my_charset_bin is good enough for numbers
-  str_value.set(value, &my_charset_bin);
+  str_value.set_int(value, unsigned_flag, &my_charset_bin);
   str->append(str_value);
 }
 
@@ -2240,7 +2308,7 @@ String *Item_uint::val_str(String *str)
 {
   // following assert is redundant, because fixed=1 assigned in constructor
   DBUG_ASSERT(fixed == 1);
-  str->set((ulonglong) value, &my_charset_bin);
+  str->set((ulonglong) value, collation.collation);
   return str;
 }
 
@@ -2340,7 +2408,7 @@ double Item_decimal::val_real()
 
 String *Item_decimal::val_str(String *result)
 {
-  result->set_charset(&my_charset_bin);
+  result->set_charset(&my_charset_numeric);
   my_decimal2string(E_DEC_FATAL_ERROR, &decimal_value, 0, 0, 0, result);
   return result;
 }
@@ -4467,17 +4535,33 @@ bool Item_field::fix_fields(THD *thd, Item **reference)
               It's not an Item_field in the select list so we must make a new
               Item_ref to point to the Item in the select list and replace the
               Item_field created by the parser with the new Item_ref.
+
+              NOTE: If we are fixing an alias reference inside ORDER/GROUP BY
+              item tree, then we use new Item_ref as an intermediate value
+              to resolve referenced item only.
+              In this case the new Item_ref item is unused.
             */
             Item_ref *rf= new Item_ref(context, db_name,table_name,field_name);
             if (!rf)
               return 1;
-            thd->change_item_tree(reference, rf);
+
+            bool save_group_fix_field= thd->lex->current_select->group_fix_field;
             /*
-              Because Item_ref never substitutes itself with other items 
-              in Item_ref::fix_fields(), we can safely use the original 
-              pointer to it even after fix_fields()
-             */
-            return rf->fix_fields(thd, reference) ||  rf->check_cols(1);
+              No need for recursive resolving of aliases.
+            */
+            thd->lex->current_select->group_fix_field= 0;
+
+            bool ret= rf->fix_fields(thd, (Item **) &rf) || rf->check_cols(1);
+            thd->lex->current_select->group_fix_field= save_group_fix_field;
+            if (ret)
+              return TRUE;
+
+            if (save_group_fix_field && alias_name_used)
+              thd->change_item_tree(reference, *rf->ref);
+            else
+              thd->change_item_tree(reference, rf);
+
+            return FALSE;
           }
         }
       }
@@ -4866,7 +4950,7 @@ void Item::init_make_field(Send_field *tmp_field,
   tmp_field->col_name=		name;
   tmp_field->charsetnr=         collation.collation->number;
   tmp_field->flags=             (maybe_null ? 0 : NOT_NULL_FLAG) | 
-                                (my_binary_compare(collation.collation) ?
+                                (my_binary_compare(charset_for_protocol()) ?
                                  BINARY_FLAG : 0);
   tmp_field->type=              field_type_arg;
   tmp_field->length=max_length;
@@ -5312,7 +5396,7 @@ int Item::save_in_field(Field *field, bool no_conversions)
     field->set_notnull();
     error=field->store(nr, unsigned_flag);
   }
-  return error ? error : (field->table->in_use->is_error() ? 2 : 0);
+  return error ? error : (field->table->in_use->is_error() ? 1 : 0);
 }
 
 
@@ -6704,7 +6788,8 @@ int Item_default_value::save_in_field(Field *field_arg, bool no_conversions)
 {
   if (!arg)
   {
-    if (field_arg->flags & NO_DEFAULT_VALUE_FLAG)
+    if (field_arg->flags & NO_DEFAULT_VALUE_FLAG &&
+        field_arg->real_type() != MYSQL_TYPE_ENUM)
     {
       if (field_arg->reset())
       {
diff --git a/sql/item.h b/sql/item.h
index b7e6cc6c204..9f64c0110ec 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -44,9 +44,10 @@ class Item_field;
 
 #define MY_COLL_ALLOW_SUPERSET_CONV   1
 #define MY_COLL_ALLOW_COERCIBLE_CONV  2
-#define MY_COLL_ALLOW_CONV            3
 #define MY_COLL_DISALLOW_NONE         4
-#define MY_COLL_CMP_CONV              7
+
+#define MY_COLL_ALLOW_CONV (MY_COLL_ALLOW_SUPERSET_CONV | MY_COLL_ALLOW_COERCIBLE_CONV)
+#define MY_COLL_CMP_CONV   (MY_COLL_ALLOW_CONV | MY_COLL_DISALLOW_NONE)
 
 class DTCollation {
 public:
@@ -91,6 +92,12 @@ public:
     derivation= derivation_arg;
     repertoire= repertoire_arg;
   }
+  void set_numeric()
+  {
+    collation= &my_charset_numeric;
+    derivation= DERIVATION_NUMERIC;
+    repertoire= MY_REPERTOIRE_NUMERIC;
+  }
   void set(CHARSET_INFO *collation_arg)
   {
     collation= collation_arg;
@@ -105,6 +112,7 @@ public:
   {
     switch(derivation)
     {
+      case DERIVATION_NUMERIC:   return "NUMERIC";
       case DERIVATION_IGNORABLE: return "IGNORABLE";
       case DERIVATION_COERCIBLE: return "COERCIBLE";
       case DERIVATION_IMPLICIT:  return "IMPLICIT";
@@ -690,6 +698,77 @@ public:
       If value is not null null_value flag will be reset to FALSE.
   */
   virtual String *val_str(String *str)=0;
+
+  /*
+    Returns string representation of this item in ASCII format.
+
+    SYNOPSIS
+      val_str_ascii()
+      str - similar to val_str();
+
+    NOTE
+      This method is introduced for performance optimization purposes.
+
+      1. val_str() result of some Items in string context
+      depends on @@character_set_results.
+      @@character_set_results can be set to a "real multibyte" character
+      set like UCS2, UTF16, UTF32. (We'll use only UTF32 in the examples
+      below for convenience.)
+
+      So the default string result of such functions
+      in these circumstances is real multi-byte character set, like UTF32.
+
+      For example, all numbers in string context
+      return result in @@character_set_results:
+
+      SELECT CONCAT(20010101); -> UTF32
+
+      We do sprintf() first (to get ASCII representation)
+      and then convert to UTF32;
+      
+      So these kind "data sources" can use ASCII representation
+      internally, but return multi-byte data only because
+      @@character_set_results wants so.
+      Therefore, conversion from ASCII to UTF32 is applied internally.
+
+
+      2. Some other functions need in fact ASCII input.
+
+      For example,
+        inet_aton(), GeometryFromText(), Convert_TZ(), GET_FORMAT().
+
+      Similar, fields of certain type, like DATE, TIME,
+      when you insert string data into them, expect in fact ASCII input.
+      If they get non-ASCII input, for example UTF32, they
+      convert input from UTF32 to ASCII, and then use ASCII
+      representation to do further processing.
+
+
+      3. Now imagine we pass result of a data source of the first type
+         to a data destination of the second type.
+
+      What happens:
+        a. data source converts data from ASCII to UTF32, because
+           @@character_set_results wants so and passes the result to
+           data destination.
+        b. data destination gets UTF32 string.
+        c. data destination converts UTF32 string to ASCII,
+           because it needs ASCII representation to be able to handle data
+           correctly.
+
+      As a result we get two steps of unnecessary conversion:
+      From ASCII to UTF32, then from UTF32 to ASCII.
+
+      A better way to handle these situations is to pass ASCII
+      representation directly from the source to the destination.
+
+      This is why val_str_ascii() introduced.
+
+    RETURN
+      Similar to val_str()
+  */
+  virtual String *val_str_ascii(String *str);
+  
   /*
     Return decimal representation of item with fixed point.
 
@@ -864,6 +943,16 @@ public:
   static CHARSET_INFO *default_charset();
   virtual CHARSET_INFO *compare_collation() { return NULL; }
 
+  /*
+    For backward compatibility, to make numeric
+    data types return "binary" charset in client-side metadata.
+  */
+  virtual CHARSET_INFO *charset_for_protocol(void) const
+  {
+    return result_type() == STRING_RESULT ? collation.collation :
+                                            &my_charset_bin;
+  };
+
   virtual bool walk(Item_processor processor, bool walk_subquery, uchar *arg)
   {
     return (this->*processor)(arg);
@@ -918,6 +1007,7 @@ public:
   virtual bool change_context_processor(uchar *context) { return 0; }
   virtual bool reset_query_id_processor(uchar *query_id_arg) { return 0; }
   virtual bool is_expensive_processor(uchar *arg) { return 0; }
+  virtual bool find_item_processor(uchar *arg) { return this == (void *) arg; }
   virtual bool register_field_in_read_map(uchar *arg) { return 0; }
 
   virtual bool cache_const_expr_analyzer(uchar **arg);
@@ -1069,6 +1159,20 @@ public:
     { return Field::GEOM_GEOMETRY; };
   String *check_well_formed_result(String *str, bool send_error= 0);
   bool eq_by_collation(Item *item, bool binary_cmp, CHARSET_INFO *cs); 
+  uint32 max_char_length() const
+  { return max_length / collation.collation->mbmaxlen; }
+  void fix_length_and_charset(uint32 max_char_length_arg, CHARSET_INFO *cs)
+  {
+    max_length= max_char_length_arg * cs->mbmaxlen;
+    collation.collation= cs;
+  }
+  void fix_char_length(uint32 max_char_length_arg)
+  { max_length= max_char_length_arg * collation.collation->mbmaxlen; }
+  void fix_length_and_charset_datetime(uint32 max_char_length_arg)
+  {
+    collation.set(&my_charset_numeric, DERIVATION_NUMERIC, MY_REPERTOIRE_ASCII);
+    fix_char_length(max_char_length_arg);
+  }
 };
 
 
@@ -1371,12 +1475,30 @@ bool agg_item_set_converter(DTCollation &coll, const char *fname,
                             Item **args, uint nargs, uint flags, int item_sep);
 bool agg_item_charsets(DTCollation &c, const char *name,
                        Item **items, uint nitems, uint flags, int item_sep);
-
+inline bool
+agg_item_charsets_for_string_result(DTCollation &c, const char *name,
+                                    Item **items, uint nitems,
+                                    int item_sep= 1)
+{
+  uint flags= MY_COLL_ALLOW_SUPERSET_CONV |
+              MY_COLL_ALLOW_COERCIBLE_CONV;
+  return agg_item_charsets(c, name, items, nitems, flags, item_sep);
+}
+inline bool
+agg_item_charsets_for_comparison(DTCollation &c, const char *name,
+                                 Item **items, uint nitems,
+                                 int item_sep= 1)
+{
+  uint flags= MY_COLL_ALLOW_SUPERSET_CONV |
+              MY_COLL_ALLOW_COERCIBLE_CONV |
+              MY_COLL_DISALLOW_NONE;
+  return agg_item_charsets(c, name, items, nitems, flags, item_sep);
+}
 
 class Item_num: public Item_basic_constant
 {
 public:
-  Item_num() {}                               /* Remove gcc warning */
+  Item_num() { collation.set_numeric(); } /* Remove gcc warning */
   virtual Item_num *neg()= 0;
   Item *safe_charset_converter(CHARSET_INFO *tocs);
   bool check_partition_func_processor(uchar *int_arg) { return FALSE;}
@@ -1561,6 +1683,8 @@ public:
     DBUG_ASSERT(field_type() == MYSQL_TYPE_GEOMETRY);
     return field->get_geometry_type();
   }
+  CHARSET_INFO *charset_for_protocol(void) const
+  { return field->charset_for_protocol(); }
   friend class Item_default_value;
   friend class Item_insert_value;
   friend class st_select_lex_unit;
@@ -2316,7 +2440,10 @@ public:
     return ref ? (*ref)->real_item() : this;
   }
   bool walk(Item_processor processor, bool walk_subquery, uchar *arg)
-  { return (*ref)->walk(processor, walk_subquery, arg); }
+  {
+    return (*ref)->walk(processor, walk_subquery, arg) ||
+           (this->*processor)(arg);
+  }
   virtual void print(String *str, enum_query_type query_type);
   bool result_as_longlong()
   {
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index ed465cbe280..08383325580 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -483,7 +483,7 @@ void Item_bool_func2::fix_length_and_dec()
   DTCollation coll;
   if (args[0]->result_type() == STRING_RESULT &&
       args[1]->result_type() == STRING_RESULT &&
-      agg_arg_charsets(coll, args, 2, MY_COLL_CMP_CONV, 1))
+      agg_arg_charsets_for_comparison(coll, args, 2))
     return;
     
   args[0]->cmp_context= args[1]->cmp_context=
@@ -934,6 +934,7 @@ int Arg_comparator::set_cmp_func(Item_result_field *owner_arg,
     func= &Arg_comparator::compare_datetime;
     get_value_a_func= &get_datetime_value;
     get_value_b_func= &get_datetime_value;
+    cmp_collation.set(&my_charset_numeric);
     return 0;
   }
   else if (type == STRING_RESULT && (*a)->field_type() == MYSQL_TYPE_TIME &&
@@ -2173,7 +2174,7 @@ void Item_func_between::fix_length_and_dec()
   if ( agg_cmp_type(&cmp_type, args, 3))
     return;
   if (cmp_type == STRING_RESULT &&
-      agg_arg_charsets(cmp_collation, args, 3, MY_COLL_CMP_CONV, 1))
+      agg_arg_charsets_for_comparison(cmp_collation, args, 3))
    return;
 
   /*
@@ -2374,7 +2375,7 @@ Item_func_ifnull::fix_length_and_dec()
 
   switch (hybrid_type) {
   case STRING_RESULT:
-    agg_arg_charsets(collation, args, arg_count, MY_COLL_CMP_CONV, 1);
+    agg_arg_charsets_for_comparison(collation, args, arg_count);
     break;
   case DECIMAL_RESULT:
   case REAL_RESULT:
@@ -2549,12 +2550,12 @@ Item_func_if::fix_length_and_dec()
     agg_result_type(&cached_result_type, args+1, 2);
     if (cached_result_type == STRING_RESULT)
     {
-      if (agg_arg_charsets(collation, args+1, 2, MY_COLL_ALLOW_CONV, 1))
+      if (agg_arg_charsets_for_string_result(collation, args + 1, 2))
         return;
     }
     else
     {
-      collation.set(&my_charset_bin);	// Number
+      collation.set_numeric(); // Number
     }
     cached_field_type= agg_field_type(args + 1, 2);
   }
@@ -2640,7 +2641,7 @@ Item_func_nullif::fix_length_and_dec()
     unsigned_flag= args[0]->unsigned_flag;
     cached_result_type= args[0]->result_type();
     if (cached_result_type == STRING_RESULT &&
-        agg_arg_charsets(collation, args, arg_count, MY_COLL_CMP_CONV, 1))
+        agg_arg_charsets_for_comparison(collation, args, arg_count))
       return;
   }
 }
@@ -2915,9 +2916,13 @@ void Item_func_case::fix_length_and_dec()
     agg[nagg++]= args[else_expr_num];
   
   agg_result_type(&cached_result_type, agg, nagg);
-  if ((cached_result_type == STRING_RESULT) &&
-      agg_arg_charsets(collation, agg, nagg, MY_COLL_ALLOW_CONV, 1))
-    return;
+  if (cached_result_type == STRING_RESULT)
+  {
+    if (agg_arg_charsets_for_string_result(collation, agg, nagg))
+      return;
+  }
+  else
+    collation.set_numeric();
   
   cached_field_type= agg_field_type(agg, nagg);
   /*
@@ -2942,7 +2947,7 @@ void Item_func_case::fix_length_and_dec()
       {
         DBUG_ASSERT((Item_result)i != ROW_RESULT);
         if ((Item_result)i == STRING_RESULT &&
-            agg_arg_charsets(cmp_collation, agg, nagg, MY_COLL_CMP_CONV, 1))
+            agg_arg_charsets_for_comparison(cmp_collation, agg, nagg))
           return;
         if (!(cmp_items[i]=
             cmp_item::get_comparator((Item_result)i,
@@ -3105,7 +3110,7 @@ void Item_func_coalesce::fix_length_and_dec()
   case STRING_RESULT:
     count_only_length();
     decimals= NOT_FIXED_DEC;
-    agg_arg_charsets(collation, args, arg_count, MY_COLL_ALLOW_CONV, 1);
+    agg_arg_charsets_for_string_result(collation, args, arg_count);
     break;
   case DECIMAL_RESULT:
     count_decimal_length();
@@ -3750,7 +3755,7 @@ void Item_func_in::fix_length_and_dec()
   if (type_cnt == 1)
   {
     if (cmp_type == STRING_RESULT && 
-        agg_arg_charsets(cmp_collation, args, arg_count, MY_COLL_CMP_CONV, 1))
+        agg_arg_charsets_for_comparison(cmp_collation, args, arg_count))
       return;
     arg_types_compatible= TRUE;
   }
@@ -3928,8 +3933,7 @@ void Item_func_in::fix_length_and_dec()
         if (found_types & (1 << i) && !cmp_items[i])
         {
           if ((Item_result)i == STRING_RESULT &&
-              agg_arg_charsets(cmp_collation, args, arg_count,
-                               MY_COLL_CMP_CONV, 1))
+              agg_arg_charsets_for_comparison(cmp_collation, args, arg_count))
             return;
           if (!cmp_items[i] && !(cmp_items[i]=
               cmp_item::get_comparator((Item_result)i,
@@ -4758,7 +4762,7 @@ Item_func_regex::fix_fields(THD *thd, Item **ref)
   max_length= 1;
   decimals= 0;
 
-  if (agg_arg_charsets(cmp_collation, args, 2, MY_COLL_CMP_CONV, 1))
+  if (agg_arg_charsets_for_comparison(cmp_collation, args, 2))
     return TRUE;
 
   regex_lib_flags= (cmp_collation.collation->state &
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index ef4eef3276c..3232a77ac3e 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -637,6 +637,11 @@ public:
   {
     Item_func::print(str, query_type);
   }
+  void fix_length_and_dec()
+  {
+    Item_bool_func2::fix_length_and_dec();
+    fix_char_length(2); // returns "1" or "0" or "-1"
+  }
 };
 
 
@@ -1477,9 +1482,21 @@ public:
   Item_cond(THD *thd, Item_cond *item);
   Item_cond(List<Item> &nlist)
     :Item_bool_func(), list(nlist), abort_on_null(0) {}
-  bool add(Item *item) { return list.push_back(item); }
-  bool add_at_head(Item *item) { return list.push_front(item); }
-  void add_at_head(List<Item> *nlist) { list.prepand(nlist); }
+  bool add(Item *item)
+  {
+    DBUG_ASSERT(item);
+    return list.push_back(item);
+  }
+  bool add_at_head(Item *item)
+  {
+    DBUG_ASSERT(item);
+    return list.push_front(item);
+  }
+  void add_at_head(List<Item> *nlist)
+  {
+    DBUG_ASSERT(nlist->elements);
+    list.prepand(nlist);
+  }
   bool fix_fields(THD *, Item **ref);
 
   enum Type type() const { return COND_ITEM; }
diff --git a/sql/item_func.cc b/sql/item_func.cc
index e49ee4346b1..ca8f5d00bb1 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -438,13 +438,15 @@ Field *Item_func::tmp_table_field(TABLE *table)
 
   switch (result_type()) {
   case INT_RESULT:
-    if (max_length > MY_INT32_NUM_DECIMAL_DIGITS)
-      field= new Field_longlong(max_length, maybe_null, name, unsigned_flag);
+    if (max_char_length() > MY_INT32_NUM_DECIMAL_DIGITS)
+      field= new Field_longlong(max_char_length(), maybe_null, name,
+                                unsigned_flag);
     else
-      field= new Field_long(max_length, maybe_null, name, unsigned_flag);
+      field= new Field_long(max_char_length(), maybe_null, name,
+                            unsigned_flag);
     break;
   case REAL_RESULT:
-    field= new Field_double(max_length, maybe_null, name, decimals);
+    field= new Field_double(max_char_length(), maybe_null, name, decimals);
     break;
   case STRING_RESULT:
     return make_string_field(table);
@@ -485,7 +487,7 @@ String *Item_real_func::val_str(String *str)
   double nr= val_real();
   if (null_value)
     return 0; /* purecov: inspected */
-  str->set_real(nr,decimals, &my_charset_bin);
+  str->set_real(nr, decimals, collation.collation);
   return str;
 }
 
@@ -624,7 +626,7 @@ String *Item_int_func::val_str(String *str)
   longlong nr=val_int();
   if (null_value)
     return 0;
-  str->set_int(nr, unsigned_flag, &my_charset_bin);
+  str->set_int(nr, unsigned_flag, collation.collation);
   return str;
 }
 
@@ -744,6 +746,7 @@ String *Item_func_numhybrid::val_str(String *str)
     if (!(val= decimal_op(&decimal_value)))
       return 0;                                 // null is set
     my_decimal_round(E_DEC_FATAL_ERROR, val, decimals, FALSE, val);
+    str->set_charset(collation.collation);
     my_decimal2string(E_DEC_FATAL_ERROR, val, 0, 0, 0, str);
     break;
   }
@@ -752,7 +755,7 @@ String *Item_func_numhybrid::val_str(String *str)
     longlong nr= int_op();
     if (null_value)
       return 0; /* purecov: inspected */
-    str->set_int(nr, unsigned_flag, &my_charset_bin);
+    str->set_int(nr, unsigned_flag, collation.collation);
     break;
   }
   case REAL_RESULT:
@@ -760,7 +763,7 @@ String *Item_func_numhybrid::val_str(String *str)
     double nr= real_op();
     if (null_value)
       return 0; /* purecov: inspected */
-    str->set_real(nr,decimals,&my_charset_bin);
+    str->set_real(nr, decimals, collation.collation);
     break;
   }
   case STRING_RESULT:
@@ -895,6 +898,7 @@ longlong Item_func_signed::val_int_from_str(int *error)
   uint32 length;
   String tmp(buff,sizeof(buff), &my_charset_bin), *res;
   longlong value;
+  CHARSET_INFO *cs;
 
   /*
     For a string result, we must first get the string and then convert it
@@ -910,9 +914,10 @@ longlong Item_func_signed::val_int_from_str(int *error)
   null_value= 0;
   start= (char *)res->ptr();
   length= res->length();
+  cs= res->charset();
 
   end= start + length;
-  value= my_strtoll10(start, &end, error);
+  value= cs->cset->strtoll10(cs, start, &end, error);
   if (*error > 0 || end != start+ length)
   {
     char err_buff[128];
@@ -2263,7 +2268,7 @@ void Item_func_min_max::fix_length_and_dec()
   }
   if (cmp_type == STRING_RESULT)
   {
-    agg_arg_charsets(collation, args, arg_count, MY_COLL_CMP_CONV, 1);
+    agg_arg_charsets_for_comparison(collation, args, arg_count);
     if (datetime_found)
     {
       thd= current_thd;
@@ -2271,9 +2276,13 @@ void Item_func_min_max::fix_length_and_dec()
     }
   }
   else if ((cmp_type == DECIMAL_RESULT) || (cmp_type == INT_RESULT))
-    max_length= my_decimal_precision_to_length_no_truncation(max_int_part +
-                                                             decimals, decimals,
-                                                             unsigned_flag);
+  {
+    collation.set_numeric();
+    fix_char_length(my_decimal_precision_to_length_no_truncation(max_int_part +
+                                                                 decimals,
+                                                                 decimals,
+                                                                 unsigned_flag));
+  }
   cached_field_type= agg_field_type(args, arg_count);
 }
 
@@ -2343,7 +2352,7 @@ String *Item_func_min_max::val_str(String *str)
     longlong nr=val_int();
     if (null_value)
       return 0;
-    str->set_int(nr, unsigned_flag, &my_charset_bin);
+    str->set_int(nr, unsigned_flag, collation.collation);
     return str;
   }
   case DECIMAL_RESULT:
@@ -2359,7 +2368,7 @@ String *Item_func_min_max::val_str(String *str)
     double nr= val_real();
     if (null_value)
       return 0; /* purecov: inspected */
-    str->set_real(nr,decimals,&my_charset_bin);
+    str->set_real(nr, decimals, collation.collation);
     return str;
   }
   case STRING_RESULT:
@@ -2530,7 +2539,7 @@ longlong Item_func_coercibility::val_int()
 void Item_func_locate::fix_length_and_dec()
 {
   max_length= MY_INT32_NUM_DECIMAL_DIGITS;
-  agg_arg_charsets(cmp_collation, args, 2, MY_COLL_CMP_CONV, 1);
+  agg_arg_charsets_for_comparison(cmp_collation, args, 2);
 }
 
 
@@ -2654,7 +2663,7 @@ void Item_func_field::fix_length_and_dec()
   for (uint i=1; i < arg_count ; i++)
     cmp_type= item_cmp_type(cmp_type, args[i]->result_type());
   if (cmp_type == STRING_RESULT)
-    agg_arg_charsets(cmp_collation, args, arg_count, MY_COLL_CMP_CONV, 1);
+    agg_arg_charsets_for_comparison(cmp_collation, args, arg_count);
 }
 
 
@@ -2721,7 +2730,7 @@ void Item_func_find_in_set::fix_length_and_dec()
       }
     }
   }
-  agg_arg_charsets(cmp_collation, args, 2, MY_COLL_CMP_CONV, 1);
+  agg_arg_charsets_for_comparison(cmp_collation, args, 2);
 }
 
 static const char separator=',';
@@ -3960,7 +3969,9 @@ bool Item_func_set_user_var::fix_fields(THD *thd, Item **ref)
   */
   null_item= (args[0]->type() == NULL_ITEM);
   if (!entry->collation.collation || !null_item)
-    entry->collation.set(args[0]->collation.collation, DERIVATION_IMPLICIT);
+    entry->collation.set(args[0]->collation.derivation == DERIVATION_NUMERIC ?
+                         default_charset() : args[0]->collation.collation,
+                         DERIVATION_IMPLICIT);
   collation.set(entry->collation.collation, DERIVATION_IMPLICIT);
   cached_result_type= args[0]->result_type();
   return FALSE;
@@ -3971,9 +3982,15 @@ void
 Item_func_set_user_var::fix_length_and_dec()
 {
   maybe_null=args[0]->maybe_null;
-  max_length=args[0]->max_length;
   decimals=args[0]->decimals;
-  collation.set(args[0]->collation.collation, DERIVATION_IMPLICIT);
+  collation.set(DERIVATION_IMPLICIT);
+  if (args[0]->collation.derivation == DERIVATION_NUMERIC)
+    fix_length_and_charset(args[0]->max_char_length(), default_charset());
+  else
+  {
+    fix_length_and_charset(args[0]->max_char_length(),
+                           args[0]->collation.collation);
+  }
 }
 
 
@@ -4168,16 +4185,16 @@ String *user_var_entry::val_str(my_bool *null_value, String *str,
 
   switch (type) {
   case REAL_RESULT:
-    str->set_real(*(double*) value, decimals, &my_charset_bin);
+    str->set_real(*(double*) value, decimals, collation.collation);
     break;
   case INT_RESULT:
     if (!unsigned_flag)
-      str->set(*(longlong*) value, &my_charset_bin);
+      str->set(*(longlong*) value, collation.collation);
     else
-      str->set(*(ulonglong*) value, &my_charset_bin);
+      str->set(*(ulonglong*) value, collation.collation);
     break;
   case DECIMAL_RESULT:
-    my_decimal2string(E_DEC_FATAL_ERROR, (my_decimal *)value, 0, 0, 0, str);
+    str_set_decimal((my_decimal *) value, str, collation.collation);
     break;
   case STRING_RESULT:
     if (str->copy(value, length, collation.collation))
@@ -4335,13 +4352,13 @@ Item_func_set_user_var::update()
   case REAL_RESULT:
   {
     res= update_hash((void*) &save_result.vreal,sizeof(save_result.vreal),
-		     REAL_RESULT, &my_charset_bin, DERIVATION_IMPLICIT, 0);
+		     REAL_RESULT, default_charset(), DERIVATION_IMPLICIT, 0);
     break;
   }
   case INT_RESULT:
   {
     res= update_hash((void*) &save_result.vint, sizeof(save_result.vint),
-                     INT_RESULT, &my_charset_bin, DERIVATION_IMPLICIT,
+                     INT_RESULT, default_charset(), DERIVATION_IMPLICIT,
                      unsigned_flag);
     break;
   }
@@ -4365,7 +4382,7 @@ Item_func_set_user_var::update()
     else
       res= update_hash((void*) save_result.vdec,
                        sizeof(my_decimal), DECIMAL_RESULT,
-                       &my_charset_bin, DERIVATION_IMPLICIT, 0);
+                       default_charset(), DERIVATION_IMPLICIT, 0);
     break;
   }
   case ROW_RESULT:
@@ -4798,17 +4815,17 @@ void Item_func_get_user_var::fix_length_and_dec()
     collation.set(var_entry->collation);
     switch(m_cached_result_type) {
     case REAL_RESULT:
-      max_length= DBL_DIG + 8;
+      fix_char_length(DBL_DIG + 8);
       break;
     case INT_RESULT:
-      max_length= MAX_BIGINT_WIDTH;
+      fix_char_length(MAX_BIGINT_WIDTH);
       decimals=0;
       break;
     case STRING_RESULT:
       max_length= MAX_BLOB_WIDTH;
       break;
     case DECIMAL_RESULT:
-      max_length= DECIMAL_MAX_STR_LENGTH;
+      fix_char_length(DECIMAL_MAX_STR_LENGTH);
       decimals= DECIMAL_MAX_SCALE;
       break;
     case ROW_RESULT:                            // Keep compiler happy
@@ -4994,12 +5011,14 @@ void Item_func_get_system_var::fix_length_and_dec()
     case SHOW_INT:
     case SHOW_HA_ROWS:
       unsigned_flag= TRUE;
-      max_length= MY_INT64_NUM_DECIMAL_DIGITS;
+      collation.set_numeric();
+      fix_char_length(MY_INT64_NUM_DECIMAL_DIGITS);
       decimals=0;
       break;
     case SHOW_LONGLONG:
       unsigned_flag= TRUE;
-      max_length= MY_INT64_NUM_DECIMAL_DIGITS;
+      collation.set_numeric();
+      fix_char_length(MY_INT64_NUM_DECIMAL_DIGITS);
       decimals=0;
       break;
     case SHOW_CHAR:
@@ -5033,13 +5052,15 @@ void Item_func_get_system_var::fix_length_and_dec()
     case SHOW_BOOL:
     case SHOW_MY_BOOL:
       unsigned_flag= FALSE;
-      max_length= 1;
+      collation.set_numeric();
+      fix_char_length(1);
       decimals=0;
       break;
     case SHOW_DOUBLE:
       unsigned_flag= FALSE;
       decimals= 6;
-      max_length= DBL_DIG + 6;
+      collation.set_numeric();
+      fix_char_length(DBL_DIG + 6);
       break;
     default:
       my_error(ER_VAR_CANT_BE_READ, MYF(0), var->name.str);
@@ -5401,8 +5422,8 @@ longlong Item_func_inet_aton::val_int()
   char buff[36];
   int dot_count= 0;
 
-  String *s,tmp(buff,sizeof(buff),&my_charset_bin);
-  if (!(s = args[0]->val_str(&tmp)))		// If null value
+  String *s, tmp(buff, sizeof(buff), &my_charset_latin1);
+  if (!(s = args[0]->val_str_ascii(&tmp)))       // If null value
     goto err;
   null_value=0;
 
@@ -5410,7 +5431,7 @@ longlong Item_func_inet_aton::val_int()
   while (p < end)
   {
     c = *p++;
-    int digit = (int) (c - '0');		// Assume ascii
+    int digit = (int) (c - '0');
     if (digit >= 0 && digit <= 9)
     {
       if ((byte_result = byte_result * 10 + digit) > 255)
@@ -5560,8 +5581,8 @@ bool Item_func_match::fix_fields(THD *thd, Item **ref)
     return 1;
   }
   table->fulltext_searched=1;
-  return agg_arg_collations_for_comparison(cmp_collation,
-                                           args+1, arg_count-1, 0);
+  return agg_item_collations_for_comparison(cmp_collation, func_name(),
+                                            args+1, arg_count-1, 0);
 }
 
 bool Item_func_match::fix_index()
diff --git a/sql/item_func.h b/sql/item_func.h
index 6bfdae8d56d..fc7f8708a45 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -160,22 +160,24 @@ public:
 
   my_decimal *val_decimal(my_decimal *);
 
-  bool agg_arg_collations(DTCollation &c, Item **items, uint nitems,
-                          uint flags)
+  bool agg_arg_charsets(DTCollation &c, Item **items, uint nitems,
+                        uint flags, int item_sep)
   {
-    return agg_item_collations(c, func_name(), items, nitems, flags, 1);
+    return agg_item_charsets(c, func_name(), items, nitems, flags, item_sep);
   }
-  bool agg_arg_collations_for_comparison(DTCollation &c,
-                                         Item **items, uint nitems,
-                                         uint flags)
+  bool agg_arg_charsets_for_string_result(DTCollation &c,
+                                          Item **items, uint nitems,
+                                          int item_sep= 1)
   {
-    return agg_item_collations_for_comparison(c, func_name(),
-                                              items, nitems, flags);
+    return agg_item_charsets_for_string_result(c, func_name(),
+                                               items, nitems, item_sep);
   }
-  bool agg_arg_charsets(DTCollation &c, Item **items, uint nitems,
-                        uint flags, int item_sep)
+  bool agg_arg_charsets_for_comparison(DTCollation &c,
+                                       Item **items, uint nitems,
+                                       int item_sep= 1)
   {
-    return agg_item_charsets(c, func_name(), items, nitems, flags, item_sep);
+    return agg_item_charsets_for_comparison(c, func_name(),
+                                            items, nitems, item_sep);
   }
   bool walk(Item_processor processor, bool walk_subquery, uchar *arg);
   Item *transform(Item_transformer transformer, uchar *arg);
@@ -226,10 +228,10 @@ public:
 class Item_real_func :public Item_func
 {
 public:
-  Item_real_func() :Item_func() {}
-  Item_real_func(Item *a) :Item_func(a) {}
-  Item_real_func(Item *a,Item *b) :Item_func(a,b) {}
-  Item_real_func(List<Item> &list) :Item_func(list) {}
+  Item_real_func() :Item_func() { collation.set_numeric(); }
+  Item_real_func(Item *a) :Item_func(a) { collation.set_numeric(); }
+  Item_real_func(Item *a,Item *b) :Item_func(a,b) { collation.set_numeric(); }
+  Item_real_func(List<Item> &list) :Item_func(list) { collation.set_numeric(); }
   String *val_str(String*str);
   my_decimal *val_decimal(my_decimal *decimal_value);
   longlong val_int()
@@ -246,13 +248,13 @@ protected:
   Item_result hybrid_type;
 public:
   Item_func_numhybrid(Item *a) :Item_func(a), hybrid_type(REAL_RESULT)
-  {}
+  { collation.set_numeric(); }
   Item_func_numhybrid(Item *a,Item *b)
     :Item_func(a,b), hybrid_type(REAL_RESULT)
-  {}
+  { collation.set_numeric(); }
   Item_func_numhybrid(List<Item> &list)
     :Item_func(list), hybrid_type(REAL_RESULT)
-  {}
+  { collation.set_numeric(); }
 
   enum Item_result result_type () const { return hybrid_type; }
   void fix_length_and_dec();
@@ -335,13 +337,18 @@ class Item_num_op :public Item_func_numhybrid
 class Item_int_func :public Item_func
 {
 public:
-  Item_int_func() :Item_func() { max_length= 21; }
-  Item_int_func(Item *a) :Item_func(a) { max_length= 21; }
-  Item_int_func(Item *a,Item *b) :Item_func(a,b) { max_length= 21; }
+  Item_int_func() :Item_func()
+  { collation.set_numeric(); fix_char_length(21); }
+  Item_int_func(Item *a) :Item_func(a)
+  { collation.set_numeric(); fix_char_length(21); }
+  Item_int_func(Item *a,Item *b) :Item_func(a,b)
+  { collation.set_numeric(); fix_char_length(21); }
   Item_int_func(Item *a,Item *b,Item *c) :Item_func(a,b,c)
-  { max_length= 21; }
-  Item_int_func(List<Item> &list) :Item_func(list) { max_length= 21; }
-  Item_int_func(THD *thd, Item_int_func *item) :Item_func(thd, item) {}
+  { collation.set_numeric(); fix_char_length(21); }
+  Item_int_func(List<Item> &list) :Item_func(list)
+  { collation.set_numeric(); fix_char_length(21); }
+  Item_int_func(THD *thd, Item_int_func *item) :Item_func(thd, item)
+  { collation.set_numeric(); }
   double val_real();
   String *val_str(String*str);
   enum Item_result result_type () const { return INT_RESULT; }
@@ -370,7 +377,7 @@ public:
   longlong val_int();
   longlong val_int_from_str(int *error);
   void fix_length_and_dec()
-  { max_length=args[0]->max_length; unsigned_flag=0; }
+  { fix_char_length(args[0]->max_char_length()); unsigned_flag=0; }
   virtual void print(String *str, enum_query_type query_type);
   uint decimal_precision() const { return args[0]->decimal_precision(); }
 };
@@ -383,7 +390,8 @@ public:
   const char *func_name() const { return "cast_as_unsigned"; }
   void fix_length_and_dec()
   {
-    max_length= min(args[0]->max_length, DECIMAL_MAX_PRECISION + 2);
+    fix_char_length(min(args[0]->max_char_length(),
+                        DECIMAL_MAX_PRECISION + 2));
     unsigned_flag=1;
   }
   longlong val_int();
@@ -398,8 +406,9 @@ public:
   Item_decimal_typecast(Item *a, int len, int dec) :Item_func(a)
   {
     decimals= dec;
-    max_length= my_decimal_precision_to_length_no_truncation(len, dec,
-                                                             unsigned_flag);
+    collation.set_numeric();
+    fix_char_length(my_decimal_precision_to_length_no_truncation(len, dec,
+                                                                 unsigned_flag));
   }
   String *val_str(String *str);
   double val_real();
diff --git a/sql/item_geofunc.cc b/sql/item_geofunc.cc
index 8c38cb2a859..c5554265fbe 100644
--- a/sql/item_geofunc.cc
+++ b/sql/item_geofunc.cc
@@ -52,7 +52,7 @@ String *Item_func_geometry_from_text::val_str(String *str)
   DBUG_ASSERT(fixed == 1);
   Geometry_buffer buffer;
   String arg_val;
-  String *wkt= args[0]->val_str(&arg_val);
+  String *wkt= args[0]->val_str_ascii(&arg_val);
 
   if ((null_value= args[0]->null_value))
     return 0;
@@ -110,7 +110,7 @@ String *Item_func_geometry_from_wkb::val_str(String *str)
 }
 
 
-String *Item_func_as_wkt::val_str(String *str)
+String *Item_func_as_wkt::val_str_ascii(String *str)
 {
   DBUG_ASSERT(fixed == 1);
   String arg_val;
@@ -134,6 +134,7 @@ String *Item_func_as_wkt::val_str(String *str)
 
 void Item_func_as_wkt::fix_length_and_dec()
 {
+  collation.set(default_charset(), DERIVATION_COERCIBLE, MY_REPERTOIRE_ASCII);
   max_length=MAX_BLOB_WIDTH;
   maybe_null= 1;
 }
@@ -157,7 +158,7 @@ String *Item_func_as_wkb::val_str(String *str)
 }
 
 
-String *Item_func_geometry_type::val_str(String *str)
+String *Item_func_geometry_type::val_str_ascii(String *str)
 {
   DBUG_ASSERT(fixed == 1);
   String *swkb= args[0]->val_str(str);
diff --git a/sql/item_geofunc.h b/sql/item_geofunc.h
index 9a55ea7d5b1..25755de1e2c 100644
--- a/sql/item_geofunc.h
+++ b/sql/item_geofunc.h
@@ -57,12 +57,12 @@ public:
   String *val_str(String *);
 };
 
-class Item_func_as_wkt: public Item_str_func
+class Item_func_as_wkt: public Item_str_ascii_func
 {
 public:
-  Item_func_as_wkt(Item *a): Item_str_func(a) {}
+  Item_func_as_wkt(Item *a): Item_str_ascii_func(a) {}
   const char *func_name() const { return "astext"; }
-  String *val_str(String *);
+  String *val_str_ascii(String *);
   void fix_length_and_dec();
 };
 
@@ -75,16 +75,17 @@ public:
   enum_field_types field_type() const  { return MYSQL_TYPE_BLOB; }
 };
 
-class Item_func_geometry_type: public Item_str_func
+class Item_func_geometry_type: public Item_str_ascii_func
 {
 public:
-  Item_func_geometry_type(Item *a): Item_str_func(a) {}
-  String *val_str(String *);
+  Item_func_geometry_type(Item *a): Item_str_ascii_func(a) {}
+  String *val_str_ascii(String *);
   const char *func_name() const { return "geometrytype"; }
   void fix_length_and_dec() 
   {
-     max_length=20; // "GeometryCollection" is the most long
-     maybe_null= 1;
+    // "GeometryCollection" is the longest
+    fix_length_and_charset(20, default_charset());
+    maybe_null= 1;
   };
 };
 
diff --git a/sql/item_row.cc b/sql/item_row.cc
index 28de03bf049..29b37eb2bc0 100644
--- a/sql/item_row.cc
+++ b/sql/item_row.cc
@@ -71,7 +71,12 @@ bool Item_row::fix_fields(THD *thd, Item **ref)
     Item *item= *arg;
     used_tables_cache |= item->used_tables();
     const_item_cache&= item->const_item() && !with_null;
-    if (const_item_cache)
+    /*
+      Some subqueries transformations aren't done in the view_prepare_mode thus
+      is_null() will fail. So we skip is_null() calculation for CREATE VIEW as
+      not necessary.
+    */
+    if (const_item_cache && !thd->lex->view_prepare_mode)
     {
       if (item->cols() > 1)
 	with_null|= item->null_inside();
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index c33e0f4c6fb..dab29bb58c4 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -41,6 +41,38 @@ C_MODE_END
 
 String my_empty_string("",default_charset_info);
 
+/*
+  For the Items which have only val_str_ascii() method
+  and don't have their own "native" val_str(),
+  we provide a "wrapper" method to convert from ASCII
+  to Item character set when it's necessary.
+  Conversion happens only in case of "tricky" Item character set (e.g. UCS2).
+  Normally conversion does not happen, and val_str_ascii() is immediately
+  returned instead.
+*/
+String *Item_str_ascii_func::val_str(String *str)
+{
+  DBUG_ASSERT(fixed == 1);
+
+  if (!(collation.collation->state & MY_CS_NONASCII))
+    return val_str_ascii(str);
+  
+  DBUG_ASSERT(str != &ascii_buf);
+  
+  uint errors;
+  String *res= val_str_ascii(&ascii_buf);
+  if (!res)
+    return 0;
+  
+  if ((null_value= str->copy(res->ptr(), res->length(),
+                             &my_charset_latin1, collation.collation,
+                             &errors)))
+    return 0;
+  
+  return str;
+}
+
+
 
 /*
   Convert an array of bytes to a hexadecimal representation.
@@ -112,7 +144,7 @@ longlong Item_str_func::val_int()
 }
 
 
-String *Item_func_md5::val_str(String *str)
+String *Item_func_md5::val_str_ascii(String *str)
 {
   DBUG_ASSERT(fixed == 1);
   String * sptr= args[0]->val_str(str);
@@ -139,7 +171,6 @@ String *Item_func_md5::val_str(String *str)
 
 void Item_func_md5::fix_length_and_dec()
 {
-  max_length=32;
   /*
     The MD5() function treats its parameter as being a case sensitive. Thus
     we set binary collation on it so different instances of MD5() will be
@@ -148,10 +179,11 @@ void Item_func_md5::fix_length_and_dec()
   args[0]->collation.set(
       get_charset_by_csname(args[0]->collation.collation->csname,
                             MY_CS_BINSORT,MYF(0)), DERIVATION_COERCIBLE);
+  fix_length_and_charset(32, default_charset());
 }
 
 
-String *Item_func_sha::val_str(String *str)
+String *Item_func_sha::val_str_ascii(String *str)
 {
   DBUG_ASSERT(fixed == 1);
   String * sptr= args[0]->val_str(str);
@@ -181,7 +213,6 @@ String *Item_func_sha::val_str(String *str)
 
 void Item_func_sha::fix_length_and_dec()
 {
-  max_length=SHA1_HASH_SIZE*2; // size of hex representation of hash
   /*
     The SHA() function treats its parameter as being a case sensitive. Thus
     we set binary collation on it so different instances of MD5() will be
@@ -190,6 +221,8 @@ void Item_func_sha::fix_length_and_dec()
   args[0]->collation.set(
       get_charset_by_csname(args[0]->collation.collation->csname,
                             MY_CS_BINSORT,MYF(0)), DERIVATION_COERCIBLE);
+  // size of hex representation of hash
+  fix_length_and_charset(SHA1_HASH_SIZE * 2, default_charset());
 }
 
 
@@ -414,7 +447,7 @@ void Item_func_concat::fix_length_and_dec()
 {
   ulonglong max_result_length= 0;
 
-  if (agg_arg_charsets(collation, args, arg_count, MY_COLL_ALLOW_CONV, 1))
+  if (agg_arg_charsets_for_string_result(collation, args, arg_count))
     return;
 
   for (uint i=0 ; i < arg_count ; i++)
@@ -772,7 +805,7 @@ void Item_func_concat_ws::fix_length_and_dec()
 {
   ulonglong max_result_length;
 
-  if (agg_arg_charsets(collation, args, arg_count, MY_COLL_ALLOW_CONV, 1))
+  if (agg_arg_charsets_for_string_result(collation, args, arg_count))
     return;
 
   /*
@@ -843,8 +876,8 @@ String *Item_func_reverse::val_str(String *str)
 
 void Item_func_reverse::fix_length_and_dec()
 {
-  collation.set(args[0]->collation);
-  max_length = args[0]->max_length;
+  agg_arg_charsets_for_string_result(collation, args, 1);
+  fix_char_length(args[0]->max_char_length());
 }
 
 /**
@@ -987,7 +1020,7 @@ void Item_func_replace::fix_length_and_dec()
   }
   max_length= (ulong) max_result_length;
   
-  if (agg_arg_charsets(collation, args, 3, MY_COLL_CMP_CONV, 1))
+  if (agg_arg_charsets_for_comparison(collation, args, 3))
     return;
 }
 
@@ -1046,7 +1079,7 @@ void Item_func_insert::fix_length_and_dec()
   ulonglong max_result_length;
 
   // Handle character set for args[0] and args[3].
-  if (agg_arg_charsets(collation, &args[0], 2, MY_COLL_ALLOW_CONV, 3))
+  if (agg_arg_charsets_for_string_result(collation, args, 2, 3))
     return;
   max_result_length= ((ulonglong) args[0]->max_length+
                       (ulonglong) args[3]->max_length);
@@ -1094,7 +1127,7 @@ String *Item_str_conv::val_str(String *str)
 
 void Item_func_lcase::fix_length_and_dec()
 {
-  collation.set(args[0]->collation);
+  agg_arg_charsets_for_string_result(collation, args, 1);
   multiply= collation.collation->casedn_multiply;
   converter= collation.collation->cset->casedn;
   max_length= args[0]->max_length * multiply;
@@ -1102,7 +1135,7 @@ void Item_func_lcase::fix_length_and_dec()
 
 void Item_func_ucase::fix_length_and_dec()
 {
-  collation.set(args[0]->collation);
+  agg_arg_charsets_for_string_result(collation, args, 1);
   multiply= collation.collation->caseup_multiply;
   converter= collation.collation->cset->caseup;
   max_length= args[0]->max_length * multiply;
@@ -1150,7 +1183,7 @@ void Item_str_func::left_right_max_length()
 
 void Item_func_left::fix_length_and_dec()
 {
-  collation.set(args[0]->collation);
+  agg_arg_charsets_for_string_result(collation, args, 1);
   left_right_max_length();
 }
 
@@ -1183,7 +1216,7 @@ String *Item_func_right::val_str(String *str)
 
 void Item_func_right::fix_length_and_dec()
 {
-  collation.set(args[0]->collation);
+  agg_arg_charsets_for_string_result(collation, args, 1);
   left_right_max_length();
 }
 
@@ -1239,7 +1272,7 @@ void Item_func_substr::fix_length_and_dec()
 {
   max_length=args[0]->max_length;
 
-  collation.set(args[0]->collation);
+  agg_arg_charsets_for_string_result(collation, args, 1);
   if (args[1]->const_item())
   {
     int32 start= (int32) args[1]->val_int();
@@ -1264,7 +1297,7 @@ void Item_func_substr_index::fix_length_and_dec()
 { 
   max_length= args[0]->max_length;
 
-  if (agg_arg_charsets(collation, args, 2, MY_COLL_CMP_CONV, 1))
+  if (agg_arg_charsets_for_comparison(collation, args, 2))
     return;
 }
 
@@ -1594,7 +1627,7 @@ void Item_func_trim::fix_length_and_dec()
   max_length= args[0]->max_length;
   if (arg_count == 1)
   {
-    collation.set(args[0]->collation);
+    agg_arg_charsets_for_string_result(collation, args, 1);
     remove.set_charset(collation.collation);
     remove.set_ascii(" ",1);
   }
@@ -1602,7 +1635,7 @@ void Item_func_trim::fix_length_and_dec()
   {
     // Handle character set for args[1] and args[0].
     // Note that we pass args[1] as the first item, and args[0] as the second.
-    if (agg_arg_charsets(collation, &args[1], 2, MY_COLL_CMP_CONV, -1))
+    if (agg_arg_charsets_for_comparison(collation, &args[1], 2, -1))
       return;
   }
 }
@@ -1627,7 +1660,7 @@ void Item_func_trim::print(String *str, enum_query_type query_type)
 
 /* Item_func_password */
 
-String *Item_func_password::val_str(String *str)
+String *Item_func_password::val_str_ascii(String *str)
 {
   DBUG_ASSERT(fixed == 1);
   String *res= args[0]->val_str(str); 
@@ -1636,7 +1669,7 @@ String *Item_func_password::val_str(String *str)
   if (res->length() == 0)
     return &my_empty_string;
   my_make_scrambled_password(tmp_value, res->ptr(), res->length());
-  str->set(tmp_value, SCRAMBLED_PASSWORD_CHAR_LENGTH, res->charset());
+  str->set(tmp_value, SCRAMBLED_PASSWORD_CHAR_LENGTH, &my_charset_latin1);
   return str;
 }
 
@@ -1651,7 +1684,7 @@ char *Item_func_password::alloc(THD *thd, const char *password,
 
 /* Item_func_old_password */
 
-String *Item_func_old_password::val_str(String *str)
+String *Item_func_old_password::val_str_ascii(String *str)
 {
   DBUG_ASSERT(fixed == 1);
   String *res= args[0]->val_str(str);
@@ -1660,7 +1693,7 @@ String *Item_func_old_password::val_str(String *str)
   if (res->length() == 0)
     return &my_empty_string;
   my_make_scrambled_password_323(tmp_value, res->ptr(), res->length());
-  str->set(tmp_value, SCRAMBLED_PASSWORD_CHAR_LENGTH_323, res->charset());
+  str->set(tmp_value, SCRAMBLED_PASSWORD_CHAR_LENGTH_323, &my_charset_latin1);
   return str;
 }
 
@@ -1880,7 +1913,7 @@ bool Item_func_current_user::fix_fields(THD *thd, Item **ref)
 
 void Item_func_soundex::fix_length_and_dec()
 {
-  collation.set(args[0]->collation);
+  agg_arg_charsets_for_string_result(collation, args, 1);
   max_length=args[0]->max_length;
   set_if_bigger(max_length, 4 * collation.collation->mbminlen);
   tmp_value.set_charset(collation.collation);
@@ -2186,7 +2219,7 @@ void Item_func_elt::fix_length_and_dec()
   max_length=0;
   decimals=0;
 
-  if (agg_arg_charsets(collation, args+1, arg_count-1, MY_COLL_ALLOW_CONV, 1))
+  if (agg_arg_charsets_for_string_result(collation, args + 1, arg_count - 1))
     return;
 
   for (uint i= 1 ; i < arg_count ; i++)
@@ -2253,7 +2286,7 @@ void Item_func_make_set::fix_length_and_dec()
 {
   max_length=arg_count-1;
 
-  if (agg_arg_charsets(collation, args, arg_count, MY_COLL_ALLOW_CONV, 1))
+  if (agg_arg_charsets_for_string_result(collation, args, arg_count))
     return;
   
   for (uint i=0 ; i < arg_count ; i++)
@@ -2371,17 +2404,27 @@ String *Item_func_char::val_str(String *str)
     int32 num=(int32) args[i]->val_int();
     if (!args[i]->null_value)
     {
-      char char_num= (char) num;
-      if (num&0xFF000000L) {
-        str->append((char)(num>>24));
-        goto b2;
-      } else if (num&0xFF0000L) {
-    b2:        str->append((char)(num>>16));
-        goto b1;
-      } else if (num&0xFF00L) {
-    b1:        str->append((char)(num>>8));
+      char tmp[4];
+      if (num & 0xFF000000L)
+      {
+        mi_int4store(tmp, num);
+        str->append(tmp, 4, &my_charset_bin);
+      }
+      else if (num & 0xFF0000L)
+      {
+        mi_int3store(tmp, num);
+        str->append(tmp, 3, &my_charset_bin);
+      }
+      else if (num & 0xFF00L)
+      {
+        mi_int2store(tmp, num);
+        str->append(tmp, 2, &my_charset_bin);
+      }
+      else
+      {
+        tmp[0]= (char) num;
+        str->append(tmp, 1, &my_charset_bin);
       }
-      str->append(&char_num, 1);
     }
   }
   str->realloc(str->length());			// Add end 0 (for Purify)
@@ -2413,7 +2456,7 @@ inline String* alloc_buffer(String *res,String *str,String *tmp_value,
 
 void Item_func_repeat::fix_length_and_dec()
 {
-  collation.set(args[0]->collation);
+  agg_arg_charsets_for_string_result(collation, args, 1);
   if (args[1]->const_item())
   {
     /* must be longlong to avoid truncation */
@@ -2497,7 +2540,7 @@ err:
 void Item_func_rpad::fix_length_and_dec()
 {
   // Handle character set for args[0] and args[2].
-  if (agg_arg_charsets(collation, &args[0], 2, MY_COLL_ALLOW_CONV, 2))
+  if (agg_arg_charsets_for_string_result(collation, &args[0], 2, 2))
     return;
   if (args[1]->const_item())
   {
@@ -2599,7 +2642,7 @@ String *Item_func_rpad::val_str(String *str)
 void Item_func_lpad::fix_length_and_dec()
 {
   // Handle character set for args[0] and args[2].
-  if (agg_arg_charsets(collation, &args[0], 2, MY_COLL_ALLOW_CONV, 2))
+  if (agg_arg_charsets_for_string_result(collation, &args[0], 2, 2))
     return;
   
   if (args[1]->const_item())
@@ -2769,7 +2812,7 @@ String *Item_func_conv_charset::val_str(String *str)
 void Item_func_conv_charset::fix_length_and_dec()
 {
   collation.set(conv_charset, DERIVATION_IMPLICIT);
-  max_length = args[0]->max_length*conv_charset->mbmaxlen;
+  fix_char_length(args[0]->max_char_length());
 }
 
 void Item_func_conv_charset::print(String *str, enum_query_type query_type)
@@ -2859,7 +2902,7 @@ String *Item_func_charset::val_str(String *str)
   DBUG_ASSERT(fixed == 1);
   uint dummy_errors;
 
-  CHARSET_INFO *cs= args[0]->collation.collation; 
+  CHARSET_INFO *cs= args[0]->charset_for_protocol(); 
   null_value= 0;
   str->copy(cs->csname, (uint) strlen(cs->csname),
 	    &my_charset_latin1, collation.collation, &dummy_errors);
@@ -2870,7 +2913,7 @@ String *Item_func_collation::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
   uint dummy_errors;
-  CHARSET_INFO *cs= args[0]->collation.collation; 
+  CHARSET_INFO *cs= args[0]->charset_for_protocol(); 
 
   null_value= 0;
   str->copy(cs->name, (uint) strlen(cs->name),
@@ -2904,7 +2947,7 @@ String *Item_func_hex::val_str(String *str)
     if ((null_value= args[0]->null_value))
       return 0;
     ptr= longlong2str(dec,ans,16);
-    if (str->copy(ans,(uint32) (ptr-ans),default_charset()))
+    if (str->copy(ans,(uint32) (ptr-ans), &my_charset_numeric))
       return &my_empty_string;			// End of memory
     return str;
   }
@@ -3109,11 +3152,11 @@ void Item_func_export_set::fix_length_and_dec()
 {
   uint length=max(args[1]->max_length,args[2]->max_length);
   uint sep_length=(arg_count > 3 ? args[3]->max_length : 1);
-  max_length=length*64+sep_length*63;
 
-  if (agg_arg_charsets(collation, args+1, min(4,arg_count)-1,
-                       MY_COLL_ALLOW_CONV, 1))
+  if (agg_arg_charsets_for_string_result(collation,
+                                         args + 1, min(4, arg_count) - 1))
     return;
+  fix_char_length(length * 64 + sep_length * 63);
 }
 
 String* Item_func_inet_ntoa::val_str(String* str)
@@ -3132,6 +3175,7 @@ String* Item_func_inet_ntoa::val_str(String* str)
   if ((null_value= (args[0]->null_value || n > (ulonglong) LL(4294967295))))
     return 0;					// Null value
 
+  str->set_charset(collation.collation);
   str->length(0);
   int4store(buf,n);
 
@@ -3149,11 +3193,11 @@ String* Item_func_inet_ntoa::val_str(String* str)
     num[0]=(char) n1+'0';
     num[1]=(char) n2+'0';
     num[2]=(char) c+'0';
-    uint length=(n1 ? 4 : n2 ? 3 : 2);		// Remove pre-zero
-
-    (void) str->append(num+4-length,length);
+    uint length= (n1 ? 4 : n2 ? 3 : 2);         // Remove pre-zero
+    uint dot_length= (p <= buf) ? 1 : 0;
+    (void) str->append(num + 4 - length, length - dot_length,
+                       &my_charset_latin1);
   }
-  str->length(str->length()-1);			// Remove last '.';
   return str;
 }
 
diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h
index 09a7da021c0..c9d62ac1052 100644
--- a/sql/item_strfunc.h
+++ b/sql/item_strfunc.h
@@ -41,28 +41,41 @@ public:
   bool fix_fields(THD *thd, Item **ref);
 };
 
-class Item_func_md5 :public Item_str_func
+
+
+/*
+  Functions that return values with ASCII repertoire
+*/
+class Item_str_ascii_func :public Item_str_func
+{
+  String ascii_buf;
+public:
+  Item_str_ascii_func() :Item_str_func() {}
+  Item_str_ascii_func(Item *a) :Item_str_func(a) {}
+  Item_str_ascii_func(Item *a,Item *b) :Item_str_func(a,b) {}
+  Item_str_ascii_func(Item *a,Item *b,Item *c) :Item_str_func(a,b,c) {}
+  String *val_str_convert_from_ascii(String *str, String *ascii_buf);
+  String *val_str(String *str);
+  virtual String *val_str_ascii(String *)= 0;
+};
+
+
+class Item_func_md5 :public Item_str_ascii_func
 {
   String tmp_value;
 public:
-  Item_func_md5(Item *a) :Item_str_func(a)
-  {
-    collation.set(&my_charset_bin);
-  }
-  String *val_str(String *);
+  Item_func_md5(Item *a) :Item_str_ascii_func(a) {}
+  String *val_str_ascii(String *);
   void fix_length_and_dec();
   const char *func_name() const { return "md5"; }
 };
 
 
-class Item_func_sha :public Item_str_func
+class Item_func_sha :public Item_str_ascii_func
 {
 public:
-  Item_func_sha(Item *a) :Item_str_func(a)
-  {
-    collation.set(&my_charset_bin);
-  }
-  String *val_str(String *);    
+  Item_func_sha(Item *a) :Item_str_ascii_func(a) {}
+  String *val_str_ascii(String *);    
   void fix_length_and_dec();      
   const char *func_name() const { return "sha"; }	
 };
@@ -263,13 +276,16 @@ public:
   authentication procedure works, see comments in password.c.
 */
 
-class Item_func_password :public Item_str_func
+class Item_func_password :public Item_str_ascii_func
 {
   char tmp_value[SCRAMBLED_PASSWORD_CHAR_LENGTH+1]; 
 public:
-  Item_func_password(Item *a) :Item_str_func(a) {}
-  String *val_str(String *str);
-  void fix_length_and_dec() { max_length= SCRAMBLED_PASSWORD_CHAR_LENGTH; }
+  Item_func_password(Item *a) :Item_str_ascii_func(a) {}
+  String *val_str_ascii(String *str);
+  void fix_length_and_dec()
+  {
+    fix_length_and_charset(SCRAMBLED_PASSWORD_CHAR_LENGTH, default_charset());
+  }
   const char *func_name() const { return "password"; }
   static char *alloc(THD *thd, const char *password, size_t pass_len);
 };
@@ -282,13 +298,16 @@ public:
   function.
 */
 
-class Item_func_old_password :public Item_str_func
+class Item_func_old_password :public Item_str_ascii_func
 {
   char tmp_value[SCRAMBLED_PASSWORD_CHAR_LENGTH_323+1];
 public:
-  Item_func_old_password(Item *a) :Item_str_func(a) {}
-  String *val_str(String *str);
-  void fix_length_and_dec() { max_length= SCRAMBLED_PASSWORD_CHAR_LENGTH_323; } 
+  Item_func_old_password(Item *a) :Item_str_ascii_func(a) {}
+  String *val_str_ascii(String *str);
+  void fix_length_and_dec()
+  {
+    fix_length_and_charset(SCRAMBLED_PASSWORD_CHAR_LENGTH_323, default_charset());
+  } 
   const char *func_name() const { return "old_password"; }
   static char *alloc(THD *thd, const char *password, size_t pass_len);
 };
@@ -688,7 +707,7 @@ public:
   void fix_length_and_dec() 
   { 
     decimals= 0; 
-    max_length= 3 * 8 + 7; 
+    fix_length_and_charset(3 * 8 + 7, default_charset()); 
     maybe_null= 1;
   }
 };
@@ -848,14 +867,11 @@ class Item_func_uuid: public Item_str_func
 {
 public:
   Item_func_uuid(): Item_str_func() {}
-  void fix_length_and_dec() {
-    collation.set(system_charset_info);
-    /*
-       NOTE! uuid() should be changed to use 'ascii'
-       charset when hex(), format(), md5(), etc, and implicit
-       number-to-string conversion will use 'ascii'
-    */
-    max_length= UUID_LENGTH * system_charset_info->mbmaxlen;
+  void fix_length_and_dec()
+  {
+    collation.set(system_charset_info,
+                  DERIVATION_COERCIBLE, MY_REPERTOIRE_ASCII);
+    fix_char_length(UUID_LENGTH);
   }
   const char *func_name() const{ return "uuid"; }
   String *val_str(String *);
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index a61c5d59d67..cb0fd82bd54 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -3202,11 +3202,9 @@ Item_func_group_concat::fix_fields(THD *thd, Item **ref)
       return TRUE;
   }
 
-  if (agg_item_charsets(collation, func_name(),
-                        args,
-			/* skip charset aggregation for order columns */
-			arg_count - arg_count_order,
-			MY_COLL_ALLOW_CONV, 1))
+  /* skip charset aggregation for order columns */
+  if (agg_item_charsets_for_string_result(collation, func_name(),
+                                          args, arg_count - arg_count_order))
     return 1;
 
   result.set_charset(collation.collation);
@@ -3395,6 +3393,8 @@ String* Item_func_group_concat::val_str(String* str)
 
 void Item_func_group_concat::print(String *str, enum_query_type query_type)
 {
+  /* orig_args is not filled with valid values until fix_fields() */
+  Item **pargs= fixed ? orig_args : args;
   str->append(STRING_WITH_LEN("group_concat("));
   if (distinct)
     str->append(STRING_WITH_LEN("distinct "));
@@ -3402,7 +3402,7 @@ void Item_func_group_concat::print(String *str, enum_query_type query_type)
   {
     if (i)
       str->append(',');
-    args[i]->print(str, query_type);
+    pargs[i]->print(str, query_type);
   }
   if (arg_count_order)
   {
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index 392bc936f17..49d31bf1252 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -57,7 +57,7 @@ static bool make_datetime(date_time_format_types format, MYSQL_TIME *ltime,
 			  String *str)
 {
   char *buff;
-  CHARSET_INFO *cs= &my_charset_bin;
+  CHARSET_INFO *cs= &my_charset_numeric;
   uint length= MAX_DATE_STRING_REP_LENGTH;
 
   if (str->alloc(length))
@@ -1586,9 +1586,7 @@ bool Item_func_from_days::get_date(MYSQL_TIME *ltime, uint fuzzy_date)
 
 void Item_func_curdate::fix_length_and_dec()
 {
-  collation.set(&my_charset_bin);
-  decimals=0; 
-  max_length=MAX_DATE_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
+  Item_date::fix_length_and_dec();
 
   store_now_in_TIME(&ltime);
   
@@ -1649,7 +1647,7 @@ bool Item_func_curdate::get_date(MYSQL_TIME *res,
 String *Item_func_curtime::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
-  str_value.set(buff, buff_length, &my_charset_bin);
+  str_value.set(buff, buff_length, &my_charset_latin1);
   return &str_value;
 }
 
@@ -1659,11 +1657,10 @@ void Item_func_curtime::fix_length_and_dec()
   MYSQL_TIME ltime;
 
   decimals= DATETIME_DEC;
-  collation.set(&my_charset_bin);
   store_now_in_TIME(&ltime);
   value= TIME_to_ulonglong_time(&ltime);
   buff_length= (uint) my_time_to_str(&ltime, buff);
-  max_length= buff_length;
+  fix_length_and_charset_datetime(buff_length);
 }
 
 
@@ -1698,7 +1695,7 @@ void Item_func_curtime_utc::store_now_in_TIME(MYSQL_TIME *now_time)
 String *Item_func_now::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
-  str_value.set(buff,buff_length, &my_charset_bin);
+  str_value.set(buff, buff_length, &my_charset_numeric);
   return &str_value;
 }
 
@@ -1706,13 +1703,12 @@ String *Item_func_now::val_str(String *str)
 void Item_func_now::fix_length_and_dec()
 {
   decimals= DATETIME_DEC;
-  collation.set(&my_charset_bin);
 
   store_now_in_TIME(&ltime);
   value= (longlong) TIME_to_ulonglong_datetime(&ltime);
 
   buff_length= (uint) my_datetime_to_str(&ltime, buff);
-  max_length= buff_length;
+  fix_length_and_charset_datetime(buff_length);
 }
 
 
@@ -1776,7 +1772,7 @@ String *Item_func_sysdate_local::val_str(String *str)
   DBUG_ASSERT(fixed == 1);
   store_now_in_TIME(&ltime);
   buff_length= (uint) my_datetime_to_str(&ltime, buff);
-  str_value.set(buff, buff_length, &my_charset_bin);
+  str_value.set(buff, buff_length, &my_charset_numeric);
   return &str_value;
 }
 
@@ -1800,8 +1796,7 @@ double Item_func_sysdate_local::val_real()
 void Item_func_sysdate_local::fix_length_and_dec()
 {
   decimals= 0;
-  collation.set(&my_charset_bin);
-  max_length= MAX_DATETIME_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
+  fix_length_and_charset_datetime(MAX_DATETIME_WIDTH);
 }
 
 
@@ -2005,7 +2000,8 @@ String *Item_func_date_format::val_str(String *str)
   {
     String *res;
     if (!(res=args[0]->val_str(str)) ||
-	(str_to_time_with_warn(res->ptr(), res->length(), &l_time)))
+	(str_to_time_with_warn(res->charset(), res->ptr(), res->length(),
+	                       &l_time)))
       goto null_date;
 
     l_time.year=l_time.month=l_time.day=0;
@@ -2049,9 +2045,8 @@ null_date:
 void Item_func_from_unixtime::fix_length_and_dec()
 { 
   thd= current_thd;
-  collation.set(&my_charset_bin);
   decimals= DATETIME_DEC;
-  max_length=MAX_DATETIME_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
+  fix_length_and_charset_datetime(MAX_DATETIME_WIDTH);
   maybe_null= 1;
   thd->time_zone_used= 1;
 }
@@ -2109,9 +2104,8 @@ bool Item_func_from_unixtime::get_date(MYSQL_TIME *ltime,
 
 void Item_func_convert_tz::fix_length_and_dec()
 {
-  collation.set(&my_charset_bin);
   decimals= 0;
-  max_length= MAX_DATETIME_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
+  fix_length_and_charset_datetime(MAX_DATETIME_WIDTH);
   maybe_null= 1;
 }
 
@@ -2155,13 +2149,13 @@ bool Item_func_convert_tz::get_date(MYSQL_TIME *ltime,
 
   if (!from_tz_cached)
   {
-    from_tz= my_tz_find(thd, args[1]->val_str(&str));
+    from_tz= my_tz_find(thd, args[1]->val_str_ascii(&str));
     from_tz_cached= args[1]->const_item();
   }
 
   if (!to_tz_cached)
   {
-    to_tz= my_tz_find(thd, args[2]->val_str(&str));
+    to_tz= my_tz_find(thd, args[2]->val_str_ascii(&str));
     to_tz_cached= args[2]->const_item();
   }
 
@@ -2195,9 +2189,8 @@ void Item_date_add_interval::fix_length_and_dec()
 {
   enum_field_types arg0_field_type;
 
-  collation.set(&my_charset_bin);
   maybe_null=1;
-  max_length=MAX_DATETIME_FULL_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
+  fix_length_and_charset_datetime(MAX_DATETIME_FULL_WIDTH);
   value.alloc(max_length);
 
   /*
@@ -2375,7 +2368,9 @@ longlong Item_extract::val_int()
   else
   {
     String *res= args[0]->val_str(&value);
-    if (!res || str_to_time_with_warn(res->ptr(), res->length(), &ltime))
+    if (!res ||
+        str_to_time_with_warn(res->charset(), res->ptr(), res->length(),
+                              &ltime))
     {
       null_value=1;
       return 0;
@@ -2814,7 +2809,7 @@ void Item_func_add_time::fix_length_and_dec()
 {
   enum_field_types arg0_field_type;
   decimals=0;
-  max_length=MAX_DATETIME_FULL_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
+  fix_length_and_charset_datetime(MAX_DATETIME_FULL_WIDTH);
   maybe_null= 1;
 
   /*
@@ -3225,12 +3220,12 @@ void Item_func_timestamp_diff::print(String *str, enum_query_type query_type)
 }
 
 
-String *Item_func_get_format::val_str(String *str)
+String *Item_func_get_format::val_str_ascii(String *str)
 {
   DBUG_ASSERT(fixed == 1);
   const char *format_name;
   KNOWN_DATE_TIME_FORMAT *format;
-  String *val= args[0]->val_str(str);
+  String *val= args[0]->val_str_ascii(str);
   ulong val_len;
 
   if ((null_value= args[0]->null_value))
@@ -3249,7 +3244,7 @@ String *Item_func_get_format::val_str(String *str)
 		      (const uchar *) format_name, val_len))
     {
       const char *format_str= get_date_time_format_str(format, type);
-      str->set(format_str, (uint) strlen(format_str), &my_charset_bin);
+      str->set(format_str, (uint) strlen(format_str), &my_charset_numeric);
       return str;
     }
   }
diff --git a/sql/item_timefunc.h b/sql/item_timefunc.h
index 860bd983184..20e4d6488d6 100644
--- a/sql/item_timefunc.h
+++ b/sql/item_timefunc.h
@@ -122,23 +122,22 @@ public:
 class Item_func_month :public Item_func
 {
 public:
-  Item_func_month(Item *a) :Item_func(a) {}
+  Item_func_month(Item *a) :Item_func(a) { collation.set_numeric(); }
   longlong val_int();
   double val_real()
   { DBUG_ASSERT(fixed == 1); return (double) Item_func_month::val_int(); }
   String *val_str(String *str) 
   {
-    str->set(val_int(), &my_charset_bin);
+    str->set(val_int(), collation.collation);
     return null_value ? 0 : str;
   }
   const char *func_name() const { return "month"; }
   enum Item_result result_type () const { return INT_RESULT; }
   void fix_length_and_dec() 
   { 
-    collation.set(&my_charset_bin);
-    decimals=0;
-    max_length=2*MY_CHARSET_BIN_MB_MAXLEN;
-    maybe_null=1; 
+    decimals= 0;
+    fix_char_length(2);
+    maybe_null= 1;
   }
   bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
 };
@@ -165,9 +164,9 @@ public:
   const char *func_name() const { return "dayofyear"; }
   void fix_length_and_dec() 
   { 
-    decimals=0;
-    max_length=3*MY_CHARSET_BIN_MB_MAXLEN;
-    maybe_null=1; 
+    decimals= 0;
+    fix_char_length(3);
+    maybe_null= 1;
   }
   bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
 };
@@ -290,7 +289,7 @@ class Item_func_weekday :public Item_func
   bool odbc_type;
 public:
   Item_func_weekday(Item *a,bool type_arg)
-    :Item_func(a), odbc_type(type_arg) {}
+    :Item_func(a), odbc_type(type_arg) { collation.set_numeric(); }
   longlong val_int();
   double val_real() { DBUG_ASSERT(fixed == 1); return (double) val_int(); }
   String *val_str(String *str)
@@ -306,10 +305,9 @@ public:
   enum Item_result result_type () const { return INT_RESULT; }
   void fix_length_and_dec()
   {
-    collation.set(&my_charset_bin);
-    decimals=0;
-    max_length=1*MY_CHARSET_BIN_MB_MAXLEN;
-    maybe_null=1;
+    decimals= 0;
+    fix_char_length(1);
+    maybe_null= 1;
   }
   bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
 };
@@ -379,15 +377,15 @@ public:
   Item_date(Item *a) :Item_func(a) {}
   enum Item_result result_type () const { return STRING_RESULT; }
   enum_field_types field_type() const { return MYSQL_TYPE_DATE; }
+  CHARSET_INFO *charset_for_protocol(void) const { return &my_charset_bin; }
   String *val_str(String *str);
   longlong val_int();
   double val_real() { return val_real_from_decimal(); }
   const char *func_name() const { return "date"; }
   void fix_length_and_dec()
   { 
-    collation.set(&my_charset_bin);
-    decimals=0;
-    max_length=MAX_DATE_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
+    decimals= 0;
+    fix_length_and_charset_datetime(MAX_DATE_WIDTH);
   }
   Field *tmp_table_field(TABLE *table)
   {
@@ -414,6 +412,7 @@ public:
   Item_date_func(Item *a,Item *b) :Item_str_func(a,b) {}
   Item_date_func(Item *a,Item *b, Item *c) :Item_str_func(a,b,c) {}
   enum_field_types field_type() const { return MYSQL_TYPE_DATETIME; }
+  CHARSET_INFO *charset_for_protocol(void) const { return &my_charset_bin; }
   Field *tmp_table_field(TABLE *table)
   {
     return tmp_table_field_from_field_type(table, 0);
@@ -440,10 +439,11 @@ public:
   Item_str_timefunc(Item *a,Item *b) :Item_str_func(a,b) {}
   Item_str_timefunc(Item *a, Item *b, Item *c) :Item_str_func(a, b ,c) {}
   enum_field_types field_type() const { return MYSQL_TYPE_TIME; }
+  CHARSET_INFO *charset_for_protocol(void) const { return &my_charset_bin; }
   void fix_length_and_dec()
   {
     decimals= DATETIME_DEC;
-    max_length=MAX_TIME_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
+    fix_length_and_charset_datetime(MAX_TIME_WIDTH);
   }
   Field *tmp_table_field(TABLE *table)
   {
@@ -701,7 +701,6 @@ public:
   void fix_length_and_dec()
   { 
     Item_str_timefunc::fix_length_and_dec();
-    collation.set(&my_charset_bin);
     maybe_null=1;
   }
   const char *func_name() const { return "sec_to_time"; }
@@ -774,13 +773,7 @@ public:
 class Item_typecast_maybe_null :public Item_typecast
 {
 public:
-  Item_typecast_maybe_null(Item *a) :Item_typecast(a) {}
-  void fix_length_and_dec()
-  {
-    collation.set(&my_charset_bin);
-    max_length=args[0]->max_length;
-    maybe_null= 1;
-  }
+  Item_typecast_maybe_null(Item *a) :Item_typecast(a) { maybe_null= 1; }
 };
 
 
@@ -813,16 +806,12 @@ public:
   bool get_time(MYSQL_TIME *ltime);
   const char *cast_type() const { return "date"; }
   enum_field_types field_type() const { return MYSQL_TYPE_DATE; }
+  CHARSET_INFO *charset_for_protocol(void) const { return &my_charset_bin; }
   Field *tmp_table_field(TABLE *table)
   {
     return tmp_table_field_from_field_type(table, 0);
-  }  
-  void fix_length_and_dec()
-  {
-    collation.set(&my_charset_bin);
-    max_length= 10;
-    maybe_null= 1;
   }
+  void fix_length_and_dec() { fix_length_and_charset_datetime(10); }
   bool result_as_longlong() { return TRUE; }
   longlong val_int();
   double val_real() { return (double) val_int(); }
@@ -847,6 +836,7 @@ public:
   bool get_time(MYSQL_TIME *ltime);
   const char *cast_type() const { return "time"; }
   enum_field_types field_type() const { return MYSQL_TYPE_TIME; }
+  CHARSET_INFO *charset_for_protocol(void) const { return &my_charset_bin; }
   Field *tmp_table_field(TABLE *table)
   {
     return tmp_table_field_from_field_type(table, 0);
@@ -863,6 +853,8 @@ public:
   {
     return save_time_in_field(field);
   }
+  void fix_length_and_dec()
+  { fix_length_and_charset_datetime(args[0]->max_char_length()); }
 };
 
 
@@ -874,15 +866,14 @@ public:
   String *val_str(String *str);
   const char *cast_type() const { return "datetime"; }
   enum_field_types field_type() const { return MYSQL_TYPE_DATETIME; }
+  CHARSET_INFO *charset_for_protocol(void) const { return &my_charset_bin; }
   Field *tmp_table_field(TABLE *table)
   {
     return tmp_table_field_from_field_type(table, 0);
   }
   void fix_length_and_dec()
   {
-    collation.set(&my_charset_bin);
-    maybe_null= 1;
-    max_length= MAX_DATETIME_FULL_WIDTH * MY_CHARSET_BIN_MB_MAXLEN;
+    fix_length_and_charset_datetime(MAX_DATETIME_FULL_WIDTH);
     decimals= DATETIME_DEC;
   }
   bool result_as_longlong() { return TRUE; }
@@ -907,10 +898,11 @@ public:
   String *val_str(String *str);
   const char *func_name() const { return "makedate"; }
   enum_field_types field_type() const { return MYSQL_TYPE_DATE; }
+  CHARSET_INFO *charset_for_protocol(void) const { return &my_charset_bin; }
   void fix_length_and_dec()
   { 
     decimals=0;
-    max_length=MAX_DATE_WIDTH*MY_CHARSET_BIN_MB_MAXLEN;
+    fix_length_and_charset_datetime(MAX_DATE_WIDTH);
     maybe_null= 1;
   }
   longlong val_int();
@@ -929,6 +921,7 @@ public:
   String *val_str(String *str);
   enum_field_types field_type() const { return cached_field_type; }
   void fix_length_and_dec();
+  CHARSET_INFO *charset_for_protocol(void) const { return &my_charset_bin; }
 
   Field *tmp_table_field(TABLE *table)
   {
@@ -1019,20 +1012,20 @@ enum date_time_format
   USA_FORMAT, JIS_FORMAT, ISO_FORMAT, EUR_FORMAT, INTERNAL_FORMAT
 };
 
-class Item_func_get_format :public Item_str_func
+class Item_func_get_format :public Item_str_ascii_func
 {
 public:
   const timestamp_type type; // keep it public
   Item_func_get_format(timestamp_type type_arg, Item *a)
-    :Item_str_func(a), type(type_arg)
+    :Item_str_ascii_func(a), type(type_arg)
   {}
-  String *val_str(String *str);
+  String *val_str_ascii(String *str);
   const char *func_name() const { return "get_format"; }
   void fix_length_and_dec()
   {
     maybe_null= 1;
     decimals=0;
-    max_length=17*MY_CHARSET_BIN_MB_MAXLEN;
+    fix_length_and_charset(17, default_charset());
   }
   virtual void print(String *str, enum_query_type query_type);
 };
diff --git a/sql/item_xmlfunc.cc b/sql/item_xmlfunc.cc
index 3e20b90e68e..3621733f456 100644
--- a/sql/item_xmlfunc.cc
+++ b/sql/item_xmlfunc.cc
@@ -2569,7 +2569,7 @@ void Item_xml_str_func::fix_length_and_dec()
 
   nodeset_func= 0;
 
-  if (agg_arg_charsets(collation, args, arg_count, MY_COLL_CMP_CONV, 1))
+  if (agg_arg_charsets_for_comparison(collation, args, arg_count))
     return;
 
   if (collation.collation->mbminlen > 1)
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 87ac577d7b2..6412f0047d1 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3276,6 +3276,18 @@ int Query_log_event::do_apply_event(Relay_log_info const *rli,
       const char* found_semicolon= NULL;
       mysql_parse(thd, thd->query(), thd->query_length(), &found_semicolon);
       log_slow_statement(thd);
+
+      /*
+        Resetting the enable_slow_log thd variable.
+
+        We need to reset it back to the opt_log_slow_slave_statements
+        value after the statement execution (and slow logging
+        is done). It might have changed if the statement was an
+        admin statement (in which case, down in mysql_parse execution
+        thd->enable_slow_log is set to the value of
+        opt_log_slow_admin_statements).
+      */
+      thd->enable_slow_log= opt_log_slow_slave_statements;
     }
     else
     {
@@ -8901,7 +8913,7 @@ static bool record_compare(TABLE *table)
   DBUG_DUMP("record[1]", table->record[1], table->s->reclength);
 
   bool result= FALSE;
-  uchar saved_x[2], saved_filler[2];
+  uchar saved_x[2]= {0, 0}, saved_filler[2]= {0, 0};
 
   if (table->s->null_bytes > 0)
   {
diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index 2d559804534..9b46ad83b14 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -304,7 +304,7 @@ static bool record_compare(TABLE *table)
    */
 
   bool result= FALSE;
-  uchar saved_x[2], saved_filler[2];
+  uchar saved_x[2]= {0, 0}, saved_filler[2]= {0, 0};
 
   if (table->s->null_bytes > 0)
   {
diff --git a/sql/my_decimal.cc b/sql/my_decimal.cc
index 16d07526a0f..234a0a98782 100644
--- a/sql/my_decimal.cc
+++ b/sql/my_decimal.cc
@@ -110,10 +110,63 @@ int my_decimal2string(uint mask, const my_decimal *d,
                          &length, (int)fixed_prec, fixed_dec,
                          filler);
   str->length(length);
+  str->set_charset(&my_charset_numeric);
   return check_result(mask, result);
 }
 
 
+/**
+  @brief Converting decimal to string with character set conversion
+
+  @details Convert given my_decimal to String; allocate buffer as needed.
+
+  @param[in]   mask        what problems to warn on (mask of E_DEC_* values)
+  @param[in]   val         the decimal to print
+  @param[in]   fixed_prec  overall number of digits if ZEROFILL, 0 otherwise
+  @param[in]   fixed_dec   number of decimal places (if fixed_prec != 0)
+  @param[in]   filler      what char to pad with (ZEROFILL et al.)
+  @param[out]  *str        where to store the resulting string
+  @param[in]   cs          character set
+
+  @return error coce
+    @retval E_DEC_OK
+    @retval E_DEC_TRUNCATED
+    @retval E_DEC_OVERFLOW
+    @retval E_DEC_OOM
+
+  Would be great to make it a method of the String class,
+  but this would need to include
+  my_decimal.h from sql_string.h and sql_string.cc, which is not desirable.
+*/
+bool
+str_set_decimal(uint mask, const my_decimal *val,
+                uint fixed_prec, uint fixed_dec, char filler,
+                String *str, CHARSET_INFO *cs)
+{
+  if (!(cs->state & MY_CS_NONASCII))
+  {
+    /* For ASCII-compatible character sets we can use my_decimal2string */
+    my_decimal2string(mask, val, fixed_prec, fixed_dec, filler, str);
+    str->set_charset(cs);
+    return FALSE;
+  }
+  else
+  {
+    /*
+      For ASCII-incompatible character sets (like UCS2) we
+      call my_decimal2string() on a temporary buffer first,
+      and then convert the result to the target character
+      with help of str->copy().
+    */
+    uint errors;
+    char buf[DECIMAL_MAX_STR_LENGTH];
+    String tmp(buf, sizeof(buf), &my_charset_latin1);
+    my_decimal2string(mask, val, fixed_prec, fixed_dec, filler, &tmp);
+    return str->copy(tmp.ptr(), tmp.length(), &my_charset_latin1, cs, &errors);
+  }
+}
+
+
 /*
   Convert from decimal to binary representation
 
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index d5df0ab5930..3f2b9cf3093 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -218,7 +218,8 @@ extern CHARSET_INFO *error_message_charset_info;
 
 enum Derivation
 {
-  DERIVATION_IGNORABLE= 5,
+  DERIVATION_IGNORABLE= 6,
+  DERIVATION_NUMERIC= 5,
   DERIVATION_COERCIBLE= 4,
   DERIVATION_SYSCONST= 3,
   DERIVATION_IMPLICIT= 2,
@@ -226,6 +227,8 @@ enum Derivation
   DERIVATION_EXPLICIT= 0
 };
 
+#define my_charset_numeric      my_charset_latin1
+#define MY_REPERTOIRE_NUMERIC   MY_REPERTOIRE_ASCII
 
 typedef struct my_locale_errmsgs
 {
@@ -870,6 +873,16 @@ typedef Comp_creator* (*chooser_compare_func_creator)(bool invert);
 #include "item.h"
 extern my_decimal decimal_zero;
 
+/* my_decimal.cc */
+bool str_set_decimal(uint mask, const my_decimal *val, uint fixed_prec,
+                     uint fixed_dec, char filler, String *str,
+                     CHARSET_INFO *cs);
+inline bool str_set_decimal(const my_decimal *val, String *str,
+                            CHARSET_INFO *cs)
+{
+  return str_set_decimal(E_DEC_FATAL_ERROR, val, 0, 0, 0, str, cs);
+}
+
 /* sql_parse.cc */
 void free_items(Item *item);
 void cleanup_items(Item *item);
@@ -1192,7 +1205,7 @@ int setup_group(THD *thd, Item **ref_pointer_array, TABLE_LIST *tables,
 		List<Item> &fields, List<Item> &all_fields, ORDER *order,
 		bool *hidden_group_fields);
 bool fix_inner_refs(THD *thd, List<Item> &all_fields, SELECT_LEX *select,
-                   Item **ref_pointer_array);
+                   Item **ref_pointer_array, ORDER *group_list= NULL);
 
 bool handle_select(THD *thd, LEX *lex, select_result *result,
                    ulong setup_tables_done_option);
@@ -2240,8 +2253,17 @@ ulong convert_month_to_period(ulong month);
 void get_date_from_daynr(long daynr,uint *year, uint *month,
 			 uint *day);
 my_time_t TIME_to_timestamp(THD *thd, const MYSQL_TIME *t, my_bool *not_exist);
-bool str_to_time_with_warn(const char *str,uint length,MYSQL_TIME *l_time);
-timestamp_type str_to_datetime_with_warn(const char *str, uint length,
+/* Character set-aware version of str_to_time() */
+bool str_to_time(CHARSET_INFO *cs, const char *str,uint length,
+                 MYSQL_TIME *l_time, int *warning);
+/* Character set-aware version of str_to_datetime() */
+timestamp_type str_to_datetime(CHARSET_INFO *cs,
+                               const char *str, uint length,
+                               MYSQL_TIME *l_time, uint flags, int *was_cut);
+bool str_to_time_with_warn(CHARSET_INFO *cs, const char *str,uint length,
+                           MYSQL_TIME *l_time);
+timestamp_type str_to_datetime_with_warn(CHARSET_INFO *cs,
+                                         const char *str, uint length,
                                          MYSQL_TIME *l_time, uint flags);
 void localtime_to_TIME(MYSQL_TIME *to, struct tm *from);
 void calc_time_from_sec(MYSQL_TIME *to, long seconds, long microseconds);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 41563db4abb..3799562968e 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -5690,8 +5690,9 @@ pthread_handler_t handle_connections_namedpipes(void *arg)
     create_new_thread(thd);
   }
   CloseHandle(connectOverlapped.hEvent);
+  DBUG_LEAVE;
   decrement_handler_count();
-  DBUG_RETURN(0);
+  return 0;
 }
 #endif /* _WIN32 */
 
@@ -5927,9 +5928,9 @@ error:
   if (handle_connect_file_map)	CloseHandle(handle_connect_file_map);
   if (event_connect_answer)	CloseHandle(event_connect_answer);
   if (smem_event_connect_request) CloseHandle(smem_event_connect_request);
-
+  DBUG_LEAVE;
   decrement_handler_count();
-  DBUG_RETURN(0);
+  return 0;
 }
 #endif /* HAVE_SMEM */
 #endif /* EMBEDDED_LIBRARY */
@@ -5974,7 +5975,7 @@ struct my_option my_long_options[]=
    "Tells the master it should log updates for the specified database, and exclude all others not explicitly mentioned.",
    0, 0, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"binlog-ignore-db", OPT_BINLOG_IGNORE_DB,
-   "Tells the master that updates to the given database should not be logged tothe binary log.",
+   "Tells the master that updates to the given database should not be logged to the binary log.",
    0, 0, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"binlog-row-event-max-size", 0,
    "The maximum size of a row-based binary log event in bytes. Rows will be "
@@ -6010,7 +6011,7 @@ struct my_option my_long_options[]=
   {"collation-server", 0, "Set the default collation.",
    (uchar**) &default_collation_name, (uchar**) &default_collation_name,
    0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
-  {"console", OPT_CONSOLE, "Write error output on screen; Don't remove the console window on windows.",
+  {"console", OPT_CONSOLE, "Write error output on screen; don't remove the console window on windows.",
    (uchar**) &opt_console, (uchar**) &opt_console, 0, GET_BOOL, NO_ARG, 0, 0, 0,
    0, 0, 0},
   {"core-file", OPT_WANT_CORE, "Write core on errors.", 0, 0, 0, GET_NO_ARG,
@@ -6043,7 +6044,7 @@ struct my_option my_long_options[]=
    0, 0, 0, 0},
 #endif /* HAVE_STACK_TRACE_ON_SEGV */
   /* See how it's handled in get_one_option() */
-  {"exit-info", 'T', "Used for debugging;  Use at your own risk!", 0, 0, 0,
+  {"exit-info", 'T', "Used for debugging. Use at your own risk.", 0, 0, 0,
    GET_LONG, OPT_ARG, 0, 0, 0, 0, 0, 0},
   {"external-locking", 0, "Use system (external) locking (disabled by default).  With this option enabled you can run myisamchk to test (not repair) tables while the MySQL server is running. Disable with --skip-external-locking.",
    (uchar**) &opt_external_locking, (uchar**) &opt_external_locking,
@@ -6051,7 +6052,7 @@ struct my_option my_long_options[]=
   /* We must always support the next option to make scripts like mysqltest
      easier to do */
   {"gdb", 0,
-   "Set up signals usable for debugging",
+   "Set up signals usable for debugging.",
    (uchar**) &opt_debugging, (uchar**) &opt_debugging,
    0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
 #ifdef HAVE_LARGE_PAGE_OPTION
@@ -6112,7 +6113,7 @@ struct my_option my_long_options[]=
    0, 0, 0, 0, 0, 0},
   {"log-tc", 0,
    "Path to transaction coordinator log (used for transactions that affect "
-   "more than one storage engine, when binary log is disabled)",
+   "more than one storage engine, when binary log is disabled).",
    (uchar**) &opt_tc_log_file, (uchar**) &opt_tc_log_file, 0, GET_STR,
    REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
 #ifdef HAVE_MMAP
@@ -6142,15 +6143,17 @@ thread is in the master's binlogs.",
   {"memlock", 0, "Lock mysqld in memory.", (uchar**) &locked_in_memory,
    (uchar**) &locked_in_memory, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"one-thread", OPT_ONE_THREAD,
-   "(deprecated): Only use one thread (for debugging under Linux). Use thread-handling=no-threads instead",
+   "(Deprecated): Only use one thread (for debugging under Linux). Use "
+   "thread-handling=no-threads instead.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"old-style-user-limits", 0,
-   "Enable old-style user limits (before 5.0.3 user resources were counted per each user+host vs. per account)",
+   "Enable old-style user limits (before 5.0.3, user resources were counted "
+   "per each user+host vs. per account).",
    (uchar**) &opt_old_style_user_limits, (uchar**) &opt_old_style_user_limits,
    0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"port-open-timeout", 0,
    "Maximum time in seconds to wait for the port to become free. "
-   "(Default: no wait)", (uchar**) &mysqld_port_timeout,
+   "(Default: No wait).", (uchar**) &mysqld_port_timeout,
    (uchar**) &mysqld_port_timeout, 0, GET_UINT, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"replicate-do-db", OPT_REPLICATE_DO_DB,
    "Tells the slave thread to restrict replication to the specified database. To specify more than one database, use the directive multiple times, once for each database. Note that this will only work if you do not use cross-database queries such as UPDATE some_db.some_table SET foo='bar' while having selected a different or no database. If you need cross database updates to work, make sure you have 3.23.28 or later, and use replicate-wild-do-table=db_name.%.",
@@ -6197,12 +6200,12 @@ Can't be set to 1 if --log-slave-updates is used.",
    REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
 #endif
   {"show-slave-auth-info", 0,
-   "Show user and password in SHOW SLAVE HOSTS on this master",
+   "Show user and password in SHOW SLAVE HOSTS on this master.",
    (uchar**) &opt_show_slave_auth_info, (uchar**) &opt_show_slave_auth_info, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
 #ifndef DISABLE_GRANT_OPTIONS
   {"skip-grant-tables", 0,
-   "Start without grant tables. This gives all users FULL ACCESS to all tables!",
+   "Start without grant tables. This gives all users FULL ACCESS to all tables.",
    (uchar**) &opt_noacl, (uchar**) &opt_noacl, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0,
    0},
 #endif
@@ -6211,7 +6214,7 @@ Can't be set to 1 if --log-slave-updates is used.",
   {"skip-name-resolve", OPT_SKIP_RESOLVE,
    "Don't resolve hostnames. All hostnames are IP's or 'localhost'.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
-  {"skip-new", OPT_SKIP_NEW, "Don't use new, possible wrong routines.",
+  {"skip-new", OPT_SKIP_NEW, "Don't use new, possibly wrong routines.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"skip-slave-start", 0,
    "If set, slave is not autostarted.", (uchar**) &opt_skip_slave_start,
@@ -6283,7 +6286,7 @@ Can't be set to 1 if --log-slave-updates is used.",
    IF_PURIFY(0,1), 0, 0, 0, 0, 0},
   {"user", 'u', "Run mysqld daemon as user.", 0, 0, 0, GET_STR, REQUIRED_ARG,
    0, 0, 0, 0, 0, 0},
-  {"verbose", 'v', "Used with --help option for detailed help",
+  {"verbose", 'v', "Used with --help option for detailed help.",
    (uchar**) &opt_verbose, (uchar**) &opt_verbose, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0,
    0, 0},
   {"version", 'V', "Output version information and exit.", 0, 0, 0, GET_NO_ARG,
@@ -6895,27 +6898,27 @@ static void usage(void)
     default_collation_name= (char*) default_charset_info->name;
   print_version();
   puts("\
-Copyright (C) 2000-2008 MySQL AB, by Monty and others\n\
+Copyright (C) 2000-2008 MySQL AB, by Monty and others.\n\
 Copyright (C) 2008,2009 Sun Microsystems, Inc.\n\
 This software comes with ABSOLUTELY NO WARRANTY. This is free software,\n\
 and you are welcome to modify and redistribute it under the GPL license\n\n\
-Starts the MySQL database server\n");
+Starts the MySQL database server.\n");
 
   printf("Usage: %s [OPTIONS]\n", my_progname);
   if (!opt_verbose)
-    puts("\nFor more help options (several pages), use mysqld --verbose --help");
+    puts("\nFor more help options (several pages), use mysqld --verbose --help.");
   else
   {
 #ifdef __WIN__
   puts("NT and Win32 specific options:\n\
-  --install                     Install the default service (NT)\n\
-  --install-manual              Install the default service started manually (NT)\n\
-  --install service_name        Install an optional service (NT)\n\
-  --install-manual service_name Install an optional service started manually (NT)\n\
-  --remove                      Remove the default service from the service list (NT)\n\
-  --remove service_name         Remove the service_name from the service list (NT)\n\
-  --enable-named-pipe           Only to be used for the	default server (NT)\n\
-  --standalone                  Dummy option to start as a standalone server (NT)\
+  --install                     Install the default service (NT).\n\
+  --install-manual              Install the default service started manually (NT).\n\
+  --install service_name        Install an optional service (NT).\n\
+  --install-manual service_name Install an optional service started manually (NT).\n\
+  --remove                      Remove the default service from the service list (NT).\n\
+  --remove service_name         Remove the service_name from the service list (NT).\n\
+  --enable-named-pipe           Only to be used for the default server (NT).\n\
+  --standalone                  Dummy option to start as a standalone server (NT).\
 ");
   puts("");
 #endif
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 68285563239..5483ed237db 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -1198,11 +1198,7 @@ QUICK_RANGE_SELECT::~QUICK_RANGE_SELECT()
     if (file) 
     {
       range_end();
-      if (head->key_read)
-      {
-        head->key_read= 0;
-        file->extra(HA_EXTRA_NO_KEYREAD);
-      }
+      head->set_keyread(FALSE);
       if (free_file)
       {
         DBUG_PRINT("info", ("Freeing separate handler 0x%lx (free: %d)", (long) file,
@@ -1404,10 +1400,7 @@ end:
   head->file= file;
   /* We don't have to set 'head->keyread' here as the 'file' is unique */
   if (!head->no_keyread)
-  {
-    head->key_read= 1;
     head->mark_columns_used_by_index(index);
-  }
   head->prepare_for_position();
   head->file= org_file;
   bitmap_copy(&column_bitmap, head->read_set);
@@ -8344,7 +8337,7 @@ int QUICK_INDEX_MERGE_SELECT::read_keys_and_merge()
   DBUG_ENTER("QUICK_INDEX_MERGE_SELECT::read_keys_and_merge");
 
   /* We're going to just read rowids. */
-  file->extra(HA_EXTRA_KEYREAD);
+  head->set_keyread(TRUE);
   head->prepare_for_position();
 
   cur_quick_it.rewind();
@@ -8420,7 +8413,7 @@ int QUICK_INDEX_MERGE_SELECT::read_keys_and_merge()
   delete unique;
   doing_pk_scan= FALSE;
   /* index_merge currently doesn't support "using index" at all */
-  file->extra(HA_EXTRA_NO_KEYREAD);
+  head->set_keyread(FALSE);
   init_read_record(&read_record, thd, head, (SQL_SELECT*) 0, 1 , 1, TRUE);
   DBUG_RETURN(result);
 }
@@ -10838,7 +10831,7 @@ int QUICK_GROUP_MIN_MAX_SELECT::reset(void)
   int result;
   DBUG_ENTER("QUICK_GROUP_MIN_MAX_SELECT::reset");
 
-  file->extra(HA_EXTRA_KEYREAD); /* We need only the key attributes */
+  head->set_keyread(TRUE); /* We need only the key attributes */
   if ((result= file->ha_index_init(index,1)))
     DBUG_RETURN(result);
   if (quick_prefix_select && quick_prefix_select->reset())
diff --git a/sql/opt_sum.cc b/sql/opt_sum.cc
index d85e976e9c8..ae54a462c67 100644
--- a/sql/opt_sum.cc
+++ b/sql/opt_sum.cc
@@ -325,11 +325,7 @@ int opt_sum_query(TABLE_LIST *tables, List<Item> &all_fields,COND *conds)
 	  if (!error && reckey_in_range(0, &ref, item_field->field, 
 			                conds, range_fl, prefix_len))
 	    error= HA_ERR_KEY_NOT_FOUND;
-          if (table->key_read)
-          {
-            table->key_read= 0;
-            table->file->extra(HA_EXTRA_NO_KEYREAD);
-          }
+          table->set_keyread(FALSE);
           table->file->ha_index_end();
           if (error)
 	  {
@@ -415,11 +411,7 @@ int opt_sum_query(TABLE_LIST *tables, List<Item> &all_fields,COND *conds)
 	  if (!error && reckey_in_range(1, &ref, item_field->field,
 			                conds, range_fl, prefix_len))
 	    error= HA_ERR_KEY_NOT_FOUND;
-          if (table->key_read)
-          {
-            table->key_read=0;
-            table->file->extra(HA_EXTRA_NO_KEYREAD);
-          }
+          table->set_keyread(FALSE);
           table->file->ha_index_end();
           if (error)
           {
@@ -880,10 +872,7 @@ static bool find_key_for_maxmin(bool max_fl, TABLE_REF *ref,
             converted (for example to upper case)
           */
           if (field->part_of_key.is_set(idx))
-          {
-            table->key_read= 1;
-            table->file->extra(HA_EXTRA_KEYREAD);
-          }
+            table->set_keyread(TRUE);
           return 1;
         }
       }
diff --git a/sql/protocol.cc b/sql/protocol.cc
index 855a6842f1f..fd943c98bd6 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -736,10 +736,10 @@ bool Protocol::send_result_set_metadata(List<Item> *list, uint flags)
       /* Store fixed length fields */
       pos= (char*) local_packet->ptr()+local_packet->length();
       *pos++= 12;				// Length of packed fields
-      if (item->collation.collation == &my_charset_bin || thd_charset == NULL)
+      if (item->charset_for_protocol() == &my_charset_bin || thd_charset == NULL)
       {
         /* No conversion */
-        int2store(pos, field.charsetnr);
+        int2store(pos, item->charset_for_protocol()->number);
         int4store(pos+2, field.length);
       }
       else
diff --git a/sql/share/Makefile.am b/sql/share/Makefile.am
index da19ed1c5a7..cd4f7a83279 100644
--- a/sql/share/Makefile.am
+++ b/sql/share/Makefile.am
@@ -23,7 +23,7 @@ dist-hook:
 	  test -d $(distdir)/$$dir || mkdir $(distdir)/$$dir; \
 	  $(INSTALL_DATA) $(srcdir)/$$dir/*.* $(distdir)/$$dir; \
 	done; \
-	sleep 1 ; touch $(srcdir)/*/errmsg.sys
+	sleep 1 ; touch $(builddir)/*/errmsg.sys
 	$(INSTALL_DATA) $(srcdir)/charsets/README $(distdir)/charsets
 	$(INSTALL_DATA) $(srcdir)/charsets/Index.xml $(distdir)/charsets
 
@@ -40,12 +40,12 @@ install-data-local:
 	for lang in @AVAILABLE_LANGUAGES@; \
 	do \
 		$(mkinstalldirs) $(DESTDIR)$(pkgdatadir)/$$lang; \
-		$(INSTALL_DATA) $(srcdir)/$$lang/errmsg.sys \
+		$(INSTALL_DATA) $(builddir)/$$lang/errmsg.sys \
 			        $(DESTDIR)$(pkgdatadir)/$$lang/errmsg.sys; \
 	done
 	$(mkinstalldirs) $(DESTDIR)$(pkgdatadir)/charsets
-	$(INSTALL_DATA) $(srcdir)/errmsg-utf8.txt \
-	$(DESTDIR)$(pkgdatadir)/errmsg-utf8.txt; \
+	$(INSTALL_DATA) $(builddir)/errmsg-utf8.txt \
+        $(DESTDIR)$(pkgdatadir)/errmsg-utf8.txt; \
 	$(INSTALL_DATA) $(srcdir)/charsets/README $(DESTDIR)$(pkgdatadir)/charsets/README
 	$(INSTALL_DATA) $(srcdir)/charsets/*.xml $(DESTDIR)$(pkgdatadir)/charsets
 
@@ -54,7 +54,7 @@ uninstall-local:
 	@RM@ -f -r $(DESTDIR)$(pkgdatadir)
 
 distclean-local:
-	@RM@ -f */errmsg.sys
+	@RM@ -f $(builddir)/*/errmsg.sys
 
 # Do nothing
 link_sources:
diff --git a/sql/sp_cache.cc b/sql/sp_cache.cc
index a585004b1e8..09d347bf951 100644
--- a/sql/sp_cache.cc
+++ b/sql/sp_cache.cc
@@ -184,8 +184,9 @@ sp_head *sp_cache_lookup(sp_cache **cp, sp_name *name)
     sp_cache_invalidate()
       
   NOTE
-    This is called when a VIEW definition is modifed. We can't destroy sp_head
-    objects here as one may modify VIEW definitions from prelocking-free SPs.
+    This is called when a VIEW definition is created or modified (and in some
+    other contexts). We can't destroy sp_head objects here as one may modify
+    VIEW definitions from prelocking-free SPs.
 */
 
 void sp_cache_invalidate()
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index d2ac5d638d7..c6bf0e381fb 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -1848,6 +1848,8 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
 {
   bool err_status= FALSE;
   uint params = m_pcont->context_var_count();
+  /* Query start time may be reset in a multi-stmt SP; keep this for later. */
+  ulonglong utime_before_sp_exec= thd->utime_after_lock;
   sp_rcontext *save_spcont, *octx;
   sp_rcontext *nctx = NULL;
   bool save_enable_slow_log= false;
@@ -2054,6 +2056,7 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
 
   delete nctx;
   thd->spcont= save_spcont;
+  thd->utime_after_lock= utime_before_sp_exec;
 
   DBUG_RETURN(err_status);
 }
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 0dcfafed882..b4c9aa576d0 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2226,6 +2226,7 @@ void wait_for_condition(THD *thd, mysql_mutex_t *mutex, mysql_cond_t *cond)
   proc_info=thd->proc_info;
   thd_proc_info(thd, "Waiting for table");
   DBUG_ENTER("wait_for_condition");
+  DEBUG_SYNC(thd, "waiting_for_table");
   if (!thd->killed)
     mysql_cond_wait(cond, mutex);
 
@@ -4359,6 +4360,7 @@ open_tables_acquire_upgradable_mdl(THD *thd, TABLE_LIST *tables_start,
   TABLE_LIST *table;
 
   DBUG_ASSERT(!thd->locked_tables_mode);
+  DEBUG_SYNC(thd, "open_tables_acquire_upgradable_mdl");
 
   for (table= tables_start; table && table != tables_end;
        table= table->next_global)
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index b7ded3b632f..4b21bc283e2 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -612,10 +612,10 @@ bool THD::handle_condition(uint sql_errno,
 
   for (Internal_error_handler *error_handler= m_internal_handler;
        error_handler;
-       error_handler= m_internal_handler->m_prev_internal_handler)
+       error_handler= error_handler->m_prev_internal_handler)
   {
-    if (error_handler-> handle_condition(this, sql_errno, sqlstate, level, msg,
-                                         cond_hdl))
+    if (error_handler->handle_condition(this, sql_errno, sqlstate, level, msg,
+					cond_hdl))
     {
       return TRUE;
     }
@@ -625,10 +625,12 @@ bool THD::handle_condition(uint sql_errno,
 }
 
 
-void THD::pop_internal_handler()
+Internal_error_handler *THD::pop_internal_handler()
 {
   DBUG_ASSERT(m_internal_handler != NULL);
+  Internal_error_handler *popped_handler= m_internal_handler;
   m_internal_handler= m_internal_handler->m_prev_internal_handler;
+  return popped_handler;
 }
 
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 617c13a0a08..538a8a42ef6 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2594,7 +2594,7 @@ public:
   /**
     Remove the error handler last pushed.
   */
-  void pop_internal_handler();
+  Internal_error_handler *pop_internal_handler();
 
   /**
     Raise an exception condition.
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 6da734592dc..14b5e160629 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -1675,6 +1675,7 @@ void st_select_lex::init_query()
   having= prep_having= where= prep_where= 0;
   olap= UNSPECIFIED_OLAP_TYPE;
   having_fix_field= 0;
+  group_fix_field= 0;
   context.select_lex= this;
   context.init();
   /*
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 7ec87806ea5..0b27f73e763 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -677,6 +677,8 @@ public:
   bool  braces;   	/* SELECT ... UNION (SELECT ... ) <- this braces */
   /* TRUE when having fix field called in processing of this SELECT */
   bool having_fix_field;
+  /* TRUE when GROUP BY fix field called in processing of this SELECT */
+  bool group_fix_field;
   /* List of references to fields referenced from inner selects */
   List<Item_outer_ref> inner_refs_list;
   /* Number of Item_sum-derived objects in this SELECT */
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 468f81a7d87..099d1a87bfd 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -290,6 +290,7 @@ bool handle_select(THD *thd, LEX *lex, select_result *result,
     all_fields        List of all fields used in select
     select            Current select
     ref_pointer_array Array of references to Items used in current select
+    group_list        GROUP BY list (is NULL by default)
 
   DESCRIPTION
     The function serves 3 purposes - adds fields referenced from inner
@@ -308,6 +309,8 @@ bool handle_select(THD *thd, LEX *lex, select_result *result,
       function is aggregated in the select where the outer field was
       resolved or in some more inner select then the Item_direct_ref
       class should be used.
+      Also it should be used if we are grouping by a subquery containing
+      the outer field.
     The resolution is done here and not at the fix_fields() stage as
     it can be done only after sum functions are fixed and pulled up to
     selects where they are have to be aggregated.
@@ -324,7 +327,7 @@ bool handle_select(THD *thd, LEX *lex, select_result *result,
 
 bool
 fix_inner_refs(THD *thd, List<Item> &all_fields, SELECT_LEX *select,
-                 Item **ref_pointer_array)
+                 Item **ref_pointer_array, ORDER *group_list)
 {
   Item_outer_ref *ref;
   bool res= FALSE;
@@ -374,6 +377,22 @@ fix_inner_refs(THD *thd, List<Item> &all_fields, SELECT_LEX *select,
         }
       }
     }
+    else
+    {
+      /*
+        Check if GROUP BY item trees contain the outer ref:
+        in this case we have to use Item_direct_ref instead of Item_ref.
+      */
+      for (ORDER *group= group_list; group; group= group->next)
+      {
+        if ((*group->item)->walk(&Item::find_item_processor, TRUE,
+                                 (uchar *) ref))
+        {
+          direct_ref= TRUE;
+          break;
+        }
+      }
+    }
     new_ref= direct_ref ?
               new Item_direct_ref(ref->context, item_ref, ref->table_name,
                           ref->field_name, ref->alias_name_used) :
@@ -580,7 +599,8 @@ JOIN::prepare(Item ***rref_pointer_array,
   }
 
   if (select_lex->inner_refs_list.elements &&
-      fix_inner_refs(thd, all_fields, select_lex, ref_pointer_array))
+      fix_inner_refs(thd, all_fields, select_lex, ref_pointer_array,
+                     group_list))
     DBUG_RETURN(-1);
 
   if (group_list)
@@ -2334,7 +2354,13 @@ JOIN::destroy()
 	tab->cleanup();
     }
     tmp_join->tmp_join= 0;
-    tmp_table_param.copy_field= 0;
+    /*
+      We need to clean up tmp_table_param for reusable JOINs (having non-zero
+      and different from self tmp_join) because it's not being cleaned up
+      anywhere else (as we need to keep the join is reusable).
+    */
+    tmp_table_param.cleanup();
+    tmp_table_param.copy_field= tmp_join->tmp_table_param.copy_field= 0;
     DBUG_RETURN(tmp_join->destroy());
   }
   cond_equal= 0;
@@ -5939,6 +5965,12 @@ JOIN::make_simple_join(JOIN *parent, TABLE *tmp_table)
   const_table_map= 0;
   tmp_table_param.field_count= tmp_table_param.sum_func_count=
     tmp_table_param.func_count= 0;
+  /*
+    We need to destruct the copy_field (allocated in create_tmp_table())
+    before setting it to 0 if the join is not "reusable".
+  */
+  if (!tmp_join || tmp_join != this) 
+    tmp_table_param.cleanup(); 
   tmp_table_param.copy_field= tmp_table_param.copy_field_end=0;
   first_record= sort_and_group=0;
   send_records= (ha_rows) 0;
@@ -6709,10 +6741,7 @@ make_join_readinfo(JOIN *join, ulonglong options)
     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);
-      }
+        table->set_keyread(TRUE);
       break;
     case JT_ALL:
       /*
@@ -6773,10 +6802,7 @@ make_join_readinfo(JOIN *join, ulonglong options)
 	  if (tab->select && tab->select->quick &&
               tab->select->quick->index != MAX_KEY && //not index_merge
 	      table->covering_keys.is_set(tab->select->quick->index))
-	  {
-	    table->key_read=1;
-	    table->file->extra(HA_EXTRA_KEYREAD);
-	  }
+            table->set_keyread(TRUE);
 	  else if (!table->covering_keys.is_clear_all() &&
 		   !(tab->select && tab->select->quick))
 	  {					// Only read index tree
@@ -6860,11 +6886,7 @@ void JOIN_TAB::cleanup()
   limit= 0;
   if (table)
   {
-    if (table->key_read)
-    {
-      table->key_read= 0;
-      table->file->extra(HA_EXTRA_NO_KEYREAD);
-    }
+    table->set_keyread(FALSE);
     table->file->ha_index_or_rnd_end();
     /*
       We need to reset this for next select
@@ -7107,9 +7129,11 @@ eq_ref_table(JOIN *join, ORDER *start_order, JOIN_TAB *tab)
       }
       if (order)
       {
-	found++;
-	DBUG_ASSERT(!(order->used & map));
-	order->used|=map;
+        if (!(order->used & map))
+        {
+          found++;
+          order->used|= map;
+        }
 	continue;				// Used in ORDER BY
       }
       if (!only_eq_ref_tables(join,start_order, (*ref_item)->used_tables()))
@@ -8277,7 +8301,8 @@ static Item *eliminate_item_equal(COND *cond, COND_EQUAL *upper_levels,
   else
   {
     DBUG_ASSERT(cond->type() == Item::COND_ITEM);
-    ((Item_cond *) cond)->add_at_head(&eq_list);
+    if (eq_list.elements)
+      ((Item_cond *) cond)->add_at_head(&eq_list);
   }
 
   cond->quick_fix_field();
@@ -9953,7 +9978,11 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
   KEY_PART_INFO *key_part_info;
   Item **copy_func;
   MI_COLUMNDEF *recinfo;
-  uint total_uneven_bit_length= 0;
+  /*
+    total_uneven_bit_length is uneven bit length for visible fields
+    hidden_uneven_bit_length is uneven bit length for hidden fields
+  */
+  uint total_uneven_bit_length= 0, hidden_uneven_bit_length= 0;
   bool force_copy_fields= param->force_copy_fields;
   /* Treat sum functions as normal ones when loose index scan is used. */
   save_sum_fields|= param->precomputed_group_by;
@@ -10232,6 +10261,14 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
       */
       param->hidden_field_count= fieldnr;
       null_count= 0;
+      /*
+        On last hidden field we store uneven bit length in
+        hidden_uneven_bit_length and proceed calculation of
+        uneven bits for visible fields into
+        total_uneven_bit_length variable.
+      */
+      hidden_uneven_bit_length= total_uneven_bit_length;
+      total_uneven_bit_length= 0;
     }
   }
   DBUG_ASSERT(fieldnr == (uint) (reg_field - table->field));
@@ -10277,7 +10314,8 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
     else
       null_count++;
   }
-  hidden_null_pack_length=(hidden_null_count+7)/8;
+  hidden_null_pack_length= (hidden_null_count + 7 +
+                            hidden_uneven_bit_length) / 8;
   null_pack_length= (hidden_null_pack_length +
                      (null_count + total_uneven_bit_length + 7) / 8);
   reclength+=null_pack_length;
@@ -11691,21 +11729,45 @@ flush_cached_records(JOIN *join,JOIN_TAB *join_tab,bool skip_last)
       return NESTED_LOOP_KILLED; // Aborted by user /* purecov: inspected */
     }
     SQL_SELECT *select=join_tab->select;
-    if (rc == NESTED_LOOP_OK &&
-        (!join_tab->cache.select || !join_tab->cache.select->skip_record()))
+    if (rc == NESTED_LOOP_OK)
     {
-      uint i;
-      reset_cache_read(&join_tab->cache);
-      for (i=(join_tab->cache.records- (skip_last ? 1 : 0)) ; i-- > 0 ;)
+      bool consider_record= !join_tab->cache.select || 
+        !join_tab->cache.select->skip_record();
+
+      /*
+        Check for error: skip_record() can execute code by calling
+        Item_subselect::val_*. We need to check for errors (if any)
+        after such call.
+      */
+      if (join->thd->is_error())
       {
-	read_cached_record(join_tab);
-	if (!select || !select->skip_record())
+        reset_cache_write(&join_tab->cache);
+        return NESTED_LOOP_ERROR;
+      }
+
+      if (consider_record)  
+      {
+        uint i;
+        reset_cache_read(&join_tab->cache);
+        for (i=(join_tab->cache.records- (skip_last ? 1 : 0)) ; i-- > 0 ;)
         {
-          rc= (join_tab->next_select)(join,join_tab+1,0);
-	  if (rc != NESTED_LOOP_OK && rc != NESTED_LOOP_NO_MORE_ROWS)
+          read_cached_record(join_tab);
+          if (!select || !select->skip_record())
           {
-            reset_cache_write(&join_tab->cache);
-            return rc;
+            /*
+              Check for error: skip_record() can execute code by calling
+              Item_subselect::val_*. We need to check for errors (if any)
+              after such call.
+              */
+            if (join->thd->is_error())
+              rc= NESTED_LOOP_ERROR;
+            else
+              rc= (join_tab->next_select)(join,join_tab+1,0);
+            if (rc != NESTED_LOOP_OK && rc != NESTED_LOOP_NO_MORE_ROWS)
+            {
+              reset_cache_write(&join_tab->cache);
+              return rc;
+            }
           }
         }
       }
@@ -11790,16 +11852,11 @@ join_read_const_table(JOIN_TAB *tab, POSITION *pos)
 	!table->no_keyread &&
         (int) table->reginfo.lock_type <= (int) TL_READ_HIGH_PRIORITY)
     {
-      table->key_read=1;
-      table->file->extra(HA_EXTRA_KEYREAD);
+      table->set_keyread(TRUE);
       tab->index= tab->ref.key;
     }
     error=join_read_const(tab);
-    if (table->key_read)
-    {
-      table->key_read=0;
-      table->file->extra(HA_EXTRA_NO_KEYREAD);
-    }
+    table->set_keyread(FALSE);
     if (error)
     {
       tab->info="unique row not found";
@@ -12152,12 +12209,8 @@ join_read_first(JOIN_TAB *tab)
 {
   int error;
   TABLE *table=tab->table;
-  if (!table->key_read && table->covering_keys.is_set(tab->index) &&
-      !table->no_keyread)
-  {
-    table->key_read=1;
-    table->file->extra(HA_EXTRA_KEYREAD);
-  }
+  if (table->covering_keys.is_set(tab->index) && !table->no_keyread)
+    table->set_keyread(TRUE);
   tab->table->status=0;
   tab->read_record.read_record=join_read_next;
   tab->read_record.table=table;
@@ -12191,12 +12244,8 @@ join_read_last(JOIN_TAB *tab)
 {
   TABLE *table=tab->table;
   int error;
-  if (!table->key_read && table->covering_keys.is_set(tab->index) &&
-      !table->no_keyread)
-  {
-    table->key_read=1;
-    table->file->extra(HA_EXTRA_KEYREAD);
-  }
+  if (table->covering_keys.is_set(tab->index) && !table->no_keyread)
+    table->set_keyread(TRUE);
   tab->table->status=0;
   tab->read_record.read_record=join_read_prev;
   tab->read_record.table=table;
@@ -12963,7 +13012,7 @@ static int test_if_order_by_key(ORDER *order, TABLE *table, uint idx,
   key_part_end=key_part+table->key_info[idx].key_parts;
   key_part_map const_key_parts=table->const_key_parts[idx];
   int reverse=0;
-  my_bool on_primary_key= FALSE;
+  my_bool on_pk_suffix= FALSE;
   DBUG_ENTER("test_if_order_by_key");
 
   for (; order ; order=order->next, const_key_parts>>=1)
@@ -12985,11 +13034,12 @@ static int test_if_order_by_key(ORDER *order, TABLE *table, uint idx,
         key as a suffix to the secondary keys. If it has continue to check
         the primary key as a suffix.
       */
-      if (!on_primary_key &&
+      if (!on_pk_suffix &&
           (table->file->ha_table_flags() & HA_PRIMARY_KEY_IN_READ_INDEX) &&
-          table->s->primary_key != MAX_KEY)
+          table->s->primary_key != MAX_KEY &&
+          table->s->primary_key != idx)
       {
-        on_primary_key= TRUE;
+        on_pk_suffix= TRUE;
         key_part= table->key_info[table->s->primary_key].key_part;
         key_part_end=key_part+table->key_info[table->s->primary_key].key_parts;
         const_key_parts=table->const_key_parts[table->s->primary_key];
@@ -13021,7 +13071,7 @@ static int test_if_order_by_key(ORDER *order, TABLE *table, uint idx,
     reverse=flag;				// Remember if reverse
     key_part++;
   }
-  if (on_primary_key)
+  if (on_pk_suffix)
   {
     uint used_key_parts_secondary= table->key_info[idx].key_parts;
     uint used_key_parts_pk=
@@ -13443,12 +13493,6 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
     */
     if (select_limit >= table_records)
     {
-      /* 
-        filesort() and join cache are usually faster than reading in 
-        index order and not using join cache
-        */
-      if (tab->type == JT_ALL && tab->join->tables > tab->join->const_tables + 1)
-        DBUG_RETURN(0);
       keys= *table->file->keys_to_use_for_scanning();
       keys.merge(table->covering_keys);
 
@@ -13510,8 +13554,15 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
             select_limit= table_records;
           if (group)
           {
-            rec_per_key= used_key_parts ? keyinfo->rec_per_key[used_key_parts-1]
-                                        : 1;
+            /* 
+              Used_key_parts can be larger than keyinfo->key_parts
+              when using a secondary index clustered with a primary 
+              key (e.g. as in Innodb). 
+              See Bug #28591 for details.
+            */  
+            rec_per_key= used_key_parts &&
+                         used_key_parts <= keyinfo->key_parts ?
+                         keyinfo->rec_per_key[used_key_parts-1] : 1;
             set_if_bigger(rec_per_key, 1);
             /*
               With a grouping query each group containing on average
@@ -13591,6 +13642,19 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
 	}      
       }
     }
+
+    /*
+      filesort() and join cache are usually faster than reading in 
+      index order and not using join cache, except in case that chosen
+      index is clustered primary key.
+    */
+    if ((select_limit >= table_records) &&
+        (tab->type == JT_ALL &&
+         tab->join->tables > tab->join->const_tables + 1) &&
+         ((unsigned) best_key != table->s->primary_key ||
+          !table->file->primary_key_is_clustered()))
+      DBUG_RETURN(0);
+
     if (best_key >= 0)
     {
       bool quick_created= FALSE;
@@ -13612,11 +13676,8 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
            If ref_key used index tree reading only ('Using index' in EXPLAIN),
            and best_key doesn't, then revert the decision.
         */
-        if (!table->covering_keys.is_set(best_key) && table->key_read)
-        {
-          table->key_read= 0;
-          table->file->extra(HA_EXTRA_NO_KEYREAD);          
-        }        
+        if (!table->covering_keys.is_set(best_key))
+          table->set_keyread(FALSE);
         if (!quick_created)
 	{
           tab->index= best_key;
@@ -13629,10 +13690,7 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
             select->quick= 0;
           }
           if (table->covering_keys.is_set(best_key))
-          {
-            table->key_read=1;
-            table->file->extra(HA_EXTRA_KEYREAD);
-          }
+            table->set_keyread(TRUE);
           table->file->ha_index_or_rnd_end();
           if (join->select_options & SELECT_DESCRIBE)
           {
@@ -13806,11 +13864,8 @@ create_sort_index(THD *thd, JOIN *join, ORDER *order,
         We can only use 'Only index' if quick key is same as ref_key
         and in index_merge 'Only index' cannot be used
       */
-      if (table->key_read && ((uint) tab->ref.key != select->quick->index))
-      {
-	table->key_read=0;
-	table->file->extra(HA_EXTRA_NO_KEYREAD);
-      }
+      if (((uint) tab->ref.key != select->quick->index))
+        table->set_keyread(FALSE);
     }
     else
     {
@@ -13866,11 +13921,7 @@ create_sort_index(THD *thd, JOIN *join, ORDER *order,
   tab->type=JT_ALL;				// Read with normal read_record
   tab->read_first_record= join_init_read_record;
   tab->join->examined_rows+=examined_rows;
-  if (table->key_read)				// Restore if we used indexes
-  {
-    table->key_read=0;
-    table->file->extra(HA_EXTRA_NO_KEYREAD);
-  }
+  table->set_keyread(FALSE); // Restore if we used indexes
   DBUG_RETURN(table->sort.found_records == HA_POS_ERROR);
 err:
   DBUG_RETURN(-1);
@@ -14307,7 +14358,7 @@ join_init_cache(THD *thd,JOIN_TAB *tables,uint table_count)
       {
 	used_fields--;
 	length+=field->fill_cache_field(copy);
-	if (copy->blob_field)
+	if (copy->type == CACHE_BLOB)
 	  (*blob_ptr++)=copy;
 	if (field->real_maybe_null())
 	  null_fields++;
@@ -14322,8 +14373,8 @@ join_init_cache(THD *thd,JOIN_TAB *tables,uint table_count)
     {						/* must copy null bits */
       copy->str= tables[i].table->null_flags;
       copy->length= tables[i].table->s->null_bytes;
-      copy->strip=0;
-      copy->blob_field=0;
+      copy->type=0;
+      copy->field=0;
       length+=copy->length;
       copy++;
       cache->fields++;
@@ -14333,8 +14384,8 @@ join_init_cache(THD *thd,JOIN_TAB *tables,uint table_count)
     {
       copy->str= (uchar*) &tables[i].table->null_row;
       copy->length=sizeof(tables[i].table->null_row);
-      copy->strip=0;
-      copy->blob_field=0;
+      copy->type=0;
+      copy->field=0;
       length+=copy->length;
       copy++;
       cache->fields++;
@@ -14359,9 +14410,10 @@ used_blob_length(CACHE_FIELD **ptr)
   uint length,blob_length;
   for (length=0 ; *ptr ; ptr++)
   {
-    (*ptr)->blob_length=blob_length=(*ptr)->blob_field->get_length();
+    Field_blob *field_blob= (Field_blob *) (*ptr)->field;
+    (*ptr)->blob_length=blob_length= field_blob->get_length();
     length+=blob_length;
-    (*ptr)->blob_field->get_ptr(&(*ptr)->str);
+    field_blob->get_ptr(&(*ptr)->str);
   }
   return length;
 }
@@ -14390,30 +14442,35 @@ store_record_in_cache(JOIN_CACHE *cache)
   cache->records++;
   for (copy=cache->field ; copy < end_field; copy++)
   {
-    if (copy->blob_field)
+    if (copy->type == CACHE_BLOB)
     {
+      Field_blob *blob_field= (Field_blob *) copy->field;
       if (last_record)
       {
-	copy->blob_field->get_image(pos, copy->length+sizeof(char*), 
-				    copy->blob_field->charset());
+	blob_field->get_image(pos, copy->length+sizeof(char*), 
+                              blob_field->charset());
 	pos+=copy->length+sizeof(char*);
       }
       else
       {
-	copy->blob_field->get_image(pos, copy->length, // blob length
-				    copy->blob_field->charset());
+	blob_field->get_image(pos, copy->length, // blob length
+                              blob_field->charset());
 	memcpy(pos+copy->length,copy->str,copy->blob_length);  // Blob data
 	pos+=copy->length+copy->blob_length;
       }
     }
     else
     {
-      if (copy->strip)
+      if (copy->type == CACHE_STRIPPED)
       {
 	uchar *str,*end;
-	for (str=copy->str,end= str+copy->length;
-	     end > str && end[-1] == ' ' ;
-	     end--) ;
+        Field *field= copy->field;
+        if (field && field->maybe_null() && field->is_null())
+          end= str= copy->str;
+        else
+          for (str=copy->str,end= str+copy->length;
+               end > str && end[-1] == ' ' ;
+               end--) ;
 	length=(uint) (end-str);
 	memcpy(pos+2, str, length);
         int2store(pos, length);
@@ -14462,23 +14519,24 @@ read_cached_record(JOIN_TAB *tab)
        copy < end_field;
        copy++)
   {
-    if (copy->blob_field)
+    if (copy->type == CACHE_BLOB)
     {
+      Field_blob *blob_field= (Field_blob *) copy->field;
       if (last_record)
       {
-	copy->blob_field->set_image(pos, copy->length+sizeof(char*),
-				    copy->blob_field->charset());
+	blob_field->set_image(pos, copy->length+sizeof(char*),
+                              blob_field->charset());
 	pos+=copy->length+sizeof(char*);
       }
       else
       {
-	copy->blob_field->set_ptr(pos, pos+copy->length);
-	pos+=copy->length+copy->blob_field->get_length();
+	blob_field->set_ptr(pos, pos+copy->length);
+	pos+=copy->length + blob_field->get_length();
       }
     }
     else
     {
-      if (copy->strip)
+      if (copy->type == CACHE_STRIPPED)
       {
         length= uint2korr(pos);
 	memcpy(copy->str, pos+2, length);
@@ -14689,11 +14747,29 @@ find_order_in_list(THD *thd, Item **ref_pointer_array, TABLE_LIST *tables,
 
     We check order_item->fixed because Item_func_group_concat can put
     arguments for which fix_fields already was called.
+    
+    group_fix_field= TRUE is to resolve aliases from the SELECT list
+    without creating of Item_ref-s: JOIN::exec() wraps aliased items
+    in SELECT list with Item_copy items. To re-evaluate such a tree
+    that includes Item_copy items we have to refresh Item_copy caches,
+    but:
+      - filesort() never refresh Item_copy items,
+      - end_send_group() checks every record for group boundary by the
+        test_if_group_changed function that obtain data from these
+        Item_copy items, but the copy_fields function that
+        refreshes Item copy items is called after group boundaries only -
+        that is a vicious circle.
+    So we prevent inclusion of Item_copy items.
   */
-  if (!order_item->fixed &&
+  bool save_group_fix_field= thd->lex->current_select->group_fix_field;
+  if (is_group_field)
+    thd->lex->current_select->group_fix_field= TRUE;
+  bool ret= (!order_item->fixed &&
       (order_item->fix_fields(thd, order->item) ||
        (order_item= *order->item)->check_cols(1) ||
-       thd->is_fatal_error))
+       thd->is_fatal_error));
+  thd->lex->current_select->group_fix_field= save_group_fix_field;
+  if (ret)
     return TRUE; /* Wrong field. */
 
   uint el= all_fields.elements;
@@ -15789,7 +15865,7 @@ static bool add_ref_to_table_cond(THD *thd, JOIN_TAB *join_tab)
 
   Item_cond_and *cond=new Item_cond_and();
   TABLE *table=join_tab->table;
-  int error;
+  int error= 0;
   if (!cond)
     DBUG_RETURN(TRUE);
 
@@ -15807,7 +15883,8 @@ static bool add_ref_to_table_cond(THD *thd, JOIN_TAB *join_tab)
     cond->fix_fields(thd, (Item**)&cond);
   if (join_tab->select)
   {
-    error=(int) cond->add(join_tab->select->cond);
+    if (join_tab->select->cond)
+      error=(int) cond->add(join_tab->select->cond);
     join_tab->select_cond=join_tab->select->cond=cond;
   }
   else if ((join_tab->select= make_select(join_tab->table, 0, 0, cond, 0,
diff --git a/sql/sql_select.h b/sql/sql_select.h
index c30aae4fbf9..3cec00a84c8 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -98,6 +98,10 @@ typedef struct st_table_ref
 } TABLE_REF;
 
 
+
+#define CACHE_BLOB      1        /* blob field  */
+#define CACHE_STRIPPED  2        /* field stripped of trailing spaces */
+
 /**
   CACHE_FIELD and JOIN_CACHE is used on full join to cache records in outer
   table
@@ -106,8 +110,8 @@ typedef struct st_table_ref
 typedef struct st_cache_field {
   uchar *str;
   uint length, blob_length;
-  Field_blob *blob_field;
-  bool strip;
+  Field *field;
+  uint type;    /**< category of the of the copied field (CACHE_BLOB et al.) */
 } CACHE_FIELD;
 
 
@@ -361,7 +365,25 @@ public:
   */
   bool no_const_tables; 
   
-  JOIN *tmp_join; ///< copy of this JOIN to be used with temporary tables
+  /**
+    Copy of this JOIN to be used with temporary tables.
+
+    tmp_join is used when the JOIN needs to be "reusable" (e.g. in a subquery
+    that gets re-executed several times) and we know will use temporary tables
+    for materialization. The materialization to a temporary table overwrites the
+    JOIN structure to point to the temporary table after the materialization is
+    done. This is where tmp_join is used : it's a copy of the JOIN before the
+    materialization and is used in restoring before re-execution by overwriting
+    the current JOIN structure with the saved copy.
+    Because of this we should pay extra care of not freeing up helper structures
+    that are referenced by the original contents of the JOIN. We can check for
+    this by making sure the "current" join is not the temporary copy, e.g.
+    !tmp_join || tmp_join != join
+ 
+    We should free these sub-structures at JOIN::destroy() if the "current" join
+    has a copy is not that copy.
+  */
+  JOIN *tmp_join;
   ROLLUP rollup;				///< Used with rollup
 
   bool select_distinct;				///< Set if SELECT DISTINCT
@@ -724,10 +746,11 @@ public:
      we need to check for errors executing it and react accordingly
     */
     if (!res && table->in_use->is_error())
-      res= 2;
+      res= 1; /* STORE_KEY_FATAL */
     dbug_tmp_restore_column_map(table->write_set, old_map);
     null_key= to_field->is_null() || item->null_value;
-    return (err != 0 || res > 2 ? STORE_KEY_FATAL : (store_key_result) res); 
+    return ((err != 0 || res < 0 || res > 2) ? STORE_KEY_FATAL : 
+            (store_key_result) res);
   }
 };
 
@@ -756,17 +779,17 @@ protected:
       if ((res= item->save_in_field(to_field, 1)))
       {       
         if (!err)
-          err= res;
+          err= res < 0 ? 1 : res; /* 1=STORE_KEY_FATAL */
       }
       /*
         Item::save_in_field() may call Item::val_xxx(). And if this is a subquery
         we need to check for errors executing it and react accordingly
         */
       if (!err && to_field->table->in_use->is_error())
-        err= 2;
+        err= 1; /* STORE_KEY_FATAL */
     }
     null_key= to_field->is_null() || item->null_value;
-    return (err > 2 ?  STORE_KEY_FATAL : (store_key_result) err);
+    return (err > 2 ? STORE_KEY_FATAL : (store_key_result) err);
   }
 };
 
diff --git a/sql/sql_string.cc b/sql/sql_string.cc
index e4e51aba622..75e8ca30cf0 100644
--- a/sql/sql_string.cc
+++ b/sql/sql_string.cc
@@ -412,11 +412,25 @@ bool String::append(const char *s)
 
 bool String::append(const char *s,uint32 arg_length, CHARSET_INFO *cs)
 {
-  uint32 dummy_offset;
+  uint32 offset;
   
-  if (needs_conversion(arg_length, cs, str_charset, &dummy_offset))
+  if (needs_conversion(arg_length, cs, str_charset, &offset))
   {
-    uint32 add_length= arg_length / cs->mbminlen * str_charset->mbmaxlen;
+    uint32 add_length;
+    if ((cs == &my_charset_bin) && offset)
+    {
+      DBUG_ASSERT(str_charset->mbminlen > offset);
+      offset= str_charset->mbminlen - offset; // How many characters to pad
+      add_length= arg_length + offset;
+      if (realloc(str_length + add_length))
+        return TRUE;
+      bzero((char*) Ptr + str_length, offset);
+      memcpy(Ptr + str_length + offset, s, arg_length);
+      str_length+= add_length;
+      return FALSE;
+    }
+
+    add_length= arg_length / cs->mbminlen * str_charset->mbmaxlen;
     uint dummy_errors;
     if (realloc(str_length + add_length)) 
       return TRUE;
@@ -966,6 +980,24 @@ well_formed_copy_nchars(CHARSET_INFO *to_cs,
         uint pad_length= to_cs->mbminlen - from_offset;
         bzero(to, pad_length);
         memmove(to + pad_length, from, from_offset);
+        /*
+          In some cases left zero-padding can create an incorrect character.
+          For example:
+            INSERT INTO t1 (utf32_column) VALUES (0x110000);
+          We'll pad the value to 0x00110000, which is a wrong UTF32 sequence!
+          The valid characters range is limited to 0x00000000..0x0010FFFF.
+          
+          Make sure we didn't pad to an incorrect character.
+        */
+        if (to_cs->cset->well_formed_len(to_cs,
+                                         to, to + to_cs->mbminlen, 1,
+                                         &well_formed_error) !=
+                                         to_cs->mbminlen)
+        {
+          *from_end_pos= *well_formed_error_pos= from;
+          *cannot_convert_error_pos= NULL;
+          return 0;
+        }
         nchars--;
         from+= from_offset;
         from_length-= from_offset;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 4bf56c7f8c8..8ba8c50b01e 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1896,30 +1896,16 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 
   mysql_ha_rm_tables(thd, tables);
 
-  /*
-    If we have the table in the definition cache, we don't have to check the
-    .frm file to find if the table is a normal table (not view) and what
-    engine to use.
-  */
-  mysql_mutex_lock(&LOCK_open);
+  /* Disable drop of enabled log tables, must be done before name locking */
   for (table= tables; table; table= table->next_local)
   {
-    TABLE_SHARE *share;
-    table->db_type= NULL;
-    if ((share= get_cached_table_share(table->db, table->table_name)))
-      table->db_type= share->db_type();
-
-    /* Disable drop of enabled log tables */
-    if (share && (share->table_category == TABLE_CATEGORY_LOG) &&
-        check_if_log_table(table->db_length, table->db,
+    if (check_if_log_table(table->db_length, table->db,
                            table->table_name_length, table->table_name, 1))
     {
-      mysql_mutex_unlock(&LOCK_open);
       my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
       DBUG_RETURN(1);
     }
   }
-  mysql_mutex_unlock(&LOCK_open);
 
   if (!drop_temporary)
   {
@@ -1974,7 +1960,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   {
     char *db=table->db;
     handlerton *table_type;
-    enum legacy_db_type frm_db_type;
+    enum legacy_db_type frm_db_type= DB_TYPE_UNKNOWN;
 
     DBUG_PRINT("table", ("table_l: '%s'.'%s'  table: 0x%lx  s: 0x%lx",
                          table->db, table->table_name, (long) table->table,
@@ -2042,7 +2028,6 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       built_query.append("`,");
     }
 
-    table_type= table->db_type;
     if (!drop_temporary)
     {
       if (thd->locked_tables_mode)
@@ -2072,10 +2057,12 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       TODO: Investigate what should be done to remove this lock completely.
             Is exclusive meta-data lock enough ?
     */
+    DEBUG_SYNC(thd, "rm_table_part2_before_delete_table");
+    DBUG_EXECUTE_IF("sleep_before_part2_delete_table",
+                    my_sleep(100000););
     mysql_mutex_lock(&LOCK_open);
     if (drop_temporary ||
-        ((table_type == NULL &&        
-         access(path, F_OK) &&
+        ((access(path, F_OK) &&
           ha_create_table_from_engine(thd, db, alias)) ||
          (!drop_view &&
           mysql_frm_type(thd, path, &frm_db_type) != FRMTYPE_TABLE)))
@@ -2091,15 +2078,25 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     else
     {
       char *end;
-      if (table_type == NULL)
+      /*
+        Cannot use the db_type from the table, since that might have changed
+        while waiting for the exclusive name lock. We are under LOCK_open,
+        so reading from the frm-file is safe.
+      */
+      if (frm_db_type == DB_TYPE_UNKNOWN)
       {
-	mysql_frm_type(thd, path, &frm_db_type);
-        table_type= ha_resolve_by_legacy_type(thd, frm_db_type);
+        mysql_frm_type(thd, path, &frm_db_type);
+        DBUG_PRINT("info", ("frm_db_type %d from %s", frm_db_type, path));
       }
+      table_type= ha_resolve_by_legacy_type(thd, frm_db_type);
       // Remove extension for delete
       *(end= path + path_length - reg_ext_length)= '\0';
+      DBUG_PRINT("info", ("deleting table of type %d",
+                          (table_type ? table_type->db_type : 0)));
       error= ha_delete_table(thd, table_type, path, db, table->table_name,
                              !dont_log_query);
+
+      /* No error if non existent table and 'IF EXIST' clause or view */
       if ((error == ENOENT || error == HA_ERR_NO_SUCH_TABLE) && 
 	  (if_exists || table_type == NULL))
       {
@@ -2141,6 +2138,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                                     table->table_name););
 
   }
+  DEBUG_SYNC(thd, "rm_table_part2_before_binlog");
   thd->thread_specific_used|= tmp_table_deleted;
   error= 0;
   if (wrong_tables.length())
@@ -2791,7 +2789,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
                                                sql_field->interval_list);
         List_iterator<String> int_it(sql_field->interval_list);
         String conv, *tmp;
-        char comma_buf[2];
+        char comma_buf[4]; /* 4 bytes for utf32 */
         int comma_length= cs->cset->wc_mb(cs, ',', (uchar*) comma_buf,
                                           (uchar*) comma_buf + 
                                           sizeof(comma_buf));
@@ -7127,6 +7125,9 @@ view_err:
   else
     create_info->data_file_name=create_info->index_file_name=0;
 
+  DEBUG_SYNC(thd, "alter_table_before_create_table_no_lock");
+  DBUG_EXECUTE_IF("sleep_before_create_table_no_lock",
+                  my_sleep(100000););
   /*
     Create a table with a temporary name.
     With create_info->frm_only == 1 this creates a .frm file only.
@@ -7336,6 +7337,7 @@ view_err:
     close_temporary_table(thd, new_table, 1, 0);
     new_table= 0;
   }
+  DEBUG_SYNC(thd, "alter_table_before_rename_result_table");
 
   /*
     Data is copied. Now we:
@@ -7478,6 +7480,7 @@ view_err:
   thd_proc_info(thd, "end");
 
   DBUG_EXECUTE_IF("sleep_alter_before_main_binlog", my_sleep(6000000););
+  DEBUG_SYNC(thd, "alter_table_before_main_binlog");
 
   ha_binlog_log_query(thd, create_info->db_type, LOGCOM_ALTER_TABLE,
                       thd->query(), thd->query_length(),
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index c2ab740f29b..b1367a61b37 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -330,6 +330,7 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
   String stmt_query;
   bool lock_upgrade_done= FALSE;
   MDL_ticket *mdl_ticket= NULL;
+  Query_tables_list backup;
 
   DBUG_ENTER("mysql_create_or_drop_trigger");
 
@@ -393,6 +394,12 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
   {
     bool if_exists= thd->lex->drop_if_exists;
 
+    /*
+      Protect the query table list from the temporary and potentially
+      destructive changes necessary to open the trigger's table.
+    */
+    thd->lex->reset_n_backup_query_tables_list(&backup);
+
     if (add_table_for_trigger(thd, thd->lex->spname, if_exists, & tables))
       goto end;
 
@@ -522,6 +529,10 @@ end:
   if (thd->locked_tables_mode && tables && lock_upgrade_done)
     mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
 
+  /* Restore the query table list. Used only for drop trigger. */
+  if (!create)
+    thd->lex->restore_backup_query_tables_list(&backup);
+
   if (thd->global_read_lock.has_protection())
     thd->global_read_lock.start_waiting_global_read_lock(thd);
 
@@ -1636,10 +1647,6 @@ bool add_table_for_trigger(THD *thd,
   if (load_table_name_for_trigger(thd, trg_name, &trn_path, &tbl_name))
     DBUG_RETURN(TRUE);
 
-  /* We need to reset statement table list to be PS/SP friendly. */
-  lex->query_tables= 0;
-  lex->query_tables_last= &lex->query_tables;
-
   *table= sp_add_to_query_tables(thd, lex, trg_name->m_db.str,
                                  tbl_name.str, TL_IGNORE);
 
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index a163dda2c69..1643bce8ddd 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -400,10 +400,7 @@ int mysql_update(THD *thd,
       matching rows before updating the table!
     */
     if (used_index < MAX_KEY && old_covering_keys.is_set(used_index))
-    {
-      table->key_read=1;
       table->mark_columns_used_by_index(used_index);
-    }
     else
     {
       table->use_all_columns();
@@ -852,11 +849,7 @@ int mysql_update(THD *thd,
 err:
   delete select;
   free_underlaid_joins(thd, select_lex);
-  if (table->key_read)
-  {
-    table->key_read=0;
-    table->file->extra(HA_EXTRA_NO_KEYREAD);
-  }
+  table->set_keyread(FALSE);
   thd->abort_on_warning= 0;
   DBUG_RETURN(1);
 }
@@ -1192,6 +1185,57 @@ reopen_tables:
 }
 
 
+/**
+   Implementation of the safe update options during UPDATE IGNORE. This syntax
+   causes an UPDATE statement to ignore all errors. In safe update mode,
+   however, we must never ignore the ER_UPDATE_WITHOUT_KEY_IN_SAFE_MODE. There
+   is a special hook in my_message_sql that will otherwise delete all errors
+   when the IGNORE option is specified. 
+
+   In the future, all IGNORE handling should be used with this class and all
+   traces of the hack outlined below should be removed.
+
+   - The parser detects IGNORE option and sets thd->lex->ignore= 1
+   
+   - In JOIN::optimize, if this is set, then 
+     thd->lex->current_select->no_error gets set.
+
+   - In my_message_sql(), if the flag above is set then any error is
+     unconditionally converted to a warning.
+
+   We are moving in the direction of using Internal_error_handler subclasses
+   to do all such error tweaking, please continue this effort if new bugs
+   appear.
+ */
+class Safe_dml_handler : public Internal_error_handler {
+
+private:
+  bool m_handled_error;
+
+public:
+  explicit Safe_dml_handler() : m_handled_error(FALSE) {}
+
+  bool handle_condition(THD *thd,
+                        uint sql_errno,
+                        const char* sqlstate,
+                        MYSQL_ERROR::enum_warning_level level,
+                        const char* msg,
+                        MYSQL_ERROR ** cond_hdl)
+  {
+    if (level == MYSQL_ERROR::WARN_LEVEL_ERROR &&
+        sql_errno == ER_UPDATE_WITHOUT_KEY_IN_SAFE_MODE)
+    {
+      thd->stmt_da->set_error_status(thd, sql_errno, msg, sqlstate);
+      m_handled_error= TRUE;
+      return TRUE;
+    }
+    return FALSE;
+  }
+
+  bool handled_error() { return m_handled_error; }
+
+};
+
 /*
   Setup multi-update handling and call SELECT to do the join
 */
@@ -1224,18 +1268,36 @@ bool mysql_multi_update(THD *thd,
                                MODE_STRICT_ALL_TABLES));
 
   List<Item> total_list;
+
+  Safe_dml_handler handler;
+  bool using_handler= thd->variables.option_bits & OPTION_SAFE_UPDATES;
+  if (using_handler)
+    thd->push_internal_handler(&handler);
+
   res= mysql_select(thd, &select_lex->ref_pointer_array,
-                      table_list, select_lex->with_wild,
-                      total_list,
-                      conds, 0, (ORDER *) NULL, (ORDER *)NULL, (Item *) NULL,
-                      (ORDER *)NULL,
-                      options | SELECT_NO_JOIN_CACHE | SELECT_NO_UNLOCK |
-                      OPTION_SETUP_TABLES_DONE,
-                      *result, unit, select_lex);
-  DBUG_PRINT("info",("res: %d  report_error: %d", res,
-                     (int) thd->is_error()));
+                    table_list, select_lex->with_wild,
+                    total_list,
+                    conds, 0, (ORDER *) NULL, (ORDER *)NULL, (Item *) NULL,
+                    (ORDER *)NULL,
+                    options | SELECT_NO_JOIN_CACHE | SELECT_NO_UNLOCK |
+                    OPTION_SETUP_TABLES_DONE,
+                    *result, unit, select_lex);
+
+  if (using_handler)
+  {
+    Internal_error_handler *top_handler;
+    top_handler= thd->pop_internal_handler();
+    DBUG_ASSERT(&handler == top_handler);
+  }
+
+  DBUG_PRINT("info",("res: %d  report_error: %d", res, (int) thd->is_error()));
   res|= thd->is_error();
-  if (unlikely(res))
+  /*
+    Todo: remove below code and make Safe_dml_handler do error processing
+    instead. That way we can return the actual error instead of
+    ER_UNKNOWN_ERROR.
+  */
+  if (unlikely(res) && (!using_handler || !handler.handled_error()))
   {
     /* If we had a another error reported earlier then this will be ignored */
     (*result)->send_error(ER_UNKNOWN_ERROR, ER(ER_UNKNOWN_ERROR));
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 5f61adb392d..4438f1c37b5 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -405,17 +405,14 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
 
   view= lex->unlink_first_table(&link_to_local);
 
-  if (mode != VIEW_CREATE_NEW)
+  if (mode == VIEW_ALTER && fill_defined_view_parts(thd, view))
   {
-    if (mode == VIEW_ALTER &&
-        fill_defined_view_parts(thd, view))
-    {
-      res= TRUE;
-      goto err;
-    }
-    sp_cache_invalidate();
+    res= TRUE;
+    goto err;
   }
 
+  sp_cache_invalidate();
+
   if (!lex->definer)
   {
     /*
diff --git a/sql/table.cc b/sql/table.cc
index 0e66ff9da94..fa1186c2a45 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -3410,7 +3410,7 @@ bool TABLE_LIST::prep_check_option(THD *thd, uint8 check_opt_type)
 
 void TABLE_LIST::hide_view_error(THD *thd)
 {
-  if (thd->get_internal_handler())
+  if (thd->killed || thd->get_internal_handler())
     return;
   /* Hide "Unknown column" or "Unknown function" error */
   DBUG_ASSERT(thd->is_error());
@@ -4421,7 +4421,7 @@ void TABLE::mark_columns_used_by_index(uint index)
   MY_BITMAP *bitmap= &tmp_set;
   DBUG_ENTER("TABLE::mark_columns_used_by_index");
 
-  (void) file->extra(HA_EXTRA_KEYREAD);
+  set_keyread(TRUE);
   bitmap_clear_all(bitmap);
   mark_columns_used_by_index_no_reset(index, bitmap);
   column_bitmaps_set(bitmap, bitmap);
@@ -4444,8 +4444,7 @@ void TABLE::restore_column_maps_after_mark_index()
 {
   DBUG_ENTER("TABLE::restore_column_maps_after_mark_index");
 
-  key_read= 0;
-  (void) file->extra(HA_EXTRA_NO_KEYREAD);
+  set_keyread(FALSE);
   default_column_bitmaps();
   file->column_bitmaps_signal();
   DBUG_VOID_RETURN;
diff --git a/sql/table.h b/sql/table.h
index 3d11118a3ea..3832e7c9555 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -950,6 +950,21 @@ public:
   */
   inline bool needs_reopen()
   { return !db_stat || m_needs_reopen; }
+
+  inline void set_keyread(bool flag)
+  {
+    DBUG_ASSERT(file);
+    if (flag && !key_read)
+    {
+      key_read= 1;
+      file->extra(HA_EXTRA_KEYREAD);
+    }
+    else if (!flag && key_read)
+    {
+      key_read= 0;
+      file->extra(HA_EXTRA_NO_KEYREAD);
+    }
+  }
 };
 
 
diff --git a/sql/time.cc b/sql/time.cc
index 5f804072eb0..2db7812b9ef 100644
--- a/sql/time.cc
+++ b/sql/time.cc
@@ -214,6 +214,69 @@ ulong convert_month_to_period(ulong month)
 
 
 /*
+  Convert a string to 8-bit representation,
+  for use in str_to_time/str_to_date/str_to_date.
+  
+  In the future to_ascii() can be extended to convert
+  non-ASCII digits to ASCII digits
+  (for example, ARABIC-INDIC, DEVANAGARI, BENGALI, and so on)
+  so DATE/TIME/DATETIME values understand digits in the
+  respected scripts.
+*/
+static uint
+to_ascii(CHARSET_INFO *cs,
+         const char *src, uint src_length,
+         char *dst, uint dst_length)
+                     
+{
+  int cnvres;
+  my_wc_t wc;
+  const char *srcend= src + src_length;
+  char *dst0= dst, *dstend= dst + dst_length - 1;
+  while (dst < dstend &&
+         (cnvres= (cs->cset->mb_wc)(cs, &wc,
+                                    (const uchar*) src,
+                                    (const uchar*) srcend)) > 0 &&
+         wc < 128)
+  {
+    src+= cnvres;
+    *dst++= wc;
+  }
+  *dst= '\0';
+  return dst - dst0;
+}
+
+
+/* Character set-aware version of str_to_time() */
+bool str_to_time(CHARSET_INFO *cs, const char *str,uint length,
+                 MYSQL_TIME *l_time, int *warning)
+{
+  char cnv[32];
+  if ((cs->state & MY_CS_NONASCII) != 0)
+  {
+    length= to_ascii(cs, str, length, cnv, sizeof(cnv));
+    str= cnv;
+  }
+  return str_to_time(str, length, l_time, warning);
+}
+
+
+/* Character set-aware version of str_to_datetime() */
+timestamp_type str_to_datetime(CHARSET_INFO *cs,
+                               const char *str, uint length,
+                               MYSQL_TIME *l_time, uint flags, int *was_cut)
+{
+  char cnv[32];
+  if ((cs->state & MY_CS_NONASCII) != 0)
+  {
+    length= to_ascii(cs, str, length, cnv, sizeof(cnv));
+    str= cnv;
+  }
+  return str_to_datetime(str, length, l_time, flags, was_cut);
+}
+
+
+/*
   Convert a timestamp string to a MYSQL_TIME value and produce a warning 
   if string was truncated during conversion.
 
@@ -222,14 +285,15 @@ ulong convert_month_to_period(ulong month)
 */
 
 timestamp_type
-str_to_datetime_with_warn(const char *str, uint length, MYSQL_TIME *l_time,
+str_to_datetime_with_warn(CHARSET_INFO *cs,
+                          const char *str, uint length, MYSQL_TIME *l_time,
                           uint flags)
 {
   int was_cut;
   THD *thd= current_thd;
   timestamp_type ts_type;
   
-  ts_type= str_to_datetime(str, length, l_time,
+  ts_type= str_to_datetime(cs, str, length, l_time,
                            (flags | (thd->variables.sql_mode &
                                      (MODE_INVALID_DATES |
                                       MODE_NO_ZERO_DATE))),
@@ -284,7 +348,8 @@ my_time_t TIME_to_timestamp(THD *thd, const MYSQL_TIME *t, my_bool *in_dst_time_
     See str_to_time() for more info.
 */
 bool
-str_to_time_with_warn(const char *str, uint length, MYSQL_TIME *l_time)
+str_to_time_with_warn(CHARSET_INFO *cs,
+                      const char *str, uint length, MYSQL_TIME *l_time)
 {
   int warning;
   bool ret_val= str_to_time(str, length, l_time, &warning);
@@ -697,7 +762,7 @@ void make_time(const DATE_TIME_FORMAT *format __attribute__((unused)),
 {
   uint length= (uint) my_time_to_str(l_time, (char*) str->ptr());
   str->length(length);
-  str->set_charset(&my_charset_bin);
+  str->set_charset(&my_charset_numeric);
 }
 
 
@@ -706,7 +771,7 @@ void make_date(const DATE_TIME_FORMAT *format __attribute__((unused)),
 {
   uint length= (uint) my_date_to_str(l_time, (char*) str->ptr());
   str->length(length);
-  str->set_charset(&my_charset_bin);
+  str->set_charset(&my_charset_numeric);
 }
 
 
@@ -715,7 +780,7 @@ void make_datetime(const DATE_TIME_FORMAT *format __attribute__((unused)),
 {
   uint length= (uint) my_datetime_to_str(l_time, (char*) str->ptr());
   str->length(length);
-  str->set_charset(&my_charset_bin);
+  str->set_charset(&my_charset_numeric);
 }