Merge sama.ndb.mysql.com:/export/space/pekka/ndb/version/my50-ndb

into  sama.ndb.mysql.com:/export/space/pekka/ndb/version/my50-bug31477

ndb/src/common/util/NdbOut.cpp

@@ -29,7 +29,7 @@ static const char * fms[] = {
   "%d", "0x%08x",      // Int32
   "%u", "0x%08x",      // Uint32
   "%lld", "0x%016llx", // Int64
-  "%llu", "0x%016llx"  // Uint64
+  "%llu", "0x%016llx", // Uint64
   "%llu", "0x%016llx"  // UintPtr
 };
 

ndb/src/kernel/blocks/dbtup/Dbtup.hpp

@@ -1085,7 +1085,7 @@ public:
   /*
    * TUX checks if tuple is visible to scan.
    */
-  bool tuxQueryTh(Uint32 fragPtrI, Uint32 tupAddr, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, Uint32 savePointId);
+  bool tuxQueryTh(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, bool dirty, Uint32 savePointId);
 
 private:
   BLOCK_DEFINES(Dbtup);
@@ -1942,6 +1942,8 @@ private:
   bool getPageThroughSavePoint(Operationrec* const regOperPtr,
                                Operationrec* const leaderOpPtr);
 
+  bool find_savepoint(OperationrecPtr& loopOpPtr, Uint32 savepointId);
+
   Uint32 calculateChecksum(Page* const pagePtr, Uint32 tupHeadOffset, Uint32 tupHeadSize);
   void setChecksum(Page* const pagePtr, Uint32 tupHeadOffset, Uint32 tupHeadSize);
 
@@ -2467,4 +2469,22 @@ bool Dbtup::isPageUndoLogged(Fragrecord* const regFragPtr,
   return false;
 }//Dbtup::isUndoLoggingNeeded()
 
+inline
+bool Dbtup::find_savepoint(OperationrecPtr& loopOpPtr, Uint32 savepointId)
+{
+  while (true) {
+    if (savepointId > loopOpPtr.p->savePointId) {
+      jam();
+      return true;
+    }
+    // note 5.0 has reversed next/prev pointers
+    loopOpPtr.i = loopOpPtr.p->nextActiveOp;
+    if (loopOpPtr.i == RNIL) {
+      break;
+    }
+    ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
+  }
+  return false;
+}
+
 #endif

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

@@ -246,8 +246,18 @@ Dbtup::accReadPk(Uint32 tableId, Uint32 fragId, Uint32 fragPageId, Uint32 pageIn
   return ret;
 }
 
+/*
+ * TUX index contains all tuple versions.  A scan in TUX has scanned
+ * one of them and asks if it can be returned as scan result.  This
+ * depends on trans id, dirty read flag, and savepoint within trans.
+ *
+ * Previously this faked a ZREAD operation and used getPage().
+ * In TUP getPage() is run after ACC locking, but TUX comes here
+ * before ACC access.  Instead of modifying getPage() it is more
+ * clear to do the full check here.
+ */
 bool
-Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 tupAddr, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, Uint32 savePointId)
+Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, bool dirty, Uint32 savePointId)
 {
   ljamEntry();
   FragrecordPtr fragPtr;
@@ -256,33 +266,73 @@ Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 tupAddr, Uint32 tupVersion, Uint32 tra
   TablerecPtr tablePtr;
   tablePtr.i = fragPtr.p->fragTableId;
   ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
-  // get page
   PagePtr pagePtr;
-  Uint32 fragPageId = tupAddr >> MAX_TUPLES_BITS;
-  Uint32 pageIndex = tupAddr & ((1 << MAX_TUPLES_BITS ) - 1);
-  // use temp op rec
-  Operationrec tempOp;
-  tempOp.fragPageId = fragPageId;
-  tempOp.pageIndex = pageIndex;
-  tempOp.transid1 = transId1;
-  tempOp.transid2 = transId2;
-  tempOp.savePointId = savePointId;
-  tempOp.optype = ZREAD;
-  tempOp.dirtyOp = 1;
-  if (getPage(pagePtr, &tempOp, fragPtr.p, tablePtr.p)) {
-    /*
-    * We use the normal getPage which will return the tuple to be used
-    * for this transaction and savepoint id.  If its tuple version
-    * equals the requested then we have a visible tuple otherwise not.
-    */
+  pagePtr.i = pageId;
+  ptrCheckGuard(pagePtr, cnoOfPage, page);
+
+  OperationrecPtr currOpPtr;
+  currOpPtr.i = pagePtr.p->pageWord[pageOffset];
+  if (currOpPtr.i == RNIL) {
     ljam();
-    Uint32 read_tupVersion = pagePtr.p->pageWord[tempOp.pageOffset + 1];
-    if (read_tupVersion == tupVersion) {
+    // tuple has no operation, any scan can see it
+    return true;
+  }
+  ptrCheckGuard(currOpPtr, cnoOfOprec, operationrec);
+
+  const bool sameTrans =
+    transId1 == currOpPtr.p->transid1 &&
+    transId2 == currOpPtr.p->transid2;
+
+  bool res = false;
+  OperationrecPtr loopOpPtr = currOpPtr;
+
+  if (!sameTrans) {
+    ljam();
+    if (!dirty) {
       ljam();
-      return true;
+      if (currOpPtr.p->prevActiveOp == RNIL) {
+        ljam();
+        // last op - TUX makes ACC lock request in same timeslice
+        res = true;
+      }
+    }
+    else {
+      // loop to first op (returns false)
+      find_savepoint(loopOpPtr, 0);
+      const Uint32 op_type = loopOpPtr.p->optype;
+
+      if (op_type != ZINSERT) {
+        ljam();
+        // read committed version from the page
+        const Uint32 origVersion = pagePtr.p->pageWord[pageOffset + 1];
+        if (origVersion == tupVersion) {
+          ljam();
+          res = true;
+        }
+      }
     }
   }
-  return false;
+  else {
+    ljam();
+    // for own trans, ignore dirty flag
+
+    if (find_savepoint(loopOpPtr, savePointId)) {
+      ljam();
+      const Uint32 op_type = loopOpPtr.p->optype;
+
+      if (op_type != ZDELETE) {
+        ljam();
+        // check if this op has produced the scanned version
+        Uint32 loopVersion = loopOpPtr.p->tupVersion;
+        if (loopVersion == tupVersion) {
+          ljam();
+          res = true;
+        }
+      }
+    }
+  }
+
+  return res;
 }
 
 // ordered index build

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

@@ -984,7 +984,8 @@ Dbtux::scanVisible(ScanOpPtr scanPtr, TreeEnt ent)
   const Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
   Uint32 fragBit = ent.m_fragBit;
   Uint32 tableFragPtrI = frag.m_tupTableFragPtrI[fragBit];
-  Uint32 tupAddr = getTupAddr(frag, ent);
+  Uint32 pageId = ent.m_tupLoc.getPageId();
+  Uint32 pageOffset = ent.m_tupLoc.getPageOffset();
   Uint32 tupVersion = ent.m_tupVersion;
   // check for same tuple twice in row
   if (scan.m_scanEnt.m_tupLoc == ent.m_tupLoc &&
@@ -994,8 +995,9 @@ Dbtux::scanVisible(ScanOpPtr scanPtr, TreeEnt ent)
   }
   Uint32 transId1 = scan.m_transId1;
   Uint32 transId2 = scan.m_transId2;
+  bool dirty = scan.m_readCommitted;
   Uint32 savePointId = scan.m_savePointId;
-  bool ret = c_tup->tuxQueryTh(tableFragPtrI, tupAddr, tupVersion, transId1, transId2, savePointId);
+  bool ret = c_tup->tuxQueryTh(tableFragPtrI, pageId, pageOffset, tupVersion, transId1, transId2, dirty, savePointId);
   jamEntry();
   return ret;
 }