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.

1 files changed, 128 insertions, 131 deletions

diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 1391b2c4d72..292fbba2352 100644
--- a/sql/sql_db.cc
+++ b/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 @@ exit:
 }
 
 
-/*
-  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 @@ exit:
     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);
 }