MDEV-717 LP:1003679 - Wrong binlog order on concurrent DROP schema and CREATE function.

        The cause of the issue is when DROP DATABASE takes
        metadata lock and is in progress through it's
        execution, a concurrently running CREATE FUNCTION checks
        for the existence of database which it succeeds and then it
        waits on the metadata lock.  Once DROP DATABASE writes to
        BINLOG and finally releases the metadata lock on schema
        object, the CREATE FUNCTION waiting on metadata lock
        gets in it's code path and succeeds and writes to binlog.

6 files changed, 219 insertions, 82 deletions

diff --git a/mysql-test/suite/binlog/r/binlog_concurrent_drop_create.result b/mysql-test/suite/binlog/r/binlog_concurrent_drop_create.result
new file mode 100644
index 00000000000..8db1dc4b75e
--- /dev/null
+++ b/mysql-test/suite/binlog/r/binlog_concurrent_drop_create.result
@@ -0,0 +1,52 @@
+include/master-slave.inc
+[connection master]
+RESET MASTER;
+DROP DATABASE IF EXISTS db_717;
+DROP EVENT IF EXISTS test.e_x1;
+set @saved_global_binlog_format = @@global.binlog_format;
+set @saved_local_binlog_format = @@session.binlog_format;
+SET GLOBAL BINLOG_FORMAT = STATEMENT;
+SET SESSION BINLOG_FORMAT = STATEMENT;
+CREATE DATABASE db_717;
+CREATE FUNCTION db_717.f1() RETURNS INT RETURN 1;
+DROP DATABASE db_717;
+CREATE FUNCTION db_717.f2() RETURNS INT RETURN 1;
+CREATE DATABASE db_717;
+CREATE EVENT db_717.e_x1 ON SCHEDULE EVERY 1000 HOUR DO CREATE TABLE t1(id int);
+DROP DATABASE db_717;
+CREATE EVENT db_717.e_x2 ON SCHEDULE EVERY 1000 HOUR DO CREATE TABLE t1(id int);
+CREATE DATABASE db_717;
+CREATE EVENT test.e_x1 ON SCHEDULE EVERY 1000 HOUR DO CREATE TABLE t1(id int);
+DROP DATABASE db_717;
+ALTER EVENT test.e_x1 RENAME TO db_717.e_x2;
+DROP EVENT test.e_x1;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP DATABASE IF EXISTS db_717
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP EVENT IF EXISTS test.e_x1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	CREATE DATABASE db_717
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` FUNCTION `db_717`.`f1`() RETURNS int(11)
+RETURN 1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP DATABASE db_717
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	CREATE DATABASE db_717
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` EVENT db_717.e_x1 ON SCHEDULE EVERY 1000 HOUR DO CREATE TABLE t1(id int)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP DATABASE db_717
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	CREATE DATABASE db_717
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` EVENT test.e_x1 ON SCHEDULE EVERY 1000 HOUR DO CREATE TABLE t1(id int)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	DROP DATABASE db_717
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP EVENT test.e_x1
+SET GLOBAL BINLOG_FORMAT = @saved_global_binlog_format;
+SET SESSION BINLOG_FORMAT = @saved_local_binlog_format;
+include/rpl_end.inc
diff --git a/mysql-test/suite/binlog/t/binlog_concurrent_drop_create.test b/mysql-test/suite/binlog/t/binlog_concurrent_drop_create.test
new file mode 100644
index 00000000000..c885a7291f7
--- /dev/null
+++ b/mysql-test/suite/binlog/t/binlog_concurrent_drop_create.test
@@ -0,0 +1,85 @@
+# MDEV-717
+#   DROP DATABASE and CREATE PROCEDURE|FUNCTION|EVENT
+#   statements can appear in wrong order in the binlog.
+#   
+#   Note - the test can be undeterministic.
+
+--source include/master-slave.inc
+RESET MASTER;
+
+--disable_warnings
+DROP DATABASE IF EXISTS db_717;
+DROP EVENT IF EXISTS test.e_x1;
+--enable_warnings
+
+set @saved_global_binlog_format = @@global.binlog_format;
+set @saved_local_binlog_format = @@session.binlog_format;
+SET GLOBAL BINLOG_FORMAT = STATEMENT;
+SET SESSION BINLOG_FORMAT = STATEMENT;
+
+# test function creation
+CREATE DATABASE db_717;
+ 
+CREATE FUNCTION db_717.f1() RETURNS INT RETURN 1;
+ 
+--send
+ 
+DROP DATABASE db_717;
+ 
+--connection master1
+ 
+--error 0,ER_BAD_DB_ERROR
+ 
+CREATE FUNCTION db_717.f2() RETURNS INT RETURN 1;
+ 
+--connection master
+ 
+--reap
+ 
+# test event creation
+CREATE DATABASE db_717;
+ 
+CREATE EVENT db_717.e_x1 ON SCHEDULE EVERY 1000 HOUR DO CREATE TABLE t1(id int);
+ 
+--send
+ 
+DROP DATABASE db_717;
+ 
+--connection master1
+ 
+--error 0,ER_BAD_DB_ERROR
+ 
+CREATE EVENT db_717.e_x2 ON SCHEDULE EVERY 1000 HOUR DO CREATE TABLE t1(id int);
+ 
+--connection master
+ 
+--reap
+ 
+# test event modification
+CREATE DATABASE db_717;
+ 
+CREATE EVENT test.e_x1 ON SCHEDULE EVERY 1000 HOUR DO CREATE TABLE t1(id int);
+ 
+--send
+ 
+DROP DATABASE db_717;
+ 
+--connection master1
+ 
+--error 0,ER_BAD_DB_ERROR
+ 
+ALTER EVENT test.e_x1 RENAME TO db_717.e_x2;
+ 
+--connection master
+ 
+--reap
+
+DROP EVENT test.e_x1;
+
+source include/show_binlog_events.inc;
+
+SET GLOBAL BINLOG_FORMAT = @saved_global_binlog_format;
+SET SESSION BINLOG_FORMAT = @saved_local_binlog_format;
+
+--source include/rpl_end.inc
+
diff --git a/sql/events.cc b/sql/events.cc
index b80ec993ac4..77dbb8b82e5 100644
--- a/sql/events.cc
+++ b/sql/events.cc
@@ -333,6 +333,10 @@ Events::create_event(THD *thd, Event_parse_data *parse_data)
   if (check_access(thd, EVENT_ACL, parse_data->dbname.str, NULL, NULL, 0, 0))
     DBUG_RETURN(TRUE);
 
+  if (lock_object_name(thd, MDL_key::EVENT,
+                       parse_data->dbname.str, parse_data->name.str))
+    DBUG_RETURN(TRUE);
+
   if (check_db_dir_existence(parse_data->dbname.str))
   {
     my_error(ER_BAD_DB_ERROR, MYF(0), parse_data->dbname.str);
@@ -347,10 +351,6 @@ Events::create_event(THD *thd, Event_parse_data *parse_data)
   */
   save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
 
-  if (lock_object_name(thd, MDL_key::EVENT,
-                       parse_data->dbname.str, parse_data->name.str))
-    DBUG_RETURN(TRUE);
-
   if (thd->lex->create_info.or_replace() && event_queue)
     event_queue->drop_event(thd, parse_data->dbname, parse_data->name);
 
@@ -454,6 +454,16 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
 
   if (check_access(thd, EVENT_ACL, parse_data->dbname.str, NULL, NULL, 0, 0))
     DBUG_RETURN(TRUE);
+  if (lock_object_name(thd, MDL_key::EVENT,
+                       parse_data->dbname.str, parse_data->name.str))
+    DBUG_RETURN(TRUE);
+
+  if (check_db_dir_existence(parse_data->dbname.str))
+  {
+    my_error(ER_BAD_DB_ERROR, MYF(0), parse_data->dbname.str);
+    DBUG_RETURN(TRUE);
+  }
+
 
   if (new_dbname)                               /* It's a rename */
   {
@@ -476,6 +486,13 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
     if (check_access(thd, EVENT_ACL, new_dbname->str, NULL, NULL, 0, 0))
       DBUG_RETURN(TRUE);
 
+    /*
+     Acquire mdl exclusive lock on target database name.
+    */
+    if (lock_object_name(thd, MDL_key::EVENT,
+                         new_dbname->str, new_name->str))
+      DBUG_RETURN(TRUE);
+
     /* Check that the target database exists */
     if (check_db_dir_existence(new_dbname->str))
     {
@@ -490,10 +507,6 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
   */
   save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
 
-  if (lock_object_name(thd, MDL_key::EVENT,
-                       parse_data->dbname.str, parse_data->name.str))
-    DBUG_RETURN(TRUE);
-
   /* On error conditions my_error() is called so no need to handle here */
   if (!(ret= db_repository->update_event(thd, parse_data,
                                          new_dbname, new_name)))
diff --git a/sql/sp.cc b/sql/sp.cc
index 6ec59143720..3cc24089c40 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -34,6 +34,10 @@
 
 #include <my_user.h>
 
+/* Used in error handling only */
+#define SP_TYPE_STRING(type) \
+    (type == TYPE_ENUM_FUNCTION ? "FUNCTION" : "PROCEDURE")
+     
 static int
 db_load_routine(THD *thd, stored_procedure_type type, sp_name *name,
                 sp_head **sphp,
@@ -1007,15 +1011,16 @@ sp_drop_routine_internal(THD *thd, stored_procedure_type type,
   followed by an implicit grant (sp_grant_privileges())
   and this subsequent call opens and closes mysql.procs_priv.
 
-  @return Error code. SP_OK is returned on success. Other
-  SP_ constants are used to indicate about errors.
+  @return Error status.
+    @retval FALSE on success
+    @retval TRUE on error
 */
 
-int
+bool
 sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp)
 {
   LEX *lex= thd->lex;
-  int ret;
+  bool ret= TRUE;
   TABLE *table;
   char definer_buf[USER_HOST_BUFF_SIZE];
   LEX_STRING definer;
@@ -1040,7 +1045,22 @@ sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp)
 
   /* Grab an exclusive MDL lock. */
   if (lock_object_name(thd, mdl_type, sp->m_db.str, sp->m_name.str))
-    DBUG_RETURN(SP_OPEN_TABLE_FAILED);
+  {
+    my_error(ER_BAD_DB_ERROR, MYF(0), sp->m_db.str);
+    DBUG_RETURN(TRUE);
+  }
+
+  /*
+    Check that a database directory with this name
+    exists. Design note: This won't work on virtual databases
+    like information_schema.
+  */
+  if (check_db_dir_existence(sp->m_db.str))
+  {
+    my_error(ER_BAD_DB_ERROR, MYF(0), sp->m_db.str);
+    DBUG_RETURN(TRUE);
+  }
+
 
   /* Reset sql_mode during data dictionary operations. */
   thd->variables.sql_mode= 0;
@@ -1049,7 +1069,9 @@ sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp)
   thd->count_cuted_fields= CHECK_FIELD_WARN;
 
   if (!(table= open_proc_table_for_update(thd)))
-    ret= SP_OPEN_TABLE_FAILED;
+  {
+    my_error(ER_SP_STORE_FAILED, MYF(0), SP_TYPE_STRING(type),sp->m_name.str);
+  }
   else
   {
     /* Checking if the routine already exists */
@@ -1065,11 +1087,10 @@ sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp)
         push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
                             ER_SP_ALREADY_EXISTS,
                             ER_THD(thd, ER_SP_ALREADY_EXISTS),
-                            type == TYPE_ENUM_FUNCTION ?
-                            "FUNCTION" : "PROCEDURE",
+                            SP_TYPE_STRING(type),
                             lex->spname->m_name.str);
 
-        ret= SP_OK;
+        ret= FALSE;
 
         // Setting retstr as it is used for logging.
         if (sp->m_type == TYPE_ENUM_FUNCTION)
@@ -1078,7 +1099,8 @@ sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp)
       }
       else
       {
-        ret= SP_WRITE_ROW_FAILED;
+        my_error(ER_SP_ALREADY_EXISTS, MYF(0),
+                 SP_TYPE_STRING(type), sp->m_name.str);
         goto done;
       }
     }
@@ -1090,7 +1112,8 @@ sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp)
 
     if (table->s->fields < MYSQL_PROC_FIELD_COUNT)
     {
-      ret= SP_GET_FIELD_FAILED;
+      my_error(ER_SP_STORE_FAILED, MYF(0),
+               SP_TYPE_STRING(type), sp->m_name.str);
       goto done;
     }
 
@@ -1099,12 +1122,12 @@ sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp)
                                             sp->m_name.str+sp->m_name.length) >
         table->field[MYSQL_PROC_FIELD_NAME]->char_length())
     {
-      ret= SP_BAD_IDENTIFIER;
+      my_error(ER_TOO_LONG_IDENT, MYF(0), sp->m_name.str);
       goto done;
     }
     if (sp->m_body.length > table->field[MYSQL_PROC_FIELD_BODY]->field_length)
     {
-      ret= SP_BODY_TOO_LONG;
+      my_error(ER_TOO_LONG_BODY, MYF(0), sp->m_name.str);
       goto done;
     }
 
@@ -1193,17 +1216,13 @@ sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp)
 	if (access == SP_CONTAINS_SQL ||
 	    access == SP_MODIFIES_SQL_DATA)
 	{
-	  my_message(ER_BINLOG_UNSAFE_ROUTINE,
-		     ER_THD(thd, ER_BINLOG_UNSAFE_ROUTINE), MYF(0));
-	  ret= SP_INTERNAL_ERROR;
+          my_error(ER_BINLOG_UNSAFE_ROUTINE, MYF(0));
 	  goto done;
 	}
       }
       if (!(thd->security_ctx->master_access & SUPER_ACL))
       {
-	my_message(ER_BINLOG_CREATE_ROUTINE_NEED_SUPER,
-		   ER_THD(thd, ER_BINLOG_CREATE_ROUTINE_NEED_SUPER), MYF(0));
-	ret= SP_INTERNAL_ERROR;
+        my_error(ER_BINLOG_CREATE_ROUTINE_NEED_SUPER,MYF(0));
 	goto done;
       }
     }
@@ -1234,22 +1253,24 @@ sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp)
 
     if (store_failed)
     {
-      ret= SP_FLD_STORE_FAILED;
+      my_error(ER_CANT_CREATE_SROUTINE, MYF(0), sp->m_name.str);
       goto done;
     }
 
-    ret= SP_OK;
     if (table->file->ha_write_row(table->record[0]))
-      ret= SP_WRITE_ROW_FAILED;
+    {
+      my_error(ER_SP_ALREADY_EXISTS, MYF(0),
+               SP_TYPE_STRING(type), sp->m_name.str);
+      goto done;
+    }
     /* Make change permanent and avoid 'table is marked as crashed' errors */
     table->file->extra(HA_EXTRA_FLUSH);
 
-    if (ret == SP_OK)
-      sp_cache_invalidate();
+    sp_cache_invalidate();
   }
 
 log:
-  if (ret == SP_OK && mysql_bin_log.is_open())
+  if (mysql_bin_log.is_open())
   {
     thd->clear_error();
 
@@ -1268,7 +1289,7 @@ log:
                        &(thd->lex->definer->host),
                        saved_mode))
     {
-      ret= SP_INTERNAL_ERROR;
+      my_error(ER_OUT_OF_RESOURCES, MYF(0));
       goto done;
     }
     /* restore sql_mode when binloging */
@@ -1277,9 +1298,13 @@ log:
     if (thd->binlog_query(THD::STMT_QUERY_TYPE,
                           log_query.ptr(), log_query.length(),
                           FALSE, FALSE, FALSE, 0))
-      ret= SP_INTERNAL_ERROR;
+    {
+      my_error(ER_ERROR_ON_WRITE, MYF(MY_WME), "binary log", -1);
+      goto done;
+    }
     thd->variables.sql_mode= 0;
   }
+  ret= FALSE;
 
 done:
   thd->count_cuted_fields= saved_count_cuted_fields;
diff --git a/sql/sp.h b/sql/sp.h
index 4bfb0577fcc..df60482f8fd 100644
--- a/sql/sp.h
+++ b/sql/sp.h
@@ -126,7 +126,7 @@ sp_exist_routines(THD *thd, TABLE_LIST *procs, bool is_proc);
 bool
 sp_show_create_routine(THD *thd, stored_procedure_type type, sp_name *name);
 
-int
+bool
 sp_create_routine(THD *thd, stored_procedure_type type, sp_head *sp);
 
 int
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 248a15458d7..6fcb549f097 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -118,8 +118,6 @@ static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
 */
 
 /* Used in error handling only */
-#define SP_TYPE_STRING(LP) \
-  ((LP)->sphead->m_type == TYPE_ENUM_FUNCTION ? "FUNCTION" : "PROCEDURE")
 #define SP_COM_STRING(LP) \
   ((LP)->sql_command == SQLCOM_CREATE_SPFUNCTION || \
    (LP)->sql_command == SQLCOM_ALTER_FUNCTION || \
@@ -5104,7 +5102,6 @@ end_with_restore_list:
   {
     uint namelen;
     char *name;
-    int sp_result= SP_INTERNAL_ERROR;
 
     DBUG_ASSERT(lex->sphead != 0);
     DBUG_ASSERT(lex->sphead->m_db.str); /* Must be initialized in the parser */
@@ -5115,23 +5112,12 @@ end_with_restore_list:
     if (check_db_name(&lex->sphead->m_db))
     {
       my_error(ER_WRONG_DB_NAME, MYF(0), lex->sphead->m_db.str);
-      goto create_sp_error;
+      goto error;
     }
 
     if (check_access(thd, CREATE_PROC_ACL, lex->sphead->m_db.str,
                      NULL, NULL, 0, 0))
-      goto create_sp_error;
-
-    /*
-      Check that a database directory with this name
-      exists. Design note: This won't work on virtual databases
-      like information_schema.
-    */
-    if (check_db_dir_existence(lex->sphead->m_db.str))
-    {
-      my_error(ER_BAD_DB_ERROR, MYF(0), lex->sphead->m_db.str);
-      goto create_sp_error;
-    }
+      goto error;
 
     /* Checking the drop permissions if CREATE OR REPLACE is used */
     if (lex->create_info.or_replace())
@@ -5139,7 +5125,7 @@ end_with_restore_list:
       if (check_routine_access(thd, ALTER_PROC_ACL, lex->spname->m_db.str,
                                lex->spname->m_name.str,
                                lex->sql_command == SQLCOM_DROP_PROCEDURE, 0))
-        goto create_sp_error;
+        goto error;
     }
 
     name= lex->sphead->name(&namelen);
@@ -5151,18 +5137,17 @@ end_with_restore_list:
       if (udf)
       {
         my_error(ER_UDF_EXISTS, MYF(0), name);
-        goto create_sp_error;
+        goto error;
       }
     }
 #endif
 
     if (sp_process_definer(thd))
-      goto create_sp_error;
+      goto error;
 
     WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL)
-    res= (sp_result= sp_create_routine(thd, lex->sphead->m_type, lex->sphead));
-    switch (sp_result) {
-    case SP_OK: {
+    if (!sp_create_routine(thd, lex->sphead->m_type, lex->sphead))
+    {
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
       /* only add privileges if really neccessary */
 
@@ -5227,31 +5212,8 @@ end_with_restore_list:
       }
 
 #endif
-    break;
     }
-    case SP_WRITE_ROW_FAILED:
-      my_error(ER_SP_ALREADY_EXISTS, MYF(0), SP_TYPE_STRING(lex), name);
-    break;
-    case SP_BAD_IDENTIFIER:
-      my_error(ER_TOO_LONG_IDENT, MYF(0), name);
-    break;
-    case SP_BODY_TOO_LONG:
-      my_error(ER_TOO_LONG_BODY, MYF(0), name);
-    break;
-    case SP_FLD_STORE_FAILED:
-      my_error(ER_CANT_CREATE_SROUTINE, MYF(0), name);
-      break;
-    default:
-      my_error(ER_SP_STORE_FAILED, MYF(0), SP_TYPE_STRING(lex), name);
-    break;
-    } /* end switch */
-
-    /*
-      Capture all errors within this CASE and
-      clean up the environment.
-    */
-create_sp_error:
-    if (sp_result != SP_OK )
+    else
       goto error;
     my_ok(thd);
     break; /* break super switch */