database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-07-30 16:24:05 +03:00
Code Activity

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

Browse Source 
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.
This commit is contained in:
Alexey Botchkov
2016-03-23 12:16:39 +04:00
parent e4435b5ec3
commit 2783fc7d14
6 changed files with 219 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
sql/sql_parse.cc
View File

@ -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 */