bug #21495 Alter table from x engine to ndb and back can cause issue with drop DB:disable distributed drop database if a database contains local tables

4 files changed, 56 insertions, 14 deletions

diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index 3dfca5d1bb2..350c031d6a2 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -16,6 +16,7 @@
 */
 
 #include "mysql_priv.h"
+#include "sql_show.h"
 #ifdef WITH_NDBCLUSTER_STORAGE_ENGINE
 #include "ha_ndbcluster.h"
 
@@ -1828,14 +1829,25 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
           log_query= 1;
           break;
         case SOT_DROP_DB:
-          run_query(thd, schema->query,
-                    schema->query + schema->query_length,
-                    TRUE,    /* print error */
-                    TRUE);   /* don't binlog the query */
-          /* binlog dropping database after any table operations */
-          post_epoch_log_list->push_back(schema, mem_root);
-          /* acknowledge this query _after_ epoch completion */
-          post_epoch_unlock= 1;
+	  /* Drop the database locally if it only contains ndb tables */
+	  if (! ndbcluster_check_if_local_tables_in_db(thd, schema->db))
+	  {
+	    run_query(thd, schema->query,
+		      schema->query + schema->query_length,
+		      TRUE,    /* print error */
+		      TRUE);   /* don't binlog the query */
+	    /* binlog dropping database after any table operations */
+	    post_epoch_log_list->push_back(schema, mem_root);
+	    /* acknowledge this query _after_ epoch completion */
+	    post_epoch_unlock= 1;
+	  }
+	  else
+	  {
+	    /*
+	      Database contained local tables, leave it
+	     */
+	    log_query= 1;
+	  }
           break;
         case SOT_CREATE_DB:
           /* fall through */
@@ -2334,6 +2346,32 @@ ndbcluster_check_if_local_table(const char *dbname, const char *tabname)
   DBUG_RETURN(false);
 }
 
+bool
+ndbcluster_check_if_local_tables_in_db(THD *thd, const char *dbname)
+{
+  DBUG_ENTER("ndbcluster_check_if_local_tables_in_db");
+  DBUG_PRINT("info", ("Looking for files in directory %s", dbname));
+  char *tabname;
+  List<char> files;
+  char path[FN_REFLEN];
+
+  build_table_filename(path, sizeof(path), dbname, "", "", 0);
+  if (find_files(thd, &files, dbname, path, NullS, 0) != FIND_FILES_OK)
+  {
+    DBUG_PRINT("info", ("Failed to find files"));
+    DBUG_RETURN(true);
+  }
+  DBUG_PRINT("info",("found: %d files", files.elements));
+  while ((tabname= files.pop()))
+  {
+    DBUG_PRINT("info", ("Found table %s", tabname));
+    if (ndbcluster_check_if_local_table(dbname, tabname))
+      DBUG_RETURN(true);
+  }
+  
+  DBUG_RETURN(false);
+}
+
 /*
   Common function for setting up everything for logging a table at
   create/discover.
diff --git a/sql/ha_ndbcluster_binlog.h b/sql/ha_ndbcluster_binlog.h
index 233d1a58aaa..5cf3cedf03b 100644
--- a/sql/ha_ndbcluster_binlog.h
+++ b/sql/ha_ndbcluster_binlog.h
@@ -124,6 +124,7 @@ void ndbcluster_binlog_init_handlerton();
 void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *table);
 
 bool ndbcluster_check_if_local_table(const char *dbname, const char *tabname);
+bool ndbcluster_check_if_local_tables_in_db(THD *thd, const char *dbname);
 
 int ndbcluster_create_binlog_setup(Ndb *ndb, const char *key,
                                    uint key_len,
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index c39cdfa6d7e..6cb2ed6a826 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -491,13 +491,7 @@ bool mysqld_show_column_types(THD *thd)
     FIND_FILES_DIR      no such directory, or directory can't be read
 */
 
-enum find_files_result {
-  FIND_FILES_OK,
-  FIND_FILES_OOM,
-  FIND_FILES_DIR
-};
 
-static
 find_files_result
 find_files(THD *thd, List<char> *files, const char *db,
            const char *path, const char *wild, bool dir)
diff --git a/sql/sql_show.h b/sql/sql_show.h
index 681d1232b39..29cd52eb9fd 100644
--- a/sql/sql_show.h
+++ b/sql/sql_show.h
@@ -10,6 +10,15 @@ struct st_table_list;
 typedef st_ha_create_information HA_CREATE_INFO;
 typedef st_table_list TABLE_LIST;
 
+enum find_files_result {
+  FIND_FILES_OK,
+  FIND_FILES_OOM,
+  FIND_FILES_DIR
+};
+
+find_files_result find_files(THD *thd, List<char> *files, const char *db,
+                             const char *path, const char *wild, bool dir);
+
 int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
                       HA_CREATE_INFO  *create_info_arg);
 int view_store_create_info(THD *thd, TABLE_LIST *table, String *buff);