restore of auto increment bug#5786

6 files changed, 95 insertions, 10 deletions

diff --git a/ndb/src/kernel/blocks/backup/restore/Restore.cpp b/ndb/src/kernel/blocks/backup/restore/Restore.cpp
index f60480ad7bf..9e1ad228ee5 100644
--- a/ndb/src/kernel/blocks/backup/restore/Restore.cpp
+++ b/ndb/src/kernel/blocks/backup/restore/Restore.cpp
@@ -88,7 +88,7 @@ RestoreMetaData::~RestoreMetaData(){
   allTables.clear();
 }
 
-const TableS * 
+TableS * 
 RestoreMetaData::getTable(Uint32 tableId) const {
   for(Uint32 i= 0; i < allTables.size(); i++)
     if(allTables[i]->getTableId() == tableId)
@@ -201,6 +201,8 @@ TableS::TableS(NdbTableImpl* tableImpl)
 {
   m_dictTable = tableImpl;
   m_noOfNullable = m_nullBitmaskSize = 0;
+  m_auto_val_id= ~(Uint32)0;
+  m_max_auto_val= 0;
 
   for (int i = 0; i < tableImpl->getNoOfColumns(); i++)
     createAttr(tableImpl->getColumn(i));
@@ -269,7 +271,7 @@ int TupleS::getNoOfAttributes() const {
   return m_currentTable->getNoOfAttributes();
 };
 
-const TableS * TupleS::getTable() const {
+TableS * TupleS::getTable() const {
   return m_currentTable;
 };
 
@@ -282,7 +284,7 @@ AttributeData * TupleS::getData(int i) const{
 };
 
 bool
-TupleS::prepareRecord(const TableS & tab){
+TupleS::prepareRecord(TableS & tab){
   if (allAttrData) {
     if (getNoOfAttributes() == tab.getNoOfAttributes())
     {
@@ -698,6 +700,9 @@ void TableS::createAttr(NdbDictionary::Column *column)
   d->attrId = allAttributesDesc.size();
   allAttributesDesc.push_back(d);
 
+  if (d->m_column->getAutoIncrement())
+    m_auto_val_id= d->attrId;
+
   if(d->m_column->getPrimaryKey() /* && not variable */)
   {
     m_fixedKeys.push_back(d);
diff --git a/ndb/src/kernel/blocks/backup/restore/Restore.hpp b/ndb/src/kernel/blocks/backup/restore/Restore.hpp
index 5a705740c69..8ca68fd9b23 100644
--- a/ndb/src/kernel/blocks/backup/restore/Restore.hpp
+++ b/ndb/src/kernel/blocks/backup/restore/Restore.hpp
@@ -91,9 +91,9 @@ class TupleS {
 private:
   friend class RestoreDataIterator;
   
-  const TableS *m_currentTable;
+  TableS *m_currentTable;
   AttributeData *allAttrData;
-  bool prepareRecord(const TableS &);
+  bool prepareRecord(TableS &);
   
 public:
   TupleS() {
@@ -108,7 +108,7 @@ public:
   TupleS(const TupleS& tuple); // disable copy constructor
   TupleS & operator=(const TupleS& tuple);
   int getNoOfAttributes() const;
-  const TableS * getTable() const;
+  TableS * getTable() const;
   const AttributeDesc * getDesc(int i) const;
   AttributeData * getData(int i) const;
 }; // class TupleS
@@ -130,6 +130,9 @@ class TableS {
   Uint32 m_noOfNullable;
   Uint32 m_nullBitmaskSize;
 
+  Uint32 m_auto_val_id;
+  Uint64 m_max_auto_val;
+
   int pos;
 
   void createAttr(NdbDictionary::Column *column);
@@ -170,6 +173,42 @@ public:
     return allAttributesDesc.size();
   };
   
+  bool have_auto_inc() const {
+    return m_auto_val_id != ~(Uint32)0;
+  };
+
+  bool have_auto_inc(Uint32 id) const {
+    return m_auto_val_id == id;
+  };
+
+  Uint64 get_max_auto_val() const {
+    return m_max_auto_val;
+  };
+
+  void update_max_auto_val(const char *data, int size) {
+    Uint64 val= 0;
+    switch(size){
+    case 8:
+      val= *(Uint8*)data;
+      break;
+    case 16:
+      val= *(Uint16*)data;
+      break;
+    case 24:
+      val= (0xffffff)&*(Uint32*)data;
+      break;
+    case 32:
+      val= *(Uint32*)data;
+      break;
+    case 64:
+      val= *(Uint64*)data;
+      break;
+    default:
+      return;
+    };
+    if(val > m_max_auto_val)
+      m_max_auto_val= val;
+  };
   /**
    * Get attribute descriptor
    */
@@ -245,7 +284,7 @@ public:
   Uint32 getNoOfTables() const { return allTables.size();}
   
   const TableS * operator[](int i) const { return allTables[i];}
-  const TableS * getTable(Uint32 tableId) const;
+  TableS * getTable(Uint32 tableId) const;
 
   Uint32 getStopGCP() const;
 }; // RestoreMetaData
@@ -254,7 +293,7 @@ public:
 class RestoreDataIterator : public BackupFile {
   const RestoreMetaData & m_metaData;
   Uint32 m_count;
-  const TableS* m_currentTable;
+  TableS* m_currentTable;
   TupleS m_tuple;
 
 public:
@@ -278,7 +317,7 @@ public:
     LE_UPDATE
   };
   EntryType m_type;
-  const TableS * m_table;  
+  TableS * m_table;  
   Vector<AttributeS*> m_values;
   Vector<AttributeS*> m_values_e;
   AttributeS *add_attr() {
diff --git a/ndb/src/kernel/blocks/backup/restore/consumer.hpp b/ndb/src/kernel/blocks/backup/restore/consumer.hpp
index 0b7b517c666..692c814159f 100644
--- a/ndb/src/kernel/blocks/backup/restore/consumer.hpp
+++ b/ndb/src/kernel/blocks/backup/restore/consumer.hpp
@@ -30,6 +30,7 @@ public:
   virtual void endOfTuples(){}
   virtual void logEntry(const LogEntry &){}
   virtual void endOfLogEntrys(){}
+  virtual bool finalize_table(const TableS &){return true;}
 };
 
 #endif
diff --git a/ndb/src/kernel/blocks/backup/restore/consumer_restore.cpp b/ndb/src/kernel/blocks/backup/restore/consumer_restore.cpp
index c132bccee75..fb82570b0b9 100644
--- a/ndb/src/kernel/blocks/backup/restore/consumer_restore.cpp
+++ b/ndb/src/kernel/blocks/backup/restore/consumer_restore.cpp
@@ -131,6 +131,21 @@ BackupRestore::get_table(const NdbDictionary::Table* tab){
 }
 
 bool
+BackupRestore::finalize_table(const TableS & table){
+  bool ret= true;
+  if (!m_restore && !m_restore_meta)
+    return ret;
+  if (table.have_auto_inc())
+  {
+    Uint64 max_val= table.get_max_auto_val();
+    Uint64 auto_val= m_ndb->readAutoIncrementValue(get_table(table.m_dictTable));
+    if (max_val+1 > auto_val || auto_val == ~(Uint64)0)
+      ret= m_ndb->setAutoIncrementValue(get_table(table.m_dictTable), max_val+1, false);
+  }
+  return ret;
+}
+
+bool
 BackupRestore::table(const TableS & table){
   if (!m_restore && !m_restore_meta)
     return true;
@@ -179,6 +194,9 @@ BackupRestore::table(const TableS & table){
     err << "Unable to find table: " << split[2].c_str() << endl;
     return false;
   }
+  if(m_restore_meta){
+    m_ndb->setAutoIncrementValue(tab, ~(Uint64)0, false);
+  }
   const NdbDictionary::Table* null = 0;
   m_new_tables.fill(table.m_dictTable->getTableId(), null);
   m_new_tables[table.m_dictTable->getTableId()] = tab;
@@ -316,6 +334,10 @@ void BackupRestore::tuple_a(restore_callback_t *cb)
 	int arraySize = attr_desc->arraySize;
 	char * dataPtr = attr_data->string_value;
 	Uint32 length = (size * arraySize) / 8;
+
+	if (j == 0 && tup.getTable()->have_auto_inc(i))
+	  tup.getTable()->update_max_auto_val(dataPtr,size);
+
 	if (attr_desc->m_column->getPrimaryKey())
 	{
 	  if (j == 1) continue;
@@ -510,8 +532,11 @@ BackupRestore::logEntry(const LogEntry & tup)
     int arraySize = attr->Desc->arraySize;
     const char * dataPtr = attr->Data.string_value;
     
+    if (tup.m_table->have_auto_inc(attr->Desc->attrId))
+      tup.m_table->update_max_auto_val(dataPtr,size);
+
     const Uint32 length = (size / 8) * arraySize;
-    if (attr->Desc->m_column->getPrimaryKey()) 
+    if (attr->Desc->m_column->getPrimaryKey())
       op->equal(attr->Desc->attrId, dataPtr, length);
     else
       op->setValue(attr->Desc->attrId, dataPtr, length);
diff --git a/ndb/src/kernel/blocks/backup/restore/consumer_restore.hpp b/ndb/src/kernel/blocks/backup/restore/consumer_restore.hpp
index ca336eeea4f..59e2734ea1f 100644
--- a/ndb/src/kernel/blocks/backup/restore/consumer_restore.hpp
+++ b/ndb/src/kernel/blocks/backup/restore/consumer_restore.hpp
@@ -59,6 +59,7 @@ public:
   virtual void endOfTuples();
   virtual void logEntry(const LogEntry &);
   virtual void endOfLogEntrys();
+  virtual bool finalize_table(const TableS &);
   void connectToMysql();
   Ndb * m_ndb;
   bool m_restore;
diff --git a/ndb/src/kernel/blocks/backup/restore/main.cpp b/ndb/src/kernel/blocks/backup/restore/main.cpp
index 5708415c61e..f7b1479cc93 100644
--- a/ndb/src/kernel/blocks/backup/restore/main.cpp
+++ b/ndb/src/kernel/blocks/backup/restore/main.cpp
@@ -355,6 +355,20 @@ main(int argc, const char** argv)
       logIter.validateFooter(); //not implemented
       for (i= 0; i < g_consumers.size(); i++)
 	g_consumers[i]->endOfLogEntrys();
+      for(i = 0; i<metaData.getNoOfTables(); i++)
+      {
+	if (checkSysTable(metaData[i]->getTableName()))
+	{
+	  for(Uint32 j= 0; j < g_consumers.size(); j++)
+	    if (!g_consumers[j]->finalize_table(* metaData[i]))
+	    {
+	      ndbout_c("Restore: Failed to finalize restore table: %s. "
+		       "Exiting...", 
+		       metaData[i]->getTableName());
+	      return -11;
+	    } 
+	}
+      }
     }
   }
   clearConsumers();