Commit e914e6c7 authored by unknown's avatar unknown
Browse files

Merge mysql.com:/home/jonas/src/mysql-4.1 

into mysql.com:/home/jonas/src/mysql-4.1-ndb
parents 9155e8de 89ba9400

ndb/include/kernel/signaldata/TcKeyConf.hpp

+1 −1
Original line number Diff line number Diff line
@@ -111,7 +111,7 @@ inline 
void 

TcKeyConf::setNoOfOperations(Uint32 & confInfo, Uint32 noOfOps){

  ASSERT_MAX(noOfOps, 65535, "TcKeyConf::setNoOfOperations");

  confInfo |= noOfOps;

  confInfo = (confInfo & 0xFFFF0000) | noOfOps;

}



inline

ndb/include/util/Parser.hpp

+1 −0
Original line number Diff line number Diff line
@@ -165,6 +165,7 @@ Parser<T>::Parser(const ParserRow<T> rows[], class InputStream & in, 
template<class T>

inline

Parser<T>::~Parser(){

  delete impl;

}



template<class T>

ndb/src/kernel/blocks/backup/restore/Makefile.am

+1 −1
Original line number Diff line number Diff line 


ndbtools_PROGRAMS = ndb_restore



ndb_restore_SOURCES = main.cpp Restore.cpp

ndb_restore_SOURCES = main.cpp consumer.cpp consumer_restore.cpp consumer_printer.cpp Restore.cpp



LDADD_LOC = $(top_builddir)/ndb/src/libndbclient.la

ndb/src/kernel/blocks/backup/restore/Makefile_old

deleted100644 → 0
+0 −20
Original line number Diff line number Diff line 
include .defs.mk



TYPE := *



BIN_TARGET := restore

BIN_TARGET_LIBS := 

BIN_TARGET_ARCHIVES := NDB_API



CCFLAGS_LOC = -I.. -I$(NDB_TOP)/src/ndbapi  -I$(NDB_TOP)/include/ndbapi -I$(NDB_TOP)/include/util  -I$(NDB_TOP)/include/portlib -I$(NDB_TOP)/include/kernel



#ifneq ($(MYSQLCLUSTER_TOP),)

#CCFLAGS_LOC +=-I$(MYSQLCLUSTER_TOP)/include -D USE_MYSQL

#LDFLAGS_LOC += -L$(MYSQLCLUSTER_TOP)/libmysql_r/ -lmysqlclient_r

#endif



SOURCES = main.cpp Restore.cpp



include $(NDB_TOP)/Epilogue.mk

ndb/src/kernel/blocks/backup/restore/Restore.cpp

+278 −146
Original line number Diff line number Diff line
@@ -33,32 +33,32 @@ Uint32 Twiddle32(Uint32 in); // Byte shift 32-bit data 
Uint64 Twiddle64(Uint64 in); // Byte shift 64-bit data



bool

BackupFile::Twiddle(AttributeS* attr, Uint32 arraySize){

BackupFile::Twiddle(const AttributeDesc* attr_desc, AttributeData* attr_data, Uint32 arraySize){



  if(m_hostByteOrder)

    return true;

  

  if(arraySize == 0){

    arraySize = attr->Desc->arraySize;

    arraySize = attr_desc->arraySize;

  }

  

  switch(attr->Desc->size){

  switch(attr_desc->size){

  case 8:

    

    return true;

  case 16:

    for(unsigned i = 0; i<arraySize; i++){

      attr->Data.u_int16_value[i] = Twiddle16(attr->Data.u_int16_value[i]);

      attr_data->u_int16_value[i] = Twiddle16(attr_data->u_int16_value[i]);

    }

    return true;

  case 32:

    for(unsigned i = 0; i<arraySize; i++){

      attr->Data.u_int32_value[i] = Twiddle32(attr->Data.u_int32_value[i]);

      attr_data->u_int32_value[i] = Twiddle32(attr_data->u_int32_value[i]);

    }

    return true;

  case 64:

    for(unsigned i = 0; i<arraySize; i++){

      attr->Data.u_int64_value[i] = Twiddle64(attr->Data.u_int64_value[i]);

      attr_data->u_int64_value[i] = Twiddle64(attr_data->u_int64_value[i]);

    }

    return true;

  default:
@@ -82,14 +82,14 @@ RestoreMetaData::RestoreMetaData(const char* path, Uint32 nodeId, Uint32 bNo) { 
}



RestoreMetaData::~RestoreMetaData(){

  for(int i = 0; i<allTables.size(); i++)

  for(Uint32 i= 0; i < allTables.size(); i++)

    delete allTables[i];

  allTables.clear();

}



const TableS * 

RestoreMetaData::getTable(Uint32 tableId) const {

  for(int i = 0; i<allTables.size(); i++)

  for(Uint32 i= 0; i < allTables.size(); i++)

    if(allTables[i]->getTableId() == tableId)

      return allTables[i];

  return NULL;
@@ -122,7 +122,8 @@ RestoreMetaData::readMetaTableList() { 
  

  Uint32 sectionInfo[2];

  

  if (fread(&sectionInfo, sizeof(sectionInfo), 1, m_file) != 1){

  if (buffer_read(&sectionInfo, sizeof(sectionInfo), 1) != 1){

    err << "readMetaTableList read header error" << endl;

    return 0;

  }

  sectionInfo[0] = ntohl(sectionInfo[0]);
@@ -130,11 +131,9 @@ RestoreMetaData::readMetaTableList() { 


  const Uint32 tabCount = sectionInfo[1] - 2;



  const Uint32 len = 4 * tabCount;

  if(createBuffer(len) == 0)

    abort();



  if (fread(m_buffer, 1, len, m_file) != len){

  void *tmp;

  if (buffer_get_ptr(&tmp, 4, tabCount) != tabCount){

    err << "readMetaTableList read tabCount error" << endl;

    return 0;

  }
@@ -147,7 +146,7 @@ RestoreMetaData::readMetaTableDesc() { 
  Uint32 sectionInfo[2];

  

  // Read section header 

  if (fread(&sectionInfo, sizeof(sectionInfo), 1, m_file) != 1){

  if (buffer_read(&sectionInfo, sizeof(sectionInfo), 1) != 1){

    err << "readMetaTableDesc read header error" << endl;

    return false;

  } // if
@@ -156,20 +155,15 @@ RestoreMetaData::readMetaTableDesc() { 
  

  assert(sectionInfo[0] == BackupFormat::TABLE_DESCRIPTION);

  

  // Allocate temporary storage for dictTabInfo buffer

  const Uint32 len = (sectionInfo[1] - 2);

  if (createBuffer(4 * (len+1)) == NULL) {

    err << "readMetaTableDesc allocation error" << endl;

    return false;

  } // if

  

  // Read dictTabInfo buffer

  if (fread(m_buffer, 4, len, m_file) != len){

  const Uint32 len = (sectionInfo[1] - 2);

  void *ptr;

  if (buffer_get_ptr(&ptr, 4, len) != len){

    err << "readMetaTableDesc read error" << endl;

    return false;

  } // if

  

  return parseTableDescriptor(m_buffer, len);	     

  return parseTableDescriptor((Uint32*)ptr, len);	     

}



bool
@@ -177,11 +171,10 @@ RestoreMetaData::readGCPEntry() { 


  Uint32 data[4];

  

  

  BackupFormat::CtlFile::GCPEntry * dst = 

    (BackupFormat::CtlFile::GCPEntry *)&data[0];

  

  if(fread(dst, 4, 4, m_file) != 4){

  if(buffer_read(dst, 4, 4) != 4){

    err << "readGCPEntry read error" << endl;

    return false;

  }
@@ -208,10 +201,16 @@ TableS::TableS(NdbTableImpl* tableImpl) 
  m_dictTable = tableImpl;

  m_noOfNullable = m_nullBitmaskSize = 0;



  for (Uint32 i = 0; i < tableImpl->getNoOfColumns(); i++)

  for (int i = 0; i < tableImpl->getNoOfColumns(); i++)

    createAttr(tableImpl->getColumn(i));

}



TableS::~TableS()

{

  for (Uint32 i= 0; i < allAttributesDesc.size(); i++)

    delete allAttributesDesc[i];

}



// Parse dictTabInfo buffer and pushback to to vector storage 

bool

RestoreMetaData::parseTableDescriptor(const Uint32 * data, Uint32 len)
@@ -246,56 +245,68 @@ RestoreMetaData::parseTableDescriptor(const Uint32 * data, Uint32 len) 
}



// Constructor

RestoreDataIterator::RestoreDataIterator(const RestoreMetaData & md)

  : m_metaData(md) 

RestoreDataIterator::RestoreDataIterator(const RestoreMetaData & md, void (* _free_data_callback)())

  : BackupFile(_free_data_callback), m_metaData(md)

{

  debug << "RestoreDataIterator constructor" << endl;

  setDataFile(md, 0);

}



RestoreDataIterator::~RestoreDataIterator(){

TupleS & TupleS::operator=(const TupleS& tuple)

{

  prepareRecord(*tuple.m_currentTable);



  if (allAttrData)

    memcpy(allAttrData, tuple.allAttrData, getNoOfAttributes()*sizeof(AttributeData));

  

  return *this;

};

int TupleS::getNoOfAttributes() const {

  if (m_currentTable == 0)

    return 0;

  return m_currentTable->getNoOfAttributes();

};



const TableS * TupleS::getTable() const {

  return m_currentTable;

};



const AttributeDesc * TupleS::getDesc(int i) const {

  return m_currentTable->allAttributesDesc[i];

}



AttributeData * TupleS::getData(int i) const{

  return &(allAttrData[i]);

};



bool

TupleS::prepareRecord(const TableS & tab){

  if (allAttrData) {

    if (getNoOfAttributes() == tab.getNoOfAttributes())

    {

      m_currentTable = &tab;

  for(int i = 0; i<allAttributes.size(); i++) {

    if(allAttributes[i] != NULL)

      delete allAttributes[i];

  }

  allAttributes.clear();

  AttributeS * a;

  for(int i = 0; i<tab.getNoOfAttributes(); i++){

    a = new AttributeS;

    if(a == NULL) {

      ndbout_c("Restore: Failed to allocate memory");

      return false;

      return true;

    }

    a->Desc = tab[i];

    allAttributes.push_back(a);

    delete [] allAttrData;

    m_currentTable= 0;

  }

  

  allAttrData = new AttributeData[tab.getNoOfAttributes()];

  if (allAttrData == 0)

    return false;

  

  m_currentTable = &tab;



  return true;

}



const TupleS *

RestoreDataIterator::getNextTuple(int  & res) {

  TupleS * tup = new TupleS();

  if(tup == NULL) {

    ndbout_c("Restore: Failed to allocate memory");

    res = -1;

    return NULL;

  }

  if(!tup->prepareRecord(* m_currentTable)) {

    res =-1;

    return NULL;

  }

    



RestoreDataIterator::getNextTuple(int  & res)

{

  Uint32  dataLength = 0;

  // Read record length

  if (fread(&dataLength, sizeof(dataLength), 1, m_file) != 1){

  if (buffer_read(&dataLength, sizeof(dataLength), 1) != 1){

    err << "getNextTuple:Error reading length  of data part" << endl;

    delete tup;

    res = -1;

    return NULL;

  } // if
@@ -309,34 +320,34 @@ RestoreDataIterator::getNextTuple(int & res) { 
    // End of this data fragment

    debug << "End of fragment" << endl;

    res = 0;

    delete tup;

    return NULL;

  } // if



  tup->createDataRecord(dataLenBytes);

  // Read tuple data

  if (fread(tup->getDataRecord(), 1, dataLenBytes, m_file) != dataLenBytes) {

  void *_buf_ptr;

  if (buffer_get_ptr(&_buf_ptr, 1, dataLenBytes) != dataLenBytes) {

    err << "getNextTuple:Read error: " << endl;

    delete tup;

    res = -1;

    return NULL;

  }

  

  Uint32 * ptr = tup->getDataRecord();

  Uint32 *buf_ptr = (Uint32*)_buf_ptr, *ptr = buf_ptr;

  ptr += m_currentTable->m_nullBitmaskSize;



  for(int i = 0; i < m_currentTable->m_fixedKeys.size(); i++){

    assert(ptr < tup->getDataRecord() + dataLength);

  for(Uint32 i= 0; i < m_currentTable->m_fixedKeys.size(); i++){

    assert(ptr < buf_ptr + dataLength);

 

    const Uint32 attrId = m_currentTable->m_fixedKeys[i]->attrId;

    AttributeS * attr = tup->allAttributes[attrId];



    const Uint32 sz = attr->Desc->getSizeInWords();

    AttributeData * attr_data = m_tuple.getData(attrId);

    const AttributeDesc * attr_desc = m_tuple.getDesc(attrId);



    attr->Data.null = false;

    attr->Data.void_value = ptr;

    const Uint32 sz = attr_desc->getSizeInWords();



    if(!Twiddle(attr))

    attr_data->null = false;

    attr_data->void_value = ptr;



    if(!Twiddle(attr_desc, attr_data))

      {

	res = -1;

	return NULL;
@@ -344,18 +355,20 @@ RestoreDataIterator::getNextTuple(int & res) { 
    ptr += sz;

  }



  for(int i = 0; i<m_currentTable->m_fixedAttribs.size(); i++){

    assert(ptr < tup->getDataRecord() + dataLength);

  for(Uint32 i = 0; i < m_currentTable->m_fixedAttribs.size(); i++){

    assert(ptr < buf_ptr + dataLength);



    const Uint32 attrId = m_currentTable->m_fixedAttribs[i]->attrId;

    AttributeS * attr = tup->allAttributes[attrId];



    const Uint32 sz = attr->Desc->getSizeInWords();

    AttributeData * attr_data = m_tuple.getData(attrId);

    const AttributeDesc * attr_desc = m_tuple.getDesc(attrId);



    attr->Data.null = false;

    attr->Data.void_value = ptr;

    const Uint32 sz = attr_desc->getSizeInWords();



    if(!Twiddle(attr))

    attr_data->null = false;

    attr_data->void_value = ptr;



    if(!Twiddle(attr_desc, attr_data))

      {

	res = -1;

	return NULL;
@@ -364,21 +377,23 @@ RestoreDataIterator::getNextTuple(int & res) { 
    ptr += sz;

  }



  for(int i = 0; i<m_currentTable->m_variableAttribs.size(); i++){

  for(Uint32 i = 0; i < m_currentTable->m_variableAttribs.size(); i++){

    const Uint32 attrId = m_currentTable->m_variableAttribs[i]->attrId;

    AttributeS * attr = tup->allAttributes[attrId];



    if(attr->Desc->m_column->getNullable()){

      const Uint32 ind = attr->Desc->m_nullBitIndex;

    AttributeData * attr_data = m_tuple.getData(attrId);

    const AttributeDesc * attr_desc = m_tuple.getDesc(attrId);

    

    if(attr_desc->m_column->getNullable()){

      const Uint32 ind = attr_desc->m_nullBitIndex;

      if(BitmaskImpl::get(m_currentTable->m_nullBitmaskSize, 

			  tup->getDataRecord(),ind)){

	attr->Data.null = true;

	attr->Data.void_value = NULL;

			  buf_ptr,ind)){

	attr_data->null = true;

	attr_data->void_value = NULL;

	continue;

      }

    }



    assert(ptr < tup->getDataRecord() + dataLength);

    assert(ptr < buf_ptr + dataLength);



    typedef BackupFormat::DataFile::VariableData VarData;

    VarData * data = (VarData *)ptr;
@@ -386,15 +401,15 @@ RestoreDataIterator::getNextTuple(int & res) { 
    Uint32 id = ntohl(data->Id);

    assert(id == attrId);

    

    attr->Data.null = false;

    attr->Data.void_value = &data->Data[0];

    attr_data->null = false;

    attr_data->void_value = &data->Data[0];



    /**

     * Compute array size

     */

    const Uint32 arraySize = (4 * sz) / (attr->Desc->size / 8);

    assert(arraySize >= attr->Desc->arraySize);

    if(!Twiddle(attr, attr->Desc->arraySize))

    const Uint32 arraySize = (4 * sz) / (attr_desc->size / 8);

    assert(arraySize >= attr_desc->arraySize);

    if(!Twiddle(attr_desc, attr_data, attr_desc->arraySize))

      {

	res = -1;

	return NULL;
@@ -405,15 +420,20 @@ RestoreDataIterator::getNextTuple(int & res) { 


  m_count ++;  

  res = 0;

  return tup;

  return &m_tuple;

} // RestoreDataIterator::getNextTuple



BackupFile::BackupFile(){

BackupFile::BackupFile(void (* _free_data_callback)()) 

  : free_data_callback(_free_data_callback)

{

  m_file = 0;

  m_path[0] = 0;

  m_fileName[0] = 0;

  m_buffer = 0;

  m_bufferSize = 0;



  m_buffer_sz = 64*1024;

  m_buffer = malloc(m_buffer_sz);

  m_buffer_ptr = m_buffer;

  m_buffer_data_left = 0;

}



BackupFile::~BackupFile(){
@@ -434,15 +454,54 @@ BackupFile::openFile(){ 
  return m_file != 0;

}



Uint32 *

BackupFile::createBuffer(Uint32 bytes){

  if(bytes > m_bufferSize){

    if(m_buffer != 0)

      free(m_buffer);

    m_bufferSize = m_bufferSize + 2 * bytes;

    m_buffer = (Uint32*)malloc(m_bufferSize);

Uint32 BackupFile::buffer_get_ptr_ahead(void **p_buf_ptr, Uint32 size, Uint32 nmemb)

{

  Uint32 sz = size*nmemb;

  if (sz > m_buffer_data_left) {



    if (free_data_callback)

      (*free_data_callback)();



    memcpy(m_buffer, m_buffer_ptr, m_buffer_data_left);



    size_t r = fread(((char *)m_buffer) + m_buffer_data_left, 1, m_buffer_sz - m_buffer_data_left, m_file);

    m_buffer_data_left += r;

    m_buffer_ptr = m_buffer;



    if (sz > m_buffer_data_left)

      sz = size * (m_buffer_data_left / size);

  }



  *p_buf_ptr = m_buffer_ptr;



  return sz/size;

}

Uint32 BackupFile::buffer_get_ptr(void **p_buf_ptr, Uint32 size, Uint32 nmemb)

{

  Uint32 r = buffer_get_ptr_ahead(p_buf_ptr, size, nmemb);



  m_buffer_ptr = ((char*)m_buffer_ptr)+(r*size);

  m_buffer_data_left -= (r*size);



  return r;

}



Uint32 BackupFile::buffer_read_ahead(void *ptr, Uint32 size, Uint32 nmemb)

{

  void *buf_ptr;

  Uint32 r = buffer_get_ptr_ahead(&buf_ptr, size, nmemb);

  memcpy(ptr, buf_ptr, r*size);



  return r;

}

  return m_buffer;



Uint32 BackupFile::buffer_read(void *ptr, Uint32 size, Uint32 nmemb)

{

  void *buf_ptr;

  Uint32 r = buffer_get_ptr(&buf_ptr, size, nmemb);

  memcpy(ptr, buf_ptr, r*size);



  return r;

}



void
@@ -503,7 +562,7 @@ BackupFile::readHeader(){ 
    return false;

  }

  

  if(fread(&m_fileHeader, sizeof(m_fileHeader), 1, m_file) != 1){

  if(buffer_read(&m_fileHeader, sizeof(m_fileHeader), 1) != 1){

    err << "readDataFileHeader: Error reading header" << endl;

    return false;

  }
@@ -551,14 +610,13 @@ BackupFile::validateFooter(){ 
  return true;

}



bool

RestoreDataIterator::readFragmentHeader(int & ret)

bool RestoreDataIterator::readFragmentHeader(int & ret)

{

  BackupFormat::DataFile::FragmentHeader Header;

  

  debug << "RestoreDataIterator::getNextFragment" << endl;

  

  if (fread(&Header, sizeof(Header), 1, m_file) != 1){

  if (buffer_read(&Header, sizeof(Header), 1) != 1){

    ret = 0;

    return false;

  } // if
@@ -581,6 +639,12 @@ RestoreDataIterator::readFragmentHeader(int & ret) 
    return false;

  }

  

  if(!m_tuple.prepareRecord(*m_currentTable))

  {

    ret =-1;

    return false;

  }



  info << "_____________________________________________________" << endl

       << "Restoring data in table: " << m_currentTable->getTableName() 

       << "(" << Header.TableId << ") fragment "
@@ -588,6 +652,7 @@ RestoreDataIterator::readFragmentHeader(int & ret) 
  

  m_count = 0;

  ret = 0;



  return true;

} // RestoreDataIterator::getNextFragment
@@ -596,7 +661,7 @@ bool 
RestoreDataIterator::validateFragmentFooter() {

  BackupFormat::DataFile::FragmentFooter footer;

  

  if (fread(&footer, sizeof(footer), 1, m_file) != 1){

  if (buffer_read(&footer, sizeof(footer), 1) != 1){

    err << "getFragmentFooter:Error reading fragment footer" << endl;

    return false;

  }
@@ -709,36 +774,23 @@ RestoreLogIterator::getNextLogEntry(int & res) { 
  Uint32 len= ~0;

  const Uint32 stopGCP = m_metaData.getStopGCP();

  do {

    

    if(createBuffer(4) == 0) {

    if (buffer_read_ahead(&len, sizeof(Uint32), 1) != 1){

      res= -1;

      return NULL;

      return 0;

    }

    len= ntohl(len);





    if (fread(m_buffer, sizeof(Uint32), 1, m_file) != 1){

      res = -1;

      return NULL;

    Uint32 data_len = sizeof(Uint32) + len*4;

    if (buffer_get_ptr((void **)(&logE), 1, data_len) != data_len) {

      res= -2;

      return 0;

    }

    

    m_buffer[0] = ntohl(m_buffer[0]);

    len = m_buffer[0];

    if(len == 0){

      res= 0;

      return 0;

    }



    if(createBuffer(4 * (len + 1)) == 0){

      res = -1;

      return NULL;

    }

    

    if (fread(&m_buffer[1], 4, len, m_file) != len) {

      res = -1;

      return NULL;

    }

    

    logE = (LogE *)&m_buffer[0];

    logE->TableId= ntohl(logE->TableId);

    logE->TriggerEvent= ntohl(logE->TriggerEvent);
@@ -751,9 +803,6 @@ RestoreLogIterator::getNextLogEntry(int & res) { 
    }

  } while(gcp > stopGCP + 1);



  for(int i=0; i<m_logEntry.m_values.size();i++)

    delete m_logEntry.m_values[i];

  m_logEntry.m_values.clear();

  m_logEntry.m_table = m_metaData.getTable(logE->TableId);

  switch(logE->TriggerEvent){

  case TriggerEvent::TE_INSERT:
@@ -771,17 +820,19 @@ RestoreLogIterator::getNextLogEntry(int & res) { 
  }



  const TableS * tab = m_logEntry.m_table;

  m_logEntry.clear();



  AttributeHeader * ah = (AttributeHeader *)&logE->Data[0];

  AttributeHeader *end = (AttributeHeader *)&logE->Data[len - 2];

  AttributeS *  attr;

  while(ah < end){

    attr = new AttributeS;

    attr= m_logEntry.add_attr();

    if(attr == NULL) {

      ndbout_c("Restore: Failed to allocate memory");

      res = -1;

      return NULL;

      return 0;

    }



    attr->Desc = (* tab)[ah->getAttributeId()];

    assert(attr->Desc != 0);
@@ -794,8 +845,7 @@ RestoreLogIterator::getNextLogEntry(int & res) { 
      attr->Data.void_value = ah->getDataPtr();

    }

    

    Twiddle(attr);

    m_logEntry.m_values.push_back(attr);

    Twiddle(attr->Desc, &(attr->Data));

    

    ah = ah->getNext();

  }
@@ -804,3 +854,85 @@ RestoreLogIterator::getNextLogEntry(int & res) { 
  res = 0;

  return &m_logEntry;

}



NdbOut &

operator<<(NdbOut& ndbout, const AttributeS& attr){

  const AttributeData & data = attr.Data;

  const AttributeDesc & desc = *(attr.Desc);



  if (data.null)

  {

    ndbout << "<NULL>";

    return ndbout;

  }

  

  NdbRecAttr tmprec;

  tmprec.setup(desc.m_column, (char *)data.void_value);

  ndbout << tmprec;



  return ndbout;

}



// Print tuple data

NdbOut& 

operator<<(NdbOut& ndbout, const TupleS& tuple)

{

  ndbout << tuple.getTable()->getTableName() << "; ";

  for (int i = 0; i < tuple.getNoOfAttributes(); i++) 

  {

    AttributeData * attr_data = tuple.getData(i);

    const AttributeDesc * attr_desc = tuple.getDesc(i);

    const AttributeS attr = {attr_desc, *attr_data};

    debug << i << " " << attr_desc->m_column->getName();

    ndbout << attr;

    

    if (i != (tuple.getNoOfAttributes() - 1))

      ndbout << delimiter << " ";

  } // for

  return ndbout;

}



// Print tuple data

NdbOut& 

operator<<(NdbOut& ndbout, const LogEntry& logE)

{

  switch(logE.m_type)

  {

  case LogEntry::LE_INSERT:

    ndbout << "INSERT " << logE.m_table->getTableName() << " ";

    break;

  case LogEntry::LE_DELETE:

    ndbout << "DELETE " << logE.m_table->getTableName() << " ";

    break;

  case LogEntry::LE_UPDATE:

    ndbout << "UPDATE " << logE.m_table->getTableName() << " ";

    break;

  default:

    ndbout << "Unknown log entry type (not insert, delete or update)" ;

  }

  

  for (Uint32 i= 0; i < logE.size();i++) 

  {

    const AttributeS * attr = logE[i];

    ndbout << attr->Desc->m_column->getName() << "=";

    ndbout << (* attr);

    if (i < (logE.size() - 1))

      ndbout << ", ";

  }

  return ndbout;

}





NdbOut & 

operator<<(NdbOut& ndbout, const TableS & table){

  ndbout << endl << "Table: " << table.getTableName() << endl;

  for (int j = 0; j < table.getNoOfAttributes(); j++) 

  {

    const AttributeDesc * desc = table[j];

    ndbout << desc->m_column->getName() << ": " << desc->m_column->getType();

    ndbout << " key: "  << desc->m_column->getPrimaryKey();

    ndbout << " array: " << desc->arraySize;

    ndbout << " size: " << desc->size << endl;

  } // for

  return ndbout;

}