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

21 files changed, 518 insertions, 34 deletions

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