Merge with global tree

myisam/mi_key.c:
  Auto merged
myisam/mi_open.c:
  Auto merged
mysql-test/r/information_schema.result:
  Auto merged
mysql-test/r/ps_1general.result:
  Auto merged
mysys/my_handler.c:
  Auto merged
sql/field.cc:
  Auto merged
sql/field.h:
  Auto merged
sql/field_conv.cc:
  Auto merged
sql/ha_myisam.cc:
  Auto merged
sql/sql_parse.cc:
  Auto merged
sql/sql_select.cc:
  Auto merged
sql/sql_show.cc:
  Auto merged
sql/sql_table.cc:
  Auto merged
include/my_base.h:
  Merge

57 files changed, 1405 insertions, 460 deletions

diff --git a/BUILD/SETUP.sh b/BUILD/SETUP.sh
index 8fa70aecb6a..d378276a0a3 100644
--- a/BUILD/SETUP.sh
+++ b/BUILD/SETUP.sh
@@ -45,6 +45,7 @@ cxx_warnings="$global_warnings -Woverloaded-virtual -Wsign-promo -Wreorder -Wcto
 
 alpha_cflags="-mcpu=ev6 -Wa,-mev6"	# Not used yet
 pentium_cflags="-mcpu=pentiumpro"
+pentium64_cflags="-mcpu=nocona -m64"
 ppc_cflags="-mpowerpc -mcpu=powerpc"
 sparc_cflags=""
 
diff --git a/BUILD/compile-pentium64-valgrind-max b/BUILD/compile-pentium64-valgrind-max
new file mode 100644
index 00000000000..7f78089c3e8
--- /dev/null
+++ b/BUILD/compile-pentium64-valgrind-max
@@ -0,0 +1,29 @@
+#! /bin/sh
+
+path=`dirname $0`
+. "$path/SETUP.sh"
+
+extra_flags="$pentium64_cflags $debug_cflags -USAFEMALLOC -UFORCE_INIT_OF_VARS -DHAVE_purify -DMYSQL_SERVER_SUFFIX=-valgrind-max"
+c_warnings="$c_warnings $debug_extra_warnings"
+cxx_warnings="$cxx_warnings $debug_extra_warnings"
+extra_configs="$pentium_configs $debug_configs"
+
+# We want to test isam when building with valgrind
+extra_configs="$extra_configs --with-berkeley-db --with-innodb --with-isam --with-embedded-server --with-openssl --with-vio --with-raid --with-ndbcluster"
+
+. "$path/FINISH.sh"
+
+if test -z "$just_print"
+then
+  set +v +x
+  echo "\
+******************************************************************************
+Note that by default BUILD/compile-pentium-valgrind-max calls 'configure' with
+--enable-assembler. When Valgrind detects an error involving an assembly
+function (for example an uninitialized value used as an argument of an
+assembly function), Valgrind will not print the stacktrace and 'valgrind
+--gdb-attach=yes' will not work either. If you need a stacktrace in those
+cases, you have to run BUILD/compile-pentium-valgrind-max with the
+--disable-assembler argument.
+******************************************************************************"
+fi
diff --git a/dbug/dbug.c b/dbug/dbug.c
index d21b4e7801a..91b7e7b6c4c 100644
--- a/dbug/dbug.c
+++ b/dbug/dbug.c
@@ -978,7 +978,7 @@ uint length)
     {
       fprintf(_db_fp_, "%s: ", state->func);
     }
-    sprintf(dbuff,"%s: Memory: %lx  Bytes: (%d)\n",
+    sprintf(dbuff,"%s: Memory: 0x%lx  Bytes: (%d)\n",
 	    keyword,(ulong) memory, length);
     (void) fputs(dbuff,_db_fp_);
 
diff --git a/heap/_check.c b/heap/_check.c
index a745aee48bf..4316a9926f7 100644
--- a/heap/_check.c
+++ b/heap/_check.c
@@ -123,7 +123,7 @@ static int check_one_key(HP_KEYDEF *keydef, uint keynr, ulong records,
 			        blength, records))
 	    != i)
 	{
-	  DBUG_PRINT("error",("Record in wrong link: Link %d  Record: %lx  Record-link %d", i,hash_info->ptr_to_rec,rec_link));
+	  DBUG_PRINT("error",("Record in wrong link: Link %d  Record: 0x%lx  Record-link %d", i,hash_info->ptr_to_rec,rec_link));
 	  error=1;
 	}
 	else
@@ -180,7 +180,7 @@ static int check_one_rb_key(HP_INFO *info, uint keynr, ulong records,
 		     key_length, SEARCH_FIND | SEARCH_SAME, &not_used))
       {
 	error= 1;
-	DBUG_PRINT("error",("Record in wrong link:  key: %d  Record: %lx\n", 
+	DBUG_PRINT("error",("Record in wrong link:  key: %d  Record: 0x%lx\n", 
 			    keynr, recpos));
       }
       else
diff --git a/heap/hp_create.c b/heap/hp_create.c
index d296c9db28b..0580c178498 100644
--- a/heap/hp_create.c
+++ b/heap/hp_create.c
@@ -77,14 +77,31 @@ int heap_create(const char *name, uint keys, HP_KEYDEF *keydef,
 	case HA_KEYTYPE_INT8:
 	  keyinfo->seg[j].flag|= HA_SWAP_KEY;
           break;
-        case HA_KEYTYPE_VARBINARY:
+        case HA_KEYTYPE_VARBINARY1:
           /* Case-insensitiveness is handled in coll->hash_sort */
-          keyinfo->seg[j].type= HA_KEYTYPE_VARTEXT;
+          keyinfo->seg[j].type= HA_KEYTYPE_VARTEXT1;
           /* fall_through */
-        case HA_KEYTYPE_VARTEXT:
+        case HA_KEYTYPE_VARTEXT1:
           if (!my_binary_compare(keyinfo->seg[j].charset))
             keyinfo->flag|= HA_END_SPACE_KEY;
           keyinfo->flag|= HA_VAR_LENGTH_KEY;
+          /* Save number of bytes used to store length */
+          keyinfo->seg[j].bit_start= 1;
+          break;
+        case HA_KEYTYPE_VARBINARY2:
+          /* Case-insensitiveness is handled in coll->hash_sort */
+          /* fall_through */
+        case HA_KEYTYPE_VARTEXT2:
+          if (!my_binary_compare(keyinfo->seg[j].charset))
+            keyinfo->flag|= HA_END_SPACE_KEY;
+          keyinfo->flag|= HA_VAR_LENGTH_KEY;
+          /* Save number of bytes used to store length */
+          keyinfo->seg[j].bit_start= 2;
+          /*
+            Make future comparison simpler by only having to check for
+            one type
+          */
+          keyinfo->seg[j].type= HA_KEYTYPE_VARTEXT1;
           break;
 	default:
 	  break;
diff --git a/heap/hp_delete.c b/heap/hp_delete.c
index 4adefde1fe9..5287533ae0a 100644
--- a/heap/hp_delete.c
+++ b/heap/hp_delete.c
@@ -24,7 +24,7 @@ int heap_delete(HP_INFO *info, const byte *record)
   HP_SHARE *share=info->s;
   HP_KEYDEF *keydef, *end, *p_lastinx;
   DBUG_ENTER("heap_delete");
-  DBUG_PRINT("enter",("info: %lx  record: %lx",info,record));
+  DBUG_PRINT("enter",("info: %lx  record: 0x%lx",info,record));
 
   test_active(info);
 
@@ -139,7 +139,7 @@ int hp_delete_key(HP_INFO *info, register HP_KEYDEF *keyinfo,
     /* Save for heap_rnext/heap_rprev */
     info->current_hash_ptr=last_ptr;
     info->current_ptr = last_ptr ? last_ptr->ptr_to_rec : 0;
-    DBUG_PRINT("info",("Corrected current_ptr to point at: %lx",
+    DBUG_PRINT("info",("Corrected current_ptr to point at: 0x%lx",
 		       info->current_ptr));
   }
   empty=pos;
diff --git a/heap/hp_hash.c b/heap/hp_hash.c
index 7e5f92bc7b8..3121ef71fb0 100644
--- a/heap/hp_hash.c
+++ b/heap/hp_hash.c
@@ -271,18 +271,21 @@ ulong hp_hashnr(register HP_KEYDEF *keydef, register const byte *key)
        }
        cs->coll->hash_sort(cs, pos, length, &nr, &nr2);
     }
-    else if (seg->type == HA_KEYTYPE_VARTEXT)
+    else if (seg->type == HA_KEYTYPE_VARTEXT1)  /* Any VARCHAR segments */
     {
        CHARSET_INFO *cs= seg->charset;
+       uint pack_length= 2;                     /* Key packing is constant */
        uint length= uint2korr(pos);
        if (cs->mbmaxlen > 1)
        {
          uint char_length;
-         char_length= my_charpos(cs, pos +2, pos +2 + length,
+         char_length= my_charpos(cs, pos +pack_length,
+                                 pos +pack_length + length,
                                  seg->length/cs->mbmaxlen);
          set_if_smaller(length, char_length);
        }
-       cs->coll->hash_sort(cs, pos+2, length, &nr, &nr2);
+       cs->coll->hash_sort(cs, pos+pack_length, length, &nr, &nr2);
+       key+= pack_length;
     }
     else
     {
@@ -293,6 +296,7 @@ ulong hp_hashnr(register HP_KEYDEF *keydef, register const byte *key)
       }
     }
   }
+  DBUG_PRINT("exit", ("hash: 0x%lx", nr));
   return((ulong) nr);
 }
 
@@ -300,7 +304,6 @@ ulong hp_hashnr(register HP_KEYDEF *keydef, register const byte *key)
 
 ulong hp_rec_hashnr(register HP_KEYDEF *keydef, register const byte *rec)
 {
-  /*register*/
   ulong nr=1, nr2=4;
   HA_KEYSEG *seg,*endseg;
 
@@ -327,18 +330,20 @@ ulong hp_rec_hashnr(register HP_KEYDEF *keydef, register const byte *rec)
       }
       cs->coll->hash_sort(cs, pos, char_length, &nr, &nr2);
     }
-    else if (seg->type == HA_KEYTYPE_VARTEXT)
+    else if (seg->type == HA_KEYTYPE_VARTEXT1)  /* Any VARCHAR segments */
     {
       CHARSET_INFO *cs= seg->charset;
-      uint length= uint2korr(pos);
+      uint pack_length= seg->bit_start;
+      uint length= (pack_length == 1 ? (uint) *(uchar*) pos : uint2korr(pos));
       if (cs->mbmaxlen > 1)
       {
         uint char_length;
-        char_length= my_charpos(cs, pos + 2 , pos + 2 + length,
+        char_length= my_charpos(cs, pos + pack_length,
+                                pos + pack_length + length,
                                 seg->length/cs->mbmaxlen);
         set_if_smaller(length, char_length);
       }
-      cs->coll->hash_sort(cs, pos+2, length, &nr, &nr2);
+      cs->coll->hash_sort(cs, pos+pack_length, length, &nr, &nr2);
     }
     else
     {
@@ -349,7 +354,8 @@ ulong hp_rec_hashnr(register HP_KEYDEF *keydef, register const byte *rec)
       }
     }
   }
-  return((ulong) nr);
+  DBUG_PRINT("exit", ("hash: 0x%lx", nr));
+  return(nr);
 }
 
 #else
@@ -392,10 +398,13 @@ ulong hp_hashnr(register HP_KEYDEF *keydef, register const byte *key)
     {
       seg->charset->hash_sort(seg->charset,pos,((uchar*)key)-pos,&nr,NULL);
     }
-    else if (seg->type == HA_KEYTYPE_VARTEXT)
+    else if (seg->type == HA_KEYTYPE_VARTEXT1)  /* Any VARCHAR segments */
     {
+      uint pack_length= 2;                      /* Key packing is constant */
       uint length= uint2korr(pos);
-      seg->charset->hash_sort(seg->charset, pos+2, length, &nr, NULL);
+      seg->charset->hash_sort(seg->charset, pos+pack_length, length, &nr,
+                              NULL);
+      key+= pack_length;
     }
     else
     {
@@ -406,7 +415,8 @@ ulong hp_hashnr(register HP_KEYDEF *keydef, register const byte *key)
       }
     }
   }
-  return((ulong) nr);
+  DBUG_PRINT("exit", ("hash: 0x%lx", nr));
+  return(nr);
 }
 
 	/* Calc hashvalue for a key in a record */
@@ -418,7 +428,7 @@ ulong hp_rec_hashnr(register HP_KEYDEF *keydef, register const byte *rec)
 
   for (seg=keydef->seg,endseg=seg+keydef->keysegs ; seg < endseg ; seg++)
   {
-    uchar *pos=(uchar*) rec+seg->start,*end=pos+seg->length;
+    uchar *pos=(uchar*) rec+seg->start;
     if (seg->null_bit)
     {
       if (rec[seg->null_pos] & seg->null_bit)
@@ -431,13 +441,16 @@ ulong hp_rec_hashnr(register HP_KEYDEF *keydef, register const byte *rec)
     {
       seg->charset->hash_sort(seg->charset,pos,((uchar*)key)-pos,&nr,NULL);
     }
-    else if (seg->type == HA_KEYTYPE_VARTEXT)
+    else if (seg->type == HA_KEYTYPE_VARTEXT1)  /* Any VARCHAR segments */
     {
-      uint length= uint2korr(pos);
-      seg->charset->hash_sort(seg->charset, pos+2, length, &nr, NULL);
+      uint pack_length= seg->bit_start;
+      uint length= (pack_length == 1 ? (uint) *(uchar*) pos : uint2korr(pos));
+      seg->charset->hash_sort(seg->charset, pos+pack_length,
+                              length, &nr, NULL);
     }
     else
     {
+      uchar *end= pos+seg->length;
       for ( ; pos < end ; pos++)
       {
 	nr *=16777619; 
@@ -445,7 +458,8 @@ ulong hp_rec_hashnr(register HP_KEYDEF *keydef, register const byte *rec)
       }
     }
   }
-  return((ulong) nr);
+  DBUG_PRINT("exit", ("hash: 0x%lx", nr));
+  return(nr);
 }
 
 #endif
@@ -510,13 +524,25 @@ int hp_rec_key_cmp(HP_KEYDEF *keydef, const byte *rec1, const byte *rec2,
 					  pos2,char_length2, 0))
 	return 1;
     }
-    else if (seg->type == HA_KEYTYPE_VARTEXT)
+    else if (seg->type == HA_KEYTYPE_VARTEXT1)  /* Any VARCHAR segments */
     {
-      uchar *pos1= (uchar*)rec1 + seg->start;
-      uchar *pos2= (uchar*)rec2 + seg->start;
-      uint char_length1= uint2korr(pos1);
-      uint char_length2= uint2korr(pos2);
+      uchar *pos1= (uchar*) rec1 + seg->start;
+      uchar *pos2= (uchar*) rec2 + seg->start;
+      uint char_length1, char_length2;
+      uint pack_length= seg->bit_start;
       CHARSET_INFO *cs= seg->charset;
+      if (pack_length == 1)
+      {
+        char_length1= (uint) *(uchar*) pos1++;
+        char_length2= (uint) *(uchar*) pos2++;
+      }
+      else
+      {
+        char_length1= uint2korr(pos1);
+        char_length2= uint2korr(pos2);
+        pos1+= 2;
+        pos2+= 2;
+      }
       if (cs->mbmaxlen > 1)
       {
         uint char_length= seg->length / cs->mbmaxlen;
@@ -527,8 +553,8 @@ int hp_rec_key_cmp(HP_KEYDEF *keydef, const byte *rec1, const byte *rec2,
       }
 
       if (cs->coll->strnncollsp(seg->charset,
-                                pos1+2, char_length1,
-                                pos2+2, char_length2,
+                                pos1, char_length1,
+                                pos2, char_length2,
                                 seg->flag & HA_END_SPACE_ARE_EQUAL ?
                                 0 : diff_if_only_endspace_difference))
 	return 1;
@@ -585,28 +611,31 @@ int hp_key_cmp(HP_KEYDEF *keydef, const byte *rec, const byte *key)
 					  (uchar*) key, char_length_key, 0))
 	return 1;
     }
-    else if (seg->type == HA_KEYTYPE_VARTEXT)
+    else if (seg->type == HA_KEYTYPE_VARTEXT1)  /* Any VARCHAR segments */
     {
       uchar *pos= (uchar*) rec + seg->start;
       CHARSET_INFO *cs= seg->charset;
-      uint char_length_rec= uint2korr(pos);
+      uint pack_length= seg->bit_start;
+      uint char_length_rec= (pack_length == 1 ? (uint) *(uchar*) pos :
+                             uint2korr(pos));
+      /* Key segments are always packed with 2 bytes */
       uint char_length_key= uint2korr(key);
-
+      pos+= pack_length;
+      key+= 2;                                  /* skip key pack length */
       if (cs->mbmaxlen > 1)
       {
         uint char_length= seg->length / cs->mbmaxlen;
-        char_length_key= my_charpos(cs, key+2, key +2 + char_length_key,
+        char_length_key= my_charpos(cs, key, key + char_length_key,
                                     char_length);
         set_if_smaller(char_length_key, seg->length);
-        char_length_rec= my_charpos(cs, pos +2 , pos + 2 + char_length_rec,
+        char_length_rec= my_charpos(cs, pos, pos + char_length_rec,
                                     char_length);
         set_if_smaller(char_length_rec, seg->length);
       }
 
-
       if (cs->coll->strnncollsp(seg->charset,
-                                (uchar*) pos+2, char_length_rec,
-                                (uchar*) key+2, char_length_key, 0))
+                                (uchar*) pos, char_length_rec,
+                                (uchar*) key, char_length_key, 0))
 	return 1;
     }
     else
@@ -638,6 +667,8 @@ void hp_make_key(HP_KEYDEF *keydef, byte *key, const byte *rec)
                               char_length / cs->mbmaxlen);
       set_if_smaller(char_length, seg->length); /* QQ: ok to remove? */
     }
+    if (seg->type == HA_KEYTYPE_VARTEXT1)
+      char_length+= seg->bit_start;             /* Copy also length */
     memcpy(key,rec+seg->start,(size_t) char_length);
     key+= char_length;
   }
@@ -707,11 +738,13 @@ uint hp_rb_make_key(HP_KEYDEF *keydef, byte *key,
     {
       uchar *pos=      (uchar*) rec + seg->start;
       uint length=     seg->length;
-      uint tmp_length= uint2korr(pos);
+      uint pack_length= seg->bit_start;
+      uint tmp_length= (pack_length == 1 ? (uint) *(uchar*) pos :
+                        uint2korr(pos));
       CHARSET_INFO *cs= seg->charset;
       char_length= length/cs->mbmaxlen;
 
-      pos+=2;					/* Skip VARCHAR length */
+      pos+= pack_length;			/* Skip VARCHAR length */
       set_if_smaller(length,tmp_length);
       FIX_LENGTH(cs, pos, length, char_length);
       store_key_length_inc(key,char_length);
diff --git a/heap/hp_open.c b/heap/hp_open.c
index 1fa832208fb..fd937229b0d 100644
--- a/heap/hp_open.c
+++ b/heap/hp_open.c
@@ -63,7 +63,7 @@ HP_INFO *heap_open(const char *name, int mode)
 #ifndef DBUG_OFF
   info->opt_flag= READ_CHECK_USED;		/* Check when changing */
 #endif
-  DBUG_PRINT("exit",("heap: %lx  reclength: %d  records_in_block: %d",
+  DBUG_PRINT("exit",("heap: 0x%lx  reclength: %d  records_in_block: %d",
 		     info,share->reclength,share->block.records_in_block));
   DBUG_RETURN(info);
 }
@@ -82,7 +82,7 @@ HP_SHARE *hp_find_named_heap(const char *name)
     info= (HP_SHARE*) pos->data;
     if (!strcmp(name, info->name))
     {
-      DBUG_PRINT("exit", ("Old heap_database: %lx",info));
+      DBUG_PRINT("exit", ("Old heap_database: 0x%lx",info));
       DBUG_RETURN(info);
     }
   }
diff --git a/heap/hp_rkey.c b/heap/hp_rkey.c
index a88139bbdee..f5f22a877a1 100644
--- a/heap/hp_rkey.c
+++ b/heap/hp_rkey.c
@@ -23,7 +23,7 @@ int heap_rkey(HP_INFO *info, byte *record, int inx, const byte *key,
   HP_SHARE *share= info->s;
   HP_KEYDEF *keyinfo= share->keydef + inx;
   DBUG_ENTER("heap_rkey");
-  DBUG_PRINT("enter",("base: %lx  inx: %d",info,inx));
+  DBUG_PRINT("enter",("base: 0x%lx  inx: %d",info,inx));
 
   if ((uint) inx >= share->keys)
   {
diff --git a/heap/hp_rrnd.c b/heap/hp_rrnd.c
index cce3ce24e51..4daa3a06377 100644
--- a/heap/hp_rrnd.c
+++ b/heap/hp_rrnd.c
@@ -29,7 +29,7 @@ int heap_rrnd(register HP_INFO *info, byte *record, byte *pos)
 {
   HP_SHARE *share=info->s;
   DBUG_ENTER("heap_rrnd");
-  DBUG_PRINT("enter",("info: %lx  pos: %lx",info,pos));
+  DBUG_PRINT("enter",("info: 0x%lx  pos: %lx",info,pos));
 
   info->lastinx= -1;
   if (!(info->current_ptr= pos))
@@ -44,7 +44,7 @@ int heap_rrnd(register HP_INFO *info, byte *record, byte *pos)
   }
   info->update=HA_STATE_PREV_FOUND | HA_STATE_NEXT_FOUND | HA_STATE_AKTIV;
   memcpy(record,info->current_ptr,(size_t) share->reclength);
-  DBUG_PRINT("exit",("found record at %lx",info->current_ptr));
+  DBUG_PRINT("exit",("found record at 0x%lx",info->current_ptr));
   info->current_hash_ptr=0;			/* Can't use rnext */
   DBUG_RETURN(0);
 } /* heap_rrnd */
@@ -64,7 +64,7 @@ int heap_rrnd_old(register HP_INFO *info, byte *record, ulong pos)
 {
   HP_SHARE *share=info->s;
   DBUG_ENTER("heap_rrnd");
-  DBUG_PRINT("enter",("info: %lx  pos: %ld",info,pos));
+  DBUG_PRINT("enter",("info: 0x%lx  pos: %ld",info,pos));
 
   info->lastinx= -1;
   if (pos == (ulong) -1)
@@ -98,7 +98,7 @@ end:
   }
   info->update=HA_STATE_PREV_FOUND | HA_STATE_NEXT_FOUND | HA_STATE_AKTIV;
   memcpy(record,info->current_ptr,(size_t) share->reclength);
-  DBUG_PRINT("exit",("found record at %lx",info->current_ptr));
+  DBUG_PRINT("exit",("found record at 0x%lx",info->current_ptr));
   info->current_hash_ptr=0;			/* Can't use rnext */
   DBUG_RETURN(0);
 } /* heap_rrnd */
diff --git a/heap/hp_write.c b/heap/hp_write.c
index 577c52a007d..171998e9125 100644
--- a/heap/hp_write.c
+++ b/heap/hp_write.c
@@ -138,7 +138,7 @@ static byte *next_free_record_pos(HP_SHARE *info)
     pos=info->del_link;
     info->del_link= *((byte**) pos);
     info->deleted--;
-    DBUG_PRINT("exit",("Used old position: %lx",pos));
+    DBUG_PRINT("exit",("Used old position: 0x%lx",pos));
     DBUG_RETURN(pos);
   }
   if (!(block_pos=(info->records % info->block.records_in_block)))
diff --git a/include/my_base.h b/include/my_base.h
index a1b14ecbb65..9c860c7eebd 100644
--- a/include/my_base.h
+++ b/include/my_base.h
@@ -181,9 +181,13 @@ enum ha_base_keytype {
   HA_KEYTYPE_INT24=12,
   HA_KEYTYPE_UINT24=13,
   HA_KEYTYPE_INT8=14,
-  HA_KEYTYPE_VARTEXT=15,		/* Key is sorted as letters */
-  HA_KEYTYPE_VARBINARY=16,		/* Key is sorted as unsigned chars */
-  HA_KEYTYPE_BIT=17
+  /* Varchar (0-255 bytes) with length packed with 1 byte */
+  HA_KEYTYPE_VARTEXT1=15,               /* Key is sorted as letters */
+  HA_KEYTYPE_VARBINARY1=16,             /* Key is sorted as unsigned chars */
+  /* Varchar (0-65535 bytes) with length packed with 2 bytes */
+  HA_KEYTYPE_VARTEXT2=17,		/* Key is sorted as letters */
+  HA_KEYTYPE_VARBINARY2=18,		/* Key is sorted as unsigned chars */
+  HA_KEYTYPE_BIT=18
 };
 
 #define HA_MAX_KEYTYPE	31		/* Must be log2-1 */
@@ -391,4 +395,6 @@ typedef ulong		ha_rows;
 #define MAX_FILE_SIZE	LONGLONG_MAX
 #endif
 
+#define HA_VARCHAR_PACKLENGTH(field_length) ((field_length) < 256 ? 1 :2)
+
 #endif /* _my_base_h */
diff --git a/myisam/ft_static.c b/myisam/ft_static.c
index 3b186f7b179..cdb1580e706 100644
--- a/myisam/ft_static.c
+++ b/myisam/ft_static.c
@@ -25,9 +25,9 @@ char ft_boolean_syntax[]="+ -><()~*:\"\"&|";
 
 const HA_KEYSEG ft_keysegs[FT_SEGS]={
 {
-    HA_KEYTYPE_VARTEXT,               /* type */
+    HA_KEYTYPE_VARTEXT2,              /* type */
     63,                               /* language (will be overwritten) */
-    0, 0, 0,                          /* null_bit, bit_start, bit_end */
+    0, 2, 0,                          /* null_bit, bit_start, bit_end */
     HA_VAR_LENGTH_PART | HA_PACK_KEY,      /* flag */
     HA_FT_MAXBYTELEN,                 /* length */
     HA_FT_WLEN,                       /* start */
diff --git a/myisam/ft_test1.c b/myisam/ft_test1.c
index a92c85924de..14be9aa1e8c 100644
--- a/myisam/ft_test1.c
+++ b/myisam/ft_test1.c
@@ -79,24 +79,24 @@ static int run_test(const char *filename)
   recinfo[0].length= (extra_field == FIELD_BLOB ? 4 + mi_portable_sizeof_char_ptr :
 	      extra_length);
   if (extra_field == FIELD_VARCHAR)
-    recinfo[0].length+=2;
+    recinfo[0].length+= HA_VARCHAR_PACKLENGTH(extra_length);
   recinfo[1].type=key_field;
   recinfo[1].length= (key_field == FIELD_BLOB ? 4+mi_portable_sizeof_char_ptr :
 		      key_length);
   if (key_field == FIELD_VARCHAR)
-    recinfo[1].length+=2;
+    recinfo[1].length+= HA_VARCHAR_PACKLENGTH(key_length);
 
   /* Define a key over the first column */
   keyinfo[0].seg=keyseg;
   keyinfo[0].keysegs=1;
   keyinfo[0].seg[0].type= key_type;
-  keyinfo[0].seg[0].flag= (key_field == FIELD_BLOB)?HA_BLOB_PART:
-			  (key_field == FIELD_VARCHAR)?HA_VAR_LENGTH_PART:0;
+  keyinfo[0].seg[0].flag= (key_field == FIELD_BLOB) ? HA_BLOB_PART:
+			  (key_field == FIELD_VARCHAR) ? HA_VAR_LENGTH_PART:0;
   keyinfo[0].seg[0].start=recinfo[0].length;
   keyinfo[0].seg[0].length=key_length;
   keyinfo[0].seg[0].null_bit= 0;
   keyinfo[0].seg[0].null_pos=0;
-  keyinfo[0].seg[0].language=MY_CHARSET_CURRENT;
+  keyinfo[0].seg[0].language= default_charset_info->number;
   keyinfo[0].flag = (no_fulltext?HA_PACK_KEY:HA_FULLTEXT);
 
   if (!silent)
@@ -155,33 +155,42 @@ static int run_test(const char *filename)
   if (!silent)
     printf("- Reading rows with key\n");
   for (i=0 ; i < NQUERIES ; i++)
-  { FT_DOCLIST *result;
+  {
+    FT_DOCLIST *result;
     result=ft_nlq_init_search(file,0,(char*) query[i],strlen(query[i]),1);
-    if(!result) {
+    if(!result)
+    {
       printf("Query %d: `%s' failed with errno %3d\n",i,query[i],my_errno);
       continue;
     }
     printf("Query %d: `%s'. Found: %d. Top five documents:\n",
-	    i,query[i],result->ndocs);
-    for(j=0;j<5;j++) { double w; int err;
-	err=ft_nlq_read_next(result, read_record);
-	if(err==HA_ERR_END_OF_FILE) {
-	    printf("No more matches!\n");
-	    break;
-	} else if (err) {
-	    printf("ft_read_next %d failed with errno %3d\n",j,my_errno);
-	    break;
-	}
-        w=ft_nlq_get_relevance(result);
-	if(key_field == FIELD_VARCHAR) {
-	    uint l;
-	    char *p;
-	    p=recinfo[0].length+read_record;
-	    l=uint2korr(p);
-	    printf("%10.7f: %.*s\n",w,(int) l,p+2);
-	} else
-	    printf("%10.7f: %.*s\n",w,recinfo[1].length,
-			  recinfo[0].length+read_record);
+           i,query[i],result->ndocs);
+    for (j=0;j<5;j++)
+    {
+      double w; int err;
+      err= ft_nlq_read_next(result, read_record);
+      if (err==HA_ERR_END_OF_FILE)
+      {
+        printf("No more matches!\n");
+        break;
+      }
+      else if (err)
+      {
+        printf("ft_read_next %d failed with errno %3d\n",j,my_errno);
+        break;
+      }
+      w=ft_nlq_get_relevance(result);
+      if (key_field == FIELD_VARCHAR)
+      {
+        uint l;
+        char *p;
+        p=recinfo[0].length+read_record;
+        l=uint2korr(p);
+        printf("%10.7f: %.*s\n",w,(int) l,p+2);
+      }
+      else
+        printf("%10.7f: %.*s\n",w,recinfo[1].length,
+               recinfo[0].length+read_record);
     }
     ft_nlq_close_search(result);
   }
@@ -215,9 +224,14 @@ void create_record(char *pos, int n)
   else if (recinfo[0].type == FIELD_VARCHAR)
   {
     uint tmp;
-    strnmov(pos+2,data[n].f0,keyinfo[0].seg[0].length);
-    tmp=strlen(pos+2);
-    int2store(pos,tmp);
+    /* -1 is here because pack_length is stored in seg->length */
+    uint pack_length= HA_VARCHAR_PACKLENGTH(keyinfo[0].seg[0].length-1);
+    strnmov(pos+pack_length,data[n].f0,keyinfo[0].seg[0].length);
+    tmp=strlen(pos+pack_length);
+    if (pack_length == 1)
+      *pos= (char) tmp;
+    else
+      int2store(pos,tmp);
     pos+=recinfo[0].length;
   }
   else
@@ -239,9 +253,14 @@ void create_record(char *pos, int n)
   else if (recinfo[1].type == FIELD_VARCHAR)
   {
     uint tmp;
-    strnmov(pos+2,data[n].f2,keyinfo[0].seg[0].length);
-    tmp=strlen(pos+2);
-    int2store(pos,tmp);
+    /* -1 is here because pack_length is stored in seg->length */
+    uint pack_length= HA_VARCHAR_PACKLENGTH(keyinfo[0].seg[0].length-1);
+    strnmov(pos+pack_length,data[n].f2,keyinfo[0].seg[0].length);
+    tmp=strlen(pos+1);
+    if (pack_length == 1)
+      *pos= (char) tmp;
+    else
+      int2store(pos,tmp);
     pos+=recinfo[1].length;
   }
   else
diff --git a/myisam/ft_update.c b/myisam/ft_update.c
index 8dafefe77a8..b8cd925bf4f 100644
--- a/myisam/ft_update.c
+++ b/myisam/ft_update.c
@@ -58,29 +58,27 @@ uint _mi_ft_segiterator(register FT_SEG_ITERATOR *ftsi)
   DBUG_ENTER("_mi_ft_segiterator");
 
   if (!ftsi->num)
-  {
     DBUG_RETURN(0);
-  }
-  else
-    ftsi->num--;
+
+  ftsi->num--;
   if (!ftsi->seg)
-  {
     DBUG_RETURN(1);
-  }
-  else
-    ftsi->seg--;
+
+  ftsi->seg--;
 
   if (ftsi->seg->null_bit &&
       (ftsi->rec[ftsi->seg->null_pos] & ftsi->seg->null_bit))
   {
-      ftsi->pos=0;
-      DBUG_RETURN(1);
+    ftsi->pos=0;
+    DBUG_RETURN(1);
   }
   ftsi->pos= ftsi->rec+ftsi->seg->start;
   if (ftsi->seg->flag & HA_VAR_LENGTH_PART)
   {
-    ftsi->len=uint2korr(ftsi->pos);
-    ftsi->pos+=2;				 /* Skip VARCHAR length */
+    uint pack_length= (ftsi->seg->bit_start);
+    ftsi->len= (pack_length == 1 ? (uint) *(uchar*) ftsi->pos :
+                uint2korr(ftsi->pos)); 
+    ftsi->pos+= pack_length;			 /* Skip VARCHAR length */
     DBUG_RETURN(1);
   }
   if (ftsi->seg->flag & HA_BLOB_PART)
@@ -296,9 +294,11 @@ uint _ft_make_key(MI_INFO *info, uint keynr, byte *keybuf, FT_WORD *wptr,
   DBUG_RETURN(_mi_make_key(info,keynr,(uchar*) keybuf,buf,filepos));
 }
 
+
 /*
   convert key value to ft2
 */
+
 uint _mi_ft_convert_to_ft2(MI_INFO *info, uint keynr, uchar *key)
 {
   my_off_t root;
@@ -316,9 +316,12 @@ uint _mi_ft_convert_to_ft2(MI_INFO *info, uint keynr, uchar *key)
 
   get_key_full_length_rdonly(key_length, key);
   while (_mi_ck_delete(info, keynr, key, key_length) == 0)
-    /* nothing to do here.
-       _mi_ck_delete() will populate info->ft1_to_ft2 with deleted keys
-     */;
+  {
+    /*
+      nothing to do here.
+      _mi_ck_delete() will populate info->ft1_to_ft2 with deleted keys
+     */
+  }
 
   /* creating pageful of keys */
   mi_putint(info->buff,length+2,0);
diff --git a/myisam/mi_check.c b/myisam/mi_check.c
index 112a371c9fe..b8f992dc21a 100644
--- a/myisam/mi_check.c
+++ b/myisam/mi_check.c
@@ -280,7 +280,8 @@ int chk_size(MI_CHECK *param, register MI_INFO *info)
   size=my_seek(info->s->kfile,0L,MY_SEEK_END,MYF(0));
   if ((skr=(my_off_t) info->state->key_file_length) != size)
   {
-    if (skr > size)
+    /* Don't give error if file generated by myisampack */
+    if (skr > size && info->s->state.key_map)
     {
       error=1;
       mi_check_print_error(param,
diff --git a/myisam/mi_checksum.c b/myisam/mi_checksum.c
index 95338434211..33a51068fb0 100644
--- a/myisam/mi_checksum.c
+++ b/myisam/mi_checksum.c
@@ -40,8 +40,12 @@ ha_checksum mi_checksum(MI_INFO *info, const byte *buf)
     }
     case FIELD_VARCHAR:
     {
-      length=uint2korr(buf);
-      pos=buf+2;
+      uint pack_length= HA_VARCHAR_PACKLENGTH(rec->length-1);
+      if (pack_length == 1)
+        length= (ulong) *(uchar*) buf;
+      else
+        length= uint2korr(buf);
+      pos= buf+pack_length;
       break;
     }
     default:
diff --git a/myisam/mi_create.c b/myisam/mi_create.c
index e139997e0c7..0164555272d 100644
--- a/myisam/mi_create.c
+++ b/myisam/mi_create.c
@@ -43,7 +43,7 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
   myf create_flag;
   uint fields,length,max_key_length,packed,pointer,real_length_diff,
        key_length,info_length,key_segs,options,min_key_length_skip,
-       base_pos,varchar_count,long_varchar_count,varchar_length,
+       base_pos,long_varchar_count,varchar_length,
        max_key_block_length,unique_key_parts,fulltext_keys,offset;
   ulong reclength, real_reclength,min_pack_length;
   char filename[FN_REFLEN],linkname[FN_REFLEN], *linkname_ptr;
@@ -99,7 +99,7 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
 
 	/* Start by checking fields and field-types used */
 
-  reclength=varchar_count=varchar_length=long_varchar_count=packed=
+  reclength=varchar_length=long_varchar_count=packed=
     min_pack_length=pack_reclength=0;
   for (rec=recinfo, fields=0 ;
        fields != columns ;
@@ -130,14 +130,15 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
       }
       else if (type == FIELD_VARCHAR)
       {
-	varchar_count++;
-	varchar_length+=rec->length-2;
+	varchar_length+= rec->length-1;          /* Used for min_pack_length */
 	packed--;
-	pack_reclength+=1;
-	if (test(rec->length > 257))
-	{					/* May be packed on 3 bytes */
+	pack_reclength++;
+        min_pack_length++;
+        /* We must test for 257 as length includes pack-length */
+ 	if (test(rec->length >= 257))
+	{
 	  long_varchar_count++;
-	  pack_reclength+=2;
+	  pack_reclength+= 2;			/* May be packed on 3 bytes */
 	}
       }
       else if (type != FIELD_SKIP_ZERO)
@@ -169,12 +170,8 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
   /* We can't use checksum with static length rows */
   if (!(options & HA_OPTION_PACK_RECORD))
     options&= ~HA_OPTION_CHECKSUM;
-  if (options & (HA_OPTION_PACK_RECORD | HA_OPTION_COMPRESS_RECORD))
-    min_pack_length+=varchar_count;		/* Min length to pack */
-  else
-  {
-    min_pack_length+=varchar_length+2*varchar_count;
-  }
+  if (!(options & (HA_OPTION_PACK_RECORD | HA_OPTION_COMPRESS_RECORD)))
+    min_pack_length+= varchar_length;
   if (flags & HA_CREATE_TMP_TABLE)
     options|= HA_OPTION_TMP_TABLE;
   if (flags & HA_CREATE_CHECKSUM || (options & HA_OPTION_CHECKSUM))
@@ -220,7 +217,7 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
       reclength=pointer+1;		/* reserve place for delete link */
   }
   else
-    reclength+=long_varchar_count;	/* We need space for this! */
+    reclength+= long_varchar_count;	/* We need space for varchar! */
 
   max_key_length=0; tot_length=0 ; key_segs=0;
   fulltext_keys=0;
@@ -261,7 +258,8 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
 	   j++, keyseg++)
       {
         if (keyseg->type != HA_KEYTYPE_BINARY &&
-	    keyseg->type != HA_KEYTYPE_VARBINARY)
+	    keyseg->type != HA_KEYTYPE_VARBINARY1 &&
+            keyseg->type != HA_KEYTYPE_VARBINARY2)
         {
           my_errno=HA_WRONG_CREATE_OPTION;
           goto err;
@@ -285,11 +283,22 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
 	   j++, keyseg++)
       {
         if (keyseg->type != HA_KEYTYPE_TEXT &&
-	    keyseg->type != HA_KEYTYPE_VARTEXT)
+	    keyseg->type != HA_KEYTYPE_VARTEXT1 &&
+            keyseg->type != HA_KEYTYPE_VARTEXT2)
         {
           my_errno=HA_WRONG_CREATE_OPTION;
           goto err;
         }
+        if (!(keyseg->flag & HA_BLOB_PART) &&
+	    (keyseg->type == HA_KEYTYPE_VARTEXT1 ||
+             keyseg->type == HA_KEYTYPE_VARTEXT2))
+        {
+          /* Make a flag that this is a VARCHAR */
+          keyseg->flag|= HA_VAR_LENGTH_PART;
+          /* Store in bit_start number of bytes used to pack the length */
+          keyseg->bit_start= ((keyseg->type == HA_KEYTYPE_VARTEXT1)?
+                              1 : 2);
+        }
       }
 
       fulltext_keys++;
@@ -345,10 +354,19 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
 	case HA_KEYTYPE_INT8:
 	  keyseg->flag|= HA_SWAP_KEY;
           break;
-        case HA_KEYTYPE_VARTEXT:
-        case HA_KEYTYPE_VARBINARY:
+        case HA_KEYTYPE_VARTEXT1:
+        case HA_KEYTYPE_VARTEXT2:
+        case HA_KEYTYPE_VARBINARY1:
+        case HA_KEYTYPE_VARBINARY2:
           if (!(keyseg->flag & HA_BLOB_PART))
+          {
+            /* Make a flag that this is a VARCHAR */
             keyseg->flag|= HA_VAR_LENGTH_PART;
+            /* Store in bit_start number of bytes used to pack the length */
+            keyseg->bit_start= ((keyseg->type == HA_KEYTYPE_VARTEXT1 ||
+                                 keyseg->type == HA_KEYTYPE_VARBINARY1) ?
+                                1 : 2);
+          }
           break;
 	default:
 	  break;
@@ -368,6 +386,8 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
 	}
 	if (keyseg->flag & (HA_VAR_LENGTH_PART | HA_BLOB_PART))
 	{
+          DBUG_ASSERT(!test_all_bits(keyseg->flag,
+                                    (HA_VAR_LENGTH_PART | HA_BLOB_PART)));
 	  keydef->flag|=HA_VAR_LENGTH_KEY;
 	  length++;				/* At least one length byte */
 	  options|=HA_OPTION_PACK_KEYS;		/* Using packed keys */
@@ -646,11 +666,31 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
   /* Save unique definition */
   for (i=0 ; i < share.state.header.uniques ; i++)
   {
+    HA_KEYSEG *keyseg_end;
+    keyseg= uniquedefs[i].seg;
     if (mi_uniquedef_write(file, &uniquedefs[i]))
       goto err;
-    for (j=0 ; j < uniquedefs[i].keysegs ; j++)
+    for (keyseg= uniquedefs[i].seg, keyseg_end= keyseg+ uniquedefs[i].keysegs;
+         keyseg < keyseg_end;
+         keyseg++)
     {
-      if (mi_keyseg_write(file, &uniquedefs[i].seg[j]))
+      switch (keyseg->type) {
+      case HA_KEYTYPE_VARTEXT1:
+      case HA_KEYTYPE_VARTEXT2:
+      case HA_KEYTYPE_VARBINARY1:
+      case HA_KEYTYPE_VARBINARY2:
+        if (!(keyseg->flag & HA_BLOB_PART))
+        {
+          keyseg->flag|= HA_VAR_LENGTH_PART;
+          keyseg->bit_start= ((keyseg->type == HA_KEYTYPE_VARTEXT1 ||
+                               keyseg->type == HA_KEYTYPE_VARBINARY1) ?
+                              1 : 2);
+        }
+        break;
+      default:
+        break;
+      }
+      if (mi_keyseg_write(file, keyseg))
 	goto err;
     }
   }
diff --git a/myisam/mi_dbug.c b/myisam/mi_dbug.c
index 02d1c7d05d6..531d0b9ddba 100644
--- a/myisam/mi_dbug.c
+++ b/myisam/mi_dbug.c
@@ -132,8 +132,10 @@ void _mi_print_key(FILE *stream, register HA_KEYSEG *keyseg,
       break;
     }
 #endif
-    case HA_KEYTYPE_VARTEXT:			/* VARCHAR and TEXT */
-    case HA_KEYTYPE_VARBINARY:			/* VARBINARY and BLOB */
+    case HA_KEYTYPE_VARTEXT1:                   /* VARCHAR and TEXT */
+    case HA_KEYTYPE_VARTEXT2:                   /* VARCHAR and TEXT */
+    case HA_KEYTYPE_VARBINARY1:                 /* VARBINARY and BLOB */
+    case HA_KEYTYPE_VARBINARY2:                 /* VARBINARY and BLOB */
     {
       uint tmp_length;
       get_key_length(tmp_length,key);
diff --git a/myisam/mi_dynrec.c b/myisam/mi_dynrec.c
index 0b8d3c97872..9d8e161b8fe 100644
--- a/myisam/mi_dynrec.c
+++ b/myisam/mi_dynrec.c
@@ -768,11 +768,21 @@ uint _mi_rec_pack(MI_INFO *info, register byte *to, register const byte *from)
       }
       else if (type == FIELD_VARCHAR)
       {
-	uint tmp_length=uint2korr(from);
-	store_key_length_inc(to,tmp_length);
-	memcpy(to,from+2,tmp_length);
-	to+=tmp_length;
-	continue;
+        uint pack_length= HA_VARCHAR_PACKLENGTH(rec->length -1);
+	uint tmp_length;
+        if (pack_length == 1)
+        {
+          tmp_length= (uint) *(uchar*) from;
+          *to++= *from;
+        }
+        else
+        {
+          tmp_length= uint2korr(from);
+          store_key_length_inc(to,tmp_length);
+        }
+        memcpy(to, from+pack_length,tmp_length);
+        to+= tmp_length;
+        continue;
       }
       else
       {
@@ -878,9 +888,20 @@ my_bool _mi_rec_check(MI_INFO *info,const char *record, byte *rec_buff,
       }
       else if (type == FIELD_VARCHAR)
       {
-	uint tmp_length=uint2korr(record);
-	to+=get_pack_length(tmp_length)+tmp_length;
-	continue;
+        uint pack_length= HA_VARCHAR_PACKLENGTH(rec->length -1);
+	uint tmp_length;
+        if (pack_length == 1)
+        {
+          tmp_length= (uint) *(uchar*) record;
+          to+= 1+ tmp_length;
+          continue;
+        }
+        else
+        {
+          tmp_length= uint2korr(record);
+          to+= get_pack_length(tmp_length)+tmp_length;
+        }
+        continue;
       }
       else
       {
@@ -894,9 +915,7 @@ my_bool _mi_rec_check(MI_INFO *info,const char *record, byte *rec_buff,
       }
     }
     else
-    {
-      to+=length;
-    }
+      to+= length;
   }
   if (packed_length != (uint) (to - rec_buff) + test(info->s->calc_checksum) ||
       (bit != 1 && (flag & ~(bit - 1))))
@@ -947,13 +966,27 @@ ulong _mi_rec_unpack(register MI_INFO *info, register byte *to, byte *from,
     {
       if (type == FIELD_VARCHAR)
       {
-	get_key_length(length,from);
-	if (length > rec_length-2)
-	  goto err;
-	int2store(to,length);
-	memcpy(to+2,from,length);
-	from+=length;
-	continue;
+        uint pack_length= HA_VARCHAR_PACKLENGTH(rec_length-1);
+        if (pack_length == 1)
+        {
+          length= (uint) *(uchar*) from;
+          if (length > rec_length-1)
+            goto err;
+          *to= *from++;
+        }
+        else
+        {
+          get_key_length(length, from);
+          if (length > rec_length-2)
+            goto err;
+          int2store(to,length);
+        }
+        if (from+length > from_end)
+          goto err;
+        memcpy(to+pack_length, from, length);
+        from+= length;
+        min_pack_length--;
+        continue;
       }
       if (flag & bit)
       {
@@ -1021,15 +1054,17 @@ ulong _mi_rec_unpack(register MI_INFO *info, register byte *to, byte *from,
       if (min_pack_length > (uint) (from_end - from))
 	goto err;
       min_pack_length-=rec_length;
-      memcpy(to,(byte*) from,(size_t) rec_length); from+=rec_length;
+      memcpy(to, (byte*) from, (size_t) rec_length);
+      from+=rec_length;
     }
   }
   if (info->s->calc_checksum)
     from++;
   if (to == to_end && from == from_end && (bit == 1 || !(flag & ~(bit-1))))
     DBUG_RETURN(found_length);
+
 err:
-  my_errno=HA_ERR_RECORD_DELETED;
+  my_errno= HA_ERR_WRONG_IN_RECORD;
   DBUG_PRINT("error",("to_end: %lx -> %lx  from_end: %lx -> %lx",
 		      to,to_end,from,from_end));
   DBUG_DUMP("from",(byte*) info->rec_buff,info->s->base.min_pack_length);
diff --git a/myisam/mi_key.c b/myisam/mi_key.c
index c1400033254..caca63452b0 100644
--- a/myisam/mi_key.c
+++ b/myisam/mi_key.c
@@ -34,10 +34,20 @@
 
 static int _mi_put_key_in_record(MI_INFO *info,uint keynr,byte *record);
 
-	/*
-	** Make a intern key from a record
-	** Ret: Length of key
-	*/
+/*
+  Make a intern key from a record
+
+  SYNOPSIS
+    _mi_make_key()
+    info		MyiSAM handler
+    keynr		key number
+    key			Store created key here
+    record		Record
+    filepos		Position to record in the data file
+
+  RETURN
+    Length of key
+*/
 
 uint _mi_make_key(register MI_INFO *info, uint keynr, uchar *key,
 		  const byte *record, my_off_t filepos)
@@ -117,8 +127,10 @@ uint _mi_make_key(register MI_INFO *info, uint keynr, uchar *key,
     }
     if (keyseg->flag & HA_VAR_LENGTH_PART)
     {
-      uint tmp_length=uint2korr(pos);
-      pos+=2;					/* Skip VARCHAR length */
+      uint pack_length= keyseg->bit_start;
+      uint tmp_length= (pack_length == 1 ? (uint) *(uchar*) pos :
+                        uint2korr(pos));
+      pos+= pack_length;			/* Skip VARCHAR length */
       set_if_smaller(length,tmp_length);
       FIX_LENGTH(cs, pos, length, char_length);
       store_key_length_inc(key,char_length);
@@ -398,9 +410,12 @@ static int _mi_put_key_in_record(register MI_INFO *info, uint keynr,
 	goto err;
 #endif
       /* Store key length */
-      int2store(record+keyseg->start, length);
+      if (keyseg->bit_start == 1)
+        *(uchar*) (record+keyseg->start)= (uchar) length;
+      else
+        int2store(record+keyseg->start, length);
       /* And key data */
-      memcpy(record+keyseg->start+2,(byte*) key, length);
+      memcpy(record+keyseg->start + keyseg->bit_start, (byte*) key, length);
       key+= length;
     }
     else if (keyseg->flag & HA_BLOB_PART)
diff --git a/myisam/mi_open.c b/myisam/mi_open.c
index 615e6bb5dd1..58db2e47c1f 100644
--- a/myisam/mi_open.c
+++ b/myisam/mi_open.c
@@ -314,7 +314,9 @@ MI_INFO *mi_open(const char *name, int mode, uint open_flags)
 	{
 	  disk_pos=mi_keyseg_read(disk_pos, pos);
 
-	  if (pos->type == HA_KEYTYPE_TEXT || pos->type == HA_KEYTYPE_VARTEXT)
+	  if (pos->type == HA_KEYTYPE_TEXT ||
+              pos->type == HA_KEYTYPE_VARTEXT1 ||
+              pos->type == HA_KEYTYPE_VARTEXT2)
 	  {
 	    if (!pos->language)
 	      pos->charset=default_charset_info;
@@ -389,7 +391,9 @@ MI_INFO *mi_open(const char *name, int mode, uint open_flags)
 	for (j=0 ; j < share->uniqueinfo[i].keysegs; j++,pos++)
 	{
 	  disk_pos=mi_keyseg_read(disk_pos, pos);
-	  if (pos->type == HA_KEYTYPE_TEXT || pos->type == HA_KEYTYPE_VARTEXT)
+	  if (pos->type == HA_KEYTYPE_TEXT ||
+              pos->type == HA_KEYTYPE_VARTEXT1 ||
+              pos->type == HA_KEYTYPE_VARTEXT2)
 	  {
 	    if (!pos->language)
 	      pos->charset=default_charset_info;
diff --git a/myisam/mi_packrec.c b/myisam/mi_packrec.c
index a277c2ca9d1..62d15c03266 100644
--- a/myisam/mi_packrec.c
+++ b/myisam/mi_packrec.c
@@ -91,8 +91,10 @@ static void uf_zero(MI_COLUMNDEF *rec,MI_BIT_BUFF *bit_buff,
 		    uchar *to,uchar *end);
 static void uf_blob(MI_COLUMNDEF *rec, MI_BIT_BUFF *bit_buff,
 		    uchar *to, uchar *end);
-static void uf_varchar(MI_COLUMNDEF *rec, MI_BIT_BUFF *bit_buff,
-		       uchar *to, uchar *end);
+static void uf_varchar1(MI_COLUMNDEF *rec, MI_BIT_BUFF *bit_buff,
+                        uchar *to, uchar *end);
+static void uf_varchar2(MI_COLUMNDEF *rec, MI_BIT_BUFF *bit_buff,
+                        uchar *to, uchar *end);
 static void decode_bytes(MI_COLUMNDEF *rec,MI_BIT_BUFF *bit_buff,
 			 uchar *to,uchar *end);
 static uint decode_pos(MI_BIT_BUFF *bit_buff,MI_DECODE_TREE *decode_tree);
@@ -522,14 +524,16 @@ static void (*get_unpack_function(MI_COLUMNDEF *rec))
   case FIELD_BLOB:
     return &uf_blob;
   case FIELD_VARCHAR:
-    return &uf_varchar;
+    if (rec->length <= 256)                      /* 255 + 1 byte length */
+      return &uf_varchar1;
+    return &uf_varchar2;
   case FIELD_LAST:
   default:
     return 0;			/* This should never happend */
   }
 }
 
-	/* De different functions to unpack a field */
+	/* The different functions to unpack a field */
 
 static void uf_zerofill_skip_zero(MI_COLUMNDEF *rec, MI_BIT_BUFF *bit_buff,
 				   uchar *to, uchar *end)
@@ -773,7 +777,22 @@ static void uf_blob(MI_COLUMNDEF *rec, MI_BIT_BUFF *bit_buff,
   }
 }
 
-static void uf_varchar(MI_COLUMNDEF *rec, MI_BIT_BUFF *bit_buff,
+
+static void uf_varchar1(MI_COLUMNDEF *rec, MI_BIT_BUFF *bit_buff,
+		       uchar *to, uchar *end __attribute__((unused)))
+{
+  if (get_bit(bit_buff))
+    to[0]= 0;				/* Zero lengths */
+  else
+  {
+    ulong length=get_bits(bit_buff,rec->space_length_bits);
+    *to= (uchar) length;
+    decode_bytes(rec,bit_buff,to+1,to+1+length);
+  }
+}
+
+
+static void uf_varchar2(MI_COLUMNDEF *rec, MI_BIT_BUFF *bit_buff,
 		       uchar *to, uchar *end __attribute__((unused)))
 {
   if (get_bit(bit_buff))
diff --git a/myisam/mi_search.c b/myisam/mi_search.c
index 2fef70db9f0..2259dd17fcd 100644
--- a/myisam/mi_search.c
+++ b/myisam/mi_search.c
@@ -425,7 +425,8 @@ int _mi_prefix_search(MI_INFO *info, register MI_KEYDEF *keyinfo, uchar *page,
         if (len < cmplen)
         {
 	  if ((keyinfo->seg->type != HA_KEYTYPE_TEXT &&
-	       keyinfo->seg->type != HA_KEYTYPE_VARTEXT))
+	       keyinfo->seg->type != HA_KEYTYPE_VARTEXT1 &&
+               keyinfo->seg->type != HA_KEYTYPE_VARTEXT2))
 	    my_flag= -1;
 	  else
 	  {
@@ -1371,7 +1372,8 @@ _mi_calc_var_pack_key_length(MI_KEYDEF *keyinfo,uint nod_flag,uchar *next_key,
   sort_order=0;
   if ((keyinfo->flag & HA_FULLTEXT) &&
       ((keyseg->type == HA_KEYTYPE_TEXT) ||
-       (keyseg->type == HA_KEYTYPE_VARTEXT)) &&
+       (keyseg->type == HA_KEYTYPE_VARTEXT1) ||
+       (keyseg->type == HA_KEYTYPE_VARTEXT2)) &&
       !use_strnxfrm(keyseg->charset))
     sort_order=keyseg->charset->sort_order;
 
diff --git a/myisam/mi_test1.c b/myisam/mi_test1.c
index 15ce7515ac2..aa6cd98ac8e 100644
--- a/myisam/mi_test1.c
+++ b/myisam/mi_test1.c
@@ -75,11 +75,11 @@ static int run_test(const char *filename)
   recinfo[1].length= (key_field == FIELD_BLOB ? 4+mi_portable_sizeof_char_ptr :
 		      key_length);
   if (key_field == FIELD_VARCHAR)
-    recinfo[1].length+=2;
+    recinfo[1].length+= HA_VARCHAR_PACKLENGTH(key_length);;
   recinfo[2].type=extra_field;
   recinfo[2].length= (extra_field == FIELD_BLOB ? 4 + mi_portable_sizeof_char_ptr : 24);
   if (extra_field == FIELD_VARCHAR)
-    recinfo[2].length+=2;
+    recinfo[2].length+= HA_VARCHAR_PACKLENGTH(recinfo[2].length);
   if (opt_unique)
   {
     recinfo[3].type=FIELD_CHECK;
@@ -88,6 +88,9 @@ static int run_test(const char *filename)
   rec_length=recinfo[0].length+recinfo[1].length+recinfo[2].length+
     recinfo[3].length;
 
+  if (key_type == HA_KEYTYPE_VARTEXT1 &&
+      key_length > 255)
+    key_type= HA_KEYTYPE_VARTEXT2;
 
   /* Define a key over the first column */
   keyinfo[0].seg=keyseg;
@@ -330,7 +333,8 @@ static void create_key_part(char *key,uint rownr)
   {
     sprintf(key,"%*d",keyinfo[0].seg[0].length,rownr);
   }
-  else if (keyinfo[0].seg[0].type == HA_KEYTYPE_VARTEXT)
+  else if (keyinfo[0].seg[0].type == HA_KEYTYPE_VARTEXT1 ||
+           keyinfo[0].seg[0].type == HA_KEYTYPE_VARTEXT2)
   {						/* Alpha record */
     /* Create a key that may be easily packed */
     bfill(key,keyinfo[0].seg[0].length,rownr < 10 ? 'A' : 'B');
@@ -410,11 +414,14 @@ static void create_record(char *record,uint rownr)
   }
   else if (recinfo[1].type == FIELD_VARCHAR)
   {
-    uint tmp;
-    create_key_part(pos+2,rownr);
-    tmp=strlen(pos+2);
-    int2store(pos,tmp);
-    pos+=recinfo[1].length;
+    uint tmp, pack_length= HA_VARCHAR_PACKLENGTH(recinfo[1].length-1);
+    create_key_part(pos+pack_length,rownr);
+    tmp= strlen(pos+pack_length);
+    if (pack_length == 1)
+      *(uchar*) pos= (uchar) tmp;
+    else
+      int2store(pos,tmp);
+    pos+= recinfo[1].length;
   }
   else
   {
@@ -434,10 +441,13 @@ static void create_record(char *record,uint rownr)
   }
   else if (recinfo[2].type == FIELD_VARCHAR)
   {
-    uint tmp;
-    sprintf(pos+2,"... row: %d", rownr);
-    tmp=strlen(pos+2);
-    int2store(pos,tmp);
+    uint tmp, pack_length= HA_VARCHAR_PACKLENGTH(recinfo[1].length-1);
+    sprintf(pos+pack_length, "... row: %d", rownr);
+    tmp= strlen(pos+pack_length);
+    if (pack_length == 1)
+      *(uchar*) pos= (uchar) tmp;
+    else
+      int2store(pos,tmp);
   }
   else
   {
@@ -466,8 +476,9 @@ static void update_record(char *record)
   }
   else if (recinfo[1].type == FIELD_VARCHAR)
   {
-    uint length=uint2korr(pos);
-    my_casedn(default_charset_info,pos+2,length);
+    uint pack_length= HA_VARCHAR_PACKLENGTH(recinfo[1].length-1);
+    uint length= pack_length == 1 ? (uint) *(uchar*) pos : uint2korr(pos);
+    my_casedn(default_charset_info,pos+pack_length,length);
     pos+=recinfo[1].length;
   }
   else
@@ -493,10 +504,14 @@ static void update_record(char *record)
   else if (recinfo[2].type == FIELD_VARCHAR)
   {
     /* Second field is longer than 10 characters */
-    uint length=uint2korr(pos);
-    bfill(pos+2+length,recinfo[2].length-length-2,'.');
-    length=recinfo[2].length-2;
-    int2store(pos,length);
+    uint pack_length= HA_VARCHAR_PACKLENGTH(recinfo[1].length-1);
+    uint length= pack_length == 1 ? (uint) *(uchar*) pos : uint2korr(pos);
+    bfill(pos+pack_length+length,recinfo[2].length-length-pack_length,'.');
+    length=recinfo[2].length-pack_length;
+    if (pack_length == 1)
+      *(uchar*) pos= (uchar) length;
+    else
+      int2store(pos,length);
   }
   else
   {
@@ -519,7 +534,7 @@ static struct my_option my_long_options[] =
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"insert_rows", 'i', "Undocumented", (gptr*) &insert_count,
    (gptr*) &insert_count, 0, GET_UINT, REQUIRED_ARG, 1000, 0, 0, 0, 0, 0},
-  {"key_alpha", 'a', "Undocumented",
+  {"key_alpha", 'a', "Use a key of type HA_KEYTYPE_TEXT",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"key_binary_pack", 'B', "Undocumented",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
@@ -535,9 +550,9 @@ static struct my_option my_long_options[] =
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"key_space_pack", 'p', "Undocumented",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
-  {"key_varchar", 'w', "Undocumented",
+  {"key_varchar", 'w', "Test VARCHAR keys",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
-  {"null_fields", 'N', "Undocumented",
+  {"null_fields", 'N', "Define fields with NULL",
    (gptr*) &null_fields, (gptr*) &null_fields, 0, GET_BOOL, NO_ARG,
    0, 0, 0, 0, 0, 0},
   {"row_fixed_size", 'S', "Undocumented",
@@ -604,7 +619,7 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
     key_field=FIELD_BLOB;			/* blob key */
     extra_field= FIELD_BLOB;
     pack_seg|= HA_BLOB_PART;
-    key_type= HA_KEYTYPE_VARTEXT;
+    key_type= HA_KEYTYPE_VARTEXT1;
     break;
   case 'k':
     if (key_length < 4 || key_length > MI_MAX_KEY_LENGTH)
@@ -616,11 +631,11 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
   case 'w':
     key_field=FIELD_VARCHAR;			/* varchar keys */
     extra_field= FIELD_VARCHAR;
-    key_type= HA_KEYTYPE_VARTEXT;
+    key_type= HA_KEYTYPE_VARTEXT1;
     pack_seg|= HA_VAR_LENGTH_PART;
     create_flag|= HA_PACK_RECORD;
     break;
-  case 'K':				/* Use key cacheing */
+  case 'K':                                     /* Use key cacheing */
     key_cacheing=1;
     break;
   case 'V':
diff --git a/myisam/mi_test3.c b/myisam/mi_test3.c
index 27d23317b5c..be4277cc65c 100644
--- a/myisam/mi_test3.c
+++ b/myisam/mi_test3.c
@@ -67,6 +67,7 @@ int main(int argc,char **argv)
 
   bzero((char*) keyinfo,sizeof(keyinfo));
   bzero((char*) recinfo,sizeof(recinfo));
+  bzero((char*) keyseg,sizeof(keyseg));
   keyinfo[0].seg= &keyseg[0][0];
   keyinfo[0].seg[0].start=0;
   keyinfo[0].seg[0].length=8;
diff --git a/myisam/mi_test_all.res b/myisam/mi_test_all.res
index 94355bf1aa2..16b517d3f76 100644
--- a/myisam/mi_test_all.res
+++ b/myisam/mi_test_all.res
@@ -1,3 +1,6 @@
+myisamchk: MyISAM file test1
+myisamchk: warning: Size of indexfile is: 1024          Should be: 2048
+MyISAM-table 'test1' is usable but should be fixed
 mi_test2 -s -L -K -R1 -m2000 ;  Should give error 135
 Error: 135 in write at record: 1105
 got error: 135 when using MyISAM-database
@@ -5,46 +8,46 @@ myisamchk: MyISAM file test2
 myisamchk: warning: Datafile is almost full,      65532 of      65534 used
 MyISAM-table 'test2' is usable but should be fixed
 Commands   Used count    Errors   Recover errors
-open               17         0                0
-write             850         0                0
-update             85         0                0
-delete            850         0                0
-close              17         0                0
-extra             102         0                0
-Total            1921         0                0
+open                1         0                0
+write              50         0                0
+update              5         0                0
+delete             50         0                0
+close               1         0                0
+extra               6         0                0
+Total             113         0                0
 Commands   Used count    Errors   Recover errors
-open               18         0                0
-write             900         0                0
-update             90         0                0
-delete            900         0                0
-close              18         0                0
-extra             108         0                0
-Total            2034         0                0
+open                2         0                0
+write             100         0                0
+update             10         0                0
+delete            100         0                0
+close               2         0                0
+extra              12         0                0
+Total             226         0                0
 
-real	0m1.054s
-user	0m0.410s
-sys	0m0.640s
+real	0m0.791s
+user	0m0.137s
+sys	0m0.117s
 
-real	0m1.077s
-user	0m0.550s
-sys	0m0.530s
+real	0m0.659s
+user	0m0.252s
+sys	0m0.102s
 
-real	0m1.100s
-user	0m0.420s
-sys	0m0.680s
+real	0m0.571s
+user	0m0.188s
+sys	0m0.098s
 
-real	0m0.783s
-user	0m0.590s
-sys	0m0.200s
+real	0m1.111s
+user	0m0.236s
+sys	0m0.037s
 
-real	0m0.764s
-user	0m0.560s
-sys	0m0.210s
+real	0m0.621s
+user	0m0.242s
+sys	0m0.022s
 
-real	0m0.699s
-user	0m0.570s
-sys	0m0.130s
+real	0m0.698s
+user	0m0.248s
+sys	0m0.021s
 
-real	0m0.991s
-user	0m0.630s
-sys	0m0.350s
+real	0m0.683s
+user	0m0.265s
+sys	0m0.079s
diff --git a/myisam/mi_unique.c b/myisam/mi_unique.c
index c03182456df..f2d5f01be25 100644
--- a/myisam/mi_unique.c
+++ b/myisam/mi_unique.c
@@ -95,8 +95,10 @@ ha_checksum mi_unique_hash(MI_UNIQUEDEF *def, const byte *record)
     pos= record+keyseg->start;
     if (keyseg->flag & HA_VAR_LENGTH_PART)
     {
-      uint tmp_length=uint2korr(pos);
-      pos+=2;					/* Skip VARCHAR length */
+      uint pack_length=  keyseg->bit_start;
+      uint tmp_length= (pack_length == 1 ? (uint) *(uchar*) pos :
+                        uint2korr(pos));
+      pos+= pack_length;			/* Skip VARCHAR length */
       set_if_smaller(length,tmp_length);
     }
     else if (keyseg->flag & HA_BLOB_PART)
@@ -107,7 +109,8 @@ ha_checksum mi_unique_hash(MI_UNIQUEDEF *def, const byte *record)
 	length=tmp_length;			/* The whole blob */
     }
     end= pos+length;
-    if (type == HA_KEYTYPE_TEXT || type == HA_KEYTYPE_VARTEXT)
+    if (type == HA_KEYTYPE_TEXT || type == HA_KEYTYPE_VARTEXT1 ||
+        type == HA_KEYTYPE_VARTEXT2)
     {
       keyseg->charset->coll->hash_sort(keyseg->charset,
                                        (const uchar*) pos, length, &seed1,
@@ -157,12 +160,21 @@ int mi_unique_comp(MI_UNIQUEDEF *def, const byte *a, const byte *b,
     pos_b= b+keyseg->start;
     if (keyseg->flag & HA_VAR_LENGTH_PART)
     {
-      a_length= uint2korr(pos_a);
-      b_length= uint2korr(pos_b);
-      pos_a+= 2;				/* Skip VARCHAR length */
-      pos_b+= 2;
-      set_if_smaller(a_length, keyseg->length);
-      set_if_smaller(b_length, keyseg->length);
+      uint pack_length= keyseg->bit_start;
+      if (pack_length == 1)
+      {
+        a_length= (uint) *(uchar*) pos_a++;
+        b_length= (uint) *(uchar*) pos_b++;
+      }
+      else
+      {
+        a_length= uint2korr(pos_a);
+        b_length= uint2korr(pos_b);
+        pos_a+= 2;				/* Skip VARCHAR length */
+        pos_b+= 2;
+      }
+      set_if_smaller(a_length, keyseg->length); /* Safety */
+      set_if_smaller(b_length, keyseg->length); /* safety */
     }
     else if (keyseg->flag & HA_BLOB_PART)
     {
@@ -182,7 +194,8 @@ int mi_unique_comp(MI_UNIQUEDEF *def, const byte *a, const byte *b,
       memcpy_fixed((byte*) &pos_a,pos_a+keyseg->bit_start,sizeof(char*));
       memcpy_fixed((byte*) &pos_b,pos_b+keyseg->bit_start,sizeof(char*));
     }
-    if (type == HA_KEYTYPE_TEXT || type == HA_KEYTYPE_VARTEXT)
+    if (type == HA_KEYTYPE_TEXT || type == HA_KEYTYPE_VARTEXT1 ||
+        type == HA_KEYTYPE_VARTEXT2)
     {
       if (mi_compare_text(keyseg->charset, (uchar *) pos_a, a_length,
                                            (uchar *) pos_b, b_length, 0, 1))
diff --git a/myisam/myisampack.c b/myisam/myisampack.c
index cc520847f70..bda620a594a 100644
--- a/myisam/myisampack.c
+++ b/myisam/myisampack.c
@@ -849,9 +849,11 @@ static int get_statistic(PACK_MRG_INFO *mrg,HUFF_COUNTS *huff_counts)
 	}
 	else if (count->field_type == FIELD_VARCHAR)
 	{
-	  length=uint2korr(start_pos);
-	  pos=start_pos+2;
-	  end_pos=start_pos+length;
+          uint pack_length= HA_VARCHAR_PACKLENGTH(count->field_length-1);
+	  length= (pack_length == 1 ? (uint) *(uchar*) start_pos :
+                   uint2korr(start_pos));
+	  pos= start_pos+pack_length;
+	  end_pos= pos+length;
 	  set_if_bigger(count->max_length,length);
 	}
 	if (count->field_length <= 8 &&
@@ -1833,17 +1835,19 @@ static int compress_isam_file(PACK_MRG_INFO *mrg, HUFF_COUNTS *huff_counts)
 	}
 	case FIELD_VARCHAR:
 	{
-	  ulong col_length= uint2korr(start_pos);
+          uint pack_length= HA_VARCHAR_PACKLENGTH(count->field_length-1);
+	  ulong col_length= (pack_length == 1 ? (uint) *(uchar*) start_pos :
+                             uint2korr(start_pos));
 	  if (!col_length)
 	  {
 	    write_bits(1,1);			/* Empty varchar */
 	  }
 	  else
 	  {
-	    byte *end=start_pos+2+col_length;
+	    byte *end=start_pos+pack_length+col_length;
 	    write_bits(0,1);
 	    write_bits(col_length,count->length_bits);
-	    for (start_pos+=2 ; start_pos < end ; start_pos++)
+	    for (start_pos+=pack_length ; start_pos < end ; start_pos++)
 	      write_bits(tree->code[(uchar) *start_pos],
 			 (uint) tree->code_len[(uchar) *start_pos]);
 	  }
diff --git a/mysql-test/include/varchar.inc b/mysql-test/include/varchar.inc
index 6c9b62065c5..32140bc7146 100644
--- a/mysql-test/include/varchar.inc
+++ b/mysql-test/include/varchar.inc
@@ -92,6 +92,66 @@ select sql_big_result c,count(t) from t1 group by c limit 10;
 select t,count(*) from t1 group by t limit 10;
 select t,count(t) from t1 group by t limit 10;
 select sql_big_result t,count(t) from t1 group by t limit 10;
+
+#
+# Test varchar > 255 bytes
+#
+
+alter table t1 modify v varchar(300), drop key v, drop key v_2, add key v (v);
+show create table t1;
+select count(*) from t1 where v='a';
+select count(*) from t1 where v='a  ';
+select count(*) from t1 where v between 'a' and 'a ';
+select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+select count(*) from t1 where v like 'a%';
+select count(*) from t1 where v like 'a %';
+explain select count(*) from t1 where v='a  ';
+explain select count(*) from t1 where v like 'a%';
+explain select count(*) from t1 where v between 'a' and 'a ';
+explain select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+explain select * from t1 where v='a';
+
+# GROUP BY
+
+select v,count(*) from t1 group by v limit 10;
+select v,count(t) from t1 group by v limit 10;
+select sql_big_result v,count(t) from t1 group by v limit 10;
+
+#
+# Test varchar > 255 bytes, key < 255
+#
+
+alter table t1 drop key v, add key v (v(30));
+show create table t1;
+select count(*) from t1 where v='a';
+select count(*) from t1 where v='a  ';
+select count(*) from t1 where v between 'a' and 'a ';
+select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+select count(*) from t1 where v like 'a%';
+select count(*) from t1 where v like 'a %';
+explain select count(*) from t1 where v='a  ';
+explain select count(*) from t1 where v like 'a%';
+explain select count(*) from t1 where v between 'a' and 'a ';
+explain select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+explain select * from t1 where v='a';
+
+# GROUP BY
+
+select v,count(*) from t1 group by v limit 10;
+select v,count(t) from t1 group by v limit 10;
+select sql_big_result v,count(t) from t1 group by v limit 10;
+
+#
+# Test varchar > 512 (special case for GROUP BY becasue of
+# CONVERT_IF_BIGGER_TO_BLOB define)
+#
+
+alter table t1 modify v varchar(600), drop key v, add key v (v);
+show create table t1;
+select v,count(*) from t1 group by v limit 10;
+select v,count(t) from t1 group by v limit 10;
+select sql_big_result v,count(t) from t1 group by v limit 10;
+
 drop table t1;
 
 #
diff --git a/mysql-test/r/bdb.result b/mysql-test/r/bdb.result
index 3570f74065e..337cc570298 100644
--- a/mysql-test/r/bdb.result
+++ b/mysql-test/r/bdb.result
@@ -1578,6 +1578,217 @@ f         	10
 g         	10
 h         	10
 i         	10
+alter table t1 modify v varchar(300), drop key v, drop key v_2, add key v (v);
+Warnings:
+Warning	1071	Specified key was too long; max key length is 255 bytes
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `v` varchar(300) default NULL,
+  `c` char(10) default NULL,
+  `t` text,
+  KEY `c` (`c`),
+  KEY `t` (`t`(10)),
+  KEY `v` (`v`(255))
+) ENGINE=BerkeleyDB DEFAULT CHARSET=latin1
+select count(*) from t1 where v='a';
+count(*)
+10
+select count(*) from t1 where v='a  ';
+count(*)
+10
+select count(*) from t1 where v between 'a' and 'a ';
+count(*)
+10
+select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+count(*)
+10
+select count(*) from t1 where v like 'a%';
+count(*)
+11
+select count(*) from t1 where v like 'a %';
+count(*)
+9
+explain select count(*) from t1 where v='a  ';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ref	v	v	258	const	10	Using where
+explain select count(*) from t1 where v like 'a%';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	258	NULL	11	Using where
+explain select count(*) from t1 where v between 'a' and 'a ';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	258	NULL	10	Using where
+explain select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	258	NULL	10	Using where
+explain select * from t1 where v='a';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ref	v	v	258	const	10	Using where
+select v,count(*) from t1 group by v limit 10;
+v	count(*)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select sql_big_result v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a         	10
+b         	10
+c         	10
+d         	10
+e         	10
+f         	10
+g         	10
+h         	10
+i         	10
+alter table t1 drop key v, add key v (v(30));
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `v` varchar(300) default NULL,
+  `c` char(10) default NULL,
+  `t` text,
+  KEY `c` (`c`),
+  KEY `t` (`t`(10)),
+  KEY `v` (`v`(30))
+) ENGINE=BerkeleyDB DEFAULT CHARSET=latin1
+select count(*) from t1 where v='a';
+count(*)
+10
+select count(*) from t1 where v='a  ';
+count(*)
+10
+select count(*) from t1 where v between 'a' and 'a ';
+count(*)
+10
+select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+count(*)
+10
+select count(*) from t1 where v like 'a%';
+count(*)
+11
+select count(*) from t1 where v like 'a %';
+count(*)
+9
+explain select count(*) from t1 where v='a  ';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ref	v	v	33	const	10	Using where
+explain select count(*) from t1 where v like 'a%';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	33	NULL	11	Using where
+explain select count(*) from t1 where v between 'a' and 'a ';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	33	NULL	10	Using where
+explain select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	33	NULL	10	Using where
+explain select * from t1 where v='a';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ref	v	v	33	const	10	Using where
+select v,count(*) from t1 group by v limit 10;
+v	count(*)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select sql_big_result v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a         	10
+b         	10
+c         	10
+d         	10
+e         	10
+f         	10
+g         	10
+h         	10
+i         	10
+alter table t1 modify v varchar(600), drop key v, add key v (v);
+Warnings:
+Warning	1071	Specified key was too long; max key length is 255 bytes
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `v` varchar(600) default NULL,
+  `c` char(10) default NULL,
+  `t` text,
+  KEY `c` (`c`),
+  KEY `t` (`t`(10)),
+  KEY `v` (`v`(255))
+) ENGINE=BerkeleyDB DEFAULT CHARSET=latin1
+select v,count(*) from t1 group by v limit 10;
+v	count(*)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select sql_big_result v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a         	10
+b         	10
+c         	10
+d         	10
+e         	10
+f         	10
+g         	10
+h         	10
+i         	10
 drop table t1;
 create table t1 (a char(10), unique (a));
 insert into t1 values ('a   ');
diff --git a/mysql-test/r/information_schema.result b/mysql-test/r/information_schema.result
index d421ac1d184..ea541c4d599 100644
--- a/mysql-test/r/information_schema.result
+++ b/mysql-test/r/information_schema.result
@@ -474,7 +474,7 @@ character_sets	CREATE TEMPORARY TABLE `character_sets` (
   `DEFAULT_COLLATE_NAME` varchar(60) NOT NULL default '',
   `DESCRIPTION` varchar(60) NOT NULL default '',
   `MAXLEN` bigint(3) NOT NULL default '0'
-) ENGINE=HEAP DEFAULT CHARSET=utf8 MAX_ROWS=2252
+) ENGINE=HEAP DEFAULT CHARSET=utf8 MAX_ROWS=2267
 set names latin2;
 SHOW CREATE TABLE INFORMATION_SCHEMA.character_sets;
 Table	Create Table
@@ -483,7 +483,7 @@ character_sets	CREATE TEMPORARY TABLE `character_sets` (
   `DEFAULT_COLLATE_NAME` varchar(60) NOT NULL default '',
   `DESCRIPTION` varchar(60) NOT NULL default '',
   `MAXLEN` bigint(3) NOT NULL default '0'
-) ENGINE=HEAP DEFAULT CHARSET=utf8 MAX_ROWS=2252
+) ENGINE=HEAP DEFAULT CHARSET=utf8 MAX_ROWS=2267
 set names latin1;
 create table t1 select * from information_schema.CHARACTER_SETS
 where CHARACTER_SET_NAME like "latin1";
diff --git a/mysql-test/r/innodb.result b/mysql-test/r/innodb.result
index 9a6c69b7bea..fe99961f964 100644
--- a/mysql-test/r/innodb.result
+++ b/mysql-test/r/innodb.result
@@ -1421,19 +1421,19 @@ insert t2 select * from t1;
 insert t3 select * from t1;
 checksum table t1, t2, t3, t4 quick;
 Table	Checksum
-test.t1	272226711
+test.t1	2948697075
 test.t2	NULL
 test.t3	NULL
 test.t4	NULL
 checksum table t1, t2, t3, t4;
 Table	Checksum
-test.t1	272226711
+test.t1	2948697075
 test.t2	968604391
 test.t3	968604391
 test.t4	NULL
 checksum table t1, t2, t3, t4 extended;
 Table	Checksum
-test.t1	272226711
+test.t1	3092701434
 test.t2	968604391
 test.t3	968604391
 test.t4	NULL
diff --git a/mysql-test/r/myisam.result b/mysql-test/r/myisam.result
index 0074b2672fc..98020f26e37 100644
--- a/mysql-test/r/myisam.result
+++ b/mysql-test/r/myisam.result
@@ -513,18 +513,18 @@ insert t1 values (1, "aaa", "bbb"), (NULL, "", "ccccc"), (0, NULL, "");
 insert t2 select * from t1;
 checksum table t1, t2, t3 quick;
 Table	Checksum
-test.t1	272226711
+test.t1	2948697075
 test.t2	NULL
 test.t3	NULL
 checksum table t1, t2, t3;
 Table	Checksum
-test.t1	272226711
-test.t2	272226711
+test.t1	2948697075
+test.t2	3092701434
 test.t3	NULL
 checksum table t1, t2, t3 extended;
 Table	Checksum
-test.t1	272226711
-test.t2	272226711
+test.t1	3092701434
+test.t2	3092701434
 test.t3	NULL
 drop table t1,t2;
 create table t1 (a int, key (a));
@@ -849,6 +849,213 @@ f	10
 g	10
 h	10
 i	10
+alter table t1 modify v varchar(300), drop key v, drop key v_2, add key v (v);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `v` varchar(300) default NULL,
+  `c` char(10) default NULL,
+  `t` text,
+  KEY `c` (`c`),
+  KEY `t` (`t`(10)),
+  KEY `v` (`v`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+select count(*) from t1 where v='a';
+count(*)
+10
+select count(*) from t1 where v='a  ';
+count(*)
+10
+select count(*) from t1 where v between 'a' and 'a ';
+count(*)
+10
+select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+count(*)
+10
+select count(*) from t1 where v like 'a%';
+count(*)
+11
+select count(*) from t1 where v like 'a %';
+count(*)
+9
+explain select count(*) from t1 where v='a  ';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ref	v	v	303	const	7	Using where; Using index
+explain select count(*) from t1 where v like 'a%';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	303	NULL	8	Using where; Using index
+explain select count(*) from t1 where v between 'a' and 'a ';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	303	NULL	7	Using where; Using index
+explain select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	303	NULL	7	Using where; Using index
+explain select * from t1 where v='a';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ref	v	v	303	const	7	Using where
+select v,count(*) from t1 group by v limit 10;
+v	count(*)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select sql_big_result v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+alter table t1 drop key v, add key v (v(30));
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `v` varchar(300) default NULL,
+  `c` char(10) default NULL,
+  `t` text,
+  KEY `c` (`c`),
+  KEY `t` (`t`(10)),
+  KEY `v` (`v`(30))
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+select count(*) from t1 where v='a';
+count(*)
+10
+select count(*) from t1 where v='a  ';
+count(*)
+10
+select count(*) from t1 where v between 'a' and 'a ';
+count(*)
+10
+select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+count(*)
+10
+select count(*) from t1 where v like 'a%';
+count(*)
+11
+select count(*) from t1 where v like 'a %';
+count(*)
+9
+explain select count(*) from t1 where v='a  ';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ref	v	v	33	const	7	Using where
+explain select count(*) from t1 where v like 'a%';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	33	NULL	8	Using where
+explain select count(*) from t1 where v between 'a' and 'a ';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	33	NULL	7	Using where
+explain select count(*) from t1 where v between 'a' and 'a ' and v between 'a  ' and 'b\n';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	range	v	v	33	NULL	7	Using where
+explain select * from t1 where v='a';
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ref	v	v	33	const	7	Using where
+select v,count(*) from t1 group by v limit 10;
+v	count(*)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select sql_big_result v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+alter table t1 modify v varchar(600), drop key v, add key v (v);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `v` varchar(600) default NULL,
+  `c` char(10) default NULL,
+  `t` text,
+  KEY `c` (`c`),
+  KEY `t` (`t`(10)),
+  KEY `v` (`v`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+select v,count(*) from t1 group by v limit 10;
+v	count(*)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
+select sql_big_result v,count(t) from t1 group by v limit 10;
+v	count(t)
+a
	1
+a	10
+b	10
+c	10
+d	10
+e	10
+f	10
+g	10
+h	10
+i	10
 drop table t1;
 create table t1 (a char(10), unique (a));
 insert into t1 values ('a   ');
diff --git a/mysql-test/r/ps_1general.result b/mysql-test/r/ps_1general.result
index 2019268f443..d963fd04109 100644
--- a/mysql-test/r/ps_1general.result
+++ b/mysql-test/r/ps_1general.result
@@ -278,7 +278,7 @@ t2	MyISAM	9	Fixed	0	0	0	64424509439	1024	0	NULL	#	#	#	latin1_swedish_ci	NULL
 prepare stmt4 from ' show table status from test like ''t9%'' ';
 execute stmt4;
 Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-t9	MyISAM	10	Dynamic	2	222	444	4294967295	2048	0	NULL	#	#	#	latin1_swedish_ci	NULL		
+t9	MyISAM	10	Dynamic	2	220	440	4294967295	2048	0	NULL	#	#	#	latin1_swedish_ci	NULL		
 prepare stmt4 from ' show status like ''Threads_running'' ';
 execute stmt4;
 Variable_name	Value
diff --git a/mysql-test/t/bdb.test b/mysql-test/t/bdb.test
index 01d82c0ba16..b319cf79ec2 100644
--- a/mysql-test/t/bdb.test
+++ b/mysql-test/t/bdb.test
@@ -824,7 +824,7 @@ select a from t1;
 drop table t1;
 
 #
-# bug#2686 - index_merge select on BerkeleyDB table with varchar PK causes mysqld to crash
+# bug#2686 - index_merge select on BerkeleyDB table with varchar PK crashes
 #
 
 create table t1(
@@ -842,7 +842,8 @@ select substring(pk1, 1, 4), substring(pk1, 4001),
 drop table t1;
 
 #
-# bug#2688 - Wrong index_merge query results for BDB table with variable length primary key
+# bug#2688 - Wrong index_merge query results for BDB table with
+# variable length primary key
 #
 
 create table t1 (
diff --git a/mysys/list.c b/mysys/list.c
index 64fca10dc0b..c3cd6c94b9f 100644
--- a/mysys/list.c
+++ b/mysys/list.c
@@ -28,7 +28,7 @@
 LIST *list_add(LIST *root, LIST *element)
 {
   DBUG_ENTER("list_add");
-  DBUG_PRINT("enter",("root: 0x%lx  element: %lx", root, element));
+  DBUG_PRINT("enter",("root: 0x%lx  element: 0x%lx", root, element));
   if (root)
   {
     if (root->prev)			/* If add in mid of list */
diff --git a/mysys/my_handler.c b/mysys/my_handler.c
index e4dc55964da..7c22f02fa2a 100644
--- a/mysys/my_handler.c
+++ b/mysys/my_handler.c
@@ -207,7 +207,8 @@ int ha_key_cmp(register HA_KEYSEG *keyseg, register uchar *a,
         b+=length;
       }
       break;
-    case HA_KEYTYPE_VARTEXT:
+    case HA_KEYTYPE_VARTEXT1:
+    case HA_KEYTYPE_VARTEXT2:
       {
         int a_length,b_length,pack_length;
         get_key_length(a_length,a);
@@ -229,7 +230,8 @@ int ha_key_cmp(register HA_KEYSEG *keyseg, register uchar *a,
         break;
       }
       break;
-    case HA_KEYTYPE_VARBINARY:
+    case HA_KEYTYPE_VARBINARY1:
+    case HA_KEYTYPE_VARBINARY2:
       {
         int a_length,b_length,pack_length;
         get_key_length(a_length,a);
diff --git a/mysys/raid.cc b/mysys/raid.cc
index 0b688464fb3..1d2e0cb01f0 100644
--- a/mysys/raid.cc
+++ b/mysys/raid.cc
@@ -185,7 +185,7 @@ extern "C" {
   uint my_raid_write(File fd,const byte *Buffer, uint Count, myf MyFlags)
   {
     DBUG_ENTER("my_raid_write");
-    DBUG_PRINT("enter",("Fd: %d  Buffer: %lx  Count: %u  MyFlags: %d",
+    DBUG_PRINT("enter",("Fd: %d  Buffer: 0x%lx  Count: %u  MyFlags: %d",
 		      fd, Buffer, Count, MyFlags));
     if (is_raid(fd))
     {
@@ -198,7 +198,7 @@ extern "C" {
   uint my_raid_read(File fd, byte *Buffer, uint Count, myf MyFlags)
   {
     DBUG_ENTER("my_raid_read");
-    DBUG_PRINT("enter",("Fd: %d  Buffer: %lx  Count: %u  MyFlags: %d",
+    DBUG_PRINT("enter",("Fd: %d  Buffer: 0x%lx  Count: %u  MyFlags: %d",
 		      fd, Buffer, Count, MyFlags));
     if (is_raid(fd))
     {
@@ -212,8 +212,9 @@ extern "C" {
 		     myf MyFlags)
   {
     DBUG_ENTER("my_raid_pread");
-    DBUG_PRINT("enter",("Fd: %d  Buffer: %lx  Count: %u offset: %u  MyFlags: %d",
-		      Filedes, Buffer, Count, offset, MyFlags));
+    DBUG_PRINT("enter",
+               ("Fd: %d  Buffer: 0x%lx  Count: %u offset: %u  MyFlags: %d",
+                Filedes, Buffer, Count, offset, MyFlags));
      if (is_raid(Filedes))
      {
        assert(offset != MY_FILEPOS_ERROR);
@@ -231,8 +232,9 @@ extern "C" {
 		      my_off_t offset, myf MyFlags)
   {
     DBUG_ENTER("my_raid_pwrite");
-    DBUG_PRINT("enter",("Fd: %d  Buffer: %lx  Count: %u offset: %u  MyFlags: %d",
-		      Filedes, Buffer, Count, offset, MyFlags));
+    DBUG_PRINT("enter",
+               ("Fd: %d  Buffer: 0x %lx  Count: %u offset: %u  MyFlags: %d",
+                Filedes, Buffer, Count, offset, MyFlags));
      if (is_raid(Filedes))
      {
        assert(offset != MY_FILEPOS_ERROR);
diff --git a/sql/field.cc b/sql/field.cc
index 7b555571850..ee12ce5ea68 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -4657,7 +4657,19 @@ Field *Field_string::new_field(MEM_ROOT *root, struct st_table *new_table)
 }
 
 /****************************************************************************
-** VARCHAR type  (Not available for the end user yet)
+  VARCHAR type
+  Data in field->ptr is stored as:
+    1 or 2 bytes length-prefix-header  (from Field_varstring::length_bytes)
+    data
+
+  NOTE:
+  When VARCHAR is stored in a key (for handler::index_read() etc) it's always
+  stored with a 2 byte prefix. (Just like blob keys).
+
+  Normally length_bytes is calculated as (field_length < 256 : 1 ? 2)
+  The exception is if there is a prefix key field that is part of a long
+  VARCHAR, in which case field_length for this may be 1 but the length_bytes
+  is 2.
 ****************************************************************************/
 
 
@@ -4686,8 +4698,11 @@ int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
 						    from,from+length,
 						    field_length/
 						    field_charset->mbmaxlen);
-  memcpy(ptr + HA_KEY_BLOB_LENGTH, from, copy_length);
-  int2store(ptr, copy_length);
+  memcpy(ptr + length_bytes, from, copy_length);
+  if (length_bytes == 1)
+    *ptr= (uchar) copy_length;
+  else
+    int2store(ptr, copy_length);
   
   if (copy_length < length)
     error= 1;
@@ -4700,91 +4715,117 @@ int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
 int Field_varstring::store(longlong nr)
 {
   char buff[64];
-  int  l;
-  CHARSET_INFO *cs=charset();
-  l= (cs->cset->longlong10_to_str)(cs,buff,sizeof(buff),-10,nr);
-  return Field_varstring::store(buff,(uint)l,cs);
+  uint  length;
+  length= (uint) (field_charset->cset->longlong10_to_str)(field_charset,
+                                                          buff,
+                                                          sizeof(buff),
+                                                          -10,nr);
+  return Field_varstring::store(buff, length, field_charset);
 }
 
 
 double Field_varstring::val_real(void)
 {
   int not_used;
-  uint length=uint2korr(ptr)+HA_KEY_BLOB_LENGTH;
-  CHARSET_INFO *cs=charset();
-  return my_strntod(cs, ptr+HA_KEY_BLOB_LENGTH, length, (char**)0, &not_used);
+  uint length= length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
+  return my_strntod(field_charset, ptr+length_bytes, length, (char**) 0,
+                    &not_used);
 }
 
 
 longlong Field_varstring::val_int(void)
 {
   int not_used;
-  uint length=uint2korr(ptr)+HA_KEY_BLOB_LENGTH;
-  CHARSET_INFO *cs=charset();
-  return my_strntoll(cs,ptr+HA_KEY_BLOB_LENGTH,length,10,NULL, &not_used);
+  uint length= length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
+  return my_strntoll(field_charset, ptr+length_bytes, length, 10, NULL,
+                     &not_used);
 }
 
 
 String *Field_varstring::val_str(String *val_buffer __attribute__((unused)),
 				 String *val_ptr)
 {
-  uint length=uint2korr(ptr);
-  val_ptr->set((const char*) ptr+HA_KEY_BLOB_LENGTH,length,field_charset);
+  uint length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
+  val_ptr->set((const char*) ptr+length_bytes, length, field_charset);
   return val_ptr;
 }
 
 
 int Field_varstring::cmp(const char *a_ptr, const char *b_ptr)
 {
-  uint a_length=uint2korr(a_ptr);
-  uint b_length=uint2korr(b_ptr);
+  uint a_length, b_length;
   int diff;
+
+  if (length_bytes == 1)
+  {
+    a_length= (uint) (uchar) *a_ptr;
+    b_length= (uint) (uchar) *b_ptr;
+  }
+  else
+  {
+    a_length= uint2korr(a_ptr);
+    b_length= uint2korr(b_ptr);
+  }
   diff= field_charset->coll->strnncollsp(field_charset,
                                          (const uchar*) a_ptr+
-                                         HA_KEY_BLOB_LENGTH,
+                                         length_bytes,
                                          a_length,
                                          (const uchar*) b_ptr+
-                                         HA_KEY_BLOB_LENGTH,
+                                         length_bytes,
                                          b_length,0);
   return diff;
 }
 
 
+/*
+  NOTE: varstring and blob keys are ALWAYS stored with a 2 byte length prefix
+*/
+
 int Field_varstring::key_cmp(const byte *key_ptr, uint max_key_length)
 {
   char *blob1;
-  uint length= uint2korr(ptr);
-  CHARSET_INFO *cs= charset();
-  uint char_length= max_key_length / cs->mbmaxlen;
+  uint length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
+  uint char_length= max_key_length / field_charset->mbmaxlen;
 
-  char_length= my_charpos(cs, ptr + HA_KEY_BLOB_LENGTH,
-                          ptr + HA_KEY_BLOB_LENGTH + length, char_length);
+  char_length= my_charpos(field_charset, ptr + length_bytes,
+                          ptr + length_bytes + length, char_length);
   set_if_smaller(length, char_length);
-  return cs->coll->strnncollsp(cs, 
-                               (const uchar*) ptr+2, length,
-                               (const uchar*) key_ptr+HA_KEY_BLOB_LENGTH,
-                               uint2korr(key_ptr), 0);
+  return field_charset->coll->strnncollsp(field_charset, 
+                                          (const uchar*) ptr + length_bytes,
+                                          length,
+                                          (const uchar*) key_ptr+
+                                          HA_KEY_BLOB_LENGTH,
+                                          uint2korr(key_ptr), 0);
 }
 
 
+/*
+  Compare to key segments (always 2 byte length prefix)
+
+  NOTE
+    This is used only to compare key segments created for index_read().
+    (keys are created and compared in key.cc)
+*/
+
 int Field_varstring::key_cmp(const byte *a,const byte *b)
 {
-  CHARSET_INFO *cs= charset();
-  return cs->coll->strnncollsp(cs,
-                               (const uchar*) a + HA_KEY_BLOB_LENGTH,
-                               uint2korr(a),
-                               (const uchar*) b + HA_KEY_BLOB_LENGTH,
-                               uint2korr(b),
-                               0);
+  return field_charset->coll->strnncollsp(field_charset,
+                                          (const uchar*) a +
+                                          HA_KEY_BLOB_LENGTH,
+                                          uint2korr(a),
+                                          (const uchar*) b +
+                                          HA_KEY_BLOB_LENGTH,
+                                          uint2korr(b),
+                                          0);
 }
 
 
 void Field_varstring::sort_string(char *to,uint length)
 {
-  uint tot_length= uint2korr(ptr);
+  uint tot_length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
   tot_length= my_strnxfrm(field_charset,
 			  (uchar*) to, length,
-			  (uchar*) ptr+HA_KEY_BLOB_LENGTH,
+			  (uchar*) ptr + length_bytes,
 			  tot_length);
   if (tot_length < length)
     field_charset->cset->fill(field_charset, to+tot_length,length-tot_length,
@@ -4792,6 +4833,18 @@ void Field_varstring::sort_string(char *to,uint length)
 }
 
 
+enum ha_base_keytype Field_varstring::key_type() const
+{
+  enum ha_base_keytype res;
+
+  if (binary())
+    res= length_bytes == 1 ? HA_KEYTYPE_VARBINARY1 : HA_KEYTYPE_VARBINARY2;
+  else
+    res= length_bytes == 1 ? HA_KEYTYPE_VARTEXT1 : HA_KEYTYPE_VARTEXT2;
+  return res;
+}
+
+
 void Field_varstring::sql_type(String &res) const
 {
   THD *thd= table->in_use;
@@ -4809,9 +4862,14 @@ void Field_varstring::sql_type(String &res) const
 }
 
 
+/*
+  Functions to create a packed row.
+  Here the number of length bytes are depending on the given max_length
+*/
+
 char *Field_varstring::pack(char *to, const char *from, uint max_length)
 {
-  uint length=uint2korr(from);
+  uint length= length_bytes == 1 ? (uint) (uchar) *from : uint2korr(from);
   set_if_smaller(max_length, field_length);
   if (length > max_length)
     length=max_length;
@@ -4819,39 +4877,104 @@ char *Field_varstring::pack(char *to, const char *from, uint max_length)
   if (max_length > 255)
     *to++= (char) (length >> 8);
   if (length)
-    memcpy(to, from+HA_KEY_BLOB_LENGTH, length);
+    memcpy(to, from+length_bytes, length);
   return to+length;
 }
 
 
-char *Field_varstring::pack_key(char *to, const char *from, uint max_length)
+char *Field_varstring::pack_key(char *to, const char *key, uint max_length)
 {
-  uint length=uint2korr(from);
+  uint length=  length_bytes == 1 ? (uint) (uchar) *key : uint2korr(key);
   uint char_length= ((field_charset->mbmaxlen > 1) ?
                      max_length/field_charset->mbmaxlen : max_length);
-  from+= HA_KEY_BLOB_LENGTH;
+  key+= length_bytes;
   if (length > char_length)
   {
-    char_length= my_charpos(field_charset, from, from+length, char_length);
+    char_length= my_charpos(field_charset, key, key+length, char_length);
     set_if_smaller(length, char_length);
   }
   *to++= (char) (length & 255);
   if (max_length > 255)
     *to++= (char) (length >> 8);
   if (length)
-    memcpy(to, from, length);
+    memcpy(to, key, length);
   return to+length;
 }
 
 
+/*
+  Unpack a key into a record buffer.
+
+  SYNOPSIS
+    unpack_key()
+    to                          Pointer into the record buffer.
+    key                         Pointer to the packed key.
+    max_length                  Key length limit from key description.
+
+  DESCRIPTION
+    A VARCHAR key has a maximum size of 64K-1.
+    In its packed form, the length field is one or two bytes long,
+    depending on 'max_length'.
+
+  RETURN
+    Pointer to end of 'key' (To the next key part if multi-segment key)
+*/
+
+const char *Field_varstring::unpack_key(char *to, const char *key,
+                                      uint max_length)
+{
+  /* get length of the blob key */
+  uint32 length= *((uchar*) key++);
+  if (max_length > 255)
+    length+= (*((uchar*) key++)) << 8;
+
+  /* put the length into the record buffer */
+  if (length_bytes == 1)
+    *ptr= (uchar) length;
+  else
+    int2store(ptr, length);
+  memcpy(ptr + length_bytes, key, length);
+  return key + length;
+}
+
+/*
+  Create a packed key that will be used for storage in the index tree
+
+  SYNOPSIS
+    pack_key_from_key_image()
+    to			Store packed key segment here
+    from		Key segment (as given to index_read())
+    max_length  	Max length of key
+
+  RETURN
+    end of key storage
+*/
+
+char *Field_varstring::pack_key_from_key_image(char *to, const char *from,
+                                               uint max_length)
+{
+  /* Key length is always stored as 2 bytes */
+  uint length= uint2korr(from);
+  if (length > max_length)
+    length= max_length;
+  *to++= (char) (length & 255);
+  if (max_length > 255)
+    *to++= (char) (length >> 8);
+  if (length)
+    memcpy(to, from+HA_KEY_BLOB_LENGTH, length);
+  return to+length;
+}
+
+
+/*
+  unpack field packed with Field_varstring::pack()
+*/
+
 const char *Field_varstring::unpack(char *to, const char *from)
 {
   uint length;
-  if (field_length <= 255)
-  {
+  if (length_bytes == 1)
     length= (uint) (uchar) (*to= *from++);
-    to[1]=0;
-  }
   else
   {
     length= uint2korr(from);
@@ -4859,7 +4982,7 @@ const char *Field_varstring::unpack(char *to, const char *from)
     to[1]= *from++;
   }
   if (length)
-    memcpy(to+HA_KEY_BLOB_LENGTH, from, length);
+    memcpy(to+ length_bytes, from, length);
   return from+length;
 }
 
@@ -4867,12 +4990,11 @@ const char *Field_varstring::unpack(char *to, const char *from)
 int Field_varstring::pack_cmp(const char *a, const char *b, uint key_length,
                               my_bool insert_or_update)
 {
-  uint a_length;
-  uint b_length;
+  uint a_length, b_length;
   if (key_length > 255)
   {
-    a_length=uint2korr(a); a+= HA_KEY_BLOB_LENGTH;
-    b_length=uint2korr(b); b+= HA_KEY_BLOB_LENGTH;
+    a_length=uint2korr(a); a+= 2;
+    b_length=uint2korr(b); b+= 2;
   }
   else
   {
@@ -4889,8 +5011,8 @@ int Field_varstring::pack_cmp(const char *a, const char *b, uint key_length,
 int Field_varstring::pack_cmp(const char *b, uint key_length,
                               my_bool insert_or_update)
 {
-  char *a= ptr+HA_KEY_BLOB_LENGTH;
-  uint a_length= uint2korr(ptr);
+  char *a= ptr+ length_bytes;
+  uint a_length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
   uint b_length;
   uint char_length= ((field_charset->mbmaxlen > 1) ?
                      key_length / field_charset->mbmaxlen : key_length);
@@ -4919,7 +5041,7 @@ int Field_varstring::pack_cmp(const char *b, uint key_length,
 uint Field_varstring::packed_col_length(const char *data_ptr, uint length)
 {
   if (length > 255)
-    return uint2korr(data_ptr)+HA_KEY_BLOB_LENGTH;
+    return uint2korr(data_ptr)+2;
   return (uint) ((uchar) *data_ptr)+1;
 }
 
@@ -4932,13 +5054,14 @@ uint Field_varstring::max_packed_col_length(uint max_length)
 
 void Field_varstring::get_key_image(char *buff, uint length, imagetype type)
 {
-  uint f_length= uint2korr(ptr);
+  uint f_length=  length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
   uint char_length= length / field_charset->mbmaxlen;
-  char_length= my_charpos(field_charset, ptr, ptr + HA_KEY_BLOB_LENGTH,
+  char_length= my_charpos(field_charset, ptr, ptr + length_bytes,
                           char_length);
   set_if_smaller(f_length, char_length);
+  /* Key is always stored with 2 bytes */
   int2store(buff,f_length);
-  memcpy(buff+HA_KEY_BLOB_LENGTH, ptr+HA_KEY_BLOB_LENGTH, f_length);
+  memcpy(buff+HA_KEY_BLOB_LENGTH, ptr+length_bytes, f_length);
   if (f_length < length)
   {
     /*
@@ -4952,18 +5075,12 @@ void Field_varstring::get_key_image(char *buff, uint length, imagetype type)
 
 void Field_varstring::set_key_image(char *buff,uint length)
 {
-  length=uint2korr(buff);			// Real length is here
+  length= uint2korr(buff);			// Real length is here
   (void) Field_varstring::store(buff+HA_KEY_BLOB_LENGTH, length,
                                 field_charset);
 }
 
 
-int Field_varstring::cmp_binary_offset(uint row_offset)
-{
-  return cmp_binary(ptr, ptr+row_offset);
-}
-
-
 int Field_varstring::cmp_binary(const char *a_ptr, const char *b_ptr,
                                 uint32 max_length)
 {
@@ -4971,13 +5088,41 @@ int Field_varstring::cmp_binary(const char *a_ptr, const char *b_ptr,
   uint diff;
   uint32 a_length,b_length;
 
-  a_length= uint2korr(a_ptr);
-  b_length= uint2korr(b_ptr);
+  if (length_bytes == 1)
+  {
+    a_length= (uint) (uchar) *a_ptr;
+    b_length= (uint) (uchar) *b_ptr;
+  }
+  else
+  {
+    a_length= uint2korr(a_ptr);
+    b_length= uint2korr(b_ptr);
+  }
   set_if_smaller(a_length, max_length);
   set_if_smaller(b_length, max_length);
   if (a_length != b_length)
     return 1;
-  return memcmp(a_ptr+2, b_ptr+2, a_length);
+  return memcmp(a_ptr+length_bytes, b_ptr+length_bytes, a_length);
+}
+
+
+Field *Field_varstring::new_field(MEM_ROOT *root, struct st_table *new_table)
+{
+  Field_varstring *res= (Field_varstring*) Field::new_field(root, new_table);
+  if (res)
+    res->length_bytes= length_bytes;
+  return res;
+}
+
+
+Field *Field_varstring::new_key_field(MEM_ROOT *root,
+                                      struct st_table *new_table)
+{
+  Field_varstring *res= (Field_varstring*) Field::new_field(root, new_table);
+  /* Keys length prefixes are always packed with 2 bytes */
+  if (res)
+    res->length_bytes= 2;
+  return res;
 }
 
 
@@ -5234,18 +5379,6 @@ int Field_blob::cmp(const char *a_ptr, const char *b_ptr)
 }
 
 
-int Field_blob::cmp_offset(uint row_offset)
-{
-  return Field_blob::cmp(ptr,ptr+row_offset);
-}
-
-
-int Field_blob::cmp_binary_offset(uint row_offset)
-{
-  return cmp_binary(ptr, ptr+row_offset);
-}
-
-
 int Field_blob::cmp_binary(const char *a_ptr, const char *b_ptr,
 			   uint32 max_length)
 {
@@ -5432,8 +5565,7 @@ const char *Field_blob::unpack(char *to, const char *from)
 int Field_blob::pack_cmp(const char *a, const char *b, uint key_length,
                          my_bool insert_or_update)
 {
-  uint a_length;
-  uint b_length;
+  uint a_length, b_length;
   if (key_length > 255)
   {
     a_length=uint2korr(a); a+=2;
@@ -5539,6 +5671,7 @@ const char *Field_blob::unpack_key(char *to, const char *from, uint max_length)
   return from + length;
 }
 
+
 /* Create a packed key that will be used for storage from a MySQL key */
 
 char *Field_blob::pack_key_from_key_image(char *to, const char *from,
@@ -6284,15 +6417,17 @@ void create_field::create_length_to_internal_length(void)
   case MYSQL_TYPE_STRING:
   case MYSQL_TYPE_VARCHAR:
     length*= charset->mbmaxlen;
-    key_length*= charset->mbmaxlen;
+    key_length= length;
     pack_length= calc_pack_length(sql_type, length);
     break;
   case MYSQL_TYPE_ENUM:
   case MYSQL_TYPE_SET:
+    /* Pack_length already calculated in sql_parse.cc */
     length*= charset->mbmaxlen;
+    key_length= pack_length;
     break;
   default:
-    /* do nothing */
+    key_length= pack_length= calc_pack_length(sql_type, length);
     break;
   }
 }
@@ -6323,7 +6458,7 @@ uint32 calc_pack_length(enum_field_types type,uint32 length)
   case MYSQL_TYPE_VAR_STRING:
   case MYSQL_TYPE_STRING:
   case FIELD_TYPE_DECIMAL:     return (length);
-  case MYSQL_TYPE_VARCHAR:     return (length+HA_KEY_BLOB_LENGTH);
+  case MYSQL_TYPE_VARCHAR:     return (length + (length < 256 ? 1: 2));
   case FIELD_TYPE_YEAR:
   case FIELD_TYPE_TINY	: return 1;
   case FIELD_TYPE_SHORT : return 2;
@@ -6423,7 +6558,9 @@ Field *make_field(char *ptr, uint32 field_length,
                                 unireg_check, field_name, table,
                                 field_charset);
       if (field_type == MYSQL_TYPE_VARCHAR)
-        return new Field_varstring(ptr,field_length,null_pos,null_bit,
+        return new Field_varstring(ptr,field_length,
+                                   HA_VARCHAR_PACKLENGTH(field_length),
+                                   null_pos,null_bit,
                                    unireg_check, field_name, table,
                                    field_charset);
       return 0;                                 // Error
@@ -6573,9 +6710,8 @@ create_field::create_field(Field *old_field,Field *orig_field)
   case MYSQL_TYPE_SET:
   case MYSQL_TYPE_VARCHAR:
   case MYSQL_TYPE_VAR_STRING:
-    /* These are corrected in create_length_to_internal_length */
+    /* This is corrected in create_length_to_internal_length */
     length= (length+charset->mbmaxlen-1) / charset->mbmaxlen;
-    key_length/= charset->mbmaxlen;
     break;
 #ifdef HAVE_SPATIAL
   case FIELD_TYPE_GEOMETRY:
diff --git a/sql/field.h b/sql/field.h
index de31f60350d..2d84d240839 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -139,10 +139,9 @@ public:
   virtual int cmp(const char *,const char *)=0;
   virtual int cmp_binary(const char *a,const char *b, uint32 max_length=~0L)
   { return memcmp(a,b,pack_length()); }
-  virtual int cmp_offset(uint row_offset)
-  { return memcmp(ptr,ptr+row_offset,pack_length()); }
-  virtual int cmp_binary_offset(uint row_offset)
-  { return memcmp(ptr,ptr+row_offset,pack_length()); }
+  int cmp_offset(uint row_offset) { return cmp(ptr,ptr+row_offset); }
+  int cmp_binary_offset(uint row_offset)
+  { return cmp_binary(ptr, ptr+row_offset); };
   virtual int key_cmp(const byte *a,const byte *b)
   { return cmp((char*) a,(char*) b); }
   virtual int key_cmp(const byte *str, uint length)
@@ -185,6 +184,10 @@ public:
   virtual bool can_be_compared_as_longlong() const { return FALSE; }
   virtual void free() {}
   virtual Field *new_field(MEM_ROOT *root, struct st_table *new_table);
+  virtual Field *new_key_field(MEM_ROOT *root, struct st_table *new_table)
+  {
+    return new_field(root, new_table);
+  }
   inline void move_field(char *ptr_arg,uchar *null_ptr_arg,uchar null_bit_arg)
   {
     ptr=ptr_arg; null_ptr=null_ptr_arg; null_bit=null_bit_arg;
@@ -935,26 +938,31 @@ public:
 
 class Field_varstring :public Field_str {
 public:
-  Field_varstring(char *ptr_arg, uint32 len_arg,uchar *null_ptr_arg,
+  /* Store number of bytes used to store length (1 or 2) */
+  uint32 length_bytes;
+  Field_varstring(char *ptr_arg,
+                  uint32 len_arg, uint length_bytes_arg,
+                  uchar *null_ptr_arg,
 		  uchar null_bit_arg,
 		  enum utype unireg_check_arg, const char *field_name_arg,
 		  struct st_table *table_arg, CHARSET_INFO *cs)
     :Field_str(ptr_arg, len_arg, null_ptr_arg, null_bit_arg,
-	       unireg_check_arg, field_name_arg, table_arg, cs)
+	       unireg_check_arg, field_name_arg, table_arg, cs),
+    length_bytes(length_bytes_arg)
   {}
   Field_varstring(uint32 len_arg,bool maybe_null_arg,
 		  const char *field_name_arg,
 		  struct st_table *table_arg, CHARSET_INFO *cs)
     :Field_str((char*) 0,len_arg, maybe_null_arg ? (uchar*) "": 0,0,
-	       NONE, field_name_arg, table_arg, cs)
+	       NONE, field_name_arg, table_arg, cs),
+    length_bytes(len_arg < 256 ? 1 :2)
   {}
 
   enum_field_types type() const { return MYSQL_TYPE_VARCHAR; }
-  enum ha_base_keytype key_type() const
-    { return binary() ? HA_KEYTYPE_VARBINARY : HA_KEYTYPE_VARTEXT; }
+  enum ha_base_keytype key_type() const;
   bool zero_pack() const { return 0; }
-  void reset(void) { bzero(ptr,field_length+2); }
-  uint32 pack_length() const { return (uint32) field_length+2; }
+  void reset(void) { bzero(ptr,field_length+length_bytes); }
+  uint32 pack_length() const { return (uint32) field_length+length_bytes; }
   uint32 key_length() const { return (uint32) field_length; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(longlong nr);
@@ -969,12 +977,13 @@ public:
   void sql_type(String &str) const;
   char *pack(char *to, const char *from, uint max_length=~(uint) 0);
   char *pack_key(char *to, const char *from, uint max_length);
+  char *pack_key_from_key_image(char* to, const char *from, uint max_length);
   const char *unpack(char* to, const char *from);
+  const char *unpack_key(char* to, const char *from, uint max_length);
   int pack_cmp(const char *a, const char *b, uint key_length,
                my_bool insert_or_update);
   int pack_cmp(const char *b, uint key_length,my_bool insert_or_update);
   int cmp_binary(const char *a,const char *b, uint32 max_length=~0L);
-  int cmp_binary_offset(uint row_offset);
   int key_cmp(const byte *,const byte*);
   int key_cmp(const byte *str, uint length);
   uint packed_col_length(const char *to, uint length);
@@ -984,6 +993,8 @@ public:
   bool has_charset(void) const
   { return charset() == &my_charset_bin ? FALSE : TRUE; }
   field_cast_enum field_cast_type() { return FIELD_CAST_VARSTRING; }
+  Field *new_field(MEM_ROOT *root, struct st_table *new_table);
+  Field *new_key_field(MEM_ROOT *root, struct st_table *new_table);
 };
 
 
@@ -1006,7 +1017,7 @@ public:
   }
   enum_field_types type() const { return FIELD_TYPE_BLOB;}
   enum ha_base_keytype key_type() const
-    { return binary() ? HA_KEYTYPE_VARBINARY : HA_KEYTYPE_VARTEXT; }
+    { return binary() ? HA_KEYTYPE_VARBINARY2 : HA_KEYTYPE_VARTEXT2; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(double nr);
   int  store(longlong nr);
@@ -1015,9 +1026,7 @@ public:
   String *val_str(String*,String *);
   int cmp(const char *,const char*);
   int cmp(const char *a, uint32 a_length, const char *b, uint32 b_length);
-  int cmp_offset(uint offset);
   int cmp_binary(const char *a,const char *b, uint32 max_length=~0L);
-  int cmp_binary_offset(uint row_offset);
   int key_cmp(const byte *,const byte*);
   int key_cmp(const byte *str, uint length);
   uint32 key_length() const { return 0; }
@@ -1064,9 +1073,9 @@ public:
     return 0;
   }
   char *pack(char *to, const char *from, uint max_length= ~(uint) 0);
-  const char *unpack(char *to, const char *from);
   char *pack_key(char *to, const char *from, uint max_length);
   char *pack_key_from_key_image(char* to, const char *from, uint max_length);
+  const char *unpack(char *to, const char *from);
   const char *unpack_key(char* to, const char *from, uint max_length);
   int pack_cmp(const char *a, const char *b, uint key_length,
                my_bool insert_or_update);
@@ -1101,7 +1110,7 @@ public:
      :Field_blob(len_arg, maybe_null_arg, field_name_arg,
                  table_arg, &my_charset_bin)
   { geom_type= geom_type_arg; }
-  enum ha_base_keytype key_type() const { return HA_KEYTYPE_VARBINARY; }
+  enum ha_base_keytype key_type() const { return HA_KEYTYPE_VARBINARY2; }
   enum_field_types type() const { return FIELD_TYPE_GEOMETRY; }
   void sql_type(String &str) const;
   int  store(const char *to, uint length, CHARSET_INFO *charset);
diff --git a/sql/field_conv.cc b/sql/field_conv.cc
index f0887d8343a..9fd4f0228b3 100644
--- a/sql/field_conv.cc
+++ b/sql/field_conv.cc
@@ -305,7 +305,8 @@ static void do_field_string(Copy_field *copy)
   char buff[MAX_FIELD_WIDTH];
   copy->tmp.set_quick(buff,sizeof(buff),copy->tmp.charset());
   copy->from_field->val_str(&copy->tmp);
-  copy->to_field->store(copy->tmp.c_ptr_quick(),copy->tmp.length(),copy->tmp.charset());
+  copy->to_field->store(copy->tmp.c_ptr_quick(),copy->tmp.length(),
+                        copy->tmp.charset());
 }
 
 
@@ -350,7 +351,23 @@ static void do_expand_string(Copy_field *copy)
                      copy->to_length-copy->from_length, ' ');
 }
 
-static void do_varstring(Copy_field *copy)
+
+static void do_varstring1(Copy_field *copy)
+{
+  uint length= (uint) *(uchar*) copy->from_ptr;
+  if (length > copy->to_length- 1)
+  {
+    length=copy->to_length - 1;
+    if (current_thd->count_cuted_fields)
+      copy->to_field->set_warning(MYSQL_ERROR::WARN_LEVEL_WARN,
+                                  ER_WARN_DATA_TRUNCATED, 1);
+  }
+  *(uchar*) copy->to_ptr= (uchar) length;
+  memcpy(copy->to_ptr+1, copy->from_ptr + 1, length);
+}
+
+
+static void do_varstring2(Copy_field *copy)
 {
   uint length=uint2korr(copy->from_ptr);
   if (length > copy->to_length- HA_KEY_BLOB_LENGTH)
@@ -508,9 +525,15 @@ void (*Copy_field::get_copy_func(Field *to,Field *from))(Copy_field*)
       }
       else if (to->charset() != from->charset())
 	return do_field_string;
-      else if (to->real_type() == MYSQL_TYPE_VARCHAR && to_length !=
-	       from_length)
-	return do_varstring;
+      else if (to->real_type() == MYSQL_TYPE_VARCHAR)
+      {
+        if (((Field_varstring*) to)->length_bytes !=
+            ((Field_varstring*) from)->length_bytes)
+          return do_field_string;
+        if (to_length != from_length)
+          return (((Field_varstring*) to)->length_bytes == 1 ?
+                  do_varstring1 : do_varstring2);
+      }
       else if (to_length < from_length)
 	return do_cut_string;
       else if (to_length > from_length)
@@ -590,6 +613,12 @@ void field_conv(Field *to,Field *from)
     char buff[MAX_FIELD_WIDTH];
     String result(buff,sizeof(buff),from->charset());
     from->val_str(&result);
+    /*
+      We use c_ptr_quick() here to make it easier if to is a float/double
+      as the conversion routines will do a copy of the result doesn't
+      end with \0. Can be replaced with .ptr() when we have our own
+      string->double conversion.
+    */
     to->store(result.c_ptr_quick(),result.length(),from->charset());
   }
   else if (from->result_type() == REAL_RESULT)
diff --git a/sql/ha_berkeley.cc b/sql/ha_berkeley.cc
index 6cb83624eff..322126ff47b 100644
--- a/sql/ha_berkeley.cc
+++ b/sql/ha_berkeley.cc
@@ -356,7 +356,8 @@ ulong ha_berkeley::index_flags(uint idx, uint part, bool all_parts) const
     }
     switch (table->key_info[idx].key_part[i].field->key_type()) {
     case HA_KEYTYPE_TEXT:
-    case HA_KEYTYPE_VARTEXT:
+    case HA_KEYTYPE_VARTEXT1:
+    case HA_KEYTYPE_VARTEXT2:
       /*
         As BDB stores only one copy of equal strings, we can't use key read
         on these. Binary collations do support key read though.
@@ -391,6 +392,7 @@ berkeley_cmp_packed_key(DB *file, const DBT *new_key, const DBT *saved_key)
   KEY_PART_INFO *key_part= key->key_part, *end=key_part+key->key_parts;
   uint key_length=new_key->size;
 
+  DBUG_DUMP("key_in_index", saved_key_ptr, saved_key->size);
   for (; key_part != end && (int) key_length > 0; key_part++)
   {
     int cmp;
@@ -745,11 +747,11 @@ void ha_berkeley::unpack_row(char *record, DBT *row)
 
 void ha_berkeley::unpack_key(char *record, DBT *key, uint index)
 {
-  KEY *key_info=table->key_info+index;
+  KEY *key_info= table->key_info+index;
   KEY_PART_INFO *key_part= key_info->key_part,
-		*end=key_part+key_info->key_parts;
+		*end= key_part+key_info->key_parts;
+  char *pos= (char*) key->data;
 
-  char *pos=(char*) key->data;
   for (; key_part != end; key_part++)
   {
     if (key_part->null_bit)
@@ -773,8 +775,10 @@ void ha_berkeley::unpack_key(char *record, DBT *key, uint index)
 
 
 /*
-  Create a packed key from from a row
-  This will never fail as the key buffer is pre allocated.
+  Create a packed key from a row. This key will be written as such
+  to the index tree.
+
+  This will never fail as the key buffer is pre-allocated.
 */
 
 DBT *ha_berkeley::create_key(DBT *key, uint keynr, char *buff,
@@ -820,7 +824,10 @@ DBT *ha_berkeley::create_key(DBT *key, uint keynr, char *buff,
 
 
 /*
-  Create a packed key from from a MySQL unpacked key
+  Create a packed key from from a MySQL unpacked key (like the one that is
+  sent from the index_read()
+
+  This key is to be used to read a row
 */
 
 DBT *ha_berkeley::pack_key(DBT *key, uint keynr, char *buff,
@@ -1457,7 +1464,7 @@ int ha_berkeley::read_row(int error, char *buf, uint keynr, DBT *row,
 int ha_berkeley::index_read_idx(byte * buf, uint keynr, const byte * key,
 				uint key_len, enum ha_rkey_function find_flag)
 {
-  statistic_increment(table->in_use->status_var.ha_read_key_count,&LOCK_status);
+  table->in_use->status_var.ha_read_key_count++;
   DBUG_ENTER("index_read_idx");
   current_row.flags=DB_DBT_REALLOC;
   active_index=MAX_KEY;
@@ -1476,10 +1483,9 @@ int ha_berkeley::index_read(byte * buf, const byte * key,
   int error;
   KEY *key_info= &table->key_info[active_index];
   int do_prev= 0;
-
   DBUG_ENTER("ha_berkeley::index_read");
 
-  statistic_increment(table->in_use->status_var.ha_read_key_count,&LOCK_status);
+  table->in_use->status_var.ha_read_key_count++;
   bzero((char*) &row,sizeof(row));
   if (find_flag == HA_READ_BEFORE_KEY)
   {
@@ -1679,6 +1685,7 @@ DBT *ha_berkeley::get_pos(DBT *to, byte *pos)
       pos+=key_part->field->packed_col_length((char*) pos,key_part->length);
     to->size= (uint) (pos- (byte*) to->data);
   }
+  DBUG_DUMP("key", (char*) to->data, to->size);
   return to;
 }
 
diff --git a/sql/ha_heap.cc b/sql/ha_heap.cc
index 60555d51402..1556a18bfca 100644
--- a/sql/ha_heap.cc
+++ b/sql/ha_heap.cc
@@ -488,8 +488,10 @@ int ha_heap::create(const char *name, TABLE *table_arg,
       else
       {
         if ((seg->type = field->key_type()) != (int) HA_KEYTYPE_TEXT &&
-            seg->type != HA_KEYTYPE_VARTEXT &&
-            seg->type != HA_KEYTYPE_VARBINARY)
+            seg->type != HA_KEYTYPE_VARTEXT1 &&
+            seg->type != HA_KEYTYPE_VARTEXT2 &&
+            seg->type != HA_KEYTYPE_VARBINARY1 &&
+            seg->type != HA_KEYTYPE_VARBINARY2)
           seg->type= HA_KEYTYPE_BINARY;
       }
       seg->start=   (uint) key_part->offset;
diff --git a/sql/ha_myisam.cc b/sql/ha_myisam.cc
index c1516e61c88..ce3287d8745 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -926,8 +926,11 @@ int ha_myisam::enable_indexes(uint mode)
     {
       sql_print_warning("Warning: Enabling keys got errno %d, retrying",
                         my_errno);
+      thd->clear_error();
       param.testflag&= ~(T_REP_BY_SORT | T_QUICK);
       error= (repair(thd,param,0) != HA_ADMIN_OK);
+      if (!error && thd->net.report_error)
+        error= HA_ERR_CRASHED;
     }
     info(HA_STATUS_CONST);
     thd->proc_info=save_proc_info;
@@ -1478,11 +1481,10 @@ int ha_myisam::create(const char *name, register TABLE *table_arg,
       break;
 
     if (found->flags & BLOB_FLAG)
-    {
       recinfo_pos->type= (int) FIELD_BLOB;
-    }
-    else if (!(options & HA_OPTION_PACK_RECORD) ||
-             found->type() == MYSQL_TYPE_VARCHAR)
+    else if (found->type() == MYSQL_TYPE_VARCHAR)
+      recinfo_pos->type= FIELD_VARCHAR;
+    else if (!(options & HA_OPTION_PACK_RECORD))
       recinfo_pos->type= (int) FIELD_NORMAL;
     else if (found->zero_pack())
       recinfo_pos->type= (int) FIELD_SKIP_ZERO;
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index b242698d36e..949545bcdb0 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -1214,7 +1214,7 @@ int composite_key_cmp(void* arg, byte* key1, byte* key2)
   {
     Field* f = *field;
     int len = *lengths++;
-    int res = f->key_cmp(key1, key2);
+    int res = f->cmp(key1, key2);
     if (res)
       return res;
     key1 += len;
@@ -1668,7 +1668,7 @@ int group_concat_key_cmp_with_distinct(void* arg, byte* key1,
     {
       int res;
       uint offset= (uint) (field->ptr - record);
-      if ((res= field->key_cmp(key1 + offset, key2 + offset)))
+      if ((res= field->cmp(key1 + offset, key2 + offset)))
 	return res;
     }
   }
@@ -1702,7 +1702,7 @@ int group_concat_key_cmp_with_order(void* arg, byte* key1, byte* key2)
     {
       int res;
       uint offset= (uint) (field->ptr - record);
-      if ((res= field->key_cmp(key1 + offset, key2 + offset)))
+      if ((res= field->cmp(key1 + offset, key2 + offset)))
         return (*order_item)->asc ? res : -res;
     }
   }
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index f9149f10a30..0f54f06a22b 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -746,7 +746,7 @@ int QUICK_RANGE_SELECT::init()
 void QUICK_RANGE_SELECT::range_end()
 {
   if (file->inited != handler::NONE)
-    file->ha_index_end();
+    file->ha_index_or_rnd_end();
 }
 
 
@@ -3687,7 +3687,8 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
   }
   /* Get local copy of key */
   copies= 1;
-  if (field->key_type() == HA_KEYTYPE_VARTEXT)
+  if (field->key_type() == HA_KEYTYPE_VARTEXT1 ||
+      field->key_type() == HA_KEYTYPE_VARTEXT2)
     copies= 2;
   str= str2= (char*) alloc_root(param->mem_root,
 				(key_part->store_length)*copies+1);
@@ -5888,7 +5889,7 @@ int QUICK_RANGE_SELECT::get_next()
   SYNOPSIS
     QUICK_RANGE_SELECT::get_next_prefix()
     prefix_length  length of cur_prefix
-    cur_prefix     prefix of a key to be searached for
+    cur_prefix     prefix of a key to be searched for
 
   DESCRIPTION
     Each subsequent call to the method retrieves the first record that has a
@@ -7402,7 +7403,8 @@ TRP_GROUP_MIN_MAX::make_quick(PARAM *param, bool retrieve_full_rows,
       quick->quick_prefix_select= NULL; /* Can't construct a quick select. */
     else
       /* Make a QUICK_RANGE_SELECT to be used for group prefix retrieval. */
-      quick->quick_prefix_select= get_quick_select(param, param_idx, index_tree,
+      quick->quick_prefix_select= get_quick_select(param, param_idx,
+                                                   index_tree,
                                                    &quick->alloc);
 
     /*
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index b4a2f368bc2..2500769ee30 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2467,6 +2467,8 @@ find_item_in_list(Item *find, List<Item> &items, uint *counter,
   bool found_unaliased_non_uniq= 0;
   uint unaliased_counter;
 
+  LINT_INIT(unaliased_counter);                 // Dependent on found_unaliased
+
   *unaliased= FALSE;
 
   if (find->type() == Item::FIELD_ITEM	|| find->type() == Item::REF_ITEM)
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 5720aac8dbd..9fc4cfacfd2 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -4891,11 +4891,9 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
     break;
   case MYSQL_TYPE_VARCHAR:
     /*
-      We can't use pack_length as this includes the field length
       Long VARCHAR's are automaticly converted to blobs in mysql_prepare_table
       if they don't have a default value
     */
-    new_field->key_length= new_field->length;
     max_field_charlength= MAX_FIELD_VARCHARLENGTH;
     break;
   case MYSQL_TYPE_STRING:
@@ -5096,16 +5094,12 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
     my_error(ER_WRONG_FIELD_SPEC, MYF(0), field_name);
     DBUG_RETURN(1);
   }
-  if (!new_field->pack_length)
-    new_field->pack_length= calc_pack_length(new_field->sql_type,
-                                             new_field->length);
-  if (!new_field->key_length)
-    new_field->key_length= new_field->pack_length;
   lex->create_list.push_back(new_field);
   lex->last_field=new_field;
   DBUG_RETURN(0);
 }
 
+
 /* Store position for column in ALTER TABLE .. ADD column */
 
 void store_position_for_column(const char *name)
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 8fe83e4f1d7..5425ddfb64b 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -235,9 +235,7 @@ bool handle_select(THD *thd, LEX *lex, select_result *result)
   res|= thd->net.report_error;
   if (unlikely(res))
   {
-    /*
-      If we have real error reported erly then this will be ignored
-    */
+    /* If we had a another error reported earlier then this will be ignored */
     result->send_error(ER_UNKNOWN_ERROR, ER(ER_UNKNOWN_ERROR));
     result->abort();
   }
@@ -4873,7 +4871,7 @@ static bool create_ref_for_key(JOIN *join, JOIN_TAB *j, KEYUSE *org_keyuse,
 				  &keyinfo->key_part[i],
 				  (char*) key_buff,maybe_null);
       /*
-	Remeber if we are going to use REF_OR_NULL
+	Remember if we are going to use REF_OR_NULL
 	But only if field _really_ can be null i.e. we force JT_REF
 	instead of JT_REF_OR_NULL in case if field can't be null
       */
@@ -7538,7 +7536,7 @@ static Field* create_tmp_field_from_field(THD *thd, Field* org_field,
 {
   Field *new_field;
 
-  if (convert_blob_length && org_field->flags & BLOB_FLAG)
+  if (convert_blob_length && (org_field->flags & BLOB_FLAG))
     new_field= new Field_varstring(convert_blob_length,
                                    org_field->maybe_null(),
                                    org_field->field_name, table,
@@ -7806,7 +7804,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
     else for (ORDER *tmp=group ; tmp ; tmp=tmp->next)
     {
       (*tmp->item)->marker=4;			// Store null in key
-      if ((*tmp->item)->max_length >= MAX_CHAR_WIDTH)
+      if ((*tmp->item)->max_length >= CONVERT_IF_BIGGER_TO_BLOB)
 	using_unique_constraint=1;
     }
     if (param->group_length >= MAX_BLOB_WIDTH)
@@ -8151,16 +8149,17 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
       key_part_info->null_bit=0;
       key_part_info->field=  field;
       key_part_info->offset= field->offset();
-      key_part_info->length= (uint16) field->pack_length();
+      key_part_info->length= (uint16) field->key_length();
       key_part_info->type=   (uint8) field->key_type();
       key_part_info->key_type =
 	((ha_base_keytype) key_part_info->type == HA_KEYTYPE_TEXT ||
-	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT) ?
+	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT1 ||
+	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT2) ?
 	0 : FIELDFLAG_BINARY;
       if (!using_unique_constraint)
       {
 	group->buff=(char*) group_buff;
-	if (!(group->field=field->new_field(thd->mem_root,table)))
+	if (!(group->field=field->new_key_field(thd->mem_root,table)))
 	  goto err; /* purecov: inspected */
 	if (maybe_null)
 	{
@@ -8181,7 +8180,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
 	  group->field->move_field((char*) group_buff);
         /* In GROUP BY 'a' and 'a ' are equal for VARCHAR fields */
         key_part_info->key_part_flag|= HA_END_SPACE_ARE_EQUAL;
-	group_buff+= key_part_info->length;
+	group_buff+= group->field->pack_length();
       }
       keyinfo->key_length+=  key_part_info->length;
     }
@@ -8245,7 +8244,8 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
       key_part_info->type=     (uint8) (*reg_field)->key_type();
       key_part_info->key_type =
 	((ha_base_keytype) key_part_info->type == HA_KEYTYPE_TEXT ||
-	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT) ?
+	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT1 ||
+	 (ha_base_keytype) key_part_info->type == HA_KEYTYPE_VARTEXT2) ?
 	0 : FIELDFLAG_BINARY;
     }
   }
@@ -8295,8 +8295,8 @@ static bool create_myisam_tmp_table(TABLE *table,TMP_TABLE_PARAM *param,
   MI_KEYDEF keydef;
   MI_UNIQUEDEF uniquedef;
   KEY *keyinfo=param->keyinfo;
-
   DBUG_ENTER("create_myisam_tmp_table");
+
   if (table->keys)
   {						// Get keys for ni_create
     bool using_unique_constraint=0;
@@ -8344,19 +8344,18 @@ static bool create_myisam_tmp_table(TABLE *table,TMP_TABLE_PARAM *param,
       {
 	seg->type=
 	((keyinfo->key_part[i].key_type & FIELDFLAG_BINARY) ?
-	 HA_KEYTYPE_VARBINARY : HA_KEYTYPE_VARTEXT);
-	seg->bit_start=seg->length - table->blob_ptr_size;
+	 HA_KEYTYPE_VARBINARY2 : HA_KEYTYPE_VARTEXT2);
+	seg->bit_start= field->pack_length() - table->blob_ptr_size;
 	seg->flag= HA_BLOB_PART;
 	seg->length=0;			// Whole blob in unique constraint
       }
       else
       {
-	seg->type= ((keyinfo->key_part[i].key_type & FIELDFLAG_BINARY) ?
-                    HA_KEYTYPE_BINARY : HA_KEYTYPE_TEXT);
+	seg->type= keyinfo->key_part[i].type;
         /* Tell handler if it can do suffic space compression */
 	if (field->real_type() == MYSQL_TYPE_STRING &&
 	    keyinfo->key_part[i].length > 4)
-	  seg->flag|=HA_SPACE_PACK;
+	  seg->flag|= HA_SPACE_PACK;
       }
       if (!(field->flags & NOT_NULL_FLAG))
       {
@@ -8365,7 +8364,7 @@ static bool create_myisam_tmp_table(TABLE *table,TMP_TABLE_PARAM *param,
 	/*
 	  We are using a GROUP BY on something that contains NULL
 	  In this case we have to tell MyISAM that two NULL should
-	  on INSERT be compared as equal
+	  on INSERT be regarded at the same value
 	*/
 	if (!using_unique_constraint)
 	  keydef.flag|= HA_NULL_ARE_EQUAL;
@@ -8649,21 +8648,19 @@ do_select(JOIN *join,List<Item> *fields,TABLE *table,Procedure *procedure)
   }
   if (table)
   {
-    int tmp;
+    int tmp, new_errno= 0;
     if ((tmp=table->file->extra(HA_EXTRA_NO_CACHE)))
     {
       DBUG_PRINT("error",("extra(HA_EXTRA_NO_CACHE) failed"));
-      my_errno= tmp;
-      error= -1;
+      new_errno= tmp;
     }
     if ((tmp=table->file->ha_index_or_rnd_end()))
     {
       DBUG_PRINT("error",("ha_index_or_rnd_end() failed"));
-      my_errno= tmp;
-      error= -1;
+      new_errno= tmp;
     }
-    if (error == -1)
-      table->file->print_error(my_errno,MYF(0));
+    if (new_errno)
+      table->file->print_error(new_errno,MYF(0));
   }
 #ifndef DBUG_OFF
   if (error)
@@ -9835,13 +9832,19 @@ end_update(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
     DBUG_RETURN(0);
   }
 
-  /* The null bits are already set */
+  /*
+    Copy null bits from group key to table
+    We can't copy all data as the key may have different format
+    as the row data (for example as with VARCHAR keys)
+  */
   KEY_PART_INFO *key_part;
   for (group=table->group,key_part=table->key_info[0].key_part;
        group ;
        group=group->next,key_part++)
-    memcpy(table->record[0]+key_part->offset, group->buff, key_part->length);
-
+  {
+    if (key_part->null_bit)
+      memcpy(table->record[0]+key_part->offset, group->buff, 1);
+  }
   init_tmptable_sum_functions(join->sum_funcs);
   copy_funcs(join->tmp_table_param.items_to_copy);
   if ((error=table->file->write_row(table->record[0])))
@@ -11651,8 +11654,10 @@ calc_group_buffer(JOIN *join,ORDER *group)
     {
       if (field->type() == FIELD_TYPE_BLOB)
 	key_length+=MAX_BLOB_WIDTH;		// Can't be used as a key
+      else if (field->type() == MYSQL_TYPE_VARCHAR)
+        key_length+= field->field_length + HA_KEY_BLOB_LENGTH;
       else
-	key_length+=field->pack_length();
+	key_length+= field->pack_length();
     }
     else if ((*group->item)->result_type() == REAL_RESULT)
       key_length+=sizeof(double);
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 5e42fc0ee30..be3a72836b4 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -420,12 +420,15 @@ class store_key :public Sql_alloc
     :null_ptr(null),err(0)
   {
     if (field_arg->type() == FIELD_TYPE_BLOB)
-      to_field=new Field_varstring(ptr, length, (uchar*) null, 1, 
+    {
+      /* Key segments are always packed with a 2 byte length prefix */
+      to_field=new Field_varstring(ptr, length, 2, (uchar*) null, 1, 
 				   Field::NONE, field_arg->field_name,
 				   field_arg->table, field_arg->charset());
+    }
     else
     {
-      to_field=field_arg->new_field(thd->mem_root,field_arg->table);
+      to_field=field_arg->new_key_field(thd->mem_root, field_arg->table);
       if (to_field)
 	to_field->move_field(ptr, (uchar*) null, 1);
     }
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 5a135faf52f..3023a7e869d 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -3389,7 +3389,7 @@ int make_schema_select(THD *thd, SELECT_LEX *sel,
 
 
 /*
-  Fill temporaty schema tables before SELECT
+  Fill temporary schema tables before SELECT
 
   SYNOPSIS
     get_schema_tables_result()
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 8662eb63ce6..6629122a1fa 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -509,7 +509,8 @@ int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
       String conv, *tmp;
       for (uint i= 0; (tmp= it++); i++)
       {
-        if (String::needs_conversion(tmp->length(), tmp->charset(), cs, &dummy))
+        if (String::needs_conversion(tmp->length(), tmp->charset(), cs,
+                                     &dummy))
         {
           uint cnv_errs;
           conv.copy(tmp->ptr(), tmp->length(), tmp->charset(), cs, &cnv_errs);
@@ -3698,7 +3699,7 @@ copy_data_between_tables(TABLE *from,TABLE *to,
   free_io_cache(from);
   delete [] copy;				// This is never 0
 
-  if (to->file->end_bulk_insert() && !error)
+  if (to->file->end_bulk_insert() && error <= 0)
   {
     to->file->print_error(my_errno,MYF(0));
     error=1;
diff --git a/vio/vio.c b/vio/vio.c
index a356d8edeff..39b5f843e5e 100644
--- a/vio/vio.c
+++ b/vio/vio.c
@@ -32,7 +32,8 @@ void vio_reset(Vio* vio, enum enum_vio_type type,
 		      my_bool localhost)
 {
   DBUG_ENTER("vio_reset");
-  DBUG_PRINT("enter", ("type=%d  sd=%d  localhost=%d", type, sd, localhost));
+  DBUG_PRINT("enter", ("type: %d  sd: %d  localhost: %d", type, sd,
+                       localhost));
 
   bzero((char*) vio, sizeof(*vio));
   vio->type	= type;
@@ -123,7 +124,7 @@ Vio *vio_new(my_socket sd, enum enum_vio_type type, my_bool localhost)
 {
   Vio *vio;
   DBUG_ENTER("vio_new");
-  DBUG_PRINT("enter", ("sd=%d", sd));
+  DBUG_PRINT("enter", ("sd: %d", sd));
   if ((vio = (Vio*) my_malloc(sizeof(*vio),MYF(MY_WME))))
   {
     vio_reset(vio, type, sd, 0, localhost);
diff --git a/vio/viosocket.c b/vio/viosocket.c
index 48a9058480a..2921eb7495e 100644
--- a/vio/viosocket.c
+++ b/vio/viosocket.c
@@ -33,7 +33,7 @@ int vio_read(Vio * vio, gptr buf, int size)
 {
   int r;
   DBUG_ENTER("vio_read");
-  DBUG_PRINT("enter", ("sd=%d, buf=%p, size=%d", vio->sd, buf, size));
+  DBUG_PRINT("enter", ("sd: %d, buf: 0x%p, size: %d", vio->sd, buf, size));
 
 #ifdef __WIN__
   r = recv(vio->sd, buf, size,0);
@@ -56,7 +56,7 @@ int vio_write(Vio * vio, const gptr buf, int size)
 {
   int r;
   DBUG_ENTER("vio_write");
-  DBUG_PRINT("enter", ("sd=%d, buf=%p, size=%d", vio->sd, buf, size));
+  DBUG_PRINT("enter", ("sd: %d, buf: 0x%p, size: %d", vio->sd, buf, size));
 #ifdef __WIN__
   r = send(vio->sd, buf, size,0);
 #else
@@ -168,7 +168,7 @@ int vio_keepalive(Vio* vio, my_bool set_keep_alive)
   int r=0;
   uint opt = 0;
   DBUG_ENTER("vio_keepalive");
-  DBUG_PRINT("enter", ("sd=%d, set_keep_alive=%d", vio->sd, (int)
+  DBUG_PRINT("enter", ("sd: %d, set_keep_alive: %d", vio->sd, (int)
 		       set_keep_alive));
   if (vio->type != VIO_TYPE_NAMEDPIPE)
   {
@@ -315,7 +315,7 @@ int vio_read_pipe(Vio * vio, gptr buf, int size)
 {
   DWORD length;
   DBUG_ENTER("vio_read_pipe");
-  DBUG_PRINT("enter", ("sd=%d, buf=%p, size=%d", vio->sd, buf, size));
+  DBUG_PRINT("enter", ("sd: %d, buf: 0x%p, size: %d", vio->sd, buf, size));
 
   if (!ReadFile(vio->hPipe, buf, size, &length, NULL))
     DBUG_RETURN(-1);
@@ -329,7 +329,7 @@ int vio_write_pipe(Vio * vio, const gptr buf, int size)
 {
   DWORD length;
   DBUG_ENTER("vio_write_pipe");
-  DBUG_PRINT("enter", ("sd=%d, buf=%p, size=%d", vio->sd, buf, size));
+  DBUG_PRINT("enter", ("sd: %d, buf: 0x%p, size: %d", vio->sd, buf, size));
 
   if (!WriteFile(vio->hPipe, (char*) buf, size, &length, NULL))
     DBUG_RETURN(-1);
@@ -373,7 +373,7 @@ int vio_read_shared_memory(Vio * vio, gptr buf, int size)
   char *current_postion;
 
   DBUG_ENTER("vio_read_shared_memory");
-  DBUG_PRINT("enter", ("sd=%d, buf=%p, size=%d", vio->sd, buf, size));
+  DBUG_PRINT("enter", ("sd: %d, buf: 0x%p, size: %d", vio->sd, buf, size));
 
   remain_local = size;
   current_postion=buf;
@@ -423,7 +423,7 @@ int vio_write_shared_memory(Vio * vio, const gptr buf, int size)
   char *current_postion;
 
   DBUG_ENTER("vio_write_shared_memory");
-  DBUG_PRINT("enter", ("sd=%d, buf=%p, size=%d", vio->sd, buf, size));
+  DBUG_PRINT("enter", ("sd: %d, buf: 0x%p, size: %d", vio->sd, buf, size));
 
   remain = size;
   current_postion = buf;
diff --git a/vio/viossl.c b/vio/viossl.c
index a489cb98f98..912365adca0 100644
--- a/vio/viossl.c
+++ b/vio/viossl.c
@@ -99,7 +99,7 @@ int vio_ssl_read(Vio * vio, gptr buf, int size)
 {
   int r;
   DBUG_ENTER("vio_ssl_read");
-  DBUG_PRINT("enter", ("sd=%d, buf=%p, size=%d, ssl_=%p",
+  DBUG_PRINT("enter", ("sd: %d, buf: 0x%p, size: %d, ssl_: 0x%p",
 		       vio->sd, buf, size, vio->ssl_arg));
 
   if ((r= SSL_read((SSL*) vio->ssl_arg, buf, size)) < 0)
@@ -117,7 +117,7 @@ int vio_ssl_write(Vio * vio, const gptr buf, int size)
 {
   int r;
   DBUG_ENTER("vio_ssl_write");
-  DBUG_PRINT("enter", ("sd=%d, buf=%p, size=%d", vio->sd, buf, size));
+  DBUG_PRINT("enter", ("sd: %d, buf: 0x%p, size: %d", vio->sd, buf, size));
 
   if ((r= SSL_write((SSL*) vio->ssl_arg, buf, size)) < 0)
     report_errors();
@@ -157,7 +157,7 @@ int vio_ssl_keepalive(Vio* vio, my_bool set_keep_alive)
 {
   int r=0;
   DBUG_ENTER("vio_ssl_keepalive");
-  DBUG_PRINT("enter", ("sd=%d, set_keep_alive=%d", vio->sd, (int)
+  DBUG_PRINT("enter", ("sd: %d, set_keep_alive: %d", vio->sd, (int)
 		       set_keep_alive));
   if (vio->type != VIO_TYPE_NAMEDPIPE)
   {
@@ -227,7 +227,7 @@ my_socket vio_ssl_fd(Vio* vio)
 my_bool vio_ssl_peer_addr(Vio * vio, char *buf, uint16 *port)
 {
   DBUG_ENTER("vio_ssl_peer_addr");
-  DBUG_PRINT("enter", ("sd=%d", vio->sd));
+  DBUG_PRINT("enter", ("sd: %d", vio->sd));
   if (vio->localhost)
   {
     strmov(buf,"127.0.0.1");
@@ -250,7 +250,7 @@ my_bool vio_ssl_peer_addr(Vio * vio, char *buf, uint16 *port)
     *port= 0;
 #endif
   }
-  DBUG_PRINT("exit", ("addr=%s", buf));
+  DBUG_PRINT("exit", ("addr: %s", buf));
   DBUG_RETURN(0);
 }
 
@@ -279,7 +279,7 @@ int sslaccept(struct st_VioSSLAcceptorFd* ptr, Vio* vio, long timeout)
   my_bool net_blocking;
   enum enum_vio_type old_type;  
   DBUG_ENTER("sslaccept");
-  DBUG_PRINT("enter", ("sd=%d ptr=%p", vio->sd,ptr));
+  DBUG_PRINT("enter", ("sd: %d  ptr: Ox%p", vio->sd,ptr));
 
   old_type= vio->type;
   net_blocking = vio_is_blocking(vio);
@@ -294,7 +294,8 @@ int sslaccept(struct st_VioSSLAcceptorFd* ptr, Vio* vio, long timeout)
     vio_blocking(vio, net_blocking, &unused);
     DBUG_RETURN(1);
   }
-  DBUG_PRINT("info", ("ssl_=%p  timeout=%ld",(SSL*) vio->ssl_arg, timeout));
+  DBUG_PRINT("info", ("ssl_: Ox%p  timeout: %ld",
+                      (SSL*) vio->ssl_arg, timeout));
   SSL_clear((SSL*) vio->ssl_arg);
   SSL_SESSION_set_timeout(SSL_get_session((SSL*) vio->ssl_arg), timeout);
   SSL_set_fd((SSL*) vio->ssl_arg,vio->sd);
@@ -352,7 +353,8 @@ int sslconnect(struct st_VioSSLConnectorFd* ptr, Vio* vio, long timeout)
   my_bool net_blocking;
   enum enum_vio_type old_type;  
   DBUG_ENTER("sslconnect");
-  DBUG_PRINT("enter", ("sd=%d ptr=%p ctx: %p", vio->sd,ptr,ptr->ssl_context));
+  DBUG_PRINT("enter", ("sd: %d  ptr: 0x%p  ctx: 0x%p",
+                       vio->sd,ptr,ptr->ssl_context));
 
   old_type= vio->type;
   net_blocking = vio_is_blocking(vio);
@@ -367,7 +369,8 @@ int sslconnect(struct st_VioSSLConnectorFd* ptr, Vio* vio, long timeout)
     vio_blocking(vio, net_blocking, &unused);    
     DBUG_RETURN(1);
   }
-  DBUG_PRINT("info", ("ssl_=%p  timeout=%ld",(SSL*) vio->ssl_arg, timeout));
+  DBUG_PRINT("info", ("ssl_: 0x%p  timeout: %ld",
+                      (SSL*) vio->ssl_arg, timeout));
   SSL_clear((SSL*) vio->ssl_arg);
   SSL_SESSION_set_timeout(SSL_get_session((SSL*) vio->ssl_arg), timeout);
   SSL_set_fd ((SSL*) vio->ssl_arg, vio->sd);
diff --git a/vio/viosslfactories.c b/vio/viosslfactories.c
index 498d10da0ee..44a077c33fc 100644
--- a/vio/viosslfactories.c
+++ b/vio/viosslfactories.c
@@ -80,7 +80,7 @@ static int
 vio_set_cert_stuff(SSL_CTX *ctx, const char *cert_file, const char *key_file)
 {
   DBUG_ENTER("vio_set_cert_stuff");
-  DBUG_PRINT("enter", ("ctx=%p, cert_file=%s, key_file=%s",
+  DBUG_PRINT("enter", ("ctx: %p, cert_file: %s, key_file: %s",
 		       ctx, cert_file, key_file));
   if (cert_file != NULL)
   {
@@ -131,7 +131,7 @@ vio_verify_callback(int ok, X509_STORE_CTX *ctx)
   int	err,depth;
 
   DBUG_ENTER("vio_verify_callback");
-  DBUG_PRINT("enter", ("ok=%d, ctx=%p", ok, ctx));
+  DBUG_PRINT("enter", ("ok: %d, ctx: 0x%p", ok, ctx));
   err_cert=X509_STORE_CTX_get_current_cert(ctx);
   err=	   X509_STORE_CTX_get_error(ctx);
   depth=   X509_STORE_CTX_get_error_depth(ctx);
@@ -220,7 +220,7 @@ new_VioSSLConnectorFd(const char* key_file,
   DH *dh;
   DBUG_ENTER("new_VioSSLConnectorFd");
   DBUG_PRINT("enter",
-	     ("key_file=%s, cert_file=%s, ca_path=%s, ca_file=%s, cipher=%s",
+	     ("key_file: %s, cert_file: %s, ca_path: %s, ca_file: %s, cipher: %s",
 	      key_file, cert_file, ca_path, ca_file, cipher));
 
   if (!(ptr=((struct st_VioSSLConnectorFd*)
@@ -315,7 +315,7 @@ new_VioSSLAcceptorFd(const char *key_file,
   DH *dh;
   DBUG_ENTER("new_VioSSLAcceptorFd");
   DBUG_PRINT("enter",
-	     ("key_file=%s, cert_file=%s, ca_path=%s, ca_file=%s, cipher=%s",
+	     ("key_file: %s, cert_file: %s, ca_path: %s, ca_file: %s, cipher: %s",
 	      key_file, cert_file, ca_path, ca_file, cipher));
 
   ptr= ((struct st_VioSSLAcceptorFd*)