bk commit into 5.0 tree (jonas:1.1896) BUG#12220

[jonas]

Below is the list of changes that have just been committed into a local
5.0 repository of jonas. When jonas does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet
  1.1896 05/08/22 10:59:25 jonas@eel.(none) +11 -0
  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/src/kernel/vm/SimulatedBlock.hpp
    1.12 05/08/22 10:59:21 jonas@eel.(none) +20 -0
    Move xfrm handling into SimulatedBlock

  ndb/src/kernel/vm/SimulatedBlock.cpp
    1.18 05/08/22 10:59:21 jonas@eel.(none) +115 -0
    Move xfrm handling into SimulatedBlock

  ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
    1.81 05/08/22 10:59:21 jonas@eel.(none) +20 -125
    Move xfrm handling into SimulatedBlock

  ndb/src/kernel/blocks/dbtc/Dbtc.hpp
    1.30 05/08/22 10:59:21 jonas@eel.(none) +0 -5
    Move xfrm handling into SimulatedBlock

  ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
    1.71 05/08/22 10:59:21 jonas@eel.(none) +20 -39
    Move xfrm handling into SimulatedBlock

  ndb/src/kernel/blocks/dblqh/Dblqh.hpp
    1.33 05/08/22 10:59:21 jonas@eel.(none) +1 -1
    Move xfrm handling into SimulatedBlock

  ndb/src/kernel/blocks/dbdict/Dbdict.cpp
    1.51 05/08/22 10:59:21 jonas@eel.(none) +45 -23
    Move xfrm handling into SimulatedBlock

  ndb/src/kernel/blocks/dbacc/DbaccMain.cpp
    1.54 05/08/22 10:59:21 jonas@eel.(none) +14 -115
    Move xfrm handling into SimulatedBlock

  ndb/src/kernel/blocks/dbacc/DbaccInit.cpp
    1.13 05/08/22 10:59:20 jonas@eel.(none) +0 -1
    Move xfrm handling into SimulatedBlock

  ndb/src/kernel/blocks/dbacc/Dbacc.hpp
    1.20 05/08/22 10:59:20 jonas@eel.(none) +0 -8
    Move xfrm handling into SimulatedBlock

  ndb/include/kernel/AttributeDescriptor.hpp
    1.6 05/08/22 10:59:20 jonas@eel.(none) +2 -1
    Make SimulatedBlock use AttributeDescriptor (to xfrm)

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	jonas
# Host:	eel.(none)
# Root:	/home/jonas/src/mysql-5.0

--- 1.5/ndb/include/kernel/AttributeDescriptor.hpp	2004-12-26 22:34:25 +01:00
+++ 1.6/ndb/include/kernel/AttributeDescriptor.hpp	2005-08-22 10:59:20 +02:00
@@ -23,7 +23,8 @@
   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);

--- 1.19/ndb/src/kernel/blocks/dbacc/Dbacc.hpp	2005-04-14 13:32:36 +02:00
+++ 1.20/ndb/src/kernel/blocks/dbacc/Dbacc.hpp	2005-08-22 10:59:20 +02:00
@@ -851,13 +851,6 @@
   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 @@
   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);

--- 1.12/ndb/src/kernel/blocks/dbacc/DbaccInit.cpp	2005-04-17 18:51:06 +02:00
+++ 1.13/ndb/src/kernel/blocks/dbacc/DbaccInit.cpp	2005-08-22 10:59:20 +02:00
@@ -179,7 +179,6 @@
   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);

--- 1.53/ndb/src/kernel/blocks/dbacc/DbaccMain.cpp	2005-07-25 13:23:29 +02:00
+++ 1.54/ndb/src/kernel/blocks/dbacc/DbaccMain.cpp	2005-08-22 10:59:21 +02:00
@@ -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 @@
       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 @@
   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 @@
   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::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 @@
   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)

--- 1.50/ndb/src/kernel/blocks/dbdict/Dbdict.cpp	2005-08-18 14:09:06 +02:00
+++ 1.51/ndb/src/kernel/blocks/dbdict/Dbdict.cpp	2005-08-22 10:59:21 +02:00
@@ -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 @@
   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 @@
     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 @@
     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 @@
   new (&attr) MetaData::Attribute(*attrPtr.p);
   return 0;
 }
+
+CArray<KeyDescriptor> g_key_descriptor_pool;

--- 1.32/ndb/src/kernel/blocks/dblqh/Dblqh.hpp	2005-04-14 13:46:30 +02:00
+++ 1.33/ndb/src/kernel/blocks/dblqh/Dblqh.hpp	2005-08-22 10:59:21 +02:00
@@ -2233,7 +2233,7 @@
   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);

--- 1.70/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp	2005-08-18 14:09:07 +02:00
+++ 1.71/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp	2005-08-22 10:59:21 +02:00
@@ -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 @@
 /*   FRAGMENT TO A NEW REPLICA OF THE FRAGMENT. IT DOES ALSO SHUT DOWN ALL   */
 /*   CONNECTIONS TO THE FAILED NODE.                                         */
 /*---------------------------------------------------------------------------*/
-void Dblqh::calculateHash(Signal* signal) 
+Uint32 
+Dblqh::calculateHash(Uint32 tableId, const Uint32* src) 
 {
-  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);
+  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 @@
   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 @@
   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);

--- 1.29/ndb/src/kernel/blocks/dbtc/Dbtc.hpp	2005-08-18 14:09:07 +02:00
+++ 1.30/ndb/src/kernel/blocks/dbtc/Dbtc.hpp	2005-08-22 10:59:21 +02:00
@@ -962,11 +962,6 @@
     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));

--- 1.80/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp	2005-08-18 14:09:07 +02:00
+++ 1.81/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp	2005-08-22 10:59:21 +02:00
@@ -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 @@
   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 @@
 			  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 @@
     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()
 

--- 1.17/ndb/src/kernel/vm/SimulatedBlock.cpp	2005-01-11 23:25:41 +01:00
+++ 1.18/ndb/src/kernel/vm/SimulatedBlock.cpp	2005-08-22 10:59:21 +02:00
@@ -1802,3 +1802,118 @@
 }
 
 #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;
+}

--- 1.11/ndb/src/kernel/vm/SimulatedBlock.hpp	2005-01-19 09:15:31 +01:00
+++ 1.12/ndb/src/kernel/vm/SimulatedBlock.hpp	2005-08-22 10:59:21 +02:00
@@ -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 @@
    */
   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  */