ndb - wl#2972 (5.0) fix detached trigger opType + new pgm test_event_merge

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

@@ -445,6 +445,7 @@ Dbtup::commitRecord(Signal* signal,
     befOpPtr.p->changeMask.bitOR(attributeMask);
     befOpPtr.p->gci = regOperPtr->gci;
     
+    befOpPtr.p->optype = opType;
     operPtr.p = befOpPtr.p;
     checkDetachedTriggers(signal,
                           befOpPtr.p,
@@ -483,6 +484,7 @@ Dbtup::commitRecord(Signal* signal,
     befOpPtr.p->pageIndex  = befOpPtr.p->pageIndexC;
     befOpPtr.p->gci = regOperPtr->gci;
 
+    befOpPtr.p->optype = opType;
     operPtr.p = befOpPtr.p;
     checkDetachedTriggers(signal,
                           befOpPtr.p,

ndb/src/ndbapi/NdbEventOperationImpl.cpp

@@ -250,10 +250,10 @@ NdbEventOperationImpl::execute()
   int hasSubscriber;
   int r= m_bufferHandle->prepareAddSubscribeEvent(this,
 						  hasSubscriber /*return value*/);
-  m_error.code= 4709;
 
   if (r < 0)
   {
+    m_error.code= 4709;
     DBUG_RETURN(-1);
   }
 

ndb/test/ndbapi/Makefile.am

@@ -35,7 +35,8 @@ testPartitioning \
 testBitfield \
 DbCreate DbAsyncGenerator \
 test_event_multi_table \
-testSRBank
+testSRBank \
+test_event_merge
 
 #flexTimedAsynch
 #testBlobs
@@ -80,6 +81,7 @@ DbCreate_SOURCES = bench/mainPopulate.cpp bench/dbPopulate.cpp bench/userInterfa
 DbAsyncGenerator_SOURCES = bench/mainAsyncGenerator.cpp bench/asyncGenerator.cpp bench/ndb_async2.cpp bench/dbGenerator.h bench/macros.h bench/userInterface.h bench/testData.h bench/testDefinitions.h bench/ndb_schema.hpp bench/ndb_error.hpp
 test_event_multi_table_SOURCES = test_event_multi_table.cpp
 testSRBank_SOURCES = testSRBank.cpp
+test_event_merge_SOURCES = test_event_merge.cpp
 
 INCLUDES_LOC = -I$(top_srcdir)/ndb/include/kernel
 
@@ -160,3 +162,4 @@ testScan.dsp: Makefile \
 	@$(top_srcdir)/ndb/config/win-includes $@ $(INCLUDES)
 	@$(top_srcdir)/ndb/config/win-sources $@ $(testScan_SOURCES)
 	@$(top_srcdir)/ndb/config/win-libraries $@ LINK $(LDADD)
+

ndb/test/ndbapi/test_event_merge.cpp

+/* Copyright (C) 2005 MySQL AB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 2 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+#include <ndb_global.h>
+#include <ndb_opts.h>
+#include <NdbApi.hpp>
+#include <NdbTest.hpp>
+#include <my_sys.h>
+#include <ndb_version.h>
+
+#if NDB_VERSION_D < MAKE_VERSION(5, 1, 0)
+#define version50
+#else
+#undef version50
+#endif
+
+#if !defined(min) || !defined(max)
+#define min(x, y) ((x) < (y) ? (x) : (y))
+#define max(x, y) ((x) > (y) ? (x) : (y))
+#endif
+
+/*
+ * Test composite operations on same PK via events.  The merge of event
+ * data can happen in 2 places:
+ *
+ * 1) In TUP at commit, the detached triggers report a single composite
+ * operation and its post/pre data
+ *
+ * 2) In event API version >= 5.1 separate commits within same GCI are
+ * by default merged.  This is required to read blob data via NdbBlob.
+ *
+ * This test program ignores Blob columns in version 5.0.
+ *
+ * There are 5 ways (ignoring NUL operand) to compose 2 ops:
+ *                      5.0 bugs        5.1 bugs
+ * INS o DEL = NUL
+ * INS o UPD = INS                       5.1
+ * DEL o INS = UPD      type=INS         5.1
+ * UPD o DEL = DEL      no event
+ * UPD o UPD = UPD
+ */
+
+struct Opts {
+  my_bool abort_on_error;
+  int loglevel;
+  uint loop;
+  uint maxops;
+  uint maxpk;
+  const char* opstr;
+  uint seed;
+  my_bool separate_events;
+  my_bool use_table;
+};
+
+static Opts g_opts;
+static const uint g_maxops = 10000;
+static const uint g_maxpk = 100;
+
+static Ndb_cluster_connection* g_ncc = 0;
+static Ndb* g_ndb = 0;
+static NdbDictionary::Dictionary* g_dic = 0;
+static NdbTransaction* g_con = 0;
+static NdbOperation* g_op = 0;
+
+static const char* g_tabname = "tem1";
+static const char* g_evtname = "tem1ev1";
+static const uint g_charlen = 5;
+static const char* g_csname = "latin1_swedish_ci";
+
+static const NdbDictionary::Table* g_tab = 0;
+static const NdbDictionary::Event* g_evt = 0;
+
+static NdbEventOperation* g_evt_op = 0;
+
+static uint
+urandom(uint n)
+{
+  uint r = (uint)random();
+  if (n != 0)
+    r = r % n;
+  return r;
+}
+
+static int& g_loglevel = g_opts.loglevel; // default log level
+
+#define chkdb(x) \
+  do { if (x) break; ndbout << "line " << __LINE__ << " FAIL " << #x << endl; errdb(); if (g_opts.abort_on_error) abort(); return -1; } while (0)
+
+#define chkrc(x) \
+  do { if (x) break; ndbout << "line " << __LINE__ << " FAIL " << #x << endl; if (g_opts.abort_on_error) abort(); return -1; } while (0)
+
+#define reqrc(x) \
+  do { if (x) break; ndbout << "line " << __LINE__ << " ASSERT " << #x << endl; abort(); } while (0)
+
+#define ll0(x) \
+  do { if (g_loglevel < 0) break; ndbout << x << endl; } while (0)
+
+#define ll1(x) \
+  do { if (g_loglevel < 1) break; ndbout << x << endl; } while (0)
+
+#define ll2(x) \
+  do { if (g_loglevel < 2) break; ndbout << x << endl; } while (0)
+
+static void
+errdb()
+{
+  uint any = 0;
+  if (g_ndb != 0) {
+    const NdbError& e = g_ndb->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " ndb: error " << e);
+  }
+  if (g_dic != 0) {
+    const NdbError& e = g_dic->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " dic: error " << e);
+  }
+  if (g_con != 0) {
+    const NdbError& e = g_con->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " con: error " << e);
+  }
+  if (g_op != 0) {
+    const NdbError& e = g_op->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " op: error " << e);
+  }
+  if (g_evt_op != 0) {
+    const NdbError& e = g_evt_op->getNdbError();
+    if (e.code != 0)
+      ll0(++any << " evt_op: error " << e);
+  }
+  if (! any)
+    ll0("unknown db error");
+}
+
+struct Col {
+  uint no;
+  const char* name;
+  NdbDictionary::Column::Type type;
+  bool pk;
+  bool nullable;
+  uint length;
+  uint size;
+};
+
+static Col g_col[] = {
+  { 0, "pk1", NdbDictionary::Column::Unsigned, true, false, 1, 4 },
+  { 1, "pk2", NdbDictionary::Column::Char, true, false,  g_charlen, g_charlen },
+  { 2, "seq", NdbDictionary::Column::Unsigned,  false, false, 1, 4 },
+  { 3, "cc1", NdbDictionary::Column::Char, false, true, g_charlen, g_charlen }
+};
+
+static const uint g_ncol = sizeof(g_col)/sizeof(g_col[0]);
+
+static const Col&
+getcol(uint i)
+{
+  if (i < g_ncol)
+    return g_col[i];
+  assert(false);
+  return g_col[g_ncol];
+}
+
+static const Col&
+getcol(const char* name)
+{
+  uint i;
+  for (i = 0; i < g_ncol; i++)
+    if (strcmp(g_col[i].name, name) == 0)
+      break;
+  return getcol(i);
+}
+
+static int
+createtable()
+{
+  g_tab = 0;
+  NdbDictionary::Table tab(g_tabname);
+  tab.setLogging(false);
+  CHARSET_INFO* cs;
+  chkrc((cs = get_charset_by_name(g_csname, MYF(0))) != 0);
+  uint i;
+  for (i = 0; i < g_ncol; i++) {
+    const Col& c = g_col[i];
+    NdbDictionary::Column col(c.name);
+    col.setType(c.type);
+    col.setPrimaryKey(c.pk);
+    if (! c.pk)
+      col.setNullable(true);
+    col.setLength(c.length);
+    switch (c.type) {
+    case NdbDictionary::Column::Unsigned:
+      break;
+    case NdbDictionary::Column::Char:
+      col.setLength(c.length);
+      col.setCharset(cs);
+      break;
+    default:
+      assert(false);
+      break;
+    }
+    tab.addColumn(col);
+  }
+  g_dic = g_ndb->getDictionary();
+  if (! g_opts.use_table) {
+    if (g_dic->getTable(g_tabname) != 0)
+      chkdb(g_dic->dropTable(g_tabname) == 0);
+    chkdb(g_dic->createTable(tab) == 0);
+  }
+  chkdb((g_tab = g_dic->getTable(g_tabname)) != 0);
+  g_dic = 0;
+  if (! g_opts.use_table) {
+    // extra row for GCI probe
+    chkdb((g_con = g_ndb->startTransaction()) != 0);
+    chkdb((g_op = g_con->getNdbOperation(g_tabname)) != 0);
+    chkdb(g_op->insertTuple() == 0);
+    Uint32 pk1;
+    char pk2[g_charlen];
+    pk1 = g_maxpk;
+    memset(pk2, 0x20, g_charlen);
+    chkdb(g_op->equal("pk1", (char*)&pk1) == 0);
+    chkdb(g_op->equal("pk2", (char*)&pk2[0]) == 0);
+    chkdb(g_con->execute(Commit) == 0);
+    g_ndb->closeTransaction(g_con);
+    g_op = 0;
+    g_con = 0;
+  }
+  return 0;
+}
+
+static int
+droptable()
+{
+  if (! g_opts.use_table) {
+    g_dic = g_ndb->getDictionary();
+    chkdb(g_dic->dropTable(g_tab->getName()) == 0);
+    g_tab = 0;
+    g_dic = 0;
+  }
+  return 0;
+}
+
+struct Data {
+  Uint32 pk1;
+  char pk2[g_charlen];
+  Uint32 seq;
+  char cc1[g_charlen];
+  void* ptr[g_ncol];
+  int ind[g_ncol]; // -1 = no data, 1 = NULL, 0 = not NULL
+  void init() {
+    uint i;
+    pk1 = 0;
+    memset(pk2, 0, sizeof(pk2));
+    seq = 0;
+    memset(cc1, 0, sizeof(cc1));
+    ptr[0] = &pk1;
+    ptr[1] = pk2;
+    ptr[2] = &seq;
+    ptr[3] = cc1;
+    for (i = 0; i < g_ncol; i++)
+      ind[i] = -1;
+  }
+};
+
+static NdbOut&
+operator<<(NdbOut& out, const Data& d)
+{
+  uint i;
+  for (i = 0; i < g_ncol; i++) {
+    const Col& c = getcol(i);
+    out << (i == 0 ? "" : " ") << c.name << "=";
+    if (d.ind[i] == -1)
+      continue;
+    if (d.ind[i] == 1) {
+      out << "NULL";
+      continue;
+    }
+    switch (c.type) {
+    case NdbDictionary::Column::Unsigned:
+      out << *(Uint32*)d.ptr[i];
+      break;
+    case NdbDictionary::Column::Char:
+      {
+        char buf[g_charlen + 1];
+        memcpy(buf, d.ptr[i], g_charlen);
+        uint n = g_charlen;
+        while (1) {
+          buf[n] = 0;
+          if (n == 0 || buf[n - 1] != 0x20)
+            break;
+          n--;
+        }
+        out << buf;
+      }
+      break;
+    default:
+      out << "?";
+      break;
+    }
+  }
+  return out;
+}
+
+static const uint g_optypes = 3; // real ops 0-2
+
+/*
+ * Represents single or composite operation or received event.  The
+ * post/pre data is either computed here for operations or received from
+ * the event.
+ */
+struct Op { // single or composite
+  enum Kind { OP = 1, EV = 2 };
+  enum Type { NUL = -1, INS, DEL, UPD };
+  Kind kind;
+  Type type;
+  Op* next_op; // within one commit
+  Op* next_com; // next commit chain or next event
+  uint num_op;
+  uint num_com;
+  Data data[2]; // 0-post 1-pre
+  bool match; // matched to event
+  void init() {
+    assert(kind == OP || kind == EV);
+    type = NUL;
+    next_op = next_com = 0;
+    num_op = num_com = 0;
+    data[0].init();
+    data[1].init();
+    match = false;
+  }
+};
+
+static NdbOut&
+operator<<(NdbOut& out, Op::Type t)
+{
+  switch (t) {
+  case Op::NUL:
+    out << "NUL";
+    break;
+  case Op::INS:
+    out << "INS";
+    break;
+  case Op::DEL:
+    out << "DEL";
+    break;
+  case Op::UPD:
+    out << "UPD";
+    break;
+  default:
+    out << (int)t;
+    break;
+  }
+  return out;
+}
+
+static NdbOut&
+operator<<(NdbOut& out, const Op& op)
+{
+  out << "t=" << op.type;
+  out << " " << op.data[0];
+  out << " [" << op.data[1] << "]";
+  return out;
+}
+
+static int
+seteventtype(Op* ev, NdbDictionary::Event::TableEvent te)
+{
+  Op::Type t = Op::NUL;
+  switch (te) {
+  case NdbDictionary::Event::TE_INSERT:
+    t = Op::INS;
+    break;
+  case NdbDictionary::Event::TE_DELETE:
+    t = Op::DEL;
+    break;
+  case NdbDictionary::Event::TE_UPDATE:
+    t = Op::UPD;
+    break;
+  default:
+    ll0("EVT: " << *ev << ": bad event type" << (int)te);
+    return -1;
+  }
+  ev->type = t;
+  return 0;
+}
+
+static uint g_usedops = 0;
+static uint g_usedevs = 0;
+static Op g_oplist[g_maxops];
+static Op g_evlist[g_maxops];
+static uint g_maxcom = 8; // max ops per commit
+
+static Op* g_pk_op[g_maxpk];
+static Op* g_pk_ev[g_maxpk];
+static uint g_seq = 0;
+
+static NdbRecAttr* g_ra[2][g_ncol]; // 0-post 1-pre
+static Op* g_rec_ev;
+static uint g_ev_cnt[g_maxpk];
+
+static uint
+getfreeops()
+{
+  assert(g_opts.maxops >= g_usedops);
+  return g_opts.maxops - g_usedops;
+}
+
+static uint
+getfreeevs()
+{
+  assert(g_opts.maxops >= g_usedevs);
+  return g_opts.maxops - g_usedevs;
+}
+
+static Op*
+getop()
+{
+  if (g_usedops < g_opts.maxops) {
+    Op* op = &g_oplist[g_usedops++];
+    op->kind = Op::OP;
+    op->init();
+    return op;
+  }
+  assert(false);
+  return 0;
+}
+
+static Op*
+getev()
+{
+  if (g_usedevs < g_opts.maxops) {
+    Op* ev = &g_evlist[g_usedevs++];
+    ev->kind = Op::EV;
+    ev->init();
+    return ev;
+  }
+  assert(false);
+  return 0;
+}
+
+static void
+resetmem()
+{
+  int i, j;
+  for (j = 0; j < 2; j++)
+    for (i = 0; i < g_ncol; i++)
+      g_ra[j][i] = 0;
+  g_rec_ev = 0;
+  for (i = 0; i < g_opts.maxpk; i++)
+    g_pk_op[i] = 0;
+  for (i = 0; i < g_opts.maxpk; i++)
+    g_ev_cnt[i] = 0;
+  g_seq = 0;
+  g_usedops = 0;
+  g_usedevs = 0;
+}
+
+struct Comp {
+  Op::Type t1, t2, t3;
+};
+
+static Comp
+g_comp[] = {
+  { Op::INS, Op::DEL, Op::NUL },
+  { Op::INS, Op::UPD, Op::INS },
+  { Op::DEL, Op::INS, Op::UPD },
+  { Op::UPD, Op::DEL, Op::DEL },
+  { Op::UPD, Op::UPD, Op::UPD }
+};
+
+static const uint g_ncomp = sizeof(g_comp)/sizeof(g_comp[0]);
+
+static int
+checkop(const Op* op, Uint32& pk1)
+{
+  const Data (&d)[2] = op->data;
+  Op::Type t = op->type;
+  chkrc(t == Op::NUL || t == Op::INS || t == Op::DEL || t == Op::UPD);
+  { const Col& c = getcol("pk1");
+    chkrc(d[0].ind[c.no] == 0);
+    pk1 = d[0].pk1;
+    chkrc(pk1 < g_opts.maxpk);
+  }
+  uint i;
+  for (i = 0; i < g_ncol; i++) {
+    const Col& c = getcol(i);
+    if (t != Op::NUL) {
+      if (c.pk) {
+        chkrc(d[0].ind[i] == 0); // even DEL has PK in post data
+        if (t == Op::INS) {
+          chkrc(d[1].ind[i] == -1);
+        } else if (t == Op::DEL) {
+#ifdef ndb_event_cares_about_pk_pre_data
+          chkrc(d[1].ind[i] == -1);
+#endif
+        } else {
+#ifdef ndb_event_cares_about_pk_pre_data
+          chkrc(d[1].ind[i] == 0);
+#endif
+        }
+      } else {
+        if (t == Op::INS) {
+          chkrc(d[0].ind[i] >= 0);
+          chkrc(d[1].ind[i] == -1);
+        } else if (t == Op::DEL) {
+          chkrc(d[0].ind[i] == -1);
+          chkrc(d[1].ind[i] >= 0);
+        } else if (op->kind == Op::OP) {
+          chkrc(d[0].ind[i] >= 0);
+          chkrc(d[1].ind[i] >= 0);
+        }
+      }
+    }
+  }
+  return 0;
+}
+
+static Comp*
+comptype(Op::Type t1, Op::Type t2) // only non-NUL
+{
+  uint i;
+  for (i = 0; i < g_ncomp; i++)
+    if (g_comp[i].t1 == t1 && g_comp[i].t2 == t2)
+      return &g_comp[i];
+  return 0;
+}
+
+static void
+copycol(const Col& c, const Data& d1, Data& d3)
+{
+  if ((d3.ind[c.no] = d1.ind[c.no]) != -1)
+    memmove(d3.ptr[c.no], d1.ptr[c.no], c.size);
+}
+
+static void
+copykeys(const Data& d1, Data& d3)
+{
+  uint i;
+  for (i = 0; i < g_ncol; i++) {
+    const Col& c = g_col[i];
+    if (c.pk)
+      copycol(c, d1, d3);
+  }
+}
+
+static void
+copydata(const Data& d1, Data& d3)
+{
+  uint i;
+  for (i = 0; i < g_ncol; i++) {
+    const Col& c = g_col[i];
+    copycol(c, d1, d3);
+  }
+}
+
+static void
+copyop(const Op* op1, Op* op3)
+{
+  op3->type = op1->type;
+  copydata(op1->data[0], op3->data[0]);
+  copydata(op1->data[1], op3->data[1]);
+  Uint32 pk1_tmp;
+  reqrc(checkop(op3, pk1_tmp) == 0);
+}
+
+// not needed for ops
+static void
+compdata(const Data& d1, const Data& d2, Data& d3) // d2 overrides d1
+{
+  uint i;
+  for (i = 0; i < g_ncol; i++) {
+    const Col& c = g_col[i];
+    const Data* d = 0;
+    if (d1.ind[i] == -1 && d2.ind[i] == -1)
+      d3.ind[i] = -1;
+    else if (d1.ind[i] == -1 && d2.ind[i] != -1)
+      d = &d2;
+    else if (d1.ind[i] != -1 && d2.ind[i] == -1)
+      d = &d1;
+    else
+      d = &d2;
+    if (d != 0)
+      copycol(c, *d, d3);
+  }
+}
+
+static int
+compop(const Op* op1, const Op* op2, Op* op3) // op1 o op2 = op3
+{
+  Comp* comp;
+  if (op2->type == Op::NUL) {
+    copyop(op1, op3);
+    return 0;
+  }
+  if (op1->type == Op::NUL) {
+    copyop(op2, op3);
+    return 0;
+  }
+  chkrc((comp = comptype(op1->type, op2->type)) != 0);
+  op3->type = comp->t3;
+  copykeys(op2->data[0], op3->data[0]);
+  if (op3->type != Op::DEL)
+    copydata(op2->data[0], op3->data[0]);
+  if (op3->type != Op::INS)
+    copydata(op1->data[1], op3->data[1]);
+  Uint32 pk1_tmp;
+  reqrc(checkop(op3, pk1_tmp) == 0);
+  // not eliminating identical post-pre fields
+  return 0;
+}
+
+static int
+createevent()
+{
+  ll1("createevent");
+  g_evt = 0;
+  g_dic = g_ndb->getDictionary();
+  NdbDictionary::Event evt(g_evtname);
+  evt.setTable(*g_tab);
+  evt.addTableEvent(NdbDictionary::Event::TE_ALL);
+  // pk always
+  evt.addEventColumn("pk1");
+  evt.addEventColumn("pk2");
+  // simple cols
+  evt.addEventColumn("seq");
+  evt.addEventColumn("cc1");
+  if (g_dic->getEvent(evt.getName()) != 0)
+    chkdb(g_dic->dropEvent(evt.getName()) == 0);
+  chkdb(g_dic->createEvent(evt) == 0);
+  chkdb((g_evt = g_dic->getEvent(evt.getName())) != 0);
+  g_dic = 0;
+  return 0;
+}
+
+static int
+dropevent()
+{
+  ll1("dropevent");
+  g_dic = g_ndb->getDictionary();
+  chkdb(g_dic->dropEvent(g_evt->getName()) == 0);
+  g_evt = 0;
+  g_dic = 0;
+  return 0;
+}
+
+static int
+createeventop()
+{
+  ll1("createeventop");
+#ifdef version50
+  uint bsz = 10 * g_opts.maxops;
+  chkdb((g_evt_op = g_ndb->createEventOperation(g_evt->getName(), bsz)) != 0);
+#else
+  chkdb((g_evt_op = g_ndb->createEventOperation(g_evt->getName())) != 0);
+#endif
+  uint i;
+  for (i = 0; i < g_ncol; i++) {
+    const Col& c = g_col[i];
+    Data (&d)[2] = g_rec_ev->data;
+    switch (c.type) {
+    case NdbDictionary::Column::Unsigned:
+    case NdbDictionary::Column::Char:
+      chkdb((g_ra[0][i] = g_evt_op->getValue(c.name, (char*)d[0].ptr[i])) != 0);
+      chkdb((g_ra[1][i] = g_evt_op->getPreValue(c.name, (char*)d[1].ptr[i])) != 0);
+      break;
+    default:
+      assert(false);
+      break;
+    }
+  }
+  return 0;
+}
+
+static int
+dropeventop()
+{
+  ll1("dropeventop");
+  chkdb(g_ndb->dropEventOperation(g_evt_op) == 0);
+  g_evt_op = 0;
+  return 0;
+}
+
+static int
+waitgci() // wait for event to be installed and for at least 1 GCI to pass
+{
+  const uint ngci = 3;
+  ll1("waitgci " << ngci);
+  Uint32 gci[2];
+  uint i = 0;
+  while (1) {
+    chkdb((g_con = g_ndb->startTransaction()) != 0);
+    { // forced to exec a dummy op
+      Uint32 pk1;
+      char pk2[g_charlen];
+      pk1 = g_maxpk;
+      memset(pk2, 0x20, g_charlen);
+      chkdb((g_op = g_con->getNdbOperation(g_tabname)) != 0);
+      chkdb(g_op->readTuple() == 0);
+      chkdb(g_op->equal("pk1", (char*)&pk1) == 0);
+      chkdb(g_op->equal("pk2", (char*)&pk2[0]) == 0);
+      chkdb(g_con->execute(Commit) == 0);
+      g_op = 0;
+    }
+    gci[i] = g_con->getGCI();
+    g_ndb->closeTransaction(g_con);
+    g_con = 0;
+    if (i == 1 && gci[0] + ngci <= gci[1]) {
+      ll1("waitgci: " << gci[0] << " " << gci[1]);
+      break;
+    }
+    i = 1;
+  }
+  return 0;
+}
+
+static int
+makeop(Op* op, Uint32 pk1, Op::Type t, const Op* prev_op)
+{
+  op->type = t;
+  if (t != Op::INS)
+    copydata(prev_op->data[0], op->data[1]);
+  uint i;
+  for (i = 0; i < g_ncol; i++) {
+    const Col& c = getcol(i);
+    Data (&d)[2] = op->data;
+    if (i == getcol("pk1").no) {
+      d[0].pk1 = pk1;
+      d[0].ind[i] = 0;
+      continue;
+    }
+    if (i == getcol("pk2").no) {
+      sprintf(d[0].pk2, "%-*u", g_charlen, d[0].pk1);
+      d[0].ind[i] = 0;
+      continue;
+    }
+    if (t == Op::DEL) {
+      d[0].ind[i] = -1;
+      continue;
+    }
+    if (i == getcol("seq").no) {
+      d[0].seq = g_seq++;
+      d[0].ind[i] = 0;
+      continue;
+    }
+    uint u;
+    u = urandom(100);
+    if (c.nullable && u < 20) {
+      d[0].ind[i] = 1;
+      continue;
+    }
+    switch (c.type) {
+    case NdbDictionary::Column::Unsigned:
+      {
+        u = urandom(0);
+        Uint32* p = (Uint32*)d[0].ptr[i];
+        *p = u;
+      }
+      break;
+    case NdbDictionary::Column::Char:
+      {
+        u = urandom(g_charlen);
+        char* p = (char*)d[0].ptr[i];
+        uint j;
+        for (j = 0; j < g_charlen; j++) {
+          uint v = urandom(3);
+          p[j] = j < u ? "abcde"[v] : 0x20;
+        }
+      }
+      break;
+    default:
+      assert(false);
+      break;
+    }
+    d[0].ind[i] = 0;
+  }
+  Uint32 pk1_tmp = ~(Uint32)0;
+  chkrc(checkop(op, pk1_tmp) == 0);
+  reqrc(pk1 == pk1_tmp);
+  return 0;
+}
+
+static void
+makeop(Op* tot_op, Op* com_op, Uint32 pk1, Op::Type t)
+{
+  Op tmp_op;
+  tmp_op.kind = Op::OP;
+  Op* op = getop();
+  reqrc(makeop(op, pk1, t, tot_op) == 0);
+  // add to end
+  Op* last_op = com_op;
+  while (last_op->next_op != 0)
+    last_op = last_op->next_op;
+  last_op->next_op = op;
+  // merge into chain head
+  tmp_op.init();
+  reqrc(compop(com_op, op, &tmp_op) == 0);
+  copyop(&tmp_op, com_op);
+  // merge into total op
+  tmp_op.init();
+  reqrc(compop(tot_op, op, &tmp_op) == 0);
+  copyop(&tmp_op, tot_op);
+  // counts
+  com_op->num_op += 1;
+  tot_op->num_op += 1;
+}
+
+static void
+makeops()
+{
+  ll1("makeops");
+  uint resv = g_opts.opstr == 0 ? 2 * g_opts.maxpk : 0; // for final deletes
+  uint next = g_opts.opstr == 0 ? g_maxcom : strlen(g_opts.opstr);
+  Op tmp_op;
+  tmp_op.kind = Op::OP;
+  Uint32 pk1 = 0;
+  while (getfreeops() >= resv + 2 + next && pk1 < g_opts.maxpk) {
+    if (g_opts.opstr == 0)
+      pk1 = urandom(g_opts.maxpk);
+    ll2("makeops: pk1=" << pk1 << " free=" << getfreeops());
+    // total op on the pk so far
+    // optype either NUL=initial/deleted or INS=created
+    Op* tot_op = g_pk_op[pk1];
+    if (tot_op == 0)
+      tot_op = g_pk_op[pk1] = getop(); //1
+    assert(tot_op->type == Op::NUL || tot_op->type == Op::INS);
+    // add new commit chain to end
+    Op* last_com = tot_op;
+    while (last_com->next_com != 0)
+      last_com = last_com->next_com;
+    Op* com_op = getop(); //2
+    last_com->next_com = com_op;
+    // length of random chain
+    uint len = ~0;
+    if (g_opts.opstr == 0)
+      len = 1 + urandom(g_maxcom - 1);
+    ll2("makeops: com chain");
+    uint n = 0;
+    while (1) {
+      // random or from g_opts.opstr
+      Op::Type t;
+      if (g_opts.opstr == 0) {
+        if (n == len)
+          break;
+        do {
+          t = (Op::Type)urandom(g_optypes);
+        } while (tot_op->type == Op::NUL && (t == Op::DEL || t == Op::UPD) ||
+                 tot_op->type == Op::INS && t == Op::INS);
+      } else {
+        uint m = strlen(g_opts.opstr);
+        uint k = tot_op->num_com + tot_op->num_op;
+        assert(k < m);
+        char c = g_opts.opstr[k];
+        if (c == 'c') {
+          if (k + 1 == m)
+            pk1 += 1;
+          break;
+        }
+        const char* p = "idu";
+        const char* q = strchr(p, c);
+        assert(q != 0);
+        t = (Op::Type)(q - p);
+      }
+      makeop(tot_op, com_op, pk1, t);
+      assert(tot_op->type == Op::NUL || tot_op->type == Op::INS);
+      n++;
+    }
+    tot_op->num_com += 1;
+  }
+  assert(getfreeops() >= resv);
+  // terminate with DEL if necessary
+  for (pk1 = 0; pk1 < g_opts.maxpk; pk1++) {
+    Op* tot_op = g_pk_op[pk1];
+    if (tot_op == 0)
+      continue;
+    if (tot_op->type == Op::NUL)
+      continue;
+    assert(g_opts.opstr == 0);
+    Op* last_com = tot_op;
+    while (last_com->next_com != 0)
+      last_com = last_com->next_com;
+    Op* com_op = getop(); //1
+    last_com->next_com = com_op;
+    makeop(tot_op, com_op, pk1, Op::DEL);
+    assert(tot_op->type == Op::NUL);
+    tot_op->num_com += 1;
+  }
+}
+
+static int
+addndbop(Op* op)
+{
+  chkdb((g_op = g_con->getNdbOperation(g_tabname)) != 0);
+  switch (op->type) {
+  case Op::INS:
+    chkdb(g_op->insertTuple() == 0);
+    break;
+  case Op::DEL:
+    chkdb(g_op->deleteTuple() == 0);
+    break;
+  case Op::UPD:
+    chkdb(g_op->updateTuple() == 0);
+    break;
+  default:
+    assert(false);
+    break;
+  }
+  uint i;
+  for (i = 0; i < g_ncol; i++) {
+    const Col& c = getcol(i);
+    const Data& d = op->data[0];
+    if (! c.pk)
+      continue;
+    chkdb(g_op->equal(c.name, (char*)d.ptr[i]) == 0);
+  }
+  if (op->type != Op::DEL) {
+    for (i = 0; i < g_ncol; i++) {
+      const Col& c = getcol(i);
+      const Data& d = op->data[0];
+      if (c.pk)
+        continue;
+      if (d.ind[i] == -1)
+        continue;
+      const char* ptr = d.ind[i] == 0 ? (char*)d.ptr[i] : 0;
+      chkdb(g_op->setValue(c.name, ptr) == 0);
+    }
+  }
+  g_op = 0;
+  return 0;
+}
+
+static int
+runops()
+{
+  ll1("runops");
+  Uint32 pk1;
+  const Op* com_op[g_maxpk];
+  uint left = 0;
+  for (pk1 = 0; pk1 < g_opts.maxpk; pk1++) {
+    com_op[pk1] = 0;
+    // total op on the pk
+    Op* tot_op = g_pk_op[pk1];
+    if (tot_op == 0)
+      continue;
+    // first commit chain
+    assert(tot_op->next_com != 0);
+    com_op[pk1] = tot_op->next_com;
+    left++;
+  }
+  while (left != 0) {
+    pk1 = urandom(g_opts.maxpk);
+    if (com_op[pk1] == 0)
+      continue;
+    // do the ops in one transaction
+    ll2("runops: pk1=" << pk1);
+    chkdb((g_con = g_ndb->startTransaction()) != 0);
+    // first op in chain
+    Op* op = com_op[pk1]->next_op;
+    assert(op != 0);
+    while (op != 0) {
+      ll2("add op:" << *op);
+      chkrc(addndbop(op) == 0);
+      op = op->next_op;
+    }
+    chkdb(g_con->execute(Commit) == 0);
+    g_ndb->closeTransaction(g_con);
+    g_con = 0;
+    // next chain
+    com_op[pk1] = com_op[pk1]->next_com;
+    if (com_op[pk1] == 0) {
+      assert(left != 0);
+      left--;
+    }
+  }
+  assert(left == 0);
+  return 0;
+}
+
+static int
+matchevent(Op* ev)
+{
+  Op::Type t = ev->type;
+  Data (&d)[2] = ev->data;
+  // get PK
+  Uint32 pk1 = d[0].pk1;
+  chkrc(pk1 < g_opts.maxpk);
+  // on error repeat and print details
+  uint loop = 0;
+  while (loop <= 1) {
+    uint g_loglevel = loop == 0 ? g_opts.loglevel : 2;
+    ll1("matchevent: pk1=" << pk1 << " type=" << t);
+    ll2("EVT: " << *ev);
+    Op* tot_op = g_pk_op[pk1];
+    Op* com_op = tot_op ? tot_op->next_com : 0;
+    uint cnt = 0;
+    bool ok = false;
+    while (com_op != 0) {
+      ll2("COM: " << *com_op);
+      Op* op = com_op->next_op;
+      assert(op != 0);
+      while (op != 0) {
+        ll2("---: " << *op);
+        op = op->next_op;
+      }
+      if (com_op->type != Op::NUL) {
+        if (com_op->type == t) {
+          const Data (&d2)[2] = com_op->data;
+          if (t == Op::INS && d2[0].seq == d[0].seq ||
+              t == Op::DEL && d2[1].seq == d[1].seq ||
+              t == Op::UPD && d2[0].seq == d[0].seq) {
+            if (cnt == g_ev_cnt[pk1]) {
+              if (! com_op->match) {
+                ll2("match pos " << cnt);
+                ok = com_op->match = true;
+              } else {
+                ll2("duplicate match");
+              }
+            } else {
+              ll2("match bad pos event=" << g_ev_cnt[pk1] << " op=" << cnt);
+            }
+          }
+        }
+        cnt++;
+      }
+      com_op = com_op->next_com;
+    }
+    if (ok)
+      return 0;
+    ll2("no match");
+    if (g_loglevel >= 2)
+      return -1;
+    loop++;
+  }
+  return 0;
+}
+
+static int
+matchevents()
+{
+  uint nomatch = 0;
+  Uint32 pk1;
+  for (pk1 = 0; pk1 < g_opts.maxpk; pk1++) {
+    Op* tot_ev = g_pk_ev[pk1];
+    if (tot_ev == 0)
+      continue;
+    Op* com_ev = tot_ev->next_com;
+    while (com_ev != 0) {
+      if (matchevent(com_ev) < 0)
+        nomatch++;
+      g_ev_cnt[pk1]++;
+      com_ev = com_ev->next_com;
+    }
+  }
+  chkrc(nomatch == 0);
+  return 0;
+}
+
+static int
+matchops()
+{
+  Uint32 pk1;
+  for (pk1 = 0; pk1 < g_opts.maxpk; pk1++) {
+    Op* tot_op = g_pk_op[pk1];
+    if (tot_op == 0)
+      continue;
+    Op* com_op = tot_op->next_com;
+    while (com_op != 0) {
+      if (com_op->type != Op::NUL && ! com_op->match) {
+        ll0("COM: " << *com_op);
+        Op* op = com_op->next_op;
+        assert(op != 0);
+        while (op != 0) {
+          ll0("---: " << *op);
+          op = op->next_op;
+        }
+        ll0("no matching event");
+        return -1;
+      }
+      com_op = com_op->next_com;
+    }
+  }
+  return 0;
+}
+
+static int
+runevents()
+{
+  ll1("runevents");
+  NdbEventOperation* evt_op;
+  uint npoll = 3;
+  while (npoll != 0) {
+    npoll--;
+    int ret;
+    ll1("poll");
+    ret = g_ndb->pollEvents(1000);
+    if (ret <= 0)
+      continue;
+    while (1) {
+      g_rec_ev->init();
+      Data (&d)[2] = g_rec_ev->data;
+#ifdef version50
+      int overrun = g_opts.maxops;
+      chkdb((ret = g_evt_op->next(&overrun)) >= 0);
+      chkrc(overrun == 0);
+      if (ret == 0)
+        break;
+#else
+      NdbEventOperation* tmp_op = g_ndb->nextEvent();
+      if (tmp_op == 0)
+        break;
+      reqrc(g_evt_op == tmp_op);
+#endif
+      chkrc(seteventtype(g_rec_ev, g_evt_op->getEventType()) == 0);
+      // get indicators
+      { int i, j;
+        for (j = 0; j < 2; j++)
+          for (i = 0; i < g_ncol; i++)
+            d[j].ind[i] = g_ra[j][i]->isNULL();
+      }
+      ll2("runevents: EVT: " << *g_rec_ev);
+      // check basic sanity
+      Uint32 pk1 = ~(Uint32)0;
+      chkrc(checkop(g_rec_ev, pk1) == 0);
+      // add to events
+      chkrc(getfreeevs() >= 2);
+      Op* tot_ev = g_pk_ev[pk1];
+      if (tot_ev == 0)
+        tot_ev = g_pk_ev[pk1] = getev(); //1
+      Op* last_com = tot_ev;
+      while (last_com->next_com != 0)
+        last_com = last_com->next_com;
+      // copy and add
+      Op* ev = getev(); //3
+      copyop(g_rec_ev, ev);
+      last_com->next_com = ev;
+    }
+  }
+  chkrc(matchevents() == 0);
+  chkrc(matchops() == 0);
+  return 0;
+}
+
+static void
+setseed(int n)
+{
+  uint seed;
+  if (n == -1) {
+    if (g_opts.seed == 0)
+      return;
+    if (g_opts.seed != -1)
+      seed = (uint)g_opts.seed;
+    else
+      seed = 1 + (ushort)getpid();
+  } else {
+    if (g_opts.seed != 0)
+      return;
+    seed = n;
+  }
+  ll0("seed=" << seed);
+  srandom(seed);
+}
+
+static int
+runtest()
+{
+  setseed(-1);
+  chkrc(createtable() == 0);
+  chkrc(createevent() == 0);
+  uint n;
+  for (n = 0; n < g_opts.loop; n++) {
+    ll0("loop " << n);
+    setseed(n);
+    resetmem();
+    g_rec_ev = getev();
+    chkrc(createeventop() == 0);
+    chkdb(g_evt_op->execute() == 0);
+    chkrc(waitgci() == 0);
+    makeops();
+    chkrc(runops() == 0);
+    chkrc(runevents() == 0);
+    chkrc(dropeventop() == 0);
+  }
+  chkrc(dropevent() == 0);
+  chkrc(droptable() == 0);
+  return 0;
+}
+
+NDB_STD_OPTS_VARS;
+
+static struct my_option
+my_long_options[] =
+{
+  NDB_STD_OPTS("test_event_merge"),
+  { "abort-on-error", 1008, "Do abort() on any error",
+    (gptr*)&g_opts.abort_on_error, (gptr*)&g_opts.abort_on_error, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "loglevel", 1001, "Logging level in this program (default 0)",
+    (gptr*)&g_opts.loglevel, (gptr*)&g_opts.loglevel, 0,
+    GET_INT, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { "loop", 1002, "Number of test loops (default 1, 0=forever)",
+    (gptr*)&g_opts.loop, (gptr*)&g_opts.loop, 0,
+    GET_INT, REQUIRED_ARG, 1, 0, 0, 0, 0, 0 },
+  { "maxops", 1003, "Number of PK operations (default 2000)",
+    (gptr*)&g_opts.maxops, (gptr*)&g_opts.maxops, 0,
+    GET_UINT, REQUIRED_ARG, 2000, 0, g_maxops, 0, 0, 0 },
+  { "maxpk", 1004, "Number of different PK values (default 10)",
+    (gptr*)&g_opts.maxpk, (gptr*)&g_opts.maxpk, 0,
+    GET_UINT, REQUIRED_ARG, 10, 1, g_maxpk, 0, 0, 0 },
+  { "opstr", 1005, "Ops to run e.g. idiucdc (c = commit, default random)",
+    (gptr*)&g_opts.opstr, (gptr*)&g_opts.opstr, 0,
+    GET_STR_ALLOC, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { "seed", 1006, "Random seed (0=loop number, default -1=random)",
+    (gptr*)&g_opts.seed, (gptr*)&g_opts.seed, 0,
+    GET_INT, REQUIRED_ARG, -1, 0, 0, 0, 0, 0 },
+  { "separate-events", 1007, "Do not combine events per GCI >5.0",
+    (gptr*)&g_opts.separate_events, (gptr*)&g_opts.separate_events, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "use-table", 1008, "Use existing table 'tem1'",
+    (gptr*)&g_opts.use_table, (gptr*)&g_opts.use_table, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { 0, 0, 0,
+    0, 0, 0,
+    GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0 }
+};
+
+static void
+usage()
+{
+  my_print_help(my_long_options);
+}
+
+static int
+checkopts()
+{
+  if (g_opts.opstr != 0) {
+    const char* s = g_opts.opstr;
+    uint n = strlen(s);
+    if (n < 3 || s[0] != 'i' || s[n-2] != 'd' || s[n-1] != 'c')
+      return -1;
+    while (*s != 0)
+      if (strchr("iduc", *s++) == 0)
+        return -1;
+  }
+  return 0;
+}
+
+int
+main(int argc, char** argv)
+{
+  ndb_init();
+  const char* progname =
+    strchr(argv[0], '/') ? strrchr(argv[0], '/') + 1 : argv[0];
+  uint i;
+  ndbout << progname;
+  for (i = 1; i < argc; i++)
+    ndbout << " " << argv[i];
+  ndbout << endl;
+  int ret;
+  ret = handle_options(&argc, &argv, my_long_options, ndb_std_get_one_option);
+  if (ret != 0 || argc != 0 || checkopts() != 0)
+    return NDBT_ProgramExit(NDBT_WRONGARGS);
+  g_ncc = new Ndb_cluster_connection();
+  if (g_ncc->connect(30) == 0) {
+    g_ndb = new Ndb(g_ncc, "TEST_DB");
+    if (g_ndb->init() == 0 && g_ndb->waitUntilReady(30) == 0) {
+      if (runtest() == 0)
+        return NDBT_ProgramExit(NDBT_OK);
+    }
+  }
+  delete g_ndb;
+  delete g_ncc;
+  return NDBT_ProgramExit(NDBT_FAILED);
+}