Merge orca.ndb.mysql.com:/export/home/space/pekka/ndb/version/my50-bug20446

into  orca.ndb.mysql.com:/export/home/space/pekka/ndb/version/my51-bug20446

storage/ndb/src/kernel/blocks/dbtux/Dbtux.hpp

@@ -192,6 +192,7 @@ private:
     unsigned m_tupVersion : 15; // version
     TreeEnt();
     // methods
+    bool eqtuple(const TreeEnt ent) const;
     bool eq(const TreeEnt ent) const;
     int cmp(const TreeEnt ent) const;
   };
@@ -270,8 +271,7 @@ private:
   struct TreePos {
     TupLoc m_loc;               // physical node address
     Uint16 m_pos;               // position 0 to m_occup
-    Uint8 m_match;              // at an existing entry
-    Uint8 m_dir;                // see scanNext()
+    Uint8 m_dir;                // see scanNext
     TreePos();
   };
 
@@ -362,12 +362,13 @@ private:
     enum {
       Undef = 0,
       First = 1,                // before first entry
-      Current = 2,              // at current before locking
-      Blocked = 3,              // at current waiting for ACC lock
-      Locked = 4,               // at current and locked or no lock needed
-      Next = 5,                 // looking for next extry
-      Last = 6,                 // after last entry
-      Aborting = 7,             // lock wait at scan close
+      Current = 2,              // at some entry
+      Found = 3,                // return current as next scan result
+      Blocked = 4,              // found and waiting for ACC lock
+      Locked = 5,               // found and locked or no lock needed
+      Next = 6,                 // looking for next extry
+      Last = 7,                 // after last entry
+      Aborting = 8,             // lock wait at scan close
       Invalid = 9               // cannot return REF to LQH currently
     };
     Uint16 m_state;
@@ -544,6 +545,7 @@ private:
   void readKeyAttrs(const Frag& frag, TreeEnt ent, unsigned start, Data keyData);
   void readTablePk(const Frag& frag, TreeEnt ent, Data pkData, unsigned& pkSize);
   void copyAttrs(const Frag& frag, ConstData data1, Data data2, unsigned maxlen2 = MaxAttrDataSize);
+  void unpackBound(const ScanBound& bound, Data data);
 
   /*
    * DbtuxMeta.cpp
@@ -618,7 +620,9 @@ private:
   void execACCKEYREF(Signal* signal);
   void execACC_ABORTCONF(Signal* signal);
   void scanFirst(ScanOpPtr scanPtr);
+  void scanFind(ScanOpPtr scanPtr);
   void scanNext(ScanOpPtr scanPtr, bool fromMaintReq);
+  bool scanCheck(ScanOpPtr scanPtr, TreeEnt ent);
   bool scanVisible(ScanOpPtr scanPtr, TreeEnt ent);
   void scanClose(Signal* signal, ScanOpPtr scanPtr);
   void addAccLockOp(ScanOp& scan, Uint32 accLockOp);
@@ -628,8 +632,8 @@ private:
   /*
    * DbtuxSearch.cpp
    */
-  void searchToAdd(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos);
-  void searchToRemove(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos);
+  bool searchToAdd(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos);
+  bool searchToRemove(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos);
   void searchToScan(Frag& frag, ConstData boundInfo, unsigned boundCount, bool descending, TreePos& treePos);
   void searchToScanAscending(Frag& frag, ConstData boundInfo, unsigned boundCount, TreePos& treePos);
   void searchToScanDescending(Frag& frag, ConstData boundInfo, unsigned boundCount, TreePos& treePos);
@@ -793,6 +797,14 @@ Dbtux::TreeEnt::TreeEnt() :
 {
 }
 
+inline bool
+Dbtux::TreeEnt::eqtuple(const TreeEnt ent) const
+{
+  return
+    m_tupLoc == ent.m_tupLoc &&
+    m_fragBit == ent.m_fragBit;
+}
+
 inline bool
 Dbtux::TreeEnt::eq(const TreeEnt ent) const
 {
@@ -812,10 +824,25 @@ Dbtux::TreeEnt::cmp(const TreeEnt ent) const
     return -1;
   if (m_tupLoc.getPageOffset() > ent.m_tupLoc.getPageOffset())
     return +1;
-  if (m_tupVersion < ent.m_tupVersion)
-    return -1;
-  if (m_tupVersion > ent.m_tupVersion)
-    return +1;
+  /*
+   * Guess if one tuple version has wrapped around.  This is well
+   * defined ordering on existing versions since versions are assigned
+   * consecutively and different versions exists only on uncommitted
+   * tuple.  Assuming max 2**14 uncommitted ops on same tuple.
+   */
+  const unsigned version_wrap_limit = (1 << (ZTUP_VERSION_BITS - 1));
+  if (m_tupVersion < ent.m_tupVersion) {
+    if (ent.m_tupVersion - m_tupVersion < version_wrap_limit)
+      return -1;
+    else
+      return +1;
+  }
+  if (m_tupVersion > ent.m_tupVersion) {
+    if (m_tupVersion - ent.m_tupVersion < version_wrap_limit)
+      return +1;
+    else
+      return -1;
+  }
   return 0;
 }
 
@@ -883,7 +910,6 @@ inline
 Dbtux::TreePos::TreePos() :
   m_loc(),
   m_pos(ZNIL),
-  m_match(false),
   m_dir(255)
 {
 }

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

@@ -280,7 +280,6 @@ operator<<(NdbOut& out, const Dbtux::TreePos& pos)
   out << "[TreePos " << hex << &pos;
   out << " [loc " << pos.m_loc << "]";
   out << " [pos " << dec << pos.m_pos << "]";
-  out << " [match " << dec << pos.m_match << "]";
   out << " [dir " << dec << pos.m_dir << "]";
   out << "]";
   return out;

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

@@ -318,4 +318,17 @@ Dbtux::copyAttrs(const Frag& frag, ConstData data1, Data data2, unsigned maxlen2
 #endif
 }
 
+void
+Dbtux::unpackBound(const ScanBound& bound, Data dest)
+{
+  ScanBoundIterator iter;
+  bound.first(iter);
+  const unsigned n = bound.getSize();
+  unsigned j;
+  for (j = 0; j < n; j++) {
+    dest[j] = *iter.data;
+    bound.next(iter);
+  }
+}
+
 BLOCK_FUNCTIONS(Dbtux)

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

@@ -111,16 +111,17 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal)
   // do the operation
   req->errorCode = 0;
   TreePos treePos;
+  bool ok;
   switch (opCode) {
   case TuxMaintReq::OpAdd:
     jam();
-    searchToAdd(frag, c_searchKey, ent, treePos);
+    ok = searchToAdd(frag, c_searchKey, ent, treePos);
 #ifdef VM_TRACE
     if (debugFlags & DebugMaint) {
-      debugOut << treePos << (treePos.m_match ? " - error" : "") << endl;
+      debugOut << treePos << (! ok ? " - error" : "") << endl;
     }
 #endif
-    if (treePos.m_match) {
+    if (! ok) {
       jam();
       // there is no "Building" state so this will have to do
       if (indexPtr.p->m_state == Index::Online) {
@@ -150,13 +151,13 @@ Dbtux::execTUX_MAINT_REQ(Signal* signal)
     break;
   case TuxMaintReq::OpRemove:
     jam();
-    searchToRemove(frag, c_searchKey, ent, treePos);
+    ok = searchToRemove(frag, c_searchKey, ent, treePos);
 #ifdef VM_TRACE
     if (debugFlags & DebugMaint) {
-      debugOut << treePos << (! treePos.m_match ? " - error" : "") << endl;
+      debugOut << treePos << (! ok ? " - error" : "") << endl;
     }
 #endif
-    if (! treePos.m_match) {
+    if (! ok) {
       jam();
       // there is no "Building" state so this will have to do
       if (indexPtr.p->m_state == Index::Online) {

storage/ndb/src/kernel/blocks/dbtux/DbtuxSearch.cpp

@@ -21,22 +21,18 @@
  * Search for entry to add.
  *
  * Similar to searchToRemove (see below).
- *
- * TODO optimize for initial equal attrs in node min/max
  */
-void
+bool
 Dbtux::searchToAdd(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos)
 {
   const TreeHead& tree = frag.m_tree;
   const unsigned numAttrs = frag.m_numAttrs;
   NodeHandle currNode(frag);
   currNode.m_loc = tree.m_root;
-  // assume success
-  treePos.m_match = false;
   if (currNode.m_loc == NullTupLoc) {
     // empty tree
     jam();
-    return;
+    return true;
   }
   NodeHandle glbNode(frag);     // potential g.l.b of final node
   /*
@@ -94,9 +90,8 @@ Dbtux::searchToAdd(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos&
       jam();
       treePos.m_loc = currNode.m_loc;
       treePos.m_pos = 0;
-      // failed
-      treePos.m_match = true;
-      return;
+      // entry found - error
+      return false;
     }
     break;
   }
@@ -104,7 +99,7 @@ Dbtux::searchToAdd(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos&
   treePos.m_loc = currNode.m_loc;
   // binary search
   int lo = -1;
-  unsigned hi = currNode.getOccup();
+  int hi = currNode.getOccup();
   int ret;
   while (1) {
     jam();
@@ -126,9 +121,8 @@ Dbtux::searchToAdd(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos&
       lo = j;
     else {
       treePos.m_pos = j;
-      // failed
-      treePos.m_match = true;
-      return;
+      // entry found - error
+      return false;
     }
     if (hi - lo == 1)
       break;
@@ -136,22 +130,23 @@ Dbtux::searchToAdd(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos&
   if (ret < 0) {
     jam();
     treePos.m_pos = hi;
-    return;
+    return true;
   }
   if (hi < currNode.getOccup()) {
     jam();
     treePos.m_pos = hi;
-    return;
+    return true;
   }
   if (bottomNode.isNull()) {
     jam();
     treePos.m_pos = hi;
-    return;
+    return true;
   }
   jam();
   // backwards compatible for now
   treePos.m_loc = bottomNode.m_loc;
   treePos.m_pos = 0;
+  return true;
 }
 
 /*
@@ -163,21 +158,17 @@ Dbtux::searchToAdd(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos&
  * then the saved node is the g.l.b of the final node and we move back
  * to it.
  */
-void
+bool
 Dbtux::searchToRemove(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos)
 {
   const TreeHead& tree = frag.m_tree;
   const unsigned numAttrs = frag.m_numAttrs;
   NodeHandle currNode(frag);
   currNode.m_loc = tree.m_root;
-  // assume success
-  treePos.m_match = true;
   if (currNode.m_loc == NullTupLoc) {
-    // empty tree
+    // empty tree - failed
     jam();
-    // failed
-    treePos.m_match = false;
-    return;
+    return false;
   }
   NodeHandle glbNode(frag);     // potential g.l.b of final node
   while (true) {
@@ -229,7 +220,7 @@ Dbtux::searchToRemove(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePo
       jam();
       treePos.m_loc = currNode.m_loc;
       treePos.m_pos = 0;
-      return;
+      return true;
     }
     break;
   }
@@ -242,12 +233,12 @@ Dbtux::searchToRemove(Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePo
     if (searchEnt.eq(currNode.getEnt(j))) {
       jam();
       treePos.m_pos = j;
-      return;
+      return true;
     }
   }
   treePos.m_pos = currNode.getOccup();
-  // failed
-  treePos.m_match = false;
+  // not found - failed
+  return false;
 }
 
 /*
@@ -278,8 +269,6 @@ Dbtux::searchToScanAscending(Frag& frag, ConstData boundInfo, unsigned boundCoun
   currNode.m_loc = tree.m_root;
   NodeHandle glbNode(frag);     // potential g.l.b of final node
   NodeHandle bottomNode(frag);
-  // always before entry
-  treePos.m_match = false;
   while (true) {
     jam();
     selectNode(currNode, currNode.m_loc);
@@ -315,7 +304,7 @@ Dbtux::searchToScanAscending(Frag& frag, ConstData boundInfo, unsigned boundCoun
         treePos.m_dir = 3;
         return;
       }
-    } else if (ret > 0) {
+    } else {
       // bound is at or right of this node
       jam();
       const TupLoc loc = currNode.getLink(1);
@@ -327,8 +316,6 @@ Dbtux::searchToScanAscending(Frag& frag, ConstData boundInfo, unsigned boundCoun
         currNode.m_loc = loc;
         continue;
       }
-    } else {
-      ndbrequire(false);
     }
     break;
   }
@@ -369,8 +356,6 @@ Dbtux::searchToScanDescending(Frag& frag, ConstData boundInfo, unsigned boundCou
   currNode.m_loc = tree.m_root;
   NodeHandle glbNode(frag);     // potential g.l.b of final node
   NodeHandle bottomNode(frag);
-  // always before entry
-  treePos.m_match = false;
   while (true) {
     jam();
     selectNode(currNode, currNode.m_loc);
@@ -403,7 +388,7 @@ Dbtux::searchToScanDescending(Frag& frag, ConstData boundInfo, unsigned boundCou
         // empty result set
         return;
       }
-    } else if (ret > 0) {
+    } else {
       // bound is at or right of this node
       jam();
       const TupLoc loc = currNode.getLink(1);
@@ -415,8 +400,6 @@ Dbtux::searchToScanDescending(Frag& frag, ConstData boundInfo, unsigned boundCou
         currNode.m_loc = loc;
         continue;
       }
-    } else {
-      ndbrequire(false);
     }
     break;
   }