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

mysql-test/r/ndb_multi.result

@@ -97,3 +97,27 @@ c1
 3
 5
 drop table t1;
+create database db;
+use db;
+create table t1(x int) engine=ndb;
+use db;
+show tables;
+Tables_in_db
+t1
+drop database db;
+show tables;
+ERROR 42000: Unknown database 'db'
+create database db;
+use db;
+create table t1(x int) engine=ndb;
+use db;
+create table t2(x int) engine=myisam;
+show tables;
+Tables_in_db
+t1
+t2
+drop database db;
+show tables;
+Tables_in_db
+t2
+drop database db;

mysql-test/t/ndb_multi.test

@@ -87,3 +87,40 @@ connection server1;
 select * from t1 order by c1;
 drop table t1;
 # End of 4.1 tests
+
+# Check distributed drop of database in 5.1
+create database db;
+use db;
+create table t1(x int) engine=ndb;
+
+connection server2;
+use db;
+show tables;
+
+connection server1;
+drop database db;
+
+connection server2;
+--error 1049
+show tables;
+
+connection server1;
+
+# bug#21495
+create database db;
+use db;
+create table t1(x int) engine=ndb;
+
+connection server2;
+use db;
+create table t2(x int) engine=myisam;
+show tables;
+
+connection server1;
+drop database db;
+
+connection server2;
+show tables;
+drop database db;
+
+connection server1;

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.

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,

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)

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);