added testcase for multiple tables and events

ndb/test/ndbapi/Makefile.am

@@ -33,7 +33,8 @@ testDeadlock \
 test_event ndbapi_slow_select testReadPerf testLcp \
 testPartitioning \
 testBitfield \
-DbCreate DbAsyncGenerator
+DbCreate DbAsyncGenerator \
+test_event_multi_table
 
 #flexTimedAsynch
 #testBlobs
@@ -76,6 +77,7 @@ testPartitioning_SOURCES = testPartitioning.cpp
 testBitfield_SOURCES = testBitfield.cpp
 DbCreate_SOURCES = bench/mainPopulate.cpp bench/dbPopulate.cpp bench/userInterface.cpp bench/dbPopulate.h bench/userInterface.h bench/testData.h bench/testDefinitions.h bench/ndb_schema.hpp bench/ndb_error.hpp
 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
 
 INCLUDES_LOC = -I$(top_srcdir)/ndb/include/kernel
 

ndb/test/ndbapi/test_event.cpp

@@ -14,11 +14,11 @@
    along with this program; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
-#include "NDBT_Test.hpp"
+#include <NDBT_Test.hpp>
-#include "NDBT_ReturnCodes.h"
+#include <NDBT_ReturnCodes.h>
-#include "HugoTransactions.hpp"
+#include <HugoTransactions.hpp>
-#include "UtilTransactions.hpp"
+#include <UtilTransactions.hpp>
-#include "TestNdbEventOperation.hpp"
+#include <TestNdbEventOperation.hpp>
 
 #define GETNDB(ps) ((NDBT_NdbApiStep*)ps)->getNdb()
 
@@ -263,7 +263,10 @@ int runEventApplier(NDBT_Context* ctx, NDBT_Step* step)
     //printf("now waiting for event...\n");
     res= GETNDB(step)->pollEvents(1000); // wait for event or 1000 ms
     if (res <= 0)
+    {
+      ndbout_c("********************");
       continue;
+    }
 
     //printf("got data! %d\n", r);
     int overrun= 0;

ndb/test/ndbapi/test_event_multi_table.cpp

@@ -0,0 +1,487 @@
+/* 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 <NDBT_Test.hpp>
+#include <NDBT_ReturnCodes.h>
+#include <HugoTransactions.hpp>
+#include <UtilTransactions.hpp>
+#include <TestNdbEventOperation.hpp>
+
+static void usage()
+{
+  ndb_std_print_version();
+}
+
+static int start_transaction(Ndb *ndb, Vector<HugoOperations*> &ops)
+{
+  if (ops[0]->startTransaction(ndb) != NDBT_OK)
+    return -1;
+  NdbTransaction * t= ops[0]->getTransaction();
+  for (int i= ops.size()-1; i > 0; i--)
+  {
+    ops[i]->setTransaction(t);
+  }
+  return 0;
+}
+
+static int close_transaction(Ndb *ndb, Vector<HugoOperations*> &ops)
+{
+  if (ops[0]->closeTransaction(ndb) != NDBT_OK)
+    return -1;
+  for (int i= ops.size()-1; i > 0; i--)
+  {
+    ops[i]->setTransaction(NULL);
+  }
+  return 0;
+}
+
+static int execute_commit(Ndb *ndb, Vector<HugoOperations*> &ops)
+{
+  if (ops[0]->execute_Commit(ndb) != NDBT_OK)
+    return -1;
+  return 0;
+}
+
+static int copy_events(Ndb *ndb,
+		       Vector<NdbEventOperation *> &ops,
+		       Vector<const NdbDictionary::Table *> &tabs,
+		       Vector<Vector<NdbRecAttr *> > &values)
+{
+  DBUG_ENTER("copy_events");
+  int r= 0;
+  while (1)
+  {
+    int res= ndb->pollEvents(1000); // wait for event or 1000 ms
+    DBUG_PRINT("info", ("pollEvents res=%d", r));
+    if (res <= 0)
+    {
+      break;
+    }
+    for (unsigned i_ops= 0; i_ops < ops.size(); i_ops++)
+    {
+      NdbEventOperation *pOp= ops[i_ops];
+      const NdbDictionary::Table *table= tabs[i_ops];
+      Vector<NdbRecAttr *> &recAttr= values[i_ops];
+
+      int overrun= 0;
+      unsigned i;
+      unsigned n_columns= table->getNoOfColumns();
+      while (pOp->next(&overrun) > 0)
+      {
+	if (overrun)
+	{
+	  g_err << "buffer overrun\n";
+	  DBUG_RETURN(-1);
+	}
+	r++;
+	
+	Uint32 gci= pOp->getGCI();
+
+	if (!pOp->isConsistent()) {
+	  g_err << "A node failure has occured and events might be missing\n";
+	  DBUG_RETURN(-1);
+	}
+	
+	int noRetries= 0;
+	do
+	{
+	  NdbTransaction *trans= ndb->startTransaction();
+	  if (trans == 0)
+	  {
+	    g_err << "startTransaction failed "
+		  << ndb->getNdbError().code << " "
+		  << ndb->getNdbError().message << endl;
+	    DBUG_RETURN(-1);
+	  }
+	  
+	  NdbOperation *op= trans->getNdbOperation(table);
+	  if (op == 0)
+	  {
+	    g_err << "getNdbOperation failed "
+		  << trans->getNdbError().code << " "
+		  << trans->getNdbError().message << endl;
+	    DBUG_RETURN(-1);
+	  }
+	  
+	  switch (pOp->getEventType()) {
+	  case NdbDictionary::Event::TE_INSERT:
+	    if (op->insertTuple())
+	    {
+	      g_err << "insertTuple "
+		    << op->getNdbError().code << " "
+		    << op->getNdbError().message << endl;
+	      DBUG_RETURN(-1);
+	    }
+	    break;
+	  case NdbDictionary::Event::TE_DELETE:
+	    if (op->deleteTuple())
+	    {
+	      g_err << "deleteTuple "
+		    << op->getNdbError().code << " "
+		    << op->getNdbError().message << endl;
+	      DBUG_RETURN(-1);
+	    }
+	    break;
+	  case NdbDictionary::Event::TE_UPDATE:
+	    if (op->updateTuple())
+	    {
+	      g_err << "updateTuple "
+		    << op->getNdbError().code << " "
+		    << op->getNdbError().message << endl;
+	      DBUG_RETURN(-1);
+	    }
+	    break;
+	  default:
+	    abort();
+	  }
+	  
+	  for (i= 0; i < n_columns; i++)
+	  {
+	    if (recAttr[i]->isNULL())
+	    {
+	      if (table->getColumn(i)->getPrimaryKey())
+	      {
+		g_err << "internal error: primary key isNull()="
+		      << recAttr[i]->isNULL() << endl;
+		DBUG_RETURN(NDBT_FAILED);
+	      }
+	      switch (pOp->getEventType()) {
+	      case NdbDictionary::Event::TE_INSERT:
+		if (recAttr[i]->isNULL() < 0)
+		{
+		  g_err << "internal error: missing value for insert\n";
+		  DBUG_RETURN(NDBT_FAILED);
+		}
+		break;
+	      case NdbDictionary::Event::TE_DELETE:
+		break;
+	      case NdbDictionary::Event::TE_UPDATE:
+		break;
+	      default:
+		abort();
+	      }
+	    }
+	    if (table->getColumn(i)->getPrimaryKey() &&
+		op->equal(i,recAttr[i]->aRef()))
+	    {
+	      g_err << "equal " << i << " "
+		    << op->getNdbError().code << " "
+		    << op->getNdbError().message << endl;
+	      DBUG_RETURN(NDBT_FAILED);
+	    }
+	  }
+	  
+	  switch (pOp->getEventType()) {
+	  case NdbDictionary::Event::TE_INSERT:
+	    for (i= 0; i < n_columns; i++)
+	    {
+	      if (!table->getColumn(i)->getPrimaryKey() &&
+		  op->setValue(i,recAttr[i]->isNULL() ? 0:recAttr[i]->aRef()))
+	      {
+		g_err << "setValue(insert) " << i << " "
+		      << op->getNdbError().code << " "
+		      << op->getNdbError().message << endl;
+		DBUG_RETURN(-1);
+	      }
+	    }
+	    break;
+	  case NdbDictionary::Event::TE_DELETE:
+	    break;
+	  case NdbDictionary::Event::TE_UPDATE:
+	    for (i= 0; i < n_columns; i++)
+	    {
+	      if (!table->getColumn(i)->getPrimaryKey() &&
+		  recAttr[i]->isNULL() >= 0 &&
+		  op->setValue(i,recAttr[i]->isNULL() ? 0:recAttr[i]->aRef()))
+	      {
+		g_err << "setValue(update) " << i << " "
+		      << op->getNdbError().code << " "
+		      << op->getNdbError().message << endl;
+		DBUG_RETURN(NDBT_FAILED);
+	      }
+	    }
+	    break;
+	  case NdbDictionary::Event::TE_ALL:
+	    abort();
+	  }
+	  if (trans->execute(Commit) == 0)
+	  {
+	    trans->close();
+	    // everything ok
+	    break;
+	  }
+	  if (noRetries++ == 10 ||
+	      trans->getNdbError().status != NdbError::TemporaryError)
+	  {
+	    g_err << "execute " << r << " failed "
+		  << trans->getNdbError().code << " "
+		  << trans->getNdbError().message << endl;
+	    trans->close();
+	    DBUG_RETURN(-1);
+	  }
+	  trans->close();
+	  NdbSleep_MilliSleep(100); // sleep before retying
+	} while(1);
+      }
+    }
+  }
+  DBUG_RETURN(r);
+}
+
+static int verify_copy(Ndb *ndb,
+		       Vector<const NdbDictionary::Table *> &tabs1,
+		       Vector<const NdbDictionary::Table *> &tabs2)
+{
+  for (unsigned i= 0; i < tabs1.size(); i++)
+    if (tabs1[i])
+    {
+      HugoTransactions hugoTrans(*tabs1[i]);
+      if (hugoTrans.compare(ndb, tabs2[i]->getName(), 0))
+	return -1;
+    }
+  return 0;
+}
+
+NDB_STD_OPTS_VARS;
+
+static const char* _dbname = "TEST_DB";
+static struct my_option my_long_options[] =
+{
+  NDB_STD_OPTS(""),
+  { "database", 'd', "Name of database table is in",
+    (gptr*) &_dbname, (gptr*) &_dbname, 0,
+    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
+};
+
+int
+main(int argc, char** argv)
+{
+  NDB_INIT(argv[0]);
+  const char *load_default_groups[]= { "mysql_cluster",0 };
+  load_defaults("my",load_default_groups,&argc,&argv);
+
+  int ho_error;
+#ifndef DBUG_OFF
+  opt_debug= "d:t:F:L";
+#endif
+  if ((ho_error=handle_options(&argc, &argv, my_long_options, 
+			       ndb_std_get_one_option)))
+    return NDBT_ProgramExit(NDBT_WRONGARGS);
+
+  DBUG_ENTER("main");
+  Ndb_cluster_connection con(opt_connect_str);
+  if(con.connect(12, 5, 1))
+  {
+    DBUG_RETURN(NDBT_ProgramExit(NDBT_FAILED));
+  }
+  
+
+  Ndb ndb(&con,_dbname);
+  ndb.init();
+  while (ndb.waitUntilReady() != 0);
+
+  NdbDictionary::Dictionary * dict = ndb.getDictionary();
+  int no_error= 1;
+  int i;
+
+  // create all tables
+  Vector<const NdbDictionary::Table*> pTabs;
+  for (i= 0; no_error && argc; argc--, i++)
+  {
+    dict->dropTable(argv[i]);
+    NDBT_Tables::createTable(&ndb, argv[i]);
+    const NdbDictionary::Table *pTab= dict->getTable(argv[i]);
+    if (pTab == 0)
+    {
+      ndbout << "Failed to create table" << endl;
+      ndbout << dict->getNdbError() << endl;
+      no_error= 0;
+      break;
+    }
+    pTabs.push_back(pTab);
+  }
+  pTabs.push_back(NULL);
+
+  // create an event for each table
+  for (i= 0; no_error && pTabs[i]; i++)
+  {
+    HugoTransactions ht(*pTabs[i]);
+    if (ht.createEvent(&ndb)){
+      no_error= 0;
+      break;
+    }
+  }
+
+  // create an event operation for each event
+  Vector<NdbEventOperation *> pOps;
+  for (i= 0; no_error && pTabs[i]; i++)
+  {
+    char buf[1024];
+    sprintf(buf, "%s_EVENT", pTabs[i]->getName());
+    NdbEventOperation *pOp= ndb.createEventOperation(buf, 1000);
+    if ( pOp == NULL )
+    {
+      no_error= 0;
+      break;
+    }
+    pOps.push_back(pOp);
+  }
+
+  // get storage for each event operation
+  Vector<Vector<NdbRecAttr*> > values;
+  Vector<Vector<NdbRecAttr*> > pre_values;
+  for (i= 0; no_error && pTabs[i]; i++)
+  {
+    int n_columns= pTabs[i]->getNoOfColumns();
+    Vector<NdbRecAttr*> tmp_a;
+    Vector<NdbRecAttr*> tmp_b;
+    for (int j = 0; j < n_columns; j++) {
+      tmp_a.push_back(pOps[i]->getValue(pTabs[i]->getColumn(j)->getName()));
+      tmp_b.push_back(pOps[i]->getPreValue(pTabs[i]->getColumn(j)->getName()));
+    }
+    values.push_back(tmp_a);
+    pre_values.push_back(tmp_b);
+  }
+
+  // start receiving events
+  for (i= 0; no_error && pTabs[i]; i++)
+  {
+    if ( pOps[i]->execute() )
+    {
+      no_error= 0;
+      break;
+    }
+  }
+
+  // create a "shadow" table for each table
+  Vector<const NdbDictionary::Table*> pShadowTabs;
+  for (i= 0; no_error && pTabs[i]; i++)
+  {
+    char buf[1024];
+    sprintf(buf, "%s_SHADOW", pTabs[i]->getName());
+
+    dict->dropTable(buf);
+    if (dict->getTable(buf))
+    {
+      no_error= 0;
+      break;
+    }
+
+    NdbDictionary::Table table_shadow(*pTabs[i]);
+    table_shadow.setName(buf);
+    dict->createTable(table_shadow);
+    pShadowTabs.push_back(dict->getTable(buf));
+    if (!pShadowTabs[i])
+    {
+      no_error= 0;
+      break;
+    }
+  }
+
+  // create a hugo operation per table
+  Vector<HugoOperations *> hugo_ops;
+  for (i= 0; no_error && pTabs[i]; i++)
+  {
+    hugo_ops.push_back(new HugoOperations(*pTabs[i]));
+  }
+
+  sleep(5);
+
+  // insert 3 records per table
+  do {
+    if (start_transaction(&ndb, hugo_ops))
+    {
+      no_error= 0;
+      break;
+    }
+    for (i= 0; no_error && pTabs[i]; i++)
+    {
+      hugo_ops[i]->pkInsertRecord(&ndb, 0, 3);
+    }
+    if (execute_commit(&ndb, hugo_ops))
+    {
+      no_error= 0;
+      break;
+    }
+    if(close_transaction(&ndb, hugo_ops))
+    {
+      no_error= 0;
+      break;
+    }
+  } while(0);
+
+  // copy events and verify
+  do {
+    if (copy_events(&ndb, pOps, pShadowTabs, values) < 0)
+    {
+      no_error= 0;
+      break;
+    }
+    if (verify_copy(&ndb, pTabs, pShadowTabs))
+    {
+      no_error= 0;
+      break;
+    }
+  } while (0);
+
+  // update 2 records in first table
+  do {
+    if (start_transaction(&ndb, hugo_ops))
+    {
+      no_error= 0;
+      break;
+    }
+
+    hugo_ops[0]->pkUpdateRecord(&ndb, 2);
+
+    if (execute_commit(&ndb, hugo_ops))
+    {
+      no_error= 0;
+      break;
+    }
+    if(close_transaction(&ndb, hugo_ops))
+    {
+      no_error= 0;
+      break;
+    }
+  } while(0);
+
+  // copy events and verify
+  do {
+    if (copy_events(&ndb, pOps, pShadowTabs, values) < 0)
+    {
+      no_error= 0;
+      break;
+    }
+    if (verify_copy(&ndb, pTabs, pShadowTabs))
+    {
+      no_error= 0;
+      break;
+    }
+  } while (0);
+
+  if (no_error)
+    DBUG_RETURN(NDBT_ProgramExit(NDBT_OK));
+  DBUG_RETURN(NDBT_ProgramExit(NDBT_FAILED));
+}
+
+template class Vector<HugoOperations *>;
+template class Vector<NdbEventOperation *>;
+template class Vector<NdbRecAttr*>;
+template class Vector<Vector<NdbRecAttr*> >;