MDEV-32932 Port backup features from ES

Added support to BACKUP STAGE to maria-backup

This is a port of the code from ES 10.6
See MDEV-5336 for backup stages description.

The following old options are not supported by the new code:
--rsync             ; This is because rsync will not work on tables
                      that are in used.
--no-backup-locks   ; This is disabled as mariadb-backup will always
                      use backup locks for better performance.

extra/mariabackup/CMakeLists.txt

@@ -31,6 +31,7 @@ ENDIF()
 INCLUDE_DIRECTORIES(
   ${CMAKE_SOURCE_DIR}/include
   ${CMAKE_SOURCE_DIR}/sql
+  ${CMAKE_SOURCE_DIR}/storage/maria
   ${CMAKE_CURRENT_SOURCE_DIR}/quicklz
   ${CMAKE_CURRENT_SOURCE_DIR}
   )
@@ -71,8 +72,12 @@ MYSQL_ADD_EXECUTABLE(mariadb-backup
   xbstream_write.cc
   backup_mysql.cc
   backup_copy.cc
-  xb_plugin.cc
+  encryption_plugin.cc
   ${PROJECT_BINARY_DIR}/sql/sql_builtin.cc
+  aria_backup_client.cc
+  thread_pool.cc
+  ddl_log.cc
+  common_engine.cc
   ${PROJECT_SOURCE_DIR}/sql/net_serv.cc
   ${PROJECT_SOURCE_DIR}/libmysqld/libmysql.c
   COMPONENT backup
@@ -81,7 +86,8 @@ MYSQL_ADD_EXECUTABLE(mariadb-backup
 # Export all symbols on Unix, for better crash callstacks
 SET_TARGET_PROPERTIES(mariadb-backup PROPERTIES ENABLE_EXPORTS TRUE)
 
-TARGET_LINK_LIBRARIES(mariadb-backup sql sql_builtins)
+TARGET_LINK_LIBRARIES(mariadb-backup  sql sql_builtins aria)
+
 IF(NOT HAVE_SYSTEM_REGEX)
   TARGET_LINK_LIBRARIES(mariadb-backup pcre2-posix)
 ENDIF()

extra/mariabackup/aria_backup_client.h

+#pragma once
+#include "my_global.h"
+#include "datasink.h"
+#include "backup_mysql.h"
+#include "thread_pool.h"
+#include "xtrabackup.h"
+
+namespace aria {
+
+bool prepare(const char *target_dir);
+
+class BackupImpl;
+
+class Backup {
+	public:
+		Backup(const char *datadir_path,
+		       const char *aria_log_path,
+		       ds_ctxt_t *datasink,
+			std::vector<MYSQL *> &con_pool, ThreadPool &thread_pool);
+		~Backup();
+		Backup (Backup &&other) = delete;
+		Backup & operator= (Backup &&other) = delete;
+		Backup(const Backup &) = delete;
+		Backup & operator= (const Backup &) = delete;
+		bool init();
+		bool start(bool no_lock);
+		bool wait_for_finish();
+		bool copy_offline_tables(
+			const std::unordered_set<table_key_t> *exclude_tables, bool no_lock,
+			bool copy_stats);
+		bool finalize();
+		bool copy_log_tail();
+		void set_post_copy_table_hook(const post_copy_table_hook_t &hook);
+	private:
+		BackupImpl *m_backup_impl;
+};
+
+} // namespace aria

extra/mariabackup/backup_copy.h

@@ -2,6 +2,7 @@
 #ifndef XTRABACKUP_BACKUP_COPY_H
 #define XTRABACKUP_BACKUP_COPY_H
 
+#include <functional>
 #include <my_global.h>
 #include <mysql.h>
 #include "datasink.h"
@@ -21,8 +22,7 @@ bool
 equal_paths(const char *first, const char *second);
 
 /** Start --backup */
-bool backup_start(ds_ctxt *ds_data, ds_ctxt *ds_meta,
-                  CorruptedPages &corrupted_pages);
+bool backup_files(ds_ctxt *ds_data, const char *from);
 /** Release resources after backup_start() */
 void backup_release();
 /** Finish after backup_start() and backup_release() */
@@ -38,7 +38,25 @@ is_path_separator(char);
 bool
 directory_exists(const char *dir, bool create);
 
-lsn_t
-get_current_lsn(MYSQL *connection);
-
+bool has_rocksdb_plugin();
+void rocksdb_create_checkpoint();
+void foreach_file_in_db_dirs(
+	const char *dir_path, std::function<bool(const char *)> func);
+void foreach_file_in_datadir(
+	const char *dir_path, std::function<bool(const char *)> func);
+bool ends_with(const char *str, const char *suffix);
+bool starts_with(const char *str, const char *prefix);
+void parse_db_table_from_file_path(
+	const char *filepath, char *dbname, char *tablename);
+const char *trim_dotslash(const char *path);
+bool backup_files_from_datadir(ds_ctxt_t *ds_data,
+                               const char *dir_path,
+                               const char *prefix);
+
+bool is_system_table(const char *dbname, const char *tablename);
+std::unique_ptr<std::vector<std::string>>
+	find_files(const char *dir_path, const char *prefix, const char *suffix);
+bool file_exists(const char *filename);
+bool
+filename_matches(const char *filename, const char **ext_list);
 #endif

extra/mariabackup/backup_debug.h

 #pragma once
 #include "my_dbug.h"
+
 #ifndef DBUG_OFF
 char *dbug_mariabackup_get_val(const char *event, fil_space_t::name_type key);
 /*
@@ -14,11 +15,21 @@ To use this facility, you need to
    for the variable)
 3. start mariabackup with --dbug=+d,debug_mariabackup_events
 */
-#define DBUG_EXECUTE_FOR_KEY(EVENT, KEY, CODE)			\
-	DBUG_EXECUTE_IF("mariabackup_inject_code",		\
-	{ char *dbug_val= dbug_mariabackup_get_val(EVENT, KEY);	\
-	  if (dbug_val) CODE })
+extern void dbug_mariabackup_event(
+	const char *event, const fil_space_t::name_type key, bool need_lock);
+#define DBUG_MARIABACKUP_EVENT(A, B) \
+	DBUG_EXECUTE_IF("mariabackup_events", \
+		dbug_mariabackup_event(A,B,false););
+#define DBUG_MARIABACKUP_EVENT_LOCK(A, B) \
+	DBUG_EXECUTE_IF("mariabackup_events", \
+		dbug_mariabackup_event(A,B, true););
+#define DBUG_EXECUTE_FOR_KEY(EVENT, KEY, CODE) \
+	DBUG_EXECUTE_IF("mariabackup_inject_code", {\
+		char *dbug_val = dbug_mariabackup_get_val(EVENT, KEY); \
+		if (dbug_val && *dbug_val) CODE \
+	})
 #else
+#define DBUG_MARIABACKUP_EVENT(A,B)
+#define DBUG_MARIABACKUP_EVENT_LOCK(A,B)
 #define DBUG_EXECUTE_FOR_KEY(EVENT, KEY, CODE)
 #endif
-

extra/mariabackup/backup_mysql.cc

@@ -60,10 +60,11 @@ Street, Fifth Floor, Boston, MA 02110-1335 USA
 #include "backup_copy.h"
 #include "backup_mysql.h"
 #include "mysqld.h"
-#include "xb_plugin.h"
+#include "encryption_plugin.h"
 #include <sstream>
 #include <sql_error.h>
 #include "page0zip.h"
+#include "backup_debug.h"
 
 char *tool_name;
 char tool_args[2048];
@@ -71,7 +72,7 @@ char tool_args[2048];
 ulong mysql_server_version;
 
 /* server capabilities */
-bool have_backup_locks = false;
+bool have_changed_page_bitmaps = false;
 bool have_lock_wait_timeout = false;
 bool have_galera_enabled = false;
 bool have_multi_threaded_slave = false;
@@ -251,13 +252,14 @@ struct mysql_variable {
 
 
 static
-void
+uint
 read_mysql_variables(MYSQL *connection, const char *query, mysql_variable *vars,
 	bool vertical_result)
 {
 	MYSQL_RES *mysql_result;
 	MYSQL_ROW row;
 	mysql_variable *var;
+	uint n_values=0;
 
 	mysql_result = xb_mysql_query(connection, query, true);
 
@@ -271,6 +273,7 @@ read_mysql_variables(MYSQL *connection, const char *query, mysql_variable *vars,
 				if (strcmp(var->name, name) == 0
 				    && value != NULL) {
 					*(var->value) = strdup(value);
+					n_values++;
 				}
 			}
 		}
@@ -287,6 +290,7 @@ read_mysql_variables(MYSQL *connection, const char *query, mysql_variable *vars,
 					if (strcmp(var->name, name) == 0
 					    && value != NULL) {
 						*(var->value) = strdup(value);
+						n_values++;
 					}
 				}
 				++i;
@@ -295,6 +299,7 @@ read_mysql_variables(MYSQL *connection, const char *query, mysql_variable *vars,
 	}
 
 	mysql_free_result(mysql_result);
+	return n_values;
 }
 
 
@@ -359,7 +364,6 @@ bool get_mysql_vars(MYSQL *connection)
 {
   char *gtid_mode_var= NULL;
   char *version_var= NULL;
-  char *have_backup_locks_var= NULL;
   char *log_bin_var= NULL;
   char *lock_wait_timeout_var= NULL;
   char *wsrep_on_var= NULL;
@@ -384,7 +388,6 @@ bool get_mysql_vars(MYSQL *connection)
   bool ret= true;
 
   mysql_variable mysql_vars[]= {
-      {"have_backup_locks", &have_backup_locks_var},
       {"log_bin", &log_bin_var},
       {"lock_wait_timeout", &lock_wait_timeout_var},
       {"gtid_mode", &gtid_mode_var},
@@ -409,11 +412,6 @@ bool get_mysql_vars(MYSQL *connection)
 
   read_mysql_variables(connection, "SHOW VARIABLES", mysql_vars, true);
 
-  if (have_backup_locks_var != NULL && !opt_no_backup_locks)
-  {
-    have_backup_locks= true;
-  }
-
   if (opt_binlog_info == BINLOG_INFO_AUTO)
   {
     if (log_bin_var != NULL && !strcmp(log_bin_var, "ON"))
@@ -867,11 +865,11 @@ static void stop_query_killer()
 
 
 /*********************************************************************//**
-Function acquires either a backup tables lock, if supported
-by the server, or a global read lock (FLUSH TABLES WITH READ LOCK)
-otherwise.
+Function acquires backup locks
 @returns true if lock acquired */
-bool lock_tables(MYSQL *connection)
+
+bool
+lock_for_backup_stage_start(MYSQL *connection)
 {
   if (have_lock_wait_timeout || opt_lock_wait_timeout)
   {
@@ -884,12 +882,6 @@ bool lock_tables(MYSQL *connection)
     xb_mysql_query(connection, buf, false);
   }
 
-  if (have_backup_locks)
-  {
-    msg("Executing LOCK TABLES FOR BACKUP...");
-    xb_mysql_query(connection, "LOCK TABLES FOR BACKUP", false);
-    return (true);
-  }
 
   if (opt_lock_wait_timeout)
   {
@@ -914,8 +906,6 @@ bool lock_tables(MYSQL *connection)
 
   xb_mysql_query(connection, "BACKUP STAGE START", true);
   DBUG_MARIABACKUP_EVENT("after_backup_stage_start", {});
-  xb_mysql_query(connection, "BACKUP STAGE BLOCK_COMMIT", true);
-  DBUG_MARIABACKUP_EVENT("after_backup_stage_block_commit", {});
   /* Set the maximum supported session value for
   lock_wait_timeout to prevent unnecessary timeouts when the
   global value is changed from the default */
@@ -931,24 +921,68 @@ bool lock_tables(MYSQL *connection)
   return (true);
 }
 
-/*********************************************************************//**
-If backup locks are used, execute LOCK BINLOG FOR BACKUP provided that we are
-not in the --no-lock mode and the lock has not been acquired already.
-@returns true if lock acquired */
 bool
-lock_binlog_maybe(MYSQL *connection)
-{
-	if (have_backup_locks && !opt_no_lock && !binlog_locked) {
-		msg("Executing LOCK BINLOG FOR BACKUP...");
-		xb_mysql_query(connection, "LOCK BINLOG FOR BACKUP", false);
-		binlog_locked = true;
+lock_for_backup_stage_flush(MYSQL *connection) {
+	if (opt_kill_long_queries_timeout) {
+		start_query_killer();
+	}
+	xb_mysql_query(connection, "BACKUP STAGE FLUSH", true);
+	if (opt_kill_long_queries_timeout) {
+		stop_query_killer();
+	}
+	return true;
+}
+
+bool
+lock_for_backup_stage_block_ddl(MYSQL *connection) {
+	if (opt_kill_long_queries_timeout) {
+		start_query_killer();
+	}
+	xb_mysql_query(connection, "BACKUP STAGE BLOCK_DDL", true);
+	DBUG_MARIABACKUP_EVENT("after_backup_stage_block_ddl", {});
+	if (opt_kill_long_queries_timeout) {
+		stop_query_killer();
+	}
+	return true;
+}
 
-		return(true);
+bool
+lock_for_backup_stage_commit(MYSQL *connection) {
+	if (opt_kill_long_queries_timeout) {
+		start_query_killer();
 	}
+	xb_mysql_query(connection, "BACKUP STAGE BLOCK_COMMIT", true);
+	DBUG_MARIABACKUP_EVENT("after_backup_stage_block_commit", {});
+	if (opt_kill_long_queries_timeout) {
+		stop_query_killer();
+	}
+	return true;
+}
 
-	return(false);
+bool backup_lock(MYSQL *con, const char *table_name) {
+	static const std::string backup_lock_prefix("BACKUP LOCK ");
+	std::string backup_lock_query = backup_lock_prefix + table_name;
+	xb_mysql_query(con, backup_lock_query.c_str(), true);
+	return true;
 }
 
+bool backup_unlock(MYSQL *con) {
+	xb_mysql_query(con, "BACKUP UNLOCK", true);
+	return true;
+}
+
+std::unordered_set<std::string>
+get_tables_in_use(MYSQL *con) {
+	std::unordered_set<std::string> result;
+	MYSQL_RES *q_res =
+		xb_mysql_query(con, "SHOW OPEN TABLES WHERE In_use = 1", true);
+	while (MYSQL_ROW row = mysql_fetch_row(q_res)) {
+		auto tk = table_key(row[0], row[1]);
+		msg("Table %s is in use", tk.c_str());
+		result.insert(std::move(tk));
+	}
+	return result;
+}
 
 /*********************************************************************//**
 Releases either global read lock acquired with FTWRL and the binlog
@@ -1383,77 +1417,12 @@ write_slave_info(ds_ctxt *datasink, MYSQL *connection)
 
 
 /*********************************************************************//**
-Retrieves MySQL Galera and
-saves it in a file. It also prints it to stdout. */
+Old function, not needed anymore with BACKUP LOCKS
+*/
 bool
 write_galera_info(ds_ctxt *datasink, MYSQL *connection)
 {
-	char *state_uuid = NULL, *state_uuid55 = NULL;
-	char *last_committed = NULL, *last_committed55 = NULL;
-	char *domain_id = NULL, *domain_id55 = NULL;
-	bool result;
-
-	mysql_variable status[] = {
-		{"Wsrep_local_state_uuid", &state_uuid},
-		{"wsrep_local_state_uuid", &state_uuid55},
-		{"Wsrep_last_committed", &last_committed},
-		{"wsrep_last_committed", &last_committed55},
-		{NULL, NULL}
-	};
-
-	mysql_variable value[] = {
-		{"Wsrep_gtid_domain_id", &domain_id},
-		{"wsrep_gtid_domain_id", &domain_id55},
-		{NULL, NULL}
-	};
-
-	/* When backup locks are supported by the server, we should skip
-	creating xtrabackup_galera_info file on the backup stage, because
-	wsrep_local_state_uuid and wsrep_last_committed will be inconsistent
-	without blocking commits. The state file will be created on the prepare
-	stage using the WSREP recovery procedure. */
-	if (have_backup_locks) {
-		return(true);
-	}
-
-	read_mysql_variables(connection, "SHOW STATUS", status, true);
-
-	if ((state_uuid == NULL && state_uuid55 == NULL)
-		|| (last_committed == NULL && last_committed55 == NULL)) {
-		msg("Warning: failed to get master wsrep state from SHOW STATUS.");
-		result = true;
-		goto cleanup;
-	}
-
-	read_mysql_variables(connection, "SHOW VARIABLES LIKE 'wsrep%'", value, true);
-
-	if (domain_id == NULL && domain_id55 == NULL) {
-		msg("Warning: failed to get master wsrep state from SHOW VARIABLES.");
-		result = true;
-		goto cleanup;
-	}
-
-	result = datasink->backup_file_printf(XTRABACKUP_GALERA_INFO,
-		"%s:%s %s\n", state_uuid ? state_uuid : state_uuid55,
-			      last_committed ? last_committed : last_committed55,
-			      domain_id ? domain_id : domain_id55);
-
-	if (result)
-	{
-	  result= datasink->backup_file_printf(XTRABACKUP_DONOR_GALERA_INFO,
-		"%s:%s %s\n", state_uuid ? state_uuid : state_uuid55,
-			      last_committed ? last_committed : last_committed55,
-			      domain_id ? domain_id : domain_id55);
-	}
-	if (result)
-	{
-		write_current_binlog_file(datasink, connection);
-	}
-
-cleanup:
-	free_mysql_variables(status);
-
-	return(result);
+  return true;                                  // Success
 }
 
 
@@ -1496,8 +1465,6 @@ write_current_binlog_file(ds_ctxt *datasink, MYSQL *connection)
 	if (gtid_exists) {
 		size_t log_bin_dir_length;
 
-		lock_binlog_maybe(connection);
-
 		xb_mysql_query(connection, "FLUSH BINARY LOGS", false);
 
 		read_mysql_variables(connection, "SHOW MASTER STATUS",
@@ -1856,13 +1823,13 @@ bool write_backup_config_file(ds_ctxt *datasink)
 		srv_log_file_size,
 		srv_page_size,
 		srv_undo_dir,
-		srv_undo_tablespaces,
+                (uint) srv_undo_tablespaces,
 		page_zip_level,
 		innobase_buffer_pool_filename ?
 			"innodb_buffer_pool_filename=" : "",
 		innobase_buffer_pool_filename ?
 			innobase_buffer_pool_filename : "",
-		xb_plugin_get_config());
+		encryption_plugin_get_config());
 		return rc;
 }
 
@@ -1987,3 +1954,23 @@ mdl_unlock_all()
   mysql_close(mdl_con);
   spaceid_to_tablename.clear();
 }
+
+ulonglong get_current_lsn(MYSQL *connection)
+{
+	static const char lsn_prefix[] = "\nLog sequence number ";
+	ulonglong lsn = 0;
+	if (MYSQL_RES *res = xb_mysql_query(connection,
+					    "SHOW ENGINE INNODB STATUS",
+					    true, false)) {
+		if (MYSQL_ROW row = mysql_fetch_row(res)) {
+			const char *p= strstr(row[2], lsn_prefix);
+			DBUG_ASSERT(p);
+			if (p) {
+				p += sizeof lsn_prefix - 1;
+				lsn = lsn_t(strtoll(p, NULL, 10));
+			}
+		}
+		mysql_free_result(res);
+	}
+	return lsn;
+}

extra/mariabackup/backup_mysql.h

@@ -2,12 +2,15 @@
 #define XTRABACKUP_BACKUP_MYSQL_H
 
 #include <mysql.h>
+#include <string>
+#include <unordered_set>
+#include "datasink.h"
 
 /* MariaDB version */
 extern ulong mysql_server_version;
 
 /* server capabilities */
-extern bool have_backup_locks;
+extern bool have_changed_page_bitmaps;
 extern bool have_lock_wait_timeout;
 extern bool have_galera_enabled;
 extern bool have_multi_threaded_slave;
@@ -71,7 +74,21 @@ bool
 lock_binlog_maybe(MYSQL *connection);
 
 bool
-lock_tables(MYSQL *connection);
+lock_for_backup_stage_start(MYSQL *connection);
+
+bool
+lock_for_backup_stage_flush(MYSQL *connection);
+
+bool
+lock_for_backup_stage_block_ddl(MYSQL *connection);
+
+bool
+lock_for_backup_stage_commit(MYSQL *connection);
+
+bool backup_lock(MYSQL *con, const char *table_name);
+bool backup_unlock(MYSQL *con);
+
+std::unordered_set<std::string> get_tables_in_use(MYSQL *con);
 
 bool
 wait_for_safe_slave(MYSQL *connection);
@@ -82,5 +99,6 @@ write_galera_info(ds_ctxt *datasink, MYSQL *connection);
 bool
 write_slave_info(ds_ctxt *datasink, MYSQL *connection);
 
+ulonglong get_current_lsn(MYSQL *connection);
 
 #endif

extra/mariabackup/common_engine.h

+#pragma once
+#include "my_global.h"
+#include "backup_mysql.h"
+#include "datasink.h"
+#include "thread_pool.h"
+#include "xtrabackup.h"
+
+#include <unordered_set>
+#include <string>
+#include <vector>
+
+namespace common_engine {
+
+class BackupImpl;
+
+class Backup {
+	public:
+		Backup(const char *datadir_path, ds_ctxt_t *datasink,
+				std::vector<MYSQL *> &con_pool, ThreadPool &thread_pool);
+		~Backup();
+		Backup (Backup &&other) = delete;
+		Backup & operator= (Backup &&other) = delete;
+		Backup(const Backup &) = delete;
+		Backup & operator= (const Backup &) = delete;
+		bool scan(
+				const std::unordered_set<table_key_t> &exclude_tables,
+				std::unordered_set<table_key_t> *out_processed_tables,
+				bool no_lock, bool collect_log_and_stats);
+		bool copy_log_tables(bool finalize);
+		bool copy_stats_tables();
+		bool wait_for_finish();
+		bool close_log_tables();
+		void set_post_copy_table_hook(const post_copy_table_hook_t &hook);
+	private:
+		BackupImpl *m_backup_impl;
+};
+
+} // namespace common_engine
+

extra/mariabackup/datasink.cc

@@ -80,11 +80,11 @@ ds_create(const char *root, ds_type_t type)
 /************************************************************************
 Open a datasink file */
 ds_file_t *
-ds_open(ds_ctxt_t *ctxt, const char *path, MY_STAT *stat)
+ds_open(ds_ctxt_t *ctxt, const char *path, const MY_STAT *stat, bool rewrite)
 {
 	ds_file_t	*file;
 
-	file = ctxt->datasink->open(ctxt, path, stat);
+	file = ctxt->datasink->open(ctxt, path, stat, rewrite);
 	if (file != NULL) {
 		file->datasink = ctxt->datasink;
 	}
@@ -104,6 +104,30 @@ ds_write(ds_file_t *file, const void *buf, size_t len)
 	return file->datasink->write(file, (const uchar *)buf, len);
 }
 
+int ds_seek_set(ds_file_t *file, my_off_t offset) {
+	DBUG_ASSERT(file);
+	DBUG_ASSERT(file->datasink);
+	if (file->datasink->seek_set)
+		return file->datasink->seek_set(file, offset);
+	return 0;
+}
+
+int ds_rename(ds_ctxt_t *ctxt, const char *old_path, const char *new_path) {
+	DBUG_ASSERT(ctxt);
+	DBUG_ASSERT(ctxt->datasink);
+	if (ctxt->datasink->rename)
+		return ctxt->datasink->rename(ctxt, old_path, new_path);
+	return 0;
+}
+
+int ds_remove(ds_ctxt_t *ctxt, const char *path) {
+	DBUG_ASSERT(ctxt);
+	DBUG_ASSERT(ctxt->datasink);
+	if (ctxt->datasink->remove)
+		return ctxt->datasink->mremove(ctxt, path);
+	return 0;
+}
+
 /************************************************************************
 Close a datasink file.
 @return 0 on success, 1, on error. */

extra/mariabackup/datasink.h

@@ -43,7 +43,8 @@ typedef struct ds_ctxt {
         */
         bool copy_file(const char *src_file_path,
                        const char *dst_file_path,
-                       uint thread_n);
+                       uint thread_n,
+                       bool rewrite = false);
 
         bool move_file(const char *src_file_path,
                        const char *dst_file_path,
@@ -76,10 +77,15 @@ typedef struct {
 
 struct datasink_struct {
 	ds_ctxt_t *(*init)(const char *root);
-	ds_file_t *(*open)(ds_ctxt_t *ctxt, const char *path, MY_STAT *stat);
+	ds_file_t *(*open)(ds_ctxt_t *ctxt, const char *path,
+		const MY_STAT *stat, bool rewrite);
 	int (*write)(ds_file_t *file, const unsigned char *buf, size_t len);
+	int (*seek_set)(ds_file_t *file, my_off_t offset);
 	int (*close)(ds_file_t *file);
 	int (*remove)(const char *path);
+	// TODO: consider to return bool from "rename" and "remove"
+	int (*rename)(ds_ctxt_t *ctxt, const char *old_path, const char *new_path);
+	int (*mremove)(ds_ctxt_t *ctxt, const char *path);
 	void (*deinit)(ds_ctxt_t *ctxt);
 };
 
@@ -106,12 +112,17 @@ ds_ctxt_t *ds_create(const char *root, ds_type_t type);
 
 /************************************************************************
 Open a datasink file */
-ds_file_t *ds_open(ds_ctxt_t *ctxt, const char *path, MY_STAT *stat);
+ds_file_t *ds_open(
+	ds_ctxt_t *ctxt, const char *path, const MY_STAT *stat, bool rewrite = false);
 
 /************************************************************************
 Write to a datasink file.
 @return 0 on success, 1 on error. */
 int ds_write(ds_file_t *file, const void *buf, size_t len);
+int ds_seek_set(ds_file_t *file, my_off_t offset);
+
+int ds_rename(ds_ctxt_t *ctxt, const char *old_path, const char *new_path);
+int ds_remove(ds_ctxt_t *ctxt, const char *path);
 
 /************************************************************************
 Close a datasink file.

extra/mariabackup/ddl_log.h

+#pragma once
+#include "my_global.h"
+#include "datasink.h"
+#include "aria_backup_client.h"
+#include <string>
+#include <memory>
+#include <vector>
+#include <unordered_map>
+
+namespace ddl_log {
+
+typedef std::unordered_map<std::string, std::string> tables_t;
+bool backup(const char *datadir_path, ds_ctxt_t *ds, const tables_t &tables);
+
+} // namespace ddl_log

extra/mariabackup/ds_buffer.cc

@@ -44,7 +44,7 @@ typedef struct {
 
 static ds_ctxt_t *buffer_init(const char *root);
 static ds_file_t *buffer_open(ds_ctxt_t *ctxt, const char *path,
-			      MY_STAT *mystat);
+			      const MY_STAT *mystat, bool rewrite);
 static int buffer_write(ds_file_t *file, const uchar *buf, size_t len);
 static int buffer_close(ds_file_t *file);
 static void buffer_deinit(ds_ctxt_t *ctxt);
@@ -53,8 +53,11 @@ datasink_t datasink_buffer = {
 	&buffer_init,
 	&buffer_open,
 	&buffer_write,
+	nullptr,
 	&buffer_close,
 	&dummy_remove,
+	nullptr,
+	nullptr,
 	&buffer_deinit
 };
 
@@ -84,8 +87,10 @@ buffer_init(const char *root)
 }
 
 static ds_file_t *
-buffer_open(ds_ctxt_t *ctxt, const char *path, MY_STAT *mystat)
+buffer_open(ds_ctxt_t *ctxt, const char *path,
+	const MY_STAT *mystat, bool rewrite)
 {
+  DBUG_ASSERT(rewrite == false);
 	ds_buffer_ctxt_t	*buffer_ctxt;
 	ds_ctxt_t		*pipe_ctxt;
 	ds_file_t		*dst_file;

extra/mariabackup/ds_compress.cc

@@ -65,7 +65,7 @@ extern ulonglong	xtrabackup_compress_chunk_size;
 
 static ds_ctxt_t *compress_init(const char *root);
 static ds_file_t *compress_open(ds_ctxt_t *ctxt, const char *path,
-				MY_STAT *mystat);
+				const MY_STAT *mystat, bool rewrite);
 static int compress_write(ds_file_t *file, const uchar *buf, size_t len);
 static int compress_close(ds_file_t *file);
 static void compress_deinit(ds_ctxt_t *ctxt);
@@ -74,8 +74,11 @@ datasink_t datasink_compress = {
 	&compress_init,
 	&compress_open,
 	&compress_write,
+	nullptr,
 	&compress_close,
 	&dummy_remove,
+	nullptr,
+	nullptr,
 	&compress_deinit
 };
 
@@ -116,8 +119,10 @@ compress_init(const char *root)
 
 static
 ds_file_t *
-compress_open(ds_ctxt_t *ctxt, const char *path, MY_STAT *mystat)
+compress_open(ds_ctxt_t *ctxt, const char *path,
+	const MY_STAT *mystat, bool rewrite)
 {
+  DBUG_ASSERT(rewrite == false);
 	ds_compress_ctxt_t	*comp_ctxt;
 	ds_ctxt_t		*dest_ctxt;
  	ds_file_t		*dest_file;

extra/mariabackup/ds_local.cc

@@ -42,8 +42,9 @@ typedef struct {
 
 static ds_ctxt_t *local_init(const char *root);
 static ds_file_t *local_open(ds_ctxt_t *ctxt, const char *path,
-			     MY_STAT *mystat);
+			     const MY_STAT *mystat, bool rewrite);
 static int local_write(ds_file_t *file, const uchar *buf, size_t len);
+static int local_seek_set(ds_file_t *file, my_off_t offset);
 static int local_close(ds_file_t *file);
 static void local_deinit(ds_ctxt_t *ctxt);
 
@@ -52,13 +53,20 @@ static int local_remove(const char *path)
 	return unlink(path);
 }
 
+static int local_rename(
+	ds_ctxt_t *ctxt, const char *old_path, const char *new_path);
+static int local_mremove(ds_ctxt_t *ctxt, const char *path);
+
 extern "C" {
 datasink_t datasink_local = {
 	&local_init,
 	&local_open,
 	&local_write,
+	&local_seek_set,
 	&local_close,
 	&local_remove,
+	&local_rename,
+	&local_mremove,
 	&local_deinit
 };
 }
@@ -89,7 +97,7 @@ local_init(const char *root)
 static
 ds_file_t *
 local_open(ds_ctxt_t *ctxt, const char *path,
-	   MY_STAT *mystat __attribute__((unused)))
+	   const MY_STAT *mystat __attribute__((unused)), bool rewrite)
 {
 	char 		fullpath[FN_REFLEN];
 	char		dirpath[FN_REFLEN];
@@ -111,8 +119,10 @@ local_open(ds_ctxt_t *ctxt, const char *path,
 		return NULL;
 	}
 
-	fd = my_create(fullpath, 0, O_WRONLY | O_BINARY | O_EXCL | O_NOFOLLOW,
-		     MYF(MY_WME));
+	// TODO: check in Windows and set the corresponding flags on fail
+	fd = my_create(fullpath, 0,
+		O_WRONLY | O_BINARY | (rewrite ? O_TRUNC : O_EXCL) | O_NOFOLLOW,
+		MYF(MY_WME));
 	if (fd < 0) {
 		return NULL;
 	}
@@ -194,8 +204,8 @@ static void init_ibd_data(ds_local_file_t *local_file, const uchar *buf, size_t
 		return;
 	}
 
-	auto flags = mach_read_from_4(&buf[FIL_PAGE_DATA + FSP_SPACE_FLAGS]);
-	auto ssize = FSP_FLAGS_GET_PAGE_SSIZE(flags);
+	uint32_t flags = mach_read_from_4(&buf[FIL_PAGE_DATA + FSP_SPACE_FLAGS]);
+	uint32_t ssize = FSP_FLAGS_GET_PAGE_SSIZE(flags);
 	local_file->pagesize= ssize == 0 ? UNIV_PAGE_SIZE_ORIG : ((UNIV_ZIP_SIZE_MIN >> 1) << ssize);
 	local_file->compressed = fil_space_t::full_crc32(flags)
 		? fil_space_t::is_compressed(flags)
@@ -239,6 +249,15 @@ local_write(ds_file_t *file, const uchar *buf, size_t len)
 	return 1;
 }
 
+static
+int
+local_seek_set(ds_file_t *file, my_off_t offset) {
+	ds_local_file_t *local_file= (ds_local_file_t *)file->ptr;
+	if (my_seek(local_file->fd, offset, SEEK_SET, MYF(0)) == MY_FILEPOS_ERROR)
+		return 1;
+	return 0;
+}
+
 /* Set EOF at file's current position.*/
 static int set_eof(File fd)
 {
@@ -276,3 +295,77 @@ local_deinit(ds_ctxt_t *ctxt)
 	my_free(ctxt->root);
 	my_free(ctxt);
 }
+
+
+static int local_rename(
+	ds_ctxt_t *ctxt, const char *old_path, const char *new_path) {
+	char		full_old_path[FN_REFLEN];
+	char		full_new_path[FN_REFLEN];
+	fn_format(full_old_path, old_path, ctxt->root, "", MYF(MY_RELATIVE_PATH));
+	fn_format(full_new_path, new_path, ctxt->root, "", MYF(MY_RELATIVE_PATH));
+	// Ignore errors as .frm files can me copied separately.
+	// TODO: return error processing here after the corresponding changes in
+	// xtrabackup.cc
+	(void)my_rename(full_old_path, full_new_path, MYF(0));
+//	if (my_rename(full_old_path, full_new_path, MYF(0))) {
+//		msg("Failed to rename file %s to %s", old_path, new_path);
+//		return 1;
+//	}
+	return 0;
+}
+
+// It's ok if destination does not contain the file or folder
+static int local_mremove(ds_ctxt_t *ctxt, const char *path) {
+	char		full_path[FN_REFLEN];
+	fn_format(full_path, path, ctxt->root, "", MYF(MY_RELATIVE_PATH));
+	size_t full_path_len = strlen(full_path);
+	if (full_path[full_path_len - 1] == '*') {
+		full_path[full_path_len - 1] = '\0';
+		char *preffix = strrchr(full_path, '/');
+		const char *full_path_dir = full_path;
+		size_t preffix_len;
+		if (preffix) {
+			preffix_len = (full_path_len - 1) - (preffix - full_path);
+			*(preffix++) = '\0';
+		}
+		else {
+			preffix = full_path;
+			preffix_len = full_path_len - 1;
+			full_path_dir= IF_WIN(".\\", "./");
+		}
+		if (!preffix_len)
+			return 0;
+		MY_DIR *dir= my_dir(full_path_dir, 0);
+		if (!dir)
+			return 0;
+		for (size_t i = 0; i < dir->number_of_files; ++i) {
+			char full_fpath[FN_REFLEN];
+			if (strncmp(dir->dir_entry[i].name, preffix, preffix_len))
+				continue;
+			fn_format(full_fpath, dir->dir_entry[i].name,
+					full_path_dir, "", MYF(MY_RELATIVE_PATH));
+			(void)my_delete(full_fpath, MYF(0));
+		}
+		my_dirend(dir);
+	}
+	else {
+		MY_STAT stat;
+		if (!my_stat(full_path, &stat, MYF(0)))
+			return 0;
+		MY_DIR *dir= my_dir(full_path, 0);
+		if (!dir) {
+			// TODO: check for error here if necessary
+			(void)my_delete(full_path, MYF(0));
+			return 0;
+		}
+		for (size_t i = 0; i < dir->number_of_files; ++i) {
+			char		full_fpath[FN_REFLEN];
+			fn_format(full_fpath, dir->dir_entry[i].name,
+					full_path, "", MYF(MY_RELATIVE_PATH));
+			(void)my_delete(full_fpath, MYF(0));
+		}
+		my_dirend(dir);
+		(void)my_rmtree(full_path, MYF(0));
+	}
+	return 0;
+}

extra/mariabackup/ds_stdout.cc

@@ -30,7 +30,7 @@ typedef struct {
 
 static ds_ctxt_t *stdout_init(const char *root);
 static ds_file_t *stdout_open(ds_ctxt_t *ctxt, const char *path,
-			     MY_STAT *mystat);
+			const MY_STAT *mystat, bool rewrite);
 static int stdout_write(ds_file_t *file, const uchar *buf, size_t len);
 static int stdout_close(ds_file_t *file);
 static void stdout_deinit(ds_ctxt_t *ctxt);
@@ -39,8 +39,11 @@ datasink_t datasink_stdout = {
 	&stdout_init,
 	&stdout_open,
 	&stdout_write,
+	nullptr,
 	&stdout_close,
 	&dummy_remove,
+	nullptr,
+	nullptr,
 	&stdout_deinit
 };
 
@@ -61,8 +64,9 @@ static
 ds_file_t *
 stdout_open(ds_ctxt_t *ctxt __attribute__((unused)),
 	    const char *path __attribute__((unused)),
-	    MY_STAT *mystat __attribute__((unused)))
+	    const MY_STAT *mystat __attribute__((unused)), bool rewrite)
 {
+  DBUG_ASSERT(rewrite == false);
 	ds_stdout_file_t 	*stdout_file;
 	ds_file_t		*file;
 	size_t			pathlen;

extra/mariabackup/ds_tmpfile.cc

@@ -41,7 +41,7 @@ typedef struct {
 
 static ds_ctxt_t *tmpfile_init(const char *root);
 static ds_file_t *tmpfile_open(ds_ctxt_t *ctxt, const char *path,
-			       MY_STAT *mystat);
+			const MY_STAT *mystat, bool rewrite);
 static int tmpfile_write(ds_file_t *file, const uchar *buf, size_t len);
 static int tmpfile_close(ds_file_t *file);
 static void tmpfile_deinit(ds_ctxt_t *ctxt);
@@ -50,8 +50,11 @@ datasink_t datasink_tmpfile = {
 	&tmpfile_init,
 	&tmpfile_open,
 	&tmpfile_write,
+	nullptr,
 	&tmpfile_close,
 	&dummy_remove,
+	nullptr,
+	nullptr,
 	&tmpfile_deinit
 };
 
@@ -80,8 +83,9 @@ tmpfile_init(const char *root)
 
 static ds_file_t *
 tmpfile_open(ds_ctxt_t *ctxt, const char *path,
-			       MY_STAT *mystat)
+			const MY_STAT *mystat, bool rewrite)
 {
+  DBUG_ASSERT(rewrite == false);
 	ds_tmpfile_ctxt_t	*tmpfile_ctxt;
 	char			 tmp_path[FN_REFLEN];
 	ds_tmp_file_t		*tmp_file;

extra/mariabackup/ds_xbstream.cc

@@ -40,24 +40,31 @@ General streaming interface */
 
 static ds_ctxt_t *xbstream_init(const char *root);
 static ds_file_t *xbstream_open(ds_ctxt_t *ctxt, const char *path,
-			      MY_STAT *mystat);
+			const MY_STAT *mystat, bool rewrite);
 static int xbstream_write(ds_file_t *file, const uchar *buf, size_t len);
+static int xbstream_seek_set(ds_file_t *file, my_off_t offset);
 static int xbstream_close(ds_file_t *file);
 static void xbstream_deinit(ds_ctxt_t *ctxt);
 
+static int xbstream_rename(
+	ds_ctxt_t *ctxt, const char *old_path, const char *new_path);
+static int xbstream_mremove(ds_ctxt_t *ctxt, const char *path);
+
 datasink_t datasink_xbstream = {
 	&xbstream_init,
 	&xbstream_open,
 	&xbstream_write,
+	&xbstream_seek_set,
 	&xbstream_close,
 	&dummy_remove,
+	&xbstream_rename,
+	&xbstream_mremove,
 	&xbstream_deinit
 };
 
 static
 ssize_t
-my_xbstream_write_callback(xb_wstream_file_t *f __attribute__((unused)),
-		       void *userdata, const void *buf, size_t len)
+my_xbstream_write_callback(void *userdata, const void *buf, size_t len)
 {
 	ds_stream_ctxt_t	*stream_ctxt;
 
@@ -89,7 +96,7 @@ xbstream_init(const char *root __attribute__((unused)))
 		goto err;
 	}
 
-	xbstream = xb_stream_write_new();
+	xbstream = xb_stream_write_new(my_xbstream_write_callback, stream_ctxt);
 	if (xbstream == NULL) {
 		msg("xb_stream_write_new() failed.");
 		goto err;
@@ -108,7 +115,8 @@ xbstream_init(const char *root __attribute__((unused)))
 
 static
 ds_file_t *
-xbstream_open(ds_ctxt_t *ctxt, const char *path, MY_STAT *mystat)
+xbstream_open(ds_ctxt_t *ctxt, const char *path,
+	const MY_STAT *mystat, bool rewrite)
 {
 	ds_file_t		*file;
 	ds_stream_file_t	*stream_file;
@@ -144,9 +152,7 @@ xbstream_open(ds_ctxt_t *ctxt, const char *path, MY_STAT *mystat)
 
 	xbstream = stream_ctxt->xbstream;
 
-	xbstream_file = xb_stream_write_open(xbstream, path, mystat,
-		                             stream_ctxt,
-					     my_xbstream_write_callback);
+	xbstream_file = xb_stream_write_open(xbstream, path, mystat, rewrite);
 
 	if (xbstream_file == NULL) {
 		msg("xb_stream_write_open() failed.");
@@ -190,6 +196,45 @@ xbstream_write(ds_file_t *file, const uchar *buf, size_t len)
 	return 0;
 }
 
+static
+int
+xbstream_seek_set(ds_file_t *file, my_off_t offset)
+{
+	ds_stream_file_t	*stream_file;
+	xb_wstream_file_t	*xbstream_file;
+
+
+	stream_file = (ds_stream_file_t *) file->ptr;
+
+	xbstream_file = stream_file->xbstream_file;
+
+	if (xb_stream_write_seek_set(xbstream_file, offset)) {
+		msg("xb_stream_write_seek_set() failed.");
+		return 1;
+	}
+
+	return 0;
+}
+
+static
+int
+xbstream_mremove(ds_ctxt_t *ctxt, const char *path) {
+	ds_stream_ctxt_t	*stream_ctxt =
+		reinterpret_cast<ds_stream_ctxt_t *>(ctxt->ptr);
+	xb_wstream_t		*xbstream = stream_ctxt->xbstream;
+	return xb_stream_write_remove(xbstream, path);
+}
+
+static
+int
+xbstream_rename(
+	ds_ctxt_t *ctxt, const char *old_path, const char *new_path) {
+	ds_stream_ctxt_t	*stream_ctxt =
+		reinterpret_cast<ds_stream_ctxt_t *>(ctxt->ptr);
+	xb_wstream_t		*xbstream = stream_ctxt->xbstream;
+	return xb_stream_write_rename(xbstream, old_path, new_path);
+}
+
 static
 int
 xbstream_close(ds_file_t *file)

extra/mariabackup/xb_plugin.cc → extra/mariabackup/encryption_plugin.cc

-/* Copyright (c) 2017, 2022, MariaDB Corporation.
+/* Copyright (c) 2017, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -17,18 +17,18 @@
 #include <mysqld.h>
 #include <mysql.h>
 #include <xtrabackup.h>
-#include <xb_plugin.h>
+#include <encryption_plugin.h>
 #include <sql_plugin.h>
 #include <sstream>
 #include <vector>
 #include <common.h>
 #include <backup_mysql.h>
-#include <srv0srv.h>
+#include <log0crypt.h>
 
 
 extern struct st_maria_plugin *mysql_optional_plugins[];
 extern struct st_maria_plugin *mysql_mandatory_plugins[];
-static void xb_plugin_init(int argc, char **argv);
+static void encryption_plugin_init(int argc, char **argv);
 
 extern char *xb_plugin_load;
 extern char *xb_plugin_dir;
@@ -42,7 +42,7 @@ const char *QUERY_PLUGIN =
 " OR (plugin_type = 'DAEMON' AND plugin_name LIKE 'provider\\_%')"
 " AND plugin_status='ACTIVE'";
 
-std::string xb_plugin_config;
+std::string encryption_plugin_config;
 
 static void add_to_plugin_load_list(const char *plugin_def)
 {
@@ -52,16 +52,16 @@ static void add_to_plugin_load_list(const char *plugin_def)
 static char XTRABACKUP_EXE[] = "xtrabackup";
 
 /*
-  Read "plugin-load" value from backup-my.cnf during prepare phase.
+  Read "plugin-load" value (encryption plugin) from backup-my.cnf during
+  prepare phase.
   The value is stored during backup phase.
 */
-static std::string get_plugin_from_cnf(const char *dir)
+static std::string get_encryption_plugin_from_cnf()
 {
-  std::string path = dir + std::string("/backup-my.cnf");
-  FILE *f = fopen(path.c_str(), "r");
+  FILE *f = fopen("backup-my.cnf", "r");
   if (!f)
   {
-    die("Can't open %s for reading", path.c_str());
+    die("Can't open backup-my.cnf for reading");
   }
   char line[512];
   std::string plugin_load;
@@ -80,7 +80,7 @@ static std::string get_plugin_from_cnf(const char *dir)
 }
 
 
-void xb_plugin_backup_init(MYSQL *mysql)
+void encryption_plugin_backup_init(MYSQL *mysql)
 {
   MYSQL_RES *result;
   MYSQL_ROW row;
@@ -163,7 +163,7 @@ void xb_plugin_backup_init(MYSQL *mysql)
     mysql_free_result(result);
   }
 
-  xb_plugin_config = oss.str();
+  encryption_plugin_config = oss.str();
 
   argc = 0;
   argv[argc++] = XTRABACKUP_EXE;
@@ -175,23 +175,23 @@ void xb_plugin_backup_init(MYSQL *mysql)
   }
   argv[argc] = 0;
 
-  xb_plugin_init(argc, argv);
+  encryption_plugin_init(argc, argv);
 }
 
-const char *xb_plugin_get_config()
+const char *encryption_plugin_get_config()
 {
-  return xb_plugin_config.c_str();
+  return encryption_plugin_config.c_str();
 }
 
 extern int finalize_encryption_plugin(st_plugin_int *plugin);
 
 
-void xb_plugin_prepare_init(int argc, char **argv, const char *dir)
+void encryption_plugin_prepare_init(int argc, char **argv)
 {
-  std::string plugin_load= get_plugin_from_cnf(dir ? dir : ".");
+  std::string plugin_load= get_encryption_plugin_from_cnf();
   if (plugin_load.size())
   {
-    msg("Loading plugins from %s", plugin_load.c_str());
+    msg("Loading encryption plugin from %s", plugin_load.c_str());
   }
   else
   {
@@ -211,19 +211,19 @@ void xb_plugin_prepare_init(int argc, char **argv, const char *dir)
   new_argv[0] = XTRABACKUP_EXE;
   memcpy(&new_argv[1], argv, argc*sizeof(char *));
 
-  xb_plugin_init(argc+1, new_argv);
+  encryption_plugin_init(argc+1, new_argv);
 
   delete[] new_argv;
 }
 
-static void xb_plugin_init(int argc, char **argv)
+static void encryption_plugin_init(int argc, char **argv)
 {
   /* Patch optional and mandatory plugins, we only need to load the one in xb_plugin_load. */
   mysql_optional_plugins[0] = mysql_mandatory_plugins[0] = 0;
   plugin_maturity = MariaDB_PLUGIN_MATURITY_UNKNOWN; /* mariabackup accepts all plugins */
-  msg("Loading plugins");
+  msg("Loading encryption plugin");
   for (int i= 1; i < argc; i++)
-    msg("\t Plugin parameter :  '%s'", argv[i]);
+    msg("\t Encryption plugin parameter :  '%s'", argv[i]);
   plugin_init(&argc, argv, PLUGIN_INIT_SKIP_PLUGIN_TABLE);
 }
 

extra/mariabackup/encryption_plugin.h

+#include <mysql.h>
+#include <string>
+extern void encryption_plugin_backup_init(MYSQL *mysql);
+extern const char* encryption_plugin_get_config();
+extern void encryption_plugin_prepare_init(int argc, char **argv);
+
+//extern  void encryption_plugin_init(int argc, char **argv);

extra/mariabackup/innobackupex.cc

@@ -78,7 +78,6 @@ my_bool opt_ibx_galera_info = FALSE;
 my_bool opt_ibx_slave_info = FALSE;
 my_bool opt_ibx_no_lock = FALSE;
 my_bool opt_ibx_safe_slave_backup = FALSE;
-my_bool opt_ibx_rsync = FALSE;
 my_bool opt_ibx_force_non_empty_dirs = FALSE;
 my_bool opt_ibx_noversioncheck = FALSE;
 my_bool opt_ibx_no_backup_locks = FALSE;
@@ -297,15 +296,6 @@ static struct my_option ibx_long_options[] =
 	 (uchar *) &opt_ibx_safe_slave_backup,
 	 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
 
-	{"rsync", OPT_RSYNC, "Uses the rsync utility to optimize local file "
-	 "transfers. When this option is specified, innobackupex uses rsync "
-	 "to copy all non-InnoDB files instead of spawning a separate cp for "
-	 "each file, which can be much faster for servers with a large number "
-	 "of databases or tables.  This option cannot be used together with "
-	 "--stream.",
-	 (uchar *) &opt_ibx_rsync, (uchar *) &opt_ibx_rsync,
-	 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
-
 	{"force-non-empty-directories", OPT_FORCE_NON_EMPTY_DIRS, "This "
 	 "option, when specified, makes --copy-back or --move-back transfer "
 	 "files to non-empty directories. Note that no existing files will be "
@@ -864,10 +854,8 @@ ibx_init()
 	opt_slave_info = opt_ibx_slave_info;
 	opt_no_lock = opt_ibx_no_lock;
 	opt_safe_slave_backup = opt_ibx_safe_slave_backup;
-	opt_rsync = opt_ibx_rsync;
 	opt_force_non_empty_dirs = opt_ibx_force_non_empty_dirs;
 	opt_noversioncheck = opt_ibx_noversioncheck;
-	opt_no_backup_locks = opt_ibx_no_backup_locks;
 	opt_decompress = opt_ibx_decompress;
 
 	opt_incremental_history_name = opt_ibx_incremental_history_name;

extra/mariabackup/thread_pool.cc

+#include "thread_pool.h"
+#include "common.h"
+
+bool ThreadPool::start(size_t threads_count) {
+	if (!m_stopped)
+		return false;
+	m_stopped = false;
+	for (unsigned i = 0; i < threads_count; ++i)
+		m_threads.emplace_back(&ThreadPool::thread_func, this, i);
+	return true;
+}
+
+void ThreadPool::stop() {
+	if (m_stopped)
+		return;
+	m_stop = true;
+	m_cv.notify_all();
+	for (auto &t : m_threads)
+		t.join();
+	m_stopped = true;
+};
+
+void ThreadPool::push(ThreadPool::job_t &&j) {
+	std::unique_lock<std::mutex> lock(m_mutex);
+	m_jobs.push(j);
+	lock.unlock();
+	m_cv.notify_one();
+}
+
+void ThreadPool::thread_func(unsigned thread_num) {
+	if (my_thread_init())
+		die("Can't init mysql thread");
+	std::unique_lock<std::mutex> lock(m_mutex);
+	while(true) {
+		if (m_stop)
+			goto exit;
+		while (!m_jobs.empty()) {
+			if (m_stop)
+				goto exit;
+			job_t j = std::move(m_jobs.front());
+			m_jobs.pop();
+			lock.unlock();
+			j(thread_num);
+			lock.lock();
+		}
+		m_cv.wait(lock, [&] { return m_stop || !m_jobs.empty(); });
+	}
+exit:
+	my_thread_end();
+}

extra/mariabackup/thread_pool.h

+#pragma once
+#include <queue>
+#include <vector>
+#include <functional>
+#include <thread>
+#include <mutex>
+#include <condition_variable>
+#include <atomic>
+#include "trx0sys.h"
+
+class ThreadPool {
+public:
+	typedef std::function<void(unsigned)> job_t;
+
+	ThreadPool() { m_stop = false; m_stopped = true; }
+	ThreadPool (ThreadPool &&other) = delete;
+	ThreadPool &  operator= (ThreadPool &&other) = delete;
+	ThreadPool(const ThreadPool &) = delete;
+	ThreadPool & operator= (const ThreadPool &) = delete;
+
+	bool start(size_t threads_count);
+	void stop();
+	void push(job_t &&j);
+	size_t threads_count() const { return m_threads.size(); }
+private:
+	void thread_func(unsigned thread_num);
+	std::mutex m_mutex;
+	std::condition_variable m_cv;
+	std::queue<job_t> m_jobs;
+	std::atomic<bool> m_stop;
+	std::atomic<bool> m_stopped;
+	std::vector<std::thread> m_threads;
+};
+
+class TasksGroup {
+public:
+	TasksGroup(ThreadPool &thread_pool) : m_thread_pool(thread_pool) {
+		m_tasks_count = 0;
+		m_tasks_result = 1;
+	}
+	void push_task(ThreadPool::job_t &&j) {
+		++m_tasks_count;
+		m_thread_pool.push(std::forward<ThreadPool::job_t>(j));
+	}
+	void finish_task(int res) {
+		--m_tasks_count;
+		m_tasks_result.fetch_and(res);
+	}
+	int get_result() const { return m_tasks_result; }
+	bool is_finished() const {
+		return !m_tasks_count;
+	}
+	bool wait_for_finish() {
+		while (!is_finished())
+                        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+		return get_result();
+	}
+private:
+	ThreadPool &m_thread_pool;
+	std::atomic<size_t> m_tasks_count;
+	std::atomic<int> m_tasks_result;
+};

extra/mariabackup/write_filt.cc

@@ -144,6 +144,18 @@ wf_incremental_process(xb_write_filt_ctxt_t *ctxt, ds_file_t *dstfile)
 			return false;
 		}
 
+		/* Check whether TRX_SYS page has been changed */
+		if (mach_read_from_4(page + FIL_PAGE_SPACE_ID)
+				== TRX_SYS_SPACE
+		    && mach_read_from_4(page + FIL_PAGE_OFFSET)
+				== TRX_SYS_PAGE_NO) {
+			msg(cursor->thread_n,
+			    "--incremental backup is impossible if "
+			    "the server had been restarted with "
+			    "different innodb_undo_tablespaces.");
+			return false;
+		}
+
 		/* updated page */
 		if (cp->npages == page_size / 4) {
 			/* flush buffer */

extra/mariabackup/wsrep.cc

@@ -55,6 +55,9 @@ permission notice:
 #define XB_GALERA_INFO_FILENAME "xtrabackup_galera_info"
 #define XB_GALERA_DONOR_INFO_FILENAME "donor_galera_info"
 
+/* backup copy of galera info file as sent by donor */
+#define XB_GALERA_INFO_FILENAME_SST "xtrabackup_galera_info_SST"
+
 /***********************************************************************
 Store Galera checkpoint info in the 'xtrabackup_galera_info' file, if that
 information is present in the trx system header. Otherwise, do nothing. */
@@ -68,21 +71,47 @@ xb_write_galera_info(bool incremental_prepare)
 	long long	seqno;
 	MY_STAT		statinfo;
 
-	/* Do not overwrite an existing file to be compatible with
-	servers with older server versions */
-	if (!incremental_prepare &&
-		my_stat(XB_GALERA_INFO_FILENAME, &statinfo, MYF(0)) != NULL) {
-
-		return;
-	}
-
 	xid.null();
 
+	/* try to read last wsrep XID from innodb rsegs, we will use it
+	   instead of galera info file received from donor
+	*/
 	if (!trx_rseg_read_wsrep_checkpoint(xid)) {
-
+		/* no worries yet, SST may have brought in galera info file
+		   from some old MariaDB version, which does not support
+		   wsrep XID storing in innodb rsegs
+		*/
 		return;
 	}
 
+	/* if SST brought in galera info file, copy it as *_SST file
+	   this will not be used, saved just for future reference
+	*/
+	if (my_stat(XB_GALERA_INFO_FILENAME, &statinfo, MYF(0)) != NULL) {
+		FILE* fp_in  = fopen(XB_GALERA_INFO_FILENAME, "r");
+		FILE* fp_out = fopen(XB_GALERA_INFO_FILENAME_SST, "w");
+
+		char buf[BUFSIZ] = {'\0'};
+		size_t size;
+		while ((size = fread(buf, 1, BUFSIZ, fp_in))) {
+			if (fwrite(buf, 1, size, fp_out) != strlen(buf)) {
+				die(
+				    "could not write to "
+				    XB_GALERA_INFO_FILENAME_SST
+				    ", errno = %d\n",
+				    errno);
+			}
+		}
+		if (!feof(fp_in)) {
+			die(
+			    XB_GALERA_INFO_FILENAME_SST
+			    " not fully copied\n"
+			);
+		}
+		fclose(fp_out);
+		fclose(fp_in);
+	}
+
 	wsrep_uuid_t uuid;
 	memcpy(uuid.data, wsrep_xid_uuid(&xid), sizeof(uuid.data));
 	if (wsrep_uuid_print(&uuid, uuid_str,
@@ -97,7 +126,6 @@ xb_write_galera_info(bool incremental_prepare)
 		    "could not create " XB_GALERA_INFO_FILENAME
 		    ", errno = %d\n",
 		    errno);
-		exit(EXIT_FAILURE);
 	}
 
 	seqno = wsrep_xid_seqno(&xid);

extra/mariabackup/xb_plugin.h

-#include <mysql.h>
-#include <string>
-extern void xb_plugin_backup_init(MYSQL *mysql);
-extern const char* xb_plugin_get_config();
-extern void xb_plugin_prepare_init(int argc, char **argv, const char *dir);

extra/mariabackup/xbstream.cc

@@ -262,7 +262,7 @@ mode_create(int argc, char **argv)
 		return 1;
 	}
 
-	stream = xb_stream_write_new();
+	stream = xb_stream_write_new(nullptr, nullptr);
 	if (stream == NULL) {
 		msg("%s: xb_stream_write_new() failed.", my_progname);
 		return 1;
@@ -287,7 +287,7 @@ mode_create(int argc, char **argv)
 			goto err;
 		}
 
-		file = xb_stream_write_open(stream, filepath, &mystat, NULL, NULL);
+		file = xb_stream_write_open(stream, filepath, &mystat, false);
 		if (file == NULL) {
 			goto err;
 		}
@@ -314,7 +314,8 @@ mode_create(int argc, char **argv)
 
 static
 file_entry_t *
-file_entry_new(extract_ctxt_t *ctxt, const char *path, uint pathlen)
+file_entry_new(extract_ctxt_t *ctxt, const char *path, uint pathlen,
+	uchar chunk_flags)
 {
 	file_entry_t	*entry;
 	ds_file_t	*file;
@@ -331,7 +332,8 @@ file_entry_new(extract_ctxt_t *ctxt, const char *path, uint pathlen)
 	}
 	entry->pathlen = pathlen;
 
-	file = ds_open(ctxt->ds_ctxt, path, NULL);
+	file = ds_open(ctxt->ds_ctxt, path, NULL,
+		chunk_flags == XB_STREAM_FLAG_REWRITE);
 
 	if (file == NULL) {
 		msg("%s: failed to create file.", my_progname);
@@ -412,10 +414,50 @@ extract_worker_thread_func(void *arg)
 							(uchar *) chunk.path,
 							chunk.pathlen);
 
+		if (entry && (chunk.type == XB_CHUNK_TYPE_REMOVE ||
+			chunk.type == XB_CHUNK_TYPE_RENAME)) {
+			msg("%s: rename and remove chunks can not be applied to opened file: %s",
+				my_progname, chunk.path);
+			pthread_mutex_unlock(ctxt->mutex);
+			break;
+		}
+
+		if (chunk.type == XB_CHUNK_TYPE_REMOVE) {
+			if (ds_remove(ctxt->ds_ctxt, chunk.path)) {
+				msg("%s: error on file removing: %s", my_progname, chunk.path);
+				pthread_mutex_unlock(ctxt->mutex);
+				res = XB_STREAM_READ_ERROR;
+				break;
+			}
+			pthread_mutex_unlock(ctxt->mutex);
+			continue;
+		}
+
+		if (chunk.type == XB_CHUNK_TYPE_RENAME) {
+			if (my_hash_search(ctxt->filehash,
+				reinterpret_cast<const uchar *>(chunk.data), chunk.length)) {
+				msg("%s: rename chunks can not be applied to opened file: %s",
+					my_progname, reinterpret_cast<const uchar *>(chunk.data));
+				pthread_mutex_unlock(ctxt->mutex);
+				break;
+			}
+			if (ds_rename(ctxt->ds_ctxt, chunk.path,
+				reinterpret_cast<const char *>(chunk.data))) {
+				msg("%s: error on file renaming: %s to %s", my_progname,
+					reinterpret_cast<const char *>(chunk.data), chunk.path);
+				pthread_mutex_unlock(ctxt->mutex);
+				res = XB_STREAM_READ_ERROR;
+				break;
+			}
+			pthread_mutex_unlock(ctxt->mutex);
+			continue;
+		}
+
 		if (entry == NULL) {
 			entry = file_entry_new(ctxt,
 					       chunk.path,
-					       chunk.pathlen);
+					       chunk.pathlen,
+								 chunk.flags);
 			if (entry == NULL) {
 				pthread_mutex_unlock(ctxt->mutex);
 				break;
@@ -432,6 +474,18 @@ extract_worker_thread_func(void *arg)
 
 		pthread_mutex_unlock(ctxt->mutex);
 
+		if (chunk.type == XB_CHUNK_TYPE_SEEK) {
+			if (ds_seek_set(entry->file, chunk.offset)) {
+				msg("%s: my_seek() failed.", my_progname);
+				pthread_mutex_unlock(&entry->mutex);
+				res = XB_STREAM_READ_ERROR;
+				break;
+			}
+			entry->offset = chunk.offset;
+			pthread_mutex_unlock(&entry->mutex);
+			continue;
+		}
+
 		res = xb_stream_validate_checksum(&chunk);
 
 		if (res != XB_STREAM_READ_CHUNK) {

extra/mariabackup/xbstream.h

@@ -29,6 +29,7 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 /* Chunk flags */
 /* Chunk can be ignored if unknown version/format */
 #define XB_STREAM_FLAG_IGNORABLE 0x01
+#define XB_STREAM_FLAG_REWRITE   0x02
 
 /* Magic + flags + type + path len */
 #define CHUNK_HEADER_CONSTANT_LEN ((sizeof(XB_STREAM_CHUNK_MAGIC) - 1) + \
@@ -48,18 +49,21 @@ typedef enum {
 /************************************************************************
 Write interface. */
 
-typedef ssize_t xb_stream_write_callback(xb_wstream_file_t *file,
+typedef ssize_t xb_stream_write_callback(
 					 void *userdata,
 					 const void *buf, size_t len);
 
-xb_wstream_t *xb_stream_write_new(void);
-
+xb_wstream_t *xb_stream_write_new(
+	xb_stream_write_callback *write_callback, void *user_data);
 xb_wstream_file_t *xb_stream_write_open(xb_wstream_t *stream, const char *path,
-					MY_STAT *mystat, void *userdata,
-					xb_stream_write_callback *onwrite);
+					const MY_STAT *mystat, bool rewrite);
 
 int xb_stream_write_data(xb_wstream_file_t *file, const void *buf, size_t len);
-
+int xb_stream_write_seek_set(xb_wstream_file_t *file, my_off_t offset);
+int xb_stream_write_remove(xb_wstream_t *stream, const char *path);
+int
+xb_stream_write_rename(
+	xb_wstream_t *stream, const char *old_path, const char *new_path);
 int xb_stream_write_close(xb_wstream_file_t *file);
 
 int xb_stream_write_done(xb_wstream_t *stream);
@@ -76,6 +80,9 @@ typedef enum {
 typedef enum {
 	XB_CHUNK_TYPE_UNKNOWN = '\0',
 	XB_CHUNK_TYPE_PAYLOAD = 'P',
+	XB_CHUNK_TYPE_RENAME = 'R',
+	XB_CHUNK_TYPE_REMOVE = 'D',
+	XB_CHUNK_TYPE_SEEK = 'S',
 	XB_CHUNK_TYPE_EOF = 'E'
 } xb_chunk_type_t;
 

extra/mariabackup/xbstream_read.cc

@@ -59,6 +59,9 @@ validate_chunk_type(uchar code)
 {
 	switch ((xb_chunk_type_t) code) {
 	case XB_CHUNK_TYPE_PAYLOAD:
+	case XB_CHUNK_TYPE_RENAME:
+	case XB_CHUNK_TYPE_REMOVE:
+	case XB_CHUNK_TYPE_SEEK:
 	case XB_CHUNK_TYPE_EOF:
 		return (xb_chunk_type_t) code;
 	default:
@@ -159,57 +162,91 @@ xb_stream_read_chunk(xb_rstream_t *stream, xb_rstream_chunk_t *chunk)
 	}
 	chunk->path[pathlen] = '\0';
 
-	if (chunk->type == XB_CHUNK_TYPE_EOF) {
+	if (chunk->type == XB_CHUNK_TYPE_EOF ||
+			chunk->type == XB_CHUNK_TYPE_REMOVE) {
 		return XB_STREAM_READ_CHUNK;
 	}
 
-	/* Payload length */
-	F_READ(tmpbuf, 16);
-	ullval = uint8korr(tmpbuf);
-	if (ullval > (ulonglong) SIZE_T_MAX) {
-		msg("xb_stream_read_chunk(): chunk length is too large at "
-		    "offset 0x%llx: 0x%llx.", (ulonglong) stream->offset,
-		    ullval);
-		goto err;
+	if (chunk->type == XB_CHUNK_TYPE_RENAME) {
+		F_READ(tmpbuf, 4);
+		size_t new_pathlen = uint4korr(tmpbuf);
+		if (new_pathlen >= FN_REFLEN) {
+			msg("xb_stream_read_chunk(): path length (%lu) for new name of 'rename'"
+				"	chunk is too large", (ulong) new_pathlen);
+			goto err;
+		}
+		chunk->length = new_pathlen;
+		stream->offset +=4;
 	}
-	chunk->length = (size_t) ullval;
-	stream->offset += 8;
-
-	/* Payload offset */
-	ullval = uint8korr(tmpbuf + 8);
-	if (ullval > (ulonglong) MY_OFF_T_MAX) {
-		msg("xb_stream_read_chunk(): chunk offset is too large at "
-		    "offset 0x%llx: 0x%llx.", (ulonglong) stream->offset,
-		    ullval);
-		goto err;
+	else if (chunk->type == XB_CHUNK_TYPE_SEEK) {
+		F_READ(tmpbuf, 8);
+		chunk->offset = uint8korr(tmpbuf);
+		stream->offset += 8;
+		return XB_STREAM_READ_CHUNK;
+	}
+	else {
+		/* Payload length */
+		F_READ(tmpbuf, 16);
+		ullval = uint8korr(tmpbuf);
+		if (ullval > (ulonglong) SIZE_T_MAX) {
+			msg("xb_stream_read_chunk(): chunk length is too large at "
+					"offset 0x%llx: 0x%llx.", (ulonglong) stream->offset,
+					ullval);
+			goto err;
+		}
+		chunk->length = (size_t) ullval;
+		stream->offset += 8;
+
+		/* Payload offset */
+		ullval = uint8korr(tmpbuf + 8);
+		if (ullval > (ulonglong) MY_OFF_T_MAX) {
+			msg("xb_stream_read_chunk(): chunk offset is too large at "
+					"offset 0x%llx: 0x%llx.", (ulonglong) stream->offset,
+					ullval);
+			goto err;
+		}
+		chunk->offset = (my_off_t) ullval;
+		stream->offset += 8;
 	}
-	chunk->offset = (my_off_t) ullval;
-	stream->offset += 8;
 
-	/* Reallocate the buffer if needed */
-	if (chunk->length > chunk->buflen) {
-		chunk->data = my_realloc(PSI_NOT_INSTRUMENTED, chunk->data, chunk->length,
-					 MYF(MY_WME | MY_ALLOW_ZERO_PTR));
+	/* Reallocate the buffer if needed, take into account trailing '\0' for
+	new file name in the case of XB_CHUNK_TYPE_RENAME */
+	if (chunk->length + 1 > chunk->buflen) {
+		chunk->data = my_realloc(PSI_NOT_INSTRUMENTED, chunk->data,
+					chunk->length + 1, MYF(MY_WME | MY_ALLOW_ZERO_PTR));
 		if (chunk->data == NULL) {
 			msg("xb_stream_read_chunk(): failed to increase buffer "
-			    "to %lu bytes.", (ulong) chunk->length);
+			    "to %lu bytes.", (ulong) chunk->length + 1);
 			goto err;
 		}
-		chunk->buflen = chunk->length;
+		chunk->buflen = chunk->length + 1;
 	}
 
-	/* Checksum */
-	F_READ(tmpbuf, 4);
-	chunk->checksum = uint4korr(tmpbuf);
-	chunk->checksum_offset = stream->offset;
-
-	/* Payload */
-	if (chunk->length > 0) {
+	if (chunk->type == XB_CHUNK_TYPE_RENAME) {
+		if (chunk->length == 0) {
+			msg("xb_stream_read_chunk(): failed to read new name for file to rename "
+				": %s", chunk->path);
+			goto err;
+		}
 		F_READ(chunk->data, chunk->length);
 		stream->offset += chunk->length;
+		reinterpret_cast<char *>(chunk->data)[chunk->length] = '\0';
+		++chunk->length;
 	}
+	else {
+		/* Checksum */
+		F_READ(tmpbuf, 4);
+		chunk->checksum = uint4korr(tmpbuf);
+		chunk->checksum_offset = stream->offset;
+
+		/* Payload */
+		if (chunk->length > 0) {
+			F_READ(chunk->data, chunk->length);
+			stream->offset += chunk->length;
+		}
 
-	stream->offset += 4;
+		stream->offset += 4;
+	}
 
 	return XB_STREAM_READ_CHUNK;
 

extra/mariabackup/xbstream_write.cc

@@ -21,6 +21,7 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 #include <my_global.h>
 #include <my_base.h>
 #include <zlib.h>
+#include <stdint.h>
 #include "common.h"
 #include "xbstream.h"
 
@@ -29,6 +30,8 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 
 struct xb_wstream_struct {
 	pthread_mutex_t	mutex;
+	xb_stream_write_callback *write;
+	void *user_data;
 };
 
 struct xb_wstream_file_struct {
@@ -39,8 +42,7 @@ struct xb_wstream_file_struct {
 	char		*chunk_ptr;
 	size_t		chunk_free;
 	my_off_t	offset;
-	void		*userdata;
-	xb_stream_write_callback *write;
+  bool rewrite;
 };
 
 static int xb_stream_flush(xb_wstream_file_t *file);
@@ -50,7 +52,7 @@ static int xb_stream_write_eof(xb_wstream_file_t *file);
 
 static
 ssize_t
-xb_stream_default_write_callback(xb_wstream_file_t *file __attribute__((unused)),
+xb_stream_default_write_callback(
 				 void *userdata __attribute__((unused)),
 				 const void *buf, size_t len)
 {
@@ -60,21 +62,31 @@ xb_stream_default_write_callback(xb_wstream_file_t *file __attribute__((unused))
 }
 
 xb_wstream_t *
-xb_stream_write_new(void)
+xb_stream_write_new(
+	xb_stream_write_callback *write_callback, void *user_data)
 {
 	xb_wstream_t	*stream;
 
 	stream = (xb_wstream_t *) my_malloc(PSI_NOT_INSTRUMENTED, sizeof(xb_wstream_t), MYF(MY_FAE));
 	pthread_mutex_init(&stream->mutex, NULL);
+	if (write_callback) {
+#ifdef _WIN32
+		setmode(fileno(stdout), _O_BINARY);
+#endif
+		stream->write = write_callback;
+		stream->user_data = user_data;
+	}
+	else {
+		stream->write = xb_stream_default_write_callback;
+		stream->user_data = user_data;
+	}
 
 	return stream;;
 }
 
 xb_wstream_file_t *
 xb_stream_write_open(xb_wstream_t *stream, const char *path,
-		     MY_STAT *mystat __attribute__((unused)),
-		     void *userdata,
-		     xb_stream_write_callback *onwrite)
+		const MY_STAT *mystat __attribute__((unused)), bool rewrite)
 {
 	xb_wstream_file_t	*file;
 	size_t			path_len;
@@ -109,16 +121,7 @@ xb_stream_write_open(xb_wstream_t *stream, const char *path,
 	file->offset = 0;
 	file->chunk_ptr = file->chunk;
 	file->chunk_free = XB_STREAM_MIN_CHUNK_SIZE;
-	if (onwrite) {
-#ifdef _WIN32
-		setmode(fileno(stdout), _O_BINARY);
-#endif
-		file->userdata = userdata;
-		file->write = onwrite;
-	} else {
-		file->userdata = NULL;
-		file->write = xb_stream_default_write_callback;
-	}
+	file->rewrite = rewrite;
 
 	return file;
 }
@@ -202,7 +205,8 @@ xb_stream_write_chunk(xb_wstream_file_t *file, const void *buf, size_t len)
 	memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
 	ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;
 
-	*ptr++ = 0;                              /* Chunk flags */
+	*ptr++ =
+    file->rewrite ? XB_STREAM_FLAG_REWRITE : 0; /* Chunk flags */
 
 	*ptr++ = (uchar) XB_CHUNK_TYPE_PAYLOAD;  /* Chunk type */
 
@@ -227,11 +231,11 @@ xb_stream_write_chunk(xb_wstream_file_t *file, const void *buf, size_t len)
 
 	xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));
 
-	if (file->write(file, file->userdata, tmpbuf, ptr-tmpbuf) == -1)
+	if (stream->write(stream->user_data, tmpbuf, ptr-tmpbuf) == -1)
 		goto err;
 
 
-	if (file->write(file, file->userdata, buf, len) == -1) /* Payload */
+	if (stream->write(stream->user_data, buf, len) == -1) /* Payload */
 		goto err;
 
 	file->offset+= len;
@@ -247,6 +251,38 @@ xb_stream_write_chunk(xb_wstream_file_t *file, const void *buf, size_t len)
 	return 1;
 }
 
+int xb_stream_write_seek_set(xb_wstream_file_t *file, my_off_t offset)
+{
+	/* Chunk magic + flags + chunk type + path_len + path + offset */
+	uchar		tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 + 4 +
+			       FN_REFLEN + 8];
+	int error = 0;
+	xb_wstream_t	*stream = file->stream;
+	uchar		*ptr = tmpbuf;
+	/* Chunk magic */
+	memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
+	ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;
+	*ptr++ = 0;                              /* Chunk flags */
+	*ptr++ = (uchar) XB_CHUNK_TYPE_SEEK;   /* Chunk type */
+	int4store(ptr, file->path_len);          /* Path length */
+	ptr += 4;
+	memcpy(ptr, file->path, file->path_len); /* Path */
+	ptr += file->path_len;
+	int8store(ptr, static_cast<int64_t>(offset));  /* Offset */
+	ptr += 8;
+	if (xb_stream_flush(file))
+		return 1;
+	pthread_mutex_lock(&stream->mutex);
+	if (stream->write(stream->user_data, tmpbuf, ptr-tmpbuf) == -1)
+		error = 1;
+	if (!error)
+		file->offset = offset;
+	pthread_mutex_unlock(&stream->mutex);
+	if (xb_stream_flush(file))
+		return 1;
+	return error;
+}
+
 static
 int
 xb_stream_write_eof(xb_wstream_file_t *file)
@@ -278,7 +314,7 @@ xb_stream_write_eof(xb_wstream_file_t *file)
 
 	xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));
 
-	if (file->write(file, file->userdata, tmpbuf,
+	if (stream->write(stream->user_data, tmpbuf,
 			(ulonglong) (ptr - tmpbuf)) == -1)
 		goto err;
 
@@ -291,3 +327,77 @@ xb_stream_write_eof(xb_wstream_file_t *file)
 
 	return 1;
 }
+
+
+int
+xb_stream_write_remove(xb_wstream_t *stream, const char *path) {
+	/* Chunk magic + flags + chunk type + path_len + path */
+	uchar		tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 + 4 + FN_REFLEN];
+	uchar		*ptr = tmpbuf;
+	/* Chunk magic */
+	memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
+	ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;
+
+	*ptr++ = 0;                              /* Chunk flags */
+
+	*ptr++ = (uchar) XB_CHUNK_TYPE_REMOVE;   /* Chunk type */
+	size_t path_len = strlen(path);
+	int4store(ptr, path_len);               /* Path length */
+	ptr += 4;
+
+	memcpy(ptr, path, path_len);            /* Path */
+	ptr += path_len;
+
+	xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));
+
+	pthread_mutex_lock(&stream->mutex);
+
+	ssize_t result = stream->write(stream->user_data, tmpbuf,
+		(ulonglong) (ptr - tmpbuf));
+
+	pthread_mutex_unlock(&stream->mutex);
+
+	return result < 0;
+
+}
+
+int
+xb_stream_write_rename(
+	xb_wstream_t *stream, const char *old_path, const char *new_path) {
+	/* Chunk magic + flags + chunk type + path_len + path + path_len + path*/
+	uchar		tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 +
+		4 + FN_REFLEN + 4 + FN_REFLEN];
+	uchar		*ptr = tmpbuf;
+	/* Chunk magic */
+	memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
+	ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;
+
+	*ptr++ = 0;                              /* Chunk flags */
+
+	*ptr++ = (uchar) XB_CHUNK_TYPE_RENAME;   /* Chunk type */
+	size_t path_len = strlen(old_path);
+	int4store(ptr, path_len);               /* Path length */
+	ptr += 4;
+
+	memcpy(ptr, old_path, path_len);            /* Path */
+	ptr += path_len;
+
+	path_len = strlen(new_path);
+	int4store(ptr, path_len);               /* Path length */
+	ptr += 4;
+
+	memcpy(ptr, new_path, path_len);            /* Path */
+	ptr += path_len;
+
+	xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));
+
+	pthread_mutex_lock(&stream->mutex);
+
+	ssize_t result = stream->write(stream->user_data, tmpbuf,
+		(ulonglong) (ptr - tmpbuf));
+
+	pthread_mutex_unlock(&stream->mutex);
+
+	return result < 0;
+}
+

extra/mariabackup/xtrabackup.h

@@ -26,6 +26,13 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 #include "xbstream.h"
 #include "fil0fil.h"
 #include <set>
+#include "handler.h"
+
+#include <utility>
+#include <vector>
+#include <tuple>
+#include <functional>
+
 
 #define XB_TOOL_NAME "mariadb-backup"
 #define XB_HISTORY_TABLE "mysql.mariadb_backup_history"
@@ -110,7 +117,7 @@ extern my_bool		xtrabackup_decrypt_decompress;
 extern char		*innobase_data_file_path;
 extern longlong		innobase_page_size;
 
-extern int		xtrabackup_parallel;
+extern uint		xtrabackup_parallel;
 
 extern my_bool		xb_close_files;
 extern const char	*xtrabackup_compress_alg;
@@ -129,7 +136,6 @@ extern my_bool		opt_galera_info;
 extern my_bool		opt_slave_info;
 extern my_bool		opt_no_lock;
 extern my_bool		opt_safe_slave_backup;
-extern my_bool		opt_rsync;
 extern my_bool		opt_force_non_empty_dirs;
 extern my_bool		opt_noversioncheck;
 extern my_bool		opt_no_backup_locks;
@@ -286,15 +292,40 @@ fil_file_readdir_next_file(
 	os_file_stat_t* info);	/*!< in/out: buffer where the
 				info is returned */
 
-#ifndef DBUG_OFF
-#include <fil0fil.h>
-extern void dbug_mariabackup_event(const char *event,
-                            const fil_space_t::name_type key);
+const char *convert_dst(const char *dst);
 
-#define DBUG_MARIABACKUP_EVENT(A, B)                                          \
-  DBUG_EXECUTE_IF("mariabackup_events", dbug_mariabackup_event(A, B);)
-#else
-#define DBUG_MARIABACKUP_EVENT(A, B) /* empty */
-#endif // DBUG_OFF
+std::string get_table_version_from_image(const std::vector<uchar> &frm_image);
+std::pair<bool, legacy_db_type>
+	get_table_engine_from_image(const std::vector<uchar> &frm_image);
+std::string read_table_version_id(File file);
+
+std::string convert_tablename_to_filepath(
+	const char *data_dir_path, const std::string &db, const std::string &table);
+
+std::tuple<std::string, std::string, std::string>
+convert_filepath_to_tablename(const char *filepath);
+
+typedef std::string table_key_t;
+
+inline table_key_t table_key(const std::string &db, const std::string &table) {
+	return std::string(db).append(".").append(table);
+};
+
+inline table_key_t table_key(const char *db, const char *table) {
+	return std::string(db).append(".").append(table);
+};
+
+typedef std::function<void(std::string, std::string, std::string)>
+  post_copy_table_hook_t;
+
+my_bool
+check_if_skip_table(
+/******************/
+	const char*	name);	/*!< in: path to the table */
+
+bool is_log_table(const char *dbname, const char *tablename);
+bool is_stats_table(const char *dbname, const char *tablename);
 
+extern my_bool xtrabackup_copy_back;
+extern my_bool xtrabackup_move_back;
 #endif /* XB_XTRABACKUP_H */

mysql-test/include/aria_log_control_load.inc

+#
+# This file loads aria_log_control file into a user variable @aria_log_control.
+# Set $ARIA_DATADIR before including this file
+#
+
+--disable_query_log
+--copy_file $ARIA_DATADIR/aria_log_control $MYSQLTEST_VARDIR/aria_log_control_tmp
+--chmod 0777 $MYSQLTEST_VARDIR/aria_log_control_tmp
+--eval SET @aria_log_control=(SELECT LOAD_FILE('$MYSQLTEST_VARDIR/aria_log_control_tmp'))
+--remove_file $MYSQLTEST_VARDIR/aria_log_control_tmp
+--enable_query_log

mysql-test/suite/mariabackup/absolute_ibdata_paths.test

@@ -30,7 +30,7 @@ let $_innodb_data_file_path=`select @@innodb_data_file_path`;
 let $_innodb_data_home_dir=`select @@innodb_data_home_dir`;
 
 --disable_result_log
-exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --target-dir=$targetdir;
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --parallel=10 --target-dir=$targetdir;
 --enable_result_log
 exec $XTRABACKUP  --prepare --target-dir=$targetdir;
 

mysql-test/suite/mariabackup/alter_copy_race.result

@@ -4,7 +4,7 @@ INSERT into t1 values(1);
 connect  con2, localhost, root,,;
 connection con2;
 set lock_wait_timeout=1;
-SET debug_sync='copy_data_between_tables_before_reset_backup_lock  SIGNAL go WAIT_FOR after_backup_stage_block_commit';
+SET debug_sync='copy_data_between_tables_before_reset_backup_lock  SIGNAL go WAIT_FOR after_backup_stage_block_ddl';
 SET debug_sync='alter_table_after_temp_table_drop SIGNAL temp_table_dropped';
 SET debug_sync='now WAIT_FOR after_backup_stage_start';ALTER TABLE test.t1 FORCE, algorithm=COPY;|
 connection default;

mysql-test/suite/mariabackup/alter_copy_race.test

@@ -18,7 +18,7 @@ INSERT into t1 values(1);
 connect  con2, localhost, root,,;
 connection con2;
 set lock_wait_timeout=1;
-SET debug_sync='copy_data_between_tables_before_reset_backup_lock  SIGNAL go WAIT_FOR after_backup_stage_block_commit';
+SET debug_sync='copy_data_between_tables_before_reset_backup_lock  SIGNAL go WAIT_FOR after_backup_stage_block_ddl';
 SET debug_sync='alter_table_after_temp_table_drop SIGNAL temp_table_dropped';
 DELIMITER |;
 send  SET debug_sync='now WAIT_FOR after_backup_stage_start';ALTER TABLE test.t1 FORCE, algorithm=COPY;|
@@ -27,7 +27,7 @@ connection default;
 
 # setup mariabackup events
 let after_backup_stage_start=SET debug_sync='now SIGNAL after_backup_stage_start WAIT_FOR go';
-let after_backup_stage_block_commit=SET debug_sync='now SIGNAL after_backup_stage_block_commit';
+let after_backup_stage_block_ddl=SET debug_sync='now SIGNAL after_backup_stage_block_ddl';
 let backup_fix_ddl=SET debug_sync='now WAIT_FOR temp_table_dropped';
 --disable_result_log
 exec  $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --dbug=+d,mariabackup_events;

mysql-test/suite/mariabackup/apply-log-only-incr.test

@@ -19,7 +19,7 @@ dec $n;
 }
 
 --disable_result_log
-exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$basedir;
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --parallel=10 --target-dir=$basedir;
 --enable_result_log
 let $n=100;
 while ($n) {
@@ -36,7 +36,7 @@ disconnect flush_log;
 connection default;
 
 --disable_result_log
-exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --ftwrl-wait-timeout=5 --ftwrl-wait-threshold=300 --ftwrl-wait-query-type=all --target-dir=$incremental_dir --incremental-basedir=$basedir ;
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --parallel=10 --ftwrl-wait-timeout=5 --ftwrl-wait-threshold=300 --ftwrl-wait-query-type=all --target-dir=$incremental_dir --incremental-basedir=$basedir ;
 exec $XTRABACKUP --prepare --verbose --target-dir=$basedir ;
 --enable_result_log
 

mysql-test/suite/mariabackup/apply-log-only.test

@@ -8,7 +8,7 @@ start transaction;
 INSERT INTO t VALUES(1);
 
 --disable_result_log
-exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$basedir;
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --parallel=10 --target-dir=$basedir;
 --enable_result_log
 
 exec $XTRABACKUP --prepare --target-dir=$basedir ;

mysql-test/suite/mariabackup/aria_backup.opt

+--loose-partition --loose-aria-log-file-size=8388608

mysql-test/suite/mariabackup/aria_log.opt

+--loose-aria-log-file-size=8388608

mysql-test/suite/mariabackup/aria_log_dir_path.result

@@ -35,7 +35,6 @@ DROP TABLE t1;
 SET @@global.aria_checkpoint_interval=DEFAULT /*Force checkpoint*/;
 SHOW ENGINE aria logs;
 Type	Name	Status
-Aria	aria_log.00000001	free
 Aria	aria_log.00000002	in use
 # Restarting mariadbd with default parameters
 # restart

mysql-test/suite/mariabackup/aria_log_dir_path.test

@@ -48,7 +48,6 @@ SET @@global.aria_checkpoint_interval=DEFAULT /*Force checkpoint*/;
 --replace_regex /Size +[0-9]+ ; .+aria_log/aria_log/
 SHOW ENGINE aria logs;
 
-
 --echo # mariadb-backup --backup
 --disable_result_log
 --mkdir $targetdir
@@ -61,7 +60,6 @@ SHOW ENGINE aria logs;
 --exec $XTRABACKUP  --prepare --target-dir=$targetdir
 --enable_result_log
 
-
 --echo # shutdown server
 --disable_result_log
 --source include/shutdown_mysqld.inc
@@ -70,12 +68,14 @@ SHOW ENGINE aria logs;
 --echo # remove aria-log-dir-path
 --rmdir $ARIA_LOGDIR_FS
 
+
 --echo # mariadb-backup --copy-back
 --let $mariadb_backup_parameters=--defaults-file=$MYSQLTEST_VARDIR/my.cnf --copy-back --datadir=$datadir --target-dir=$targetdir --parallel=2 --throttle=1 --aria-log-dir-path=$ARIA_LOGDIR_MARIADB
 --replace_result $MYSQL_TEST_DIR MYSQL_TEST_DIR $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
 --exec echo "# with parameters: $mariadb_backup_parameters"
 --exec $XTRABACKUP $mariadb_backup_parameters
 
+
 --echo # starting server
 --let $restart_parameters=$server_parameters
 --source include/start_mysqld.inc
@@ -91,7 +91,7 @@ DROP TABLE t1;
 --echo # Testing aria log files after --copy-back
 SET @@global.aria_checkpoint_interval=DEFAULT /*Force checkpoint*/;
 --file_exists $ARIA_LOGDIR_FS/aria_log_control
---file_exists $ARIA_LOGDIR_FS/aria_log.00000001
+#--file_exists $ARIA_LOGDIR_FS/aria_log.00000001
 --file_exists $ARIA_LOGDIR_FS/aria_log.00000002
 --error 1
 --file_exists $ARIA_LOGDIR_FS/aria_log.00000003

mysql-test/suite/mariabackup/aria_log_dir_path_rel.result

@@ -35,7 +35,6 @@ DROP TABLE t1;
 SET @@global.aria_checkpoint_interval=DEFAULT /*Force checkpoint*/;
 SHOW ENGINE aria logs;
 Type	Name	Status
-Aria	aria_log.00000001	free
 Aria	aria_log.00000002	in use
 # Restarting mariadbd with default parameters
 # restart

mysql-test/suite/mariabackup/aria_log_rotate_during_backup.opt

+--loose-aria-log-file-size=8388608
+--loose-restart-for-aria_log_rotate_during_backup="This is needed to recreate datadir, to have Aria start logs from aria_log.00000001"

mysql-test/suite/mariabackup/aria_log_rotate_during_backup.result

+SHOW VARIABLES LIKE 'aria_log_file_size';
+Variable_name	Value
+aria_log_file_size	8388608
+CREATE PROCEDURE display_aria_log_control(ctrl BLOB)
+BEGIN
+SELECT HEX(REVERSE(SUBSTRING(ctrl, 42, 4))) AS last_logno;
+END;
+$$
+CREATE PROCEDURE populate_t1()
+BEGIN
+FOR id IN 0..9 DO
+INSERT INTO test.t1 (id, txt) VALUES (id, REPEAT(id,1024*1024));
+END FOR;
+END;
+$$
+CREATE TABLE test.t1(id INT, txt LONGTEXT) ENGINE=Aria;
+# MYSQLD_DATADIR/aria_log_control before --backup
+CALL display_aria_log_control(@aria_log_control);
+last_logno
+00000001
+# Running --backup
+# MYSQLD_DATADIR/aria_log_control after --backup
+CALL display_aria_log_control(@aria_log_control);
+last_logno
+00000002
+# targetdir/aria_log_control after --backup
+CALL display_aria_log_control(@aria_log_control);
+last_logno
+00000001
+# Running --prepare
+# targetdir/aria_log_control after --prepare
+CALL display_aria_log_control(@aria_log_control);
+last_logno
+00000002
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart
+# MYSQLD_DATADIR/aria_log_control after --copy-back
+CALL display_aria_log_control(@aria_log_control);
+last_logno
+00000002
+# Checking that after --restore all t1 data is there
+SELECT id, LENGTH(txt) FROM t1 ORDER BY id;
+id	LENGTH(txt)
+0	1048576
+1	1048576
+2	1048576
+3	1048576
+4	1048576
+5	1048576
+6	1048576
+7	1048576
+8	1048576
+9	1048576
+DROP TABLE t1;
+DROP PROCEDURE populate_t1;
+DROP PROCEDURE display_aria_log_control;

mysql-test/suite/mariabackup/aria_log_rotate_during_backup.test

+--source include/have_debug.inc
+--source include/have_aria.inc
+
+SHOW VARIABLES LIKE 'aria_log_file_size';
+
+--let $MYSQLD_DATADIR= `select @@datadir`
+--let $targetdir=$MYSQLTEST_VARDIR/tmp/backup
+mkdir $targetdir;
+
+
+DELIMITER $$;
+CREATE PROCEDURE display_aria_log_control(ctrl BLOB)
+BEGIN
+  SELECT HEX(REVERSE(SUBSTRING(ctrl, 42, 4))) AS last_logno;
+END;
+$$
+DELIMITER ;$$
+
+DELIMITER $$;
+CREATE PROCEDURE populate_t1()
+BEGIN
+  FOR id IN 0..9 DO
+    INSERT INTO test.t1 (id, txt) VALUES (id, REPEAT(id,1024*1024));
+  END FOR;
+END;
+$$
+DELIMITER ;$$
+
+
+CREATE TABLE test.t1(id INT, txt LONGTEXT) ENGINE=Aria;
+
+--echo # MYSQLD_DATADIR/aria_log_control before --backup
+--let ARIA_DATADIR=$MYSQLD_DATADIR
+--source include/aria_log_control_load.inc
+CALL display_aria_log_control(@aria_log_control);
+
+
+--echo # Running --backup
+--let after_scanning_log_files=CALL test.populate_t1
+--disable_result_log
+--exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --parallel=10  --target-dir=$targetdir  --dbug=+d,mariabackup_events 2>&1
+--let after_scanning_log_files=
+--enable_result_log
+
+--echo # MYSQLD_DATADIR/aria_log_control after --backup
+--let ARIA_DATADIR=$MYSQLD_DATADIR
+--source include/aria_log_control_load.inc
+CALL display_aria_log_control(@aria_log_control);
+
+--echo # targetdir/aria_log_control after --backup
+--let ARIA_DATADIR=$targetdir
+--source include/aria_log_control_load.inc
+CALL display_aria_log_control(@aria_log_control);
+
+
+--echo # Running --prepare
+--disable_result_log
+--exec $XTRABACKUP  --prepare --target-dir=$targetdir
+--enable_result_log
+
+--echo # targetdir/aria_log_control after --prepare
+--let ARIA_DATADIR=$targetdir
+--source include/aria_log_control_load.inc
+CALL display_aria_log_control(@aria_log_control);
+
+
+--disable_result_log
+--source include/restart_and_restore.inc
+--enable_result_log
+
+--echo # MYSQLD_DATADIR/aria_log_control after --copy-back
+--let ARIA_DATADIR=$MYSQLD_DATADIR
+--source include/aria_log_control_load.inc
+CALL display_aria_log_control(@aria_log_control);
+
+--echo # Checking that after --restore all t1 data is there
+SELECT id, LENGTH(txt) FROM t1 ORDER BY id;
+DROP TABLE t1;
+rmdir $targetdir;
+
+DROP PROCEDURE populate_t1;
+DROP PROCEDURE display_aria_log_control;

mysql-test/suite/mariabackup/auth_plugin_win.test

@@ -22,7 +22,7 @@ eval GRANT ALL PRIVILEGES ON *.* to '$USERNAME';
 
 let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
 --disable_result_log
-exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf -u $USERNAME --backup --protocol=pipe --target-dir=$targetdir;
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf -u $USERNAME --backup --parallel=10 --protocol=pipe --target-dir=$targetdir;
 --enable_result_log
 --replace_result $USERNAME USERNAME
 eval DROP USER '$USERNAME';