Restore.cpp

/*
   Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.

   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 "Restore.hpp"
#include <NdbTCP.h>
#include <NdbMem.h>
#include <OutputStream.hpp>
#include <Bitmask.hpp>

#include <AttributeHeader.hpp>
#include <trigger_definitions.h>
#include <SimpleProperties.hpp>
#include <signaldata/DictTabInfo.hpp>
#include <ndb_limits.h>
#include <NdbAutoPtr.hpp>
#include "../src/ndbapi/NdbDictionaryImpl.hpp"

#include "../../../../sql/ha_ndbcluster_tables.h"
extern NdbRecordPrintFormat g_ndbrecord_print_format;
extern bool ga_skip_unknown_objects;
extern bool ga_skip_broken_objects;

Uint16 Twiddle16(Uint16 in); // Byte shift 16-bit data
Uint32 Twiddle32(Uint32 in); // Byte shift 32-bit data
Uint64 Twiddle64(Uint64 in); // Byte shift 64-bit data


/*
  TwiddleUtil

  Utility class used when swapping byteorder
  of one attribute in a table

*/

class TwiddleUtil {
  Uint32 m_twiddle_size;
  Uint32 m_twiddle_array_size;
public:
  TwiddleUtil(); // Not implemented
  TwiddleUtil(const TwiddleUtil&); // Not implemented

  TwiddleUtil(const AttributeDesc * const attr_desc) {
    const NdbDictionary::Column::Type attribute_type =
      attr_desc->m_column->getType();

    switch(attribute_type){
    case NdbDictionary::Column::Datetime:
      // Datetime is stored as 8x8, should be twiddled as 64 bit
      assert(attr_desc->size == 8);
      assert(attr_desc->arraySize == 8);
      m_twiddle_size = 64;
      m_twiddle_array_size = 1;
      break;

    case NdbDictionary::Column::Timestamp:
      // Timestamp is stored as 4x8, should be twiddled as 32 bit
      assert(attr_desc->size == 8);
      assert(attr_desc->arraySize == 4);
      m_twiddle_size = 32;
      m_twiddle_array_size = 1;
      break;

    case NdbDictionary::Column::Blob:
    case NdbDictionary::Column::Text:
      if (attr_desc->m_column->getArrayType() ==
          NdbDictionary::Column::ArrayTypeFixed)
      {
        // Length of fixed size blob which is stored in first 64 bit's
        // has to be twiddled, the remaining byte stream left as is
        assert(attr_desc->size == 8);
        assert(attr_desc->arraySize > 8);
        m_twiddle_size = 64;
        m_twiddle_array_size = 1;
        break;
      }
      // Fallthrough for blob/text with ArrayTypeVar

    default:
      // Default twiddling parameters
      m_twiddle_size = attr_desc->size;
      m_twiddle_array_size = attr_desc->arraySize;
      break;
    }

    assert(m_twiddle_array_size);
    assert(m_twiddle_size);
  }

  bool is_aligned (void* data_ptr) const {
    switch (m_twiddle_size){
    case 8:
      // Always aligned
      return true;
      break;
    case 16:
      return ((((size_t)data_ptr) & 1) == 0);
      break;
    case 32:
      return ((((size_t)data_ptr) & 3) == 0);
      break;
    case 64:
      return ((((size_t)data_ptr) & 7) == 0);
      break;
    default:
      abort();
      break;
    }
    return false; // Never reached
  }

  void twiddle_aligned(void* const data_ptr) const {
    // Make sure the data pointer is properly aligned
    assert(is_aligned(data_ptr));

    switch(m_twiddle_size){
    case 8:
      // Nothing to swap
      break;
    case 16:
    {
      Uint16* ptr = (Uint16*)data_ptr;
      for (Uint32 i = 0; i < m_twiddle_array_size; i++){
        *ptr = Twiddle16(*ptr);
        ptr++;
      }
      break;
    }
    case 32:
    {
      Uint32* ptr = (Uint32*)data_ptr;
      for (Uint32 i = 0; i < m_twiddle_array_size; i++){
        *ptr = Twiddle32(*ptr);
        ptr++;
      }
      break;
    }
    case 64:
    {
      Uint64* ptr = (Uint64*)data_ptr;
      for (Uint32 i = 0; i < m_twiddle_array_size; i++){
        *ptr = Twiddle64(*ptr);
        ptr++;
      }
      break;
    }
    default:
      abort();
    } // switch
  }
};


/*
  BackupFile::twiddle_attribute

  Swap the byte order of one attribute whose data may or may not
  be properly aligned for the current datatype

*/

void
BackupFile::twiddle_atribute(const AttributeDesc * const attr_desc,
                             AttributeData* attr_data)
{
  TwiddleUtil map(attr_desc);

  // Check if data is aligned properly
  void* data_ptr = (char*)attr_data->void_value;
  Uint32 data_sz = attr_desc->getSizeInBytes();
  bool aligned= map.is_aligned(data_ptr);
  if (!aligned)
  {
    // The pointer is not properly aligned, copy the data
    // to aligned memory before twiddling
    m_twiddle_buffer.assign(data_ptr, data_sz);
    data_ptr = m_twiddle_buffer.get_data();
  }

  // Swap the byteorder of the aligned data
  map.twiddle_aligned(data_ptr);

  if (!aligned)
  {
    // Copy data back from aligned memory
    memcpy(attr_data->void_value,
           m_twiddle_buffer.get_data(),
           data_sz);
  }
}


/*
  BackupFile::Twiddle

  Swap the byteorder for one attribute if it was stored
  in different byteorder than current host

*/

bool
BackupFile::Twiddle(const AttributeDesc * const attr_desc,
                    AttributeData* attr_data)
{
  // Check parameters are not NULL
  assert(attr_desc);
  assert(attr_data);

  // Make sure there is data to fiddle with
  assert(!attr_data->null);
  assert(attr_data->void_value);

  if(unlikely(!m_hostByteOrder))
  {
    // The data file is not in host byte order, the
    // attribute need byte order swapped
    twiddle_atribute(attr_desc, attr_data);
  }
#ifdef VM_TRACE
  else
  {
    // Increase test converage in debug mode by doing
    // a double byte order swap to prove that both ways work
    twiddle_atribute(attr_desc, attr_data);
    twiddle_atribute(attr_desc, attr_data);
  }
#endif

  return true;
}


FilteredNdbOut err(* new FileOutputStream(stderr), 0, 0);
FilteredNdbOut info(* new FileOutputStream(stdout), 1, 1);
FilteredNdbOut debug(* new FileOutputStream(stdout), 2, 0);

// To decide in what byte order data is
const Uint32 magicByteOrder = 0x12345678;
const Uint32 swappedMagicByteOrder = 0x78563412;

RestoreMetaData::RestoreMetaData(const char* path, Uint32 nodeId, Uint32 bNo) {
  
  debug << "RestoreMetaData constructor" << endl;
  setCtlFile(nodeId, bNo, path);
}

RestoreMetaData::~RestoreMetaData(){
  for(Uint32 i= 0; i < allTables.size(); i++)
  {
    TableS *table = allTables[i];
    for(Uint32 j= 0; j < table->m_fragmentInfo.size(); j++)
      delete table->m_fragmentInfo[j];
    delete table;
  }
  allTables.clear();
}

TableS * 
RestoreMetaData::getTable(Uint32 tableId) const {
  for(Uint32 i= 0; i < allTables.size(); i++)
    if(allTables[i]->getTableId() == tableId)
      return allTables[i];
  return NULL;
}

Uint32
RestoreMetaData::getStopGCP() const {
  return m_stopGCP;
}

int
RestoreMetaData::loadContent() 
{
  Uint32 noOfTables = readMetaTableList();
  if(noOfTables == 0) {
    return 1;
  }
  for(Uint32 i = 0; i<noOfTables; i++){
    if(!readMetaTableDesc()){
      return 0;
    }
  }
  if (!markSysTables())
    return 0;
  if (!fixBlobs())
    return 0;
  if(!readGCPEntry())
    return 0;

  if(!readFragmentInfo())
    return 0;
  return 1;
}

Uint32
RestoreMetaData::readMetaTableList() {
  
  Uint32 sectionInfo[2];
  
  if (buffer_read(&sectionInfo, sizeof(sectionInfo), 1) != 1){
    err << "readMetaTableList read header error" << endl;
    return 0;
  }
  sectionInfo[0] = ntohl(sectionInfo[0]);
  sectionInfo[1] = ntohl(sectionInfo[1]);

  const Uint32 tabCount = sectionInfo[1] - 2;

  void *tmp;
  if (buffer_get_ptr(&tmp, 4, tabCount) != tabCount){
    err << "readMetaTableList read tabCount error" << endl;
    return 0;
  }
  
  return tabCount;
}

bool
RestoreMetaData::readMetaTableDesc() {
  
  Uint32 sectionInfo[3];
  
  // Read section header 
  Uint32 sz = sizeof(sectionInfo) >> 2;
  if (m_fileHeader.NdbVersion < NDBD_ROWID_VERSION ||
      isDrop6(m_fileHeader.NdbVersion))
  {
    sz = 2;
    sectionInfo[2] = htonl(DictTabInfo::UserTable);
  }
  if (buffer_read(&sectionInfo, 4*sz, 1) != 1){
    err << "readMetaTableDesc read header error" << endl;
    return false;
  } // if
  sectionInfo[0] = ntohl(sectionInfo[0]);
  sectionInfo[1] = ntohl(sectionInfo[1]);
  sectionInfo[2] = ntohl(sectionInfo[2]);
  
  assert(sectionInfo[0] == BackupFormat::TABLE_DESCRIPTION);
  
  // Read dictTabInfo buffer
  const Uint32 len = (sectionInfo[1] - sz);
  void *ptr;
  if (buffer_get_ptr(&ptr, 4, len) != len){
    err << "readMetaTableDesc read error" << endl;
    return false;
  } // if
  
  int errcode = 0;
  DictObject obj = { sectionInfo[2], 0 };
  switch(obj.m_objType){
  case DictTabInfo::SystemTable:
  case DictTabInfo::UserTable:
  case DictTabInfo::UniqueHashIndex:
  case DictTabInfo::OrderedIndex:
    return parseTableDescriptor((Uint32*)ptr, len);          
    break;
  case DictTabInfo::Tablespace:
  {
    NdbDictionary::Tablespace * dst = new NdbDictionary::Tablespace;
    errcode = 
      NdbDictInterface::parseFilegroupInfo(NdbTablespaceImpl::getImpl(* dst), 
                                           (Uint32*)ptr, len);
    if (errcode)
      delete dst;
    obj.m_objPtr = dst;
    debug << hex << obj.m_objPtr << " " 
           << dec << dst->getObjectId() << " " << dst->getName() << endl;
    break;
  }
  case DictTabInfo::LogfileGroup:
  {
    NdbDictionary::LogfileGroup * dst = new NdbDictionary::LogfileGroup;
    errcode = 
      NdbDictInterface::parseFilegroupInfo(NdbLogfileGroupImpl::getImpl(* dst),
                                           (Uint32*)ptr, len);
    if (errcode)
      delete dst;
    obj.m_objPtr = dst;
    debug << hex << obj.m_objPtr << " " 
           << dec << dst->getObjectId() << " " << dst->getName() << endl;
    break;
  }
  case DictTabInfo::Datafile:
  {
    NdbDictionary::Datafile * dst = new NdbDictionary::Datafile;
    errcode = 
      NdbDictInterface::parseFileInfo(NdbDatafileImpl::getImpl(* dst), 
                                      (Uint32*)ptr, len);
    if (errcode)
      delete dst;
    obj.m_objPtr = dst;
    debug << hex << obj.m_objPtr << " "
           << dec << dst->getObjectId() << " " << dst->getPath() << endl;
    break;
  }
  case DictTabInfo::Undofile:
  {
    NdbDictionary::Undofile * dst = new NdbDictionary::Undofile;
    errcode = 
      NdbDictInterface::parseFileInfo(NdbUndofileImpl::getImpl(* dst), 
                                      (Uint32*)ptr, len);
    if (errcode)
      delete dst;
    obj.m_objPtr = dst;
    debug << hex << obj.m_objPtr << " " 
           << dec << dst->getObjectId() << " " << dst->getPath() << endl;
    break;
  }
  case DictTabInfo::HashMap:
  {
    NdbDictionary::HashMap * dst = new NdbDictionary::HashMap;
    errcode =
      NdbDictInterface::parseHashMapInfo(NdbHashMapImpl::getImpl(* dst),
                                         (Uint32*)ptr, len);
    if (errcode)
      delete dst;
    obj.m_objPtr = dst;

    if (!m_hostByteOrder)
    {
      Vector<Uint32> values;
      Uint32 len = dst->getMapLen();
      Uint32 zero = 0;
      values.fill(len - 1, zero);
      dst->getMapValues(values.getBase(), values.size());
      for (Uint32 i = 0; i<len; i++)
      {
        values[i] = Twiddle16(values[i]);
      }
      dst->setMap(values.getBase(), values.size());
    }

    m_objects.push(obj, 0); // Put first
    return true;
    break;
  }
  default:
    if (ga_skip_unknown_objects)
    {
      info << "Skipping schema object with unknown table type "
           << sectionInfo[2] << endl;
      return true;
    }
    else
    {
      err << "Unsupported table type!! " << sectionInfo[2] << endl;
      return false;
    }
  }
  if (errcode)
  {
    err << "Unable to parse dict info..." 
        << sectionInfo[2] << " " << errcode << endl;
    return false;
  }

  for(Uint32 i = 0; i<m_objects.size(); i++)
  {
    switch(sectionInfo[2]){
    case DictTabInfo::Tablespace:
      if (DictTabInfo::isFile(m_objects[i].m_objType))
      {
        m_objects.push(obj, i);
        goto end;
      }
      break;
    case DictTabInfo::LogfileGroup:
    {
      if (DictTabInfo::isFile(m_objects[i].m_objType) ||
          m_objects[i].m_objType == DictTabInfo::Tablespace)
      {
        m_objects.push(obj, i);
        goto end;
      }
      break;
    }
    default:
      m_objects.push_back(obj);
      goto end;
    }
  }
  m_objects.push_back(obj);
  
end:
  return true;
}

#define OLD_NDB_REP_DB  "cluster"
#define OLD_NDB_APPLY_TABLE "apply_status"
#define OLD_NDB_SCHEMA_TABLE "schema"

bool
RestoreMetaData::markSysTables()
{
  Uint32 i;
  for (i = 0; i < getNoOfTables(); i++) {
    TableS* table = allTables[i];
    table->m_local_id = i;
    const char* tableName = table->getTableName();
    if ( // XXX should use type
        strcmp(tableName, "SYSTAB_0") == 0 ||
        strcmp(tableName, "NDB$EVENTS_0") == 0 ||
        strcmp(tableName, "sys/def/SYSTAB_0") == 0 ||
        strcmp(tableName, "sys/def/NDB$EVENTS_0") == 0 ||
        // index stats tables and indexes
        strncmp(tableName, NDB_INDEX_STAT_PREFIX,
                sizeof(NDB_INDEX_STAT_PREFIX)-1) == 0 ||
        strstr(tableName, "/" NDB_INDEX_STAT_PREFIX) != 0 ||
        /*
          The following is for old MySQL versions,
           before we changed the database name of the tables from
           "cluster_replication" -> "cluster" -> "mysql"
        */
        strcmp(tableName, "cluster_replication/def/" OLD_NDB_APPLY_TABLE) == 0 ||
        strcmp(tableName, OLD_NDB_REP_DB "/def/" OLD_NDB_APPLY_TABLE) == 0 ||
        strcmp(tableName, OLD_NDB_REP_DB "/def/" OLD_NDB_SCHEMA_TABLE) == 0 ||
        strcmp(tableName, NDB_REP_DB "/def/" NDB_APPLY_TABLE) == 0 ||
        strcmp(tableName, NDB_REP_DB "/def/" NDB_SCHEMA_TABLE)== 0 )
    {
      table->m_isSysTable = true;
      if (strcmp(tableName, "SYSTAB_0") == 0 ||
          strcmp(tableName, "sys/def/SYSTAB_0") == 0)
        table->m_isSYSTAB_0 = true;
    }
  }
  for (i = 0; i < getNoOfTables(); i++) {
    TableS* blobTable = allTables[i];
    const char* blobTableName = blobTable->getTableName();
    // yet another match blob
    int cnt, id1, id2;
    char buf[256];
    cnt = sscanf(blobTableName, "%[^/]/%[^/]/NDB$BLOB_%d_%d",
                 buf, buf, &id1, &id2);
    if (cnt == 4) {
      Uint32 j;
      for (j = 0; j < getNoOfTables(); j++) {
        TableS* table = allTables[j];
        if (table->getTableId() == (Uint32) id1) {
          if (table->m_isSysTable)
            blobTable->m_isSysTable = true;
          blobTable->m_main_table = table;
          blobTable->m_main_column_id = id2;
          break;
        }
      }
      if (j == getNoOfTables()) {
        err << "Restore: Bad primary table id in " << blobTableName << endl;
        return false;
      }
    }
  }
  return true;
}

bool
RestoreMetaData::fixBlobs()
{
  Uint32 i;
  for (i = 0; i < getNoOfTables(); i++) {
    TableS* table = allTables[i];
    assert(table->m_dictTable != NULL);
    NdbTableImpl& t = NdbTableImpl::getImpl(*table->m_dictTable);
    const Uint32 noOfBlobs = t.m_noOfBlobs;
    if (noOfBlobs == 0)
      continue;
    Uint32 n = 0;
    Uint32 j;
    for (j = 0; n < noOfBlobs; j++) {
      NdbColumnImpl* c = t.getColumn(j);
      assert(c != NULL);
      if (!c->getBlobType())
        continue;
      // tinyblobs are counted in noOfBlobs...
      n++;
      if (c->getPartSize() == 0)
        continue;
      Uint32 k;
      TableS* blobTable = NULL;
      for (k = 0; k < getNoOfTables(); k++) {
        TableS* tmp = allTables[k];
        if (tmp->m_main_table == table &&
            tmp->m_main_column_id == j) {
          blobTable = tmp;
          break;
        }
      }
      if (blobTable == NULL)
      {
        table->m_broken = true;
        /* Corrupt backup, has main table, but no blob table */
        err << "Table " << table->m_dictTable->getName()
            << " has blob column " << j << " (" 
            << c->m_name.c_str()
            << ") with missing parts table in backup."
            << endl;
        if (ga_skip_broken_objects)
        {
          continue;
        }
        else
        {
          return false;
        }
      }
      assert(blobTable->m_dictTable != NULL);
      NdbTableImpl& bt = NdbTableImpl::getImpl(*blobTable->m_dictTable);
      const char* colName = c->m_blobVersion == 1 ? "DATA" : "NDB$DATA";
      const NdbColumnImpl* bc = bt.getColumn(colName);
      assert(bc != NULL);
      assert(c->m_storageType == NDB_STORAGETYPE_MEMORY);
      c->m_storageType = bc->m_storageType;
    }
  }
  return true;
}

bool
RestoreMetaData::readGCPEntry() {

  BackupFormat::CtlFile::GCPEntry dst;
  
  if(buffer_read(&dst, 1, sizeof(dst)) != sizeof(dst)){
    err << "readGCPEntry read error" << endl;
    return false;
  }
  
  dst.SectionType = ntohl(dst.SectionType);
  dst.SectionLength = ntohl(dst.SectionLength);
  
  if(dst.SectionType != BackupFormat::GCP_ENTRY){
    err << "readGCPEntry invalid format" << endl;
    return false;
  }
  
  dst.StartGCP = ntohl(dst.StartGCP);
  dst.StopGCP = ntohl(dst.StopGCP);
  
  m_startGCP = dst.StartGCP;
  m_stopGCP = dst.StopGCP;
  return true;
}

bool
RestoreMetaData::readFragmentInfo()
{
  BackupFormat::CtlFile::FragmentInfo fragInfo;
  TableS * table = 0;
  Uint32 tableId = RNIL;

  while (buffer_read(&fragInfo, 4, 2) == 2)
  {
    fragInfo.SectionType = ntohl(fragInfo.SectionType);
    fragInfo.SectionLength = ntohl(fragInfo.SectionLength);

    if (fragInfo.SectionType != BackupFormat::FRAGMENT_INFO)
    {
      err << "readFragmentInfo invalid section type: " <<
        fragInfo.SectionType << endl;
      return false;
    }

    if (buffer_read(&fragInfo.TableId, (fragInfo.SectionLength-2)*4, 1) != 1)
    {
      err << "readFragmentInfo invalid section length: " <<
        fragInfo.SectionLength << endl;
      return false;
    }

    fragInfo.TableId = ntohl(fragInfo.TableId);
    if (fragInfo.TableId != tableId)
    {
      tableId = fragInfo.TableId;
      table = getTable(tableId);
    }

    FragmentInfo * tmp = new FragmentInfo;
    tmp->fragmentNo = ntohl(fragInfo.FragmentNo);
    tmp->noOfRecords = ntohl(fragInfo.NoOfRecordsLow) +
      (((Uint64)ntohl(fragInfo.NoOfRecordsHigh)) << 32);
    tmp->filePosLow = ntohl(fragInfo.FilePosLow);
    tmp->filePosHigh = ntohl(fragInfo.FilePosHigh);

    table->m_fragmentInfo.push_back(tmp);
    table->m_noOfRecords += tmp->noOfRecords;
  }
  return true;
}

TableS::TableS(Uint32 version, NdbTableImpl* tableImpl)
  : m_dictTable(tableImpl)
{
  m_dictTable = tableImpl;
  m_noOfNullable = m_nullBitmaskSize = 0;
  m_auto_val_attrib = 0;
  m_max_auto_val= 0;
  m_noOfRecords= 0;
  backupVersion = version;
  m_isSysTable = false;
  m_isSYSTAB_0 = false;
  m_broken = false;
  m_main_table = NULL;
  m_main_column_id = ~(Uint32)0;
  
  for (int i = 0; i < tableImpl->getNoOfColumns(); i++)
    createAttr(tableImpl->getColumn(i));
}

TableS::~TableS()
{
  for (Uint32 i= 0; i < allAttributesDesc.size(); i++)
  {
    if (allAttributesDesc[i]->parameter)
      free(allAttributesDesc[i]->parameter);
    delete allAttributesDesc[i];
  }
}


// Parse dictTabInfo buffer and pushback to to vector storage 
bool
RestoreMetaData::parseTableDescriptor(const Uint32 * data, Uint32 len)
{
  NdbTableImpl* tableImpl = 0;
  int ret = NdbDictInterface::parseTableInfo
    (&tableImpl, data, len, false,
     isDrop6(m_fileHeader.NdbVersion) ? MAKE_VERSION(5,1,2) :
     m_fileHeader.NdbVersion);
  
  if (ret != 0) {
    err << "parseTableInfo " << " failed" << endl;
    return false;
  }
  if(tableImpl == 0)
    return false;

  debug << "parseTableInfo " << tableImpl->getName() << " done" << endl;
  TableS * table = new TableS(m_fileHeader.NdbVersion, tableImpl);
  if(table == NULL) {
    return false;
  }

  debug << "Parsed table id " << table->getTableId() << endl;
  debug << "Parsed table #attr " << table->getNoOfAttributes() << endl;
  debug << "Parsed table schema version not used " << endl;

  debug << "Pushing table " << table->getTableName() << endl;
  debug << "   with " << table->getNoOfAttributes() << " attributes" << endl;
  
  allTables.push_back(table);

  return true;
}

// Constructor
RestoreDataIterator::RestoreDataIterator(const RestoreMetaData & md, void (* _free_data_callback)())
  : BackupFile(_free_data_callback), m_metaData(md)
{
  debug << "RestoreDataIterator constructor" << endl;
  setDataFile(md, 0);

  m_bitfield_storage_len = 8192;
  m_bitfield_storage_ptr = (Uint32*)malloc(4*m_bitfield_storage_len);
  m_bitfield_storage_curr_ptr = m_bitfield_storage_ptr;
  m_row_bitfield_len = 0;
}

RestoreDataIterator::~RestoreDataIterator()
{
  free_bitfield_storage();
}

void
RestoreDataIterator::init_bitfield_storage(const NdbDictionary::Table* tab)
{
  Uint32 len = 0;
  for (Uint32 i = 0; i<(Uint32)tab->getNoOfColumns(); i++)
  {
    if (tab->getColumn(i)->getType() == NdbDictionary::Column::Bit)
    {
      len += (tab->getColumn(i)->getLength() + 31) >> 5;
    }
  }

  m_row_bitfield_len = len;
}

void
RestoreDataIterator::reset_bitfield_storage()
{
  m_bitfield_storage_curr_ptr = m_bitfield_storage_ptr;
}

void
RestoreDataIterator::free_bitfield_storage()
{
  if (m_bitfield_storage_ptr)
    free(m_bitfield_storage_ptr);
  m_bitfield_storage_ptr = 0;
  m_bitfield_storage_curr_ptr = 0;
  m_bitfield_storage_len = 0;
}

Uint32
RestoreDataIterator::get_free_bitfield_storage() const
{
  
  return Uint32((m_bitfield_storage_ptr + m_bitfield_storage_len) - 
    m_bitfield_storage_curr_ptr);
}

Uint32*
RestoreDataIterator::get_bitfield_storage(Uint32 len)
{
  Uint32 * currptr = m_bitfield_storage_curr_ptr;
  Uint32 * nextptr = currptr + len;
  Uint32 * endptr = m_bitfield_storage_ptr + m_bitfield_storage_len;

  if (nextptr <= endptr)
  {
    m_bitfield_storage_curr_ptr = nextptr;
    return currptr;
  }
  
  abort();
  return 0;
}

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();
}

TableS * TupleS::getTable() const {
  return m_currentTable;
}

AttributeDesc * TupleS::getDesc(int i) const {
  return m_currentTable->allAttributesDesc[i];
}

AttributeData * TupleS::getData(int i) const{
  return &(allAttrData[i]);
}

bool
TupleS::prepareRecord(TableS & tab){
  if (allAttrData) {
    if (getNoOfAttributes() == tab.getNoOfAttributes())
    {
      m_currentTable = &tab;
      return true;
    }
    delete [] allAttrData;
    m_currentTable= 0;
  }
  
  allAttrData = new AttributeData[tab.getNoOfAttributes()];
  if (allAttrData == 0)
    return false;
  
  m_currentTable = &tab;

  return true;
}

static
inline
Uint8*
pad(Uint8* src, Uint32 align, Uint32 bitPos)
{
  UintPtr ptr = UintPtr(src);
  switch(align){
  case DictTabInfo::aBit:
  case DictTabInfo::a32Bit:
  case DictTabInfo::a64Bit:
  case DictTabInfo::a128Bit:
    return (Uint8*)(((ptr + 3) & ~(UintPtr)3) + 4 * ((bitPos + 31) >> 5));
charpad:
  case DictTabInfo::an8Bit:
  case DictTabInfo::a16Bit:
    return src + 4 * ((bitPos + 31) >> 5);
  default:
#ifdef VM_TRACE
    abort();
#endif
    goto charpad;
  }
}

const TupleS *
RestoreDataIterator::getNextTuple(int  & res)
{
  if (m_currentTable->backupVersion >= NDBD_RAW_LCP)
  {
    if (m_row_bitfield_len >= get_free_bitfield_storage())
    {
      if (free_data_callback)
        (*free_data_callback)();
      reset_bitfield_storage();
    }
  }
  
  Uint32  dataLength = 0;
  // Read record length
  if (buffer_read(&dataLength, sizeof(dataLength), 1) != 1){
    err << "getNextTuple:Error reading length  of data part" << endl;
    res = -1;
    return NULL;
  } // if
  
  // Convert length from network byte order
  dataLength = ntohl(dataLength);
  const Uint32 dataLenBytes = 4 * dataLength;
  
  if (dataLength == 0) {
    // Zero length for last tuple
    // End of this data fragment
    debug << "End of fragment" << endl;
    res = 0;
    return NULL;
  } // if

  // Read tuple data
  void *_buf_ptr;
  if (buffer_get_ptr(&_buf_ptr, 1, dataLenBytes) != dataLenBytes) {
    err << "getNextTuple:Read error: " << endl;
    res = -1;
    return NULL;
  }

  Uint32 *buf_ptr = (Uint32*)_buf_ptr;
  if (m_currentTable->backupVersion >= NDBD_RAW_LCP)
  {
    res = readTupleData_packed(buf_ptr, dataLength);
  }
  else
  {
    res = readTupleData_old(buf_ptr, dataLength);
  }
  
  if (res)
  {
    return NULL;
  }

  m_count ++;  
  res = 0;
  return &m_tuple;
} // RestoreDataIterator::getNextTuple

TableS *
RestoreDataIterator::getCurrentTable()
{
  return m_currentTable;
}

int
RestoreDataIterator::readTupleData_packed(Uint32 *buf_ptr, 
                                          Uint32 dataLength)
{
  Uint32 * ptr = buf_ptr;
  Uint32 rp = * ptr;
  if(unlikely(!m_hostByteOrder))
    rp = Twiddle32(rp);

  AttributeHeader ah(rp);
  assert(ah.getAttributeId() == AttributeHeader::READ_PACKED);
  Uint32 bmlen = ah.getByteSize();
  assert((bmlen & 3) == 0);
  Uint32 bmlen32 = bmlen / 4;

  if (!m_hostByteOrder)
  {
    for (Uint32 i = 0; i < 1 + bmlen32; i++)
    {
      ptr[i] = Twiddle32(ptr[i]);
    }
  }
  
  const NdbDictionary::Table* tab = m_currentTable->m_dictTable;
  
  // All columns should be present...
  assert(((tab->getNoOfColumns() + 31) >> 5) <= (int)bmlen32);
  
  const Uint32 * bmptr = ptr + 1;
  Uint8* src = (Uint8*)(bmptr + bmlen32);
  Uint32 bmpos = 0;
  Uint32 bitPos = 0;
  for (Uint32 i = 0; i < (Uint32)tab->getNoOfColumns(); i++, bmpos++)
  {
    // All columns should be present
    assert(BitmaskImpl::get(bmlen32, bmptr, bmpos));
    const NdbColumnImpl & col = NdbColumnImpl::getImpl(* tab->getColumn(i));
    AttributeData * attr_data = m_tuple.getData(i);
    const AttributeDesc * attr_desc = m_tuple.getDesc(i);
    if (col.getNullable())
    {
      bmpos++;
      if (BitmaskImpl::get(bmlen32, bmptr, bmpos))
      {
        attr_data->null = true;
        attr_data->void_value = NULL;
        continue;
      }
    }
    
    attr_data->null = false;
    
    Uint32 align = col.m_orgAttrSize;
    Uint32 attrSize = col.m_attrSize;
    Uint32 array = col.m_arraySize;
    Uint32 len = col.m_length;
    Uint32 sz = attrSize * array;
    Uint32 arrayType = col.m_arrayType;
    
    switch(align){
    case DictTabInfo::aBit:{ // Bit
      src = pad(src, 0, 0);
      Uint32* src32 = (Uint32*)src;
      
      Uint32 len32 = (len + 31) >> 5;
      Uint32* tmp = get_bitfield_storage(len32);
      attr_data->null = false;
      attr_data->void_value = tmp;
      attr_data->size = 4*len32;
      
      if (m_hostByteOrder)
      {
        BitmaskImpl::getField(1 + len32, src32, bitPos, len, tmp);
      }
      else
      {
        Uint32 ii;
        for (ii = 0; ii< (1 + len32); ii++)
          src32[ii] = Twiddle32(src32[ii]);
        BitmaskImpl::getField(1 + len32, (Uint32*)src, bitPos, len, tmp);
        for (ii = 0; ii< (1 + len32); ii++)
          src32[ii] = Twiddle32(src32[ii]);
      }
      
      src += 4 * ((bitPos + len) >> 5);
      bitPos = (bitPos + len) & 31;
      goto next;
    }
    default:
      src = pad(src, align, bitPos);
    }
    switch(arrayType){
    case NDB_ARRAYTYPE_FIXED:
      break;
    case NDB_ARRAYTYPE_SHORT_VAR:
      sz = 1 + src[0];
      break;
    case NDB_ARRAYTYPE_MEDIUM_VAR:
      sz = 2 + src[0] + 256 * src[1];
      break;
    default:
      abort();
    }
    
    attr_data->void_value = src;
    attr_data->size = sz;
    
    if(!Twiddle(attr_desc, attr_data))
    {
      return -1;
    }
    
    bitPos = 0;
    src += sz;
next:
    (void)1;
  }
  return 0;
}

int
RestoreDataIterator::readTupleData_old(Uint32 *buf_ptr, 
                                       Uint32 dataLength)
{
  Uint32 * ptr = buf_ptr;
  ptr += m_currentTable->m_nullBitmaskSize;
  Uint32 i;
  for(i= 0; i < m_currentTable->m_fixedKeys.size(); i++){
    assert(ptr < buf_ptr + dataLength);
 
    const Uint32 attrId = m_currentTable->m_fixedKeys[i]->attrId;

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

    const Uint32 sz = attr_desc->getSizeInWords();

    attr_data->null = false;
    attr_data->void_value = ptr;
    attr_data->size = 4*sz;

    if(!Twiddle(attr_desc, attr_data))
    {
      return -1;
    }
    ptr += sz;
  }
  
  for(i = 0; i < m_currentTable->m_fixedAttribs.size(); i++){
    assert(ptr < buf_ptr + dataLength);

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

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

    const Uint32 sz = attr_desc->getSizeInWords();

    attr_data->null = false;
    attr_data->void_value = ptr;
    attr_data->size = 4*sz;

    if(!Twiddle(attr_desc, attr_data))
    {
      return -1;
    }
    
    ptr += sz;
  }

  // init to NULL
  for(i = 0; i < m_currentTable->m_variableAttribs.size(); i++){
    const Uint32 attrId = m_currentTable->m_variableAttribs[i]->attrId;

    AttributeData * attr_data = m_tuple.getData(attrId);
    
    attr_data->null = true;
    attr_data->void_value = NULL;
  }

  int res;
  if (!isDrop6(m_currentTable->backupVersion))
  {
    if ((res = readVarData(buf_ptr, ptr, dataLength)))
      return res;
  }
  else
  {
    if ((res = readVarData_drop6(buf_ptr, ptr, dataLength)))
      return res;
  }

  return 0;
}

int
RestoreDataIterator::readVarData(Uint32 *buf_ptr, Uint32 *ptr,
                                  Uint32 dataLength)
{
  while (ptr + 2 < buf_ptr + dataLength)
  {
    typedef BackupFormat::DataFile::VariableData VarData;
    VarData * data = (VarData *)ptr;
    Uint32 sz = ntohl(data->Sz);
    Uint32 attrId = ntohl(data->Id); // column_no

    AttributeData * attr_data = m_tuple.getData(attrId);
    const AttributeDesc * attr_desc = m_tuple.getDesc(attrId);
    
    // just a reminder - remove when backwards compat implemented
    if (m_currentTable->backupVersion < MAKE_VERSION(5,1,3) && 
        attr_desc->m_column->getNullable())
    {
      const Uint32 ind = attr_desc->m_nullBitIndex;
      if(BitmaskImpl::get(m_currentTable->m_nullBitmaskSize, 
                          buf_ptr,ind))
      {
        attr_data->null = true;
        attr_data->void_value = NULL;
        continue;
      }
    }

    if (m_currentTable->backupVersion < MAKE_VERSION(5,1,3))
    {
      sz *= 4;
    }
    
    attr_data->null = false;
    attr_data->void_value = &data->Data[0];
    attr_data->size = sz;

    //convert the length of blob(v1) and text(v1)
    if(!Twiddle(attr_desc, attr_data))
    {
      return -1;
    }
    
    ptr += ((sz + 3) >> 2) + 2;
  }

  assert(ptr == buf_ptr + dataLength);

  return 0;
}


int
RestoreDataIterator::readVarData_drop6(Uint32 *buf_ptr, Uint32 *ptr,
                                       Uint32 dataLength)
{
  Uint32 i;
  for (i = 0; i < m_currentTable->m_variableAttribs.size(); i++)
  {
    const Uint32 attrId = m_currentTable->m_variableAttribs[i]->attrId;

    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, 
                          buf_ptr,ind))
      {
        attr_data->null = true;
        attr_data->void_value = NULL;
        continue;
      }
    }

    assert(ptr < buf_ptr + dataLength);

    typedef BackupFormat::DataFile::VariableData VarData;
    VarData * data = (VarData *)ptr;
    Uint32 sz = ntohl(data->Sz);
    Uint32 id = ntohl(data->Id);
    assert(id == attrId);

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

    if (!Twiddle(attr_desc, attr_data))
    {
      return -1;
    }
    ptr += (sz + 2);
  }
  assert(ptr == buf_ptr + dataLength);
  return 0;
}

BackupFile::BackupFile(void (* _free_data_callback)()) 
  : free_data_callback(_free_data_callback)
{
  memset(&m_file,0,sizeof(m_file));
  m_path[0] = 0;
  m_fileName[0] = 0;

  m_buffer_sz = 64*1024;
  m_buffer = malloc(m_buffer_sz);
  m_buffer_ptr = m_buffer;
  m_buffer_data_left = 0;

  m_file_size = 0;
  m_file_pos = 0;
  m_is_undolog = false;
}

BackupFile::~BackupFile()
{
  (void)ndbzclose(&m_file);

  if(m_buffer != 0)
    free(m_buffer);
}

bool
BackupFile::openFile(){
  (void)ndbzclose(&m_file);
  m_file_size = 0;
  m_file_pos = 0;

  info.setLevel(254);
  info << "Opening file '" << m_fileName << "'\n";
  int r= ndbzopen(&m_file, m_fileName, O_RDONLY);

  if(r != 1)
    return false;

  size_t size;
  if (ndbz_file_size(&m_file, &size) == 0)
  {
    m_file_size = (Uint64)size;
    info << "File size " << m_file_size << " bytes\n";
  }
  else
  {
    info << "Progress reporting degraded output since fstat failed,"
         << "errno: " << errno << endl;
    m_file_size = 0;
  }

  return true;
}

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)();

    reset_buffers();

    if (m_is_undolog)
    {
      /* move the left data to the end of buffer
       */
      size_t r = 0;
      int error;
      /* move the left data to the end of buffer
       * m_buffer_ptr point the end of the left data. buffer_data_start point the start of left data
       * m_buffer_data_left is the length of left data.
       */
      Uint64 file_left_entry_data = 0;
      Uint32 buffer_free_space = m_buffer_sz - m_buffer_data_left;
      void * buffer_end = (char *)m_buffer + m_buffer_sz;
      void * buffer_data_start = (char *)m_buffer_ptr - m_buffer_data_left;

      memmove((char *)buffer_end - m_buffer_data_left, buffer_data_start, m_buffer_data_left);
      buffer_data_start = (char *)buffer_end - m_buffer_data_left;
      /*
       * For undo log file we should read log entris backwards from log file.
       *   That mean the first entries should start at sizeof(m_fileHeader).
       *   The end of the last entries should be the end of log file(EOF-1).
       * If ther are entries left in log file to read.
       *   m_file_pos should bigger than sizeof(m_fileHeader).
       * If the length of left log entries less than the residual length of buffer,
       *   we just need to read all the left entries from log file into the buffer.
       *   and all the left entries in log file should been read into buffer. Or
       * If the length of left entries is bigger than the residual length of buffer,
       *   we should fill the buffer because the current buffer can't contain
           all the left log entries, we should read more times.
       * 
       */
      if (m_file_pos > sizeof(m_fileHeader))
      {
        /*
         * We read(consume) data from the end of the buffer.
         * If the left data is not enough for next read in buffer,
         *   we move the residual data to the end of buffer.
         *   Then we will fill the start of buffer with new data from log file.
         * eg. If the buffer length is 10. "+" denotes useless content.
         *                          top        end
         *   Bytes in file        abcdefgh0123456789
         *   Byte in buffer       0123456789             --after first read
         *   Consume datas...     (6789) (2345)
         *   Bytes in buffer      01++++++++             --after several consumes
         *   Move data to end     ++++++++01
         *   Bytes in buffer      abcdefgh01             --after second read
         */
        file_left_entry_data = m_file_pos - sizeof(m_fileHeader);
        if (file_left_entry_data <= buffer_free_space)
        {
          /* All remaining data fits in space available in buffer. 
           * Read data into buffer before existing data.
           */
          // Move to the start of data to be read
          ndbzseek(&m_file, sizeof(m_fileHeader), SEEK_SET);
          r = ndbzread(&m_file,
                       (char *)buffer_data_start - file_left_entry_data,
                       Uint32(file_left_entry_data),
                       &error);
          //move back
          ndbzseek(&m_file, sizeof(m_fileHeader), SEEK_SET);
        }
        else
        {
          // Fill remaing space at start of buffer with data from file.
          ndbzseek(&m_file, m_file_pos-buffer_free_space, SEEK_SET);
          r = ndbzread(&m_file, ((char *)m_buffer), buffer_free_space, &error);
          ndbzseek(&m_file, m_file_pos-buffer_free_space, SEEK_SET);
        }
      }
      m_file_pos -= r;
      m_buffer_data_left += (Uint32)r;
      //move to the end of buffer
      m_buffer_ptr = buffer_end;
    }
    else
    {
      memmove(m_buffer, m_buffer_ptr, m_buffer_data_left);
      int error;
      Uint32 r = ndbzread(&m_file,
                          ((char *)m_buffer) + m_buffer_data_left,
                          m_buffer_sz - m_buffer_data_left, &error);
      m_file_pos += r;
      m_buffer_data_left += r;
      m_buffer_ptr = m_buffer;
    }

    if (sz > m_buffer_data_left)
      sz = size * (m_buffer_data_left / size);
  }

  /*
   * For undolog, the m_buffer_ptr points to the end of the left data.
   * After we get data from the end of buffer, the data-end move forward.
   *   So we should move m_buffer_ptr to the right place.
   */
  if(m_is_undolog)
    *p_buf_ptr = (char *)m_buffer_ptr - sz;
  else
    *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);

  if(m_is_undolog)
  {
    /* we read from end of buffer to start of buffer.
     * m_buffer_ptr keep at the end of real data in buffer.
     */
    m_buffer_ptr = ((char*)m_buffer_ptr)-(r*size);
    m_buffer_data_left -= (r*size);
  }
  else
  {
    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;
}

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
BackupFile::setCtlFile(Uint32 nodeId, Uint32 backupId, const char * path){
  m_nodeId = nodeId;
  m_expectedFileHeader.BackupId = backupId;
  m_expectedFileHeader.FileType = BackupFormat::CTL_FILE;

  char name[PATH_MAX]; const Uint32 sz = sizeof(name);
  BaseString::snprintf(name, sz, "BACKUP-%u.%d.ctl", backupId, nodeId);  
  setName(path, name);
}

void
BackupFile::setDataFile(const BackupFile & bf, Uint32 no){
  m_nodeId = bf.m_nodeId;
  m_expectedFileHeader = bf.m_fileHeader;
  m_expectedFileHeader.FileType = BackupFormat::DATA_FILE;
  
  char name[PATH_MAX]; const Uint32 sz = sizeof(name);
  BaseString::snprintf(name, sz, "BACKUP-%u-%d.%d.Data", 
           m_expectedFileHeader.BackupId, no, m_nodeId);
  setName(bf.m_path, name);
}

void
BackupFile::setLogFile(const BackupFile & bf, Uint32 no){
  m_nodeId = bf.m_nodeId;
  m_expectedFileHeader = bf.m_fileHeader;
  m_expectedFileHeader.FileType = BackupFormat::LOG_FILE;
  
  char name[PATH_MAX]; const Uint32 sz = sizeof(name);
  BaseString::snprintf(name, sz, "BACKUP-%u.%d.log", 
           m_expectedFileHeader.BackupId, m_nodeId);
  setName(bf.m_path, name);
}

void
BackupFile::setName(const char * p, const char * n){
  const Uint32 sz = sizeof(m_path);
  if(p != 0 && strlen(p) > 0){
    if(p[strlen(p)-1] == DIR_SEPARATOR[0]){
      BaseString::snprintf(m_path, sz, "%s", p);
    } else {
      BaseString::snprintf(m_path, sz, "%s%s", p, DIR_SEPARATOR);
    }
  } else {
    m_path[0] = 0;
  }

  BaseString::snprintf(m_fileName, sizeof(m_fileName), "%s%s", m_path, n);
  debug << "Filename = " << m_fileName << endl;
}

bool
BackupFile::readHeader(){
  if(!openFile()){
    return false;
  }
  
  Uint32 oldsz = sizeof(BackupFormat::FileHeader_pre_backup_version);
  if(buffer_read(&m_fileHeader, oldsz, 1) != 1){
    err << "readDataFileHeader: Error reading header" << endl;
    return false;
  }
  
  // Convert from network to host byte order for platform compatibility
  /*
    Due to some optimization going on when using gcc 4.2.3 we
    have to read 'backup_version' into tmp variable. If
    'm_fileHeader.BackupVersion' is used directly in the if statement
    below it will have the wrong value.
  */
  Uint32 backup_version = ntohl(m_fileHeader.BackupVersion);
  m_fileHeader.BackupVersion = backup_version;
  m_fileHeader.SectionType = ntohl(m_fileHeader.SectionType);
  m_fileHeader.SectionLength = ntohl(m_fileHeader.SectionLength);
  m_fileHeader.FileType = ntohl(m_fileHeader.FileType);
  m_fileHeader.BackupId = ntohl(m_fileHeader.BackupId);
  m_fileHeader.BackupKey_0 = ntohl(m_fileHeader.BackupKey_0);
  m_fileHeader.BackupKey_1 = ntohl(m_fileHeader.BackupKey_1);

  if (backup_version >= NDBD_RAW_LCP)
  {
    if (buffer_read(&m_fileHeader.NdbVersion, 
                    sizeof(m_fileHeader) - oldsz, 1) != 1)
    {
      err << "readDataFileHeader: Error reading header" << endl;
      return false;
    }
    
    m_fileHeader.NdbVersion = ntohl(m_fileHeader.NdbVersion);
    m_fileHeader.MySQLVersion = ntohl(m_fileHeader.MySQLVersion);
  }
  else
  {
    m_fileHeader.NdbVersion = m_fileHeader.BackupVersion;
    m_fileHeader.MySQLVersion = 0;
  }
  
  debug << "FileHeader: " << m_fileHeader.Magic << " " <<
    m_fileHeader.BackupVersion << " " <<
    m_fileHeader.SectionType << " " <<
    m_fileHeader.SectionLength << " " <<
    m_fileHeader.FileType << " " <<
    m_fileHeader.BackupId << " " <<
    m_fileHeader.BackupKey_0 << " " <<
    m_fileHeader.BackupKey_1 << " " <<
    m_fileHeader.ByteOrder << endl;
  
  debug << "ByteOrder is " << m_fileHeader.ByteOrder << endl;
  debug << "magicByteOrder is " << magicByteOrder << endl;
  

  if (m_fileHeader.FileType != m_expectedFileHeader.FileType &&
      !(m_expectedFileHeader.FileType == BackupFormat::LOG_FILE &&
      m_fileHeader.FileType == BackupFormat::UNDO_FILE)){
    // UNDO_FILE will do in case where we expect LOG_FILE
    abort();
  }
  
  if(m_fileHeader.FileType == BackupFormat::UNDO_FILE){
      m_is_undolog = true;
      /* move pointer to end of data part. 
         move 4 bytes from the end of file 
         because footer contain 4 bytes 0 at the end of file.
         we discard the remain data stored in m_buffer.
      */
      size_t size;
      if (ndbz_file_size(&m_file, &size) == 0)
        m_file_size = (Uint64)size;
      ndbzseek(&m_file, 4, SEEK_END);
      m_file_pos = m_file_size - 4;
      m_buffer_data_left = 0;
      m_buffer_ptr = m_buffer;
  }

  // Check for BackupFormat::FileHeader::ByteOrder if swapping is needed
  if (m_fileHeader.ByteOrder == magicByteOrder) {
    m_hostByteOrder = true;
  } else if (m_fileHeader.ByteOrder == swappedMagicByteOrder){
    m_hostByteOrder = false;
  } else {
    abort();
  }
  
  return true;
} // BackupFile::readHeader

bool
BackupFile::validateFooter(){
  return true;
}

bool RestoreDataIterator::readFragmentHeader(int & ret, Uint32 *fragmentId)
{
  BackupFormat::DataFile::FragmentHeader Header;
  
  debug << "RestoreDataIterator::getNextFragment" << endl;
  
  while (1)
  {
    /* read first part of header */
    if (buffer_read(&Header, 8, 1) != 1)
    {
      ret = 0;
      return false;
    } // if

    /* skip if EMPTY_ENTRY */
    Header.SectionType  = ntohl(Header.SectionType);
    Header.SectionLength  = ntohl(Header.SectionLength);
    if (Header.SectionType == BackupFormat::EMPTY_ENTRY)
    {
      void *tmp;
      if (Header.SectionLength < 2)
      {
        err << "getFragmentFooter:Error reading fragment footer" << endl;
        return false;
      }
      if (Header.SectionLength > 2)
        buffer_get_ptr(&tmp, Header.SectionLength*4-8, 1);
      continue;
    }
    break;
  }
  /* read rest of header */
  if (buffer_read(((char*)&Header)+8, Header.SectionLength*4-8, 1) != 1)
  {
    ret = 0;
    return false;
  }
  Header.TableId  = ntohl(Header.TableId);
  Header.FragmentNo  = ntohl(Header.FragmentNo);
  Header.ChecksumType  = ntohl(Header.ChecksumType);
  
  debug << "FragmentHeader: " << Header.SectionType 
        << " " << Header.SectionLength 
        << " " << Header.TableId 
        << " " << Header.FragmentNo 
        << " " << Header.ChecksumType << endl;
  
  m_currentTable = m_metaData.getTable(Header.TableId);
  if(m_currentTable == 0){
    ret = -1;
    return false;
  }
  
  if(!m_tuple.prepareRecord(*m_currentTable))
  {
    ret =-1;
    return false;
  }

  init_bitfield_storage(m_currentTable->m_dictTable);

  info.setLevel(254);
  info << "_____________________________________________________" << endl
       << "Processing data in table: " << m_currentTable->getTableName() 
       << "(" << Header.TableId << ") fragment " 
       << Header.FragmentNo << endl;
  
  m_count = 0;
  ret = 0;
  *fragmentId = Header.FragmentNo;
  return true;
} // RestoreDataIterator::getNextFragment


bool
RestoreDataIterator::validateFragmentFooter() {
  BackupFormat::DataFile::FragmentFooter footer;
  
  if (buffer_read(&footer, sizeof(footer), 1) != 1){
    err << "getFragmentFooter:Error reading fragment footer" << endl;
    return false;
  } 
  
  // TODO: Handle footer, nothing yet
  footer.SectionType  = ntohl(footer.SectionType);
  footer.SectionLength  = ntohl(footer.SectionLength);
  footer.TableId  = ntohl(footer.TableId);
  footer.FragmentNo  = ntohl(footer.FragmentNo);
  footer.NoOfRecords  = ntohl(footer.NoOfRecords);
  footer.Checksum  = ntohl(footer.Checksum);

  assert(m_count == footer.NoOfRecords);
  
  return true;
} // RestoreDataIterator::getFragmentFooter

AttributeDesc::AttributeDesc(NdbDictionary::Column *c)
  : m_column(c), truncation_detected(false)
{
  size = 8*NdbColumnImpl::getImpl(* c).m_attrSize;
  arraySize = NdbColumnImpl::getImpl(* c).m_arraySize;
}

void TableS::createAttr(NdbDictionary::Column *column)
{
  AttributeDesc * d = new AttributeDesc(column);
  if(d == NULL) {
    ndbout_c("Restore: Failed to allocate memory");
    abort();
  }
  d->attrId = allAttributesDesc.size();
  d->convertFunc = NULL;
  d->parameter = NULL;
  d->m_exclude = false;
  allAttributesDesc.push_back(d);

  if (d->m_column->getAutoIncrement())
    m_auto_val_attrib = d;

  if(d->m_column->getPrimaryKey() && backupVersion <= MAKE_VERSION(4,1,7))
  {
    m_fixedKeys.push_back(d);
    return;
  }
  
  if (d->m_column->getArrayType() == NDB_ARRAYTYPE_FIXED &&
      ! d->m_column->getNullable())
  {
    m_fixedAttribs.push_back(d);
    return;
  }

  // just a reminder - does not solve backwards compat
  if (backupVersion < MAKE_VERSION(5,1,3) || isDrop6(backupVersion))
  {
    d->m_nullBitIndex = m_noOfNullable; 
    m_noOfNullable++;
    m_nullBitmaskSize = (m_noOfNullable + 31) / 32;
  }
  m_variableAttribs.push_back(d);
} // TableS::createAttr

bool
TableS::get_auto_data(const TupleS & tuple, Uint32 * syskey, Uint64 * nextid) const
{
  /*
    Read current (highest) auto_increment value for
    a table. Currently there can only be one per table.
    The values are stored in sustable SYSTAB_0 as
    {SYSKEY,NEXTID} values where SYSKEY (32-bit) is
    the table_id and NEXTID (64-bit) is the next auto_increment
    value in the sequence (note though that sequences of
    values can have been fetched and that are cached in NdbAPI).
    SYSTAB_0 can contain other data so we need to check that
    the found SYSKEY value is a valid table_id (< 0x10000000).
   */
  AttributeData * attr_data = tuple.getData(0);
  AttributeDesc * attr_desc = tuple.getDesc(0);
  const AttributeS attr1 = {attr_desc, *attr_data};
  memcpy(syskey ,attr1.Data.u_int32_value, sizeof(Uint32));
  attr_data = tuple.getData(1);
  attr_desc = tuple.getDesc(1);
  const AttributeS attr2 = {attr_desc, *attr_data};
  memcpy(nextid, attr2.Data.u_int64_value, sizeof(Uint64));
  if (*syskey < 0x10000000)
  {
    return true;
  }
  return false;
};

Uint16 Twiddle16(Uint16 in)
{
  Uint16 retVal = 0;

  retVal = ((in & 0xFF00) >> 8) |
    ((in & 0x00FF) << 8);

  return(retVal);
} // Twiddle16

Uint32 Twiddle32(Uint32 in)
{
  Uint32 retVal = 0;

  retVal = ((in & 0x000000FF) << 24) | 
    ((in & 0x0000FF00) << 8)  |
    ((in & 0x00FF0000) >> 8)  |
    ((in & 0xFF000000) >> 24);
  
  return(retVal);
} // Twiddle32

Uint64 Twiddle64(Uint64 in)
{
  Uint64 retVal = 0;

  retVal = 
    ((in & (Uint64)0x00000000000000FFLL) << 56) | 
    ((in & (Uint64)0x000000000000FF00LL) << 40) | 
    ((in & (Uint64)0x0000000000FF0000LL) << 24) | 
    ((in & (Uint64)0x00000000FF000000LL) << 8) | 
    ((in & (Uint64)0x000000FF00000000LL) >> 8) | 
    ((in & (Uint64)0x0000FF0000000000LL) >> 24) | 
    ((in & (Uint64)0x00FF000000000000LL) >> 40) | 
    ((in & (Uint64)0xFF00000000000000LL) >> 56);

  return(retVal);
} // Twiddle64


RestoreLogIterator::RestoreLogIterator(const RestoreMetaData & md)
  : m_metaData(md) 
{
  debug << "RestoreLog constructor" << endl;
  setLogFile(md, 0);

  m_count = 0;
  m_last_gci = 0;
}

const LogEntry *
RestoreLogIterator::getNextLogEntry(int & res) {
  // Read record length
  const Uint32 stopGCP = m_metaData.getStopGCP();
  Uint32 tableId;
  Uint32 triggerEvent;
  Uint32 frag_id;
  Uint32 *attr_data;
  Uint32 attr_data_len;
  do {
    Uint32 len;
    Uint32 *logEntryPtr;
    if(m_is_undolog){
      int read_result = 0;
      read_result = buffer_read(&len, sizeof(Uint32), 1);
      //no more log data to read
      if (read_result == 0 ) {
        res = 0;
        return 0;
      }
      if (read_result != 1) {
        res= -1;
        return 0;
      }
    }
    else{
      if (buffer_read_ahead(&len, sizeof(Uint32), 1) != 1){
        res= -1;
        return 0;
      }
    }
    len= ntohl(len);

    Uint32 data_len = sizeof(Uint32) + len*4;
    if (buffer_get_ptr((void **)(&logEntryPtr), 1, data_len) != data_len) {
      res= -2;
      return 0;
    }
    
    if(len == 0){
      res= 0;
      return 0;
    }

    if (unlikely(m_metaData.getFileHeader().NdbVersion < NDBD_FRAGID_VERSION ||
                 isDrop6(m_metaData.getFileHeader().NdbVersion)))
    {
      /*
        FragId was introduced in LogEntry in version
        5.1.6
        We set FragId to 0 in older versions (these versions
        do not support restore of user defined partitioned
        tables.
      */
      typedef BackupFormat::LogFile::LogEntry_no_fragid LogE_no_fragid;
      LogE_no_fragid * logE_no_fragid= (LogE_no_fragid *)logEntryPtr;
      tableId= ntohl(logE_no_fragid->TableId);
      triggerEvent= ntohl(logE_no_fragid->TriggerEvent);
      frag_id= 0;
      attr_data= &logE_no_fragid->Data[0];
      attr_data_len= len - ((offsetof(LogE_no_fragid, Data) >> 2) - 1);
    }
    else /* normal case */
    {
      typedef BackupFormat::LogFile::LogEntry LogE;
      LogE * logE= (LogE *)logEntryPtr;
      tableId= ntohl(logE->TableId);
      triggerEvent= ntohl(logE->TriggerEvent);
      frag_id= ntohl(logE->FragId);
      attr_data= &logE->Data[0];
      attr_data_len= len - ((offsetof(LogE, Data) >> 2) - 1);
    }
    
    const bool hasGcp= (triggerEvent & 0x10000) != 0;
    triggerEvent &= 0xFFFF;

    if(hasGcp){
      // last attr_data is gci info
      attr_data_len--;
      m_last_gci = ntohl(*(attr_data + attr_data_len));
    }
  } while(m_last_gci > stopGCP + 1);

  m_logEntry.m_table = m_metaData.getTable(tableId);
  /* We should 'invert' the operation type when we restore an Undo log.
   *   To undo an insert operation, a delete is required.
   *   To undo a delete operation, an insert is required.
   * The backup have collected 'before values' for undoing 'delete+update' to make this work.
   * To undo insert, we only need primary key.
   */
  switch(triggerEvent){
  case TriggerEvent::TE_INSERT:
    if(m_is_undolog)
      m_logEntry.m_type = LogEntry::LE_DELETE;
    else
      m_logEntry.m_type = LogEntry::LE_INSERT;
    break;
  case TriggerEvent::TE_UPDATE:
    m_logEntry.m_type = LogEntry::LE_UPDATE;
    break;
  case TriggerEvent::TE_DELETE:
    if(m_is_undolog)
      m_logEntry.m_type = LogEntry::LE_INSERT;
    else
      m_logEntry.m_type = LogEntry::LE_DELETE;
    break;
  default:
    res = -1;
    return NULL;
  }

  const TableS * tab = m_logEntry.m_table;
  m_logEntry.clear();

  AttributeHeader * ah = (AttributeHeader *)attr_data;
  AttributeHeader *end = (AttributeHeader *)(attr_data + attr_data_len);
  AttributeS * attr;
  m_logEntry.m_frag_id = frag_id;
  while(ah < end){
    attr = m_logEntry.add_attr();
    if(attr == NULL) {
      ndbout_c("Restore: Failed to allocate memory");
      res = -1;
      return 0;
    }

    if(unlikely(!m_hostByteOrder))
      *(Uint32*)ah = Twiddle32(*(Uint32*)ah);

    attr->Desc = tab->getAttributeDesc(ah->getAttributeId());
    assert(attr->Desc != 0);

    const Uint32 sz = ah->getDataSize();
    if(sz == 0){
      attr->Data.null = true;
      attr->Data.void_value = NULL;
    } else {
      attr->Data.null = false;
      attr->Data.void_value = ah->getDataPtr();
      Twiddle(attr->Desc, &(attr->Data));
    }
    
    
    ah = ah->getNext();
  }

  m_count ++;
  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 << g_ndbrecord_print_format.null_string;
    return ndbout;
  }
  
  NdbRecAttr tmprec(0);
  tmprec.setup(desc.m_column, 0);
  Uint32 length = (desc.size)/8 * (desc.arraySize);
  tmprec.receive_data((Uint32*)data.void_value, length);

  ndbrecattr_print_formatted(ndbout, tmprec, g_ndbrecord_print_format);

  return ndbout;
}

// Print tuple data
NdbOut& 
operator<<(NdbOut& ndbout, const TupleS& tuple)
{
  for (int i = 0; i < tuple.getNoOfAttributes(); i++) 
  {
    if (i > 0)
      ndbout << g_ndbrecord_print_format.fields_terminated_by;
    AttributeData * attr_data = tuple.getData(i);
    AttributeDesc * attr_desc = tuple.getDesc(i);
    const AttributeS attr = {attr_desc, *attr_data};
    debug << i << " " << attr_desc->m_column->getName();
    ndbout << attr;
  } // 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;
}

#include <NDBT.hpp>

NdbOut & 
operator<<(NdbOut& ndbout, const TableS & table){
  
  ndbout << (* (NDBT_Table*)table.m_dictTable) << endl;
  return ndbout;
}

template class Vector<TableS*>;
template class Vector<AttributeS*>;
template class Vector<AttributeDesc*>;
template class Vector<FragmentInfo*>;
template class Vector<DictObject>;