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