Bug#26899 ndb_restore cannot restore selected tables and databases

Bug#26900 ndb_restore printout option does not give structured data
- add data stucturing options
- add database and table selection options


ndb/include/ndbapi/NdbRecAttr.hpp:
  add formatting option to ndb recattr printing
ndb/include/util/OutputStream.hpp:
  getter for FILE*
ndb/src/ndbapi/NdbRecAttr.cpp:
  add formatting option to ndb recattr printing
ndb/tools/restore/Restore.cpp:
  add formatting option to restore printing
  adoption to 5.1 source code
ndb/tools/restore/Restore.hpp:
  restore adoption to 5.1
ndb/tools/restore/consumer_printer.cpp:
  add formatting option to restore printing
ndb/tools/restore/restore_main.cpp:
  added option to restore only selected databases and tables
  added option to get CSV printout to file and stdout
  add formatting option to restore printing
  adoption to 5.1 source code

ndb/include/ndbapi/NdbRecAttr.hpp

@@ -441,6 +441,25 @@ NdbRecAttr::isNULL() const
 
 class NdbOut& operator <<(class NdbOut&, const NdbRecAttr &);
 
+class NdbRecordPrintFormat
+{
+public:
+  NdbRecordPrintFormat();
+  virtual ~NdbRecordPrintFormat();
+  const char *lines_terminated_by;
+  const char *fields_terminated_by;
+  const char *start_array_enclosure;
+  const char *end_array_enclosure;
+  const char *fields_enclosed_by;
+  const char *fields_optionally_enclosed_by;
+  const char *hex_prefix;
+  const char *null_string;
+  int hex_format;
+};
+NdbOut&
+ndbrecattr_print_formatted(NdbOut& out, const NdbRecAttr &r,
+                           const NdbRecordPrintFormat &f);
+
 #endif // ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
 
 #endif

ndb/include/util/OutputStream.hpp

@@ -34,6 +34,7 @@ class FileOutputStream : public OutputStream {
   FILE * f;
 public:
   FileOutputStream(FILE * file = stdout);
+  FILE *getFile() { return f; }
 
   int print(const char * fmt, ...);
   int println(const char * fmt, ...);

ndb/src/ndbapi/NdbRecAttr.cpp

@@ -140,8 +140,24 @@ NdbRecAttr::receive_data(const Uint32 * data, Uint32 sz){
   return false;
 }
 
+NdbRecordPrintFormat::NdbRecordPrintFormat()
+{
+  fields_terminated_by= ";";
+  start_array_enclosure= "[";
+  end_array_enclosure= "]";
+  fields_enclosed_by= "";
+  fields_optionally_enclosed_by= "\"";
+  lines_terminated_by= "\n";
+  hex_prefix= "H'";
+  null_string= "[NULL]";
+  hex_format= 0;
+}
+NdbRecordPrintFormat::~NdbRecordPrintFormat() {};
+static const NdbRecordPrintFormat default_print_format;
+
 static void
-ndbrecattr_print_string(NdbOut& out, const char *type,
+ndbrecattr_print_string(NdbOut& out, const NdbRecordPrintFormat &f,
+                        const char *type, bool is_binary,
 			const char *aref, unsigned sz)
 {
   const unsigned char* ref = (const unsigned char*)aref;
@@ -150,6 +166,25 @@ ndbrecattr_print_string(NdbOut& out, const char *type,
   for (i=sz-1; i >= 0; i--)
     if (ref[i] == 0) sz--;
     else break;
+  if (!is_binary)
+  {
+    // trailing spaces are not printed
+    for (i=sz-1; i >= 0; i--)
+      if (ref[i] == 32) sz--;
+      else break;
+  }
+  if (is_binary && f.hex_format)
+  {
+    if (sz == 0)
+    {
+      out.print("0x0");
+      return;
+    }
+    out.print("0x");
+    for (len = 0; len < (int)sz; len++)
+      out.print("%02X", (int)ref[len]);
+    return;
+  }
   if (sz == 0) return; // empty
 
   for (len=0; len < (int)sz && ref[i] != 0; len++)
@@ -170,37 +205,56 @@ ndbrecattr_print_string(NdbOut& out, const char *type,
     for (i= len+1; ref[i] != 0; i++)
     out.print("%u]",len-i);
     assert((int)sz > i);
-    ndbrecattr_print_string(out,type,aref+i,sz-i);
+    ndbrecattr_print_string(out,f,type,is_binary,aref+i,sz-i);
   }
 }
 
-NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
+NdbOut&
+ndbrecattr_print_formatted(NdbOut& out, const NdbRecAttr &r,
+                           const NdbRecordPrintFormat &f)
 {
   if (r.isNULL())
   {
-    out << "[NULL]";
+    out << f.null_string;
     return out;
   }
   
   const NdbDictionary::Column* c = r.getColumn();
   uint length = c->getLength();
-  if (length > 1)
-    out << "[";
-
-  for (Uint32 j = 0; j < length; j++) 
   {
-    if (j > 0)
-      out << " ";
-
+    const char *fields_optionally_enclosed_by;
+    if (f.fields_enclosed_by[0] == '\0')
+      fields_optionally_enclosed_by=
+        f.fields_optionally_enclosed_by;
+    else
+      fields_optionally_enclosed_by= "";
+    out << f.fields_enclosed_by;
+    Uint32 j;
     switch(r.getType()){
     case NdbDictionary::Column::Bigunsigned:
       out << r.u_64_value();
       break;
     case NdbDictionary::Column::Bit:
-      out << hex << "H'" << r.u_32_value() << dec;
+      for (j = (length-1)/32 + 1; j > 0; j--)
+        if (*((Uint32*)r.aRef() + j - 1))
+          break;
+      if (j == 0)
+      {
+        out << "0x0";
+        break;
+      }
+      out << f.hex_prefix << "0x";
+      for (; j > 0; j--)
+        out.print("%X", *((Uint32*)r.aRef() + j - 1));
       break;
     case NdbDictionary::Column::Unsigned:
-      out << r.u_32_value();
+      if (length > 1)
+        out << f.start_array_enclosure;
+      out << *(Uint32*)r.aRef();
+      for (j = 1; j < length; j++)
+        out << " " << *((Uint32*)r.aRef() + j);
+      if (length > 1)
+        out << f.end_array_enclosure;
       break;
     case NdbDictionary::Column::Smallunsigned:
       out << r.u_short_value();
@@ -221,25 +275,37 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
       out << (int) r.char_value();
       break;
     case NdbDictionary::Column::Binary:
+      if (!f.hex_format)
+        out << fields_optionally_enclosed_by;
       j = r.arraySize();
-      ndbrecattr_print_string(out,"Binary", r.aRef(), j);
+      ndbrecattr_print_string(out,f,"Binary", true, r.aRef(), j);
+      if (!f.hex_format)
+        out << fields_optionally_enclosed_by;
       break;
     case NdbDictionary::Column::Char:
+      out << fields_optionally_enclosed_by;
       j = length;
-      ndbrecattr_print_string(out,"Char", r.aRef(), r.arraySize());
+      ndbrecattr_print_string(out,f,"Char", false, r.aRef(), r.arraySize());
+      out << fields_optionally_enclosed_by;
       break;
     case NdbDictionary::Column::Varchar:
     {
+      out << fields_optionally_enclosed_by;
       unsigned len = *(const unsigned char*)r.aRef();
-      ndbrecattr_print_string(out,"Varchar", r.aRef()+1,len);
+      ndbrecattr_print_string(out,f,"Varchar", false, r.aRef()+1,len);
       j = length;
+      out << fields_optionally_enclosed_by;
     }
     break;
     case NdbDictionary::Column::Varbinary:
     {
+      if (!f.hex_format)
+        out << fields_optionally_enclosed_by;
       unsigned len = *(const unsigned char*)r.aRef();
-      ndbrecattr_print_string(out,"Varbinary", r.aRef()+1,len);
+      ndbrecattr_print_string(out,f,"Varbinary", true, r.aRef()+1,len);
       j = length;
+      if (!f.hex_format)
+        out << fields_optionally_enclosed_by;
     }
     break;
     case NdbDictionary::Column::Float:
@@ -368,16 +434,28 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
     break;
     case NdbDictionary::Column::Longvarchar:
     {
+      out << fields_optionally_enclosed_by;
+      unsigned len = uint2korr(r.aRef());
+      ndbrecattr_print_string(out,f,"Longvarchar", false, r.aRef()+2,len);
+      j = length;
+      out << fields_optionally_enclosed_by;
+    }
+    break;
+    case NdbDictionary::Column::Longvarbinary:
+    {
+      if (!f.hex_format)
+        out << fields_optionally_enclosed_by;
       unsigned len = uint2korr(r.aRef());
-      ndbrecattr_print_string(out,"Longvarchar", r.aRef()+2,len);
+      ndbrecattr_print_string(out,f,"Longvarbinary", true, r.aRef()+2,len);
       j = length;
+      if (!f.hex_format)
+        out << fields_optionally_enclosed_by;
     }
     break;
 
     case NdbDictionary::Column::Undefined:
     case NdbDictionary::Column::Mediumint:
     case NdbDictionary::Column::Mediumunsigned:
-    case NdbDictionary::Column::Longvarbinary:
     unknown:
     //default: /* no print functions for the rest, just print type */
     out << (int) r.getType();
@@ -386,16 +464,17 @@ NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
       out << " " << j << " times";
     break;
     }
-  }
-
-  if (length > 1)
-  {
-    out << "]";
+    out << f.fields_enclosed_by;
   }
 
   return out;
 }
 
+NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
+{
+  return ndbrecattr_print_formatted(out, r, default_print_format);
+}
+
 Int64
 NdbRecAttr::int64_value() const 
 {

ndb/tools/restore/Restore.cpp

@@ -23,6 +23,8 @@
 #include <SimpleProperties.hpp>
 #include <signaldata/DictTabInfo.hpp>
 
+extern NdbRecordPrintFormat g_ndbrecord_print_format;
+
 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
@@ -118,6 +120,8 @@ RestoreMetaData::loadContent()
       return 0;
     }
   }
+  if (! markSysTables())
+    return 0;
   if(!readGCPEntry())
     return 0;
 
@@ -175,6 +179,49 @@ RestoreMetaData::readMetaTableDesc() {
   return parseTableDescriptor((Uint32*)ptr, len);	     
 }
 
+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)
+      table->isSysTable = 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->isSysTable)
+            blobTable->isSysTable = true;
+          blobTable->m_main_table = table;
+          break;
+        }
+      }
+      if (j == getNoOfTables()) {
+        err << "Restore: Bad primary table id in " << blobTableName << endl;
+        return false;
+      }
+    }
+  }
+  return true;
+}
+
 bool
 RestoreMetaData::readGCPEntry() {
 
@@ -259,6 +306,8 @@ TableS::TableS(Uint32 version, NdbTableImpl* tableImpl)
   m_max_auto_val= 0;
   m_noOfRecords= 0;
   backupVersion = version;
+  isSysTable = false;
+  m_main_table = NULL;
   
   for (int i = 0; i < tableImpl->getNoOfColumns(); i++)
     createAttr(tableImpl->getColumn(i));
@@ -704,6 +753,7 @@ bool RestoreDataIterator::readFragmentHeader(int & ret)
     return false;
   }
 
+  info.setLevel(254);
   info << "_____________________________________________________" << endl
        << "Processing data in table: " << m_currentTable->getTableName() 
        << "(" << Header.TableId << ") fragment " 
@@ -924,13 +974,13 @@ operator<<(NdbOut& ndbout, const AttributeS& attr){
 
   if (data.null)
   {
-    ndbout << "<NULL>";
+    ndbout << g_ndbrecord_print_format.null_string;
     return ndbout;
   }
   
   NdbRecAttr tmprec(0);
   tmprec.setup(desc.m_column, (char *)data.void_value);
-  ndbout << tmprec;
+  ndbrecattr_print_formatted(ndbout, tmprec, g_ndbrecord_print_format);
 
   return ndbout;
 }
@@ -939,17 +989,15 @@ operator<<(NdbOut& ndbout, const AttributeS& attr){
 NdbOut& 
 operator<<(NdbOut& ndbout, const TupleS& tuple)
 {
-  ndbout << tuple.getTable()->getTableName() << "; ";
   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);
     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;
 }

ndb/tools/restore/Restore.hpp

@@ -25,8 +25,6 @@
 #include <ndb_version.h>
 #include <version.h>
 
-static const char * delimiter = ";"; // Delimiter in file dump
-
 const int FileNameLenC = 256;
 const int TableNameLenC = 256;
 const int AttrNameLenC = 256;
@@ -143,6 +141,10 @@ class TableS {
 
   int pos;
 
+  bool isSysTable;
+  TableS *m_main_table;
+  Uint32 m_local_id;
+
   Uint64 m_noOfRecords;
   Vector<FragmentInfo *> m_fragmentInfo;
 
@@ -156,6 +158,9 @@ public:
   Uint32 getTableId() const { 
     return m_dictTable->getTableId(); 
   }
+  Uint32 getLocalId() const { 
+    return m_local_id; 
+  }
   Uint32 getNoOfRecords() const { 
     return m_noOfRecords; 
   }
@@ -235,6 +240,14 @@ public:
     return allAttributesDesc[attributeId]; 
   }
 
+  bool getSysTable() const {
+    return isSysTable;
+  }
+
+  const TableS *getMainTable() const {
+    return m_main_table;
+  }
+
   TableS& operator=(TableS& org) ; 
 }; // TableS;
 
@@ -285,6 +298,7 @@ class RestoreMetaData : public BackupFile {
   Vector<TableS *> allTables;
   bool readMetaFileHeader();
   bool readMetaTableDesc();
+  bool markSysTables();
 		
   bool readGCPEntry();
   bool readFragmentInfo();

ndb/tools/restore/consumer_printer.cpp

@@ -14,6 +14,9 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 #include "consumer_printer.hpp"
+extern FilteredNdbOut info;
+extern NdbRecordPrintFormat g_ndbrecord_print_format;
+extern const char *tab_path;
 
 bool
 BackupPrinter::table(const TableS & tab)
@@ -21,7 +24,8 @@ BackupPrinter::table(const TableS & tab)
   if (m_print || m_print_meta) 
   {
     m_ndbout << tab;
-    ndbout_c("Successfully printed table: %s", tab.m_dictTable->getName());
+    info.setLevel(254);
+    info << "Successfully printed table: ", tab.m_dictTable->getName();
   }
   return true;
 }
@@ -31,7 +35,14 @@ BackupPrinter::tuple(const TupleS & tup)
 {
   m_dataCount++;
   if (m_print || m_print_data)
-    m_ndbout << tup << endl;  
+  {
+    if (m_ndbout.m_out == info.m_out)
+    {
+      info.setLevel(254);
+      info << tup.getTable()->getTableName() << "; ";
+    }
+    m_ndbout << tup << g_ndbrecord_print_format.lines_terminated_by;  
+  }
 }
 
 void
@@ -47,7 +58,8 @@ BackupPrinter::endOfLogEntrys()
 {
   if (m_print || m_print_log) 
   {
-    ndbout << "Printed " << m_dataCount << " tuples and "
+    info.setLevel(254);
+    info << "Printed " << m_dataCount << " tuples and "
          << m_logCount << " log entries" 
          << " to stdout." << endl;
   }

ndb/tools/restore/restore_main.cpp

@@ -18,7 +18,9 @@
 #include <Vector.hpp>
 #include <ndb_limits.h>
 #include <NdbTCP.h>
+#include <NdbMem.h>
 #include <NdbOut.hpp>
+#include <OutputStream.hpp>
 #include <NDBT_ReturnCodes.h>
 
 #include "consumer_restore.hpp"
@@ -33,8 +35,18 @@ static int ga_nParallelism = 128;
 static int ga_backupId = 0;
 static bool ga_dont_ignore_systab_0 = false;
 static Vector<class BackupConsumer *> g_consumers;
+static BackupPrinter* g_printer = NULL;
 
-static const char* ga_backupPath = "." DIR_SEPARATOR;
+static const char* default_backupPath = "." DIR_SEPARATOR;
+static const char* ga_backupPath = default_backupPath;
+
+const char *opt_ndb_database= NULL;
+const char *opt_ndb_table= NULL;
+unsigned int opt_verbose;
+unsigned int opt_hex_format;
+Vector<BaseString> g_databases;
+Vector<BaseString> g_tables;
+NdbRecordPrintFormat g_ndbrecord_print_format;
 
 NDB_STD_OPTS_VARS;
 
@@ -53,6 +65,32 @@ BaseString g_options("ndb_restore");
 
 const char *load_default_groups[]= { "mysql_cluster","ndb_restore",0 };
 
+enum ndb_restore_options {
+  OPT_PRINT= NDB_STD_OPTIONS_LAST,
+  OPT_PRINT_DATA,
+  OPT_PRINT_LOG,
+  OPT_PRINT_META,
+  OPT_BACKUP_PATH,
+  OPT_HEX_FORMAT,
+  OPT_FIELDS_ENCLOSED_BY,
+  OPT_FIELDS_TERMINATED_BY,
+  OPT_FIELDS_OPTIONALLY_ENCLOSED_BY,
+  OPT_LINES_TERMINATED_BY,
+  OPT_APPEND,
+  OPT_VERBOSE
+};
+/*
+  the below formatting options follow the formatting from mysqldump
+  do not change unless to adopt to changes in mysqldump
+*/
+static const char *opt_fields_enclosed_by= "";
+static const char *opt_fields_terminated_by= ";";
+static const char *opt_fields_optionally_enclosed_by= "";
+static const char *opt_lines_terminated_by= "\n";
+
+static const char *tab_path= NULL;
+static int opt_append;
+
 static struct my_option my_long_options[] =
 {
   NDB_STD_OPTS("ndb_restore"),
@@ -78,22 +116,56 @@ static struct my_option my_long_options[] =
     "(parallelism can be 1 to 1024)", 
     (gptr*) &ga_nParallelism, (gptr*) &ga_nParallelism, 0,
     GET_INT, REQUIRED_ARG, 128, 1, 1024, 0, 1, 0 },
-  { "print", 256, "Print data and log to stdout",
+  { "print", OPT_PRINT, "Print data and log to stdout",
     (gptr*) &_print, (gptr*) &_print, 0,
     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
-  { "print_data", 257, "Print data to stdout", 
+  { "print_data", OPT_PRINT_DATA, "Print data to stdout", 
     (gptr*) &_print_data, (gptr*) &_print_data, 0,
     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
-  { "print_meta", 258, "Print meta data to stdout",
+  { "print_meta", OPT_PRINT_META, "Print meta data to stdout",
     (gptr*) &_print_meta, (gptr*) &_print_meta,  0,
     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
-  { "print_log", 259, "Print log to stdout",
+  { "print_log", OPT_PRINT_LOG, "Print log to stdout",
     (gptr*) &_print_log, (gptr*) &_print_log,  0,
     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "backup_path", OPT_BACKUP_PATH, "Path to backup files",
+    (gptr*) &ga_backupPath, (gptr*) &ga_backupPath, 0,
+    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
   { "dont_ignore_systab_0", 'f',
     "Experimental. Do not ignore system table during restore.", 
     (gptr*) &ga_dont_ignore_systab_0, (gptr*) &ga_dont_ignore_systab_0, 0,
     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "fields-enclosed-by", OPT_FIELDS_ENCLOSED_BY,
+    "Fields are enclosed by ...",
+    (gptr*) &opt_fields_enclosed_by, (gptr*) &opt_fields_enclosed_by, 0,
+    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { "fields-terminated-by", OPT_FIELDS_TERMINATED_BY,
+    "Fields are terminated by ...",
+    (gptr*) &opt_fields_terminated_by,
+    (gptr*) &opt_fields_terminated_by, 0,
+    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { "fields-optionally-enclosed-by", OPT_FIELDS_OPTIONALLY_ENCLOSED_BY,
+    "Fields are optionally enclosed by ...",
+    (gptr*) &opt_fields_optionally_enclosed_by,
+    (gptr*) &opt_fields_optionally_enclosed_by, 0,
+    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { "hex", OPT_HEX_FORMAT, "print binary types in hex format", 
+    (gptr*) &opt_hex_format, (gptr*) &opt_hex_format, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "tab", 'T', "Creates tab separated textfile for each table to "
+    "given path. (creates .txt files)",
+   (gptr*) &tab_path, (gptr*) &tab_path, 0,
+    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  { "append", OPT_APPEND, "for --tab append data to file", 
+    (gptr*) &opt_append, (gptr*) &opt_append, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
+  { "lines-terminated-by", OPT_LINES_TERMINATED_BY, "",
+    (gptr*) &opt_lines_terminated_by, (gptr*) &opt_lines_terminated_by, 0,
+    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { "verbose", OPT_VERBOSE,
+    "verbosity", 
+    (gptr*) &opt_verbose, (gptr*) &opt_verbose, 0,
+    GET_INT, REQUIRED_ARG, 1, 0, 255, 0, 0, 0 },
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 
@@ -119,19 +191,26 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
 #endif
   ndb_std_get_one_option(optid, opt, argument);
   switch (optid) {
+  case OPT_VERBOSE:
+    info.setThreshold(255-opt_verbose);
+    break;
   case 'n':
     if (ga_nodeId == 0)
     {
-      printf("Error in --nodeid,-n setting, see --help\n");
+      err << "Error in --nodeid,-n setting, see --help";
       exit(NDBT_ProgramExit(NDBT_WRONGARGS));
     }
+    info.setLevel(254);
+    info << "Nodeid = " << ga_nodeId << endl;
     break;
   case 'b':
     if (ga_backupId == 0)
     {
-      printf("Error in --backupid,-b setting, see --help\n");
+      err << "Error in --backupid,-b setting, see --help";
       exit(NDBT_ProgramExit(NDBT_WRONGARGS));
     }
+    info.setLevel(254);
+    info << "Backup Id = " << ga_backupId << endl;
     break;
   }
   return 0;
@@ -139,20 +218,26 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
 bool
 readArguments(int *pargc, char*** pargv) 
 {
+  Uint32 i;
+  debug << "Load defaults" << endl;
+  const char *load_default_groups[]= { "mysql_cluster","ndb_restore",0 };
+
   load_defaults("my",load_default_groups,pargc,pargv);
+  debug << "handle_options" << endl;
   if (handle_options(pargc, pargv, my_long_options, get_one_option))
   {
     exit(NDBT_ProgramExit(NDBT_WRONGARGS));
   }
 
-  BackupPrinter* printer = new BackupPrinter();
-  if (printer == NULL)
+  g_printer = new BackupPrinter();
+  if (g_printer == NULL)
     return false;
 
   BackupRestore* restore = new BackupRestore(ga_nParallelism);
   if (restore == NULL) 
   {
-    delete printer;
+    delete g_printer;
+    g_printer = NULL;
     return false;
   }
 
@@ -160,22 +245,22 @@ readArguments(int *pargc, char*** pargv)
   {
     ga_print = true;
     ga_restore = true;
-    printer->m_print = true;
+    g_printer->m_print = true;
   } 
   if (_print_meta) 
   {
     ga_print = true;
-    printer->m_print_meta = true;
+    g_printer->m_print_meta = true;
   }
   if (_print_data) 
   {
     ga_print = true;
-    printer->m_print_data = true;
+    g_printer->m_print_data = true;
   }
   if (_print_log) 
   {
     ga_print = true;
-    printer->m_print_log = true;
+    g_printer->m_print_log = true;
   }
 
   if (_restore_data)
@@ -191,19 +276,64 @@ readArguments(int *pargc, char*** pargv)
   }
 
   {
-    BackupConsumer * c = printer;
+    BackupConsumer * c = g_printer;
     g_consumers.push_back(c);
   }
   {
     BackupConsumer * c = restore;
     g_consumers.push_back(c);
   }
+  for (;;)
+  {
+    int i= 0;
+    if (ga_backupPath == default_backupPath)
+    {
       // Set backup file path
-  if (*pargv[0] != NULL) 
+      if ((*pargv)[i] == NULL)
+        break;
+      ga_backupPath = (*pargv)[i++];
+    }
+    if ((*pargv)[i] == NULL)
+      break;
+    g_databases.push_back((*pargv)[i++]);
+    while ((*pargv)[i] != NULL)
     {
-    ga_backupPath = *pargv[0];
+      g_tables.push_back((*pargv)[i++]);
     }
-
+    break;
+  }
+  info.setLevel(254);
+  info << "backup path = " << ga_backupPath << endl;
+  if (g_databases.size() > 0)
+  {
+    info << "Restoring only from database " << g_databases[0].c_str() << endl;
+    if (g_tables.size() > 0)
+      info << "Restoring only tables:";
+    for (unsigned i= 0; i < g_tables.size(); i++)
+    {
+      info << " " << g_tables[i].c_str();
+    }
+    if (g_tables.size() > 0)
+      info << endl;
+  }
+  /*
+    the below formatting follows the formatting from mysqldump
+    do not change unless to adopt to changes in mysqldump
+  */
+  g_ndbrecord_print_format.fields_enclosed_by=
+    opt_fields_enclosed_by;
+  g_ndbrecord_print_format.fields_terminated_by=
+    opt_fields_terminated_by;
+  g_ndbrecord_print_format.fields_optionally_enclosed_by=
+    opt_fields_optionally_enclosed_by;
+  g_ndbrecord_print_format.lines_terminated_by=
+    opt_lines_terminated_by;
+  if (g_ndbrecord_print_format.fields_optionally_enclosed_by[0] == '\0')
+    g_ndbrecord_print_format.null_string= "\\N";
+  else
+    g_ndbrecord_print_format.null_string= "";
+  g_ndbrecord_print_format.hex_prefix= "";
+  g_ndbrecord_print_format.hex_format= opt_hex_format;
   return true;
 }
 
@@ -215,14 +345,81 @@ clearConsumers()
   g_consumers.clear();
 }
 
-static bool
-checkSysTable(const char *tableName) 
+static inline bool
+checkSysTable(const TableS* table)
+{
+  return ga_dont_ignore_systab_0 || ! table->getSysTable();
+}
+
+static inline bool
+checkSysTable(const RestoreMetaData& metaData, uint i)
+{
+  assert(i < metaData.getNoOfTables());
+  return checkSysTable(metaData[i]);
+}
+
+static inline bool
+isBlobTable(const TableS* table)
 {
-  return ga_dont_ignore_systab_0 ||
-    (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);
+  return table->getMainTable() != NULL;
+}
+
+static inline bool
+isIndex(const TableS* table)
+{
+  const NdbTableImpl & tmptab = NdbTableImpl::getImpl(* table->m_dictTable);
+  return (int) tmptab.m_indexType != (int) NdbDictionary::Index::Undefined;
+}
+
+static inline bool
+checkDbAndTableName(const TableS* table)
+{
+  if (g_tables.size() == 0 &&
+      g_databases.size() == 0)
+    return true;
+  if (g_databases.size() == 0)
+    g_databases.push_back("TEST_DB");
+
+  // Filter on the main table name for indexes and blobs
+  const char *table_name;
+  if (isBlobTable(table))
+    table_name= table->getMainTable()->getTableName();
+  else if (isIndex(table))
+    table_name=
+      NdbTableImpl::getImpl(*table->m_dictTable).m_primaryTable.c_str();
+  else
+    table_name= table->getTableName();
+
+  unsigned i;
+  for (i= 0; i < g_databases.size(); i++)
+  {
+    if (strncmp(table_name, g_databases[i].c_str(),
+                g_databases[i].length()) == 0 &&
+        table_name[g_databases[i].length()] == '/')
+    {
+      // we have a match
+      if (g_databases.size() > 1 || g_tables.size() == 0)
+        return true;
+      break;
+    }
+  }
+  if (i == g_databases.size())
+    return false; // no match found
+
+  while (*table_name != '/') table_name++;
+  table_name++;
+  while (*table_name != '/') table_name++;
+  table_name++;
+
+  for (i= 0; i < g_tables.size(); i++)
+  {
+    if (strcmp(table_name, g_tables[i].c_str()) == 0)
+    {
+      // we have a match
+      return true;
+    }
+  }
+  return false;
 }
 
 static void
@@ -247,6 +444,7 @@ main(int argc, char** argv)
 {
   NDB_INIT(argv[0]);
 
+  debug << "Start readArguments" << endl;
   if (!readArguments(&argc, &argv))
   {
     exitHandler(NDBT_FAILED);
@@ -265,10 +463,11 @@ main(int argc, char** argv)
   /**
    * we must always load meta data, even if we will only print it to stdout
    */
+  debug << "Start restoring meta data" << endl;
   RestoreMetaData metaData(ga_backupPath, ga_nodeId, ga_backupId);
   if (!metaData.readHeader())
   {
-    ndbout << "Failed to read " << metaData.getFilename() << endl << endl;
+    err << "Failed to read " << metaData.getFilename() << endl << endl;
     exitHandler(NDBT_FAILED);
   }
 
@@ -276,66 +475,108 @@ main(int argc, char** argv)
   const Uint32 version = tmp.NdbVersion;
   
   char buf[NDB_VERSION_STRING_BUF_SZ];
-  ndbout << "Ndb version in backup files: " 
+  info.setLevel(254);
+  info << "Ndb version in backup files: " 
        <<  getVersionString(version, 0, buf, sizeof(buf)) << endl;
   
   /**
    * check wheater we can restore the backup (right version).
    */
+  if (version > NDB_VERSION)
+  {
+    err << "Restore program older than backup version. Not supported. "
+        << "Use new restore program" << endl;
+    exitHandler(NDBT_FAILED);
+  }
+
+  debug << "Load content" << endl;
   int res  = metaData.loadContent();
   
   if (res == 0)
   {
-    ndbout_c("Restore: Failed to load content");
+    err << "Restore: Failed to load content" << endl;
     exitHandler(NDBT_FAILED);
   }
-  
+  debug << "Get no of Tables" << endl; 
   if (metaData.getNoOfTables() == 0) 
   {
-    ndbout_c("Restore: The backup contains no tables ");
+    err << "The backup contains no tables" << endl;
     exitHandler(NDBT_FAILED);
   }
-
+  debug << "Validate Footer" << endl;
 
   if (!metaData.validateFooter()) 
   {
-    ndbout_c("Restore: Failed to validate footer.");
+    err << "Restore: Failed to validate footer." << endl;
     exitHandler(NDBT_FAILED);
   }
-
+  debug << "Init Backup objects" << endl;
   Uint32 i;
   for(i= 0; i < g_consumers.size(); i++)
   {
     if (!g_consumers[i]->init())
     {
       clearConsumers();
+      err << "Failed to initialize consumers" << endl;
       exitHandler(NDBT_FAILED);
     }
 
   }
 
+  Vector<OutputStream *> table_output(metaData.getNoOfTables());
+  debug << "Restoring tables" << endl;
   for(i = 0; i<metaData.getNoOfTables(); i++)
   {
-    if (checkSysTable(metaData[i]->getTableName()))
+    const TableS *table= metaData[i];
+    table_output.push_back(NULL);
+    if (!checkDbAndTableName(table))
+      continue;
+    if (checkSysTable(table))
+    {
+      if (!tab_path || isBlobTable(table) || isIndex(table))
       {
+        table_output[i]= ndbout.m_out;
+      }
+      else
+      {
+        FILE* res;
+        char filename[FN_REFLEN], tmp_path[FN_REFLEN];
+        const char *table_name;
+        table_name= table->getTableName();
+        while (*table_name != '/') table_name++;
+        table_name++;
+        while (*table_name != '/') table_name++;
+        table_name++;
+        convert_dirname(tmp_path, tab_path, NullS);
+        res= my_fopen(fn_format(filename, table_name, tmp_path, ".txt", 4),
+                      opt_append ?
+                      O_WRONLY|O_APPEND|O_CREAT :
+                      O_WRONLY|O_TRUNC|O_CREAT,
+                      MYF(MY_WME));
+        if (res == 0)
+        {
+          exitHandler(NDBT_FAILED);
+        }
+        FileOutputStream *f= new FileOutputStream(res);
+        table_output[i]= f;
+      }
       for(Uint32 j= 0; j < g_consumers.size(); j++)
-	if (!g_consumers[j]->table(* metaData[i]))
+	if (!g_consumers[j]->table(* table))
 	{
-	  ndbout_c("Restore: Failed to restore table: %s. "
-		   "Exiting...", 
-		   metaData[i]->getTableName());
+	  err << "Restore: Failed to restore table: ";
+          err << table->getTableName() << " ... Exiting " << endl;
 	  exitHandler(NDBT_FAILED);
         }
     }
   }
-  
+  debug << "Close tables" << endl; 
   for(i= 0; i < g_consumers.size(); i++)
     if (!g_consumers[i]->endOfTables())
     {
-      ndbout_c("Restore: Failed while closing tables");
+      err << "Restore: Failed while closing tables" << endl;
       exitHandler(NDBT_FAILED);
     } 
-  
+  debug << "Iterate over data" << endl; 
   if (ga_restore || ga_print) 
   {
     if(_restore_data || _print_data)
@@ -345,7 +586,7 @@ main(int argc, char** argv)
       // Read data file header
       if (!dataIter.readHeader())
       {
-	ndbout << "Failed to read header of data file. Exiting..." ;
+	err << "Failed to read header of data file. Exiting..." << endl;
 	exitHandler(NDBT_FAILED);
       }
       
@@ -355,20 +596,26 @@ main(int argc, char** argv)
 	const TupleS* tuple;
 	while ((tuple = dataIter.getNextTuple(res= 1)) != 0)
 	{
-	  if (checkSysTable(tuple->getTable()->getTableName()))
+          const TableS* table = tuple->getTable();
+          OutputStream *output = table_output[table->getLocalId()];
+          if (!output)
+            continue;
+          OutputStream *tmp = ndbout.m_out;
+          ndbout.m_out = output;
           for(Uint32 i= 0; i < g_consumers.size(); i++) 
             g_consumers[i]->tuple(* tuple);
+          ndbout.m_out =  tmp;
 	} // while (tuple != NULL);
 	
 	if (res < 0)
 	{
-	  ndbout_c("Restore: An error occured while restoring data. "
-		   "Exiting...");
+	  err <<" Restore: An error occured while restoring data. Exiting...";
+          err << endl;
 	  exitHandler(NDBT_FAILED);
 	}
 	if (!dataIter.validateFragmentFooter()) {
-	  ndbout_c("Restore: Error validating fragment footer. "
-		   "Exiting...");
+	  err << "Restore: Error validating fragment footer. ";
+          err << "Exiting..." << endl;
 	  exitHandler(NDBT_FAILED);
 	}
       } // while (dataIter.readFragmentHeader(res))
@@ -376,7 +623,7 @@ main(int argc, char** argv)
       if (res < 0)
       {
 	err << "Restore: An error occured while restoring data. Exiting... "
-	    << "res=" << res << endl;
+	    << "res= " << res << endl;
 	exitHandler(NDBT_FAILED);
       }
       
@@ -399,7 +646,10 @@ main(int argc, char** argv)
       const LogEntry * logEntry = 0;
       while ((logEntry = logIter.getNextLogEntry(res= 0)) != 0)
       {
-	if (checkSysTable(logEntry->m_table->getTableName()))
+        const TableS* table = logEntry->m_table;
+        OutputStream *output = table_output[table->getLocalId()];
+        if (!output)
+          continue;
         for(Uint32 i= 0; i < g_consumers.size(); i++)
           g_consumers[i]->logEntry(* logEntry);
       }
@@ -418,20 +668,20 @@ main(int argc, char** argv)
     {
       for(i = 0; i<metaData.getNoOfTables(); i++)
       {
-	if (checkSysTable(metaData[i]->getTableName()))
-	{
+        const TableS* table = metaData[i];
+        OutputStream *output = table_output[table->getLocalId()];
+        if (!output)
+          continue;
         for(Uint32 j= 0; j < g_consumers.size(); j++)
-	    if (!g_consumers[j]->finalize_table(* metaData[i]))
+          if (!g_consumers[j]->finalize_table(*table))
           {
-	      ndbout_c("Restore: Failed to finalize restore table: %s. "
-		       "Exiting...", 
-		       metaData[i]->getTableName());
+            err << "Restore: Failed to finalize restore table: %s. ";
+            err << "Exiting... " << metaData[i]->getTableName() << endl;
             exitHandler(NDBT_FAILED);
           }
       }
     }
   }
-  }
   for(Uint32 i= 0; i < g_consumers.size(); i++) 
   {
     if (g_consumers[i]->has_temp_error())
@@ -439,12 +689,27 @@ main(int argc, char** argv)
       clearConsumers();
       ndbout_c("\nRestore successful, but encountered temporary error, "
                "please look at configuration.");
-      return NDBT_ProgramExit(NDBT_TEMPORARY);
     }               
   }
   
   clearConsumers();
+
+  for(i = 0; i < metaData.getNoOfTables(); i++)
+  {
+    if (table_output[i] &&
+        table_output[i] != ndbout.m_out)
+    {
+      my_fclose(((FileOutputStream *)table_output[i])->getFile(), MYF(MY_WME));
+      delete table_output[i];
+      table_output[i] = NULL;
+    }
+  }
+
+  if (opt_verbose)
     return NDBT_ProgramExit(NDBT_OK);
+  else
+    return 0;
 } // main
 
 template class Vector<BackupConsumer*>;
+template class Vector<OutputStream*>;