Merge with new VARCHAR code

configure.in:
  Auto merged
BitKeeper/deleted/.del-acinclude.m4~f4ab416bac5003:
  Auto merged
BitKeeper/deleted/.del-ha_isam.cc~4dce65904db2675e:
  Auto merged
BitKeeper/deleted/.del-ha_isammrg.cc~dc682e4755d77a2e:
  Auto merged
client/mysqldump.c:
  Auto merged
client/mysqltest.c:
  Auto merged
heap/hp_create.c:
  Auto merged
heap/hp_delete.c:
  Auto merged
heap/hp_hash.c:
  Auto merged
heap/hp_write.c:
  Auto merged
include/decimal.h:
  Auto merged
include/m_ctype.h:
  Auto merged
libmysql/libmysql.c:
  Auto merged
libmysqld/Makefile.am:
  Auto merged
myisam/mi_check.c:
  Auto merged
myisam/mi_create.c:
  Auto merged
myisam/mi_write.c:
  Auto merged
mysql-test/r/ctype_tis620.result:
  Auto merged
mysql-test/r/ctype_ucs.result:
  Auto merged
mysql-test/r/myisam.result:
  Auto merged
mysql-test/r/mysqldump.result:
  Auto merged
mysql-test/r/order_by.result:
  Auto merged
mysql-test/r/ps.result:
  Auto merged
mysql-test/r/ps_1general.result:
  Auto merged
mysql-test/r/ps_2myisam.result:
  Auto merged
mysql-test/r/ps_3innodb.result:
  Auto merged
mysql-test/r/ps_4heap.result:
  Auto merged
mysql-test/r/ps_5merge.result:
  Auto merged
mysql-test/r/ps_6bdb.result:
  Auto merged
mysql-test/r/select.result:
  Auto merged
mysql-test/r/strict.result:
  Auto merged
mysql-test/r/subselect.result:
  Auto merged
mysql-test/r/type_blob.result:
  Auto merged
mysql-test/t/ctype_ucs.test:
  Auto merged
mysql-test/t/endspace.test:
  Auto merged
mysql-test/t/myisam.test:
  Auto merged
mysql-test/t/ps_1general.test:
  Auto merged
mysql-test/t/strict.test:
  Auto merged
mysql-test/t/type_blob.test:
  Auto merged
ndb/src/common/util/NdbSqlUtil.cpp:
  Auto merged
scripts/mysql_fix_privilege_tables.sh:
  Auto merged
sql/field.h:
  Auto merged
sql/field_conv.cc:
  Auto merged
sql/ha_heap.cc:
  Auto merged
sql/ha_innodb.cc:
  Auto merged
sql/ha_innodb.h:
  Auto merged
sql/ha_myisam.cc:
  Auto merged
sql/ha_ndbcluster.cc:
  Auto merged
sql/handler.h:
  Auto merged
sql/item_cmpfunc.cc:
  Auto merged
sql/item_sum.cc:
  Auto merged
sql/opt_range.cc:
  Auto merged
sql/opt_sum.cc:
  Auto merged
sql/protocol.cc:
  Auto merged
sql/sql_base.cc:
  Auto merged
sql/sql_help.cc:
  Auto merged
sql/sql_select.cc:
  Auto merged
sql/sql_yacc.yy:
  Auto merged
sql/structs.h:
  Auto merged
sql/table.cc:
  Auto merged
strings/ctype-czech.c:
  Auto merged
strings/ctype-uca.c:
  Auto merged
strings/ctype-utf8.c:
  Auto merged
strings/ctype-win1250ch.c:
  Auto merged
strings/decimal.c:
  Auto merged
tests/client_test.c:
  Auto merged
mysql-test/r/bdb.result:
  Merge with VARCHAR code
mysql-test/r/heap.result:
  Merge with VARCHAR code
mysql-test/r/innodb.result:
  Merge with VARCHAR code
mysql-test/r/select.result.es:
  Merge with VARCHAR code
mysql-test/t/bdb.test:
  Merge with VARCHAR code
mysql-test/t/heap.test:
  Merge with VARCHAR code
mysql-test/t/innodb.test:
  Merge with VARCHAR code
sql/field.cc:
  Merge with VARCHAR code
sql/item.cc:
  Merge with VARCHAR code
sql/sql_acl.cc:
  Merge with VARCHAR code
sql/sql_parse.cc:
  Merge with VARCHAR code
sql/sql_table.cc:
  Merge with VARCHAR code
sql/sql_update.cc:
  Merge with VARCHAR code
sql/table.h:
  Merge with VARCHAR code
strings/ctype-mb.c:
  Don't pad my_like_range with max_str for simple LIKE expression
strings/ctype-tis620.c:
  Merge with VARCHAR code
strings/ctype-ucs2.c:
  Added new argument to my_strnncollsp_ucs2()
  Simply code

35 files changed, 815 insertions, 1234 deletions

diff --git a/sql/Makefile.am b/sql/Makefile.am
index a3935835a2d..7d2eda1e222 100644
--- a/sql/Makefile.am
+++ b/sql/Makefile.am
@@ -29,8 +29,7 @@ libexec_PROGRAMS =	mysqld
 noinst_PROGRAMS =	gen_lex_hash
 bin_PROGRAMS =		mysql_tzinfo_to_sql
 gen_lex_hash_LDFLAGS =  @NOINST_LDFLAGS@
-LDADD =			@isam_libs@ \
-			$(top_builddir)/myisam/libmyisam.a \
+LDADD =			$(top_builddir)/myisam/libmyisam.a \
 			$(top_builddir)/myisammrg/libmyisammrg.a \
 			$(top_builddir)/heap/libheap.a \
 			$(top_builddir)/vio/libvio.a \
@@ -51,7 +50,7 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			procedure.h sql_class.h sql_lex.h sql_list.h \
 			sql_manager.h sql_map.h sql_string.h unireg.h \
 			field.h handler.h mysqld_suffix.h \
-			ha_isammrg.h ha_isam.h ha_myisammrg.h\
+			ha_myisammrg.h\
 			ha_heap.h ha_myisam.h ha_berkeley.h ha_innodb.h \
 			ha_ndbcluster.h opt_range.h protocol.h \
 			sql_select.h structs.h table.h sql_udf.h hash_filo.h\
@@ -86,7 +85,7 @@ mysqld_SOURCES =	sql_lex.cc sql_handler.cc \
 		   	records.cc filesort.cc handler.cc \
 		        ha_heap.cc ha_myisam.cc ha_myisammrg.cc \
 	                ha_berkeley.cc ha_innodb.cc \
-			ha_isam.cc ha_isammrg.cc ha_ndbcluster.cc \
+			ha_ndbcluster.cc \
 			sql_db.cc sql_table.cc sql_rename.cc sql_crypt.cc \
 			sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \
 			sql_udf.cc sql_analyse.cc sql_analyse.h sql_cache.cc \
diff --git a/sql/field.cc b/sql/field.cc
index 73caddf1d0b..c06b7006a2b 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -429,8 +429,7 @@ uint Field::fill_cache_field(CACHE_FIELD *copy)
     copy->length-=table->blob_ptr_size;
     return copy->length;
   }
-  else if (!zero_pack() && (type() == FIELD_TYPE_STRING && copy->length > 4 ||
-			    type() == FIELD_TYPE_VAR_STRING))
+  else if (!zero_pack() && (type() == FIELD_TYPE_STRING && copy->length >= 4))
     copy->strip=1;				/* Remove end space */
   else
     copy->strip=0;
@@ -479,6 +478,26 @@ bool Field::optimize_range(uint idx, uint part)
   return test(table->file->index_flags(idx, part, 1) & HA_READ_RANGE);
 }
 
+
+Field *Field::new_field(MEM_ROOT *root, struct st_table *new_table)
+{
+  Field *tmp;
+  if (!(tmp= (Field*) memdup_root(root,(char*) this,size_of())))
+    return 0;
+
+  if (tmp->table->maybe_null)
+    tmp->flags&= ~NOT_NULL_FLAG;
+  tmp->table= new_table;
+  tmp->key_start.init(0);
+  tmp->part_of_key.init(0);
+  tmp->part_of_sortkey.init(0);
+  tmp->unireg_check=Field::NONE;
+  tmp->flags&= (NOT_NULL_FLAG | BLOB_FLAG | UNSIGNED_FLAG |
+                ZEROFILL_FLAG | BINARY_FLAG | ENUM_FLAG | SET_FLAG);
+  tmp->reset_fields();
+  return tmp;
+}
+
 /****************************************************************************
   Field_null, a field that always return NULL
 ****************************************************************************/
@@ -4432,7 +4451,7 @@ int Field_string::cmp(const char *a_ptr, const char *b_ptr)
     return field_charset->coll->strnncollsp(field_charset,
                                             (const uchar*) a_ptr, field_length,
                                             (const uchar*) b_ptr,
-                                            field_length);
+                                            field_length, 0);
   }
   if (field_charset->mbmaxlen != 1)
   {
@@ -4461,20 +4480,22 @@ void Field_string::sql_type(String &res) const
 {
   THD *thd= table->in_use;
   CHARSET_INFO *cs=res.charset();
-  ulong length= cs->cset->snprintf(cs,(char*) res.ptr(),
-			    res.alloced_length(), "%s(%d)",
-			    (field_length > 3 &&
-			     (table->db_options_in_use &
-			      HA_OPTION_PACK_RECORD) ?
-			      (has_charset() ? "varchar" : "varbinary") :
+  ulong length;
+
+  length= cs->cset->snprintf(cs,(char*) res.ptr(),
+                             res.alloced_length(), "%s(%d)",
+                             ((type() == MYSQL_TYPE_VAR_STRING &&
+                               !thd->variables.new_mode) ?
+                              (has_charset() ? "varchar" : "varbinary") :
 			      (has_charset() ? "char" : "binary")),
-			    (int) field_length / charset()->mbmaxlen);
+                             (int) field_length / charset()->mbmaxlen);
   res.length(length);
   if ((thd->variables.sql_mode & (MODE_MYSQL323 | MODE_MYSQL40)) &&
       has_charset() && (charset()->state & MY_CS_BINSORT))
     res.append(" binary");
 }
 
+
 char *Field_string::pack(char *to, const char *from, uint max_length)
 {
   uint length=      min(field_length,max_length);
@@ -4508,10 +4529,27 @@ const char *Field_string::unpack(char *to, const char *from)
 }
 
 
-int Field_string::pack_cmp(const char *a, const char *b, uint length)
+/*
+  Compare two packed keys
+
+  SYNOPSIS
+    pack_cmp()
+     a			New key
+     b			Original key
+     length		Key length
+     insert_or_update	1 if this is an insert or update
+
+  RETURN
+    < 0	  a < b
+    0	  a = b
+    > 0   a > b
+*/
+
+int Field_string::pack_cmp(const char *a, const char *b, uint length,
+                           my_bool insert_or_update)
 {
   uint a_length, b_length;
-  if (field_length > 255)
+  if (length > 255)
   {
     a_length= uint2korr(a);
     b_length= uint2korr(b);
@@ -4523,29 +4561,51 @@ int Field_string::pack_cmp(const char *a, const char *b, uint length)
     a_length= (uint) (uchar) *a++;
     b_length= (uint) (uchar) *b++;
   }
-  return my_strnncoll(field_charset,
-		      (const uchar*)a,a_length,
-		      (const uchar*)b,b_length);
+  return field_charset->coll->strnncollsp(field_charset,
+                                          (const uchar*) a, a_length,
+                                          (const uchar*) b, b_length,
+                                          insert_or_update);
 }
 
 
-int Field_string::pack_cmp(const char *b, uint length)
+/*
+  Compare a packed key against row
+
+  SYNOPSIS
+    pack_cmp()
+     key		Original key
+     length		Key length. (May be less than field length)
+     insert_or_update	1 if this is an insert or update
+
+  RETURN
+    < 0	  row < key
+    0	  row = key
+    > 0   row > key
+*/
+
+int Field_string::pack_cmp(const char *key, uint length,
+                           my_bool insert_or_update)
 {
-  uint b_length;
-  if (field_length > 255)
+  uint row_length, key_length;
+  char *end;
+  if (length > 255)
   {
-    b_length= uint2korr(b);
-    b+= 2;
+    key_length= uint2korr(key);
+    key+= 2;
   }
   else
-    b_length= (uint) (uchar) *b++;
-  char *end= ptr + field_length;
+    key_length= (uint) (uchar) *key++;
+  
+  /* Only use 'length' of key, not field_length */
+  end= ptr + length;
   while (end > ptr && end[-1] == ' ')
     end--;
-  uint a_length = (uint) (end - ptr);
-  return my_strnncoll(field_charset,
-		     (const uchar*)ptr,a_length,
-		     (const uchar*)b, b_length);
+  row_length= (uint) (end - ptr);
+
+  return field_charset->coll->strnncollsp(field_charset,
+                                          (const uchar*) ptr, row_length,
+                                          (const uchar*) key, key_length,
+                                          insert_or_update);
 }
 
 
@@ -4553,16 +4613,31 @@ uint Field_string::packed_col_length(const char *data_ptr, uint length)
 {
   if (length > 255)
     return uint2korr(data_ptr)+2;
-  else
-    return (uint) ((uchar) *data_ptr)+1;
+  return (uint) ((uchar) *data_ptr)+1;
 }
 
+
 uint Field_string::max_packed_col_length(uint max_length)
 {
   return (max_length > 255 ? 2 : 1)+max_length;
 }
 
 
+Field *Field_string::new_field(MEM_ROOT *root, struct st_table *new_table)
+{
+  if (type() != MYSQL_TYPE_VAR_STRING || table == new_table)
+    return Field::new_field(root, new_table);
+
+  /*
+    Old VARCHAR field which should be modified to a VARCHAR on copy
+    This is done to ensure that ALTER TABLE will convert old VARCHAR fields
+    to now VARCHAR fields.
+  */
+  return new Field_varstring(field_length, maybe_null(),
+                             field_name, new_table,
+                             charset());
+}
+
 /****************************************************************************
 ** VARCHAR type  (Not available for the end user yet)
 ****************************************************************************/
@@ -4571,7 +4646,7 @@ uint Field_string::max_packed_col_length(uint max_length)
 int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
 {
   int error= 0;
-  uint32 not_used;
+  uint32 not_used, copy_length;
   char buff[80];
   String tmpstr(buff,sizeof(buff), &my_charset_bin);
 
@@ -4585,15 +4660,21 @@ int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
     if (conv_errors)
       error= 1;
   }
-  if (length > field_length)
-  {
-    length=field_length;
+  /* 
+    Make sure we don't break a multibyte sequence
+    as well as don't copy a malformed data.
+  */
+  copy_length= field_charset->cset->well_formed_len(field_charset,
+						    from,from+length,
+						    field_length/
+						    field_charset->mbmaxlen);
+  memcpy(ptr + HA_KEY_BLOB_LENGTH, from, copy_length);
+  int2store(ptr, copy_length);
+  
+  if (copy_length < length)
     error= 1;
-  }
   if (error)
     set_warning(MYSQL_ERROR::WARN_LEVEL_WARN, ER_WARN_DATA_TRUNCATED, 1);
-  memcpy(ptr+HA_KEY_BLOB_LENGTH,from,length);
-  int2store(ptr, length);
   return error;
 }
 
@@ -4640,17 +4721,49 @@ int Field_varstring::cmp(const char *a_ptr, const char *b_ptr)
   uint a_length=uint2korr(a_ptr);
   uint b_length=uint2korr(b_ptr);
   int diff;
-  diff= my_strnncoll(field_charset,
-		     (const uchar*) a_ptr+HA_KEY_BLOB_LENGTH,
-		     min(a_length,b_length),
-		     (const uchar*) b_ptr+HA_KEY_BLOB_LENGTH,
-		     min(a_length,b_length));
-  return diff ? diff : (int) (a_length - b_length);
+  diff= field_charset->coll->strnncollsp(field_charset,
+                                         (const uchar*) a_ptr+
+                                         HA_KEY_BLOB_LENGTH,
+                                         a_length,
+                                         (const uchar*) b_ptr+
+                                         HA_KEY_BLOB_LENGTH,
+                                         b_length,0);
+  return diff;
+}
+
+
+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;
+
+  char_length= my_charpos(cs, ptr + HA_KEY_BLOB_LENGTH,
+                          ptr + HA_KEY_BLOB_LENGTH + 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);
 }
 
+
+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);
+}
+
+
 void Field_varstring::sort_string(char *to,uint length)
 {
-  uint tot_length=uint2korr(ptr);
+  uint tot_length= uint2korr(ptr);
   tot_length= my_strnxfrm(field_charset,
 			  (uchar*) to, length,
 			  (uchar*) ptr+HA_KEY_BLOB_LENGTH,
@@ -4670,9 +4783,11 @@ void Field_varstring::sql_type(String &res) const
   res.length(length);
 }
 
+
 char *Field_varstring::pack(char *to, const char *from, uint max_length)
 {
   uint length=uint2korr(from);
+  set_if_smaller(max_length, field_length);
   if (length > max_length)
     length=max_length;
   *to++= (char) (length & 255);
@@ -4687,12 +4802,14 @@ char *Field_varstring::pack(char *to, const char *from, uint max_length)
 char *Field_varstring::pack_key(char *to, const char *from, uint max_length)
 {
   uint length=uint2korr(from);
-  uint char_length= (field_charset->mbmaxlen > 1) ?
-              max_length/field_charset->mbmaxlen : max_length;
-  from+=HA_KEY_BLOB_LENGTH;
+  uint char_length= ((field_charset->mbmaxlen > 1) ?
+                     max_length/field_charset->mbmaxlen : max_length);
+  from+= HA_KEY_BLOB_LENGTH;
   if (length > char_length)
+  {
     char_length= my_charpos(field_charset, from, from+length, char_length);
-  set_if_smaller(length, char_length);
+    set_if_smaller(length, char_length);
+  }
   *to++= (char) (length & 255);
   if (max_length > 255)
     *to++= (char) (length >> 8);
@@ -4705,16 +4822,16 @@ char *Field_varstring::pack_key(char *to, const char *from, uint max_length)
 const char *Field_varstring::unpack(char *to, const char *from)
 {
   uint length;
-  if (field_length > 255)
+  if (field_length <= 255)
   {
     length= (uint) (uchar) (*to= *from++);
     to[1]=0;
   }
   else
   {
-    length=uint2korr(from);
-    to[0] = *from++;
-    to[1] = *from++;
+    length= uint2korr(from);
+    to[0]= *from++;
+    to[1]= *from++;
   }
   if (length)
     memcpy(to+HA_KEY_BLOB_LENGTH, from, length);
@@ -4722,77 +4839,122 @@ const char *Field_varstring::unpack(char *to, const char *from)
 }
 
 
-int Field_varstring::pack_cmp(const char *a, const char *b, uint key_length)
+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;
   if (key_length > 255)
   {
-    a_length=uint2korr(a); a+= 2;
-    b_length=uint2korr(b); b+= 2;
+    a_length=uint2korr(a); a+= HA_KEY_BLOB_LENGTH;
+    b_length=uint2korr(b); b+= HA_KEY_BLOB_LENGTH;
   }
   else
   {
     a_length= (uint) (uchar) *a++;
     b_length= (uint) (uchar) *b++;
   }
-  return my_strnncoll(field_charset,
-		     (const uchar*) a, a_length,
-		     (const uchar*) b, b_length);
+  return field_charset->coll->strnncollsp(field_charset,
+                                          (const uchar*) a, a_length,
+                                          (const uchar*) b, b_length,
+                                          insert_or_update);
 }
 
-int Field_varstring::pack_cmp(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);
   uint b_length;
+  uint char_length= ((field_charset->mbmaxlen > 1) ?
+                     key_length / field_charset->mbmaxlen : key_length);
+
   if (key_length > 255)
   {
-    b_length=uint2korr(b); b+= 2;
+    b_length=uint2korr(b); b+= HA_KEY_BLOB_LENGTH;
   }
   else
-  {
     b_length= (uint) (uchar) *b++;
+
+  if (a_length > char_length)
+  {
+    char_length= my_charpos(field_charset, a, a+a_length, char_length);
+    set_if_smaller(a_length, char_length);
   }
-  return my_strnncoll(field_charset,
-		     (const uchar*) a, a_length,
-		     (const uchar*) b, b_length);
+
+  return field_charset->coll->strnncollsp(field_charset,
+                                          (const uchar*) a,
+                                          a_length,
+                                          (const uchar*) b, b_length,
+                                          insert_or_update);
 }
 
+
 uint Field_varstring::packed_col_length(const char *data_ptr, uint length)
 {
   if (length > 255)
     return uint2korr(data_ptr)+HA_KEY_BLOB_LENGTH;
-  else
-    return (uint) ((uchar) *data_ptr)+1;
+  return (uint) ((uchar) *data_ptr)+1;
 }
 
+
 uint Field_varstring::max_packed_col_length(uint max_length)
 {
   return (max_length > 255 ? 2 : 1)+max_length;
 }
 
-void Field_varstring::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
-				    imagetype type)
+
+void Field_varstring::get_key_image(char *buff, uint length, imagetype type)
 {
-  uint f_length=uint2korr(ptr);
-  if (f_length > length)
-    f_length= length;
-  int2store(buff,length);
-  memcpy(buff+HA_KEY_BLOB_LENGTH, ptr+HA_KEY_BLOB_LENGTH, length);
-#ifdef HAVE_purify
+  uint f_length= uint2korr(ptr);
+  uint char_length= length / field_charset->mbmaxlen;
+  char_length= my_charpos(field_charset, ptr, ptr + HA_KEY_BLOB_LENGTH,
+                          char_length);
+  set_if_smaller(f_length, char_length);
+  int2store(buff,f_length);
+  memcpy(buff+HA_KEY_BLOB_LENGTH, ptr+HA_KEY_BLOB_LENGTH, f_length);
   if (f_length < length)
+  {
+    /*
+      Must clear this as we do a memcmp in opt_range.cc to detect
+      identical keys
+    */
     bzero(buff+HA_KEY_BLOB_LENGTH+f_length, (length-f_length));
-#endif
+  }
 }
 
-void Field_varstring::set_key_image(char *buff,uint length, CHARSET_INFO *cs)
+
+void Field_varstring::set_key_image(char *buff,uint length)
 {
   length=uint2korr(buff);			// Real length is here
-  (void) Field_varstring::store(buff+HA_KEY_BLOB_LENGTH, length, cs);
+  (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)
+{
+  char *a,*b;
+  uint diff;
+  uint32 a_length,b_length;
+
+  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);
+}
+
 
 /****************************************************************************
 ** blob type
@@ -5030,10 +5192,10 @@ String *Field_blob::val_str(String *val_buffer __attribute__((unused)),
 int Field_blob::cmp(const char *a,uint32 a_length, const char *b,
 		    uint32 b_length)
 {
-  return field_charset->coll->strnncoll(field_charset, 
-                                        (const uchar*)a, a_length,
-                                        (const uchar*)b, b_length,
-                                        0);
+  return field_charset->coll->strnncollsp(field_charset, 
+                                          (const uchar*)a, a_length,
+                                          (const uchar*)b, b_length,
+                                          0);
 }
 
 
@@ -5080,8 +5242,7 @@ int Field_blob::cmp_binary(const char *a_ptr, const char *b_ptr,
 
 /* The following is used only when comparing a key */
 
-void Field_blob::get_key_image(char *buff,uint length,
-			       CHARSET_INFO *cs, imagetype type)
+void Field_blob::get_key_image(char *buff, uint length, imagetype type)
 {
   uint32 blob_length= get_length(ptr);
   char *blob;
@@ -5116,8 +5277,9 @@ void Field_blob::get_key_image(char *buff,uint length,
 #endif /*HAVE_SPATIAL*/
 
   get_ptr(&blob);
-  uint char_length= length / cs->mbmaxlen;
-  char_length= my_charpos(cs, blob, blob + blob_length, char_length);
+  uint char_length= length / field_charset->mbmaxlen;
+  char_length= my_charpos(field_charset, blob, blob + blob_length,
+                          char_length);
   set_if_smaller(blob_length, char_length);
 
   if ((uint32) length > blob_length)
@@ -5133,10 +5295,11 @@ void Field_blob::get_key_image(char *buff,uint length,
   memcpy(buff+HA_KEY_BLOB_LENGTH, blob, length);
 }
 
-void Field_blob::set_key_image(char *buff,uint length, CHARSET_INFO *cs)
+
+void Field_blob::set_key_image(char *buff,uint length)
 {
   length= uint2korr(buff);
-  (void) Field_blob::store(buff+HA_KEY_BLOB_LENGTH, length, cs);
+  (void) Field_blob::store(buff+HA_KEY_BLOB_LENGTH, length, field_charset);
 }
 
 
@@ -5149,7 +5312,7 @@ int Field_blob::key_cmp(const byte *key_ptr, uint max_key_length)
   uint char_length= max_key_length / cs->mbmaxlen;
   char_length= my_charpos(cs, blob1, blob1+blob_length, char_length);
   set_if_smaller(blob_length, char_length);
-  return Field_blob::cmp(blob1,min(blob_length, max_key_length),
+  return Field_blob::cmp(blob1, blob_length,
 			 (char*) key_ptr+HA_KEY_BLOB_LENGTH,
 			 uint2korr(key_ptr));
 }
@@ -5241,7 +5404,8 @@ const char *Field_blob::unpack(char *to, const char *from)
 
 /* Keys for blobs are like keys on varchars */
 
-int Field_blob::pack_cmp(const char *a, const char *b, uint key_length)
+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;
@@ -5255,13 +5419,15 @@ int Field_blob::pack_cmp(const char *a, const char *b, uint key_length)
     a_length= (uint) (uchar) *a++;
     b_length= (uint) (uchar) *b++;
   }
-  return my_strnncoll(field_charset,
-		     (const uchar*) a, a_length,
-		     (const uchar*) b, b_length);
+  return field_charset->coll->strnncollsp(field_charset,
+                                          (const uchar*) a, a_length,
+                                          (const uchar*) b, b_length,
+                                          insert_or_update);
 }
 
 
-int Field_blob::pack_cmp(const char *b, uint key_length)
+int Field_blob::pack_cmp(const char *b, uint key_length,
+                         my_bool insert_or_update)
 {
   char *a;
   memcpy_fixed(&a,ptr+packlength,sizeof(char*));
@@ -5275,12 +5441,11 @@ int Field_blob::pack_cmp(const char *b, uint key_length)
     b_length=uint2korr(b); b+=2;
   }
   else
-  {
     b_length= (uint) (uchar) *b++;
-  }
-  return my_strnncoll(field_charset,
-		     (const uchar*) a, a_length,
-		     (const uchar*) b, b_length);
+  return field_charset->coll->strnncollsp(field_charset,
+                                          (const uchar*) a, a_length,
+                                          (const uchar*) b, b_length,
+                                          insert_or_update);
 }
 
 /* Create a packed key that will be used for storage from a MySQL row */
@@ -5290,8 +5455,8 @@ char *Field_blob::pack_key(char *to, const char *from, uint max_length)
   char *save=ptr;
   ptr=(char*) from;
   uint32 length=get_length();			// Length of from string
-  uint char_length= (field_charset->mbmaxlen > 1) ?
-              max_length/field_charset->mbmaxlen : max_length;
+  uint char_length= ((field_charset->mbmaxlen > 1) ?
+                     max_length/field_charset->mbmaxlen : max_length);
   if (length)
     get_ptr((char**) &from);
   if (length > char_length)
@@ -5365,14 +5530,15 @@ char *Field_blob::pack_key_from_key_image(char *to, const char *from,
   return to+length;
 }
 
+
 uint Field_blob::packed_col_length(const char *data_ptr, uint length)
 {
   if (length > 255)
     return uint2korr(data_ptr)+2;
-  else
-    return (uint) ((uchar) *data_ptr)+1;
+  return (uint) ((uchar) *data_ptr)+1;
 }
 
+
 uint Field_blob::max_packed_col_length(uint max_length)
 {
   return (max_length > 255 ? 2 : 1)+max_length;
@@ -5381,8 +5547,7 @@ uint Field_blob::max_packed_col_length(uint max_length)
 
 #ifdef HAVE_SPATIAL
 
-void Field_geom::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
-			       imagetype type)
+void Field_geom::get_key_image(char *buff, uint length, imagetype type)
 {
   char *blob;
   const char *dummy;
@@ -5411,11 +5576,6 @@ void Field_geom::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
 }
 
 
-void Field_geom::set_key_image(char *buff, uint length, CHARSET_INFO *cs)
-{
-  Field_blob::set_key_image(buff, length, cs);
-}
-
 void Field_geom::sql_type(String &res) const
 {
   CHARSET_INFO *cs= &my_charset_latin1;
@@ -5864,7 +6024,7 @@ bool Field_num::eq_def(Field *field)
 
 
 /*****************************************************************************
-** Handling of field and create_field
+  Handling of field and create_field
 *****************************************************************************/
 
 void create_field::create_length_to_internal_length(void)
@@ -5876,9 +6036,10 @@ void create_field::create_length_to_internal_length(void)
   case MYSQL_TYPE_BLOB:
   case MYSQL_TYPE_VAR_STRING:
   case MYSQL_TYPE_STRING:
+  case MYSQL_TYPE_VARCHAR:
     length*= charset->mbmaxlen;
-    pack_length= calc_pack_length(sql_type == FIELD_TYPE_VAR_STRING ?
-                                  FIELD_TYPE_STRING : sql_type, length);
+    key_length*= charset->mbmaxlen;
+    pack_length= calc_pack_length(sql_type, length);
     break;
   case MYSQL_TYPE_ENUM:
   case MYSQL_TYPE_SET:
@@ -5890,6 +6051,22 @@ void create_field::create_length_to_internal_length(void)
   }
 }
 
+
+enum_field_types get_blob_type_from_length(ulong length)
+{
+  enum_field_types type;
+  if (length < 256)
+    type= FIELD_TYPE_TINY_BLOB;
+  else if (length < 65536)
+    type= FIELD_TYPE_BLOB;
+  else if (length < 256L*256L*256L)
+    type= FIELD_TYPE_MEDIUM_BLOB;
+  else
+    type= FIELD_TYPE_LONG_BLOB;
+  return type;
+}
+
+
 /*
   Make a field from the .frm file info
 */
@@ -5897,9 +6074,10 @@ void create_field::create_length_to_internal_length(void)
 uint32 calc_pack_length(enum_field_types type,uint32 length)
 {
   switch (type) {
+  case MYSQL_TYPE_VAR_STRING:
   case FIELD_TYPE_STRING:
-  case FIELD_TYPE_DECIMAL: return (length);
-  case FIELD_TYPE_VAR_STRING: return (length+HA_KEY_BLOB_LENGTH);
+  case FIELD_TYPE_DECIMAL:     return (length);
+  case MYSQL_TYPE_VARCHAR:     return (length+HA_KEY_BLOB_LENGTH);
   case FIELD_TYPE_YEAR:
   case FIELD_TYPE_TINY	: return 1;
   case FIELD_TYPE_SHORT : return 2;
@@ -5972,12 +6150,16 @@ Field *make_field(char *ptr, uint32 field_length,
   {
     if (!f_is_packed(pack_flag))
     {
-      if (field_type == FIELD_TYPE_STRING ||
+      if (field_type == MYSQL_TYPE_STRING ||
           field_type == FIELD_TYPE_DECIMAL ||   // 3.23 or 4.0 string
-          field_type == FIELD_TYPE_VAR_STRING)
+          field_type == MYSQL_TYPE_VAR_STRING)
         return new Field_string(ptr,field_length,null_pos,null_bit,
                                 unireg_check, field_name, table,
                                 field_charset);
+      if (field_type == MYSQL_TYPE_VARCHAR)
+        return new Field_varstring(ptr,field_length,null_pos,null_bit,
+                                   unireg_check, field_name, table,
+                                   field_charset);
       return 0;                                 // Error
     }
 
@@ -6090,44 +6272,51 @@ create_field::create_field(Field *old_field,Field *orig_field)
   flags=      old_field->flags;
   unireg_check=old_field->unireg_check;
   pack_length=old_field->pack_length();
+  key_length= old_field->key_length();
   sql_type=   old_field->real_type();
   charset=    old_field->charset();		// May be NULL ptr
   comment=    old_field->comment;
+  decimals=   old_field->decimals();
 
   /* Fix if the original table had 4 byte pointer blobs */
   if (flags & BLOB_FLAG)
     pack_length= (pack_length- old_field->table->blob_ptr_size +
 		  portable_sizeof_char_ptr);
 
-  switch (sql_type)
-  {
-    case FIELD_TYPE_BLOB:
-      switch (pack_length - portable_sizeof_char_ptr)
-      {
-        case  1: sql_type= FIELD_TYPE_TINY_BLOB; break;
-        case  2: sql_type= FIELD_TYPE_BLOB; break;
-        case  3: sql_type= FIELD_TYPE_MEDIUM_BLOB; break;
-        default: sql_type= FIELD_TYPE_LONG_BLOB; break;
-      }
-      length=(length+charset->mbmaxlen-1)/charset->mbmaxlen; // QQ: Probably not needed
-      break;
-    case MYSQL_TYPE_ENUM:
-    case MYSQL_TYPE_SET:
-    case FIELD_TYPE_STRING:
-    case FIELD_TYPE_VAR_STRING:
-      length=(length+charset->mbmaxlen-1)/charset->mbmaxlen;
-      break;
-    default:
-      break;
-  }
-
-  decimals= old_field->decimals();
-  if (sql_type == FIELD_TYPE_STRING)
-  {
+  switch (sql_type) {
+  case FIELD_TYPE_BLOB:
+    switch (pack_length - portable_sizeof_char_ptr) {
+    case  1: sql_type= FIELD_TYPE_TINY_BLOB; break;
+    case  2: sql_type= FIELD_TYPE_BLOB; break;
+    case  3: sql_type= FIELD_TYPE_MEDIUM_BLOB; break;
+    default: sql_type= FIELD_TYPE_LONG_BLOB; break;
+    }
+    length=(length+charset->mbmaxlen-1) / charset->mbmaxlen;
+    key_length/= charset->mbmaxlen;
+    break;
+  case FIELD_TYPE_STRING:
     /* Change CHAR -> VARCHAR if dynamic record length */
-    sql_type=old_field->type();
-    decimals=0;
+    if (old_field->type() == MYSQL_TYPE_VAR_STRING)
+      sql_type= MYSQL_TYPE_VARCHAR;
+    /* fall through */
+
+  case MYSQL_TYPE_ENUM:
+  case MYSQL_TYPE_SET:
+  case MYSQL_TYPE_VARCHAR:
+  case MYSQL_TYPE_VAR_STRING:
+    /* These are 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:
+    geom_type= ((Field_geom*)old_field)->geom_type;
+    break;
+#endif
+  default:
+    break;
   }
+
   if (flags & (ENUM_FLAG | SET_FLAG))
     interval= ((Field_enum*) old_field)->typelib;
   else
@@ -6152,12 +6341,6 @@ create_field::create_field(Field *old_field,Field *orig_field)
       def= new Item_string(pos, tmp.length(), charset);
     }
   }
-#ifdef HAVE_SPATIAL
-  if (sql_type == FIELD_TYPE_GEOMETRY)
-  {
-    geom_type= ((Field_geom*)old_field)->geom_type;
-  }
-#endif
 }
 
 
diff --git a/sql/field.h b/sql/field.h
index b4e8e61e3fb..058c386954f 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -184,27 +184,7 @@ public:
   */
   virtual bool can_be_compared_as_longlong() const { return FALSE; }
   virtual void free() {}
-  Field *new_field(MEM_ROOT *root, struct st_table *new_table)
-  {
-    Field *tmp= (Field*) memdup_root(root,(char*) this,size_of());
-    if (tmp)
-    {
-      if (tmp->table->maybe_null)
-	tmp->flags&= ~NOT_NULL_FLAG;
-      tmp->table= new_table;
-      tmp->key_start.init(0);
-      tmp->part_of_key.init(0);
-      tmp->part_of_sortkey.init(0);
-      tmp->unireg_check=Field::NONE;
-      tmp->flags&= (NOT_NULL_FLAG | BLOB_FLAG | UNSIGNED_FLAG |
-		    ZEROFILL_FLAG | BINARY_FLAG | ENUM_FLAG | SET_FLAG);
-#ifdef PROBABLY_WRONG
-      tmp->table_name= new_table->table_name;
-#endif
-      tmp->reset_fields();
-    }
-    return tmp;
-  }
+  virtual Field *new_field(MEM_ROOT *root, struct st_table *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;
@@ -220,11 +200,10 @@ public:
     { memcpy(buff,ptr,length); }
   inline void set_image(char *buff,uint length, CHARSET_INFO *cs)
     { memcpy(ptr,buff,length); }
-  virtual void get_key_image(char *buff,uint length, CHARSET_INFO *cs,
-			     imagetype type)
-    { get_image(buff,length,cs); }
-  virtual void set_key_image(char *buff,uint length, CHARSET_INFO *cs)
-    { set_image(buff,length,cs); }
+  virtual void get_key_image(char *buff, uint length, imagetype type)
+    { get_image(buff,length, &my_charset_bin); }
+  virtual void set_key_image(char *buff,uint length)
+    { set_image(buff,length, &my_charset_bin); }
   inline longlong val_int_offset(uint row_offset)
     {
       ptr+=row_offset;
@@ -263,9 +242,11 @@ public:
   virtual uint max_packed_col_length(uint max_length)
   { return max_length;}
 
-  virtual int pack_cmp(const char *a,const char *b, uint key_length_arg)
+  virtual int pack_cmp(const char *a,const char *b, uint key_length_arg,
+                       my_bool insert_or_update)
   { return cmp(a,b); }
-  virtual int pack_cmp(const char *b, uint key_length_arg)
+  virtual int pack_cmp(const char *b, uint key_length_arg,
+                       my_bool insert_or_update)
   { return cmp(ptr,b); }
   uint offset();			// Should be inline ...
   void copy_from_tmp(int offset);
@@ -908,8 +889,8 @@ public:
   enum_field_types type() const
   {
     return ((table && table->db_create_options & HA_OPTION_PACK_RECORD &&
-	     field_length >= 4) ?
-	    FIELD_TYPE_VAR_STRING : FIELD_TYPE_STRING);
+	     field_length >= 4) && table->frm_version < FRM_VER_TRUE_VARCHAR ?
+	    MYSQL_TYPE_VAR_STRING : MYSQL_TYPE_STRING);
   }
   enum ha_base_keytype key_type() const
     { return binary() ? HA_KEYTYPE_BINARY : HA_KEYTYPE_TEXT; }
@@ -926,8 +907,9 @@ public:
   void sql_type(String &str) const;
   char *pack(char *to, const char *from, uint max_length=~(uint) 0);
   const char *unpack(char* to, const char *from);
-  int pack_cmp(const char *a,const char *b,uint key_length);
-  int pack_cmp(const char *b,uint key_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);
   uint packed_col_length(const char *to, uint length);
   uint max_packed_col_length(uint max_length);
   uint size_of() const { return sizeof(*this); }
@@ -935,6 +917,7 @@ public:
   bool has_charset(void) const
   { return charset() == &my_charset_bin ? FALSE : TRUE; }
   field_cast_enum field_cast_type() { return FIELD_CAST_STRING; }
+  Field *new_field(MEM_ROOT *root, struct st_table *new_table);
 };
 
 
@@ -954,7 +937,7 @@ public:
 	       NONE, field_name_arg, table_arg, cs)
   {}
 
-  enum_field_types type() const { return FIELD_TYPE_VAR_STRING; }
+  enum_field_types type() const { return MYSQL_TYPE_VARCHAR; }
   enum ha_base_keytype key_type() const
     { return binary() ? HA_KEYTYPE_VARBINARY : HA_KEYTYPE_VARTEXT; }
   bool zero_pack() const { return 0; }
@@ -969,18 +952,23 @@ public:
   String *val_str(String*,String *);
   int cmp(const char *,const char*);
   void sort_string(char *buff,uint length);
-  void get_key_image(char *buff,uint length, CHARSET_INFO *cs, imagetype type);
-  void set_key_image(char *buff,uint length, CHARSET_INFO *cs);
+  void get_key_image(char *buff,uint length, imagetype type);
+  void set_key_image(char *buff,uint length);
   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);
   const char *unpack(char* to, const char *from);
-  int pack_cmp(const char *a, const char *b, uint key_length);
-  int pack_cmp(const char *b, uint key_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);
   uint max_packed_col_length(uint max_length);
   uint size_of() const { return sizeof(*this); }
-  enum_field_types real_type() const { return FIELD_TYPE_VAR_STRING; }
+  enum_field_types real_type() const { return MYSQL_TYPE_VARCHAR; }
   bool has_charset(void) const
   { return charset() == &my_charset_bin ? FALSE : TRUE; }
   field_cast_enum field_cast_type() { return FIELD_CAST_VARSTRING; }
@@ -1049,8 +1037,8 @@ public:
       store_length(length);
       memcpy_fixed(ptr+packlength,&data,sizeof(char*));
     }
-  void get_key_image(char *buff,uint length, CHARSET_INFO *cs, imagetype type);
-  void set_key_image(char *buff,uint length, CHARSET_INFO *cs);
+  void get_key_image(char *buff,uint length, imagetype type);
+  void set_key_image(char *buff,uint length);
   void sql_type(String &str) const;
   inline bool copy()
   { char *tmp;
@@ -1068,8 +1056,9 @@ public:
   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_key(char* to, const char *from, uint max_length);
-  int pack_cmp(const char *a, const char *b, uint key_length);
-  int pack_cmp(const char *b, uint key_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);
   uint packed_col_length(const char *col_ptr, uint length);
   uint max_packed_col_length(uint max_length);
   void free() { value.free(); }
@@ -1082,6 +1071,7 @@ public:
   uint32 max_length();
 };
 
+
 #ifdef HAVE_SPATIAL
 class Field_geom :public Field_blob {
 public:
@@ -1106,12 +1096,12 @@ public:
   int  store(double nr) { return 1; }
   int  store(longlong nr) { return 1; }
 
-  void get_key_image(char *buff,uint length, CHARSET_INFO *cs,imagetype type);
-  void set_key_image(char *buff,uint length, CHARSET_INFO *cs);
+  void get_key_image(char *buff,uint length,imagetype type);
   field_cast_enum field_cast_type() { return FIELD_CAST_GEOM; }
 };
 #endif /*HAVE_SPATIAL*/
 
+
 class Field_enum :public Field_str {
 protected:
   uint packlength;
@@ -1192,8 +1182,8 @@ public:
   LEX_STRING comment;			// Comment for field
   Item	*def;				// Default value
   enum	enum_field_types sql_type;
-  uint32 length;
-  uint decimals,flags,pack_length;
+  ulong length;
+  uint  decimals, flags, pack_length, key_length;
   Field::utype unireg_check;
   TYPELIB *interval;			// Which interval to use
   List<String> interval_list;
@@ -1258,6 +1248,7 @@ Field *make_field(char *ptr, uint32 field_length,
 		  TYPELIB *interval, const char *field_name,
 		  struct st_table *table);
 uint pack_length_to_packflag(uint type);
+enum_field_types get_blob_type_from_length(ulong length);
 uint32 calc_pack_length(enum_field_types type,uint32 length);
 int set_field_to_null(Field *field);
 int set_field_to_null_with_conversions(Field *field, bool no_conversions);
diff --git a/sql/field_conv.cc b/sql/field_conv.cc
index 336408c5aa9..c4dbb9ab647 100644
--- a/sql/field_conv.cc
+++ b/sql/field_conv.cc
@@ -353,15 +353,16 @@ static void do_expand_string(Copy_field *copy)
 static void do_varstring(Copy_field *copy)
 {
   uint length=uint2korr(copy->from_ptr);
-  if (length > copy->to_length-2)
+  if (length > copy->to_length- HA_KEY_BLOB_LENGTH)
   {
-    length=copy->to_length-2;
+    length=copy->to_length-HA_KEY_BLOB_LENGTH;
     if (current_thd->count_cuted_fields)
       copy->to_field->set_warning(MYSQL_ERROR::WARN_LEVEL_WARN,
                                   ER_WARN_DATA_TRUNCATED, 1);
   }
   int2store(copy->to_ptr,length);
-  memcpy(copy->to_ptr+2, copy->from_ptr,length);
+  memcpy(copy->to_ptr+HA_KEY_BLOB_LENGTH, copy->from_ptr + HA_KEY_BLOB_LENGTH,
+         length);
 }
 
 /***************************************************************************
@@ -504,7 +505,7 @@ 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() == FIELD_TYPE_VAR_STRING && to_length !=
+      else if (to->real_type() == MYSQL_TYPE_VARCHAR && to_length !=
 	       from_length)
 	return do_varstring;
       else if (to_length < from_length)
diff --git a/sql/ha_berkeley.cc b/sql/ha_berkeley.cc
index ff6b10fe504..6cb83624eff 100644
--- a/sql/ha_berkeley.cc
+++ b/sql/ha_berkeley.cc
@@ -394,6 +394,7 @@ berkeley_cmp_packed_key(DB *file, const DBT *new_key, const DBT *saved_key)
   for (; key_part != end && (int) key_length > 0; key_part++)
   {
     int cmp;
+    uint length;
     if (key_part->null_bit)
     {
       if (*new_key_ptr != *saved_key_ptr++)
@@ -402,11 +403,12 @@ berkeley_cmp_packed_key(DB *file, const DBT *new_key, const DBT *saved_key)
       if (!*new_key_ptr++)
 	continue;
     }
-    if ((cmp=key_part->field->pack_cmp(new_key_ptr,saved_key_ptr,
-				       key_part->length)))
+    if ((cmp= key_part->field->pack_cmp(new_key_ptr,saved_key_ptr,
+                                        key_part->length,
+                                        key->table->insert_or_update)))
       return cmp;
-    uint length=key_part->field->packed_col_length(new_key_ptr,
-						   key_part->length);
+    length= key_part->field->packed_col_length(new_key_ptr,
+                                               key_part->length);
     new_key_ptr+=length;
     key_length-=length;
     saved_key_ptr+=key_part->field->packed_col_length(saved_key_ptr,
@@ -432,7 +434,7 @@ berkeley_cmp_fix_length_key(DB *file, const DBT *new_key, const DBT *saved_key)
   for (; key_part != end && (int) key_length > 0 ; key_part++)
   {
     int cmp;
-    if ((cmp=key_part->field->pack_cmp(new_key_ptr,saved_key_ptr,0)))
+    if ((cmp=key_part->field->pack_cmp(new_key_ptr,saved_key_ptr,0,0)))
       return cmp;
     new_key_ptr+=key_part->length;
     key_length-= key_part->length;
@@ -442,6 +444,7 @@ berkeley_cmp_fix_length_key(DB *file, const DBT *new_key, const DBT *saved_key)
 }
 #endif
 
+
 /* Compare key against row */
 
 static bool
@@ -453,6 +456,7 @@ berkeley_key_cmp(TABLE *table, KEY *key_info, const char *key, uint key_length)
   for (; key_part != end && (int) key_length > 0; key_part++)
   {
     int cmp;
+    uint length;
     if (key_part->null_bit)
     {
       key_length--;
@@ -466,15 +470,20 @@ berkeley_key_cmp(TABLE *table, KEY *key_info, const char *key, uint key_length)
       if (!*key++)				// Null value
 	continue;
     }
-    if ((cmp=key_part->field->pack_cmp(key,key_part->length)))
+    /*
+      Last argument has to be 0 as we are also using this to function to see
+      if a key like 'a  ' matched a row with 'a'
+    */
+    if ((cmp= key_part->field->pack_cmp(key, key_part->length, 0)))
       return cmp;
-    uint length=key_part->field->packed_col_length(key,key_part->length);
-    key+=length;
-    key_length-=length;
+    length= key_part->field->packed_col_length(key,key_part->length);
+    key+= length;
+    key_length-= length;
   }
   return 0;					// Identical keys
 }
 
+
 int ha_berkeley::open(const char *name, int mode, uint test_if_locked)
 {
   char name_buff[FN_REFLEN];
@@ -865,6 +874,7 @@ int ha_berkeley::write_row(byte * record)
   if ((error=pack_row(&row, record,1)))
     DBUG_RETURN(error); /* purecov: inspected */
 
+  table->insert_or_update= 1;                   // For handling of VARCHAR
   if (table->keys + test(hidden_primary_key) == 1)
   {
     error=file->put(file, transaction, create_key(&prim_key, primary_key,
@@ -950,6 +960,7 @@ int ha_berkeley::write_row(byte * record)
 	break;
     }
   }
+  table->insert_or_update= 0;
   if (error == DB_KEYEXIST)
     error=HA_ERR_FOUND_DUPP_KEY;
   else if (!error)
@@ -974,7 +985,7 @@ int ha_berkeley::key_cmp(uint keynr, const byte * old_row,
 	  (new_row[key_part->null_offset] & key_part->null_bit))
 	return 1;
     }
-    if (key_part->key_part_flag & (HA_BLOB_PART | HA_VAR_LENGTH))
+    if (key_part->key_part_flag & (HA_BLOB_PART | HA_VAR_LENGTH_PART))
     {
 
       if (key_part->field->cmp_binary((char*) (old_row + key_part->offset),
@@ -1109,6 +1120,7 @@ int ha_berkeley::update_row(const byte * old_row, byte * new_row)
   if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
     table->timestamp_field->set_time();
 
+  table->insert_or_update= 1;                   // For handling of VARCHAR
   if (hidden_primary_key)
   {
     primary_key_changed=0;
@@ -1161,6 +1173,7 @@ int ha_berkeley::update_row(const byte * old_row, byte * new_row)
 	      if (new_error)
 		error = new_error;
 	    }
+            table->insert_or_update= 0;
 	    DBUG_RETURN(error);			// Fatal error /* purecov: inspected */
 	  }
 	  changed_keys.set_bit(keynr);
@@ -1193,8 +1206,9 @@ int ha_berkeley::update_row(const byte * old_row, byte * new_row)
 				 thd_options);
 	if (new_error)
 	{
-	  error=new_error;			// This shouldn't happen /* purecov: inspected */
-	  break; /* purecov: inspected */
+          /* This shouldn't happen */
+	  error=new_error;			/* purecov: inspected */
+	  break;                                /* purecov: inspected */
 	}
       }
     }
@@ -1206,6 +1220,7 @@ int ha_berkeley::update_row(const byte * old_row, byte * new_row)
     if (error != DB_LOCK_DEADLOCK)
       break;
   }
+  table->insert_or_update= 0;
   if (error == DB_KEYEXIST)
     error=HA_ERR_FOUND_DUPP_KEY;
   DBUG_RETURN(error);
@@ -1476,7 +1491,8 @@ int ha_berkeley::index_read(byte * buf, const byte * key,
     find_flag= HA_READ_AFTER_KEY;
     do_prev= 1;
   }
-  if (key_len == key_info->key_length)
+  if (key_len == key_info->key_length &&
+      !table->key_info[active_index].flags & HA_END_SPACE_KEY)
   {
     if (find_flag == HA_READ_AFTER_KEY)
       key_info->handler.bdb_return_if_eq= 1;
@@ -1573,7 +1589,8 @@ int ha_berkeley::index_next_same(byte * buf, const byte *key, uint keylen)
   statistic_increment(table->in_use->status_var.ha_read_next_count,
 		      &LOCK_status);
   bzero((char*) &row,sizeof(row));
-  if (keylen == table->key_info[active_index].key_length)
+  if (keylen == table->key_info[active_index].key_length &&
+      !table->key_info[active_index].flags & HA_END_SPACE_KEY)
     error=read_row(cursor->c_get(cursor, &last_key, &row, DB_NEXT_DUP),
 		   (char*) buf, active_index, &row, &last_key, 1);
   else
@@ -1814,7 +1831,8 @@ int ha_berkeley::external_lock(THD *thd, int lock_type)
 	  !thd->transaction.all.bdb_tid)
       {
 	/* We have to start a master transaction */
-	DBUG_PRINT("trans",("starting transaction all"));
+	DBUG_PRINT("trans",("starting transaction all:  options: 0x%lx",
+                            (ulong) thd->options));
 	if ((error=txn_begin(db_env, 0,
 			     (DB_TXN**) &thd->transaction.all.bdb_tid,
 			     0)))
@@ -2068,19 +2086,35 @@ ha_rows ha_berkeley::records_in_range(uint keynr, key_range *start_key,
   DB_KEY_RANGE start_range, end_range;
   DB *kfile=key_file[keynr];
   double start_pos,end_pos,rows;
-  DBUG_ENTER("records_in_range");
-
-  if ((start_key && kfile->key_range(kfile,transaction,
-                                     pack_key(&key, keynr, key_buff,
-                                              start_key->key,
-                                              start_key->length),
-                                     &start_range,0)) ||
-      (end_key && kfile->key_range(kfile,transaction,
-				   pack_key(&key, keynr, key_buff,
-                                            end_key->key,
-                                            end_key->length),
-				   &end_range,0)))
-    DBUG_RETURN(HA_BERKELEY_RANGE_COUNT); // Better than returning an error /* purecov: inspected */
+  bool error;
+  KEY *key_info= &table->key_info[keynr];
+  DBUG_ENTER("ha_berkeley::records_in_range");
+
+  /* Ensure we get maximum range, even for varchar keys with different space */
+  key_info->handler.bdb_return_if_eq= -1;
+  error= ((start_key && kfile->key_range(kfile,transaction,
+                                         pack_key(&key, keynr, key_buff,
+                                                  start_key->key,
+                                                  start_key->length),
+                                         &start_range,0)));
+  if (error)
+  {
+    key_info->handler.bdb_return_if_eq= 0;
+    // Better than returning an error
+    DBUG_RETURN(HA_BERKELEY_RANGE_COUNT);       /* purecov: inspected */
+  }
+  key_info->handler.bdb_return_if_eq= 1;
+  error= (end_key && kfile->key_range(kfile,transaction,
+                                      pack_key(&key, keynr, key_buff,
+                                               end_key->key,
+                                               end_key->length),
+                                      &end_range,0));
+  key_info->handler.bdb_return_if_eq= 0;
+  if (error)
+  {
+    // Better than returning an error
+    DBUG_RETURN(HA_BERKELEY_RANGE_COUNT);       /* purecov: inspected */
+  }
 
   if (!start_key)
     start_pos= 0.0;
@@ -2563,6 +2597,7 @@ end:
   DBUG_VOID_RETURN;
 }
 
+
 /*
   Return an estimated of the number of rows in the table.
   Used when sorting to allocate buffers and by the optimizer.
@@ -2588,11 +2623,11 @@ int ha_berkeley::cmp_ref(const byte *ref1, const byte *ref2)
   {
     field=  key_part->field; 
     result= field->pack_cmp((const char*)ref1, (const char*)ref2, 
-                            key_part->length);
+                            key_part->length, 0);
     if (result)
       return result;
-    ref1 += field->packed_col_length((const char*)ref1, key_part->length);
-    ref2 += field->packed_col_length((const char*)ref2, key_part->length);
+    ref1+= field->packed_col_length((const char*)ref1, key_part->length);
+    ref2+= field->packed_col_length((const char*)ref2, key_part->length);
   }
 
   return 0;
diff --git a/sql/ha_heap.cc b/sql/ha_heap.cc
index 46824033db9..29508207a70 100644
--- a/sql/ha_heap.cc
+++ b/sql/ha_heap.cc
@@ -481,16 +481,20 @@ int ha_heap::create(const char *name, TABLE *table_arg,
     {
       uint flag=    key_part->key_type;
       Field *field= key_part->field;
+
       if (pos->algorithm == HA_KEY_ALG_BTREE)
 	seg->type= field->key_type();
       else
       {
-        if ((seg->type = field->key_type()) != (int) HA_KEYTYPE_TEXT)
+        if ((seg->type = field->key_type()) != (int) HA_KEYTYPE_TEXT &&
+            seg->type != HA_KEYTYPE_VARTEXT &&
+            seg->type != HA_KEYTYPE_VARBINARY)
           seg->type= HA_KEYTYPE_BINARY;
       }
       seg->start=   (uint) key_part->offset;
       seg->length=  (uint) key_part->length;
-      seg->flag =   0;
+      seg->flag=    key_part->key_part_flag;
+
       seg->charset= field->charset();
       if (field->null_ptr)
       {
diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index c6ad646be4d..f387b79ac2d 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -2010,11 +2010,12 @@ innobase_mysql_cmp(
 	switch (mysql_tp) {
 
 	case FIELD_TYPE_STRING:
-	case FIELD_TYPE_VAR_STRING:
+	case MYSQL_TYPE_VAR_STRING:
 	case FIELD_TYPE_TINY_BLOB:
 	case FIELD_TYPE_MEDIUM_BLOB:
 	case FIELD_TYPE_BLOB:
 	case FIELD_TYPE_LONG_BLOB:
+        case MYSQL_TYPE_VARCHAR:
 		/* Use the charset number to pick the right charset struct for
 		the comparison. Since the MySQL function get_charset may be
 		slow before Bar removes the mutex operation there, we first
@@ -2042,7 +2043,7 @@ innobase_mysql_cmp(
 
                 ret = charset->coll->strnncollsp(charset,
                                   a, a_length,
-                                  b, b_length);
+                                                 b, b_length, 0);
 		if (ret < 0) {
 		        return(-1);
 		} else if (ret > 0) {
@@ -2071,17 +2072,11 @@ get_innobase_type_from_mysql_type(
 	8 bits: this is used in ibuf and also when DATA_NOT_NULL is ORed to
 	the type */
 
-	DBUG_ASSERT((ulint)FIELD_TYPE_STRING < 256);
-	DBUG_ASSERT((ulint)FIELD_TYPE_VAR_STRING < 256);
-	DBUG_ASSERT((ulint)FIELD_TYPE_DOUBLE < 256);
-	DBUG_ASSERT((ulint)FIELD_TYPE_FLOAT < 256);
-	DBUG_ASSERT((ulint)FIELD_TYPE_DECIMAL < 256);
-
 	switch (field->type()) {
 	        /* NOTE that we only allow string types in DATA_MYSQL
 		and DATA_VARMYSQL */
-		case FIELD_TYPE_VAR_STRING: if (field->binary()) {
-
+                case MYSQL_TYPE_VAR_STRING:
+                case MYSQL_TYPE_VARCHAR: if (field->binary()) {
 						return(DATA_BINARY);
 					} else if (strcmp(
 						  field->charset()->name,
@@ -2648,12 +2643,12 @@ innobase_convert_and_store_changed_col(
 
 	if (len == UNIV_SQL_NULL) {
 		data = NULL;
-	} else if (col_type == DATA_VARCHAR || col_type == DATA_BINARY
-		   || col_type == DATA_VARMYSQL) {
-	        /* Remove trailing spaces */
-        	while (len > 0 && data[len - 1] == ' ') {
-	                len--;
-	        }
+        } else if (col_type == DATA_VARCHAR || col_type == DATA_BINARY
+                   || col_type == DATA_VARMYSQL) {
+                /* Remove trailing spaces */
+                while (len > 0 && data[len - 1] == ' ') {
+                        len--;
+                }
 	} else if (col_type == DATA_INT) {
 		/* Store integer data in InnoDB in a big-endian
 		format, sign bit negated, if signed */
diff --git a/sql/ha_innodb.h b/sql/ha_innodb.h
index a64bbc665c1..b3b8d1a29e8 100644
--- a/sql/ha_innodb.h
+++ b/sql/ha_innodb.h
@@ -84,6 +84,7 @@ class ha_innobase: public handler
 			  HA_CAN_SQL_HANDLER |
 			  HA_NOT_EXACT_COUNT |
 			  HA_PRIMARY_KEY_IN_READ_INDEX |
+                          HA_NO_VARCHAR |
 			  HA_TABLE_SCAN_ON_INDEX),
 	  last_dup_key((uint) -1),
 	  start_of_scan(0),
diff --git a/sql/ha_isam.cc b/sql/ha_isam.cc
deleted file mode 100644
index e1fd17f4399..00000000000
--- a/sql/ha_isam.cc
+++ /dev/null
@@ -1,411 +0,0 @@
-/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
-
-   This program is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published by
-   the Free Software Foundation; either version 2 of the License, or
-   (at your option) any later version.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
-
-
-#ifdef __GNUC__
-#pragma implementation			// gcc: Class implementation
-#endif
-
-#include "mysql_priv.h"
-#ifdef HAVE_ISAM
-#include <m_ctype.h>
-#include <myisampack.h>
-#include "ha_isam.h"
-#ifndef MASTER
-#include "../srclib/isam/isamdef.h"
-#else
-#include "../isam/isamdef.h"
-#endif
-
-/*****************************************************************************
-** isam tables
-*****************************************************************************/
-
-
-const char **ha_isam::bas_ext() const
-{ static const char *ext[]= { ".ISM",".ISD", NullS }; return ext; }
-
-int ha_isam::open(const char *name, int mode, uint test_if_locked)
-{
-  char name_buff[FN_REFLEN];
-  if (!(file=nisam_open(fn_format(name_buff,name,"","",2 | 4), mode,
-		     test_if_locked)))
-    return (my_errno ? my_errno : -1);
-
-  if (!(test_if_locked == HA_OPEN_WAIT_IF_LOCKED ||
-	test_if_locked == HA_OPEN_ABORT_IF_LOCKED))
-    (void) nisam_extra(file,HA_EXTRA_NO_WAIT_LOCK);
-  info(HA_STATUS_NO_LOCK | HA_STATUS_VARIABLE | HA_STATUS_CONST);
-  if (!(test_if_locked & HA_OPEN_WAIT_IF_LOCKED))
-    (void) nisam_extra(file,HA_EXTRA_WAIT_LOCK);
-  if (!table->db_record_offset)
-    int_table_flags|=HA_REC_NOT_IN_SEQ;
-  return (0);
-}
-
-int ha_isam::close(void)
-{
-  return !nisam_close(file) ? 0 : my_errno ? my_errno : -1;
-}
-
-uint ha_isam::min_record_length(uint options) const
-{
-  return (options & HA_OPTION_PACK_RECORD) ? 1 : 5;
-}
-
-
-int ha_isam::write_row(byte * buf)
-{
-  statistic_increment(table->in_use->status_var.ha_write_count, &LOCK_status);
-  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
-    table->timestamp_field->set_time();
-  if (table->next_number_field && buf == table->record[0])
-    update_auto_increment();
-  return !nisam_write(file,buf) ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isam::update_row(const byte * old_data, byte * new_data)
-{
-  statistic_increment(table->in_use->status_var.ha_update_count, &LOCK_status);
-  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
-    table->timestamp_field->set_time();
-  return !nisam_update(file,old_data,new_data) ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isam::delete_row(const byte * buf)
-{
-  statistic_increment(table->in_use->status_var.ha_delete_count, &LOCK_status);
-  return !nisam_delete(file,buf) ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isam::index_read(byte * buf, 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);
-  int error=nisam_rkey(file, buf, active_index, key, key_len, find_flag);
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isam::index_read_idx(byte * buf, uint index, 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);
-  int error=nisam_rkey(file, buf, index, key, key_len, find_flag);
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isam::index_read_last(byte * buf, const byte * key, uint key_len)
-{
-  statistic_increment(table->in_use->status_var.ha_read_key_count,
-		      &LOCK_status);
-  int error=nisam_rkey(file, buf, active_index, key, key_len,
-		       HA_READ_PREFIX_LAST);
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isam::index_next(byte * buf)
-{
-  statistic_increment(table->in_use->status_var.ha_read_next_count,
-		      &LOCK_status);
-  int error=nisam_rnext(file,buf,active_index);
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : HA_ERR_END_OF_FILE;
-}
-
-int ha_isam::index_prev(byte * buf)
-{
-  statistic_increment(table->in_use->status_var.ha_read_prev_count,
-		      &LOCK_status);
-  int error=nisam_rprev(file,buf, active_index);
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : HA_ERR_END_OF_FILE;
-}
-
-int ha_isam::index_first(byte * buf)
-{
-  statistic_increment(table->in_use->status_var.ha_read_first_count,
-		      &LOCK_status);
-  int error=nisam_rfirst(file, buf, active_index);
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : HA_ERR_END_OF_FILE;
-}
-
-int ha_isam::index_last(byte * buf)
-{
-  statistic_increment(table->in_use->status_var.ha_read_last_count,
-		      &LOCK_status);
-  int error=nisam_rlast(file, buf, active_index);
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : HA_ERR_END_OF_FILE;
-}
-
-int ha_isam::rnd_init(bool scan)
-{
-  return nisam_extra(file,HA_EXTRA_RESET) ? 0 : my_errno ? my_errno : -1;;
-}
-
-int ha_isam::rnd_next(byte *buf)
-{
-  statistic_increment(table->in_use->status_var.ha_read_rnd_next_count,
-		      &LOCK_status);
-  int error=nisam_rrnd(file, buf, NI_POS_ERROR);
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isam::rnd_pos(byte * buf, byte *pos)
-{
-  statistic_increment(table->in_use->status_var.ha_read_rnd_count,
-		      &LOCK_status);
-  int error=nisam_rrnd(file, buf, (ulong) my_get_ptr(pos,ref_length));
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : -1;
-}
-
-void ha_isam::position(const byte *record)
-{
-  my_off_t position=nisam_position(file);
-  if (position == (my_off_t) ~ (ulong) 0)
-    position=HA_OFFSET_ERROR;
-  my_store_ptr(ref, ref_length, position);
-}
-
-void ha_isam::info(uint flag)
-{
-  N_ISAMINFO info;
-  (void) nisam_info(file,&info,flag);
-  if (flag & HA_STATUS_VARIABLE)
-  {
-    records = info.records;
-    deleted = info.deleted;
-    data_file_length=info.data_file_length;
-    index_file_length=info.index_file_length;
-    delete_length = info.delete_length;
-    check_time  = info.isamchk_time;
-    mean_rec_length=info.mean_reclength;
-  }
-  if (flag & HA_STATUS_CONST)
-  {
-    max_data_file_length=info.max_data_file_length;
-    max_index_file_length=info.max_index_file_length;
-    create_time = info.create_time;
-    sortkey = info.sortkey;
-    block_size=nisam_block_size;
-    table->keys	   = min(table->keys,info.keys);
-    table->keys_in_use.set_prefix(table->keys);
-    table->db_options_in_use= info.options;
-    table->db_record_offset=
-      (table->db_options_in_use &
-       (HA_OPTION_PACK_RECORD | HA_OPTION_COMPRESS_RECORD)) ? 0 :
-      table->reclength;
-    if (!table->tmp_table)
-    {
-      ulong *rec_per_key=info.rec_per_key;
-      for (uint i=0 ; i < table->keys ; i++)
-      {
-	table->key_info[i].rec_per_key[table->key_info[i].key_parts-1]=
-	  *(rec_per_key++);
-      }
-    }
-    ref_length=4;
-  }
-  if (flag & HA_STATUS_ERRKEY)
-  {
-    errkey  = info.errkey;
-    my_store_ptr(dupp_ref, ref_length, info.dupp_key_pos);
-  }
-  if (flag & HA_STATUS_TIME)
-    update_time = info.update_time;
-}
-
-
-int ha_isam::extra(enum ha_extra_function operation)
-{
-  if ((specialflag & SPECIAL_SAFE_MODE || test_flags & TEST_NO_EXTRA) &&
-      (operation == HA_EXTRA_WRITE_CACHE ||
-       operation == HA_EXTRA_KEYREAD))
-    return 0;
-  return nisam_extra(file,operation);
-}
-
-int ha_isam::external_lock(THD *thd, int lock_type)
-{
-  if (!table->tmp_table)
-    return nisam_lock_database(file,lock_type);
-  return 0;
-}
-
-
-THR_LOCK_DATA **ha_isam::store_lock(THD *thd,
-				    THR_LOCK_DATA **to,
-				    enum thr_lock_type lock_type)
-{
-  if (lock_type != TL_IGNORE && file->lock.type == TL_UNLOCK)
-    file->lock.type=lock_type;
-  *to++= &file->lock;
-  return to;
-}
-
-
-int ha_isam::create(const char *name, register TABLE *form,
-		    HA_CREATE_INFO *create_info)
-
-{
-  uint options=form->db_options_in_use;
-  int error;
-  uint i,j,recpos,minpos,fieldpos,temp_length,length;
-  enum ha_base_keytype type;
-  char buff[FN_REFLEN];
-  KEY *pos;
-  N_KEYDEF keydef[MAX_KEY];
-  N_RECINFO *recinfo,*recinfo_pos;
-  DBUG_ENTER("ha_isam::create");
-
-  type=HA_KEYTYPE_BINARY;				// Keep compiler happy
-  if (!(recinfo= (N_RECINFO*) my_malloc((form->fields*2+2)*sizeof(N_RECINFO),
-					MYF(MY_WME))))
-    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
-
-  pos=form->key_info;
-  for (i=0; i < form->keys ; i++, pos++)
-  {
-    keydef[i].base.flag= (pos->flags & HA_NOSAME);
-    for (j=0 ; (int7) j < pos->key_parts ; j++)
-    {
-      keydef[i].seg[j].base.flag=pos->key_part[j].key_part_flag;
-      Field *field=pos->key_part[j].field;
-      type=field->key_type();
-
-      if ((options & HA_OPTION_PACK_KEYS ||
-	   (pos->flags & (HA_PACK_KEY | HA_BINARY_PACK_KEY |
-			  HA_SPACE_PACK_USED))) &&
-	  pos->key_part[j].length > 8 &&
-	  (type == HA_KEYTYPE_TEXT ||
-	   type == HA_KEYTYPE_NUM ||
-	   (type == HA_KEYTYPE_BINARY && !field->zero_pack())))
-      {
-	if (j == 0)
-	  keydef[i].base.flag|=HA_PACK_KEY;
-	if (!(field->flags & ZEROFILL_FLAG) &&
-	    (field->type() == FIELD_TYPE_STRING ||
-	     field->type() == FIELD_TYPE_VAR_STRING ||
-	     ((int) (pos->key_part[j].length - field->decimals()))
-	     >= 4))
-	  keydef[i].seg[j].base.flag|=HA_SPACE_PACK;
-      }
-      keydef[i].seg[j].base.type=(int) type;
-      keydef[i].seg[j].base.start=  pos->key_part[j].offset;
-      keydef[i].seg[j].base.length= pos->key_part[j].length;
-    }
-    keydef[i].seg[j].base.type=(int) HA_KEYTYPE_END;	/* End of key-parts */
-  }
-
-  recpos=0; recinfo_pos=recinfo;
-  while (recpos < (uint) form->reclength)
-  {
-    Field **field,*found=0;
-    minpos=form->reclength; length=0;
-
-    for (field=form->field ; *field ; field++)
-    {
-      if ((fieldpos=(*field)->offset()) >= recpos &&
-	  fieldpos <= minpos)
-      {
-	/* skip null fields */
-	if (!(temp_length= (*field)->pack_length()))
-	  continue;				/* Skip null-fields */
-	if (! found || fieldpos < minpos ||
-	    (fieldpos == minpos && temp_length < length))
-	{
-	  minpos=fieldpos; found= *field; length=temp_length;
-	}
-      }
-    }
-    DBUG_PRINT("loop",("found: 0x%lx  recpos: %d  minpos: %d  length: %d",
-		       found,recpos,minpos,length));
-    if (recpos != minpos)
-    {						// Reserved space (Null bits?)
-      recinfo_pos->base.type=(int) FIELD_NORMAL;
-      recinfo_pos++->base.length= (uint16) (minpos-recpos);
-    }
-    if (! found)
-      break;
-
-    if (found->flags & BLOB_FLAG)
-    {
-      /* ISAM can only handle blob pointers of sizeof(char(*)) */
-      recinfo_pos->base.type= (int) FIELD_BLOB;
-      if (options & HA_OPTION_LONG_BLOB_PTR)
-	length= length-portable_sizeof_char_ptr+sizeof(char*);
-    }
-    else if (!(options & HA_OPTION_PACK_RECORD))
-      recinfo_pos->base.type= (int) FIELD_NORMAL;
-    else if (found->zero_pack())
-      recinfo_pos->base.type= (int) FIELD_SKIP_ZERO;
-    else
-      recinfo_pos->base.type= (int) ((length <= 3 ||
-				      (found->flags & ZEROFILL_FLAG)) ?
-				     FIELD_NORMAL :
-				     found->type() == FIELD_TYPE_STRING ||
-				     found->type() == FIELD_TYPE_VAR_STRING ?
-				     FIELD_SKIP_ENDSPACE :
-				     FIELD_SKIP_PRESPACE);
-    recinfo_pos++ ->base.length=(uint16) length;
-    recpos=minpos+length;
-    DBUG_PRINT("loop",("length: %d  type: %d",
-		       recinfo_pos[-1].base.length,recinfo_pos[-1].base.type));
-
-    if ((found->flags & BLOB_FLAG) && (options & HA_OPTION_LONG_BLOB_PTR) &&
-	sizeof(char*) != portable_sizeof_char_ptr)
-    {						// Not used space
-      recinfo_pos->base.type=(int) FIELD_ZERO;
-      recinfo_pos++->base.length=
-	(uint16) (portable_sizeof_char_ptr-sizeof(char*));
-      recpos+= (portable_sizeof_char_ptr-sizeof(char*));
-    }
-  }
-  recinfo_pos->base.type= (int) FIELD_LAST;	/* End of fieldinfo */
-  error=nisam_create(fn_format(buff,name,"","",2+4+16),form->keys,keydef,
-		     recinfo,(ulong) form->max_rows, (ulong) form->min_rows,
-		     0, 0, 0L);
-  my_free((gptr) recinfo,MYF(0));
-  DBUG_RETURN(error);
-
-}
-
-static key_range no_range= { (byte*) 0, 0, HA_READ_KEY_EXACT };
-
-ha_rows ha_isam::records_in_range(uint inx, key_range *min_key,
-                                  key_range *max_key)
-{
-  /* ISAM checks if 'key' pointer <> 0 to know if there is no range */
-  if (!min_key)
-    min_key= &no_range;
-  if (!max_key)
-    max_key= &no_range;
-  return (ha_rows) nisam_records_in_range(file,
-                                          (int) inx,
-                                          min_key->key, min_key->length,
-                                          min_key->flag,
-                                          max_key->key, max_key->length,
-                                          max_key->flag);
-}
-#endif /* HAVE_ISAM */
diff --git a/sql/ha_isam.h b/sql/ha_isam.h
deleted file mode 100644
index b3e932696cb..00000000000
--- a/sql/ha_isam.h
+++ /dev/null
@@ -1,79 +0,0 @@
-/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
-
-   This program is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published by
-   the Free Software Foundation; either version 2 of the License, or
-   (at your option) any later version.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
-
-
-#ifdef __GNUC__
-#pragma interface			/* gcc class implementation */
-#endif
-
-/* class for the the myisam handler */
-
-#include <nisam.h>
-
-class ha_isam: public handler
-{
-  N_INFO *file;
-  /* We need this as table_flags() may change after open() */
-  ulong int_table_flags;
-
- public:
-  ha_isam(TABLE *table)
-    :handler(table), file(0),
-    int_table_flags(HA_READ_RND_SAME |
-		    HA_DUPP_POS | HA_NOT_DELETE_WITH_CACHE | HA_FILE_BASED)
-  {}
-  ~ha_isam() {}
-  ulong index_flags(uint idx, uint part, bool all_parts) const
-  { return HA_READ_NEXT; } // but no HA_READ_PREV here!!!
-  const char *table_type() const { return "ISAM"; }
-  const char *index_type(uint key_number) { return "BTREE"; }
-  const char **bas_ext() const;
-  ulong table_flags() const { return int_table_flags; }
-  uint max_supported_record_length() const { return HA_MAX_REC_LENGTH; }
-  uint max_supported_keys()          const { return N_MAXKEY; }
-  uint max_supported_key_parts()     const { return N_MAXKEY_SEG; }
-  uint max_supported_key_length()    const { return N_MAX_KEY_LENGTH; }
-  uint min_record_length(uint options) const;
-  bool low_byte_first() const { return 0; }
-
-  int open(const char *name, int mode, uint test_if_locked);
-  int close(void);
-  int write_row(byte * buf);
-  int update_row(const byte * old_data, byte * new_data);
-  int delete_row(const byte * buf);
-  int index_read(byte * buf, const byte * key,
-		 uint key_len, enum ha_rkey_function find_flag);
-  int index_read_idx(byte * buf, uint idx, const byte * key,
-		     uint key_len, enum ha_rkey_function find_flag);
-  int index_read_last(byte * buf, const byte * key, uint key_len);
-  int index_next(byte * buf);
-  int index_prev(byte * buf);
-  int index_first(byte * buf);
-  int index_last(byte * buf);
-  int rnd_init(bool scan);
-  int rnd_next(byte *buf);
-  int rnd_pos(byte * buf, byte *pos);
-  void position(const byte *record);
-  void info(uint);
-  int extra(enum ha_extra_function operation);
-  int external_lock(THD *thd, int lock_type);
-  ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key);
-
-  int create(const char *name, TABLE *form, HA_CREATE_INFO *create_info);
-  THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
-			     enum thr_lock_type lock_type);
-};
-
diff --git a/sql/ha_isammrg.cc b/sql/ha_isammrg.cc
deleted file mode 100644
index c63548ec5cf..00000000000
--- a/sql/ha_isammrg.cc
+++ /dev/null
@@ -1,211 +0,0 @@
-/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
-
-   This program is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published by
-   the Free Software Foundation; either version 2 of the License, or
-   (at your option) any later version.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
-
-
-#ifdef __GNUC__
-#pragma implementation				// gcc: Class implementation
-#endif
-
-#include "mysql_priv.h"
-#ifdef HAVE_ISAM
-#include <m_ctype.h>
-#ifndef MASTER
-#include "../srclib/merge/mrg_def.h"
-#else
-#include "../merge/mrg_def.h"
-#endif
-#include "ha_isammrg.h"
-
-/*****************************************************************************
-** ISAM MERGE tables
-*****************************************************************************/
-
-const char **ha_isammrg::bas_ext() const
-{ static const char *ext[]= { ".MRG", NullS }; return ext; }
-
-int ha_isammrg::open(const char *name, int mode, uint test_if_locked)
-{
-  char name_buff[FN_REFLEN];
-  if (!(file=mrg_open(fn_format(name_buff,name,"","",2 | 4), mode,
-		       test_if_locked)))
-    return (my_errno ? my_errno : -1);
-
-  if (!(test_if_locked == HA_OPEN_WAIT_IF_LOCKED ||
-	test_if_locked == HA_OPEN_ABORT_IF_LOCKED))
-    mrg_extra(file,HA_EXTRA_NO_WAIT_LOCK);
-  info(HA_STATUS_NO_LOCK | HA_STATUS_VARIABLE | HA_STATUS_CONST);
-  if (!(test_if_locked & HA_OPEN_WAIT_IF_LOCKED))
-    mrg_extra(file,HA_EXTRA_WAIT_LOCK);
-  if (table->reclength != mean_rec_length)
-  {
-    DBUG_PRINT("error",("reclength: %d  mean_rec_length: %d",
-			table->reclength, mean_rec_length));
-    mrg_close(file);
-    file=0;
-    return ER_WRONG_MRG_TABLE;
-  }
-  return (0);
-}
-
-int ha_isammrg::close(void)
-{
-  return !mrg_close(file) ? 0 : my_errno ? my_errno : -1;
-}
-
-uint ha_isammrg::min_record_length(uint options) const
-{
-  return (options & HA_OPTION_PACK_RECORD) ? 1 : 5;
-}
-
-int ha_isammrg::write_row(byte * buf)
-{
-  return (my_errno=HA_ERR_WRONG_COMMAND);
-}
-
-int ha_isammrg::update_row(const byte * old_data, byte * new_data)
-{
-  statistic_increment(table->in_use->status_var.ha_update_count, &LOCK_status);
-  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
-    table->timestamp_field->set_time();
-  return !mrg_update(file,old_data,new_data) ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isammrg::delete_row(const byte * buf)
-{
-  statistic_increment(table->in_use->status_var.ha_delete_count, &LOCK_status);
-  return !mrg_delete(file,buf) ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isammrg::index_read(byte * buf, const byte * key,
-			   uint key_len, enum ha_rkey_function find_flag)
-{
-  return (my_errno=HA_ERR_WRONG_COMMAND);
-}
-
-int ha_isammrg::index_read_idx(byte * buf, uint index, const byte * key,
-			       uint key_len, enum ha_rkey_function find_flag)
-{
-  return (my_errno=HA_ERR_WRONG_COMMAND);
-}
-
-int ha_isammrg::index_next(byte * buf)
-{
-  return (my_errno=HA_ERR_WRONG_COMMAND);
-}
-
-int ha_isammrg::index_prev(byte * buf)
-{
-  return (my_errno=HA_ERR_WRONG_COMMAND);
-}
-
-int ha_isammrg::index_first(byte * buf)
-{
-  return (my_errno=HA_ERR_WRONG_COMMAND);
-}
-
-int ha_isammrg::index_last(byte * buf)
-{
-  return (my_errno=HA_ERR_WRONG_COMMAND);
-}
-
-int ha_isammrg::rnd_init(bool scan)
-{
-  return !mrg_extra(file,HA_EXTRA_RESET) ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isammrg::rnd_next(byte *buf)
-{
-  statistic_increment(table->in_use->status_var.ha_read_rnd_next_count,
-		      &LOCK_status);
-  int error=mrg_rrnd(file, buf, ~(mrg_off_t) 0);
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isammrg::rnd_pos(byte * buf, byte *pos)
-{
-  statistic_increment(table->in_use->status_var.ha_read_rnd_count,
-		      &LOCK_status);
-  int error=mrg_rrnd(file, buf, (ulong) my_get_ptr(pos,ref_length));
-  table->status=error ? STATUS_NOT_FOUND: 0;
-  return !error ? 0 : my_errno ? my_errno : -1;
-}
-
-void ha_isammrg::position(const byte *record)
-{
-  ulong position= mrg_position(file);
-  my_store_ptr(ref, ref_length, (my_off_t) position);
-}
-
-
-void ha_isammrg::info(uint flag)
-{
-  MERGE_INFO info;
-  (void) mrg_info(file,&info,flag);
-  records = (ha_rows) info.records;
-  deleted = (ha_rows) info.deleted;
-  data_file_length=info.data_file_length;
-  errkey  = info.errkey;
-  table->keys_in_use.clear_all();               // No keys yet
-  table->db_options_in_use    = info.options;
-  mean_rec_length=info.reclength;
-  block_size=0;
-  update_time=0;
-  ref_length=4;					// Should be big enough
-}
-
-
-int ha_isammrg::extra(enum ha_extra_function operation)
-{
-  return !mrg_extra(file,operation) ? 0 : my_errno ? my_errno : -1;
-}
-
-int ha_isammrg::external_lock(THD *thd, int lock_type)
-{
-  return !mrg_lock_database(file,lock_type) ? 0 : my_errno ? my_errno : -1;
-}
-
-uint ha_isammrg::lock_count(void) const
-{
-  return file->tables;
-}
-
-THR_LOCK_DATA **ha_isammrg::store_lock(THD *thd,
-				       THR_LOCK_DATA **to,
-				       enum thr_lock_type lock_type)
-{
-  MRG_TABLE *open_table;
-
-  for (open_table=file->open_tables ;
-       open_table != file->end_table ;
-       open_table++)
-  {
-    *(to++)= &open_table->table->lock;
-    if (lock_type != TL_IGNORE && open_table->table->lock.type == TL_UNLOCK)
-      open_table->table->lock.type=lock_type;
-  }
-  return to;
-}
-
-
-int ha_isammrg::create(const char *name, register TABLE *form,
-		       HA_CREATE_INFO *create_info)
-
-{
-  char buff[FN_REFLEN];
-  return mrg_create(fn_format(buff,name,"","",2+4+16),0);
-}
-#endif /* HAVE_ISAM */
diff --git a/sql/ha_isammrg.h b/sql/ha_isammrg.h
deleted file mode 100644
index 657e5060272..00000000000
--- a/sql/ha_isammrg.h
+++ /dev/null
@@ -1,69 +0,0 @@
-/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
-
-   This program is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published by
-   the Free Software Foundation; either version 2 of the License, or
-   (at your option) any later version.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
-
-
-#ifdef __GNUC__
-#pragma interface			/* gcc class implementation */
-#endif
-
-/* class for the the myisam merge handler */
-
-#include <merge.h>
-
-class ha_isammrg: public handler
-{
-  MRG_INFO *file;
-
- public:
-  ha_isammrg(TABLE *table): handler(table), file(0) {}
-  ~ha_isammrg() {}
-  const char *table_type() const { return "MRG_ISAM"; }
-  const char **bas_ext() const;
-  ulong table_flags() const { return (HA_READ_RND_SAME |
-				      HA_REC_NOT_IN_SEQ | HA_FILE_BASED); }
-  ulong index_flags(uint idx, uint part, bool all_parts) const
-  { DBUG_ASSERT(0); return 0; }
-
-  uint max_supported_keys()          const { return 0; }
-  bool low_byte_first()	   const { return 0; }
-  uint min_record_length(uint options) const;
-
-  int open(const char *name, int mode, uint test_if_locked);
-  int close(void);
-  int write_row(byte * buf);
-  int update_row(const byte * old_data, byte * new_data);
-  int delete_row(const byte * buf);
-  int index_read(byte * buf, const byte * key,
-		 uint key_len, enum ha_rkey_function find_flag);
-  int index_read_idx(byte * buf, uint indx, const byte * key,
-		     uint key_len, enum ha_rkey_function find_flag);
-  int index_next(byte * buf);
-  int index_prev(byte * buf);
-  int index_first(byte * buf);
-  int index_last(byte * buf);
-  int rnd_init(bool scan);
-  int rnd_next(byte *buf);
-  int rnd_pos(byte * buf, byte *pos);
-  void position(const byte *record);
-  void info(uint);
-  int extra(enum ha_extra_function operation);
-  int external_lock(THD *thd, int lock_type);
-  uint lock_count(void) const;
-  int create(const char *name, TABLE *form, HA_CREATE_INFO *create_info);
-  THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
-			     enum thr_lock_type lock_type);
-  uint8 table_cache_type() { return HA_CACHE_TBL_NOCACHE; }
-};
diff --git a/sql/ha_myisam.cc b/sql/ha_myisam.cc
index 8b8824448ba..726647cd131 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -1377,9 +1377,9 @@ int ha_myisam::create(const char *name, register TABLE *table_arg,
     keydef[i].keysegs=pos->key_parts;
     for (j=0 ; j < pos->key_parts ; j++)
     {
-      keydef[i].seg[j].flag=pos->key_part[j].key_part_flag;
       Field *field=pos->key_part[j].field;
       type=field->key_type();
+      keydef[i].seg[j].flag=pos->key_part[j].key_part_flag;
 
       if (options & HA_OPTION_PACK_KEYS ||
 	  (pos->flags & (HA_PACK_KEY | HA_BINARY_PACK_KEY |
@@ -1394,8 +1394,8 @@ int ha_myisam::create(const char *name, register TABLE *table_arg,
 	  if (j == 0)
 	    keydef[i].flag|=HA_PACK_KEY;
 	  if (!(field->flags & ZEROFILL_FLAG) &&
-	      (field->type() == FIELD_TYPE_STRING ||
-	       field->type() == FIELD_TYPE_VAR_STRING ||
+	      (field->type() == MYSQL_TYPE_STRING ||
+	       field->type() == MYSQL_TYPE_VAR_STRING ||
 	       ((int) (pos->key_part[j].length - field->decimals()))
 	       >= 4))
 	    keydef[i].seg[j].flag|=HA_SPACE_PACK;
@@ -1474,30 +1474,31 @@ int ha_myisam::create(const char *name, register TABLE *table_arg,
     {
       recinfo_pos->type= (int) FIELD_BLOB;
     }
-    else if (!(options & HA_OPTION_PACK_RECORD))
+    else if (!(options & HA_OPTION_PACK_RECORD) ||
+             found->type() == MYSQL_TYPE_VARCHAR)
       recinfo_pos->type= (int) FIELD_NORMAL;
     else if (found->zero_pack())
       recinfo_pos->type= (int) FIELD_SKIP_ZERO;
     else
       recinfo_pos->type= (int) ((length <= 3 ||
-				      (found->flags & ZEROFILL_FLAG)) ?
-				     FIELD_NORMAL :
-				     found->type() == FIELD_TYPE_STRING ||
-				     found->type() == FIELD_TYPE_VAR_STRING ?
-				     FIELD_SKIP_ENDSPACE :
-				     FIELD_SKIP_PRESPACE);
+                                 (found->flags & ZEROFILL_FLAG)) ?
+                                FIELD_NORMAL :
+                                found->type() == MYSQL_TYPE_STRING ||
+                                found->type() == MYSQL_TYPE_VAR_STRING ?
+                                FIELD_SKIP_ENDSPACE :
+                                FIELD_SKIP_PRESPACE);
     if (found->null_ptr)
     {
       recinfo_pos->null_bit=found->null_bit;
       recinfo_pos->null_pos= (uint) (found->null_ptr-
-				  (uchar*) table_arg->record[0]);
+                                     (uchar*) table_arg->record[0]);
     }
     else
     {
       recinfo_pos->null_bit=0;
       recinfo_pos->null_pos=0;
     }
-    (recinfo_pos++) ->length=(uint16) length;
+    (recinfo_pos++)->length= (uint16) length;
     recpos=minpos+length;
     DBUG_PRINT("loop",("length: %d  type: %d",
 		       recinfo_pos[-1].length,recinfo_pos[-1].type));
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 9133de9c546..f838bfcd8c3 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -406,6 +406,7 @@ static inline bool ndb_supported_type(enum_field_types type)
     return TRUE;
   case MYSQL_TYPE_NULL:   
   case MYSQL_TYPE_GEOMETRY:
+  case MYSQL_TYPE_VARCHAR:
     break;
   }
   return FALSE;
@@ -1843,7 +1844,7 @@ int ha_ndbcluster::key_cmp(uint keynr, const byte * old_row,
 	  (new_row[key_part->null_offset] & key_part->null_bit))
 	return 1;
     }
-    if (key_part->key_part_flag & (HA_BLOB_PART | HA_VAR_LENGTH))
+    if (key_part->key_part_flag & (HA_BLOB_PART | HA_VAR_LENGTH_PART))
     {
 
       if (key_part->field->cmp_binary((char*) (old_row + key_part->offset),
@@ -3774,6 +3775,7 @@ ha_ndbcluster::ha_ndbcluster(TABLE *table_arg):
   m_table_flags(HA_REC_NOT_IN_SEQ |
 		HA_NULL_IN_KEY |
 		HA_AUTO_PART_KEY |
+                HA_NO_VARCHAR |
 		HA_NO_PREFIX_CHAR_KEYS),
   m_share(0),
   m_use_write(FALSE),
diff --git a/sql/handler.h b/sql/handler.h
index c70ea266734..2800c9b07ec 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -74,6 +74,7 @@
 #define HA_HAS_CHECKSUM        (1 << 24)
 /* Table data are stored in separate files (for lower_case_table_names) */
 #define HA_FILE_BASED	       (1 << 26)
+#define HA_NO_VARCHAR	       (1 << 27)
 
 
 /* bits in index_flags(index_number) for what you can do with index */
@@ -252,6 +253,7 @@ typedef struct st_ha_create_information
   uint merge_insert_method;
   bool table_existed;			/* 1 in create if table existed */
   bool frm_only;                        /* 1 if no ha_create_table() */
+  bool varchar;                         /* 1 if table has a VARCHAR */
 } HA_CREATE_INFO;
 
 
diff --git a/sql/item.cc b/sql/item.cc
index 03bbbe2ad49..d40721e9ae2 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -188,7 +188,6 @@ void Item_ident::cleanup()
 {
   DBUG_ENTER("Item_ident::cleanup");
 #ifdef CANT_BE_USED_AS_MEMORY_IS_FREED
-  DBUG_PRINT("enter", ("b:%s(%s), t:%s(%s), f:%s(%s)",
 		       db_name ? db_name : "(null)",
                        orig_db_name ? orig_db_name : "(null)",
 		       table_name ? table_name : "(null)",
@@ -2237,18 +2236,29 @@ void Item::make_field(Send_field *tmp_field)
 
 void Item_empty_string::make_field(Send_field *tmp_field)
 {
-  init_make_field(tmp_field,FIELD_TYPE_VAR_STRING);
+  init_make_field(tmp_field, MYSQL_TYPE_VARCHAR);
 }
 
 
 enum_field_types Item::field_type() const
 {
-  return ((result_type() == STRING_RESULT) ? FIELD_TYPE_VAR_STRING :
+  return ((result_type() == STRING_RESULT) ? MYSQL_TYPE_VARCHAR :
 	  (result_type() == INT_RESULT) ? FIELD_TYPE_LONGLONG :
 	  FIELD_TYPE_DOUBLE);
 }
 
 
+/*
+  Create a field based on field_type of argument
+
+  For now, this is only used to create a field for
+  IFNULL(x,something)
+
+  RETURN
+    0  error
+    #  Created field
+*/
+
 Field *Item::tmp_table_field_from_field_type(TABLE *table)
 {
   /*
@@ -2304,12 +2314,17 @@ Field *Item::tmp_table_field_from_field_type(TABLE *table)
   case MYSQL_TYPE_ENUM:
   case MYSQL_TYPE_SET:
   case MYSQL_TYPE_VAR_STRING:
-    if (max_length > 255)
-      break;					// If blob
+    if (max_length > MAX_FIELD_CHARLENGTH)
+      break;					// convert to blob
+    return new Field_varstring(max_length, maybe_null, name, table,
+			       collation.collation);
+  case MYSQL_TYPE_VARCHAR:
+    if (max_length > CONVERT_IF_BIGGER_TO_BLOB)
+      break;					// convert to blob
     return new Field_varstring(max_length, maybe_null, name, table,
 			       collation.collation);
   case MYSQL_TYPE_STRING:
-    if (max_length > 255)			// If blob
+    if (max_length > MAX_FIELD_CHARLENGTH)	// If blob
       break;
     return new Field_string(max_length, maybe_null, name, table,
 			    collation.collation);
@@ -2637,6 +2652,7 @@ bool Item::send(Protocol *protocol, String *buffer)
   case MYSQL_TYPE_GEOMETRY:
   case MYSQL_TYPE_STRING:
   case MYSQL_TYPE_VAR_STRING:
+  case MYSQL_TYPE_VARCHAR:
   {
     String *res;
     if ((res=val_str(buffer)))
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 7611b70c6a4..90eee9e76d1 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -1726,7 +1726,7 @@ static int srtcmp_in(CHARSET_INFO *cs, const String *x,const String *y)
 {
   return cs->coll->strnncollsp(cs,
                                (uchar *) x->ptr(),x->length(),
-                               (uchar *) y->ptr(),y->length());
+                               (uchar *) y->ptr(),y->length(), 0);
 }
 
 
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index d0bf98e7adb..b242698d36e 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -1194,7 +1194,7 @@ int simple_str_key_cmp(void* arg, byte* key1, byte* key2)
   uint len=item->key_length;
   return cs->coll->strnncollsp(cs, 
 			       (const uchar*) key1, len, 
-			       (const uchar*) key2, len);
+			       (const uchar*) key2, len, 0);
 }
 
 /*
@@ -1344,10 +1344,9 @@ bool Item_sum_count_distinct::setup(THD *thd)
 	about other fields
       */
       Field* field = table->field[0];
-      switch(field->type())
-      {
-      case FIELD_TYPE_STRING:
-      case FIELD_TYPE_VAR_STRING:
+      switch (field->type()) {
+      case MYSQL_TYPE_STRING:
+      case MYSQL_TYPE_VAR_STRING:
 	if (field->binary())
 	{
 	  compare_key = (qsort_cmp2)simple_raw_key_cmp;
diff --git a/sql/key.cc b/sql/key.cc
index 15d84b73beb..dfd924f1dc7 100644
--- a/sql/key.cc
+++ b/sql/key.cc
@@ -106,14 +106,21 @@ void key_copy(byte *to_key, byte *from_record, KEY *key_info, uint key_length)
     {
       char *pos;
       ulong blob_length= ((Field_blob*) key_part->field)->get_length();
-      key_length-= 2;
+      key_length-= HA_KEY_BLOB_LENGTH;
       ((Field_blob*) key_part->field)->get_ptr(&pos);
       length=min(key_length, key_part->length);
       set_if_smaller(blob_length, length);
       int2store(to_key, (uint) blob_length);
-      to_key+= 2;					// Skip length info
+      to_key+= HA_KEY_BLOB_LENGTH;			// Skip length info
       memcpy(to_key, pos, blob_length);
     }
+    else if (key_part->key_part_flag & HA_VAR_LENGTH_PART)
+    {
+      key_length-= HA_KEY_BLOB_LENGTH;
+      length= min(key_length, key_part->length);
+      key_part->field->get_key_image(to_key, length, Field::itRAW);
+      to_key+= HA_KEY_BLOB_LENGTH;
+    }
     else
     {
       length= min(key_length, key_part->length);
@@ -166,12 +173,19 @@ void key_restore(byte *to_record, byte *from_key, KEY *key_info,
     if (key_part->key_part_flag & HA_BLOB_PART)
     {
       uint blob_length= uint2korr(from_key);
-      from_key+= 2;
-      key_length-= 2;
+      from_key+= HA_KEY_BLOB_LENGTH;
+      key_length-= HA_KEY_BLOB_LENGTH;
       ((Field_blob*) key_part->field)->set_ptr((ulong) blob_length,
 					       (char*) from_key);
       length= key_part->length;
     }
+    else if (key_part->key_part_flag & HA_VAR_LENGTH_PART)
+    {
+      key_length-= HA_KEY_BLOB_LENGTH;
+      length= min(key_length, key_part->length);
+      key_part->field->set_key_image(from_key, length);
+      from_key+= HA_KEY_BLOB_LENGTH;
+    }
     else
     {
       length= min(key_length, key_part->length);
@@ -226,9 +240,9 @@ bool key_cmp_if_same(TABLE *table,const byte *key,uint idx,uint key_length)
       }
       key++;
     }
-    if (key_part->key_part_flag & (HA_BLOB_PART | HA_VAR_LENGTH))
+    if (key_part->key_part_flag & (HA_BLOB_PART | HA_VAR_LENGTH_PART))
     {
-      if (key_part->field->key_cmp(key, key_part->length+ HA_KEY_BLOB_LENGTH))
+      if (key_part->field->key_cmp(key, key_part->length))
 	return 1;
       length=key_part->length+HA_KEY_BLOB_LENGTH;
     }
@@ -248,7 +262,7 @@ bool key_cmp_if_same(TABLE *table,const byte *key,uint idx,uint key_length)
         }
 	if (cs->coll->strnncollsp(cs,
                                   (const uchar*) key, length,
-                                  (const uchar*) pos, char_length))
+                                  (const uchar*) pos, char_length, 0))
 	  return 1;
       }
       else if (memcmp(key,table->record[0]+key_part->offset,length))
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index e92d266ba28..f9149f10a30 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -1749,8 +1749,10 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
           objects are not allowed so don't use ROR-intersection for
           table deletes.
         */
-        if ((thd->lex->sql_command != SQLCOM_DELETE) )//&& 
-//          (thd->lex->sql_command != SQLCOM_UPDATE))
+        if ((thd->lex->sql_command != SQLCOM_DELETE))
+#ifdef NOT_USED
+          if ((thd->lex->sql_command != SQLCOM_UPDATE))
+#endif
         {
           /*
             Get best non-covering ROR-intersection plan and prepare data for
@@ -3693,8 +3695,7 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
     DBUG_RETURN(0);
   if (maybe_null)
     *str= (char) field->is_real_null();		// Set to 1 if null
-  field->get_key_image(str+maybe_null, key_part->length,
-		       field->charset(), key_part->image_type);
+  field->get_key_image(str+maybe_null, key_part->length, key_part->image_type);
   if (copies == 2)
   {
     /*
@@ -8444,7 +8445,7 @@ print_key(KEY_PART *key_part,const char *key,uint used_length)
       key++;					// Skip null byte
       store_length--;
     }
-    field->set_key_image((char*) key, key_part->length, field->charset());
+    field->set_key_image((char*) key, key_part->length);
     field->val_str(&tmp);
     fwrite(tmp.ptr(),sizeof(char),tmp.length(),DBUG_FILE);
     if (key+store_length < key_end)
diff --git a/sql/opt_sum.cc b/sql/opt_sum.cc
index 55efcce1c19..80226dcfa2c 100644
--- a/sql/opt_sum.cc
+++ b/sql/opt_sum.cc
@@ -562,8 +562,7 @@ static bool matching_cond(bool max_fl, TABLE_REF *ref, KEY *keyinfo,
       store_val_in_field(part->field, args[between && max_fl ? 2 : 1]);
       if (part->null_bit) 
         *key_ptr++= (byte) test(part->field->is_null());
-      part->field->get_key_image((char*) key_ptr, part->length,
-                                 part->field->charset(), Field::itRAW);
+      part->field->get_key_image((char*) key_ptr, part->length, Field::itRAW);
     }
     if (is_field_part)
     {
diff --git a/sql/protocol.cc b/sql/protocol.cc
index 11a915ec151..d2e63539610 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -511,6 +511,11 @@ bool Protocol::send_fields(List<Item> *list, uint flags)
     CHARSET_INFO *cs= system_charset_info;
     Send_field field;
     item->make_field(&field);
+
+    /* Keep things compatible for old clients */
+    if (field.type == MYSQL_TYPE_VARCHAR)
+      field.type= MYSQL_TYPE_VAR_STRING;
+
     prot.prepare_for_resend();
 
     if (thd->client_capabilities & CLIENT_PROTOCOL_41)
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 4d4923d16b8..17007cbcc5d 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -1053,9 +1053,8 @@ static void acl_insert_db(const char *user, const char *host, const char *db,
 ulong acl_get(const char *host, const char *ip,
               const char *user, const char *db, my_bool db_is_pattern)
 {
-  ulong host_access,db_access;
+  ulong host_access= ~0, db_access= 0;
   uint i,key_length;
-  db_access=0; host_access= ~0;
   char key[ACL_KEY_LENGTH],*tmp_db,*end;
   acl_entry *entry;
   DBUG_ENTER("acl_get");
@@ -1438,6 +1437,7 @@ static bool update_user_table(THD *thd, const char *host, const char *user,
   TABLE_LIST tables;
   TABLE *table;
   bool error=1;
+  char user_key[MAX_KEY_LENGTH];
   DBUG_ENTER("update_user_table");
   DBUG_PRINT("enter",("user: %s  host: %s",user,host));
 
@@ -1467,11 +1467,12 @@ static bool update_user_table(THD *thd, const char *host, const char *user,
     DBUG_RETURN(1); /* purecov: deadcode */
   table->field[0]->store(host,(uint) strlen(host), system_charset_info);
   table->field[1]->store(user,(uint) strlen(user), system_charset_info);
+  key_copy(user_key, table->record[0], table->key_info,
+           table->key_info->key_length);
 
   table->file->extra(HA_EXTRA_RETRIEVE_ALL_COLS);
-  if (table->file->index_read_idx(table->record[0],0,
-				  (byte*) table->field[0]->ptr,
-				  table->key_info[0].key_length,
+  if (table->file->index_read_idx(table->record[0], 0,
+				  user_key, table->key_info->key_length,
 				  HA_READ_KEY_EXACT))
   {
     my_message(ER_PASSWORD_NO_MATCH, ER(ER_PASSWORD_NO_MATCH),
@@ -1531,8 +1532,10 @@ static int replace_user_table(THD *thd, TABLE *table, const LEX_USER &combo,
   const char *password= "";
   uint password_len= 0;
   char what= (revoke_grant) ? 'N' : 'Y';
-  DBUG_ENTER("replace_user_table");
+  byte user_key[MAX_KEY_LENGTH];
   LEX *lex= thd->lex;
+  DBUG_ENTER("replace_user_table");
+
   safe_mutex_assert_owner(&acl_cache->lock);
 
   if (combo.password.str && combo.password.str[0])
@@ -1549,11 +1552,13 @@ static int replace_user_table(THD *thd, TABLE *table, const LEX_USER &combo,
 
   table->field[0]->store(combo.host.str,combo.host.length, system_charset_info);
   table->field[1]->store(combo.user.str,combo.user.length, system_charset_info);
+  key_copy(user_key, table->record[0], table->key_info,
+           table->key_info->key_length);
+
   table->file->extra(HA_EXTRA_RETRIEVE_ALL_COLS);
   if (table->file->index_read_idx(table->record[0], 0,
-				  (byte*) table->field[0]->ptr,
-				  table->key_info[0].key_length,
-				  HA_READ_KEY_EXACT))
+                                  user_key, table->key_info->key_length,
+                                  HA_READ_KEY_EXACT))
   {
     /* what == 'N' means revoke */
     if (what == 'N')
@@ -1742,6 +1747,7 @@ static int replace_db_table(TABLE *table, const char *db,
   bool old_row_exists=0;
   int error;
   char what= (revoke_grant) ? 'N' : 'Y';
+  byte user_key[MAX_KEY_LENGTH];
   DBUG_ENTER("replace_db_table");
 
   if (!initialized)
@@ -1760,11 +1766,13 @@ static int replace_db_table(TABLE *table, const char *db,
   table->field[0]->store(combo.host.str,combo.host.length, system_charset_info);
   table->field[1]->store(db,(uint) strlen(db), system_charset_info);
   table->field[2]->store(combo.user.str,combo.user.length, system_charset_info);
+  key_copy(user_key, table->record[0], table->key_info,
+           table->key_info->key_length);
+
   table->file->extra(HA_EXTRA_RETRIEVE_ALL_COLS);
   if (table->file->index_read_idx(table->record[0],0,
-				  (byte*) table->field[0]->ptr,
-				  table->key_info[0].key_length,
-				  HA_READ_KEY_EXACT))
+                                  user_key, table->key_info->key_length,
+                                  HA_READ_KEY_EXACT))
   {
     if (what == 'N')
     { // no row, no revoke
@@ -1935,22 +1943,25 @@ GRANT_TABLE::GRANT_TABLE(TABLE *form, TABLE *col_privs)
                    0,0,0, (hash_get_key) get_key_column,0,0);
   if (cols)
   {
-    int key_len;
+    uint key_prefix_len;
+    KEY_PART_INFO *key_part= col_privs->key_info->key_part;
     col_privs->field[0]->store(orig_host,(uint) strlen(orig_host),
                                system_charset_info);
     col_privs->field[1]->store(db,(uint) strlen(db), system_charset_info);
     col_privs->field[2]->store(user,(uint) strlen(user), system_charset_info);
     col_privs->field[3]->store(tname,(uint) strlen(tname), system_charset_info);
-    key_len=(col_privs->field[0]->pack_length()+
-             col_privs->field[1]->pack_length()+
-             col_privs->field[2]->pack_length()+
-             col_privs->field[3]->pack_length());
-    key_copy(key,col_privs->record[0],col_privs->key_info,key_len);
+
+    key_prefix_len= (key_part[0].store_length +
+                     key_part[1].store_length +
+                     key_part[2].store_length +
+                     key_part[3].store_length);
+    key_copy(key, col_privs->record[0], col_privs->key_info, key_prefix_len);
     col_privs->field[4]->store("",0, &my_charset_latin1);
+
     col_privs->file->ha_index_init(0);
     if (col_privs->file->index_read(col_privs->record[0],
-                                    (byte*) col_privs->field[0]->ptr,
-                                    key_len, HA_READ_KEY_EXACT))
+                                    (byte*) key,
+                                    key_prefix_len, HA_READ_KEY_EXACT))
     {
       cols = 0; /* purecov: deadcode */
       col_privs->file->ha_index_end();
@@ -1972,7 +1983,7 @@ GRANT_TABLE::GRANT_TABLE(TABLE *form, TABLE *col_privs)
       }
       my_hash_insert(&hash_columns, (byte *) mem_check);
     } while (!col_privs->file->index_next(col_privs->record[0]) &&
-             !key_cmp_if_same(col_privs,key,0,key_len));
+             !key_cmp_if_same(col_privs,key,0,key_prefix_len));
     col_privs->file->ha_index_end();
   }
 }
@@ -2053,19 +2064,22 @@ static int replace_column_table(GRANT_TABLE *g_t,
 				ulong rights, bool revoke_grant)
 {
   int error=0,result=0;
-  uint key_length;
   byte key[MAX_KEY_LENGTH];
+  uint key_prefix_length;
+  KEY_PART_INFO *key_part= table->key_info->key_part;
   DBUG_ENTER("replace_column_table");
 
   table->field[0]->store(combo.host.str,combo.host.length, system_charset_info);
   table->field[1]->store(db,(uint) strlen(db), system_charset_info);
   table->field[2]->store(combo.user.str,combo.user.length, system_charset_info);
   table->field[3]->store(table_name,(uint) strlen(table_name), system_charset_info);
-  key_length=(table->field[0]->pack_length()+ table->field[1]->pack_length()+
-	      table->field[2]->pack_length()+ table->field[3]->pack_length());
-  key_copy(key,table->record[0],table->key_info,key_length);
 
-  rights &= COL_ACLS;				// Only ACL for columns
+  /* Get length of 3 first key parts */
+  key_prefix_length= (key_part[0].store_length + key_part[1].store_length +
+                      key_part[2].store_length + key_part[3].store_length);
+  key_copy(key, table->record[0], table->key_info, key_prefix_length);
+
+  rights&= COL_ACLS;				// Only ACL for columns
 
   /* first fix privileges for all columns in column list */
 
@@ -2076,14 +2090,20 @@ static int replace_column_table(GRANT_TABLE *g_t,
   {
     ulong privileges = xx->rights;
     bool old_row_exists=0;
-    key_restore(table->record[0],key,table->key_info,key_length);
+    byte user_key[MAX_KEY_LENGTH];
+
+    key_restore(table->record[0],key,table->key_info,
+                key_prefix_length);
     table->field[4]->store(xx->column.ptr(),xx->column.length(),
                            system_charset_info);
+    /* Get key for the first 4 columns */
+    key_copy(user_key, table->record[0], table->key_info,
+             table->key_info->key_length);
 
     table->file->extra(HA_EXTRA_RETRIEVE_ALL_COLS);
-    if (table->file->index_read(table->record[0],(byte*) table->field[0]->ptr,
-				table->key_info[0].key_length,
-				HA_READ_KEY_EXACT))
+    if (table->file->index_read(table->record[0], user_key,
+				table->key_info->key_length,
+                                HA_READ_KEY_EXACT))
     {
       if (revoke_grant)
       {
@@ -2095,7 +2115,8 @@ static int replace_column_table(GRANT_TABLE *g_t,
       }
       old_row_exists = 0;
       restore_record(table,default_values);		// Get empty record
-      key_restore(table->record[0],key,table->key_info,key_length);
+      key_restore(table->record[0],key,table->key_info,
+                  key_prefix_length);
       table->field[4]->store(xx->column.ptr(),xx->column.length(),
                              system_charset_info);
     }
@@ -2152,10 +2173,14 @@ static int replace_column_table(GRANT_TABLE *g_t,
 
   if (revoke_grant)
   {
+    byte user_key[MAX_KEY_LENGTH];
+    key_copy(user_key, table->record[0], table->key_info,
+             key_prefix_length);
+
     table->file->extra(HA_EXTRA_RETRIEVE_ALL_COLS);
-    if (table->file->index_read(table->record[0], (byte*) table->field[0]->ptr,
-                                key_length,
-				HA_READ_KEY_EXACT))
+    if (table->file->index_read(table->record[0], user_key,
+				key_prefix_length,
+                                HA_READ_KEY_EXACT))
       goto end;
 
     /* Scan through all rows with the same host,db,user and table */
@@ -2169,7 +2194,8 @@ static int replace_column_table(GRANT_TABLE *g_t,
       {
 	GRANT_COLUMN *grant_column = NULL;
 	char  colum_name_buf[HOSTNAME_LENGTH+1];
-	String column_name(colum_name_buf,sizeof(colum_name_buf),system_charset_info);
+	String column_name(colum_name_buf,sizeof(colum_name_buf),
+                           system_charset_info);
 
 	privileges&= ~rights;
 	table->field[6]->store((longlong)
@@ -2205,7 +2231,7 @@ static int replace_column_table(GRANT_TABLE *g_t,
 	}
       }
     } while (!table->file->index_next(table->record[0]) &&
-	     !key_cmp_if_same(table,key,0,key_length));
+	     !key_cmp_if_same(table, key, 0, key_prefix_length));
   }
 
 end:
@@ -2224,6 +2250,7 @@ static int replace_table_table(THD *thd, GRANT_TABLE *grant_table,
   int old_row_exists = 1;
   int error=0;
   ulong store_table_rights, store_col_rights;
+  byte user_key[MAX_KEY_LENGTH];
   DBUG_ENTER("replace_table_table");
 
   strxmov(grantor, thd->user, "@", thd->host_or_ip, NullS);
@@ -2245,10 +2272,12 @@ static int replace_table_table(THD *thd, GRANT_TABLE *grant_table,
   table->field[2]->store(combo.user.str,combo.user.length, system_charset_info);
   table->field[3]->store(table_name,(uint) strlen(table_name), system_charset_info);
   store_record(table,record[1]);			// store at pos 1
+  key_copy(user_key, table->record[0], table->key_info,
+           table->key_info->key_length);
+
   table->file->extra(HA_EXTRA_RETRIEVE_ALL_COLS);
-  if (table->file->index_read_idx(table->record[0],0,
-				  (byte*) table->field[0]->ptr,
-				  table->key_info[0].key_length,
+  if (table->file->index_read_idx(table->record[0], 0,
+                                  user_key, table->key_info->key_length,
 				  HA_READ_KEY_EXACT))
   {
     /*
@@ -3702,6 +3731,8 @@ static int modify_grant_table(TABLE *table, Field *host_field,
                               Field *user_field, LEX_USER *user_to)
 {
   int error;
+  TABLE *table;
+  byte user_key[MAX_KEY_LENGTH];
   DBUG_ENTER("modify_grant_table");
 
   if (user_to)
@@ -3769,6 +3800,7 @@ static int handle_grant_table(TABLE_LIST *tables, uint table_no, bool drop,
   char *user_str= user_from->user.str;
   const char *host;
   const char *user;
+  uint key_prefix_length;
   DBUG_ENTER("handle_grant_table");
 
   if (! table_no) // mysql.user table
@@ -3786,8 +3818,13 @@ static int handle_grant_table(TABLE_LIST *tables, uint table_no, bool drop,
                        table->real_name, user_str, host_str));
     host_field->store(host_str, user_from->host.length, system_charset_info);
     user_field->store(user_str, user_from->user.length, system_charset_info);
+
+    key_prefix_length= (table->key_info->key_part[0].store_length +
+                        table->key_info->key_part[1].store_length);
+    key_copy(user_key, table->record[0], table->key_info, key_prefix_length);
+
     if ((error= table->file->index_read_idx(table->record[0], 0,
-                                            (byte*) host_field->ptr, 0,
+                                            user_key, key_prefix_length,
                                             HA_READ_KEY_EXACT)))
     {
       if (error != HA_ERR_KEY_NOT_FOUND)
@@ -4058,7 +4095,8 @@ static int handle_grant_data(TABLE_LIST *tables, bool drop,
   else
   {
     /* Handle user array. */
-    if ((handle_grant_struct(0, drop, user_from, user_to) && ! result) || found)
+    if ((handle_grant_struct(0, drop, user_from, user_to) && ! result) ||
+        found)
     {
       result= 1; /* At least one record/element found. */
       /* If search is requested, we do not need to search further. */
@@ -4120,6 +4158,7 @@ static int handle_grant_data(TABLE_LIST *tables, bool drop,
   DBUG_RETURN(result);
 }
 
+
 static void append_user(String *str, LEX_USER *user)
 {
   if (str->length())
@@ -4131,6 +4170,7 @@ static void append_user(String *str, LEX_USER *user)
   str->append('\'');
 }
 
+
 /*
   Create a list of users.
 
@@ -4298,6 +4338,7 @@ bool mysql_rename_user(THD *thd, List <LEX_USER> &list)
   DBUG_RETURN(result);
 }
 
+
 /*
   Revoke all privileges from a list of users.
 
@@ -4332,9 +4373,8 @@ bool mysql_revoke_all(THD *thd,  List <LEX_USER> &list)
   {
     if (!check_acl_user(lex_user, &counter))
     {
-      sql_print_error("REVOKE ALL PRIVILEGES, GRANT: User '%s'@'%s' not exists",
-		      lex_user->user.str,
-		      lex_user->host.str);
+      sql_print_error("REVOKE ALL PRIVILEGES, GRANT: User '%s'@'%s' does not "
+                      "exists", lex_user->user.str, lex_user->host.str);
       result= -1;
       continue;
     }
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 3ead7163171..b4a2f368bc2 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1063,31 +1063,6 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     for (uint i=0 ; i < table->fields ; i++)
       table->field[i]->table_name=table->table_name;
   }
-#if MYSQL_VERSION_ID < 40100
-  /*
-    If per-connection "new" variable (represented by variables.new_mode)
-    is set then we should pretend that the length of TIMESTAMP field is 19.
-    The cheapest (from perfomance viewpoint) way to achieve that is to set
-    field_length of all Field_timestamp objects in a table after opening
-    it (to 19 if new_mode is true or to original field length otherwise).
-    We save value of new_mode variable in TABLE::timestamp_mode to
-    not perform this setup if new_mode value is the same between sequential
-    table opens.
-  */
-  my_bool new_mode= thd->variables.new_mode;
-  if (table->timestamp_mode != new_mode)
-  {
-    for (uint i=0 ; i < table->fields ; i++)
-    {
-      Field *field= table->field[i];
-
-      if (field->type() == FIELD_TYPE_TIMESTAMP)
-        field->field_length= new_mode ? 19 :
-                             ((Field_timestamp *)(field))->orig_field_length;
-    }
-    table->timestamp_mode= new_mode;
-  }
-#endif
   /* These variables are also set in reopen_table() */
   table->tablenr=thd->current_tablenr++;
   table->used_fields=0;
diff --git a/sql/sql_help.cc b/sql/sql_help.cc
index bcde1dc46f2..f71e9a7133a 100644
--- a/sql/sql_help.cc
+++ b/sql/sql_help.cc
@@ -286,8 +286,7 @@ int get_topics_for_keyword(THD *thd, TABLE *topics, TABLE *relations,
   relations->file->ha_index_init(iindex_relations);
 
   rkey_id->store((longlong) key_id);
-  rkey_id->get_key_image(buff, rkey_id->pack_length(), rkey_id->charset(),
-			 Field::itRAW);
+  rkey_id->get_key_image(buff, rkey_id->pack_length(), Field::itRAW);
   int key_res= relations->file->index_read(relations->record[0],
 					   (byte *)buff, rkey_id->pack_length(),
 					   HA_READ_KEY_EXACT);
@@ -300,8 +299,7 @@ int get_topics_for_keyword(THD *thd, TABLE *topics, TABLE *relations,
     longlong topic_id= rtopic_id->val_int();
     Field *field= find_fields[help_topic_help_topic_id].field;
     field->store((longlong) topic_id);
-    field->get_key_image(topic_id_buff, field->pack_length(), field->charset(),
-			 Field::itRAW);
+    field->get_key_image(topic_id_buff, field->pack_length(), Field::itRAW);
 
     if (!topics->file->index_read(topics->record[0], (byte *)topic_id_buff,
 				  field->pack_length(), HA_READ_KEY_EXACT))
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 168632f7578..3d4252a2b17 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -118,15 +118,20 @@ static void unlock_locked_tables(THD *thd)
 static bool end_active_trans(THD *thd)
 {
   int error=0;
+  DBUG_ENTER("end_active_trans");
   if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN |
 		      OPTION_TABLE_LOCK))
   {
+    DBUG_PRINT("info",("options: 0x%lx", (ulong) thd->options));
     thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_STATUS_NO_TRANS_UPDATE);
+    /* Safety if one did "drop table" on locked tables */
+    if (!thd->locked_tables)
+      thd->options&= ~OPTION_TABLE_LOCK;
     thd->server_status&= ~SERVER_STATUS_IN_TRANS;
     if (ha_commit(thd))
       error=1;
   }
-  return error;
+  DBUG_RETURN(error);
 }
 
 
@@ -4736,7 +4741,7 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
   LEX  *lex= thd->lex;
   uint allowed_type_modifier=0;
   uint sign_len;
-  char warn_buff[MYSQL_ERRMSG_SIZE];
+  ulong max_field_charlength= MAX_FIELD_CHARLENGTH;
   DBUG_ENTER("add_field_to_list");
 
   if (strlen(field_name) > NAME_LEN)
@@ -4812,7 +4817,7 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
   new_field->length=0;
   new_field->change=change;
   new_field->interval=0;
-  new_field->pack_length=0;
+  new_field->pack_length= new_field->key_length= 0;
   new_field->charset=cs;
   new_field->geom_type= (Field::geometry_type) uint_geom_type;
 
@@ -4884,36 +4889,22 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
 	new_field->length++;
     }
     break;
-  case FIELD_TYPE_STRING:
-  case FIELD_TYPE_VAR_STRING:
-    if (new_field->length <= MAX_FIELD_CHARLENGTH || default_value)
-      break;
-    /* Convert long CHAR() and VARCHAR columns to TEXT or BLOB */
-    new_field->sql_type= FIELD_TYPE_BLOB;
-    sprintf(warn_buff, ER(ER_AUTO_CONVERT), field_name, "CHAR",
-	    (cs == &my_charset_bin) ? "BLOB" : "TEXT");
-    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_AUTO_CONVERT,
-		 warn_buff);
-    /* fall through */
+  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:
+    break;
   case FIELD_TYPE_BLOB:
   case FIELD_TYPE_TINY_BLOB:
   case FIELD_TYPE_LONG_BLOB:
   case FIELD_TYPE_MEDIUM_BLOB:
   case FIELD_TYPE_GEOMETRY:
-    if (new_field->length)
-    {
-      /* The user has given a length to the blob column */
-      if (new_field->length < 256)
-	type= FIELD_TYPE_TINY_BLOB;
-      if (new_field->length < 65536)
-	type= FIELD_TYPE_BLOB;
-      else if (new_field->length < 256L*256L*256L)
-	type= FIELD_TYPE_MEDIUM_BLOB;
-      else
-	type= FIELD_TYPE_LONG_BLOB;
-      new_field->length= 0;
-    }
-    new_field->sql_type= type;
     if (default_value)				// Allow empty as default value
     {
       String str,*res;
@@ -5055,8 +5046,8 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
         when we know the character set of the column
       */
       new_field->length= 1;
+      break;
     }
-    break;
   case FIELD_TYPE_ENUM:
     {
       // Should be safe
@@ -5067,18 +5058,23 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
       while ((tmp= it++))
         new_field->interval_list.push_back(tmp);
       new_field->length= 1; // See comment for FIELD_TYPE_SET above.
-    }
+      break;
+   }
+  case MYSQL_TYPE_VAR_STRING:
+    DBUG_ASSERT(0);                             // Impossible
     break;
   }
 
-  if ((new_field->length > MAX_FIELD_CHARLENGTH && type != FIELD_TYPE_SET && 
-       type != FIELD_TYPE_ENUM) ||
-      (!new_field->length && !(new_field->flags & BLOB_FLAG) &&
-       type != FIELD_TYPE_STRING &&
-       type != FIELD_TYPE_VAR_STRING && type != FIELD_TYPE_GEOMETRY))
+  if (!(new_field->flags & BLOB_FLAG) &&
+      ((new_field->length > max_field_charlength && type != FIELD_TYPE_SET && 
+        type != FIELD_TYPE_ENUM &&
+        (type != MYSQL_TYPE_VARCHAR || default_value)) ||
+       (!new_field->length &&
+        type != MYSQL_TYPE_STRING &&
+        type != MYSQL_TYPE_VARCHAR && type != FIELD_TYPE_GEOMETRY)))
   {
     my_error(ER_TOO_BIG_FIELDLENGTH, MYF(0),
-                    field_name, MAX_FIELD_CHARLENGTH);/* purecov: inspected */
+             field_name, max_field_charlength); /* purecov: inspected */
     DBUG_RETURN(1);				/* purecov: inspected */
   }
   type_modifier&= AUTO_INCREMENT_FLAG;
@@ -5088,11 +5084,10 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
     DBUG_RETURN(1);
   }
   if (!new_field->pack_length)
-    new_field->pack_length=calc_pack_length(new_field->sql_type ==
-					    FIELD_TYPE_VAR_STRING ?
-					    FIELD_TYPE_STRING :
-					    new_field->sql_type,
-					    new_field->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);
@@ -5136,7 +5131,6 @@ static void remove_escape(char *name)
   {
 #ifdef USE_MB
     int l;
-/*    if ((l = ismbchar(name, name+MBMAXLEN))) { Wei He: I think it's wrong */
     if (use_mb(system_charset_info) &&
         (l = my_ismbchar(system_charset_info, name, strend)))
     {
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 1049c074ce1..e8f5c6585df 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -7539,7 +7539,8 @@ static Field* create_tmp_field_from_field(THD *thd, Field* org_field,
   Field *new_field;
 
   if (convert_blob_length && org_field->flags & BLOB_FLAG)
-    new_field= new Field_varstring(convert_blob_length, org_field->maybe_null(),
+    new_field= new Field_varstring(convert_blob_length,
+                                   org_field->maybe_null(),
                                    org_field->field_name, table,
                                    org_field->charset());
   else
@@ -7552,7 +7553,8 @@ static Field* create_tmp_field_from_field(THD *thd, Field* org_field,
       new_field->field_name= item->name;
     if (org_field->maybe_null())
       new_field->flags&= ~NOT_NULL_FLAG;	// Because of outer join
-    if (org_field->type() == FIELD_TYPE_VAR_STRING)
+    if (org_field->type() == MYSQL_TYPE_VAR_STRING ||
+        org_field->type() == MYSQL_TYPE_VARCHAR)
       table->db_create_options|= HA_OPTION_PACK_RECORD;
   }
   return new_field;
@@ -7759,6 +7761,11 @@ Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
   for send_fields
 */
 
+#define STRING_TOTAL_LENGTH_TO_PACK_ROWS 128
+#define AVG_STRING_LENGTH_TO_PACK_ROWS   64
+#define RATIO_TO_PACK_ROWS	       2
+#define MIN_STRING_LENGTH_TO_PACK_ROWS   10
+
 TABLE *
 create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
 		 ORDER *group, bool distinct, bool save_sum_fields,
@@ -7766,10 +7773,13 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
 		 char *table_alias)
 {
   TABLE *table;
-  uint	i,field_count,reclength,null_count,null_pack_length,
-        hidden_null_count, hidden_null_pack_length, hidden_field_count,
-	blob_count,group_null_items;
-  bool	using_unique_constraint=0;
+  uint	i,field_count,null_count,null_pack_length;
+  uint  hidden_null_count, hidden_null_pack_length, hidden_field_count;
+  uint  blob_count,group_null_items, string_count;
+  uint  temp_pool_slot=MY_BIT_NONE;
+  ulong reclength, string_total_length;
+  bool  using_unique_constraint= 0;
+  bool  use_packed_rows= 0;
   bool  not_all_columns= !(select_options & TMP_TABLE_ALL_COLUMNS);
   char	*tmpname,path[FN_REFLEN], filename[FN_REFLEN];
   byte	*pos,*group_buff;
@@ -7780,8 +7790,6 @@ 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 temp_pool_slot=MY_BIT_NONE;
-
   DBUG_ENTER("create_tmp_table");
   DBUG_PRINT("enter",("distinct: %d  save_sum_fields: %d  rows_limit: %lu  group: %d",
 		      (int) distinct, (int) save_sum_fields,
@@ -7876,7 +7884,8 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
 
   /* Calculate which type of fields we will store in the temporary table */
 
-  reclength=blob_count=null_count=hidden_null_count=group_null_items=0;
+  reclength= string_total_length= 0;
+  blob_count= string_count= null_count= hidden_null_count= group_null_items= 0;
   param->using_indirect_summary_function=0;
 
   List_iterator_fast<Item> li(fields);
@@ -7923,6 +7932,12 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
 	    *blob_field++= new_field;
 	    blob_count++;
 	  }
+          if (new_field->real_type() == MYSQL_TYPE_STRING ||
+              new_field->real_type() == MYSQL_TYPE_VARCHAR)
+          {
+            string_count++;
+            string_total_length+= new_field->pack_length();
+          }
           thd->change_item_tree(argp, new Item_field(new_field));
 	  if (!(new_field->flags & NOT_NULL_FLAG))
           {
@@ -8016,6 +8031,12 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
   reclength+=null_pack_length;
   if (!reclength)
     reclength=1;				// Dummy select
+  /* Use packed rows if there is blobs or a lot of space to gain */
+  if (blob_count ||
+      string_total_length >= STRING_TOTAL_LENGTH_TO_PACK_ROWS &&
+      (reclength / string_total_length <= RATIO_TO_PACK_ROWS ||
+       string_total_length / string_count >= AVG_STRING_LENGTH_TO_PACK_ROWS))
+    use_packed_rows= 1;
 
   table->fields=field_count;
   table->reclength=reclength;
@@ -8090,10 +8111,9 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
     recinfo->length=length;
     if (field->flags & BLOB_FLAG)
       recinfo->type= (int) FIELD_BLOB;
-    else if (!field->zero_pack() &&
-	     (field->type() == FIELD_TYPE_STRING ||
-	      field->type() == FIELD_TYPE_VAR_STRING) &&
-	     length >= 10 && blob_count)
+    else if (use_packed_rows &&
+             field->real_type() == MYSQL_TYPE_STRING &&
+	     length >= MIN_STRING_LENGTH_TO_PACK_ROWS)
       recinfo->type=FIELD_SKIP_ENDSPACE;
     else
       recinfo->type=FIELD_NORMAL;
@@ -8168,6 +8188,8 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
 	}
 	else
 	  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;
       }
       keyinfo->key_length+=  key_part_info->length;
@@ -8338,12 +8360,10 @@ static bool create_myisam_tmp_table(TABLE *table,TMP_TABLE_PARAM *param,
       }
       else
       {
-	seg->type=
-	  ((keyinfo->key_part[i].key_type & FIELDFLAG_BINARY) ?
-	   HA_KEYTYPE_BINARY : HA_KEYTYPE_TEXT);
-	if (!(field->flags & ZEROFILL_FLAG) &&
-	    (field->type() == FIELD_TYPE_STRING ||
-	     field->type() == FIELD_TYPE_VAR_STRING) &&
+	seg->type= ((keyinfo->key_part[i].key_type & FIELDFLAG_BINARY) ?
+                    HA_KEYTYPE_BINARY : HA_KEYTYPE_TEXT);
+        /* 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;
       }
@@ -9846,6 +9866,7 @@ end_update(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
   DBUG_RETURN(0);
 }
 
+
 /* Like end_update, but this is done with unique constraints instead of keys */
 
 static int
@@ -9993,11 +10014,11 @@ static bool test_if_ref(Item_field *left_item,Item *right_item)
 	/*
 	  We can remove binary fields and numerical fields except float,
 	  as float comparison isn't 100 % secure
-	  We have to keep binary strings to be able to check for end spaces
+	  We have to keep normal strings to be able to check for end spaces
 	*/
 	if (field->binary() &&
-	    field->real_type() != FIELD_TYPE_STRING &&
-	    field->real_type() != FIELD_TYPE_VAR_STRING &&
+	    field->real_type() != MYSQL_TYPE_STRING &&
+	    field->real_type() != MYSQL_TYPE_VARCHAR &&
 	    (field->type() != FIELD_TYPE_FLOAT || field->decimals() == 0))
 	{
 	  return !store_val_in_field(field,right_item);
diff --git a/sql/sql_string.cc b/sql/sql_string.cc
index 69b530911f4..226a80201a1 100644
--- a/sql/sql_string.cc
+++ b/sql/sql_string.cc
@@ -720,8 +720,8 @@ void String::qs_append(uint i)
 int sortcmp(const String *s,const String *t, CHARSET_INFO *cs)
 {
  return cs->coll->strnncollsp(cs,
-                             (unsigned char *) s->ptr(),s->length(),
-			     (unsigned char *) t->ptr(),t->length());
+                              (unsigned char *) s->ptr(),s->length(),
+                              (unsigned char *) t->ptr(),t->length(), 0);
 }
 
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index c0ddfdb2f07..29a92e901c8 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -433,6 +433,9 @@ void calculate_interval_lengths(CHARSET_INFO *cs, TYPELIB *interval,
   DESCRIPTION
     Prepares the table and key structures for table creation.
 
+  NOTES
+    sets create_info->varchar if the table has a varchar or blob.
+
   RETURN VALUES
     0	ok
     -1	error
@@ -447,20 +450,24 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
   const char	*key_name;
   create_field	*sql_field,*dup_field;
   uint		field,null_fields,blob_columns;
+  uint		max_key_length= file->max_key_length();
   ulong		pos;
   KEY		*key_info;
   KEY_PART_INFO *key_part_info;
   int		timestamps= 0, timestamps_with_niladic= 0;
   int		field_no,dup_no;
   int		select_field_pos,auto_increment=0;
+  List_iterator<create_field> it(fields),it2(fields);
   DBUG_ENTER("mysql_prepare_table");
 
-  List_iterator<create_field> it(fields),it2(fields);
   select_field_pos=fields.elements - select_field_count;
   null_fields=blob_columns=0;
+  create_info->varchar= 0;
 
   for (field_no=0; (sql_field=it++) ; field_no++)
   {
+    CHARSET_INFO *save_cs;
+
     if (!sql_field->charset)
       sql_field->charset= create_info->default_table_charset;
     /*
@@ -472,13 +479,13 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
     if (create_info->table_charset && sql_field->charset != &my_charset_bin)
       sql_field->charset= create_info->table_charset;
 
-    CHARSET_INFO *savecs= sql_field->charset;
+    save_cs= sql_field->charset;
     if ((sql_field->flags & BINCMP_FLAG) &&
 	!(sql_field->charset= get_charset_by_csname(sql_field->charset->csname,
 						    MY_CS_BINSORT,MYF(0))))
     {
       char tmp[64];
-      strmake(strmake(tmp, savecs->csname, sizeof(tmp)-4), "_bin", 4);
+      strmake(strmake(tmp, save_cs->csname, sizeof(tmp)-4), "_bin", 4);
       my_error(ER_UNKNOWN_COLLATION, MYF(0), tmp);
       DBUG_RETURN(-1);
     }
@@ -569,10 +576,41 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
     }
 
     sql_field->create_length_to_internal_length();
+    if (sql_field->length > MAX_FIELD_VARCHARLENGTH &&
+        !(sql_field->flags & BLOB_FLAG))
+    {
+      /* Convert long VARCHAR columns to TEXT or BLOB */
+      char warn_buff[MYSQL_ERRMSG_SIZE];
+
+      if (sql_field->def)
+      {
+        my_error(ER_TOO_BIG_FIELDLENGTH, MYF(0), sql_field->field_name,
+                 MAX_FIELD_VARCHARLENGTH / sql_field->charset->mbmaxlen);
+        DBUG_RETURN(-1);
+      }
+      sql_field->sql_type= FIELD_TYPE_BLOB;
+      sql_field->flags|= BLOB_FLAG;
+      sprintf(warn_buff, ER(ER_AUTO_CONVERT), sql_field->field_name,
+              "VARCHAR",
+              (sql_field->charset == &my_charset_bin) ? "BLOB" : "TEXT");
+      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_AUTO_CONVERT,
+                   warn_buff);
+    }
+    
+    if ((sql_field->flags & BLOB_FLAG) && sql_field->length)
+    {
+      if (sql_field->sql_type == FIELD_TYPE_BLOB)
+      {
+        /* The user has given a length to the blob column */
+        sql_field->sql_type= get_blob_type_from_length(sql_field->length);
+        sql_field->pack_length= calc_pack_length(sql_field->sql_type, 0);
+      }
+      sql_field->length= 0;                     // Probably from an item
+    }
 
-    /* Don't pack keys in old tables if the user has requested this */
+    /* Don't pack rows in old tables if the user has requested this */
     if ((sql_field->flags & BLOB_FLAG) ||
-	sql_field->sql_type == FIELD_TYPE_VAR_STRING &&
+	sql_field->sql_type == MYSQL_TYPE_VARCHAR &&
 	create_info->row_type != ROW_TYPE_FIXED)
     {
       db_options|=HA_OPTION_PACK_RECORD;
@@ -646,6 +684,7 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
       sql_field->length=8;			// Unireg field length
       sql_field->unireg_check=Field::BLOB_FIELD;
       blob_columns++;
+      create_info->varchar= 1;
       break;
     case FIELD_TYPE_GEOMETRY:
 #ifdef HAVE_SPATIAL
@@ -662,17 +701,37 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
       sql_field->length=8;			// Unireg field length
       sql_field->unireg_check=Field::BLOB_FIELD;
       blob_columns++;
+      create_info->varchar= 1;
       break;
 #else
       my_error(ER_FEATURE_DISABLED, MYF(0),
                sym_group_geom.name, sym_group_geom.needed_define);
       DBUG_RETURN(-1);
 #endif /*HAVE_SPATIAL*/
-    case FIELD_TYPE_VAR_STRING:
-    case FIELD_TYPE_STRING:
+    case MYSQL_TYPE_VARCHAR:
+#ifndef QQ_ALL_HANDLERS_SUPPORT_VARCHAR
+      if (file->table_flags() & HA_NO_VARCHAR)
+      {
+        /* convert VARCHAR to CHAR because handler is not yet up to date */
+        sql_field->sql_type=    MYSQL_TYPE_VAR_STRING;
+        sql_field->pack_length= calc_pack_length(sql_field->sql_type,
+                                                 (uint) sql_field->length);
+        if ((sql_field->length / sql_field->charset->mbmaxlen) >
+            MAX_FIELD_CHARLENGTH)
+        {
+          my_printf_error(ER_TOO_BIG_FIELDLENGTH, ER(ER_TOO_BIG_FIELDLENGTH),
+                          MYF(0), sql_field->field_name, MAX_FIELD_CHARLENGTH);
+          DBUG_RETURN(-1);
+        }
+      }
+      else
+#endif
+        create_info->varchar= 1;
+      /* fall through */
+    case MYSQL_TYPE_STRING:
       sql_field->pack_flag=0;
       if (sql_field->charset->state & MY_CS_BINSORT)
-	sql_field->pack_flag|=FIELDFLAG_BINARY;
+	sql_field->pack_flag|= FIELDFLAG_BINARY;
       break;
     case FIELD_TYPE_ENUM:
       sql_field->pack_flag=pack_length_to_packflag(sql_field->pack_length) |
@@ -953,6 +1012,7 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
     CHARSET_INFO *ft_key_charset=0;  // for FULLTEXT
     for (uint column_nr=0 ; (column=cols++) ; column_nr++)
     {
+      uint length;
       key_part_spec *dup_column;
 
       it.rewind();
@@ -981,8 +1041,8 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
       cols2.rewind();
       if (key->type == Key::FULLTEXT)
       {
-	if ((sql_field->sql_type != FIELD_TYPE_STRING &&
-	     sql_field->sql_type != FIELD_TYPE_VAR_STRING &&
+	if ((sql_field->sql_type != MYSQL_TYPE_STRING &&
+	     sql_field->sql_type != MYSQL_TYPE_VARCHAR &&
 	     !f_is_blob(sql_field->pack_flag)) ||
 	    sql_field->charset == &my_charset_bin ||
 	    sql_field->charset->mbminlen > 1 || // ucs2 doesn't work yet
@@ -1019,15 +1079,15 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
 	  }
 	}
 #ifdef HAVE_SPATIAL
-	if (key->type  == Key::SPATIAL)
+	if (key->type == Key::SPATIAL)
 	{
-	  if (!column->length )
+	  if (!column->length)
 	  {
 	    /*
-	    BAR: 4 is: (Xmin,Xmax,Ymin,Ymax), this is for 2D case
-		 Lately we'll extend this code to support more dimensions
+              4 is: (Xmin,Xmax,Ymin,Ymax), this is for 2D case
+              Lately we'll extend this code to support more dimensions
 	    */
-	    column->length=4*sizeof(double);
+	    column->length= 4*sizeof(double);
 	  }
 	}
 #endif
@@ -1063,7 +1123,8 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
       key_part_info->fieldnr= field;
       key_part_info->offset=  (uint16) sql_field->offset;
       key_part_info->key_type=sql_field->pack_flag;
-      uint length=sql_field->pack_length;
+      length= sql_field->key_length;
+
       if (column->length)
       {
 	if (f_is_blob(sql_field->pack_flag))
@@ -1128,12 +1189,13 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
       /* Use packed keys for long strings on the first column */
       if (!(db_options & HA_OPTION_NO_PACK_KEYS) &&
 	  (length >= KEY_DEFAULT_PACK_LENGTH &&
-	   (sql_field->sql_type == FIELD_TYPE_STRING ||
-	    sql_field->sql_type == FIELD_TYPE_VAR_STRING ||
+	   (sql_field->sql_type == MYSQL_TYPE_STRING ||
+	    sql_field->sql_type == MYSQL_TYPE_VARCHAR ||
 	    sql_field->pack_flag & FIELDFLAG_BLOB)))
       {
-	if (column_nr == 0 && (sql_field->pack_flag & FIELDFLAG_BLOB))
-	  key_info->flags|= HA_BINARY_PACK_KEY;
+	if (column_nr == 0 && (sql_field->pack_flag & FIELDFLAG_BLOB) ||
+            sql_field->sql_type == MYSQL_TYPE_VARCHAR)
+	  key_info->flags|= HA_BINARY_PACK_KEY | HA_VAR_LENGTH_KEY;
 	else
 	  key_info->flags|= HA_PACK_KEY;
       }
@@ -1173,7 +1235,6 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
     if (!(key_info->flags & HA_NULL_PART_KEY))
       unique_key=1;
     key_info->key_length=(uint16) key_length;
-    uint max_key_length= file->max_key_length();
     if (key_length > max_key_length && key->type != Key::FULLTEXT)
     {
       my_error(ER_TOO_LONG_KEY,MYF(0),max_key_length);
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 8ea7cdb34fa..8b3cdda5d9e 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -2773,13 +2773,13 @@ type:
 					  Lex->charset=&my_charset_bin;
 					  $$=FIELD_TYPE_STRING; }
 	| varchar '(' NUM ')' opt_binary { Lex->length=$3.str;
-					  $$=FIELD_TYPE_VAR_STRING; }
+					  $$= MYSQL_TYPE_VARCHAR; }
 	| nvarchar '(' NUM ')'		{ Lex->length=$3.str;
-					  $$=FIELD_TYPE_VAR_STRING;
+					  $$= MYSQL_TYPE_VARCHAR;
 					  Lex->charset=national_charset_info; }
 	| VARBINARY '(' NUM ')' 	{ Lex->length=$3.str;
 					  Lex->charset=&my_charset_bin;
-					  $$=FIELD_TYPE_VAR_STRING; }
+					  $$= MYSQL_TYPE_VARCHAR; }
 	| YEAR_SYM opt_len field_options { $$=FIELD_TYPE_YEAR; }
 	| DATE_SYM			{ $$=FIELD_TYPE_DATE; }
 	| TIME_SYM			{ $$=FIELD_TYPE_TIME; }
diff --git a/sql/strfunc.cc b/sql/strfunc.cc
index b5255e9be06..2253f48e558 100644
--- a/sql/strfunc.cc
+++ b/sql/strfunc.cc
@@ -136,7 +136,7 @@ uint find_type2(TYPELIB *typelib, const char *x, uint length, CHARSET_INFO *cs)
   int find,pos,findpos;
   const char *j;
   DBUG_ENTER("find_type2");
-  DBUG_PRINT("enter",("x: '%s'  lib: 0x%lx",x,typelib));
+  DBUG_PRINT("enter",("x: '%.*s'  lib: 0x%lx", length, x, typelib));
 
   if (!typelib->count)
   {
diff --git a/sql/structs.h b/sql/structs.h
index c02f8c502a2..5d0c7bc4f1f 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -98,6 +98,7 @@ typedef struct st_key {
   union {
     int  bdb_return_if_eq;
   } handler;
+  struct st_table *table;
 } KEY;
 
 
diff --git a/sql/table.cc b/sql/table.cc
index 63575f30326..b4a07448b14 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -139,7 +139,8 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   *fn_ext(outparam->path)='\0';		// Remove extension
 
   if (head[0] != (uchar) 254 || head[1] != 1 ||
-      (head[2] != FRM_VER && head[2] != FRM_VER+1 && head[2] != FRM_VER+3))
+      (head[2] != FRM_VER && head[2] != FRM_VER+1 &&
+       ! (head[2] >= FRM_VER+3 && head[2] <= FRM_VER+4)))
     goto err_not_open;				/* purecov: inspected */
   new_field_pack_flag=head[27];
   new_frm_ver= (head[2] - FRM_VER);
@@ -215,7 +216,8 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 
   for (i=0 ; i < keys ; i++, keyinfo++)
   {
-    if (new_frm_ver == 3)
+    keyinfo->table= outparam;
+    if (new_frm_ver >= 3)
     {
       keyinfo->flags=	   (uint) uint2korr(strpos) ^ HA_NOSAME;
       keyinfo->key_length= (uint) uint2korr(strpos+2);
@@ -433,7 +435,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
     Field::geometry_type geom_type= Field::GEOM_GEOMETRY;
     LEX_STRING comment;
 
-    if (new_frm_ver == 3)
+    if (new_frm_ver >= 3)
     {
       /* new frm file in 4.1 */
       field_length= uint2korr(strpos+3);
@@ -441,11 +443,10 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
       pack_flag=    uint2korr(strpos+8);
       unireg_type=  (uint) strpos[10];
       interval_nr=  (uint) strpos[12];
-
       uint comment_length=uint2korr(strpos+15);
       field_type=(enum_field_types) (uint) strpos[13];
 
-      // charset and geometry_type share the same byte in frm
+      /* charset and geometry_type share the same byte in frm */
       if (field_type == FIELD_TYPE_GEOMETRY)
       {
 #ifdef HAVE_SPATIAL
@@ -609,10 +610,12 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 	    keyinfo->key_length+= HA_KEY_NULL_LENGTH;
 	  }
 	  if (field->type() == FIELD_TYPE_BLOB ||
-	      field->real_type() == FIELD_TYPE_VAR_STRING)
+	      field->real_type() == MYSQL_TYPE_VARCHAR)
 	  {
 	    if (field->type() == FIELD_TYPE_BLOB)
 	      key_part->key_part_flag|= HA_BLOB_PART;
+            else
+              key_part->key_part_flag|= HA_VAR_LENGTH_PART;
 	    keyinfo->extra_length+=HA_KEY_BLOB_LENGTH;
 	    key_part->store_length+=HA_KEY_BLOB_LENGTH;
 	    keyinfo->key_length+= HA_KEY_BLOB_LENGTH;
@@ -1232,7 +1235,11 @@ File create_frm(register my_string name, uint reclength, uchar *fileinfo,
   if ((file=my_create(name,CREATE_MODE,O_RDWR | O_TRUNC,MYF(MY_WME))) >= 0)
   {
     bzero((char*) fileinfo,64);
-    fileinfo[0]=(uchar) 254; fileinfo[1]= 1; fileinfo[2]= FRM_VER+3; // Header
+    /* header */
+    fileinfo[0]=(uchar) 254;
+    fileinfo[1]= 1;
+    fileinfo[2]= FRM_VER+3+ test(create_info->varchar);
+
     fileinfo[3]= (uchar) ha_checktype(create_info->db_type);
     fileinfo[4]=1;
     int2store(fileinfo+6,IO_SIZE);		/* Next block starts here */
@@ -1385,7 +1392,7 @@ bool check_db_name(char *name)
     if (use_mb(system_charset_info))
     {
       int len=my_ismbchar(system_charset_info, name, 
-		name+system_charset_info->mbmaxlen);
+                          name+system_charset_info->mbmaxlen);
       if (len)
       {
         name += len;
@@ -1462,7 +1469,7 @@ bool check_column_name(const char *name)
     if (use_mb(system_charset_info))
     {
       int len=my_ismbchar(system_charset_info, name, 
-		name+system_charset_info->mbmaxlen);
+                          name+system_charset_info->mbmaxlen);
       if (len)
       {
         name += len;
@@ -1497,7 +1504,8 @@ db_type get_table_type(const char *name)
   error=my_read(file,(byte*) head,4,MYF(MY_NABP));
   my_close(file,MYF(0));
   if (error || head[0] != (uchar) 254 || head[1] != 1 ||
-      (head[2] != FRM_VER && head[2] != FRM_VER+1 && head[2] != FRM_VER+3))
+      (head[2] != FRM_VER && head[2] != FRM_VER+1 &&
+       (head[2] < FRM_VER+3 || head[2] > FRM_VER+4)))
     DBUG_RETURN(DB_TYPE_UNKNOWN);
   DBUG_RETURN(ha_checktype((enum db_type) (uint) *(head+3)));
 }
diff --git a/sql/table.h b/sql/table.h
index ed9c1445cdf..f5f2a76c6f1 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -160,7 +160,8 @@ struct st_table {
   my_bool no_keyread, no_cache;
   my_bool clear_query_id;               /* To reset query_id for tables and cols */
   my_bool auto_increment_field_not_null;
-  my_bool alias_name_used; /* true if table_name is alias */
+  my_bool insert_or_update;             /* Can be used by the handler */
+  my_bool alias_name_used;		/* true if table_name is alias */
   Field *next_number_field,		/* Set if next_number is activated */
 	*found_next_number_field,	/* Set on open */
         *rowid_field;
diff --git a/sql/unireg.h b/sql/unireg.h
index 2879e30d861..31b28da2423 100644
--- a/sql/unireg.h
+++ b/sql/unireg.h
@@ -60,6 +60,9 @@
 
 #define MAX_MBWIDTH		3		/* Max multibyte sequence */
 #define MAX_FIELD_CHARLENGTH	255
+#define MAX_FIELD_VARCHARLENGTH	65535
+#define CONVERT_IF_BIGGER_TO_BLOB 512		/* Used for CREATE ... SELECT */
+
 /* Max column width +1 */
 #define MAX_FIELD_WIDTH		(MAX_FIELD_CHARLENGTH*MAX_MBWIDTH+1)
 
@@ -145,6 +148,7 @@
 #define TE_INFO_LENGTH 3
 #define MTYP_NOEMPTY_BIT 128
 
+#define FRM_VER_TRUE_VARCHAR (FRM_VER+4)
 /*
   Minimum length pattern before Turbo Boyer-Moore is used
   for SELECT "text" LIKE "%pattern%", excluding the two