test_event_multi_table.cpp

/*
   Copyright (C) 2005, 2006, 2008 MySQL AB
    All rights reserved. Use is subject to license terms.

   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; version 2 of the License.

   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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  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>
#include <NdbRestarter.hpp>
#include <NdbRestarts.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)
{
  DBUG_ENTER("copy_events");
  int r= 0;
  NdbDictionary::Dictionary * dict = ndb->getDictionary();
  while (1)
  {
    int res= ndb->pollEvents(1000); // wait for event or 1000 ms
    DBUG_PRINT("info", ("pollEvents res=%d", res));
    if (res <= 0)
    {
      break;
    }
    int error= 0;
    NdbEventOperation *pOp;
    while ((pOp= ndb->nextEvent()))
    {
      char buf[1024];
      sprintf(buf, "%s_SHADOW", pOp->getTable()->getName());
      const NdbDictionary::Table *table= dict->getTable(buf);

      if (table == 0)
      {
        g_err << "unable to find table " << buf << endl;
        DBUG_RETURN(-1);
      }

      if (pOp->isOverrun())
      {
        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 (const NdbRecAttr *pk= pOp->getFirstPkAttr(); pk; pk= pk->next())
          {
            if (pk->isNULL())
            {
              g_err << "internal error: primary key isNull()="
                    << pk->isNULL() << endl;
              DBUG_RETURN(NDBT_FAILED);
            }
            if (op->equal(pk->getColumn()->getColumnNo(),pk->aRef()))
            {
              g_err << "equal " << pk->getColumn()->getColumnNo() << " "
                    << op->getNdbError().code << " "
                    << op->getNdbError().message << endl;
              DBUG_RETURN(NDBT_FAILED);
            }
          }
        }
        
        switch (pOp->getEventType()) {
        case NdbDictionary::Event::TE_INSERT:
        {
          for (const NdbRecAttr *data= pOp->getFirstDataAttr(); data; data= data->next())
          {
            if (data->isNULL() < 0 ||
                op->setValue(data->getColumn()->getColumnNo(),
                             data->isNULL() ? 0:data->aRef()))
            {
              g_err << "setValue(insert) " << data->getColumn()->getColumnNo() << " "
                    << op->getNdbError().code << " "
                    << op->getNdbError().message << endl;
              DBUG_RETURN(-1);
            }
          }
          break;
        }
        case NdbDictionary::Event::TE_DELETE:
          break;
        case NdbDictionary::Event::TE_UPDATE:
        {
          for (const NdbRecAttr *data= pOp->getFirstDataAttr(); data; data= data->next())
          {
            if (data->isNULL() >= 0 &&
                op->setValue(data->getColumn()->getColumnNo(),
                             data->isNULL() ? 0:data->aRef()))
            {
              g_err << "setValue(update) " << data->getColumn()->getColumnNo() << " "
                    << 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);
    } // for
    if (error)
    {
      g_err << "nextEvent()\n";
      DBUG_RETURN(-1);
    }
  } // 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;
}

static const char* _dbname = "TEST_DB";
struct my_option my_long_options[] =
{
  NDB_STD_OPTS(""),
  { "database", 'd', "Name of database table is in",
    (uchar**) &_dbname, (uchar**) &_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;
  if (argc == 0)
  {
    NDBT_Tables::dropAllTables(&ndb);
    NDBT_Tables::createAllTables(&ndb);
    for (i= 0; no_error && i < NDBT_Tables::getNumTables(); i++)
    {
      const NdbDictionary::Table *pTab= dict->getTable(NDBT_Tables::getTable(i)->getName());
      if (pTab == 0)
      {
        ndbout << "Failed to create table" << endl;
        ndbout << dict->getNdbError() << endl;
        no_error= 0;
        break;
      }
      pTabs.push_back(pTab);
    }
  }
  else
  {
    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
  for (i= 0; no_error && pTabs[i]; i++)
  {
    int n_columns= pTabs[i]->getNoOfColumns();
    for (int j = 0; j < n_columns; j++) {
      pOps[i]->getValue(pTabs[i]->getColumn(j)->getName());
      pOps[i]->getPreValue(pTabs[i]->getColumn(j)->getName());
    }
  }

  // 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]));
  }

  int n_records= 3;
  // insert n_records 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, n_records);
    }
    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) < 0)
    {
      no_error= 0;
      break;
    }
    if (verify_copy(&ndb, pTabs, pShadowTabs))
    {
      no_error= 0;
      break;
    }
  } while (0);

  // update n_records-1 records in first table
  do {
    if (start_transaction(&ndb, hugo_ops))
    {
      no_error= 0;
      break;
    }

    hugo_ops[0]->pkUpdateRecord(&ndb, n_records-1);

    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) < 0)
    {
      no_error= 0;
      break;
    }
    if (verify_copy(&ndb, pTabs, pShadowTabs))
    {
      no_error= 0;
      break;
    }
  } while (0);


  {
    NdbRestarts restarts;
    for (int j= 0; j < 10; j++)
    {
      // restart a node
      if (no_error)
      {
        int timeout = 240;
        if (restarts.executeRestart("RestartRandomNodeAbort", timeout))
        {
          no_error= 0;
          break;
        }
      }

      // update all n_records records on all tables
      if (start_transaction(&ndb, hugo_ops))
      {
        no_error= 0;
        break;
      }

      for (int r= 0; r < n_records; r++)
      {
        for (i= 0; pTabs[i]; i++)
        {
          hugo_ops[i]->pkUpdateRecord(&ndb, r);
        }
      }
      if (execute_commit(&ndb, hugo_ops))
      {
        no_error= 0;
        break;
      }
      if(close_transaction(&ndb, hugo_ops))
      {
        no_error= 0;
        break;
      }

      // copy events and verify
      if (copy_events(&ndb) < 0)
      {
        no_error= 0;
        break;
      }
      if (verify_copy(&ndb, pTabs, pShadowTabs))
      {
        no_error= 0;
        break;
      }
    }
  }

  // drop the event operations
  for (i= 0; i < (int)pOps.size(); i++)
  {
    if (ndb.dropEventOperation(pOps[i]))
    {
      no_error= 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*> >;