Copy "safe" Aria files under DDL lock only to allow concurrent DML.

Author: mariadb-andrzejjarzabek

sql/sql_backup.cc

@@ -262,6 +262,18 @@ static bool run_backup_end_phase(THD *thd, backup_end_phase phase, bool failed)
                              failed;
 }
 
+static bool upgrade_and_process(THD *thd,
+                                MDL_request& mdl_request,
+                                enum_mdl_type lock_level,
+                                backup_proc_phase phase)
+{
+  bool fail= thd->mdl_context.upgrade_shared_lock(mdl_request.ticket,
+                                                  lock_level,
+                                                  thd->variables.lock_wait_timeout) ||
+             start_proc_phase(thd, phase);
+  return fail || run_backup_processing(thd);
+}
+
 bool Sql_cmd_backup::execute(THD *thd)
 {
   if (check_global_access(thd, RELOAD_ACL) ||
@@ -310,20 +322,12 @@ bool Sql_cmd_backup::execute(THD *thd)
 
   if (!fail)
   {
-    fail= thd->mdl_context.upgrade_shared_lock(mdl_request.ticket, 
-                                               MDL_BACKUP_WAIT_DDL,
-                                               thd->variables.lock_wait_timeout) ||
-          start_proc_phase(thd, BACKUP_PROC_PHASE_BACKUP_LOCKED);
-    fail= fail || run_backup_processing(thd);
-  }
-
-  if (!fail)
-  {
-    fail= thd->mdl_context.upgrade_shared_lock(mdl_request.ticket,
-                                               MDL_BACKUP_WAIT_COMMIT,
-                                               thd->variables.lock_wait_timeout) ||
-          start_proc_phase(thd, BACKUP_PROC_PHASE_COMMIT_LOCKED);
-    fail= fail || run_backup_processing(thd);
+    fail= start_proc_phase(thd, BACKUP_PROC_PHASE_BACKUP_LOCKED) ||
+          run_backup_processing(thd) ||
+          upgrade_and_process(thd, mdl_request, MDL_BACKUP_WAIT_DDL,
+                              BACKUP_PROC_PHASE_DDL_LOCKED) ||
+          upgrade_and_process(thd, mdl_request, MDL_BACKUP_WAIT_COMMIT,
+                              BACKUP_PROC_PHASE_COMMIT_LOCKED);
   }
 
   fail= run_backup_end_phase(thd, BACKUP_END_PHASE_COMMIT_LOCKED, fail) || fail;