bug#12220 - ndb - node recovery with charsets

  LQH computes incorrect hash values during NR (as it doesn't concider charsets)

  Solution: make LQH compute correct hash :-)
           1) move xfrm_key into SimulatedBlock so that there's _one_ impl.
           2) make TC, ACC, LQH use same impl.


ndb/include/kernel/AttributeDescriptor.hpp:
  Make SimulatedBlock use AttributeDescriptor (to xfrm)
ndb/src/kernel/blocks/dbacc/Dbacc.hpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbacc/DbaccInit.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbacc/DbaccMain.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbdict/Dbdict.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dblqh/Dblqh.hpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dblqh/DblqhMain.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbtc/Dbtc.hpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbtc/DbtcMain.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/vm/SimulatedBlock.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/vm/SimulatedBlock.hpp:
  Move xfrm handling into SimulatedBlock

11 files changed, 238 insertions, 319 deletions

diff --git a/ndb/include/kernel/AttributeDescriptor.hpp b/ndb/include/kernel/AttributeDescriptor.hpp
index af28e777213..2fe7c9f0973 100644
--- a/ndb/include/kernel/AttributeDescriptor.hpp
+++ b/ndb/include/kernel/AttributeDescriptor.hpp
@@ -23,7 +23,8 @@ class AttributeDescriptor {
   friend class Dbacc;
   friend class Dbtup;
   friend class Dbtux;
-  
+  friend class SimulatedBlock;
+
 private:
   static void setType(Uint32 &, Uint32 type);
   static void setSize(Uint32 &, Uint32 size);
diff --git a/ndb/src/kernel/blocks/dbacc/Dbacc.hpp b/ndb/src/kernel/blocks/dbacc/Dbacc.hpp
index a2d6fe4d64a..8800388e96c 100644
--- a/ndb/src/kernel/blocks/dbacc/Dbacc.hpp
+++ b/ndb/src/kernel/blocks/dbacc/Dbacc.hpp
@@ -851,13 +851,6 @@ struct Tabrec {
   Uint32 fragptrholder[MAX_FRAG_PER_NODE];
   Uint32 tabUserPtr;
   BlockReference tabUserRef;
-
-  Uint8 noOfKeyAttr;
-  Uint8 hasCharAttr;
-  struct KeyAttr {
-    Uint32 attributeDescriptor;
-    CHARSET_INFO* charsetInfo;
-  } keyAttr[MAX_ATTRIBUTES_IN_INDEX];
 };
   typedef Ptr<Tabrec> TabrecPtr;
 
@@ -903,7 +896,6 @@ private:
   void execACCKEYREQ(Signal* signal);
   void execACCSEIZEREQ(Signal* signal);
   void execACCFRAGREQ(Signal* signal);
-  void execTC_SCHVERREQ(Signal* signal);
   void execACC_SRREQ(Signal* signal);
   void execNEXT_SCANREQ(Signal* signal);
   void execACC_ABORTREQ(Signal* signal);
diff --git a/ndb/src/kernel/blocks/dbacc/DbaccInit.cpp b/ndb/src/kernel/blocks/dbacc/DbaccInit.cpp
index 90839163a72..d4ff20e4c43 100644
--- a/ndb/src/kernel/blocks/dbacc/DbaccInit.cpp
+++ b/ndb/src/kernel/blocks/dbacc/DbaccInit.cpp
@@ -179,7 +179,6 @@ Dbacc::Dbacc(const class Configuration & conf):
   addRecSignal(GSN_ACCKEYREQ, &Dbacc::execACCKEYREQ);
   addRecSignal(GSN_ACCSEIZEREQ, &Dbacc::execACCSEIZEREQ);
   addRecSignal(GSN_ACCFRAGREQ, &Dbacc::execACCFRAGREQ);
-  addRecSignal(GSN_TC_SCHVERREQ, &Dbacc::execTC_SCHVERREQ);
   addRecSignal(GSN_ACC_SRREQ, &Dbacc::execACC_SRREQ);
   addRecSignal(GSN_NEXT_SCANREQ, &Dbacc::execNEXT_SCANREQ);
   addRecSignal(GSN_ACC_ABORTREQ, &Dbacc::execACC_ABORTREQ);
diff --git a/ndb/src/kernel/blocks/dbacc/DbaccMain.cpp b/ndb/src/kernel/blocks/dbacc/DbaccMain.cpp
index aefd2612151..37b6825cbd2 100644
--- a/ndb/src/kernel/blocks/dbacc/DbaccMain.cpp
+++ b/ndb/src/kernel/blocks/dbacc/DbaccMain.cpp
@@ -28,7 +28,8 @@
 #include <signaldata/FsRemoveReq.hpp>
 #include <signaldata/DropTab.hpp>
 #include <signaldata/DumpStateOrd.hpp>
-#include <SectionReader.hpp>
+#include <KeyDescriptor.hpp>
+
 
 // TO_DO_RONM is a label for comments on what needs to be improved in future versions
 // when more time is given.
@@ -1037,12 +1038,6 @@ void Dbacc::initialiseTableRec(Signal* signal)
       tabptr.p->fragholder[i] = RNIL;
       tabptr.p->fragptrholder[i] = RNIL;
     }//for
-    tabptr.p->noOfKeyAttr = 0;
-    tabptr.p->hasCharAttr = 0;
-    for (Uint32 k = 0; k < MAX_ATTRIBUTES_IN_INDEX; k++) {
-      tabptr.p->keyAttr[k].attributeDescriptor = 0;
-      tabptr.p->keyAttr[k].charsetInfo = 0;
-    }
   }//for
 }//Dbacc::initialiseTableRec()
 
@@ -1172,8 +1167,8 @@ void Dbacc::execACCFRAGREQ(Signal* signal)
   Uint32 userPtr = req->userPtr;
   BlockReference retRef = req->userRef;
   rootfragrecptr.p->rootState = ACTIVEROOT;
-  AccFragConf * const conf = (AccFragConf*)&signal->theData[0];
 
+  AccFragConf * const conf = (AccFragConf*)&signal->theData[0];
   conf->userPtr = userPtr;
   conf->rootFragPtr = rootfragrecptr.i;
   conf->fragId[0] = rootfragrecptr.p->fragmentid[0];
@@ -1197,65 +1192,6 @@ void Dbacc::addFragRefuse(Signal* signal, Uint32 errorCode)
   return;
 }//Dbacc::addFragRefuseEarly()
 
-void
-Dbacc::execTC_SCHVERREQ(Signal* signal)
-{
-  jamEntry();
-  if (! assembleFragments(signal)) {
-    jam();
-    return;
-  }
-  tabptr.i = signal->theData[0];
-  ptrCheckGuard(tabptr, ctablesize, tabrec);
-  Uint32 noOfKeyAttr = signal->theData[6];
-  ndbrequire(noOfKeyAttr <= MAX_ATTRIBUTES_IN_INDEX);
-  Uint32 hasCharAttr = 0;
-
-  SegmentedSectionPtr s0Ptr;
-  signal->getSection(s0Ptr, 0);
-  SectionReader r0(s0Ptr, getSectionSegmentPool());
-  Uint32 i = 0;
-  while (i < noOfKeyAttr) {
-    jam();
-    Uint32 attributeDescriptor = ~0;
-    Uint32 csNumber = ~0;
-    if (! r0.getWord(&attributeDescriptor) ||
-        ! r0.getWord(&csNumber)) {
-      jam();
-      break;
-    }
-    CHARSET_INFO* cs = 0;
-    if (csNumber != 0) {
-      cs = all_charsets[csNumber];
-      ndbrequire(cs != 0);
-      hasCharAttr = 1;
-    }
-    tabptr.p->keyAttr[i].attributeDescriptor = attributeDescriptor;
-    tabptr.p->keyAttr[i].charsetInfo = cs;
-    i++;
-  }
-  ndbrequire(i == noOfKeyAttr);
-  releaseSections(signal);
-
-  tabptr.p->noOfKeyAttr = noOfKeyAttr;
-  tabptr.p->hasCharAttr = hasCharAttr;
-
-  // copy char attr flag to each fragment
-  for (Uint32 i1 = 0; i1 < MAX_FRAG_PER_NODE; i1++) {
-    jam();
-    if (tabptr.p->fragptrholder[i1] != RNIL) {
-      rootfragrecptr.i = tabptr.p->fragptrholder[i1];
-      ptrCheckGuard(rootfragrecptr, crootfragmentsize, rootfragmentrec);
-      for (Uint32 i2 = 0; i2 < 2; i2++) {
-        fragrecptr.i = rootfragrecptr.p->fragmentptr[i2];
-        ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
-        fragrecptr.p->hasCharAttr = hasCharAttr;
-      }
-    }
-  }
-
-  // no reply to DICT
-}
 
 void
 Dbacc::execDROP_TAB_REQ(Signal* signal){
@@ -1841,55 +1777,14 @@ void Dbacc::execACCKEYREQ(Signal* signal)
 void
 Dbacc::xfrmKeyData(Signal* signal)
 {
-  tabptr.i = fragrecptr.p->myTableId;
-  ptrCheckGuard(tabptr, ctablesize, tabrec);
-
-  Uint32 dst[1024 * MAX_XFRM_MULTIPLY];
-  Uint32 dstSize = (sizeof(dst) >> 2);
+  Uint32 table = fragrecptr.p->myTableId;
+  Uint32 dst[MAX_KEY_SIZE_IN_WORDS * MAX_XFRM_MULTIPLY];
+  Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX];
   Uint32* src = &signal->theData[7];
-  const Uint32 noOfKeyAttr = tabptr.p->noOfKeyAttr;
-  Uint32 dstPos = 0;
-  Uint32 srcPos = 0;
-  Uint32 i = 0;
-
-  while (i < noOfKeyAttr) {
-    const Tabrec::KeyAttr& keyAttr = tabptr.p->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 == 0) {
-      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;
-      // see comment in DbtcMain.cpp
-      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;
-    i++;
-  }
-  memcpy(src, dst, dstPos << 2);
-  operationRecPtr.p->xfrmtupkeylen = dstPos;
+  Uint32 len = xfrm_key(table, src, dst, sizeof(dst) >> 2, keyPartLen);
+  ndbrequire(len); // 0 means error
+  memcpy(src, dst, len << 2);
+  operationRecPtr.p->xfrmtupkeylen = len;
 }
 
 void Dbacc::accIsLockedLab(Signal* signal) 
@@ -8024,6 +7919,10 @@ void Dbacc::initFragAdd(Signal* signal,
   Uint32 Tmp2 = regFragPtr.p->maxloadfactor - regFragPtr.p->minloadfactor;
   Tmp2 = Tmp1 * Tmp2;
   regFragPtr.p->slackCheck = Tmp2;
+
+  Uint32 hasCharAttr = g_key_descriptor_pool.getPtr(req->tableId)->hasCharAttr;
+  regFragPtr.p->hasCharAttr = hasCharAttr;
+
 }//Dbacc::initFragAdd()
 
 void Dbacc::initFragGeneral(FragmentrecPtr regFragPtr)
diff --git a/ndb/src/kernel/blocks/dbdict/Dbdict.cpp b/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
index 7d036b0e988..2725e7d14fe 100644
--- a/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
+++ b/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
@@ -27,6 +27,7 @@
 #include <SectionReader.hpp>
 #include <SimpleProperties.hpp>
 #include <AttributeHeader.hpp>
+#include <KeyDescriptor.hpp>
 #include <signaldata/DictSchemaInfo.hpp>
 #include <signaldata/DictTabInfo.hpp>
 #include <signaldata/DropTabFile.hpp>
@@ -1750,6 +1751,7 @@ void Dbdict::execREAD_CONFIG_REQ(Signal* signal)
   c_schemaPageRecordArray.setSize(2 * NDB_SF_MAX_PAGES);
   c_tableRecordPool.setSize(tablerecSize);
   c_tableRecordHash.setSize(tablerecSize);
+  g_key_descriptor_pool.setSize(tablerecSize);
   c_triggerRecordPool.setSize(c_maxNoOfTriggers);
   c_triggerRecordHash.setSize(c_maxNoOfTriggers);
   c_opRecordPool.setSize(256);   // XXX need config params
@@ -4450,6 +4452,44 @@ Dbdict::execADD_FRAGREQ(Signal* signal) {
     sendSignal(DBLQH_REF, GSN_LQHFRAGREQ, signal, 
 	       LqhFragReq::SignalLength, JBB);
   }
+
+  /**
+   * Create KeyDescriptor
+   */
+  KeyDescriptor* desc= g_key_descriptor_pool.getPtr(tabPtr.i);
+  new (desc) KeyDescriptor();
+
+  Uint32 key = 0;
+  Uint32 tAttr = tabPtr.p->firstAttribute;
+  while (tAttr != RNIL) 
+  {
+    jam();
+    AttributeRecord* aRec = c_attributeRecordPool.getPtr(tAttr);
+    if (aRec->tupleKey) 
+    {
+      desc->noOfKeyAttr ++;
+      desc->keyAttr[key].attributeDescriptor = aRec->attributeDescriptor;
+      
+      Uint32 csNumber = (aRec->extPrecision >> 16);
+      if(csNumber)
+      {
+	desc->keyAttr[key].charsetInfo = all_charsets[csNumber];
+	ndbrequire(all_charsets[csNumber]);
+	desc->hasCharAttr = 1;
+      }
+      else
+      {
+	desc->keyAttr[key].charsetInfo = 0;	  
+      }
+      if(AttributeDescriptor::getDKey(aRec->attributeDescriptor))
+      {
+	desc->noOfDistrKeys ++;
+      }
+      key++;
+    }
+    tAttr = aRec->nextAttrInTable;
+  }
+  ndbrequire(key == tabPtr.p->noOfPrimkey);
 }
 
 void
@@ -4644,31 +4684,11 @@ Dbdict::execTAB_COMMITCONF(Signal* signal){
     signal->theData[4] = (Uint32)tabPtr.p->tableType;
     signal->theData[5] = createTabPtr.p->key;
     signal->theData[6] = (Uint32)tabPtr.p->noOfPrimkey;
-
-    Uint32 buf[2 * MAX_ATTRIBUTES_IN_INDEX];
-    Uint32 sz = 0;
-    Uint32 tAttr = tabPtr.p->firstAttribute;
-    while (tAttr != RNIL) {
-      jam();
-      AttributeRecord* aRec = c_attributeRecordPool.getPtr(tAttr);
-      if (aRec->tupleKey) {
-        buf[sz++] = aRec->attributeDescriptor;
-        buf[sz++] = (aRec->extPrecision >> 16); // charset number
-      }
-      tAttr = aRec->nextAttrInTable;
-    }
-    ndbrequire((int)sz == 2 * tabPtr.p->noOfPrimkey);
-
-    LinearSectionPtr lsPtr[3];
-    lsPtr[0].p = buf;
-    lsPtr[0].sz = sz;
-    // note: ACC does not reply
-    if (tabPtr.p->isTable() || tabPtr.p->isHashIndex())
-      sendSignal(DBACC_REF, GSN_TC_SCHVERREQ, signal, 7, JBB, lsPtr, 1);
-    sendSignal(DBTC_REF, GSN_TC_SCHVERREQ, signal, 7, JBB, lsPtr, 1);
+    
+    sendSignal(DBTC_REF, GSN_TC_SCHVERREQ, signal, 7, JBB);
     return;
   }
-
+  
   ndbrequire(false);
 }
 
@@ -12342,3 +12362,5 @@ Dbdict::getMetaAttribute(MetaData::Attribute& attr, const MetaData::Table& table
   new (&attr) MetaData::Attribute(*attrPtr.p);
   return 0;
 }
+
+CArray<KeyDescriptor> g_key_descriptor_pool;
diff --git a/ndb/src/kernel/blocks/dblqh/Dblqh.hpp b/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
index fa7e8667e27..ce84834e808 100644
--- a/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
+++ b/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
@@ -2233,7 +2233,7 @@ private:
   void LQHKEY_abort(Signal* signal, int errortype);
   void LQHKEY_error(Signal* signal, int errortype);
   void nextRecordCopy(Signal* signal);
-  void calculateHash(Signal* signal);
+  Uint32 calculateHash(Uint32 tableId, const Uint32* src);
   void continueAfterCheckLcpStopBlocked(Signal* signal);
   void checkLcpStopBlockedLab(Signal* signal);
   void sendCommittedTc(Signal* signal, BlockReference atcBlockref);
diff --git a/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp b/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
index 5a35b5eefc0..71b5aaf4163 100644
--- a/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
+++ b/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
@@ -55,6 +55,7 @@
 #include <signaldata/AlterTab.hpp>
 
 #include <signaldata/LCP.hpp>
+#include <KeyDescriptor.hpp>
 
 // Use DEBUG to print messages that should be
 // seen only when we debug the product
@@ -9013,44 +9014,17 @@ void Dblqh::sendScanFragConf(Signal* signal, Uint32 scanCompleted)
 /*   FRAGMENT TO A NEW REPLICA OF THE FRAGMENT. IT DOES ALSO SHUT DOWN ALL   */
 /*   CONNECTIONS TO THE FAILED NODE.                                         */
 /*---------------------------------------------------------------------------*/
-void Dblqh::calculateHash(Signal* signal) 
-{
-  DatabufPtr locDatabufptr;
-  UintR  Ti;
-  UintR  Tdata0;
-  UintR  Tdata1;
-  UintR  Tdata2;
-  UintR  Tdata3;
-  UintR*  Tdata32;
-  Uint64 Tdata[512];
-
-  Tdata32 = (UintR*)&Tdata[0];
-
-  Tdata0 = tcConnectptr.p->tupkeyData[0];
-  Tdata1 = tcConnectptr.p->tupkeyData[1];
-  Tdata2 = tcConnectptr.p->tupkeyData[2];
-  Tdata3 = tcConnectptr.p->tupkeyData[3];
-  Tdata32[0] = Tdata0;
-  Tdata32[1] = Tdata1;
-  Tdata32[2] = Tdata2;
-  Tdata32[3] = Tdata3;
-  locDatabufptr.i = tcConnectptr.p->firstTupkeybuf;
-  Ti = 4;
-  while (locDatabufptr.i != RNIL) {
-    ptrCheckGuard(locDatabufptr, cdatabufFileSize, databuf);
-    Tdata0 = locDatabufptr.p->data[0];
-    Tdata1 = locDatabufptr.p->data[1];
-    Tdata2 = locDatabufptr.p->data[2];
-    Tdata3 = locDatabufptr.p->data[3];
-    Tdata32[Ti    ] = Tdata0;
-    Tdata32[Ti + 1] = Tdata1;
-    Tdata32[Ti + 2] = Tdata2;
-    Tdata32[Ti + 3] = Tdata3;
-    locDatabufptr.i = locDatabufptr.p->nextDatabuf;
-    Ti += 4;
-  }//while
-  tcConnectptr.p->hashValue = 
-    md5_hash((Uint64*)&Tdata32[0], (UintR)tcConnectptr.p->primKeyLen);
+Uint32 
+Dblqh::calculateHash(Uint32 tableId, const Uint32* src) 
+{
+  jam();
+  Uint64 Tmp[(MAX_KEY_SIZE_IN_WORDS*MAX_XFRM_MULTIPLY) >> 1];
+  Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX];
+  Uint32 keyLen = xfrm_key(tableId, src, (Uint32*)Tmp, sizeof(Tmp) >> 2, 
+			   keyPartLen);
+  ndbrequire(keyLen);
+  
+  return md5_hash(Tmp, keyLen);
 }//Dblqh::calculateHash()
 
 /* *************************************** */
@@ -9384,7 +9358,7 @@ void Dblqh::copyTupkeyConfLab(Signal* signal)
   const TupKeyConf * const tupKeyConf = (TupKeyConf *)signal->getDataPtr();
 
   UintR readLength = tupKeyConf->readLength;
-
+  Uint32 tableId = tcConnectptr.p->tableref;
   scanptr.i = tcConnectptr.p->tcScanRec;
   c_scanRecordPool.getPtr(scanptr);
   ScanRecord* scanP = scanptr.p;
@@ -9411,7 +9385,14 @@ void Dblqh::copyTupkeyConfLab(Signal* signal)
   Uint32 len= tcConnectptr.p->primKeyLen = readPrimaryKeys(scanP, tcConP, tmp);
   
   // Calculate hash (no need to linearies key)
-  tcConnectptr.p->hashValue = md5_hash((Uint64*)tmp, len);
+  if (g_key_descriptor_pool.getPtr(tableId)->hasCharAttr)
+  {
+    tcConnectptr.p->hashValue = calculateHash(tableId, tmp);
+  }
+  else
+  {
+    tcConnectptr.p->hashValue = md5_hash((Uint64*)tmp, len);
+  }
 
   // Move into databuffer to make packLqhkeyreqLab happy
   memcpy(tcConP->tupkeyData, tmp, 4*4);
diff --git a/ndb/src/kernel/blocks/dbtc/Dbtc.hpp b/ndb/src/kernel/blocks/dbtc/Dbtc.hpp
index cd5eedd89ad..bf9f421a0e3 100644
--- a/ndb/src/kernel/blocks/dbtc/Dbtc.hpp
+++ b/ndb/src/kernel/blocks/dbtc/Dbtc.hpp
@@ -962,11 +962,6 @@ public:
     Uint8 hasCharAttr;
     Uint8 noOfDistrKeys;
     
-    struct KeyAttr {
-      Uint32 attributeDescriptor;
-      CHARSET_INFO* charsetInfo;
-    } keyAttr[MAX_ATTRIBUTES_IN_INDEX];
-
     bool checkTable(Uint32 schemaVersion) const {
       return enabled && !dropping && 
 	(table_version_major(schemaVersion) == table_version_major(currentSchemaVersion));
diff --git a/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp b/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
index 12f3a278f0e..0ff6d8477b9 100644
--- a/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
+++ b/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
@@ -66,6 +66,7 @@
 #include <signaldata/DictTabInfo.hpp>
 #include <AttributeDescriptor.hpp>
 #include <SectionReader.hpp>
+#include <KeyDescriptor.hpp>
 
 #include <NdbOut.hpp>
 #include <DebuggerNames.hpp>
@@ -329,42 +330,16 @@ void Dbtc::execTC_SCHVERREQ(Signal* signal)
   BlockReference retPtr = signal->theData[5];
   Uint32 noOfKeyAttr = signal->theData[6];
   ndbrequire(noOfKeyAttr <= MAX_ATTRIBUTES_IN_INDEX);
-  Uint32 hasCharAttr = 0;
-  Uint32 noOfDistrKeys = 0;
-  SegmentedSectionPtr s0Ptr;
-  signal->getSection(s0Ptr, 0);
-  SectionReader r0(s0Ptr, getSectionSegmentPool());
-  Uint32 i = 0;
-  while (i < noOfKeyAttr) {
-    jam();
-    Uint32 attributeDescriptor = ~0;
-    Uint32 csNumber = ~0;
-    if (! r0.getWord(&attributeDescriptor) ||
-        ! r0.getWord(&csNumber)) {
-      jam();
-      break;
-    }
-    CHARSET_INFO* cs = 0;
-    if (csNumber != 0) {
-      cs = all_charsets[csNumber];
-      ndbrequire(cs != 0);
-      hasCharAttr = 1;
-    }
-    
-    noOfDistrKeys += AttributeDescriptor::getDKey(attributeDescriptor);
-    tabptr.p->keyAttr[i].attributeDescriptor = attributeDescriptor;
-    tabptr.p->keyAttr[i].charsetInfo = cs;
-    i++;
-  }
-  ndbrequire(i == noOfKeyAttr);
-  releaseSections(signal);
+
+  const KeyDescriptor* desc = g_key_descriptor_pool.getPtr(tabptr.i);
+  ndbrequire(noOfKeyAttr == desc->noOfKeyAttr);
 
   ndbrequire(tabptr.p->enabled == false);
   tabptr.p->enabled = true;
   tabptr.p->dropping = false;
-  tabptr.p->noOfKeyAttr = noOfKeyAttr;
-  tabptr.p->hasCharAttr = hasCharAttr;
-  tabptr.p->noOfDistrKeys = noOfDistrKeys;
+  tabptr.p->noOfKeyAttr = desc->noOfKeyAttr;
+  tabptr.p->hasCharAttr = desc->hasCharAttr;
+  tabptr.p->noOfDistrKeys = desc->noOfDistrKeys;
   
   signal->theData[0] = tabptr.i;
   signal->theData[1] = retPtr;
@@ -2323,113 +2298,37 @@ Dbtc::handle_special_hash(Uint32 dstHash[4], Uint32* src, Uint32 srcLen,
 			  Uint32 tabPtrI,
 			  bool distr)
 {
-  Uint64 Tmp[MAX_KEY_SIZE_IN_WORDS * 4 * MAX_XFRM_MULTIPLY];
-  const Uint32 dstSize = sizeof(Tmp) / 4;
+  Uint64 Tmp[MAX_KEY_SIZE_IN_WORDS * MAX_XFRM_MULTIPLY];
   const TableRecord* tabPtrP = &tableRecord[tabPtrI];
-  const Uint32 noOfKeyAttr = tabPtrP->noOfKeyAttr;
-  Uint32 noOfDistrKeys = tabPtrP->noOfDistrKeys;
   const bool hasCharAttr = tabPtrP->hasCharAttr;
+  const bool hasDistKeys = tabPtrP->noOfDistrKeys > 0;
   
   Uint32 *dst = (Uint32*)Tmp;
   Uint32 dstPos = 0;
-  Uint32 srcPos = 0;
   Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX];
-  if(hasCharAttr){
-    Uint32 i = 0;
-    while (i < noOfKeyAttr) {
-      const TableRecord::KeyAttr& keyAttr = tabPtrP->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;
-      keyPartLen[i++] = dstWords;
-    }
+  Uint32 * keyPartLenPtr;
+  if(hasCharAttr)
+  {
+    keyPartLenPtr = keyPartLen;
+    dstPos = xfrm_key(tabPtrI, src, dst, sizeof(Tmp) >> 2, keyPartLenPtr);
+    ndbrequire(dstPos);
   } 
   else 
   {
     dst = src;
     dstPos = srcLen;
+    keyPartLenPtr = 0;
   }
   
   md5_hash(dstHash, (Uint64*)dst, dstPos);
   
-  if(distr && noOfDistrKeys)
+  if(distr && hasDistKeys)
   {
     jam();
-    src = dst;
-    dstPos = 0;
-    Uint32 i = 0;
-    if(hasCharAttr)
-    {
-      while (i < noOfKeyAttr && noOfDistrKeys) 
-      {
-	const TableRecord::KeyAttr& keyAttr = tabPtrP->keyAttr[i];
-	Uint32 len = keyPartLen[i];
-	if(AttributeDescriptor::getDKey(keyAttr.attributeDescriptor))
-	{
-	  noOfDistrKeys--;
-	  memmove(dst+dstPos, src, len << 2);
-	  dstPos += len;
-	}
-	src += len;
-	i++;
-      }
-    }
-    else
-    {
-      while (i < noOfKeyAttr && noOfDistrKeys) 
-      {
-	const TableRecord::KeyAttr& keyAttr = tabPtrP->keyAttr[i];
-	Uint32 len = 
-	  AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
-	len = (len + 3) / 4;
-	if(AttributeDescriptor::getDKey(keyAttr.attributeDescriptor))
-	{
-	  noOfDistrKeys--;
-	  memmove(dst+dstPos, src, len << 2);
-	  dstPos += len;
-	}
-	src += len;
-	i++;
-      }
-    }
+    
     Uint32 tmp[4];
-    md5_hash(tmp, (Uint64*)dst, dstPos);
+    Uint32 len = create_distr_key(tabPtrI, dst, keyPartLenPtr);
+    md5_hash(tmp, (Uint64*)dst, len);
     dstHash[1] = tmp[1];
   }
   return true;  // success
@@ -10204,10 +10103,6 @@ void Dbtc::initTable(Signal* signal)
     tabptr.p->noOfKeyAttr = 0;
     tabptr.p->hasCharAttr = 0;
     tabptr.p->noOfDistrKeys = 0;
-    for (unsigned k = 0; k < MAX_ATTRIBUTES_IN_INDEX; k++) {
-      tabptr.p->keyAttr[k].attributeDescriptor = 0;
-      tabptr.p->keyAttr[k].charsetInfo = 0;
-    }
   }//for
 }//Dbtc::initTable()
 
diff --git a/ndb/src/kernel/vm/SimulatedBlock.cpp b/ndb/src/kernel/vm/SimulatedBlock.cpp
index 35c0781a24d..ef9f2c3c716 100644
--- a/ndb/src/kernel/vm/SimulatedBlock.cpp
+++ b/ndb/src/kernel/vm/SimulatedBlock.cpp
@@ -1802,3 +1802,118 @@ SimulatedBlock::init_globals_list(void ** tmp, size_t cnt){
 }
 
 #endif
+
+#include "KeyDescriptor.hpp"
+
+Uint32
+SimulatedBlock::xfrm_key(Uint32 tab, const Uint32* src, 
+			 Uint32 *dst, Uint32 dstSize,
+			 Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX]) const
+{
+  const KeyDescriptor * desc = g_key_descriptor_pool.getPtr(tab);
+  const Uint32 noOfKeyAttr = desc->noOfKeyAttr;
+
+  Uint32 i = 0;
+  Uint32 srcPos = 0;
+  Uint32 dstPos = 0;
+  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;
+    keyPartLen[i++] = dstWords;
+  }
+
+  return dstPos;
+}
+
+Uint32
+SimulatedBlock::create_distr_key(Uint32 tableId,
+				 Uint32 *data, 
+				 const Uint32 
+				 keyPartLen[MAX_ATTRIBUTES_IN_INDEX]) const 
+{
+  const KeyDescriptor* desc = g_key_descriptor_pool.getPtr(tableId);
+  const Uint32 noOfKeyAttr = desc->noOfKeyAttr;
+  Uint32 noOfDistrKeys = desc->noOfDistrKeys;
+  
+  Uint32 *src = data;
+  Uint32 *dst = data;
+  Uint32 i = 0;
+  Uint32 dstPos = 0;
+  
+  if(keyPartLen)
+  {
+    while (i < noOfKeyAttr && noOfDistrKeys) 
+    {
+      Uint32 attr = desc->keyAttr[i].attributeDescriptor;
+      Uint32 len = keyPartLen[i];
+      if(AttributeDescriptor::getDKey(attr))
+      {
+	noOfDistrKeys--;
+	memmove(dst+dstPos, src, len << 2);
+	dstPos += len;
+      }
+      src += len;
+      i++;
+    }
+  }
+  else
+  {
+    while (i < noOfKeyAttr && noOfDistrKeys) 
+    {
+      Uint32 attr = desc->keyAttr[i].attributeDescriptor;
+      Uint32 len = AttributeDescriptor::getSizeInWords(attr);
+      if(AttributeDescriptor::getDKey(attr))
+      {
+	noOfDistrKeys--;
+	memmove(dst+dstPos, src, len << 2);
+	dstPos += len;
+      }
+      src += len;
+      i++;
+    }
+  }
+  return dstPos;
+}
diff --git a/ndb/src/kernel/vm/SimulatedBlock.hpp b/ndb/src/kernel/vm/SimulatedBlock.hpp
index 787d14ca5cb..50c85a0b274 100644
--- a/ndb/src/kernel/vm/SimulatedBlock.hpp
+++ b/ndb/src/kernel/vm/SimulatedBlock.hpp
@@ -20,11 +20,13 @@
 #include <NdbTick.h>
 #include <kernel_types.h>
 #include <ndb_version.h>
+#include <ndb_limits.h>
 
 #include "VMSignal.hpp"
 #include <RefConvert.hpp>
 #include <BlockNumbers.h>
 #include <GlobalSignalNumbers.h>
+
 #include "pc.hpp"
 #include <NodeInfo.hpp>
 #include <NodeState.hpp>
@@ -385,6 +387,24 @@ protected:
    */
   const NodeInfo & getNodeInfo(NodeId nodeId) const;
   NodeInfo & setNodeInfo(NodeId);
+
+  /**********************
+   * Xfrm stuff
+   */
+  
+  /**
+   * @return length
+   */
+  Uint32 xfrm_key(Uint32 tab, const Uint32* src, 
+		  Uint32 *dst, Uint32 dstLen,
+		  Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX]) const;
+  
+  /**
+   *
+   */
+  Uint32 create_distr_key(Uint32 tableId,
+			  Uint32 *data, 
+			  const Uint32 keyPaLen[MAX_ATTRIBUTES_IN_INDEX])const;
   
 private:
   NewVARIABLE* NewVarRef;      /* New Base Address Table for block  */