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

7 files changed, 546 insertions, 108 deletions

diff --git a/ndb/include/ndbapi/NdbRecAttr.hpp b/ndb/include/ndbapi/NdbRecAttr.hpp
index df7f22fef60..159c7e7ce77 100644
--- a/ndb/include/ndbapi/NdbRecAttr.hpp
+++ b/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
diff --git a/ndb/include/util/OutputStream.hpp b/ndb/include/util/OutputStream.hpp
index 460915e12e7..19aa1838c4b 100644
--- a/ndb/include/util/OutputStream.hpp
+++ b/ndb/include/util/OutputStream.hpp
@@ -34,7 +34,8 @@ 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, ...);
   void flush() { fflush(f); }
diff --git a/ndb/src/ndbapi/NdbRecAttr.cpp b/ndb/src/ndbapi/NdbRecAttr.cpp
index abfbd76d2c3..b029f9826d0 100644
--- a/ndb/src/ndbapi/NdbRecAttr.cpp
+++ b/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,"Longvarchar", r.aRef()+2,len);
+      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,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 
 {
diff --git a/ndb/tools/restore/Restore.cpp b/ndb/tools/restore/Restore.cpp
index 8b2e9a799a4..7ca7e91acf0 100644
--- a/ndb/tools/restore/Restore.cpp
+++ b/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;
 
@@ -176,6 +180,49 @@ RestoreMetaData::readMetaTableDesc() {
 }
 
 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() {
 
   Uint32 data[4];
@@ -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;
 }
diff --git a/ndb/tools/restore/Restore.hpp b/ndb/tools/restore/Restore.hpp
index b132dda374d..afffc1a7b1e 100644
--- a/ndb/tools/restore/Restore.hpp
+++ b/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();
diff --git a/ndb/tools/restore/consumer_printer.cpp b/ndb/tools/restore/consumer_printer.cpp
index 8fe9805c39c..e0525522284 100644
--- a/ndb/tools/restore/consumer_printer.cpp
+++ b/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,8 +58,9 @@ BackupPrinter::endOfLogEntrys()
 {
   if (m_print || m_print_log) 
   {
-    ndbout << "Printed " << m_dataCount << " tuples and "
-	   << m_logCount << " log entries" 
-	   << " to stdout." << endl;
+    info.setLevel(254);
+    info << "Printed " << m_dataCount << " tuples and "
+         << m_logCount << " log entries" 
+         << " to stdout." << endl;
   }
 }
diff --git a/ndb/tools/restore/restore_main.cpp b/ndb/tools/restore/restore_main.cpp
index 0110782ff39..5f31a86e270 100644
--- a/ndb/tools/restore/restore_main.cpp
+++ b/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);
   }
-  // Set backup file path
-  if (*pargv[0] != NULL) 
+  for (;;)
+  {
+    int i= 0;
+    if (ga_backupPath == default_backupPath)
+    {
+      // Set backup file path
+      if ((*pargv)[i] == NULL)
+        break;
+      ga_backupPath = (*pargv)[i++];
+    }
+    if ((*pargv)[i] == NULL)
+      break;
+    g_databases.push_back((*pargv)[i++]);
+    while ((*pargv)[i] != NULL)
+    {
+      g_tables.push_back((*pargv)[i++]);
+    }
+    break;
+  }
+  info.setLevel(254);
+  info << "backup path = " << ga_backupPath << endl;
+  if (g_databases.size() > 0)
   {
-    ga_backupPath = *pargv[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 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)
 {
-  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);
+  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: " 
-	 <<  getVersionString(version, 0, buf, sizeof(buf)) << endl;
+  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()))
-	    for(Uint32 i= 0; i < g_consumers.size(); i++) 
-	      g_consumers[i]->tuple(* tuple);
+          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,9 +646,12 @@ main(int argc, char** argv)
       const LogEntry * logEntry = 0;
       while ((logEntry = logIter.getNextLogEntry(res= 0)) != 0)
       {
-	if (checkSysTable(logEntry->m_table->getTableName()))
-	  for(Uint32 i= 0; i < g_consumers.size(); i++)
-	    g_consumers[i]->logEntry(* logEntry);
+        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);
       }
       if (res < 0)
       {
@@ -418,17 +668,17 @@ main(int argc, char** argv)
     {
       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());
-	      exitHandler(NDBT_FAILED);
-	    } 
-	}
+        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(*table))
+          {
+            err << "Restore: Failed to finalize restore table: %s. ";
+            err << "Exiting... " << metaData[i]->getTableName() << endl;
+            exitHandler(NDBT_FAILED);
+          }
       }
     }
   }
@@ -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;
     }
   }
 
-  clearConsumers();
-  return NDBT_ProgramExit(NDBT_OK);
+  if (opt_verbose)
+    return NDBT_ProgramExit(NDBT_OK);
+  else
+    return 0;
 } // main
 
 template class Vector<BackupConsumer*>;
+template class Vector<OutputStream*>;