Fixed bug #13191.

INSERT...ON DUPLICATE KEY UPDATE may cause error 1032: 
"Can't find record in ..." if we are inserting into
InnoDB table unique index of partial key with
underlying UTF-8 string field.

This error occurs because INSERT...ON DUPLICATE uses a wrong
procedure to copy string fields of multi-byte character sets
for index search.


mysql-test/t/innodb_mysql.test:
  Added test case for bug #13191.
mysql-test/r/innodb_mysql.result:
  Added test case for bug #13191.
sql/field.h:
  Fixed bug #13191.
  Field_string::get_key_image() virtual function was overloaded
  to implement copying of variable length character (UTF-8) fields.
  Field::get_key_image() function prototype has been changed to
  return byte size of copied data.
sql/field.cc:
  Fixed bug #13191.
  Field_string::get_key_image() virtual function was overloaded
  to implement copying of variable length character (UTF-8) fields.
  Field::get_key_image() function prototype has been changed to
  return byte size of copied data.
sql/key.cc:
  Fixed bug #13191.
  INSERT...ON DUPLICATE KEY UPDATE may cause error 1032: 
  "Can't find record in ...".
  This error occurs because INSERT...ON DUPLICATE uses
  a wrong procedure to copy field parts for index search.
  key_copy() function has been fixed.

mysql-test/r/innodb_mysql.result

@@ -128,4 +128,37 @@ show /*!50002 GLOBAL */ status like 'Handler_rollback';
 Variable_name	Value
 Handler_rollback	0
 drop table t1;
+CREATE TABLE t1(c1 TEXT, UNIQUE (c1(1)), cnt INT DEFAULT 1)
+ENGINE=INNODB CHARACTER SET UTF8;
+INSERT INTO t1 (c1) VALUES ('1a');
+SELECT * FROM t1;
+c1	cnt
+1a	1
+INSERT INTO t1 (c1) VALUES ('1b') ON DUPLICATE KEY UPDATE cnt=cnt+1;
+SELECT * FROM t1;
+c1	cnt
+1a	2
+DROP TABLE t1;
+CREATE TABLE t1(c1 VARCHAR(2), UNIQUE (c1(1)), cnt INT DEFAULT 1)
+ENGINE=INNODB CHARACTER SET UTF8;
+INSERT INTO t1 (c1) VALUES ('1a');
+SELECT * FROM t1;
+c1	cnt
+1a	1
+INSERT INTO t1 (c1) VALUES ('1b') ON DUPLICATE KEY UPDATE cnt=cnt+1;
+SELECT * FROM t1;
+c1	cnt
+1a	2
+DROP TABLE t1;
+CREATE TABLE t1(c1 CHAR(2), UNIQUE (c1(1)), cnt INT DEFAULT 1)
+ENGINE=INNODB CHARACTER SET UTF8;
+INSERT INTO t1 (c1) VALUES ('1a');
+SELECT * FROM t1;
+c1	cnt
+1a	1
+INSERT INTO t1 (c1) VALUES ('1b') ON DUPLICATE KEY UPDATE cnt=cnt+1;
+SELECT * FROM t1;
+c1	cnt
+1a	2
+DROP TABLE t1;
 End of 4.1 tests

mysql-test/t/innodb_mysql.test

@@ -161,4 +161,34 @@ show /*!50002 GLOBAL */ status like 'Handler_rollback';
 connection default;
 drop table t1;
 disconnect con1;
+
+#
+# Bug #13191: INSERT...ON DUPLICATE KEY UPDATE of UTF-8 string fields
+# used in partial unique indices.
+#
+
+CREATE TABLE t1(c1 TEXT, UNIQUE (c1(1)), cnt INT DEFAULT 1)
+  ENGINE=INNODB CHARACTER SET UTF8;
+INSERT INTO t1 (c1) VALUES ('1a');
+SELECT * FROM t1;
+INSERT INTO t1 (c1) VALUES ('1b') ON DUPLICATE KEY UPDATE cnt=cnt+1;
+SELECT * FROM t1;
+DROP TABLE t1;
+
+CREATE TABLE t1(c1 VARCHAR(2), UNIQUE (c1(1)), cnt INT DEFAULT 1)
+  ENGINE=INNODB CHARACTER SET UTF8;
+INSERT INTO t1 (c1) VALUES ('1a');
+SELECT * FROM t1;
+INSERT INTO t1 (c1) VALUES ('1b') ON DUPLICATE KEY UPDATE cnt=cnt+1;
+SELECT * FROM t1;
+DROP TABLE t1;
+
+CREATE TABLE t1(c1 CHAR(2), UNIQUE (c1(1)), cnt INT DEFAULT 1)
+  ENGINE=INNODB CHARACTER SET UTF8;
+INSERT INTO t1 (c1) VALUES ('1a');
+SELECT * FROM t1;
+INSERT INTO t1 (c1) VALUES ('1b') ON DUPLICATE KEY UPDATE cnt=cnt+1;
+SELECT * FROM t1;
+DROP TABLE t1;
+
 --echo End of 4.1 tests

sql/field.cc

@@ -5204,6 +5204,16 @@ uint Field_string::max_packed_col_length(uint max_length)
   return (max_length > 255 ? 2 : 1)+max_length;
 }
 
+uint Field_string::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
+                                 imagetype type_arg)
+{
+  uint bytes = my_charpos(cs, ptr, ptr + field_length,
+                          length / field_charset->mbmaxlen);
+  memcpy(buff, ptr, bytes);
+  if (bytes < length)
+    bzero(buff + bytes, length - bytes);
+  return bytes;
+}
 
 /****************************************************************************
 ** VARCHAR type  (Not available for the end user yet)
@@ -5414,7 +5424,7 @@ uint Field_varstring::max_packed_col_length(uint max_length)
   return (max_length > 255 ? 2 : 1)+max_length;
 }
 
-void Field_varstring::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
+uint Field_varstring::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
                                     imagetype type)
 {
   uint f_length=uint2korr(ptr);
@@ -5426,6 +5436,7 @@ void Field_varstring::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
   if (f_length < length)
     bzero(buff+HA_KEY_BLOB_LENGTH+f_length, (length-f_length));
 #endif
+  return HA_KEY_BLOB_LENGTH+f_length;
 }
 
 void Field_varstring::set_key_image(char *buff,uint length, CHARSET_INFO *cs)
@@ -5724,7 +5735,7 @@ int Field_blob::cmp_binary(const char *a_ptr, const char *b_ptr,
 
 /* The following is used only when comparing a key */
 
-void Field_blob::get_key_image(char *buff,uint length,
+uint Field_blob::get_key_image(char *buff,uint length,
                                CHARSET_INFO *cs, imagetype type)
 {
   uint32 blob_length= get_length(ptr);
@@ -5737,16 +5748,17 @@ void Field_blob::get_key_image(char *buff,uint length,
     MBR mbr;
     Geometry_buffer buffer;
     Geometry *gobj;
+    const uint image_length= SIZEOF_STORED_DOUBLE*4;
 
     if (blob_length < SRID_SIZE)
     {
-      bzero(buff, SIZEOF_STORED_DOUBLE*4);
-      return;
+      bzero(buff, image_length);
+      return image_length;
     }
     get_ptr(&blob);
     gobj= Geometry::construct(&buffer, blob, blob_length);
     if (gobj->get_mbr(&mbr, &dummy))
-      bzero(buff, SIZEOF_STORED_DOUBLE*4);
+      bzero(buff, image_length);
     else
     {
       float8store(buff,    mbr.xmin);
@@ -5754,7 +5766,7 @@ void Field_blob::get_key_image(char *buff,uint length,
       float8store(buff+16, mbr.ymin);
       float8store(buff+24, mbr.ymax);
     }
-    return;
+    return image_length;
   }
 #endif /*HAVE_SPATIAL*/
 
@@ -5774,6 +5786,7 @@ void Field_blob::get_key_image(char *buff,uint length,
   }
   int2store(buff,length);
   memcpy(buff+HA_KEY_BLOB_LENGTH, blob, length);
+  return HA_KEY_BLOB_LENGTH+length;
 }
 
 void Field_blob::set_key_image(char *buff,uint length, CHARSET_INFO *cs)
@@ -6021,7 +6034,7 @@ uint Field_blob::max_packed_col_length(uint max_length)
 
 #ifdef HAVE_SPATIAL
 
-void Field_geom::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
+uint Field_geom::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
                                imagetype type)
 {
   char *blob;
@@ -6030,16 +6043,17 @@ void Field_geom::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
   ulong blob_length= get_length(ptr);
   Geometry_buffer buffer;
   Geometry *gobj;
+  const uint image_length= SIZEOF_STORED_DOUBLE*4;
 
   if (blob_length < SRID_SIZE)
   {
-    bzero(buff, SIZEOF_STORED_DOUBLE*4);
-    return;
+    bzero(buff, image_length);
+    return image_length;
   }
   get_ptr(&blob);
   gobj= Geometry::construct(&buffer, blob, blob_length);
   if (gobj->get_mbr(&mbr, &dummy))
-    bzero(buff, SIZEOF_STORED_DOUBLE*4);
+    bzero(buff, image_length);
   else
   {
     float8store(buff, mbr.xmin);
@@ -6047,6 +6061,7 @@ void Field_geom::get_key_image(char *buff, uint length, CHARSET_INFO *cs,
     float8store(buff + 16, mbr.ymin);
     float8store(buff + 24, mbr.ymax);
   }
+  return image_length;
 }
 
 

sql/field.h

@@ -228,9 +228,43 @@ class Field
     { memcpy(buff,ptr,length); }
   inline void set_image(char *buff,uint length, CHARSET_INFO *cs)
     { memcpy(ptr,buff,length); }
-  virtual void get_key_image(char *buff,uint length, CHARSET_INFO *cs,
+
+
+  /*
+    Copy a field part into an output buffer.
+
+    SYNOPSIS
+      Field::get_key_image()
+      buff   [out] output buffer
+      length       output buffer size
+      cs           charset, always same as this->charset(),
+                   (to be removed in 5.x)
+      type         itMBR for geometry blobs, otherwise itRAW
+
+    DESCRIPTION
+      This function makes a copy of field part of size equal to or
+      less than "length" parameter value.
+      For fields of string types (CHAR, VARCHAR, TEXT) the rest of buffer
+      is padded by zero byte.
+
+    NOTES
+      For variable length character fields (i.e. UTF-8) the "length"
+      parameter means a number of output buffer bytes as if all field
+      characters have maximal possible size (mbmaxlen). In the other words,
+      "length" parameter is a number of characters multiplied by
+      field_charset->mbmaxlen.
+
+    RETURN
+      Number of copied bytes (excluding padded zero bytes -- see above).
+  */
+
+  virtual uint get_key_image(char *buff, uint length,
+                             CHARSET_INFO *cs,
                              imagetype type)
-    { get_image(buff,length,cs); }
+  {
+    get_image(buff,length,cs);
+    return length;
+  }
   virtual void set_key_image(char *buff,uint length, CHARSET_INFO *cs)
     { set_image(buff,length,cs); }
   inline longlong val_int_offset(uint row_offset)
@@ -947,6 +981,8 @@ class Field_string :public Field_str {
   enum_field_types real_type() const { return FIELD_TYPE_STRING; }
   bool has_charset(void) const
   { return charset() == &my_charset_bin ? FALSE : TRUE; }
+  virtual uint get_key_image(char *buff,uint length, CHARSET_INFO *cs,
+                             imagetype type);
 };
 
 
@@ -981,7 +1017,7 @@ class Field_varstring :public Field_str {
   String *val_str(String*,String *);
   int cmp(const char *,const char*);
   void sort_string(char *buff,uint length);
-  void get_key_image(char *buff,uint length, CHARSET_INFO *cs, imagetype type);
+  uint get_key_image(char *buff,uint length, CHARSET_INFO *cs, imagetype type);
   void set_key_image(char *buff,uint length, CHARSET_INFO *cs);
   void sql_type(String &str) const;
   char *pack(char *to, const char *from, uint max_length=~(uint) 0);
@@ -1060,7 +1096,7 @@ class Field_blob :public Field_str {
       store_length(length);
       memcpy_fixed(ptr+packlength,&data,sizeof(char*));
     }
-  void get_key_image(char *buff,uint length, CHARSET_INFO *cs, imagetype type);
+  uint get_key_image(char *buff,uint length, CHARSET_INFO *cs, imagetype type);
   void set_key_image(char *buff,uint length, CHARSET_INFO *cs);
   void sql_type(String &str) const;
   inline bool copy()
@@ -1117,7 +1153,7 @@ class Field_geom :public Field_blob {
   int  store(longlong nr) { return 1; }
   int  reset(void) { return !maybe_null() || Field_blob::reset(); }
 
-  void get_key_image(char *buff,uint length, CHARSET_INFO *cs,imagetype type);
+  uint get_key_image(char *buff,uint length, CHARSET_INFO *cs,imagetype type);
   void set_key_image(char *buff,uint length, CHARSET_INFO *cs);
 };
 #endif /*HAVE_SPATIAL*/

sql/key.cc

@@ -89,20 +89,21 @@ void key_copy(byte *key,TABLE *table,uint idx,uint key_length)
     }
     if (key_part->key_part_flag & HA_BLOB_PART)
     {
-      char *pos;
-      ulong blob_length=((Field_blob*) key_part->field)->get_length();
-      key_length-=2;
-      ((Field_blob*) key_part->field)->get_ptr(&pos);
-      length=min(key_length,key_part->length);
-      set_if_smaller(blob_length,length);
-      int2store(key,(uint) blob_length);
-      key+=2;					// Skip length info
-      memcpy(key,pos,blob_length);
+      key_length-= HA_KEY_BLOB_LENGTH;
+      length= min(key_length, key_part->length);
+      key_part->field->get_key_image((char *) key, length,
+                                     key_part->field->charset(),
+                                     Field::itRAW);
+      key+= HA_KEY_BLOB_LENGTH;
     }
     else
     {
-      length=min(key_length,key_part->length);
-      memcpy(key,table->record[0]+key_part->offset,(size_t) length);
+      length= min(key_length, key_part->length);
+      Field *field= key_part->field;
+      CHARSET_INFO *cs= field->charset();
+      uint bytes= field->get_key_image(key, length, cs, Field::itRAW);
+      if (bytes < length)
+        cs->cset->fill(cs, key + bytes, length - bytes, ' ');
     }
     key+=length;
     key_length-=length;