Ndb backup/restore also handle indexes

ndb/src/kernel/blocks/backup/Backup.cpp:
  Save meta data for indexes (but not actual data)
ndb/src/kernel/blocks/backup/Backup.hpp:
  Save meta data for indexes (but not actual data)
ndb/src/kernel/blocks/backup/restore/consumer.hpp:
  Introduced endOfTables into Consumer interface
ndb/src/kernel/blocks/backup/restore/consumer_restore.cpp:
  Store indexes and create then when endOfTables is called
ndb/src/kernel/blocks/backup/restore/consumer_restore.hpp:
  Store indexes and create then when endOfTables is called
ndb/src/kernel/blocks/backup/restore/main.cpp:
  Run endOfTables
ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  Split getIndexImpl into 2 methods
  (one being used by restore)
ndb/src/ndbapi/NdbDictionaryImpl.hpp:
  Split getIndexImpl into 2 methods
  (one being used by restore)

ndb/src/kernel/blocks/backup/Backup.cpp

@@ -885,7 +885,7 @@ Backup::execBACKUP_REQ(Signal* signal)
   }//if
 
   ndbrequire(ptr.p->pages.empty());
-  ndbrequire(ptr.p->tables.empty());
+  ndbrequire(ptr.p->tables.isEmpty());
   
   ptr.p->masterData.state.forceState(INITIAL);
   ptr.p->masterData.state.setState(DEFINING);
@@ -2484,8 +2484,7 @@ Backup::execLIST_TABLES_CONF(Signal* signal)
     jam();
     Uint32 tableId = ListTablesConf::getTableId(conf->tableData[i]);
     Uint32 tableType = ListTablesConf::getTableType(conf->tableData[i]);
-    if (tableType != DictTabInfo::SystemTable &&
-        tableType != DictTabInfo::UserTable) {
+    if (!DictTabInfo::isTable(tableType) && !DictTabInfo::isIndex(tableType)){
       jam();
       continue;
     }//if
@@ -2864,7 +2863,12 @@ Backup::execGET_TABINFO_CONF(Signal* signal)
     return;
   }//if
 
+  TablePtr tmp = tabPtr;
   ptr.p->tables.next(tabPtr);
+  if(DictTabInfo::isIndex(tmp.p->tableType)){
+    ptr.p->tables.release(tmp);
+  }
+  
   if(tabPtr.i == RNIL) {
     jam();
     
@@ -2906,6 +2910,10 @@ Backup::parseTableDescription(Signal* signal, BackupRecordPtr ptr, Uint32 len)
   
   TablePtr tabPtr;
   ndbrequire(findTable(ptr, tabPtr, tmpTab.TableId));
+  if(DictTabInfo::isIndex(tabPtr.p->tableType)){
+    jam();
+    return tabPtr;
+  }
   
   /**
    * Initialize table object

ndb/src/kernel/blocks/backup/Backup.hpp

@@ -441,7 +441,7 @@ public:
     Uint32 startGCP;
     Uint32 currGCP;
     Uint32 stopGCP;
-    SLList<Table> tables;
+    DLList<Table> tables;
     SLList<TriggerRecord> triggers;
     
     SLList<BackupFile> files; 

ndb/src/kernel/blocks/backup/restore/consumer.hpp

@@ -24,6 +24,7 @@ public:
   virtual ~BackupConsumer() { }
   virtual bool init() { return true;}
   virtual bool table(const TableS &){return true;}
+  virtual bool endOfTables() { return true; }
   virtual void tuple(const TupleS &){}
   virtual void tuple_free(){}
   virtual void endOfTuples(){}

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

@@ -16,6 +16,7 @@
 
 #include "consumer_restore.hpp"
 #include <NdbSleep.h>
+#include <NdbDictionaryImpl.hpp>
 
 extern FilteredNdbOut err;
 extern FilteredNdbOut info;
@@ -142,6 +143,12 @@ BackupRestore::table(const TableS & table){
   if(match_blob(name) >= 0)
     return true;
   
+  const NdbTableImpl & tmptab = NdbTableImpl::getImpl(* table.m_dictTable);
+  if(tmptab.m_indexType != NdbDictionary::Index::Undefined){
+    m_indexes.push_back(table.m_dictTable);
+    return true;
+  }
+  
   BaseString tmp(name);
   Vector<BaseString> split;
   if(tmp.split(split, "/") != 3){
@@ -178,6 +185,65 @@ BackupRestore::table(const TableS & table){
   return true;
 }
 
+bool
+BackupRestore::endOfTables(){
+  if(!m_restore_meta)
+    return true;
+
+  NdbDictionary::Dictionary* dict = m_ndb->getDictionary();
+  for(size_t i = 0; i<m_indexes.size(); i++){
+    const NdbTableImpl & indtab = NdbTableImpl::getImpl(* m_indexes[i]);
+
+    BaseString tmp(indtab.m_primaryTable.c_str());
+    Vector<BaseString> split;
+    if(tmp.split(split, "/") != 3){
+      err << "Invalid table name format " << indtab.m_primaryTable.c_str()
+	  << endl;
+      return false;
+    }
+    
+    m_ndb->setDatabaseName(split[0].c_str());
+    m_ndb->setSchemaName(split[1].c_str());
+    
+    const NdbDictionary::Table * prim = dict->getTable(split[2].c_str());
+    if(prim == 0){
+      err << "Unable to find base table \"" << split[2].c_str() 
+	  << "\" for index "
+	  << indtab.getName() << endl;
+      return false;
+    }
+    NdbTableImpl& base = NdbTableImpl::getImpl(*prim);
+    NdbIndexImpl* idx;
+    int id;
+    char idxName[255], buf[255];
+    if(sscanf(indtab.getName(), "%[^/]/%[^/]/%d/%s",
+	      buf, buf, &id, idxName) != 4){
+      err << "Invalid index name format " << indtab.getName() << endl;
+      return false;
+    }
+    if(NdbDictInterface::create_index_obj_from_table(&idx, &indtab, &base))
+    {
+      err << "Failed to create index " << idxName
+	  << " on " << split[2].c_str() << endl;
+	return false;
+    }
+    idx->setName(idxName);
+    if(dict->createIndex(* idx) != 0)
+    {
+      delete idx;
+      err << "Failed to create index " << idxName
+	  << " on " << split[2].c_str() << endl
+	  << dict->getNdbError() << endl;
+
+      return false;
+    }
+    delete idx;
+    info << "Successfully created index " << idxName
+	 << " on " << split[2].c_str() << endl;
+  }
+  return true;
+}
+
 void BackupRestore::tuple(const TupleS & tup)
 {
   if (!m_restore) 

ndb/src/kernel/blocks/backup/restore/consumer_restore.hpp

@@ -49,6 +49,7 @@ public:
   virtual bool init();
   virtual void release();
   virtual bool table(const TableS &);
+  virtual bool endOfTables();
   virtual void tuple(const TupleS &);
   virtual void tuple_free();
   virtual void tuple_a(restore_callback_t *cb);
@@ -83,6 +84,8 @@ public:
     const NdbDictionary::Table* m_new_table;
   } m_cache;
   const NdbDictionary::Table* get_table(const NdbDictionary::Table* );
+
+  Vector<const NdbDictionary::Table*> m_indexes;
 };
 
 #endif

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

@@ -276,6 +276,13 @@ main(int argc, const char** argv)
     }
   }
   
+  for(i= 0; i < g_consumers.size(); i++)
+    if (!g_consumers[i]->endOfTables())
+    {
+      ndbout_c("Restore: Failed while closing tables");
+      return -11;
+    } 
+  
   if (ga_restore || ga_print) 
   {
     if (ga_restore) 

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -1796,14 +1796,30 @@ NdbDictionaryImpl::getIndexImpl(const char * externalName,
   /**
    * Create index impl
    */
-  NdbIndexImpl* idx = new NdbIndexImpl();
+  NdbIndexImpl* idx;
+  if(NdbDictInterface::create_index_obj_from_table(&idx, tab, prim) == 0){
+    idx->m_table = tab;
+    idx->m_internalName.assign(internalName);
+    // TODO Assign idx to tab->m_index
+    // Don't do it right now since assign can't asign a table with index
+    // tab->m_index = idx;
+    return idx;
+  }
+  return 0;
+}
+
+int
+NdbDictInterface::create_index_obj_from_table(NdbIndexImpl** dst,
+					      const NdbTableImpl* tab,
+					      const NdbTableImpl* prim){
+  NdbIndexImpl *idx = new NdbIndexImpl();
   idx->m_version = tab->m_version;
   idx->m_status = tab->m_status;
   idx->m_indexId = tab->m_tableId;
-  idx->m_internalName.assign(internalName);
-  idx->m_externalName.assign(externalName);
+  idx->m_externalName.assign(tab->getName());
   idx->m_tableName.assign(prim->m_externalName);
   idx->m_type = tab->m_indexType;
+  idx->m_logging = tab->m_logging;
   // skip last attribute (NDB$PK or NDB$TNODE)
   for(unsigned i = 0; i+1<tab->m_columns.size(); i++){
     NdbColumnImpl* col = new NdbColumnImpl;
@@ -1820,11 +1836,8 @@ NdbDictionaryImpl::getIndexImpl(const char * externalName,
     col->m_keyInfoPos = key_id;
   }
 
-  idx->m_table = tab;
-  // TODO Assign idx to tab->m_index
-  // Don't do it right now since assign can't asign a table with index
-  // tab->m_index = idx;
-  return idx;
+  * dst = idx;
+  return 0;
 }
 
 /*****************************************************************

ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -304,6 +304,10 @@ public:
 			    const Uint32 * data, Uint32 len,
 			    bool fullyQualifiedNames);
   
+  static int create_index_obj_from_table(NdbIndexImpl ** dst, 
+					 const NdbTableImpl*,
+					 const NdbTableImpl*);
+  
   NdbError & m_error;
 private:
   Uint32 m_reference;