ndb - bug#14007 5.0 *** does not automerge into 5.1 ***

mysql-test/r/ndb_charset.result:
  bug#14007 5.0
mysql-test/t/ndb_charset.test:
  bug#14007 5.0
ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp:
  bug#14007 5.0
ndb/src/kernel/vm/SimulatedBlock.cpp:
  bug#14007 5.0
ndb/src/kernel/vm/SimulatedBlock.hpp:
  bug#14007 5.0

3 files changed, 62 insertions, 58 deletions

diff --git a/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp b/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
index db916b2d4d2..8a55777ac05 100644
--- a/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
+++ b/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
@@ -685,22 +685,16 @@ Dbtup::checkUpdateOfPrimaryKey(Uint32* updateBuffer, Tablerec* const regTabPtr)
   Uint32 attrDescriptor = tableDescriptor[attrDescriptorIndex].tabDescr;
   Uint32 attributeOffset = tableDescriptor[attrDescriptorIndex + 1].tabDescr;
 
-  Uint32 xfrmBuffer[1 + MAX_KEY_SIZE_IN_WORDS * 1]; // strxfrm_multiply == 1
+  Uint32 xfrmBuffer[1 + MAX_KEY_SIZE_IN_WORDS * MAX_XFRM_MULTIPLY];
   Uint32 charsetFlag = AttributeOffset::getCharsetFlag(attributeOffset);
   if (charsetFlag) {
-    Uint32 csPos = AttributeOffset::getCharsetPos(attributeOffset);
-    CHARSET_INFO* cs = regTabPtr->charsetArray[csPos];
-    Uint32 sizeInBytes = AttributeDescriptor::getSizeInBytes(attrDescriptor);
-    Uint32 sizeInWords = AttributeDescriptor::getSizeInWords(attrDescriptor);
-    const uchar* srcPtr = (uchar*)&updateBuffer[1];
-    uchar* dstPtr = (uchar*)&xfrmBuffer[1];
-    Uint32 n =
-      (*cs->coll->strnxfrm)(cs, dstPtr, sizeInBytes, srcPtr, sizeInBytes);
-    // pad with blanks (unlikely) and zeroes to match NDB API behaviour
-    while (n < sizeInBytes)
-      dstPtr[n++] = 0x20;
-    while (n < 4 * sizeInWords)
-      dstPtr[n++] = 0;
+    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;
   }
diff --git a/ndb/src/kernel/vm/SimulatedBlock.cpp b/ndb/src/kernel/vm/SimulatedBlock.cpp
index d708052ca4e..4625cd43640 100644
--- a/ndb/src/kernel/vm/SimulatedBlock.cpp
+++ b/ndb/src/kernel/vm/SimulatedBlock.cpp
@@ -1868,49 +1868,9 @@ SimulatedBlock::xfrm_key(Uint32 tab, const Uint32* src,
   while (i < noOfKeyAttr) 
   {
     const KeyDescriptor::KeyAttr& keyAttr = desc->keyAttr[i];
-    
-    Uint32 srcBytes = 
-      AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
-    Uint32 srcWords = (srcBytes + 3) / 4;
-    Uint32 dstWords = ~0;
-    uchar* dstPtr = (uchar*)&dst[dstPos];
-    const uchar* srcPtr = (const uchar*)&src[srcPos];
-    CHARSET_INFO* cs = keyAttr.charsetInfo;
-    
-    if (cs == NULL) 
-    {
-      jam();
-      memcpy(dstPtr, srcPtr, srcWords << 2);
-      dstWords = srcWords;
-    } 
-    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);
-    }
-    dstPos += dstWords;
-    srcPos += srcWords;
+    Uint32 dstWords =
+      xfrm_attr(keyAttr.attributeDescriptor, keyAttr.charsetInfo,
+                src, srcPos, dst, dstPos, dstSize);
     keyPartLen[i++] = dstWords;
   }
 
@@ -1918,6 +1878,52 @@ SimulatedBlock::xfrm_key(Uint32 tab, const Uint32* src,
 }
 
 Uint32
+SimulatedBlock::xfrm_attr(Uint32 attrDesc, CHARSET_INFO* cs,
+                          const Uint32* src, Uint32 & srcPos,
+                          Uint32* dst, Uint32 & dstPos, Uint32 dstSize) const
+{
+  Uint32 srcBytes = AttributeDescriptor::getSizeInBytes(attrDesc);
+  Uint32 srcWords = (srcBytes + 3) / 4;
+  Uint32 dstWords = ~0;
+  uchar* dstPtr = (uchar*)&dst[dstPos];
+  const uchar* srcPtr = (const uchar*)&src[srcPos];
+  
+  if (cs == NULL) 
+  {
+    jam();
+    memcpy(dstPtr, srcPtr, srcWords << 2);
+    dstWords = srcWords;
+  } 
+  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);
+  }
+  dstPos += dstWords;
+  srcPos += srcWords;
+  return dstWords;
+}
+
+Uint32
 SimulatedBlock::create_distr_key(Uint32 tableId,
 				 Uint32 *data, 
 				 const Uint32 
diff --git a/ndb/src/kernel/vm/SimulatedBlock.hpp b/ndb/src/kernel/vm/SimulatedBlock.hpp
index ce77fa916d8..b7bd8c57ee8 100644
--- a/ndb/src/kernel/vm/SimulatedBlock.hpp
+++ b/ndb/src/kernel/vm/SimulatedBlock.hpp
@@ -395,8 +395,12 @@ protected:
    * @return length
    */
   Uint32 xfrm_key(Uint32 tab, const Uint32* src, 
-		  Uint32 *dst, Uint32 dstLen,
+		  Uint32 *dst, Uint32 dstSize,
 		  Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX]) const;
+
+  Uint32 xfrm_attr(Uint32 attrDesc, CHARSET_INFO* cs,
+                   const Uint32* src, Uint32 & srcPos,
+                   Uint32* dst, Uint32 & dstPos, Uint32 dstSize) const;
   
   /**
    *