MDEV-31039 mariadb-backup: remove global variables ds_data and ds_meta

This is a non-functional change.

simplifying the code logic:
- removing global variables ds_data and ds_meta
- passing these variables as parameters to functions instead
- adding helper classes: Datasink_free_list and Backup_datasinks
- moving some function accepting a ds_ctxt parameter
  as methods to ds_ctxt.

9 files changed, 289 insertions, 221 deletions

diff --git a/extra/mariabackup/backup_copy.cc b/extra/mariabackup/backup_copy.cc
index 608be7125bd..ffaf6dc98e4 100644
--- a/extra/mariabackup/backup_copy.cc
+++ b/extra/mariabackup/backup_copy.cc
@@ -73,9 +73,8 @@ bool binlog_locked;
 
 static void rocksdb_create_checkpoint();
 static bool has_rocksdb_plugin();
-static void copy_or_move_dir(const char *from, const char *to, bool copy, bool allow_hardlinks);
-static void rocksdb_backup_checkpoint();
-static void rocksdb_copy_back();
+static void rocksdb_backup_checkpoint(ds_ctxt *ds_data);
+static void rocksdb_copy_back(ds_ctxt *ds_data);
 
 static bool is_abs_path(const char *path)
 {
@@ -131,7 +130,7 @@ struct datadir_thread_ctxt_t {
 	bool			ret;
 };
 
-static bool backup_files_from_datadir(const char *dir_path);
+static bool backup_files_from_datadir(ds_ctxt *ds_data, const char *dir_path);
 
 /************************************************************************
 Retirn true if character if file separator */
@@ -804,7 +803,7 @@ if passes the rules for partial backup.
 @return true if file backed up or skipped successfully. */
 static
 bool
-datafile_copy_backup(const char *filepath, uint thread_n)
+datafile_copy_backup(ds_ctxt *ds_data, const char *filepath, uint thread_n)
 {
 	const char *ext_list[] = {"frm", "isl", "MYD", "MYI", "MAD", "MAI",
 		"MRG", "TRG", "TRN", "ARM", "ARZ", "CSM", "CSV", "opt", "par",
@@ -825,7 +824,7 @@ datafile_copy_backup(const char *filepath, uint thread_n)
 	}
 
 	if (filename_matches(filepath, ext_list)) {
-		return copy_file(ds_data, filepath, filepath, thread_n);
+		return ds_data->copy_file(filepath, filepath, thread_n);
 	}
 
 	return(true);
@@ -866,7 +865,8 @@ datafile_rsync_backup(const char *filepath, bool save_to_list, FILE *f)
 	return(true);
 }
 
-bool backup_file_print_buf(const char *filename, const char *buf, int buf_len)
+bool ds_ctxt_t::backup_file_print_buf(const char *filename,
+                                      const char *buf, int buf_len)
 {
 	ds_file_t	*dstfile	= NULL;
 	MY_STAT		 stat;			/* unused for now */
@@ -877,7 +877,7 @@ bool backup_file_print_buf(const char *filename, const char *buf, int buf_len)
 	stat.st_size = buf_len;
 	stat.st_mtime = my_time(0);
 
-	dstfile = ds_open(ds_data, filename, &stat);
+	dstfile = ds_open(this, filename, &stat);
 	if (dstfile == NULL) {
 		msg("error: Can't open the destination stream for %s",
 			filename);
@@ -916,9 +916,9 @@ error_close:
 	return true;
 };
 
-static
 bool
-backup_file_vprintf(const char *filename, const char *fmt, va_list ap)
+ds_ctxt_t::backup_file_vprintf(const char *filename,
+                               const char *fmt, va_list ap)
 {
 	char		*buf		= 0;
 	int		 buf_len;
@@ -929,7 +929,7 @@ backup_file_vprintf(const char *filename, const char *fmt, va_list ap)
 }
 
 bool
-backup_file_printf(const char *filename, const char *fmt, ...)
+ds_ctxt_t::backup_file_printf(const char *filename, const char *fmt, ...)
 {
 	bool result;
 	va_list ap;
@@ -1055,16 +1055,15 @@ static int fix_win_file_permissions(const char *file)
 Copy file for backup/restore.
 @return true in case of success. */
 bool
-copy_file(ds_ctxt_t *datasink,
-	  const char *src_file_path,
-	  const char *dst_file_path,
-	  uint thread_n)
+ds_ctxt_t::copy_file(const char *src_file_path,
+                     const char *dst_file_path,
+                     uint thread_n)
 {
 	char			 dst_name[FN_REFLEN];
 	ds_file_t		*dstfile = NULL;
 	datafile_cur_t		 cursor;
 	xb_fil_cur_result_t	 res;
-	DBUG_ASSERT(datasink->datasink->remove);
+	DBUG_ASSERT(datasink->remove);
 	const char	*dst_path =
 		(xtrabackup_copy_back || xtrabackup_move_back)?
 		dst_file_path : trim_dotslash(dst_file_path);
@@ -1075,7 +1074,7 @@ copy_file(ds_ctxt_t *datasink,
 
 	strncpy(dst_name, cursor.rel_path, sizeof(dst_name));
 
-	dstfile = ds_open(datasink, dst_path, &cursor.statinfo);
+	dstfile = ds_open(this, dst_path, &cursor.statinfo);
 	if (dstfile == NULL) {
 		msg(thread_n,"error: "
 			"cannot open the destination stream for %s", dst_name);
@@ -1112,7 +1111,7 @@ copy_file(ds_ctxt_t *datasink,
 error:
 	datafile_close(&cursor);
 	if (dstfile != NULL) {
-		datasink->datasink->remove(dstfile->path);
+		datasink->remove(dstfile->path);
 		ds_close(dstfile);
 	}
 
@@ -1126,12 +1125,10 @@ error_close:
 Try to move file by renaming it. If source and destination are on
 different devices fall back to copy and unlink.
 @return true in case of success. */
-static
 bool
-move_file(ds_ctxt_t *datasink,
-	  const char *src_file_path,
-	  const char *dst_file_path,
-	  const char *dst_dir, uint thread_n)
+ds_ctxt_t::move_file(const char *src_file_path,
+                     const char *dst_file_path,
+                     const char *dst_dir, uint thread_n)
 {
 	char errbuf[MYSYS_STRERROR_SIZE];
 	char dst_file_path_abs[FN_REFLEN];
@@ -1158,7 +1155,7 @@ move_file(ds_ctxt_t *datasink,
 	if (my_rename(src_file_path, dst_file_path_abs, MYF(0)) != 0) {
 		if (my_errno == EXDEV) {
 			/* Fallback to copy/unlink */
-			if(!copy_file(datasink, src_file_path,
+			if(!copy_file(src_file_path,
 					dst_file_path, thread_n))
 					return false;
 			msg(thread_n,"Removing %s", src_file_path);
@@ -1242,13 +1239,13 @@ Copy or move file depending on current mode.
 @return true in case of success. */
 static
 bool
-copy_or_move_file(const char *src_file_path,
+copy_or_move_file(ds_ctxt *datasink0, const char *src_file_path,
 		  const char *dst_file_path,
 		  const char *dst_dir,
 		  uint thread_n,
 		 bool copy = xtrabackup_copy_back)
 {
-	ds_ctxt_t *datasink = ds_data;		/* copy to datadir by default */
+	ds_ctxt_t *datasink = datasink0; /* copy to datadir by default */
 	char filedir[FN_REFLEN];
 	size_t filedir_len;
 	bool ret;
@@ -1296,13 +1293,13 @@ copy_or_move_file(const char *src_file_path,
 	}
 
 	ret = (copy ?
-		copy_file(datasink, src_file_path, dst_file_path, thread_n) :
-		move_file(datasink, src_file_path, dst_file_path,
+		datasink->copy_file(src_file_path, dst_file_path, thread_n) :
+		datasink->move_file(src_file_path, dst_file_path,
 			  dst_dir, thread_n));
 
 cleanup:
 
-	if (datasink != ds_data) {
+	if (datasink != datasink0) {
 		ds_destroy(datasink);
 	}
 
@@ -1314,7 +1311,7 @@ cleanup:
 
 static
 bool
-backup_files(const char *from, bool prep_mode)
+backup_files(ds_ctxt *ds_data, const char *from, bool prep_mode)
 {
 	char rsync_tmpfile_name[FN_REFLEN];
 	FILE *rsync_tmpfile = NULL;
@@ -1352,7 +1349,7 @@ backup_files(const char *from, bool prep_mode)
 				ret = datafile_rsync_backup(node.filepath,
 					!prep_mode, rsync_tmpfile);
 			} else {
-				ret = datafile_copy_backup(node.filepath, 1);
+				ret = datafile_copy_backup(ds_data, node.filepath, 1);
 			}
 			if (!ret) {
 				msg("Failed to copy file %s", node.filepath);
@@ -1363,7 +1360,7 @@ backup_files(const char *from, bool prep_mode)
 			char path[FN_REFLEN];
 			snprintf(path, sizeof(path),
 				 "%s/db.opt", node.filepath);
-			if (!(ret = backup_file_printf(
+			if (!(ret = ds_data->backup_file_printf(
 					trim_dotslash(path), "%s", ""))) {
 				msg("Failed to create file %s", path);
 				goto out;
@@ -1452,7 +1449,6 @@ out:
 	return(ret);
 }
 
-void backup_fix_ddl(CorruptedPages &);
 
 lsn_t get_current_lsn(MYSQL *connection)
 {
@@ -1477,7 +1473,8 @@ lsn_t get_current_lsn(MYSQL *connection)
 lsn_t server_lsn_after_lock;
 extern void backup_wait_for_lsn(lsn_t lsn);
 /** Start --backup */
-bool backup_start(CorruptedPages &corrupted_pages)
+bool backup_start(ds_ctxt *ds_data, ds_ctxt *ds_meta,
+                  CorruptedPages &corrupted_pages)
 {
 	if (!opt_no_lock) {
 		if (opt_safe_slave_backup) {
@@ -1486,7 +1483,7 @@ bool backup_start(CorruptedPages &corrupted_pages)
 			}
 		}
 
-		if (!backup_files(fil_path_to_mysql_datadir, true)) {
+		if (!backup_files(ds_data, fil_path_to_mysql_datadir, true)) {
 			return(false);
 		}
 
@@ -1498,11 +1495,11 @@ bool backup_start(CorruptedPages &corrupted_pages)
 		server_lsn_after_lock = get_current_lsn(mysql_connection);
 	}
 
-	if (!backup_files(fil_path_to_mysql_datadir, false)) {
+	if (!backup_files(ds_data, fil_path_to_mysql_datadir, false)) {
 		return(false);
 	}
 
-	if (!backup_files_from_datadir(fil_path_to_mysql_datadir)) {
+	if (!backup_files_from_datadir(ds_data, fil_path_to_mysql_datadir)) {
 		return false;
 	}
 
@@ -1512,7 +1509,7 @@ bool backup_start(CorruptedPages &corrupted_pages)
 
 	msg("Waiting for log copy thread to read lsn %llu", (ulonglong)server_lsn_after_lock);
 	backup_wait_for_lsn(server_lsn_after_lock);
-	backup_fix_ddl(corrupted_pages);
+	corrupted_pages.backup_fix_ddl(ds_data, ds_meta);
 
 	// There is no need to stop slave thread before coping non-Innodb data when
 	// --no-lock option is used because --no-lock option requires that no DDL or
@@ -1528,7 +1525,7 @@ bool backup_start(CorruptedPages &corrupted_pages)
 	if (opt_slave_info) {
 		lock_binlog_maybe(mysql_connection);
 
-		if (!write_slave_info(mysql_connection)) {
+		if (!write_slave_info(ds_data, mysql_connection)) {
 			return(false);
 		}
 	}
@@ -1540,7 +1537,7 @@ bool backup_start(CorruptedPages &corrupted_pages)
 	avoid that is to have a single process, i.e. merge innobackupex and
 	xtrabackup. */
 	if (opt_galera_info) {
-		if (!write_galera_info(mysql_connection)) {
+		if (!write_galera_info(ds_data, mysql_connection)) {
 			return(false);
 		}
 	}
@@ -1548,7 +1545,7 @@ bool backup_start(CorruptedPages &corrupted_pages)
 	if (opt_binlog_info == BINLOG_INFO_ON) {
 
 		lock_binlog_maybe(mysql_connection);
-		write_binlog_info(mysql_connection);
+		write_binlog_info(ds_data, mysql_connection);
 	}
 
 	if (have_flush_engine_logs && !opt_no_lock) {
@@ -1585,20 +1582,20 @@ void backup_release()
 static const char *default_buffer_pool_file = "ib_buffer_pool";
 
 /** Finish after backup_start() and backup_release() */
-bool backup_finish()
+bool backup_finish(ds_ctxt *ds_data)
 {
 	/* Copy buffer pool dump or LRU dump */
 	if (!opt_rsync && opt_galera_info) {
 		if (buffer_pool_filename && file_exists(buffer_pool_filename)) {
-			copy_file(ds_data, buffer_pool_filename, default_buffer_pool_file, 0);
+			ds_data->copy_file(buffer_pool_filename, default_buffer_pool_file, 0);
 		}
 		if (file_exists("ib_lru_dump")) {
-			copy_file(ds_data, "ib_lru_dump", "ib_lru_dump", 0);
+			ds_data->copy_file("ib_lru_dump", "ib_lru_dump", 0);
 		}
 	}
 
 	if (has_rocksdb_plugin()) {
-		rocksdb_backup_checkpoint();
+		rocksdb_backup_checkpoint(ds_data);
 	}
 
 	msg("Backup created in directory '%s'", xtrabackup_target_dir);
@@ -1610,11 +1607,11 @@ bool backup_finish()
 			mysql_slave_position);
 	}
 
-	if (!write_backup_config_file()) {
+	if (!write_backup_config_file(ds_data)) {
 		return(false);
 	}
 
-	if (!write_xtrabackup_info(mysql_connection, XTRABACKUP_INFO,
+	if (!write_xtrabackup_info(ds_data, mysql_connection, XTRABACKUP_INFO,
 				    opt_history != 0, true)) {
 		return(false);
 	}
@@ -1681,6 +1678,7 @@ ibx_copy_incremental_over_full()
 	bool ret = true;
 	char path[FN_REFLEN];
 	int i;
+	ds_ctxt *ds_data= NULL;
 
         DBUG_ASSERT(!opt_galera_info);
 	datadir_node_init(&node);
@@ -1708,15 +1706,15 @@ ibx_copy_incremental_over_full()
 				unlink(node.filepath_rel);
 			}
 
-			if (!(ret = copy_file(ds_data, node.filepath,
-						node.filepath_rel, 1))) {
+			if (!(ret = ds_data->copy_file(node.filepath,
+						       node.filepath_rel, 1))) {
 				msg("Failed to copy file %s",
 					node.filepath);
 				goto cleanup;
 			}
 		}
 
-		if (!(ret = backup_files_from_datadir(xtrabackup_incremental_dir)))
+		if (!(ret = backup_files_from_datadir(ds_data, xtrabackup_incremental_dir)))
 			goto cleanup;
 
 		/* copy supplementary files */
@@ -1731,7 +1729,7 @@ ibx_copy_incremental_over_full()
 				if (file_exists(sup_files[i])) {
 					unlink(sup_files[i]);
 				}
-				copy_file(ds_data, path, sup_files[i], 0);
+				ds_data->copy_file(path, sup_files[i], 0);
 			}
 		}
 
@@ -1745,7 +1743,7 @@ ibx_copy_incremental_over_full()
 			if (my_mkdir(ROCKSDB_BACKUP_DIR, 0777, MYF(0))) {
 				die("my_mkdir failed for " ROCKSDB_BACKUP_DIR);
 			}
-			copy_or_move_dir(path, ROCKSDB_BACKUP_DIR, true, true);
+			ds_data->copy_or_move_dir(path, ROCKSDB_BACKUP_DIR, true, true);
 		}
 		ibx_incremental_drop_databases(xtrabackup_target_dir,
 					       xtrabackup_incremental_dir);
@@ -1894,7 +1892,7 @@ copy_back()
 
 	dst_dir = dst_dir_buf.make(srv_undo_dir);
 
-	ds_data = ds_create(dst_dir, DS_TYPE_LOCAL);
+	ds_ctxt *ds_tmp = ds_create(dst_dir, DS_TYPE_LOCAL);
 
 	for (uint i = 1; i <= TRX_SYS_MAX_UNDO_SPACES; i++) {
 		char filename[20];
@@ -1902,14 +1900,14 @@ copy_back()
 		if (!file_exists(filename)) {
 			break;
 		}
-		if (!(ret = copy_or_move_file(filename, filename,
+		if (!(ret = copy_or_move_file(ds_tmp, filename, filename,
 					      dst_dir, 1))) {
 			goto cleanup;
 		}
 	}
 
-	ds_destroy(ds_data);
-	ds_data = NULL;
+	ds_destroy(ds_tmp);
+	ds_tmp = NULL;
 
 	/* copy redo logs */
 
@@ -1918,7 +1916,7 @@ copy_back()
 	/* --backup generates a single ib_logfile0, which we must copy
 	if it exists. */
 
-	ds_data = ds_create(dst_dir, DS_TYPE_LOCAL);
+	ds_tmp = ds_create(dst_dir, DS_TYPE_LOCAL);
 	MY_STAT stat_arg;
 	if (!my_stat("ib_logfile0", &stat_arg, MYF(0)) || !stat_arg.st_size) {
 		/* After completed --prepare, redo log files are redundant.
@@ -1932,17 +1930,17 @@ copy_back()
 				 dst_dir, i);
 			unlink(filename);
 		}
-	} else if (!(ret = copy_or_move_file("ib_logfile0", "ib_logfile0",
+	} else if (!(ret = copy_or_move_file(ds_tmp, "ib_logfile0", "ib_logfile0",
 					     dst_dir, 1))) {
 		goto cleanup;
 	}
-	ds_destroy(ds_data);
+	ds_destroy(ds_tmp);
 
 	/* copy innodb system tablespace(s) */
 
 	dst_dir = dst_dir_buf.make(innobase_data_home_dir);
 
-	ds_data = ds_create(dst_dir, DS_TYPE_LOCAL);
+	ds_tmp = ds_create(dst_dir, DS_TYPE_LOCAL);
 
 	for (Tablespace::const_iterator iter(srv_sys_space.begin()),
 	     end(srv_sys_space.end());
@@ -1950,16 +1948,16 @@ copy_back()
 	     ++iter) {
 		const char *filename = base_name(iter->name());
 
-		if (!(ret = copy_or_move_file(filename, iter->name(),
+		if (!(ret = copy_or_move_file(ds_tmp, filename, iter->name(),
 					      dst_dir, 1))) {
 			goto cleanup;
 		}
 	}
 
-	ds_destroy(ds_data);
+	ds_destroy(ds_tmp);
 
 	/* copy the rest of tablespaces */
-	ds_data = ds_create(mysql_data_home, DS_TYPE_LOCAL);
+	ds_tmp = ds_create(mysql_data_home, DS_TYPE_LOCAL);
 
 	it = datadir_iter_new(".", false);
 
@@ -2046,7 +2044,7 @@ copy_back()
 			continue;
 		}
 
-		if (!(ret = copy_or_move_file(node.filepath, node.filepath_rel,
+		if (!(ret = copy_or_move_file(ds_tmp, node.filepath, node.filepath_rel,
 					      mysql_data_home, 1))) {
 			goto cleanup;
 		}
@@ -2056,12 +2054,12 @@ copy_back()
 
         if (file_exists(default_buffer_pool_file) &&
             innobase_buffer_pool_filename) {
-		copy_or_move_file(default_buffer_pool_file,
+		copy_or_move_file(ds_tmp, default_buffer_pool_file,
 				  innobase_buffer_pool_filename,
 				  mysql_data_home, 0);
 	}
 
-	rocksdb_copy_back();
+	rocksdb_copy_back(ds_tmp);
 
 cleanup:
 	if (it != NULL) {
@@ -2070,11 +2068,11 @@ cleanup:
 
 	datadir_node_free(&node);
 
-	if (ds_data != NULL) {
-		ds_destroy(ds_data);
+	if (ds_tmp != NULL) {
+		ds_destroy(ds_tmp);
 	}
 
-	ds_data = NULL;
+	ds_tmp = NULL;
 
 	sync_check_close();
 	return(ret);
@@ -2182,7 +2180,7 @@ decrypt_decompress()
 	}
 
 	/* copy the rest of tablespaces */
-	ds_data = ds_create(".", DS_TYPE_LOCAL);
+	ds_ctxt *ds_tmp = ds_create(".", DS_TYPE_LOCAL);
 
 	it = datadir_iter_new(".", false);
 
@@ -2195,11 +2193,11 @@ decrypt_decompress()
 		datadir_iter_free(it);
 	}
 
-	if (ds_data != NULL) {
-		ds_destroy(ds_data);
+	if (ds_tmp != NULL) {
+		ds_destroy(ds_tmp);
 	}
 
-	ds_data = NULL;
+	ds_tmp = NULL;
 
 	sync_check_close();
 
@@ -2211,7 +2209,7 @@ decrypt_decompress()
   Do not copy the Innodb files (ibdata1, redo log files),
   as this is done in a separate step.
 */
-static bool backup_files_from_datadir(const char *dir_path)
+static bool backup_files_from_datadir(ds_ctxt *ds_data, const char *dir_path)
 {
 	os_file_dir_t dir = os_file_opendir(dir_path);
 	if (dir == IF_WIN(INVALID_HANDLE_VALUE, nullptr)) return false;
@@ -2241,7 +2239,7 @@ static bool backup_files_from_datadir(const char *dir_path)
 
 		std::string full_path(dir_path);
 		full_path.append(1, OS_PATH_SEPARATOR).append(info.name);
-		if (!(ret = copy_file(ds_data, full_path.c_str() , info.name, 1)))
+		if (!(ret = ds_data->copy_file(full_path.c_str() , info.name, 1)))
 			break;
 	}
 	os_file_closedir(dir);
@@ -2291,13 +2289,14 @@ static char *trim_trailing_dir_sep(char *path)
 Create a file hardlink.
 @return true on success, false on error.
 */
-static bool make_hardlink(const char *from_path, const char *to_path)
+bool
+ds_ctxt_t::make_hardlink(const char *from_path, const char *to_path)
 {
 	DBUG_EXECUTE_IF("no_hardlinks", return false;);
 	char to_path_full[FN_REFLEN];
 	if (!is_abs_path(to_path))
 	{
-		fn_format(to_path_full, to_path, ds_data->root, "", MYF(MY_RELATIVE_PATH));
+		fn_format(to_path_full, to_path, root, "", MYF(MY_RELATIVE_PATH));
 	}
 	else
 	{
@@ -2318,7 +2317,9 @@ static bool make_hardlink(const char *from_path, const char *to_path)
  Has optimization that allows to use hardlinks when possible
  (source and destination are directories on the same device)
 */
-static void copy_or_move_dir(const char *from, const char *to, bool do_copy, bool allow_hardlinks)
+void
+ds_ctxt_t::copy_or_move_dir(const char *from, const char *to,
+                            bool do_copy, bool allow_hardlinks)
 {
 	datadir_node_t node;
 	datadir_node_init(&node);
@@ -2346,8 +2347,8 @@ static void copy_or_move_dir(const char *from, const char *to, bool do_copy, boo
 		if (!rc) 
 		{
 			rc = (do_copy ?
-				copy_file(ds_data, from_path, to_path, 1) :
-				move_file(ds_data, from_path, node.filepath_rel,
+				copy_file(from_path, to_path, 1) :
+				move_file(from_path, node.filepath_rel,
 					to, 1));
 		}
 		if (!rc)
@@ -2444,7 +2445,7 @@ static void rocksdb_create_checkpoint()
   remove temp.checkpoint directory (in server's datadir)
   and release user level lock acquired inside rocksdb_create_checkpoint().
 */
-static void rocksdb_backup_checkpoint()
+static void rocksdb_backup_checkpoint(ds_ctxt *ds_data)
 {
 	msg("Backing up rocksdb files.");
 	char rocksdb_backup_dir[FN_REFLEN];
@@ -2456,7 +2457,7 @@ static void rocksdb_backup_checkpoint()
 			die("Can't create rocksdb backup directory %s", rocksdb_backup_dir);
 		}
 	}
-	copy_or_move_dir(rocksdb_checkpoint_dir, ROCKSDB_BACKUP_DIR, true, backup_to_directory);
+	ds_data->copy_or_move_dir(rocksdb_checkpoint_dir, ROCKSDB_BACKUP_DIR, true, backup_to_directory);
 	rocksdb_remove_checkpoint_directory();
 	rocksdb_unlock_checkpoint();
 }
@@ -2464,7 +2465,7 @@ static void rocksdb_backup_checkpoint()
 /*
   Copies #rocksdb directory to the $rockdb_data_dir, on copy-back
 */
-static void rocksdb_copy_back() {
+static void rocksdb_copy_back(ds_ctxt *ds_data) {
 	if (access(ROCKSDB_BACKUP_DIR, 0))
 		return;
 	char rocksdb_home_dir[FN_REFLEN];
@@ -2476,5 +2477,5 @@ static void rocksdb_copy_back() {
 		xb_rocksdb_datadir?trim_dotslash(xb_rocksdb_datadir): ROCKSDB_BACKUP_DIR);
 	}
 	mkdirp(rocksdb_home_dir, 0777, MYF(0));
-	copy_or_move_dir(ROCKSDB_BACKUP_DIR, rocksdb_home_dir, xtrabackup_copy_back, xtrabackup_copy_back);
+	ds_data->copy_or_move_dir(ROCKSDB_BACKUP_DIR, rocksdb_home_dir, xtrabackup_copy_back, xtrabackup_copy_back);
 }
diff --git a/extra/mariabackup/backup_copy.h b/extra/mariabackup/backup_copy.h
index 62b2b1bc232..b4a323f2e89 100644
--- a/extra/mariabackup/backup_copy.h
+++ b/extra/mariabackup/backup_copy.h
@@ -14,30 +14,18 @@
 
 extern bool binlog_locked;
 
-bool
-backup_file_printf(const char *filename, const char *fmt, ...)
-		ATTRIBUTE_FORMAT(printf, 2, 0);
-
 /************************************************************************
 Return true if first and second arguments are the same path. */
 bool
 equal_paths(const char *first, const char *second);
 
-/************************************************************************
-Copy file for backup/restore.
-@return true in case of success. */
-bool
-copy_file(ds_ctxt_t *datasink,
-	  const char *src_file_path,
-	  const char *dst_file_path,
-	  uint thread_n);
-
 /** Start --backup */
-bool backup_start(CorruptedPages &corrupted_pages);
+bool backup_start(ds_ctxt *ds_data, ds_ctxt *ds_meta,
+                  CorruptedPages &corrupted_pages);
 /** Release resources after backup_start() */
 void backup_release();
 /** Finish after backup_start() and backup_release() */
-bool backup_finish();
+bool backup_finish(ds_ctxt *ds_data);
 bool
 apply_log_finish();
 bool
@@ -51,6 +39,5 @@ directory_exists(const char *dir, bool create);
 
 lsn_t
 get_current_lsn(MYSQL *connection);
-bool backup_file_print_buf(const char *filename, const char *buf, int buf_len);
 
 #endif
diff --git a/extra/mariabackup/backup_mysql.cc b/extra/mariabackup/backup_mysql.cc
index b895f8c4561..6003bfb36c4 100644
--- a/extra/mariabackup/backup_mysql.cc
+++ b/extra/mariabackup/backup_mysql.cc
@@ -1383,7 +1383,7 @@ variable.
 @returns true on success
 */
 bool
-write_slave_info(MYSQL *connection)
+write_slave_info(ds_ctxt *datasink, MYSQL *connection)
 {
   String sql, comment;
   bool show_all_slaves_status= false;
@@ -1413,7 +1413,8 @@ write_slave_info(MYSQL *connection)
   }
 
   mysql_slave_position= strdup(comment.c_ptr());
-  return backup_file_print_buf(XTRABACKUP_SLAVE_INFO, sql.ptr(), sql.length());
+  return datasink->backup_file_print_buf(XTRABACKUP_SLAVE_INFO,
+                               sql.ptr(), sql.length());
 }
 
 
@@ -1421,7 +1422,7 @@ write_slave_info(MYSQL *connection)
 Retrieves MySQL Galera and
 saves it in a file. It also prints it to stdout. */
 bool
-write_galera_info(MYSQL *connection)
+write_galera_info(ds_ctxt *datasink, MYSQL *connection)
 {
 	char *state_uuid = NULL, *state_uuid55 = NULL;
 	char *last_committed = NULL, *last_committed55 = NULL;
@@ -1453,12 +1454,12 @@ write_galera_info(MYSQL *connection)
 		goto cleanup;
 	}
 
-	result = backup_file_printf(XTRABACKUP_GALERA_INFO,
+	result = datasink->backup_file_printf(XTRABACKUP_GALERA_INFO,
 		"%s:%s\n", state_uuid ? state_uuid : state_uuid55,
 			last_committed ? last_committed : last_committed55);
 	if (result)
 	{
-		write_current_binlog_file(connection);
+		write_current_binlog_file(datasink, connection);
 	}
 
 cleanup:
@@ -1472,7 +1473,7 @@ cleanup:
 Flush and copy the current binary log file into the backup,
 if GTID is enabled */
 bool
-write_current_binlog_file(MYSQL *connection)
+write_current_binlog_file(ds_ctxt *datasink, MYSQL *connection)
 {
 	char *executed_gtid_set = NULL;
 	char *gtid_binlog_state = NULL;
@@ -1542,7 +1543,7 @@ write_current_binlog_file(MYSQL *connection)
 
 		snprintf(filepath, sizeof(filepath), "%s%c%s",
 			 log_bin_dir, FN_LIBCHAR, log_bin_file);
-		result = copy_file(ds_data, filepath, log_bin_file, 0);
+		result = datasink->copy_file(filepath, log_bin_file, 0);
 	}
 
 cleanup:
@@ -1558,7 +1559,7 @@ cleanup:
 Retrieves MySQL binlog position and
 saves it in a file. It also prints it to stdout. */
 bool
-write_binlog_info(MYSQL *connection)
+write_binlog_info(ds_ctxt *datasink, MYSQL *connection)
 {
 	char *filename = NULL;
 	char *position = NULL;
@@ -1603,14 +1604,14 @@ write_binlog_info(MYSQL *connection)
 			"filename '%s', position '%s', "
 			"GTID of the last change '%s'",
 			filename, position, gtid) != -1);
-		result = backup_file_printf(XTRABACKUP_BINLOG_INFO,
+		result = datasink->backup_file_printf(XTRABACKUP_BINLOG_INFO,
 					    "%s\t%s\t%s\n", filename, position,
 					    gtid);
 	} else {
 		ut_a(asprintf(&mysql_binlog_position,
 			"filename '%s', position '%s'",
 			filename, position) != -1);
-		result = backup_file_printf(XTRABACKUP_BINLOG_INFO,
+		result = datasink->backup_file_printf(XTRABACKUP_BINLOG_INFO,
 					    "%s\t%s\n", filename, position);
 	}
 
@@ -1650,8 +1651,9 @@ PERCONA_SCHEMA.xtrabackup_history and writes a new history record to the
 table containing all the history info particular to the just completed
 backup. */
 bool
-write_xtrabackup_info(MYSQL *connection, const char * filename, bool history,
-                       bool stream)
+write_xtrabackup_info(ds_ctxt *datasink,
+                      MYSQL *connection, const char * filename, bool history,
+                      bool stream)
 {
 
 	bool result = true;
@@ -1727,7 +1729,7 @@ write_xtrabackup_info(MYSQL *connection, const char * filename, bool history,
 	}
 
 	if (stream) {
-		backup_file_printf(filename, "%s", buf);
+		datasink->backup_file_printf(filename, "%s", buf);
 	} else {
 		fp = fopen(filename, "w");
 		if (!fp) {
@@ -1848,9 +1850,9 @@ static std::string make_local_paths(const char *data_file_path)
 	return buf.str();
 }
 
-bool write_backup_config_file()
+bool write_backup_config_file(ds_ctxt *datasink)
 {
-	int rc= backup_file_printf("backup-my.cnf",
+	int rc= datasink->backup_file_printf("backup-my.cnf",
 		"# This MySQL options file was generated by innobackupex.\n\n"
 		"# The MySQL server\n"
 		"[mysqld]\n"
diff --git a/extra/mariabackup/backup_mysql.h b/extra/mariabackup/backup_mysql.h
index b61fa2362c6..d80f3bb7bc1 100644
--- a/extra/mariabackup/backup_mysql.h
+++ b/extra/mariabackup/backup_mysql.h
@@ -62,17 +62,18 @@ void
 unlock_all(MYSQL *connection);
 
 bool
-write_current_binlog_file(MYSQL *connection);
+write_current_binlog_file(ds_ctxt *datasink, MYSQL *connection);
 
 bool
-write_binlog_info(MYSQL *connection);
+write_binlog_info(ds_ctxt *datasink, MYSQL *connection);
 
 bool
-write_xtrabackup_info(MYSQL *connection, const char * filename, bool history,
-                       bool stream);
+write_xtrabackup_info(ds_ctxt *datasink,
+                      MYSQL *connection, const char * filename, bool history,
+                      bool stream);
 
 bool
-write_backup_config_file();
+write_backup_config_file(ds_ctxt *datasink);
 
 bool
 lock_binlog_maybe(MYSQL *connection);
@@ -84,10 +85,10 @@ bool
 wait_for_safe_slave(MYSQL *connection);
 
 bool
-write_galera_info(MYSQL *connection);
+write_galera_info(ds_ctxt *datasink, MYSQL *connection);
 
 bool
-write_slave_info(MYSQL *connection);
+write_slave_info(ds_ctxt *datasink, MYSQL *connection);
 
 
 #endif
diff --git a/extra/mariabackup/datasink.h b/extra/mariabackup/datasink.h
index 4bede4ec9e7..57468e0c9c7 100644
--- a/extra/mariabackup/datasink.h
+++ b/extra/mariabackup/datasink.h
@@ -37,6 +37,35 @@ typedef struct ds_ctxt {
 	char 		*root;
 	void		*ptr;
 	struct ds_ctxt	*pipe_ctxt;
+	/*
+          Copy file for backup/restore.
+          @return true in case of success.
+        */
+        bool copy_file(const char *src_file_path,
+                       const char *dst_file_path,
+                       uint thread_n);
+
+        bool move_file(const char *src_file_path,
+                       const char *dst_file_path,
+                       const char *dst_dir,
+                       uint thread_n);
+
+        bool make_hardlink(const char *from_path, const char *to_path);
+
+        void copy_or_move_dir(const char *from, const char *to,
+                              bool do_copy, bool allow_hardlinks);
+
+        bool backup_file_vprintf(const char *filename,
+                                 const char *fmt, va_list ap);
+
+        bool backup_file_print_buf(const char *filename,
+                                   const char *buf,
+                                   int buf_len);
+
+        bool backup_file_printf(const char *filename,
+                                const char *fmt, ...)
+                                ATTRIBUTE_FORMAT(printf, 2, 0);
+
 } ds_ctxt_t;
 
 typedef struct {
diff --git a/extra/mariabackup/write_filt.cc b/extra/mariabackup/write_filt.cc
index c144a60e378..dd77a9bb953 100644
--- a/extra/mariabackup/write_filt.cc
+++ b/extra/mariabackup/write_filt.cc
@@ -31,7 +31,8 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 
 /************************************************************************
 Write-through page write filter. */
-static my_bool wf_wt_init(xb_write_filt_ctxt_t *ctxt, char *dst_name,
+static my_bool wf_wt_init(ds_ctxt *ds_meta,
+                          xb_write_filt_ctxt_t *ctxt, char *dst_name,
 			  xb_fil_cur_t *cursor, CorruptedPages *corrupted_pages);
 static my_bool wf_wt_process(xb_write_filt_ctxt_t *ctxt, ds_file_t *dstfile);
 
@@ -44,7 +45,8 @@ xb_write_filt_t wf_write_through = {
 
 /************************************************************************
 Incremental page write filter. */
-static my_bool wf_incremental_init(xb_write_filt_ctxt_t *ctxt, char *dst_name,
+static my_bool wf_incremental_init(ds_ctxt *ds_meta,
+                                   xb_write_filt_ctxt_t *ctxt, char *dst_name,
 				   xb_fil_cur_t *cursor, CorruptedPages *corrupted_pages);
 static my_bool wf_incremental_process(xb_write_filt_ctxt_t *ctxt,
 				      ds_file_t *dstfile);
@@ -64,7 +66,8 @@ Initialize incremental page write filter.
 
 @return TRUE on success, FALSE on error. */
 static my_bool
-wf_incremental_init(xb_write_filt_ctxt_t *ctxt, char *dst_name,
+wf_incremental_init(ds_ctxt *ds_meta,
+                    xb_write_filt_ctxt_t *ctxt, char *dst_name,
 		    xb_fil_cur_t *cursor, CorruptedPages *corrupted_pages)
 {
 	char				meta_name[FN_REFLEN];
@@ -88,7 +91,7 @@ wf_incremental_init(xb_write_filt_ctxt_t *ctxt, char *dst_name,
 		 XB_DELTA_INFO_SUFFIX);
 	const xb_delta_info_t	info(cursor->page_size, cursor->zip_size,
 				     cursor->space_id);
-	if (!xb_write_delta_metadata(meta_name, &info)) {
+	if (!xb_write_delta_metadata(ds_meta, meta_name, &info)) {
 		msg(cursor->thread_n,"Error: "
 		    "failed to write meta info for %s",
 		    cursor->rel_path);
@@ -195,7 +198,8 @@ Initialize the write-through page write filter.
 
 @return TRUE on success, FALSE on error. */
 static my_bool
-wf_wt_init(xb_write_filt_ctxt_t *ctxt, char *dst_name __attribute__((unused)),
+wf_wt_init(ds_ctxt *ds_meta __attribute__((unused)),
+           xb_write_filt_ctxt_t *ctxt, char *dst_name __attribute__((unused)),
 	   xb_fil_cur_t *cursor, CorruptedPages *)
 {
 	ctxt->cursor = cursor;
diff --git a/extra/mariabackup/write_filt.h b/extra/mariabackup/write_filt.h
index 6c3ef24291f..a0ce0778a7f 100644
--- a/extra/mariabackup/write_filt.h
+++ b/extra/mariabackup/write_filt.h
@@ -45,7 +45,8 @@ typedef struct {
 
 
 typedef struct {
-	my_bool	(*init)(xb_write_filt_ctxt_t *ctxt, char *dst_name,
+	my_bool	(*init)(ds_ctxt *ds_meta,
+	                xb_write_filt_ctxt_t *ctxt, char *dst_name,
 			xb_fil_cur_t *cursor, CorruptedPages *corrupted_pages);
 	my_bool	(*process)(xb_write_filt_ctxt_t *ctxt, ds_file_t *dstfile);
 	my_bool	(*finalize)(xb_write_filt_ctxt_t *, ds_file_t *dstfile);
diff --git a/extra/mariabackup/xtrabackup.cc b/extra/mariabackup/xtrabackup.cc
index ad35749d0ff..ee12034c910 100644
--- a/extra/mariabackup/xtrabackup.cc
+++ b/extra/mariabackup/xtrabackup.cc
@@ -291,10 +291,66 @@ char *xb_plugin_dir;
 char *xb_plugin_load;
 my_bool xb_close_files;
 
-/* Datasinks */
-ds_ctxt_t       *ds_data     = NULL;
-ds_ctxt_t       *ds_meta     = NULL;
-ds_ctxt_t       *ds_redo     = NULL;
+
+class Datasink_free_list
+{
+protected:
+  /*
+    Simple datasink creation tracking...
+    add datasinks in the reverse order you want them destroyed.
+  */
+#define XTRABACKUP_MAX_DATASINKS 10
+  ds_ctxt_t  *m_datasinks_to_destroy[XTRABACKUP_MAX_DATASINKS];
+  uint        m_actual_datasinks_to_destroy;
+public:
+  Datasink_free_list()
+   :m_actual_datasinks_to_destroy(0)
+  { }
+
+  void add_datasink_to_destroy(ds_ctxt_t *ds)
+  {
+    xb_ad(m_actual_datasinks_to_destroy < XTRABACKUP_MAX_DATASINKS);
+    m_datasinks_to_destroy[m_actual_datasinks_to_destroy] = ds;
+    m_actual_datasinks_to_destroy++;
+  }
+
+  /*
+    Destroy datasinks.
+    Destruction is done in the specific order to not violate their order in the
+    pipeline so that each datasink is able to flush data down the pipeline.
+  */
+  void destroy()
+  {
+    for (uint i= m_actual_datasinks_to_destroy; i > 0; i--)
+    {
+      ds_destroy(m_datasinks_to_destroy[i - 1]);
+      m_datasinks_to_destroy[i - 1] = NULL;
+    }
+  }
+};
+
+
+class Backup_datasinks: public Datasink_free_list
+{
+public:
+  ds_ctxt_t       *m_data;
+  ds_ctxt_t       *m_meta;
+  ds_ctxt_t       *m_redo;
+
+  Backup_datasinks()
+   :m_data(NULL),
+    m_meta(NULL),
+    m_redo(NULL)
+  { }
+  void init();
+  void destroy()
+  {
+    Datasink_free_list::destroy();
+    *this= Backup_datasinks();
+  }
+  bool backup_low();
+};
+
 
 static bool	innobackupex_mode = false;
 
@@ -440,7 +496,8 @@ void CorruptedPages::rename_space(ulint space_id, const std::string &new_name)
   ut_a(!pthread_mutex_unlock(&m_mutex));
 }
 
-bool CorruptedPages::print_to_file(const char *filename) const
+bool CorruptedPages::print_to_file(ds_ctxt *ds_data,
+                                   const char *filename) const
 {
   std::ostringstream out;
   ut_a(!pthread_mutex_lock(&m_mutex));
@@ -468,8 +525,8 @@ bool CorruptedPages::print_to_file(const char *filename) const
     out << "\n";
   }
   ut_a(!pthread_mutex_unlock(&m_mutex));
-  if (xtrabackup_backup)
-    return backup_file_print_buf(filename, out.str().c_str(),
+  if (ds_data)
+    return ds_data->backup_file_print_buf(filename, out.str().c_str(),
                                  static_cast<int>(out.str().size()));
   std::ofstream outfile;
   outfile.open(filename);
@@ -587,19 +644,6 @@ void CorruptedPages::zero_out_free_pages()
   ut_free(buf);
 }
 
-/* Simple datasink creation tracking...add datasinks in the reverse order you
-want them destroyed. */
-#define XTRABACKUP_MAX_DATASINKS	10
-static	ds_ctxt_t	*datasinks[XTRABACKUP_MAX_DATASINKS];
-static	uint		actual_datasinks = 0;
-static inline
-void
-xtrabackup_add_datasink(ds_ctxt_t *ds)
-{
-	xb_ad(actual_datasinks < XTRABACKUP_MAX_DATASINKS);
-	datasinks[actual_datasinks] = ds; actual_datasinks++;
-}
-
 typedef void (*process_single_tablespace_func_t)(const char *dirname,
                                                  const char *filname,
                                                  bool is_remote,
@@ -955,6 +999,7 @@ typedef struct {
 	pthread_mutex_t*	count_mutex;
 	os_thread_id_t		id;
 	CorruptedPages *corrupted_pages;
+	Backup_datasinks *datasinks;
 } data_thread_ctxt_t;
 
 /* ======== for option and variables ======== */
@@ -2456,7 +2501,8 @@ xb_read_delta_metadata(const char *filepath, xb_delta_info_t *info)
 Write meta info for an incremental delta.
 @return TRUE on success, FALSE on failure. */
 my_bool
-xb_write_delta_metadata(const char *filename, const xb_delta_info_t *info)
+xb_write_delta_metadata(ds_ctxt *ds_meta,
+                        const char *filename, const xb_delta_info_t *info)
 {
 	ds_file_t	*f;
 	char		buf[64];
@@ -2771,7 +2817,9 @@ xb_get_copy_action(const char *dflt)
 for full backup, pages filter for incremental backup, etc.
 
 @return FALSE on success and TRUE on error */
-static my_bool xtrabackup_copy_datafile(fil_node_t *node, uint thread_n,
+static my_bool xtrabackup_copy_datafile(ds_ctxt *ds_data,
+                                        ds_ctxt *ds_meta,
+                                        fil_node_t *node, uint thread_n,
                                         const char *dest_name,
                                         const xb_write_filt_t &write_filter,
                                         CorruptedPages &corrupted_pages)
@@ -2839,7 +2887,7 @@ static my_bool xtrabackup_copy_datafile(fil_node_t *node, uint thread_n,
 	ut_a(write_filter.process != NULL);
 
 	if (write_filter.init != NULL &&
-		!write_filter.init(&write_filt_ctxt, dst_name, &cursor,
+		!write_filter.init(ds_meta, &write_filt_ctxt, dst_name, &cursor,
 			opt_log_innodb_page_corruption ? &corrupted_pages : NULL)) {
 		msg (thread_n, "mariabackup: error: failed to initialize page write filter.");
 		goto error;
@@ -3204,7 +3252,8 @@ DECLARE_THREAD(data_copy_thread_func)(
 		DBUG_EXECUTE_FOR_KEY("wait_innodb_redo_before_copy", node->space->name,
 			backup_wait_for_lsn(get_current_lsn(mysql_connection)););
 		/* copy the datafile */
-		if (xtrabackup_copy_datafile(node, num, NULL,
+		if (xtrabackup_copy_datafile(ctxt->datasinks->m_data,
+					     ctxt->datasinks->m_meta, node, num, NULL,
 			xtrabackup_incremental ? wf_incremental : wf_write_through,
 			*ctxt->corrupted_pages))
 			die("failed to copy datafile.");
@@ -3230,22 +3279,21 @@ Otherwise (i.e. when streaming in the 'tar' format) we need 2 separate datasinks
 for the data stream (and don't allow parallel data copying) and for metainfo
 files (including ib_logfile0). The second datasink writes to temporary
 files first, and then streams them in a serialized way when closed. */
-static void
-xtrabackup_init_datasinks(void)
+void Backup_datasinks::init()
 {
 	/* Start building out the pipelines from the terminus back */
 	if (xtrabackup_stream) {
 		/* All streaming goes to stdout */
-		ds_data = ds_meta = ds_redo = ds_create(xtrabackup_target_dir,
-						        DS_TYPE_STDOUT);
+		m_data = m_meta = m_redo = ds_create(xtrabackup_target_dir,
+						     DS_TYPE_STDOUT);
 	} else {
 		/* Local filesystem */
-		ds_data = ds_meta = ds_redo = ds_create(xtrabackup_target_dir,
-						        DS_TYPE_LOCAL);
+		m_data = m_meta = m_redo = ds_create(xtrabackup_target_dir,
+						     DS_TYPE_LOCAL);
 	}
 
 	/* Track it for destruction */
-	xtrabackup_add_datasink(ds_data);
+	add_datasink_to_destroy(m_data);
 
 	/* Stream formatting */
 	if (xtrabackup_stream) {
@@ -3254,66 +3302,50 @@ xtrabackup_init_datasinks(void)
 	 ut_a(xtrabackup_stream_fmt == XB_STREAM_FMT_XBSTREAM);
 	 ds = ds_create(xtrabackup_target_dir, DS_TYPE_XBSTREAM);
 
-		xtrabackup_add_datasink(ds);
+		add_datasink_to_destroy(ds);
 
-		ds_set_pipe(ds, ds_data);
-		ds_data = ds;
+		ds_set_pipe(ds, m_data);
+		m_data = ds;
 
 
-		ds_redo = ds_meta = ds_data;
+		m_redo = m_meta = m_data;
 	}
 
-	/* Compression for ds_data and ds_redo */
+	/* Compression for m_data and m_redo */
 	if (xtrabackup_compress) {
 		ds_ctxt_t	*ds;
 
 		/* Use a 1 MB buffer for compressed output stream */
 		ds = ds_create(xtrabackup_target_dir, DS_TYPE_BUFFER);
 		ds_buffer_set_size(ds, 1024 * 1024);
-		xtrabackup_add_datasink(ds);
-		ds_set_pipe(ds, ds_data);
-		if (ds_data != ds_redo) {
-			ds_data = ds;
+		add_datasink_to_destroy(ds);
+		ds_set_pipe(ds, m_data);
+		if (m_data != m_redo) {
+			m_data = ds;
 			ds = ds_create(xtrabackup_target_dir, DS_TYPE_BUFFER);
 			ds_buffer_set_size(ds, 1024 * 1024);
-			xtrabackup_add_datasink(ds);
-			ds_set_pipe(ds, ds_redo);
-			ds_redo = ds;
+			add_datasink_to_destroy(ds);
+			ds_set_pipe(ds, m_redo);
+			m_redo = ds;
 		} else {
-			ds_redo = ds_data = ds;
+			m_redo = m_data = ds;
 		}
 
 		ds = ds_create(xtrabackup_target_dir, DS_TYPE_COMPRESS);
-		xtrabackup_add_datasink(ds);
-		ds_set_pipe(ds, ds_data);
-		if (ds_data != ds_redo) {
-			ds_data = ds;
+		add_datasink_to_destroy(ds);
+		ds_set_pipe(ds, m_data);
+		if (m_data != m_redo) {
+			m_data = ds;
 			ds = ds_create(xtrabackup_target_dir, DS_TYPE_COMPRESS);
-			xtrabackup_add_datasink(ds);
-			ds_set_pipe(ds, ds_redo);
-			ds_redo = ds;
+			add_datasink_to_destroy(ds);
+			ds_set_pipe(ds, m_redo);
+			m_redo = ds;
 		} else {
-			ds_redo = ds_data = ds;
+			m_redo = m_data = ds;
 		}
 	}
 }
 
-/************************************************************************
-Destroy datasinks.
-
-Destruction is done in the specific order to not violate their order in the
-pipeline so that each datasink is able to flush data down the pipeline. */
-static void xtrabackup_destroy_datasinks(void)
-{
-	for (uint i = actual_datasinks; i > 0; i--) {
-		ds_destroy(datasinks[i-1]);
-		datasinks[i-1] = NULL;
-	}
-	ds_data = NULL;
-	ds_meta = NULL;
-	ds_redo = NULL;
-}
-
 #define SRV_MAX_N_PENDING_SYNC_IOS	100
 
 /** Initialize the tablespace cache subsystem. */
@@ -4447,7 +4479,7 @@ static void stop_backup_threads()
 
 /** Implement the core of --backup
 @return	whether the operation succeeded */
-static bool xtrabackup_backup_low()
+bool Backup_datasinks::backup_low()
 {
 	ut_ad(!metadata_to_lsn);
 
@@ -4499,7 +4531,7 @@ static bool xtrabackup_backup_low()
 	}
 	metadata_last_lsn = log_copy_scanned_lsn;
 
-	if (!xtrabackup_stream_metadata(ds_meta)) {
+	if (!xtrabackup_stream_metadata(m_meta)) {
 		msg("Error: failed to stream metadata.");
 		return false;
 	}
@@ -4515,7 +4547,8 @@ static bool xtrabackup_backup_low()
 		}
 		sprintf(filename, "%s/%s", xtrabackup_extra_lsndir,
 			XTRABACKUP_INFO);
-		if (!write_xtrabackup_info(mysql_connection, filename, false, false)) {
+		if (!write_xtrabackup_info(m_data,
+		                           mysql_connection, filename, false, false)) {
 			msg("Error: failed to write info "
 			 "to '%s'.", filename);
 			return false;
@@ -4535,6 +4568,7 @@ static bool xtrabackup_backup_func()
 	pthread_mutex_t		 count_mutex;
 	CorruptedPages corrupted_pages;
 	data_thread_ctxt_t 	*data_threads;
+	Backup_datasinks backup_datasinks;
 	pthread_mutex_init(&backup_mutex, NULL);
 	pthread_cond_init(&scanned_lsn_cond, NULL);
 
@@ -4686,7 +4720,7 @@ reread_log_header:
 
 	log_mutex_exit();
 
-	xtrabackup_init_datasinks();
+	backup_datasinks.init();
 
 	if (!select_history()) {
 		goto fail;
@@ -4694,7 +4728,7 @@ reread_log_header:
 
 	/* open the log file */
 	memset(&stat_info, 0, sizeof(MY_STAT));
-	dst_log_file = ds_open(ds_redo, "ib_logfile0", &stat_info);
+	dst_log_file = ds_open(backup_datasinks.m_redo, "ib_logfile0", &stat_info);
 	if (dst_log_file == NULL) {
 		msg("Error: failed to open the target stream for "
 		    "'ib_logfile0'.");
@@ -4812,6 +4846,7 @@ fail_before_log_copying_thread_start:
 		data_threads[i].count = &count;
 		data_threads[i].count_mutex = &count_mutex;
 		data_threads[i].corrupted_pages = &corrupted_pages;
+		data_threads[i].datasinks= &backup_datasinks;
 		os_thread_create(data_copy_thread_func, data_threads + i,
 				 &data_threads[i].id);
 	}
@@ -4832,10 +4867,13 @@ fail_before_log_copying_thread_start:
 	datafiles_iter_free(it);
 	}
 
-	bool ok = backup_start(corrupted_pages);
+	DBUG_ASSERT(backup_datasinks.m_data);
+	DBUG_ASSERT(backup_datasinks.m_meta);
+	bool ok = backup_start(backup_datasinks.m_data,
+	                       backup_datasinks.m_meta, corrupted_pages);
 
 	if (ok) {
-		ok = xtrabackup_backup_low();
+		ok = backup_datasinks.backup_low();
 
 		backup_release();
 
@@ -4845,12 +4883,13 @@ fail_before_log_copying_thread_start:
 		);
 
 		if (ok) {
-			backup_finish();
+			backup_finish(backup_datasinks.m_data);
 		}
 	}
 
 	if (opt_log_innodb_page_corruption)
-		ok = corrupted_pages.print_to_file(MB_CORRUPTED_PAGES_FILE);
+		ok = corrupted_pages.print_to_file(backup_datasinks.m_data,
+						   MB_CORRUPTED_PAGES_FILE);
 
 	if (!ok) {
 		goto fail;
@@ -4859,7 +4898,7 @@ fail_before_log_copying_thread_start:
 	if (changed_page_bitmap) {
 		xb_page_bitmap_deinit(changed_page_bitmap);
 	}
-	xtrabackup_destroy_datasinks();
+	backup_datasinks.destroy();
 
 	msg("Redo log (from LSN " LSN_PF " to " LSN_PF
 	    ") was copied.", checkpoint_lsn_start, log_copy_scanned_lsn);
@@ -4906,7 +4945,7 @@ FTWRL.  This ensures consistent backup in presence of DDL.
   It is the responsibility of the prepare phase to deal with .new, .ren, and .del
   files.
 */
-void backup_fix_ddl(CorruptedPages &corrupted_pages)
+void CorruptedPages::backup_fix_ddl(ds_ctxt *ds_data, ds_ctxt *ds_meta)
 {
 	std::set<std::string> new_tables;
 	std::set<std::string> dropped_tables;
@@ -4929,7 +4968,7 @@ void backup_fix_ddl(CorruptedPages &corrupted_pages)
 
 		if (ddl_tracker.drops.find(id) != ddl_tracker.drops.end()) {
 			dropped_tables.insert(name);
-			corrupted_pages.drop_space(id);
+			drop_space(id);
 			continue;
 		}
 
@@ -4951,20 +4990,20 @@ void backup_fix_ddl(CorruptedPages &corrupted_pages)
 				of it because of optimized DDL. We emulate a drop/create.*/
 				dropped_tables.insert(name);
 				if (opt_log_innodb_page_corruption)
-					corrupted_pages.drop_space(id);
+					drop_space(id);
 				new_tables.insert(new_name);
 			} else {
 				/* Renamed, and no optimized DDL*/
 				renamed_tables[name] = new_name;
 				if (opt_log_innodb_page_corruption)
-					corrupted_pages.rename_space(id, new_name);
+					rename_space(id, new_name);
 			}
 		} else if (has_optimized_ddl) {
 			/* Table was recreated, or optimized DDL ran.
 			In both cases we need a full copy in the backup.*/
 			new_tables.insert(name);
 			if (opt_log_innodb_page_corruption)
-				corrupted_pages.drop_space(id);
+				drop_space(id);
 		}
 	}
 
@@ -4985,7 +5024,7 @@ void backup_fix_ddl(CorruptedPages &corrupted_pages)
 			dropped_tables.erase(name);
 			new_tables.insert(name);
 			if (opt_log_innodb_page_corruption)
-				corrupted_pages.drop_space(id);
+				drop_space(id);
 		}
 	}
 
@@ -4994,7 +5033,8 @@ void backup_fix_ddl(CorruptedPages &corrupted_pages)
 		iter != renamed_tables.end(); ++iter) {
 		const std::string old_name = iter->first;
 		std::string new_name = iter->second;
-		backup_file_printf((old_name + ".ren").c_str(), "%s", new_name.c_str());
+		DBUG_ASSERT(ds_data);
+		ds_data->backup_file_printf((old_name + ".ren").c_str(), "%s", new_name.c_str());
 	}
 
 	// Mark tablespaces for drop
@@ -5002,7 +5042,7 @@ void backup_fix_ddl(CorruptedPages &corrupted_pages)
 		iter != dropped_tables.end();
 		iter++) {
 		const std::string name(*iter);
-		backup_file_printf((name + ".del").c_str(), "%s", "");
+		ds_data->backup_file_printf((name + ".del").c_str(), "%s", "");
 	}
 
 	//  Load and copy new tables.
@@ -5046,8 +5086,9 @@ void backup_fix_ddl(CorruptedPages &corrupted_pages)
 			continue;
 		std::string dest_name(node->space->name);
 		dest_name.append(".new");
-		xtrabackup_copy_datafile(node, 0, dest_name.c_str(), wf_write_through,
-			corrupted_pages);
+		xtrabackup_copy_datafile(ds_data, ds_meta,
+		                         node, 0, dest_name.c_str(),
+		                         wf_write_through, *this);
 	}
 
 	datafiles_iter_free(it);
@@ -6163,7 +6204,7 @@ static bool xtrabackup_prepare_func(char** argv)
           }
         }
         else
-          corrupted_pages.print_to_file(MB_CORRUPTED_PAGES_FILE);
+          corrupted_pages.print_to_file(NULL, MB_CORRUPTED_PAGES_FILE);
 
         if (xtrabackup_rollback_xa)
         {
diff --git a/extra/mariabackup/xtrabackup.h b/extra/mariabackup/xtrabackup.h
index 854456c3afd..de3a96443a3 100644
--- a/extra/mariabackup/xtrabackup.h
+++ b/extra/mariabackup/xtrabackup.h
@@ -46,11 +46,13 @@ public:
   bool contains(ulint space_id, ulint page_no) const;
   void drop_space(ulint space_id);
   void rename_space(ulint space_id, const std::string &new_name);
-  bool print_to_file(const char *file_name) const;
+  bool print_to_file(ds_ctxt *ds_data, const char *file_name) const;
   void read_from_file(const char *file_name);
   bool empty() const;
   void zero_out_free_pages();
 
+  void backup_fix_ddl(ds_ctxt *ds_data, ds_ctxt *ds_meta);
+
 private:
   void add_page_no_lock(const char *space_name, ulint space_id, ulint page_no,
                         bool convert_space_name);
@@ -63,6 +65,7 @@ private:
   container_t m_spaces;
 };
 
+
 /* value of the --incremental option */
 extern lsn_t incremental_lsn;
 
@@ -76,8 +79,6 @@ extern char		*xb_rocksdb_datadir;
 extern my_bool	xb_backup_rocksdb;
 
 extern uint		opt_protocol;
-extern ds_ctxt_t	*ds_meta;
-extern ds_ctxt_t	*ds_data;
 
 /* The last checkpoint LSN at the backup startup time */
 extern lsn_t checkpoint_lsn_start;
@@ -177,7 +178,8 @@ extern ulong opt_binlog_info;
 extern ulong xtrabackup_innodb_force_recovery;
 
 void xtrabackup_io_throttling(void);
-my_bool xb_write_delta_metadata(const char *filename,
+my_bool xb_write_delta_metadata(ds_ctxt *ds_meta,
+                                const char *filename,
 				const xb_delta_info_t *info);
 
 /************************************************************************