Bug #57663 Concurrent statement using stored function and DROP DATABASE

           breaks SBR

This pre-requisite patch refactors the code for dropping tables, used
by DROP TABLE and DROP DATABASE. The patch moves the code for acquiring
metadata locks out of mysql_rm_table_part2() and makes it the
responsibility of the caller. This in preparation of changing the
DROP DATABASE implementation to acquire all metadata locks before any
changes are made. mysql_rm_table_part2() is renamed
mysql_rm_table_no_locks() to reflect the change.

mysql-test/r/partition_debug_sync.result

@@ -25,7 +25,7 @@ ALTER TABLE t1 REMOVE PARTITIONING;
 # Con default
 SET DEBUG_SYNC= 'now WAIT_FOR removing_partitioning';
 SET DEBUG_SYNC= 'mdl_acquire_lock_wait SIGNAL waiting_for_alter';
-SET DEBUG_SYNC= 'rm_table_part2_before_delete_table WAIT_FOR partitioning_removed';
+SET DEBUG_SYNC= 'rm_table_no_locks_before_delete_table WAIT_FOR partitioning_removed';
 DROP TABLE IF EXISTS t1;
 # Con 1
 SET SESSION debug= "-d,sleep_before_create_table_no_lock";
@@ -51,12 +51,12 @@ SET DEBUG_SYNC= 'alter_table_before_open_tables SIGNAL removing_partitions WAIT_
 SET DEBUG_SYNC= 'alter_table_before_rename_result_table WAIT_FOR delete_done';
 ALTER TABLE t2 REMOVE PARTITIONING;
 # Con default
-SET SESSION debug= "+d,sleep_before_part2_delete_table";
+SET SESSION debug= "+d,sleep_before_no_locks_delete_table";
 SET DEBUG_SYNC= 'now WAIT_FOR removing_partitions';
-SET DEBUG_SYNC= 'rm_table_part2_before_delete_table SIGNAL waiting_for_alter';
-SET DEBUG_SYNC= 'rm_table_part2_before_binlog SIGNAL delete_done';
+SET DEBUG_SYNC= 'rm_table_no_locks_before_delete_table SIGNAL waiting_for_alter';
+SET DEBUG_SYNC= 'rm_table_no_locks_before_binlog SIGNAL delete_done';
 DROP TABLE IF EXISTS t2;
-SET SESSION debug= "-d,sleep_before_part2_delete_table";
+SET SESSION debug= "-d,sleep_before_no_locks_delete_table";
 # Con 1
 ERROR 42S02: Table 'test.t2' doesn't exist
 SET DEBUG_SYNC= 'RESET';

mysql-test/t/partition_debug_sync.test

@@ -38,7 +38,7 @@ connection default;
 --echo # Con default
 SET DEBUG_SYNC= 'now WAIT_FOR removing_partitioning';
 SET DEBUG_SYNC= 'mdl_acquire_lock_wait SIGNAL waiting_for_alter';
-SET DEBUG_SYNC= 'rm_table_part2_before_delete_table WAIT_FOR partitioning_removed';
+SET DEBUG_SYNC= 'rm_table_no_locks_before_delete_table WAIT_FOR partitioning_removed';
 DROP TABLE IF EXISTS t1;
 --echo # Con 1
 connection con1;
@@ -70,12 +70,12 @@ SET DEBUG_SYNC= 'alter_table_before_rename_result_table WAIT_FOR delete_done';
 --send ALTER TABLE t2 REMOVE PARTITIONING
 connection default;
 --echo # Con default
-SET SESSION debug= "+d,sleep_before_part2_delete_table";
+SET SESSION debug= "+d,sleep_before_no_locks_delete_table";
 SET DEBUG_SYNC= 'now WAIT_FOR removing_partitions';
-SET DEBUG_SYNC= 'rm_table_part2_before_delete_table SIGNAL waiting_for_alter';
-SET DEBUG_SYNC= 'rm_table_part2_before_binlog SIGNAL delete_done';
+SET DEBUG_SYNC= 'rm_table_no_locks_before_delete_table SIGNAL waiting_for_alter';
+SET DEBUG_SYNC= 'rm_table_no_locks_before_binlog SIGNAL delete_done';
 DROP TABLE IF EXISTS t2;
-SET SESSION debug= "-d,sleep_before_part2_delete_table";
+SET SESSION debug= "-d,sleep_before_no_locks_delete_table";
 --echo # Con 1
 connection con1;
 --error ER_NO_SUCH_TABLE

sql/sql_db.cc

@@ -28,6 +28,8 @@
 #include "sql_acl.h"                     // SELECT_ACL, DB_ACLS,
                                          // acl_get, check_grant_db
 #include "log_event.h"                   // Query_log_event
+#include "sql_base.h"                    // lock_table_names, tdc_remove_table
+#include "sql_handler.h"                 // mysql_ha_rm_tables
 #include <mysys_err.h>
 #include "sp.h"
 #include "events.h"
@@ -944,6 +946,7 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
   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));
 
@@ -1040,8 +1043,39 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
       }
     }
   }
+
+  /*
+    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;
+      }
+    }
+  }
+
+  /* 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_part2(thd, tot_list, 1, 0, 1, 1)))
+      (tot_list && mysql_rm_table_no_locks(thd, tot_list, true,
+                                           false, true, true)))
     goto err;
 
   my_dirend(dirp);  

sql/sql_table.h

@@ -174,8 +174,9 @@ bool mysql_checksum_table(THD* thd, TABLE_LIST* table_list,
                           HA_CHECK_OPT* check_opt);
 bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
                     my_bool drop_temporary);
-int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
-                         bool drop_temporary, bool drop_view, bool log_query);
+int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
+                            bool drop_temporary, bool drop_view,
+                            bool log_query);
 bool quick_rm_table(handlerton *base,const char *db,
                     const char *table_name, uint flags);
 void close_cached_table(THD *thd, TABLE *table);