Bug #16851, log floods with ndb discover messages

+ fix crashing bug when discovering ndb tables outside select context

sql/ha_ndbcluster.cc

@@ -5388,6 +5388,18 @@ static void ndbcluster_drop_database(char *path)
 /*
   find all tables in ndb and discover those needed
 */
+int ndb_create_table_from_engine(THD *thd, const char *db,
+                                 const char *table_name)
+{
+  LEX *old_lex= thd->lex, newlex;
+  thd->lex= &newlex;
+  newlex.current_select= NULL;
+  lex_start(thd, (const uchar*) "", 0);
+  int res= ha_create_table_from_engine(thd, db, table_name);
+  thd->lex= old_lex;
+  return res;
+}
+
 int ndbcluster_find_all_files(THD *thd)
 {
   DBUG_ENTER("ndbcluster_find_all_files");
@@ -5440,8 +5452,15 @@ int ndbcluster_find_all_files(THD *thd)
       if (ndbtab->getFrmLength() == 0)
         continue;
     
-      strxnmov(key, FN_LEN-1, mysql_data_home, "/",
-               elmt.database, "/", elmt.name, NullS);
+      /* check if database exists */
+      char *end= strxnmov(key, FN_LEN-1, mysql_data_home, "/",
+                          elmt.database, NullS);
+      if (my_access(key, F_OK))
+      {
+        /* no such database defined, skip table */
+        continue;
+      }
+      strxnmov(end, FN_LEN-1-(key-end), "/", elmt.name, NullS);
       const void *data= 0, *pack_data= 0;
       uint length, pack_length;
       int discover= 0;
@@ -5471,7 +5490,7 @@ int ndbcluster_find_all_files(THD *thd)
       {
         /* ToDo 4.1 database needs to be created if missing */
         pthread_mutex_lock(&LOCK_open);
-        if (ha_create_table_from_engine(thd, elmt.database, elmt.name))
+        if (ndb_create_table_from_engine(thd, elmt.database, elmt.name))
         {
           /* ToDo 4.1 handle error */
         }
@@ -5699,7 +5718,7 @@ int ndbcluster_find_files(THD *thd,const char *db,const char *path,
   while ((file_name=it2++))
   {  
     DBUG_PRINT("info", ("Table %s need discovery", file_name));
-    if (ha_create_table_from_engine(thd, db, file_name) == 0)
+    if (ndb_create_table_from_engine(thd, db, file_name) == 0)
       files->push_back(thd->strdup(file_name)); 
   }
 

sql/ha_ndbcluster_binlog.cc

@@ -1314,7 +1314,7 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
           /* fall through */
         case SOT_ALTER_TABLE:
           pthread_mutex_lock(&LOCK_open);
-          if (ha_create_table_from_engine(thd, schema->db, schema->name))
+          if (ndb_create_table_from_engine(thd, schema->db, schema->name))
           {
             sql_print_error("Could not discover table '%s.%s' from "
                             "binlog schema event '%s' from node %d",

sql/ha_ndbcluster_binlog.h

@@ -90,7 +90,8 @@ int ndbcluster_handle_drop_table(Ndb *ndb, const char *event_name,
                                  NDB_SHARE *share);
 void ndb_rep_event_name(String *event_name,
                         const char *db, const char *tbl);
-
+int ndb_create_table_from_engine(THD *thd, const char *db,
+                                 const char *table_name);
 int ndbcluster_binlog_start();
 pthread_handler_t ndb_binlog_thread_func(void *arg);