diff --git a/mysql-test/suite/backup/backup_aria_concurrent.result b/mysql-test/suite/backup/backup_aria_concurrent.result
new file mode 100644
index 0000000000000..c7862eaa8600e
--- /dev/null
+++ b/mysql-test/suite/backup/backup_aria_concurrent.result
@@ -0,0 +1,20 @@
+15 30 7500
+Back up the database
+BACKUP SERVER TO '$target_directory' 4 CONCURRENT;
+Restore the database
+# restart: --datadir=MYSQLTEST_VARDIR/some_directory
+Check contents after restore
+SELECT COUNT(*) FROM table_checks;
+COUNT(*)
+400
+SELECT * FROM table_checks WHERE sum_id <> 15;
+tbl_name	sum_id	str_len	blob_len	num_rows
+SELECT * FROM table_checks WHERE str_len <> 30;
+tbl_name	sum_id	str_len	blob_len	num_rows
+SELECT * FROM table_checks WHERE blob_len <> 7500;
+tbl_name	sum_id	str_len	blob_len	num_rows
+SELECT * FROM table_checks WHERE num_rows <> 5;
+tbl_name	sum_id	str_len	blob_len	num_rows
+Restart database in original data directory
+# restart
+Clean up
diff --git a/mysql-test/suite/backup/backup_aria_concurrent.test b/mysql-test/suite/backup/backup_aria_concurrent.test
new file mode 100644
index 0000000000000..390d28b6a3efb
--- /dev/null
+++ b/mysql-test/suite/backup/backup_aria_concurrent.test
@@ -0,0 +1,143 @@
+
+--source include/have_aria.inc
+
+--disable_query_log
+
+
+DELIMITER //;
+CREATE PROCEDURE populate_data(IN t_name VARCHAR(64), IN num_rows INT)
+BEGIN
+  DECLARE i INT DEFAULT 1;
+  SET @query = CONCAT('INSERT INTO ', t_name, ' (id, str_val, blob_val) VALUES (?, ?, ?)');
+  PREPARE stmt FROM @query;
+
+  WHILE i <= num_rows DO
+    SET @str = CONCAT('_row_', i);
+    # Generate a predictable but repeating blob based on the row index
+    SET @blb = REPEAT(CHAR(97 + (i % 26)), 1500);
+    EXECUTE stmt USING i, @str, @blb;
+    SET i = i + 1;
+  END WHILE;
+
+  DEALLOCATE PREPARE stmt;
+END//
+DELIMITER ;//
+
+# Create this many tables transactional and non-transactional each
+let $tab_num= 200;
+let $num_rows= 5;
+
+let $i = 1;
+while ($i <= $tab_num) {
+
+  let $tr=0;
+  while ($tr <= 1) {
+
+    let $suff= _$i;
+    let $table_name= ta_tr$tr$suff;
+
+    eval CREATE TABLE $table_name (
+      id INT PRIMARY KEY,
+      str_val VARCHAR(255),
+      blob_val BLOB,
+      INDEX idx_str (str_val)
+    ) ENGINE=Aria TRANSACTIONAL=$tr;
+
+    eval CALL populate_data('$table_name', $num_rows);
+
+    inc $tr;
+  }
+
+  inc $i;
+}
+
+--enable_query_log
+
+# All tables have the same data, so we query only one  for reference
+
+let $sum_id= `SELECT SUM(id) FROM ta_tr0_1`;
+let $str_len= `SELECT SUM(LENGTH(str_val)) FROM ta_tr0_1`;
+let $blob_len= `SELECT SUM(LENGTH(blob_val)) FROM ta_tr0_1`;
+
+echo $sum_id $str_len $blob_len;
+
+--let $target_directory=$MYSQLTEST_VARDIR/some_directory
+
+# Clean up after a previous failed test, in case we are retrying.
+--error 0,1
+--rmdir $target_directory
+
+--echo Back up the database
+evalp BACKUP SERVER TO '$target_directory' 4 CONCURRENT;
+
+--echo Restore the database
+--let $restart_parameters=--datadir=$target_directory
+--source include/restart_mysqld.inc
+
+--echo Check contents after restore
+
+--disable_query_log
+CREATE TEMPORARY TABLE table_checks (
+  tbl_name VARCHAR(64),
+  sum_id INT,
+  str_len INT,
+  blob_len INT,
+  num_rows INT
+) ENGINE=MEMORY;
+
+let $i = 1;
+while ($i <= $tab_num) {
+  let $tr=0;
+  while ($tr <= 1) {
+
+    let $suff= _$i;
+    let $table_name= ta_tr$tr$suff;
+
+    let $r_sum_id= `SELECT SUM(id) FROM $table_name`;
+    let $r_str_len= `SELECT SUM(LENGTH(str_val)) FROM $table_name`;
+    let $r_blob_len= `SELECT SUM(LENGTH(blob_val)) FROM $table_name`;
+    let $r_num_rows= `SELECT COUNT(*) FROM $table_name`;
+
+    eval INSERT INTO table_checks VALUES ('$table_name', $r_sum_id, $r_str_len, $r_blob_len, $r_num_rows);
+
+    inc $tr;
+  }
+  inc $i;
+}
+
+--enable_query_log
+
+SELECT COUNT(*) FROM table_checks;
+
+# We expect results in the table to always match the results captured before the BACKUP
+# Returned rowsets should be empty
+eval SELECT * FROM table_checks WHERE sum_id <> $sum_id;
+eval SELECT * FROM table_checks WHERE str_len <> $str_len;
+eval SELECT * FROM table_checks WHERE blob_len <> $blob_len;
+eval SELECT * FROM table_checks WHERE num_rows <> $num_rows;
+
+--echo Restart database in original data directory
+--let $restart_parameters=
+--source include/restart_mysqld.inc
+
+--echo Clean up
+
+--disable_query_log
+
+let $i = 1;
+while ($i <= $tab_num) {
+  let $tr=0;
+  while ($tr <= 1) {
+    let $suff= _$i;
+    let $table_name= ta_tr$tr$suff;
+    eval DROP TABLE $table_name;
+    inc $tr;
+  }
+  inc $i;
+}
+
+DROP PROCEDURE populate_data;
+
+--enable_query_log
+
+--rmdir $target_directory
diff --git a/mysql-test/suite/backup/backup_aria_log_dir.result b/mysql-test/suite/backup/backup_aria_log_dir.result
new file mode 100644
index 0000000000000..d16ca8158cb45
--- /dev/null
+++ b/mysql-test/suite/backup/backup_aria_log_dir.result
@@ -0,0 +1,53 @@
+# restart: --aria-log-dir-path=MYSQLTEST_VARDIR/log_directory
+CREATE TABLE t (
+id INT PRIMARY KEY,
+str_val VARCHAR(255),
+blob_val BLOB,
+INDEX idx_str (str_val)
+) ENGINE=Aria TRANSACTIONAL=1;
+CREATE PROCEDURE populate_data(IN num_rows INT)
+BEGIN
+DECLARE i INT DEFAULT 0;
+WHILE i < num_rows DO
+SET @str = CONCAT('_row_', i);
+SET @blb = REPEAT(CHAR(97 + (i % 26)), 1500);
+INSERT INTO t (id, str_val, blob_val) VALUES (i, @str, @blb);
+SET i = i + 1;
+END WHILE;
+END//
+CALL populate_data(10000);
+SELECT COUNT(*) from t;
+COUNT(*)
+10000
+SELECT SUM(id) FROM t;
+SUM(id)
+49995000
+SELECT SUM(LENGTH(str_val)) FROM t;
+SUM(LENGTH(str_val))
+88890
+SELECT SUM(LENGTH(blob_val)) FROM t;
+SUM(LENGTH(blob_val))
+15000000
+Back up the database
+BACKUP SERVER TO '$target_directory';
+Restore the database
+# restart: --datadir=MYSQLTEST_VARDIR/some_directory
+Check contents after restore
+SELECT COUNT(*) from t;
+COUNT(*)
+10000
+SELECT SUM(id) FROM t;
+SUM(id)
+49995000
+SELECT SUM(LENGTH(str_val)) FROM t;
+SUM(LENGTH(str_val))
+88890
+SELECT SUM(LENGTH(blob_val)) FROM t;
+SUM(LENGTH(blob_val))
+15000000
+Restart database in original log and data directories
+# restart: --aria-log-dir-path=MYSQLTEST_VARDIR/log_directory
+Clean up
+DROP PROCEDURE populate_data;
+DROP TABLE t;
+# restart
diff --git a/mysql-test/suite/backup/backup_aria_log_dir.test b/mysql-test/suite/backup/backup_aria_log_dir.test
new file mode 100644
index 0000000000000..7b8123dcd37b6
--- /dev/null
+++ b/mysql-test/suite/backup/backup_aria_log_dir.test
@@ -0,0 +1,74 @@
+--source include/have_aria.inc
+
+--let $log_directory=$MYSQLTEST_VARDIR/log_directory
+--let $target_directory=$MYSQLTEST_VARDIR/some_directory
+
+# Clean up after a previous failed test, in case we are retrying.
+--error 0,1
+--rmdir $log_directory
+--error 0,1
+--rmdir $target_directory
+
+--mkdir $log_directory
+
+--let $orig_restart_parameters=--aria-log-dir-path=$log_directory
+--let $restart_parameters=$orig_restart_parameters
+--source include/restart_mysqld.inc
+
+CREATE TABLE t (
+  id INT PRIMARY KEY,
+  str_val VARCHAR(255),
+  blob_val BLOB,
+  INDEX idx_str (str_val)
+) ENGINE=Aria TRANSACTIONAL=1;
+
+--disable_warnings
+DELIMITER //;
+CREATE PROCEDURE populate_data(IN num_rows INT)
+BEGIN
+  DECLARE i INT DEFAULT 0;
+  WHILE i < num_rows DO
+    SET @str = CONCAT('_row_', i);
+    SET @blb = REPEAT(CHAR(97 + (i % 26)), 1500);
+    INSERT INTO t (id, str_val, blob_val) VALUES (i, @str, @blb);
+    SET i = i + 1;
+  END WHILE;
+END//
+DELIMITER ;//
+--enable_warnings
+
+CALL populate_data(10000);
+
+SELECT COUNT(*) from t;
+SELECT SUM(id) FROM t;
+SELECT SUM(LENGTH(str_val)) FROM t;
+SELECT SUM(LENGTH(blob_val)) FROM t;
+
+--echo Back up the database
+evalp BACKUP SERVER TO '$target_directory';
+
+--echo Restore the database
+--let $restart_parameters=--datadir=$target_directory
+--source include/restart_mysqld.inc
+
+--echo Check contents after restore
+
+SELECT COUNT(*) from t;
+SELECT SUM(id) FROM t;
+SELECT SUM(LENGTH(str_val)) FROM t;
+SELECT SUM(LENGTH(blob_val)) FROM t;
+
+--echo Restart database in original log and data directories
+--let $restart_parameters=$orig_restart_parameters
+--source include/restart_mysqld.inc
+
+--echo Clean up
+
+DROP PROCEDURE populate_data;
+DROP TABLE t;
+
+--let $restart_parameters=
+--source include/restart_mysqld.inc
+
+--rmdir $target_directory
+--rmdir $log_directory
diff --git a/mysql-test/suite/backup/backup_nonacid.result b/mysql-test/suite/backup/backup_nonacid.result
new file mode 100644
index 0000000000000..9027dd735bedf
--- /dev/null
+++ b/mysql-test/suite/backup/backup_nonacid.result
@@ -0,0 +1,56 @@
+CREATE TABLE t_archive (id int unsigned) ENGINE=ARCHIVE;
+INSERT INTO t_archive VALUES (2), (3), (5), (7), (11);
+CREATE DATABASE d;
+CREATE TABLE d.t_csv (id int unsigned NOT NULL) ENGINE=CSV;
+INSERT INTO d.t_csv VALUES (4), (26), (41), (60), (83), (109);
+CREATE TABLE t_myisam1 (id int unsigned) ENGINE=MyISAM;
+INSERT INTO t_myisam1 VALUES (1), (1), (2), (3), (5), (8);
+CREATE TABLE t_myisam2 (id int unsigned) ENGINE=MyISAM;
+INSERT INTO t_myisam2 VALUES (13), (21), (34), (55), (89), (144);
+CREATE TABLE t_mrg (id int unsigned) ENGINE=MRG_MyISAM UNION=(t_myisam1, t_myisam2);
+BACKUP SERVER TO '$target_directory';
+# restart: --datadir=MYSQLTEST_VARDIR/some_directory
+SELECT * FROM t_archive ORDER BY id;
+id
+2
+3
+5
+7
+11
+SELECT * FROM d.t_csv ORDER BY id;
+id
+4
+26
+41
+60
+83
+109
+SELECT * FROM t_myisam1 ORDER BY id;
+id
+1
+1
+2
+3
+5
+8
+SELECT * FROM t_mrg ORDER BY id;
+id
+1
+1
+2
+3
+5
+8
+13
+21
+34
+55
+89
+144
+# restart
+DROP TABLE t_archive;
+DROP TABLE d.t_csv;
+DROP TABLE t_myisam1;
+DROP TABLE t_myisam2;
+DROP TABLE t_mrg;
+DROP DATABASE d;
diff --git a/mysql-test/suite/backup/backup_nonacid.test b/mysql-test/suite/backup/backup_nonacid.test
new file mode 100644
index 0000000000000..eac863a191685
--- /dev/null
+++ b/mysql-test/suite/backup/backup_nonacid.test
@@ -0,0 +1,45 @@
+--source include/have_csv.inc
+--source include/have_archive.inc
+
+CREATE TABLE t_archive (id int unsigned) ENGINE=ARCHIVE;
+INSERT INTO t_archive VALUES (2), (3), (5), (7), (11);
+
+CREATE DATABASE d;
+CREATE TABLE d.t_csv (id int unsigned NOT NULL) ENGINE=CSV;
+INSERT INTO d.t_csv VALUES (4), (26), (41), (60), (83), (109);
+
+CREATE TABLE t_myisam1 (id int unsigned) ENGINE=MyISAM;
+INSERT INTO t_myisam1 VALUES (1), (1), (2), (3), (5), (8);
+
+CREATE TABLE t_myisam2 (id int unsigned) ENGINE=MyISAM;
+INSERT INTO t_myisam2 VALUES (13), (21), (34), (55), (89), (144);
+
+CREATE TABLE t_mrg (id int unsigned) ENGINE=MRG_MyISAM UNION=(t_myisam1, t_myisam2);
+
+--let $target_directory=$MYSQLTEST_VARDIR/some_directory
+
+# Clean up after a previous failed test, in case we are retrying.
+--error 0,1
+--rmdir $target_directory
+
+evalp BACKUP SERVER TO '$target_directory';
+
+--let $restart_parameters=--datadir=$target_directory
+--source include/restart_mysqld.inc
+
+SELECT * FROM t_archive ORDER BY id;
+SELECT * FROM d.t_csv ORDER BY id;
+SELECT * FROM t_myisam1 ORDER BY id;
+SELECT * FROM t_mrg ORDER BY id;
+
+--let $restart_parameters=
+--source include/restart_mysqld.inc
+
+DROP TABLE t_archive;
+DROP TABLE d.t_csv;
+DROP TABLE t_myisam1;
+DROP TABLE t_myisam2;
+DROP TABLE t_mrg;
+DROP DATABASE d;
+
+--rmdir $target_directory
diff --git a/mysql-test/suite/backup/backup_nonacid_partitioned.result b/mysql-test/suite/backup/backup_nonacid_partitioned.result
new file mode 100644
index 0000000000000..7090a0fe688bb
--- /dev/null
+++ b/mysql-test/suite/backup/backup_nonacid_partitioned.result
@@ -0,0 +1,65 @@
+CREATE TABLE t_myisam (id int unsigned, KEY(id)) ENGINE=MyISAM;
+INSERT INTO t_myisam VALUES (1), (2), (3), (5), (8), (13);
+CREATE TABLE t_myisam_part (id int unsigned, KEY(id)) ENGINE=MyISAM
+PARTITION BY HASH(id) PARTITIONS 2;
+INSERT INTO t_myisam_part VALUES (1), (2), (3), (5), (8), (13);
+CREATE TABLE t_archive (id int unsigned) ENGINE=ARCHIVE;
+INSERT INTO t_archive VALUES (2), (3), (5), (7), (11);
+CREATE TABLE t_archive_part (id int unsigned) ENGINE=ARCHIVE
+PARTITION BY HASH(id) PARTITIONS 2;
+INSERT INTO t_archive_part VALUES (2), (3), (5), (7), (11);
+BACKUP SERVER TO '$target_directory';
+# restart: --datadir=MYSQLTEST_VARDIR/backup_nonacid_partitioned
+SELECT COUNT(*) FROM t_myisam;
+COUNT(*)
+6
+SELECT * FROM t_myisam ORDER BY id;
+id
+1
+2
+3
+5
+8
+13
+CHECK TABLE t_myisam;
+Table	Op	Msg_type	Msg_text
+test.t_myisam	check	status	OK
+SELECT COUNT(*) FROM t_archive;
+COUNT(*)
+5
+SELECT * FROM t_archive ORDER BY id;
+id
+2
+3
+5
+7
+11
+SELECT COUNT(*) FROM t_myisam_part;
+COUNT(*)
+6
+SELECT * FROM t_myisam_part ORDER BY id;
+id
+1
+2
+3
+5
+8
+13
+CHECK TABLE t_myisam_part;
+Table	Op	Msg_type	Msg_text
+test.t_myisam_part	check	status	OK
+SELECT COUNT(*) FROM t_archive_part;
+COUNT(*)
+5
+SELECT * FROM t_archive_part ORDER BY id;
+id
+2
+3
+5
+7
+11
+# restart
+DROP TABLE t_myisam;
+DROP TABLE t_myisam_part;
+DROP TABLE t_archive;
+DROP TABLE t_archive_part;
diff --git a/mysql-test/suite/backup/backup_nonacid_partitioned.test b/mysql-test/suite/backup/backup_nonacid_partitioned.test
new file mode 100644
index 0000000000000..e5a545c456090
--- /dev/null
+++ b/mysql-test/suite/backup/backup_nonacid_partitioned.test
@@ -0,0 +1,67 @@
+--source include/have_partition.inc
+--source include/have_archive.inc
+
+#
+# BACKUP SERVER must flush non-transactional partitioned tables before
+# copying their files.
+#
+# tc_purge_for_backup() decides whether a table has to be closed before its
+# files are copied by switching on TABLE_SHARE::db_type()->db_type. For a
+# partitioned table that value is DB_TYPE_PARTITION_DB -- the real engine is
+# kept in TABLE_SHARE::default_part_plugin -- so the switch falls through to
+# "return false" and the table is never purged, while copy_misc_files() does
+# copy its per-partition .MYD/.MYI/.ARZ/.ARM files (and .par).
+#
+# The unpartitioned tables are a control group: they take the MyISAM/ARCHIVE
+# arms of the switch, are closed before the copy, and must survive intact.
+# Both groups are expected to come back complete and consistent.
+#
+
+CREATE TABLE t_myisam (id int unsigned, KEY(id)) ENGINE=MyISAM;
+INSERT INTO t_myisam VALUES (1), (2), (3), (5), (8), (13);
+
+CREATE TABLE t_myisam_part (id int unsigned, KEY(id)) ENGINE=MyISAM
+PARTITION BY HASH(id) PARTITIONS 2;
+INSERT INTO t_myisam_part VALUES (1), (2), (3), (5), (8), (13);
+
+CREATE TABLE t_archive (id int unsigned) ENGINE=ARCHIVE;
+INSERT INTO t_archive VALUES (2), (3), (5), (7), (11);
+
+CREATE TABLE t_archive_part (id int unsigned) ENGINE=ARCHIVE
+PARTITION BY HASH(id) PARTITIONS 2;
+INSERT INTO t_archive_part VALUES (2), (3), (5), (7), (11);
+
+--let $target_directory=$MYSQLTEST_VARDIR/backup_nonacid_partitioned
+
+# Clean up after a previous failed test, in case we are retrying.
+--error 0,1
+--rmdir $target_directory
+
+evalp BACKUP SERVER TO '$target_directory';
+
+--let $restart_parameters=--datadir=$target_directory
+--source include/restart_mysqld.inc
+
+# Control group: unpartitioned tables.
+SELECT COUNT(*) FROM t_myisam;
+SELECT * FROM t_myisam ORDER BY id;
+CHECK TABLE t_myisam;
+SELECT COUNT(*) FROM t_archive;
+SELECT * FROM t_archive ORDER BY id;
+
+# Same tables, partitioned.
+SELECT COUNT(*) FROM t_myisam_part;
+SELECT * FROM t_myisam_part ORDER BY id;
+CHECK TABLE t_myisam_part;
+SELECT COUNT(*) FROM t_archive_part;
+SELECT * FROM t_archive_part ORDER BY id;
+
+--let $restart_parameters=
+--source include/restart_mysqld.inc
+
+DROP TABLE t_myisam;
+DROP TABLE t_myisam_part;
+DROP TABLE t_archive;
+DROP TABLE t_archive_part;
+
+--rmdir $target_directory
diff --git a/sql/sql_backup.cc b/sql/sql_backup.cc
index f4ad618b4694a..4a6c1a752cf52 100644
--- a/sql/sql_backup.cc
+++ b/sql/sql_backup.cc
@@ -23,6 +23,13 @@
 #include "my_atomic_wrapper.h"
 #include "tpool.h"
 #include "aligned.h"
+#include "table_cache.h"
+
+#include <string>
+#include <set>
+#include <algorithm>
+
+static constexpr const char zerobuf[511]{};
 
 #if defined __linux__ || defined __FreeBSD__
 using copying_step= ssize_t(int,int,size_t,off_t*);
@@ -62,6 +69,29 @@ using tpool::pread;
 using tpool::pwrite;
 #else
 # include <sys/mman.h>
+
+/** Obtain file descriptor to source directory.
+@return  File descriptor or -1 on error
+#note  Should be called when BACKUP SERVER is in progress.
+       Call is thread-safe and may incur synchronization cost.
+       Value may be stored safely for the duration of backup.
+       Lifetime is managed by SQL layer. Error is reported by my_error.
+*/
+int get_datadir_fd()
+{
+  /* Implemented by lazy initialization of static local variable on first call.
+  mysql_real_data_home is guaranteed to be populated from the requirement that
+  the call should happen from withing the backup process.
+  The descriptor is then kept open throughout the lifetime of the process
+  (it is never explicitly closed). */
+  static int datadir_fd= open(mysql_real_data_home, O_DIRECTORY);
+  if (datadir_fd < 0)
+  {
+    my_error(ER_CANT_READ_DIR, MYF(0), mysql_real_data_home, errno);
+  }
+  return datadir_fd;
+}
+
 /**
    Copy a file using a memory mapping.
    @tparam stream true=write to a stream, false=pwrite to a file
@@ -168,7 +198,98 @@ static ssize_t pread_write(IF_WIN(const native_file_handle&,int) in_fd,
 #ifdef __APPLE__
 /* The inline copy_entire_file() invokes fcopyfile() */
 #elif defined _WIN32
-/* CopyFileEx() should be used */
+/** Copy entire file.
+ @param src_path  path file file to copy
+ @param dst_path  path of file to copy to
+ @param target    backup target
+ @param sink      worker context
+ @return error code (non-positive)
+ @retval 0      on success
+ @note Wrapper for CopyFileExA, will report error using my_error  */
+extern "C"
+int copy_entire_file(const char *src_path, const char *dst_path,
+                     const struct backup_target *target,
+                     const struct backup_sink *sink)
+{
+   if (sink->stream == sink->NO_STREAM)
+  {
+    const std::string full_dst_path{make_path(target->path, dst_path)};
+    if (!CopyFileEx(src_path, full_dst_path.c_str(), nullptr, nullptr, nullptr,
+                    COPY_FILE_NO_BUFFERING))
+    {
+      my_osmaperr(GetLastError());
+      my_error(ER_CANT_CREATE_FILE, MYF(0), full_dst_path.c_str(), errno);
+      return 1;
+    }
+  }
+  else
+  {
+    HANDLE src, dst{sink->stream};
+    for (;;)
+    {
+      src= CreateFile(src_path, GENERIC_READ,
+                      FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
+                      my_win_file_secattr(), OPEN_EXISTING,
+                      FILE_ATTRIBUTE_NORMAL, nullptr);
+      if (src != INVALID_HANDLE_VALUE)
+        break;
+
+      switch (GetLastError()) {
+      case ERROR_SHARING_VIOLATION:
+      case ERROR_LOCK_VIOLATION:
+        std::this_thread::sleep_for(std::chrono::seconds(1));
+        continue;
+      }
+
+      my_osmaperr(GetLastError());
+      my_error(ER_FILE_NOT_FOUND, MYF(ME_ERROR_LOG), src_path, errno);
+      return -1;
+    }
+
+    LARGE_INTEGER li;
+    if (!GetFileSizeEx(src, &li))
+    {
+    write_error:
+      my_osmaperr(GetLastError());
+      my_error(ER_ERROR_ON_WRITE, MYF(0), dst_path, errno);
+      if (src != INVALID_HANDLE_VALUE)
+        CloseHandle(src);
+      return -1;
+    }
+
+    if (backup_stream_start(dst, dst_path, 0644, li.QuadPart, nullptr, 0) ||
+        backup_stream_append_plain(src, dst, 0, li.QuadPart))
+      goto write_error;
+
+    if (size_t pad= size_t(li.LowPart) & 511)
+      if (backup_stream_write(dst, zerobuf, 512 - pad))
+        goto write_error;
+    if (!CloseHandle(src))
+    {
+      src= INVALID_HANDLE_VALUE;
+      goto write_error;
+    }
+  }
+  return 0;
+}
+
+/** Copy entire file from data directory target, preserving path.
+@param path    relative path of file
+@param target  backup target
+@param sink    worker context
+@return error code
+@retval 0      on success
+@note   The file will be copied to the same path relative to
+        target directory. Any intermediate directories must
+        already exist in the target. */
+extern "C" int copy_datafile_to_target(const char *path,
+                                       const struct backup_target *target,
+                                       const backup_sink *sink)
+{
+  return copy_entire_file(make_path(mysql_real_data_home, path).c_str(),
+                          path, target, sink);
+}
+
 #else
 /** Copy a file (whole content).
 @param src  source file descriptor
@@ -181,6 +302,85 @@ extern "C" int copy_entire_file(int src, int dst)
 }
 #endif
 
+#ifndef _WIN32
+/** Copy an entire file to target.
+@param src_fd  source file descriptor
+@param target  backup target
+@return error code (non-positive)
+@retval 0   on success
+@note   Any intermediate directories must already exist in the target. */
+# ifdef __cplusplus
+extern "C"
+# endif
+int copy_fd_to_target(int src_fd,
+                      const struct backup_target *target,
+                      const char *path,
+                      const struct backup_sink *sink)
+{
+  int ret_val= 0;
+  int tgt_fd{sink->stream};
+  if (tgt_fd == sink->NO_STREAM)
+  {
+    tgt_fd= openat(target->fd, path,
+                    O_CREAT | O_EXCL | O_WRONLY, 0666);
+
+    if (tgt_fd < 0)
+    {
+      my_error(ER_CANT_CREATE_FILE, MYF(0), path, errno);
+      ret_val= 1;
+    }
+    else
+    {
+      ret_val= copy_entire_file(src_fd, tgt_fd);
+      if (ret_val | close(tgt_fd))
+      {
+      write_error:
+        my_error(ER_ERROR_ON_WRITE, MYF(0), path, errno);
+        ret_val= 1;
+      }
+    }
+  }
+  else
+  {
+    uint64_t end= uint64_t(lseek(src_fd, 0, SEEK_END));
+    if (backup_stream_start(tgt_fd, path, 0644, end, nullptr, 0) ||
+        backup_stream_append(src_fd, tgt_fd, 0, end))
+      goto write_error;
+    if (size_t pad= size_t(end) & 511)
+      if (backup_stream_write(tgt_fd, zerobuf, 512 - pad))
+        goto write_error;
+  }
+
+  return ret_val;
+}
+
+/** Copy entire file from data directory target, preserving path.
+@param datadir_fd  file descriptor of source directory
+@param path        relative path of file
+@param target      backup target
+@param sink        worker context
+@return error code
+@retval 0      on success
+@note   The file will be copied to the same path relative to
+        target directory. Any intermediate directories must
+        already exist in the target. */
+extern "C" int copy_datafile_to_target(int datadir_fd,
+                                       const char *path,
+                                       const struct backup_target *target,
+                                       const backup_sink *sink)
+{
+  int src_fd = openat(datadir_fd, path, O_RDONLY);
+  if (src_fd < 0)
+  {
+    my_error(ER_CANT_OPEN_FILE, MYF(0), path, errno);
+    return 1;
+  }
+  int ret_val=  copy_fd_to_target(src_fd, target, path, sink);
+  close(src_fd);
+  return ret_val;
+}
+#endif
+
 /** Copy a portion of a file.
 @param src   source file descriptor
 @param dst   target to append src to
@@ -215,6 +415,132 @@ extern "C" int copy_file(IF_WIN(const native_file_handle&,int) src,
   return int(ret);
 }
 
+/** Ensure a file can be copied to a subdirectory in target.
+May create the subdirectory.
+@param target  backup target
+@param name    subdirectory name
+@return error code (non-positive)
+@retval 0      on success
+@note   If the directory is created, the directory containing it must
+        already exist: nested directory creation is not supported. */
+extern "C" int ensure_target_subdir(const struct backup_target *target,
+                                    const char* name)
+{
+
+#ifdef _WIN32
+  const std::string dir_path{make_path(target->path, name)};
+  if (CreateDirectory(dir_path.c_str(), nullptr))
+    return 0;
+  DWORD err= GetLastError();
+  if (err == ERROR_ALREADY_EXISTS)
+    return 0;
+  my_osmaperr(err);
+#else
+  if (likely(!mkdirat(target->fd, name, 0777) || errno == EEXIST))
+    return 0;
+#endif
+  my_error(ER_CANT_CREATE_FILE, MYF(0), name, errno);
+  return 1;
+}
+
+/* all extensions have the same length, adjust if that changes */
+static constexpr size_t ext_len= 4;
+
+/* Files not copied by plugin backup implementations: files managed by
+SQL layer and miscellaneous engine files to be copied bunde DDL lock */
+static constexpr const char* misc_exts[] {".frm", ".par", ".MYD", ".MYI", ".MRG",
+                                          ".ARM", ".ARZ", ".CSM", ".CSV"};
+static constexpr const char db_opt_name[] {"db.opt"};
+static constexpr size_t db_opt_len= sizeof(db_opt_name) - 1;
+
+static bool match_ext(const char* ext1, const char* ext2) noexcept
+{
+  return memcmp(ext1,
+                ext2,
+                ext_len) == 0;
+}
+
+static bool match_misc_ext(const char* file_ext) noexcept
+{
+  return std::find_if(std::begin(misc_exts), std::end(misc_exts),
+                    [file_ext](const char* misc_ext) {
+                      return match_ext(file_ext, misc_ext);
+                    }) != std::end(misc_exts);
+}
+
+static bool is_db_opt(const char* filename, size_t filename_len)
+{
+  return filename_len == db_opt_len &&
+         memcmp(filename, db_opt_name, db_opt_len) == 0;
+}
+
+static bool is_misc_file(const char* filename)
+{
+  size_t filename_len= strlen(filename);
+  if (filename_len < ext_len)
+    return false;
+  const char *file_ext = filename + filename_len - ext_len;
+  return match_misc_ext(file_ext) || is_db_opt(filename, filename_len);
+}
+
+std::string make_path(const char *base_path, const char *filename) noexcept
+{
+  std::string path;
+  const size_t base_len= strlen(base_path);
+  const size_t filename_len= strlen(filename);
+  path.reserve(base_len + filename_len + 1);
+  path.append(base_path, base_len);
+  path+= '/';
+  path.append(filename, filename_len);
+  return path;
+}
+
+static bool copy_misc_files(const backup_target *target,
+                            const backup_sink *sink)
+{
+#ifndef _WIN32
+  int datadir_fd= get_datadir_fd();
+  if (datadir_fd < 0)
+    return true;
+#endif
+  std::unordered_set<std::string> ensured_dirs;
+  Dir_scan datadir;
+  if (datadir.initialize(mysql_real_data_home, MYF(MY_WANT_STAT)))
+    return true;
+  for (const fileinfo &fi : datadir.contents())
+  {
+    if ((fi.mystat->st_mode & S_IFMT) == S_IFDIR)
+    {
+      const char* dir_name= fi.name;
+      if(sink->stream == sink->NO_STREAM &&
+         ensured_dirs.insert(dir_name).second)
+      {
+        if (ensure_target_subdir(target, dir_name))
+          return true;
+      }
+      const std::string dir_path{make_path(mysql_real_data_home, dir_name)};
+      Dir_scan dbdir;
+      if (dbdir.initialize(dir_path.c_str(), MYF(0)))
+        return true;
+      for (const fileinfo &fi : dbdir.contents())
+      {
+        if (is_misc_file(fi.name))
+        {
+          const std::string path= make_path(dir_name, fi.name);
+          if (copy_datafile_to_target(
+#ifndef _WIN32
+                                      datadir_fd,
+#endif
+                                      path.c_str(), target, sink))
+            return true;
+        }
+      }
+    }
+  }
+  return false;
+}
+
+
 /** Append to the configuration file.
 @param target   backup target directory
 @param config   the configuration file snippet to append
@@ -558,6 +884,14 @@ bool Sql_cmd_backup::execute(THD *thd)
     }
   backup_phase_start:
     target_phase->phase= backup_phase(phase);
+
+    if (phase == BACKUP_PHASE_NO_DDL)
+    {
+      tc_purge_for_backup();
+      if ((fail= copy_misc_files(&target_phase->target, &target_phase->sink)))
+        break;
+    }
+
     fail= plugin_foreach_with_mask(thd, backup_start,
                                    MYSQL_STORAGE_ENGINE_PLUGIN,
                                    PLUGIN_IS_DELETED|PLUGIN_IS_READY,
diff --git a/sql/sql_backup_interface.h b/sql/sql_backup_interface.h
index 94892ba7591ab..e2362f99a022a 100644
--- a/sql/sql_backup_interface.h
+++ b/sql/sql_backup_interface.h
@@ -13,7 +13,17 @@
    along with this program; if not, write to the Free Software
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335  USA */
 
+#include <mysqld_error.h>
+#include <my_dir.h>
+#include <string.h>
+
+#ifdef __cplusplus
+# include "span.h"
+# include <string>
+#endif
+
 struct backup_target;
+struct backup_sink;
 
 /** A payload chunk in a sparse file that is being streamed */
 struct backup_chunk
@@ -25,14 +35,65 @@ struct backup_chunk
 };
 
 #ifdef _WIN32
-/* Use CopyFileEx() to copy entire files */
+/** Copy entire file.
+ @param src_path  path file file to copy
+ @param dst_path  path of file to copy to
+ @param target    backup target
+ @param sink      worker context
+ @return error code (non-positive)
+ @retval 0      on success
+ @note Wrapper for CopyFileExA, will report error using my_error  */
+# ifdef __cplusplus
+extern "C"
+# endif
+int copy_entire_file(const char *src_path,
+                     const char *dst_path,
+                     const struct backup_target *target,
+                     const struct backup_sink *sink);
+
+/** Copy entire file from data directory target, preserving path.
+@param path    relative path of file
+@param target  backup target
+@param sink    worker context
+@return error code
+@retval 0      on success
+@note   The file will be copied to the same path relative to
+        target directory. Any intermediate directories must
+        already exist in the target. */
+# ifdef __cplusplus
+extern "C"
+# endif
+int copy_datafile_to_target(const char *path,
+                            const struct backup_target *target,
+                            const struct backup_sink *sink);
+
 struct native_file_handle;
-#elif defined __APPLE__
+#else
+
+/** Copy entire file from data directory target, preserving path.
+@param datadir_fd  file descriptor of source directory
+@param path        relative path of file
+@param target      backup target
+@param sink        worker context
+@return error code
+@retval 0      on success
+@note   The file will be copied to the same path relative to
+        target directory. Any intermediate directories must
+        already exist in the target. */
+# ifdef __cplusplus
+extern "C"
+# endif
+int copy_datafile_to_target(int datadir_fd,
+                            const char *path,
+                            const struct backup_target *target,
+                            const struct backup_sink *sink);
+
+# if defined __APPLE__
 /* You should invoke fclonefileat(2) manually before attempting
 copy_entire_file() or copy_file() */
-# include <sys/attr.h>
-# include <sys/clonefile.h>
-# include <copyfile.h>
+#  include <sys/attr.h>
+#  include <sys/clonefile.h>
+#  include <copyfile.h>
 /** Copy an entire file.
 @param src  source file descriptor
 @param dst  target to append src to
@@ -42,20 +103,51 @@ inline int copy_entire_file(int src, int dst)
 {
   return fcopyfile(src, dst, NULL, COPYFILE_ALL | COPYFILE_CLONE);
 }
-#else
-# ifdef __cplusplus
+# else
+#  ifdef __cplusplus
 extern "C"
-# endif
+#  endif
 /** Copy an entire file.
 @param src  source file descriptor
 @param dst  target to append src to
 @return error code (non-positive)
 @retval 0   on success */
 int copy_entire_file(int src, int dst);
+# endif
+
+/** Obtain file descriptor to source directory.
+@return  File descriptor or -1 on error
+#note  Should be called when BACKUP SERVER is in progress.
+       Call is thread-safe and may incur synchronization cost.
+       Value may be stored safely for the duration of backup.
+       Lifetime is managed by SQL layer. Error is reported by my_error.
+*/
+# ifdef __cplusplus
+extern "C"
+# endif
+int get_datadir_fd();
+
+/** Copy an entire file to target.
+@param src_fd  source file descriptor
+@param target  backup target
+@param path    target file path
+@param sink    worker context
+@return error code (non-positive)
+@retval 0   on success
+@note   Any intermediate directories must already exist in the target. */
+# ifdef __cplusplus
+extern "C"
+# endif
+int copy_fd_to_target(int src_fd,
+                      const struct backup_target *target,
+                      const char *path,
+                      const struct backup_sink *sink);
+
 #endif
 
 #ifdef __cplusplus
 extern "C"
+{
 #endif
 /** Copy a portion of a file.
 @param src   source file descriptor
@@ -68,9 +160,16 @@ int copy_file(IF_WIN(const native_file_handle&,int) src,
               IF_WIN(const native_file_handle&,int) dst,
               uint64_t start, uint64_t end);
 
-#ifdef __cplusplus
-extern "C"
-#endif
+/** Ensure a file can be copied to a subdirectory in target.
+May create the subdirectory.
+@param target  backup target
+@param name    subdirectory name
+@return error code (non-positive)
+@retval 0      on success
+@note   If the directory is created, the directory containing it must
+        already exist: nested directory creation is not supported. */
+int ensure_target_subdir(const struct backup_target *target, const char* name);
+
 /** Append to the configuration file.
 @param target   backup target directory
 @param config   the configuration file snippet to append
@@ -80,9 +179,7 @@ extern "C"
 int backup_config_append(IF_WIN(const char*, int) target,
                          const char *config, size_t size);
 
-#ifdef __cplusplus
-extern "C"
-#endif
+
 /** Append to the configuration file.
 @param target   backup stream
 @param config   the configuration file snippet to append
@@ -92,9 +189,6 @@ extern "C"
 int backup_stream_config(IF_WIN(HANDLE, int) stream,
                          const char *config, size_t size);
 
-#ifdef __cplusplus
-extern "C"
-#endif
 /** Start streaming a file.
 @param target   backup target
 @param name     file name
@@ -108,9 +202,6 @@ int backup_stream_start(IF_WIN(HANDLE, int) stream,
                         const char *name, mode_t mode, uint64_t size,
                         const struct backup_chunk *chunks, size_t n_chunks);
 
-#ifdef __cplusplus
-extern "C"
-#endif
 /**
    Write data to a stream.
    @param stream  backup stream
@@ -122,9 +213,6 @@ extern "C"
 int backup_stream_write(IF_WIN(HANDLE, int) stream, const void *buf,
                         size_t size);
 
-#ifdef __cplusplus
-extern "C"
-#endif
 /**
    Append a file snippet to the stream,
    after a corresponding call to backup_stream_start().
@@ -143,9 +231,6 @@ int backup_stream_append(IF_WIN(const native_file_handle&,int) src,
                          uint64_t start, uint64_t end);
 
 #ifdef __linux__
-# ifdef __cplusplus
-extern "C"
-# endif
 /**
    Append an immutable snippet of a file to the stream,
    allowing Linux sendfile(2) to be invoked.
@@ -168,11 +253,49 @@ int backup_stream_append_async(int src, int stream,
 #endif
 
 #ifdef _WIN32
-# ifdef __cplusplus
-extern "C"
-# endif
 int backup_stream_append_plain(HANDLE src, HANDLE stream,
                                uint64_t start, uint64_t end);
 #else
 # define backup_stream_append_plain backup_stream_append
 #endif
+
+#ifdef __cplusplus
+} // extern "C"
+
+ /* RAII wrapper for my_dir() */
+class Dir_scan
+{
+public:
+  Dir_scan() = default;
+  bool initialize(const char* path, myf flags) noexcept
+  {
+    dir_info= my_dir(path, flags);
+    if (!dir_info)
+    {
+      my_error(ER_CANT_READ_DIR, MYF(0), path, my_errno);
+      return true;
+    }
+    return false;
+  }
+
+  ~Dir_scan() noexcept
+  {
+    my_dirend(dir_info);
+  }
+
+  Dir_scan(const Dir_scan&) = delete;
+  Dir_scan& operator=(const Dir_scan&) = delete;
+
+  st_::span<const fileinfo> contents() const
+  {
+    assert(dir_info);
+    return {dir_info->dir_entry, dir_info->number_of_files};
+  }
+
+private:
+  MY_DIR *dir_info {nullptr};
+};
+
+std::string make_path(const char *base_path, const char *filename) noexcept;
+
+#endif
diff --git a/sql/table_cache.cc b/sql/table_cache.cc
index 60d26b8c7f3e1..db2f142eed68b 100644
--- a/sql/table_cache.cc
+++ b/sql/table_cache.cc
@@ -324,7 +324,6 @@ void tc_purge()
     intern_close_table(table);
 }
 
-
 /**
   Add new TABLE object to table cache.
 
@@ -537,6 +536,152 @@ static void tdc_delete_share_from_hash(TDC_element *element)
   DBUG_VOID_RETURN;
 }
 
+/* The engine that owns the files of a table. For a partitioned table
+share, db_type() is the partitioning engine, while the engine that owns
+the per-partition files is default_part_plugin. */
+static handlerton *backup_files_hton(const TABLE_SHARE& share)
+{
+  handlerton *hton= share.db_type();
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  if (hton == partition_hton && share.default_part_plugin)
+    hton= plugin_hton(share.default_part_plugin);
+#endif
+  return hton;
+}
+
+/* Tables for non-ACID engies for which files are copied directly
+from the SQL layer need to be closed. Additionally non-transactional
+Aria tables are closed and purged from the cache. */
+static bool table_needs_flush_for_backup(const TABLE_SHARE& share)
+{
+  legacy_db_type engine= backup_files_hton(share)->db_type;
+  switch (engine)
+  {
+  case DB_TYPE_MYISAM:
+  case DB_TYPE_MRG_MYISAM:
+  case DB_TYPE_CSV_DB:
+  case DB_TYPE_ARCHIVE_DB:
+    return true;
+  case DB_TYPE_ARIA:
+    return share.transactional != HA_CHOICE_YES;
+  case DB_TYPE_INNODB:
+  default:
+    return false;
+  }
+}
+
+/* Move the TABLE objects for tables to be copied from the free list
+(where they should all be due to locking) */
+static void tc_get_tables_to_purge_for_backup(TDC_element *element,
+                                                 Share_free_tables::List *purge_tables)
+{
+  for (uint32 i= 0; i < tc_instances; i++)
+  {
+    mysql_mutex_lock(&tc[i].LOCK_table_cache);
+    auto free_list= element->free_tables[i].list;
+    Share_free_tables::List::Iterator it(free_list);
+    while (TABLE *table= it++)
+    {
+      if (table_needs_flush_for_backup(*table->s))
+      {
+        free_list.remove(table);
+        tc[i].records--;
+        tc[i].free_tables.remove(table);
+        DBUG_ASSERT(element->all_tables_refs == 0);
+        element->all_tables.remove(table);
+        purge_tables->push_front(table);
+      }
+    }
+    mysql_mutex_unlock(&tc[i].LOCK_table_cache);
+  }
+#ifndef DBUG_OFF
+  /* Verify in debug that none of the tables for which files are to be
+  copied remain in the cache (due to not being on the free list) */
+  All_share_tables_list::Iterator it(element->all_tables);
+  while (auto table= it++)
+  {
+    DBUG_ASSERT (!table_needs_flush_for_backup(*table->s));
+  }
+#endif
+}
+
+static my_bool tc_backup_purge_callback(void *_element, void *_purge_tables)
+{
+  TDC_element *element= static_cast<TDC_element *>(_element);
+  Share_free_tables::List *purge_tables=
+      static_cast<Share_free_tables::List *>(_purge_tables);
+  mysql_mutex_lock(&element->LOCK_table_share);
+  tc_get_tables_to_purge_for_backup(element, purge_tables);
+  mysql_mutex_unlock(&element->LOCK_table_share);
+  return FALSE;
+}
+
+/* Only ARCHIVE tables need to be purged from TDC,
+as it closes the archive */
+static bool tdc_needs_purge_for_backup(TDC_element *element)
+{
+  if (!element->share)
+    return false;
+
+  if (!element->share->db_plugin)
+    return false;
+
+  return backup_files_hton(*element->share)->db_type == DB_TYPE_ARCHIVE_DB;
+}
+
+/* Selectlively purge the TDC */
+static void tdc_purge_for_backup()
+{
+  /* Singly-linked list of purged elements */
+  TDC_element *purge_list {nullptr};
+  mysql_mutex_lock(&LOCK_unused_shares);
+  decltype(unused_shares)::Iterator it(unused_shares);
+  while (TDC_element *element= it++)
+  {
+    mysql_mutex_lock(&element->LOCK_table_share);
+    if (tdc_needs_purge_for_backup(element))
+    {
+      /* Table is not being accessed due to backup lock and
+      table cache entries should have been purged previously */
+      DBUG_ASSERT (element->ref_count == 0);
+      unused_shares.remove(element);
+      element->prev= 0;
+      element->next= purge_list;
+      purge_list= element;
+    }
+    else
+      mysql_mutex_unlock(&element->LOCK_table_share);
+  }
+  mysql_mutex_unlock(&LOCK_unused_shares);
+  while (purge_list)
+  {
+    TDC_element *element= purge_list;
+    purge_list= element->next;
+    element->next= nullptr;
+    tdc_delete_share_from_hash(element);
+  }
+}
+
+/**
+  Perform TC/TDC actions needed to copy files from the data directory
+  as part of backup, after acquiring a DDL-level MDL backup lock.
+  Implementation is tied to which storage engines are supported from
+  the SQL layer.
+*/
+void tc_purge_for_backup()
+{
+  DBUG_ENTER("tc_purge_for_backup");
+  /* This is similart to calling tc_backup adn then
+  tdc_backup(true), but only appied to some tables. */
+  Share_free_tables::List purge_tables;
+
+  tdc_iterate(0, tc_backup_purge_callback, &purge_tables);
+  while (auto table= purge_tables.pop_front())
+    intern_close_table(table);
+
+  tdc_purge_for_backup();
+  DBUG_VOID_RETURN;
+}
 
 /**
   Prepare table share for use with table definition cache.
diff --git a/sql/table_cache.h b/sql/table_cache.h
index 4847e53b63c9d..811ad0e43880d 100644
--- a/sql/table_cache.h
+++ b/sql/table_cache.h
@@ -94,6 +94,14 @@ extern void tc_add_table(THD *thd, TABLE *table);
 extern void tc_release_table(TABLE *table);
 extern TABLE *tc_acquire_table(THD *thd, TDC_element *element);
 
+/**
+  Perform TC/TDC actions needed to copy files from the data directory
+  as part of backup, after acquiring a DDL-level MDL backup lock.
+  Implementation is tied to which storage engines are supported from
+  the SQL layer.
+*/
+void tc_purge_for_backup();
+
 /**
   Create a table cache key for non-temporary table.
 
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 802ad0f9fcc19..991ce6f7611eb 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -3944,7 +3944,7 @@ static int ha_maria_init(void *p)
   maria_hton->end_backup= maria_end_backup;
   maria_hton->update_optimizer_costs= aria_update_optimizer_costs;
   maria_hton->backup_start= aria_backup_start;
-  //maria_hton->backup_step= aria_backup_step;
+  maria_hton->backup_step= aria_backup_step;
   maria_hton->backup_end= aria_backup_end;
 
   /* TODO: decide if we support Maria being used for log tables */
diff --git a/storage/maria/ma_backup_server.cc b/storage/maria/ma_backup_server.cc
index 25777dff3c19a..93862680a1ecd 100644
--- a/storage/maria/ma_backup_server.cc
+++ b/storage/maria/ma_backup_server.cc
@@ -16,12 +16,12 @@
 #include "maria_def.h"
 #include "ma_backup_server.h"
 #include "mysqld_error.h"
-#if 1 // tc_purge(), tdc_purge()
-# include "sql_class.h"
-# include "table_cache.h"
-#endif
-#include <memory>
+#include "sql_class.h"
+#include <aria_backup.h>
+#include <mysqld_error.h>
+#include <atomic>
 #include <string>
+#include <string_view>
 #include <utility>
 #include <vector>
 #include "span.h"
@@ -33,7 +33,6 @@
 
 namespace
 {
-  /** Backup state; protected by log_sys.latch */
   class Aria_backup
   {
   public:
@@ -41,8 +40,6 @@ namespace
     ~Aria_backup()
     {
 #ifndef _WIN32
-      if (datadir_fd >= 0)
-        std::ignore= close(datadir_fd);
       if (logdir_fd >= 0)
         std::ignore= close(logdir_fd);
 #endif
@@ -57,12 +54,9 @@ namespace
       (mysql_real_data_home), while the transaction logs and control file
       live under aria_log_dir_path (maria_data_root). These differ when
       aria_log_dir_path is set, so open and scan them separately. */
-      datadir_fd= open(mysql_real_data_home, O_DIRECTORY);
+      datadir_fd= get_datadir_fd();
       if (datadir_fd < 0)
-      {
-        my_error(ER_CANT_READ_DIR, MYF(0), mysql_real_data_home, errno);
         return true;
-      }
       logdir_fd= open(maria_data_root, O_DIRECTORY);
       if (logdir_fd < 0)
       {
@@ -76,38 +70,116 @@ namespace
       return false;
     }
 
-    int end(const backup_target &target, const backup_sink &sink) noexcept
+    bool start_copy_dml_safe(const backup_target *target, const backup_sink *sink) noexcept
+    {
+      assert(translog_purge_disabled);
+      if (scan_dbdirs())
+        return true;
+      flatten_table_lists();
+      if (sink->stream == sink->NO_STREAM)
+        return ensure_target_dirs(target);
+      return false;
+    }
+
+    bool start_copy_unsafe() noexcept
+    {
+      if (scan_logs())
+        return true;
+      return false;
+    }
+
+    /* Copy an Aria table that is safe to be copied while concurrent DML
+    is in progress. */
+    int dml_safe_copy_step(const backup_target *target, const backup_sink *sink) noexcept
+    {
+      return copy_from_list_step(flat_table_list, tables_copied,
+                                 [this, target, sink](const table_ref &table) noexcept
+                                 {
+                                   return copy_table(target, sink, table);
+                                 });
+    }
+
+    /* Copy an entity that is not safe to copy if there are concurrent
+    writes to it. One entity is copied, of the first category that has
+    any remaning entities to be copied. Returns the total number of
+    entities to be copied in all categories. Categories in order:
+     - log control file
+     - log files
+     - Aria tables
+     - other ("miscellaneous") files
+    */
+    int unsafe_copy_step(const backup_target *target, const backup_sink *sink) noexcept
+    {
+      /* If control file is always the first file copied and there is only
+      one, it is never included in the "steps remaining" calculation.
+      Should the order be changed, the calculation needs to be updated for
+      the control file as well. */
+      if (have_control_file)
+      {
+        bool already_copied= control_file_copied.exchange(true);
+        if (!already_copied)
+        {
+          if (copy_control_file(target, sink) != 0)
+            return -1;
+          size_t current_copied= log_files_copied.load(std::memory_order_relaxed);
+          return (current_copied < log_files.size()) ?
+                 static_cast<int>(log_files.size() - current_copied) :
+                 0;
+        }
+      }
+
+      return copy_from_list_step(log_files, log_files_copied,
+                                 [this, target, sink](const std::string &path) noexcept
+                                 {
+                                   return copy_log_file(target, sink, path.c_str());
+                                 });
+    }
+
+    int end() noexcept
     {
-      int ret_val= perform_backup(target, sink);
       assert(translog_purge_disabled);
       translog_purge_disabled= false;
       translog_enable_purge();
-      return ret_val;
+      return 0;
     }
   private:
 #ifndef _WIN32
-    /** The server data directory (Aria table files) */
+    /** The server data directory */
     int datadir_fd{-1};
     /** The Aria log directory aria_log_dir_path (logs, control file) */
     int logdir_fd{-1};
 #endif
     /** whether the Aria translog_disable_purge() is in effect */
     bool translog_purge_disabled{false};
-    static constexpr const char zerobuf[511]{};
+
+    /* File extensions are 4 characters long (dot and 3 letter extension) */
+    static constexpr size_t ext_len= 4;
+    static constexpr const char* data_ext {MARIA_NAME_DEXT};
+    static constexpr const char* index_ext {MARIA_NAME_IEXT};
+    static constexpr LEX_CSTRING log_file_prefix {C_STRING_WITH_LEN("aria_log.")};
+    static constexpr LEX_CSTRING tmp_prefix {C_STRING_WITH_LEN(tmp_file_prefix)};
+    static constexpr LEX_CSTRING control_file_name {C_STRING_WITH_LEN("aria_log_control")};
+    using dir_name = std::string;
     using dir_contents = std::vector<std::string>;
-    using database_dir = std::pair<std::string, dir_contents>;
-    std::vector<database_dir> database_dirs;
+    using database_dir = std::pair<dir_name, dir_contents>;
+    using database_dirs = std::vector<database_dir>;
+    /* Collection of tables to be backed up. */
+    database_dirs tables;
+    /* Aria log files */
     std::vector<std::string> log_files;
+
     bool have_control_file = false;
 
-    int perform_backup(const backup_target &target, const backup_sink &sink)
-      noexcept
-    {
-      return scan_datadir() || copy_databases(target, sink) ||
-        copy_control_file(target, sink) ||
-        translog_flush(translog_get_horizon()) ||
-        copy_logs(target, sink);
-    }
+    /* Refer to a string stored elsewhere */
+    using dir_ref= std::string_view;
+    using tablename_ref= std::string_view;
+    using table_ref= std::pair<dir_ref, tablename_ref>;
+    using table_list= std::vector<table_ref>;
+
+    table_list flat_table_list;
+    std::atomic<size_t> tables_copied {0};
+    std::atomic<size_t> log_files_copied {0};
+    std::atomic<bool> control_file_copied {false};
 
     ATTRIBUTE_COLD ATTRIBUTE_NOINLINE
       static int dir_error(const char *name) noexcept
@@ -116,9 +188,8 @@ namespace
       return 1;
     }
 
-    int scan_datadir() noexcept
+    int scan_dbdirs() noexcept
     {
-      /* Scan the server data directory for Aria table files. */
       MY_DIR *data_dir= my_dir(mysql_real_data_home, MYF(MY_WANT_STAT));
       if (!data_dir)
         return dir_error(mysql_real_data_home);
@@ -127,27 +198,14 @@ namespace
              st_::span<const fileinfo>{data_dir->dir_entry,
                                        data_dir->number_of_files})
         if ((fi.mystat->st_mode & S_IFMT) == S_IFDIR)
-          if ((fail= scan_database_dir(fi.name)) != 0)
-            break;
+        {
+          fail= scan_database_dir(fi.name);
+          if (fail != 0)
+            goto func_exit;
+        }
+    func_exit:
       my_dirend(data_dir);
-      if (fail)
-        return fail;
-
-      /* Scan aria_log_dir_path for the transaction logs and control file. */
-      MY_DIR *log_dir= my_dir(maria_data_root, MYF(MY_WANT_STAT));
-      if (!log_dir)
-        return dir_error(maria_data_root);
-      for (const fileinfo &fi :
-             st_::span<const fileinfo>{log_dir->dir_entry,
-                                       log_dir->number_of_files})
-      {
-        if (!strncmp(fi.name, C_STRING_WITH_LEN("aria_log.")))
-          log_files.emplace_back(fi.name);
-        else if (!strcmp(fi.name, "aria_log_control"))
-          have_control_file = true;
-      }
-      my_dirend(log_dir);
-      return 0;
+      return fail;
     }
 
     int scan_database_dir(const char* dir_name) noexcept
@@ -156,250 +214,178 @@ namespace
       MY_DIR *dir_info= my_dir(dir_path.c_str(), MYF(MY_WANT_STAT));
       if (!dir_info)
         return dir_error(dir_path.c_str());
-      std::vector<std::string> files_to_backup;
+      dir_contents dir_tables;
       for (const fileinfo &fi :
              st_::span<const fileinfo>{dir_info->dir_entry,
                                        dir_info->number_of_files})
-        if (is_db_file(fi.name))
-          files_to_backup.emplace_back(fi.name);
-      if (!files_to_backup.empty())
-        database_dirs.emplace_back(dir_name, std::move(files_to_backup));
+      {
+        const LEX_CSTRING filename {fi.name, strlen(fi.name)};
+        if (filename.length >= ext_len)
+        {
+          /* Length of filename without extension. */
+          size_t base_filename_len= filename.length - ext_len;
+          const char* suffix = filename.str + base_filename_len;
+          if(match_ext(suffix, index_ext))
+          {
+            if (!is_tmp_table(filename))
+            {
+              dir_tables.emplace_back(filename.str, base_filename_len);
+            }
+          }
+        }
+      }
+      if (!dir_tables.empty())
+        tables.emplace_back(dir_name, std::move(dir_tables));
       my_dirend(dir_info);
       return 0;
     }
 
-    int copy_databases(const backup_target &target, const backup_sink &sink)
-      noexcept
+    static bool is_tmp_table(const LEX_CSTRING &filename) noexcept
+    {
+      return begins_with(filename, tmp_prefix);
+    }
+
+    void flatten_table_lists() noexcept
     {
-      for (const database_dir &dir : database_dirs)
+      flatten_table_list(tables, flat_table_list);
+    }
+
+    static void flatten_table_list(const database_dirs& dirs, table_list& list) noexcept
+    {
+      for (const database_dir& dir : dirs)
       {
-        if (sink.stream != sink.NO_STREAM);
-        else if (int fail= ensure_target_subdir(target, dir.first.c_str()))
-          return fail;
-        if (int fail= copy_database(target, sink, dir))
-          return fail;
+        for (const std::string& table : dir.second)
+          list.emplace_back(dir.first, table);
       }
+    }
+
+    int scan_logs() noexcept
+    {
+      const char *base_dir= maria_data_root;
+      MY_DIR *dir_info= my_dir(base_dir, MYF(MY_WANT_STAT));
+      if (!dir_info)
+        return dir_error(base_dir);
+      for (const fileinfo &fi :
+             st_::span<const fileinfo>{dir_info->dir_entry,
+                                       dir_info->number_of_files})
+      {
+        const LEX_CSTRING filename {fi.name, strlen(fi.name)};
+        if (begins_with(filename, log_file_prefix))
+          log_files.emplace_back(LEX_STRING_WITH_LEN(filename));
+        else if (is_control_file_name(filename))
+          have_control_file = true;
+      }
+      my_dirend(dir_info);
       return 0;
     }
 
-    /*
-       Create directory in the target directory if it does not exist.
-       Return 0 on success, non-0 on failure. Set errno in case of failure
-    */
-    int ensure_target_subdir(const backup_target &target, const char *name)
-      noexcept
+    bool ensure_target_dirs(const backup_target *target) noexcept
     {
-#ifdef _WIN32
-      if (CreateDirectory(make_path(target.path, name).c_str(), nullptr))
-        return 0;
-      DWORD err= GetLastError();
-      if (err == ERROR_ALREADY_EXISTS)
-        return 0;
-      my_osmaperr(err);
-#else
-      if (likely(!mkdirat(target.fd, name, 0777) || errno == EEXIST))
-        return 0;
-#endif
-      my_error(ER_CANT_CREATE_FILE, MYF(0), name, errno);
-      return 1;
+      for (const database_dir &dir : tables)
+        if(::ensure_target_subdir(target, dir.first.c_str()) != 0)
+          return true;
+      return false;
     }
 
-    int copy_database(const backup_target &target, const backup_sink &sink,
-                      const database_dir& dir) noexcept
+    template<typename T, typename Fn>
+    static int copy_from_list_step(const std::vector<T> &list,
+                                   std::atomic<size_t> &copied,
+                                   Fn copy_action)
     {
-      std::string file_path;
-      for (const std::string &file : dir.second)
+      size_t idx= copied.fetch_add(1, std::memory_order_relaxed);
+      if (idx < list.size())
       {
-        file_path= dir.first;
-        file_path.push_back('/');
-        file_path.append(file);
-        if (int fail= copy_file(target, sink, file_path.c_str(), false))
-          return fail;
+        if (copy_action(list[idx]) != 0)
+            return -1;
+        return static_cast<int>(list.size() - idx - 1U);
       }
+
       return 0;
     }
 
-    int copy_control_file(const backup_target &target, const backup_sink &sink)
-      noexcept
+    int copy_table(const backup_target *target, const backup_sink *sink,
+                   const table_ref& table) noexcept
+    {
+      dir_ref dir_name = table.first;
+      tablename_ref table_name = table.second;
+      std::string index_path;
+      index_path.reserve(dir_name.size() + table_name.size() + 5);
+      index_path= dir_name;
+      index_path += '/';
+      index_path.append(table_name.begin(), table_name.end());
+      std::string data_path;
+      data_path.reserve(dir_name.size() + table_name.size() + 5);
+      data_path= index_path;
+      index_path+= index_ext;
+      data_path+= data_ext;
+
+      return copy_table_file(target, sink, index_path) ||
+             copy_table_file(target, sink, data_path);
+    }
+
+    int copy_control_file(const backup_target *target, const backup_sink *sink) noexcept
     {
       if (!have_control_file)
         return 0;
-      return copy_file(target, sink, "aria_log_control", true);
+      return copy_log_file(target, sink, control_file_name.str);
     }
 
-    int copy_logs(const backup_target &target, const backup_sink &sink)
-      noexcept
+    int copy_table_file(const backup_target *target,
+                        const backup_sink *sink, 
+                        const std::string &path) const noexcept
     {
-      for (const std::string &file : log_files)
-        if (int fail= copy_file(target, sink, file.c_str(), true))
-          return fail;
-      return 0;
+      return copy_table_file(target, sink, path.c_str());
+    }
+
+    int copy_table_file(const backup_target *target,
+                        const backup_sink *sink,
+                        const char *path) const noexcept
+    {
+#ifdef _WIN32
+      return ::copy_datafile_to_target(path, target, sink);
+#else
+      return ::copy_datafile_to_target(datadir_fd, path, target, sink);
+#endif
     }
 
-    int copy_file(const backup_target &target, const backup_sink &sink,
-                  const char *path, bool is_log) const noexcept
+    int copy_log_file(const backup_target *target,
+                      const backup_sink *sink,
+                      const char *filename)
     {
 #ifndef _WIN32
-      int ret_val{0};
-      int src_fd{openat(is_log ? logdir_fd : datadir_fd, path, O_RDONLY)};
+      int src_fd = openat(logdir_fd, filename, O_RDONLY);
       if (src_fd < 0)
       {
-        my_error(ER_CANT_OPEN_FILE, MYF(0), path, errno);
+        my_error(ER_CANT_OPEN_FILE, MYF(0),
+                 make_path(maria_data_root, filename).c_str(),
+                 errno);
         return 1;
       }
-      int tgt_fd{sink.stream};
-      if (tgt_fd == sink.NO_STREAM)
-      {
-        tgt_fd= openat(target.fd, path,
-                       O_CREAT | O_EXCL | O_WRONLY, 0666);
-        if (tgt_fd < 0)
-        {
-          my_error(ER_CANT_CREATE_FILE, MYF(0), path, errno);
-          ret_val= 1;
-        }
-        else
-        {
-          ret_val= copy_entire_file(src_fd, tgt_fd);
-          if (ret_val | close(tgt_fd))
-          {
-          write_error:
-            my_error(ER_ERROR_ON_WRITE, MYF(0), path, errno);
-            ret_val= 1;
-          }
-        }
-      }
-      else
-      {
-        uint64_t end= uint64_t(lseek(src_fd, 0, SEEK_END));
-        if (backup_stream_start(tgt_fd, path, 0644, end, nullptr, 0) ||
-            backup_stream_append(src_fd, tgt_fd, 0, end))
-          goto write_error;
-        if (size_t pad= size_t(end) & 511)
-          if (backup_stream_write(tgt_fd, zerobuf, 512 - pad))
-            goto write_error;
-      }
-
+      int ret_val=  copy_fd_to_target(src_fd, target, filename, sink);
       close(src_fd);
       return ret_val;
 #else
-      const std::string src_path
-        {make_path(is_log ? maria_data_root : mysql_real_data_home, path)};
-
-      if (sink.stream == sink.NO_STREAM)
-      {
-        std::string dest_path{make_path(target.path, path)};
-        if (!CopyFileEx(src_path.c_str(), dest_path.c_str(),
-                        nullptr, nullptr, nullptr, COPY_FILE_NO_BUFFERING))
-        {
-          my_osmaperr(GetLastError());
-          my_error(ER_CANT_CREATE_FILE, MYF(0), dest_path.c_str(), errno);
-          return 1;
-        }
-      }
-      else
-      {
-        HANDLE src, dst{sink.stream};
-        for (;;)
-        {
-          src= CreateFile(src_path.c_str(), GENERIC_READ,
-                          FILE_SHARE_READ|FILE_SHARE_WRITE|FILE_SHARE_DELETE,
-                          my_win_file_secattr(), OPEN_EXISTING,
-                          FILE_ATTRIBUTE_NORMAL, nullptr);
-          if (src != INVALID_HANDLE_VALUE)
-            break;
-          switch (GetLastError()) {
-          case ERROR_SHARING_VIOLATION:
-          case ERROR_LOCK_VIOLATION:
-            std::this_thread::sleep_for(std::chrono::seconds(1));
-            continue;
-          }
-
-          my_osmaperr(GetLastError());
-          my_error(ER_FILE_NOT_FOUND, MYF(ME_ERROR_LOG), src_path.c_str(),
-                   errno);
-          return -1;
-        }
-
-        LARGE_INTEGER li;
-        if (!GetFileSizeEx(src, &li))
-        {
-        write_error:
-          my_osmaperr(GetLastError());
-          my_error(ER_ERROR_ON_WRITE, MYF(0), path, errno);
-          if (src != INVALID_HANDLE_VALUE)
-            CloseHandle(src);
-          return -1;
-        }
-
-        if (backup_stream_start(dst, path, 0644, li.QuadPart, nullptr, 0) ||
-            backup_stream_append_plain(src, dst, 0, li.QuadPart))
-          goto write_error;
-
-        if (size_t pad= size_t(li.LowPart) & 511)
-          if (backup_stream_write(dst, zerobuf, 512 - pad))
-            goto write_error;
-        if (!CloseHandle(src))
-        {
-          src= INVALID_HANDLE_VALUE;
-          goto write_error;
-        }
-      }
-      return 0;
+      return copy_entire_file(make_path(maria_data_root, filename).c_str(),
+                              filename, target, sink);
 #endif
     }
 
+    static bool match_ext(const char* ext1, const char* ext2) noexcept
+    {
+      return memcmp(ext1, ext2, ext_len) == 0;
+    }
 
-    static bool is_db_file(const char* file_name) noexcept
+    static bool begins_with(const LEX_CSTRING &str, const LEX_CSTRING &prefix) noexcept
     {
-      size_t len= strlen(file_name);
-      if (len < 4)
+      if (str.length < prefix.length)
         return false;
-      uint32_t suffix;
-      memcpy(&suffix, file_name + len - 4, 4);
-      switch (suffix) {
-      default:
-        return len == 6 && !memcmp(file_name, C_STRING_WITH_LEN("db.opt"));
-#ifdef WORDS_BIGENDIAN
-      case 0x2e41524d: /* .ARM ENGINE=ARCHIVE metadata */
-      case 0x2e41525a: /* .ARZ ENGINE=ARCHIVE compressed data */
-      case 0x2e43534d: /* .CSM ENGINE=CSV metadata */
-      case 0x2e435356: /* .CSV ENGINE=CSV data ("comma separated values") */
-      case 0x2e4d4144: /* .MAD ENGINE=Aria data heap */
-      case 0x2e4d4149: /* .MAI ENGINE=Aria indexes */
-      case 0x2e4d5247: /* .MRG ENGINE=MRG_MyISAM */
-      case 0x2e4d5944: /* .MYD ENGINE=MyISAM data heap */
-      case 0x2e4d5949: /* .MYI ENGINE=MyISAM indexes */
-      case 0x2e66726d: /* .frm form (SHOW CREATE TABLE) */
-      case 0x2e706172: /* .par PARTITION metadata */
-#else
-      case 0x4d52412e: /* .ARM ENGINE=ARCHIVE metadata */
-      case 0x5a52412e: /* .ARZ ENGINE=ARCHIVE compressed data */
-      case 0x4d53432e: /* .CSM ENGINE=CSV metadata */
-      case 0x5653432e: /* .CSV ENGINE=CSV data ("comma separated values") */
-      case 0x44414d2e: /* .MAD ENGINE=Aria data heap */
-      case 0x49414d2e: /* .MAI ENGINE=Aria indexes */
-      case 0x47524d2e: /* .MRG ENGINE=MRG_MyISAM */
-      case 0x44594d2e: /* .MYD ENGINE=MyISAM data heap */
-      case 0x49594d2e: /* .MYI ENGINE=MyISAM indexes */
-      case 0x6d72662e: /* .frm form (SHOW CREATE TABLE) */
-      case 0x7261702e: /* .par PARTITION metadata */
-#endif
-        return true;
-      }
+      return memcmp(str.str, prefix.str, prefix.length) == 0;
     }
 
-    /**
-       Construct a file path.
-       @param dir   directory name
-       @param name  file name
-       @return dir/name
-    */
-    static std::string make_path(const char *dir, const char *name)
+    static bool is_control_file_name(const LEX_CSTRING &str)
     {
-      std::string path{dir};
-      path.push_back('/');
-      path.append(name);
-      return path;
+      return str.length == control_file_name.length &&
+        memcmp(str.str, control_file_name.str, control_file_name.length) == 0;
     }
   };
 }
@@ -407,43 +393,71 @@ namespace
 void *aria_backup_start(THD *thd, const backup_target *target,
                         backup_phase phase, const backup_sink *sink) noexcept
 {
-  switch (phase) {
-  case BACKUP_PHASE_PREPARE_START:
+  Aria_backup *aria_backup {};
+  if (phase == BACKUP_PHASE_PREPARE_START)
+  {
     return 0;
-  default:
-    return sink->ha_data;
-  case BACKUP_PHASE_NO_COMMIT:
+  }
+  else if (phase == BACKUP_PHASE_START)
+  {
     assert(!sink->ha_data);
-    Aria_backup *aria_backup{new Aria_backup};
+    aria_backup= new Aria_backup();
     if (aria_backup->initialize())
     {
       delete aria_backup;
-      return reinterpret_cast<void*>(-1);
+      goto error;
     }
     return aria_backup;
   }
+
+  assert (sink->ha_data != reinterpret_cast<void*>(-1));
+  aria_backup= static_cast<Aria_backup*>(sink->ha_data);
+  assert(aria_backup);
+  switch(phase)
+  {
+  case BACKUP_PHASE_NO_DDL:
+    if (aria_backup->start_copy_dml_safe(target, sink))
+      goto error;
+    break;
+  case BACKUP_PHASE_NO_COMMIT:
+    if (aria_backup->start_copy_unsafe())
+      goto error;
+    break;
+  default:
+    break;
+  }
+  return sink->ha_data;
+error:
+  return reinterpret_cast<void*>(-1);
 }
 
-#if 0 // FIXME: implement the actual copying here
-int aria_backup_step(THD*, const backup_target*, backup_phase,
-                     const backup_sink*) noexcept
+
+int aria_backup_step(THD*, const backup_target *target, backup_phase phase,
+                     const backup_sink *sink) noexcept
 {
-  return 0;
+  assert (sink->ha_data != reinterpret_cast<void*>(-1));
+  Aria_backup *aria_backup= static_cast<Aria_backup*>(sink->ha_data);
+  assert(aria_backup);
+  switch (phase)
+  {
+  case BACKUP_PHASE_NO_DDL:
+    return aria_backup->dml_safe_copy_step(target, sink);
+  case BACKUP_PHASE_NO_COMMIT:
+    return aria_backup->unsafe_copy_step(target, sink);
+  default:
+    return 0;
+  }
 }
-#endif
 
 int aria_backup_end(THD *thd, const backup_target *target, backup_phase phase,
                     const backup_sink *sink) noexcept
 {
+  assert (sink->ha_data != reinterpret_cast<void*>(-1));
   Aria_backup *aria_backup= static_cast<Aria_backup*>(sink->ha_data);
   switch (phase) {
   case BACKUP_PHASE_NO_COMMIT:
     assert(aria_backup);
-#if 1 // FIXME: invoke these only for Aria, MyISAM, CSV but not others
-    tc_purge();
-    tdc_purge(true);
-#endif
-    return aria_backup->end(*target, *sink);
+    return aria_backup->end();
   case BACKUP_PHASE_FINISH:
     delete aria_backup;
     /* fall through */