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

@@ -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,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); }

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 
 {

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,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;
   }
 }