Combine GLOBAL and COMMIT namespaces into BACKUP namespace.

Part of MDEV-5336 Implement LOCK FOR BACKUP

Other things:
- Added printing of MDL locks to DBUG.

mysql-test/include/check_ftwrl_incompatible.inc

@@ -68,8 +68,7 @@ connection $con_aux1;
 --enable_query_log
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where (state = "Waiting for global read lock" or
+  where state = "Waiting for backup lock" and
-         state = "Waiting for commit lock") and
         info = "$statement";
 --source include/wait_condition.inc
 --disable_result_log
@@ -116,8 +115,7 @@ connection $con_aux2;
 --enable_query_log
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where (state = "Waiting for global read lock" or
+  where state = "Waiting for backup lock" and
-         state = "Waiting for commit lock") and
         info = "flush tables with read lock";
 --source include/wait_condition.inc
 --disable_result_log

mysql-test/main/create_or_replace.result

@@ -262,7 +262,7 @@ create table mysqltest2.t2 like test.t1;
 lock table test.t1 write, mysqltest2.t2 write;
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-#	MDL_INTENTION_EXCLUSIVE	NULL	Global read lock		
+#	MDL_BACKUP_STMT	NULL	Backup lock		
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	mysqltest2	
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	test	
 #	MDL_SHARED_NO_READ_WRITE	NULL	Table metadata lock	mysqltest2	t2
@@ -274,7 +274,7 @@ Tables_in_test
 t2
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-#	MDL_INTENTION_EXCLUSIVE	NULL	Global read lock		
+#	MDL_BACKUP_STMT	NULL	Backup lock		
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	mysqltest2	
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	test	
 #	MDL_SHARED_NO_READ_WRITE	NULL	Table metadata lock	mysqltest2	t2
@@ -289,7 +289,7 @@ create table mysqltest2.t2 like test.t1;
 lock table test.t1 write, mysqltest2.t2 write;
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-#	MDL_INTENTION_EXCLUSIVE	NULL	Global read lock		
+#	MDL_BACKUP_STMT	NULL	Backup lock		
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	mysqltest2	
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	test	
 #	MDL_SHARED_NO_READ_WRITE	NULL	Table metadata lock	mysqltest2	t2
@@ -301,7 +301,7 @@ Tables_in_test
 t2
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-#	MDL_INTENTION_EXCLUSIVE	NULL	Global read lock		
+#	MDL_BACKUP_STMT	NULL	Backup lock		
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	mysqltest2	
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	test	
 #	MDL_SHARED_NO_READ_WRITE	NULL	Table metadata lock	mysqltest2	t2
@@ -398,28 +398,28 @@ create table t1 (a int);
 lock table t1 write, t2 read;
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-#	MDL_INTENTION_EXCLUSIVE	NULL	Global read lock		
+#	MDL_BACKUP_STMT	NULL	Backup lock		
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	test	
 #	MDL_SHARED_NO_READ_WRITE	NULL	Table metadata lock	test	t1
 #	MDL_SHARED_READ	NULL	Table metadata lock	test	t2
 create or replace table t1 (i int);
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-#	MDL_INTENTION_EXCLUSIVE	NULL	Global read lock		
+#	MDL_BACKUP_STMT	NULL	Backup lock		
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	test	
 #	MDL_SHARED_NO_READ_WRITE	NULL	Table metadata lock	test	t1
 #	MDL_SHARED_READ	NULL	Table metadata lock	test	t2
 create or replace table t1 like t2;
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-#	MDL_INTENTION_EXCLUSIVE	NULL	Global read lock		
+#	MDL_BACKUP_STMT	NULL	Backup lock		
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	test	
 #	MDL_SHARED_NO_READ_WRITE	NULL	Table metadata lock	test	t1
 #	MDL_SHARED_READ	NULL	Table metadata lock	test	t2
 create or replace table t1 select 1 as f1;
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-#	MDL_INTENTION_EXCLUSIVE	NULL	Global read lock		
+#	MDL_BACKUP_STMT	NULL	Backup lock		
 #	MDL_INTENTION_EXCLUSIVE	NULL	Schema metadata lock	test	
 #	MDL_SHARED_NO_READ_WRITE	NULL	Table metadata lock	test	t1
 #	MDL_SHARED_READ	NULL	Table metadata lock	test	t2

mysql-test/main/flush.test

@@ -557,7 +557,7 @@ connection con2;
 --echo # Wait until INSERT starts to wait for FTWRL to go away.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock"
+  where state = "Waiting for backup lock"
   and info = "insert into t2 values (1)";
 --source include/wait_condition.inc
 

mysql-test/main/flush_block_commit.test

@@ -32,7 +32,7 @@ connection con2;
 --echo # Wait until COMMIT gets blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and info = "COMMIT";
+  where state = "Waiting for backup lock" and info = "COMMIT";
 --source include/wait_condition.inc
 --echo # Verify that 'con1' was blocked and data did not move.
 SELECT * FROM t1;

mysql-test/main/flush_block_commit_notembedded.test

@@ -46,7 +46,7 @@ begin;
 connection con1;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "insert into t1 values (1)";
 --source include/wait_condition.inc
 unlock tables;

mysql-test/main/flush_read_lock.test

@@ -259,7 +259,7 @@ connection $con_aux1;
 --echo # Wait until COMMIT is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and
+  where state = "Waiting for backup lock" and
         info = "commit";
 --source include/wait_condition.inc
 unlock tables;
@@ -281,7 +281,7 @@ connection $con_aux2;
 --echo # Wait until FTWRL is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and
+  where state = "Waiting for backup lock" and
         info = "flush tables with read lock";
 --source include/wait_condition.inc
 set debug_sync='now SIGNAL go';
@@ -565,7 +565,7 @@ connection $con_aux1;
 --echo # Check that EXECUTE is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "insert into t1_base values (1)";
 --source include/wait_condition.inc
 unlock tables;
@@ -582,7 +582,7 @@ connection $con_aux2;
 --echo # Wait until FTWRL is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "flush tables with read lock";
 --source include/wait_condition.inc
 set debug_sync='now SIGNAL go';
@@ -1011,7 +1011,7 @@ connection $con_aux1;
 --echo # Check that LOCK TABLES WRITE is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "lock tables t1_base write";
 --source include/wait_condition.inc
 unlock tables;
@@ -1055,7 +1055,7 @@ connection $con_aux1;
 --echo # Check that OPTIMIZE TABLE is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "optimize table t1_base";
 --source include/wait_condition.inc
 unlock tables;
@@ -1221,7 +1221,7 @@ connection $con_aux1;
 --echo # Check that REPAIR TABLE is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "repair table t1_base";
 --source include/wait_condition.inc
 unlock tables;
@@ -1420,7 +1420,7 @@ connection $con_aux1;
 --echo # Wait until SET AUTOCOMMIT=1 is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and
+  where state = "Waiting for backup lock" and
         info = "set autocommit= 1";
 --source include/wait_condition.inc
 unlock tables;
@@ -1442,7 +1442,7 @@ connection $con_aux2;
 --echo # Wait until FTWRL is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and
+  where state = "Waiting for backup lock" and
         info = "flush tables with read lock";
 --source include/wait_condition.inc
 set debug_sync='now SIGNAL go';
@@ -1621,7 +1621,7 @@ connection $con_aux1;
 --echo # Wait until XA COMMIT is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and
+  where state = "Waiting for backup lock" and
         info = "xa commit 'test1'";
 --source include/wait_condition.inc
 unlock tables;
@@ -1645,7 +1645,7 @@ connection $con_aux2;
 --echo # Wait until FTWRL is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and
+  where state = "Waiting for backup lock" and
         info = "flush tables with read lock";
 --source include/wait_condition.inc
 set debug_sync='now SIGNAL go';
@@ -1724,7 +1724,7 @@ connection $con_aux1;
 --echo # Check that ANALYZE TABLE is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and
+  where state = "Waiting for backup lock" and
         info = "analyze table t3_trans";
 --source include/wait_condition.inc
 unlock tables;
@@ -1799,7 +1799,7 @@ connection $con_aux1;
 --echo # Check that CHECK TABLE is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and
+  where state = "Waiting for backup lock" and
         info = "check table t1_base";
 --source include/wait_condition.inc
 unlock tables;
@@ -1817,7 +1817,7 @@ connection $con_aux1;
 --echo # Check that ALTER TABLE is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock" and
+  where state = "Waiting for backup lock" and
         info = "alter table t1_temp add column c1 int";
 --source include/wait_condition.inc
 unlock tables;
@@ -1870,7 +1870,7 @@ connection $con_aux2;
 --echo # Wait until FTWRL is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "flush tables with read lock";
 --source include/wait_condition.inc
 --echo # Try to run another INSERT and see that it is blocked.
@@ -1879,7 +1879,7 @@ connection con3;
 --echo # Wait until new INSERT is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "insert into t2_base values (1)";
 --echo # Unblock INSERT in the first connection.
 set debug_sync='now SIGNAL go';

mysql-test/main/flush_read_lock_kill.test

@@ -51,7 +51,7 @@ SELECT ((@id := kill_id) - kill_id) FROM t1 LIMIT 1;
 --echo # to active COMMIT
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for commit lock"
+  where state = "Waiting for backup lock"
   and info = "flush tables with read lock";
 --source include/wait_condition.inc
 

mysql-test/main/lock_multi.test

@@ -229,7 +229,7 @@ connection writer;
 # Sleep a bit till the flush of connection locker is in work and hangs
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "FLUSH TABLES WITH READ LOCK";
 --source include/wait_condition.inc
 # This must not block.
@@ -261,7 +261,7 @@ connection writer;
 # Sleep a bit till the flush of connection locker is in work and hangs
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "FLUSH TABLES WITH READ LOCK";
 --source include/wait_condition.inc
 --error ER_TABLE_NOT_LOCKED
@@ -298,10 +298,10 @@ DROP DATABASE mysqltest_1;
 # When fixed: Reject dropping db because of the read lock.
 connection con1;
 # Wait a bit so that the session con2 is in state
-# "Waiting for global read lock"
+# "Waiting for backup lock"
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock"
+  where state = "Waiting for backup lock"
   and info = "DROP DATABASE mysqltest_1";
 --source include/wait_condition.inc
 --error ER_CANT_UPDATE_WITH_READLOCK
@@ -377,7 +377,7 @@ send flush tables with read lock;
 connection con5;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "flush tables with read lock";
 --source include/wait_condition.inc
 --echo # global read lock is taken
@@ -386,7 +386,7 @@ send select * from t2 for update;
 connection con5;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "select * from t2 for update";
 --source include/wait_condition.inc
 --echo # waiting for release of read lock
@@ -432,7 +432,7 @@ send update t2 set a = 1;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "update t2 set a = 1";
 --source include/wait_condition.inc
 --echo # statement is waiting for release of read lock
@@ -454,7 +454,7 @@ send lock tables t2 write;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "lock tables t2 write";
 --source include/wait_condition.inc
 --echo # statement is waiting for release of read lock
@@ -542,7 +542,7 @@ connection flush;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "flush tables with read lock";
 --source include/wait_condition.inc
 alter table t1 add column j int;
@@ -550,14 +550,14 @@ connect (insert,localhost,root,,test,,);
 connection insert;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "flush tables with read lock";
 --source include/wait_condition.inc
 --send insert into t1 values (1,2);
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "insert into t1 values (1,2)";
 --source include/wait_condition.inc
 unlock tables;
@@ -565,7 +565,7 @@ connection flush;
 --reap
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock" and
+  where state = "Waiting for backup lock" and
         info = "insert into t1 values (1,2)";
 --source include/wait_condition.inc
 select * from t1;
@@ -598,12 +598,12 @@ connection flush;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock";
+  where state = "Waiting for backup lock";
 --source include/wait_condition.inc
 flush tables;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock";
+  where state = "Waiting for backup lock";
 --source include/wait_condition.inc
 unlock tables;
 connection flush;
@@ -664,12 +664,12 @@ connection flush;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock";
+  where state = "Waiting for backup lock";
 --source include/wait_condition.inc
 flush tables;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock";
+  where state = "Waiting for backup lock";
 --source include/wait_condition.inc
 drop table t1;
 connection flush;

mysql-test/main/mdl.result

@@ -9,13 +9,13 @@ CREATE TABLE t1(a INT) ENGINE=InnoDB;
 LOCK TABLES t1 WRITE CONCURRENT, t1 AS t2 READ;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-MDL_INTENTION_EXCLUSIVE	Global read lock		
+MDL_BACKUP_STMT	Backup lock		
 MDL_SHARED_NO_READ_WRITE	Table metadata lock	test	t1
 UNLOCK TABLES;
 LOCK TABLES t1 AS t2 READ, t1 WRITE CONCURRENT;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
 LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
-MDL_INTENTION_EXCLUSIVE	Global read lock		
+MDL_BACKUP_STMT	Backup lock		
 MDL_SHARED_WRITE	Table metadata lock	test	t1
 MDL_SHARED_READ_ONLY	Table metadata lock	test	t1
 UNLOCK TABLES;

mysql-test/main/mdl_sync.test

@@ -3969,7 +3969,7 @@ connection con2;
 
 connection default;
 let $wait_condition=SELECT COUNT(*)=1 FROM information_schema.processlist
-  WHERE state='Waiting for global read lock'
+  WHERE state='Waiting for backup lock'
   AND info='CREATE TABLE db1.t2(a INT)';
 --source include/wait_condition.inc
 UNLOCK TABLES;
@@ -3987,7 +3987,7 @@ connection con2;
 
 connection default;
 let $wait_condition=SELECT COUNT(*)=1 FROM information_schema.processlist
-  WHERE state='Waiting for global read lock'
+  WHERE state='Waiting for backup lock'
   AND info='ALTER DATABASE db1 DEFAULT CHARACTER SET utf8';
 --source include/wait_condition.inc
 UNLOCK TABLES;

mysql-test/main/trigger_notembedded.test

@@ -907,7 +907,7 @@ connection flush;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for global read lock";
+  where state = "Waiting for backup lock";
 --source include/wait_condition.inc
 create trigger t1_bi before insert on t1 for each row begin end;
 unlock tables;

mysql-test/suite/compat/oracle/r/sp-package-mdl.result

@@ -61,8 +61,8 @@ TABLE_NAME	pkg1.p1
 CONN	2
 INFO	DROP PACKAGE pkg1
 STATE	Waiting for stored package body metadata lock
-LOCK_MODE	MDL_INTENTION_EXCLUSIVE
+LOCK_MODE	MDL_BACKUP_STMT
-LOCK_TYPE	Global read lock
+LOCK_TYPE	Backup lock
 TABLE_NAME	
 CONN	2
 INFO	DROP PACKAGE pkg1

mysql-test/suite/perfschema/r/stage_mdl_global.result

@@ -22,7 +22,7 @@ call dump_one_thread('user2');
 username	event_name	sql_text
 user2	statement/sql/insert	insert into test.t1 values (1), (2), (3)
 username	event_name	nesting_event_type
-user2	stage/sql/Waiting for global read lock	STATEMENT
+user2	stage/sql/Waiting for backup lock	STATEMENT
 username	event_name	nesting_event_type
 user2	stage/sql/Init	STATEMENT
 user2	stage/sql/Checking permissions	STATEMENT

mysql-test/suite/perfschema/t/stage_mdl_global.test

@@ -9,7 +9,7 @@ flush tables with read lock;
 
 connect (con2, localhost, user2, , );
 
-# Will wait on con1, "Waiting for global read lock"
+# Will wait on con1, "Waiting for backup lock"
 --send
 insert into test.t1 values (1), (2), (3);
 
@@ -26,7 +26,7 @@ let $wait_condition=
 let $wait_condition=
   select count(*) = 1 from performance_schema.threads
   where `TYPE`='FOREGROUND' and PROCESSLIST_USER like 'user2'
-  and PROCESSLIST_STATE = 'Waiting for global read lock';
+  and PROCESSLIST_STATE = 'Waiting for backup lock';
 --source include/wait_condition.inc
 
 call dump_one_thread('user1');

plugin/metadata_lock_info/metadata_lock_info.cc

@@ -21,7 +21,7 @@
 #include "sql_show.h"
 
 static const LEX_STRING metadata_lock_info_lock_name[] = {
-  { C_STRING_WITH_LEN("Global read lock") },
+  { C_STRING_WITH_LEN("Backup lock") },
   { C_STRING_WITH_LEN("Schema metadata lock") },
   { C_STRING_WITH_LEN("Table metadata lock") },
   { C_STRING_WITH_LEN("Stored function metadata lock") },
@@ -29,23 +29,9 @@ static const LEX_STRING metadata_lock_info_lock_name[] = {
   { C_STRING_WITH_LEN("Stored package body metadata lock") },
   { C_STRING_WITH_LEN("Trigger metadata lock") },
   { C_STRING_WITH_LEN("Event metadata lock") },
-  { C_STRING_WITH_LEN("Commit lock") },
   { C_STRING_WITH_LEN("User lock") },
 };
 
-static const LEX_STRING metadata_lock_info_lock_mode[] = {
-  { C_STRING_WITH_LEN("MDL_INTENTION_EXCLUSIVE") },
-  { C_STRING_WITH_LEN("MDL_SHARED") },
-  { C_STRING_WITH_LEN("MDL_SHARED_HIGH_PRIO") },
-  { C_STRING_WITH_LEN("MDL_SHARED_READ") },
-  { C_STRING_WITH_LEN("MDL_SHARED_WRITE") },
-  { C_STRING_WITH_LEN("MDL_SHARED_UPGRADABLE") },
-  { C_STRING_WITH_LEN("MDL_SHARED_READ_ONLY") },
-  { C_STRING_WITH_LEN("MDL_SHARED_NO_WRITE") },
-  { C_STRING_WITH_LEN("MDL_SHARED_NO_READ_WRITE") },
-  { C_STRING_WITH_LEN("MDL_EXCLUSIVE") },
-};
-
 static ST_FIELD_INFO i_s_metadata_lock_info_fields_info[] =
 {
   {"THREAD_ID", 20, MYSQL_TYPE_LONGLONG, 0,
@@ -71,22 +57,21 @@ struct st_i_s_metadata_param
 
 int i_s_metadata_lock_info_fill_row(
   MDL_ticket *mdl_ticket,
-  void *arg
+  void *arg,
+  bool granted
 ) {
   st_i_s_metadata_param *param = (st_i_s_metadata_param *) arg;
   THD *thd = param->thd;
   TABLE *table = param->table;
   DBUG_ENTER("i_s_metadata_lock_info_fill_row");
   MDL_context *mdl_ctx = mdl_ticket->get_ctx();
-  enum_mdl_type mdl_ticket_type = mdl_ticket->get_type();
   MDL_key *mdl_key = mdl_ticket->get_key();
   MDL_key::enum_mdl_namespace mdl_namespace = mdl_key->mdl_namespace();
+  if (!granted)
+    DBUG_RETURN(0);
   table->field[0]->store((longlong) mdl_ctx->get_thread_id(), TRUE);
   table->field[1]->set_notnull();
-  table->field[1]->store(
+  table->field[1]->store(mdl_ticket->get_type_name(), system_charset_info);
-    metadata_lock_info_lock_mode[(int) mdl_ticket_type].str,
-    metadata_lock_info_lock_mode[(int) mdl_ticket_type].length,
-    system_charset_info);
   table->field[2]->set_null();
   table->field[3]->set_notnull();
   table->field[3]->store(
@@ -122,8 +107,6 @@ static int i_s_metadata_lock_info_init(
 
   compile_time_assert(sizeof(metadata_lock_info_lock_name)/sizeof(LEX_STRING)
                       == MDL_key::NAMESPACE_END);
-  compile_time_assert(sizeof(metadata_lock_info_lock_mode)/sizeof(LEX_STRING)
-                      == MDL_TYPE_END);
 
   ST_SCHEMA_TABLE *schema = (ST_SCHEMA_TABLE *) p;
   DBUG_ENTER("i_s_metadata_lock_info_init");

plugin/metadata_lock_info/mysql-test/metadata_lock_info/r/global_read_lock.result

@@ -3,8 +3,7 @@ lock_mode	lock_duration	lock_type	table_schema	table_name
 FLUSH TABLES WITH READ LOCK;
 SELECT lock_mode, lock_duration, lock_type, table_schema, table_name FROM information_schema.metadata_lock_info;
 lock_mode	lock_duration	lock_type	table_schema	table_name
-MDL_SHARED	NULL	Commit lock		
+MDL_BACKUP_FTWRL2	NULL	Backup lock		
-MDL_SHARED	NULL	Global read lock		
 UNLOCK TABLES;
 SELECT lock_mode, lock_duration, lock_type, table_schema, table_name FROM information_schema.metadata_lock_info;
 lock_mode	lock_duration	lock_type	table_schema	table_name

sql/handler.cc

@@ -1431,8 +1431,7 @@ int ha_commit_trans(THD *thd, bool all)
       We allow the owner of FTWRL to COMMIT; we assume that it knows
       what it does.
     */
-    mdl_request.init(MDL_key::COMMIT, "", "", MDL_INTENTION_EXCLUSIVE,
+    mdl_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_COMMIT, MDL_EXPLICIT);
-                     MDL_EXPLICIT);
 
     if (!WSREP(thd) &&
       thd->mdl_context.acquire_lock(&mdl_request,

sql/lock.cc

@@ -863,8 +863,7 @@ bool lock_schema_name(THD *thd, const char *db)
 
   if (thd->global_read_lock.can_acquire_protection())
     return TRUE;
-  global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
+  global_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_STMT, MDL_STATEMENT);
-                      MDL_STATEMENT);
   mdl_request.init(MDL_key::SCHEMA, db, "", MDL_EXCLUSIVE, MDL_TRANSACTION);
 
   mdl_requests.push_front(&mdl_request);
@@ -922,8 +921,7 @@ bool lock_object_name(THD *thd, MDL_key::enum_mdl_namespace mdl_type,
 
   if (thd->global_read_lock.can_acquire_protection())
     return TRUE;
-  global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
+  global_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_STMT, MDL_STATEMENT);
-                      MDL_STATEMENT);
   schema_request.init(MDL_key::SCHEMA, db, "", MDL_INTENTION_EXCLUSIVE,
                       MDL_TRANSACTION);
   mdl_request.init(mdl_type, db, name, MDL_EXCLUSIVE, MDL_TRANSACTION);
@@ -1018,15 +1016,17 @@ bool Global_read_lock::lock_global_read_lock(THD *thd)
 
     mysql_ha_cleanup_no_free(thd);
 
-    DBUG_ASSERT(! thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
+    DBUG_ASSERT(! thd->mdl_context.is_lock_owner(MDL_key::BACKUP, "", "",
-                                                 MDL_SHARED));
+                                                 MDL_BACKUP_FTWRL1));
-    mdl_request.init(MDL_key::GLOBAL, "", "", MDL_SHARED, MDL_EXPLICIT);
+    DBUG_ASSERT(! thd->mdl_context.is_lock_owner(MDL_key::BACKUP, "", "",
+                                                 MDL_BACKUP_FTWRL2));
+    mdl_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_FTWRL1, MDL_EXPLICIT);
 
     if (thd->mdl_context.acquire_lock(&mdl_request,
                                       thd->variables.lock_wait_timeout))
       DBUG_RETURN(1);
 
-    m_mdl_global_shared_lock= mdl_request.ticket;
+    m_mdl_global_read_lock= mdl_request.ticket;
     m_state= GRL_ACQUIRED;
   }
   /*
@@ -1055,7 +1055,7 @@ void Global_read_lock::unlock_global_read_lock(THD *thd)
 {
   DBUG_ENTER("unlock_global_read_lock");
 
-  DBUG_ASSERT(m_mdl_global_shared_lock && m_state);
+  DBUG_ASSERT(m_mdl_global_read_lock && m_state);
 
   if (thd->global_disable_checkpoint)
   {
@@ -1066,11 +1066,11 @@ void Global_read_lock::unlock_global_read_lock(THD *thd)
     }
   }
 
-  if (m_mdl_blocks_commits_lock)
+  thd->mdl_context.release_lock(m_mdl_global_read_lock);
-  {
+
-    thd->mdl_context.release_lock(m_mdl_blocks_commits_lock);
-    m_mdl_blocks_commits_lock= NULL;
 #ifdef WITH_WSREP
+  if (m_state == GRL_ACQUIRED_AND_BLOCKS_COMMIT)
+  {
     if (WSREP(thd) || wsrep_node_is_donor())
     {
       wsrep_locked_seqno= WSREP_SEQNO_UNDEFINED;
@@ -1083,14 +1083,13 @@ void Global_read_lock::unlock_global_read_lock(THD *thd)
         {
           WSREP_WARN("resync failed %d for FTWRL: db: %s, query: %s",
                      ret, thd->get_db(), thd->query());
-          DBUG_VOID_RETURN;
         }
       }
     }
-#endif /* WITH_WSREP */
   }
-  thd->mdl_context.release_lock(m_mdl_global_shared_lock);
+#endif /* WITH_WSREP */
-  m_mdl_global_shared_lock= NULL;
+
+  m_mdl_global_read_lock= NULL;
   m_state= GRL_NONE;
 
   DBUG_VOID_RETURN;
@@ -1114,7 +1113,6 @@ void Global_read_lock::unlock_global_read_lock(THD *thd)
 
 bool Global_read_lock::make_global_read_lock_block_commit(THD *thd)
 {
-  MDL_request mdl_request;
   DBUG_ENTER("make_global_read_lock_block_commit");
   /*
     If we didn't succeed lock_global_read_lock(), or if we already suceeded
@@ -1124,22 +1122,11 @@ bool Global_read_lock::make_global_read_lock_block_commit(THD *thd)
   if (m_state != GRL_ACQUIRED)
     DBUG_RETURN(0);
 
-#ifdef WITH_WSREP
+  if (thd->mdl_context.upgrade_shared_lock(m_mdl_global_read_lock,
-  if (WSREP(thd) && m_mdl_blocks_commits_lock)
+                                           MDL_BACKUP_FTWRL2,
-  {
+                                           thd->variables.lock_wait_timeout))
-    WSREP_DEBUG("GRL was in block commit mode when entering "
-		"make_global_read_lock_block_commit");
-    DBUG_RETURN(FALSE);
-  }
-#endif /* WITH_WSREP */
-
-  mdl_request.init(MDL_key::COMMIT, "", "", MDL_SHARED, MDL_EXPLICIT);
-
-  if (thd->mdl_context.acquire_lock(&mdl_request,
-                                    thd->variables.lock_wait_timeout))
     DBUG_RETURN(TRUE);
 
-  m_mdl_blocks_commits_lock= mdl_request.ticket;
   m_state= GRL_ACQUIRED_AND_BLOCKS_COMMIT;
 
 #ifdef WITH_WSREP
@@ -1190,7 +1177,11 @@ bool Global_read_lock::make_global_read_lock_block_commit(THD *thd)
     {
       WSREP_ERROR("Failed to pause provider: %lld (%s)", -ret, strerror(-ret));
 
-      DBUG_ASSERT(m_mdl_blocks_commits_lock == NULL);
+      /*
+        For some reason Galera wants to crash here in debug build.
+        It is equivalent of original assertion.
+      */
+      DBUG_ASSERT(0);
       wsrep_locked_seqno= WSREP_SEQNO_UNDEFINED;
       my_error(ER_LOCK_DEADLOCK, MYF(0));
       DBUG_RETURN(TRUE);
@@ -1209,10 +1200,8 @@ bool Global_read_lock::make_global_read_lock_block_commit(THD *thd)
 
 void Global_read_lock::set_explicit_lock_duration(THD *thd)
 {
-  if (m_mdl_global_shared_lock)
+  if (m_mdl_global_read_lock)
-    thd->mdl_context.set_lock_duration(m_mdl_global_shared_lock, MDL_EXPLICIT);
+    thd->mdl_context.set_lock_duration(m_mdl_global_read_lock, MDL_EXPLICIT);
-  if (m_mdl_blocks_commits_lock)
-    thd->mdl_context.set_lock_duration(m_mdl_blocks_commits_lock, MDL_EXPLICIT);
 }
 
 /**

sql/mdl.cc

@@ -80,7 +80,7 @@ static void init_mdl_psi_keys(void)
 
 PSI_stage_info MDL_key::m_namespace_to_wait_state_name[NAMESPACE_END]=
 {
-  {0, "Waiting for global read lock", 0},
+  {0, "Waiting for backup lock", 0},
   {0, "Waiting for schema metadata lock", 0},
   {0, "Waiting for table metadata lock", 0},
   {0, "Waiting for stored function metadata lock", 0},
@@ -88,10 +88,33 @@ PSI_stage_info MDL_key::m_namespace_to_wait_state_name[NAMESPACE_END]=
   {0, "Waiting for stored package body metadata lock", 0},
   {0, "Waiting for trigger metadata lock", 0},
   {0, "Waiting for event metadata lock", 0},
-  {0, "Waiting for commit lock", 0},
   {0, "User lock", 0} /* Be compatible with old status. */
 };
 
+
+static const LEX_STRING lock_types[]=
+{
+  { C_STRING_WITH_LEN("MDL_INTENTION_EXCLUSIVE") },
+  { C_STRING_WITH_LEN("MDL_SHARED") },
+  { C_STRING_WITH_LEN("MDL_SHARED_HIGH_PRIO") },
+  { C_STRING_WITH_LEN("MDL_SHARED_READ") },
+  { C_STRING_WITH_LEN("MDL_SHARED_WRITE") },
+  { C_STRING_WITH_LEN("MDL_SHARED_UPGRADABLE") },
+  { C_STRING_WITH_LEN("MDL_SHARED_READ_ONLY") },
+  { C_STRING_WITH_LEN("MDL_SHARED_NO_WRITE") },
+  { C_STRING_WITH_LEN("MDL_SHARED_NO_READ_WRITE") },
+  { C_STRING_WITH_LEN("MDL_EXCLUSIVE") },
+};
+
+
+static const LEX_STRING backup_lock_types[]=
+{
+  { C_STRING_WITH_LEN("MDL_BACKUP_FTWRL1") },
+  { C_STRING_WITH_LEN("MDL_BACKUP_FTWRL2") },
+  { C_STRING_WITH_LEN("MDL_BACKUP_STMT") },
+  { C_STRING_WITH_LEN("MDL_BACKUP_COMMIT") }
+};
+
 #ifdef HAVE_PSI_INTERFACE
 void MDL_key::init_psi_keys()
 {
@@ -128,11 +151,9 @@ public:
   LF_PINS *get_pins() { return lf_hash_get_pins(&m_locks); }
 private:
   LF_HASH m_locks; /**< All acquired locks in the server. */
-  /** Pre-allocated MDL_lock object for GLOBAL namespace. */
+  /** Pre-allocated MDL_lock object for BACKUP namespace. */
-  MDL_lock *m_global_lock;
+  MDL_lock *m_backup_lock;
-  /** Pre-allocated MDL_lock object for COMMIT namespace. */
+  friend int mdl_iterate(mdl_iterator_callback, void *);
-  MDL_lock *m_commit_lock;
-  friend int mdl_iterate(int (*)(MDL_ticket *, void *), void *);
 };
 
 
@@ -328,9 +349,10 @@ public:
 
   /**
     Helper struct which defines how different types of locks are handled
-    for a specific MDL_lock. In practice we use only two strategies: "scoped"
+    for a specific MDL_lock. In practice we use only three strategies:
-    lock strategy for locks in GLOBAL, COMMIT and SCHEMA namespaces and
+    "backup" lock strategy for locks in BACKUP namespace, "scoped" lock
-    "object" lock strategy for all other namespaces.
+    strategy for locks in SCHEMA namespace and "object" lock strategy for
+    all other namespaces.
   */
   struct MDL_lock_strategy
   {
@@ -427,6 +449,41 @@ public:
     static const bitmap_t m_waiting_incompatible[MDL_TYPE_END];
   };
 
+
+  struct MDL_backup_lock: public MDL_lock_strategy
+  {
+    MDL_backup_lock() {}
+    virtual const bitmap_t *incompatible_granted_types_bitmap() const
+    { return m_granted_incompatible; }
+    virtual const bitmap_t *incompatible_waiting_types_bitmap() const
+    { return m_waiting_incompatible; }
+    virtual bool needs_notification(const MDL_ticket *ticket) const
+    {
+      return ticket->get_type() == MDL_BACKUP_FTWRL1;
+    }
+
+    /**
+      Threads having aninsert delayed thread may hold STMT lock. We
+      need to kill such threads in order to get backup lock for FTWRL
+      or BACKUP statements. We do this my calling code outside of MDL.
+    */
+    virtual bool conflicting_locks(const MDL_ticket *ticket) const
+    {
+      return ticket->get_type() == MDL_BACKUP_STMT;
+    }
+
+    /*
+      In backup namespace DML/DDL may starve because of concurrent FTWRL or
+      BACKUP statements. This scenario is partically useless in real world,
+      so we just return 0 here.
+    */
+    virtual bitmap_t hog_lock_types_bitmap() const
+    { return 0; }
+  private:
+    static const bitmap_t m_granted_incompatible[MDL_BACKUP_END];
+    static const bitmap_t m_waiting_incompatible[MDL_BACKUP_END];
+  };
+
 public:
   /** The key of the object (data) being protected. */
   MDL_key key;
@@ -538,10 +595,9 @@ public:
   MDL_lock(const MDL_key *key_arg)
   : key(key_arg),
     m_hog_lock_count(0),
-    m_strategy(&m_scoped_lock_strategy)
+    m_strategy(&m_backup_lock_strategy)
   {
-    DBUG_ASSERT(key_arg->mdl_namespace() == MDL_key::GLOBAL ||
+    DBUG_ASSERT(key_arg->mdl_namespace() == MDL_key::BACKUP);
-                key_arg->mdl_namespace() == MDL_key::COMMIT);
     mysql_prlock_init(key_MDL_lock_rwlock, &m_rwlock);
   }
 
@@ -557,8 +613,7 @@ public:
   static void lf_hash_initializer(LF_HASH *hash __attribute__((unused)),
                                   MDL_lock *lock, MDL_key *key_arg)
   {
-    DBUG_ASSERT(key_arg->mdl_namespace() != MDL_key::GLOBAL &&
+    DBUG_ASSERT(key_arg->mdl_namespace() != MDL_key::BACKUP);
-                key_arg->mdl_namespace() != MDL_key::COMMIT);
     new (&lock->key) MDL_key(key_arg);
     if (key_arg->mdl_namespace() == MDL_key::SCHEMA)
       lock->m_strategy= &m_scoped_lock_strategy;
@@ -568,11 +623,13 @@ public:
 
   const MDL_lock_strategy *m_strategy;
 private:
+  static const MDL_backup_lock m_backup_lock_strategy;
   static const MDL_scoped_lock m_scoped_lock_strategy;
   static const MDL_object_lock m_object_lock_strategy;
 };
 
 
+const MDL_lock::MDL_backup_lock MDL_lock::m_backup_lock_strategy;
 const MDL_lock::MDL_scoped_lock MDL_lock::m_scoped_lock_strategy;
 const MDL_lock::MDL_object_lock MDL_lock::m_object_lock_strategy;
 
@@ -636,7 +693,7 @@ void mdl_destroy()
 
 struct mdl_iterate_arg
 {
-  int (*callback)(MDL_ticket *ticket, void *arg);
+  mdl_iterator_callback callback;
   void *argument;
 };
 
@@ -649,16 +706,19 @@ static my_bool mdl_iterate_lock(MDL_lock *lock, mdl_iterate_arg *arg)
     must be empty for such locks anyway.
   */
   mysql_prlock_rdlock(&lock->m_rwlock);
-  MDL_lock::Ticket_iterator ticket_it(lock->m_granted);
+  MDL_lock::Ticket_iterator granted_it(lock->m_granted);
+  MDL_lock::Ticket_iterator waiting_it(lock->m_waiting);
   MDL_ticket *ticket;
-  while ((ticket= ticket_it++) && !(res= arg->callback(ticket, arg->argument)))
+  while ((ticket= granted_it++) && !(res= arg->callback(ticket, arg->argument, true)))
+    /* no-op */;
+  while ((ticket= waiting_it++) && !(res= arg->callback(ticket, arg->argument, false)))
     /* no-op */;
   mysql_prlock_unlock(&lock->m_rwlock);
   return MY_TEST(res);
 }
 
 
-int mdl_iterate(int (*callback)(MDL_ticket *ticket, void *arg), void *arg)
+int mdl_iterate(mdl_iterator_callback callback, void *arg)
 {
   DBUG_ENTER("mdl_iterate");
   mdl_iterate_arg argument= { callback, arg };
@@ -667,8 +727,7 @@ int mdl_iterate(int (*callback)(MDL_ticket *ticket, void *arg), void *arg)
 
   if (pins)
   {
-    res= mdl_iterate_lock(mdl_locks.m_global_lock, &argument) ||
+    res= mdl_iterate_lock(mdl_locks.m_backup_lock, &argument) ||
-         mdl_iterate_lock(mdl_locks.m_commit_lock, &argument) ||
          lf_hash_iterate(&mdl_locks.m_locks, pins,
                          (my_hash_walk_action) mdl_iterate_lock, &argument);
     lf_hash_put_pins(pins);
@@ -689,11 +748,9 @@ my_hash_value_type mdl_hash_function(CHARSET_INFO *cs,
 
 void MDL_map::init()
 {
-  MDL_key global_lock_key(MDL_key::GLOBAL, "", "");
+  MDL_key backup_lock_key(MDL_key::BACKUP, "", "");
-  MDL_key commit_lock_key(MDL_key::COMMIT, "", "");
 
-  m_global_lock= new (std::nothrow) MDL_lock(&global_lock_key);
+  m_backup_lock= new (std::nothrow) MDL_lock(&backup_lock_key);
-  m_commit_lock= new (std::nothrow) MDL_lock(&commit_lock_key);
 
   lf_hash_init(&m_locks, sizeof(MDL_lock), LF_HASH_UNIQUE, 0, 0,
                mdl_locks_key, &my_charset_bin);
@@ -711,8 +768,7 @@ void MDL_map::init()
 
 void MDL_map::destroy()
 {
-  delete m_global_lock;
+  delete m_backup_lock;
-  delete m_commit_lock;
 
   DBUG_ASSERT(!my_atomic_load32(&m_locks.count));
   lf_hash_destroy(&m_locks);
@@ -732,26 +788,18 @@ MDL_lock* MDL_map::find_or_insert(LF_PINS *pins, const MDL_key *mdl_key)
 {
   MDL_lock *lock;
 
-  if (mdl_key->mdl_namespace() == MDL_key::GLOBAL ||
+  if (mdl_key->mdl_namespace() == MDL_key::BACKUP)
-      mdl_key->mdl_namespace() == MDL_key::COMMIT)
   {
     /*
-      Avoid locking any m_mutex when lock for GLOBAL or COMMIT namespace is
+      Return pointer to pre-allocated MDL_lock instance. Such an optimization
-      requested. Return pointer to pre-allocated MDL_lock instance instead.
+      allows to save one hash lookup for any statement changing data.
-      Such an optimization allows to save one mutex lock/unlock for any
-      statement changing data.
 
-      It works since these namespaces contain only one element so keys
+      It works since this namespace contains only one element so keys
       for them look like '<namespace-id>\0\0'.
     */
     DBUG_ASSERT(mdl_key->length() == 3);
-
+    mysql_prlock_wrlock(&m_backup_lock->m_rwlock);
-    lock= (mdl_key->mdl_namespace() == MDL_key::GLOBAL) ? m_global_lock :
+    return m_backup_lock;
-                                                          m_commit_lock;
-
-    mysql_prlock_wrlock(&lock->m_rwlock);
-
-    return lock;
   }
 
 retry:
@@ -780,22 +828,18 @@ retry:
 unsigned long
 MDL_map::get_lock_owner(LF_PINS *pins, const MDL_key *mdl_key)
 {
-  MDL_lock *lock;
   unsigned long res= 0;
 
-  if (mdl_key->mdl_namespace() == MDL_key::GLOBAL ||
+  if (mdl_key->mdl_namespace() == MDL_key::BACKUP)
-      mdl_key->mdl_namespace() == MDL_key::COMMIT)
   {
-    lock= (mdl_key->mdl_namespace() == MDL_key::GLOBAL) ? m_global_lock :
+    mysql_prlock_rdlock(&m_backup_lock->m_rwlock);
-                                                          m_commit_lock;
+    res= m_backup_lock->get_lock_owner();
-    mysql_prlock_rdlock(&lock->m_rwlock);
+    mysql_prlock_unlock(&m_backup_lock->m_rwlock);
-    res= lock->get_lock_owner();
-    mysql_prlock_unlock(&lock->m_rwlock);
   }
   else
   {
-    lock= (MDL_lock*) lf_hash_search(&m_locks, pins, mdl_key->ptr(),
+    MDL_lock *lock= (MDL_lock*) lf_hash_search(&m_locks, pins, mdl_key->ptr(),
-                                     mdl_key->length());
+                                               mdl_key->length());
     if (lock)
     {
       /*
@@ -820,13 +864,9 @@ MDL_map::get_lock_owner(LF_PINS *pins, const MDL_key *mdl_key)
 
 void MDL_map::remove(LF_PINS *pins, MDL_lock *lock)
 {
-  if (lock->key.mdl_namespace() == MDL_key::GLOBAL ||
+  if (lock->key.mdl_namespace() == MDL_key::BACKUP)
-      lock->key.mdl_namespace() == MDL_key::COMMIT)
   {
-    /*
+    /* Never destroy pre-allocated MDL_lock object in BACKUP namespace. */
-      Never destroy pre-allocated MDL_lock objects for GLOBAL and
-      COMMIT namespaces.
-    */
     mysql_prlock_unlock(&lock->m_rwlock);
     return;
   }
@@ -975,7 +1015,7 @@ void MDL_ticket::destroy(MDL_ticket *ticket)
 
 uint MDL_ticket::get_deadlock_weight() const
 {
-  return (m_lock->key.mdl_namespace() == MDL_key::GLOBAL ||
+  return (m_lock->key.mdl_namespace() == MDL_key::BACKUP ||
           m_type >= MDL_SHARED_UPGRADABLE ?
           DEADLOCK_WEIGHT_DDL : DEADLOCK_WEIGHT_DML);
 }
@@ -1369,7 +1409,7 @@ void MDL_lock::reschedule_waiters()
 /**
   Compatibility (or rather "incompatibility") matrices for scoped metadata
   lock.
-  Scoped locks are GLOBAL READ LOCK, COMMIT and database (or schema) locks.
+  Scoped locks are database (or schema) locks.
   Arrays of bitmaps which elements specify which granted/waiting locks
   are incompatible with type of lock being requested.
 
@@ -1537,6 +1577,60 @@ MDL_lock::MDL_object_lock::m_waiting_incompatible[MDL_TYPE_END]=
 };
 
 
+/**
+  Compatibility (or rather "incompatibility") matrices for backup metadata
+  lock. Arrays of bitmaps which elements specify which granted/waiting locks
+  are incompatible with type of lock being requested.
+
+  The first array specifies if particular type of request can be satisfied
+  if there is granted backup lock of certain type.
+
+             | Type of active  |
+     Request |   backup lock   |
+      type   | F1  F2   S   C  |
+    ---------+-----------------+
+    FTWRL1   |  +   +   -   +  |
+    FTWRL2   |  +   +   -   -  |
+    STMT     |  -   -   +   +  |
+    COMMIT   |  +   -   +   +  |
+
+  The second array specifies if particular type of request can be satisfied
+  if there is already waiting request for the backup lock of certain type.
+  I.e. it specifies what is the priority of different lock types.
+
+             |     Pending     |
+     Request |   backup lock   |
+      type   | F1  F2   S   C  |
+    ---------+-----------------+
+    FTWRL1   |  +   +   +   +  |
+    FTWRL2   |  +   +   +   +  |
+    STMT     |  -   -   +   +  |
+    COMMIT   |  +   -   +   +  |
+
+  Here: "+" -- means that request can be satisfied
+        "-" -- means that request can't be satisfied and should wait
+*/
+
+const MDL_lock::bitmap_t
+MDL_lock::MDL_backup_lock::m_granted_incompatible[MDL_BACKUP_END]=
+{
+  MDL_BIT(MDL_BACKUP_STMT),
+  MDL_BIT(MDL_BACKUP_STMT) | MDL_BIT(MDL_BACKUP_COMMIT),
+  MDL_BIT(MDL_BACKUP_FTWRL1) | MDL_BIT(MDL_BACKUP_FTWRL2),
+  MDL_BIT(MDL_BACKUP_FTWRL2)
+};
+
+
+const MDL_lock::bitmap_t
+MDL_lock::MDL_backup_lock::m_waiting_incompatible[MDL_BACKUP_END]=
+{
+  0,
+  0,
+  MDL_BIT(MDL_BACKUP_FTWRL1) | MDL_BIT(MDL_BACKUP_FTWRL2),
+  MDL_BIT(MDL_BACKUP_FTWRL2)
+};
+
+
 /**
   Check if request for the metadata lock can be satisfied given its
   current state.
@@ -1587,7 +1681,7 @@ MDL_lock::can_grant_lock(enum_mdl_type type_arg,
         {
 #ifdef WITH_WSREP
           if (wsrep_thd_is_BF(requestor_ctx->get_thd(),false) &&
-              key.mdl_namespace() == MDL_key::GLOBAL)
+              key.mdl_namespace() == MDL_key::BACKUP)
           {
             WSREP_DEBUG("global lock granted for BF: %lu %s",
                         thd_get_thread_id(requestor_ctx->get_thd()),
@@ -1621,7 +1715,7 @@ MDL_lock::can_grant_lock(enum_mdl_type type_arg,
   else
   {
     if (wsrep_thd_is_BF(requestor_ctx->get_thd(), false) &&
-	key.mdl_namespace() == MDL_key::GLOBAL)
+	key.mdl_namespace() == MDL_key::BACKUP)
     {
       WSREP_DEBUG("global lock granted for BF (waiting queue): %lu %s",
                   thd_get_thread_id(requestor_ctx->get_thd()),
@@ -1742,6 +1836,27 @@ bool MDL_ticket::is_incompatible_when_waiting(enum_mdl_type type) const
 }
 
 
+static const LEX_STRING
+*get_mdl_lock_name(MDL_key::enum_mdl_namespace mdl_namespace,
+                   enum_mdl_type type)
+{
+  return mdl_namespace == MDL_key::BACKUP ?
+         &backup_lock_types[type] :
+         &lock_types[type];
+}
+
+
+const LEX_STRING *MDL_ticket::get_type_name() const
+{
+  return get_mdl_lock_name(get_key()->mdl_namespace(), m_type);
+}
+
+const LEX_STRING *MDL_ticket::get_type_name(enum_mdl_type type) const
+{
+  return get_mdl_lock_name(get_key()->mdl_namespace(), type);
+}
+
+
 /**
   Check whether the context already holds a compatible lock ticket
   on an object.
@@ -1775,8 +1890,10 @@ MDL_context::find_ticket(MDL_request *mdl_request,
       if (mdl_request->key.is_equal(&ticket->m_lock->key) &&
           ticket->has_stronger_or_equal_type(mdl_request->type))
       {
-        DBUG_PRINT("info", ("Adding mdl lock %d to %d",
+        DBUG_PRINT("info", ("Adding mdl lock %s to %s",
-                            mdl_request->type, ticket->m_type));
+                            get_mdl_lock_name(mdl_request->key.mdl_namespace(),
+                                              mdl_request->type)->str,
+                            ticket->get_type_name()->str));
         *result_duration= duration;
         return ticket;
       }
@@ -2064,7 +2181,10 @@ MDL_context::acquire_lock(MDL_request *mdl_request, double lock_wait_timeout)
   MDL_ticket *ticket;
   MDL_wait::enum_wait_status wait_status;
   DBUG_ENTER("MDL_context::acquire_lock");
-  DBUG_PRINT("enter", ("lock_type: %d", mdl_request->type));
+  DBUG_PRINT("enter", ("lock_type: %s  timeout: %f",
+                       get_mdl_lock_name(mdl_request->key.mdl_namespace(),
+                                         mdl_request->type)->str,
+                       lock_wait_timeout));
 
   if (try_acquire_lock_impl(mdl_request, &ticket))
     DBUG_RETURN(TRUE);
@@ -2180,6 +2300,7 @@ MDL_context::acquire_lock(MDL_request *mdl_request, double lock_wait_timeout)
     switch (wait_status)
     {
     case MDL_wait::VICTIM:
+      mdl_dbug_print_locks();
       my_error(ER_LOCK_DEADLOCK, MYF(0));
       break;
     case MDL_wait::TIMEOUT:
@@ -2319,7 +2440,9 @@ MDL_context::upgrade_shared_lock(MDL_ticket *mdl_ticket,
   MDL_savepoint mdl_svp= mdl_savepoint();
   bool is_new_ticket;
   DBUG_ENTER("MDL_context::upgrade_shared_lock");
-  DBUG_PRINT("enter",("new_type: %d  lock_wait_timeout: %f", new_type,
+  DBUG_PRINT("enter",("old_type: %s  new_type: %s  lock_wait_timeout: %f",
+                      mdl_ticket->get_type_name()->str,
+                      mdl_ticket->get_type_name(new_type)->str,
                       lock_wait_timeout));
   DEBUG_SYNC(get_thd(), "mdl_upgrade_lock");
 
@@ -3015,31 +3138,12 @@ bool MDL_context::has_explicit_locks()
 }
 
 #ifdef WITH_WSREP
-static
-const char *wsrep_get_mdl_type_name(enum_mdl_type type)
-{
-  switch (type)
-  {
-  case MDL_INTENTION_EXCLUSIVE  : return "intention exclusive";
-  case MDL_SHARED               : return "shared";
-  case MDL_SHARED_HIGH_PRIO     : return "shared high prio";
-  case MDL_SHARED_READ          : return "shared read";
-  case MDL_SHARED_WRITE         : return "shared write";
-  case MDL_SHARED_UPGRADABLE    : return "shared upgradable";
-  case MDL_SHARED_NO_WRITE      : return "shared no write";
-  case MDL_SHARED_NO_READ_WRITE : return "shared no read write";
-  case MDL_EXCLUSIVE            : return "exclusive";
-  default: break;
-  }
-  return "UNKNOWN";
-}
-
 static
 const char *wsrep_get_mdl_namespace_name(MDL_key::enum_mdl_namespace ns)
 {
   switch (ns)
   {
-  case MDL_key::GLOBAL    : return "GLOBAL";
+  case MDL_key::BACKUP    : return "BACKUP";
   case MDL_key::SCHEMA    : return "SCHEMA";
   case MDL_key::TABLE     : return "TABLE";
   case MDL_key::FUNCTION  : return "FUNCTION";
@@ -3047,7 +3151,6 @@ const char *wsrep_get_mdl_namespace_name(MDL_key::enum_mdl_namespace ns)
   case MDL_key::PACKAGE_BODY: return "PACKAGE BODY";
   case MDL_key::TRIGGER   : return "TRIGGER";
   case MDL_key::EVENT     : return "EVENT";
-  case MDL_key::COMMIT    : return "COMMIT";
   case MDL_key::USER_LOCK : return "USER_LOCK";
   default: break;
   }
@@ -3060,10 +3163,41 @@ void MDL_ticket::wsrep_report(bool debug)
 
   const PSI_stage_info *psi_stage= m_lock->key.get_wait_state_name();
   WSREP_DEBUG("MDL ticket: type: %s space: %s db: %s name: %s (%s)",
-              wsrep_get_mdl_type_name(get_type()),
+              get_type_name()->str,
               wsrep_get_mdl_namespace_name(m_lock->key.mdl_namespace()),
               m_lock->key.db_name(),
               m_lock->key.name(),
               psi_stage->m_name);
 }
 #endif /* WITH_WSREP */
+
+
+#ifndef DBUG_OFF
+
+/*
+  Print a list of all locks to DBUG trace to help with debugging
+*/
+
+static int mdl_dbug_print_lock(MDL_ticket *mdl_ticket, void *arg, bool granted)
+{
+  String *tmp= (String*) arg;
+  char buffer[128];
+  MDL_key *mdl_key= mdl_ticket->get_key();
+  size_t length;
+  length= my_snprintf(buffer, sizeof(buffer)-1,
+                      "\nname: %s  db: %.*s  key_name: %.*s (%s)",
+                      mdl_ticket->get_type_name()->str,
+                      (int) mdl_key->db_name_length(), mdl_key->db_name(),
+                      (int) mdl_key->name_length(),    mdl_key->name(),
+                      granted ? "granted" : "waiting");
+  tmp->append(buffer, length);
+  return 0;
+}
+
+void mdl_dbug_print_locks()
+{
+  String tmp;
+  mdl_iterate(mdl_dbug_print_lock, (void*) &tmp);
+  DBUG_PRINT("mdl_locks", ("%s", tmp.c_ptr()));
+}
+#endif /* DBUG_OFF */

sql/mdl.h

@@ -113,7 +113,7 @@ public:
   @sa Comments for MDL_object_lock::can_grant_lock() and
       MDL_scoped_lock::can_grant_lock() for details.
 
-  Scoped locks are GLOBAL READ LOCK, COMMIT and database (or schema) locks.
+  Scoped locks are database (or schema) locks.
   The object locks are for tables, triggers etc.
 */
 
@@ -247,6 +247,33 @@ enum enum_mdl_type {
 };
 
 
+/** Backup locks */
+
+/**
+  Blocks (or is blocked by) statements that intend to modify data. Acquired
+  before commit lock by FLUSH TABLES WITH READ LOCK.
+*/
+#define MDL_BACKUP_FTWRL1 enum_mdl_type(0)
+
+/**
+  Blocks (or is blocked by) commits. Acquired after global read lock by
+  FLUSH TABLES WITH READ LOCK.
+*/
+#define MDL_BACKUP_FTWRL2 enum_mdl_type(1)
+
+/**
+  Must be acquired by statements that intend to modify data.
+*/
+#define MDL_BACKUP_STMT enum_mdl_type(2)
+
+/**
+  Must be acquired during commit.
+*/
+#define MDL_BACKUP_COMMIT enum_mdl_type(3)
+#define MDL_BACKUP_END enum_mdl_type(4)
+
+
+
 /** Duration of metadata lock. */
 
 enum enum_mdl_duration {
@@ -292,10 +319,13 @@ public:
   /**
     Object namespaces.
     Sic: when adding a new member to this enum make sure to
-    update m_namespace_to_wait_state_name array in mdl.cc!
+    update m_namespace_to_wait_state_name array in mdl.cc and
+    metadata_lock_info_lock_name in metadata_lock_info.cc!
 
     Different types of objects exist in different namespaces
+     - SCHEMA is for databases (to protect against DROP DATABASE)
      - TABLE is for tables and views.
+     - BACKUP is for locking DML, DDL and COMMIT's during BACKUP STAGES
      - FUNCTION is for stored functions.
      - PROCEDURE is for stored procedures.
      - TRIGGER is for triggers.
@@ -304,7 +334,7 @@ public:
     it's necessary to have a separate namespace for them since
     MDL_key is also used outside of the MDL subsystem.
   */
-  enum enum_mdl_namespace { GLOBAL=0,
+  enum enum_mdl_namespace { BACKUP=0,
                             SCHEMA,
                             TABLE,
                             FUNCTION,
@@ -312,7 +342,6 @@ public:
                             PACKAGE_BODY,
                             TRIGGER,
                             EVENT,
-                            COMMIT,
                             USER_LOCK,           /* user level locks. */
                             /* This should be the last ! */
                             NAMESPACE_END };
@@ -620,6 +649,8 @@ public:
            m_type == MDL_EXCLUSIVE;
   }
   enum_mdl_type get_type() const { return m_type; }
+  const LEX_STRING *get_type_name() const;
+  const LEX_STRING *get_type_name(enum_mdl_type type) const;
   MDL_lock *get_lock() const { return m_lock; }
   MDL_key *get_key() const;
   void downgrade_lock(enum_mdl_type type);
@@ -1011,6 +1042,13 @@ extern "C" int thd_is_connected(MYSQL_THD thd);
 */
 extern "C" ulong max_write_lock_count;
 
+typedef int (*mdl_iterator_callback)(MDL_ticket *ticket, void *arg,
+                                     bool granted);
 extern MYSQL_PLUGIN_IMPORT
-int mdl_iterate(int (*callback)(MDL_ticket *ticket, void *arg), void *arg);
+int mdl_iterate(mdl_iterator_callback callback, void *arg);
-#endif
+#ifndef DBUG_OFF
+void mdl_dbug_print_locks();
+#else
+  static inline void mdl_dbug_print_locks() {}
+#endif /* DBUG_OFF */
+#endif /* MDL_H */

sql/sp.cc

@@ -1794,8 +1794,8 @@ bool lock_db_routines(THD *thd, const char *db)
   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, "", "",
+  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::BACKUP, "", "",
-                                             MDL_INTENTION_EXCLUSIVE) &&
+                                             MDL_BACKUP_STMT) &&
               thd->mdl_context.is_lock_owner(MDL_key::SCHEMA, db, "",
                                              MDL_EXCLUSIVE));
   DBUG_RETURN(thd->mdl_context.acquire_locks(&mdl_requests,

sql/sp.h

@@ -522,12 +522,11 @@ inline const Sp_handler *Sp_handler::handler(MDL_key::enum_mdl_namespace type)
     return &sp_handler_procedure;
   case MDL_key::PACKAGE_BODY:
     return &sp_handler_package_body;
-  case MDL_key::GLOBAL:
+  case MDL_key::BACKUP:
   case MDL_key::SCHEMA:
   case MDL_key::TABLE:
   case MDL_key::TRIGGER:
   case MDL_key::EVENT:
-  case MDL_key::COMMIT:
   case MDL_key::USER_LOCK:
   case MDL_key::NAMESPACE_END:
     break;

sql/sql_base.cc

@@ -1864,7 +1864,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, Open_table_context *ot_ctx)
       if (thd->global_read_lock.can_acquire_protection())
         DBUG_RETURN(TRUE);
 
-      protection_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
+      protection_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_STMT,
                               MDL_STATEMENT);
 
       /*
@@ -2207,8 +2207,8 @@ TABLE *find_table_for_mdl_upgrade(THD *thd, const char *db,
     cases don't take a global IX lock in order to be compatible with
     global read lock.
   */
-  if (unlikely(!thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
+  if (unlikely(!thd->mdl_context.is_lock_owner(MDL_key::BACKUP, "", "",
-                                               MDL_INTENTION_EXCLUSIVE)))
+                                               MDL_BACKUP_STMT)))
   {
     error= ER_TABLE_NOT_LOCKED_FOR_WRITE;
     goto err_exit;
@@ -3941,7 +3941,7 @@ lock_table_names(THD *thd, const DDL_options_st &options,
     */
     if (thd->global_read_lock.can_acquire_protection())
       DBUG_RETURN(TRUE);
-    global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
+    global_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_STMT,
                         MDL_STATEMENT);
     mdl_requests.push_front(&global_request);
 

sql/sql_class.h

@@ -1953,8 +1953,7 @@ public:
 
   Global_read_lock()
     : m_state(GRL_NONE),
-      m_mdl_global_shared_lock(NULL),
+      m_mdl_global_read_lock(NULL)
-      m_mdl_blocks_commits_lock(NULL)
   {}
 
   bool lock_global_read_lock(THD *thd);
@@ -1978,17 +1977,11 @@ public:
 private:
   enum_grl_state m_state;
   /**
-    In order to acquire the global read lock, the connection must
+    Global read lock is acquired in two steps:
-    acquire shared metadata lock in GLOBAL namespace, to prohibit
+    1. acquire MDL_BACKUP_FTWRL1 in BACKUP namespace to prohibit DDL and DML
-    all DDL.
+    2. upgrade to MDL_BACKUP_FTWRL2 to prohibit commits
   */
-  MDL_ticket *m_mdl_global_shared_lock;
+  MDL_ticket *m_mdl_global_read_lock;
-  /**
-    Also in order to acquire the global read lock, the connection
-    must acquire a shared metadata lock in COMMIT namespace, to
-    prohibit commits.
-  */
-  MDL_ticket *m_mdl_blocks_commits_lock;
 };
 
 

sql/sql_insert.cc

@@ -550,7 +550,7 @@ bool open_and_lock_for_insert_delayed(THD *thd, TABLE_LIST *table_list)
   if (thd->global_read_lock.can_acquire_protection())
     DBUG_RETURN(TRUE);
 
-  protection_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
+  protection_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_STMT,
                           MDL_STATEMENT);
 
   if (thd->mdl_context.acquire_lock(&protection_request,
@@ -2375,8 +2375,8 @@ bool delayed_get_table(THD *thd, MDL_request *grl_protection_request,
       di->table_list.alias.length= di->table_list.table_name.length= di->thd.query_length();
       di->table_list.db= di->thd.db;
       /* We need the tickets so that they can be cloned in handle_delayed_insert */
-      di->grl_protection.init(MDL_key::GLOBAL, "", "",
+      di->grl_protection.init(MDL_key::BACKUP, "", "",
-                              MDL_INTENTION_EXCLUSIVE, MDL_STATEMENT);
+                              MDL_BACKUP_STMT, MDL_STATEMENT);
       di->grl_protection.ticket= grl_protection_request->ticket;
       init_mdl_requests(&di->table_list);
       di->table_list.mdl_request.ticket= table_list->mdl_request.ticket;

sql/sql_reload.cc

@@ -308,8 +308,8 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
             with global read lock.
           */
           if (thd->open_tables &&
-              !thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
+              !thd->mdl_context.is_lock_owner(MDL_key::BACKUP, "", "",
-                                              MDL_INTENTION_EXCLUSIVE))
+                                              MDL_BACKUP_STMT))
           {
             my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0),
                      thd->open_tables->s->table_name.str);

sql/sql_table.cc

@@ -9191,12 +9191,12 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
       }
 
       /*
-        Global intention exclusive lock must have been already acquired when
+        Protection against global read lock must have been acquired when table
-        table to be altered was open, so there is no need to do it here.
+        to be altered was being opened.
       */
-      DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::GLOBAL,
+      DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::BACKUP,
                                                  "", "",
-                                                 MDL_INTENTION_EXCLUSIVE));
+                                                 MDL_BACKUP_STMT));
 
       if (thd->mdl_context.acquire_locks(&mdl_requests,
                                          thd->variables.lock_wait_timeout))

sql/transaction.cc

@@ -963,7 +963,7 @@ bool trans_xa_commit(THD *thd)
 
       We allow FLUSHer to COMMIT; we assume FLUSHer knows what it does.
     */
-    mdl_request.init(MDL_key::COMMIT, "", "", MDL_INTENTION_EXCLUSIVE,
+    mdl_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_COMMIT,
                      MDL_TRANSACTION);
 
     if (thd->mdl_context.acquire_lock(&mdl_request,