ndb - wl-2451: Increase max schema object to > 1600

mysql-test/Makefile.am:
  divide schemafile in blocks to allow unlimited # objects
ndb/include/debugger/SignalLoggerManager.hpp:
  divide schemafile in blocks to allow unlimited # objects
ndb/include/kernel/ndb_limits.h:
  divide schemafile in blocks to allow unlimited # objects
ndb/src/common/debugger/SignalLoggerManager.cpp:
  divide schemafile in blocks to allow unlimited # objects
ndb/src/kernel/blocks/dbdict/Dbdict.cpp:
  divide schemafile in blocks to allow unlimited # objects
ndb/src/kernel/blocks/dbdict/Dbdict.hpp:
  divide schemafile in blocks to allow unlimited # objects
ndb/src/kernel/blocks/dbdict/SchemaFile.hpp:
  divide schemafile in blocks to allow unlimited # objects
ndb/src/kernel/blocks/dbdict/printSchemaFile.cpp:
  divide schemafile in blocks to allow unlimited # objects
ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  divide schemafile in blocks to allow unlimited # objects
ndb/test/ndbapi/testDict.cpp:
  divide schemafile in blocks to allow unlimited # objects

mysql-test/Makefile.am

@@ -59,6 +59,9 @@ dist-hook:
 	$(INSTALL_DATA) $(srcdir)/std_data/*.frm $(distdir)/std_data
 
 install-data-local:
+	true
+
+x-install-data-local:
 	$(mkinstalldirs) \
 		$(DESTDIR)$(testdir)/t \
 		$(DESTDIR)$(testdir)/r \

ndb/include/debugger/SignalLoggerManager.hpp

@@ -87,7 +87,7 @@ public:
   /**
    * Generic messages in the signal log
    */
-  void log(BlockNumber bno, const char * msg);
+  void log(BlockNumber bno, const char * msg, ...);
   
   /**
    * LogModes

ndb/include/kernel/ndb_limits.h

@@ -50,7 +50,7 @@
  **/
 #define MAX_TUPLES_PER_PAGE 8191
 #define MAX_TUPLES_BITS 13 		/* 13 bits = 8191 tuples per page */
-#define MAX_TABLES 1600
+#define MAX_TABLES 20320                /* SchemaFile.hpp */
 #define MAX_TAB_NAME_SIZE 128
 #define MAX_ATTR_NAME_SIZE 32
 #define MAX_ATTR_DEFAULT_VALUE_SIZE 128

ndb/src/common/debugger/SignalLoggerManager.cpp

@@ -383,7 +383,7 @@ SignalLoggerManager::sendSignalWithDelay(Uint32 delayInMilliSeconds,
  * Generic messages in the signal log
  */
 void
-SignalLoggerManager::log(BlockNumber bno, const char * msg)
+SignalLoggerManager::log(BlockNumber bno, const char * msg, ...)
 {
   // Normalise blocknumber for use in logModes array
   const BlockNumber bno2 = bno - MIN_BLOCK_NO;
@@ -391,7 +391,12 @@ SignalLoggerManager::log(BlockNumber bno, const char * msg)
 
   if(outputStream != 0 &&
      logModes[bno2] != LogOff){
-    fprintf(outputStream, "%s: %s\n", getBlockName(bno, "API"), msg);
+    va_list ap;
+    va_start(ap, msg);
+    fprintf(outputStream, "%s: ", getBlockName(bno, "API"));
+    vfprintf(outputStream, msg, ap);
+    fprintf(outputStream, "\n", msg);
+    va_end(ap);
   }
 }
 

ndb/src/kernel/blocks/dbdict/Dbdict.hpp

@@ -78,7 +78,8 @@
 /*--------------------------------------------------------------*/
 // Page constants
 /*--------------------------------------------------------------*/
-#define ZALLOCATE 1 //Variable number of page for NDBFS
+#define ZBAT_SCHEMA_FILE 0 //Variable number of page for NDBFS
+#define ZBAT_TABLE_FILE 1 //Variable number of page for NDBFS
 #define ZPAGE_HEADER_SIZE 32
 #define ZPOS_PAGE_SIZE 16
 #define ZPOS_CHECKSUM 17
@@ -92,7 +93,7 @@
 #define ZSIZE_OF_PAGES_IN_WORDS 8192
 #define ZLOG_SIZE_OF_PAGES_IN_WORDS 13
 #define ZMAX_PAGES_OF_TABLE_DEFINITION 8
-#define ZNUMBER_OF_PAGES (2 * ZMAX_PAGES_OF_TABLE_DEFINITION + 2)
+#define ZNUMBER_OF_PAGES (ZMAX_PAGES_OF_TABLE_DEFINITION + 1)
 #define ZNO_OF_FRAGRECORD 5
 
 /*--------------------------------------------------------------*/
@@ -429,6 +430,12 @@ public:
   typedef Ptr<PageRecord> PageRecordPtr;
   CArray<PageRecord> c_pageRecordArray;
 
+  struct SchemaPageRecord {
+    Uint32 word[NDB_SF_PAGE_SIZE_IN_WORDS];
+  };
+
+  CArray<SchemaPageRecord> c_schemaPageRecordArray;
+
   /**
    * A page for create index table signal.
    */
@@ -655,16 +662,20 @@ private:
   struct ReadSchemaRecord {
     /** Page Id of schema page */
     Uint32 pageId;
+    /** First page to read */
+    Uint32 firstPage;
+    /** Number of pages to read */
+    Uint32 noOfPages;
     /** State, indicates from where it was called */
     enum SchemaReadState {
       IDLE = 0,
-      INITIAL_READ = 1
+      INITIAL_READ_HEAD = 1,
+      INITIAL_READ = 2
     };
     SchemaReadState schemaReadState;
   };
   ReadSchemaRecord c_readSchemaRecord;
 
-private:
   /**
    * This record stores all the state needed 
    * when a schema file is being written to disk
@@ -672,6 +683,12 @@ private:
   struct WriteSchemaRecord {
     /** Page Id of schema page */
     Uint32 pageId;
+    /** Rewrite entire file */
+    Uint32 newFile;
+    /** First page to write */
+    Uint32 firstPage;
+    /** Number of pages to write */
+    Uint32 noOfPages;
     /** Schema Files Handled, local state variable */
     Uint32 noOfSchemaFilesHandled;
 
@@ -752,21 +769,33 @@ private:
    * Word 4: Currently zero
    ****************************************************************************/
   struct SchemaRecord {
-    /**    Schema page       */
+    /**    Schema file first page (0)   */
     Uint32 schemaPage;
 
-    /**    Old Schema page (used at node restart)   */
+    /**    Old Schema file first page (used at node restart)    */
     Uint32 oldSchemaPage;
     
     Callback m_callback;
   };
   SchemaRecord c_schemaRecord;
 
-  void initSchemaFile(SchemaFile *, Uint32 sz);
-  void computeChecksum(SchemaFile *);
-  bool validateChecksum(const SchemaFile *);
-  SchemaFile::TableEntry * getTableEntry(void * buf, Uint32 tableId, 
-					 bool allowTooBig = false);
+  /*
+   * Schema file, list of schema pages.  Use an array until a pool
+   * exists and NDBFS interface can use it.
+   */
+  struct XSchemaFile {
+    SchemaFile* schemaPage;
+    Uint32 noOfPages;
+  };
+  // 0-normal 1-old
+  XSchemaFile c_schemaFile[2];
+
+  void initSchemaFile(XSchemaFile *, Uint32 firstPage, Uint32 lastPage,
+                      bool initEntries);
+  void resizeSchemaFile(XSchemaFile * xsf, Uint32 noOfPages);
+  void computeChecksum(XSchemaFile *, Uint32 pageNo);
+  bool validateChecksum(const XSchemaFile *);
+  SchemaFile::TableEntry * getTableEntry(XSchemaFile *, Uint32 tableId);
 
   Uint32 computeChecksum(const Uint32 * src, Uint32 len);
 
@@ -1631,7 +1660,8 @@ private:
   void openSchemaFile(Signal* signal,
                       Uint32 fileNo,
                       Uint32 fsPtr,
-                      bool writeFlag);
+                      bool writeFlag,
+                      bool newFile);
   void writeSchemaFile(Signal* signal, Uint32 filePtr, Uint32 fsPtr);
   void writeSchemaConf(Signal* signal,
                                FsConnectRecordPtr fsPtr);
@@ -1673,6 +1703,7 @@ private:
   void readSchemaRef(Signal* signal, FsConnectRecordPtr fsPtr);
   void closeReadSchemaConf(Signal* signal,
                            FsConnectRecordPtr fsPtr);
+  bool convertSchemaFileTo_5_0_5(XSchemaFile*);
 
   /* ------------------------------------------------------------ */
   // Get table definitions

ndb/src/kernel/blocks/dbdict/SchemaFile.hpp

@@ -18,16 +18,35 @@
 #define DBDICT_SCHEMA_FILE_HPP
 
 #include <ndb_types.h>
+#include <ndb_version.h>
 #include <string.h>
 
+#define NDB_SF_MAGIC                    "NDBSCHMA"
+
+// page size 4k
+#define NDB_SF_PAGE_SIZE_IN_WORDS_LOG2  10
+#define NDB_SF_PAGE_SIZE_IN_WORDS       (1 << NDB_SF_PAGE_SIZE_IN_WORDS_LOG2)
+#define NDB_SF_PAGE_SIZE                (NDB_SF_PAGE_SIZE_IN_WORDS << 2)
+
+// 4k = (1 + 127) * 32
+#define NDB_SF_PAGE_ENTRIES             127
+
+// 160 pages = 20320 objects
+#define NDB_SF_MAX_PAGES                160
+
+// versions where format changed
+#define NDB_SF_VERSION_5_0_5            MAKE_VERSION(5, 0, 5)
+
+// One page in schema file.
 struct SchemaFile {
+  // header size 32 bytes
   char Magic[8];
   Uint32 ByteOrder;
   Uint32 NdbVersion;
   Uint32 FileSize; // In bytes
-  Uint32 Unused;
-  
-  Uint32 CheckSum;
+  Uint32 PageNumber;
+  Uint32 CheckSum; // Of this page
+  Uint32 NoOfTableEntries; // On this page (NDB_SF_PAGE_ENTRIES)
   
   enum TableState {
     INIT = 0,
@@ -38,20 +57,33 @@ struct SchemaFile {
     ALTER_TABLE_COMMITTED = 5
   };
 
+  // entry size 32 bytes
   struct TableEntry {
     Uint32 m_tableState;
     Uint32 m_tableVersion;
     Uint32 m_tableType;
     Uint32 m_noOfPages;
     Uint32 m_gcp;
+    Uint32 m_unused[3];
     
     bool operator==(const TableEntry& o) const { 
       return memcmp(this, &o, sizeof(* this))== 0;
     }
   };
+
+  // pre-5.0.5
+  struct TableEntry_old {
+    Uint32 m_tableState;
+    Uint32 m_tableVersion;
+    Uint32 m_tableType;
+    Uint32 m_noOfPages;
+    Uint32 m_gcp;
+  };
   
-  Uint32 NoOfTableEntries;
-  TableEntry TableEntries[1];
+  union {
+  TableEntry TableEntries[NDB_SF_PAGE_ENTRIES];
+  TableEntry_old TableEntries_old[1];
+  };
 };
 
 #endif

ndb/src/kernel/blocks/dbdict/printSchemaFile.cpp

 #if 0
 make -f Makefile -f - printSchemaFile <<'_eof_'
-printSchemaFile: printSchemaFile.cpp
+printSchemaFile: printSchemaFile.cpp SchemaFile.hpp
 	$(CXXCOMPILE) -o $@ $@.cpp -L../../../common/util/.libs -lgeneral
 _eof_
 exit $?
@@ -24,19 +24,28 @@ exit $?
 
 
 #include <ndb_global.h>
+#include <ndb_version.h>
 
 #include <NdbMain.h>
 #include <NdbOut.hpp>
 #include <SchemaFile.hpp>
 
-void 
-usage(const char * prg){
-  ndbout << "Usage " << prg 
-	 << " P0.SchemaLog" << endl;  
+static const char* progname = 0;
+static bool allflag = false;
+static bool checkonly = false;
+static int xitcode = 0;
+
+static void 
+usage()
+{
+  ndbout << "Usage " << progname
+         << " [-ac]"
+         << " P0.SchemaLog" << endl;  
 }
 
-void
-fill(const char * buf, int mod){
+static void
+fill(const char * buf, int mod)
+{
   int len = strlen(buf)+1;
   ndbout << buf << " ";
   while((len % mod) != 0){
@@ -45,19 +54,34 @@ fill(const char * buf, int mod){
   }
 }
 
-void 
-print(const char * filename, const SchemaFile * file){
-  ndbout << "----- Schemafile: " << filename << " -----" << endl;
-  ndbout_c("Magic: %.*s ByteOrder: %.8x NdbVersion: %d FileSize: %d",
-	   sizeof(file->Magic), file->Magic, 
-	   file->ByteOrder, 
-	   file->NdbVersion,
-	   file->FileSize);
-
-  for(Uint32 i = 0; i<file->NoOfTableEntries; i++){
-    SchemaFile::TableEntry te = file->TableEntries[i];
-    if(te.m_tableState != SchemaFile::INIT){
-      ndbout << "Table " << i << ": State = " << te.m_tableState 
+static void
+print_head(const char * filename, const SchemaFile * sf)
+{
+  if (! checkonly) {
+    ndbout << "----- Schemafile: " << filename << " -----" << endl;
+    ndbout_c("Magic: %.*s ByteOrder: %.8x NdbVersion: %d.%d.%d FileSize: %d",
+             sizeof(sf->Magic),
+             sf->Magic, 
+             sf->ByteOrder, 
+             sf->NdbVersion >> 16,
+             (sf->NdbVersion >> 8) & 0xFF,
+             sf->NdbVersion & 0xFF,
+             sf->FileSize);
+  }
+}
+
+static void 
+print_old(const char * filename, const SchemaFile * sf)
+{
+  print_head(filename, sf);
+
+  for (Uint32 i = 0; i < sf->NoOfTableEntries; i++) {
+    SchemaFile::TableEntry_old te = sf->TableEntries_old[i];
+    if (allflag ||
+        (te.m_tableState != SchemaFile::INIT &&
+         te.m_tableState != SchemaFile::DROP_TABLE_COMMITTED)) {
+      ndbout << "Table " << i << ":"
+             << " State = " << te.m_tableState 
 	     << " version = " << te.m_tableVersion
              << " type = " << te.m_tableType
 	     << " noOfPages = " << te.m_noOfPages
@@ -66,47 +90,114 @@ print(const char * filename, const SchemaFile * file){
   }
 }
 
-NDB_COMMAND(printSchemafile, 
-	    "printSchemafile", "printSchemafile", "Prints a schemafile", 16384){ 
-  if(argc < 2){
-    usage(argv[0]);
-    return 0;
-  }
+static void 
+print(const char * filename, const SchemaFile * xsf, Uint32 sz)
+{
+  int retcode = 0;
 
-  const char * filename = argv[1];
+  print_head(filename, xsf);
 
-  struct stat sbuf;
-  const int res = stat(filename, &sbuf);
-  if(res != 0){
-    ndbout << "Could not find file: \"" << filename << "\"" << endl;
-    return 0;
+  assert(sizeof(SchemaFile) == NDB_SF_PAGE_SIZE);
+  if (xsf->FileSize != sz || xsf->FileSize % NDB_SF_PAGE_SIZE != 0) {
+    ndbout << "***** invalid FileSize " << xsf->FileSize << endl;
+    retcode = 1;
   }
-  const Uint32 bytes = sbuf.st_size;
-  
-  Uint32 * buf = new Uint32[bytes/4+1];
-  
-  FILE * f = fopen(filename, "rb");
-  if(f == 0){
-    ndbout << "Failed to open file" << endl;
-    delete [] buf;
-    return 0;
+  Uint32 noOfPages = xsf->FileSize / NDB_SF_PAGE_SIZE;
+  for (Uint32 n = 0; n < noOfPages; n++) {
+    if (! checkonly) {
+      ndbout << "----- Page: " << n << " (" << noOfPages << ") -----" << endl;
+    }
+    const SchemaFile * sf = &xsf[n];
+    if (sf->FileSize != xsf->FileSize) {
+      ndbout << "***** page " << n << " FileSize changed to " << sf->FileSize << "!=" << xsf->FileSize << endl;
+      retcode = 1;
+    }
+    Uint32 cs = 0;
+    for (Uint32 j = 0; j < NDB_SF_PAGE_SIZE_IN_WORDS; j++)
+      cs ^= ((const Uint32*)sf)[j];
+    if (cs != 0) {
+      ndbout << "***** page " << n << " invalid CheckSum" << endl;
+      retcode = 1;
+    }
+    if (sf->NoOfTableEntries != NDB_SF_PAGE_ENTRIES) {
+      ndbout << "***** page " << n << " invalid NoOfTableEntries " << sf->NoOfTableEntries << endl;
+      retcode = 1;
+    }
+    for (Uint32 i = 0; i < NDB_SF_PAGE_ENTRIES; i++) {
+      SchemaFile::TableEntry te = sf->TableEntries[i];
+      Uint32 j = n * NDB_SF_PAGE_ENTRIES + i;
+      if (allflag ||
+          (te.m_tableState != SchemaFile::INIT &&
+           te.m_tableState != SchemaFile::DROP_TABLE_COMMITTED)) {
+        if (! checkonly)
+          ndbout << "Table " << j << ":"
+                 << " State = " << te.m_tableState 
+                 << " version = " << te.m_tableVersion
+                 << " type = " << te.m_tableType
+                 << " noOfPages = " << te.m_noOfPages
+                 << " gcp: " << te.m_gcp << endl;
+      }
+      if (te.m_unused[0] != 0 || te.m_unused[1] != 0 || te.m_unused[2] != 0) {
+        ndbout << "***** entry " << j << " garbage in m_unused[3]" << endl;
+        retcode = 1;
+      }
+    }
   }
-  Uint32 sz = fread(buf, 1, bytes, f);
-  fclose(f);
-  if(sz != bytes){
-    ndbout << "Failure while reading file" << endl;
-    delete [] buf;
-    return 0;
+
+  if (retcode != 0)
+    xitcode = 1;
+  else if (checkonly)
+    ndbout << "ok: " << filename << endl;
+}
+
+NDB_COMMAND(printSchemafile, 
+	    "printSchemafile", "printSchemafile", "Prints a schemafile", 16384)
+{ 
+  progname = argv[0];
+
+  while (argv[1][0] == '-') {
+    if (strchr(argv[1], 'a') != 0)
+      allflag = true;
+    if (strchr(argv[1], 'c') != 0)
+      checkonly = true;
+    argc--, argv++;
   }
-  
-  print(filename, (SchemaFile *)&buf[0]);
 
-  Uint32 chk = 0, i;
-  for (i = 0; i < bytes/4; i++)
-    chk ^= buf[i];
-  if (chk != 0)
-    ndbout << "Invalid checksum!" << endl;
+  while (argc > 1) {
+    const char * filename = argv[1];
+    argc--, argv++;
+
+    struct stat sbuf;
+    const int res = stat(filename, &sbuf);
+    if (res != 0) {
+      ndbout << "Could not find file: \"" << filename << "\"" << endl;
+      return 1;
+    }
+    const Uint32 bytes = sbuf.st_size;
+    
+    Uint32 * buf = new Uint32[bytes/4+1];
+    
+    FILE * f = fopen(filename, "rb");
+    if (f == 0) {
+      ndbout << "Failed to open file" << endl;
+      delete [] buf;
+      return 1;
+    }
+    Uint32 sz = fread(buf, 1, bytes, f);
+    fclose(f);
+    if (sz != bytes) {
+      ndbout << "Failure while reading file" << endl;
+      delete [] buf;
+      return 1;
+    }
+
+    SchemaFile* sf = (SchemaFile *)&buf[0];
+    if (sf->NdbVersion < NDB_SF_VERSION_5_0_5)
+      print_old(filename, sf);
+    else
+      print(filename, sf, sz);
+    delete [] buf;
+  }
 
-  delete [] buf;
-  return 0;
+  return xitcode;
 }

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -208,11 +208,13 @@ NdbColumnImpl::equal(const NdbColumnImpl& col) const
   if(m_nullable != col.m_nullable){
     DBUG_RETURN(false);
   }
+#ifdef ndb_dictionary_dkey_fixed
   if(m_pk){
     if(m_distributionKey != col.m_distributionKey){
       DBUG_RETURN(false);
     }
   }
+#endif
   if (m_precision != col.m_precision ||
       m_scale != col.m_scale ||
       m_length != col.m_length ||

ndb/test/ndbapi/testDict.cpp

@@ -428,103 +428,99 @@ int runUseTableUntilStopped(NDBT_Context* ctx, NDBT_Step* step){
 }
 
 
-int runCreateMaxTables(NDBT_Context* ctx, NDBT_Step* step){
-  int failures = 0;
+int
+runCreateMaxTables(NDBT_Context* ctx, NDBT_Step* step)
+{
   char tabName[256];
   int numTables = ctx->getProperty("tables", 1000);
   Ndb* pNdb = GETNDB(step);
-
-  for (int i = 0; i < numTables && failures < 5; i++){
+  NdbDictionary::Dictionary* pDic = pNdb->getDictionary();
+  int i = 0;
+  for (i = 0; i < numTables; i++) {
     BaseString::snprintf(tabName, 256, "MAXTAB%d", i);
-
-    if (pNdb->waitUntilReady(30) != 0){
+    if (pNdb->waitUntilReady(30) != 0) {
       // Db is not ready, return with failure
       return NDBT_FAILED;
     }
-  
     const NdbDictionary::Table* pTab = ctx->getTab();
-    ndbout << "|- " << tabName << endl;
-
+    //ndbout << "|- " << tabName << endl;
     // Set new name for T1
     NdbDictionary::Table newTab(* pTab);
     newTab.setName(tabName);
-
+    // Drop any old (or try to)
+    (void)pDic->dropTable(newTab.getName());
     // Try to create table in db
-    if (newTab.createTableInDb(pNdb) != 0){
-      ndbout << tabName << " coult not be created"<< endl;
-      failures++;
-      continue;
+    if (newTab.createTableInDb(pNdb) != 0) {
+      ndbout << tabName << " could not be created: "
+             << pDic->getNdbError() << endl;
+      if (pDic->getNdbError().code == 707 ||
+          pDic->getNdbError().code == 708 ||
+          pDic->getNdbError().code == 826 ||
+          pDic->getNdbError().code == 827)
+        break;
+      return NDBT_FAILED;
     }
-    
     // Verify that table exists in db    
     const NdbDictionary::Table* pTab3 = 
       NDBT_Table::discoverTableFromDb(pNdb, tabName) ;
     if (pTab3 == NULL){
-      ndbout << tabName << " was not found in DB"<< endl;
-      failures++;
-      continue;
+      ndbout << tabName << " was not found in DB: "
+             << pDic->getNdbError() << endl;
+      return NDBT_FAILED;
     }
-    
-    if (pTab->equal(*pTab3) == false){
-      ndbout << "It was not equal" << endl;
-      failures++;
+    if (! newTab.equal(*pTab3)) {
+      ndbout << "It was not equal" << endl; abort();
+      return NDBT_FAILED;
     }
-
-    int records = 1000;
+    int records = ctx->getNumRecords();
     HugoTransactions hugoTrans(*pTab3);
-    if (hugoTrans.loadTable(pNdb, records) != 0){
+    if (hugoTrans.loadTable(pNdb, records) != 0) {
       ndbout << "It can NOT be loaded" << endl;
-    } else{
-      ndbout << "It can be loaded" << endl;
-      
-      UtilTransactions utilTrans(*pTab3);
-      if (utilTrans.clearTable(pNdb, records, 64) != 0){
-	ndbout << "It can NOT be cleared" << endl;
-      } else{
-	ndbout << "It can be cleared" << endl;
-      }	  
+      return NDBT_FAILED;
+    }
+    UtilTransactions utilTrans(*pTab3);
+    if (utilTrans.clearTable(pNdb, records, 64) != 0) {
+      ndbout << "It can NOT be cleared" << endl;
+      return NDBT_FAILED;
     }
-    
   }
-  if (pNdb->waitUntilReady(30) != 0){
+  if (pNdb->waitUntilReady(30) != 0) {
     // Db is not ready, return with failure
     return NDBT_FAILED;
   }
+  ctx->setProperty("maxtables", i);
   // HURRAAA!
   return NDBT_OK;
 }
 
-int runDropMaxTables(NDBT_Context* ctx, NDBT_Step* step){
-  int result = NDBT_OK;
+int runDropMaxTables(NDBT_Context* ctx, NDBT_Step* step)
+{
   char tabName[256];
-  int numTables = ctx->getProperty("tables", 1000);
+  int numTables = ctx->getProperty("maxtables", (Uint32)0);
   Ndb* pNdb = GETNDB(step);
-
-  for (int i = 0; i < numTables; i++){
+  NdbDictionary::Dictionary* pDic = pNdb->getDictionary();
+  for (int i = 0; i < numTables; i++) {
     BaseString::snprintf(tabName, 256, "MAXTAB%d", i);
-
-    if (pNdb->waitUntilReady(30) != 0){
+    if (pNdb->waitUntilReady(30) != 0) {
       // Db is not ready, return with failure
       return NDBT_FAILED;
     }
-  
     // Verify that table exists in db    
     const NdbDictionary::Table* pTab3 = 
       NDBT_Table::discoverTableFromDb(pNdb, tabName) ;
-    if (pTab3 == NULL){
-      ndbout << tabName << " was not found in DB"<< endl;
-      continue;
+    if (pTab3 == NULL) {
+      ndbout << tabName << " was not found in DB: "
+             << pDic->getNdbError() << endl;
+      return NDBT_FAILED;
     }
-
-
     // Try to drop table in db
-    if (pNdb->getDictionary()->dropTable(pTab3->getName()) != 0){
-      ndbout << tabName << " coult not be dropped"<< endl;
-      result = NDBT_FAILED;
+    if (pDic->dropTable(pTab3->getName()) != 0) {
+      ndbout << tabName << " could not be dropped: "
+             << pDic->getNdbError() << endl;
+      return NDBT_FAILED;
     }
-    
   }
-  return result;
+  return NDBT_OK;
 }
 
 int runTestFragmentTypes(NDBT_Context* ctx, NDBT_Step* step){
@@ -1622,7 +1618,7 @@ TESTCASE("CreateMaxTables",
 	 "Create tables until db says that it can't create any more\n"){
   TC_PROPERTY("tables", 1000);
   INITIALIZER(runCreateMaxTables);
-  FINALIZER(runDropMaxTables);
+  INITIALIZER(runDropMaxTables);
 }
 TESTCASE("PkSizes", 
 	 "Create tables with all different primary key sizes.\n"\