Commit fc351758 authored by pekka@sama.ndb.mysql.com's avatar pekka@sama.ndb.mysql.com
Browse files

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

into  sama.ndb.mysql.com:/export/space/pekka/ndb/version/my51-bug31477
parents 1715bb18 7bf78e09

mysql-test/r/ndb_bug31477.result

0 → 100644
+98 −0
Original line number Diff line number Diff line 
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

0 → 100644
+109 −0
Original line number Diff line number Diff line 
--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

+1 −1
Original line number Diff line number Diff line
@@ -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

+20 −1
Original line number Diff line number Diff line
@@ -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

+76 −33
Original line number Diff line number Diff line
@@ -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