Merge 10.1 into 10.2

Also, include fixes by Vladislav Vaintroub to the
aws_key_management plugin. The AWS C++ SDK specifically depends on
OPENSSL_LIBRARIES, not generic SSL_LIBRARIES (such as YaSSL).

23 files changed, 829 insertions, 290 deletions

diff --git a/storage/xtradb/CMakeLists.txt b/storage/xtradb/CMakeLists.txt
index ba0797dd422..4f9d2bd2cbb 100644
--- a/storage/xtradb/CMakeLists.txt
+++ b/storage/xtradb/CMakeLists.txt
@@ -510,3 +510,6 @@ MYSQL_ADD_PLUGIN(xtradb ${INNOBASE_SOURCES} STORAGE_ENGINE
 IF(TARGET xtradb AND NOT XTRADB_OK)
   MESSAGE(FATAL_ERROR "Percona XtraDB is not supported on this platform")
 ENDIF()
+
+ADD_SUBDIRECTORY(${CMAKE_SOURCE_DIR}/extra/mariabackup ${CMAKE_BINARY_DIR}/extra/mariabackup)
+
diff --git a/storage/xtradb/btr/btr0btr.cc b/storage/xtradb/btr/btr0btr.cc
index c94b539c2c7..d84c93f8b3e 100644
--- a/storage/xtradb/btr/btr0btr.cc
+++ b/storage/xtradb/btr/btr0btr.cc
@@ -722,7 +722,6 @@ btr_root_fseg_validate(
 /**************************************************************//**
 Gets the root node of a tree and x- or s-latches it.
 @return	root page, x- or s-latched */
-static
 buf_block_t*
 btr_root_block_get(
 /*===============*/
@@ -1531,7 +1530,6 @@ btr_node_ptr_set_child_page_no(
 /************************************************************//**
 Returns the child page of a node pointer and x-latches it.
 @return	child page, x-latched */
-static
 buf_block_t*
 btr_node_ptr_get_child(
 /*===================*/
diff --git a/storage/xtradb/buf/buf0buf.cc b/storage/xtradb/buf/buf0buf.cc
index ebf6bb10caa..c57dab79ef7 100644
--- a/storage/xtradb/buf/buf0buf.cc
+++ b/storage/xtradb/buf/buf0buf.cc
@@ -65,6 +65,15 @@ Created 11/5/1995 Heikki Tuuri
 #include "fil0pagecompress.h"
 #include "ha_prototypes.h"
 
+/** Decrypt a page.
+@param[in,out]	bpage	Page control block
+@param[in,out]	space	tablespace
+@return whether the operation was successful */
+static
+bool
+buf_page_decrypt_after_read(buf_page_t* bpage, fil_space_t* space)
+	MY_ATTRIBUTE((nonnull));
+
 /* prototypes for new functions added to ha_innodb.cc */
 trx_t* innobase_get_trx();
 
@@ -548,16 +557,13 @@ buf_block_alloc(
 }
 #endif /* !UNIV_HOTBACKUP */
 
-/********************************************************************//**
-Checks if a page is all zeroes.
-@return	TRUE if the page is all zeroes */
+/** Check if a page is all zeroes.
+@param[in]	read_buf	database page
+@param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
+@return	whether the page is all zeroes */
 UNIV_INTERN
 bool
-buf_page_is_zeroes(
-/*===============*/
-	const byte*	read_buf,	/*!< in: a database page */
-	const ulint	zip_size)	/*!< in: size of compressed page;
-					0 for uncompressed pages */
+buf_page_is_zeroes(const byte* read_buf, ulint zip_size)
 {
 	const ulint page_size = zip_size ? zip_size : UNIV_PAGE_SIZE;
 
@@ -673,8 +679,7 @@ buf_page_is_checksum_valid_none(
 	       && checksum_field1 == BUF_NO_CHECKSUM_MAGIC);
 }
 
-/********************************************************************//**
-Checks if a page is corrupt.
+/** Check if a page is corrupt.
 @param[in]	check_lsn		true if LSN should be checked
 @param[in]	read_buf		Page to be checked
 @param[in]	zip_size		compressed size or 0
@@ -4529,34 +4534,30 @@ buf_mark_space_corrupt(
 	mutex_exit(&buf_pool->LRU_list_mutex);
 }
 
-/********************************************************************//**
-Check if page is maybe compressed, encrypted or both when we encounter
+/** Check if page is maybe compressed, encrypted or both when we encounter
 corrupted page. Note that we can't be 100% sure if page is corrupted
 or decrypt/decompress just failed.
-@param[in,out]	bpage		Page
-@return DB_SUCCESS if page has been read and is not corrupted,
-@retval DB_PAGE_CORRUPTED if page based on checksum check is corrupted,
-@retval DB_DECRYPTION_FAILED if page post encryption checksum matches but
+@param[in,out]	bpage		page
+@param[in,out]	space		tablespace from fil_space_acquire_for_io()
+@return	whether the operation succeeded
+@retval	DB_SUCCESS		if page has been read and is not corrupted
+@retval	DB_PAGE_CORRUPTED	if page based on checksum check is corrupted
+@retval	DB_DECRYPTION_FAILED	if page post encryption checksum matches but
 after decryption normal page checksum does not match.
-@retval DB_TABLESPACE_DELETED if accessed tablespace is not found */
+@retval	DB_TABLESPACE_DELETED	if accessed tablespace is not found */
 static
 dberr_t
-buf_page_check_corrupt(buf_page_t*	bpage)
+buf_page_check_corrupt(buf_page_t* bpage, fil_space_t* space)
 {
+	ut_ad(space->n_pending_ios > 0);
+
 	ulint zip_size = buf_page_get_zip_size(bpage);
 	byte* dst_frame = (zip_size) ? bpage->zip.data :
 		((buf_block_t*) bpage)->frame;
-	FilSpace space(bpage->space, true);
 	bool still_encrypted = false;
 	dberr_t err = DB_SUCCESS;
 	bool corrupted = false;
-	fil_space_crypt_t* crypt_data = NULL;
-
-	if (!space()) {
-		return(DB_TABLESPACE_DELETED);
-	}
-
-	crypt_data = space()->crypt_data;
+	fil_space_crypt_t* crypt_data = space->crypt_data;
 
 	/* In buf_decrypt_after_read we have either decrypted the page if
 	page post encryption checksum matches and used key_id is found
@@ -4568,12 +4569,12 @@ buf_page_check_corrupt(buf_page_t*	bpage)
 		crypt_data->type != CRYPT_SCHEME_UNENCRYPTED &&
 		!bpage->encrypted &&
 		fil_space_verify_crypt_checksum(dst_frame, zip_size,
-			space(), bpage->offset));
-
+			space, bpage->offset));
 	if (!still_encrypted) {
 		/* If traditional checksums match, we assume that page is
 		not anymore encrypted. */
-		corrupted = buf_page_is_corrupted(true, dst_frame, zip_size, space());
+		corrupted = buf_page_is_corrupted(true, dst_frame, zip_size,
+						  space);
 
 		if (!corrupted) {
 			bpage->encrypted = false;
@@ -4596,7 +4597,7 @@ buf_page_check_corrupt(buf_page_t*	bpage)
 			", page number=%u]"
 			" in file %s cannot be decrypted.",
 			bpage->space, bpage->offset,
-			space()->name);
+			space->name);
 
 		ib_logf(IB_LOG_LEVEL_INFO,
 			"However key management plugin or used key_version " ULINTPF
@@ -4614,26 +4615,23 @@ buf_page_check_corrupt(buf_page_t*	bpage)
 	return (err);
 }
 
-/********************************************************************//**
-Completes an asynchronous read or write request of a file page to or from
-the buffer pool.
+/** Complete a read or write request of a file page to or from the buffer pool.
 @param[in,out]		bpage		Page to complete
-@return DB_SUCCESS if page has been read and is not corrupted,
-DB_PAGE_CORRUPTED if page based on checksum check is corrupted,
-DB_DECRYPTION_FAILED if page post encryption checksum matches but
-after decryption normal page checksum does not match.
-in write only DB_SUCCESS is possible. */
+@return whether the operation succeeded
+@retval	DB_SUCCESS		always when writing, or if a read page was OK
+@retval	DB_PAGE_CORRUPTED	if the checksum fails on a page read
+@retval	DB_DECRYPTION_FAILED	if page post encryption checksum matches but
+				after decryption normal page checksum does
+				not match */
 UNIV_INTERN
 dberr_t
-buf_page_io_complete(
-	buf_page_t*	bpage)
+buf_page_io_complete(buf_page_t* bpage)
 {
 	enum buf_io_fix	io_type;
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
 	const ibool	uncompressed = (buf_page_get_state(bpage)
 					== BUF_BLOCK_FILE_PAGE);
 	bool		have_LRU_mutex = false;
-	fil_space_t*	space = NULL;
 	byte*		frame = NULL;
 	dberr_t 	err = DB_SUCCESS;
 
@@ -4653,7 +4651,13 @@ buf_page_io_complete(
 		ulint	read_space_id = 0;
 		uint	key_version = 0;
 
-		buf_page_decrypt_after_read(bpage);
+		ut_ad(bpage->zip.data || ((buf_block_t*)bpage)->frame);
+		fil_space_t* space = fil_space_acquire_for_io(bpage->space);
+		if (!space) {
+			return(DB_TABLESPACE_DELETED);
+		}
+
+		buf_page_decrypt_after_read(bpage, space);
 
 		if (buf_page_get_zip_size(bpage)) {
 			frame = bpage->zip.data;
@@ -4727,7 +4731,7 @@ buf_page_io_complete(
 
 		if (UNIV_LIKELY(!bpage->is_corrupt ||
 				!srv_pass_corrupt_table)) {
-			err = buf_page_check_corrupt(bpage);
+			err = buf_page_check_corrupt(bpage, space);
 		}
 
 database_corrupted:
@@ -4740,6 +4744,7 @@ database_corrupted:
 					buf_mark_space_corrupt(bpage);
 					ib_logf(IB_LOG_LEVEL_INFO,
 						"Simulated page corruption");
+					fil_space_release_for_io(space);
 					return(err);
 				}
 				err = DB_SUCCESS;
@@ -4747,9 +4752,6 @@ database_corrupted:
 			);
 
 			if (err == DB_PAGE_CORRUPTED) {
-				fil_system_enter();
-				space = fil_space_get_by_id(bpage->space);
-
 				ib_logf(IB_LOG_LEVEL_ERROR,
 					"Database page corruption on disk"
 					" or a failed file read of tablespace %s"
@@ -4760,8 +4762,6 @@ database_corrupted:
 					space->name,
 					bpage->space, bpage->offset);
 
-				fil_system_exit();
-
 				buf_page_print(frame, buf_page_get_zip_size(bpage),
 					BUF_PAGE_PRINT_NO_CRASH);
 
@@ -4798,6 +4798,7 @@ database_corrupted:
 				table as corrupted instead of crashing server */
 				if (bpage->space > TRX_SYS_SPACE) {
 					buf_mark_space_corrupt(bpage);
+					fil_space_release_for_io(space);
 					return(err);
 				} else {
 					ib_logf(IB_LOG_LEVEL_FATAL,
@@ -4836,6 +4837,8 @@ database_corrupted:
 			}
 
 		}
+
+		fil_space_release_for_io(space);
 	} else {
 		/* io_type == BUF_IO_WRITE */
 		if (bpage->slot) {
@@ -6306,16 +6309,17 @@ buf_page_encrypt_before_write(
 	return dst_frame;
 }
 
-/********************************************************************//**
-Decrypt page after it has been read from disk
-@param[in,out]	bpage		Page control block
-@return true if successfull, false if something went wrong
-*/
-UNIV_INTERN
+/** Decrypt a page.
+@param[in,out]	bpage	Page control block
+@param[in,out]	space	tablespace
+@return whether the operation was successful */
+static
 bool
-buf_page_decrypt_after_read(
-	buf_page_t*	bpage)
+buf_page_decrypt_after_read(buf_page_t* bpage, fil_space_t* space)
 {
+	ut_ad(space->n_pending_ios > 0);
+	ut_ad(space->id == bpage->space);
+
 	ulint zip_size = buf_page_get_zip_size(bpage);
 	ulint size = (zip_size) ? zip_size : UNIV_PAGE_SIZE;
 
@@ -6333,12 +6337,10 @@ buf_page_decrypt_after_read(
 		return (true);
 	}
 
-	FilSpace space(bpage->space, false, true);
-
 	/* Page is encrypted if encryption information is found from
 	tablespace and page contains used key_version. This is true
 	also for pages first compressed and then encrypted. */
-	if (!space() || !space()->crypt_data) {
+	if (!space->crypt_data) {
 		key_version = 0;
 	}
 
@@ -6375,8 +6377,8 @@ buf_page_decrypt_after_read(
 
 				/* Mark page encrypted in case it should
 				be. */
-				if (key_version && space()->crypt_data &&
-				    space()->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED) {
+				if (space->crypt_data->type
+				    != CRYPT_SCHEME_UNENCRYPTED) {
 					bpage->encrypted = true;
 				}
 
@@ -6391,12 +6393,8 @@ buf_page_decrypt_after_read(
 #endif
 
 			/* decrypt using crypt_buf to dst_frame */
-			byte* res = fil_space_decrypt(space(),
-						slot->crypt_buf,
-						dst_frame,
-						&bpage->encrypted);
-
-			if (!res) {
+			if (!fil_space_decrypt(space, slot->crypt_buf,
+					       dst_frame, &bpage->encrypted)) {
 				success = false;
 			}
 
@@ -6427,5 +6425,6 @@ buf_page_decrypt_after_read(
 		}
 	}
 
+	ut_ad(space->n_pending_ios > 0);
 	return (success);
 }
diff --git a/storage/xtradb/buf/buf0flu.cc b/storage/xtradb/buf/buf0flu.cc
index 62a2468ba66..1f5c3993be7 100644
--- a/storage/xtradb/buf/buf0flu.cc
+++ b/storage/xtradb/buf/buf0flu.cc
@@ -873,7 +873,7 @@ buf_flush_write_block_low(
 	buf_flush_t	flush_type,	/*!< in: type of flush */
 	bool		sync)		/*!< in: true if sync IO request */
 {
-	fil_space_t*	space = fil_space_acquire(bpage->space, true);
+	fil_space_t*	space = fil_space_acquire_for_io(bpage->space);
 	if (!space) {
 		return;
 	}
@@ -995,6 +995,13 @@ buf_flush_write_block_low(
 		ut_ad(flush_type == BUF_FLUSH_SINGLE_PAGE);
 		fil_flush(space);
 
+		/* The tablespace could already have been dropped,
+		because fil_io(request, sync) would already have
+		decremented the node->n_pending. However,
+		buf_page_io_complete() only needs to look up the
+		tablespace during read requests, not during writes. */
+		ut_ad(buf_page_get_io_fix_unlocked(bpage) == BUF_IO_WRITE);
+
 #ifdef UNIV_DEBUG
 		dberr_t err =
 #endif
@@ -1003,7 +1010,7 @@ buf_flush_write_block_low(
 		ut_ad(err == DB_SUCCESS);
 	}
 
-	fil_space_release(space);
+	fil_space_release_for_io(space);
 
 	/* Increment the counter of I/O operations used
 	for selecting LRU policy. */
@@ -2864,7 +2871,7 @@ DECLARE_THREAD(buf_flush_page_cleaner_thread)(
 		success = buf_flush_list(PCT_IO(100), LSN_MAX, &n_flushed);
 		buf_flush_wait_batch_end(NULL, BUF_FLUSH_LIST);
 
-	} while (!success || n_flushed > 0);
+	} while (!success || n_flushed > 0 || (IS_XTRABACKUP() && buf_get_n_pending_read_ios() > 0));
 
 	/* Some sanity checks */
 	ut_a(srv_get_active_thread_type() == SRV_NONE);
diff --git a/storage/xtradb/buf/buf0rea.cc b/storage/xtradb/buf/buf0rea.cc
index e275eead4cc..85b04d37a08 100644
--- a/storage/xtradb/buf/buf0rea.cc
+++ b/storage/xtradb/buf/buf0rea.cc
@@ -955,11 +955,8 @@ buf_read_ibuf_merge_pages(
 
 tablespace_deleted:
 			/* We have deleted or are deleting the single-table
-			tablespace: remove the entries for that page */
-
-			ibuf_merge_or_delete_for_page(NULL, space_ids[i],
-						      page_nos[i],
-						      zip_size, FALSE);
+			tablespace: remove the entries for tablespace. */
+			ibuf_delete_for_discarded_space(space_ids[i]);
 			break;
 		case DB_DECRYPTION_FAILED:
 			ib_logf(IB_LOG_LEVEL_ERROR,
diff --git a/storage/xtradb/dict/dict0load.cc b/storage/xtradb/dict/dict0load.cc
index 2dbde465369..4991c4f3fcc 100644
--- a/storage/xtradb/dict/dict0load.cc
+++ b/storage/xtradb/dict/dict0load.cc
@@ -945,6 +945,10 @@ dict_insert_tablespace_and_filepath(
 	return(err);
 }
 
+/* Set by Xtrabackup */
+my_bool (*dict_check_if_skip_table)(const char*	name) = 0;
+
+
 /********************************************************************//**
 This function looks at each table defined in SYS_TABLES.  It checks the
 tablespace for any table with a space_id > 0.  It looks up the tablespace
@@ -1064,6 +1068,9 @@ loop:
 
 		bool		is_temp = false;
 		bool		discarded = false;
+		bool		print_error_if_does_not_exist;
+		bool		remove_from_data_dict_if_does_not_exist;
+
 		ib_uint32_t	flags2 = static_cast<ib_uint32_t>(
 			mach_read_from_4(field));
 
@@ -1089,6 +1096,19 @@ loop:
 			goto loop;
 		}
 
+
+		ut_a(!IS_XTRABACKUP() || dict_check_if_skip_table);
+
+		if (is_temp || discarded ||
+			(IS_XTRABACKUP() && dict_check_if_skip_table(name))) {
+			print_error_if_does_not_exist = false;
+		}
+		else {
+			print_error_if_does_not_exist = true;
+		}
+
+		remove_from_data_dict_if_does_not_exist = IS_XTRABACKUP() && !(is_temp || discarded);
+
 		mtr_commit(&mtr);
 
 		switch (dict_check) {
@@ -1096,8 +1116,8 @@ loop:
 			/* All tablespaces should have been found in
 			fil_load_single_table_tablespaces(). */
 			if (fil_space_for_table_exists_in_mem(
-				space_id, name, !(is_temp || discarded),
-				false, NULL, 0, flags)
+				space_id, name, print_error_if_does_not_exist,
+				remove_from_data_dict_if_does_not_exist , false, NULL, 0, flags)
 			    && !(is_temp || discarded)) {
 				/* If user changes the path of .ibd files in
 				   *.isl files before doing crash recovery ,
@@ -1130,7 +1150,7 @@ loop:
 			trx_resurrect_table_locks(). */
 			if (fil_space_for_table_exists_in_mem(
 				    space_id, name, false,
-				    false, NULL, 0, flags)) {
+				    false, false, NULL, 0, flags)) {
 				break;
 			}
 			/* fall through */
@@ -2383,7 +2403,7 @@ err_exit:
 		table->file_unreadable = true;
 
 	} else if (!fil_space_for_table_exists_in_mem(
-			table->space, name, false, true, heap,
+			table->space, name, false, IS_XTRABACKUP(), true, heap,
 			table->id, table->flags)) {
 
 		if (DICT_TF2_FLAG_IS_SET(table, DICT_TF2_TEMPORARY)) {
diff --git a/storage/xtradb/fil/fil0crypt.cc b/storage/xtradb/fil/fil0crypt.cc
index c1ec1c7b1fd..a7373f451bc 100644
--- a/storage/xtradb/fil/fil0crypt.cc
+++ b/storage/xtradb/fil/fil0crypt.cc
@@ -683,7 +683,7 @@ fil_space_encrypt(
 	}
 
 	fil_space_crypt_t* crypt_data = space->crypt_data;
-	ut_ad(space->n_pending_ops);
+	ut_ad(space->n_pending_ios > 0);
 	ulint zip_size = fsp_flags_get_zip_size(space->flags);
 	byte* tmp = fil_encrypt_buf(crypt_data, space->id, offset, lsn, src_frame, zip_size, dst_frame);
 
@@ -860,7 +860,7 @@ fil_space_decrypt(
 	*decrypted = false;
 
 	ut_ad(space->crypt_data != NULL && space->crypt_data->is_encrypted());
-	ut_ad(space->n_pending_ops > 0);
+	ut_ad(space->n_pending_ios > 0);
 
 	bool encrypted = fil_space_decrypt(
 				space->crypt_data,
diff --git a/storage/xtradb/fil/fil0fil.cc b/storage/xtradb/fil/fil0fil.cc
index 7ac08cc0e97..e39be46840c 100644
--- a/storage/xtradb/fil/fil0fil.cc
+++ b/storage/xtradb/fil/fil0fil.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2014, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2014, 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 the Free Software
@@ -67,9 +67,11 @@ static ulint srv_data_read, srv_data_written;
 #include <fcntl.h>
 #endif
 #include "row0mysql.h"
+#include "trx0purge.h"
 
 MYSQL_PLUGIN_IMPORT extern my_bool lower_case_file_system;
 
+
 /*
 		IMPLEMENTATION OF THE TABLESPACE MEMORY CACHE
 		=============================================
@@ -247,18 +249,16 @@ fil_node_complete_io(
 	ulint		type);	/*!< in: OS_FILE_WRITE or OS_FILE_READ; marks
 				the node as modified if
 				type == OS_FILE_WRITE */
-/*******************************************************************//**
-Frees a space object from the tablespace memory cache. Closes the files in
-the chain but does not delete them. There must not be any pending i/o's or
+/** Free a space object from the tablespace memory cache. Close the files in
+the chain but do not delete them. There must not be any pending i/o's or
 flushes on the files.
-@return TRUE on success */
+The fil_system->mutex will be released.
+@param[in]	id		tablespace ID
+@param[in]	x_latched	whether the caller holds exclusive space->latch
+@return whether the tablespace existed */
 static
-ibool
-fil_space_free(
-/*===========*/
-	ulint		id,		/* in: space id */
-	ibool		x_latched);	/* in: TRUE if caller has space->latch
-					in X mode */
+bool
+fil_space_free_and_mutex_exit(ulint id, bool x_latched);
 /********************************************************************//**
 Reads data from a space to a buffer. Remember that the possible incomplete
 blocks at the end of file are ignored: they are not taken into account when
@@ -369,7 +369,6 @@ fil_node_get_space_id(
 
 /*******************************************************************//**
 Returns the table space by a given name, NULL if not found. */
-UNIV_INLINE
 fil_space_t*
 fil_space_get_by_name(
 /*==================*/
@@ -1368,18 +1367,14 @@ retry:
 	}
 }
 
-/*******************************************************************//**
-Frees a file node object from a tablespace memory cache. */
+/** Prepare a data file object for freeing.
+@param[in,out]	space	tablespace
+@param[in,out]	node	data file */
 static
 void
-fil_node_free(
-/*==========*/
-	fil_node_t*	node,	/*!< in, own: file node */
-	fil_system_t*	system,	/*!< in: tablespace memory cache */
-	fil_space_t*	space)	/*!< in: space where the file node is chained */
+fil_node_free_part1(fil_space_t* space, fil_node_t* node)
 {
-	ut_ad(node && system && space);
-	ut_ad(mutex_own(&(system->mutex)));
+	ut_ad(mutex_own(&fil_system->mutex));
 	ut_a(node->magic_n == FIL_NODE_MAGIC_N);
 	ut_a(node->n_pending == 0);
 	ut_a(!node->being_extended);
@@ -1402,12 +1397,22 @@ fil_node_free(
 			space->is_in_unflushed_spaces = false;
 
 			UT_LIST_REMOVE(unflushed_spaces,
-				       system->unflushed_spaces,
+				       fil_system->unflushed_spaces,
 				       space);
 		}
 
-		fil_node_close_file(node, system);
+		fil_node_close_file(node, fil_system);
 	}
+}
+
+/** Free a data file object.
+@param[in,out]	space	tablespace
+@param[in]	node	data file */
+static
+void
+fil_node_free_part2(fil_space_t* space, fil_node_t* node)
+{
+	ut_ad(!node->open);
 
 	space->size -= node->size;
 
@@ -1447,7 +1452,8 @@ fil_space_truncate_start(
 
 		trunc_len -= node->size * UNIV_PAGE_SIZE;
 
-		fil_node_free(node, fil_system, space);
+		fil_node_free_part1(space, node);
+		fil_node_free_part2(space, node);
 	}
 
 	mutex_exit(&fil_system->mutex);
@@ -1539,10 +1545,9 @@ fil_space_create(
 				"from the cache with id %lu",
 				name, (ulong) id);
 
-			ibool	success = fil_space_free(space->id, FALSE);
+			bool	success = fil_space_free_and_mutex_exit(
+				space->id, false);
 			ut_a(success);
-
-			mutex_exit(&fil_system->mutex);
 		}
 
 	} while (space != 0);
@@ -1574,12 +1579,13 @@ fil_space_create(
 
 		if (!fil_system->space_id_reuse_warned) {
 			fil_system->space_id_reuse_warned = TRUE;
-
-			ib_logf(IB_LOG_LEVEL_WARN,
-				"Allocated tablespace %lu, old maximum "
-				"was %lu",
-				(ulong) id,
-				(ulong) fil_system->max_assigned_id);
+			if (!IS_XTRABACKUP()) {
+				ib_logf(IB_LOG_LEVEL_WARN,
+					"Allocated tablespace %lu, old maximum "
+					"was %lu",
+					(ulong)id,
+					(ulong)fil_system->max_assigned_id);
+			}
 		}
 
 		fil_system->max_assigned_id = id;
@@ -1696,19 +1702,16 @@ fil_assign_new_space_id(
 	return(success);
 }
 
-/*******************************************************************//**
-Frees a space object from the tablespace memory cache. Closes the files in
-the chain but does not delete them. There must not be any pending i/o's or
+/** Free a space object from the tablespace memory cache. Close the files in
+the chain but do not delete them. There must not be any pending i/o's or
 flushes on the files.
-@return	TRUE if success */
+The fil_system->mutex will be released.
+@param[in]	id		tablespace ID
+@param[in]	x_latched	whether the caller holds exclusive space->latch
+@return whether the tablespace existed */
 static
-ibool
-fil_space_free(
-/*===========*/
-					/* out: TRUE if success */
-	ulint		id,		/* in: space id */
-	ibool		x_latched)	/* in: TRUE if caller has space->latch
-					in X mode */
+bool
+fil_space_free_and_mutex_exit(ulint id, bool x_latched)
 {
 	fil_space_t*	space;
 	fil_space_t*	fnamespace;
@@ -1718,13 +1721,11 @@ fil_space_free(
 	space = fil_space_get_by_id(id);
 
 	if (!space) {
-		ut_print_timestamp(stderr);
-		fprintf(stderr,
-			"  InnoDB: Error: trying to remove tablespace %lu"
-			" from the cache but\n"
-			"InnoDB: it is not there.\n", (ulong) id);
-
-		return(FALSE);
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"trying to remove non-existing tablespace " ULINTPF,
+			id);
+		mutex_exit(&fil_system->mutex);
+		return(false);
 	}
 
 	HASH_DELETE(fil_space_t, hash, fil_system->spaces, id, space);
@@ -1756,11 +1757,25 @@ fil_space_free(
 	ut_a(space->magic_n == FIL_SPACE_MAGIC_N);
 	ut_a(0 == space->n_pending_flushes);
 
+	for (fil_node_t* node = UT_LIST_GET_FIRST(space->chain);
+	     node != NULL;
+	     node = UT_LIST_GET_NEXT(chain, node)) {
+		fil_node_free_part1(space, node);
+	}
+
+	mutex_exit(&fil_system->mutex);
+
+	/* Wait for fil_space_release_for_io(); after
+	fil_space_detach(), the tablespace cannot be found, so
+	fil_space_acquire_for_io() would return NULL */
+	while (space->n_pending_ios) {
+		os_thread_sleep(100);
+	}
+
 	for (fil_node_t* fil_node = UT_LIST_GET_FIRST(space->chain);
 	     fil_node != NULL;
 	     fil_node = UT_LIST_GET_FIRST(space->chain)) {
-
-		fil_node_free(fil_node, fil_system, space);
+		fil_node_free_part2(space, fil_node);
 	}
 
 	ut_a(0 == UT_LIST_GET_LEN(space->chain));
@@ -2160,7 +2175,11 @@ fil_close_all_files(void)
 
 		space = UT_LIST_GET_NEXT(space_list, space);
 
-		fil_space_free(prev_space->id, FALSE);
+		/* This is executed during shutdown. No other thread
+		can create or remove tablespaces while we are not
+		holding fil_system->mutex. */
+		fil_space_free_and_mutex_exit(prev_space->id, false);
+		mutex_enter(&fil_system->mutex);
 	}
 
 	mutex_exit(&fil_system->mutex);
@@ -2208,7 +2227,11 @@ fil_close_log_files(
 		space = UT_LIST_GET_NEXT(space_list, space);
 
 		if (free) {
-			fil_space_free(prev_space->id, FALSE);
+			/* This is executed during startup. No other thread
+			can create or remove tablespaces while we are not
+			holding fil_system->mutex. */
+			fil_space_free_and_mutex_exit(prev_space->id, false);
+			mutex_enter(&fil_system->mutex);
 		}
 	}
 
@@ -2413,6 +2436,19 @@ fil_read_first_page(
 	const char*	check_msg = NULL;
 	fil_space_crypt_t* cdata;
 
+	if (IS_XTRABACKUP() && srv_backup_mode) {
+		/* Files smaller than page size may occur
+		in xtrabackup, when server creates new file
+		but has not yet written into it, or wrote only
+		partially. Checks size here, to avoid exit in os_file_read.
+		This file will be skipped by xtrabackup if it is too small.
+		*/
+		os_offset_t	file_size;
+		file_size = os_file_get_size(data_file);
+		if (file_size < FIL_IBD_FILE_INITIAL_SIZE*UNIV_PAGE_SIZE) {
+			return "File size is less than minimum";
+		}
+	}
 	buf = static_cast<byte*>(ut_malloc(2 * UNIV_PAGE_SIZE));
 
 	/* Align the memory for a possible read from a raw device */
@@ -2443,7 +2479,9 @@ fil_read_first_page(
 			}
 		}
 
-		check_msg = fil_check_first_page(page, *space_id, *flags);
+        if (!(IS_XTRABACKUP() && srv_backup_mode)) {
+		  check_msg = fil_check_first_page(page, *space_id, *flags);
+		}
 	}
 
 	flushed_lsn = mach_read_from_8(page +
@@ -3003,15 +3041,13 @@ fil_close_tablespace(
 	/* If the free is successful, the X lock will be released before
 	the space memory data structure is freed. */
 
-	if (!fil_space_free(id, TRUE)) {
+	if (!fil_space_free_and_mutex_exit(id, TRUE)) {
 		rw_lock_x_unlock(&space->latch);
 		err = DB_TABLESPACE_NOT_FOUND;
 	} else {
 		err = DB_SUCCESS;
 	}
 
-	mutex_exit(&fil_system->mutex);
-
 	/* If it is a delete then also delete any generated files, otherwise
 	when we drop the database the remove directory will fail. */
 
@@ -3120,12 +3156,10 @@ fil_delete_tablespace(
 		ut_a(node->n_pending == 0);
 	}
 
-	if (!fil_space_free(id, TRUE)) {
+	if (!fil_space_free_and_mutex_exit(id, true)) {
 		err = DB_TABLESPACE_NOT_FOUND;
 	}
 
-	mutex_exit(&fil_system->mutex);
-
 	if (err != DB_SUCCESS) {
 		rw_lock_x_unlock(&space->latch);
 	} else if (!os_file_delete(innodb_file_data_key, path)
@@ -3137,7 +3171,7 @@ fil_delete_tablespace(
 		err = DB_IO_ERROR;
 	}
 
-	if (err == DB_SUCCESS) {
+	if (err == DB_SUCCESS && !IS_XTRABACKUP()) {
 #ifndef UNIV_HOTBACKUP
 		/* Write a log record about the deletion of the .ibd
 		file, so that mysqlbackup can replay it in the
@@ -3536,7 +3570,7 @@ skip_second_rename:
 	mutex_exit(&fil_system->mutex);
 
 #ifndef UNIV_HOTBACKUP
-	if (success && !recv_recovery_on) {
+	if (success && !recv_recovery_on && !IS_XTRABACKUP()) {
 		mtr_t		mtr;
 
 		mtr_start(&mtr);
@@ -3782,7 +3816,18 @@ fil_create_new_single_table_tablespace(
 	ibool		success;
 	/* TRUE if a table is created with CREATE TEMPORARY TABLE */
 	bool		is_temp = !!(flags2 & DICT_TF2_TEMPORARY);
-	bool		has_data_dir = FSP_FLAGS_HAS_DATA_DIR(flags) != 0;
+
+
+	/* For XtraBackup recovery we force remote tablespaces to be local,
+	i.e. never execute the code path corresponding to has_data_dir == true.
+	We don't create .isl files either, because we rely on innobackupex to
+	copy them under a global lock, and use them to copy remote tablespaces
+	to their proper locations on --copy-back.
+
+	See also MySQL bug #72022: dir_path is always NULL for remote
+	tablespaces when a MLOG_FILE_CREATE* log record is replayed (the remote
+	directory is not available from MLOG_FILE_CREATE*). */
+	bool		has_data_dir = FSP_FLAGS_HAS_DATA_DIR(flags) != 0 && !IS_XTRABACKUP();
 	ulint		atomic_writes = FSP_FLAGS_GET_ATOMIC_WRITES(flags);
 	fil_space_crypt_t *crypt_data = NULL;
 
@@ -3964,6 +4009,7 @@ fil_create_new_single_table_tablespace(
 	}
 
 #ifndef UNIV_HOTBACKUP
+	if (!IS_XTRABACKUP())
 	{
 		mtr_t		mtr;
 		ulint		mlog_file_flag = 0;
@@ -4004,6 +4050,138 @@ error_exit_3:
 	return(err);
 }
 
+#include "pars0pars.h"
+#include "que0que.h"
+#include "dict0priv.h"
+static
+void
+fil_remove_invalid_table_from_data_dict(const char *name)
+{
+	trx_t*		trx;
+	pars_info_t*	info = NULL;
+
+	trx = trx_allocate_for_mysql();
+	trx_start_for_ddl(trx, TRX_DICT_OP_TABLE);
+
+	ut_ad(mutex_own(&dict_sys->mutex));
+
+	trx->op_info = "removing invalid table from data dictionary";
+
+	info = pars_info_create();
+
+	pars_info_add_str_literal(info, "table_name", name);
+
+	que_eval_sql(info,
+		"PROCEDURE DROP_TABLE_PROC () IS\n"
+		"sys_foreign_id CHAR;\n"
+		"table_id CHAR;\n"
+		"index_id CHAR;\n"
+		"foreign_id CHAR;\n"
+		"found INT;\n"
+
+		"DECLARE CURSOR cur_fk IS\n"
+		"SELECT ID FROM SYS_FOREIGN\n"
+		"WHERE FOR_NAME = :table_name\n"
+		"AND TO_BINARY(FOR_NAME)\n"
+		"  = TO_BINARY(:table_name)\n"
+		"LOCK IN SHARE MODE;\n"
+
+		"DECLARE CURSOR cur_idx IS\n"
+		"SELECT ID FROM SYS_INDEXES\n"
+		"WHERE TABLE_ID = table_id\n"
+		"LOCK IN SHARE MODE;\n"
+
+		"BEGIN\n"
+		"SELECT ID INTO table_id\n"
+		"FROM SYS_TABLES\n"
+		"WHERE NAME = :table_name\n"
+		"LOCK IN SHARE MODE;\n"
+		"IF (SQL % NOTFOUND) THEN\n"
+		"       RETURN;\n"
+		"END IF;\n"
+		"found := 1;\n"
+		"SELECT ID INTO sys_foreign_id\n"
+		"FROM SYS_TABLES\n"
+		"WHERE NAME = 'SYS_FOREIGN'\n"
+		"LOCK IN SHARE MODE;\n"
+		"IF (SQL % NOTFOUND) THEN\n"
+		"       found := 0;\n"
+		"END IF;\n"
+		"IF (:table_name = 'SYS_FOREIGN') THEN\n"
+		"       found := 0;\n"
+		"END IF;\n"
+		"IF (:table_name = 'SYS_FOREIGN_COLS') THEN\n"
+		"       found := 0;\n"
+		"END IF;\n"
+		"OPEN cur_fk;\n"
+		"WHILE found = 1 LOOP\n"
+		"       FETCH cur_fk INTO foreign_id;\n"
+		"       IF (SQL % NOTFOUND) THEN\n"
+		"               found := 0;\n"
+		"       ELSE\n"
+		"               DELETE FROM SYS_FOREIGN_COLS\n"
+		"               WHERE ID = foreign_id;\n"
+		"               DELETE FROM SYS_FOREIGN\n"
+		"               WHERE ID = foreign_id;\n"
+		"       END IF;\n"
+		"END LOOP;\n"
+		"CLOSE cur_fk;\n"
+		"found := 1;\n"
+		"OPEN cur_idx;\n"
+		"WHILE found = 1 LOOP\n"
+		"       FETCH cur_idx INTO index_id;\n"
+		"       IF (SQL % NOTFOUND) THEN\n"
+		"               found := 0;\n"
+		"       ELSE\n"
+		"               DELETE FROM SYS_FIELDS\n"
+		"               WHERE INDEX_ID = index_id;\n"
+		"               DELETE FROM SYS_INDEXES\n"
+		"               WHERE ID = index_id\n"
+		"               AND TABLE_ID = table_id;\n"
+		"       END IF;\n"
+		"END LOOP;\n"
+		"CLOSE cur_idx;\n"
+		"DELETE FROM SYS_COLUMNS\n"
+		"WHERE TABLE_ID = table_id;\n"
+		"DELETE FROM SYS_TABLES\n"
+		"WHERE NAME = :table_name;\n"
+		"END;\n"
+		, FALSE, trx);
+
+	/* SYS_DATAFILES and SYS_TABLESPACES do not necessarily exist
+	on XtraBackup recovery. See comments around
+	dict_create_or_check_foreign_constraint_tables() in
+	innobase_start_or_create_for_mysql(). */
+	if (dict_table_get_low("SYS_DATAFILES") != NULL) {
+		info = pars_info_create();
+
+		pars_info_add_str_literal(info, "table_name", name);
+
+		que_eval_sql(info,
+			"PROCEDURE DROP_TABLE_PROC () IS\n"
+			"space_id INT;\n"
+
+			"BEGIN\n"
+			"SELECT SPACE INTO space_id\n"
+			"FROM SYS_TABLES\n"
+			"WHERE NAME = :table_name;\n"
+			"IF (SQL % NOTFOUND) THEN\n"
+			"       RETURN;\n"
+			"END IF;\n"
+			"DELETE FROM SYS_TABLESPACES\n"
+			"WHERE SPACE = space_id;\n"
+			"DELETE FROM SYS_DATAFILES\n"
+			"WHERE SPACE = space_id;\n"
+			"END;\n"
+			, FALSE, trx);
+	}
+
+	trx_commit_for_mysql(trx);
+
+	trx_free_for_mysql(trx);
+}
+
+
 #ifndef UNIV_HOTBACKUP
 /********************************************************************//**
 Report information about a bad tablespace. */
@@ -4144,8 +4322,10 @@ fil_open_single_table_tablespace(
 	in the default location. If it is remote, it should not be here. */
 	def.filepath = fil_make_ibd_name(tablename, false);
 
-	/* The path_in was read from SYS_DATAFILES. */
-	if (path_in) {
+	/* The path_in was read from SYS_DATAFILES.
+	We skip SYS_DATAFILES validation and remote tablespaces discovery for
+	XtraBackup, as all tablespaces are local for XtraBackup recovery. */
+	if (path_in && !IS_XTRABACKUP()) {
 		if (strcmp(def.filepath, path_in)) {
 			dict.filepath = mem_strdup(path_in);
 			/* possibility of multiple files. */
@@ -4287,12 +4467,19 @@ fil_open_single_table_tablespace(
 		/* The following call prints an error message */
 		os_file_get_last_error(true);
 
-		ib_logf(IB_LOG_LEVEL_ERROR,
+		ib_logf(IS_XTRABACKUP() ? IB_LOG_LEVEL_WARN : IB_LOG_LEVEL_ERROR,
 			"Could not find a valid tablespace file for '%s'. "
 			"See " REFMAN "innodb-troubleshooting-datadict.html "
 			"for how to resolve the issue.",
 			tablename);
+		if (IS_XTRABACKUP() && fix_dict) {
+			ib_logf(IB_LOG_LEVEL_WARN,
+			"It will be removed from the data dictionary.");
 
+			if (purge_sys) {
+				fil_remove_invalid_table_from_data_dict(tablename);
+			}
+		}
 		err = DB_CORRUPTION;
 
 		goto cleanup_and_exit;
@@ -4717,6 +4904,11 @@ check_first_page:
 	}
 
 	if (!fsp->success) {
+		if (IS_XTRABACKUP()) {
+			/* Do not attempt restore from doublewrite buffer
+			  in Xtrabackup, this does not work.*/
+			return;
+		}
 		if (!restore_attempted) {
 			if (!fil_user_tablespace_find_space_id(fsp)) {
 				return;
@@ -4784,6 +4976,10 @@ fil_load_single_table_tablespace(
 	os_offset_t	size;
 	fil_space_t*	space;
 
+	fsp_open_info*	fsp;
+	ulong		minimum_size;
+	ibool		file_space_create_success;
+
 	memset(&def, 0, sizeof(def));
 	memset(&remote, 0, sizeof(remote));
 
@@ -4839,6 +5035,7 @@ fil_load_single_table_tablespace(
 # endif /* !UNIV_HOTBACKUP */
 #endif
 
+
 	/* Check for a link file which locates a remote tablespace. */
 	remote.success = fil_open_linked_file(
 		tablename, &remote.filepath, &remote.file, FALSE);
@@ -4849,6 +5046,17 @@ fil_load_single_table_tablespace(
 		if (!remote.success) {
 			os_file_close(remote.file);
 			mem_free(remote.filepath);
+
+			if (srv_backup_mode && (remote.id == ULINT_UNDEFINED
+				|| remote.id == 0)) {
+
+				/* Ignore files that have uninitialized space
+				IDs on the backup stage. This means that a
+				tablespace has just been created and we will
+				replay the corresponding log records on
+				prepare. */
+				goto func_exit_after_close;
+			}
 		}
 	}
 
@@ -4863,6 +5071,18 @@ fil_load_single_table_tablespace(
 		fil_validate_single_table_tablespace(tablename, &def);
 		if (!def.success) {
 			os_file_close(def.file);
+
+			if (IS_XTRABACKUP() && srv_backup_mode && (def.id == ULINT_UNDEFINED
+				|| def.id == 0)) {
+
+				/* Ignore files that have uninitialized space
+				IDs on the backup stage. This means that a
+				tablespace has just been created and we will
+				replay the corresponding log records on
+				prepare. */
+
+				goto func_exit_after_close;
+			}
 		}
 	}
 
@@ -4948,7 +5168,7 @@ will_not_choose:
 	/* At this point, only one tablespace is open */
 	ut_a(def.success == !remote.success);
 
-	fsp_open_info*	fsp = def.success ? &def : &remote;
+	fsp = def.success ? &def : &remote;
 
 	/* Get and test the file size. */
 	size = os_file_get_size(fsp->file);
@@ -4967,19 +5187,14 @@ will_not_choose:
 
 	/* Every .ibd file is created >= 4 pages in size. Smaller files
 	cannot be ok. */
-	ulong minimum_size = FIL_IBD_FILE_INITIAL_SIZE * UNIV_PAGE_SIZE;
+	minimum_size = FIL_IBD_FILE_INITIAL_SIZE * UNIV_PAGE_SIZE;
 	if (size < minimum_size) {
-#ifndef UNIV_HOTBACKUP
 		ib_logf(IB_LOG_LEVEL_ERROR,
 			"The size of single-table tablespace file %s "
 			"is only " UINT64PF ", should be at least %lu!",
 			fsp->filepath, size, minimum_size);
 		os_file_close(fsp->file);
 		goto no_good_file;
-#else
-		fsp->id = ULINT_UNDEFINED;
-		fsp->flags = 0;
-#endif /* !UNIV_HOTBACKUP */
 	}
 
 #ifdef UNIV_HOTBACKUP
@@ -5050,6 +5265,7 @@ will_not_choose:
 	}
 	mutex_exit(&fil_system->mutex);
 #endif /* UNIV_HOTBACKUP */
+
 	/* Adjust the memory-based flags that would normally be set by
 	dict_tf_to_fsp_flags(). In recovery, we have no data dictionary. */
 	if (FSP_FLAGS_HAS_PAGE_COMPRESSION(fsp->flags)) {
@@ -5060,7 +5276,7 @@ will_not_choose:
 	/* We will leave atomic_writes at ATOMIC_WRITES_DEFAULT.
 	That will be adjusted in fil_space_for_table_exists_in_mem(). */
 
-	ibool file_space_create_success = fil_space_create(
+	file_space_create_success = fil_space_create(
 		tablename, fsp->id, fsp->flags, FIL_TABLESPACE,
 		fsp->crypt_data, false);
 
@@ -5088,13 +5304,56 @@ will_not_choose:
 	}
 
 func_exit:
-	os_file_close(fsp->file);
+	/* We reuse file handles on the backup stage in XtraBackup to avoid
+	inconsistencies between the file name and the actual tablespace contents
+	if a DDL occurs between a fil_load_single_table_tablespaces() call and
+	the actual copy operation. */
+	if (IS_XTRABACKUP() && srv_backup_mode && !srv_close_files) {
+
+		fil_node_t*	node;
+		fil_space_t*	space;
+
+		mutex_enter(&fil_system->mutex);
+
+		space = fil_space_get_by_id(fsp->id);
+
+		if (space) {
+			node = UT_LIST_GET_LAST(space->chain);
+
+			/* The handle will be closed by xtrabackup in
+			xtrabackup_copy_datafile(). We set node->open to TRUE to
+			make sure no one calls fil_node_open_file()
+			(i.e. attempts to reopen the tablespace by name) during
+			the backup stage. */
+
+			node->open = TRUE;
+			node->handle = fsp->file;
+
+			/* The following is copied from fil_node_open_file() to
+			pass fil_system validaty checks. We cannot use
+			fil_node_open_file() directly, as that would re-open the
+			file by name and create another file handle. */
+
+			fil_system->n_open++;
+			fil_n_file_opened++;
+
+			if (fil_space_belongs_in_lru(space)) {
+
+				/* Put the node to the LRU list */
+				UT_LIST_ADD_FIRST(LRU, fil_system->LRU, node);
+			}
+		}
+
+		mutex_exit(&fil_system->mutex);
+	}
+	else {
+		os_file_close(fsp->file);
+	}
+
 
-#ifdef UNIV_HOTBACKUP
 func_exit_after_close:
-#else
 	ut_ad(!mutex_own(&fil_system->mutex));
-#endif
+
 	mem_free(tablename);
 	if (remote.success) {
 		mem_free(remote.filepath);
@@ -5108,7 +5367,7 @@ directory. We retry 100 times if os_file_readdir_next_file() returns -1. The
 idea is to read as much good data as we can and jump over bad data.
 @return 0 if ok, -1 if error even after the retries, 1 if at the end
 of the directory */
-static
+UNIV_INTERN
 int
 fil_file_readdir_next_file(
 /*=======================*/
@@ -5138,6 +5397,9 @@ fil_file_readdir_next_file(
 	return(-1);
 }
 
+
+my_bool(*fil_check_if_skip_database_by_path)(const char* name);
+
 #define CHECK_TIME_EVERY_N_FILES   10
 /********************************************************************//**
 At the server startup, if we need crash recovery, scans the database
@@ -5149,7 +5411,7 @@ space id is != 0.
 @return	DB_SUCCESS or error number */
 UNIV_INTERN
 dberr_t
-fil_load_single_table_tablespaces(void)
+fil_load_single_table_tablespaces(ibool (*pred)(const char*, const char*))
 /*===================================*/
 {
 	int		ret;
@@ -5208,7 +5470,19 @@ fil_load_single_table_tablespaces(void)
 			    "%s/%s", fil_path_to_mysql_datadir, dbinfo.name);
 		srv_normalize_path_for_win(dbpath);
 
-		dbdir = os_file_opendir(dbpath, FALSE);
+		if (IS_XTRABACKUP()) {
+			ut_a(fil_check_if_skip_database_by_path);
+			if (fil_check_if_skip_database_by_path(dbpath)) {
+				fprintf(stderr, "Skipping db: %s\n", dbpath);
+				dbdir = NULL;
+			} else {
+				/* We want wrong directory permissions to be a fatal
+				error for XtraBackup. */
+				dbdir = os_file_opendir(dbpath, TRUE);
+			}
+		} else {
+			dbdir = os_file_opendir(dbpath, FALSE);
+		}
 
 		if (dbdir != NULL) {
 
@@ -5224,14 +5498,20 @@ fil_load_single_table_tablespaces(void)
 					goto next_file_item;
 				}
 
-				/* We found a symlink or a file */
+				/* We found a symlink or a file
+
+				Ignore .isl files on XtraBackup
+				recovery, all tablespaces must be local. */
 				if (strlen(fileinfo.name) > 4
 				    && (0 == strcmp(fileinfo.name
 						   + strlen(fileinfo.name) - 4,
 						   ".ibd")
-					|| 0 == strcmp(fileinfo.name
-						   + strlen(fileinfo.name) - 4,
-						   ".isl"))) {
+					|| ((!IS_XTRABACKUP() || srv_backup_mode) 
+							&& 0 == strcmp(fileinfo.name
+							 + strlen(fileinfo.name) - 4,
+							".isl")))
+					 && (!pred ||
+						pred(dbinfo.name, fileinfo.name))) {
 					/* The name ends in .ibd or .isl;
 					try opening the file */
 					fil_load_single_table_tablespace(
@@ -5387,6 +5667,9 @@ fil_space_for_table_exists_in_mem(
 					information to the .err log if a
 					matching tablespace is not found from
 					memory */
+	bool		remove_from_data_dict_if_does_not_exist,
+					/*!< in: remove from the data dictionary
+					if tablespace does not exist */
 	bool		adjust_space,	/*!< in: whether to adjust space id
 					when find table space mismatch */
 	mem_heap_t*	heap,		/*!< in: heap memory */
@@ -5457,6 +5740,11 @@ fil_space_for_table_exists_in_mem(
 		if (fnamespace == NULL) {
 			if (print_error_if_does_not_exist) {
 				fil_report_missing_tablespace(name, id);
+				if (IS_XTRABACKUP() && remove_from_data_dict_if_does_not_exist) {
+					ib_logf(IB_LOG_LEVEL_WARN,
+						"It will be removed from "
+						"the data dictionary.");
+				}
 			}
 		} else {
 			ut_print_timestamp(stderr);
@@ -5918,7 +6206,7 @@ UNIV_INTERN
 ulint
 fil_space_get_block_size(const fil_space_t* space, unsigned offset)
 {
-	ut_ad(space->n_pending_ops > 0);
+	ut_ad(space->n_pending_ios > 0);
 
 	ulint block_size = 512;
 
@@ -6119,7 +6407,7 @@ _fil_io(
 	/* Check that at least the start offset is within the bounds of a
 	single-table tablespace, including rollback tablespaces. */
 	if (UNIV_UNLIKELY(node->size <= block_offset)
-	    && space->id != 0 && space->purpose == FIL_TABLESPACE) {
+		&& space->id != 0 && space->purpose == FIL_TABLESPACE) {
 
 		fil_report_invalid_page_access(
 			block_offset, space_id, space->name, byte_offset,
@@ -6355,7 +6643,7 @@ UNIV_INTERN
 void
 fil_flush(fil_space_t* space)
 {
-	ut_ad(space->n_pending_ops > 0);
+	ut_ad(space->n_pending_ios > 0);
 
 	if (!space->is_stopping()) {
 		mutex_enter(&fil_system->mutex);
@@ -7284,13 +7572,11 @@ Used by background threads that do not necessarily hold proper locks
 for concurrency control.
 @param[in]	id	tablespace ID
 @param[in]	silent	whether to silently ignore missing tablespaces
-@param[in]	for_io	whether to look up the tablespace while performing I/O
-			(possibly executing TRUNCATE)
 @return	the tablespace
 @retval	NULL if missing or being deleted or truncated */
 UNIV_INTERN
 fil_space_t*
-fil_space_acquire_low(ulint id, bool silent, bool for_io)
+fil_space_acquire_low(ulint id, bool silent)
 {
 	fil_space_t*	space;
 
@@ -7303,7 +7589,7 @@ fil_space_acquire_low(ulint id, bool silent, bool for_io)
 			ib_logf(IB_LOG_LEVEL_WARN, "Trying to access missing"
 				" tablespace " ULINTPF ".", id);
 		}
-	} else if (!for_io && space->is_stopping()) {
+	} else if (space->is_stopping()) {
 		space = NULL;
 	} else {
 		space->n_pending_ops++;
@@ -7314,8 +7600,44 @@ fil_space_acquire_low(ulint id, bool silent, bool for_io)
 	return(space);
 }
 
+/** Acquire a tablespace for reading or writing a block,
+when it could be dropped concurrently.
+@param[in]	id	tablespace ID
+@return	the tablespace
+@retval	NULL if missing */
+UNIV_INTERN
+fil_space_t*
+fil_space_acquire_for_io(ulint id)
+{
+	mutex_enter(&fil_system->mutex);
+
+	fil_space_t* space = fil_space_get_by_id(id);
+
+	if (space) {
+		space->n_pending_ios++;
+	}
+
+	mutex_exit(&fil_system->mutex);
+
+	return(space);
+}
+
+/** Release a tablespace acquired with fil_space_acquire_for_io().
+@param[in,out]	space	tablespace to release  */
+UNIV_INTERN
+void
+fil_space_release_for_io(fil_space_t* space)
+{
+	mutex_enter(&fil_system->mutex);
+	ut_ad(space->magic_n == FIL_SPACE_MAGIC_N);
+	ut_ad(space->n_pending_ios > 0);
+	space->n_pending_ios--;
+	mutex_exit(&fil_system->mutex);
+}
+
 /** Release a tablespace acquired with fil_space_acquire().
 @param[in,out]	space	tablespace to release  */
+UNIV_INTERN
 void
 fil_space_release(fil_space_t* space)
 {
@@ -7334,6 +7656,7 @@ blocks a concurrent operation from dropping the tablespace.
 If NULL, use the first fil_space_t on fil_system->space_list.
 @return pointer to the next fil_space_t.
 @retval NULL if this was the last*/
+UNIV_INTERN
 fil_space_t*
 fil_space_next(fil_space_t* prev_space)
 {
@@ -7401,6 +7724,7 @@ blocks a concurrent operation from dropping the tablespace.
 If NULL, use the first fil_space_t on fil_system->space_list.
 @return pointer to the next fil_space_t.
 @retval NULL if this was the last*/
+UNIV_INTERN
 fil_space_t*
 fil_space_keyrotate_next(
 	fil_space_t*	prev_space)
diff --git a/storage/xtradb/handler/ha_innodb.cc b/storage/xtradb/handler/ha_innodb.cc
index 91c9fc7824b..4353bc1d288 100644
--- a/storage/xtradb/handler/ha_innodb.cc
+++ b/storage/xtradb/handler/ha_innodb.cc
@@ -289,7 +289,8 @@ static TYPELIB innodb_stats_method_typelib = {
 
 /** Possible values for system variables "innodb_checksum_algorithm" and
 "innodb_log_checksum_algorithm". */
-static const char* innodb_checksum_algorithm_names[] = {
+UNIV_INTERN
+const char* innodb_checksum_algorithm_names[] = {
 	"CRC32",
 	"STRICT_CRC32",
 	"INNODB",
@@ -301,7 +302,8 @@ static const char* innodb_checksum_algorithm_names[] = {
 
 /** Used to define an enumerate type of the system variables
 innodb_checksum_algorithm and innodb_log_checksum_algorithm. */
-static TYPELIB innodb_checksum_algorithm_typelib = {
+UNIV_INTERN
+TYPELIB innodb_checksum_algorithm_typelib = {
 	array_elements(innodb_checksum_algorithm_names) - 1,
 	"innodb_checksum_algorithm_typelib",
 	innodb_checksum_algorithm_names,
@@ -2018,7 +2020,9 @@ thd_supports_xa(
 	THD*	thd)	/*!< in: thread handle, or NULL to query
 			the global innodb_supports_xa */
 {
-	return(THDVAR(thd, support_xa));
+	/* THDVAR cannot be used in xtrabackup,
+	plugin variables  for innodb are not loaded. */
+	return (thd || !IS_XTRABACKUP())? THDVAR(thd, support_xa): FALSE;
 }
 
 /** Get the value of innodb_tmpdir.
@@ -2051,7 +2055,9 @@ thd_fake_changes(
 	THD*	thd)	/*!< in: thread handle, or NULL to query
 			the global innodb_supports_xa */
 {
-	return(THDVAR((THD*) thd, fake_changes));
+	/* THDVAR cannot be used in xtrabackup,
+	plugin variables  for innodb are not loaded */
+	return (thd || !IS_XTRABACKUP())? THDVAR((THD*) thd, fake_changes) : FALSE ;
 }
 
 /******************************************************************//**
@@ -2091,7 +2097,10 @@ thd_flush_log_at_trx_commit(
 /*================================*/
 	void*	thd)
 {
-	return(THDVAR((THD*) thd, flush_log_at_trx_commit));
+	/* THDVAR cannot be used in xtrabackup,
+	plugin variables  for innodb are not loaded,
+	this makes xtrabackup crash when trying to use them. */
+	return (thd || !IS_XTRABACKUP())? THDVAR((THD*)thd, flush_log_at_trx_commit) : FALSE;
 }
 
 /********************************************************************//**
@@ -3058,7 +3067,7 @@ trx_is_started(
 /****************************************************************//**
 Update log_checksum_algorithm_ptr with a pointer to the function corresponding
 to a given checksum algorithm. */
-static
+
 void
 innodb_log_checksum_func_update(
 /*============================*/
@@ -17397,11 +17406,11 @@ innodb_log_archive_update(
 
 	if (in_val) {
 		/* turn archiving on */
-		srv_log_archive_on = innobase_log_archive = 1;
+		innobase_log_archive = srv_log_archive_on = 1;
 		log_archive_archivelog();
 	} else {
 		/* turn archivng off */
-		srv_log_archive_on = innobase_log_archive = 0;
+		innobase_log_archive = srv_log_archive_on = 0;
 		log_archive_noarchivelog();
 	}
 }
@@ -21956,22 +21965,27 @@ ib_logf(
 	str = static_cast<char*>(malloc(BUFSIZ));
 	my_vsnprintf(str, BUFSIZ, format, args);
 #endif /* __WIN__ */
-
-	switch(level) {
-	case IB_LOG_LEVEL_INFO:
-		sql_print_information("InnoDB: %s", str);
-		break;
-	case IB_LOG_LEVEL_WARN:
-		sql_print_warning("InnoDB: %s", str);
-		break;
-	case IB_LOG_LEVEL_ERROR:
-		sql_print_error("InnoDB: %s", str);
-		sd_notifyf(0, "STATUS=InnoDB: Error: %s", str);
-		break;
-	case IB_LOG_LEVEL_FATAL:
-		sql_print_error("InnoDB: %s", str);
-		sd_notifyf(0, "STATUS=InnoDB: Fatal: %s", str);
-		break;
+	if (!IS_XTRABACKUP()) {
+		switch (level) {
+		case IB_LOG_LEVEL_INFO:
+			sql_print_information("InnoDB: %s", str);
+			break;
+		case IB_LOG_LEVEL_WARN:
+			sql_print_warning("InnoDB: %s", str);
+			break;
+		case IB_LOG_LEVEL_ERROR:
+			sql_print_error("InnoDB: %s", str);
+			sd_notifyf(0, "STATUS=InnoDB: Error: %s", str);
+			break;
+		case IB_LOG_LEVEL_FATAL:
+			sql_print_error("InnoDB: %s", str);
+			sd_notifyf(0, "STATUS=InnoDB: Fatal: %s", str);
+			break;
+		}
+	}
+	else {
+		/* Don't use server logger for XtraBackup, just print to stderr. */
+		fprintf(stderr, "InnoDB: %s\n", str);
 	}
 
 	va_end(args);
diff --git a/storage/xtradb/include/buf0buf.h b/storage/xtradb/include/buf0buf.h
index 08c3a765a8b..9b4276efaa8 100644
--- a/storage/xtradb/include/buf0buf.h
+++ b/storage/xtradb/include/buf0buf.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2013, 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 the Free Software
@@ -675,13 +675,13 @@ buf_page_is_checksum_valid_none(
 	ulint		checksum_field2)
 	MY_ATTRIBUTE((warn_unused_result));
 
-/********************************************************************//**
-Checks if a page is corrupt.
+/** Check if a page is corrupt.
 @param[in]	check_lsn		true if LSN should be checked
 @param[in]	read_buf		Page to be checked
 @param[in]	zip_size		compressed size or 0
 @param[in]	space			Pointer to tablespace
 @return	true if corrupted, false if not */
+UNIV_INTERN
 bool
 buf_page_is_corrupted(
 	bool			check_lsn,
@@ -689,15 +689,13 @@ buf_page_is_corrupted(
 	ulint			zip_size,
 	const fil_space_t* 	space)
 	MY_ATTRIBUTE((warn_unused_result));
-/********************************************************************//**
-Checks if a page is all zeroes.
-@return	TRUE if the page is all zeroes */
+/** Check if a page is all zeroes.
+@param[in]	read_buf	database page
+@param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
+@return	whether the page is all zeroes */
+UNIV_INTERN
 bool
-buf_page_is_zeroes(
-/*===============*/
-	const byte*	read_buf,	/*!< in: a database page */
-	const ulint	zip_size);	/*!< in: size of compressed page;
-					0 for uncompressed pages */
+buf_page_is_zeroes(const byte* read_buf, ulint zip_size);
 #ifndef UNIV_HOTBACKUP
 /**********************************************************************//**
 Gets the space id, page offset, and byte offset within page of a
@@ -1259,18 +1257,18 @@ buf_page_init_for_read(
 				version of the tablespace in case we have done
 				DISCARD + IMPORT */
 	ulint		offset);/*!< in: page number */
-/********************************************************************//**
-Completes an asynchronous read or write request of a file page to or from
-the buffer pool.
-@param[in,out]	bpage		pointer to the block in question
-@return DB_SUCCESS if page has been read and is not corrupted,
-DB_PAGE_CORRUPTED if page based on checksum check is corrupted,
-DB_DECRYPTION_FAILED if page post encryption checksum matches but
-after decryption normal page checksum does not match.*/
+/** Complete a read or write request of a file page to or from the buffer pool.
+@param[in,out]		bpage		Page to complete
+@return whether the operation succeeded
+@retval	DB_SUCCESS		always when writing, or if a read page was OK
+@retval	DB_PAGE_CORRUPTED	if the checksum fails on a page read
+@retval	DB_DECRYPTION_FAILED	if page post encryption checksum matches but
+				after decryption normal page checksum does
+				not match */
 UNIV_INTERN
 dberr_t
-buf_page_io_complete(
-	buf_page_t*	bpage);
+buf_page_io_complete(buf_page_t* bpage)
+	MY_ATTRIBUTE((nonnull));
 /********************************************************************//**
 Calculates a folded value of a file page address to use in the page hash
 table.
diff --git a/storage/xtradb/include/fil0fil.h b/storage/xtradb/include/fil0fil.h
index 0f7526a8e5e..2f03d2aa0f5 100644
--- a/storage/xtradb/include/fil0fil.h
+++ b/storage/xtradb/include/fil0fil.h
@@ -136,6 +136,7 @@ extern fil_addr_t	fil_addr_null;
 
 #define FIL_PAGE_DATA		38	/*!< start of the data on the page */
 /* Following are used when page compression is used */
+
 #define FIL_PAGE_COMPRESSED_SIZE 2      /*!< Number of bytes used to store
 					actual payload data size on
 					compressed pages. */
@@ -313,13 +314,21 @@ struct fil_space_t {
 	ulint		n_pending_flushes; /*!< this is positive when flushing
 				the tablespace to disk; dropping of the
 				tablespace is forbidden if this is positive */
-	ulint		n_pending_ops;/*!< this is positive when we
-				have pending operations against this
-				tablespace. The pending operations can
-				be ibuf merges or lock validation code
-				trying to read a block.
-				Dropping of the tablespace is forbidden
-				if this is positive */
+	/** Number of pending buffer pool operations accessing the tablespace
+	without holding a table lock or dict_operation_lock S-latch
+	that would prevent the table (and tablespace) from being
+	dropped. An example is change buffer merge.
+	The tablespace cannot be dropped while this is nonzero,
+	or while fil_node_t::n_pending is nonzero.
+	Protected by fil_system->mutex. */
+	ulint		n_pending_ops;
+	/** Number of pending block read or write operations
+	(when a write is imminent or a read has recently completed).
+	The tablespace object cannot be freed while this is nonzero,
+	but it can be detached from fil_system.
+	Note that fil_node_t::n_pending tracks actual pending I/O requests.
+	Protected by fil_system->mutex. */
+	ulint		n_pending_ios;
 	hash_node_t	hash;	/*!< hash chain node */
 	hash_node_t	name_hash;/*!< hash chain the name_hash table */
 #ifndef UNIV_HOTBACKUP
@@ -651,13 +660,11 @@ Used by background threads that do not necessarily hold proper locks
 for concurrency control.
 @param[in]	id	tablespace ID
 @param[in]	silent	whether to silently ignore missing tablespaces
-@param[in]	for_io	whether to look up the tablespace while performing I/O
-			(possibly executing TRUNCATE)
 @return	the tablespace
 @retval	NULL if missing or being deleted or truncated */
 UNIV_INTERN
 fil_space_t*
-fil_space_acquire_low(ulint id, bool silent, bool for_io = false)
+fil_space_acquire_low(ulint id, bool silent)
 	MY_ATTRIBUTE((warn_unused_result));
 
 /** Acquire a tablespace when it could be dropped concurrently.
@@ -670,31 +677,45 @@ for concurrency control.
 @retval	NULL if missing or being deleted or truncated */
 inline
 fil_space_t*
-fil_space_acquire(ulint id, bool for_io = false)
+fil_space_acquire(ulint id)
 {
-	return (fil_space_acquire_low(id, false, for_io));
+	return(fil_space_acquire_low(id, false));
 }
 
 /** Acquire a tablespace that may not exist.
 Used by background threads that do not necessarily hold proper locks
 for concurrency control.
 @param[in]	id	tablespace ID
-@param[in]	for_io	whether to look up the tablespace while performing I/O
-			(possibly executing TRUNCATE)
 @return	the tablespace
 @retval	NULL if missing or being deleted */
 inline
 fil_space_t*
-fil_space_acquire_silent(ulint id, bool for_io = false)
+fil_space_acquire_silent(ulint id)
 {
-	return (fil_space_acquire_low(id, true, for_io));
+	return(fil_space_acquire_low(id, true));
 }
 
 /** Release a tablespace acquired with fil_space_acquire().
 @param[in,out]	space	tablespace to release  */
+UNIV_INTERN
 void
 fil_space_release(fil_space_t* space);
 
+/** Acquire a tablespace for reading or writing a block,
+when it could be dropped concurrently.
+@param[in]	id	tablespace ID
+@return	the tablespace
+@retval	NULL if missing */
+UNIV_INTERN
+fil_space_t*
+fil_space_acquire_for_io(ulint id);
+
+/** Release a tablespace acquired with fil_space_acquire_for_io().
+@param[in,out]	space	tablespace to release  */
+UNIV_INTERN
+void
+fil_space_release_for_io(fil_space_t* space);
+
 /** Return the next fil_space_t.
 Once started, the caller must keep calling this until it returns NULL.
 fil_space_acquire() and fil_space_release() are invoked here which
@@ -703,6 +724,7 @@ blocks a concurrent operation from dropping the tablespace.
 If NULL, use the first fil_space_t on fil_system->space_list.
 @return pointer to the next fil_space_t.
 @retval NULL if this was the last  */
+UNIV_INTERN
 fil_space_t*
 fil_space_next(
 	fil_space_t*	prev_space)
@@ -716,6 +738,7 @@ blocks a concurrent operation from dropping the tablespace.
 If NULL, use the first fil_space_t on fil_system->space_list.
 @return pointer to the next fil_space_t.
 @retval NULL if this was the last*/
+UNIV_INTERN
 fil_space_t*
 fil_space_keyrotate_next(
 	fil_space_t*	prev_space)
@@ -732,12 +755,9 @@ public:
 	/** Constructor: Look up the tablespace and increment the
 	reference count if found.
 	@param[in]	space_id	tablespace ID
-	@param[in]	silent		whether not print any errors
-	@param[in]	for_io		whether to look up the tablespace
-					while performing I/O
-					(possibly executing TRUNCATE) */
-	explicit FilSpace(ulint space_id, bool silent = false, bool for_io = false)
-		: m_space(fil_space_acquire_low(space_id, silent, for_io)) {}
+	@param[in]	silent		whether not to print any errors */
+	explicit FilSpace(ulint space_id, bool silent = false)
+		: m_space(fil_space_acquire_low(space_id, silent)) {}
 
 	/** Assignment operator: This assumes that fil_space_acquire()
 	has already been done for the fil_space_t. The caller must
@@ -1042,7 +1062,7 @@ space id is != 0.
 @return	DB_SUCCESS or error number */
 UNIV_INTERN
 dberr_t
-fil_load_single_table_tablespaces(void);
+fil_load_single_table_tablespaces(ibool (*pred)(const char*, const char*)=0);
 /*===================================*/
 /*******************************************************************//**
 Returns TRUE if a single-table tablespace does not exist in the memory cache,
@@ -1081,6 +1101,9 @@ fil_space_for_table_exists_in_mem(
 					information to the .err log if a
 					matching tablespace is not found from
 					memory */
+	bool		remove_from_data_dict_if_does_not_exist,
+					/*!< in: remove from the data dictionary
+					if tablespace does not exist */
 	bool		adjust_space,	/*!< in: whether to adjust space id
 					when find table space mismatch */
 	mem_heap_t*	heap,		/*!< in: heap memory */
diff --git a/storage/xtradb/include/srv0srv.h b/storage/xtradb/include/srv0srv.h
index c18bc7c1fc3..74c118b7660 100644
--- a/storage/xtradb/include/srv0srv.h
+++ b/storage/xtradb/include/srv0srv.h
@@ -496,7 +496,9 @@ as enum type because the configure option takes unsigned integer type. */
 extern ulong	srv_innodb_stats_method;
 
 #ifdef UNIV_LOG_ARCHIVE
-extern ibool		srv_log_archive_on;
+extern bool		srv_log_archive_on;
+extern bool		srv_archive_recovery;
+extern ib_uint64_t	srv_archive_recovery_limit_lsn;
 #endif /* UNIV_LOG_ARCHIVE */
 
 extern char*	srv_file_flush_method_str;
@@ -547,6 +549,14 @@ extern ulong	srv_pass_corrupt_table;
 
 extern ulong	srv_log_checksum_algorithm;
 
+extern bool	srv_apply_log_only;
+
+extern bool	srv_backup_mode;
+extern bool	srv_close_files;
+extern bool	srv_xtrabackup;
+
+#define IS_XTRABACKUP() (srv_xtrabackup)
+
 extern my_bool	srv_force_primary_key;
 
 /* Helper macro to support srv_pass_corrupt_table checks. If 'cond' is FALSE,
diff --git a/storage/xtradb/include/trx0sys.h b/storage/xtradb/include/trx0sys.h
index 0c18b657fd7..9bfffd09532 100644
--- a/storage/xtradb/include/trx0sys.h
+++ b/storage/xtradb/include/trx0sys.h
@@ -336,8 +336,9 @@ trx_sys_update_wsrep_checkpoint(
         trx_sysf_t*     sys_header,  /*!< in: sys_header */
         mtr_t*          mtr);        /*!< in: mtr       */
 
-void
-/** Read WSREP checkpoint XID from sys header. */
+/** Read WSREP checkpoint XID from sys header. 
+@return  true on success, false on error. */
+bool
 trx_sys_read_wsrep_checkpoint(
         XID* xid); /*!< out: WSREP XID */
 #endif /* WITH_WSREP */
diff --git a/storage/xtradb/include/univ.i b/storage/xtradb/include/univ.i
index e698f08f15b..310053b9145 100644
--- a/storage/xtradb/include/univ.i
+++ b/storage/xtradb/include/univ.i
@@ -635,7 +635,7 @@ functions. */
 
 #ifdef __WIN__
 #define usleep(a) Sleep((a)/1000)
-typedef ulint os_thread_ret_t;
+typedef DWORD os_thread_ret_t;
 #define OS_THREAD_DUMMY_RETURN return(0)
 #else
 typedef void* os_thread_ret_t;
diff --git a/storage/xtradb/log/log0crypt.cc b/storage/xtradb/log/log0crypt.cc
index e6b5c845757..f6c1416d81a 100644
--- a/storage/xtradb/log/log0crypt.cc
+++ b/storage/xtradb/log/log0crypt.cc
@@ -25,8 +25,7 @@ Modified           Jan Lindström jan.lindstrom@mariadb.com
 *******************************************************/
 #include "m_string.h"
 #include "log0crypt.h"
-#include <my_crypt.h>
-#include <my_crypt.h>
+#include <mysql/service_my_crypt.h>
 
 #include "log0log.h"
 #include "srv0start.h" // for srv_start_lsn
@@ -34,8 +33,6 @@ Modified           Jan Lindström jan.lindstrom@mariadb.com
 
 #include "ha_prototypes.h" // IB_LOG_
 
-#include "my_crypt.h"
-
 /* Used for debugging */
 // #define DEBUG_CRYPT 1
 #define UNENCRYPTED_KEY_VER	0
diff --git a/storage/xtradb/log/log0log.cc b/storage/xtradb/log/log0log.cc
index 25c8ed06981..309de7daaf8 100644
--- a/storage/xtradb/log/log0log.cc
+++ b/storage/xtradb/log/log0log.cc
@@ -2629,7 +2629,7 @@ loop:
 	start_lsn += len;
 	buf += len;
 
-	if (recv_sys->report(ut_time())) {
+	if (recv_sys && recv_sys->report(ut_time())) {
 		ib_logf(IB_LOG_LEVEL_INFO, "Read redo log up to LSN=" LSN_PF,
 			start_lsn);
 		sd_notifyf(0, "STATUS=Read redo log up to LSN=" LSN_PF,
diff --git a/storage/xtradb/log/log0recv.cc b/storage/xtradb/log/log0recv.cc
index d6d31b0c572..978e6051711 100644
--- a/storage/xtradb/log/log0recv.cc
+++ b/storage/xtradb/log/log0recv.cc
@@ -692,7 +692,6 @@ recv_synchronize_groups(
 /***********************************************************************//**
 Checks the consistency of the checkpoint info
 @return	TRUE if ok */
-static
 ibool
 recv_check_cp_is_consistent(
 /*========================*/
@@ -722,7 +721,7 @@ recv_check_cp_is_consistent(
 /********************************************************//**
 Looks for the maximum consistent checkpoint from the log groups.
 @return	error code or DB_SUCCESS */
-static MY_ATTRIBUTE((nonnull, warn_unused_result))
+MY_ATTRIBUTE((nonnull, warn_unused_result))
 dberr_t
 recv_find_max_checkpoint(
 /*=====================*/
@@ -3474,6 +3473,7 @@ recv_recovery_from_checkpoint_finish(void)
 #ifdef __WIN__
 	if (recv_writer_thread_handle) {
 		CloseHandle(recv_writer_thread_handle);
+		recv_writer_thread_handle = 0;
 	}
 #endif /* __WIN__ */
 
@@ -3700,6 +3700,102 @@ recv_reset_log_files_for_backup(
 }
 #endif /* UNIV_HOTBACKUP */
 
+/******************************************************//**
+Checks the 4-byte checksum to the trailer checksum field of a log
+block.  We also accept a log block in the old format before
+InnoDB-3.23.52 where the checksum field contains the log block number.
+@return TRUE if ok, or if the log block may be in the format of InnoDB
+version predating 3.23.52 */
+UNIV_INTERN
+ibool
+log_block_checksum_is_ok_or_old_format(
+/*===================================*/
+	const byte*	block)	/*!< in: pointer to a log block */
+{
+#ifdef UNIV_LOG_DEBUG
+	return(TRUE);
+#endif /* UNIV_LOG_DEBUG */
+
+	ulint block_checksum = log_block_get_checksum(block);
+
+	if (UNIV_LIKELY(srv_log_checksum_algorithm ==
+			SRV_CHECKSUM_ALGORITHM_NONE ||
+			log_block_calc_checksum(block) == block_checksum)) {
+
+		return(TRUE);
+	}
+
+	if (srv_log_checksum_algorithm == SRV_CHECKSUM_ALGORITHM_STRICT_CRC32 ||
+	    srv_log_checksum_algorithm == SRV_CHECKSUM_ALGORITHM_STRICT_INNODB ||
+	    srv_log_checksum_algorithm == SRV_CHECKSUM_ALGORITHM_STRICT_NONE) {
+
+		const char*	algo = NULL;
+
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"log block checksum mismatch: expected " ULINTPF ", "
+			"calculated checksum " ULINTPF,
+			block_checksum,
+			log_block_calc_checksum(block));
+
+		if (block_checksum == LOG_NO_CHECKSUM_MAGIC) {
+
+			algo = "none";
+		} else if (block_checksum ==
+			   log_block_calc_checksum_crc32(block)) {
+
+			algo = "crc32";
+		} else if (block_checksum ==
+			   log_block_calc_checksum_innodb(block)) {
+
+			algo = "innodb";
+		}
+
+		if (algo) {
+
+			const char*	current_algo;
+
+			current_algo = buf_checksum_algorithm_name(
+				(srv_checksum_algorithm_t)
+				srv_log_checksum_algorithm);
+
+			ib_logf(IB_LOG_LEVEL_ERROR,
+				"current InnoDB log checksum type: %s, "
+				"detected log checksum type: %s",
+				current_algo,
+				algo);
+		}
+
+		ib_logf(IB_LOG_LEVEL_FATAL,
+			"STRICT method was specified for innodb_log_checksum, "
+			"so we intentionally assert here.");
+	}
+
+	ut_ad(srv_log_checksum_algorithm == SRV_CHECKSUM_ALGORITHM_CRC32 ||
+	      srv_log_checksum_algorithm == SRV_CHECKSUM_ALGORITHM_INNODB);
+
+	if (block_checksum == LOG_NO_CHECKSUM_MAGIC ||
+	    block_checksum == log_block_calc_checksum_crc32(block) ||
+	    block_checksum == log_block_calc_checksum_innodb(block)) {
+
+		return(TRUE);
+	}
+
+	if (log_block_get_hdr_no(block) == block_checksum) {
+
+		/* We assume the log block is in the format of
+		InnoDB version < 3.23.52 and the block is ok */
+#if 0
+		fprintf(stderr,
+			"InnoDB: Scanned old format < InnoDB-3.23.52"
+			" log block number %lu\n",
+			log_block_get_hdr_no(block));
+#endif
+		return(TRUE);
+	}
+
+	return(FALSE);
+}
+
 void recv_dblwr_t::add(byte* page)
 {
 	pages.push_back(page);
diff --git a/storage/xtradb/os/os0file.cc b/storage/xtradb/os/os0file.cc
index 03200fee80b..4f219b18428 100644
--- a/storage/xtradb/os/os0file.cc
+++ b/storage/xtradb/os/os0file.cc
@@ -130,7 +130,7 @@ UNIV_INTERN os_ib_mutex_t	os_file_seek_mutexes[OS_FILE_N_SEEK_MUTEXES];
 #define OS_AIO_MERGE_N_CONSECUTIVE	64
 
 #ifdef WITH_INNODB_DISALLOW_WRITES
-#define WAIT_ALLOW_WRITES() os_event_wait(srv_allow_writes_event)
+#define WAIT_ALLOW_WRITES() if (!IS_XTRABACKUP()) os_event_wait(srv_allow_writes_event)
 #else
 #define WAIT_ALLOW_WRITES() do { } while (0)
 #endif /* WITH_INNODB_DISALLOW_WRITES */
@@ -1001,7 +1001,6 @@ os_file_lock(
 #ifndef UNIV_HOTBACKUP
 /****************************************************************//**
 Creates the seek mutexes used in positioned reads and writes. */
-static
 void
 os_io_init_simple(void)
 /*===================*/
@@ -1640,6 +1639,10 @@ os_file_create_simple_no_error_handling_func(
 		return((os_file_t) -1);
 	}
 
+	if (IS_XTRABACKUP()) {
+		share_mode |= FILE_SHARE_DELETE | FILE_SHARE_WRITE;
+	}
+
 	file = CreateFile((LPCTSTR) name,
 			  access,
 			  share_mode,
@@ -1921,7 +1924,10 @@ os_file_create_func(
 
 	create_mode &= ~OS_FILE_ON_ERROR_NO_EXIT;
 	create_mode &= ~OS_FILE_ON_ERROR_SILENT;
-
+	if (srv_backup_mode){
+		/* Permit others to write, while I'm reading. */
+		share_mode |= FILE_SHARE_WRITE;
+	}
 	if (create_mode == OS_FILE_OPEN_RAW) {
 
 		ut_a(!srv_read_only_mode);
@@ -3525,7 +3531,7 @@ os_file_get_status(
 			fh = CreateFile(
 				(LPCTSTR) path,		// File to open
 				access,
-				0,			// No sharing
+				FILE_SHARE_READ|FILE_SHARE_WRITE,
 				NULL,			// Default security
 				OPEN_EXISTING,		// Existing file only
 				FILE_ATTRIBUTE_NORMAL,	// Normal file
diff --git a/storage/xtradb/row/row0mysql.cc b/storage/xtradb/row/row0mysql.cc
index a4141c36eda..b4d359d5009 100644
--- a/storage/xtradb/row/row0mysql.cc
+++ b/storage/xtradb/row/row0mysql.cc
@@ -4483,8 +4483,9 @@ row_drop_table_for_mysql(
 			if (!is_temp
 			    && !fil_space_for_table_exists_in_mem(
 					space_id, tablename,
-					print_msg, false, NULL, 0,
+					print_msg, IS_XTRABACKUP() && print_msg, false, NULL, 0,
 					table_flags)) {
+
 				/* This might happen if we are dropping a
 				discarded tablespace */
 				err = DB_SUCCESS;
diff --git a/storage/xtradb/srv/srv0srv.cc b/storage/xtradb/srv/srv0srv.cc
index 1c40baee04b..bf4b9124da7 100644
--- a/storage/xtradb/srv/srv0srv.cc
+++ b/storage/xtradb/srv/srv0srv.cc
@@ -203,7 +203,7 @@ performance killer causing calling thread to context switch. Besides, Innodb
 is preallocating large number (often millions) of os_events. With kernel event
 objects it takes a big chunk out of non-paged pool, which is better suited
 for tasks like IO than for storing idle event objects. */
-UNIV_INTERN ibool	srv_use_native_conditions = FALSE;
+UNIV_INTERN ibool	srv_use_native_conditions = TRUE;
 #endif /* __WIN__ */
 
 UNIV_INTERN ulint	srv_n_data_files = 0;
@@ -366,7 +366,9 @@ readahead request. */
 UNIV_INTERN ulong	srv_read_ahead_threshold	= 56;
 
 #ifdef UNIV_LOG_ARCHIVE
-UNIV_INTERN ibool		srv_log_archive_on	= FALSE;
+UNIV_INTERN bool		srv_log_archive_on;
+UNIV_INTERN bool		srv_archive_recovery;
+UNIV_INTERN ib_uint64_t	srv_archive_recovery_limit_lsn;
 #endif /* UNIV_LOG_ARCHIVE */
 
 /* This parameter is used to throttle the number of insert buffers that are
@@ -522,6 +524,12 @@ UNIV_INTERN ulong	srv_doublewrite_batch_size	= 120;
 
 UNIV_INTERN ulong	srv_replication_delay		= 0;
 
+UNIV_INTERN bool	srv_apply_log_only;
+
+UNIV_INTERN bool	srv_backup_mode;
+UNIV_INTERN bool	srv_close_files;
+UNIV_INTERN bool	srv_xtrabackup;
+
 UNIV_INTERN ulong	srv_pass_corrupt_table = 0; /* 0:disable 1:enable */
 
 UNIV_INTERN ulong	srv_log_checksum_algorithm =
diff --git a/storage/xtradb/srv/srv0start.cc b/storage/xtradb/srv/srv0start.cc
index c5c594c82a7..aa51012816d 100644
--- a/storage/xtradb/srv/srv0start.cc
+++ b/storage/xtradb/srv/srv0start.cc
@@ -140,7 +140,7 @@ SRV_SHUTDOWN_CLEANUP and then to SRV_SHUTDOWN_LAST_PHASE, and so on */
 UNIV_INTERN enum srv_shutdown_state	srv_shutdown_state = SRV_SHUTDOWN_NONE;
 
 /** Files comprising the system tablespace */
-static os_file_t	files[1000];
+os_file_t	files[1000];
 
 /** io_handler_thread parameters for thread identification */
 static ulint		n[SRV_MAX_N_IO_THREADS];
@@ -826,7 +826,7 @@ open_log_file(
 /*********************************************************************//**
 Creates or opens database data files and closes them.
 @return	DB_SUCCESS or error code */
-static MY_ATTRIBUTE((nonnull, warn_unused_result))
+MY_ATTRIBUTE((nonnull, warn_unused_result))
 dberr_t
 open_or_create_data_files(
 /*======================*/
@@ -1080,8 +1080,10 @@ skip_size_check:
 			/* This is the earliest location where we can load
 			the double write buffer. */
 			if (i == 0) {
+				/* XtraBackup never loads corrupted pages from
+				the doublewrite buffer */
 				buf_dblwr_init_or_load_pages(
-					files[i], srv_data_file_names[i], true);
+					files[i], srv_data_file_names[i], !IS_XTRABACKUP());
 			}
 
 			bool retry = true;
@@ -1365,12 +1367,15 @@ srv_undo_tablespace_open(
 /********************************************************************
 Opens the configured number of undo tablespaces.
 @return	DB_SUCCESS or error code */
-static
 dberr_t
 srv_undo_tablespaces_init(
 /*======================*/
 	ibool		create_new_db,		/*!< in: TRUE if new db being
 						created */
+	ibool		backup_mode,		/*!< in: TRUE disables reading
+						the system tablespace (used in
+						XtraBackup), FALSE is passed on
+						recovery. */
 	const ulint	n_conf_tablespaces,	/*!< in: configured undo
 						tablespaces */
 	ulint*		n_opened)		/*!< out: number of UNDO
@@ -1424,7 +1429,7 @@ srv_undo_tablespaces_init(
 	we build the undo_tablespace_ids ourselves since they don't
 	already exist. */
 
-	if (!create_new_db) {
+	if (!create_new_db && !backup_mode) {
 		n_undo_tablespaces = trx_rseg_get_n_undo_tablespaces(
 			undo_tablespace_ids);
 	} else {
@@ -2287,11 +2292,11 @@ innobase_start_or_create_for_mysql(void)
 					max_flushed_lsn = min_flushed_lsn
 						= log_get_lsn();
 					goto files_checked;
-				} else if (i < 2) {
-					/* must have at least 2 log files */
-					ib_logf(IB_LOG_LEVEL_ERROR,
-						"Only one log file found.");
-					return(err);
+				} else if (i < 2 && !IS_XTRABACKUP()) {
+						/* must have at least 2 log files */
+						ib_logf(IB_LOG_LEVEL_ERROR,
+							"Only one log file found.");
+						return(err);
 				}
 
 				/* opened all files */
@@ -2385,6 +2390,7 @@ files_checked:
 
 	err = srv_undo_tablespaces_init(
 		create_new_db,
+		FALSE,
 		srv_undo_tablespaces,
 		&srv_undo_tablespaces_open);
 
@@ -2658,6 +2664,17 @@ files_checked:
 			dict_check_tablespaces_and_store_max_id(dict_check);
 		}
 
+		if (IS_XTRABACKUP()
+			&& !srv_backup_mode
+			&& srv_read_only_mode
+			&& srv_log_file_size_requested != srv_log_file_size) {
+
+			ib_logf(IB_LOG_LEVEL_WARN,
+				"Log files size mismatch, ignored in readonly mode");
+			srv_log_file_size_requested = srv_log_file_size;
+		}
+
+
 		if (!srv_force_recovery
 		    && !recv_sys->found_corrupt_log
 		    && (srv_log_file_size_requested != srv_log_file_size
@@ -3292,7 +3309,8 @@ innobase_shutdown_for_mysql(void)
 
 	srv_was_started = FALSE;
 	srv_start_has_been_called = FALSE;
-
+	/* reset io_tid_i, in case current process does second innodb start (xtrabackup might do that).*/
+	io_tid_i = 0;
 	return(DB_SUCCESS);
 }
 #endif /* !UNIV_HOTBACKUP */
diff --git a/storage/xtradb/trx/trx0sys.cc b/storage/xtradb/trx/trx0sys.cc
index 1c4fb19430e..558fe8a2c49 100644
--- a/storage/xtradb/trx/trx0sys.cc
+++ b/storage/xtradb/trx/trx0sys.cc
@@ -404,7 +404,7 @@ trx_sys_update_wsrep_checkpoint(
 
 }
 
-void
+bool
 trx_sys_read_wsrep_checkpoint(XID* xid)
 /*===================================*/
 {
@@ -427,7 +427,7 @@ trx_sys_read_wsrep_checkpoint(XID* xid)
                 xid->formatID = -1;
                 trx_sys_update_wsrep_checkpoint(xid, sys_header, &mtr);
                 mtr_commit(&mtr);
-                return;
+                return false;
         }
 
         xid->formatID     = (int)mach_read_from_4(
@@ -444,6 +444,7 @@ trx_sys_read_wsrep_checkpoint(XID* xid)
                   XIDDATASIZE);
 
 	mtr_commit(&mtr);
+	return true;
 }
 
 #endif /* WITH_WSREP */
@@ -1339,14 +1340,17 @@ trx_sys_close(void)
 	trx_purge_sys_close();
 
 	/* Free the double write data structures. */
-	buf_dblwr_free();
+	if (buf_dblwr) {
+		buf_dblwr_free();
+	}
 
-	ut_a(UT_LIST_GET_LEN(trx_sys->ro_trx_list) == 0);
 
 	/* Only prepared transactions may be left in the system. Free them. */
 	ut_a(UT_LIST_GET_LEN(trx_sys->rw_trx_list) == trx_sys->n_prepared_trx
 	     || srv_read_only_mode
-	     || srv_force_recovery >= SRV_FORCE_NO_TRX_UNDO);
+	     || srv_force_recovery >= SRV_FORCE_NO_TRX_UNDO
+	     || (IS_XTRABACKUP() && srv_apply_log_only));
+
 
 	while ((trx = UT_LIST_GET_FIRST(trx_sys->rw_trx_list)) != NULL) {
 		trx_free_prepared(trx);
@@ -1377,10 +1381,12 @@ trx_sys_close(void)
 		UT_LIST_REMOVE(view_list, trx_sys->view_list, prev_view);
 	}
 
-	ut_a(UT_LIST_GET_LEN(trx_sys->view_list) == 0);
-	ut_a(UT_LIST_GET_LEN(trx_sys->ro_trx_list) == 0);
-	ut_a(UT_LIST_GET_LEN(trx_sys->rw_trx_list) == 0);
-	ut_a(UT_LIST_GET_LEN(trx_sys->mysql_trx_list) == 0);
+	if (!IS_XTRABACKUP() || !srv_apply_log_only) {
+		ut_a(UT_LIST_GET_LEN(trx_sys->view_list) == 0);
+		ut_a(UT_LIST_GET_LEN(trx_sys->ro_trx_list) == 0);
+		ut_a(UT_LIST_GET_LEN(trx_sys->rw_trx_list) == 0);
+		ut_a(UT_LIST_GET_LEN(trx_sys->mysql_trx_list) == 0);
+	}
 
 	mutex_free(&trx_sys->mutex);
 
@@ -1427,6 +1433,9 @@ ulint
 trx_sys_any_active_transactions(void)
 /*=================================*/
 {
+	if (IS_XTRABACKUP() && srv_apply_log_only) {
+		return(0);
+	}
 	mutex_enter(&trx_sys->mutex);
 
 	ulint	total_trx = UT_LIST_GET_LEN(trx_sys->mysql_trx_list);
diff --git a/storage/xtradb/trx/trx0trx.cc b/storage/xtradb/trx/trx0trx.cc
index ed80bafa6c6..d0cb4a883cc 100644
--- a/storage/xtradb/trx/trx0trx.cc
+++ b/storage/xtradb/trx/trx0trx.cc
@@ -721,9 +721,16 @@ trx_resurrect_insert(
 
 			if (srv_force_recovery == 0) {
 
-				trx->state = TRX_STATE_PREPARED;
-				trx_sys->n_prepared_trx++;
-				trx_sys->n_prepared_recovered_trx++;
+				/* XtraBackup should rollback prepared XA
+				transactions */
+				if (IS_XTRABACKUP()) {
+					trx->state = TRX_STATE_ACTIVE;
+				}
+				else {
+					trx->state = TRX_STATE_PREPARED;
+					trx_sys->n_prepared_trx++;
+					trx_sys->n_prepared_recovered_trx++;
+				}
 			} else {
 				fprintf(stderr,
 					"InnoDB: Since innodb_force_recovery"
@@ -790,13 +797,16 @@ trx_resurrect_update_in_prepared_state(
 
 		if (srv_force_recovery == 0) {
 			if (trx_state_eq(trx, TRX_STATE_NOT_STARTED)) {
-				trx_sys->n_prepared_trx++;
-				trx_sys->n_prepared_recovered_trx++;
+				if (!IS_XTRABACKUP()) {
+					trx_sys->n_prepared_trx++;
+					trx_sys->n_prepared_recovered_trx++;
+				}
 			} else {
 				ut_ad(trx_state_eq(trx, TRX_STATE_PREPARED));
 			}
-
-			trx->state = TRX_STATE_PREPARED;
+			/* XtraBackup should rollback prepared XA
+			transactions */
+			trx->state = IS_XTRABACKUP()?TRX_STATE_ACTIVE: TRX_STATE_PREPARED;
 		} else {
 			fprintf(stderr,
 				"InnoDB: Since innodb_force_recovery"