bug#2686 - index_merge select on BerkeleyDB table with varchar PK causes mysqld to crash.

Added put_length() to get_length() and unpack_key() to pack_key().
Keys were packed with the minimum size of the length field for the key part and 
unpacked with length size of the base column. 
For the purpose of optimal key packing we have the method pack_key(), while rows are 
packed with pack(). Now keys are unpacked with unpack_key() and no longer with 
unpack() which is used for rows.

mysql-test/r/bdb.result

@@ -1190,3 +1190,19 @@ a
 A
 a
 drop table t1;
+create table t1(
+pk1 text not null, pk2 text not null, pk3 char(4),
+key1 int, key2 int,
+primary key(pk1(4), pk2(4), pk3), key(key1), key(key2)
+) engine=bdb;
+insert into t1 values (concat('aaa-', repeat('A', 4000)),
+concat('eee-', repeat('e', 4000)), 'a++a', 1, 1);
+insert into t1 values (concat('bbb-', repeat('B', 4000)),
+concat('ggg-', repeat('G', 4000)), 'b++b', 1, 1);
+select substring(pk1, 1, 4), substring(pk1, 4001),
+substring(pk2, 1, 4), substring(pk2, 4001), pk3, key1, key2
+from t1 force index(key1, key2) where key1 < 3 or key2 < 3;
+substring(pk1, 1, 4)	substring(pk1, 4001)	substring(pk2, 1, 4)	substring(pk2, 4001)	pk3	key1	key2
+aaa-	AAAA	eee-	eeee	a++a	1	1
+bbb-	BBBB	ggg-	GGGG	b++b	1	1
+drop table t1;

mysql-test/t/bdb.test

@@ -829,3 +829,22 @@ alter table t1 modify a char(10) binary;
 explain select a from t1;
 select a from t1;
 drop table t1;
+
+#
+# bug#2686 - index_merge select on BerkeleyDB table with varchar PK causes mysqld to crash
+#
+
+create table t1(
+  pk1 text not null, pk2 text not null, pk3 char(4),
+  key1 int, key2 int,
+  primary key(pk1(4), pk2(4), pk3), key(key1), key(key2)
+) engine=bdb;
+insert into t1 values (concat('aaa-', repeat('A', 4000)),
+  concat('eee-', repeat('e', 4000)), 'a++a', 1, 1);
+insert into t1 values (concat('bbb-', repeat('B', 4000)),
+  concat('ggg-', repeat('G', 4000)), 'b++b', 1, 1);
+select substring(pk1, 1, 4), substring(pk1, 4001),
+       substring(pk2, 1, 4), substring(pk2, 4001), pk3, key1, key2
+       from t1 force index(key1, key2) where key1 < 3 or key2 < 3;
+drop table t1;
+

sql/field.cc

@@ -4167,6 +4167,42 @@ uint32 Field_blob::get_length(const char *pos)
 }
 
 
+/*
+  Put a blob length field into a record buffer.
+
+  SYNOPSIS
+    Field_blob::put_length()
+    pos                         Pointer into the record buffer.
+    length                      The length value to put.
+
+  DESCRIPTION
+    Depending on the maximum length of a blob, its length field is
+    put into 1 to 4 bytes. This is a property of the blob object,
+    described by 'packlength'.
+
+  RETURN
+    nothing
+*/
+
+void Field_blob::put_length(char *pos, uint32 length)
+{
+  switch (packlength) {
+  case 1:
+    *pos= (char) length;
+    break;
+  case 2:
+    int2store(pos, length);
+    break;
+  case 3:
+    int3store(pos, length);
+    break;
+  case 4:
+    int4store(pos, length);
+    break;
+  }
+}
+
+
 void Field_blob::store(const char *from,uint len)
 {
   if (!len)
@@ -4525,6 +4561,50 @@ char *Field_blob::pack_key(char *to, const char *from, uint max_length)
   return to+length;
 }
 
+
+/*
+  Unpack a blob key into a record buffer.
+
+  SYNOPSIS
+    Field_blob::unpack_key()
+    to                          Pointer into the record buffer.
+    from                        Pointer to the packed key.
+    max_length                  Key length limit from key description.
+
+  DESCRIPTION
+    A blob 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'.
+    Depending on the maximum length of a blob, its length field is
+    put into 1 to 4 bytes. This is a property of the blob object,
+    described by 'packlength'.
+    Blobs are internally stored apart from the record buffer, which
+    contains a pointer to the blob buffer.
+
+  RETURN
+    Pointer into 'from' past the last byte copied from packed key.
+*/
+
+const char *Field_blob::unpack_key(char *to, const char *from, uint max_length)
+{
+  /* get length of the blob key */
+  uint32 length= *((uchar*) from++);
+  if (max_length > 255)
+    length+= (*((uchar*) from++)) << 8;
+
+  /* put the length into the record buffer */
+  put_length(to, length);
+
+  /* put the address of the blob buffer or NULL */
+  if (length)
+    memcpy_fixed(to + packlength, &from, sizeof(from));
+  else
+    bzero(to + packlength, sizeof(from));
+
+  /* point to first byte of next field in 'from' */
+  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,

sql/field.h

@@ -189,6 +189,10 @@ class Field {
   {
     return pack(to,from,max_length);
   }
+  virtual const char *unpack_key(char* to, const char *from, uint max_length)
+  {
+    return unpack(to,from);
+  }
   virtual uint packed_col_length(const char *to, uint length)
   { return length;}
   virtual uint max_packed_col_length(uint max_length)
@@ -890,6 +894,7 @@ class Field_blob :public Field_str {
   inline uint32 get_length(uint row_offset=0)
   { return get_length(ptr+row_offset); }
   uint32 get_length(const char *ptr);
+  void put_length(char *pos, uint32 length);
   bool binary() const { return binary_flag; }
   inline void get_ptr(char **str)
     {
@@ -923,6 +928,7 @@ class Field_blob :public Field_str {
   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_key(char* to, const char *from, uint max_length);
   int pack_cmp(const char *a, const char *b, uint key_length);
   int pack_cmp(const char *b, uint key_length);
   uint packed_col_length(const char *col_ptr, uint length);

sql/ha_berkeley.cc

@@ -720,8 +720,8 @@ void ha_berkeley::unpack_key(char *record, DBT *key, uint index)
       }
       record[key_part->null_offset]&= ~key_part->null_bit;
     }
-    pos= (char*) key_part->field->unpack(record + key_part->field->offset(),
-					 pos);
+    pos= (char*) key_part->field->unpack_key(record + key_part->field->offset(),
+                                             pos, key_part->length);
   }
 }