Add support for up to VARCHAR (size up to 65535)

Renamed HA_VAR_LENGTH to HA_VAR_LENGTH_PART
Renamed in all files FIELD_TYPE_STRING and FIELD_TYPE_VAR_STRING to MYSQL_TYPE_STRING and MYSQL_TYPE_VAR_STRING to make it easy to catch all possible errors
Added support for VARCHAR KEYS to heap
Removed support for ISAM
Now only long VARCHAR columns are changed to TEXT on demand (not CHAR)
Internal temporary files can now use fixed length tables if the used VARCHAR columns are short


BitKeeper/deleted/.del-ha_isam.cc~4dce65904db2675e:
  Delete: sql/ha_isam.cc
BitKeeper/deleted/.del-_cache.c~b5d80b5c3ae233b1:
  Delete: isam/_cache.c
BitKeeper/deleted/.del-_dbug.c~88d7964ae5e3c9bd:
  Delete: isam/_dbug.c
BitKeeper/deleted/.del-_dynrec.c~48dd758f5a5450df:
  Delete: isam/_dynrec.c
BitKeeper/deleted/.del-_key.c~ce62d47a6c681084:
  Delete: isam/_key.c
BitKeeper/deleted/.del-_locking.c~dea4cdc6ea425c67:
  Delete: isam/_locking.c
BitKeeper/deleted/.del-_packrec.c~47ae1b16c007e9be:
  Delete: isam/_packrec.c
BitKeeper/deleted/.del-_page.c~148b1a613d052ee8:
  Delete: isam/_page.c
BitKeeper/deleted/.del-_search.c~f509292aa1ff18ff:
  Delete: isam/_search.c
BitKeeper/deleted/.del-_statrec.c~58d9263b3475d58b:
  Delete: isam/_statrec.c
BitKeeper/deleted/.del-changed.c~d075de80a314b02d:
  Delete: isam/changed.c
BitKeeper/deleted/.del-close.c~fd62629496ee5bcc:
  Delete: isam/close.c
BitKeeper/deleted/.del-create.c~96cecc433c0c2242:
  Delete: isam/create.c
BitKeeper/deleted/.del-delete.c~65ee8daaa75a14b6:
  Delete: isam/delete.c
BitKeeper/deleted/.del-extra.c~706f29d72beb2565:
  Delete: isam/extra.c
BitKeeper/deleted/.del-info.c~96cfb747af8da0d:
  Delete: isam/info.c
BitKeeper/deleted/.del-isamchk.c~c0f59c2687d2248f:
  Delete: isam/isamchk.c
BitKeeper/deleted/.del-isamlog.c~85b6b31c6e2b8519:
  Delete: isam/isamlog.c
BitKeeper/deleted/.del-log.c~55a973013d55cade:
  Delete: isam/log.c
BitKeeper/deleted/.del-open.c~95b3b75042fae00a:
  Delete: isam/open.c
BitKeeper/deleted/.del-pack_isam.c~43801f0df7504834:
  Delete: isam/pack_isam.c
BitKeeper/deleted/.del-panic.c~f7fd71605324f8f3:
  Delete: isam/panic.c
BitKeeper/deleted/.del-range.c~142f1f8ac4948082:
  Delete: isam/range.c
BitKeeper/deleted/.del-rfirst.c~66f494291dc005d3:
  Delete: isam/rfirst.c
BitKeeper/deleted/.del-rkey.c~cc54c6498352f999:
  Delete: isam/rkey.c
BitKeeper/deleted/.del-rlast.c~d1fe1866139e9866:
  Delete: isam/rlast.c
BitKeeper/deleted/.del-rnext.c~b308eaa1e11ea7de:
  Delete: isam/rnext.c
BitKeeper/deleted/.del-rprev.c~b359f71fdea4bbce:
  Delete: isam/rprev.c
BitKeeper/deleted/.del-rrnd.c~7fcfcce88d4a5200:
  Delete: isam/rrnd.c
BitKeeper/deleted/.del-rsame.c~75a62d5548103a15:
  Delete: isam/rsame.c
BitKeeper/deleted/.del-rsamepos.c~5b5652dd2cda6d5d:
  Delete: isam/rsamepos.c
BitKeeper/deleted/.del-sort.c~e2e56b5a37ce86f4:
  Delete: isam/sort.c
BitKeeper/deleted/.del-static.c~3a1354b84f4a9cc7:
  Delete: isam/static.c
BitKeeper/deleted/.del-test1.c~64d52e9412d457ed:
  Delete: isam/test1.c
BitKeeper/deleted/.del-test2.c~2f9a632cab572958:
  Delete: isam/test2.c
BitKeeper/deleted/.del-test3.c~e8a7a4afe8a087:
  Delete: isam/test3.c
BitKeeper/deleted/.del-isamdef.h~ac8d49e7e2201c66:
  Delete: isam/isamdef.h
BitKeeper/deleted/.del-update.c~670264f51dc44934:
  Delete: isam/update.c
BitKeeper/deleted/.del-write.c~8f1918b1f6770e54:
  Delete: isam/write.c
BitKeeper/deleted/.del-Makefile.am~6cfa0db5e7778d09:
  Delete: isam/Makefile.am
BitKeeper/deleted/.del-make-ccc~3ee55391eda0b0ab:
  Delete: isam/make-ccc
BitKeeper/deleted/.del-ChangeLog~208984fb7a51e568:
  Delete: isam/ChangeLog
BitKeeper/deleted/.del-test_all.res~c2aafb49a3a77db7:
  Delete: isam/test_all.res
BitKeeper/deleted/.del-test_all~93c701e44a9c5b65:
  Delete: isam/test_all
BitKeeper/deleted/.del-.cvsignore~54f6f0f2d5012561:
  Delete: isam/.cvsignore
BitKeeper/deleted/.del-ha_isammrg.cc~dc682e4755d77a2e:
  Delete: sql/ha_isammrg.cc
BitKeeper/deleted/.del-ha_isam.h~bf53d533be3d3927:
  Delete: sql/ha_isam.h
BitKeeper/deleted/.del-ha_isammrg.h~66fd2e5bfe7207dc:
  Delete: sql/ha_isammrg.h
acinclude.m4:
  Remove ISAM
client/mysqldump.c:
  FIELD_TYPE -> MYSQL_TYPE
client/mysqltest.c:
  Add missing DBUG_RETURN
configure.in:
  Remove ISAM
heap/heapdef.h:
  Add support for VARCHAR
heap/hp_create.c:
  Add support for VARCHAR
heap/hp_delete.c:
  Add support for VARCHAR
heap/hp_hash.c:
  Add support for VARCHAR
  (VARCHAR keys was not supported before)
heap/hp_rkey.c:
  Add support for VARCHAR
heap/hp_update.c:
  Add support for VARCHAR
heap/hp_write.c:
  Add support for VARCHAR
  (Added flag SEARCH_UPDATE to mark that this is an update)
include/decimal.h:
  Remove not needed my_global.h
include/m_ctype.h:
  Add support for VARCHAR
include/my_base.h:
  Add support for VARCHAR
include/my_handler.h:
  Moved general purpose macro from MyISAM code
include/mysql_com.h:
  Add support for VARCHAR
libmysql/libmysql.c:
  Add support for VARCHAR
libmysqld/Makefile.am:
  Removed ISAM
myisam/ft_static.c:
  Add support for VARCHAR
myisam/ft_test1.c:
  Add support for VARCHAR
myisam/ft_update.c:
  Add support for VARCHAR
myisam/mi_check.c:
  Add support for VARCHAR
myisam/mi_create.c:
  Add support for VARCHAR
  - VARCHAR key segments are marked with HA_VAR_LENGTH_PART
myisam/mi_key.c:
  Add support for VARCHAR
  Fixed bug in old VARCHAR code when reading index-only
myisam/mi_range.c:
  Fixed comment style
myisam/mi_rnext_same.c:
  Handle case where equal keys can be of different length
myisam/mi_search.c:
  Add support for VARCHAR
myisam/mi_test1.c:
  Add support for VARCHAR
myisam/mi_unique.c:
  Add support for VARCHAR
  (Some new code to handle keys that are equal but have different lengths)
myisam/mi_write.c:
  Fixed comment
myisam/myisamchk.c:
  Better infotext if wrong type
mysql-test/r/bdb.result:
  Updated old result and new results for VARCHAR
mysql-test/r/create.result:
  Updated old result and new results for VARCHAR
mysql-test/r/ctype_tis620.result:
  Updated old result and new results for VARCHAR
  (Old code sorted tis620 wrong)
mysql-test/r/ctype_ucs.result:
  Updated old result and new results for VARCHAR
mysql-test/r/endspace.result:
  Updated old result and new results for VARCHAR
mysql-test/r/func_like.result:
  Updated old result and new results for VARCHAR
mysql-test/r/heap.result:
  Updated old result and new results for VARCHAR
mysql-test/r/innodb.result:
  Updated old result. This will change a bit when also InnoDB supports VARCHAR
mysql-test/r/merge.result:
  Updated old result and new results for VARCHAR
mysql-test/r/myisam.result:
  Updated old result and new results for VARCHAR
mysql-test/r/mysqldump.result:
  Updated old result and new results for VARCHAR
mysql-test/r/order_by.result:
  Updated old result and new results for VARCHAR
  (Key length is different for VARCHAR)
mysql-test/r/ps.result:
  Updated old result and new results for VARCHAR
mysql-test/r/ps_1general.result:
  Updated results for new .frm version
  Don't print seconds in show full process list as this may change
mysql-test/r/ps_2myisam.result:
  Updated old result and new results for VARCHAR
mysql-test/r/ps_3innodb.result:
  Updated old result and new results for VARCHAR
mysql-test/r/ps_4heap.result:
  Updated old result and new results for VARCHAR
mysql-test/r/ps_5merge.result:
  Updated old result and new results for VARCHAR
mysql-test/r/ps_6bdb.result:
  Updated old result and new results for VARCHAR
mysql-test/r/select.result.es:
  Updated results by hand
mysql-test/r/select.result:
  Updated old result and new results for VARCHAR
mysql-test/r/select_found.result:
  Updated old result and new results for VARCHAR
mysql-test/r/show_check.result:
  Updated old result and new results for VARCHAR
mysql-test/r/strict.result:
  Updated old result and new results for VARCHAR
mysql-test/r/subselect.result:
  Updated old result and new results for VARCHAR
mysql-test/r/system_mysql_db.result:
  Updated old result and new results for VARCHAR
mysql-test/r/type_blob.result:
  Updated old result and new results for VARCHAR
mysql-test/r/type_ranges.result:
  Updated old result and new results for VARCHAR
mysql-test/r/type_ranges.result.es:
  Updated some results by hand
mysql-test/t/bdb.test:
  Test VARCHAR
mysql-test/t/ctype_ucs.test:
  Some fixes related to VARCHAR
mysql-test/t/endspace.test:
  Fixes to make it easier to compare columns with end space
mysql-test/t/heap.test:
  Test VARCHAR
mysql-test/t/innodb.test:
  Prepare for testing VARCHAR
mysql-test/t/myisam.test:
  Test VARCHAR
mysql-test/t/ps_1general.test:
  Don't show seconds for show processlist
mysql-test/t/ps_4heap.test:
  Update for VARCHAR
mysql-test/t/strict.test:
  Fix test for VARCHAR
mysql-test/t/type_blob.test:
  Update test for VARCHAR
  Note that now you can't store 'a' and 'a ' in an unique varchar/text index if the column is not binary
mysys/my_handler.c:
  Add support for VARCHAR
ndb/src/common/util/NdbSqlUtil.cpp:
  Fix for usage of strnncollsp
scripts/mysql_fix_privilege_tables.sh:
  Simple fix so that my_print_defaults works
sql/Makefile.am:
  Remove ISAM
sql/field.cc:
  Add support for VARCHAR
  Fixed the keys for blob's are compared with strnncollsp
  Ensure that old tables from MySQL 4.0 works as they did before.
  (Old VARCHAR will be converted to new VARCHAR on ALTER TABLE)
sql/field.h:
  Add support for VARCHAR
sql/field_conv.cc:
  Change FIELD_TYPE_VAR_STRING -> MYSQL_TYPE_VARCHAR
  Added usage of HA_KEY_BLOB_LENGTH
sql/ha_berkeley.cc:
  Add support for VARCHAR
  Added usage of table->insert_or_update if we would ever want to know in key_cmp if we are changing keys
sql/ha_heap.cc:
  Add support for VARCHAR
sql/ha_innodb.cc:
  Changed MYSQL_TYPE_VAR_STRING to MYSQL_TYPE_VARCHAR.
  Waiting for Heikki to add full VARCHAR support
sql/ha_innodb.h:
  InnoDB doesn't support full VARCHAR yet
sql/ha_myisam.cc:
  Add support for VARCHAR
sql/ha_ndbcluster.cc:
  Add support for VARCHAR
sql/handler.h:
  Added HA_NO_VARCHAR for table handler that doesn't support VARCHAR. In this case MySQL will create a normal CHAR instead
sql/item.cc:
  Fixed access of already freed memory
  Added support of VARCHAR
  - varchar length is now checked in mysql_prepare
sql/item_cmpfunc.cc:
  Added new parameter to strncollsp
sql/item_sum.cc:
  Added new parameter to strncollsp
  FIELD_TYPE -> MYSQL_TYPE
sql/key.cc:
  Add support for VARCHAR
sql/opt_range.cc:
  Remove character set parameter from set_key_image()
sql/opt_sum.cc:
  Remove character set parameter from set_key_image()
sql/protocol.cc:
  Return MYSQL_TYPE_VAR_STRING instead of MYSQL_TYPE_VARCHAR to clients (to not cause compatiblity problems)
sql/sql_acl.cc:
  Change key handling code so that we can use CHAR or VARCHAR for the user table columns
sql/sql_base.cc:
  Remove old, not used code
sql/sql_help.cc:
  Remove charset from get_key_image
sql/sql_parse.cc:
  Ensure that OPTION_TABLE_LOCK is cleared ASAP; This fixed a problem in BDB transaction code when one used LOCK TABLES on a BDB table
  Added support for VARCHAR
  Moved field length checking and VARCHAR -> TEXT convert to mysql_prepare (as we need the know the character set for the column)
sql/sql_select.cc:
  Added support of VARCHAR
  Added heuristic to use fixed size rows for tmp tables if we are using only a few short VARCHAR's
sql/sql_string.cc:
  Added extra argument to strnncollsp
sql/sql_table.cc:
  Add support for VARCHAR
  Automaticly convert (with warning) big VARCHAR (but not CHAR) to TEXT
  If handler doesn't support VARCHAR convert VARCHAR to CHAR
sql/sql_update.cc:
  Fixed compiler warning
sql/sql_yacc.yy:
  Add support for VARCHAR
sql/strfunc.cc:
  Fixed valgrind warning
sql/structs.h:
  Added 'table' to KEY structure to make life easier for some handler functions
sql/table.cc:
  Add support for VARCHAR
  - New .frm version
  - FIELD_TYPE -> MYSQL_TYPE
sql/table.h:
  Added insert_or_update; A bool flag a handler can set/reset if needed (for handler internal usage)
sql/unireg.h:
  Add support for VARCHAR
strings/ctype-big5.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
  Changed my_like_range... to correctly calculate min_length & max_length
strings/ctype-bin.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
strings/ctype-czech.c:
  Changed my_like_range... to correctly calculate min_length & max_length
strings/ctype-gbk.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
  Changed my_like_range... to correctly calculate min_length & max_length
strings/ctype-latin1.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
strings/ctype-mb.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
  Changed my_like_range... to correctly calculate min_length & max_length
strings/ctype-simple.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
  Changed my_like_range... to correctly calculate min_length & max_length
strings/ctype-sjis.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
  Changed my_like_range... to correctly calculate min_length & max_length
strings/ctype-tis620.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
  Changed my_like_range... to correctly calculate min_length & max_length
strings/ctype-uca.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
strings/ctype-ucs2.c:
  Changed my_like_range... to correctly calculate min_length & max_length
strings/ctype-utf8.c:
  Added new argument to strnncollsp() to allow one to define if end space are significant or not
strings/ctype-win1250ch.c:
  Changed my_like_range... to correctly calculate min_length & max_length
strings/decimal.c:
  Fixed include files usage
  Fixed some compiler warnings
tests/client_test.c:
  Ensure tests works with VARCHAR

36 files changed, 854 insertions, 1256 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 1275e1bbb8e..b4b248a9747 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
 ****************************************************************************/
@@ -4427,7 +4446,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);
   }
   return my_strnncoll(field_charset,(const uchar*) a_ptr, field_length,
                                     (const uchar*) b_ptr, field_length);
@@ -4447,20 +4466,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);
@@ -4494,10 +4515,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);
@@ -4509,29 +4547,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);
 }
 
 
@@ -4539,16 +4599,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)
 ****************************************************************************/
@@ -4557,7 +4632,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);
 
@@ -4571,15 +4646,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;
 }
 
@@ -4626,17 +4707,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,
@@ -4656,9 +4769,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);
@@ -4673,12 +4788,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);
@@ -4691,16 +4808,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);
@@ -4708,77 +4825,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
@@ -5016,10 +5178,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);
 }
 
 
@@ -5066,8 +5228,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;
@@ -5102,8 +5263,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)
@@ -5119,10 +5281,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);
 }
 
 
@@ -5135,7 +5298,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));
 }
@@ -5227,7 +5390,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;
@@ -5241,13 +5405,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*));
@@ -5261,12 +5427,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 */
@@ -5276,8 +5441,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)
@@ -5351,14 +5516,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;
@@ -5367,8 +5533,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;
@@ -5397,11 +5562,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;
@@ -5850,33 +6010,49 @@ 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)
 {
-  switch (sql_type)
-  {
-    case MYSQL_TYPE_TINY_BLOB:
-    case MYSQL_TYPE_MEDIUM_BLOB:
-    case MYSQL_TYPE_LONG_BLOB:
-    case MYSQL_TYPE_BLOB:
-    case MYSQL_TYPE_VAR_STRING:
-    case MYSQL_TYPE_STRING:
-      length*= charset->mbmaxlen;
-      pack_length= calc_pack_length(sql_type == FIELD_TYPE_VAR_STRING ?
-				    FIELD_TYPE_STRING : sql_type, length);
-      break;
-    case MYSQL_TYPE_ENUM:
-    case MYSQL_TYPE_SET:
-      length*= charset->mbmaxlen;
-      break;
-    default:
-      /* do nothing */
-      break;
+  switch (sql_type) {
+  case MYSQL_TYPE_TINY_BLOB:
+  case MYSQL_TYPE_MEDIUM_BLOB:
+  case MYSQL_TYPE_LONG_BLOB:
+  case MYSQL_TYPE_BLOB:
+  case MYSQL_TYPE_VAR_STRING:
+  case MYSQL_TYPE_STRING:
+  case MYSQL_TYPE_VARCHAR:
+    length*= charset->mbmaxlen;
+    key_length*= charset->mbmaxlen;
+    pack_length= calc_pack_length(sql_type, length);
+    break;
+  case MYSQL_TYPE_ENUM:
+  case MYSQL_TYPE_SET:
+    length*= charset->mbmaxlen;
+    break;
+  default:
+    /* do nothing */
+    break;
   }
 }
 
+
+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
 */
@@ -5884,9 +6060,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;
@@ -5958,9 +6135,19 @@ Field *make_field(char *ptr, uint32 field_length,
   if (f_is_alpha(pack_flag))
   {
     if (!f_is_packed(pack_flag))
-      return new Field_string(ptr,field_length,null_pos,null_bit,
-			      unireg_check, field_name, table, field_charset);
-
+    {
+      if (field_type == MYSQL_TYPE_STRING ||
+          field_type == FIELD_TYPE_DECIMAL ||   // 3.23 or 4.0 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;
+    }
     uint pack_length=calc_pack_length((enum_field_types)
 				      f_packtype(pack_flag),
 				      field_length);
@@ -6070,42 +6257,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 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
@@ -6130,12 +6326,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 50ea1450085..662aa3074dd 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -188,27 +188,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;
@@ -224,11 +204,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;
@@ -267,9 +246,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);
@@ -912,8 +893,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; }
@@ -930,8 +911,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); }
@@ -939,6 +921,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);
 };
 
 
@@ -958,7 +941,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; }
@@ -973,18 +956,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; }
@@ -1053,8 +1041,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;
@@ -1072,8 +1060,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(); }
@@ -1086,6 +1075,7 @@ public:
   uint32 max_length();
 };
 
+
 #ifdef HAVE_SPATIAL
 class Field_geom :public Field_blob {
 public:
@@ -1110,12 +1100,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;
@@ -1196,8 +1186,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
   CHARSET_INFO *charset;
@@ -1261,6 +1251,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 890687fc925..5074f4e0026 100644
--- a/sql/field_conv.cc
+++ b/sql/field_conv.cc
@@ -355,15 +355,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);
 }
 
 /***************************************************************************
@@ -506,7 +507,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 3bb8383e488..bb96075831a 100644
--- a/sql/ha_heap.cc
+++ b/sql/ha_heap.cc
@@ -427,16 +427,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 370458c6e01..b3b3a70b8ef 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -1869,11 +1869,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
@@ -1901,7 +1902,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) {
@@ -1930,17 +1931,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,
@@ -2463,12 +2458,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 7e337afed0e..d7c22318671 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 d703df7d2e3..00000000000
--- a/sql/ha_isam.cc
+++ /dev/null
@@ -1,408 +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(current_thd->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(current_thd->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(current_thd->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(current_thd->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(current_thd->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(current_thd->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(current_thd->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(current_thd->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(current_thd->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(current_thd->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(current_thd->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(current_thd->status_var.ha_read_rnd_count,
-		      &LOCK_status);
-  int error=nisam_rrnd(file, buf, (ulong) ha_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;
-  ha_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;
-    ha_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 5a070087724..00000000000
--- a/sql/ha_isammrg.cc
+++ /dev/null
@@ -1,210 +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(current_thd->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(current_thd->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(current_thd->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(current_thd->status_var.ha_read_rnd_count, &LOCK_status);
-  int error=mrg_rrnd(file, buf, (ulong) ha_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);
-  ha_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 7482c6d5fa8..eb7843b6598 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -1369,9 +1369,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 |
@@ -1386,8 +1386,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;
@@ -1466,30 +1466,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 b2e115e9779..3013c6ae669 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -387,6 +387,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;
@@ -1764,7 +1765,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),
@@ -3611,6 +3612,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 c408425ed60..67610187e27 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -73,6 +73,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 */
@@ -214,6 +215,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 d78c6f5b8ba..e255793681a 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -162,10 +162,15 @@ Item_ident::Item_ident(THD *thd, Item_ident *item)
 void Item_ident::cleanup()
 {
   DBUG_ENTER("Item_ident::cleanup");
-  DBUG_PRINT("enter", ("b:%s(%s), t:%s(%s), f:%s(%s)",
-		       db_name, orig_db_name,
-		       table_name, orig_table_name,
-		       field_name, orig_field_name));
+#ifdef CANT_BE_USED_AS_MEMORY_IS_FREED
+  DBUG_PRINT("enter", ("db: %s (%s)  table: %s (%s)  field: %s (%s)",
+		       db_name ? db_name : "null",
+                       orig_db_name ? orig_db_name : "null",
+		       table_name ? table_name : "null",
+                       orig_table_name ? orig_table_name : "null",
+		       field_name ? field_name : "null",
+                       orig_field_name ? orig_field_name : "null"));
+#endif
   Item::cleanup();
   db_name= orig_db_name; 
   table_name= orig_table_name;
@@ -1935,18 +1940,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)
 {
   /*
@@ -2002,12 +2018,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);
@@ -2336,6 +2357,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 905250ed96f..6505d6153af 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 72c85e2dd40..2b9596e9d9f 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -1195,7 +1195,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);
 }
 
 /*
@@ -1345,10 +1345,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 f036cbc799b..8f066354c58 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
@@ -3687,8 +3689,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)
   {
     /*
@@ -8437,7 +8438,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 30033bc39eb..690daa5e82b 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 2b0ae60f944..3ad25600230 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -545,6 +545,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 e7b3e13274f..8ef1473e71a 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -1052,9 +1052,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;
 
@@ -1437,6 +1436,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));
 
@@ -1466,9 +1466,11 @@ 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), &my_charset_latin1);
   table->field[1]->store(user,(uint) strlen(user), &my_charset_latin1);
+  key_copy(user_key, table->record[0], table->key_info,
+           table->key_info->key_length);
 
-  if (table->file->index_read_idx(table->record[0],0,
-				  (byte*) table->field[0]->ptr,0,
+  if (table->file->index_read_idx(table->record[0], 0,
+				  user_key, table->key_info->key_length,
 				  HA_READ_KEY_EXACT))
   {
     my_error(ER_PASSWORD_NO_MATCH,MYF(0));	/* purecov: deadcode */
@@ -1527,6 +1529,7 @@ 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';
+  byte user_key[MAX_KEY_LENGTH];
   DBUG_ENTER("replace_user_table");
   safe_mutex_assert_owner(&acl_cache->lock);
 
@@ -1546,9 +1549,12 @@ static int replace_user_table(THD *thd, TABLE *table, const LEX_USER &combo,
 
   table->field[0]->store(combo.host.str,combo.host.length, &my_charset_latin1);
   table->field[1]->store(combo.user.str,combo.user.length, &my_charset_latin1);
+  key_copy(user_key, table->record[0], table->key_info,
+           table->key_info->key_length);
+
   if (table->file->index_read_idx(table->record[0], 0,
-			      (byte*) table->field[0]->ptr,0,
-			      HA_READ_KEY_EXACT))
+                                  user_key, table->key_info->key_length,
+                                  HA_READ_KEY_EXACT))
   {
     /* what == 'N' means revoke */
     if (what == 'N')
@@ -1736,6 +1742,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)
@@ -1754,8 +1761,12 @@ static int replace_db_table(TABLE *table, const char *db,
   table->field[0]->store(combo.host.str,combo.host.length, &my_charset_latin1);
   table->field[1]->store(db,(uint) strlen(db), &my_charset_latin1);
   table->field[2]->store(combo.user.str,combo.user.length, &my_charset_latin1);
-  if (table->file->index_read_idx(table->record[0],0,(byte*) table->field[0]->ptr,0,
-			      HA_READ_KEY_EXACT))
+  key_copy(user_key, table->record[0], table->key_info,
+           table->key_info->key_length);
+
+  if (table->file->index_read_idx(table->record[0],0,
+                                  user_key, table->key_info->key_length,
+                                  HA_READ_KEY_EXACT))
   {
     if (what == 'N')
     { // no row, no revoke
@@ -1922,22 +1933,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),
                                &my_charset_latin1);
     col_privs->field[1]->store(db,(uint) strlen(db), &my_charset_latin1);
     col_privs->field[2]->store(user,(uint) strlen(user), &my_charset_latin1);
     col_privs->field[3]->store(tname,(uint) strlen(tname), &my_charset_latin1);
-    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();
@@ -1959,7 +1973,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();
   }
 }
@@ -2034,19 +2048,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, &my_charset_latin1);
   table->field[1]->store(db,(uint) strlen(db), &my_charset_latin1);
   table->field[2]->store(combo.user.str,combo.user.length, &my_charset_latin1);
-  table->field[3]->store(table_name,(uint) strlen(table_name), &my_charset_latin1);
-  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);
+  table->field[3]->store(table_name,(uint) strlen(table_name),
+                         &my_charset_latin1);
+  /* 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
+  rights&= COL_ACLS;				// Only ACL for columns
 
   /* first fix privileges for all columns in column list */
 
@@ -2057,12 +2074,19 @@ 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(),
                            &my_charset_latin1);
+    /* Get key for the first 4 columns */
+    key_copy(user_key, table->record[0], table->key_info,
+             table->key_info->key_length);
 
-    if (table->file->index_read(table->record[0],(byte*) table->field[0]->ptr,
-				0, 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)
       {
@@ -2073,7 +2097,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(),
                              &my_charset_latin1);
     }
@@ -2130,8 +2155,12 @@ static int replace_column_table(GRANT_TABLE *g_t,
 
   if (revoke_grant)
   {
-    if (table->file->index_read(table->record[0], (byte*) table->field[0]->ptr,
-				key_length, HA_READ_KEY_EXACT))
+    byte user_key[MAX_KEY_LENGTH];
+    key_copy(user_key, table->record[0], table->key_info,
+             table->key_info->key_length);
+    if (table->file->index_read(table->record[0], user_key,
+				table->key_info->key_length,
+                                HA_READ_KEY_EXACT))
       goto end;
 
     /* Scan through all rows with the same host,db,user and table */
@@ -2145,7 +2174,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),&my_charset_latin1);
+	String column_name(colum_name_buf,sizeof(colum_name_buf),
+                           &my_charset_latin1);
 
 	privileges&= ~rights;
 	table->field[6]->store((longlong)
@@ -2181,7 +2211,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:
@@ -2200,6 +2230,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);
@@ -2220,9 +2251,11 @@ static int replace_table_table(THD *thd, GRANT_TABLE *grant_table,
   table->field[2]->store(combo.user.str,combo.user.length, &my_charset_latin1);
   table->field[3]->store(table_name,(uint) strlen(table_name), &my_charset_latin1);
   store_record(table,record[1]);			// store at pos 1
+  key_copy(user_key, table->record[0], table->key_info,
+           table->key_info->key_length);
 
-  if (table->file->index_read_idx(table->record[0],0,
-				  (byte*) table->field[0]->ptr,0,
+  if (table->file->index_read_idx(table->record[0], 0,
+                                  user_key, table->key_info->key_length,
 				  HA_READ_KEY_EXACT))
   {
     /*
@@ -3631,7 +3664,8 @@ int mysql_drop_user(THD *thd, List <LEX_USER> &list)
   ACL_USER *acl_user;
   ACL_DB *acl_db;
   TABLE_LIST tables[4];
-
+  TABLE *table;
+  byte user_key[MAX_KEY_LENGTH];
   DBUG_ENTER("mysql_drop_user");
 
   if ((result= open_grant_tables(thd, tables)))
@@ -3707,22 +3741,25 @@ int mysql_drop_user(THD *thd, List <LEX_USER> &list)
       continue;
     }
 
-    tables[0].table->field[0]->store(user_name->host.str,(uint)
-				     user_name->host.length,
-				     system_charset_info);
-    tables[0].table->field[1]->store(user_name->user.str,(uint)
-				     user_name->user.length,
-				     system_charset_info);
-    if (!tables[0].table->file->index_read_idx(tables[0].table->record[0],0,
-					       (byte*) tables[0].table->
-					       field[0]->ptr,0,
-					       HA_READ_KEY_EXACT))
+    table= tables[0].table;
+    table->field[0]->store(user_name->host.str,(uint)
+                           user_name->host.length,
+                           system_charset_info);
+    table->field[1]->store(user_name->user.str,(uint)
+                           user_name->user.length,
+                           system_charset_info);
+    key_copy(user_key, table->record[0], table->key_info,
+             table->key_info->key_length);
+
+    if (!table->file->index_read_idx(table->record[0],0,
+                                     user_key,
+                                     table->key_info->key_length,
+                                     HA_READ_KEY_EXACT))
     {
       int error;
-      if ((error = tables[0].table->file->delete_row(tables[0].table->
-						     record[0])))
+      if ((error= table->file->delete_row(table->record[0])))
       {
-	tables[0].table->file->print_error(error, MYF(0));
+	table->file->print_error(error, MYF(0));
 	DBUG_RETURN(-1);
       }
       delete_dynamic_element(&acl_users, acl_userd);
@@ -3737,6 +3774,7 @@ int mysql_drop_user(THD *thd, List <LEX_USER> &list)
   DBUG_RETURN(result);
 }
 
+
 int mysql_revoke_all(THD *thd,  List <LEX_USER> &list)
 {
   uint counter, revoked;
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 39604ce4deb..649494603b7 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1053,31 +1053,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 b3d7bebe96a..0d6e42b7a1b 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 4353206064e..20111ac6c1a 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);
 }
 
 
@@ -4729,7 +4734,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)
@@ -4805,7 +4810,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;
 
@@ -4877,36 +4882,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;
@@ -5037,10 +5028,10 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
       if (new_field->pack_length > 4)
 	new_field->pack_length=8;
       new_field->interval=interval;
-      uint dummy_max_length;
+      uint dummy_max_length, field_length;
       calculate_interval_lengths(thd, interval,
-                                 &dummy_max_length, &new_field->length);
-      new_field->length+= (interval->count - 1);
+                                 &dummy_max_length, &field_length);
+      new_field->length= field_length+ (interval->count - 1);
       set_if_smaller(new_field->length,MAX_FIELD_WIDTH-1);
       if (default_value)
       {
@@ -5067,10 +5058,11 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
       new_field->interval=interval;
       new_field->pack_length=interval->count < 256 ? 1 : 2; // Should be safe
 
-      uint dummy_tot_length;
+      uint dummy_tot_length, field_length;
       calculate_interval_lengths(thd, interval,
-                                 &new_field->length, &dummy_tot_length);
-      set_if_smaller(new_field->length,MAX_FIELD_WIDTH-1);
+                                 &field_length, &dummy_tot_length);
+      set_if_smaller(field_length, MAX_FIELD_WIDTH-1);
+      new_field->length= field_length;
       if (default_value)
       {
 	String str,*res;
@@ -5084,16 +5076,21 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
       }
       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)))
   {
     net_printf(thd,ER_TOO_BIG_FIELDLENGTH,field_name,
-	       MAX_FIELD_CHARLENGTH);		/* purecov: inspected */
+	       max_field_charlength);		/* purecov: inspected */
     DBUG_RETURN(1);				/* purecov: inspected */
   }
   type_modifier&= AUTO_INCREMENT_FLAG;
@@ -5103,11 +5100,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);
@@ -5151,7 +5147,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 85feea3d51a..41dd136344a 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -7484,7 +7484,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
@@ -7497,7 +7498,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;
@@ -7704,6 +7706,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,
@@ -7711,10 +7718,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];
   byte	*pos,*group_buff;
@@ -7725,8 +7735,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,
@@ -7819,7 +7827,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);
@@ -7866,6 +7875,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))
           {
@@ -7959,6 +7974,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;
@@ -8033,10 +8054,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;
@@ -8110,6 +8130,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;
@@ -8280,12 +8302,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;
       }
@@ -9790,6 +9810,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
@@ -9937,11 +9958,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 e73bf81eb74..7499b1ca69c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -404,6 +404,9 @@ void check_duplicates_in_interval(const char *set_or_name,
   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
@@ -418,20 +421,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;
     /*
@@ -443,22 +450,53 @@ 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);
     }
 
     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;
@@ -532,6 +570,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
@@ -549,17 +588,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_printf_error(ER_FEATURE_DISABLED,ER(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) |
@@ -840,6 +899,8 @@ 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;
+
       it.rewind();
       field=0;
       while ((sql_field=it++) &&
@@ -862,8 +923,8 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
       */
       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
@@ -904,15 +965,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
@@ -948,7 +1009,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))
@@ -1014,12 +1076,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;
       }
@@ -1058,7 +1121,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_update.cc b/sql/sql_update.cc
index 7bd445088cf..01396ce12a1 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -701,7 +701,7 @@ int mysql_multi_update_prepare(THD *thd)
     */
     List_iterator_fast<Item> it(*fields);
     Item *item;
-    while (item= it++)
+    while ((item= it++))
     {
       item->cleanup();
     }
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 309a9249d0e..7ea473ef81f 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -2735,13 +2735,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 cc053e2e2fd..1195948bbf2 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -93,6 +93,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 0e8045abacf..e7781f6e287 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
@@ -590,10 +591,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;
@@ -1213,7 +1216,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 */
@@ -1366,7 +1373,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;
@@ -1443,7 +1450,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;
@@ -1478,7 +1485,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 af7d90a8291..aef9a77152b 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -160,6 +160,7 @@ 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 insert_or_update;             /* Can be used by the handler */
   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