bk commit into 5.1 tree (pekka:1.2330)

[pekka]

Below is the list of changes that have just been committed into a local
5.1 repository of pekka. When pekka 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@stripped, 2006-10-25 17:32:32+02:00, pekka@stripped +1 -0
  Merge orca.ndb.mysql.com:/export/home/space/pekka/ndb/version/my50-tux
  into  orca.ndb.mysql.com:/export/home/space/pekka/ndb/version/my51-tux
  MERGE: 1.1810.2203.1

  storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp@stripped, 2006-10-25 17:32:19+02:00,
pekka@stripped +0 -0
    Auto merged
    MERGE: 1.27.3.2

  storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp@stripped, 2006-10-25 17:32:19+02:00,
pekka@stripped +0 -0
    Merge rename: ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp ->
storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp

# 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:	pekka
# Host:	orca.ndb.mysql.com
# Root:	/export/home/space/pekka/ndb/version/my51-tux/RESYNC

--- 1.27.3.1/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp	2006-10-25 17:32:52 +02:00
+++ 1.33/storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp	2006-10-25 17:32:52 +02:00
@@ -35,7 +35,7 @@
     fragPtr.i = RNIL;
     for (unsigned i = 0; i < indexPtr.p->m_numFrags; i++) {
       jam();
-      if (indexPtr.p->m_fragId[i] == req->fragmentNo << 1) {
+      if (indexPtr.p->m_fragId[i] == req->fragmentNo) {
         jam();
         c_fragPool.getPtr(fragPtr, indexPtr.p->m_fragPtrI[i]);
         break;
@@ -161,8 +161,19 @@
     Uint32 dstWords = 0;
     if (! ah->isNULL()) {
       jam();
+      const uchar* srcPtr = (const uchar*)&data[offset + 2];
       const DescAttr& descAttr = descEnt.m_descAttr[attrId];
-      Uint32 srcBytes = AttributeDescriptor::getSizeInBytes(descAttr.m_attrDesc);
+      Uint32 typeId = descAttr.m_typeId;
+      Uint32 maxBytes = AttributeDescriptor::getSizeInBytes(descAttr.m_attrDesc);
+      Uint32 lb, len;
+      bool ok = NdbSqlUtil::get_var_length(typeId, srcPtr, maxBytes, lb, len);
+      if (! ok) {
+        jam();
+        scan.m_state = ScanOp::Invalid;
+        sig->errorCode = TuxBoundInfo::InvalidCharFormat;
+        return;
+      }
+      Uint32 srcBytes = lb + len;
       Uint32 srcWords = (srcBytes + 3) / 4;
       if (srcWords != dataSize) {
         jam();
@@ -171,27 +182,17 @@
         return;
       }
       uchar* dstPtr = (uchar*)&xfrmData[dstPos + 2];
-      const uchar* srcPtr = (const uchar*)&data[offset + 2];
       if (descAttr.m_charset == 0) {
         memcpy(dstPtr, srcPtr, srcWords << 2);
         dstWords = srcWords;
       } else {
         jam();
-        Uint32 typeId = descAttr.m_typeId;
-        Uint32 lb, len;
-        bool ok = NdbSqlUtil::get_var_length(typeId, srcPtr, srcBytes, lb, len);
-        if (! ok) {
-          jam();
-          scan.m_state = ScanOp::Invalid;
-          sig->errorCode = TuxBoundInfo::InvalidCharFormat;
-          return;
-        }
         CHARSET_INFO* cs = all_charsets[descAttr.m_charset];
         Uint32 xmul = cs->strxfrm_multiply;
         if (xmul == 0)
           xmul = 1;
         // see comment in DbtcMain.cpp
-        Uint32 dstLen = xmul * (srcBytes - lb);
+        Uint32 dstLen = xmul * (maxBytes - lb);
         if (dstLen > ((dstSize - dstPos) << 2)) {
           jam();
           scan.m_state = ScanOp::Invalid;
@@ -320,7 +321,7 @@
       conf->scanPtr = scan.m_userPtr;
       unsigned signalLength = 1;
       sendSignal(scanPtr.p->m_userRef, GSN_NEXT_SCANCONF,
-          signal, signalLength, JBB);
+		 signal, signalLength, JBB);
       return;
     }
     break;
@@ -343,7 +344,8 @@
       lockReq->returnCode = RNIL;
       lockReq->requestInfo = AccLockReq::AbortWithConf;
       lockReq->accOpPtr = scan.m_accLockOp;
-      EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
+      EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, 
+		     AccLockReq::UndoSignalLength);
       jamEntry();
       ndbrequire(lockReq->returnCode == AccLockReq::Success);
       scan.m_state = ScanOp::Aborting;
@@ -354,9 +356,10 @@
       ndbrequire(scan.m_accLockOp != RNIL);
       AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
       lockReq->returnCode = RNIL;
-      lockReq->requestInfo = AccLockReq::Unlock;
+      lockReq->requestInfo = AccLockReq::Abort;
       lockReq->accOpPtr = scan.m_accLockOp;
-      EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
+      EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, 
+		     AccLockReq::UndoSignalLength);
       jamEntry();
       ndbrequire(lockReq->returnCode == AccLockReq::Success);
       scan.m_accLockOp = RNIL;
@@ -410,7 +413,7 @@
     NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
     conf->scanPtr = scan.m_userPtr;
     conf->accOperationPtr = RNIL;       // no tuple returned
-    conf->fragId = frag.m_fragId | ent.m_fragBit;
+    conf->fragId = frag.m_fragId;
     unsigned signalLength = 3;
     // if TC has ordered scan close, it will be detected here
     sendSignal(scan.m_userRef, GSN_NEXT_SCANCONF,
@@ -449,8 +452,8 @@
       lockReq->userPtr = scanPtr.i;
       lockReq->userRef = reference();
       lockReq->tableId = scan.m_tableId;
-      lockReq->fragId = frag.m_fragId | ent.m_fragBit;
-      lockReq->fragPtrI = frag.m_accTableFragPtrI[ent.m_fragBit];
+      lockReq->fragId = frag.m_fragId;
+      lockReq->fragPtrI = frag.m_accTableFragPtrI;
       const Uint32* const buf32 = static_cast<Uint32*>(pkData);
       const Uint64* const buf64 = reinterpret_cast<const Uint64*>(buf32);
       lockReq->hashValue = md5_hash(buf64, pkSize);
@@ -541,7 +544,7 @@
       accLockOp = (Uint32)-1;
     }
     conf->accOperationPtr = accLockOp;
-    conf->fragId = frag.m_fragId | ent.m_fragBit;
+    conf->fragId = frag.m_fragId;
     conf->localKey[0] = getTupAddr(frag, ent);
     conf->localKey[1] = 0;
     conf->localKeyLength = 1;
@@ -607,7 +610,7 @@
     jam();
     AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
     lockReq->returnCode = RNIL;
-    lockReq->requestInfo = AccLockReq::Unlock;
+    lockReq->requestInfo = AccLockReq::Abort;
     lockReq->accOpPtr = scan.m_accLockOp;
     EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
     jamEntry();
@@ -984,14 +987,13 @@
 {
   const ScanOp& scan = *scanPtr.p;
   const Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
-  Uint32 fragBit = ent.m_fragBit;
-  Uint32 tableFragPtrI = frag.m_tupTableFragPtrI[fragBit];
-  Uint32 fragId = frag.m_fragId | fragBit;
+  Uint32 tableFragPtrI = frag.m_tupTableFragPtrI;
+  Uint32 fragId = frag.m_fragId;
   Uint32 tupAddr = getTupAddr(frag, ent);
   Uint32 tupVersion = ent.m_tupVersion;
   // check for same tuple twice in row
-  if (scan.m_scanEnt.m_tupLoc == ent.m_tupLoc &&
-      scan.m_scanEnt.m_fragBit == fragBit) {
+  if (scan.m_scanEnt.m_tupLoc == ent.m_tupLoc)
+  {
     jam();
     return false;
   }