Added support for BACKUP LOCK / BACKUP UNLOCK

mysql-test/main/backup_locks.result

+#
+# Test lock taken
+#
+BACKUP LOCK test.t1;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_SHARED_HIGH_PRIO	Table metadata lock	test	t1
+BACKUP UNLOCK;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+BACKUP LOCK t1;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_SHARED_HIGH_PRIO	Table metadata lock	test	t1
+BACKUP UNLOCK;
+BACKUP LOCK non_existing.t1;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_SHARED_HIGH_PRIO	Table metadata lock	non_existing	t1
+BACKUP UNLOCK;
+#
+# Test that backup lock protects against ddl
+#
+connect  con1,localhost,root,,;
+connection default;
+create table t1 (a int) engine=innodb;
+insert into t1 values (1);
+backup lock t1;
+select * from t1;
+a
+1
+connection con1;
+drop table t1;
+connection default;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+LOCK_MODE	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+MDL_SHARED_HIGH_PRIO	Table metadata lock	test	t1
+MDL_INTENTION_EXCLUSIVE	Schema metadata lock	test	
+select * from t1;
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+backup unlock;
+connection con1;
+connection default;
+disconnect con1;
+show tables;
+Tables_in_test

mysql-test/main/backup_locks.test

+########################################################################
+# Tests BACKUP STAGE locking
+########################################################################
+
+--source include/have_innodb.inc
+--source include/have_metadata_lock_info.inc
+--source include/not_embedded.inc
+
+--echo #
+--echo # Test lock taken
+--echo #
+
+BACKUP LOCK test.t1;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+BACKUP UNLOCK;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+BACKUP LOCK t1;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+BACKUP UNLOCK;
+BACKUP LOCK non_existing.t1;
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+BACKUP UNLOCK;
+
+--echo #
+--echo # Test that backup lock protects against ddl
+--echo #
+
+connect (con1,localhost,root,,);
+
+connection default;
+create table t1 (a int) engine=innodb;
+insert into t1 values (1);
+backup lock t1;
+select * from t1;
+connection con1;
+--send drop table t1
+connection default;
+let $wait_condition=
+    select count(*) = 1 from information_schema.processlist
+    where state = "Waiting for table metadata lock";
+--source include/wait_condition.inc
+SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info;
+--error ER_LOCK_DEADLOCK
+select * from t1;
+backup unlock;
+connection con1;
+--reap
+connection default;
+disconnect con1;
+show tables;

sql/backup.cc

@@ -354,3 +354,32 @@ bool backup_reset_alter_copy_lock(THD *thd)
                                               thd->variables.lock_wait_timeout);
   return res;
 }
+
+
+/*****************************************************************************
+ Backup locks
+ These functions are used by maria_backup to ensure that there are no active
+ ddl's on the object the backup is going to copy
+*****************************************************************************/
+
+
+bool backup_lock(THD *thd, TABLE_LIST *table)
+{
+  backup_unlock(thd);
+  table->mdl_request.duration= MDL_EXPLICIT;
+  if (thd->mdl_context.acquire_lock(&table->mdl_request,
+                                    thd->variables.lock_wait_timeout))
+    return 1;
+  thd->mdl_backup_lock= table->mdl_request.ticket;
+  return 0;
+}
+
+
+/* Release old backup lock if it exists */
+
+void backup_unlock(THD *thd)
+{
+  if (thd->mdl_backup_lock)
+    thd->mdl_context.release_lock(thd->mdl_backup_lock);
+  thd->mdl_backup_lock= 0;
+}

sql/backup.h

@@ -28,4 +28,7 @@ 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);
+
+bool backup_lock(THD *thd, TABLE_LIST *table);
+void backup_unlock(THD *thd);
 #endif /* BACKUP_INCLUDED */

sql/mysqld.cc

@@ -3667,6 +3667,7 @@ SHOW_VAR com_status_vars[]= {
   {"analyze",              STMT_STATUS(SQLCOM_ANALYZE)},
   {"assign_to_keycache",   STMT_STATUS(SQLCOM_ASSIGN_TO_KEYCACHE)},
   {"backup",               STMT_STATUS(SQLCOM_BACKUP)},
+  {"backup_lock",          STMT_STATUS(SQLCOM_BACKUP_LOCK)},
   {"begin",                STMT_STATUS(SQLCOM_BEGIN)},
   {"binlog",               STMT_STATUS(SQLCOM_BINLOG_BASE64_EVENT)},
   {"call_procedure",       STMT_STATUS(SQLCOM_CALL)},

sql/sql_class.cc

@@ -667,6 +667,7 @@ THD::THD(my_thread_id id, bool is_wsrep_applier, bool skip_global_sys_var_lock)
   main_da.init();
 
   mdl_context.init(this);
+  mdl_backup_lock= 0;
 
   /*
     Pass nominal parameters to init_alloc_root only to ensure that
@@ -1488,6 +1489,8 @@ void THD::cleanup(void)
   mdl_context.release_transactional_locks();
 
   backup_end(this);
+  backup_unlock(this);
+
   /* Release the global read lock, if acquired. */
   if (global_read_lock.is_acquired())
     global_read_lock.unlock_global_read_lock(this);

sql/sql_class.h

@@ -2179,7 +2179,7 @@ class THD :public Statement,
     rpl_io_thread_info *rpl_io_info;
     rpl_sql_thread_info *rpl_sql_info;
   } system_thread_info;
-  MDL_ticket *mdl_backup_ticket;
+  MDL_ticket *mdl_backup_ticket, *mdl_backup_lock;
 
   void reset_for_next_command(bool do_clear_errors= 1);
   /*

sql/sql_cmd.h

@@ -108,7 +108,7 @@ enum enum_sql_command {
   SQLCOM_SHOW_STATUS_PACKAGE,
   SQLCOM_SHOW_STATUS_PACKAGE_BODY,
   SQLCOM_SHOW_PACKAGE_BODY_CODE,
-  SQLCOM_BACKUP,
+  SQLCOM_BACKUP, SQLCOM_BACKUP_LOCK,
 
   /*
     When a command is added here, be sure it's also added in mysqld.cc

sql/sql_parse.cc

@@ -772,6 +772,7 @@ void init_update_queries(void)
   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;
+  sql_command_flags[SQLCOM_BACKUP_LOCK]=        0;
 
   /*
     The following statements can deal with temporary tables,
@@ -5233,6 +5234,17 @@ mysql_execute_command(THD *thd)
     if (!(res= run_backup_stage(thd, lex->backup_stage)))
       my_ok(thd);
     break;
+  case SQLCOM_BACKUP_LOCK:
+    if (check_global_access(thd, RELOAD_ACL))
+      goto error;
+    /* first table is set for lock. For unlock the list is empty */
+    if (first_table)
+      res= backup_lock(thd, first_table);
+    else
+      backup_unlock(thd);
+    if (!res)
+      my_ok(thd);
+    break;
   case SQLCOM_CREATE_DB:
   {
     if (prepare_db_action(thd, lex->create_info.or_replace() ?

sql/sql_yacc.yy

@@ -1999,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 backup
+        analyze_stmt_command backup backup_statements
         query verb_clause create change select select_into
         do drop insert replace insert2
         insert_values update delete truncate rename compound_statement
@@ -14510,18 +14510,34 @@ opt_table_list:
         ;
 
 backup:
-        BACKUP_SYM STAGE_SYM ident
+        BACKUP_SYM backup_statements {}
+	;
+
+backup_statements:
+	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,
+          if ((type= find_type($2.str, &backup_stage_names,
                                FIND_TYPE_NO_PREFIX)) <= 0)
-            my_yyabort_error((ER_BACKUP_UNKNOWN_STAGE, MYF(0), $3.str));
+            my_yyabort_error((ER_BACKUP_UNKNOWN_STAGE, MYF(0), $2.str));
           Lex->sql_command= SQLCOM_BACKUP;
           Lex->backup_stage= (backup_stages) (type-1);
           break;
         }
+	| LOCK_SYM table_ident
+          {
+	    if (unlikely(!Select->add_table_to_list(thd, $2, NULL, 0,
+                                                    TL_READ, MDL_SHARED_HIGH_PRIO)))
+             MYSQL_YYABORT;
+            Lex->sql_command= SQLCOM_BACKUP_LOCK;
+          }
+        | UNLOCK_SYM
+          {
+	    /* Table list is empty for unlock */
+            Lex->sql_command= SQLCOM_BACKUP_LOCK;
+          }
         ;
 
 opt_delete_gtid_domain:

sql/sql_yacc_ora.yy

@@ -1505,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 backup
+        analyze_stmt_command backup backup_statements
         query verb_clause create change select select_into
         do drop insert replace insert2
         insert_values update delete truncate rename compound_statement
@@ -14566,18 +14566,34 @@ opt_table_list:
         ;
 
 backup:
-        BACKUP_SYM STAGE_SYM ident
+        BACKUP_SYM backup_statements {}
+	;
+
+backup_statements:
+	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,
+          if ((type= find_type($2.str, &backup_stage_names,
                                FIND_TYPE_NO_PREFIX)) <= 0)
-            my_yyabort_error((ER_BACKUP_UNKNOWN_STAGE, MYF(0), $3.str));
+            my_yyabort_error((ER_BACKUP_UNKNOWN_STAGE, MYF(0), $2.str));
           Lex->sql_command= SQLCOM_BACKUP;
           Lex->backup_stage= (backup_stages) (type-1);
           break;
         }
+	| LOCK_SYM table_ident
+          {
+	    if (unlikely(!Select->add_table_to_list(thd, $2, NULL, 0,
+                                                    TL_READ, MDL_SHARED_HIGH_PRIO)))
+             MYSQL_YYABORT;
+            Lex->sql_command= SQLCOM_BACKUP_LOCK;
+          }
+        | UNLOCK_SYM
+          {
+	    /* Table list is empty for unlock */
+            Lex->sql_command= SQLCOM_BACKUP_LOCK;
+          }
         ;
 
 opt_delete_gtid_domain: