ndb - bug#31477 post-merge fixes in 5.1

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

@@ -1558,7 +1558,7 @@ public:
   /*
    * TUX checks if tuple is visible to scan.
    */
-  bool tuxQueryTh(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, bool dirty, Uint32 savePointId);
+  bool tuxQueryTh(Uint32 fragPtrI, Uint32 pageId, Uint32 pageIndex, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, bool dirty, Uint32 savepointId);
 
   int load_diskpage(Signal*, Uint32 opRec, Uint32 fragPtrI, 
 		    Uint32 local_key, Uint32 flags);
@@ -3041,16 +3041,15 @@ inline
 bool Dbtup::find_savepoint(OperationrecPtr& loopOpPtr, Uint32 savepointId)
 {
   while (true) {
-    if (savepointId > loopOpPtr.p->savePointId) {
+    if (savepointId > loopOpPtr.p->savepointId) {
       jam();
       return true;
     }
-    // note 5.0 has reversed next/prev pointers
-    loopOpPtr.i = loopOpPtr.p->nextActiveOp;
+    loopOpPtr.i = loopOpPtr.p->prevActiveOp;
     if (loopOpPtr.i == RNIL) {
       break;
     }
-    ptrCheckGuard(loopOpPtr, cnoOfOprec, operationrec);
+    c_operation_pool.getPtr(loopOpPtr);
   }
   return false;
 }

storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp

@@ -358,21 +358,7 @@ Dbtup::setup_read(KeyReqStruct *req_struct,
       dirty= false;
     }
 
-    OperationrecPtr prevOpPtr = currOpPtr;  
-    bool found= false;
-    while(true) 
-    {
-      if (savepointId > currOpPtr.p->savepointId) {
-	found= true;
-	break;
-      }
-      if (currOpPtr.p->is_first_operation()){
-	break;
-      }
-      prevOpPtr= currOpPtr;
-      currOpPtr.i = currOpPtr.p->prevActiveOp;
-      c_operation_pool.getPtr(currOpPtr);
-    }
+    bool found= find_savepoint(currOpPtr, savepointId);
     
     Uint32 currOp= currOpPtr.p->op_struct.op_type;
     

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

@@ -15,6 +15,7 @@
 
 #define DBTUP_C
 #define DBTUP_INDEX_CPP
+#include <Dblqh.hpp>
 #include "Dbtup.hpp"
 #include <RefConvert.hpp>
 #include <ndb_limits.h>
@@ -330,7 +331,14 @@ Dbtup::accReadPk(Uint32 tableId, Uint32 fragId, Uint32 fragPageId, Uint32 pageIn
  * clear to do the full check here.
  */
 bool
-Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, bool dirty, Uint32 savePointId)
+Dbtup::tuxQueryTh(Uint32 fragPtrI,
+                  Uint32 pageId,
+                  Uint32 pageIndex,
+                  Uint32 tupVersion,
+                  Uint32 transId1,
+                  Uint32 transId2,
+                  bool dirty,
+                  Uint32 savepointId)
 {
   jamEntry();
   FragrecordPtr fragPtr;
@@ -341,30 +349,40 @@ Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupV
   ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
   PagePtr pagePtr;
   pagePtr.i = pageId;
-  ptrCheckGuard(pagePtr, cnoOfPage, page);
+  c_page_pool.getPtr(pagePtr);
+
+  KeyReqStruct req_struct;
+
+  {
+    Operationrec tmpOp;
+    tmpOp.m_tuple_location.m_page_no = pageId;
+    tmpOp.m_tuple_location.m_page_idx = pageIndex;
+    setup_fixed_part(&req_struct, &tmpOp, tablePtr.p);
+  }
+
+  Tuple_header* tuple_ptr = req_struct.m_tuple_ptr;
 
   OperationrecPtr currOpPtr;
-  currOpPtr.i = pagePtr.p->pageWord[pageOffset];
+  currOpPtr.i = tuple_ptr->m_operation_ptr_i;
   if (currOpPtr.i == RNIL) {
-    ljam();
+    jam();
     // tuple has no operation, any scan can see it
     return true;
   }
-  ptrCheckGuard(currOpPtr, cnoOfOprec, operationrec);
+  c_operation_pool.getPtr(currOpPtr);
 
   const bool sameTrans =
-    transId1 == currOpPtr.p->transid1 &&
-    transId2 == currOpPtr.p->transid2;
+    c_lqh->is_same_trans(currOpPtr.p->userpointer, transId1, transId2);
 
   bool res = false;
   OperationrecPtr loopOpPtr = currOpPtr;
 
   if (!sameTrans) {
-    ljam();
+    jam();
     if (!dirty) {
-      ljam();
-      if (currOpPtr.p->prevActiveOp == RNIL) {
-        ljam();
+      jam();
+      if (currOpPtr.p->nextActiveOp == RNIL) {
+        jam();
         // last op - TUX makes ACC lock request in same timeslice
         res = true;
       }
@@ -372,33 +390,33 @@ Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupV
     else {
       // loop to first op (returns false)
       find_savepoint(loopOpPtr, 0);
-      const Uint32 op_type = loopOpPtr.p->optype;
+      const Uint32 op_type = loopOpPtr.p->op_struct.op_type;
 
       if (op_type != ZINSERT) {
-        ljam();
-        // read committed version from the page
-        const Uint32 origVersion = pagePtr.p->pageWord[pageOffset + 1];
+        jam();
+        // read committed version
+        const Uint32 origVersion = tuple_ptr->get_tuple_version();
         if (origVersion == tupVersion) {
-          ljam();
+          jam();
           res = true;
         }
       }
     }
   }
   else {
-    ljam();
+    jam();
     // for own trans, ignore dirty flag
 
-    if (find_savepoint(loopOpPtr, savePointId)) {
-      ljam();
-      const Uint32 op_type = loopOpPtr.p->optype;
+    if (find_savepoint(loopOpPtr, savepointId)) {
+      jam();
+      const Uint32 op_type = loopOpPtr.p->op_struct.op_type;
 
       if (op_type != ZDELETE) {
-        ljam();
+        jam();
         // check if this op has produced the scanned version
         Uint32 loopVersion = loopOpPtr.p->tupVersion;
         if (loopVersion == tupVersion) {
-          ljam();
+          jam();
           res = true;
         }
       }