Merge bk-internal.mysql.com:/home/bk/mysql-5.1-new

into  mysql.com:/opt/local/work/mysql-5.1-new

mysql-test/r/ndb_charset.result

@@ -306,11 +306,21 @@ count(*)
 drop table t1;
 create table t1 (
 a char(10) primary key
-) engine=ndb;
-insert into t1 values ('jonas % ');
-replace into t1 values ('jonas % ');
-replace into t1 values ('jonas % ');
+) engine=ndbcluster default charset=latin1;
+insert into t1 values ('aaabb');
+select * from t1;
+a
+aaabb
+replace into t1 set a = 'AAABB';
+select * from t1;
+a
+AAABB
+replace into t1 set a = 'aAaBb';
+select * from t1;
+a
+aAaBb
+replace into t1 set a = 'aaabb';
 select * from t1;
 a
-jonas %
+aaabb
 drop table t1;

mysql-test/t/ndb_charset.test

@@ -237,13 +237,18 @@ drop table t1;
 #select a,b,length(a),length(b) from t1 where a='c' and b='c';
 #drop table t1;
 
-# bug 
+# bug#14007
 create table t1 (
   a char(10) primary key
-) engine=ndb;
-insert into t1 values ('jonas % ');
-replace into t1 values ('jonas % ');
-replace into t1 values ('jonas % ');
+) engine=ndbcluster default charset=latin1;
+
+insert into t1 values ('aaabb');
+select * from t1;
+replace into t1 set a = 'AAABB';
+select * from t1;
+replace into t1 set a = 'aAaBb';
+select * from t1;
+replace into t1 set a = 'aaabb';
 select * from t1;
 drop table t1;
 

storage/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp

@@ -800,22 +800,37 @@ Dbtup::checkUpdateOfPrimaryKey(KeyReqStruct* req_struct,
 {
   Uint32 keyReadBuffer[MAX_KEY_SIZE_IN_WORDS];
   Uint32 attributeHeader;
-  TableDescriptor* attr_descr= req_struct->attr_descr;
-  AttributeHeader* ahOut= (AttributeHeader*)&attributeHeader;
+  TableDescriptor* attr_descr = req_struct->attr_descr;
+  AttributeHeader* ahOut = (AttributeHeader*)&attributeHeader;
   AttributeHeader ahIn(*updateBuffer);
-  Uint32 attributeId= ahIn.getAttributeId();
-  Uint32 attrDescriptorIndex= attributeId << ZAD_LOG_SIZE;
-  Uint32 attrDescriptor= attr_descr[attrDescriptorIndex].tabDescr;
-  Uint32 attributeOffset= attr_descr[attrDescriptorIndex + 1].tabDescr;
-  ReadFunction f= regTabPtr->readFunctionArray[attributeId];
+  Uint32 attributeId = ahIn.getAttributeId();
+  Uint32 attrDescriptorIndex = attributeId << ZAD_LOG_SIZE;
+  Uint32 attrDescriptor = attr_descr[attrDescriptorIndex].tabDescr;
+  Uint32 attributeOffset = attr_descr[attrDescriptorIndex + 1].tabDescr;
+
+  Uint32 xfrmBuffer[1 + MAX_KEY_SIZE_IN_WORDS * MAX_XFRM_MULTIPLY];
+  Uint32 charsetFlag = AttributeOffset::getCharsetFlag(attributeOffset);
+  if (charsetFlag) {
+    Uint32 csIndex = AttributeOffset::getCharsetPos(attributeOffset);
+    CHARSET_INFO* cs = regTabPtr->charsetArray[csIndex];
+    Uint32 srcPos = 0;
+    Uint32 dstPos = 0;
+    xfrm_attr(attrDescriptor, cs, &updateBuffer[1], srcPos,
+              &xfrmBuffer[1], dstPos, MAX_KEY_SIZE_IN_WORDS * MAX_XFRM_MULTIPLY);
+    ahIn.setDataSize(dstPos);
+    xfrmBuffer[0] = ahIn.m_value;
+    updateBuffer = xfrmBuffer;
+  }
+
+  ReadFunction f = regTabPtr->readFunctionArray[attributeId];
 
   AttributeHeader::init(&attributeHeader, attributeId, 0);
-  req_struct->out_buf_index= 0;
-  req_struct->max_read= MAX_KEY_SIZE_IN_WORDS;
-  req_struct->attr_descriptor= attrDescriptor;
+  req_struct->out_buf_index = 0;
+  req_struct->max_read = MAX_KEY_SIZE_IN_WORDS;
+  req_struct->attr_descriptor = attrDescriptor;
 
   bool tmp = req_struct->xfrm_flag;
-  req_struct->xfrm_flag = false;
+  req_struct->xfrm_flag = true;
   ndbrequire((this->*f)(&keyReadBuffer[0],
                         req_struct,
                         ahOut,

storage/ndb/src/kernel/vm/SimulatedBlock.cpp

@@ -1882,79 +1882,9 @@ SimulatedBlock::xfrm_key(Uint32 tab, const Uint32* src,
   while (i < noOfKeyAttr) 
   {
     const KeyDescriptor::KeyAttr& keyAttr = desc->keyAttr[i];
-    
-    Uint32 array = 
-      AttributeDescriptor::getArrayType(keyAttr.attributeDescriptor);
-    Uint32 srcBytes = 
-      AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
-
-    Uint32 srcWords = ~0;
-    Uint32 dstWords = ~0;
-    uchar* dstPtr = (uchar*)&dst[dstPos];
-    const uchar* srcPtr = (const uchar*)&src[srcPos];
-    CHARSET_INFO* cs = keyAttr.charsetInfo;
-    
-    if (cs == NULL)
-    {
-      jam();
-      Uint32 len;
-      switch(array){
-      case NDB_ARRAYTYPE_SHORT_VAR:
-	len = 1 + srcPtr[0];
-	break;
-      case NDB_ARRAYTYPE_MEDIUM_VAR:
-	len = 2 + srcPtr[0] + (srcPtr[1] << 8);
-	break;
-#ifndef VM_TRACE
-      default:
-#endif
-      case NDB_ARRAYTYPE_FIXED:
-	len = srcBytes;
-      }
-      srcWords = (len + 3) >> 2;
-      dstWords = srcWords;
-      memcpy(dstPtr, srcPtr, dstWords << 2);
-      
-      if (0)
-      {
-	ndbout_c("srcPos: %d dstPos: %d len: %d srcWords: %d dstWords: %d",
-		 srcPos, dstPos, len, srcWords, dstWords);
-	
-	for(Uint32 i = 0; i<srcWords; i++)
-	  printf("%.8x ", src[srcPos + i]);
-	printf("\n");
-      }
-    } 
-    else
-    {
-      jam();
-      Uint32 typeId =
-	AttributeDescriptor::getType(keyAttr.attributeDescriptor);
-      Uint32 lb, len;
-      bool ok = NdbSqlUtil::get_var_length(typeId, srcPtr, srcBytes, lb, len);
-      ndbrequire(ok);
-      Uint32 xmul = cs->strxfrm_multiply;
-      if (xmul == 0)
-	xmul = 1;
-      /*
-       * Varchar is really Char.  End spaces do not matter.  To get
-       * same hash we blank-pad to maximum length via strnxfrm.
-       * TODO use MySQL charset-aware hash function instead
-       */
-      Uint32 dstLen = xmul * (srcBytes - lb);
-      ndbrequire(dstLen <= ((dstSize - dstPos) << 2));
-      int n = NdbSqlUtil::strnxfrm_bug7284(cs, dstPtr, dstLen, srcPtr + lb, len);
-      ndbrequire(n != -1);
-      while ((n & 3) != 0) 
-      {
-	dstPtr[n++] = 0;
-      }
-      dstWords = (n >> 2);
-      srcWords = (lb + len + 3) >> 2; 
-    }
-
-    dstPos += dstWords;
-    srcPos += srcWords;
+    Uint32 dstWords =
+      xfrm_attr(keyAttr.attributeDescriptor, keyAttr.charsetInfo,
+                src, srcPos, dst, dstPos, dstSize);
     keyPartLen[i++] = dstWords;
   }
 
@@ -1969,6 +1899,84 @@ SimulatedBlock::xfrm_key(Uint32 tab, const Uint32* src,
   return dstPos;
 }
 
+Uint32
+SimulatedBlock::xfrm_attr(Uint32 attrDesc, CHARSET_INFO* cs,
+                          const Uint32* src, Uint32 & srcPos,
+                          Uint32* dst, Uint32 & dstPos, Uint32 dstSize) const
+{
+  Uint32 array = 
+    AttributeDescriptor::getArrayType(attrDesc);
+  Uint32 srcBytes = 
+    AttributeDescriptor::getSizeInBytes(attrDesc);
+
+  Uint32 srcWords = ~0;
+  Uint32 dstWords = ~0;
+  uchar* dstPtr = (uchar*)&dst[dstPos];
+  const uchar* srcPtr = (const uchar*)&src[srcPos];
+  
+  if (cs == NULL)
+  {
+    jam();
+    Uint32 len;
+    switch(array){
+    case NDB_ARRAYTYPE_SHORT_VAR:
+      len = 1 + srcPtr[0];
+      break;
+    case NDB_ARRAYTYPE_MEDIUM_VAR:
+      len = 2 + srcPtr[0] + (srcPtr[1] << 8);
+      break;
+#ifndef VM_TRACE
+    default:
+#endif
+    case NDB_ARRAYTYPE_FIXED:
+      len = srcBytes;
+    }
+    srcWords = (len + 3) >> 2;
+    dstWords = srcWords;
+    memcpy(dstPtr, srcPtr, dstWords << 2);
+    
+    if (0)
+    {
+      ndbout_c("srcPos: %d dstPos: %d len: %d srcWords: %d dstWords: %d",
+               srcPos, dstPos, len, srcWords, dstWords);
+      
+      for(Uint32 i = 0; i<srcWords; i++)
+        printf("%.8x ", src[srcPos + i]);
+      printf("\n");
+    }
+  } 
+  else
+  {
+    jam();
+    Uint32 typeId =
+      AttributeDescriptor::getType(attrDesc);
+    Uint32 lb, len;
+    bool ok = NdbSqlUtil::get_var_length(typeId, srcPtr, srcBytes, lb, len);
+    ndbrequire(ok);
+    Uint32 xmul = cs->strxfrm_multiply;
+    if (xmul == 0)
+      xmul = 1;
+    /*
+     * Varchar end-spaces are ignored in comparisons.  To get same hash
+     * we blank-pad to maximum length via strnxfrm.
+     */
+    Uint32 dstLen = xmul * (srcBytes - lb);
+    ndbrequire(dstLen <= ((dstSize - dstPos) << 2));
+    int n = NdbSqlUtil::strnxfrm_bug7284(cs, dstPtr, dstLen, srcPtr + lb, len);
+    ndbrequire(n != -1);
+    while ((n & 3) != 0) 
+    {
+      dstPtr[n++] = 0;
+    }
+    dstWords = (n >> 2);
+    srcWords = (lb + len + 3) >> 2; 
+  }
+
+  dstPos += dstWords;
+  srcPos += srcWords;
+  return dstWords;
+}
+
 Uint32
 SimulatedBlock::create_distr_key(Uint32 tableId,
 				 Uint32 *data,