Added syntax and implementation for BACKUP STAGE's

Part of MDEV-5336 Implement LOCK FOR BACKUP

- Changed check of Global_only_lock to also include BACKUP lock.
- We store latest MDL_BACKUP_DDL lock in thd->mdl_backup_ticket to be able
  to downgrade lock during copy_data_between_tables()

38 files changed, 3530 insertions, 41 deletions

diff --git a/libmysqld/CMakeLists.txt b/libmysqld/CMakeLists.txt
index 537f40cbbbf..dfb5d4ce86f 100644
--- a/libmysqld/CMakeLists.txt
+++ b/libmysqld/CMakeLists.txt
@@ -118,7 +118,7 @@ SET(SQL_EMBEDDED_SOURCES emb_qcache.cc libmysqld.c lib_sql.cc
            ../sql/ha_sequence.cc ../sql/ha_sequence.h
            ../sql/temporary_tables.cc
            ../sql/session_tracker.cc
-           ../sql/proxy_protocol.cc
+           ../sql/proxy_protocol.cc ../sql/backup.cc
            ../sql/sql_tvc.cc ../sql/sql_tvc.h
            ../sql/opt_split.cc
            ../sql/item_vers.cc
diff --git a/mysql-test/main/backup_aria.result b/mysql-test/main/backup_aria.result
new file mode 100644
index 00000000000..d537711404f
--- /dev/null
+++ b/mysql-test/main/backup_aria.result
@@ -0,0 +1,158 @@
+connect  con1,localhost,root,,;
+SET SESSION lock_wait_timeout = 1;
+#-----------------------------------------------------------------------
+# Single-threaded tests
+#-----------------------------------------------------------------------
+# Show the fate and impact of some SELECT /HANDLER ... READ
+# sliding through the sequence.
+CREATE TABLE t_permanent_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+INSERT INTO t_permanent_aria SET col1 = 1;
+BACKUP STAGE START;
+SELECT COUNT(*) FROM t_permanent_aria;
+COUNT(*)
+1
+HANDLER t_permanent_aria OPEN;
+HANDLER t_permanent_aria READ FIRST;
+col1
+1
+HANDLER t_permanent_aria CLOSE;
+BACKUP STAGE FLUSH;
+SELECT COUNT(*) FROM t_permanent_aria;
+COUNT(*)
+1
+HANDLER t_permanent_aria OPEN;
+HANDLER t_permanent_aria READ FIRST;
+col1
+1
+HANDLER t_permanent_aria CLOSE;
+BACKUP STAGE BLOCK_DDL;
+SELECT COUNT(*) FROM t_permanent_aria;
+COUNT(*)
+1
+HANDLER t_permanent_aria OPEN;
+HANDLER t_permanent_aria READ FIRST;
+col1
+1
+HANDLER t_permanent_aria CLOSE;
+BACKUP STAGE BLOCK_COMMIT;
+SELECT COUNT(*) FROM t_permanent_aria;
+COUNT(*)
+1
+HANDLER t_permanent_aria OPEN;
+HANDLER t_permanent_aria READ FIRST;
+col1
+1
+HANDLER t_permanent_aria CLOSE;
+BACKUP STAGE END;
+# In case the backup lock is taken by the current connection than
+# - DML modifying some permanent table is not allowed
+BACKUP STAGE START;
+SET AUTOCOMMIT = 0;
+INSERT INTO t_permanent_aria SET col1 = 1;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+SET AUTOCOMMIT = 1;
+INSERT INTO t_permanent_aria SET col1 = 1;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+# - DDL creating or renaming a permanent table or a procedure is not
+#   allowed.
+#   The latter tries to modify a permanent system table.
+CREATE TABLE throw_away (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+RENAME TABLE t_permanent_aria To throw_away;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+# - DDL creating a temporary table is allowed.
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+# - DML modifying that temporary table is allowed.
+INSERT INTO t_temporary_aria SET col1 = 1;
+SELECT COUNT(*) FROM t_temporary_aria;
+COUNT(*)
+1
+BACKUP STAGE END;
+# Show the fate and impact of some auto committed INSERT into temporary
+# table sliding through the sequence.
+SET AUTOCOMMIT = 1;
+BACKUP STAGE START;
+INSERT INTO t_temporary_aria SET col1 = 1;
+BACKUP STAGE FLUSH;
+INSERT INTO t_temporary_aria SET col1 = 1;
+BACKUP STAGE BLOCK_DDL;
+INSERT INTO t_temporary_aria SET col1 = 1;
+BACKUP STAGE BLOCK_COMMIT;
+INSERT INTO t_temporary_aria SET col1 = 1;
+BACKUP STAGE END;
+SELECT COUNT(*) FROM t_temporary_aria;
+COUNT(*)
+5
+# Show the fate and impact of some DROP/CREATE TEMPORARY TABLE sliding
+# through the sequence.
+SET AUTOCOMMIT = 1;
+BACKUP STAGE START;
+DROP TEMPORARY TABLE t_temporary_aria;
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+BACKUP STAGE FLUSH;
+DROP TEMPORARY TABLE t_temporary_aria;
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+BACKUP STAGE BLOCK_DDL;
+DROP TEMPORARY TABLE t_temporary_aria;
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+BACKUP STAGE BLOCK_COMMIT;
+DROP TEMPORARY TABLE t_temporary_aria;
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+BACKUP STAGE END;
+# Show that even more DDL on the temporary table is allowed.
+BACKUP STAGE START;
+TRUNCATE t_temporary_aria;
+ALTER TABLE t_temporary_aria ADD COLUMN col2 INT;
+ALTER TABLE t_temporary_aria ADD KEY idx(col2);
+BACKUP STAGE END;
+DROP TABLE t_permanent_aria;
+#-----------------------------------------------------------------------
+# Show that non transactional tables locks with BACKUP STAGE FLUSH
+#-----------------------------------------------------------------------
+set session lock_wait_timeout=default;
+create table t1 (a int) engine=aria transactional=0;
+insert into t1 values (1), (2);
+connection con1;
+backup stage start;
+backup stage flush;
+connection default;
+select * from t1;
+a
+1
+2
+SET STATEMENT lock_wait_timeout=0 FOR INSERT INTO t1 values (3);
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+INSERT INTO t1 values (4);
+connection con1;
+backup stage end;
+connection default;
+select * from t1;
+a
+1
+2
+4
+drop table t1;
+#-----------------------------------------------------------------------
+# Show that transactional tables doesn't lock with BACKUP STAGE FLUSH
+#-----------------------------------------------------------------------
+set session lock_wait_timeout=default;
+create table t1 (a int) engine=aria transactional=1 page_checksum=1;
+insert into t1 values (1), (2);
+connection con1;
+backup stage start;
+backup stage flush;
+connection default;
+INSERT INTO t1 values (4);
+connection con1;
+backup stage end;
+connection default;
+select * from t1;
+a
+1
+2
+4
+drop table t1;
+#
+# Cleanup
+#
+disconnect con1;
diff --git a/mysql-test/main/backup_aria.test b/mysql-test/main/backup_aria.test
new file mode 100644
index 00000000000..7b741b829a1
--- /dev/null
+++ b/mysql-test/main/backup_aria.test
@@ -0,0 +1,157 @@
+########################################################################
+# Tests for Implement LOCK FOR BACKUP (MDEV-5336)
+########################################################################
+# Check a non transactional table per ENGINE = Aria TRANSACTIONAL = 0.
+#
+
+--source include/not_embedded.inc
+# As non transactional engine we use Aria with TRANSACTIONAL = 0
+--source include/have_aria.inc
+
+# Following connections are used in a few of the following tests
+connect (con1,localhost,root,,);
+
+SET SESSION lock_wait_timeout = 1;
+
+--echo #-----------------------------------------------------------------------
+--echo # Single-threaded tests
+--echo #-----------------------------------------------------------------------
+
+--echo # Show the fate and impact of some SELECT /HANDLER ... READ
+--echo # sliding through the sequence.
+CREATE TABLE t_permanent_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+INSERT INTO t_permanent_aria SET col1 = 1;
+BACKUP STAGE START;
+SELECT COUNT(*) FROM t_permanent_aria;
+HANDLER t_permanent_aria OPEN;
+HANDLER t_permanent_aria READ FIRST;
+HANDLER t_permanent_aria CLOSE;
+BACKUP STAGE FLUSH;
+SELECT COUNT(*) FROM t_permanent_aria;
+HANDLER t_permanent_aria OPEN;
+HANDLER t_permanent_aria READ FIRST;
+HANDLER t_permanent_aria CLOSE;
+BACKUP STAGE BLOCK_DDL;
+SELECT COUNT(*) FROM t_permanent_aria;
+HANDLER t_permanent_aria OPEN;
+HANDLER t_permanent_aria READ FIRST;
+HANDLER t_permanent_aria CLOSE;
+BACKUP STAGE BLOCK_COMMIT;
+SELECT COUNT(*) FROM t_permanent_aria;
+HANDLER t_permanent_aria OPEN;
+HANDLER t_permanent_aria READ FIRST;
+HANDLER t_permanent_aria CLOSE;
+BACKUP STAGE END;
+
+--echo # In case the backup lock is taken by the current connection than
+--echo # - DML modifying some permanent table is not allowed
+BACKUP STAGE START;
+SET AUTOCOMMIT = 0;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+INSERT INTO t_permanent_aria SET col1 = 1;
+SET AUTOCOMMIT = 1;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+INSERT INTO t_permanent_aria SET col1 = 1;
+
+--echo # - DDL creating or renaming a permanent table or a procedure is not
+--echo #   allowed.
+--echo #   The latter tries to modify a permanent system table.
+
+--error ER_BACKUP_LOCK_IS_ACTIVE
+CREATE TABLE throw_away (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+RENAME TABLE t_permanent_aria To throw_away;
+--echo # - DDL creating a temporary table is allowed.
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+--echo # - DML modifying that temporary table is allowed.
+INSERT INTO t_temporary_aria SET col1 = 1;
+SELECT COUNT(*) FROM t_temporary_aria;
+BACKUP STAGE END;
+
+--echo # Show the fate and impact of some auto committed INSERT into temporary
+--echo # table sliding through the sequence.
+SET AUTOCOMMIT = 1;
+BACKUP STAGE START;
+INSERT INTO t_temporary_aria SET col1 = 1;
+BACKUP STAGE FLUSH;
+INSERT INTO t_temporary_aria SET col1 = 1;
+BACKUP STAGE BLOCK_DDL;
+INSERT INTO t_temporary_aria SET col1 = 1;
+BACKUP STAGE BLOCK_COMMIT;
+INSERT INTO t_temporary_aria SET col1 = 1;
+BACKUP STAGE END;
+SELECT COUNT(*) FROM t_temporary_aria;
+
+--echo # Show the fate and impact of some DROP/CREATE TEMPORARY TABLE sliding
+--echo # through the sequence.
+
+SET AUTOCOMMIT = 1;
+BACKUP STAGE START;
+DROP TEMPORARY TABLE t_temporary_aria;
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+BACKUP STAGE FLUSH;
+DROP TEMPORARY TABLE t_temporary_aria;
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+BACKUP STAGE BLOCK_DDL;
+DROP TEMPORARY TABLE t_temporary_aria;
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+BACKUP STAGE BLOCK_COMMIT;
+DROP TEMPORARY TABLE t_temporary_aria;
+CREATE TEMPORARY TABLE t_temporary_aria (col1 INT) ENGINE = Aria TRANSACTIONAL = 0;
+BACKUP STAGE END;
+--echo # Show that even more DDL on the temporary table is allowed.
+BACKUP STAGE START;
+TRUNCATE t_temporary_aria;
+ALTER TABLE t_temporary_aria ADD COLUMN col2 INT;
+ALTER TABLE t_temporary_aria ADD KEY idx(col2);
+BACKUP STAGE END;
+
+DROP TABLE t_permanent_aria;
+
+--echo #-----------------------------------------------------------------------
+--echo # Show that non transactional tables locks with BACKUP STAGE FLUSH
+--echo #-----------------------------------------------------------------------
+
+set session lock_wait_timeout=default;
+create table t1 (a int) engine=aria transactional=0;
+insert into t1 values (1), (2);
+
+connection con1;
+backup stage start;
+backup stage flush;
+connection default;
+select * from t1;
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR INSERT INTO t1 values (3);
+--send INSERT INTO t1 values (4)
+connection con1;
+backup stage end;
+connection default;
+--reap # send
+select * from t1;
+drop table t1;
+
+--echo #-----------------------------------------------------------------------
+--echo # Show that transactional tables doesn't lock with BACKUP STAGE FLUSH
+--echo #-----------------------------------------------------------------------
+
+set session lock_wait_timeout=default;
+create table t1 (a int) engine=aria transactional=1 page_checksum=1;
+insert into t1 values (1), (2);
+
+connection con1;
+backup stage start;
+backup stage flush;
+connection default;
+INSERT INTO t1 values (4);
+connection con1;
+backup stage end;
+connection default;
+select * from t1;
+drop table t1;
+
+--echo #
+--echo # Cleanup
+--echo #
+
+disconnect con1;
diff --git a/mysql-test/main/backup_interaction.result b/mysql-test/main/backup_interaction.result
new file mode 100644
index 00000000000..20ba8fa0811
--- /dev/null
+++ b/mysql-test/main/backup_interaction.result
@@ -0,0 +1,520 @@
+#
+# Check backup and FTWRL
+#
+flush tables with read lock;
+backup stage start;
+ERROR HY000: Can't execute the query because you have a conflicting read lock
+unlock tables;
+backup stage start;
+flush tables with read lock;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+backup stage end;
+#
+# Check backup and FLUSH TABLES
+#
+flush tables;
+backup stage start;
+flush tables;
+backup stage end;
+#
+# Check BACKUP STAGE under lock tables
+#
+create table t1 (a int);
+lock table t1 write;
+backup stage start;
+ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction
+backup stage end;
+ERROR HY000: You must start backup with "BACKUP STAGE START"
+unlock tables;
+lock table t1 read;
+backup stage start;
+ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction
+backup stage end;
+ERROR HY000: You must start backup with "BACKUP STAGE START"
+unlock tables;
+#
+# Check lock tables under BACKUP STAGE
+#
+backup stage start;
+unlock tables;
+select lock_mode from information_schema.metadata_lock_info;
+lock_mode
+MDL_BACKUP_START
+lock table t1 write;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+lock table t1 read;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+unlock tables;
+backup stage end;
+drop table t1;
+#
+# Check setting readonly under BACKUP STAGE
+#
+backup stage start;
+set @@global.read_only=1;
+ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction
+backup stage end;
+# also make sure going back from read-only mode is not allowed
+set @@global.read_only=1;
+backup stage start;
+set @@global.read_only=0;
+ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction
+backup stage end;
+set @@global.read_only=0;
+#
+# Check BACKUP STAGE under read_only
+#
+set @@global.read_only=1;
+backup stage start;
+backup stage end;
+set @@global.read_only=0;
+#
+# Check that we can't create tables during backup
+#
+backup stage start;
+create table t1 (a int);
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+backup stage end;
+# also make sure we can't write to a table during backup
+create table t1(a INT);
+backup stage start;
+insert into t1 values(1);
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+insert delayed into t1 values(1);
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+update t1 set a=1;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+delete from t1;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+truncate table t1;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+drop table t1;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+backup stage end;
+drop table t1;
+#
+# BACKUP STAGE performs implicit commits
+#
+create table t1(a int) engine=InnoDB;
+begin;
+insert into t1 values(1);
+select lock_mode from information_schema.metadata_lock_info;
+lock_mode
+MDL_SHARED_WRITE
+backup stage start;
+select lock_mode from information_schema.metadata_lock_info;
+lock_mode
+MDL_BACKUP_START
+backup stage block_commit;
+commit;
+backup stage end;
+drop table t1;
+# Ensure that BACKUP STAGE ... does AUTOCOMMIT like most DDL.
+# Sideeffect:
+# Show the impact of not yet committed INSERT before sequence start
+# and ROLLBACK sliding through the sequence.
+CREATE TABLE t1 (col1 INT) ENGINE = InnoDB;
+SET AUTOCOMMIT = 0;
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+ROLLBACK;
+BACKUP STAGE END;
+#----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+ROLLBACK;
+BACKUP STAGE END;
+#----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+ROLLBACK;
+BACKUP STAGE END;
+#----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+ROLLBACK;
+BACKUP STAGE END;
+#----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+ROLLBACK;
+SELECT COUNT(*) = 5 AS expect_1 FROM t1;
+expect_1
+1
+# Show the impact of not yet committed INSERT before sequence start
+# and a COMMIT sliding through the sequence.
+SET AUTOCOMMIT = 0;
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+COMMIT;
+BACKUP STAGE END;
+#----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+COMMIT;
+BACKUP STAGE END;
+#----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+COMMIT;
+BACKUP STAGE END;
+#----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+COMMIT;
+BACKUP STAGE END;
+#----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+COMMIT;
+SELECT COUNT(*) = 10 AS expect_1 FROM t1;
+expect_1
+1
+DELETE FROM t1;
+COMMIT;
+drop table t1;
+#
+# CHECK: RO transaction under BACKUP STAGE is a potential deadlock
+# OTOH we most probably allow them under FTWRL as well
+#
+CREATE TABLE t1 (col1 INT) ENGINE = InnoDB;
+insert into t1 values (1);
+backup stage start;
+backup stage block_commit;
+begin;
+select * from t1;
+col1
+1
+select lock_mode from information_schema.metadata_lock_info;
+lock_mode
+MDL_BACKUP_WAIT_COMMIT
+MDL_SHARED_READ
+backup stage end;
+select lock_mode from information_schema.metadata_lock_info;
+lock_mode
+drop table t1;
+#
+# Check that handler are closed by backup stage block_ddl
+#
+create table t1 (a int, key a (a));
+insert into t1 (a) values (1), (2), (3), (4), (5);
+handler t1 open;
+handler t1 read a prev;
+a
+5
+backup stage start;
+handler t1 read a prev;
+a
+4
+backup stage flush;
+backup stage block_ddl;
+handler t1 read a prev;
+a
+5
+backup stage block_commit;
+handler t1 read a prev;
+a
+4
+backup stage end;
+handler t1 close;
+drop table t1;
+# Show the fate and impact of some SELECT /HANDLER ... READ
+# sliding through the sequence.
+CREATE TABLE t1_innodb (col1 INT) ENGINE = InnoDB;
+INSERT INTO t1_innodb values (1),(2),(3);
+COMMIT;
+CREATE TABLE t1_myisam (col1 INT) ENGINE = MyISAM;
+INSERT INTO t1_myisam values (1),(2),(3);
+BACKUP STAGE START;
+SELECT COUNT(*) FROM t1_innodb;
+COUNT(*)
+3
+SELECT * FROM t1_innodb;
+col1
+1
+2
+3
+HANDLER t1_innodb OPEN;
+HANDLER t1_innodb READ FIRST;
+col1
+1
+HANDLER t1_innodb CLOSE;
+SELECT COUNT(*) FROM t1_myisam;
+COUNT(*)
+3
+HANDLER t1_myisam OPEN;
+HANDLER t1_myisam READ FIRST;
+col1
+1
+HANDLER t1_myisam CLOSE;
+BACKUP STAGE FLUSH;
+SELECT COUNT(*) FROM t1_innodb;
+COUNT(*)
+3
+HANDLER t1_innodb OPEN;
+HANDLER t1_innodb READ FIRST;
+col1
+1
+HANDLER t1_innodb CLOSE;
+SELECT COUNT(*) FROM t1_myisam;
+COUNT(*)
+3
+HANDLER t1_myisam OPEN;
+HANDLER t1_myisam READ FIRST;
+col1
+1
+HANDLER t1_myisam CLOSE;
+BACKUP STAGE BLOCK_DDL;
+SELECT COUNT(*) FROM t1_innodb;
+COUNT(*)
+3
+HANDLER t1_innodb OPEN;
+HANDLER t1_innodb READ FIRST;
+col1
+1
+HANDLER t1_innodb CLOSE;
+SELECT COUNT(*) FROM t1_myisam;
+COUNT(*)
+3
+HANDLER t1_myisam OPEN;
+HANDLER t1_myisam READ FIRST;
+col1
+1
+HANDLER t1_myisam CLOSE;
+BACKUP STAGE BLOCK_COMMIT;
+SELECT COUNT(*) FROM t1_innodb;
+COUNT(*)
+3
+HANDLER t1_innodb OPEN;
+HANDLER t1_innodb READ FIRST;
+col1
+1
+HANDLER t1_innodb CLOSE;
+SELECT COUNT(*) FROM t1_myisam;
+COUNT(*)
+3
+HANDLER t1_myisam OPEN;
+HANDLER t1_myisam READ FIRST;
+col1
+1
+HANDLER t1_myisam CLOSE;
+BACKUP STAGE END;
+drop table t1_innodb,t1_myisam;
+# Show the fate and impact of some SET GLOBAL tx_read_only = 1/0
+# sliding through the sequence.
+BACKUP STAGE START;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+BACKUP STAGE FLUSH;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+BACKUP STAGE BLOCK_DDL;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+BACKUP STAGE BLOCK_COMMIT;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+BACKUP STAGE END;
+# Show the fate and impact of some SET SESSION sql_log_bin = 0/1
+# sliding through the sequence.
+COMMIT;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE START;
+SET SESSION sql_log_bin = 0;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE FLUSH;
+SET SESSION sql_log_bin = 0;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE BLOCK_DDL;
+SET SESSION sql_log_bin = 0;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE BLOCK_COMMIT;
+SET SESSION sql_log_bin = 0;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE END;
+#----
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE START;
+SET SESSION sql_log_bin = 1;
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE FLUSH;
+SET SESSION sql_log_bin = 1;
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE BLOCK_DDL;
+SET SESSION sql_log_bin = 1;
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE BLOCK_COMMIT;
+SET SESSION sql_log_bin = 1;
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE END;
+SET SESSION sql_log_bin = 1;
+#-----------------------------------------------------------------------
+# BACKUP STAGE statements are not allowed in stored routines
+#-----------------------------------------------------------------------
+CREATE TABLE t1 (col1 INT);
+CREATE PROCEDURE p1()
+BEGIN
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+END|
+ERROR 0A000: BACKUP STAGE is not allowed in stored procedures
+CREATE FUNCTION f1 (s CHAR(20)) RETURNS INT DETERMINISTIC
+BEGIN
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+RETURN 1;
+END|
+ERROR 0A000: BACKUP STAGE is not allowed in stored procedures
+CREATE TRIGGER tr1 BEFORE INSERT ON t1 FOR EACH ROW
+BEGIN
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+END|
+ERROR 0A000: BACKUP STAGE is not allowed in stored procedures
+DROP TABLE t1;
+#-----------------------------------------------------------------------
+# Check BACKUP status variables
+#-----------------------------------------------------------------------
+SET SESSION lock_wait_timeout = 1;
+FLUSH STATUS;
+# Show how the status variable 'Com_backup' changes after BACKUP STAGE ..
+SHOW STATUS LIKE 'Com_backup';
+Variable_name	Value
+Com_backup	0
+BACKUP STAGE START;
+SHOW STATUS LIKE 'Com_backup';
+Variable_name	Value
+Com_backup	1
+BACKUP STAGE START;
+ERROR HY000: Backup stage 'START' is same or before current backup stage 'START'
+SHOW STATUS LIKE 'Com_backup';
+Variable_name	Value
+Com_backup	2
+BACKUP STAGE FLUSH;
+SHOW STATUS LIKE 'Com_backup';
+Variable_name	Value
+Com_backup	3
+BACKUP STAGE BLOCK_DDL;
+SHOW STATUS LIKE 'Com_backup';
+Variable_name	Value
+Com_backup	4
+BACKUP STAGE BLOCK_COMMIT;
+SHOW STATUS LIKE 'Com_backup';
+Variable_name	Value
+Com_backup	5
+BACKUP STAGE END;
+# In case the backup lock is taken by the current connection than
+# - DML modifying some permanent table is not allowed
+CREATE TABLE t1_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t1_myisam (col1 INT) ENGINE = MyISAM;
+BACKUP STAGE START;
+SET AUTOCOMMIT = 0;
+INSERT INTO t1_innodb SET col1 = 1;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+SET AUTOCOMMIT = 1;
+INSERT INTO t1_innodb SET col1 = 1;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+INSERT INTO t1_myisam SET col1 = 1;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+# - DDL creating or renaming a permanent table or a procedure etc.
+#   is not allowed.                                                               
+CREATE TABLE throw_away (col1 INT) ENGINE = InnoDB;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+RENAME TABLE t1_innodb To throw_away;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+CREATE PROCEDURE p1() SELECT 13;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+CREATE PROCEDURE p1() SELECT 13;
+ERROR HY000: Can't execute the command as you have a BACKUP STAGE active
+BACKUP STAGE END;
+DROP TABLE t1_innodb;
+DROP TABLE t1_myisam;
+#
+# Creating and modifying TEMPORARY TABLES are allowed
+#
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_DDL;
+CREATE TEMPORARY TABLE tmp (col1 INT);
+DROP TEMPORARY TABLE tmp;
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TEMPORARY TABLE t_temporary_myisam (col1 INT) ENGINE = MyISAM;
+# - DML modifying that temporary table is allowed.
+INSERT INTO t_temporary_innodb SET col1 = 1;
+SELECT COUNT(*) FROM t_temporary_innodb;
+COUNT(*)
+1
+INSERT INTO t_temporary_myisam SET col1 = 1;
+SELECT COUNT(*) FROM t_temporary_myisam;
+COUNT(*)
+1
+BACKUP STAGE END;
+# Show the fate and impact of some auto committed INSERT into temporary
+# table sliding through the sequence.
+SET AUTOCOMMIT = 1;
+BACKUP STAGE START;
+INSERT INTO t_temporary_innodb SET col1 = 1;
+INSERT INTO t_temporary_myisam SET col1 = 1;
+BACKUP STAGE FLUSH;
+INSERT INTO t_temporary_innodb SET col1 = 1;
+INSERT INTO t_temporary_myisam SET col1 = 1;
+BACKUP STAGE BLOCK_DDL;
+INSERT INTO t_temporary_innodb SET col1 = 1;
+INSERT INTO t_temporary_myisam SET col1 = 1;
+BACKUP STAGE BLOCK_COMMIT;
+INSERT INTO t_temporary_innodb SET col1 = 1;
+INSERT INTO t_temporary_myisam SET col1 = 1;
+BACKUP STAGE END;
+SELECT COUNT(*) FROM t_temporary_innodb;
+COUNT(*)
+5
+# Show the fate and impact of some DROP/CREATE TEMPORARY TABLE sliding
+# through the sequence.
+SET AUTOCOMMIT = 1;
+BACKUP STAGE START;
+DROP TEMPORARY TABLE t_temporary_innodb;
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+BACKUP STAGE FLUSH;
+DROP TEMPORARY TABLE t_temporary_innodb;
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+BACKUP STAGE BLOCK_DDL;
+DROP TEMPORARY TABLE t_temporary_innodb;
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+BACKUP STAGE BLOCK_COMMIT;
+DROP TEMPORARY TABLE t_temporary_innodb;
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+BACKUP STAGE END;
+# Show that even more DDL on the temporary table is allowed.
+BACKUP STAGE START;
+TRUNCATE t_temporary_innodb;
+ALTER TABLE t_temporary_innodb ADD COLUMN col2 INT;
+ALTER TABLE t_temporary_innodb ADD KEY idx(col2);
+BACKUP STAGE END;
diff --git a/mysql-test/main/backup_interaction.test b/mysql-test/main/backup_interaction.test
new file mode 100644
index 00000000000..f5362417c98
--- /dev/null
+++ b/mysql-test/main/backup_interaction.test
@@ -0,0 +1,503 @@
+########################################################################
+# Tests how BACKUP STAGE interacts with other commands (MDEV-5336)
+########################################################################
+
+--source include/have_innodb.inc
+--source include/have_metadata_lock_info.inc
+
+--echo #
+--echo # Check backup and FTWRL
+--echo #
+
+flush tables with read lock;
+--error ER_CANT_UPDATE_WITH_READLOCK
+backup stage start;
+unlock tables;
+backup stage start;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+flush tables with read lock;
+backup stage end;
+
+--echo #
+--echo # Check backup and FLUSH TABLES
+--echo #
+
+flush tables;
+backup stage start;
+flush tables;
+backup stage end;
+
+--echo #
+--echo # Check BACKUP STAGE under lock tables
+--echo #
+
+create table t1 (a int);
+lock table t1 write;
+--error ER_LOCK_OR_ACTIVE_TRANSACTION
+backup stage start;
+--error ER_BACKUP_NOT_RUNNING
+backup stage end;
+unlock tables;
+
+lock table t1 read;
+--error ER_LOCK_OR_ACTIVE_TRANSACTION
+backup stage start;
+--error ER_BACKUP_NOT_RUNNING
+backup stage end;
+unlock tables;
+
+
+--echo #
+--echo # Check lock tables under BACKUP STAGE
+--echo #
+backup stage start;
+unlock tables;
+select lock_mode from information_schema.metadata_lock_info;
+
+--error ER_BACKUP_LOCK_IS_ACTIVE
+lock table t1 write;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+lock table t1 read;
+unlock tables;
+backup stage end;
+drop table t1;
+
+
+--echo #
+--echo # Check setting readonly under BACKUP STAGE
+--echo #
+backup stage start;
+--error ER_LOCK_OR_ACTIVE_TRANSACTION
+set @@global.read_only=1;
+backup stage end;
+
+--echo # also make sure going back from read-only mode is not allowed
+set @@global.read_only=1;
+backup stage start;
+--error ER_LOCK_OR_ACTIVE_TRANSACTION
+set @@global.read_only=0;
+backup stage end;
+set @@global.read_only=0;
+
+--echo #
+--echo # Check BACKUP STAGE under read_only
+--echo #
+
+set @@global.read_only=1;
+backup stage start;
+backup stage end;
+set @@global.read_only=0;
+
+--echo #
+--echo # Check that we can't create tables during backup
+--echo #
+
+backup stage start;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+create table t1 (a int);
+backup stage end;
+
+--echo # also make sure we can't write to a table during backup
+create table t1(a INT);
+backup stage start;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+insert into t1 values(1);
+--error ER_BACKUP_LOCK_IS_ACTIVE
+insert delayed into t1 values(1);
+--error ER_BACKUP_LOCK_IS_ACTIVE
+update t1 set a=1;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+delete from t1;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+truncate table t1;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+drop table t1;
+backup stage end;
+drop table t1;
+
+--echo #
+--echo # BACKUP STAGE performs implicit commits
+--echo #
+create table t1(a int) engine=InnoDB;
+begin;
+insert into t1 values(1);
+select lock_mode from information_schema.metadata_lock_info;
+backup stage start;
+select lock_mode from information_schema.metadata_lock_info;
+backup stage block_commit;
+commit;
+backup stage end;
+drop table t1;
+
+--echo # Ensure that BACKUP STAGE ... does AUTOCOMMIT like most DDL.
+--echo # Sideeffect:
+--echo # Show the impact of not yet committed INSERT before sequence start
+--echo # and ROLLBACK sliding through the sequence.
+
+CREATE TABLE t1 (col1 INT) ENGINE = InnoDB;
+SET AUTOCOMMIT = 0;
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+ROLLBACK;
+BACKUP STAGE END;
+--echo #----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+ROLLBACK;
+BACKUP STAGE END;
+--echo #----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+ROLLBACK;
+BACKUP STAGE END;
+--echo #----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+ROLLBACK;
+BACKUP STAGE END;
+--echo #----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+ROLLBACK;
+SELECT COUNT(*) = 5 AS expect_1 FROM t1;
+
+--echo # Show the impact of not yet committed INSERT before sequence start
+--echo # and a COMMIT sliding through the sequence.
+
+SET AUTOCOMMIT = 0;
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+COMMIT;
+BACKUP STAGE END;
+--echo #----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+COMMIT;
+BACKUP STAGE END;
+--echo #----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+COMMIT;
+BACKUP STAGE END;
+--echo #----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+COMMIT;
+BACKUP STAGE END;
+--echo #----
+INSERT INTO t1 SET col1 = 1;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+COMMIT;
+SELECT COUNT(*) = 10 AS expect_1 FROM t1;
+DELETE FROM t1;
+COMMIT;
+drop table t1;
+
+--echo #
+--echo # CHECK: RO transaction under BACKUP STAGE is a potential deadlock
+--echo # OTOH we most probably allow them under FTWRL as well
+--echo #
+
+CREATE TABLE t1 (col1 INT) ENGINE = InnoDB;
+insert into t1 values (1);
+backup stage start;
+backup stage block_commit;
+begin;
+select * from t1;
+select lock_mode from information_schema.metadata_lock_info;
+backup stage end;
+select lock_mode from information_schema.metadata_lock_info;
+drop table t1;
+
+--echo #
+--echo # Check that handler are closed by backup stage block_ddl
+--echo #
+
+create table t1 (a int, key a (a));
+insert into t1 (a) values (1), (2), (3), (4), (5);
+handler t1 open;
+handler t1 read a prev;
+backup stage start;
+handler t1 read a prev;
+backup stage flush;
+backup stage block_ddl;
+handler t1 read a prev;
+backup stage block_commit;
+handler t1 read a prev;
+backup stage end;
+handler t1 close;
+drop table t1;
+
+--echo # Show the fate and impact of some SELECT /HANDLER ... READ
+--echo # sliding through the sequence.
+
+CREATE TABLE t1_innodb (col1 INT) ENGINE = InnoDB;
+INSERT INTO t1_innodb values (1),(2),(3);
+COMMIT;
+CREATE TABLE t1_myisam (col1 INT) ENGINE = MyISAM;
+INSERT INTO t1_myisam values (1),(2),(3);
+BACKUP STAGE START;
+SELECT COUNT(*) FROM t1_innodb;
+SELECT * FROM t1_innodb;
+HANDLER t1_innodb OPEN;
+HANDLER t1_innodb READ FIRST;
+HANDLER t1_innodb CLOSE;
+SELECT COUNT(*) FROM t1_myisam;
+HANDLER t1_myisam OPEN;
+HANDLER t1_myisam READ FIRST;
+HANDLER t1_myisam CLOSE;
+BACKUP STAGE FLUSH;
+SELECT COUNT(*) FROM t1_innodb;
+HANDLER t1_innodb OPEN;
+HANDLER t1_innodb READ FIRST;
+HANDLER t1_innodb CLOSE;
+SELECT COUNT(*) FROM t1_myisam;
+HANDLER t1_myisam OPEN;
+HANDLER t1_myisam READ FIRST;
+HANDLER t1_myisam CLOSE;
+BACKUP STAGE BLOCK_DDL;
+SELECT COUNT(*) FROM t1_innodb;
+HANDLER t1_innodb OPEN;
+HANDLER t1_innodb READ FIRST;
+HANDLER t1_innodb CLOSE;
+SELECT COUNT(*) FROM t1_myisam;
+HANDLER t1_myisam OPEN;
+HANDLER t1_myisam READ FIRST;
+HANDLER t1_myisam CLOSE;
+BACKUP STAGE BLOCK_COMMIT;
+SELECT COUNT(*) FROM t1_innodb;
+HANDLER t1_innodb OPEN;
+HANDLER t1_innodb READ FIRST;
+HANDLER t1_innodb CLOSE;
+SELECT COUNT(*) FROM t1_myisam;
+HANDLER t1_myisam OPEN;
+HANDLER t1_myisam READ FIRST;
+HANDLER t1_myisam CLOSE;
+BACKUP STAGE END;
+drop table t1_innodb,t1_myisam;
+
+--echo # Show the fate and impact of some SET GLOBAL tx_read_only = 1/0
+--echo # sliding through the sequence.
+
+BACKUP STAGE START;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+BACKUP STAGE FLUSH;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+BACKUP STAGE BLOCK_DDL;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+BACKUP STAGE BLOCK_COMMIT;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+BACKUP STAGE END;
+
+--echo # Show the fate and impact of some SET SESSION sql_log_bin = 0/1
+--echo # sliding through the sequence.
+COMMIT;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE START;
+SET SESSION sql_log_bin = 0;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE FLUSH;
+SET SESSION sql_log_bin = 0;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE BLOCK_DDL;
+SET SESSION sql_log_bin = 0;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE BLOCK_COMMIT;
+SET SESSION sql_log_bin = 0;
+SET SESSION sql_log_bin = 1;
+BACKUP STAGE END;
+
+--echo #----
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE START;
+SET SESSION sql_log_bin = 1;
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE FLUSH;
+SET SESSION sql_log_bin = 1;
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE BLOCK_DDL;
+SET SESSION sql_log_bin = 1;
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE BLOCK_COMMIT;
+SET SESSION sql_log_bin = 1;
+SET SESSION sql_log_bin = 0;
+BACKUP STAGE END;
+SET SESSION sql_log_bin = 1;
+
+--echo #-----------------------------------------------------------------------
+--echo # BACKUP STAGE statements are not allowed in stored routines
+--echo #-----------------------------------------------------------------------
+
+CREATE TABLE t1 (col1 INT);
+
+delimiter |;
+--error ER_SP_BADSTATEMENT
+CREATE PROCEDURE p1()
+BEGIN
+  BACKUP STAGE START;
+  BACKUP STAGE FLUSH;
+  BACKUP STAGE BLOCK_DDL;
+  BACKUP STAGE BLOCK_COMMIT;
+  BACKUP STAGE END;
+END|
+
+--error ER_SP_BADSTATEMENT
+CREATE FUNCTION f1 (s CHAR(20)) RETURNS INT DETERMINISTIC
+BEGIN
+  BACKUP STAGE START;
+  BACKUP STAGE FLUSH;
+  BACKUP STAGE BLOCK_DDL;
+  BACKUP STAGE BLOCK_COMMIT;
+  BACKUP STAGE END;
+  RETURN 1;
+END|
+
+--error ER_SP_BADSTATEMENT
+CREATE TRIGGER tr1 BEFORE INSERT ON t1 FOR EACH ROW
+BEGIN
+    BACKUP STAGE START;
+    BACKUP STAGE FLUSH;
+    BACKUP STAGE BLOCK_DDL;
+    BACKUP STAGE BLOCK_COMMIT;
+    BACKUP STAGE END;
+END|
+
+delimiter ;|
+DROP TABLE t1;
+
+--echo #-----------------------------------------------------------------------
+--echo # Check BACKUP status variables
+--echo #-----------------------------------------------------------------------
+
+SET SESSION lock_wait_timeout = 1;
+
+FLUSH STATUS;
+
+# MDEV-5336 introduces the status variable Com_backup
+--echo # Show how the status variable 'Com_backup' changes after BACKUP STAGE ..
+
+SHOW STATUS LIKE 'Com_backup';
+BACKUP STAGE START;
+SHOW STATUS LIKE 'Com_backup';
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE START;
+SHOW STATUS LIKE 'Com_backup';
+BACKUP STAGE FLUSH;
+SHOW STATUS LIKE 'Com_backup';
+BACKUP STAGE BLOCK_DDL;
+SHOW STATUS LIKE 'Com_backup';
+BACKUP STAGE BLOCK_COMMIT;
+SHOW STATUS LIKE 'Com_backup';
+BACKUP STAGE END;
+
+--echo # In case the backup lock is taken by the current connection than
+--echo # - DML modifying some permanent table is not allowed
+
+CREATE TABLE t1_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t1_myisam (col1 INT) ENGINE = MyISAM;
+
+BACKUP STAGE START;
+SET AUTOCOMMIT = 0;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+INSERT INTO t1_innodb SET col1 = 1;
+SET AUTOCOMMIT = 1;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+INSERT INTO t1_innodb SET col1 = 1;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+INSERT INTO t1_myisam SET col1 = 1;
+--echo # - DDL creating or renaming a permanent table or a procedure etc.
+--echo #   is not allowed.                                                               
+--error ER_BACKUP_LOCK_IS_ACTIVE
+CREATE TABLE throw_away (col1 INT) ENGINE = InnoDB;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+RENAME TABLE t1_innodb To throw_away;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+CREATE PROCEDURE p1() SELECT 13;
+--error ER_BACKUP_LOCK_IS_ACTIVE
+CREATE PROCEDURE p1() SELECT 13;
+BACKUP STAGE END;
+DROP TABLE t1_innodb;
+DROP TABLE t1_myisam;
+
+--echo #
+--echo # Creating and modifying TEMPORARY TABLES are allowed
+--echo #
+
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_DDL;
+CREATE TEMPORARY TABLE tmp (col1 INT);
+DROP TEMPORARY TABLE tmp;
+
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TEMPORARY TABLE t_temporary_myisam (col1 INT) ENGINE = MyISAM;
+--echo # - DML modifying that temporary table is allowed.
+INSERT INTO t_temporary_innodb SET col1 = 1;
+SELECT COUNT(*) FROM t_temporary_innodb;
+INSERT INTO t_temporary_myisam SET col1 = 1;
+SELECT COUNT(*) FROM t_temporary_myisam;
+BACKUP STAGE END;
+
+--echo # Show the fate and impact of some auto committed INSERT into temporary
+--echo # table sliding through the sequence.
+SET AUTOCOMMIT = 1;
+BACKUP STAGE START;
+INSERT INTO t_temporary_innodb SET col1 = 1;
+INSERT INTO t_temporary_myisam SET col1 = 1;
+BACKUP STAGE FLUSH;
+INSERT INTO t_temporary_innodb SET col1 = 1;
+INSERT INTO t_temporary_myisam SET col1 = 1;
+BACKUP STAGE BLOCK_DDL;
+INSERT INTO t_temporary_innodb SET col1 = 1;
+INSERT INTO t_temporary_myisam SET col1 = 1;
+BACKUP STAGE BLOCK_COMMIT;
+INSERT INTO t_temporary_innodb SET col1 = 1;
+INSERT INTO t_temporary_myisam SET col1 = 1;
+BACKUP STAGE END;
+SELECT COUNT(*) FROM t_temporary_innodb;
+
+--echo # Show the fate and impact of some DROP/CREATE TEMPORARY TABLE sliding
+--echo # through the sequence.
+SET AUTOCOMMIT = 1;
+BACKUP STAGE START;
+DROP TEMPORARY TABLE t_temporary_innodb;
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+BACKUP STAGE FLUSH;
+DROP TEMPORARY TABLE t_temporary_innodb;
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+BACKUP STAGE BLOCK_DDL;
+DROP TEMPORARY TABLE t_temporary_innodb;
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+BACKUP STAGE BLOCK_COMMIT;
+DROP TEMPORARY TABLE t_temporary_innodb;
+CREATE TEMPORARY TABLE t_temporary_innodb (col1 INT) ENGINE = InnoDB;
+BACKUP STAGE END;
+--echo # Show that even more DDL on the temporary table is allowed.
+BACKUP STAGE START;
+TRUNCATE t_temporary_innodb;
+ALTER TABLE t_temporary_innodb ADD COLUMN col2 INT;
+ALTER TABLE t_temporary_innodb ADD KEY idx(col2);
+BACKUP STAGE END;
diff --git a/mysql-test/main/backup_lock.result b/mysql-test/main/backup_lock.result
new file mode 100644
index 00000000000..40072aa0684
--- /dev/null
+++ b/mysql-test/main/backup_lock.result
@@ -0,0 +1,221 @@
+#
+# Testing which locks we get from all stages
+#
+BACKUP STAGE START;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_BACKUP_START	Backup lock		
+BACKUP STAGE FLUSH;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_BACKUP_FLUSH	Backup lock		
+BACKUP STAGE BLOCK_DDL;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_BACKUP_WAIT_DDL	Backup lock		
+BACKUP STAGE BLOCK_COMMIT;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_BACKUP_WAIT_COMMIT	Backup lock		
+BACKUP STAGE END;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+#
+# testing BACKUP STAGE LOCK's
+#
+connect  con1,localhost,root,,;
+connect  con2,localhost,root,,;
+connection default;
+#
+# testing if BACKUP STAGE FLUSH causes deadlocks with ALTER TABLE
+#
+create table t1 (a int) engine=innodb;
+start transaction;
+insert into t1 values (1);
+connection con1;
+alter table t1 add column (j int), algorithm copy;
+connection con2;
+backup stage start;
+backup stage flush;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_BACKUP_DDL	Backup lock		
+MDL_BACKUP_FLUSH	Backup lock		
+MDL_SHARED_WRITE	Table metadata lock	test	t1
+MDL_SHARED_UPGRADABLE	Table metadata lock	test	t1
+MDL_INTENTION_EXCLUSIVE	Schema metadata lock	test	
+SET STATEMENT max_statement_time=1 FOR backup stage block_ddl;
+ERROR 70100: Query execution was interrupted (max_statement_time exceeded)
+backup stage block_ddl;
+connection default;
+commit;
+SELECT * FROM t1;
+a
+1
+SET STATEMENT lock_wait_timeout=0 FOR INSERT INTO t1 values (2);
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+INSERT INTO t1 values (2,0);;
+connection con2;
+backup stage end;
+connection con1;
+connection default;
+select * from t1;
+a	j
+1	NULL
+2	0
+drop table t1;
+# Test with inline alter table, which doesn't block block_commit
+create table t1 (a int) engine=innodb;
+start transaction;
+insert into t1 values (1);
+connection con1;
+alter table t1 add column (j int);
+connection con2;
+backup stage start;
+backup stage flush;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_BACKUP_ALTER_COPY	Backup lock		
+MDL_BACKUP_FLUSH	Backup lock		
+MDL_SHARED_WRITE	Table metadata lock	test	t1
+MDL_SHARED_UPGRADABLE	Table metadata lock	test	t1
+MDL_INTENTION_EXCLUSIVE	Schema metadata lock	test	
+backup stage block_ddl;
+backup stage block_commit;
+connection default;
+SELECT * FROM t1;
+a
+1
+commit;
+connection con2;
+backup stage end;
+connection con1;
+connection default;
+drop table t1;
+#
+# testing if BACKUP STAGE FLUSH causes deadlocks with DROP TABLE
+#
+create table t1 (a int) engine=innodb;
+start transaction;
+insert into t1 values (1);
+connection con1;
+SET STATEMENT lock_wait_timeout=0 FOR DROP TABLE t1;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+DROP TABLE t1;
+connection con2;
+backup stage start;
+backup stage flush;
+SET STATEMENT lock_wait_timeout=0 FOR SELECT * FROM t1;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+SET STATEMENT lock_wait_timeout=0 FOR backup stage block_ddl;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_BACKUP_DDL	Backup lock		
+MDL_BACKUP_FLUSH	Backup lock		
+MDL_SHARED_WRITE	Table metadata lock	test	t1
+MDL_INTENTION_EXCLUSIVE	Schema metadata lock	test	
+backup stage end;
+connection default;
+commit;
+connection con1;
+connection default;
+#
+# Check if backup stage block_dll + concurrent drop table blocks select
+#
+create table t1 (a int) engine=innodb;
+backup stage start;
+backup stage block_ddl;
+connection con1;
+DROP TABLE t1;
+connection con2;
+connection con2;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_BACKUP_WAIT_DDL	Backup lock		
+SELECT * FROM t1;
+a
+connection default;
+backup stage end;
+connection con1;
+connection default;
+#
+# Check if backup stage block_dll overrides ddl lock for drop table
+#
+create table t1 (a int) engine=innodb;
+start transaction;
+insert into t1 values (1);
+connection con1;
+SET STATEMENT lock_wait_timeout=0 FOR DROP TABLE t1;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+DROP TABLE t1;
+connection con2;
+backup stage start;
+backup stage flush;
+backup stage block_ddl;
+connection default;
+commit;
+connection con2;
+backup stage end;
+connection con1;
+connection default;
+#
+# Check if BACKUP STAGE BLOCK_COMMIT blocks commit
+#
+create table t1 (a int) engine=innodb;
+start transaction;
+insert into t1 values (1);
+connection con1;
+backup stage start;
+backup stage block_commit;
+connection default;
+commit;
+connection con1;
+backup stage end;
+connection default;
+select * from t1;
+a
+1
+drop table t1;
+disconnect con1;
+disconnect con2;
+#
+# Test backup stage and flush tables
+#
+BACKUP STAGE START ;
+BACKUP STAGE BLOCK_DDL ;
+FLUSH TABLES;
+CREATE TEMPORARY TABLE t12345678_tmp (col1 INT);
+drop table t12345678_tmp;
+BACKUP STAGE END;
+#
+# Test BACKUP STAGES with lock timeouts
+#
+SET GLOBAL lock_wait_timeout=0;
+CREATE TABLE t_permanent_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t_permanent_myisam (col1 INT) ENGINE = MyISAM;
+INSERT INTO t_permanent_innodb SET col1 = 1;
+INSERT INTO t_permanent_myisam SET col1 = 1;
+CREATE TABLE t_con1_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t_con1_myisam (col1 INT) ENGINE = MyISAM;
+connect con1,localhost,root,,;
+SET AUTOCOMMIT = 0;
+connection default;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+connection con1;
+UPDATE t_permanent_innodb SET col1 = 8;
+UPDATE t_permanent_myisam SET col1 = 8;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+DROP TABLE t_con1_innodb;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+DROP TABLE t_con1_myisam;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection default;
+BACKUP STAGE END;
+DROP TABLE t_permanent_myisam, t_permanent_innodb;
+DROP TABLE t_con1_innodb, t_con1_myisam;
+disconnect con1;
+set global lock_wait_timeout=default;
diff --git a/mysql-test/main/backup_lock.test b/mysql-test/main/backup_lock.test
new file mode 100644
index 00000000000..aafeb3a2d4b
--- /dev/null
+++ b/mysql-test/main/backup_lock.test
@@ -0,0 +1,290 @@
+########################################################################
+# Tests BACKUP STAGE locking
+########################################################################
+
+--source include/have_innodb.inc
+--source include/have_metadata_lock_info.inc
+--source include/not_embedded.inc
+
+--echo #
+--echo # Testing which locks we get from all stages
+--echo #
+
+BACKUP STAGE START;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+BACKUP STAGE FLUSH;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+BACKUP STAGE BLOCK_DDL;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+BACKUP STAGE BLOCK_COMMIT;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+BACKUP STAGE END;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+
+--echo #
+--echo # testing BACKUP STAGE LOCK's
+--echo #
+
+# Following connections are used in a few of the following tests
+connect (con1,localhost,root,,);
+connect (con2,localhost,root,,);
+connection default;
+
+--echo #
+--echo # testing if BACKUP STAGE FLUSH causes deadlocks with ALTER TABLE
+--echo #
+
+create table t1 (a int) engine=innodb;
+
+start transaction;
+# Acquires MDL lock
+insert into t1 values (1);
+
+connection con1;
+# Waits on MDL
+--send alter table t1 add column (j int), algorithm copy
+
+connection con2;
+let $wait_condition=
+    select count(*) = 1 from information_schema.processlist
+    where state = "Waiting for table metadata lock";
+--source include/wait_condition.inc
+backup stage start;
+backup stage flush;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+#
+# Do first test with max_statement_time, other tests later are done with
+# lock_wait_timeout. This is mostly to ensure that both methods works
+#
+--error ER_STATEMENT_TIMEOUT
+SET STATEMENT max_statement_time=1 FOR backup stage block_ddl;
+--send backup stage block_ddl
+
+connection default;
+let $wait_condition=
+    select count(*) = 1 from information_schema.processlist
+    where state = "Waiting for backup lock";
+--source include/wait_condition.inc
+commit;
+# The following select works because alter table is waiting for DDL lock
+SELECT * FROM t1;
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR INSERT INTO t1 values (2);
+--send INSERT INTO t1 values (2,0);
+connection con2;
+--reap # BLOCK_DDL
+backup stage end;
+connection con1;
+--reap  # ALTER TABLE
+connection default;
+--reap  # INSERT
+select * from t1;
+drop table t1;
+
+--echo # Test with inline alter table, which doesn't block block_commit
+
+create table t1 (a int) engine=innodb;
+
+start transaction;
+# Acquires MDL lock
+insert into t1 values (1);
+
+connection con1;
+# Waits on MDL
+--send alter table t1 add column (j int)
+
+connection con2;
+let $wait_condition=
+    select count(*) = 1 from information_schema.processlist
+    where state = "Waiting for table metadata lock";
+--source include/wait_condition.inc
+backup stage start;
+backup stage flush;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+backup stage block_ddl;
+backup stage block_commit;
+connection default;
+SELECT * FROM t1;
+--send commit
+connection con2;
+let $wait_condition=
+    select count(*) = 1 from information_schema.processlist
+    where state = "Waiting for backup lock";
+backup stage end;
+connection con1;
+--reap  # ALTER TABLE
+connection default;
+--reap # commit
+drop table t1;
+
+--echo #
+--echo # testing if BACKUP STAGE FLUSH causes deadlocks with DROP TABLE
+--echo #
+
+create table t1 (a int) engine=innodb;
+start transaction;
+# Acquires MDL lock
+insert into t1 values (1);
+
+connection con1;
+# Waits on MDL
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR DROP TABLE t1;
+--send DROP TABLE t1
+
+connection con2;
+backup stage start;
+backup stage flush;
+let $wait_condition=
+    select count(*) = 1 from information_schema.processlist
+    where state = "Waiting for table metadata lock";
+--source include/wait_condition.inc
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR SELECT * FROM t1;
+
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR backup stage block_ddl;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+backup stage end;
+
+connection default;
+commit;
+connection con1;
+--reap  # DROP TABLE
+connection default;
+
+--echo #
+--echo # Check if backup stage block_dll + concurrent drop table blocks select
+--echo #
+
+create table t1 (a int) engine=innodb;
+backup stage start;
+backup stage block_ddl;
+connection con1;
+--send DROP TABLE t1
+connection con2;
+let $wait_condition=
+    select count(*) = 1 from information_schema.processlist
+    where state = "Waiting for backup lock";
+--source include/wait_condition.inc
+connection con2;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+# Check that select's are not blocked
+SELECT * FROM t1;
+connection default;
+backup stage end;
+connection con1;
+--reap
+connection default;
+
+--echo #
+--echo # Check if backup stage block_dll overrides ddl lock for drop table
+--echo #
+
+create table t1 (a int) engine=innodb;
+start transaction;
+# Acquires MDL lock
+insert into t1 values (1);
+
+connection con1;
+# Waits on MDL
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR DROP TABLE t1;
+--send DROP TABLE t1
+
+connection con2;
+backup stage start;
+backup stage flush;
+--send backup stage block_ddl
+connection default;
+let $wait_condition=
+    select count(*) = 1 from information_schema.processlist
+    where state = "Waiting for backup lock";
+--source include/wait_condition.inc
+commit;
+connection con2;
+--reap
+backup stage end;
+connection con1;
+--reap  # DROP TABLE
+connection default;
+
+--echo #
+--echo # Check if BACKUP STAGE BLOCK_COMMIT blocks commit
+--echo #
+
+create table t1 (a int) engine=innodb;
+start transaction;
+# Acquires MDL lock
+insert into t1 values (1);
+
+connection con1;
+backup stage start;
+backup stage block_commit;
+connection default;
+--send commit
+connection con1;
+let $wait_condition=
+    select count(*) = 1 from information_schema.processlist
+    where state = "Waiting for backup lock";
+backup stage end;
+connection default;
+--reap # commit
+select * from t1;
+drop table t1;
+
+#
+# End of tests using con1 and con2
+#
+disconnect con1;
+disconnect con2;
+
+--echo #
+--echo # Test backup stage and flush tables
+--echo #
+
+BACKUP STAGE START ;
+BACKUP STAGE BLOCK_DDL ;
+FLUSH TABLES;
+CREATE TEMPORARY TABLE t12345678_tmp (col1 INT);
+drop table t12345678_tmp;
+BACKUP STAGE END;
+
+--echo #
+--echo # Test BACKUP STAGES with lock timeouts
+--echo #
+
+SET GLOBAL lock_wait_timeout=0;
+CREATE TABLE t_permanent_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t_permanent_myisam (col1 INT) ENGINE = MyISAM;
+INSERT INTO t_permanent_innodb SET col1 = 1;
+
+INSERT INTO t_permanent_myisam SET col1 = 1;
+CREATE TABLE t_con1_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t_con1_myisam (col1 INT) ENGINE = MyISAM;
+
+--connect(con1,localhost,root,,)
+SET AUTOCOMMIT = 0;
+
+--connection default
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+
+--connection con1
+UPDATE t_permanent_innodb SET col1 = 8;
+--error ER_LOCK_WAIT_TIMEOUT
+UPDATE t_permanent_myisam SET col1 = 8;
+--error ER_LOCK_WAIT_TIMEOUT
+DROP TABLE t_con1_innodb;
+
+--error ER_LOCK_WAIT_TIMEOUT
+DROP TABLE t_con1_myisam;
+
+--connection default
+BACKUP STAGE END;
+DROP TABLE t_permanent_myisam, t_permanent_innodb;
+DROP TABLE t_con1_innodb, t_con1_myisam;
+--disconnect con1
+set global lock_wait_timeout=default;
diff --git a/mysql-test/main/backup_priv.result b/mysql-test/main/backup_priv.result
new file mode 100644
index 00000000000..4169f58f40f
--- /dev/null
+++ b/mysql-test/main/backup_priv.result
@@ -0,0 +1,40 @@
+#
+# Test privileges for BACKUP STAGES
+#
+set sql_mode="";
+GRANT RELOAD ON *.* TO user1@localhost;
+GRANT CREATE, DROP ON *.* TO user2@localhost;
+connect con1, localhost, user1;
+BACKUP STAGE START;
+BACKUP STAGE END;
+# change_user must release backup lock
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+SELECT lock_mode FROM information_schema.metadata_lock_info WHERE lock_type='Backup lock';
+lock_mode
+MDL_BACKUP_FLUSH
+SELECT lock_mode FROM information_schema.metadata_lock_info WHERE lock_type='Backup lock';
+lock_mode
+disconnect con1;
+connection default;
+# A very low privileged user (-> con4) cannot acquire the backup lock
+connect con1, localhost, user2;
+BACKUP STAGE START;
+ERROR 42000: Access denied; you need (at least one of) the RELOAD privilege(s) for this operation
+BACKUP STAGE FLUSH;
+ERROR 42000: Access denied; you need (at least one of) the RELOAD privilege(s) for this operation
+BACKUP STAGE BLOCK_DDL;
+ERROR 42000: Access denied; you need (at least one of) the RELOAD privilege(s) for this operation
+BACKUP STAGE BLOCK_COMMIT;
+ERROR 42000: Access denied; you need (at least one of) the RELOAD privilege(s) for this operation
+BACKUP STAGE END;
+ERROR 42000: Access denied; you need (at least one of) the RELOAD privilege(s) for this operation
+disconnect con1;
+connection default;
+DROP USER user1@localhost, user2@localhost;
+#
+# Test using BACKUP STAGES in a SP
+#
+create procedure foo42()
+BACKUP STAGE START;
+ERROR 0A000: BACKUP STAGE is not allowed in stored procedures
diff --git a/mysql-test/main/backup_priv.test b/mysql-test/main/backup_priv.test
new file mode 100644
index 00000000000..93b69af0b67
--- /dev/null
+++ b/mysql-test/main/backup_priv.test
@@ -0,0 +1,52 @@
+--source include/have_innodb.inc
+--source include/not_embedded.inc
+--source include/have_metadata_lock_info.inc
+
+--echo #
+--echo # Test privileges for BACKUP STAGES
+--echo #
+
+set sql_mode="";
+
+GRANT RELOAD ON *.* TO user1@localhost;
+GRANT CREATE, DROP ON *.* TO user2@localhost;
+
+--connect(con1, localhost, user1)
+BACKUP STAGE START;
+BACKUP STAGE END;
+--echo # change_user must release backup lock
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+SELECT lock_mode FROM information_schema.metadata_lock_info WHERE lock_type='Backup lock';
+change_user user2;
+SELECT lock_mode FROM information_schema.metadata_lock_info WHERE lock_type='Backup lock';
+--disconnect con1
+--source include/wait_until_disconnected.inc
+--connection default
+
+--echo # A very low privileged user (-> con4) cannot acquire the backup lock
+
+--connect(con1, localhost, user2)
+--error ER_SPECIFIC_ACCESS_DENIED_ERROR
+BACKUP STAGE START;
+--error ER_SPECIFIC_ACCESS_DENIED_ERROR
+BACKUP STAGE FLUSH;
+--error ER_SPECIFIC_ACCESS_DENIED_ERROR
+BACKUP STAGE BLOCK_DDL;
+--error ER_SPECIFIC_ACCESS_DENIED_ERROR
+BACKUP STAGE BLOCK_COMMIT;
+--error ER_SPECIFIC_ACCESS_DENIED_ERROR
+BACKUP STAGE END;
+--disconnect con1
+--source include/wait_until_disconnected.inc
+
+--connection default
+DROP USER user1@localhost, user2@localhost;
+
+--echo #
+--echo # Test using BACKUP STAGES in a SP
+--echo #
+
+--error ER_SP_BADSTATEMENT
+create procedure foo42()
+  BACKUP STAGE START;
diff --git a/mysql-test/main/backup_stages.result b/mysql-test/main/backup_stages.result
new file mode 100644
index 00000000000..caea1fda0b9
--- /dev/null
+++ b/mysql-test/main/backup_stages.result
@@ -0,0 +1,335 @@
+#-----------------------------------------------------------------------
+# Multi-threaded tests
+#-----------------------------------------------------------------------
+# Show that only one connection can hold the backup lock.
+connection default;
+BACKUP STAGE START;
+connect con1,localhost,root,,;
+SET STATEMENT lock_wait_timeout=0 FOR BACKUP STAGE START;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+BACKUP STAGE START;
+connection default;
+# Show that the connection con1 has to wait for the backup lock and the
+# corresponding representation within the processlist.
+SET @con1_id = <con1_id>;
+SELECT ID, USER, COMMAND, STATE, INFO, STAGE, MAX_STAGE, INFO_BINARY
+FROM information_schema.processlist WHERE id = @con1_id;
+ID	USER	COMMAND	STATE	INFO	STAGE	MAX_STAGE	INFO_BINARY
+<con1_id>	root	Query	Waiting for backup lock	BACKUP STAGE START	0	0	BACKUP STAGE START
+BACKUP STAGE END;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_BACKUP_START	Backup lock		
+connection con1;
+# The connection default has removed the backup lock.
+# And so the current connection con1 can reap for its BACKUP STAGE START
+connect con2,localhost,root,,;
+# The connection con2 cannot continue the work of con1 by setting the
+# next BACKUP STAGE FLUSH.
+BACKUP STAGE FLUSH;
+ERROR HY000: You must start backup with "BACKUP STAGE START"
+BACKUP STAGE START;
+connection default;
+SET @con2_id = <con2_id>;
+# Connection con2 waits for the backup lock held by con1.
+SELECT ID, USER, COMMAND, STATE, INFO, STAGE, MAX_STAGE, INFO_BINARY
+FROM information_schema.processlist WHERE id = @con2_id;
+ID	USER	COMMAND	STATE	INFO	STAGE	MAX_STAGE	INFO_BINARY
+<con2_id>	root	Query	Waiting for backup lock	BACKUP STAGE START	0	0	BACKUP STAGE START
+disconnect con1;
+connection con2;
+# Connection con1 frees the backup lock held by disconnecting.
+# So connection con2 gets the backup lock.
+connect con3,localhost,root,,;
+BACKUP STAGE START;
+connection default;
+SET @con3_id = <con3_id>;
+# Connection con3 waits for the backup lock held by con2.
+SELECT ID, USER, COMMAND, STATE, INFO, STAGE, MAX_STAGE, INFO_BINARY
+FROM information_schema.processlist WHERE id = @con3_id;
+ID	USER	COMMAND	STATE	INFO	STAGE	MAX_STAGE	INFO_BINARY
+<con3_id>	root	Query	Waiting for backup lock	BACKUP STAGE START	0	0	BACKUP STAGE START
+KILL CONNECTION @con2_id;
+connection con3;
+# Connection con2 frees the backup lock held by getting killed.
+# So connection con3 gets the backup lock.
+BACKUP STAGE END;
+disconnect con3;
+connection default;
+CREATE TABLE t_permanent_innodb (col1 INT) ENGINE = InnoDB;
+INSERT INTO t_permanent_innodb SET col1 = 1;
+CREATE TABLE t_permanent_myisam (col1 INT) ENGINE = MyISAM;
+INSERT INTO t_permanent_myisam SET col1 = 1;
+connect backup,localhost,root,,;
+connect con11,localhost,root,,;
+SET AUTOCOMMIT = 0;
+set session lock_wait_timeout=0;
+connect con12,localhost,root,,;
+SET AUTOCOMMIT = 1;
+# Between (connection default) BACKUP STAGE START and FLUSH
+# no restrictions for concurrent sessions regarding DDL or DML
+# affecting transactional/non transactional permanent tables.
+connection backup;
+BACKUP STAGE START;
+connection con11;
+UPDATE t_permanent_innodb SET col1 = 2;
+UPDATE t_permanent_myisam SET col1 = 2;
+SELECT COUNT(*) FROM t_permanent_innodb;
+COUNT(*)
+1
+HANDLER t_permanent_innodb OPEN;
+HANDLER t_permanent_innodb READ FIRST;
+col1
+2
+HANDLER t_permanent_innodb CLOSE;
+SELECT COUNT(*) FROM t_permanent_myisam;
+COUNT(*)
+1
+HANDLER t_permanent_myisam OPEN;
+HANDLER t_permanent_myisam READ FIRST;
+col1
+2
+HANDLER t_permanent_myisam CLOSE;
+CREATE TABLE t_con1_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t_con1_myisam (col1 INT) ENGINE = InnoDB;
+ALTER TABLE t_permanent_innodb ADD COLUMN col2 INT;
+ALTER TABLE t_permanent_myisam ADD COLUMN col2 INT;
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_innodb;
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_myisam;
+connection con12;
+UPDATE t_permanent_innodb SET col1 = 3;
+UPDATE t_permanent_myisam SET col1 = 3;
+# Between (connection default) BACKUP STAGE FLUSH and BLOCK_DDL
+# concurrent sessions
+# - can change transactional permanent tables with DDL and DML
+# - can run DROP/CREATE transactional/non transactional TABLE
+# - cannot modify non transactional permanent tables with DDL or DML
+connection backup;
+BACKUP STAGE FLUSH;
+connection con11;
+UPDATE t_permanent_innodb SET col1 = 4;
+SET STATEMENT lock_wait_timeout=0 FOR UPDATE t_permanent_myisam SET col1 = 4;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+SELECT COUNT(*) FROM t_permanent_innodb;
+COUNT(*)
+1
+HANDLER t_permanent_innodb OPEN;
+HANDLER t_permanent_innodb READ FIRST;
+col1	col2
+4	NULL
+HANDLER t_permanent_innodb CLOSE;
+SELECT COUNT(*) FROM t_permanent_myisam;
+COUNT(*)
+1
+HANDLER t_permanent_myisam OPEN;
+HANDLER t_permanent_myisam READ FIRST;
+col1	col2
+3	NULL
+HANDLER t_permanent_myisam CLOSE;
+DROP TABLE t_con1_innodb;
+DROP TABLE t_con1_myisam;
+CREATE TABLE t_con1_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t_con1_myisam (col1 INT) ENGINE = InnoDB;
+ALTER TABLE t_permanent_innodb ADD COLUMN col3 INT;
+SET STATEMENT lock_wait_timeout=0 FOR ALTER TABLE t_permanent_myisam ADD COLUMN col3 INT;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_innodb;
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_myisam;
+connection con12;
+UPDATE t_permanent_innodb SET col1 = 5;
+# Between (connection default) BACKUP STAGE BLOCK_DDL and BLOCK_COMMIT
+# concurrent sessions
+# - can change transactional permanent tables with DML
+# - cannot run DDL
+# - cannot change non transactional permanent tables with DML
+connection backup;
+BACKUP STAGE BLOCK_DDL;
+connection con11;
+UPDATE t_permanent_innodb SET col1 = 6;
+UPDATE t_permanent_myisam SET col1 = 6;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+SELECT COUNT(*) FROM t_permanent_innodb;
+COUNT(*)
+1
+HANDLER t_permanent_innodb OPEN;
+HANDLER t_permanent_innodb READ FIRST;
+col1	col2	col3
+6	NULL	NULL
+HANDLER t_permanent_innodb CLOSE;
+SELECT COUNT(*) FROM t_permanent_myisam;
+COUNT(*)
+1
+HANDLER t_permanent_myisam OPEN;
+HANDLER t_permanent_myisam READ FIRST;
+col1	col2
+3	NULL
+HANDLER t_permanent_myisam CLOSE;
+DROP TABLE t_con1_innodb;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+CREATE TABLE throw_away (col1 INT) ENGINE = InnoDB;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+ALTER TABLE t_permanent_innodb ADD COLUMN col4 INT;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_innodb;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_myisam;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection con12;
+UPDATE t_permanent_innodb SET col1 = 7;
+# Between (connection default) BACKUP STAGE BLOCK_COMMIT and END
+# concurrent sessions
+# - can change transactional permanent tables with DML
+# - cannot run DDL
+# - cannot change non transactional permanent tables with DML
+connection backup;
+BACKUP STAGE BLOCK_COMMIT;
+connection con11;
+UPDATE t_permanent_innodb SET col1 = 8;
+UPDATE t_permanent_myisam SET col1 = 8;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+SELECT COUNT(*) FROM t_permanent_innodb;
+COUNT(*)
+1
+HANDLER t_permanent_innodb OPEN;
+HANDLER t_permanent_innodb READ FIRST;
+col1	col2	col3
+8	NULL	NULL
+HANDLER t_permanent_innodb CLOSE;
+SELECT COUNT(*) FROM t_permanent_myisam;
+COUNT(*)
+1
+HANDLER t_permanent_myisam OPEN;
+HANDLER t_permanent_myisam READ FIRST;
+col1	col2
+3	NULL
+HANDLER t_permanent_myisam CLOSE;
+DROP TABLE t_con1_innodb;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+DROP TABLE t_con1_myisam;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection con12;
+SET STATEMENT lock_wait_timeout=0 FOR UPDATE t_permanent_innodb SET col1 = 9;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection backup;
+BACKUP STAGE END;
+connection con11;
+COMMIT;
+SELECT * FROM t_permanent_innodb ORDER BY col1;
+col1	col2	col3
+7	NULL	NULL
+SELECT * FROM t_permanent_myisam ORDER BY col1;
+col1	col2
+3	NULL
+SET AUTOCOMMIT = 0;
+SET GLOBAL tx_read_only = 1;
+connection con12;
+BACKUP STAGE START;
+BACKUP STAGE END;
+SET GLOBAL tx_read_only = 0;
+DROP VIEW v_some_view;
+DROP TABLE t_con1_innodb;
+DROP TABLE t_con1_myisam;
+# Connection backup holds the backup log and is on some stage.
+# Connection con11 tries to LOCK TABLEs or to set read_only.
+connection backup;
+BACKUP STAGE START;
+connection con11;
+# Between BACKUP STAGE START and FLUSH:
+# No restrictions for other connection around LOCK TABLES or read-only.
+LOCK TABLES t_permanent_innodb READ;
+LOCK TABLES t_permanent_myisam READ;
+LOCK TABLES t_permanent_innodb WRITE;
+LOCK TABLES t_permanent_myisam WRITE;
+UNLOCK TABLES;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+connection backup;
+BACKUP STAGE FLUSH;
+connection con11;
+# Between BACKUP STAGE FLUSH and BLOCK_COMMIT:
+# Connection con11 not holding the backup lock cannot
+# LOCK WRITE non transactional table.
+LOCK TABLES t_permanent_innodb READ;
+LOCK TABLES t_permanent_myisam READ;
+LOCK TABLES t_permanent_innodb WRITE;
+LOCK TABLES t_permanent_myisam WRITE;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+UNLOCK TABLES;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+connection backup;
+BACKUP STAGE BLOCK_DDL;
+connection con11;
+# Between BACKUP STAGE FLUSH and BLOCK_COMMIT:
+# Connection con11 not holding the backup lock cannot
+# LOCK WRITE transactional or non transactional table.
+LOCK TABLES t_permanent_innodb READ;
+LOCK TABLES t_permanent_myisam READ;
+LOCK TABLES t_permanent_innodb WRITE;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+LOCK TABLES t_permanent_myisam WRITE;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+UNLOCK TABLES;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+connection backup;
+BACKUP STAGE BLOCK_COMMIT;
+connection con11;
+# Between BACKUP BLOCK_COMMIT FLUSH and END:
+# Connection con11 not holding the backup lock cannot
+# LOCK WRITE transactional or non transactional table.
+LOCK TABLES t_permanent_innodb READ;
+LOCK TABLES t_permanent_myisam READ;
+LOCK TABLES t_permanent_innodb WRITE;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+LOCK TABLES t_permanent_myisam WRITE;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+UNLOCK TABLES;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+connection backup;
+BACKUP STAGE END;
+DROP TABLE t_permanent_innodb;
+DROP TABLE t_permanent_myisam;
+#
+# Log tables
+#
+connection backup;
+SET @old_general_log = @@general_log;
+SET @old_slow_query_log = @@slow_query_log;
+SET @old_log_output = @@log_output;
+SET GLOBAL log_output = 'TABLE';
+SET GLOBAL general_log = ON;
+SET GLOBAL slow_query_log = ON;
+connection con11;
+SET @old_long_query_time = @@SESSION.long_query_time;
+SET SESSION long_query_time = 0;
+connection backup;
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+connection con11;
+SELECT 1;
+1
+1
+connection backup;
+SELECT 1;
+1
+1
+connection con11;
+SET SESSION long_query_time = @old_long_query_time;
+connection backup;
+BACKUP STAGE END;
+SET GLOBAL log_output = @old_log_output;
+SET GLOBAL slow_query_log = @old_slow_query_log;
+SET GLOBAL general_log = @old_general_log;
+#-----------------------------------------------------------------------
+# Cleanup
+#-----------------------------------------------------------------------
+SET GLOBAL lock_wait_timeout = <old_lock_wait_timeout>;
+disconnect con2;
+disconnect con11;
+disconnect con12;
+disconnect backup;
+connection default;
diff --git a/mysql-test/main/backup_stages.test b/mysql-test/main/backup_stages.test
new file mode 100644
index 00000000000..ba9c15a1d7e
--- /dev/null
+++ b/mysql-test/main/backup_stages.test
@@ -0,0 +1,385 @@
+########################################################################
+# Test what is locked in each stage for LOCK FOR BACKUP (MDEV-5336)
+########################################################################
+
+--source include/not_embedded.inc
+# A transactional engine
+--source include/have_innodb.inc
+--source include/have_metadata_lock_info.inc
+# As non transactional engine we have MyISAM anyway.
+
+# Save the initial number of concurrent sessions.
+--source include/count_sessions.inc
+
+let $old_lock_wait_timeout = `SELECT @@global.lock_wait_timeout`;
+
+--echo #-----------------------------------------------------------------------
+--echo # Multi-threaded tests
+--echo #-----------------------------------------------------------------------
+--echo # Show that only one connection can hold the backup lock.
+
+--connection default
+let $default_id= `SELECT CONNECTION_ID()`;
+BACKUP STAGE START;
+
+# con1, root high privileged user
+--connect(con1,localhost,root,,)
+let $con1_id= `SELECT CONNECTION_ID()`;
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR BACKUP STAGE START;
+send BACKUP STAGE START;
+--connection default
+
+--echo # Show that the connection con1 has to wait for the backup lock and the
+--echo # corresponding representation within the processlist.
+
+--replace_result $con1_id <con1_id>
+
+eval SET @con1_id = $con1_id;
+# Output expected here is
+# ID        USER COMMAND STATE                   INFO               INFO_BINARY
+# <con1_id> root   Query Waiting for backup lock BACKUP STAGE START BACKUP STAGE START
+let $wait_condition=
+    SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.PROCESSLIST
+    WHERE STATE = "Waiting for backup lock" and INFO = "BACKUP STAGE START";
+--source include/wait_condition.inc
+--replace_column 1 <con1_id>
+eval
+SELECT ID, USER, COMMAND, STATE, INFO, STAGE, MAX_STAGE, INFO_BINARY
+FROM information_schema.processlist WHERE id = @con1_id;
+# con1 uses @@global.lock_wait_timeout
+
+BACKUP STAGE END;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+
+--connection con1
+--echo # The connection default has removed the backup lock.
+--echo # And so the current connection con1 can reap for its BACKUP STAGE START
+--reap
+
+# con2, root high privileged user
+--connect(con2,localhost,root,,)
+let $con2_id= `SELECT CONNECTION_ID()`;
+--echo # The connection con2 cannot continue the work of con1 by setting the
+--echo # next BACKUP STAGE FLUSH.
+--error ER_BACKUP_NOT_RUNNING
+BACKUP STAGE FLUSH;
+send
+BACKUP STAGE START;
+
+--connection default
+--replace_result $con2_id <con2_id>
+eval SET @con2_id = $con2_id;
+--echo # Connection con2 waits for the backup lock held by con1.
+--source include/wait_condition.inc
+--replace_column 1 <con2_id>
+eval
+SELECT ID, USER, COMMAND, STATE, INFO, STAGE, MAX_STAGE, INFO_BINARY
+FROM information_schema.processlist WHERE id = @con2_id;
+
+--disconnect con1
+
+--connection con2
+--echo # Connection con1 frees the backup lock held by disconnecting.
+--echo # So connection con2 gets the backup lock.
+--reap
+
+--connect(con3,localhost,root,,)
+let $con3_id= `SELECT CONNECTION_ID()`;
+send
+BACKUP STAGE START;
+
+--connection default
+--replace_result $con3_id <con3_id>
+eval SET @con3_id = $con3_id;
+--echo # Connection con3 waits for the backup lock held by con2.
+--source include/wait_condition.inc
+--replace_column 1 <con3_id>
+eval
+SELECT ID, USER, COMMAND, STATE, INFO, STAGE, MAX_STAGE, INFO_BINARY
+FROM information_schema.processlist WHERE id = @con3_id;
+KILL CONNECTION @con2_id;
+
+--connection con3
+--echo # Connection con2 frees the backup lock held by getting killed.
+--echo # So connection con3 gets the backup lock.
+--reap
+BACKUP STAGE END;
+disconnect con3;
+--connection default
+
+CREATE TABLE t_permanent_innodb (col1 INT) ENGINE = InnoDB;
+INSERT INTO t_permanent_innodb SET col1 = 1;
+CREATE TABLE t_permanent_myisam (col1 INT) ENGINE = MyISAM;
+INSERT INTO t_permanent_myisam SET col1 = 1;
+
+# backup : Try the BACKUP STAGE sequence
+# con11  : Try DDL + DML with AUTOCOMMIT = 0
+# con12  : Try DML with AUTOCOMMIT = 1
+--connect(backup,localhost,root,,)
+--connect(con11,localhost,root,,)
+SET AUTOCOMMIT = 0;
+set session lock_wait_timeout=0;
+--connect(con12,localhost,root,,)
+SET AUTOCOMMIT = 1;
+
+--echo # Between (connection default) BACKUP STAGE START and FLUSH
+--echo # no restrictions for concurrent sessions regarding DDL or DML
+--echo # affecting transactional/non transactional permanent tables.
+
+--connection backup
+BACKUP STAGE START;
+--connection con11
+UPDATE t_permanent_innodb SET col1 = 2;
+UPDATE t_permanent_myisam SET col1 = 2;
+SELECT COUNT(*) FROM t_permanent_innodb;
+HANDLER t_permanent_innodb OPEN;
+HANDLER t_permanent_innodb READ FIRST;
+HANDLER t_permanent_innodb CLOSE;
+SELECT COUNT(*) FROM t_permanent_myisam;
+HANDLER t_permanent_myisam OPEN;
+HANDLER t_permanent_myisam READ FIRST;
+HANDLER t_permanent_myisam CLOSE;
+CREATE TABLE t_con1_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t_con1_myisam (col1 INT) ENGINE = InnoDB;
+ALTER TABLE t_permanent_innodb ADD COLUMN col2 INT;
+ALTER TABLE t_permanent_myisam ADD COLUMN col2 INT;
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_innodb;
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_myisam;
+--connection con12
+UPDATE t_permanent_innodb SET col1 = 3;
+UPDATE t_permanent_myisam SET col1 = 3;
+
+--echo # Between (connection default) BACKUP STAGE FLUSH and BLOCK_DDL
+--echo # concurrent sessions
+--echo # - can change transactional permanent tables with DDL and DML
+--echo # - can run DROP/CREATE transactional/non transactional TABLE
+--echo # - cannot modify non transactional permanent tables with DDL or DML
+
+--connection backup
+BACKUP STAGE FLUSH;
+--connection con11
+UPDATE t_permanent_innodb SET col1 = 4;
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR UPDATE t_permanent_myisam SET col1 = 4;
+SELECT COUNT(*) FROM t_permanent_innodb;
+HANDLER t_permanent_innodb OPEN;
+HANDLER t_permanent_innodb READ FIRST;
+HANDLER t_permanent_innodb CLOSE;
+SELECT COUNT(*) FROM t_permanent_myisam;
+HANDLER t_permanent_myisam OPEN;
+HANDLER t_permanent_myisam READ FIRST;
+HANDLER t_permanent_myisam CLOSE;
+DROP TABLE t_con1_innodb;
+DROP TABLE t_con1_myisam;
+CREATE TABLE t_con1_innodb (col1 INT) ENGINE = InnoDB;
+CREATE TABLE t_con1_myisam (col1 INT) ENGINE = InnoDB;
+ALTER TABLE t_permanent_innodb ADD COLUMN col3 INT;
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR ALTER TABLE t_permanent_myisam ADD COLUMN col3 INT;
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_innodb;
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_myisam;
+--connection con12
+UPDATE t_permanent_innodb SET col1 = 5;
+
+
+--echo # Between (connection default) BACKUP STAGE BLOCK_DDL and BLOCK_COMMIT
+--echo # concurrent sessions
+--echo # - can change transactional permanent tables with DML
+--echo # - cannot run DDL
+--echo # - cannot change non transactional permanent tables with DML
+
+--connection backup
+BACKUP STAGE BLOCK_DDL;
+--connection con11
+
+UPDATE t_permanent_innodb SET col1 = 6;
+--error ER_LOCK_WAIT_TIMEOUT
+UPDATE t_permanent_myisam SET col1 = 6;
+SELECT COUNT(*) FROM t_permanent_innodb;
+HANDLER t_permanent_innodb OPEN;
+HANDLER t_permanent_innodb READ FIRST;
+HANDLER t_permanent_innodb CLOSE;
+SELECT COUNT(*) FROM t_permanent_myisam;
+HANDLER t_permanent_myisam OPEN;
+HANDLER t_permanent_myisam READ FIRST;
+HANDLER t_permanent_myisam CLOSE;
+--error ER_LOCK_WAIT_TIMEOUT
+DROP TABLE t_con1_innodb;
+--error ER_LOCK_WAIT_TIMEOUT
+CREATE TABLE throw_away (col1 INT) ENGINE = InnoDB;
+--error ER_LOCK_WAIT_TIMEOUT
+ALTER TABLE t_permanent_innodb ADD COLUMN col4 INT;
+--error ER_LOCK_WAIT_TIMEOUT
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_innodb;
+--error ER_LOCK_WAIT_TIMEOUT
+CREATE OR REPLACE VIEW v_some_view AS SELECT * FROM t_permanent_myisam;
+--connection con12
+UPDATE t_permanent_innodb SET col1 = 7;
+
+
+--echo # Between (connection default) BACKUP STAGE BLOCK_COMMIT and END
+--echo # concurrent sessions
+--echo # - can change transactional permanent tables with DML
+--echo # - cannot run DDL
+--echo # - cannot change non transactional permanent tables with DML
+
+--connection backup
+BACKUP STAGE BLOCK_COMMIT;
+--connection con11
+UPDATE t_permanent_innodb SET col1 = 8;
+--error ER_LOCK_WAIT_TIMEOUT
+UPDATE t_permanent_myisam SET col1 = 8;
+SELECT COUNT(*) FROM t_permanent_innodb;
+HANDLER t_permanent_innodb OPEN;
+HANDLER t_permanent_innodb READ FIRST;
+HANDLER t_permanent_innodb CLOSE;
+SELECT COUNT(*) FROM t_permanent_myisam;
+HANDLER t_permanent_myisam OPEN;
+HANDLER t_permanent_myisam READ FIRST;
+HANDLER t_permanent_myisam CLOSE;
+--error ER_LOCK_WAIT_TIMEOUT
+DROP TABLE t_con1_innodb;
+--error ER_LOCK_WAIT_TIMEOUT
+DROP TABLE t_con1_myisam;
+--connection con12
+--error ER_LOCK_WAIT_TIMEOUT
+SET STATEMENT lock_wait_timeout=0 FOR UPDATE t_permanent_innodb SET col1 = 9;
+
+--connection backup
+BACKUP STAGE END;
+--connection con11
+COMMIT;
+SELECT * FROM t_permanent_innodb ORDER BY col1;
+SELECT * FROM t_permanent_myisam ORDER BY col1;
+SET AUTOCOMMIT = 0;
+SET GLOBAL tx_read_only = 1;
+--connection con12
+BACKUP STAGE START;
+BACKUP STAGE END;
+SET GLOBAL tx_read_only = 0;
+DROP VIEW v_some_view;
+DROP TABLE t_con1_innodb;
+DROP TABLE t_con1_myisam;
+
+--echo # Connection backup holds the backup log and is on some stage.
+--echo # Connection con11 tries to LOCK TABLEs or to set read_only.
+
+--connection backup
+BACKUP STAGE START;
+--connection con11
+--echo # Between BACKUP STAGE START and FLUSH:
+--echo # No restrictions for other connection around LOCK TABLES or read-only.
+LOCK TABLES t_permanent_innodb READ;
+LOCK TABLES t_permanent_myisam READ;
+LOCK TABLES t_permanent_innodb WRITE;
+LOCK TABLES t_permanent_myisam WRITE;
+UNLOCK TABLES;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+--connection backup
+BACKUP STAGE FLUSH;
+--connection con11
+--echo # Between BACKUP STAGE FLUSH and BLOCK_COMMIT:
+--echo # Connection con11 not holding the backup lock cannot
+--echo # LOCK WRITE non transactional table.
+LOCK TABLES t_permanent_innodb READ;
+LOCK TABLES t_permanent_myisam READ;
+LOCK TABLES t_permanent_innodb WRITE;
+--error ER_LOCK_WAIT_TIMEOUT
+LOCK TABLES t_permanent_myisam WRITE;
+UNLOCK TABLES;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+--connection backup
+BACKUP STAGE BLOCK_DDL;
+--connection con11
+--echo # Between BACKUP STAGE FLUSH and BLOCK_COMMIT:
+--echo # Connection con11 not holding the backup lock cannot
+--echo # LOCK WRITE transactional or non transactional table.
+LOCK TABLES t_permanent_innodb READ;
+LOCK TABLES t_permanent_myisam READ;
+--error ER_LOCK_WAIT_TIMEOUT
+LOCK TABLES t_permanent_innodb WRITE;
+--error ER_LOCK_WAIT_TIMEOUT
+LOCK TABLES t_permanent_myisam WRITE;
+UNLOCK TABLES;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+--connection backup
+BACKUP STAGE BLOCK_COMMIT;
+--connection con11
+--echo # Between BACKUP BLOCK_COMMIT FLUSH and END:
+--echo # Connection con11 not holding the backup lock cannot
+--echo # LOCK WRITE transactional or non transactional table.
+LOCK TABLES t_permanent_innodb READ;
+LOCK TABLES t_permanent_myisam READ;
+--error ER_LOCK_WAIT_TIMEOUT
+LOCK TABLES t_permanent_innodb WRITE;
+--error ER_LOCK_WAIT_TIMEOUT
+LOCK TABLES t_permanent_myisam WRITE;
+UNLOCK TABLES;
+SET GLOBAL tx_read_only = 1;
+SET GLOBAL tx_read_only = 0;
+--connection backup
+BACKUP STAGE END;
+
+DROP TABLE t_permanent_innodb;
+DROP TABLE t_permanent_myisam;
+
+--echo #
+--echo # Log tables
+--echo #
+
+--connection backup
+
+SET @old_general_log = @@general_log;
+SET @old_slow_query_log = @@slow_query_log;
+SET @old_log_output = @@log_output;
+
+SET GLOBAL log_output = 'TABLE';
+SET GLOBAL general_log = ON;
+SET GLOBAL slow_query_log = ON;
+
+
+--connection con11
+SET @old_long_query_time = @@SESSION.long_query_time;
+SET SESSION long_query_time = 0;
+
+--connection backup
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+
+--connection con11
+SELECT 1;
+
+--connection backup
+SELECT 1;
+
+--connection con11
+SET SESSION long_query_time = @old_long_query_time;
+
+--connection backup
+BACKUP STAGE END;
+
+SET GLOBAL log_output = @old_log_output;
+SET GLOBAL slow_query_log = @old_slow_query_log;
+SET GLOBAL general_log = @old_general_log;
+
+--echo #-----------------------------------------------------------------------
+--echo # Cleanup
+--echo #-----------------------------------------------------------------------
+
+--replace_result $old_lock_wait_timeout <old_lock_wait_timeout>
+eval
+SET GLOBAL lock_wait_timeout = $old_lock_wait_timeout;
+
+--disconnect con2
+--disconnect con11
+--disconnect con12
+--disconnect backup
+
+--connection default
+--source include/wait_until_count_sessions.inc
diff --git a/mysql-test/main/backup_syntax.result b/mysql-test/main/backup_syntax.result
new file mode 100644
index 00000000000..f8f623cda0c
--- /dev/null
+++ b/mysql-test/main/backup_syntax.result
@@ -0,0 +1,163 @@
+#-----------------------------------------------------------------------
+# Basic syntax checks
+#-----------------------------------------------------------------------
+# Check existing BACKUP STAGE statements in the sequence to be used.
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+# Check invalid variants of BACKUP .... syntax.
+BACKUP LOG;
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'LOG' at line 1
+BACKUP LOCK;
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'LOCK' at line 1
+BACKUP STAGE;
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near '' at line 1
+BACKUP STAGE LOCK;
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'LOCK' at line 1
+BACKUP STAGE not_existing;
+ERROR HY000: Unknown backup stage: 'not_existing'. Stage should be one of START, FLUSH, BLOCK_DDL, BLOCK_COMMIT or END
+#-----------------------------------------------------------------------
+# BACKUP STAGE statements in various orders.
+#-----------------------------------------------------------------------
+# All BACKUP STAGE statements != 'BACKUP STAGE START' expect that a
+# backup lock (generated by BACKUP STAGE START) already exists.
+#
+backup stage start;
+backup stage flush;
+backup stage start;
+ERROR HY000: Backup stage 'START' is same or before current backup stage 'FLUSH'
+backup stage start;
+ERROR HY000: Backup stage 'START' is same or before current backup stage 'FLUSH'
+backup stage block_commit;
+backup stage flush;
+ERROR HY000: Backup stage 'FLUSH' is same or before current backup stage 'BLOCK_COMMIT'
+backup stage flush;
+ERROR HY000: Backup stage 'FLUSH' is same or before current backup stage 'BLOCK_COMMIT'
+backup stage end;
+backup stage flush;
+ERROR HY000: You must start backup with "BACKUP STAGE START"
+BACKUP STAGE END;
+ERROR HY000: You must start backup with "BACKUP STAGE START"
+BACKUP STAGE BLOCK_COMMIT;
+ERROR HY000: You must start backup with "BACKUP STAGE START"
+BACKUP STAGE BLOCK_DDL;
+ERROR HY000: You must start backup with "BACKUP STAGE START"
+BACKUP STAGE FLUSH;
+ERROR HY000: You must start backup with "BACKUP STAGE START"
+# Ordered "give up" with 'BACKUP STAGE END' because of whatever reason.
+# Some existing backup lock assumed a 'BACKUP STAGE END' is allowed in
+# every situation.
+BACKUP STAGE START;
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE END;
+# Orders with BACKUP STAGE FLUSH omitted.
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+# Orders with BACKUP STAGE BLOCK_DDL omitted.
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+# Orders with BACKUP STAGE BLOCK_COMMIT omitted.
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE END;
+# Orders with doubled BACKUP STAGE statements.
+# We get an error but that seems to have no bad impact on the state.
+# And so we are allowed to go on with BACKUP STAGE statements.
+BACKUP STAGE START;
+BACKUP STAGE START;
+ERROR HY000: Backup stage 'START' is same or before current backup stage 'START'
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE FLUSH;
+ERROR HY000: Backup stage 'FLUSH' is same or before current backup stage 'FLUSH'
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_DDL;
+ERROR HY000: Backup stage 'BLOCK_DDL' is same or before current backup stage 'BLOCK_DDL'
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE BLOCK_COMMIT;
+ERROR HY000: Backup stage 'BLOCK_COMMIT' is same or before current backup stage 'BLOCK_COMMIT'
+BACKUP STAGE END;
+# Scrambled orders.
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE START;
+ERROR HY000: Backup stage 'START' is same or before current backup stage 'FLUSH'
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE START;
+ERROR HY000: Backup stage 'START' is same or before current backup stage 'BLOCK_DDL'
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE START;
+ERROR HY000: Backup stage 'START' is same or before current backup stage 'BLOCK_COMMIT'
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE FLUSH;
+ERROR HY000: Backup stage 'FLUSH' is same or before current backup stage 'BLOCK_DDL'
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE FLUSH;
+ERROR HY000: Backup stage 'FLUSH' is same or before current backup stage 'BLOCK_COMMIT'
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE BLOCK_DDL;
+ERROR HY000: Backup stage 'BLOCK_DDL' is same or before current backup stage 'BLOCK_COMMIT'
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE FLUSH;
+ERROR HY000: Backup stage 'FLUSH' is same or before current backup stage 'BLOCK_DDL'
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+#----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE BLOCK_DDL;
+ERROR HY000: Backup stage 'BLOCK_DDL' is same or before current backup stage 'BLOCK_COMMIT'
+BACKUP STAGE END;
+#
+# Check Oracle syntax
+#
+set SQL_MODE=Oracle;
+backup stage start;
+backup stage end;
+set SQL_MODE=default;
diff --git a/mysql-test/main/backup_syntax.test b/mysql-test/main/backup_syntax.test
new file mode 100644
index 00000000000..f02c69bdd85
--- /dev/null
+++ b/mysql-test/main/backup_syntax.test
@@ -0,0 +1,181 @@
+########################################################################
+# Tests things releated to syntax of BACKUP STAGE (MDEV-5336)
+########################################################################
+
+--source include/have_innodb.inc
+--source include/have_metadata_lock_info.inc
+
+--echo #-----------------------------------------------------------------------
+--echo # Basic syntax checks
+--echo #-----------------------------------------------------------------------
+
+--echo # Check existing BACKUP STAGE statements in the sequence to be used.
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+
+--echo # Check invalid variants of BACKUP .... syntax.
+--error ER_PARSE_ERROR
+BACKUP LOG;
+--error ER_PARSE_ERROR
+BACKUP LOCK;
+--error ER_PARSE_ERROR
+BACKUP STAGE;
+--error ER_PARSE_ERROR
+BACKUP STAGE LOCK;
+--error ER_BACKUP_UNKNOWN_STAGE
+BACKUP STAGE not_existing;
+
+--echo #-----------------------------------------------------------------------
+--echo # BACKUP STAGE statements in various orders.
+--echo #-----------------------------------------------------------------------
+--echo # All BACKUP STAGE statements != 'BACKUP STAGE START' expect that a
+--echo # backup lock (generated by BACKUP STAGE START) already exists.
+--echo #
+
+backup stage start;
+backup stage flush;
+--error ER_BACKUP_WRONG_STAGE
+backup stage start;
+--error ER_BACKUP_WRONG_STAGE
+backup stage start;
+backup stage block_commit;
+--error ER_BACKUP_WRONG_STAGE
+backup stage flush;
+--error ER_BACKUP_WRONG_STAGE
+backup stage flush;
+backup stage end;
+--error ER_BACKUP_NOT_RUNNING
+backup stage flush;
+--error ER_BACKUP_NOT_RUNNING
+BACKUP STAGE END;
+--error ER_BACKUP_NOT_RUNNING
+BACKUP STAGE BLOCK_COMMIT;
+--error ER_BACKUP_NOT_RUNNING
+BACKUP STAGE BLOCK_DDL;
+--error ER_BACKUP_NOT_RUNNING
+BACKUP STAGE FLUSH;
+
+--echo # Ordered "give up" with 'BACKUP STAGE END' because of whatever reason.
+--echo # Some existing backup lock assumed a 'BACKUP STAGE END' is allowed in
+--echo # every situation.
+
+BACKUP STAGE START;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE END;
+--echo # Orders with BACKUP STAGE FLUSH omitted.
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+--echo # Orders with BACKUP STAGE BLOCK_DDL omitted.
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+--echo # Orders with BACKUP STAGE BLOCK_COMMIT omitted.
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE END;
+
+--echo # Orders with doubled BACKUP STAGE statements.
+--echo # We get an error but that seems to have no bad impact on the state.
+--echo # And so we are allowed to go on with BACKUP STAGE statements.
+
+BACKUP STAGE START;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE FLUSH;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+
+--echo # Scrambled orders.
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_DDL;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_COMMIT;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE START;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_COMMIT;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE FLUSH;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE BLOCK_COMMIT;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE BLOCK_DDL;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_COMMIT;
+BACKUP STAGE END;
+--echo #----
+BACKUP STAGE START;
+BACKUP STAGE FLUSH;
+BACKUP STAGE BLOCK_COMMIT;
+--error ER_BACKUP_WRONG_STAGE
+BACKUP STAGE BLOCK_DDL;
+BACKUP STAGE END;
+
+--echo #
+--echo # Check Oracle syntax
+--echo #
+
+set SQL_MODE=Oracle;
+backup stage start;
+backup stage end;
+set SQL_MODE=default;
diff --git a/mysql-test/main/deprecated_features.result b/mysql-test/main/deprecated_features.result
index fc6c86d065d..2c77d745e2e 100644
--- a/mysql-test/main/deprecated_features.result
+++ b/mysql-test/main/deprecated_features.result
@@ -5,7 +5,7 @@ ERROR HY000: Unknown system variable 'table_type'
 select @@table_type='MyISAM';
 ERROR HY000: Unknown system variable 'table_type'
 backup table t1 to 'data.txt';
-ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'backup table t1 to 'data.txt'' at line 1
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'table t1 to 'data.txt'' at line 1
 restore table t1 from 'data.txt';
 ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'restore table t1 from 'data.txt'' at line 1
 show plugin;
diff --git a/mysql-test/main/mysqld--help.result b/mysql-test/main/mysqld--help.result
index 1e6cdb3ec2c..8faf332a7dd 100644
--- a/mysql-test/main/mysqld--help.result
+++ b/mysql-test/main/mysqld--help.result
@@ -1595,7 +1595,7 @@ performance-schema-max-rwlock-instances -1
 performance-schema-max-socket-classes 10
 performance-schema-max-socket-instances -1
 performance-schema-max-stage-classes 160
-performance-schema-max-statement-classes 200
+performance-schema-max-statement-classes 201
 performance-schema-max-table-handles -1
 performance-schema-max-table-instances -1
 performance-schema-max-thread-classes 50
diff --git a/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result b/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result
index 3efdcca8e44..1dac71788bd 100644
--- a/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result
+++ b/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result
@@ -3066,9 +3066,9 @@ READ_ONLY	YES
 COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	PERFORMANCE_SCHEMA_MAX_STATEMENT_CLASSES
 SESSION_VALUE	NULL
-GLOBAL_VALUE	200
+GLOBAL_VALUE	201
 GLOBAL_VALUE_ORIGIN	COMPILE-TIME
-DEFAULT_VALUE	200
+DEFAULT_VALUE	201
 VARIABLE_SCOPE	GLOBAL
 VARIABLE_TYPE	BIGINT UNSIGNED
 VARIABLE_COMMENT	Maximum number of statement instruments.
diff --git a/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result b/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
index e49e96e3587..4c9b38c48b7 100644
--- a/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
+++ b/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
@@ -3290,9 +3290,9 @@ READ_ONLY	YES
 COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	PERFORMANCE_SCHEMA_MAX_STATEMENT_CLASSES
 SESSION_VALUE	NULL
-GLOBAL_VALUE	200
+GLOBAL_VALUE	201
 GLOBAL_VALUE_ORIGIN	COMPILE-TIME
-DEFAULT_VALUE	200
+DEFAULT_VALUE	201
 VARIABLE_SCOPE	GLOBAL
 VARIABLE_TYPE	BIGINT UNSIGNED
 VARIABLE_COMMENT	Maximum number of statement instruments.
diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index 08457214ea5..697b794f39f 100644
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -136,7 +136,7 @@ SET (SQL_SOURCE
                opt_split.cc
 	       ${WSREP_SOURCES}
                table_cache.cc encryption.cc temporary_tables.cc
-               proxy_protocol.cc
+               proxy_protocol.cc backup.cc
                ${CMAKE_CURRENT_BINARY_DIR}/sql_builtin.cc
                ${CMAKE_CURRENT_BINARY_DIR}/sql_yacc.cc
                ${CMAKE_CURRENT_BINARY_DIR}/sql_yacc_ora.cc
diff --git a/sql/backup.cc b/sql/backup.cc
new file mode 100644
index 00000000000..e022a7c2c04
--- /dev/null
+++ b/sql/backup.cc
@@ -0,0 +1,356 @@
+/* Copyright (c) 2018, MariaDB Corporation
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+/*
+  Implementation of BACKUP STAGE, an interface for external backup tools.
+
+  TODO:
+  - At backup_start() we call ha_prepare_for_backup() for all active
+    storage engines.  If someone tries to load a new storage engine
+    that requires prepare_for_backup() for it to work, that storage
+    engines has to be blocked from loading until backup finishes.
+    As we currently don't have any loadable storage engine that
+    requires this and we have not implemented that part.
+    This can easily be done by adding a
+    PLUGIN_CANT_BE_LOADED_WHILE_BACKUP_IS_RUNNING flag to
+    maria_declare_plugin and check this before calling
+    plugin_initialize()
+*/
+
+#include "mariadb.h"
+#include "sql_class.h"
+#include "sql_base.h"                           // flush_tables
+#include "sql_insert.h"                         // kill_delayed_threads
+#include "sql_handler.h"                        // mysql_ha_cleanup_no_free
+#include <my_sys.h>
+
+static const char *stage_names[]=
+{"START", "FLUSH", "BLOCK_DDL", "BLOCK_COMMIT", "END", 0};
+
+TYPELIB backup_stage_names=
+{ array_elements(stage_names)-1, "", stage_names, 0 };
+
+static MDL_ticket *backup_flush_ticket;
+
+static bool backup_start(THD *thd);
+static bool backup_flush(THD *thd);
+static bool backup_block_ddl(THD *thd);
+static bool backup_block_commit(THD *thd);
+
+/**
+  Run next stage of backup
+*/
+
+void backup_init()
+{
+  backup_flush_ticket= 0;
+}
+
+bool run_backup_stage(THD *thd, backup_stages stage)
+{
+  backup_stages next_stage;
+  DBUG_ENTER("run_backup_stage");
+
+  if (thd->current_backup_stage == BACKUP_FINISHED)
+  {
+    if (stage != BACKUP_START)
+    {
+      my_error(ER_BACKUP_NOT_RUNNING, MYF(0));
+      DBUG_RETURN(1);
+    }
+    next_stage= BACKUP_START;
+  }
+  else
+  {
+    if ((uint) thd->current_backup_stage >= (uint) stage)
+    {
+      my_error(ER_BACKUP_WRONG_STAGE, MYF(0), stage_names[stage],
+               stage_names[thd->current_backup_stage]);
+      DBUG_RETURN(1);
+    }
+    if (stage == BACKUP_END)
+    {
+      /*
+        If end is given, jump directly to stage end. This is to allow one
+        to abort backup quickly.
+      */
+      next_stage= stage;
+    }
+    else
+    {
+      /* Go trough all not used stages until we reach 'stage' */
+      next_stage= (backup_stages) ((uint) thd->current_backup_stage + 1);
+    }
+  }
+
+  do
+  {
+    bool res;
+    backup_stages previous_stage= thd->current_backup_stage;
+    thd->current_backup_stage= next_stage;
+    switch (next_stage) {
+    case BACKUP_START:
+      if (!(res= backup_start(thd)))
+        break;
+      /* Reset backup stage to start for next backup try */
+      previous_stage= BACKUP_FINISHED;
+      break;
+    case BACKUP_FLUSH:
+      res= backup_flush(thd);
+      break;
+    case BACKUP_WAIT_FOR_FLUSH:
+      res= backup_block_ddl(thd);
+      break;
+    case BACKUP_LOCK_COMMIT:
+      res= backup_block_commit(thd);
+      break;
+    case BACKUP_END:
+      res= backup_end(thd);
+      break;
+    case BACKUP_FINISHED:
+      DBUG_ASSERT(0);
+      res= 0;
+    }
+    if (res)
+    {
+      thd->current_backup_stage= previous_stage;
+      my_error(ER_BACKUP_STAGE_FAILED, MYF(0), stage_names[(uint) stage]);
+      DBUG_RETURN(1);
+    }
+    next_stage= (backup_stages) ((uint) next_stage + 1);
+  } while ((uint) next_stage <= (uint) stage);
+
+  DBUG_RETURN(0);
+}
+
+
+/**
+  Start the backup
+
+  - Wait for previous backup to stop running
+  - Start service to log changed tables (TODO)
+  - Block purge of redo files (Required at least for Aria)
+  - An handler can optionally do a checkpoint of all tables,
+    to speed up the recovery stage of the backup.
+*/
+
+static bool backup_start(THD *thd)
+{
+  MDL_request mdl_request;
+  DBUG_ENTER("backup_start");
+
+  thd->current_backup_stage= BACKUP_FINISHED;   // For next test
+  if (thd->has_read_only_protection())
+    DBUG_RETURN(1);
+  thd->current_backup_stage= BACKUP_START;
+
+  if (thd->locked_tables_mode)
+  {
+    my_error(ER_LOCK_OR_ACTIVE_TRANSACTION, MYF(0));
+    DBUG_RETURN(1);
+  }
+
+  mdl_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_START, MDL_EXPLICIT);
+  if (thd->mdl_context.acquire_lock(&mdl_request,
+                                    thd->variables.lock_wait_timeout))
+    DBUG_RETURN(1);
+
+  backup_flush_ticket= mdl_request.ticket;
+
+  ha_prepare_for_backup();
+  DBUG_RETURN(0);
+}
+
+/**
+   backup_flush()
+
+   - FLUSH all changes for not active non transactional tables, except
+     for statistics and log tables. Close the tables, to ensure they
+     are marked as closed after backup.
+
+   - BLOCK all NEW write locks for all non transactional tables
+     (except statistics and log tables).  Already granted locks are
+     not affected (Running statements with non transaction tables will
+     continue running).
+
+   - The following DDL's doesn't have to be blocked as they can't set
+     the table in a non consistent state:
+     CREATE, RENAME, DROP
+*/
+
+static bool backup_flush(THD *thd)
+{
+  DBUG_ENTER("backup_flush");
+  /*
+    Lock all non transactional normal tables to be used in new DML's
+  */
+  if (thd->mdl_context.upgrade_shared_lock(backup_flush_ticket,
+                                           MDL_BACKUP_FLUSH,
+                                           thd->variables.lock_wait_timeout))
+    DBUG_RETURN(1);
+
+  /*
+    Free unused tables and table shares so that mariabackup knows what
+    is safe to copy
+  */
+  tc_purge(false);
+  tdc_purge(true);
+
+  DBUG_RETURN(0);
+}
+
+/**
+  backup_block_ddl()
+
+  - Kill all insert delay handlers, to ensure that all non transactional
+    tables are closed (can be improved in the future).
+
+  - Close handlers as other threads may wait for these, which can cause deadlocks.
+
+  - Wait for all statements using write locked non-transactional tables to end.
+
+  - Mark all not used active non transactional tables (except
+    statistics and log tables) to be closed with
+    handler->extra(HA_EXTRA_FLUSH)
+
+  - Block TRUNCATE TABLE, CREATE TABLE, DROP TABLE and RENAME
+    TABLE. Block also start of a new ALTER TABLE and the final rename
+    phase of ALTER TABLE.  Running ALTER TABLES are not blocked.  Both normal
+    and inline ALTER TABLE'S should be blocked when copying is completed but
+    before final renaming of the tables / new table is activated.
+    This will probably require a callback from the InnoDB code.
+*/
+
+static bool backup_block_ddl(THD *thd)
+{
+  DBUG_ENTER("backup_block_ddl");
+
+  kill_delayed_threads();
+  mysql_ha_cleanup_no_free(thd);
+
+  /* Wait until all non trans statements has ended */
+  if (thd->mdl_context.upgrade_shared_lock(backup_flush_ticket,
+                                           MDL_BACKUP_WAIT_FLUSH,
+                                           thd->variables.lock_wait_timeout))
+    DBUG_RETURN(1);
+
+  /*
+    Remove not used tables from the table share.  Flush all changes to
+    non transaction tables and mark those that are not in use in write
+    operations as closed. From backup purposes it's not critical if
+    flush_tables() returns an error. It's ok to continue with next
+    backup stage even if we got an error.
+  */
+  (void) flush_tables(thd, FLUSH_NON_TRANS_TABLES);
+
+  /*
+    block new DDL's, in addition to all previous blocks
+    We didn't do this lock above, as we wanted DDL's to be executed while
+    we wait for non transactional tables (which may take a while).
+ */
+  if (thd->mdl_context.upgrade_shared_lock(backup_flush_ticket,
+                                           MDL_BACKUP_WAIT_DDL,
+                                           thd->variables.lock_wait_timeout))
+  {
+    /*
+      Could be a timeout. Downgrade lock to what is was before this function
+      was called so that this function can be called again
+    */
+    backup_flush_ticket->downgrade_lock(MDL_BACKUP_FLUSH);
+    DBUG_RETURN(1);
+  }
+  DBUG_RETURN(0);
+}
+
+/**
+   backup_block_commit()
+
+   Block commits, writes to log and statistics tables and binary log
+*/
+
+static bool backup_block_commit(THD *thd)
+{
+  DBUG_ENTER("backup_block_commit");
+  if (thd->mdl_context.upgrade_shared_lock(backup_flush_ticket,
+                                           MDL_BACKUP_WAIT_COMMIT,
+                                           thd->variables.lock_wait_timeout))
+    DBUG_RETURN(1);
+  flush_tables(thd, FLUSH_SYS_TABLES);
+  DBUG_RETURN(0);
+}
+
+/**
+   backup_end()
+
+   Safe to run, even if backup has not been run by this thread.
+   This is for example the case when a THD ends.
+*/
+
+bool backup_end(THD *thd)
+{
+  DBUG_ENTER("backup_end");
+
+  if (thd->current_backup_stage != BACKUP_FINISHED)
+  {
+    ha_end_backup();
+    thd->current_backup_stage= BACKUP_FINISHED;
+    thd->mdl_context.release_lock(backup_flush_ticket);
+  }
+  DBUG_RETURN(0);
+}
+
+
+/**
+   backup_set_alter_copy_lock()
+
+   @param thd
+   @param table  From table that is part of ALTER TABLE. This is only used
+                 for the assert to ensure we use this function correctly.
+
+   Downgrades the MDL_BACKUP_DDL lock to MDL_BACKUP_ALTER_COPY to allow
+   copy of altered table to proceed under MDL_BACKUP_WAIT_DDL
+
+   Note that in some case when using non transactional tables,
+   the lock may be of type MDL_BACKUP_DML.
+*/
+
+void backup_set_alter_copy_lock(THD *thd, TABLE *table)
+{
+  MDL_ticket *ticket= thd->mdl_backup_ticket;
+
+  /* Ticket maybe NULL in case of LOCK TABLES or for temporary tables*/
+  DBUG_ASSERT(ticket || thd->locked_tables_mode ||
+              table->s->tmp_table != NO_TMP_TABLE);
+  if (ticket)
+    ticket->downgrade_lock(MDL_BACKUP_ALTER_COPY);
+}
+
+/**
+   backup_reset_alter_copy_lock
+
+   Upgrade the lock of the original ALTER table MDL_BACKUP_DDL
+   Can fail if MDL lock was killed
+*/
+
+bool backup_reset_alter_copy_lock(THD *thd)
+{
+  bool res= 0;
+  MDL_ticket *ticket= thd->mdl_backup_ticket;
+
+  /* Ticket maybe NULL in case of LOCK TABLES or for temporary tables*/
+  if (ticket)
+    res= thd->mdl_context.upgrade_shared_lock(ticket, MDL_BACKUP_DDL,
+                                              thd->variables.lock_wait_timeout);
+  return res;
+}
diff --git a/sql/backup.h b/sql/backup.h
new file mode 100644
index 00000000000..e6f290ae9c8
--- /dev/null
+++ b/sql/backup.h
@@ -0,0 +1,31 @@
+#ifndef BACKUP_INCLUDED
+#define BACKUP_INCLUDED
+/* Copyright (c) 2018, MariaDB Corporation
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+enum backup_stages
+{
+  BACKUP_START, BACKUP_FLUSH, BACKUP_WAIT_FOR_FLUSH, BACKUP_LOCK_COMMIT,
+  BACKUP_END, BACKUP_FINISHED
+};
+
+extern TYPELIB backup_stage_names;
+
+void backup_init();
+bool run_backup_stage(THD *thd, backup_stages stage);
+bool backup_end(THD *thd);
+void backup_set_alter_copy_lock(THD *thd, TABLE *altered_table);
+bool backup_reset_alter_copy_lock(THD *thd);
+#endif /* BACKUP_INCLUDED */
diff --git a/sql/lex.h b/sql/lex.h
index da20468d593..5ffe07fa415 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -589,6 +589,7 @@ static SYMBOL symbols[] = {
   { "SONAME",		SYM(SONAME_SYM)},
   { "SOUNDS",		SYM(SOUNDS_SYM)},
   { "SOURCE",           SYM(SOURCE_SYM)},
+  { "STAGE",            SYM(STAGE_SYM)},
   { "STORED",           SYM(STORED_SYM)},
   { "SPATIAL",		SYM(SPATIAL_SYM)},
   { "SPECIFIC",         SYM(SPECIFIC_SYM)},
diff --git a/sql/lock.cc b/sql/lock.cc
index f6a4ea00a33..1564059bb20 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -861,7 +861,7 @@ bool lock_schema_name(THD *thd, const char *db)
     return TRUE;
   }
 
-  if (thd->global_read_lock.can_acquire_protection())
+  if (thd->has_read_only_protection())
     return TRUE;
   global_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_DDL, MDL_STATEMENT);
   mdl_request.init(MDL_key::SCHEMA, db, "", MDL_EXCLUSIVE, MDL_TRANSACTION);
@@ -919,7 +919,7 @@ bool lock_object_name(THD *thd, MDL_key::enum_mdl_namespace mdl_type,
   DBUG_ASSERT(name);
   DEBUG_SYNC(thd, "before_wait_locked_pname");
 
-  if (thd->global_read_lock.can_acquire_protection())
+  if (thd->has_read_only_protection())
     return TRUE;
   global_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_DDL, MDL_STATEMENT);
   schema_request.init(MDL_key::SCHEMA, db, "", MDL_INTENTION_EXCLUSIVE,
@@ -1028,6 +1028,12 @@ bool Global_read_lock::lock_global_read_lock(THD *thd)
     MDL_request mdl_request;
     bool result;
 
+    if (thd->current_backup_stage != BACKUP_FINISHED)
+    {
+      my_error(ER_BACKUP_LOCK_IS_ACTIVE, MYF(0));
+      DBUG_RETURN(1);
+    }
+
     mysql_ha_cleanup_no_free(thd);
 
     DBUG_ASSERT(! thd->mdl_context.is_lock_owner(MDL_key::BACKUP, "", "",
diff --git a/sql/mdl.cc b/sql/mdl.cc
index 40ef4c04676..c06a212ed19 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -1627,7 +1627,7 @@ MDL_lock::MDL_object_lock::m_waiting_incompatible[MDL_TYPE_END]=
      Request  |               Pending backup lock                       |
       type    | S0  S1  S2  S3  S4  F1  F2   D  TD  SD   DD  BL  AC  C  |
     ----------+---------------------------------------------------------+
-    S0        |  -   -   -   -   -   +   +   +   +   +   +   +   +   +  |
+    S0        |  +   -   -   -   -   +   +   +   +   +   +   +   +   +  |
     S1        |  +   +   +   +   +   +   +   +   +   +   +   +   +   +  |
     S2        |  +   +   +   +   +   +   +   +   +   +   +   +   +   +  |
     S3        |  +   +   +   +   +   +   +   +   +   +   +   +   +   +  |
@@ -1682,7 +1682,7 @@ const MDL_lock::bitmap_t
 MDL_lock::MDL_backup_lock::m_waiting_incompatible[MDL_BACKUP_END]=
 {
   /* MDL_BACKUP_START */
-  MDL_BIT(MDL_BACKUP_START) | MDL_BIT(MDL_BACKUP_FLUSH) | MDL_BIT(MDL_BACKUP_WAIT_FLUSH) | MDL_BIT(MDL_BACKUP_WAIT_DDL) | MDL_BIT(MDL_BACKUP_WAIT_COMMIT),
+  MDL_BIT(MDL_BACKUP_FLUSH) | MDL_BIT(MDL_BACKUP_WAIT_FLUSH) | MDL_BIT(MDL_BACKUP_WAIT_DDL) | MDL_BIT(MDL_BACKUP_WAIT_COMMIT),
   0,
   0,
   0,
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 0f116395fe0..d4a1042c230 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -3667,6 +3667,7 @@ SHOW_VAR com_status_vars[]= {
   {"alter_user",           STMT_STATUS(SQLCOM_ALTER_USER)},
   {"analyze",              STMT_STATUS(SQLCOM_ANALYZE)},
   {"assign_to_keycache",   STMT_STATUS(SQLCOM_ASSIGN_TO_KEYCACHE)},
+  {"backup",               STMT_STATUS(SQLCOM_BACKUP)},
   {"begin",                STMT_STATUS(SQLCOM_BEGIN)},
   {"binlog",               STMT_STATUS(SQLCOM_BINLOG_BASE64_EVENT)},
   {"call_procedure",       STMT_STATUS(SQLCOM_CALL)},
@@ -4935,6 +4936,7 @@ static int init_server_components()
   my_rnd_init(&sql_rand,(ulong) server_start_time,(ulong) server_start_time/2);
   setup_fpu();
   init_thr_lock();
+  backup_init();
 
 #ifndef EMBEDDED_LIBRARY
   if (init_thr_timer(thread_scheduler->max_threads + extra_max_connections))
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index e73666cfb58..02b06c95453 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7921,3 +7921,13 @@ ER_KEY_DOESNT_SUPPORT
         eng "%s index %`s does not support this operation"
 ER_ALTER_OPERATION_TABLE_OPTIONS_NEED_REBUILD
 	eng "Changing table options requires the table to be rebuilt"
+ER_BACKUP_LOCK_IS_ACTIVE
+        eng "Can't execute the command as you have a BACKUP STAGE active"
+ER_BACKUP_NOT_RUNNING
+        eng "You must start backup with \"BACKUP STAGE START\""
+ER_BACKUP_WRONG_STAGE
+        eng "Backup stage '%s' is same or before current backup stage '%s'"
+ER_BACKUP_STAGE_FAILED
+        eng "Backup stage '%s' failed"
+ER_BACKUP_UNKNOWN_STAGE
+        eng "Unknown backup stage: '%s'. Stage should be one of START, FLUSH, BLOCK_DDL, BLOCK_COMMIT or END"
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index d6b9e967f63..45dea349af4 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2100,7 +2100,7 @@ retry_share:
       MDL_request protection_request;
       MDL_deadlock_handler mdl_deadlock_handler(ot_ctx);
 
-      if (thd->global_read_lock.can_acquire_protection())
+      if (thd->has_read_only_protection())
       {
         MYSQL_UNBIND_TABLE(table->file);
         tc_release_table(table);
@@ -3978,7 +3978,7 @@ lock_table_names(THD *thd, const DDL_options_st &options,
       by acquiring global intention exclusive lock with statement
       duration.
     */
-    if (thd->global_read_lock.can_acquire_protection())
+    if (thd->has_read_only_protection())
       DBUG_RETURN(TRUE);
     global_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_DDL,
                         MDL_STATEMENT);
@@ -3997,6 +3997,8 @@ lock_table_names(THD *thd, const DDL_options_st &options,
     if (create_table)
       thd->push_internal_handler(&error_handler);  // Avoid warnings & errors
     bool res= thd->mdl_context.acquire_locks(&mdl_requests, lock_wait_timeout);
+    if (!(flags & MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK))
+      thd->mdl_backup_ticket= global_request.ticket;
     if (create_table)
       thd->pop_internal_handler();
     if (!res)
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 52ebc186b1a..8a8d637815b 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1219,6 +1219,7 @@ void THD::init(bool skip_lock)
   first_successful_insert_id_in_prev_stmt= 0;
   first_successful_insert_id_in_prev_stmt_for_binlog= 0;
   first_successful_insert_id_in_cur_stmt= 0;
+  current_backup_stage= BACKUP_FINISHED;
 #ifdef WITH_WSREP
   wsrep_exec_mode= wsrep_applier ? REPL_RECV :  LOCAL_STATE;
   wsrep_conflict_state= NO_CONFLICT;
@@ -1487,6 +1488,7 @@ void THD::cleanup(void)
   */
   mdl_context.release_transactional_locks();
 
+  backup_end(this);
   /* Release the global read lock, if acquired. */
   if (global_read_lock.is_acquired())
     global_read_lock.unlock_global_read_lock(this);
@@ -2494,6 +2496,16 @@ void THD::update_charset()
                               &not_used);
 }
 
+void THD::give_protection_error()
+{
+  if (current_backup_stage != BACKUP_FINISHED)
+    my_error(ER_BACKUP_LOCK_IS_ACTIVE, MYF(0));
+  else
+  {
+    DBUG_ASSERT(global_read_lock.is_acquired());
+    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
+  }
+}
 
 /* routings to adding tables to list of changed in transaction tables */
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 38ff66fe1b5..49d34e3bdca 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -38,15 +38,14 @@
 #include "thr_timer.h"
 #include "thr_malloc.h"
 #include "log_slow.h"      /* LOG_SLOW_DISABLE_... */
-
 #include "sql_digest_stream.h"            // sql_digest_state
-
 #include <mysql/psi/mysql_stage.h>
 #include <mysql/psi/mysql_statement.h>
 #include <mysql/psi/mysql_idle.h>
 #include <mysql/psi/mysql_table.h>
 #include <mysql_com_server.h>
 #include "session_tracker.h"
+#include "backup.h"
 
 extern "C"
 void set_thd_stage_info(void *thd,
@@ -1958,19 +1957,6 @@ public:
 
   bool lock_global_read_lock(THD *thd);
   void unlock_global_read_lock(THD *thd);
-  /**
-    Check if this connection can acquire protection against GRL and
-    emit error if otherwise.
-  */
-  bool can_acquire_protection() const
-  {
-    if (m_state)
-    {
-      my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
-      return TRUE;
-    }
-    return FALSE;
-  }
   bool make_global_read_lock_block_commit(THD *thd);
   bool is_acquired() const { return m_state != GRL_NONE; }
   void set_explicit_lock_duration(THD *thd);
@@ -2193,6 +2179,7 @@ public:
     rpl_io_thread_info *rpl_io_info;
     rpl_sql_thread_info *rpl_sql_info;
   } system_thread_info;
+  MDL_ticket *mdl_backup_ticket;
 
   void reset_for_next_command(bool do_clear_errors= 1);
   /*
@@ -2978,6 +2965,7 @@ public:
   uint	     tmp_table, global_disable_checkpoint;
   uint	     server_status,open_options;
   enum enum_thread_type system_thread;
+  enum backup_stages current_backup_stage;
   /*
     Current or next transaction isolation level.
     When a connection is established, the value is taken from
@@ -3615,6 +3603,15 @@ public:
   {
     return server_status & SERVER_STATUS_IN_TRANS;
   }
+  void give_protection_error();
+  inline bool has_read_only_protection()
+  {
+    if (current_backup_stage == BACKUP_FINISHED &&
+        !global_read_lock.is_acquired())
+      return FALSE;
+    give_protection_error();
+    return TRUE;
+  }
   inline bool fill_derived_tables()
   {
     return !stmt_arena->is_stmt_prepare() && !lex->only_view_structure();
diff --git a/sql/sql_cmd.h b/sql/sql_cmd.h
index 1c4c89eb132..345ed57ee30 100644
--- a/sql/sql_cmd.h
+++ b/sql/sql_cmd.h
@@ -108,6 +108,7 @@ enum enum_sql_command {
   SQLCOM_SHOW_STATUS_PACKAGE,
   SQLCOM_SHOW_STATUS_PACKAGE_BODY,
   SQLCOM_SHOW_PACKAGE_BODY_CODE,
+  SQLCOM_BACKUP,
 
   /*
     When a command is added here, be sure it's also added in mysqld.cc
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index c741400b741..df896c14a00 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -547,7 +547,7 @@ bool open_and_lock_for_insert_delayed(THD *thd, TABLE_LIST *table_list)
     If this goes ok, the tickets are cloned and added to the list of granted
     locks held by the handler thread.
   */
-  if (thd->global_read_lock.can_acquire_protection())
+  if (thd->has_read_only_protection())
     DBUG_RETURN(TRUE);
 
   protection_request.init(MDL_key::BACKUP, "", "", MDL_BACKUP_DML,
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 3604082612d..40ba3b6e7b7 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -3139,6 +3139,7 @@ public:
   uint profile_query_id;
   uint profile_options;
   uint grant, grant_tot_col, which_columns;
+  enum backup_stages backup_stage;
   enum Foreign_key::fk_match_opt fk_match_option;
   enum_fk_option fk_update_opt;
   enum_fk_option fk_delete_opt;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 1b1c154d78a..f357f8b0275 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -770,6 +770,7 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_CREATE_SERVER]=      CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ALTER_SERVER]=       CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_SERVER]=        CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_BACKUP]=             CF_AUTO_COMMIT_TRANS;
 
   /*
     The following statements can deal with temporary tables,
@@ -5170,7 +5171,8 @@ end_with_restore_list:
       thd->mdl_context.release_transactional_locks();
       thd->variables.option_bits&= ~(OPTION_TABLE_LOCK);
     }
-    if (thd->global_read_lock.is_acquired())
+    if (thd->global_read_lock.is_acquired() &&
+        thd->current_backup_stage == BACKUP_FINISHED)
       thd->global_read_lock.unlock_global_read_lock(thd);
     if (res)
       goto error;
@@ -5185,6 +5187,13 @@ end_with_restore_list:
     if (res)
       goto error;
 
+    /* We can't have any kind of table locks while backup is active */
+    if (thd->current_backup_stage != BACKUP_FINISHED)
+    {
+      my_error(ER_BACKUP_LOCK_IS_ACTIVE, MYF(0));
+      goto error;
+    }
+
     /*
       Here we have to pre-open temporary tables for LOCK TABLES.
 
@@ -5217,6 +5226,12 @@ end_with_restore_list:
       my_ok(thd);
     }
     break;
+  case SQLCOM_BACKUP:
+    if (check_global_access(thd, RELOAD_ACL))
+      goto error;
+    if (!(res= run_backup_stage(thd, lex->backup_stage)))
+      my_ok(thd);
+    break;
   case SQLCOM_CREATE_DB:
   {
     if (prepare_db_action(thd, lex->create_info.or_replace() ?
diff --git a/sql/sql_reload.cc b/sql/sql_reload.cc
index 334e1697abd..48c5e9552e5 100644
--- a/sql/sql_reload.cc
+++ b/sql/sql_reload.cc
@@ -221,7 +221,9 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
               !thd->mdl_context.has_locks() ||
               thd->handler_tables_hash.records ||
               thd->ull_hash.records ||
-              thd->global_read_lock.is_acquired());
+              thd->global_read_lock.is_acquired() ||
+              thd->current_backup_stage != BACKUP_FINISHED
+              );
 
   /*
     Note that if REFRESH_READ_LOCK bit is set then REFRESH_TABLES is set too
@@ -243,6 +245,7 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
         my_error(ER_LOCK_OR_ACTIVE_TRANSACTION, MYF(0));
         return 1;
       }
+
       /*
 	Writing to the binlog could cause deadlocks, as we don't log
 	UNLOCK TABLES
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index f325f3de6cb..dfda896be69 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -7397,6 +7397,9 @@ static bool mysql_inplace_alter_table(THD *thd,
   thd->count_cuted_fields = CHECK_FIELD_WARN;
   DBUG_ENTER("mysql_inplace_alter_table");
 
+  /* Downgrade DDL lock while we are waiting for exclusive lock below */
+  backup_set_alter_copy_lock(thd, table);
+
   /*
     Upgrade to EXCLUSIVE lock if:
     - This is requested by the storage engine
@@ -7469,9 +7472,7 @@ static bool mysql_inplace_alter_table(THD *thd,
       thd->mdl_context.upgrade_shared_lock(table->mdl_ticket,
                                            MDL_SHARED_NO_WRITE,
                                            thd->variables.lock_wait_timeout))
-  {
     goto cleanup;
-  }
 
   // It's now safe to take the table level lock.
   if (lock_tables(thd, table_list, alter_ctx->tables_opened, 0))
@@ -7508,9 +7509,7 @@ static bool mysql_inplace_alter_table(THD *thd,
 
   if (table->file->ha_prepare_inplace_alter_table(altered_table,
                                                   ha_alter_info))
-  {
     goto rollback;
-  }
 
   /*
     Downgrade the lock if storage engine has told us that exclusive lock was
@@ -7552,6 +7551,10 @@ static bool mysql_inplace_alter_table(THD *thd,
   if (wait_while_table_is_used(thd, table, HA_EXTRA_PREPARE_FOR_RENAME))
     goto rollback;
 
+  /* Set MDL_BACKUP_DDL */
+  if (backup_reset_alter_copy_lock(thd))
+    goto rollback;
+
   /*
     If we are killed after this point, we should ignore and continue.
     We have mostly completed the operation at this point, there should
@@ -7611,7 +7614,7 @@ static bool mysql_inplace_alter_table(THD *thd,
     Rename to the new name (if needed) will be handled separately below.
 
     TODO: remove this check of thd->is_error() (now it intercept
-    errors in some val_*() methoids and bring some single place to
+    errors in some val_*() methods and bring some single place to
     such error interception).
   */
   if (mysql_rename_table(db_type, &alter_ctx->new_db, &alter_ctx->tmp_name,
@@ -9083,6 +9086,7 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
   uint tables_opened;
 
   thd->open_options|= HA_OPEN_FOR_ALTER;
+  thd->mdl_backup_ticket= 0;
   bool error= open_tables(thd, &table_list, &tables_opened, 0,
                           &alter_prelocking_strategy);
   thd->open_options&= ~HA_OPEN_FOR_ALTER;
@@ -10236,6 +10240,8 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     DBUG_RETURN(-1);
   }
 
+  backup_set_alter_copy_lock(thd, from);
+
   alter_table_manage_keys(to, from->file->indexes_are_disabled(), keys_onoff);
 
   from->default_column_bitmaps();
@@ -10517,6 +10523,9 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   cleanup_done= 1;
   to->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
+  if (backup_reset_alter_copy_lock(thd))
+    error= 1;
+
   if (unlikely(mysql_trans_commit_alter_copy_data(thd)))
     error= 1;
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 1d9fde183cd..3f416bc3445 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1503,6 +1503,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  <kwd>  SQL_CALC_FOUND_ROWS
 %token  <kwd>  SQL_NO_CACHE_SYM
 %token  <kwd>  SQL_THREAD
+%token  <kwd>  STAGE_SYM
 %token  <kwd>  STARTS_SYM
 %token  <kwd>  START_SYM                     /* SQL-2003-R */
 %token  <kwd>  STATEMENT_SYM
@@ -1998,7 +1999,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %type <select_order> opt_order_clause order_clause order_list
 
 %type <NONE>
-        analyze_stmt_command
+        analyze_stmt_command backup
         query verb_clause create change select select_into
         do drop insert replace insert2
         insert_values update delete truncate rename compound_statement
@@ -2227,6 +2228,7 @@ statement:
           alter
         | analyze
         | analyze_stmt_command
+        | backup
         | binlog_base64_event
         | call
         | change
@@ -14505,6 +14507,21 @@ opt_table_list:
         | table_list {}
         ;
 
+backup:
+        BACKUP_SYM STAGE_SYM ident
+        {
+          int type;
+          if (unlikely(Lex->sphead))
+            my_yyabort_error((ER_SP_BADSTATEMENT, MYF(0), "BACKUP STAGE"));
+          if ((type= find_type($3.str, &backup_stage_names,
+                               FIND_TYPE_NO_PREFIX)) <= 0)
+            my_yyabort_error((ER_BACKUP_UNKNOWN_STAGE, MYF(0), $3.str));
+          Lex->sql_command= SQLCOM_BACKUP;
+          Lex->backup_stage= (backup_stages) (type-1);
+          break;
+        }
+        ;
+
 opt_delete_gtid_domain:
           /* empty */ {}
         | DELETE_DOMAIN_ID_SYM '=' '(' delete_domain_id_list ')'
@@ -16077,6 +16094,7 @@ keyword_sp_var_and_label:
         | SQL_BUFFER_RESULT
         | SQL_NO_CACHE_SYM
         | SQL_THREAD
+        | STAGE_SYM
         | STARTS_SYM
         | STATEMENT_SYM
         | STATUS_SYM
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index 843309ad1a9..f3401f7e42d 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -1000,6 +1000,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  <kwd>  SQL_CALC_FOUND_ROWS
 %token  <kwd>  SQL_NO_CACHE_SYM
 %token  <kwd>  SQL_THREAD
+%token  <kwd>  STAGE_SYM
 %token  <kwd>  STARTS_SYM
 %token  <kwd>  START_SYM                     /* SQL-2003-R */
 %token  <kwd>  STATEMENT_SYM
@@ -1504,7 +1505,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %type <select_order> opt_order_clause order_clause order_list
 
 %type <NONE>
-        analyze_stmt_command
+        analyze_stmt_command backup
         query verb_clause create change select select_into
         do drop insert replace insert2
         insert_values update delete truncate rename compound_statement
@@ -1751,6 +1752,7 @@ statement:
           alter
         | analyze
         | analyze_stmt_command
+        | backup
         | binlog_base64_event
         | call
         | change
@@ -14560,6 +14562,21 @@ opt_table_list:
         | table_list {}
         ;
 
+backup:
+        BACKUP_SYM STAGE_SYM ident
+        {
+          int type;
+          if (unlikely(Lex->sphead))
+            my_yyabort_error((ER_SP_BADSTATEMENT, MYF(0), "BACKUP STAGE"));
+          if ((type= find_type($3.str, &backup_stage_names,
+                               FIND_TYPE_NO_PREFIX)) <= 0)
+            my_yyabort_error((ER_BACKUP_UNKNOWN_STAGE, MYF(0), $3.str));
+          Lex->sql_command= SQLCOM_BACKUP;
+          Lex->backup_stage= (backup_stages) (type-1);
+          break;
+        }
+        ;
+
 opt_delete_gtid_domain:
           /* empty */ {}
         | DELETE_DOMAIN_ID_SYM '=' '(' delete_domain_id_list ')'
@@ -16166,6 +16183,7 @@ keyword_sp_var_and_label:
         | SQL_BUFFER_RESULT
         | SQL_NO_CACHE_SYM
         | SQL_THREAD
+        | STAGE_SYM
         | STARTS_SYM
         | STATEMENT_SYM
         | STATUS_SYM
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index cf3067ca801..e8044ef9035 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -2585,13 +2585,15 @@ static Sys_var_ulong Sys_read_buff_size(
 static bool check_read_only(sys_var *self, THD *thd, set_var *var)
 {
   /* Prevent self dead-lock */
-  if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction())
+  if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction() ||
+      thd->current_backup_stage != BACKUP_FINISHED)
   {
     my_error(ER_LOCK_OR_ACTIVE_TRANSACTION, MYF(0));
     return true;
   }
   return false;
 }
+
 static bool fix_read_only(sys_var *self, THD *thd, enum_var_type type)
 {
   bool result= true;
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 833a39cb027..58b30a1e77f 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -1823,7 +1823,7 @@ int wsrep_to_isolation_begin(THD *thd, const char *db_, const char *table_,
   DBUG_ASSERT(thd->wsrep_exec_mode == LOCAL_STATE);
   DBUG_ASSERT(thd->wsrep_trx_meta.gtid.seqno == WSREP_SEQNO_UNDEFINED);
 
-  if (thd->global_read_lock.can_acquire_protection())
+  if (thd->has_read_only_protection())
   {
     WSREP_DEBUG("Aborting TOI: Global Read-Lock (FTWRL) in place: %s %lld",
                 thd->query(), (longlong) thd->thread_id);