# name       : innodb_fix_misc.patch
# introduced : 11 or before
# maintainer : Yasufumi
#
# Bug fix for
# http://bugs.mysql.com/56433 (always: because good for all users, and safe)
# and http://bugs.mysql.com/51325 (optional: innodb_lazy_drop_table)
# were added. They may be removed in the future when will be fixed officially.
#
#!!! notice !!!
# Any small change to this file in the main branch
# should be done or reviewed by the maintainer!
--- a/storage/innobase/buf/buf0buf.c
+++ b/storage/innobase/buf/buf0buf.c
@@ -1981,6 +1981,27 @@
 #endif /* UNIV_DEBUG || UNIV_BUF_DEBUG */
 	}
 
+	if (UNIV_UNLIKELY(bpage->space_was_being_deleted)) {
+		/* This page is obsoleted, should discard and retry */
+		rw_lock_s_unlock(&buf_pool->page_hash_latch);
+
+		mutex_enter(&buf_pool->LRU_list_mutex);
+		block_mutex = buf_page_get_mutex_enter(bpage);
+
+		if (UNIV_UNLIKELY(!block_mutex)) {
+			mutex_exit(&buf_pool->LRU_list_mutex);
+			goto lookup;
+		}
+
+		buf_LRU_free_block(bpage, TRUE, TRUE);
+
+		mutex_exit(&buf_pool->LRU_list_mutex);
+		mutex_exit(block_mutex);
+		block_mutex = NULL;
+
+		goto lookup;
+	}
+
 	if (UNIV_UNLIKELY(!bpage->zip.data)) {
 		/* There is no compressed page. */
 err_exit:
@@ -2489,6 +2510,27 @@
 		block = (buf_block_t*) buf_page_hash_get_low(
 			buf_pool, space, offset, fold);
 		if (block) {
+			if (UNIV_UNLIKELY(block->page.space_was_being_deleted)) {
+				/* This page is obsoleted, should discard and retry */
+				rw_lock_s_unlock(&buf_pool->page_hash_latch);
+
+				mutex_enter(&buf_pool->LRU_list_mutex);
+				block_mutex = buf_page_get_mutex_enter((buf_page_t*)block);
+
+				if (UNIV_UNLIKELY(!block_mutex)) {
+					mutex_exit(&buf_pool->LRU_list_mutex);
+					goto loop;
+				}
+
+				buf_LRU_free_block((buf_page_t*)block, TRUE, TRUE);
+
+				mutex_exit(&buf_pool->LRU_list_mutex);
+				mutex_exit(block_mutex);
+				block_mutex = NULL;
+
+				goto loop;
+			}
+
 			block_mutex = buf_page_get_mutex_enter((buf_page_t*)block);
 			ut_a(block_mutex);
 		}
@@ -3411,11 +3453,28 @@
 
 	fold = buf_page_address_fold(space, offset);
 
+retry:
 	//buf_pool_mutex_enter(buf_pool);
 	mutex_enter(&buf_pool->LRU_list_mutex);
 	rw_lock_x_lock(&buf_pool->page_hash_latch);
 
 	watch_page = buf_page_hash_get_low(buf_pool, space, offset, fold);
+
+	if (UNIV_UNLIKELY(watch_page && watch_page->space_was_being_deleted)) {
+		mutex_t*	block_mutex = buf_page_get_mutex_enter(watch_page);
+
+		/* This page is obsoleted, should discard and retry */
+		rw_lock_x_unlock(&buf_pool->page_hash_latch);
+		ut_a(block_mutex);
+
+		buf_LRU_free_block(watch_page, TRUE, TRUE);
+
+		mutex_exit(&buf_pool->LRU_list_mutex);
+		mutex_exit(block_mutex);
+
+		goto retry;
+	}
+
 	if (watch_page && !buf_pool_watch_is_sentinel(buf_pool, watch_page)) {
 		/* The page is already in the buffer pool. */
 		watch_page = NULL;
@@ -3546,6 +3605,7 @@
 		bpage->state	= BUF_BLOCK_ZIP_PAGE;
 		bpage->space	= space;
 		bpage->offset	= offset;
+		bpage->space_was_being_deleted = FALSE;
 
 #ifdef UNIV_DEBUG
 		bpage->in_page_hash = FALSE;
@@ -3630,6 +3690,7 @@
 
 	fold = buf_page_address_fold(space, offset);
 
+retry:
 	//buf_pool_mutex_enter(buf_pool);
 	mutex_enter(&buf_pool->LRU_list_mutex);
 	rw_lock_x_lock(&buf_pool->page_hash_latch);
@@ -3637,6 +3698,21 @@
 	block = (buf_block_t*) buf_page_hash_get_low(
 		buf_pool, space, offset, fold);
 
+	if (UNIV_UNLIKELY(block && block->page.space_was_being_deleted)) {
+		mutex_t*	block_mutex = buf_page_get_mutex_enter((buf_page_t*)block);
+
+		/* This page is obsoleted, should discard and retry */
+		rw_lock_x_unlock(&buf_pool->page_hash_latch);
+		ut_a(block_mutex);
+
+		buf_LRU_free_block((buf_page_t*)block, TRUE, TRUE);
+
+		mutex_exit(&buf_pool->LRU_list_mutex);
+		mutex_exit(block_mutex);
+
+		goto retry;
+	}
+
 	if (block
 	    && buf_page_in_file(&block->page)
 	    && !buf_pool_watch_is_sentinel(buf_pool, &block->page)) {
@@ -3990,8 +4066,11 @@
 	}
 
 	if (io_type == BUF_IO_WRITE
-	    && (buf_page_get_state(bpage) == BUF_BLOCK_ZIP_DIRTY
-		|| buf_page_get_flush_type(bpage) == BUF_FLUSH_LRU)) {
+	    && (
+#if defined UNIV_DEBUG || defined UNIV_BUF_DEBUG
+		buf_page_get_state(bpage) == BUF_BLOCK_ZIP_DIRTY ||
+#endif /* UNIV_DEBUG || UNIV_BUF_DEBUG */
+		buf_page_get_flush_type(bpage) == BUF_FLUSH_LRU)) {
 		/* to keep consistency at buf_LRU_insert_zip_clean() */
 		have_LRU_mutex = TRUE; /* optimistic */
 	}
--- a/storage/innobase/buf/buf0flu.c
+++ b/storage/innobase/buf/buf0flu.c
@@ -439,7 +439,7 @@
 
 	if (UNIV_LIKELY(bpage->in_LRU_list && buf_page_in_file(bpage))) {
 
-		return(bpage->oldest_modification == 0
+		return((bpage->oldest_modification == 0 || bpage->space_was_being_deleted)
 		       && buf_page_get_io_fix(bpage) == BUF_IO_NONE
 		       && bpage->buf_fix_count == 0);
 	}
@@ -481,6 +481,13 @@
 	    && buf_page_get_io_fix(bpage) == BUF_IO_NONE) {
 		ut_ad(bpage->in_flush_list);
 
+		if (bpage->space_was_being_deleted) {
+			/* should be removed from flush_list here */
+			/* because buf_flush_try_neighbors() cannot flush without fil_space_get_size(space) */
+			buf_flush_remove(bpage);
+			return(FALSE);
+		}
+
 		if (flush_type != BUF_FLUSH_LRU) {
 
 			return(TRUE);
--- a/storage/innobase/buf/buf0lru.c
+++ b/storage/innobase/buf/buf0lru.c
@@ -545,6 +545,62 @@
 	}
 }
 
+/******************************************************************//**
+*/
+UNIV_INTERN
+void
+buf_LRU_mark_space_was_deleted(
+/*===========================*/
+	ulint	id)	/*!< in: space id */
+{
+	ulint	i;
+
+	for (i = 0; i < srv_buf_pool_instances; i++) {
+		buf_pool_t*	buf_pool;
+		buf_page_t*	bpage;
+		buf_chunk_t*	chunk;
+		ulint		j, k;
+
+		buf_pool = buf_pool_from_array(i);
+
+		mutex_enter(&buf_pool->LRU_list_mutex);
+
+		bpage = UT_LIST_GET_FIRST(buf_pool->LRU);
+
+		while (bpage != NULL) {
+			if (buf_page_get_space(bpage) == id) {
+				bpage->space_was_being_deleted = TRUE;
+			}
+			bpage = UT_LIST_GET_NEXT(LRU, bpage);
+		}
+
+		mutex_exit(&buf_pool->LRU_list_mutex);
+
+		rw_lock_s_lock(&btr_search_latch);
+		chunk = buf_pool->chunks;
+		for (j = buf_pool->n_chunks; j--; chunk++) {
+			buf_block_t*	block	= chunk->blocks;
+			for (k = chunk->size; k--; block++) {
+				if (buf_block_get_state(block)
+				    != BUF_BLOCK_FILE_PAGE
+				    || !block->index
+				    || buf_page_get_space(&block->page) != id) {
+					continue;
+				}
+
+				rw_lock_s_unlock(&btr_search_latch);
+
+				rw_lock_x_lock(&block->lock);
+				btr_search_drop_page_hash_index(block);
+				rw_lock_x_unlock(&block->lock);
+
+				rw_lock_s_lock(&btr_search_latch);
+			}
+		}
+		rw_lock_s_unlock(&btr_search_latch);
+	}
+}
+
 #if defined UNIV_DEBUG || defined UNIV_BUF_DEBUG
 /********************************************************************//**
 Insert a compressed block into buf_pool->zip_clean in the LRU order. */
@@ -1499,6 +1555,10 @@
 		return(FALSE);
 	}
 
+	if (bpage->space_was_being_deleted && bpage->oldest_modification != 0) {
+		buf_flush_remove(bpage);
+	}
+
 #ifdef UNIV_IBUF_COUNT_DEBUG
 	ut_a(ibuf_count_get(bpage->space, bpage->offset) == 0);
 #endif /* UNIV_IBUF_COUNT_DEBUG */
--- a/storage/innobase/fil/fil0fil.c
+++ b/storage/innobase/fil/fil0fil.c
@@ -254,6 +254,7 @@
 struct fil_system_struct {
 #ifndef UNIV_HOTBACKUP
 	mutex_t		mutex;		/*!< The mutex protecting the cache */
+	mutex_t		file_extend_mutex;
 #endif /* !UNIV_HOTBACKUP */
 	hash_table_t*	spaces;		/*!< The hash table of spaces in the
 					system; they are hashed on the space
@@ -863,7 +864,7 @@
 	ut_ad(node && system);
 	ut_ad(mutex_own(&(system->mutex)));
 	ut_a(node->open);
-	ut_a(node->n_pending == 0);
+	ut_a(node->n_pending == 0 || node->space->is_being_deleted);
 	ut_a(node->n_pending_flushes == 0);
 	ut_a(node->modification_counter == node->flush_counter
 	     || srv_fast_shutdown == 2);
@@ -877,7 +878,7 @@
 	ut_a(system->n_open > 0);
 	system->n_open--;
 
-	if (node->space->purpose == FIL_TABLESPACE && !trx_sys_sys_space(node->space->id)) {
+	if (node->n_pending == 0 && node->space->purpose == FIL_TABLESPACE && !trx_sys_sys_space(node->space->id)) {
 		ut_a(UT_LIST_GET_LEN(system->LRU) > 0);
 
 		/* The node is in the LRU list, remove it */
@@ -1076,7 +1077,7 @@
 	ut_ad(node && system && space);
 	ut_ad(mutex_own(&(system->mutex)));
 	ut_a(node->magic_n == FIL_NODE_MAGIC_N);
-	ut_a(node->n_pending == 0);
+	ut_a(node->n_pending == 0 || space->is_being_deleted);
 
 	if (node->open) {
 		/* We fool the assertion in fil_node_close_file() to think
@@ -1598,6 +1599,8 @@
 
 	mutex_create(fil_system_mutex_key,
 		     &fil_system->mutex, SYNC_ANY_LATCH);
+	mutex_create(fil_system_mutex_key,
+		     &fil_system->file_extend_mutex, SYNC_OUTER_ANY_LATCH);
 
 	fil_system->spaces = hash_create(hash_size);
 	fil_system->name_hash = hash_create(hash_size);
@@ -2352,7 +2355,11 @@
 	completely and permanently. The flag is_being_deleted also prevents
 	fil_flush() from being applied to this tablespace. */
 
+	if (srv_lazy_drop_table) {
+		buf_LRU_mark_space_was_deleted(id);
+	} else {
 	buf_LRU_invalidate_tablespace(id);
+	}
 #endif
 	/* printf("Deleting tablespace %s id %lu\n", space->name, id); */
 
@@ -4739,6 +4746,10 @@
 	ulint		page_size;
 	ibool		success		= TRUE;
 
+	/* file_extend_mutex is for http://bugs.mysql.com/56433 */
+	/* to protect from the other fil_extend_space_to_desired_size() */
+	/* during temprary releasing &fil_system->mutex */
+	mutex_enter(&fil_system->file_extend_mutex);
 	fil_mutex_enter_and_prepare_for_io(space_id);
 
 	space = fil_space_get_by_id(space_id);
@@ -4750,6 +4761,7 @@
 		*actual_size = space->size;
 
 		mutex_exit(&fil_system->mutex);
+		mutex_exit(&fil_system->file_extend_mutex);
 
 		return(TRUE);
 	}
@@ -4782,6 +4794,8 @@
 		offset_low  = ((start_page_no - file_start_page_no)
 			       % (4096 * ((1024 * 1024) / page_size)))
 			* page_size;
+
+		mutex_exit(&fil_system->mutex);
 #ifdef UNIV_HOTBACKUP
 		success = os_file_write(node->name, node->handle, buf,
 					offset_low, offset_high,
@@ -4791,8 +4805,10 @@
 				 node->name, node->handle, buf,
 				 offset_low, offset_high,
 				 page_size * n_pages,
-				 NULL, NULL, NULL);
+				 NULL, NULL, space_id, NULL);
 #endif
+		mutex_enter(&fil_system->mutex);
+
 		if (success) {
 			node->size += n_pages;
 			space->size += n_pages;
@@ -4838,6 +4854,7 @@
 	printf("Extended %s to %lu, actual size %lu pages\n", space->name,
 	size_after_extend, *actual_size); */
 	mutex_exit(&fil_system->mutex);
+	mutex_exit(&fil_system->file_extend_mutex);
 
 	fil_flush(space_id, TRUE);
 
@@ -5200,6 +5217,22 @@
 		srv_data_written+= len;
 	}
 
+	/* if the table space was already deleted, space might not exist already. */
+	if (message
+	    && space_id < SRV_LOG_SPACE_FIRST_ID
+	    && ((buf_page_t*)message)->space_was_being_deleted) {
+
+		if (mode == OS_AIO_NORMAL) {
+			buf_page_io_complete(message);
+			return(DB_SUCCESS); /*fake*/
+		}
+		if (type == OS_FILE_READ) {
+			return(DB_TABLESPACE_DELETED);
+		} else {
+			return(DB_SUCCESS); /*fake*/
+		}
+	}
+
 	/* Reserve the fil_system mutex and make sure that we can open at
 	least one file while holding it, if the file is not already open */
 
@@ -5341,10 +5374,24 @@
 #else
 	/* Queue the aio request */
 	ret = os_aio(type, mode | wake_later, node->name, node->handle, buf,
-		     offset_low, offset_high, len, node, message, trx);
+		     offset_low, offset_high, len, node, message, space_id, trx);
 #endif
 	} /**/
 
+	/* if the table space was already deleted, space might not exist already. */
+	if (message
+	    && space_id < SRV_LOG_SPACE_FIRST_ID
+	    && ((buf_page_t*)message)->space_was_being_deleted) {
+
+		if (mode == OS_AIO_SYNC) {
+			if (type == OS_FILE_READ) {
+				return(DB_TABLESPACE_DELETED);
+			} else {
+				return(DB_SUCCESS); /*fake*/
+			}
+		}
+	}
+
 	ut_a(ret);
 
 	if (mode == OS_AIO_SYNC) {
@@ -5444,6 +5491,7 @@
 	fil_node_t*	fil_node;
 	void*		message;
 	ulint		type;
+	ulint		space_id = 0;
 
 	ut_ad(fil_validate_skip());
 
@@ -5451,10 +5499,10 @@
 		srv_set_io_thread_op_info(segment, "native aio handle");
 #ifdef WIN_ASYNC_IO
 		ret = os_aio_windows_handle(segment, 0, &fil_node,
-					    &message, &type);
+					    &message, &type, &space_id);
 #elif defined(LINUX_NATIVE_AIO)
 		ret = os_aio_linux_handle(segment, &fil_node,
-					  &message, &type);
+					  &message, &type, &space_id);
 #else
 		ut_error;
 		ret = 0; /* Eliminate compiler warning */
@@ -5463,7 +5511,22 @@
 		srv_set_io_thread_op_info(segment, "simulated aio handle");
 
 		ret = os_aio_simulated_handle(segment, &fil_node,
-					      &message, &type);
+					      &message, &type, &space_id);
+	}
+
+	/* if the table space was already deleted, fil_node might not exist already. */
+	if (message
+	    && space_id < SRV_LOG_SPACE_FIRST_ID
+	    && ((buf_page_t*)message)->space_was_being_deleted) {
+
+		/* intended not to be uncompress read page */
+		ut_a(buf_page_get_io_fix(message) == BUF_IO_WRITE
+		     || !buf_page_get_zip_size(message)
+		     || buf_page_get_state(message) != BUF_BLOCK_FILE_PAGE);
+
+		srv_set_io_thread_op_info(segment, "complete io for buf page");
+		buf_page_io_complete(message);
+		return;
 	}
 
 	ut_a(ret);
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -12189,6 +12189,12 @@
   "except for the deletion.",
   NULL, NULL, 0, &corrupt_table_action_typelib);
 
+static MYSQL_SYSVAR_ULONG(lazy_drop_table, srv_lazy_drop_table,
+  PLUGIN_VAR_RQCMDARG,
+  "At deleting tablespace, only miminum needed processes at the time are done. "
+  "e.g. for http://bugs.mysql.com/51325",
+  NULL, NULL, 0, 0, 1, 0);
+
 static struct st_mysql_sys_var* innobase_system_variables[]= {
   MYSQL_SYSVAR(page_size),
   MYSQL_SYSVAR(log_block_size),
@@ -12286,6 +12292,7 @@
   MYSQL_SYSVAR(flush_checkpoint_debug),
 #endif
   MYSQL_SYSVAR(corrupt_table_action),
+  MYSQL_SYSVAR(lazy_drop_table),
   NULL
 };
 
@@ -12295,7 +12302,7 @@
   &innobase_storage_engine,
   innobase_hton_name,
   plugin_author,
-  "Supports transactions, row-level locking, and foreign keys",
+  "Percona-XtraDB, Supports transactions, row-level locking, and foreign keys",
   PLUGIN_LICENSE_GPL,
   innobase_init, /* Plugin Init */
   NULL, /* Plugin Deinit */
--- a/storage/innobase/include/buf0buf.h
+++ b/storage/innobase/include/buf0buf.h
@@ -1475,6 +1475,7 @@
 					0 if the block was never accessed
 					in the buffer pool */
 	/* @} */
+	ibool		space_was_being_deleted;
 	ibool		is_corrupt;
 # if defined UNIV_DEBUG_FILE_ACCESSES || defined UNIV_DEBUG
 	ibool		file_page_was_freed;
--- a/storage/innobase/include/buf0buf.ic
+++ b/storage/innobase/include/buf0buf.ic
@@ -426,6 +426,7 @@
 	buf_block_set_state(block, BUF_BLOCK_FILE_PAGE);
 	block->page.space = space;
 	block->page.offset = page_no;
+	block->page.space_was_being_deleted = FALSE;
 }
 
 /*********************************************************************//**
--- a/storage/innobase/include/buf0lru.h
+++ b/storage/innobase/include/buf0lru.h
@@ -73,6 +73,14 @@
 buf_LRU_invalidate_tablespace(
 /*==========================*/
 	ulint	id);	/*!< in: space id */
+
+/******************************************************************//**
+*/
+UNIV_INTERN
+void
+buf_LRU_mark_space_was_deleted(
+/*===========================*/
+	ulint	id);	/*!< in: space id */
 #if defined UNIV_DEBUG || defined UNIV_BUF_DEBUG
 /********************************************************************//**
 Insert a compressed block into buf_pool->zip_clean in the LRU order. */
--- a/storage/innobase/include/os0file.h
+++ b/storage/innobase/include/os0file.h
@@ -280,9 +280,9 @@
 	pfs_os_file_close_func(file, __FILE__, __LINE__)
 
 # define os_aio(type, mode, name, file, buf, offset, offset_high,	\
-		n, message1, message2, trx)				\
+		n, message1, message2, space_id, trx)			\
 	pfs_os_aio_func(type, mode, name, file, buf, offset,		\
-			offset_high, n, message1, message2, trx,	\
+			offset_high, n, message1, message2, space_id, trx,\
 			__FILE__, __LINE__)
 
 # define os_file_read(file, buf, offset, offset_high, n)		\
@@ -326,9 +326,9 @@
 # define os_file_close(file)	os_file_close_func(file)
 
 # define os_aio(type, mode, name, file, buf, offset, offset_high,	\
-	       n, message1, message2, trx)				\
+	       n, message1, message2, space_id, trx)			\
 	os_aio_func(type, mode, name, file, buf, offset, offset_high, n,\
-		    message1, message2, trx)
+		    message1, message2, space_id, trx)
 
 # define os_file_read(file, buf, offset, offset_high, n)		\
 	os_file_read_func(file, buf, offset, offset_high, n, NULL)
@@ -757,6 +757,7 @@
 				(can be used to identify a completed
 				aio operation); ignored if mode is
                                 OS_AIO_SYNC */
+	ulint		space_id,
 	trx_t*		trx,
 	const char*	src_file,/*!< in: file name where func invoked */
 	ulint		src_line);/*!< in: line where the func invoked */
@@ -1065,6 +1066,7 @@
 				(can be used to identify a completed
 				aio operation); ignored if mode is
 				OS_AIO_SYNC */
+	ulint		space_id,
 	trx_t*		trx);
 /************************************************************************//**
 Wakes up all async i/o threads so that they know to exit themselves in
@@ -1125,7 +1127,8 @@
 				parameters are valid and can be used to
 				restart the operation, for example */
 	void**	message2,
-	ulint*	type);		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	type,		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	space_id);
 #endif
 
 /**********************************************************************//**
@@ -1147,7 +1150,8 @@
 				parameters are valid and can be used to
 				restart the operation, for example */
 	void**	message2,
-	ulint*	type);		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	type,		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	space_id);
 /**********************************************************************//**
 Validates the consistency of the aio system.
 @return	TRUE if ok */
@@ -1226,7 +1230,8 @@
 				aio operation failed, these output
 				parameters are valid and can be used to
 				restart the operation. */
-	ulint*	type);		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	type,		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	space_id);
 #endif /* LINUX_NATIVE_AIO */
 
 #ifndef UNIV_NONINL
--- a/storage/innobase/include/os0file.ic
+++ b/storage/innobase/include/os0file.ic
@@ -229,6 +229,7 @@
 				(can be used to identify a completed
 				aio operation); ignored if mode is
                                 OS_AIO_SYNC */
+	ulint		space_id,
 	trx_t*		trx,
 	const char*	src_file,/*!< in: file name where func invoked */
 	ulint		src_line)/*!< in: line where the func invoked */
@@ -245,7 +246,7 @@
 				   src_file, src_line);
 
 	result = os_aio_func(type, mode, name, file, buf, offset, offset_high,
-			     n, message1, message2, trx);
+			     n, message1, message2, space_id, trx);
 
 	register_pfs_file_io_end(locker, n);
 
--- a/storage/innobase/include/srv0srv.h
+++ b/storage/innobase/include/srv0srv.h
@@ -249,6 +249,8 @@
 extern ulint	srv_pass_corrupt_table;
 
 extern ulint	srv_dict_size_limit;
+
+extern ulint	srv_lazy_drop_table;
 /*-------------------------------------------*/
 
 extern ulint	srv_n_rows_inserted;
--- a/storage/innobase/include/sync0sync.h
+++ b/storage/innobase/include/sync0sync.h
@@ -689,6 +689,7 @@
 #define	SYNC_BUF_POOL		150	/* Buffer pool mutex */
 #define	SYNC_BUF_FLUSH_LIST	145	/* Buffer flush list mutex */
 #define SYNC_DOUBLEWRITE	140
+#define	SYNC_OUTER_ANY_LATCH	136
 #define	SYNC_ANY_LATCH		135
 #define	SYNC_MEM_HASH		131
 #define	SYNC_MEM_POOL		130
--- a/storage/innobase/include/univ.i
+++ b/storage/innobase/include/univ.i
@@ -53,6 +53,11 @@
 #define INNODB_VERSION_MINOR	1
 #define INNODB_VERSION_BUGFIX	8
 
+#ifndef PERCONA_INNODB_VERSION
+#define PERCONA_INNODB_VERSION 20.1
+#endif
+
+
 /* The following is the InnoDB version as shown in
 SELECT plugin_version FROM information_schema.plugins;
 calculated in make_version_string() in sql/sql_show.cc like this:
@@ -65,7 +70,8 @@
 #define INNODB_VERSION_STR			\
 	IB_TO_STR(INNODB_VERSION_MAJOR) "."	\
 	IB_TO_STR(INNODB_VERSION_MINOR) "."	\
-	IB_TO_STR(INNODB_VERSION_BUGFIX)
+	IB_TO_STR(INNODB_VERSION_BUGFIX) "-"	\
+	IB_TO_STR(PERCONA_INNODB_VERSION)
 
 #define REFMAN "http://dev.mysql.com/doc/refman/"	\
 	IB_TO_STR(MYSQL_MAJOR_VERSION) "."		\
--- a/storage/innobase/os/os0file.c
+++ b/storage/innobase/os/os0file.c
@@ -180,6 +180,7 @@
 					made and only the slot message
 					needs to be passed to the caller
 					of os_aio_simulated_handle */
+	ulint		space_id;
 	fil_node_t*	message1;	/*!< message which is given by the */
 	void*		message2;	/*!< the requester of an aio operation
 					and which can be used to identify
@@ -3686,7 +3687,8 @@
 				offset */
 	ulint		offset_high, /*!< in: most significant 32 bits of
 				offset */
-	ulint		len)	/*!< in: length of the block to read or write */
+	ulint		len,	/*!< in: length of the block to read or write */
+	ulint		space_id)
 {
 	os_aio_slot_t*	slot = NULL;
 #ifdef WIN_ASYNC_IO
@@ -3775,6 +3777,7 @@
 	slot->offset   = offset;
 	slot->offset_high = offset_high;
 	slot->io_already_done = FALSE;
+	slot->space_id = space_id;
 
 #ifdef WIN_ASYNC_IO
 	control = &(slot->control);
@@ -4062,6 +4065,7 @@
 				(can be used to identify a completed
 				aio operation); ignored if mode is
 				OS_AIO_SYNC */
+	ulint		space_id,
 	trx_t*		trx)
 {
 	os_aio_array_t*	array;
@@ -4157,7 +4161,7 @@
 		trx->io_read += n;
 	}
 	slot = os_aio_array_reserve_slot(type, array, message1, message2, file,
-					 name, buf, offset, offset_high, n);
+					 name, buf, offset, offset_high, n, space_id);
 	if (type == OS_FILE_READ) {
 		if (srv_use_native_aio) {
 			os_n_file_reads++;
@@ -4276,7 +4280,8 @@
 				parameters are valid and can be used to
 				restart the operation, for example */
 	void**	message2,
-	ulint*	type)		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	type,		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	space_id)
 {
 	ulint		orig_seg	= segment;
 	os_aio_array_t*	array;
@@ -4355,6 +4360,7 @@
 	*message2 = slot->message2;
 
 	*type = slot->type;
+	*space_id = slot->space_id;
 
 	if (ret && len == slot->len) {
 		ret_val = TRUE;
@@ -4583,7 +4589,8 @@
 				aio operation failed, these output
 				parameters are valid and can be used to
 				restart the operation. */
-	ulint*	type)		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	type,		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	space_id)
 {
 	ulint		segment;
 	os_aio_array_t*	array;
@@ -4656,6 +4663,7 @@
 	*message2 = slot->message2;
 
 	*type = slot->type;
+	*space_id = slot->space_id;
 
 	if ((slot->ret == 0) && (slot->n_bytes == (long)slot->len)) {
 		ret = TRUE;
@@ -4709,7 +4717,8 @@
 				parameters are valid and can be used to
 				restart the operation, for example */
 	void**	message2,
-	ulint*	type)		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	type,		/*!< out: OS_FILE_WRITE or ..._READ */
+	ulint*	space_id)
 {
 	os_aio_array_t*	array;
 	ulint		segment;
@@ -5005,6 +5014,7 @@
 	*message2 = slot->message2;
 
 	*type = slot->type;
+	*space_id = slot->space_id;
 
 	os_mutex_exit(array->mutex);
 
--- a/storage/innobase/row/row0mysql.c
+++ b/storage/innobase/row/row0mysql.c
@@ -51,6 +51,7 @@
 #include "btr0sea.h"
 #include "fil0fil.h"
 #include "ibuf0ibuf.h"
+#include "ha_prototypes.h"
 
 /** Provide optional 4.x backwards compatibility for 5.0 and above */
 UNIV_INTERN ibool	row_rollback_on_timeout	= FALSE;
@@ -1192,6 +1193,13 @@
 
 	thr = que_fork_get_first_thr(prebuilt->ins_graph);
 
+	if (!prebuilt->mysql_has_locked && !(prebuilt->table->flags & (DICT_TF2_TEMPORARY << DICT_TF2_SHIFT))) {
+		fprintf(stderr, "InnoDB: Error: row_insert_for_mysql is called without ha_innobase::external_lock()\n");
+		if (trx->mysql_thd != NULL) {
+			innobase_mysql_print_thd(stderr, trx->mysql_thd, 600);
+		}
+	}
+
 	if (prebuilt->sql_stat_start) {
 		node->state = INS_NODE_SET_IX_LOCK;
 		prebuilt->sql_stat_start = FALSE;
@@ -2576,10 +2584,29 @@
 
 			err = DB_ERROR;
 		} else {
+			dict_index_t*	index;
+
 			/* Set the flag which tells that now it is legal to
 			IMPORT a tablespace for this table */
 			table->tablespace_discarded = TRUE;
 			table->ibd_file_missing = TRUE;
+
+			/* check adaptive hash entries */
+			index = dict_table_get_first_index(table);
+			while (index) {
+				ulint ref_count = btr_search_info_get_ref_count(index->search_info);
+				if (ref_count) {
+					fprintf(stderr, "InnoDB: Warning:"
+						" hash index ref_count (%lu) is not zero"
+						" after fil_discard_tablespace().\n"
+						"index: \"%s\""
+						" table: \"%s\"\n",
+						ref_count,
+						index->name,
+						table->name);
+				}
+				index = dict_table_get_next_index(index);
+			}
 		}
 	}
 
@@ -2928,6 +2955,19 @@
 			table->space = space;
 			index = dict_table_get_first_index(table);
 			do {
+				ulint ref_count = btr_search_info_get_ref_count(index->search_info);
+				/* check adaptive hash entries */
+				if (ref_count) {
+					fprintf(stderr, "InnoDB: Warning:"
+						" hash index ref_count (%lu) is not zero"
+						" after fil_discard_tablespace().\n"
+						"index: \"%s\""
+						" table: \"%s\"\n",
+						ref_count,
+						index->name,
+						table->name);
+				}
+
 				index->space = space;
 				index = dict_table_get_next_index(index);
 			} while (index);
--- a/storage/innobase/row/row0sel.c
+++ b/storage/innobase/row/row0sel.c
@@ -3418,6 +3418,7 @@
 	ulint		offsets_[REC_OFFS_NORMAL_SIZE];
 	ulint*		offsets				= offsets_;
 	ibool		table_lock_waited		= FALSE;
+	ibool		problematic_use			= FALSE;
 
 	rec_offs_init(offsets_);
 
@@ -3796,6 +3797,17 @@
 
 	/* Do some start-of-statement preparations */
 
+	if (!prebuilt->mysql_has_locked) {
+		if (!(prebuilt->table->flags & (DICT_TF2_TEMPORARY << DICT_TF2_SHIFT))) {
+			fprintf(stderr, "InnoDB: Error: row_search_for_mysql() is called without ha_innobase::external_lock()\n");
+			if (trx->mysql_thd != NULL) {
+				innobase_mysql_print_thd(stderr, trx->mysql_thd, 600);
+			}
+		}
+		problematic_use = TRUE;
+	}
+retry_check:
+	
 	if (!prebuilt->sql_stat_start) {
 		/* No need to set an intention lock or assign a read view */
 
@@ -3806,6 +3818,18 @@
 			      " perform a consistent read\n"
 			      "InnoDB: but the read view is not assigned!\n",
 			      stderr);
+			if (problematic_use) {
+				fprintf(stderr, "InnoDB: It may be caused by calling "
+						"without ha_innobase::external_lock()\n"
+						"InnoDB: For the first-aid, avoiding the crash. "
+						"But it should be fixed ASAP.\n");
+				if (prebuilt->table->flags & (DICT_TF2_TEMPORARY << DICT_TF2_SHIFT)
+				    && trx->mysql_thd != NULL) {
+					innobase_mysql_print_thd(stderr, trx->mysql_thd, 600);
+				}
+				prebuilt->sql_stat_start = TRUE;
+				goto retry_check;
+			}
 			trx_print(stderr, trx, 600);
 			fputc('\n', stderr);
 			ut_error;
--- a/storage/innobase/srv/srv0srv.c
+++ b/storage/innobase/srv/srv0srv.c
@@ -447,6 +447,8 @@
 UNIV_INTERN ulint	srv_pass_corrupt_table = 0; /* 0:disable 1:enable */
 
 UNIV_INTERN ulint	srv_dict_size_limit = 0;
+
+UNIV_INTERN ulint	srv_lazy_drop_table = 0;
 /*-------------------------------------------*/
 UNIV_INTERN ulong	srv_n_spin_wait_rounds	= 30;
 UNIV_INTERN ulong	srv_n_free_tickets_to_enter = 500;
--- a/storage/innobase/srv/srv0start.c
+++ b/storage/innobase/srv/srv0start.c
@@ -2187,7 +2187,7 @@
 	if (srv_print_verbose_log) {
 		ut_print_timestamp(stderr);
 		fprintf(stderr,
-			" InnoDB: %s started; "
+			" Percona XtraDB (http://www.percona.com) %s started; "
 			"log sequence number %llu\n",
 			INNODB_VERSION_STR, srv_start_lsn);
 	}
--- a/storage/innobase/sync/sync0sync.c
+++ b/storage/innobase/sync/sync0sync.c
@@ -1219,6 +1219,7 @@
 	case SYNC_LOG:
 	case SYNC_LOG_FLUSH_ORDER:
 	case SYNC_ANY_LATCH:
+	case SYNC_OUTER_ANY_LATCH:
 	case SYNC_FILE_FORMAT_TAG:
 	case SYNC_DOUBLEWRITE:
 	case SYNC_SEARCH_SYS:
--- a/storage/innobase/trx/trx0purge.c
+++ b/storage/innobase/trx/trx0purge.c
@@ -1149,8 +1149,7 @@
 	/* If we cannot advance the 'purge view' because of an old
 	'consistent read view', then the DML statements cannot be delayed.
 	Also, srv_max_purge_lag <= 0 means 'infinity'. */
-	if (srv_max_purge_lag > 0
-	    && !UT_LIST_GET_LAST(trx_sys->view_list)) {
+	if (srv_max_purge_lag > 0) {
 		float	ratio = (float) trx_sys->rseg_history_len
 			/ srv_max_purge_lag;
 		if (ratio > ULINT_MAX / 10000) {