Bug #16875 Using stale MySQLD FRM files can cause restored cluster to fail

sql/ha_ndbcluster_binlog.cc:
  Bug #16875 Using stale MySQLD FRM files can cause restored cluster to fail
  - small correction of previous patch
storage/ndb/src/ndbapi/DictCache.cpp:
  Bug #16875 Using stale MySQLD FRM files can cause restored cluster to fail
  - cache not released properly
storage/ndb/tools/desc.cpp:
  Bug #16875 Using stale MySQLD FRM files can cause restored cluster to fail
  - added retry option so that ndb_desc can be used to wait for a table to be created

mysql-test/r/ndb_autodiscover3.result

@@ -14,6 +14,8 @@ a	b
 1	1
 2	1
 3	1
+select * from t2;
+ERROR 42S02: Table 'test.t2' doesn't exist
 show tables like 't2';
 Tables_in_test (t2)
 create table t2 (a int key) engine=ndbcluster;
@@ -28,6 +30,8 @@ a
 1
 2
 3
+select * from t2;
+ERROR 42S02: Table 'test.t2' doesn't exist
 show tables like 't2';
 Tables_in_test (t2)
 create table t2 (a int key) engine=ndbcluster;

mysql-test/t/ndb_autodiscover3.test

@@ -37,9 +37,11 @@ select * from t2 order by a limit 3;
 --exec $NDB_MGM --no-defaults -e "all restart -i" >> $NDB_TOOLS_OUTPUT
 --exec $NDB_TOOLS_DIR/ndb_waiter --no-defaults >> $NDB_TOOLS_OUTPUT
 # to ensure mysqld has connected again, and recreated system tables
---sleep 3
+--exec $NDB_TOOLS_DIR/ndb_desc --no-defaults -r 30 -d cluster apply_status >> $NDB_TOOLS_OUTPUT
 
 --connection server2
+--error ER_NO_SUCH_TABLE
+select * from t2;
 show tables like 't2';
 create table t2 (a int key) engine=ndbcluster;
 insert into t2 values (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);
@@ -52,9 +54,11 @@ select * from t2 order by a limit 3;
 --exec $NDB_MGM --no-defaults -e "all restart -i" >> $NDB_TOOLS_OUTPUT
 --exec $NDB_TOOLS_DIR/ndb_waiter --no-defaults >> $NDB_TOOLS_OUTPUT
 # to ensure mysqld has connected again, and recreated system tables
---sleep 3
+--exec $NDB_TOOLS_DIR/ndb_desc --no-defaults -r 30 -d cluster apply_status >> $NDB_TOOLS_OUTPUT
 
 --connection server1
+--error ER_NO_SUCH_TABLE
+select * from t2;
 show tables like 't2';
 create table t2 (a int key) engine=ndbcluster;
 insert into t2 values (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);

sql/ha_ndbcluster_binlog.cc

@@ -3318,6 +3318,7 @@ restart:
       schema_res= s_ndb->pollEvents(100, &schema_gci);
     }
     // now check that we have epochs consistant with what we had before the restart
+    DBUG_PRINT("info", ("schema_res: %d  schema_gci: %d", schema_res, schema_gci));
     if (schema_res > 0)
     {
       if (schema_gci < ndb_latest_handled_binlog_epoch)
@@ -3681,7 +3682,7 @@ restart:
     *root_ptr= old_root;
     ndb_latest_handled_binlog_epoch= ndb_latest_received_binlog_epoch;
   }
-  if (do_ndbcluster_binlog_close_connection != BCCC_exit)
+  if (do_ndbcluster_binlog_close_connection == BCCC_restart)
     goto restart;
 err:
   DBUG_PRINT("info",("Shutting down cluster binlog thread"));

storage/ndb/src/ndbapi/DictCache.cpp

@@ -178,6 +178,11 @@ GlobalDictCache::get(const char * name)
       {
         ver->m_status = DROPPED;
         retreive = true; // Break loop
+        if (ver->m_refCount == 0)
+        {
+          delete ver->m_impl;
+          versions->erase(versions->size() - 1);
+        }
         break;
       }
       ver->m_refCount++;
@@ -289,6 +294,10 @@ GlobalDictCache::get_size()
     sz += curr->theData->size();
     curr = m_tableHash.getNext(curr);
   }
+  if (sz)
+  {
+    printCache();
+  }
   return sz;
 }
 
@@ -408,6 +417,11 @@ GlobalDictCache::alter_table_rep(const char * name,
       ver.m_status = DROPPED;
       ver.m_impl->m_status = altered ? 
 	NdbDictionary::Object::Altered : NdbDictionary::Object::Invalid;
+      if (ver.m_refCount == 0)
+      {
+        delete ver.m_impl;
+        vers->erase(i);
+      }
       DBUG_VOID_RETURN;
     }
 

storage/ndb/tools/desc.cpp

@@ -18,6 +18,7 @@
 #include <ndb_opts.h>
 #include <NDBT.hpp>
 #include <NdbApi.hpp>
+#include <NdbSleep.h>
 
 void desc_AutoGrowSpecification(struct NdbDictionary::AutoGrowSpecification ags);
 int desc_logfilegroup(Ndb *myndb, char* name);
@@ -31,6 +32,7 @@ NDB_STD_OPTS_VARS;
 static const char* _dbname = "TEST_DB";
 static int _unqualified = 0;
 static int _partinfo = 0;
+static int _retries = 0;
 static struct my_option my_long_options[] =
 {
   NDB_STD_OPTS("ndb_desc"),
@@ -43,6 +45,9 @@ static struct my_option my_long_options[] =
   { "extra-partition-info", 'p', "Print more info per partition",
     (gptr*) &_partinfo, (gptr*) &_partinfo, 0,
     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 }, 
+  { "retries", 'r', "Retry every second for # retries",
+    (gptr*) &_retries, (gptr*) &_retries, 0,
+    GET_INT, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 }, 
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 static void usage()
@@ -231,7 +236,8 @@ int desc_datafile(Ndb_cluster_connection &con, Ndb *myndb, char* name)
 int desc_table(Ndb *myndb, char* name)
 {
   NdbDictionary::Dictionary * dict= myndb->getDictionary();
-  NDBT_Table* pTab = (NDBT_Table*)dict->getTable(name);
+  NDBT_Table* pTab;
+  while ((pTab = (NDBT_Table*)dict->getTable(name)) == NULL && --_retries >= 0) NdbSleep_SecSleep(1);
   if (!pTab)
     return 0;