Add 0x before pointers (to help with debugging)

Add support for VARCHAR with 1 or 2 length bytes
Enable VARCHAR packing in MyISAM files (previous patch didn't pack data properly)
Give error if we got problems in temporary tables during a SELECT
Don't use new table generated by ALTER TABLE if index generation fails
Fixed wrong call by range_end() (Could cause an ASSERT in debug mode)

14 files changed, 367 insertions, 175 deletions

diff --git a/sql/field.cc b/sql/field.cc
index dafb3dc25da..c26120734c0 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -4641,7 +4641,19 @@ Field *Field_string::new_field(MEM_ROOT *root, struct st_table *new_table)
 }
 
 /****************************************************************************
-** VARCHAR type  (Not available for the end user yet)
+  VARCHAR type
+  Data in field->ptr is stored as:
+    1 or 2 bytes length-prefix-header  (from Field_varstring::length_bytes)
+    data
+
+  NOTE:
+  When VARCHAR is stored in a key (for handler::index_read() etc) it's always
+  stored with a 2 byte prefix. (Just like blob keys).
+
+  Normally length_bytes is calculated as (field_length < 256 : 1 ? 2)
+  The exception is if there is a prefix key field that is part of a long
+  VARCHAR, in which case field_length for this may be 1 but the length_bytes
+  is 2.
 ****************************************************************************/
 
 
@@ -4670,8 +4682,11 @@ int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
 						    from,from+length,
 						    field_length/
 						    field_charset->mbmaxlen);
-  memcpy(ptr + HA_KEY_BLOB_LENGTH, from, copy_length);
-  int2store(ptr, copy_length);
+  memcpy(ptr + length_bytes, from, copy_length);
+  if (length_bytes == 1)
+    *ptr= (uchar) copy_length;
+  else
+    int2store(ptr, copy_length);
   
   if (copy_length < length)
     error= 1;
@@ -4684,91 +4699,117 @@ int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
 int Field_varstring::store(longlong nr)
 {
   char buff[64];
-  int  l;
-  CHARSET_INFO *cs=charset();
-  l= (cs->cset->longlong10_to_str)(cs,buff,sizeof(buff),-10,nr);
-  return Field_varstring::store(buff,(uint)l,cs);
+  uint  length;
+  length= (uint) (field_charset->cset->longlong10_to_str)(field_charset,
+                                                          buff,
+                                                          sizeof(buff),
+                                                          -10,nr);
+  return Field_varstring::store(buff, length, field_charset);
 }
 
 
 double Field_varstring::val_real(void)
 {
   int not_used;
-  uint length=uint2korr(ptr)+HA_KEY_BLOB_LENGTH;
-  CHARSET_INFO *cs=charset();
-  return my_strntod(cs, ptr+HA_KEY_BLOB_LENGTH, length, (char**)0, &not_used);
+  uint length= length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
+  return my_strntod(field_charset, ptr+length_bytes, length, (char**) 0,
+                    &not_used);
 }
 
 
 longlong Field_varstring::val_int(void)
 {
   int not_used;
-  uint length=uint2korr(ptr)+HA_KEY_BLOB_LENGTH;
-  CHARSET_INFO *cs=charset();
-  return my_strntoll(cs,ptr+HA_KEY_BLOB_LENGTH,length,10,NULL, &not_used);
+  uint length= length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
+  return my_strntoll(field_charset, ptr+length_bytes, length, 10, NULL,
+                     &not_used);
 }
 
 
 String *Field_varstring::val_str(String *val_buffer __attribute__((unused)),
 				 String *val_ptr)
 {
-  uint length=uint2korr(ptr);
-  val_ptr->set((const char*) ptr+HA_KEY_BLOB_LENGTH,length,field_charset);
+  uint length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
+  val_ptr->set((const char*) ptr+length_bytes, length, field_charset);
   return val_ptr;
 }
 
 
 int Field_varstring::cmp(const char *a_ptr, const char *b_ptr)
 {
-  uint a_length=uint2korr(a_ptr);
-  uint b_length=uint2korr(b_ptr);
+  uint a_length, b_length;
   int diff;
+
+  if (length_bytes == 1)
+  {
+    a_length= (uint) (uchar) *a_ptr;
+    b_length= (uint) (uchar) *b_ptr;
+  }
+  else
+  {
+    a_length= uint2korr(a_ptr);
+    b_length= uint2korr(b_ptr);
+  }
   diff= field_charset->coll->strnncollsp(field_charset,
                                          (const uchar*) a_ptr+
-                                         HA_KEY_BLOB_LENGTH,
+                                         length_bytes,
                                          a_length,
                                          (const uchar*) b_ptr+
-                                         HA_KEY_BLOB_LENGTH,
+                                         length_bytes,
                                          b_length,0);
   return diff;
 }
 
 
+/*
+  NOTE: varstring and blob keys are ALWAYS stored with a 2 byte length prefix
+*/
+
 int Field_varstring::key_cmp(const byte *key_ptr, uint max_key_length)
 {
   char *blob1;
-  uint length= uint2korr(ptr);
-  CHARSET_INFO *cs= charset();
-  uint char_length= max_key_length / cs->mbmaxlen;
+  uint length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
+  uint char_length= max_key_length / field_charset->mbmaxlen;
 
-  char_length= my_charpos(cs, ptr + HA_KEY_BLOB_LENGTH,
-                          ptr + HA_KEY_BLOB_LENGTH + length, char_length);
+  char_length= my_charpos(field_charset, ptr + length_bytes,
+                          ptr + length_bytes + length, char_length);
   set_if_smaller(length, char_length);
-  return cs->coll->strnncollsp(cs, 
-                               (const uchar*) ptr+2, length,
-                               (const uchar*) key_ptr+HA_KEY_BLOB_LENGTH,
-                               uint2korr(key_ptr), 0);
+  return field_charset->coll->strnncollsp(field_charset, 
+                                          (const uchar*) ptr + length_bytes,
+                                          length,
+                                          (const uchar*) key_ptr+
+                                          HA_KEY_BLOB_LENGTH,
+                                          uint2korr(key_ptr), 0);
 }
 
 
+/*
+  Compare to key segments (always 2 byte length prefix)
+
+  NOTE
+    This is used only to compare key segments created for index_read().
+    (keys are created and compared in key.cc)
+*/
+
 int Field_varstring::key_cmp(const byte *a,const byte *b)
 {
-  CHARSET_INFO *cs= charset();
-  return cs->coll->strnncollsp(cs,
-                               (const uchar*) a + HA_KEY_BLOB_LENGTH,
-                               uint2korr(a),
-                               (const uchar*) b + HA_KEY_BLOB_LENGTH,
-                               uint2korr(b),
-                               0);
+  return field_charset->coll->strnncollsp(field_charset,
+                                          (const uchar*) a +
+                                          HA_KEY_BLOB_LENGTH,
+                                          uint2korr(a),
+                                          (const uchar*) b +
+                                          HA_KEY_BLOB_LENGTH,
+                                          uint2korr(b),
+                                          0);
 }
 
 
 void Field_varstring::sort_string(char *to,uint length)
 {
-  uint tot_length= uint2korr(ptr);
+  uint tot_length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
   tot_length= my_strnxfrm(field_charset,
 			  (uchar*) to, length,
-			  (uchar*) ptr+HA_KEY_BLOB_LENGTH,
+			  (uchar*) ptr + length_bytes,
 			  tot_length);
   if (tot_length < length)
     field_charset->cset->fill(field_charset, to+tot_length,length-tot_length,
@@ -4776,6 +4817,18 @@ void Field_varstring::sort_string(char *to,uint length)
 }
 
 
+enum ha_base_keytype Field_varstring::key_type() const
+{
+  enum ha_base_keytype res;
+
+  if (binary())
+    res= length_bytes == 1 ? HA_KEYTYPE_VARBINARY1 : HA_KEYTYPE_VARBINARY2;
+  else
+    res= length_bytes == 1 ? HA_KEYTYPE_VARTEXT1 : HA_KEYTYPE_VARTEXT2;
+  return res;
+}
+
+
 void Field_varstring::sql_type(String &res) const
 {
   THD *thd= table->in_use;
@@ -4793,9 +4846,14 @@ void Field_varstring::sql_type(String &res) const
 }
 
 
+/*
+  Functions to create a packed row.
+  Here the number of length bytes are depending on the given max_length
+*/
+
 char *Field_varstring::pack(char *to, const char *from, uint max_length)
 {
-  uint length=uint2korr(from);
+  uint length= length_bytes == 1 ? (uint) (uchar) *from : uint2korr(from);
   set_if_smaller(max_length, field_length);
   if (length > max_length)
     length=max_length;
@@ -4803,39 +4861,104 @@ char *Field_varstring::pack(char *to, const char *from, uint max_length)
   if (max_length > 255)
     *to++= (char) (length >> 8);
   if (length)
-    memcpy(to, from+HA_KEY_BLOB_LENGTH, length);
+    memcpy(to, from+length_bytes, length);
   return to+length;
 }
 
 
-char *Field_varstring::pack_key(char *to, const char *from, uint max_length)
+char *Field_varstring::pack_key(char *to, const char *key, uint max_length)
 {
-  uint length=uint2korr(from);
+  uint length=  length_bytes == 1 ? (uint) (uchar) *key : uint2korr(key);
   uint char_length= ((field_charset->mbmaxlen > 1) ?
                      max_length/field_charset->mbmaxlen : max_length);
-  from+= HA_KEY_BLOB_LENGTH;
+  key+= length_bytes;
   if (length > char_length)
   {
-    char_length= my_charpos(field_charset, from, from+length, char_length);
+    char_length= my_charpos(field_charset, key, key+length, char_length);
     set_if_smaller(length, char_length);
   }
   *to++= (char) (length & 255);
   if (max_length > 255)
     *to++= (char) (length >> 8);
   if (length)
-    memcpy(to, from, length);
+    memcpy(to, key, length);
   return to+length;
 }
 
 
+/*
+  Unpack a key into a record buffer.
+
+  SYNOPSIS
+    unpack_key()
+    to                          Pointer into the record buffer.
+    key                         Pointer to the packed key.
+    max_length                  Key length limit from key description.
+
+  DESCRIPTION
+    A VARCHAR key has a maximum size of 64K-1.
+    In its packed form, the length field is one or two bytes long,
+    depending on 'max_length'.
+
+  RETURN
+    Pointer to end of 'key' (To the next key part if multi-segment key)
+*/
+
+const char *Field_varstring::unpack_key(char *to, const char *key,
+                                      uint max_length)
+{
+  /* get length of the blob key */
+  uint32 length= *((uchar*) key++);
+  if (max_length > 255)
+    length+= (*((uchar*) key++)) << 8;
+
+  /* put the length into the record buffer */
+  if (length_bytes == 1)
+    *ptr= (uchar) length;
+  else
+    int2store(ptr, length);
+  memcpy(ptr + length_bytes, key, length);
+  return key + length;
+}
+
+/*
+  Create a packed key that will be used for storage in the index tree
+
+  SYNOPSIS
+    pack_key_from_key_image()
+    to			Store packed key segment here
+    from		Key segment (as given to index_read())
+    max_length  	Max length of key
+
+  RETURN
+    end of key storage
+*/
+
+char *Field_varstring::pack_key_from_key_image(char *to, const char *from,
+                                               uint max_length)
+{
+  /* Key length is always stored as 2 bytes */
+  uint length= uint2korr(from);
+  if (length > max_length)
+    length= max_length;
+  *to++= (char) (length & 255);
+  if (max_length > 255)
+    *to++= (char) (length >> 8);
+  if (length)
+    memcpy(to, from+HA_KEY_BLOB_LENGTH, length);
+  return to+length;
+}
+
+
+/*
+  unpack field packed with Field_varstring::pack()
+*/
+
 const char *Field_varstring::unpack(char *to, const char *from)
 {
   uint length;
-  if (field_length <= 255)
-  {
+  if (length_bytes == 1)
     length= (uint) (uchar) (*to= *from++);
-    to[1]=0;
-  }
   else
   {
     length= uint2korr(from);
@@ -4843,7 +4966,7 @@ const char *Field_varstring::unpack(char *to, const char *from)
     to[1]= *from++;
   }
   if (length)
-    memcpy(to+HA_KEY_BLOB_LENGTH, from, length);
+    memcpy(to+ length_bytes, from, length);
   return from+length;
 }
 
@@ -4851,12 +4974,11 @@ const char *Field_varstring::unpack(char *to, const char *from)
 int Field_varstring::pack_cmp(const char *a, const char *b, uint key_length,
                               my_bool insert_or_update)
 {
-  uint a_length;
-  uint b_length;
+  uint a_length, b_length;
   if (key_length > 255)
   {
-    a_length=uint2korr(a); a+= HA_KEY_BLOB_LENGTH;
-    b_length=uint2korr(b); b+= HA_KEY_BLOB_LENGTH;
+    a_length=uint2korr(a); a+= 2;
+    b_length=uint2korr(b); b+= 2;
   }
   else
   {
@@ -4873,8 +4995,8 @@ int Field_varstring::pack_cmp(const char *a, const char *b, uint key_length,
 int Field_varstring::pack_cmp(const char *b, uint key_length,
                               my_bool insert_or_update)
 {
-  char *a= ptr+HA_KEY_BLOB_LENGTH;
-  uint a_length= uint2korr(ptr);
+  char *a= ptr+ length_bytes;
+  uint a_length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
   uint b_length;
   uint char_length= ((field_charset->mbmaxlen > 1) ?
                      key_length / field_charset->mbmaxlen : key_length);
@@ -4903,7 +5025,7 @@ int Field_varstring::pack_cmp(const char *b, uint key_length,
 uint Field_varstring::packed_col_length(const char *data_ptr, uint length)
 {
   if (length > 255)
-    return uint2korr(data_ptr)+HA_KEY_BLOB_LENGTH;
+    return uint2korr(data_ptr)+2;
   return (uint) ((uchar) *data_ptr)+1;
 }
 
@@ -4916,13 +5038,14 @@ uint Field_varstring::max_packed_col_length(uint max_length)
 
 void Field_varstring::get_key_image(char *buff, uint length, imagetype type)
 {
-  uint f_length= uint2korr(ptr);
+  uint f_length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
   uint char_length= length / field_charset->mbmaxlen;
-  char_length= my_charpos(field_charset, ptr, ptr + HA_KEY_BLOB_LENGTH,
+  char_length= my_charpos(field_charset, ptr, ptr + length_bytes,
                           char_length);
   set_if_smaller(f_length, char_length);
+  /* Key is always stored with 2 bytes */
   int2store(buff,f_length);
-  memcpy(buff+HA_KEY_BLOB_LENGTH, ptr+HA_KEY_BLOB_LENGTH, f_length);
+  memcpy(buff+HA_KEY_BLOB_LENGTH, ptr+length_bytes, f_length);
   if (f_length < length)
   {
     /*
@@ -4936,18 +5059,12 @@ void Field_varstring::get_key_image(char *buff, uint length, imagetype type)
 
 void Field_varstring::set_key_image(char *buff,uint length)
 {
-  length=uint2korr(buff);			// Real length is here
+  length= uint2korr(buff);			// Real length is here
   (void) Field_varstring::store(buff+HA_KEY_BLOB_LENGTH, length,
                                 field_charset);
 }
 
 
-int Field_varstring::cmp_binary_offset(uint row_offset)
-{
-  return cmp_binary(ptr, ptr+row_offset);
-}
-
-
 int Field_varstring::cmp_binary(const char *a_ptr, const char *b_ptr,
                                 uint32 max_length)
 {
@@ -4955,13 +5072,41 @@ int Field_varstring::cmp_binary(const char *a_ptr, const char *b_ptr,
   uint diff;
   uint32 a_length,b_length;
 
-  a_length= uint2korr(a_ptr);
-  b_length= uint2korr(b_ptr);
+  if (length_bytes == 1)
+  {
+    a_length= (uint) (uchar) *a_ptr;
+    b_length= (uint) (uchar) *b_ptr;
+  }
+  else
+  {
+    a_length= uint2korr(a_ptr);
+    b_length= uint2korr(b_ptr);
+  }
   set_if_smaller(a_length, max_length);
   set_if_smaller(b_length, max_length);
   if (a_length != b_length)
     return 1;
-  return memcmp(a_ptr+2, b_ptr+2, a_length);
+  return memcmp(a_ptr+length_bytes, b_ptr+length_bytes, a_length);
+}
+
+
+Field *Field_varstring::new_field(MEM_ROOT *root, struct st_table *new_table)
+{
+  Field_varstring *res= (Field_varstring*) Field::new_field(root, new_table);
+  if (res)
+    res->length_bytes= length_bytes;
+  return res;
+}
+
+
+Field *Field_varstring::new_key_field(MEM_ROOT *root,
+                                      struct st_table *new_table)
+{
+  Field_varstring *res= (Field_varstring*) Field::new_field(root, new_table);
+  /* Keys length prefixes are always packed with 2 bytes */
+  if (res)
+    res->length_bytes= 2;
+  return res;
 }
 
 
@@ -5218,18 +5363,6 @@ int Field_blob::cmp(const char *a_ptr, const char *b_ptr)
 }
 
 
-int Field_blob::cmp_offset(uint row_offset)
-{
-  return Field_blob::cmp(ptr,ptr+row_offset);
-}
-
-
-int Field_blob::cmp_binary_offset(uint row_offset)
-{
-  return cmp_binary(ptr, ptr+row_offset);
-}
-
-
 int Field_blob::cmp_binary(const char *a_ptr, const char *b_ptr,
 			   uint32 max_length)
 {
@@ -5416,8 +5549,7 @@ const char *Field_blob::unpack(char *to, const char *from)
 int Field_blob::pack_cmp(const char *a, const char *b, uint key_length,
                          my_bool insert_or_update)
 {
-  uint a_length;
-  uint b_length;
+  uint a_length, b_length;
   if (key_length > 255)
   {
     a_length=uint2korr(a); a+=2;
@@ -5523,6 +5655,7 @@ const char *Field_blob::unpack_key(char *to, const char *from, uint max_length)
   return from + length;
 }
 
+
 /* Create a packed key that will be used for storage from a MySQL key */
 
 char *Field_blob::pack_key_from_key_image(char *to, const char *from,
@@ -6047,15 +6180,17 @@ void create_field::create_length_to_internal_length(void)
   case MYSQL_TYPE_STRING:
   case MYSQL_TYPE_VARCHAR:
     length*= charset->mbmaxlen;
-    key_length*= charset->mbmaxlen;
+    key_length= length;
     pack_length= calc_pack_length(sql_type, length);
     break;
   case MYSQL_TYPE_ENUM:
   case MYSQL_TYPE_SET:
+    /* Pack_length already calculated in sql_parse.cc */
     length*= charset->mbmaxlen;
+    key_length= pack_length;
     break;
   default:
-    /* do nothing */
+    key_length= pack_length= calc_pack_length(sql_type, length);
     break;
   }
 }
@@ -6086,7 +6221,7 @@ uint32 calc_pack_length(enum_field_types type,uint32 length)
   case MYSQL_TYPE_VAR_STRING:
   case MYSQL_TYPE_STRING:
   case FIELD_TYPE_DECIMAL:     return (length);
-  case MYSQL_TYPE_VARCHAR:     return (length+HA_KEY_BLOB_LENGTH);
+  case MYSQL_TYPE_VARCHAR:     return (length + (length < 256 ? 1: 2));
   case FIELD_TYPE_YEAR:
   case FIELD_TYPE_TINY	: return 1;
   case FIELD_TYPE_SHORT : return 2;
@@ -6166,7 +6301,9 @@ Field *make_field(char *ptr, uint32 field_length,
                                 unireg_check, field_name, table,
                                 field_charset);
       if (field_type == MYSQL_TYPE_VARCHAR)
-        return new Field_varstring(ptr,field_length,null_pos,null_bit,
+        return new Field_varstring(ptr,field_length,
+                                   HA_VARCHAR_PACKLENGTH(field_length),
+                                   null_pos,null_bit,
                                    unireg_check, field_name, table,
                                    field_charset);
       return 0;                                 // Error
@@ -6313,9 +6450,8 @@ create_field::create_field(Field *old_field,Field *orig_field)
   case MYSQL_TYPE_SET:
   case MYSQL_TYPE_VARCHAR:
   case MYSQL_TYPE_VAR_STRING:
-    /* These are corrected in create_length_to_internal_length */
+    /* This is corrected in create_length_to_internal_length */
     length= (length+charset->mbmaxlen-1) / charset->mbmaxlen;
-    key_length/= charset->mbmaxlen;
     break;
 #ifdef HAVE_SPATIAL
   case FIELD_TYPE_GEOMETRY:
diff --git a/sql/field.h b/sql/field.h
index 4353780f9a4..521bfb818b0 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -139,10 +139,9 @@ public:
   virtual int cmp(const char *,const char *)=0;
   virtual int cmp_binary(const char *a,const char *b, uint32 max_length=~0L)
   { return memcmp(a,b,pack_length()); }
-  virtual int cmp_offset(uint row_offset)
-  { return memcmp(ptr,ptr+row_offset,pack_length()); }
-  virtual int cmp_binary_offset(uint row_offset)
-  { return memcmp(ptr,ptr+row_offset,pack_length()); }
+  int cmp_offset(uint row_offset) { return cmp(ptr,ptr+row_offset); }
+  int cmp_binary_offset(uint row_offset)
+  { return cmp_binary(ptr, ptr+row_offset); };
   virtual int key_cmp(const byte *a,const byte *b)
   { return cmp((char*) a,(char*) b); }
   virtual int key_cmp(const byte *str, uint length)
@@ -185,6 +184,10 @@ public:
   virtual bool can_be_compared_as_longlong() const { return FALSE; }
   virtual void free() {}
   virtual Field *new_field(MEM_ROOT *root, struct st_table *new_table);
+  virtual Field *new_key_field(MEM_ROOT *root, struct st_table *new_table)
+  {
+    return new_field(root, new_table);
+  }
   inline void move_field(char *ptr_arg,uchar *null_ptr_arg,uchar null_bit_arg)
   {
     ptr=ptr_arg; null_ptr=null_ptr_arg; null_bit=null_bit_arg;
@@ -925,26 +928,31 @@ public:
 
 class Field_varstring :public Field_str {
 public:
-  Field_varstring(char *ptr_arg, uint32 len_arg,uchar *null_ptr_arg,
+  /* Store number of bytes used to store length (1 or 2) */
+  uint32 length_bytes;
+  Field_varstring(char *ptr_arg,
+                  uint32 len_arg, uint length_bytes_arg,
+                  uchar *null_ptr_arg,
 		  uchar null_bit_arg,
 		  enum utype unireg_check_arg, const char *field_name_arg,
 		  struct st_table *table_arg, CHARSET_INFO *cs)
     :Field_str(ptr_arg, len_arg, null_ptr_arg, null_bit_arg,
-	       unireg_check_arg, field_name_arg, table_arg, cs)
+	       unireg_check_arg, field_name_arg, table_arg, cs),
+    length_bytes(length_bytes_arg)
   {}
   Field_varstring(uint32 len_arg,bool maybe_null_arg,
 		  const char *field_name_arg,
 		  struct st_table *table_arg, CHARSET_INFO *cs)
     :Field_str((char*) 0,len_arg, maybe_null_arg ? (uchar*) "": 0,0,
-	       NONE, field_name_arg, table_arg, cs)
+	       NONE, field_name_arg, table_arg, cs),
+    length_bytes(len_arg < 256 ? 1 :2)
   {}
 
   enum_field_types type() const { return MYSQL_TYPE_VARCHAR; }
-  enum ha_base_keytype key_type() const
-    { return binary() ? HA_KEYTYPE_VARBINARY : HA_KEYTYPE_VARTEXT; }
+  enum ha_base_keytype key_type() const;
   bool zero_pack() const { return 0; }
-  void reset(void) { bzero(ptr,field_length+2); }
-  uint32 pack_length() const { return (uint32) field_length+2; }
+  void reset(void) { bzero(ptr,field_length+length_bytes); }
+  uint32 pack_length() const { return (uint32) field_length+length_bytes; }
   uint32 key_length() const { return (uint32) field_length; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(longlong nr);
@@ -959,12 +967,13 @@ public:
   void sql_type(String &str) const;
   char *pack(char *to, const char *from, uint max_length=~(uint) 0);
   char *pack_key(char *to, const char *from, uint max_length);
+  char *pack_key_from_key_image(char* to, const char *from, uint max_length);
   const char *unpack(char* to, const char *from);
+  const char *unpack_key(char* to, const char *from, uint max_length);
   int pack_cmp(const char *a, const char *b, uint key_length,
                my_bool insert_or_update);
   int pack_cmp(const char *b, uint key_length,my_bool insert_or_update);
   int cmp_binary(const char *a,const char *b, uint32 max_length=~0L);
-  int cmp_binary_offset(uint row_offset);
   int key_cmp(const byte *,const byte*);
   int key_cmp(const byte *str, uint length);
   uint packed_col_length(const char *to, uint length);
@@ -974,6 +983,8 @@ public:
   bool has_charset(void) const
   { return charset() == &my_charset_bin ? FALSE : TRUE; }
   field_cast_enum field_cast_type() { return FIELD_CAST_VARSTRING; }
+  Field *new_field(MEM_ROOT *root, struct st_table *new_table);
+  Field *new_key_field(MEM_ROOT *root, struct st_table *new_table);
 };
 
 
@@ -996,7 +1007,7 @@ public:
   }
   enum_field_types type() const { return FIELD_TYPE_BLOB;}
   enum ha_base_keytype key_type() const
-    { return binary() ? HA_KEYTYPE_VARBINARY : HA_KEYTYPE_VARTEXT; }
+    { return binary() ? HA_KEYTYPE_VARBINARY2 : HA_KEYTYPE_VARTEXT2; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(double nr);
   int  store(longlong nr);
@@ -1005,9 +1016,7 @@ public:
   String *val_str(String*,String *);
   int cmp(const char *,const char*);
   int cmp(const char *a, uint32 a_length, const char *b, uint32 b_length);
-  int cmp_offset(uint offset);
   int cmp_binary(const char *a,const char *b, uint32 max_length=~0L);
-  int cmp_binary_offset(uint row_offset);
   int key_cmp(const byte *,const byte*);
   int key_cmp(const byte *str, uint length);
   uint32 key_length() const { return 0; }
@@ -1054,9 +1063,9 @@ public:
     return 0;
   }
   char *pack(char *to, const char *from, uint max_length= ~(uint) 0);
-  const char *unpack(char *to, const char *from);
   char *pack_key(char *to, const char *from, uint max_length);
   char *pack_key_from_key_image(char* to, const char *from, uint max_length);
+  const char *unpack(char *to, const char *from);
   const char *unpack_key(char* to, const char *from, uint max_length);
   int pack_cmp(const char *a, const char *b, uint key_length,
                my_bool insert_or_update);
@@ -1091,7 +1100,7 @@ public:
      :Field_blob(len_arg, maybe_null_arg, field_name_arg,
                  table_arg, &my_charset_bin)
   { geom_type= geom_type_arg; }
-  enum ha_base_keytype key_type() const { return HA_KEYTYPE_VARBINARY; }
+  enum ha_base_keytype key_type() const { return HA_KEYTYPE_VARBINARY2; }
   enum_field_types type() const { return FIELD_TYPE_GEOMETRY; }
   void sql_type(String &str) const;
   int  store(const char *to, uint length, CHARSET_INFO *charset);
diff --git a/sql/field_conv.cc b/sql/field_conv.cc
index f6cc851639a..1a175b95bdc 100644
--- a/sql/field_conv.cc
+++ b/sql/field_conv.cc
@@ -305,7 +305,8 @@ static void do_field_string(Copy_field *copy)
   char buff[MAX_FIELD_WIDTH];
   copy->tmp.set_quick(buff,sizeof(buff),copy->tmp.charset());
   copy->from_field->val_str(&copy->tmp);
-  copy->to_field->store(copy->tmp.c_ptr_quick(),copy->tmp.length(),copy->tmp.charset());
+  copy->to_field->store(copy->tmp.c_ptr_quick(),copy->tmp.length(),
+                        copy->tmp.charset());
 }
 
 
@@ -350,7 +351,23 @@ static void do_expand_string(Copy_field *copy)
                      copy->to_length-copy->from_length, ' ');
 }
 
-static void do_varstring(Copy_field *copy)
+
+static void do_varstring1(Copy_field *copy)
+{
+  uint length= (uint) *(uchar*) copy->from_ptr;
+  if (length > copy->to_length- 1)
+  {
+    length=copy->to_length - 1;
+    if (current_thd->count_cuted_fields)
+      copy->to_field->set_warning(MYSQL_ERROR::WARN_LEVEL_WARN,
+                                  ER_WARN_DATA_TRUNCATED, 1);
+  }
+  *(uchar*) copy->to_ptr= (uchar) length;
+  memcpy(copy->to_ptr+1, copy->from_ptr + 1, length);
+}
+
+
+static void do_varstring2(Copy_field *copy)
 {
   uint length=uint2korr(copy->from_ptr);
   if (length > copy->to_length- HA_KEY_BLOB_LENGTH)
@@ -505,9 +522,15 @@ void (*Copy_field::get_copy_func(Field *to,Field *from))(Copy_field*)
       }
       else if (to->charset() != from->charset())
 	return do_field_string;
-      else if (to->real_type() == MYSQL_TYPE_VARCHAR && to_length !=
-	       from_length)
-	return do_varstring;
+      else if (to->real_type() == MYSQL_TYPE_VARCHAR)
+      {
+        if (((Field_varstring*) to)->length_bytes !=
+            ((Field_varstring*) from)->length_bytes)
+          return do_field_string;
+        if (to_length != from_length)
+          return (((Field_varstring*) to)->length_bytes == 1 ?
+                  do_varstring1 : do_varstring2);
+      }
       else if (to_length < from_length)
 	return do_cut_string;
       else if (to_length > from_length)
@@ -587,6 +610,12 @@ void field_conv(Field *to,Field *from)
     char buff[MAX_FIELD_WIDTH];
     String result(buff,sizeof(buff),from->charset());
     from->val_str(&result);
+    /*
+      We use c_ptr_quick() here to make it easier if to is a float/double
+      as the conversion routines will do a copy of the result doesn't
+      end with \0. Can be replaced with .ptr() when we have our own
+      string->double conversion.
+    */
     to->store(result.c_ptr_quick(),result.length(),from->charset());
   }
   else if (from->result_type() == REAL_RESULT)
diff --git a/sql/ha_berkeley.cc b/sql/ha_berkeley.cc
index 6cb83624eff..322126ff47b 100644
--- a/sql/ha_berkeley.cc
+++ b/sql/ha_berkeley.cc
@@ -356,7 +356,8 @@ ulong ha_berkeley::index_flags(uint idx, uint part, bool all_parts) const
     }
     switch (table->key_info[idx].key_part[i].field->key_type()) {
     case HA_KEYTYPE_TEXT:
-    case HA_KEYTYPE_VARTEXT:
+    case HA_KEYTYPE_VARTEXT1:
+    case HA_KEYTYPE_VARTEXT2:
       /*
         As BDB stores only one copy of equal strings, we can't use key read
         on these. Binary collations do support key read though.
@@ -391,6 +392,7 @@ berkeley_cmp_packed_key(DB *file, const DBT *new_key, const DBT *saved_key)
   KEY_PART_INFO *key_part= key->key_part, *end=key_part+key->key_parts;
   uint key_length=new_key->size;
 
+  DBUG_DUMP("key_in_index", saved_key_ptr, saved_key->size);
   for (; key_part != end && (int) key_length > 0; key_part++)
   {
     int cmp;
@@ -745,11 +747,11 @@ void ha_berkeley::unpack_row(char *record, DBT *row)
 
 void ha_berkeley::unpack_key(char *record, DBT *key, uint index)
 {
-  KEY *key_info=table->key_info+index;
+  KEY *key_info= table->key_info+index;
   KEY_PART_INFO *key_part= key_info->key_part,
-		*end=key_part+key_info->key_parts;
+		*end= key_part+key_info->key_parts;
+  char *pos= (char*) key->data;
 
-  char *pos=(char*) key->data;
   for (; key_part != end; key_part++)
   {
     if (key_part->null_bit)
@@ -773,8 +775,10 @@ void ha_berkeley::unpack_key(char *record, DBT *key, uint index)
 
 
 /*
-  Create a packed key from from a row
-  This will never fail as the key buffer is pre allocated.
+  Create a packed key from a row. This key will be written as such
+  to the index tree.
+
+  This will never fail as the key buffer is pre-allocated.
 */
 
 DBT *ha_berkeley::create_key(DBT *key, uint keynr, char *buff,
@@ -820,7 +824,10 @@ DBT *ha_berkeley::create_key(DBT *key, uint keynr, char *buff,
 
 
 /*
-  Create a packed key from from a MySQL unpacked key
+  Create a packed key from from a MySQL unpacked key (like the one that is
+  sent from the index_read()
+
+  This key is to be used to read a row
 */
 
 DBT *ha_berkeley::pack_key(DBT *key, uint keynr, char *buff,
@@ -1457,7 +1464,7 @@ int ha_berkeley::read_row(int error, char *buf, uint keynr, DBT *row,
 int ha_berkeley::index_read_idx(byte * buf, uint keynr, const byte * key,
 				uint key_len, enum ha_rkey_function find_flag)
 {
-  statistic_increment(table->in_use->status_var.ha_read_key_count,&LOCK_status);
+  table->in_use->status_var.ha_read_key_count++;
   DBUG_ENTER("index_read_idx");
   current_row.flags=DB_DBT_REALLOC;
   active_index=MAX_KEY;
@@ -1476,10 +1483,9 @@ int ha_berkeley::index_read(byte * buf, const byte * key,
   int error;
   KEY *key_info= &table->key_info[active_index];
   int do_prev= 0;
-
   DBUG_ENTER("ha_berkeley::index_read");
 
-  statistic_increment(table->in_use->status_var.ha_read_key_count,&LOCK_status);
+  table->in_use->status_var.ha_read_key_count++;
   bzero((char*) &row,sizeof(row));
   if (find_flag == HA_READ_BEFORE_KEY)
   {
@@ -1679,6 +1685,7 @@ DBT *ha_berkeley::get_pos(DBT *to, byte *pos)
       pos+=key_part->field->packed_col_length((char*) pos,key_part->length);
     to->size= (uint) (pos- (byte*) to->data);
   }
+  DBUG_DUMP("key", (char*) to->data, to->size);
   return to;
 }
 
diff --git a/sql/ha_heap.cc b/sql/ha_heap.cc
index 60555d51402..1556a18bfca 100644
--- a/sql/ha_heap.cc
+++ b/sql/ha_heap.cc
@@ -488,8 +488,10 @@ int ha_heap::create(const char *name, TABLE *table_arg,
       else
       {
         if ((seg->type = field->key_type()) != (int) HA_KEYTYPE_TEXT &&
-            seg->type != HA_KEYTYPE_VARTEXT &&
-            seg->type != HA_KEYTYPE_VARBINARY)
+            seg->type != HA_KEYTYPE_VARTEXT1 &&
+            seg->type != HA_KEYTYPE_VARTEXT2 &&
+            seg->type != HA_KEYTYPE_VARBINARY1 &&
+            seg->type != HA_KEYTYPE_VARBINARY2)
           seg->type= HA_KEYTYPE_BINARY;
       }
       seg->start=   (uint) key_part->offset;
diff --git a/sql/ha_myisam.cc b/sql/ha_myisam.cc
index 726647cd131..dab3ea16377 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -926,8 +926,11 @@ int ha_myisam::enable_indexes(uint mode)
     {
       sql_print_warning("Warning: Enabling keys got errno %d, retrying",
                         my_errno);
+      thd->clear_error();
       param.testflag&= ~(T_REP_BY_SORT | T_QUICK);
       error= (repair(thd,param,0) != HA_ADMIN_OK);
+      if (!error && thd->net.report_error)
+        error= HA_ERR_CRASHED;
     }
     info(HA_STATUS_CONST);
     thd->proc_info=save_proc_info;
@@ -1471,11 +1474,10 @@ int ha_myisam::create(const char *name, register TABLE *table_arg,
       break;
 
     if (found->flags & BLOB_FLAG)
-    {
       recinfo_pos->type= (int) FIELD_BLOB;
-    }
-    else if (!(options & HA_OPTION_PACK_RECORD) ||
-             found->type() == MYSQL_TYPE_VARCHAR)
+    else if (found->type() == MYSQL_TYPE_VARCHAR)
+      recinfo_pos->type= FIELD_VARCHAR;
+    else if (!(options & HA_OPTION_PACK_RECORD))
       recinfo_pos->type= (int) FIELD_NORMAL;
     else if (found->zero_pack())
       recinfo_pos->type= (int) FIELD_SKIP_ZERO;
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index b242698d36e..949545bcdb0 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -1214,7 +1214,7 @@ int composite_key_cmp(void* arg, byte* key1, byte* key2)
   {
     Field* f = *field;
     int len = *lengths++;
-    int res = f->key_cmp(key1, key2);
+    int res = f->cmp(key1, key2);
     if (res)
       return res;
     key1 += len;
@@ -1668,7 +1668,7 @@ int group_concat_key_cmp_with_distinct(void* arg, byte* key1,
     {
       int res;
       uint offset= (uint) (field->ptr - record);
-      if ((res= field->key_cmp(key1 + offset, key2 + offset)))
+      if ((res= field->cmp(key1 + offset, key2 + offset)))
 	return res;
     }
   }
@@ -1702,7 +1702,7 @@ int group_concat_key_cmp_with_order(void* arg, byte* key1, byte* key2)
     {
       int res;
       uint offset= (uint) (field->ptr - record);
-      if ((res= field->key_cmp(key1 + offset, key2 + offset)))
+      if ((res= field->cmp(key1 + offset, key2 + offset)))
         return (*order_item)->asc ? res : -res;
     }
   }
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index f9149f10a30..0f54f06a22b 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -746,7 +746,7 @@ int QUICK_RANGE_SELECT::init()
 void QUICK_RANGE_SELECT::range_end()
 {
   if (file->inited != handler::NONE)
-    file->ha_index_end();
+    file->ha_index_or_rnd_end();
 }
 
 
@@ -3687,7 +3687,8 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
   }
   /* Get local copy of key */
   copies= 1;
-  if (field->key_type() == HA_KEYTYPE_VARTEXT)
+  if (field->key_type() == HA_KEYTYPE_VARTEXT1 ||
+      field->key_type() == HA_KEYTYPE_VARTEXT2)
     copies= 2;
   str= str2= (char*) alloc_root(param->mem_root,
 				(key_part->store_length)*copies+1);
@@ -5888,7 +5889,7 @@ int QUICK_RANGE_SELECT::get_next()
   SYNOPSIS
     QUICK_RANGE_SELECT::get_next_prefix()
     prefix_length  length of cur_prefix
-    cur_prefix     prefix of a key to be searached for
+    cur_prefix     prefix of a key to be searched for
 
   DESCRIPTION
     Each subsequent call to the method retrieves the first record that has a
@@ -7402,7 +7403,8 @@ TRP_GROUP_MIN_MAX::make_quick(PARAM *param, bool retrieve_full_rows,
       quick->quick_prefix_select= NULL; /* Can't construct a quick select. */
     else
       /* Make a QUICK_RANGE_SELECT to be used for group prefix retrieval. */
-      quick->quick_prefix_select= get_quick_select(param, param_idx, index_tree,
+      quick->quick_prefix_select= get_quick_select(param, param_idx,
+                                                   index_tree,
                                                    &quick->alloc);
 
     /*
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index b4a2f368bc2..2500769ee30 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2467,6 +2467,8 @@ find_item_in_list(Item *find, List<Item> &items, uint *counter,
   bool found_unaliased_non_uniq= 0;
   uint unaliased_counter;
 
+  LINT_INIT(unaliased_counter);                 // Dependent on found_unaliased
+
   *unaliased= FALSE;
 
   if (find->type() == Item::FIELD_ITEM	|| find->type() == Item::REF_ITEM)
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 3d4252a2b17..4810756c40d 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -4891,11 +4891,9 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
     break;
   case MYSQL_TYPE_VARCHAR:
     /*
-      We can't use pack_length as this includes the field length
       Long VARCHAR's are automaticly converted to blobs in mysql_prepare_table
       if they don't have a default value
     */
-    new_field->key_length= new_field->length;
     max_field_charlength= MAX_FIELD_VARCHARLENGTH;
     break;
   case MYSQL_TYPE_STRING:
@@ -5083,16 +5081,12 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
     my_error(ER_WRONG_FIELD_SPEC, MYF(0), field_name);
     DBUG_RETURN(1);
   }
-  if (!new_field->pack_length)
-    new_field->pack_length= calc_pack_length(new_field->sql_type,
-                                             new_field->length);
-  if (!new_field->key_length)
-    new_field->key_length= new_field->pack_length;
   lex->create_list.push_back(new_field);
   lex->last_field=new_field;
   DBUG_RETURN(0);
 }
 
+
 /* Store position for column in ALTER TABLE .. ADD column */
 
 void store_position_for_column(const char *name)
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index ed3606856a0..5f69074adcc 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -235,9 +235,7 @@ bool handle_select(THD *thd, LEX *lex, select_result *result)
   res|= thd->net.report_error;
   if (unlikely(res))
   {
-    /*
-      If we have real error reported erly then this will be ignored
-    */
+    /* If we had a another error reported earlier then this will be ignored */
     result->send_error(ER_UNKNOWN_ERROR, ER(ER_UNKNOWN_ERROR));
     result->abort();
   }
@@ -4873,7 +4871,7 @@ static bool create_ref_for_key(JOIN *join, JOIN_TAB *j, KEYUSE *org_keyuse,
 				  &keyinfo->key_part[i],
 				  (char*) key_buff,maybe_null);
       /*
-	Remeber if we are going to use REF_OR_NULL
+	Remember if we are going to use REF_OR_NULL
 	But only if field _really_ can be null i.e. we force JT_REF
 	instead of JT_REF_OR_NULL in case if field can't be null
       */
@@ -7538,7 +7536,7 @@ static Field* create_tmp_field_from_field(THD *thd, Field* org_field,
 {
   Field *new_field;
 
-  if (convert_blob_length && org_field->flags & BLOB_FLAG)
+  if (convert_blob_length && (org_field->flags & BLOB_FLAG))
     new_field= new Field_varstring(convert_blob_length,
                                    org_field->maybe_null(),
                                    org_field->field_name, table,
@@ -7805,7 +7803,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
     else for (ORDER *tmp=group ; tmp ; tmp=tmp->next)
     {
       (*tmp->item)->marker=4;			// Store null in key
-      if ((*tmp->item)->max_length >= MAX_CHAR_WIDTH)
+      if ((*tmp->item)->max_length >= CONVERT_IF_BIGGER_TO_BLOB)
 	using_unique_constraint=1;
     }
     if (param->group_length >= MAX_BLOB_WIDTH)
@@ -8147,16 +8145,17 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
       key_part_info->null_bit=0;
       key_part_info->field=  field;
       key_part_info->offset= field->offset();
-      key_part_info->length= (uint16) field->pack_length();
+      key_part_info->length= (uint16) field->key_length();
       key_part_info->type=   (uint8) field->key_type();
       key_part_info->key_type =
 	((ha_base_keytype) key_part_info->type == HA_KEYTYPE_TEXT ||
-	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT) ?
+	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT1 ||
+	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT2) ?
 	0 : FIELDFLAG_BINARY;
       if (!using_unique_constraint)
       {
 	group->buff=(char*) group_buff;
-	if (!(group->field=field->new_field(thd->mem_root,table)))
+	if (!(group->field=field->new_key_field(thd->mem_root,table)))
 	  goto err; /* purecov: inspected */
 	if (maybe_null)
 	{
@@ -8177,7 +8176,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
 	  group->field->move_field((char*) group_buff);
         /* In GROUP BY 'a' and 'a ' are equal for VARCHAR fields */
         key_part_info->key_part_flag|= HA_END_SPACE_ARE_EQUAL;
-	group_buff+= key_part_info->length;
+	group_buff+= group->field->pack_length();
       }
       keyinfo->key_length+=  key_part_info->length;
     }
@@ -8241,7 +8240,8 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
       key_part_info->type=     (uint8) (*reg_field)->key_type();
       key_part_info->key_type =
 	((ha_base_keytype) key_part_info->type == HA_KEYTYPE_TEXT ||
-	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT) ?
+	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT1 ||
+	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT2) ?
 	0 : FIELDFLAG_BINARY;
     }
   }
@@ -8291,8 +8291,8 @@ static bool create_myisam_tmp_table(TABLE *table,TMP_TABLE_PARAM *param,
   MI_KEYDEF keydef;
   MI_UNIQUEDEF uniquedef;
   KEY *keyinfo=param->keyinfo;
-
   DBUG_ENTER("create_myisam_tmp_table");
+
   if (table->keys)
   {						// Get keys for ni_create
     bool using_unique_constraint=0;
@@ -8340,19 +8340,18 @@ static bool create_myisam_tmp_table(TABLE *table,TMP_TABLE_PARAM *param,
       {
 	seg->type=
 	((keyinfo->key_part[i].key_type & FIELDFLAG_BINARY) ?
-	 HA_KEYTYPE_VARBINARY : HA_KEYTYPE_VARTEXT);
-	seg->bit_start=seg->length - table->blob_ptr_size;
+	 HA_KEYTYPE_VARBINARY2 : HA_KEYTYPE_VARTEXT2);
+	seg->bit_start= field->pack_length() - table->blob_ptr_size;
 	seg->flag= HA_BLOB_PART;
 	seg->length=0;			// Whole blob in unique constraint
       }
       else
       {
-	seg->type= ((keyinfo->key_part[i].key_type & FIELDFLAG_BINARY) ?
-                    HA_KEYTYPE_BINARY : HA_KEYTYPE_TEXT);
+	seg->type= keyinfo->key_part[i].type;
         /* Tell handler if it can do suffic space compression */
 	if (field->real_type() == MYSQL_TYPE_STRING &&
 	    keyinfo->key_part[i].length > 4)
-	  seg->flag|=HA_SPACE_PACK;
+	  seg->flag|= HA_SPACE_PACK;
       }
       if (!(field->flags & NOT_NULL_FLAG))
       {
@@ -8361,7 +8360,7 @@ static bool create_myisam_tmp_table(TABLE *table,TMP_TABLE_PARAM *param,
 	/*
 	  We are using a GROUP BY on something that contains NULL
 	  In this case we have to tell MyISAM that two NULL should
-	  on INSERT be compared as equal
+	  on INSERT be regarded at the same value
 	*/
 	if (!using_unique_constraint)
 	  keydef.flag|= HA_NULL_ARE_EQUAL;
@@ -8645,21 +8644,19 @@ do_select(JOIN *join,List<Item> *fields,TABLE *table,Procedure *procedure)
   }
   if (table)
   {
-    int tmp;
+    int tmp, new_errno= 0;
     if ((tmp=table->file->extra(HA_EXTRA_NO_CACHE)))
     {
       DBUG_PRINT("error",("extra(HA_EXTRA_NO_CACHE) failed"));
-      my_errno= tmp;
-      error= -1;
+      new_errno= tmp;
     }
     if ((tmp=table->file->ha_index_or_rnd_end()))
     {
       DBUG_PRINT("error",("ha_index_or_rnd_end() failed"));
-      my_errno= tmp;
-      error= -1;
+      new_errno= tmp;
     }
-    if (error == -1)
-      table->file->print_error(my_errno,MYF(0));
+    if (new_errno)
+      table->file->print_error(new_errno,MYF(0));
   }
 #ifndef DBUG_OFF
   if (error)
@@ -9831,13 +9828,19 @@ end_update(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
     DBUG_RETURN(0);
   }
 
-  /* The null bits are already set */
+  /*
+    Copy null bits from group key to table
+    We can't copy all data as the key may have different format
+    as the row data (for example as with VARCHAR keys)
+  */
   KEY_PART_INFO *key_part;
   for (group=table->group,key_part=table->key_info[0].key_part;
        group ;
        group=group->next,key_part++)
-    memcpy(table->record[0]+key_part->offset, group->buff, key_part->length);
-
+  {
+    if (key_part->null_bit)
+      memcpy(table->record[0]+key_part->offset, group->buff, 1);
+  }
   init_tmptable_sum_functions(join->sum_funcs);
   copy_funcs(join->tmp_table_param.items_to_copy);
   if ((error=table->file->write_row(table->record[0])))
@@ -11647,8 +11650,10 @@ calc_group_buffer(JOIN *join,ORDER *group)
     {
       if (field->type() == FIELD_TYPE_BLOB)
 	key_length+=MAX_BLOB_WIDTH;		// Can't be used as a key
+      else if (field->type() == MYSQL_TYPE_VARCHAR)
+        key_length+= field->field_length + HA_KEY_BLOB_LENGTH;
       else
-	key_length+=field->pack_length();
+	key_length+= field->pack_length();
     }
     else if ((*group->item)->result_type() == REAL_RESULT)
       key_length+=sizeof(double);
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 5e42fc0ee30..be3a72836b4 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -420,12 +420,15 @@ class store_key :public Sql_alloc
     :null_ptr(null),err(0)
   {
     if (field_arg->type() == FIELD_TYPE_BLOB)
-      to_field=new Field_varstring(ptr, length, (uchar*) null, 1, 
+    {
+      /* Key segments are always packed with a 2 byte length prefix */
+      to_field=new Field_varstring(ptr, length, 2, (uchar*) null, 1, 
 				   Field::NONE, field_arg->field_name,
 				   field_arg->table, field_arg->charset());
+    }
     else
     {
-      to_field=field_arg->new_field(thd->mem_root,field_arg->table);
+      to_field=field_arg->new_key_field(thd->mem_root, field_arg->table);
       if (to_field)
 	to_field->move_field(ptr, (uchar*) null, 1);
     }
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 71467664085..0a69ddcb40b 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -3319,7 +3319,7 @@ int make_schema_select(THD *thd, SELECT_LEX *sel,
 
 
 /*
-  Fill temporaty schema tables before SELECT
+  Fill temporary schema tables before SELECT
 
   SYNOPSIS
     get_schema_tables_result()
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 56605d1c6e0..ddd713d47cd 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -508,7 +508,8 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
       String conv, *tmp;
       for (uint i= 0; (tmp= it++); i++)
       {
-        if (String::needs_conversion(tmp->length(), tmp->charset(), cs, &dummy))
+        if (String::needs_conversion(tmp->length(), tmp->charset(), cs,
+                                     &dummy))
         {
           uint cnv_errs;
           conv.copy(tmp->ptr(), tmp->length(), tmp->charset(), cs, &cnv_errs);
@@ -3686,7 +3687,7 @@ copy_data_between_tables(TABLE *from,TABLE *to,
   free_io_cache(from);
   delete [] copy;				// This is never 0
 
-  if (to->file->end_bulk_insert() && !error)
+  if (to->file->end_bulk_insert() && error <= 0)
   {
     to->file->print_error(my_errno,MYF(0));
     error=1;