Merge pnousiainen@bk-internal.mysql.com:/home/bk/mysql-4.1-ndb

into mysql.com:/space/pekka/ndb/version/my41

ndb/include/kernel/signaldata/TupAccess.hpp

@@ -129,6 +129,8 @@ private:
 /*
  * Operate on entire tuple.  Used by TUX where the table has a single
  * Uint32 array attribute representing an index tree node.
+ *
+ * XXX this signal will be replaced by method in TUP
  */
 class TupStoreTh {
   friend class Dbtup;
@@ -153,8 +155,8 @@ private:
   Uint32 tableId;
   Uint32 fragId;
   Uint32 fragPtrI;
-  Uint32 tupAddr;
-  Uint32 tupVersion;
+  Uint32 tupAddr;       // no longer used
+  Uint32 tupVersion;    // no longer used
   Uint32 pageId;
   Uint32 pageOffset;
   Uint32 bufferId;

ndb/include/kernel/signaldata/TupFrag.hpp

@@ -69,7 +69,7 @@ class TuxFragReq {
   friend class Dblqh;
   friend class Dbtux;
 public:
-  STATIC_CONST( SignalLength = 9 );
+  STATIC_CONST( SignalLength = 14 );
 private:
   Uint32 userPtr;
   Uint32 userRef;
@@ -80,6 +80,9 @@ private:
   Uint32 fragOff;
   Uint32 tableType;
   Uint32 primaryTableId;
+  Uint32 tupIndexFragPtrI;
+  Uint32 tupTableFragPtrI[2];
+  Uint32 accTableFragPtrI[2];
 };
 
 class TuxFragConf {

ndb/src/kernel/blocks/dbacc/DbaccMain.cpp

@@ -2432,6 +2432,7 @@ void Dbacc::execACC_LOCKREQ(Signal* signal)
     }
     fragrecptr.i = req->fragPtrI;
     ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
+    ndbrequire(req->fragId == fragrecptr.p->myfid);
     // caller must be explicit here
     ndbrequire(req->accOpPtr == RNIL);
     // seize operation to hold the lock

ndb/src/kernel/blocks/dblqh/DblqhMain.cpp

@@ -1225,6 +1225,18 @@ Dblqh::sendAddFragReq(Signal* signal)
       tuxreq->fragOff = addfragptr.p->lh3DistrBits;
       tuxreq->tableType = addfragptr.p->tableType;
       tuxreq->primaryTableId = addfragptr.p->primaryTableId;
+      // pointer to index fragment in TUP
+      tuxreq->tupIndexFragPtrI =
+        addfragptr.p->addfragStatus == AddFragRecord::WAIT_TWO_TUX ?
+        fragptr.p->tupFragptr[0] : fragptr.p->tupFragptr[1];
+      // pointers to table fragments in TUP and ACC
+      FragrecordPtr tFragPtr;
+      tFragPtr.i = fragptr.p->tableFragptr;
+      ptrCheckGuard(tFragPtr, cfragrecFileSize, fragrecord);
+      tuxreq->tupTableFragPtrI[0] = tFragPtr.p->tupFragptr[0];
+      tuxreq->tupTableFragPtrI[1] = tFragPtr.p->tupFragptr[1];
+      tuxreq->accTableFragPtrI[0] = tFragPtr.p->accFragptr[0];
+      tuxreq->accTableFragPtrI[1] = tFragPtr.p->accFragptr[1];
       sendSignal(fragptr.p->tuxBlockref, GSN_TUXFRAGREQ,
           signal, TuxFragReq::SignalLength, JBB);
       return;

ndb/src/kernel/blocks/dbtux/DbtuxDebug.cpp

@@ -97,7 +97,7 @@ Dbtux::printTree(Signal* signal, Frag& frag, NdbOut& out)
   PrintPar par;
   strcpy(par.m_path, ".");
   par.m_side = 2;
-  par.m_parent = NullTupAddr;
+  par.m_parent = NullTupLoc;
   printNode(signal, frag, out, tree.m_root, par);
   out.m_out->flush();
   if (! par.m_ok) {
@@ -116,15 +116,15 @@ Dbtux::printTree(Signal* signal, Frag& frag, NdbOut& out)
 }
 
 void
-Dbtux::printNode(Signal* signal, Frag& frag, NdbOut& out, TupAddr addr, PrintPar& par)
+Dbtux::printNode(Signal* signal, Frag& frag, NdbOut& out, TupLoc loc, PrintPar& par)
 {
-  if (addr == NullTupAddr) {
+  if (loc == NullTupLoc) {
     par.m_depth = 0;
     return;
   }
   TreeHead& tree = frag.m_tree;
   NodeHandlePtr nodePtr;
-  selectNode(signal, frag, nodePtr, addr, AccFull);
+  selectNode(signal, frag, nodePtr, loc, AccFull);
   out << par.m_path << " " << *nodePtr.p << endl;
   // check children
   PrintPar cpar[2];
@@ -133,7 +133,7 @@ Dbtux::printNode(Signal* signal, Frag& frag, NdbOut& out, TupAddr addr, PrintPar
     sprintf(cpar[i].m_path, "%s%c", par.m_path, "LR"[i]);
     cpar[i].m_side = i;
     cpar[i].m_depth = 0;
-    cpar[i].m_parent = addr;
+    cpar[i].m_parent = loc;
     printNode(signal, frag, out, nodePtr.p->getLink(i), cpar[i]);
     if (! cpar[i].m_ok) {
       par.m_ok = false;
@@ -143,7 +143,7 @@ Dbtux::printNode(Signal* signal, Frag& frag, NdbOut& out, TupAddr addr, PrintPar
   if (nodePtr.p->getLink(2) != par.m_parent) {
     par.m_ok = false;
     out << par.m_path << " *** ";
-    out << "parent addr " << hex << nodePtr.p->getLink(2);
+    out << "parent loc " << hex << nodePtr.p->getLink(2);
     out << " should be " << hex << par.m_parent << endl;
   }
   if (nodePtr.p->getSide() != par.m_side) {
@@ -181,8 +181,8 @@ Dbtux::printNode(Signal* signal, Frag& frag, NdbOut& out, TupAddr addr, PrintPar
   }
   // check missed half-leaf/leaf merge
   for (unsigned i = 0; i <= 1; i++) {
-    if (nodePtr.p->getLink(i) != NullTupAddr &&
-        nodePtr.p->getLink(1 - i) == NullTupAddr &&
+    if (nodePtr.p->getLink(i) != NullTupLoc &&
+        nodePtr.p->getLink(1 - i) == NullTupLoc &&
         nodePtr.p->getOccup() + cpar[i].m_occup <= tree.m_maxOccup) {
       par.m_ok = false;
       out << par.m_path << " *** ";
@@ -194,6 +194,18 @@ Dbtux::printNode(Signal* signal, Frag& frag, NdbOut& out, TupAddr addr, PrintPar
   par.m_occup = nodePtr.p->getOccup();
 }
 
+NdbOut&
+operator<<(NdbOut& out, const Dbtux::TupLoc& loc)
+{
+  if (loc == Dbtux::NullTupLoc) {
+    out << "null";
+  } else {
+    out << hex << loc.m_pageId;
+    out << "." << dec << loc.m_pageOffset;
+  }
+  return out;
+}
+
 NdbOut&
 operator<<(NdbOut& out, const Dbtux::TreeEnt& ent)
 {
@@ -206,10 +218,13 @@ operator<<(NdbOut& out, const Dbtux::TreeEnt& ent)
 NdbOut&
 operator<<(NdbOut& out, const Dbtux::TreeNode& node)
 {
+  Dbtux::TupLoc link0(node.m_linkPI[0], node.m_linkPO[0]);
+  Dbtux::TupLoc link1(node.m_linkPI[1], node.m_linkPO[1]);
+  Dbtux::TupLoc link2(node.m_linkPI[2], node.m_linkPO[2]);
   out << "[TreeNode " << hex << &node;
-  out << " [left " << hex << node.m_link[0] << "]";
-  out << " [right " << hex << node.m_link[1] << "]";
-  out << " [up " << hex << node.m_link[2] << "]";
+  out << " [left " << link0 << "]";
+  out << " [right " << link1 << "]";
+  out << " [up " << link2 << "]";
   out << " [side " << dec << node.m_side << "]";
   out << " [occup " << dec << node.m_occup << "]";
   out << " [balance " << dec << (int)node.m_balance << "]";
@@ -238,7 +253,7 @@ NdbOut&
 operator<<(NdbOut& out, const Dbtux::TreePos& pos)
 {
   out << "[TreePos " << hex << &pos;
-  out << " [addr " << hex << pos.m_addr << "]";
+  out << " [loc " << pos.m_loc << "]";
   out << " [pos " << dec << pos.m_pos << "]";
   out << " [match " << dec << pos.m_match << "]";
   out << " [dir " << dec << pos.m_dir << "]";
@@ -338,7 +353,7 @@ operator<<(NdbOut& out, const Dbtux::NodeHandle& node)
   const Dbtux::Frag& frag = node.m_frag;
   const Dbtux::TreeHead& tree = frag.m_tree;
   out << "[NodeHandle " << hex << &node;
-  out << " [addr " << hex << node.m_addr << "]";
+  out << " [loc " << node.m_loc << "]";
   out << " [acc " << dec << node.m_acc << "]";
   out << " [flags " << hex << node.m_flags << "]";
   out << " [node " << *node.m_node << "]";

ndb/src/kernel/blocks/dbtux/DbtuxMaint.cpp

@@ -199,7 +199,7 @@ Dbtux::tupReadAttrs(Signal* signal, const Frag& frag, ReadPar& readPar)
   req->requestInfo = 0;
   req->tableId = frag.m_tableId;
   req->fragId = frag.m_fragId | (ent.m_fragBit << frag.m_fragOff);
-  req->fragPtrI = RNIL;
+  req->fragPtrI = frag.m_tupTableFragPtrI[ent.m_fragBit];
   req->tupAddr = ent.m_tupAddr;
   req->tupVersion = ent.m_tupVersion;
   req->pageId = RNIL;
@@ -246,7 +246,7 @@ Dbtux::tupReadKeys(Signal* signal, const Frag& frag, ReadPar& readPar)
   req->requestInfo = TupReadAttrs::ReadKeys;
   req->tableId = frag.m_tableId;
   req->fragId = frag.m_fragId | (ent.m_fragBit << frag.m_fragOff);
-  req->fragPtrI = RNIL;
+  req->fragPtrI = frag.m_tupTableFragPtrI[ent.m_fragBit];
   req->tupAddr = ent.m_tupAddr;
   req->tupVersion = RNIL; // not used
   req->pageId = RNIL;
@@ -284,9 +284,9 @@ Dbtux::tupStoreTh(Signal* signal, const Frag& frag, NodeHandlePtr nodePtr, Store
   req->errorCode = RNIL;
   req->tableId = frag.m_indexId;
   req->fragId = frag.m_fragId;
-  req->fragPtrI = RNIL;
-  req->tupAddr = nodePtr.p->m_addr;
-  req->tupVersion = 0;
+  req->fragPtrI = frag.m_tupIndexFragPtrI;
+  req->tupAddr = RNIL;  // no longer used
+  req->tupVersion = 0;  // no longer used
   req->pageId = nodePtr.p->m_loc.m_pageId;
   req->pageOffset = nodePtr.p->m_loc.m_pageOffset;
   req->bufferId = 0;
@@ -346,21 +346,18 @@ Dbtux::tupStoreTh(Signal* signal, const Frag& frag, NodeHandlePtr nodePtr, Store
       const Uint32* src = (const Uint32*)buffer + storePar.m_offset;
       memcpy(dst, src, storePar.m_size << 2);
     }
-    // fallthru
+    break;
   case TupStoreTh::OpInsert:
     jam();
-    // fallthru
-  case TupStoreTh::OpUpdate:
-    jam();
-    nodePtr.p->m_addr = req->tupAddr;
     nodePtr.p->m_loc.m_pageId = req->pageId;
     nodePtr.p->m_loc.m_pageOffset = req->pageOffset;
     break;
+  case TupStoreTh::OpUpdate:
+    jam();
+    break;
   case TupStoreTh::OpDelete:
     jam();
-    nodePtr.p->m_addr = NullTupAddr;
-    nodePtr.p->m_loc.m_pageId = RNIL;
-    nodePtr.p->m_loc.m_pageOffset = 0;
+    nodePtr.p->m_loc = NullTupLoc;
     break;
   default:
     ndbrequire(false);

ndb/src/kernel/blocks/dbtux/DbtuxMeta.cpp

@@ -85,6 +85,11 @@ Dbtux::execTUXFRAGREQ(Signal* signal)
     fragPtr.p->m_fragOff = req->fragOff;
     fragPtr.p->m_fragId = req->fragId;
     fragPtr.p->m_numAttrs = req->noOfAttr;
+    fragPtr.p->m_tupIndexFragPtrI = req->tupIndexFragPtrI;
+    fragPtr.p->m_tupTableFragPtrI[0] = req->tupTableFragPtrI[0];
+    fragPtr.p->m_tupTableFragPtrI[1] = req->tupTableFragPtrI[1];
+    fragPtr.p->m_accTableFragPtrI[0] = req->accTableFragPtrI[0];
+    fragPtr.p->m_accTableFragPtrI[1] = req->accTableFragPtrI[1];
     // add the fragment to the index
     indexPtr.p->m_fragId[indexPtr.p->m_numFrags] = req->fragId;
     indexPtr.p->m_fragPtrI[indexPtr.p->m_numFrags] = fragPtr.i;
@@ -197,6 +202,7 @@ Dbtux::execTUX_ADD_ATTRREQ(Signal* signal)
       jam();
       // initialize tree header
       TreeHead& tree = fragPtr.p->m_tree;
+      new (&tree) TreeHead();
       // make these configurable later
       tree.m_nodeSize = MAX_TTREE_NODE_SIZE;
       tree.m_prefSize = MAX_TTREE_PREF_SIZE;
@@ -222,8 +228,8 @@ Dbtux::execTUX_ADD_ATTRREQ(Signal* signal)
         break;
       }
       tree.m_minOccup = tree.m_maxOccup - maxSlack;
-      // root node does not exist
-      tree.m_root = NullTupAddr;
+      // root node does not exist (also set by ctor)
+      tree.m_root = NullTupLoc;
       // fragment is defined
       c_fragOpPool.release(fragOpPtr);
     }

ndb/src/kernel/blocks/dbtux/DbtuxNode.cpp

@@ -80,14 +80,14 @@ Dbtux::preallocNode(Signal* signal, Frag& frag, Uint32& errorCode)
  * Find node in the cache.  XXX too slow, use direct links instead
  */
 void
-Dbtux::findNode(Signal* signal, Frag& frag, NodeHandlePtr& nodePtr, TupAddr addr)
+Dbtux::findNode(Signal* signal, Frag& frag, NodeHandlePtr& nodePtr, TupLoc loc)
 {
   NodeHandlePtr tmpPtr;
   tmpPtr.i = frag.m_nodeList;
   while (tmpPtr.i != RNIL) {
     jam();
     c_nodeHandlePool.getPtr(tmpPtr);
-    if (tmpPtr.p->m_addr == addr) {
+    if (tmpPtr.p->m_loc == loc) {
       jam();
       nodePtr = tmpPtr;
       return;
@@ -102,18 +102,18 @@ Dbtux::findNode(Signal* signal, Frag& frag, NodeHandlePtr& nodePtr, TupAddr addr
  * Get handle for existing node.
  */
 void
-Dbtux::selectNode(Signal* signal, Frag& frag, NodeHandlePtr& nodePtr, TupAddr addr, AccSize acc)
+Dbtux::selectNode(Signal* signal, Frag& frag, NodeHandlePtr& nodePtr, TupLoc loc, AccSize acc)
 {
-  ndbrequire(addr != NullTupAddr && acc > AccNone);
+  ndbrequire(loc != NullTupLoc && acc > AccNone);
   NodeHandlePtr tmpPtr;
   // search in cache
-  findNode(signal, frag, tmpPtr, addr);
+  findNode(signal, frag, tmpPtr, loc);
   if (tmpPtr.i == RNIL) {
     jam();
     // add new node
     seizeNode(signal, frag, tmpPtr);
     ndbrequire(tmpPtr.i != RNIL);
-    tmpPtr.p->m_addr = addr;
+    tmpPtr.p->m_loc = loc;
   }
   if (tmpPtr.p->m_acc < acc) {
     jam();
@@ -276,7 +276,7 @@ Dbtux::NodeHandle::pushUp(Signal* signal, unsigned pos, const TreeEnt& ent)
     jam();
     m_tux.c_scanOpPool.getPtr(scanPtr);
     TreePos& scanPos = scanPtr.p->m_scanPos;
-    ndbrequire(scanPos.m_addr == m_addr && scanPos.m_pos < occup);
+    ndbrequire(scanPos.m_loc == m_loc && scanPos.m_pos < occup);
     if (scanPos.m_pos >= pos) {
       jam();
 #ifdef VM_TRACE
@@ -329,7 +329,7 @@ Dbtux::NodeHandle::popDown(Signal* signal, unsigned pos, TreeEnt& ent)
     jam();
     m_tux.c_scanOpPool.getPtr(scanPtr);
     TreePos& scanPos = scanPtr.p->m_scanPos;
-    ndbrequire(scanPos.m_addr == m_addr && scanPos.m_pos < occup);
+    ndbrequire(scanPos.m_loc == m_loc && scanPos.m_pos < occup);
     const Uint32 nextPtrI = scanPtr.p->m_nodeScan;
     if (scanPos.m_pos == pos) {
       jam();
@@ -349,7 +349,7 @@ Dbtux::NodeHandle::popDown(Signal* signal, unsigned pos, TreeEnt& ent)
     jam();
     m_tux.c_scanOpPool.getPtr(scanPtr);
     TreePos& scanPos = scanPtr.p->m_scanPos;
-    ndbrequire(scanPos.m_addr == m_addr && scanPos.m_pos < occup);
+    ndbrequire(scanPos.m_loc == m_loc && scanPos.m_pos < occup);
     ndbrequire(scanPos.m_pos != pos);
     if (scanPos.m_pos > pos) {
       jam();
@@ -403,7 +403,7 @@ Dbtux::NodeHandle::pushDown(Signal* signal, unsigned pos, TreeEnt& ent)
     jam();
     m_tux.c_scanOpPool.getPtr(scanPtr);
     TreePos& scanPos = scanPtr.p->m_scanPos;
-    ndbrequire(scanPos.m_addr == m_addr && scanPos.m_pos < occup);
+    ndbrequire(scanPos.m_loc == m_loc && scanPos.m_pos < occup);
     const Uint32 nextPtrI = scanPtr.p->m_nodeScan;
     if (scanPos.m_pos == 0) {
       jam();
@@ -424,7 +424,7 @@ Dbtux::NodeHandle::pushDown(Signal* signal, unsigned pos, TreeEnt& ent)
     jam();
     m_tux.c_scanOpPool.getPtr(scanPtr);
     TreePos& scanPos = scanPtr.p->m_scanPos;
-    ndbrequire(scanPos.m_addr == m_addr && scanPos.m_pos < occup);
+    ndbrequire(scanPos.m_loc == m_loc && scanPos.m_pos < occup);
     ndbrequire(scanPos.m_pos != 0);
     if (scanPos.m_pos <= pos) {
       jam();
@@ -480,7 +480,7 @@ Dbtux::NodeHandle::popUp(Signal* signal, unsigned pos, TreeEnt& ent)
     jam();
     m_tux.c_scanOpPool.getPtr(scanPtr);
     TreePos& scanPos = scanPtr.p->m_scanPos;
-    ndbrequire(scanPos.m_addr == m_addr && scanPos.m_pos < occup);
+    ndbrequire(scanPos.m_loc == m_loc && scanPos.m_pos < occup);
     const Uint32 nextPtrI = scanPtr.p->m_nodeScan;
     if (scanPos.m_pos == pos) {
       jam();
@@ -501,7 +501,7 @@ Dbtux::NodeHandle::popUp(Signal* signal, unsigned pos, TreeEnt& ent)
     jam();
     m_tux.c_scanOpPool.getPtr(scanPtr);
     TreePos& scanPos = scanPtr.p->m_scanPos;
-    ndbrequire(scanPos.m_addr == m_addr && scanPos.m_pos < occup);
+    ndbrequire(scanPos.m_loc == m_loc && scanPos.m_pos < occup);
     ndbrequire(scanPos.m_pos != pos);
     if (scanPos.m_pos < pos) {
       jam();

ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp

@@ -47,7 +47,7 @@ Dbtux::execACC_SCANREQ(Signal* signal)
     ndbrequire(frag.m_fragId < (1 << frag.m_fragOff));
     TreeHead& tree = frag.m_tree;
     // check for empty fragment
-    if (tree.m_root == NullTupAddr) {
+    if (tree.m_root == NullTupLoc) {
       jam();
       AccScanConf* const conf = (AccScanConf*)signal->getDataPtrSend();
       conf->scanPtr = req->senderData;
@@ -275,13 +275,13 @@ Dbtux::execNEXT_SCANREQ(Signal* signal)
   case NextScanReq::ZSCAN_CLOSE:
     jam();
     // unlink from tree node first to avoid state changes
-    if (scan.m_scanPos.m_addr != NullTupAddr) {
+    if (scan.m_scanPos.m_loc != NullTupLoc) {
       jam();
-      const TupAddr addr = scan.m_scanPos.m_addr;
+      const TupLoc loc = scan.m_scanPos.m_loc;
       NodeHandlePtr nodePtr;
-      selectNode(signal, frag, nodePtr, addr, AccHead);
+      selectNode(signal, frag, nodePtr, loc, AccHead);
       nodePtr.p->unlinkScan(scanPtr);
-      scan.m_scanPos.m_addr = NullTupAddr;
+      scan.m_scanPos.m_loc = NullTupLoc;
     }
     if (scan.m_lockwait) {
       jam();
@@ -407,8 +407,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
       lockReq->userRef = reference();
       lockReq->tableId = scan.m_tableId;
       lockReq->fragId = frag.m_fragId | (ent.m_fragBit << frag.m_fragOff);
-      // should cache this at fragment create
-      lockReq->fragPtrI = RNIL;
+      lockReq->fragPtrI = frag.m_accTableFragPtrI[ent.m_fragBit];
       const Uint32* const buf32 = static_cast<Uint32*>(keyPar.m_data);
       const Uint64* const buf64 = reinterpret_cast<const Uint64*>(buf32);
       lockReq->hashValue = md5_hash(buf64, keyPar.m_size);
@@ -700,14 +699,14 @@ Dbtux::scanFirst(Signal* signal, ScanOpPtr scanPtr)
   ScanOp& scan = *scanPtr.p;
   Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
   TreeHead& tree = frag.m_tree;
-  if (tree.m_root == NullTupAddr) {
+  if (tree.m_root == NullTupLoc) {
     // tree may have become empty
     jam();
     scan.m_state = ScanOp::Last;
     return;
   }
   TreePos pos;
-  pos.m_addr = tree.m_root;
+  pos.m_loc = tree.m_root;
   NodeHandlePtr nodePtr;
   // unpack lower bound
   const ScanBound& bound = *scan.m_bound[0];
@@ -725,7 +724,7 @@ Dbtux::scanFirst(Signal* signal, ScanOpPtr scanPtr)
   boundPar.m_dir = 0;
 loop: {
     jam();
-    selectNode(signal, frag, nodePtr, pos.m_addr, AccPref);
+    selectNode(signal, frag, nodePtr, pos.m_loc, AccPref);
     const unsigned occup = nodePtr.p->getOccup();
     ndbrequire(occup != 0);
     for (unsigned i = 0; i <= 1; i++) {
@@ -751,11 +750,11 @@ loop: {
       }
       if (i == 0 && ret < 0) {
         jam();
-        const TupAddr tupAddr = nodePtr.p->getLink(i);
-        if (tupAddr != NullTupAddr) {
+        const TupLoc loc = nodePtr.p->getLink(i);
+        if (loc != NullTupLoc) {
           jam();
           // continue to left subtree
-          pos.m_addr = tupAddr;
+          pos.m_loc = loc;
           goto loop;
         }
         // start scanning this node
@@ -769,11 +768,11 @@ loop: {
       }
       if (i == 1 && ret > 0) {
         jam();
-        const TupAddr tupAddr = nodePtr.p->getLink(i);
-        if (tupAddr != NullTupAddr) {
+        const TupLoc loc = nodePtr.p->getLink(i);
+        if (loc != NullTupLoc) {
           jam();
           // continue to right subtree
-          pos.m_addr = tupAddr;
+          pos.m_loc = loc;
           goto loop;
         }
         // start scanning upwards
@@ -870,7 +869,7 @@ Dbtux::scanNext(Signal* signal, ScanOpPtr scanPtr)
   TreePos pos = scan.m_scanPos;
   // get and remember original node
   NodeHandlePtr origNodePtr;
-  selectNode(signal, frag, origNodePtr, pos.m_addr, AccHead);
+  selectNode(signal, frag, origNodePtr, pos.m_loc, AccHead);
   ndbrequire(origNodePtr.p->islinkScan(scanPtr));
   // current node in loop
   NodeHandlePtr nodePtr = origNodePtr;
@@ -879,21 +878,21 @@ Dbtux::scanNext(Signal* signal, ScanOpPtr scanPtr)
     if (pos.m_dir == 2) {
       // coming up from root ends the scan
       jam();
-      pos.m_addr = NullTupAddr;
+      pos.m_loc = NullTupLoc;
       scan.m_state = ScanOp::Last;
       break;
     }
-    if (nodePtr.p->m_addr != pos.m_addr) {
+    if (nodePtr.p->m_loc != pos.m_loc) {
       jam();
-      selectNode(signal, frag, nodePtr, pos.m_addr, AccHead);
+      selectNode(signal, frag, nodePtr, pos.m_loc, AccHead);
     }
     if (pos.m_dir == 4) {
       // coming down from parent proceed to left child
       jam();
-      TupAddr addr = nodePtr.p->getLink(0);
-      if (addr != NullTupAddr) {
+      TupLoc loc = nodePtr.p->getLink(0);
+      if (loc != NullTupLoc) {
         jam();
-        pos.m_addr = addr;
+        pos.m_loc = loc;
         pos.m_dir = 4;  // unchanged
         continue;
       }
@@ -938,7 +937,7 @@ Dbtux::scanNext(Signal* signal, ScanOpPtr scanPtr)
         if (ret < 0) {
           jam();
           // hit upper bound of single range scan
-          pos.m_addr = NullTupAddr;
+          pos.m_loc = NullTupLoc;
           scan.m_state = ScanOp::Last;
           break;
         }
@@ -952,10 +951,10 @@ Dbtux::scanNext(Signal* signal, ScanOpPtr scanPtr)
         break;
       }
       // after node proceed to right child
-      TupAddr addr = nodePtr.p->getLink(1);
-      if (addr != NullTupAddr) {
+      TupLoc loc = nodePtr.p->getLink(1);
+      if (loc != NullTupLoc) {
         jam();
-        pos.m_addr = addr;
+        pos.m_loc = loc;
         pos.m_dir = 4;
         continue;
       }
@@ -965,7 +964,7 @@ Dbtux::scanNext(Signal* signal, ScanOpPtr scanPtr)
     if (pos.m_dir == 1) {
       // coming from right child proceed to parent
       jam();
-      pos.m_addr = nodePtr.p->getLink(2);
+      pos.m_loc = nodePtr.p->getLink(2);
       pos.m_dir = nodePtr.p->getSide();
       continue;
     }
@@ -975,7 +974,7 @@ Dbtux::scanNext(Signal* signal, ScanOpPtr scanPtr)
   scan.m_scanPos = pos;
   // relink
   if (scan.m_state == ScanOp::Current) {
-    ndbrequire(pos.m_addr == nodePtr.p->m_addr);
+    ndbrequire(pos.m_loc == nodePtr.p->m_loc);
     if (origNodePtr.i != nodePtr.i) {
       jam();
       origNodePtr.p->unlinkScan(scanPtr);
@@ -983,7 +982,7 @@ Dbtux::scanNext(Signal* signal, ScanOpPtr scanPtr)
     }
   } else if (scan.m_state == ScanOp::Last) {
     jam();
-    ndbrequire(pos.m_addr == NullTupAddr);
+    ndbrequire(pos.m_loc == NullTupLoc);
     origNodePtr.p->unlinkScan(scanPtr);
   } else {
     ndbrequire(false);

ndb/src/kernel/blocks/dbtux/Times.txt

+index maintenance overhead
+==========================
+
+"mc02" 2x1700 MHz linux-2.4.9 gcc-2.96 -O3 one db-node
+
+case a: index on Unsigned
+testOIBasic -case u -table 1 -index 1 -fragtype small -threads 10 -rows 100000 -subloop 1 -nologging
+
+case b: index on Varchar(5) + Varchar(5) + Varchar(20) + Unsigned
+testOIBasic -case u -table 2 -index 4 -fragtype small -threads 10 -rows 100000 -subloop 1 -nologging
+
+1 million rows, pk update without index, pk update with index
+shows ms / 1000 rows for each and pct overhead
+
+040616          mc02/a  40 ms   87 ms   114 pct
+                mc02/b  51 ms  128 ms   148 pct
+
+optim 1         mc02/a  38 ms   85 ms   124 pct
+                mc02/b  51 ms  123 ms   140 pct
+
+optim 2         mc02/a  41 ms   80 ms    96 pct
+                mc02/b  51 ms  117 ms   128 pct
+
+
+vim: set et:

ndb/test/ndbapi/testOIBasic.cpp

@@ -39,6 +39,7 @@ struct Opt {
   NdbDictionary::Object::FragmentType m_fragtype;
   const char* m_index;
   unsigned m_loop;
+  bool m_nologging;
   unsigned m_rows;
   unsigned m_scanrd;
   unsigned m_scanex;
@@ -54,6 +55,7 @@ struct Opt {
     m_fragtype(NdbDictionary::Object::FragUndefined),
     m_index(0),
     m_loop(1),
+    m_nologging(false),
     m_rows(1000),
     m_scanrd(240),
     m_scanex(240),
@@ -82,6 +84,7 @@ printhelp()
     << "  -fragtype T   fragment type single/small/medium/large" << endl
     << "  -index xyz    only given index numbers (digits 1-9)" << endl
     << "  -loop N       loop count full suite forever=0 [" << d.m_loop << "]" << endl
+    << "  -nologging    create tables in no-logging mode" << endl
     << "  -rows N       rows per thread [" << d.m_rows << "]" << endl
     << "  -scanrd N     scan read parallelism [" << d.m_scanrd << "]" << endl
     << "  -scanex N     scan exclusive parallelism [" << d.m_scanex << "]" << endl
@@ -476,7 +479,7 @@ tt1 = {
   "TT1", 5, tt1col, 4, tt1itab
 };
 
-// tt2 + tt2x1 tt2x2 tt2x3
+// tt2 + tt2x1 tt2x2 tt2x3 tt2x4
 
 static const Col
 tt2col[] = {
@@ -505,6 +508,14 @@ tt2x3col[] = {
   { 1, tt2col[4] }
 };
 
+static const ICol
+tt2x4col[] = {
+  { 0, tt2col[4] },
+  { 1, tt2col[3] },
+  { 2, tt2col[2] },
+  { 3, tt2col[1] }
+};
+
 static const ITab
 tt2x1 = {
   "TT2X1", 2, tt2x1col
@@ -520,16 +531,22 @@ tt2x3 = {
   "TT2X3", 2, tt2x3col
 };
 
+static const ITab
+tt2x4 = {
+  "TT2X4", 4, tt2x4col
+};
+
 static const ITab
 tt2itab[] = {
   tt2x1,
   tt2x2,
-  tt2x3
+  tt2x3,
+  tt2x4
 };
 
 static const Tab
 tt2 = {
-  "TT2", 5, tt2col, 3, tt2itab
+  "TT2", 5, tt2col, 4, tt2itab
 };
 
 // all tables
@@ -823,6 +840,9 @@ createtable(Par par)
   if (par.m_fragtype != NdbDictionary::Object::FragUndefined) {
     t.setFragmentType(par.m_fragtype);
   }
+  if (par.m_nologging) {
+    t.setLogging(false);
+  }
   for (unsigned k = 0; k < tab.m_cols; k++) {
     const Col& col = tab.m_col[k];
     NdbDictionary::Column c(col.m_name);
@@ -2500,9 +2520,28 @@ tbusybuild(Par par)
 }
 
 static int
-ttiming(Par par)
+ttimebuild(Par par)
 {
-  Tmr t0, t1, t2;
+  Tmr t1;
+  RUNSTEP(par, droptable, ST);
+  RUNSTEP(par, createtable, ST);
+  RUNSTEP(par, invalidatetable, MT);
+  for (unsigned i = 0; i < par.m_subloop; i++) {
+    RUNSTEP(par, pkinsert, MT);
+    t1.on();
+    RUNSTEP(par, createindex, ST);
+    t1.off(par.m_totrows);
+    RUNSTEP(par, invalidateindex, MT);
+    RUNSTEP(par, dropindex, ST);
+  }
+  LL1("build index - " << t1.time());
+  return 0;
+}
+
+static int
+ttimemaint(Par par)
+{
+  Tmr t1, t2;
   RUNSTEP(par, droptable, ST);
   RUNSTEP(par, createtable, ST);
   RUNSTEP(par, invalidatetable, MT);
@@ -2511,16 +2550,13 @@ ttiming(Par par)
     t1.on();
     RUNSTEP(par, pkupdate, MT);
     t1.off(par.m_totrows);
-    t0.on();
     RUNSTEP(par, createindex, ST);
     RUNSTEP(par, invalidateindex, MT);
-    t0.off(par.m_totrows);
     t2.on();
     RUNSTEP(par, pkupdate, MT);
     t2.off(par.m_totrows);
     RUNSTEP(par, dropindex, ST);
   }
-  LL1("build index - " << t0.time());
   LL1("update - " << t1.time());
   LL1("update indexed - " << t2.time());
   LL1("overhead - " << t2.over(t1));
@@ -2551,7 +2587,8 @@ tcaselist[] = {
   TCase("b", tpkops, "pk operations and scan reads"),
   TCase("c", tmixedops, "pk operations and scan operations"),
   TCase("d", tbusybuild, "pk operations and index build"),
-  TCase("t", ttiming, "time index build and maintenance"),
+  TCase("t", ttimebuild, "time index build"),
+  TCase("u", ttimemaint, "time index maintenance"),
   TCase("z", tdrop, "drop test tables")
 };
 
@@ -2689,6 +2726,10 @@ NDB_COMMAND(testOIBasic, "testOIBasic", "testOIBasic", "testOIBasic", 65535)
         continue;
       }
     }
+    if (strcmp(arg, "-nologging") == 0) {
+      g_opt.m_nologging = true;
+      continue;
+    }
     if (strcmp(arg, "-rows") == 0) {
       if (++argv, --argc > 0) {
         g_opt.m_rows = atoi(argv[0]);