Bug #57663 Concurrent statement using stored function and DROP DATABASE

           breaks SBR

The problem was that DROP DATABASE ignored any metadata locks on stored
functions and procedures held by other connections. This made it
possible for DROP DATABASE to drop functions/procedures that were in use
by other connections and therefore break statement based replication.
(DROP DATABASE could appear in the binlog before a statement using a
dropped function/procedure.)

This problem was an issue left unresolved by the patch for Bug#30977
where metadata locks for stored functions/procedures were introduced.

This patch fixes the problem by making sure DROP DATABASE takes
exclusive metadata locks on all stored functions/procedures to be
dropped.

Test case added to sp-lock.test.

mysql-test/r/sp-lock.result

@@ -735,5 +735,96 @@ END	latin1	latin1_swedish_ci	latin1_swedish_ci
 # Connection default;
 DROP PROCEDURE p1;
 #
+# Bug#57663 Concurrent statement using stored function and DROP DATABASE
+#           breaks SBR
+#
+DROP DATABASE IF EXISTS db1;
+DROP FUNCTION IF EXISTS f1;
+# Test 1: Check that DROP DATABASE block if a function is used
+#         by an active transaction.
+# Connection default
+CREATE DATABASE db1;
+CREATE FUNCTION db1.f1() RETURNS INTEGER RETURN 1;
+START TRANSACTION;
+SELECT db1.f1();
+db1.f1()
+1
+# Connection con1
+# Sending:
+DROP DATABASE db1;
+# Connection default
+# Waiting for DROP DATABASE to be blocked by the lock on f1()
+COMMIT;
+# Connection con1
+# Reaping: DROP DATABASE db1
+# Test 2: Check that DROP DATABASE blocks if a procedure is
+#         used by an active transaction.
+# Connection default
+CREATE DATABASE db1;
+CREATE PROCEDURE db1.p1() BEGIN END;
+CREATE FUNCTION f1() RETURNS INTEGER
+BEGIN
+CALL db1.p1();
+RETURN 1;
+END|
+START TRANSACTION;
+SELECT f1();
+f1()
+1
+# Connection con1
+# Sending:
+DROP DATABASE db1;
+# Connection default
+# Waiting for DROP DATABASE to be blocked by the lock on p1()
+COMMIT;
+# Connection con1
+# Reaping: DROP DATABASE db1
+# Test 3: Check that DROP DATABASE is not selected as a victim if a
+#         deadlock is discovered with DML statements.
+# Connection default
+CREATE DATABASE db1;
+CREATE TABLE db1.t1 (a INT);
+CREATE FUNCTION db1.f1() RETURNS INTEGER RETURN 1;
+START TRANSACTION;
+SELECT db1.f1();
+db1.f1()
+1
+# Connection con1
+# Sending:
+DROP DATABASE db1;
+# Connection default
+# Waiting for DROP DATABASE to be blocked by the lock on f1()
+SELECT * FROM db1.t1;
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+COMMIT;
+# Connection con1
+# Reaping: DROP DATABASE db1
+# Test 4: Check that active DROP DATABASE blocks stored routine DDL.
+# Connection default
+CREATE DATABASE db1;
+CREATE FUNCTION db1.f1() RETURNS INTEGER RETURN 1;
+CREATE FUNCTION db1.f2() RETURNS INTEGER RETURN 2;
+START TRANSACTION;
+SELECT db1.f2();
+db1.f2()
+2
+# Connection con1
+# Sending:
+DROP DATABASE db1;
+# Connection con2
+# Waiting for DROP DATABASE to be blocked by the lock on f2()
+# Sending:
+ALTER FUNCTION db1.f1 COMMENT "test";
+# Connection default
+# Waiting for ALTER FUNCTION to be blocked by the schema lock on db1
+COMMIT;
+# Connection con1
+# Reaping: DROP DATABASE db1
+# Connection con2
+# Reaping: ALTER FUNCTION f1 COMMENT 'test'
+ERROR 42000: FUNCTION db1.f1 does not exist
+# Connection default
+DROP FUNCTION f1;
+#
 # End of 5.5 tests
 #

mysql-test/t/sp-lock.test

@@ -971,6 +971,171 @@ connection default;
 DROP PROCEDURE p1;
 
 
+--echo #
+--echo # Bug#57663 Concurrent statement using stored function and DROP DATABASE
+--echo #           breaks SBR
+--echo #
+
+--disable_warnings
+DROP DATABASE IF EXISTS db1;
+DROP FUNCTION IF EXISTS f1;
+--enable_warnings
+
+connect(con1, localhost, root);
+connect(con2, localhost, root);
+
+--echo # Test 1: Check that DROP DATABASE block if a function is used
+--echo #         by an active transaction.
+
+--echo # Connection default
+connection default;
+CREATE DATABASE db1;
+CREATE FUNCTION db1.f1() RETURNS INTEGER RETURN 1;
+START TRANSACTION;
+SELECT db1.f1();
+
+--echo # Connection con1
+connection con1;
+--echo # Sending:
+--send DROP DATABASE db1
+
+--echo # Connection default
+connection default;
+--echo # Waiting for DROP DATABASE to be blocked by the lock on f1()
+let $wait_condition= SELECT COUNT(*)= 1 FROM information_schema.processlist
+  WHERE state= 'Waiting for stored function metadata lock'
+  AND info='DROP DATABASE db1';
+--source include/wait_condition.inc
+COMMIT;
+
+--echo # Connection con1
+connection con1;
+--echo # Reaping: DROP DATABASE db1
+--reap
+
+--echo # Test 2: Check that DROP DATABASE blocks if a procedure is
+--echo #         used by an active transaction.
+
+--echo # Connection default
+connection default;
+CREATE DATABASE db1;
+CREATE PROCEDURE db1.p1() BEGIN END;
+delimiter |;
+CREATE FUNCTION f1() RETURNS INTEGER
+BEGIN
+  CALL db1.p1();
+  RETURN 1;
+END|
+delimiter ;|
+START TRANSACTION;
+SELECT f1();
+
+--echo # Connection con1
+connection con1;
+--echo # Sending:
+--send DROP DATABASE db1
+
+--echo # Connection default
+connection default;
+--echo # Waiting for DROP DATABASE to be blocked by the lock on p1()
+let $wait_condition= SELECT COUNT(*)= 1 FROM information_schema.processlist
+  WHERE state= 'Waiting for stored procedure metadata lock'
+  AND info='DROP DATABASE db1';
+--source include/wait_condition.inc
+COMMIT;
+
+--echo # Connection con1
+connection con1;
+--echo # Reaping: DROP DATABASE db1
+--reap
+
+--echo # Test 3: Check that DROP DATABASE is not selected as a victim if a
+--echo #         deadlock is discovered with DML statements.
+
+--echo # Connection default
+connection default;
+CREATE DATABASE db1;
+CREATE TABLE db1.t1 (a INT);
+CREATE FUNCTION db1.f1() RETURNS INTEGER RETURN 1;
+START TRANSACTION;
+# DROP DATABASE will lock tables (t1) before functions (f1)
+SELECT db1.f1();
+
+--echo # Connection con1
+connection con1;
+--echo # Sending:
+--send DROP DATABASE db1
+
+--echo # Connection default
+connection default;
+--echo # Waiting for DROP DATABASE to be blocked by the lock on f1()
+let $wait_condition= SELECT COUNT(*)= 1 FROM information_schema.processlist
+  WHERE state= 'Waiting for stored function metadata lock'
+  AND info='DROP DATABASE db1';
+--source include/wait_condition.inc
+--error ER_LOCK_DEADLOCK
+SELECT * FROM db1.t1;
+COMMIT;
+
+--echo # Connection con1
+connection con1;
+--echo # Reaping: DROP DATABASE db1
+--reap
+
+--echo # Test 4: Check that active DROP DATABASE blocks stored routine DDL.
+
+--echo # Connection default
+connection default;
+CREATE DATABASE db1;
+CREATE FUNCTION db1.f1() RETURNS INTEGER RETURN 1;
+CREATE FUNCTION db1.f2() RETURNS INTEGER RETURN 2;
+START TRANSACTION;
+SELECT db1.f2();
+
+--echo # Connection con1
+connection con1;
+--echo # Sending:
+--send DROP DATABASE db1
+
+--echo # Connection con2
+connection con2;
+--echo # Waiting for DROP DATABASE to be blocked by the lock on f2()
+let $wait_condition= SELECT COUNT(*)= 1 FROM information_schema.processlist
+  WHERE state= 'Waiting for stored function metadata lock'
+  AND info='DROP DATABASE db1';
+--source include/wait_condition.inc
+--echo # Sending:
+--send ALTER FUNCTION db1.f1 COMMENT "test"
+
+--echo # Connection default
+connection default;
+--echo # Waiting for ALTER FUNCTION to be blocked by the schema lock on db1
+let $wait_condition= SELECT COUNT(*)= 1 FROM information_schema.processlist
+  WHERE state= 'Waiting for schema metadata lock'
+  AND info='ALTER FUNCTION db1.f1 COMMENT "test"';
+--source include/wait_condition.inc
+COMMIT;
+
+--echo # Connection con1
+connection con1;
+--echo # Reaping: DROP DATABASE db1
+--reap
+disconnect con1;
+--source include/wait_until_disconnected.inc
+
+--echo # Connection con2
+connection con2;
+--echo # Reaping: ALTER FUNCTION f1 COMMENT 'test'
+--error ER_SP_DOES_NOT_EXIST
+--reap
+disconnect con2;
+--source include/wait_until_disconnected.inc
+
+--echo # Connection default
+connection default;
+DROP FUNCTION f1;
+
+
 --echo #
 --echo # End of 5.5 tests
 --echo #

sql/sp.cc

@@ -1360,6 +1360,106 @@ sp_update_routine(THD *thd, int type, sp_name *name, st_sp_chistics *chistics)
 }
 
 
+/**
+  This internal handler is used to trap errors from opening mysql.proc.
+*/
+
+class Lock_db_routines_error_handler : public Internal_error_handler
+{
+public:
+  bool handle_condition(THD *thd,
+                        uint sql_errno,
+                        const char* sqlstate,
+                        MYSQL_ERROR::enum_warning_level level,
+                        const char* msg,
+                        MYSQL_ERROR ** cond_hdl)
+  {
+    if (sql_errno == ER_NO_SUCH_TABLE ||
+        sql_errno == ER_COL_COUNT_DOESNT_MATCH_CORRUPTED)
+      return true;
+    return false;
+  }
+};
+
+
+/**
+   Acquires exclusive metadata lock on all stored routines in the
+   given database.
+
+   @note Will also return false (=success) if mysql.proc can't be opened
+         or is outdated. This allows DROP DATABASE to continue in these
+         cases.
+ */
+
+bool lock_db_routines(THD *thd, char *db)
+{
+  TABLE *table;
+  uint key_len;
+  int nxtres= 0;
+  Open_tables_backup open_tables_state_backup;
+  MDL_request_list mdl_requests;
+  Lock_db_routines_error_handler err_handler;
+  DBUG_ENTER("lock_db_routines");
+
+  /*
+    mysql.proc will be re-opened during deletion, so we can ignore
+    errors when opening the table here. The error handler is
+    used to avoid getting the same warning twice.
+  */
+  thd->push_internal_handler(&err_handler);
+  table= open_proc_table_for_read(thd, &open_tables_state_backup);
+  thd->pop_internal_handler();
+  if (!table)
+  {
+    /*
+      DROP DATABASE should not fail even if mysql.proc does not exist
+      or is outdated. We therefore only abort mysql_rm_db() if we
+      have errors not handled by the error handler.
+    */
+    DBUG_RETURN(thd->is_error() || thd->killed);
+  }
+
+  table->field[MYSQL_PROC_FIELD_DB]->store(db, strlen(db), system_charset_info);
+  key_len= table->key_info->key_part[0].store_length;
+  table->file->ha_index_init(0, 1);
+
+  if (! table->file->index_read_map(table->record[0],
+                                    table->field[MYSQL_PROC_FIELD_DB]->ptr,
+                                    (key_part_map)1, HA_READ_KEY_EXACT))
+  {
+    do
+    {
+      char *sp_name= get_field(thd->mem_root,
+                               table->field[MYSQL_PROC_FIELD_NAME]);
+      longlong sp_type= table->field[MYSQL_PROC_MYSQL_TYPE]->val_int();
+      MDL_request *mdl_request= new (thd->mem_root) MDL_request;
+      mdl_request->init(sp_type == TYPE_ENUM_FUNCTION ?
+                        MDL_key::FUNCTION : MDL_key::PROCEDURE,
+                        db, sp_name, MDL_EXCLUSIVE, MDL_TRANSACTION);
+      mdl_requests.push_front(mdl_request);
+    } while (! (nxtres= table->file->index_next_same(table->record[0],
+                                         table->field[MYSQL_PROC_FIELD_DB]->ptr,
+						     key_len)));
+  }
+  table->file->ha_index_end();
+  if (nxtres != 0 && nxtres != HA_ERR_END_OF_FILE)
+  {
+    table->file->print_error(nxtres, MYF(0));
+    close_system_tables(thd, &open_tables_state_backup);
+    DBUG_RETURN(true);
+  }
+  close_system_tables(thd, &open_tables_state_backup);
+
+  /* We should already hold a global IX lock and a schema X lock. */
+  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
+                                             MDL_INTENTION_EXCLUSIVE) &&
+              thd->mdl_context.is_lock_owner(MDL_key::SCHEMA, db, "",
+                                             MDL_EXCLUSIVE));
+  DBUG_RETURN(thd->mdl_context.acquire_locks(&mdl_requests,
+                                             thd->variables.lock_wait_timeout));
+}
+
+
 /**
   Drop all routines in database 'db'
 

sql/sp.h

@@ -84,6 +84,18 @@ enum
 int
 sp_drop_db_routines(THD *thd, char *db);
 
+/**
+   Acquires exclusive metadata lock on all stored routines in the
+   given database.
+
+   @param  thd  Thread handler
+   @param  db   Database name
+
+   @retval  false  Success
+   @retval  true   Failure
+ */
+bool lock_db_routines(THD *thd, char *db);
+
 sp_head *
 sp_find_routine(THD *thd, int type, sp_name *name,
                 sp_cache **cp, bool cache_only);

sql/sql_db.cc

@@ -46,10 +46,12 @@ const char *del_exts[]= {".frm", ".BAK", ".TMD",".opt", NullS};
 static TYPELIB deletable_extentions=
 {array_elements(del_exts)-1,"del_exts", del_exts, NULL};
 
-static long mysql_rm_known_files(THD *thd, MY_DIR *dirp,
-				 const char *db, const char *path,
-                                 TABLE_LIST **dropped_tables);
-         
+static bool find_db_tables_and_rm_known_files(THD *thd, MY_DIR *dirp,
+                                              const char *db,
+                                              const char *path,
+                                              TABLE_LIST **tables,
+                                              bool *found_other_files);
+
 long mysql_rm_arc_files(THD *thd, MY_DIR *dirp, const char *org_path);
 static my_bool rm_dir_w_symlink(const char *org_path, my_bool send_error);
 static void mysql_change_db_impl(THD *thd,
@@ -738,36 +740,37 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info)
 }
 
 
-/*
-  Drop all tables in a database and the database itself
-
-  SYNOPSIS
-    mysql_rm_db()
-    thd			Thread handle
-    db			Database name in the case given by user
-		        It's already validated and set to lower case
-                        (if needed) when we come here
-    if_exists		Don't give error if database doesn't exists
-    silent		Don't generate errors
-
-  RETURN
-    FALSE ok (Database dropped)
-    ERROR Error
+/**
+  Drop all tables, routines and events in a database and the database itself.
+
+  @param  thd        Thread handle
+  @param  db         Database name in the case given by user
+                     It's already validated and set to lower case
+                     (if needed) when we come here
+  @param  if_exists  Don't give error if database doesn't exists
+  @param  silent     Don't write the statement to the binary log and don't
+                     send ok packet to the client
+
+  @retval  false  OK (Database dropped)
+  @retval  true   Error
 */
 
 bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
 {
-  long deleted=0;
-  int error= 0;
+  ulong deleted_tables= 0;
+  bool error= true;
   char	path[FN_REFLEN+16];
   MY_DIR *dirp;
   uint length;
-  TABLE_LIST* dropped_tables= 0;
+  bool found_other_files= false;
+  TABLE_LIST *tables= NULL;
+  TABLE_LIST *table;
+  Drop_table_error_handler err_handler;
   DBUG_ENTER("mysql_rm_db");
 
 
   if (lock_schema_name(thd, db))
-    DBUG_RETURN(TRUE);
+    DBUG_RETURN(true);
 
   length= build_table_filename(path, sizeof(path) - 1, db, "", "", 0);
   strmov(path+length, MY_DB_OPT_FILE);		// Append db option file name
@@ -779,20 +782,72 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   {
     if (!if_exists)
     {
-      error= -1;
       my_error(ER_DB_DROP_EXISTS, MYF(0), db);
-      goto exit;
+      DBUG_RETURN(true);
     }
     else
+    {
       push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
 			  ER_DB_DROP_EXISTS, ER(ER_DB_DROP_EXISTS), db);
+      error= false;
+      goto update_binlog;
+    }
   }
-  else
+
+  thd->push_internal_handler(&err_handler);
+
+  if (find_db_tables_and_rm_known_files(thd, dirp, db, path, &tables,
+                                        &found_other_files))
+  {
+    thd->pop_internal_handler();
+    goto exit;
+  }
+
+  /*
+    Disable drop of enabled log tables, must be done before name locking.
+    This check is only needed if we are dropping the "mysql" database.
+  */
+  if ((my_strcasecmp(system_charset_info, MYSQL_SCHEMA_NAME.str, db) == 0))
+  {
+    for (table= tables; table; table= table->next_local)
+    {
+      if (check_if_log_table(table->db_length, table->db,
+                             table->table_name_length, table->table_name, true))
+      {
+        my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+        thd->pop_internal_handler();
+        goto exit;
+      }
+    }
+  }
+
+  /* Lock all tables and stored routines about to be dropped. */
+  if (lock_table_names(thd, tables, NULL, thd->variables.lock_wait_timeout,
+                       MYSQL_OPEN_SKIP_TEMPORARY) ||
+      lock_db_routines(thd, db))
   {
-    Drop_table_error_handler err_handler;
-    thd->push_internal_handler(&err_handler);
+    thd->pop_internal_handler();
+    goto exit;
+  }
 
-    error= -1;
+  /* mysql_ha_rm_tables() requires a non-null TABLE_LIST. */
+  if (tables)
+    mysql_ha_rm_tables(thd, tables);
+
+  for (table= tables; table; table= table->next_local)
+  {
+    tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name,
+                     false);
+    deleted_tables++;
+  }
+
+  if (thd->killed ||
+      (tables && mysql_rm_table_no_locks(thd, tables, true, false, true, true)))
+  {
+    tables= NULL;
+  }
+  else
+  {
     /*
       We temporarily disable the binary log while dropping the objects
       in the database. Since the DROP DATABASE statement is always
@@ -810,23 +865,30 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       ha_drop_database(), since NDB otherwise detects the binary log
       as disabled and will not log the drop database statement on any
       other connected server.
-     */
-    if ((deleted= mysql_rm_known_files(thd, dirp, db, path,
-                                       &dropped_tables)) >= 0)
-    {
-      ha_drop_database(path);
-      tmp_disable_binlog(thd);
-      query_cache_invalidate1(db);
-      (void) sp_drop_db_routines(thd, db); /* @todo Do not ignore errors */
+    */
+
+    ha_drop_database(path);
+    tmp_disable_binlog(thd);
+    query_cache_invalidate1(db);
+    (void) sp_drop_db_routines(thd, db); /* @todo Do not ignore errors */
 #ifdef HAVE_EVENT_SCHEDULER
-      Events::drop_schema_events(thd, db);
+    Events::drop_schema_events(thd, db);
 #endif
-      error = 0;
-      reenable_binlog(thd);
-    }
-    thd->pop_internal_handler();
+    reenable_binlog(thd);
+
+    /*
+      If the directory is a symbolic link, remove the link first, then
+      remove the directory the symbolic link pointed at
+    */
+    if (found_other_files)
+      my_error(ER_DB_DROP_RMDIR, MYF(0), path, EEXIST);
+    else
+      error= rm_dir_w_symlink(path, true);
   }
-  if (!silent && deleted>=0)
+  thd->pop_internal_handler();
+
+update_binlog:
+  if (!silent && !error)
   {
     const char *query;
     ulong query_length;
@@ -861,13 +923,13 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       */
       if (mysql_bin_log.write(&qinfo))
       {
-        error= -1;
+        error= true;
         goto exit;
       }
     }
     thd->clear_error();
     thd->server_status|= SERVER_STATUS_DB_DROPPED;
-    my_ok(thd, (ulong) deleted);
+    my_ok(thd, deleted_tables);
   }
   else if (mysql_bin_log.is_open())
   {
@@ -881,7 +943,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
     query_end= query + MAX_DROP_TABLE_Q_LEN;
     db_len= strlen(db);
 
-    for (tbl= dropped_tables; tbl; tbl= tbl->next_local)
+    for (tbl= tables; tbl; tbl= tbl->next_local)
     {
       uint tbl_name_len;
 
@@ -895,7 +957,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
         */
         if (write_to_binlog(thd, query, query_pos -1 - query, db, db_len))
         {
-          error= -1;
+          error= true;
           goto exit;
         }
         query_pos= query_data_start;
@@ -915,7 +977,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       */
       if (write_to_binlog(thd, query, query_pos -1 - query, db, db_len))
       {
-        error= -1;
+        error= true;
         goto exit;
       }
     }
@@ -928,27 +990,23 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
     SELECT DATABASE() in the future). For this we free() thd->db and set
     it to 0.
   */
-  if (thd->db && !strcmp(thd->db, db) && error == 0)
+  if (thd->db && !strcmp(thd->db, db) && !error)
     mysql_change_db_impl(thd, NULL, 0, thd->variables.collation_server);
+  my_dirend(dirp);
   DBUG_RETURN(error);
 }
 
-/*
-  Removes files with known extensions.
-  thd MUST be set when calling this function!
-*/
 
-static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
-				 const char *org_path,
-                                 TABLE_LIST **dropped_tables)
+static bool find_db_tables_and_rm_known_files(THD *thd, MY_DIR *dirp,
+                                              const char *db,
+                                              const char *path,
+                                              TABLE_LIST **tables,
+                                              bool *found_other_files)
 {
-  long deleted=0;
-  ulong found_other_files=0;
   char filePath[FN_REFLEN];
   TABLE_LIST *tot_list=0, **tot_list_next_local, **tot_list_next_global;
-  TABLE_LIST *table;
-  DBUG_ENTER("mysql_rm_known_files");
-  DBUG_PRINT("enter",("path: %s", org_path));
+  DBUG_ENTER("find_db_tables_and_rm_known_files");
+  DBUG_PRINT("enter",("path: %s", path));
 
   tot_list_next_local= tot_list_next_global= &tot_list;
 
@@ -974,16 +1032,16 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
       */
       char newpath[FN_REFLEN];
       MY_DIR *new_dirp;
-      strxmov(newpath, org_path, "/", "arc", NullS);
+      strxmov(newpath, path, "/", "arc", NullS);
       (void) unpack_filename(newpath, newpath);
       if ((new_dirp = my_dir(newpath, MYF(MY_DONT_SORT))))
       {
 	DBUG_PRINT("my",("Archive subdir found: %s", newpath));
 	if ((mysql_rm_arc_files(thd, new_dirp, newpath)) < 0)
-	  goto err;
+	  DBUG_RETURN(true);
 	continue;
       }
-      found_other_files++;
+      *found_other_files= true;
       continue;
     }
     if (!(extension= strrchr(file->name, '.')))
@@ -991,7 +1049,7 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
     if (find_type(extension, &deletable_extentions,1+2) <= 0)
     {
       if (find_type(extension, ha_known_exts(),1+2) <= 0)
-	found_other_files++;
+	*found_other_files= true;
       continue;
     }
     /* just for safety we use files_charset_info */
@@ -1007,7 +1065,7 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
                                           strlen(file->name) + 1);
 
       if (!table_list)
-        goto err;
+        DBUG_RETURN(true);
       table_list->db= (char*) (table_list+1);
       table_list->db_length= strmov(table_list->db, db) - table_list->db;
       table_list->table_name= table_list->db + table_list->db_length + 1;
@@ -1032,77 +1090,16 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
       (*tot_list_next_global)= table_list;
       tot_list_next_local= &table_list->next_local;
       tot_list_next_global= &table_list->next_global;
-      deleted++;
     }
     else
     {
-      strxmov(filePath, org_path, "/", file->name, NullS);
+      strxmov(filePath, path, "/", file->name, NullS);
       if (mysql_file_delete_with_symlink(key_file_misc, filePath, MYF(MY_WME)))
-      {
-	goto err;
-      }
-    }
-  }
-
-  /*
-    Disable drop of enabled log tables, must be done before name locking.
-    This check is only needed if we are dropping the "mysql" database.
-  */
-  if ((my_strcasecmp(system_charset_info, MYSQL_SCHEMA_NAME.str, db) == 0))
-  {
-    for (table= tot_list; table; table= table->next_local)
-    {
-      if (check_if_log_table(table->db_length, table->db,
-                             table->table_name_length, table->table_name, true))
-      {
-        my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
-        goto err;
-      }
+	DBUG_RETURN(true);
     }
   }
-
-  /* mysql_ha_rm_tables() requires a non-null TABLE_LIST. */
-  if (tot_list)
-    mysql_ha_rm_tables(thd, tot_list);
-
-  if (lock_table_names(thd, tot_list, NULL, thd->variables.lock_wait_timeout,
-                       MYSQL_OPEN_SKIP_TEMPORARY))
-    goto err;
-
-  for (table= tot_list; table; table= table->next_local)
-    tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name,
-                     false);
-
-  if (thd->killed ||
-      (tot_list && mysql_rm_table_no_locks(thd, tot_list, true,
-                                           false, true, true)))
-    goto err;
-
-  my_dirend(dirp);  
-  
-  if (dropped_tables)
-    *dropped_tables= tot_list;
-  
-  /*
-    If the directory is a symbolic link, remove the link first, then
-    remove the directory the symbolic link pointed at
-  */
-  if (found_other_files)
-  {
-    my_error(ER_DB_DROP_RMDIR, MYF(0), org_path, EEXIST);
-    DBUG_RETURN(-1);
-  }
-  else
-  {
-    if (rm_dir_w_symlink(org_path, true))
-      DBUG_RETURN(-1);
-  }
-
-  DBUG_RETURN(deleted);
-
-err:
-  my_dirend(dirp);
-  DBUG_RETURN(-1);
+  *tables= tot_list;
+  DBUG_RETURN(false);
 }