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

into mc05.(none):/space2/tomas/mysql-4.1-ndb-test

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 is no longer used by TUX and can be removed
  */
 class TupStoreTh {
   friend class Dbtup;

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/dbtup/Dbtup.hpp

@@ -996,6 +996,14 @@ public:
   Dbtup(const class Configuration &);
   virtual ~Dbtup();
 
+  /*
+   * TUX index in TUP has single Uint32 array attribute which stores an
+   * index node.  TUX uses following methods.
+   */
+  int tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pageOffset, Uint32*& node);
+  void tuxFreeNode(Signal* signal, Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* node);
+  void tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& node);
+
 private:
   BLOCK_DEFINES(Dbtup);
 

ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp

@@ -179,6 +179,64 @@ Dbtup::execTUP_QUERY_TH(Signal* signal)
   return;
 }
 
+int
+Dbtup::tuxAllocNode(Signal* signal, Uint32 fragPtrI, Uint32& pageId, Uint32& pageOffset, Uint32*& node)
+{
+  FragrecordPtr fragPtr;
+  fragPtr.i = fragPtrI;
+  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
+  TablerecPtr tablePtr;
+  tablePtr.i = fragPtr.p->fragTableId;
+  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
+  PagePtr pagePtr;
+  terrorCode = 0;
+  if (! allocTh(fragPtr.p, tablePtr.p, NORMAL_PAGE, signal, pageOffset, pagePtr)) {
+    jam();
+    ndbrequire(terrorCode != 0);
+    return terrorCode;
+  }
+  pageId = pagePtr.i;
+  Uint32 attrDescIndex = tablePtr.p->tabDescriptor + (0 << ZAD_LOG_SIZE);
+  Uint32 attrDataOffset = AttributeOffset::getOffset(tableDescriptor[attrDescIndex + 1].tabDescr);
+  node = &pagePtr.p->pageWord[pageOffset] + attrDataOffset;
+  return 0;
+}
+
+void
+Dbtup::tuxFreeNode(Signal* signal, Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* node)
+{
+  FragrecordPtr fragPtr;
+  fragPtr.i = fragPtrI;
+  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
+  TablerecPtr tablePtr;
+  tablePtr.i = fragPtr.p->fragTableId;
+  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
+  PagePtr pagePtr;
+  pagePtr.i = pageId;
+  ptrCheckGuard(pagePtr, cnoOfPage, page);
+  Uint32 attrDescIndex = tablePtr.p->tabDescriptor + (0 << ZAD_LOG_SIZE);
+  Uint32 attrDataOffset = AttributeOffset::getOffset(tableDescriptor[attrDescIndex + 1].tabDescr);
+  ndbrequire(node == &pagePtr.p->pageWord[pageOffset] + attrDataOffset);
+  freeTh(fragPtr.p, tablePtr.p, signal, pagePtr.p, pageOffset);
+}
+
+void
+Dbtup::tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& node)
+{
+  FragrecordPtr fragPtr;
+  fragPtr.i = fragPtrI;
+  ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
+  TablerecPtr tablePtr;
+  tablePtr.i = fragPtr.p->fragTableId;
+  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
+  PagePtr pagePtr;
+  pagePtr.i = pageId;
+  ptrCheckGuard(pagePtr, cnoOfPage, page);
+  Uint32 attrDescIndex = tablePtr.p->tabDescriptor + (0 << ZAD_LOG_SIZE);
+  Uint32 attrDataOffset = AttributeOffset::getOffset(tableDescriptor[attrDescIndex + 1].tabDescr);
+  node = &pagePtr.p->pageWord[pageOffset] + attrDataOffset;
+}
+
 void
 Dbtup::execTUP_STORE_TH(Signal* signal)
 {

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) {
@@ -106,26 +106,24 @@ Dbtux::printTree(Signal* signal, Frag& frag, NdbOut& out)
       signal->theData[1] = 1;
       execDUMP_STATE_ORD(signal);
       if (debugFile != 0) {
-        commitNodes(signal, frag, false);
         printTree(signal, frag, debugOut);
       }
     }
     ndbrequire(false);
   }
-  commitNodes(signal, frag, false);
 }
 
 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);
-  out << par.m_path << " " << *nodePtr.p << endl;
+  NodeHandle node(frag);
+  selectNode(signal, node, loc, AccFull);
+  out << par.m_path << " " << node << endl;
   // check children
   PrintPar cpar[2];
   ndbrequire(strlen(par.m_path) + 1 < sizeof(par.m_path));
@@ -133,57 +131,57 @@ 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;
-    printNode(signal, frag, out, nodePtr.p->getLink(i), cpar[i]);
+    cpar[i].m_parent = loc;
+    printNode(signal, frag, out, node.getLink(i), cpar[i]);
     if (! cpar[i].m_ok) {
       par.m_ok = false;
     }
   }
   // check child-parent links
-  if (nodePtr.p->getLink(2) != par.m_parent) {
+  if (node.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 << node.getLink(2);
     out << " should be " << hex << par.m_parent << endl;
   }
-  if (nodePtr.p->getSide() != par.m_side) {
+  if (node.getSide() != par.m_side) {
     par.m_ok = false;
     out << par.m_path << " *** ";
-    out << "side " << dec << nodePtr.p->getSide();
+    out << "side " << dec << node.getSide();
     out << " should be " << dec << par.m_side << endl;
   }
   // check balance
   const int balance = -cpar[0].m_depth + cpar[1].m_depth;
-  if (nodePtr.p->getBalance() != balance) {
+  if (node.getBalance() != balance) {
     par.m_ok = false;
     out << par.m_path << " *** ";
-    out << "balance " << nodePtr.p->getBalance();
+    out << "balance " << node.getBalance();
     out << " should be " << balance << endl;
   }
-  if (abs(nodePtr.p->getBalance()) > 1) {
+  if (abs(node.getBalance()) > 1) {
     par.m_ok = false;
     out << par.m_path << " *** ";
-    out << "balance " << nodePtr.p->getBalance() << " is invalid" << endl;
+    out << "balance " << node.getBalance() << " is invalid" << endl;
   }
   // check occupancy
-  if (nodePtr.p->getOccup() > tree.m_maxOccup) {
+  if (node.getOccup() > tree.m_maxOccup) {
     par.m_ok = false;
     out << par.m_path << " *** ";
-    out << "occupancy " << nodePtr.p->getOccup();
+    out << "occupancy " << node.getOccup();
     out << " greater than max " << tree.m_maxOccup << endl;
   }
   // check for occupancy of interior node
-  if (nodePtr.p->getChilds() == 2 && nodePtr.p->getOccup() < tree.m_minOccup) {
+  if (node.getChilds() == 2 && node.getOccup() < tree.m_minOccup) {
     par.m_ok = false;
     out << par.m_path << " *** ";
-    out << "occupancy " << nodePtr.p->getOccup() << " of interior node";
+    out << "occupancy " << node.getOccup() << " of interior node";
     out << " less than min " << tree.m_minOccup << endl;
   }
   // 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 &&
-        nodePtr.p->getOccup() + cpar[i].m_occup <= tree.m_maxOccup) {
+    if (node.getLink(i) != NullTupLoc &&
+        node.getLink(1 - i) == NullTupLoc &&
+        node.getOccup() + cpar[i].m_occup <= tree.m_maxOccup) {
       par.m_ok = false;
       out << par.m_path << " *** ";
       out << "missed merge with child " << i << endl;
@@ -191,7 +189,19 @@ Dbtux::printNode(Signal* signal, Frag& frag, NdbOut& out, TupAddr addr, PrintPar
   }
   // return values
   par.m_depth = 1 + max(cpar[0].m_depth, cpar[1].m_depth);
-  par.m_occup = nodePtr.p->getOccup();
+  par.m_occup = node.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&
@@ -206,10 +216,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 +251,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,9 +351,8 @@ 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 << "]";
   if (node.m_acc >= Dbtux::AccPref) {
     for (unsigned i = 0; i <= 1; i++) {

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

@@ -21,6 +21,7 @@
 
 Dbtux::Dbtux(const Configuration& conf) :
   SimulatedBlock(DBTUX, conf),
+  c_tup(0),
   c_descPageList(RNIL),
 #ifdef VM_TRACE
   debugFile(0),
@@ -123,6 +124,8 @@ Dbtux::execSTTOR(Signal* signal)
   case 1:
     jam();
     CLEAR_ERROR_INSERT_VALUE;
+    c_tup = (Dbtup*)globalData.getBlock(DBTUP);
+    ndbrequire(c_tup != 0);
     break;
   case 3:
     jam();
@@ -175,12 +178,11 @@ Dbtux::execREAD_CONFIG_REQ(Signal* signal)
   c_fragPool.setSize(nFragment);
   c_descPagePool.setSize(nDescPage);
   c_fragOpPool.setSize(MaxIndexFragments);
-  c_nodeHandlePool.setSize(MaxNodeHandles);
   c_scanOpPool.setSize(nScanOp);
   c_scanBoundPool.setSize(nScanBoundWords);
   /*
    * Index id is physical array index.  We seize and initialize all
-   * index records now.  This assumes ArrayPool is an array.
+   * index records now.
    */
   IndexPtr indexPtr;
   while (1) {

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

@@ -72,7 +72,6 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal)
   }
   ndbrequire(fragPtr.i != RNIL);
   Frag& frag = *fragPtr.p;
-  ndbrequire(frag.m_nodeList == RNIL);
   // set up index entry
   TreeEnt ent;
   ent.m_tupAddr = req->tupAddr;
@@ -143,17 +142,18 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal)
     }
     /*
      * At most one new node is inserted in the operation.  We keep one
-     * free node pre-allocated so the operation cannot fail.  This also
-     * gives a real TupAddr for links to the new node.
+     * free node pre-allocated so the operation cannot fail.
      */
-    if (frag.m_nodeFree == RNIL) {
+    if (frag.m_freeLoc == NullTupLoc) {
       jam();
-      preallocNode(signal, frag, req->errorCode);
+      NodeHandle node(frag);
+      req->errorCode = allocNode(signal, node);
       if (req->errorCode != 0) {
         jam();
         break;
       }
-      ndbrequire(frag.m_nodeFree != RNIL);
+      frag.m_freeLoc = node.m_loc;
+      ndbrequire(frag.m_freeLoc != NullTupLoc);
     }
     treeAdd(signal, frag, treePos, ent);
     break;
@@ -175,7 +175,6 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal)
     break;
   }
   // commit and release nodes
-  commitNodes(signal, frag, req->errorCode == 0);
 #ifdef VM_TRACE
   if (debugFlags & DebugTree) {
     printTree(signal, frag, debugOut);
@@ -199,7 +198,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 +245,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;
@@ -270,100 +269,3 @@ Dbtux::tupReadKeys(Signal* signal, const Frag& frag, ReadPar& readPar)
   readPar.m_count = numKeys;
   readPar.m_size = copyPar.m_numwords;
 }
-
-/*
- * Operate on index node tuple in TUP.  The data is copied between node
- * cache and index storage via signal data.
- */
-void
-Dbtux::tupStoreTh(Signal* signal, const Frag& frag, NodeHandlePtr nodePtr, StorePar storePar)
-{
-  const TreeHead& tree = frag.m_tree;
-  // define the direct signal
-  TupStoreTh* req = (TupStoreTh*)signal->getDataPtrSend();
-  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->pageId = nodePtr.p->m_loc.m_pageId;
-  req->pageOffset = nodePtr.p->m_loc.m_pageOffset;
-  req->bufferId = 0;
-  req->opCode = storePar.m_opCode;
-  ndbrequire(storePar.m_offset + storePar.m_size <= tree.m_nodeSize);
-  req->dataOffset = storePar.m_offset;
-  req->dataSize = storePar.m_size;
-  // the node cache
-  ndbrequire(nodePtr.p->m_node != 0);
-  // the buffer in signal data
-  Uint32* const buffer = (Uint32*)req + TupStoreTh::SignalLength;
-  // copy in data
-  switch (storePar.m_opCode) {
-  case TupStoreTh::OpRead:
-    jam();
- #ifdef VM_TRACE
-    {
-      Uint32* dst = buffer + storePar.m_offset;
-      memset(dst, 0xa9, storePar.m_size << 2);
-    }
- #endif
-    break;
-  case TupStoreTh::OpInsert:
-    jam();
-    // fallthru
-  case TupStoreTh::OpUpdate:
-    jam();
-    // copy from cache to signal data
-    {
-      Uint32* dst = buffer + storePar.m_offset;
-      const Uint32* src = (const Uint32*)nodePtr.p->m_node + storePar.m_offset;
-      memcpy(dst, src, storePar.m_size << 2);
-    }
-    break;
-  case TupStoreTh::OpDelete:
-    jam();
-    break;
-  default:
-    ndbrequire(false);
-    break;
-  }
-  // execute
-  EXECUTE_DIRECT(DBTUP, GSN_TUP_STORE_TH, signal, TupStoreTh::SignalLength);
-  jamEntry();
-  if (req->errorCode != 0) {
-    jam();
-    storePar.m_errorCode = req->errorCode;
-    return;
-  }
-  ndbrequire(req->errorCode == 0);
-  // copy out data
-  switch (storePar.m_opCode) {
-  case TupStoreTh::OpRead:
-    jam();
-    {
-      Uint32* dst = (Uint32*)nodePtr.p->m_node + storePar.m_offset;
-      const Uint32* src = (const Uint32*)buffer + storePar.m_offset;
-      memcpy(dst, src, storePar.m_size << 2);
-    }
-    // fallthru
-  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::OpDelete:
-    jam();
-    nodePtr.p->m_addr = NullTupAddr;
-    nodePtr.p->m_loc.m_pageId = RNIL;
-    nodePtr.p->m_loc.m_pageOffset = 0;
-    break;
-  default:
-    ndbrequire(false);
-    break;
-  }
-}

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);
     }
@@ -310,12 +316,6 @@ Dbtux::dropIndex(Signal* signal, IndexPtr indexPtr, Uint32 senderRef, Uint32 sen
     unsigned i = --indexPtr.p->m_numFrags;
     FragPtr fragPtr;
     c_fragPool.getPtr(fragPtr, indexPtr.p->m_fragPtrI[i]);
-    Frag& frag = *fragPtr.p;
-    ndbrequire(frag.m_nodeList == RNIL);
-    if (frag.m_nodeFree != RNIL) {
-      c_nodeHandlePool.release(frag.m_nodeFree);
-      frag.m_nodeFree = RNIL;
-    }
     c_fragPool.release(fragPtr);
     // the real time break is not used for anything currently
     signal->theData[0] = TuxContinueB::DropIndex;

ndb/src/kernel/blocks/dbtux/Makefile.am

@@ -10,6 +10,8 @@ libdbtux_a_SOURCES = \
 			DbtuxCmp.cpp \
 			DbtuxDebug.cpp
 
+INCLUDES_LOC = -I$(top_srcdir)/ndb/src/kernel/blocks/dbtup
+
 include $(top_srcdir)/ndb/config/common.mk.am
 include $(top_srcdir)/ndb/config/type_kernel.mk.am
 

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
+the figures are based on single run on idle machine
+
+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
+
+optim 3         mc02/a  43 ms   80 ms    85 pct
+                mc02/b  54 ms  118 ms   117 pct
+
+optim 4         mc02/a  42 ms   80 ms    87 pct
+                mc02/b  51 ms  119 ms   129 pct
+
+optim 5         mc02/a  43 ms   77 ms    77 pct
+                mc02/b  54 ms  118 ms   117 pct
+
+optim 6         mc02/a  42 ms   70 ms    66 pct
+                mc02/b  53 ms  109 ms   105 pct
+
+vim: set et:

ndb/src/kernel/blocks/qmgr/QmgrMain.cpp

@@ -525,7 +525,7 @@ void Qmgr::execCM_REGREQ(Signal* signal)
   cmRegConf->dynamicId         = TdynId;
   c_clusterNodes.copyto(NdbNodeBitmask::Size, cmRegConf->allNdbNodes);
   sendSignal(Tblockref, GSN_CM_REGCONF, signal, 
-	     CmRegConf::SignalLength, JBB);
+	     CmRegConf::SignalLength, JBA);
   DEBUG_START(GSN_CM_REGCONF, refToNode(Tblockref), "");
 
   /**
@@ -847,7 +847,7 @@ void Qmgr::execCM_NODEINFOCONF(Signal* signal)
   nodePtr.i = getOwnNodeId();
   ptrAss(nodePtr, nodeRec);  
   ndbrequire(nodePtr.p->phase == ZSTARTING);
-  ndbrequire(c_start.m_gsn = GSN_CM_NODEINFOREQ);
+  ndbrequire(c_start.m_gsn == GSN_CM_NODEINFOREQ);
   c_start.m_nodes.clearWaitingFor(nodeId);
 
   /**
@@ -1019,7 +1019,7 @@ void Qmgr::execCM_ADD(Signal* signal)
     ndbrequire(addNodePtr.i == nodePtr.i);
     switch(type){
     case CmAdd::Prepare:
-      ndbrequire(c_start.m_gsn = GSN_CM_NODEINFOREQ);
+      ndbrequire(c_start.m_gsn == GSN_CM_NODEINFOREQ);
       /**
        * Wait for CM_NODEINFO_CONF
        */

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);
@@ -2202,7 +2222,6 @@ pkupdateindexbuild(Par par)
 {
   if (par.m_no == 0) {
     CHK(createindex(par) == 0);
-    CHK(invalidateindex(par) == 0);
   } else {
     CHK(pkupdate(par) == 0);
   }
@@ -2493,6 +2512,7 @@ tbusybuild(Par par)
   RUNSTEP(par, pkinsert, MT);
   for (unsigned i = 0; i < par.m_subloop; i++) {
     RUNSTEP(par, pkupdateindexbuild, MT);
+    RUNSTEP(par, invalidateindex, MT);
     RUNSTEP(par, readverify, MT);
     RUNSTEP(par, dropindex, ST);
   }
@@ -2500,9 +2520,28 @@ tbusybuild(Par par)
 }
 
 static int
-ttiming(Par par)
+ttimebuild(Par par)
+{
+  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 t0, t1, t2;
+  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]);