MDEV-16791 mariabackup : Support DDL commands during backup

extra/mariabackup/backup_copy.cc

@@ -1388,6 +1388,30 @@ backup_files(const char *from, bool prep_mode)
 	return(ret);
 }
 
+void backup_fix_ddl(void);
+
+#define LSN_PREFIX_IN_SHOW_STATUS  "\nLog sequence number "
+static lsn_t get_current_lsn(MYSQL *connection) {
+	MYSQL_RES *res = xb_mysql_query(connection, "SHOW ENGINE INNODB STATUS", true, false);
+	if (!res)
+		return 0;
+	MYSQL_ROW row = mysql_fetch_row(res);
+	DBUG_ASSERT(row);
+	if (row) {
+		const char *p = strstr(row[2],LSN_PREFIX_IN_SHOW_STATUS);
+		DBUG_ASSERT(p);
+		if (p)
+		{
+			p += sizeof(LSN_PREFIX_IN_SHOW_STATUS) - 1;
+			return (lsn_t)strtoll(p, NULL, 10);
+		}
+	}
+	mysql_free_result(res);
+	return 0;
+}
+
+lsn_t server_lsn_after_lock;
+extern void backup_wait_for_lsn(lsn_t lsn);
 /** Start --backup */
 bool backup_start()
 {
@@ -1407,6 +1431,7 @@ bool backup_start()
 		if (!lock_tables(mysql_connection)) {
 			return(false);
 		}
+		server_lsn_after_lock = get_current_lsn(mysql_connection);
 	}
 
 	if (!backup_files(fil_path_to_mysql_datadir, false)) {
@@ -1421,6 +1446,10 @@ bool backup_start()
 		rocksdb_create_checkpoint();
 	}
 
+	msg_ts("Waiting for log copy thread to read lsn %llu\n", (ulonglong)server_lsn_after_lock);
+	backup_wait_for_lsn(server_lsn_after_lock);
+	backup_fix_ddl();
+
 	// 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
 	// DML to non-transaction tables can occur.
@@ -2230,6 +2259,7 @@ static void rocksdb_lock_checkpoint()
 		msg_ts("Could not obtain rocksdb checkpont lock\n");
 		exit(EXIT_FAILURE);
 	}
+	mysql_free_result(res);
 }
 
 static void rocksdb_unlock_checkpoint()

extra/mariabackup/backup_mysql.cc

@@ -1794,7 +1794,12 @@ mdl_lock_table(ulint space_id)
     std::ostringstream lock_query;
     lock_query << "SELECT 1 FROM " << full_table_name  << " LIMIT 0";
     msg_ts("Locking MDL for %s\n", full_table_name.c_str());
-    xb_mysql_query(mdl_con, lock_query.str().c_str(), false, true);
+    if (mysql_query(mdl_con, lock_query.str().c_str())) {
+      msg_ts("Warning : locking MDL failed for space id %zu, name %s\n", space_id, full_table_name.c_str());
+    } else {
+      MYSQL_RES *r = mysql_store_result(mdl_con);
+      mysql_free_result(r);
+    }
   }
 
   pthread_mutex_unlock(&mdl_lock_con_mutex);

extra/mariabackup/datasink.c

@@ -108,6 +108,9 @@ Write to a datasink file.
 int
 ds_write(ds_file_t *file, const void *buf, size_t len)
 {
+	if (len == 0) {
+		return 0;
+	}
 	return file->datasink->write(file, (const uchar *)buf, len);
 }
 

extra/mariabackup/fil_cur.cc

@@ -131,14 +131,15 @@ Open a source file cursor and initialize the associated read filter.
 be skipped and XB_FIL_CUR_ERROR on error. */
 xb_fil_cur_result_t
 xb_fil_cur_open(
-/*============*/
+	/*============*/
 	xb_fil_cur_t*	cursor,		/*!< out: source file cursor */
 	xb_read_filt_t*	read_filter,	/*!< in/out: the read filter */
 	fil_node_t*	node,		/*!< in: source tablespace node */
-	uint		thread_n)	/*!< thread number for diagnostics */
+	uint		thread_n,	/*!< thread number for diagnostics */
+	ulonglong max_file_size)
 {
 	bool	success;
-
+	int err;
 	/* Initialize these first so xb_fil_cur_close() handles them correctly
 	in case of error */
 	cursor->orig_buf = NULL;
@@ -173,7 +174,7 @@ xb_fil_cur_open(
 			    "tablespace %s\n",
 			    thread_n, cursor->abs_path);
 
-			return(XB_FIL_CUR_ERROR);
+			return(XB_FIL_CUR_SKIP);
 		}
 		mutex_enter(&fil_system->mutex);
 
@@ -194,14 +195,31 @@ xb_fil_cur_open(
 
 	cursor->node = node;
 	cursor->file = node->handle;
-
-	if (stat(cursor->abs_path, &cursor->statinfo)) {
-		msg("[%02u] mariabackup: error: cannot stat %s\n",
+#ifdef _WIN32
+    HANDLE hDup;
+    DuplicateHandle(GetCurrentProcess(),cursor->file.m_file,
+		GetCurrentProcess(), &hDup, 0, FALSE, DUPLICATE_SAME_ACCESS);
+	int filenr = _open_osfhandle((intptr_t)hDup, 0);
+	if (filenr < 0) {
+		err = EINVAL;
+	}
+	else {
+		err = _fstat64(filenr, &cursor->statinfo);
+		close(filenr);
+	}
+#else
+	err = fstat(cursor->file.m_file, &cursor->statinfo);
+#endif
+	if (max_file_size < (ulonglong)cursor->statinfo.st_size) {
+		cursor->statinfo.st_size = (ulonglong)max_file_size;
+	}
+	if (err) {
+		msg("[%02u] mariabackup: error: cannot fstat %s\n",
 		    thread_n, cursor->abs_path);
 
 		xb_fil_cur_close(cursor);
 
-		return(XB_FIL_CUR_ERROR);
+		return(XB_FIL_CUR_SKIP);
 	}
 
 	if (srv_file_flush_method == SRV_O_DIRECT
@@ -374,7 +392,9 @@ xb_fil_cur_close(
 /*=============*/
 	xb_fil_cur_t *cursor)	/*!< in/out: source file cursor */
 {
+	if (cursor->read_filter) {
 		cursor->read_filter->deinit(&cursor->read_filter_ctxt);
+	}
 
 	free(cursor->orig_buf);
 

extra/mariabackup/fil_cur.h

@@ -57,7 +57,7 @@ struct xb_fil_cur_t {
 	ulint		space_size;	/*!< space size in pages */
 
 	/** TODO: remove this default constructor */
-	xb_fil_cur_t() : page_size(0), read_filter_ctxt() {}
+	xb_fil_cur_t() : page_size(0), read_filter(0), read_filter_ctxt() {}
 
 	/** @return whether this is not a file-per-table tablespace */
 	bool is_system() const
@@ -86,7 +86,8 @@ xb_fil_cur_open(
 	xb_fil_cur_t*	cursor,		/*!< out: source file cursor */
 	xb_read_filt_t*	read_filter,	/*!< in/out: the read filter */
 	fil_node_t*	node,		/*!< in: source tablespace node */
-	uint		thread_n);	/*!< thread number for diagnostics */
+	uint		thread_n,	/*!< thread number for diagnostics */
+	ulonglong max_file_size = ULLONG_MAX);
 
 /************************************************************************
 Reads and verifies the next block of pages from the source

extra/mariabackup/xtrabackup.cc

@@ -343,6 +343,25 @@ const char *opt_history = NULL;
 char mariabackup_exe[FN_REFLEN];
 char orig_argv1[FN_REFLEN];
 
+pthread_mutex_t backup_mutex;
+pthread_cond_t  scanned_lsn_cond;
+
+typedef std::map<space_id_t,std::string> space_id_to_name_t;
+
+struct ddl_tracker_t {
+	/** Tablspaces with their ID and name, as they were copied to backup.*/
+	space_id_to_name_t tables_in_backup;
+	/** Tablespaces for that optimized DDL without redo log was found.*/
+	std::set<space_id_t> optimized_ddl;
+	/** Drop operations found in redo log. */
+	std::set<space_id_t> drops;
+	/* For DDL operation found in redo log,  */
+	space_id_to_name_t id_to_name;
+};
+const space_id_t REMOVED_SPACE_ID = ULINT_MAX;
+static ddl_tracker_t ddl_tracker;
+
+
 /* Whether xtrabackup_binlog_info should be created on recovery */
 static bool recover_binlog_info;
 
@@ -537,49 +556,79 @@ void mdl_lock_all()
 		mdl_lock_table(node->space->id);
 	}
 	datafiles_iter_free(it);
-
-	DBUG_EXECUTE_IF("check_mdl_lock_works",
-		dbug_alter_thread_done =
-		  dbug_start_query_thread("ALTER TABLE test.t ADD COLUMN mdl_lock_column int",
-			 "Waiting for table metadata lock",1, ER_QUERY_INTERRUPTED););
 }
 
-/** Check if the space id belongs to the table which name should
-be skipped based on the --tables, --tables-file and --table-exclude
-options.
-@param[in]	space_id	space id to check
-@return true if the space id belongs to skip table/database list. */
-static bool backup_includes(space_id_t space_id)
-{
-	datafiles_iter_t *it = datafiles_iter_new(fil_system);
-	if (!it)
-		return true;
 
-	while (fil_node_t *node = datafiles_iter_next(it)){
-		if (space_id == 0
-		    || (node->space->id == space_id
-			&& !check_if_skip_table(node->space->name))) {
-
-			msg("mariabackup: Unsupported redo log detected "
-			"and it belongs to %s\n",
-			space_id ? node->name: "the InnoDB system tablespace");
+// Convert non-null terminated filename to space name
+std::string filename_to_spacename(const byte *filename, size_t len)
+{
+	// null- terminate filename
+	char *f = (char *)malloc(len + 1);
+	ut_a(f);
+	memcpy(f, filename, len);
+	f[len] = 0;
+	for (size_t i = 0; i < len; i++)
+		if (f[i] == '\\')
+			f[i] = '/';
+	char *p = strrchr(f, '.');
+	ut_a(p);
+	*p = 0;
+	char *table = strrchr(f, '/');
+	ut_a(table);
+	*table = 0;
+	char *db = strrchr(f, '/');
+	ut_a(db);
+	*table = '/';
+	return std::string(db+1);
+}
 
-			msg("mariabackup: ALTER TABLE or OPTIMIZE TABLE "
-			"was being executed during the backup.\n");
+/** Report an operation to create, delete, or rename a file during backup.
+@param[in]	space_id	tablespace identifier
+@param[in]	flags		tablespace flags (NULL if not create)
+@param[in]	name		file name (not NUL-terminated)
+@param[in]	len		length of name, in bytes
+@param[in]	new_name	new file name (NULL if not rename)
+@param[in]	new_len		length of new_name, in bytes (0 if NULL) */
+void backup_file_op(ulint space_id, const byte* flags,
+	const byte* name, ulint len,
+	const byte* new_name, ulint new_len)
+{
 
-			if (!opt_lock_ddl_per_table) {
-				msg("mariabackup: Use --lock-ddl-per-table "
-				"parameter to lock all the table before "
-				"backup operation.\n");
+	ut_ad(!flags || !new_name);
+	ut_ad(name);
+	ut_ad(len);
+	ut_ad(!new_name == !new_len);
+	pthread_mutex_lock(&backup_mutex);
+
+	if (flags) {
+		ddl_tracker.id_to_name[space_id] = filename_to_spacename(name, len);
+		msg("DDL tracking :  create %zu \"%.*s\": %x\n",
+			space_id, int(len), name, mach_read_from_4(flags));
+	}
+	else if (new_name) {
+		ddl_tracker.id_to_name[space_id] = filename_to_spacename(new_name, new_len);
+		msg("DDL tracking : rename %zu \"%.*s\",\"%.*s\"\n",
+			space_id, int(len), name, int(new_len), new_name);
+	} else {
+		ddl_tracker.drops.insert(space_id);
+		msg("DDL tracking : delete %zu \"%.*s\"\n", space_id, int(len), name);
 	}
+	pthread_mutex_unlock(&backup_mutex);
+}
 
-			datafiles_iter_free(it);
-			return false;
-		}
-	}
 
-	datafiles_iter_free(it);
-	return true;
+/** Callback whenever MLOG_INDEX_LOAD happens.
+@param[in]	space_id	space id to check
+@return false */
+void backup_optimized_ddl_op(ulint space_id)
+{
+	// TODO : handle incremental
+	if (xtrabackup_incremental)
+		return;
+
+	pthread_mutex_lock(&backup_mutex);
+	ddl_tracker.optimized_ddl.insert(space_id);
+	pthread_mutex_unlock(&backup_mutex);
 }
 
 /* ======== Date copying thread context ======== */
@@ -2336,7 +2385,7 @@ xb_get_copy_action(const char *dflt)
 
 static
 my_bool
-xtrabackup_copy_datafile(fil_node_t* node, uint thread_n)
+xtrabackup_copy_datafile(fil_node_t* node, uint thread_n, const char *dest_name=0, ulonglong max_size=ULLONG_MAX)
 {
 	char			 dst_name[FN_REFLEN];
 	ds_file_t		*dstfile = NULL;
@@ -2366,20 +2415,31 @@ xtrabackup_copy_datafile(fil_node_t* node, uint thread_n)
 		return(FALSE);
 	}
 
+	bool was_dropped;
+	pthread_mutex_lock(&backup_mutex);
+	was_dropped = (ddl_tracker.drops.find(node->space->id) != ddl_tracker.drops.end());
+	pthread_mutex_unlock(&backup_mutex);
+	if (was_dropped) {
+		fil_space_close(node->space->name);
+		goto skip;
+	}
+
 	if (!changed_page_bitmap) {
 		read_filter = &rf_pass_through;
 	}
 	else {
 		read_filter = &rf_bitmap;
 	}
-	res = xb_fil_cur_open(&cursor, read_filter, node, thread_n);
+
+	res = xb_fil_cur_open(&cursor, read_filter, node, thread_n,max_size);
 	if (res == XB_FIL_CUR_SKIP) {
 		goto skip;
 	} else if (res == XB_FIL_CUR_ERROR) {
 		goto error;
 	}
 
-	strncpy(dst_name, cursor.rel_path, sizeof(dst_name));
+	strncpy(dst_name, (dest_name)?dest_name : cursor.rel_path, sizeof(dst_name));
+
 
 	/* Setup the page write filter */
 	if (xtrabackup_incremental) {
@@ -2431,6 +2491,10 @@ xtrabackup_copy_datafile(fil_node_t* node, uint thread_n)
 		goto error;
 	}
 
+	pthread_mutex_lock(&backup_mutex);
+	ddl_tracker.tables_in_backup[node->space->id] = node_name;
+	pthread_mutex_unlock(&backup_mutex);
+
 	/* close */
 	msg_ts("[%02u]        ...done\n", thread_n);
 	xb_fil_cur_close(&cursor);
@@ -2595,7 +2659,7 @@ static bool xtrabackup_copy_logfile(bool last = false)
 		if (!start_lsn) {
 			msg("mariabackup: Error: xtrabackup_copy_logfile()"
 			    " failed.\n");
-			return(true);
+			exit(EXIT_FAILURE);
 		}
 	} while (start_lsn == end_lsn);
 
@@ -2604,12 +2668,30 @@ static bool xtrabackup_copy_logfile(bool last = false)
 	msg_ts(">> log scanned up to (" LSN_PF ")\n", start_lsn);
 
 	/* update global variable*/
+	pthread_mutex_lock(&backup_mutex);
 	log_copy_scanned_lsn = start_lsn;
+	pthread_cond_broadcast(&scanned_lsn_cond);
+	pthread_mutex_unlock(&backup_mutex);
 
 	debug_sync_point("xtrabackup_copy_logfile_pause");
 	return(false);
 }
 
+/**
+Wait until redo log copying thread processes given lsn
+*/
+void backup_wait_for_lsn(lsn_t lsn) {
+	bool completed = false;
+	pthread_mutex_lock(&backup_mutex);
+	do {
+		pthread_cond_wait(&scanned_lsn_cond, &backup_mutex);
+		completed = log_copy_scanned_lsn >= lsn;
+	} while (!completed);
+	pthread_mutex_unlock(&backup_mutex);
+}
+
+extern lsn_t server_lsn_after_lock;
+
 static os_thread_ret_t log_copying_thread(void*)
 {
 	/*
@@ -2666,6 +2748,42 @@ static os_thread_ret_t io_watching_thread(void*)
 	return(0);
 }
 
+#ifndef DBUG_OFF
+/*
+In debug mode,  execute SQL statement that was passed via environment.
+To use this facility, you need to
+
+1. Add code DBUG_EXECUTE_MARIABACKUP_EVENT("my_event_name", key););
+  to the code. key is usually a table name
+2. Set environment variable my_event_name_$key SQL statement you want to execute
+   when event occurs, in DBUG_EXECUTE_IF from above.
+   In mtr , you can set environment via 'let' statement (do not use $ as the first char
+   for the variable)
+3. start mariabackup with --dbug=+d,debug_mariabackup_events
+*/
+static void dbug_mariabackup_event(const char *event,const char *key)
+{
+	char envvar[FN_REFLEN];
+	if (key) {
+		snprintf(envvar, sizeof(envvar), "%s_%s", event, key);
+		char *slash = strchr(envvar, '/');
+		if (slash)
+			*slash = '_';
+	} else {
+		strncpy(envvar, event, sizeof(envvar));
+	}
+	char *sql = getenv(envvar);
+	if (sql) {
+		msg("dbug_mariabackup_event : executing '%s'\n", sql);
+		xb_mysql_query(mysql_connection, sql, false, true);
+	}
+
+}
+#define DBUG_MARIABACKUP_EVENT(A, B) DBUG_EXECUTE_IF("mariabackup_events", dbug_mariabackup_event(A,B););
+#else
+#define DBUG_MARIABACKUP_EVENT(A,B)
+#endif
+
 /**************************************************************************
 Datafiles copying thread.*/
 static
@@ -2688,12 +2806,18 @@ data_copy_thread_func(
 
 	while ((node = datafiles_iter_next(ctxt->it)) != NULL) {
 
+		DBUG_MARIABACKUP_EVENT("before_copy", node->space->name);
+
+
 		/* copy the datafile */
 		if(xtrabackup_copy_datafile(node, num)) {
 			msg("[%02u] mariabackup: Error: "
 			    "failed to copy datafile.\n", num);
 			exit(EXIT_FAILURE);
 		}
+
+		DBUG_MARIABACKUP_EVENT("after_copy", node->space->name);
+
 	}
 
 	pthread_mutex_lock(&ctxt->count_mutex);
@@ -2866,6 +2990,7 @@ xb_load_single_table_tablespace(
 	Datafile *file = xb_new_datafile(name, is_remote);
 
 	if (file->open_read_only(true) != DB_SUCCESS) {
+		msg("Can't open datafile %s\n", name);
 		ut_free(name);
 		exit(EXIT_FAILURE);
 	}
@@ -3182,7 +3307,7 @@ xb_load_tablespaces()
 	}
 
 	debug_sync_point("xtrabackup_load_tablespaces_pause");
-
+	DBUG_MARIABACKUP_EVENT("after_load_tablespaces", 0);
 	return(DB_SUCCESS);
 }
 
@@ -3791,6 +3916,8 @@ xtrabackup_backup_func()
 	uint			 count;
 	pthread_mutex_t		 count_mutex;
 	data_thread_ctxt_t 	*data_threads;
+	pthread_mutex_init(&backup_mutex, NULL);
+	pthread_cond_init(&scanned_lsn_cond, NULL);
 
 #ifdef USE_POSIX_FADVISE
 	msg("mariabackup: uses posix_fadvise().\n");
@@ -3817,6 +3944,7 @@ xtrabackup_backup_func()
 	srv_read_only_mode = TRUE;
 
 	srv_operation = SRV_OPERATION_BACKUP;
+	log_file_op = backup_file_op;
 	metadata_to_lsn = 0;
 
 	if (xb_close_files)
@@ -3830,6 +3958,7 @@ xtrabackup_backup_func()
 fail:
 		metadata_to_lsn = log_copying_running;
 		stop_backup_threads();
+		log_file_op = NULL;
 		if (dst_log_file) {
 			ds_close(dst_log_file);
 			dst_log_file = NULL;
@@ -4116,6 +4245,7 @@ xtrabackup_backup_func()
 		goto fail_before_log_copying_thread_start;
 
 	log_copying_stop = os_event_create(0);
+	log_optimized_ddl_op = backup_optimized_ddl_op;
 	os_thread_create(log_copying_thread, NULL, &log_copying_thread_id);
 
 	/* FLUSH CHANGED_PAGE_BITMAPS call */
@@ -4147,6 +4277,11 @@ xtrabackup_backup_func()
 
 	if (opt_lock_ddl_per_table) {
 		mdl_lock_all();
+
+		DBUG_EXECUTE_IF("check_mdl_lock_works",
+			dbug_alter_thread_done =
+			dbug_start_query_thread("ALTER TABLE test.t ADD COLUMN mdl_lock_column int",
+				"Waiting for table metadata lock", 1, ER_QUERY_INTERRUPTED););
 	}
 
 	it = datafiles_iter_new(fil_system);
@@ -4184,10 +4319,6 @@ xtrabackup_backup_func()
 	pthread_mutex_destroy(&count_mutex);
 	free(data_threads);
 	datafiles_iter_free(it);
-
-	if (changed_page_bitmap) {
-		xb_page_bitmap_deinit(changed_page_bitmap);
-	}
 	}
 
 	bool ok = backup_start();
@@ -4211,6 +4342,9 @@ xtrabackup_backup_func()
 		goto fail;
 	}
 
+	if (changed_page_bitmap) {
+		xb_page_bitmap_deinit(changed_page_bitmap);
+	}
 	xtrabackup_destroy_datasinks();
 
 	msg("mariabackup: Redo log (from LSN " LSN_PF " to " LSN_PF
@@ -4228,9 +4362,178 @@ xtrabackup_backup_func()
 	}
 
 	innodb_shutdown();
+	log_file_op = NULL;
+	pthread_mutex_destroy(&backup_mutex);
+	pthread_cond_destroy(&scanned_lsn_cond);
 	return(true);
 }
 
+
+/**
+This function handles DDL changes at the end of backup, under protection of
+FTWRL.  This ensures consistent backup in presence of DDL.
+
+- New tables, that were created during backup, are now copied into backup.
+  Also, tablespaces with optimized (no redo loggin DDL) are re-copied into 
+  backup. This tablespaces will get the extension ".new" in the backup
+
+- Tables that were renamed during backup, are marked as renamed
+  For these, file <old_name>.ren will be created.
+  The content of the file is the new tablespace name.
+
+- Tables that were deleted during backup, are marked as deleted
+  For these , an empty file <name>.del will be created
+
+  It is the responsibility of the prepare phase to deal with .new, .ren, and .del
+  files.
+*/
+void backup_fix_ddl(void)
+{
+	std::set<std::string> new_tables;
+	std::set<std::string> dropped_tables;
+	std::map<std::string, std::string> renamed_tables;
+
+	for (space_id_to_name_t::iterator iter = ddl_tracker.tables_in_backup.begin();
+		iter != ddl_tracker.tables_in_backup.end();
+		iter++) {
+
+		const std::string name = iter->second;
+		ulint id = iter->first;
+
+		if (ddl_tracker.drops.find(id) != ddl_tracker.drops.end()) {
+			dropped_tables.insert(name);
+			continue;
+		}
+
+		bool has_optimized_ddl =
+			ddl_tracker.optimized_ddl.find(id) != ddl_tracker.optimized_ddl.end();
+
+		if (ddl_tracker.id_to_name.find(id) == ddl_tracker.id_to_name.end()) {
+			if (has_optimized_ddl) {
+				new_tables.insert(name);
+			}
+			continue;
+		}
+
+		/* tablespace was affected by DDL. */
+		const std::string new_name = ddl_tracker.id_to_name[id];
+		if (new_name != name) {
+			if (has_optimized_ddl) {
+				/* table was renamed, but we need a full copy
+				of it because of optimized DDL. We emulate a drop/create.*/
+				dropped_tables.insert(name);
+				new_tables.insert(new_name);
+			} else {
+				/* Renamed, and no optimized DDL*/
+				renamed_tables[name] = 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);
+		}
+	}
+
+	/* Find tables that were created during backup (and not removed).*/
+	for(space_id_to_name_t::iterator iter = ddl_tracker.id_to_name.begin();
+		iter != ddl_tracker.id_to_name.end();
+		iter++) {
+
+		ulint id = iter->first;
+		std::string name = iter->second;
+
+		if (ddl_tracker.tables_in_backup.find(id) != ddl_tracker.tables_in_backup.end()) {
+			/* already processed above */
+			continue;
+		}
+
+		if (ddl_tracker.drops.find(id) == ddl_tracker.drops.end()) {
+			dropped_tables.erase(name);
+			new_tables.insert(name);
+		}
+	}
+
+	// Mark tablespaces for rename
+	for (std::map<std::string, std::string>::iterator iter = renamed_tables.begin();
+		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());
+	}
+
+	// Mark tablespaces for drop
+	for (std::set<std::string>::iterator iter = dropped_tables.begin();
+		iter != dropped_tables.end();
+		iter++) {
+		const std::string name(*iter);
+		backup_file_printf((name + ".del").c_str(), "%s", "");
+	}
+
+	//  Load and copy new tables.
+	//  Close all datanodes first, reload only new tables.
+	std::vector<fil_node_t *> all_nodes;
+	datafiles_iter_t *it = datafiles_iter_new(fil_system);
+	if (!it)
+		return;
+	while (fil_node_t *node = datafiles_iter_next(it)) {
+		all_nodes.push_back(node);
+	}
+	for (size_t i = 0; i < all_nodes.size(); i++) {
+		fil_node_t *n = all_nodes[i];
+		if (n->space->id == 0)
+			continue;
+		fil_space_close(n->space->name);
+		fil_space_free(n->space->id, false);
+	}
+
+
+	for (std::set<std::string>::iterator iter = new_tables.begin();
+		iter != new_tables.end(); iter++) {
+		const char *space_name = iter->c_str();
+		if (check_if_skip_table(space_name))
+			continue;
+		std::string name(*iter);
+		bool is_remote = access((name + ".ibd").c_str(), R_OK) != 0;
+		const char *extension = is_remote ? ".isl" : ".ibd";
+		name.append(extension);
+		char buf[FN_REFLEN];
+		strncpy(buf, name.c_str(), sizeof(buf));
+		const char *dbname = buf;
+		char *p = strchr(buf, '/');
+		if (p == 0) {
+			msg("Unexpected tablespace %s filename %s\n", space_name, name.c_str());
+			ut_a(0);
+		}
+		ut_a(p);
+		*p = 0;
+		const char *tablename = p + 1;
+		xb_load_single_table_tablespace(dbname, tablename, is_remote);
+	}
+
+	it = datafiles_iter_new(fil_system);
+	if (!it)
+		return;
+
+	while (fil_node_t *node = datafiles_iter_next(it)) {
+		fil_space_t * space = node->space;
+		if (!fil_is_user_tablespace_id(space->id))
+			continue;
+		std::string dest_name(node->space->name);
+		dest_name.append(".new");
+#if 0
+		bool do_full_copy = ddl_tracker.optimized_ddl.find(n->space->id) != ddl_tracker.optimized_ddl.end();
+		if (do_full_copy) {
+			msg(
+				"Performing a full copy of the tablespace %s, because optimized (without redo logging) DDL operation"
+				"ran during backup. You can use set innodb_log_optimize_ddl=OFF to improve backup performance"
+				"in the future.\n",
+				n->space->name);
+		}
+#endif
+		xtrabackup_copy_datafile(node, 0, dest_name.c_str()/*, do_full_copy ? ULONGLONG_MAX:UNIV_PAGE_SIZE */);
+	}
+}
+
 /* ================= prepare ================= */
 
 /***********************************************************************
@@ -4739,6 +5042,27 @@ xtrabackup_apply_delta(
 	return FALSE;
 }
 
+
+std::string change_extension(std::string filename, std::string new_ext) {
+	DBUG_ASSERT(new_ext.size() == 3);
+	std::string new_name(filename);
+	new_name.resize(new_name.size() - new_ext.size());
+	new_name.append(new_ext);
+	return new_name;
+}
+
+
+static void rename_file(const char *from,const char *to) {
+	msg("Renaming %s to %s\n", from, to);
+	if (my_rename(from, to, MY_WME)) {
+		msg("Cannot rename %s to %s errno %d", from, to, errno);
+		exit(EXIT_FAILURE);
+	}
+}
+
+static void rename_file(const std::string& from, const std::string &to) {
+	rename_file(from.c_str(), to.c_str());
+}
 /************************************************************************
 Callback to handle datadir entry. Function of this type will be called
 for each entry which matches the mask by xb_process_datadir.
@@ -4750,6 +5074,37 @@ typedef ibool (*handle_datadir_entry_func_t)(
 	const char*	file_name,		/*!<in: file name with suffix */
 	void*		arg);			/*!<in: caller-provided data */
 
+/** Rename, and replace destination file, if exists */
+static void rename_force(const char *from, const char *to) {
+	if (access(to, R_OK) == 0) {
+		msg("Removing %s\n", to);
+		if (my_delete(to, MYF(MY_WME))) {
+			msg("Can't remove %s, errno %d", to, errno);
+			exit(EXIT_FAILURE);
+		}
+	}
+	rename_file(from,to);
+}
+
+/* During prepare phase, rename ".new" files , that were created in backup_fix_ddl(),
+  to ".ibd".*/
+static ibool prepare_handle_new_files(
+	const char*	data_home_dir,		/*!<in: path to datadir */
+	const char*	db_name,		/*!<in: database name */
+	const char*	file_name,		/*!<in: file name with suffix */
+	void *)
+{
+
+	std::string src_path = std::string(data_home_dir) + '/' + std::string(db_name) + '/' + file_name;
+	std::string dest_path = src_path;
+
+	size_t index = dest_path.find(".new");
+	DBUG_ASSERT(index != std::string::npos);
+	dest_path.replace(index, 4, ".ibd");
+	rename_force(src_path.c_str(),dest_path.c_str());
+	return TRUE;
+}
+
 /************************************************************************
 Callback to handle datadir entry. Deletes entry if it has no matching
 fil_space in fil_system directory.
@@ -4955,6 +5310,103 @@ store_binlog_info(const char* filename, const char* name, ulonglong pos)
 	return(true);
 }
 
+/** Check if file exists*/
+static bool file_exists(std::string name)
+{
+	return access(name.c_str(), R_OK) == 0 ;
+}
+
+/** Read file content into STL string */
+static std::string read_file_as_string(const std::string file) {
+	char content[FN_REFLEN];
+	FILE *f = fopen(file.c_str(), "r");
+	if (!f) {
+		msg("Can not open %s\n", file.c_str());
+	}
+	size_t len = fread(content, 1, FN_REFLEN, f);
+	fclose(f);
+	return std::string(content, len);
+}
+
+/** Delete file- Provide verbose diagnostics and exit, if operation fails. */
+static void delete_file(const std::string& file, bool if_exists = false) {
+	if (if_exists && !file_exists(file))
+		return;
+	if (my_delete(file.c_str(), MYF(MY_WME))) {
+		msg("Can't remove %s, errno %d", file.c_str(), errno);
+		exit(EXIT_FAILURE);
+	}
+}
+
+/**
+Rename tablespace during prepare.
+Backup in its end phase may generate some .ren files, recording
+tablespaces that should be renamed in --prepare.
+*/
+static void rename_table_in_prepare(const std::string &datadir, const std::string& from , const std::string& to,
+	const char *extension=0) {
+	if (!extension) {
+		static const char *extensions_nonincremental[] = { ".ibd", 0 };
+		static const char *extensions_incremental[] = { ".ibd.delta", ".ibd.meta", 0 };
+		const char **extensions = xtrabackup_incremental_dir ?
+			extensions_incremental : extensions_nonincremental;
+		for (size_t i = 0; extensions[i]; i++) {
+			rename_table_in_prepare(datadir, from, to, extensions[i]);
+		}
+		return;
+	}
+	std::string src = std::string(datadir) + "/" + from + extension;
+	std::string dest = std::string(datadir) + "/" + to + extension;
+	std::string ren2, tmp;
+	if (file_exists(dest)) {
+		ren2= std::string(datadir) + "/" + to + ".ren";
+		if (!file_exists(ren2)) {
+			msg("ERROR : File %s was not found, but expected during rename processing\n", ren2.c_str());
+			ut_a(0);
+		}
+		tmp = to + "#";
+		rename_table_in_prepare(datadir, to, tmp);
+	}
+	rename_file(src, dest);
+	if (ren2.size()) {
+		// Make sure the temp. renamed file is processed.
+		std::string to2 = read_file_as_string(ren2);
+		rename_table_in_prepare(datadir, tmp, to2);
+		delete_file(ren2);
+	}
+}
+
+static ibool prepare_handle_ren_files(const char *datadir, const char *db, const char *filename, void *) {
+
+	std::string ren_file = std::string(datadir) + "/" + db + "/" + filename;
+	if (!file_exists(ren_file))
+		return TRUE;
+
+	std::string to = read_file_as_string(ren_file);
+	std::string source_space_name = std::string(db) + "/" + filename;
+	source_space_name.resize(source_space_name.size() - 4); // remove extension
+
+	rename_table_in_prepare(datadir, source_space_name.c_str(), to.c_str());
+	delete_file(ren_file);
+	return TRUE;
+}
+
+/* Remove tablespaces during backup, based on */
+static ibool prepare_handle_del_files(const char *datadir, const char *db, const char *filename, void *) {
+	std::string del_file = std::string(datadir) + "/" + db + "/" + filename;
+	std::string path(del_file);
+	path.resize(path.size() - 4); // remove extension;
+	if (xtrabackup_incremental) {
+		delete_file(path + ".ibd.delta", true);
+		delete_file(path + ".ibd.meta", true);
+	}
+	else {
+		delete_file(path + ".ibd", true);
+	}
+	delete_file(del_file);
+	return TRUE;
+}
+
 /** Implement --prepare
 @return	whether the operation succeeded */
 static bool
@@ -4972,6 +5424,21 @@ xtrabackup_prepare_func(char** argv)
 	}
 	msg("mariabackup: cd to %s\n", xtrabackup_real_target_dir);
 
+	fil_path_to_mysql_datadir = ".";
+
+	if (xtrabackup_incremental_dir) {
+		xb_process_datadir(xtrabackup_incremental_dir, ".new.meta", prepare_handle_new_files);
+		xb_process_datadir(xtrabackup_incremental_dir, ".new.delta", prepare_handle_new_files);
+	}
+	else {
+		xb_process_datadir(".", ".new", prepare_handle_new_files);
+	}
+	xb_process_datadir(xtrabackup_incremental_dir? xtrabackup_incremental_dir:".",
+		".ren", prepare_handle_ren_files);
+	xb_process_datadir(xtrabackup_incremental_dir ? xtrabackup_incremental_dir : ".",
+		".del", prepare_handle_del_files);
+
+
 	int argc; for (argc = 0; argv[argc]; argc++) {}
 	encryption_plugin_prepare_init(argc, argv);
 
@@ -5334,7 +5801,7 @@ handle_options(int argc, char **argv, char ***argv_client, char ***argv_server)
 	srv_operation = SRV_OPERATION_RESTORE;
 
 	files_charset_info = &my_charset_utf8_general_ci;
-	check_if_backup_includes = backup_includes;
+
 
 	setup_error_messages();
 	sys_var_init();

mysql-test/suite/mariabackup/create_during_backup.result

+# xtrabackup backup
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SELECT COUNT(*) from t1;
+COUNT(*)
+10000
+DROP TABLE t1;

mysql-test/suite/mariabackup/create_during_backup.test

+--source include/have_debug.inc
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+mkdir $targetdir;
+
+# this will table and populate it, after backup has list of tables to be copied
+--let after_load_tablespaces =CREATE TABLE test.t1 ENGINE=INNODB SELECT UUID() from test.seq_1_to_10000
+
+echo # xtrabackup backup;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --dbug=+d,mariabackup_events;
+--enable_result_log
+
+--let after_load_tables=
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# Check that new table is there after restore.
+SELECT COUNT(*) from t1;
+DROP TABLE t1;
+rmdir $targetdir;
+
+

mysql-test/suite/mariabackup/create_with_data_directory_during_backup.result

+# xtrabackup backup
+# xtrabackup prepare
+DROP TABLE t;
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SELECT * FROM t;
+i
+DROP TABLE t;

mysql-test/suite/mariabackup/create_with_data_directory_during_backup.test

+--source include/have_debug.inc
+let $table_data_dir=$MYSQLTEST_VARDIR/tmp/ddir;
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+mkdir $table_data_dir;
+--replace_result $table_data_dir table_data_dir
+--let after_load_tablespaces=CREATE TABLE test.t(i int) ENGINE=INNODB DATA DIRECTORY='$table_data_dir'
+echo # xtrabackup backup;
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+--disable_result_log
+exec   $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --target-dir=$targetdir --dbug=+d,mariabackup_events;
+--enable_result_log
+--source include/shutdown_mysqld.inc
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+--source include/start_mysqld.inc
+DROP TABLE t;
+rmdir $table_data_dir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+SELECT * FROM t;
+DROP TABLE t;
+rmdir $targetdir;
+rmdir $table_data_dir;

mysql-test/suite/mariabackup/disabled.def

-
+unsupported_redo    : MDEV-16791 allows optimized redo
\ No newline at end of file

mysql-test/suite/mariabackup/drop_table_during_backup.result

+CREATE TABLE t1 (i int) ENGINE=INNODB;
+CREATE TABLE t2 (i int) ENGINE=INNODB;
+CREATE TABLE t3 (i int) ENGINE=INNODB;
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+CREATE TABLE t2(i int);
+DROP TABLE t2;
+DROP TABLE t3;

mysql-test/suite/mariabackup/drop_table_during_backup.test

+--source include/have_debug.inc
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+CREATE TABLE t1 (i int) ENGINE=INNODB;
+CREATE TABLE t2 (i int) ENGINE=INNODB;
+CREATE TABLE t3 (i int) ENGINE=INNODB;
+--let before_copy_test_t1=DROP TABLE test.t1
+--let after_copy_test_t2=DROP TABLE test.t2;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --target-dir=$targetdir --dbug=+d,mariabackup_events;
+--enable_result_log
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+#check that the table t1 does not exist in backup
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+CREATE TABLE t2(i int);
+DROP TABLE t2;
+DROP TABLE t3;
+rmdir $targetdir;

mysql-test/suite/mariabackup/incremental_ddl_during_backup.result

+call mtr.add_suppression("InnoDB: New log files created");
+CREATE TABLE t1(i INT PRIMARY KEY) ENGINE INNODB;
+CREATE TABLE t2(i INT PRIMARY KEY) ENGINE INNODB;
+CREATE TABLE t3(i INT) ENGINE INNODB;
+# Create full backup , modify table, then create incremental/differential backup
+INSERT into t1 values(1);
+# Prepare full backup, apply incremental one
+# Restore and check results
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+SELECT * from t1_renamed;
+i
+1
+DROP TABLE t1_renamed;
+CREATE TABLE t2(i INT PRIMARY KEY) ENGINE INNODB;
+DROP TABLE t2;
+DROP TABLE t3;
+DROP TABLE t4;

mysql-test/suite/mariabackup/incremental_ddl_during_backup.test

+--source include/have_debug.inc
+
+call mtr.add_suppression("InnoDB: New log files created");
+
+let $basedir=$MYSQLTEST_VARDIR/tmp/backup;
+let $incremental_dir=$MYSQLTEST_VARDIR/tmp/backup_inc1;
+
+CREATE TABLE t1(i INT PRIMARY KEY) ENGINE INNODB;
+CREATE TABLE t2(i INT PRIMARY KEY) ENGINE INNODB;
+CREATE TABLE t3(i INT) ENGINE INNODB;
+
+echo # Create full backup , modify table, then create incremental/differential backup;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$basedir;
+--enable_result_log
+
+INSERT into t1 values(1);
+
+--let after_load_tablespaces=CREATE TABLE test.t4 ENGINE=INNODB SELECT UUID() from test.seq_1_to_10000
+--let after_copy_test_t1=RENAME TABLE  test.t1 TO test.t1_renamed
+--let after_copy_test_t2=DROP TABLE test.t2
+--let after_copy_test_t3=CREATE INDEX a_i ON test.t3(i);
+
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$incremental_dir --incremental-basedir=$basedir --dbug=+d,mariabackup_events;
+--let after_load_tablespaces=
+--disable_result_log
+echo # Prepare full backup, apply incremental one;
+exec $XTRABACKUP --apply-log-only --prepare --target-dir=$basedir;
+exec $XTRABACKUP --prepare --target-dir=$basedir --incremental-dir=$incremental_dir ;
+
+echo # Restore and check results;
+let $targetdir=$basedir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# Test that t1 does not exist, but t1_renamed does
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+
+SELECT * from t1_renamed;
+DROP TABLE t1_renamed;
+
+# Test that t2 does not exist;
+CREATE TABLE t2(i INT PRIMARY KEY) ENGINE INNODB;
+DROP TABLE t2;
+
+DROP TABLE t3;
+DROP TABLE t4;
+
+# Cleanup
+rmdir $basedir;
+rmdir $incremental_dir;

mysql-test/suite/mariabackup/mlog_index_load.result

+CREATE TABLE t1(i INT PRIMARY KEY auto_increment, a int) ENGINE INNODB;
+INSERT INTO t1(a) SELECT * from seq_1_to_10000;
+# xtrabackup backup
+t1.frm
+t1.ibd
+t1.new
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SELECT COUNT(*) from t1;
+COUNT(*)
+10000
+DROP TABLE t1;

mysql-test/suite/mariabackup/mlog_index_load.test

+--source include/have_debug.inc
+
+CREATE TABLE t1(i INT PRIMARY KEY auto_increment, a int) ENGINE INNODB;
+INSERT INTO t1(a) SELECT * from seq_1_to_10000;
+
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+
+let after_copy_test_t1=CREATE INDEX a_ind ON test.t1(a) ALGORITHM=INPLACE;
+echo # xtrabackup backup;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --dbug=+d,mariabackup_events;
+--enable_result_log
+
+--list_files $targetdir/test t1*
+--let before_copy_test_t1=
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# Check that new table is there after restore.
+SELECT COUNT(*) from t1;
+DROP TABLE t1;
+rmdir $targetdir;
+

mysql-test/suite/mariabackup/recreate_table_during_backup.result

+CREATE TABLE t1(i int) ENGINE=INNODB;
+CREATE TABLE t2(i int) ENGINE=INNODB;
+CREATE TABLE t3(a CHAR(36)) ENGINE INNODB;
+INSERT INTO t3 SELECT UUID() FROM seq_1_to_1000;
+# xtrabackup backup
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SELECT COUNT(*) from t1;
+COUNT(*)
+100
+SELECT COUNT(*) from t2;
+COUNT(*)
+1000
+SELECT COUNT(*) from t3;
+COUNT(*)
+1000
+DROP INDEX index_a ON t3;
+DROP TABLE t1;
+DROP TABLE t2;
+DROP TABLE t3;

mysql-test/suite/mariabackup/recreate_table_during_backup.test

+--source include/have_debug.inc
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+mkdir $targetdir;
+
+CREATE TABLE t1(i int) ENGINE=INNODB;
+CREATE TABLE t2(i int) ENGINE=INNODB;
+CREATE TABLE t3(a CHAR(36)) ENGINE INNODB;
+INSERT INTO t3 SELECT UUID() FROM seq_1_to_1000;
+
+# this will table and populate it, after backup has list of tables to be copied
+--let before_copy_test_t1=BEGIN NOT ATOMIC DROP TABLE test.t1;CREATE TABLE test.t1 ENGINE=INNODB SELECT UUID() from test.seq_1_to_100; END
+--let after_copy_test_t2=BEGIN NOT ATOMIC  DROP TABLE test.t2;CREATE TABLE test.t2 ENGINE=INNODB SELECT UUID() from test.seq_1_to_1000; END
+--let after_copy_test_t3=ALTER TABLE test.t3 ADD INDEX index_a(a),ALGORITHM=COPY
+echo # xtrabackup backup;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --close-files --dbug=+d,mariabackup_events;
+--enable_result_log
+--let after_load_tables=
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# Check that new table is there after restore.
+SELECT COUNT(*) from t1;
+SELECT COUNT(*) from t2;
+SELECT COUNT(*) from t3;
+DROP INDEX index_a ON t3;
+DROP TABLE t1;
+DROP TABLE t2;
+DROP TABLE t3;
+rmdir $targetdir;
+
+
+

mysql-test/suite/mariabackup/rename_during_backup.result

+CREATE TABLE t1(i int)   ENGINE INNODB;
+INSERT into t1 values(1);
+CREATE TABLE t2(i int)   ENGINE INNODB;
+INSERT INTO t2 values(2);
+CREATE TABLE t3(i int) ENGINE INNODB;
+CREATE TABLE t4(i int) ENGINE INNODB;
+CREATE TABLE a(a int)    ENGINE INNODB;
+INSERT INTO a values(1);
+CREATE TABLE b(b CHAR(1)) ENGINE INNODB;
+INSERT INTO b VALUES('b');
+CREATE TABLE a1(a1 int)    ENGINE INNODB;
+INSERT INTO a1 VALUES(1);
+CREATE TABLE b1(b1 CHAR(2)) ENGINE INNODB;
+INSERT INTO b1 VALUES('b1');
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+SELECT * from t1_renamed;
+i
+1
+DROP TABLE t1_renamed;
+CREATE TABLE t2(i int);
+DROP TABLE t2;
+SELECT * from t2_renamed;
+i
+2
+DROP TABLE t2_renamed;
+SELECT * from t3;
+i
+3
+DROP TABLE t3;
+SELECT * from t4;
+i
+DROP TABLE t4;
+CREATE TABLE tmp(i int);
+DROP TABLE tmp;
+SELECT * FROM a;
+b
+b
+SELECT * FROM b;
+a
+1
+SELECT * FROM a1;
+b1
+b1
+SELECT * FROM b1;
+a1
+1
+DROP TABLE a,b,a1,b1;

mysql-test/suite/mariabackup/rename_during_backup.test

+--source include/have_debug.inc
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+mkdir $targetdir;
+
+CREATE TABLE t1(i int)   ENGINE INNODB;
+INSERT into t1 values(1);
+
+CREATE TABLE t2(i int)   ENGINE INNODB;
+INSERT INTO t2 values(2);
+
+CREATE TABLE t3(i int) ENGINE INNODB;
+CREATE TABLE t4(i int) ENGINE INNODB;
+
+CREATE TABLE a(a int)    ENGINE INNODB;
+INSERT INTO a values(1);
+CREATE TABLE b(b CHAR(1)) ENGINE INNODB;
+INSERT INTO b VALUES('b');
+
+CREATE TABLE a1(a1 int)    ENGINE INNODB;
+INSERT INTO a1 VALUES(1);
+
+CREATE TABLE b1(b1 CHAR(2)) ENGINE INNODB;
+INSERT INTO b1 VALUES('b1');
+
+# Test renames before of after copying tablespaces
+--let before_copy_test_t1=RENAME TABLE test.t1 TO test.t1_renamed
+--let after_copy_test_t2=RENAME TABLE  test.t2 TO test.t2_renamed
+--let after_copy_test_t3=BEGIN NOT ATOMIC RENAME TABLE test.t3 TO test.t3_tmp; INSERT INTO test.t3_tmp VALUES(3); RENAME TABLE test.t3_tmp TO test.t3; END
+--let before_copy_test_t4=RENAME TABLE test.t4 TO test.t4_tmp
+--let after_copy_test_t4=RENAME TABLE test.t4_tmp TO test.t4
+
+# Test circular renames
+--let before_copy_test_b=RENAME TABLE test.a to test.tmp, test.b to test.a, test.tmp to test.b
+--let after_copy_test_b1=RENAME TABLE test.a1 to test.tmp, test.b1 to test.a1, test.tmp to test.b1
+
+--disable_result_log 
+exec  $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$targetdir  --dbug=+d,mariabackup_events;
+--enable_result_log
+
+--let before_copy_test_t1=''
+--let after_copy_test_t2=''
+--let before_copy_test_a=''
+--let after_copy_test_a1=''
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# the table was renamed from t1 to t1_renamed
+# make sure t1 does not exist, and t1_renamed does.
+
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+
+SELECT * from t1_renamed;
+DROP TABLE t1_renamed;
+
+
+CREATE TABLE t2(i int);
+DROP TABLE t2;
+
+SELECT * from t2_renamed;
+DROP TABLE t2_renamed;
+
+#rename to itself
+SELECT * from t3;
+DROP TABLE t3;
+
+SELECT * from t4;
+DROP TABLE t4;
+
+# For circular renames , make sure intermediate tables do not exist
+CREATE TABLE tmp(i int);
+DROP TABLE tmp;
+
+SELECT * FROM a;
+SELECT * FROM b;
+SELECT * FROM a1;
+SELECT * FROM b1;
+
+DROP TABLE a,b,a1,b1;
+rmdir $targetdir;
+
+

mysql-test/suite/mariabackup/rename_during_mdl_lock.result

 CREATE TABLE t1(i int) ENGINE INNODB;
-FOUND 1 /failed to execute query SELECT 1 FROM/ in backup.log
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+SELECT * from t2;
+i
 DROP TABLE t2;

mysql-test/suite/mariabackup/rename_during_mdl_lock.test

 --source include/have_debug.inc
-let $targetdir=$MYSQLTEST_VARDIR/backup;
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
 mkdir $targetdir;
 CREATE TABLE t1(i int) ENGINE INNODB;
---error 1
-exec  $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --lock-ddl-per-table --dbug=+d,rename_during_mdl_lock_table 2>$targetdir/backup.log;
+exec  $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --lock-ddl-per-table --dbug=+d,rename_during_mdl_lock_table;
 
-let SEARCH_FILE=$targetdir/backup.log;
-let SEARCH_PATTERN=failed to execute query SELECT 1 FROM;
-source include/search_pattern_in_file.inc;
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+# the table was renamed from t1 to t2
+# make sure t1 does not exist, and t2 does
 
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+SELECT * from t2;
 DROP TABLE t2;
+
 rmdir $targetdir;
+

mysql-test/suite/mariabackup/suite.opt

---innodb --loose-changed_page_bitmaps --innodb-sys-tables --innodb-flush-log-at-trx-commit=2
+--innodb --loose-changed_page_bitmaps --innodb-sys-tables --innodb-flush-log-at-trx-commit=2 --sequence

storage/innobase/include/log0recv.h

 /*****************************************************************************
 
 Copyright (c) 1997, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, MariaDB Corporation.
+Copyright (c) 2017, 2018, 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 the Free Software
@@ -153,9 +153,21 @@ bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply);
 /** Moves the parsing buffer data left to the buffer start. */
 void recv_sys_justify_left_parsing_buf();
 
-/** Backup function checks whether the space id belongs to
-the skip table list given in the mariabackup option. */
-extern bool(*check_if_backup_includes)(ulint space_id);
+/** Report optimized DDL operation (without redo log), corresponding to MLOG_INDEX_LOAD.
+@param[in]	space_id	tablespace identifier
+*/
+extern void(*log_optimized_ddl_op)(ulint space_id);
+
+/** Report an operation to create, delete, or rename a file during backup.
+@param[in]	space_id	tablespace identifier
+@param[in]	flags		tablespace flags (NULL if not create)
+@param[in]	name		file name (not NUL-terminated)
+@param[in]	len		length of name, in bytes
+@param[in]	new_name	new file name (NULL if not rename)
+@param[in]	new_len		length of new_name, in bytes (0 if NULL) */
+extern void (*log_file_op)(ulint space_id, const byte* flags,
+			   const byte* name, ulint len,
+			   const byte* new_name, ulint new_len);
 
 /** Block of log record data */
 struct recv_data_t{

storage/innobase/log/log0recv.cc

@@ -169,9 +169,21 @@ typedef std::map<
 
 static recv_spaces_t	recv_spaces;
 
-/** Backup function checks whether the space id belongs to
-the skip table list given in the mariabackup option. */
-bool(*check_if_backup_includes)(ulint space_id);
+/** Report optimized DDL operation (without redo log), corresponding to MLOG_INDEX_LOAD.
+@param[in]	space_id	tablespace identifier
+*/
+void (*log_optimized_ddl_op)(ulint space_id);
+
+/** Report an operation to create, delete, or rename a file during backup.
+@param[in]	space_id	tablespace identifier
+@param[in]	flags		tablespace flags (NULL if not create)
+@param[in]	name		file name (not NUL-terminated)
+@param[in]	len		length of name, in bytes
+@param[in]	new_name	new file name (NULL if not rename)
+@param[in]	new_len		length of new_name, in bytes (0 if NULL) */
+void (*log_file_op)(ulint space_id, const byte* flags,
+		    const byte* name, ulint len,
+		    const byte* new_name, ulint new_len);
 
 /** Process a file name from a MLOG_FILE_* record.
 @param[in,out]	name		file name
@@ -381,9 +393,13 @@ fil_name_parse(
 
 		fil_name_process(
 			reinterpret_cast<char*>(ptr), len, space_id, true);
-
-		break;
+		/* fall through */
 	case MLOG_FILE_CREATE2:
+		if (log_file_op) {
+			log_file_op(space_id,
+				    type == MLOG_FILE_CREATE2 ? ptr - 4 : NULL,
+				    ptr, len, NULL, 0);
+		}
 		break;
 	case MLOG_FILE_RENAME2:
 		if (corrupt) {
@@ -424,6 +440,11 @@ fil_name_parse(
 			reinterpret_cast<char*>(new_name), new_len,
 			space_id, false);
 
+		if (log_file_op) {
+			log_file_op(space_id, NULL,
+				    ptr, len, new_name, new_len);
+		}
+
 		if (!apply) {
 			break;
 		}
@@ -2503,11 +2524,8 @@ bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply)
 			/* fall through */
 		case MLOG_INDEX_LOAD:
 			if (type == MLOG_INDEX_LOAD) {
-				if (check_if_backup_includes
-				    && !check_if_backup_includes(space)) {
-					ut_ad(srv_operation
-					      == SRV_OPERATION_BACKUP);
-					return true;
+				if (log_optimized_ddl_op) {
+					log_optimized_ddl_op(space);
 				}
 			}
 			/* fall through */