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

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


storage/ndb/include/ndbapi/Ndb.hpp:
  Auto merged
storage/ndb/src/common/util/NdbOut.cpp:
  Auto merged
storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp:
  Auto merged
storage/ndb/test/ndbapi/testOIBasic.cpp:
  Auto merged
storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp:
  mindless merge
storage/ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp:
  mindless merge

mysql-test/r/ndb_bug31477.result

+drop table if exists t1;
+create table t1(a int primary key, b int, c int, unique(b)) engine = ndb;
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+begin;
+insert into t1 values (1,1,1);
+begin;
+update t1 set c = 2 where b = 1;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+rollback;
+rollback;
+drop table t1;
+create table t1(a int primary key, b int, c int, key(b)) engine = ndb;
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+begin;
+insert into t1 values (1,1,1);
+begin;
+update t1 set c = 2 where b = 1;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+rollback;
+rollback;
+drop table t1;
+--con1
+create table t1(a int primary key, b int, c int, key(b)) engine = ndb;
+insert into t1 values (1,1,1);
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+begin;
+update t1 set c = 10 where a = 1;
+update t1 set c = 20 where a = 1;
+update t1 set c = 30 where a = 1;
+--con1 c=30
+select * from t1 where b >= 1 order by b;
+a	b	c
+1	1	30
+2	2	2
+3	3	3
+4	4	4
+--con2 c=1
+select * from t1 where b >= 1 order by b;
+a	b	c
+1	1	1
+2	2	2
+3	3	3
+4	4	4
+--con1
+delete from t1 where a = 1;
+--con1 c=none
+select * from t1 where b >= 1 order by b;
+a	b	c
+2	2	2
+3	3	3
+4	4	4
+--con2 c=1
+select * from t1 where b >= 1 order by b;
+a	b	c
+1	1	1
+2	2	2
+3	3	3
+4	4	4
+--con1
+commit;
+--con1 c=none
+select * from t1 where b >= 1 order by b;
+a	b	c
+2	2	2
+3	3	3
+4	4	4
+--con2 c=none
+select * from t1 where b >= 1 order by b;
+a	b	c
+2	2	2
+3	3	3
+4	4	4
+--con1
+begin;
+insert into t1 values (1,1,1);
+update t1 set c = 10 where a = 1;
+update t1 set c = 20 where a = 1;
+update t1 set c = 30 where a = 1;
+--con1 c=30
+select * from t1 where b >= 1 order by b;
+a	b	c
+1	1	30
+2	2	2
+3	3	3
+4	4	4
+--con2 c=none
+select * from t1 where b >= 1 order by b;
+a	b	c
+2	2	2
+3	3	3
+4	4	4
+drop table t1;

mysql-test/t/ndb_bug31477.test

+--source include/have_ndb.inc
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+
+# setup
+
+connect (con1,localhost,root,,test);
+connect (con2,localhost,root,,test);
+
+# unique index
+connection con1;
+create table t1(a int primary key, b int, c int, unique(b)) engine = ndb;
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+
+begin;
+insert into t1 values (1,1,1);
+
+connection con2;
+begin;
+--error 1205
+update t1 set c = 2 where b = 1;
+rollback;
+
+connection con1;
+rollback;
+drop table t1;
+# ordered index
+
+connection con1;
+create table t1(a int primary key, b int, c int, key(b)) engine = ndb;
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+
+begin;
+insert into t1 values (1,1,1);
+
+connection con2;
+begin;
+--error 1205
+update t1 set c = 2 where b = 1;
+rollback;
+
+connection con1;
+rollback;
+drop table t1;
+
+# multiple versions
+
+--echo --con1
+connection con1;
+create table t1(a int primary key, b int, c int, key(b)) engine = ndb;
+insert into t1 values (1,1,1);
+insert into t1 values (2,2,2);
+insert into t1 values (3,3,3);
+insert into t1 values (4,4,4);
+
+begin;
+update t1 set c = 10 where a = 1;
+update t1 set c = 20 where a = 1;
+update t1 set c = 30 where a = 1;
+
+--echo --con1 c=30
+select * from t1 where b >= 1 order by b;
+--echo --con2 c=1
+connection con2;
+select * from t1 where b >= 1 order by b;
+
+--echo --con1
+connection con1;
+delete from t1 where a = 1;
+
+--echo --con1 c=none
+select * from t1 where b >= 1 order by b;
+--echo --con2 c=1
+connection con2;
+select * from t1 where b >= 1 order by b;
+
+--echo --con1
+connection con1;
+commit;
+
+--echo --con1 c=none
+select * from t1 where b >= 1 order by b;
+--echo --con2 c=none
+connection con2;
+select * from t1 where b >= 1 order by b;
+
+--echo --con1
+connection con1;
+begin;
+insert into t1 values (1,1,1);
+update t1 set c = 10 where a = 1;
+update t1 set c = 20 where a = 1;
+update t1 set c = 30 where a = 1;
+
+--echo --con1 c=30
+select * from t1 where b >= 1 order by b;
+--echo --con2 c=none
+connection con2;
+select * from t1 where b >= 1 order by b;
+
+# this fails with "no such table" via con2 ???
+connection con1;
+drop table t1;

storage/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
 };
 

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 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);
 
   int load_diskpage(Signal*, Uint32 opRec, Uint32 fragPtrI, 
 		    Uint32 local_key, Uint32 flags);
@@ -2421,6 +2421,7 @@ private:
 
   void setNullBits(Uint32*, Tablerec* regTabPtr);
   bool checkNullAttributes(KeyReqStruct * const, Tablerec* const);
+  bool find_savepoint(OperationrecPtr& loopOpPtr, Uint32 savepointId);
   bool setup_read(KeyReqStruct* req_struct,
 		  Operationrec* regOperPtr,
 		  Fragrecord* regFragPtr,
@@ -3036,4 +3037,22 @@ Dbtup::get_dd_ptr(PagePtr* pagePtr,
 NdbOut&
 operator<<(NdbOut&, const Dbtup::Tablerec&);
 
+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

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

@@ -319,13 +319,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)
 {
   jamEntry();
   FragrecordPtr fragPtr;
@@ -334,35 +339,73 @@ Dbtup::tuxQueryTh(Uint32 fragPtrI,
   TablerecPtr tablePtr;
   tablePtr.i= fragPtr.p->fragTableId;
   ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
-  // get page
-  Uint32 fragPageId= tupAddr >> MAX_TUPLES_BITS;
-  Uint32 pageIndex= tupAddr & ((1 << MAX_TUPLES_BITS ) - 1);
-  // use temp op rec
-  Operationrec tempOp;
-  KeyReqStruct req_struct;
-  tempOp.m_tuple_location.m_page_no= getRealpid(fragPtr.p, fragPageId);
-  tempOp.m_tuple_location.m_page_idx= pageIndex;
-  tempOp.savepointId= savePointId;
-  tempOp.op_struct.op_type= ZREAD;
-  req_struct.frag_page_id= fragPageId;
-  req_struct.trans_id1= transId1;
-  req_struct.trans_id2= transId2;
-  req_struct.dirty_op= 1;
-  
-  setup_fixed_part(&req_struct, &tempOp, tablePtr.p);
-  if (setup_read(&req_struct, &tempOp, fragPtr.p, tablePtr.p, false)) {
-    /*
-     * 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.
-     */
-    jam();
-    if (req_struct.m_tuple_ptr->get_tuple_version() == tupVersion) {
-      jam();
+  PagePtr pagePtr;
+  pagePtr.i = pageId;
+  ptrCheckGuard(pagePtr, cnoOfPage, page);
+
+  OperationrecPtr currOpPtr;
+  currOpPtr.i = pagePtr.p->pageWord[pageOffset];
+  if (currOpPtr.i == RNIL) {
+    ljam();
+    // 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();
+      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

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

@@ -986,7 +986,8 @@ Dbtux::scanVisible(ScanOpPtr scanPtr, TreeEnt ent)
   const ScanOp& scan = *scanPtr.p;
   const Frag& frag = *c_fragPool.getPtr(scan.m_fragPtrI);
   Uint32 tableFragPtrI = frag.m_tupTableFragPtrI;
-  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)
@@ -996,8 +997,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;
 }