ndb - bug#10029 fix

ndb/include/kernel/signaldata/TuxMaint.hpp:
  handle multipl index update failure atomically
ndb/src/kernel/blocks/dbtup/Dbtup.hpp:
  handle multipl index update failure atomically
ndb/src/kernel/blocks/dbtup/DbtupTrigger.cpp:
  handle multipl index update failure atomically
ndb/src/kernel/blocks/dbtup/Notes.txt:
  handle multipl index update failure atomically
ndb/src/kernel/blocks/dbtux/DbtuxNode.cpp:
  handle multipl index update failure atomically
ndb/src/ndbapi/ndberror.c:
  handle multipl index update failure atomically
ndb/test/ndbapi/testOIBasic.cpp:
  handle multipl index update failure atomically

1 files changed, 119 insertions, 41 deletions

diff --git a/ndb/test/ndbapi/testOIBasic.cpp b/ndb/test/ndbapi/testOIBasic.cpp
index e6d3844d18e..9f8da850ff4 100644
--- a/ndb/test/ndbapi/testOIBasic.cpp
+++ b/ndb/test/ndbapi/testOIBasic.cpp
@@ -228,6 +228,8 @@ struct Par : public Opt {
   bool m_verify;
   // deadlock possible
   bool m_deadlock;
+  // abort percentabge
+  unsigned m_abortpct;
   // timer location
   Par(const Opt& opt) :
     Opt(opt),
@@ -243,7 +245,8 @@ struct Par : public Opt {
     m_pctrange(0),
     m_randomkey(false),
     m_verify(false),
-    m_deadlock(false) {
+    m_deadlock(false),
+    m_abortpct(0) {
   }
 };
 
@@ -684,7 +687,7 @@ struct Con {
   NdbResultSet* m_resultset;
   enum ScanMode { ScanNo = 0, Committed, Latest, Exclusive };
   ScanMode m_scanmode;
-  enum ErrType { ErrNone = 0, ErrDeadlock, ErrOther };
+  enum ErrType { ErrNone = 0, ErrDeadlock, ErrNospace, ErrOther };
   ErrType m_errtype;
   Con() :
     m_ndb(0), m_dic(0), m_tx(0), m_op(0),
@@ -705,7 +708,7 @@ struct Con {
   int setValue(int num, const char* addr);
   int setBound(int num, int type, const void* value);
   int execute(ExecType t);
-  int execute(ExecType t, bool& deadlock);
+  int execute(ExecType t, bool& deadlock, bool& nospace);
   int openScanRead(unsigned scanbat, unsigned scanpar);
   int openScanExclusive(unsigned scanbat, unsigned scanpar);
   int executeScan();
@@ -818,17 +821,21 @@ Con::execute(ExecType t)
 }
 
 int
-Con::execute(ExecType t, bool& deadlock)
+Con::execute(ExecType t, bool& deadlock, bool& nospace)
 {
   int ret = execute(t);
-  if (ret != 0) {
-    if (deadlock && m_errtype == ErrDeadlock) {
-      LL3("caught deadlock");
-      ret = 0;
-    }
+  if (ret != 0 && deadlock && m_errtype == ErrDeadlock) {
+    LL3("caught deadlock");
+    ret = 0;
   } else {
     deadlock = false;
   }
+  if (ret != 0 && nospace && m_errtype == ErrNospace) {
+    LL3("caught nospace");
+    ret = 0;
+  } else {
+    nospace = false;
+  }
   CHK(ret == 0);
   return 0;
 }
@@ -940,6 +947,8 @@ Con::printerror(NdbOut& out)
         die += (code == g_opt.m_die);
         if (code == 266 || code == 274 || code == 296 || code == 297 || code == 499)
           m_errtype = ErrDeadlock;
+        if (code == 826 || code == 827 || code == 902)
+          m_errtype = ErrNospace;
       }
       if (m_op && m_op->getNdbError().code != 0) {
         LL0(++any << " op : error " << m_op->getNdbError());
@@ -1128,6 +1137,16 @@ irandom(unsigned n)
   return i;
 }
 
+static bool
+randompct(unsigned pct)
+{
+  if (pct == 0)
+    return false;
+  if (pct >= 100)
+    return true;
+  return urandom(100) < pct;
+}
+
 // Val - typed column value
 
 struct Val {
@@ -1565,8 +1584,8 @@ struct Set {
   // row methods
   bool exist(unsigned i) const;
   Row::Op pending(unsigned i) const;
-  void notpending(unsigned i);
-  void notpending(const Lst& lst);
+  void notpending(unsigned i, ExecType et = Commit);
+  void notpending(const Lst& lst, ExecType et = Commit);
   void calc(Par par, unsigned i);
   int insrow(Par par, unsigned i);
   int updrow(Par par, unsigned i);
@@ -1775,23 +1794,30 @@ Set::putval(unsigned i, bool force)
 }
 
 void
-Set::notpending(unsigned i)
+Set::notpending(unsigned i, ExecType et)
 {
   assert(m_row[i] != 0);
   Row& row = *m_row[i];
-  if (row.m_pending == Row::InsOp)
-    row.m_exist = true;
-  if (row.m_pending == Row::DelOp)
-    row.m_exist = false;
+  if (et == Commit) {
+    if (row.m_pending == Row::InsOp)
+      row.m_exist = true;
+    if (row.m_pending == Row::DelOp)
+      row.m_exist = false;
+  } else {
+    if (row.m_pending == Row::InsOp)
+      row.m_exist = false;
+    if (row.m_pending == Row::DelOp)
+      row.m_exist = true;
+  }
   row.m_pending = Row::NoOp;
 }
 
 void
-Set::notpending(const Lst& lst)
+Set::notpending(const Lst& lst, ExecType et)
 {
   for (unsigned j = 0; j < lst.m_cnt; j++) {
     unsigned i = lst.m_arr[j];
-    notpending(i);
+    notpending(i, et);
   }
 }
 
@@ -2121,14 +2147,20 @@ pkinsert(Par par)
     lst.push(i);
     if (lst.cnt() == par.m_batch) {
       bool deadlock = par.m_deadlock;
-      CHK(con.execute(Commit, deadlock) == 0);
+      bool nospace = true;
+      ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
+      CHK(con.execute(et, deadlock, nospace) == 0);
       con.closeTransaction();
       if (deadlock) {
         LL1("pkinsert: stop on deadlock");
         return 0;
       }
+      if (nospace) {
+        LL1("pkinsert: cnt=" << j << " stop on nospace");
+        return 0;
+      }
       set.lock();
-      set.notpending(lst);
+      set.notpending(lst, et);
       set.unlock();
       lst.reset();
       CHK(con.startTransaction() == 0);
@@ -2136,14 +2168,20 @@ pkinsert(Par par)
   }
   if (lst.cnt() != 0) {
     bool deadlock = par.m_deadlock;
-    CHK(con.execute(Commit, deadlock) == 0);
+    bool nospace = true;
+    ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
+    CHK(con.execute(et, deadlock, nospace) == 0);
     con.closeTransaction();
     if (deadlock) {
       LL1("pkinsert: stop on deadlock");
       return 0;
     }
+    if (nospace) {
+      LL1("pkinsert: end: stop on nospace");
+      return 0;
+    }
     set.lock();
-    set.notpending(lst);
+    set.notpending(lst, et);
     set.unlock();
     return 0;
   }
@@ -2160,6 +2198,7 @@ pkupdate(Par par)
   CHK(con.startTransaction() == 0);
   Lst lst;
   bool deadlock = false;
+  bool nospace = false;
   for (unsigned j = 0; j < par.m_rows; j++) {
     unsigned j2 = ! par.m_randomkey ? j : urandom(par.m_rows);
     unsigned i = thrrow(par, j2);
@@ -2175,27 +2214,37 @@ pkupdate(Par par)
     lst.push(i);
     if (lst.cnt() == par.m_batch) {
       deadlock = par.m_deadlock;
-      CHK(con.execute(Commit, deadlock) == 0);
+      nospace = true;
+      ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
+      CHK(con.execute(et, deadlock, nospace) == 0);
       if (deadlock) {
         LL1("pkupdate: stop on deadlock");
         break;
       }
+      if (nospace) {
+        LL1("pkupdate: cnt=" << j << " stop on nospace");
+        break;
+      }
       con.closeTransaction();
       set.lock();
-      set.notpending(lst);
+      set.notpending(lst, et);
       set.unlock();
       lst.reset();
       CHK(con.startTransaction() == 0);
     }
   }
-  if (! deadlock && lst.cnt() != 0) {
+  if (! deadlock && ! nospace && lst.cnt() != 0) {
     deadlock = par.m_deadlock;
-    CHK(con.execute(Commit, deadlock) == 0);
+    nospace = true;
+    ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
+    CHK(con.execute(et, deadlock, nospace) == 0);
     if (deadlock) {
       LL1("pkupdate: stop on deadlock");
+    } else if (nospace) {
+      LL1("pkupdate: end: stop on nospace");
     } else {
       set.lock();
-      set.notpending(lst);
+      set.notpending(lst, et);
       set.unlock();
     }
   }
@@ -2212,6 +2261,7 @@ pkdelete(Par par)
   CHK(con.startTransaction() == 0);
   Lst lst;
   bool deadlock = false;
+  bool nospace = false;
   for (unsigned j = 0; j < par.m_rows; j++) {
     unsigned j2 = ! par.m_randomkey ? j : urandom(par.m_rows);
     unsigned i = thrrow(par, j2);
@@ -2226,27 +2276,31 @@ pkdelete(Par par)
     lst.push(i);
     if (lst.cnt() == par.m_batch) {
       deadlock = par.m_deadlock;
-      CHK(con.execute(Commit, deadlock) == 0);
+      nospace = true;
+      ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
+      CHK(con.execute(et, deadlock, nospace) == 0);
       if (deadlock) {
         LL1("pkdelete: stop on deadlock");
         break;
       }
       con.closeTransaction();
       set.lock();
-      set.notpending(lst);
+      set.notpending(lst, et);
       set.unlock();
       lst.reset();
       CHK(con.startTransaction() == 0);
     }
   }
-  if (! deadlock && lst.cnt() != 0) {
+  if (! deadlock && ! nospace && lst.cnt() != 0) {
     deadlock = par.m_deadlock;
-    CHK(con.execute(Commit, deadlock) == 0);
+    nospace = true;
+    ExecType et = randompct(par.m_abortpct) ? Rollback : Commit;
+    CHK(con.execute(et, deadlock, nospace) == 0);
     if (deadlock) {
       LL1("pkdelete: stop on deadlock");
     } else {
       set.lock();
-      set.notpending(lst);
+      set.notpending(lst, et);
       set.unlock();
     }
   }
@@ -2730,6 +2784,10 @@ readverify(Par par)
   if (par.m_noverify)
     return 0;
   par.m_verify = true;
+  if (par.m_abortpct != 0) {
+    LL2("skip verify in this version"); // implement in 5.0 version
+    par.m_verify = false;
+  }
   CHK(pkread(par) == 0);
   CHK(scanreadall(par) == 0);
   return 0;
@@ -3028,11 +3086,11 @@ runstep(Par par, const char* fname, TFunc func, unsigned mode)
   for (n = 0; n < threads; n++) {
     LL4("start " << n);
     Thr& thr = *g_thrlist[n];
-    thr.m_par.m_tab = par.m_tab;
-    thr.m_par.m_set = par.m_set;
-    thr.m_par.m_tmr = par.m_tmr;
-    thr.m_par.m_lno = par.m_lno;
-    thr.m_par.m_slno = par.m_slno;
+    Par oldpar = thr.m_par;
+    // update parameters
+    thr.m_par = par;
+    thr.m_par.m_no = oldpar.m_no;
+    thr.m_par.m_con = oldpar.m_con;
     thr.m_func = func;
     thr.start();
   }
@@ -3144,6 +3202,24 @@ tbusybuild(Par par)
 }
 
 static int
+trollback(Par par)
+{
+  par.m_abortpct = 50;
+  RUNSTEP(par, droptable, ST);
+  RUNSTEP(par, createtable, ST);
+  RUNSTEP(par, invalidatetable, MT);
+  RUNSTEP(par, pkinsert, MT);
+  RUNSTEP(par, createindex, ST);
+  RUNSTEP(par, invalidateindex, MT);
+  RUNSTEP(par, readverify, ST);
+  for (par.m_slno = 0; par.m_slno < par.m_subloop; par.m_slno++) {
+    RUNSTEP(par, mixedoperations, MT);
+    RUNSTEP(par, readverify, ST);
+  }
+  return 0;
+}
+
+static int
 ttimebuild(Par par)
 {
   Tmr t1;
@@ -3252,10 +3328,12 @@ struct TCase {
 static const TCase
 tcaselist[] = {
   TCase("a", tbuild, "index build"),
-  TCase("b", tpkops, "pk operations"),
-  TCase("c", tpkopsread, "pk operations and scan reads"),
-  TCase("d", tmixedops, "pk operations and scan operations"),
-  TCase("e", tbusybuild, "pk operations and index build"),
+  // "b" in 5.0
+  TCase("c", tpkops, "pk operations"),
+  TCase("d", tpkopsread, "pk operations and scan reads"),
+  TCase("e", tmixedops, "pk operations and scan operations"),
+  TCase("f", tbusybuild, "pk operations and index build"),
+  TCase("g", trollback, "operations with random rollbacks"),
   TCase("t", ttimebuild, "time index build"),
   TCase("u", ttimemaint, "time index maintenance"),
   TCase("v", ttimescan, "time full scan table vs index on pk"),