2 files changed, 943 insertions, 393 deletions

diff --git a/storage/innobase/mtr/mtr0log.cc b/storage/innobase/mtr/mtr0log.cc
index 82df1df63d4..944937879c2 100644
--- a/storage/innobase/mtr/mtr0log.cc
+++ b/storage/innobase/mtr/mtr0log.cc
@@ -1,6 +1,6 @@
 /*****************************************************************************
 
-Copyright (c) 1995, 2011, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 1995, 2015, Oracle and/or its affiliates. All Rights Reserved.
 
 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
@@ -27,19 +27,19 @@ Created 12/7/1995 Heikki Tuuri
 
 #ifdef UNIV_NONINL
 #include "mtr0log.ic"
-#endif
+#endif /* UNIV_NOINL */
 
 #include "buf0buf.h"
 #include "dict0dict.h"
 #include "log0recv.h"
 #include "page0page.h"
+#include "buf0dblwr.h"
 
 #ifndef UNIV_HOTBACKUP
 # include "dict0boot.h"
 
 /********************************************************//**
 Catenates n bytes to the mtr log. */
-UNIV_INTERN
 void
 mlog_catenate_string(
 /*=================*/
@@ -47,30 +47,25 @@ mlog_catenate_string(
 	const byte*	str,	/*!< in: string to write */
 	ulint		len)	/*!< in: string length */
 {
-	dyn_array_t*	mlog;
-
 	if (mtr_get_log_mode(mtr) == MTR_LOG_NONE) {
 
 		return;
 	}
 
-	mlog = &(mtr->log);
-
-	dyn_push_string(mlog, str, len);
+	mtr->get_log()->push(str, ib_uint32_t(len));
 }
 
 /********************************************************//**
 Writes the initial part of a log record consisting of one-byte item
 type and four-byte space and page numbers. Also pushes info
 to the mtr memo that a buffer page has been modified. */
-UNIV_INTERN
 void
 mlog_write_initial_log_record(
 /*==========================*/
 	const byte*	ptr,	/*!< in: pointer to (inside) a buffer
 				frame holding the file page where
 				modification is made */
-	byte		type,	/*!< in: log item type: MLOG_1BYTE, ... */
+	mlog_id_t	type,	/*!< in: log item type: MLOG_1BYTE, ... */
 	mtr_t*		mtr)	/*!< in: mini-transaction handle */
 {
 	byte*	log_ptr;
@@ -94,23 +89,22 @@ mlog_write_initial_log_record(
 
 /********************************************************//**
 Parses an initial log record written by mlog_write_initial_log_record.
-@return	parsed record end, NULL if not a complete record */
-UNIV_INTERN
+@return parsed record end, NULL if not a complete record */
 byte*
 mlog_parse_initial_log_record(
 /*==========================*/
-	byte*	ptr,	/*!< in: buffer */
-	byte*	end_ptr,/*!< in: buffer end */
-	byte*	type,	/*!< out: log record type: MLOG_1BYTE, ... */
-	ulint*	space,	/*!< out: space id */
-	ulint*	page_no)/*!< out: page number */
+	const byte*	ptr,	/*!< in: buffer */
+	const byte*	end_ptr,/*!< in: buffer end */
+	mlog_id_t*	type,	/*!< out: log record type: MLOG_1BYTE, ... */
+	ulint*		space,	/*!< out: space id */
+	ulint*		page_no)/*!< out: page number */
 {
 	if (end_ptr < ptr + 1) {
 
 		return(NULL);
 	}
 
-	*type = (byte)((ulint)*ptr & ~MLOG_SINGLE_REC_FLAG);
+	*type = (mlog_id_t)((ulint)*ptr & ~MLOG_SINGLE_REC_FLAG);
 	ut_ad(*type <= MLOG_BIGGEST_TYPE || EXTRA_CHECK_MLOG_NUMBER(*type));
 
 	ptr++;
@@ -120,36 +114,35 @@ mlog_parse_initial_log_record(
 		return(NULL);
 	}
 
-	ptr = mach_parse_compressed(ptr, end_ptr, space);
+	*space = mach_parse_compressed(&ptr, end_ptr);
 
-	if (ptr == NULL) {
-
-		return(NULL);
+	if (ptr != NULL) {
+		*page_no = mach_parse_compressed(&ptr, end_ptr);
 	}
 
-	ptr = mach_parse_compressed(ptr, end_ptr, page_no);
-
-	return(ptr);
+	return(const_cast<byte*>(ptr));
 }
 
 /********************************************************//**
 Parses a log record written by mlog_write_ulint or mlog_write_ull.
-@return	parsed record end, NULL if not a complete record or a corrupt record */
-UNIV_INTERN
+@return parsed record end, NULL if not a complete record or a corrupt record */
 byte*
 mlog_parse_nbytes(
 /*==============*/
-	ulint	type,	/*!< in: log record type: MLOG_1BYTE, ... */
-	byte*	ptr,	/*!< in: buffer */
-	byte*	end_ptr,/*!< in: buffer end */
-	byte*	page,	/*!< in: page where to apply the log record, or NULL */
-	void*	page_zip)/*!< in/out: compressed page, or NULL */
+	mlog_id_t	type,	/*!< in: log record type: MLOG_1BYTE, ... */
+	const byte*	ptr,	/*!< in: buffer */
+	const byte*	end_ptr,/*!< in: buffer end */
+	byte*		page,	/*!< in: page where to apply the log
+				record, or NULL */
+	void*		page_zip)/*!< in/out: compressed page, or NULL */
 {
 	ulint		offset;
 	ulint		val;
 	ib_uint64_t	dval;
 
 	ut_a(type <= MLOG_8BYTES);
+	ut_a(!page || !page_zip
+	     || !fil_page_index_page_check(page));
 	if (end_ptr < ptr + 2) {
 
 		return(NULL);
@@ -170,7 +163,7 @@ mlog_parse_nbytes(
 	}
 
 	if (type == MLOG_8BYTES) {
-		ptr = mach_ull_parse_compressed(ptr, end_ptr, &dval);
+		dval = mach_u64_parse_compressed(&ptr, end_ptr);
 
 		if (ptr == NULL) {
 
@@ -186,10 +179,10 @@ mlog_parse_nbytes(
 			mach_write_to_8(page + offset, dval);
 		}
 
-		return(ptr);
+		return(const_cast<byte*>(ptr));
 	}
 
-	ptr = mach_parse_compressed(ptr, end_ptr, &val);
+	val = mach_parse_compressed(&ptr, end_ptr);
 
 	if (ptr == NULL) {
 
@@ -198,7 +191,7 @@ mlog_parse_nbytes(
 
 	switch (type) {
 	case MLOG_1BYTE:
-		if (UNIV_UNLIKELY(val > 0xFFUL)) {
+		if (val > 0xFFUL) {
 			goto corrupt;
 		}
 		if (page) {
@@ -211,7 +204,7 @@ mlog_parse_nbytes(
 		}
 		break;
 	case MLOG_2BYTES:
-		if (UNIV_UNLIKELY(val > 0xFFFFUL)) {
+		if (val > 0xFFFFUL) {
 			goto corrupt;
 		}
 		if (page) {
@@ -247,20 +240,19 @@ mlog_parse_nbytes(
 		ptr = NULL;
 	}
 
-	return(ptr);
+	return(const_cast<byte*>(ptr));
 }
 
 /********************************************************//**
 Writes 1, 2 or 4 bytes to a file page. Writes the corresponding log
 record to the mini-transaction log if mtr is not NULL. */
-UNIV_INTERN
 void
 mlog_write_ulint(
 /*=============*/
-	byte*	ptr,	/*!< in: pointer where to write */
-	ulint	val,	/*!< in: value to write */
-	byte	type,	/*!< in: MLOG_1BYTE, MLOG_2BYTES, MLOG_4BYTES */
-	mtr_t*	mtr)	/*!< in: mini-transaction handle */
+	byte*		ptr,	/*!< in: pointer where to write */
+	ulint		val,	/*!< in: value to write */
+	mlog_id_t	type,	/*!< in: MLOG_1BYTE, MLOG_2BYTES, MLOG_4BYTES */
+	mtr_t*		mtr)	/*!< in: mini-transaction handle */
 {
 	switch (type) {
 	case MLOG_1BYTE:
@@ -299,7 +291,6 @@ mlog_write_ulint(
 /********************************************************//**
 Writes 8 bytes to a file page. Writes the corresponding log
 record to the mini-transaction log, only if mtr is not NULL */
-UNIV_INTERN
 void
 mlog_write_ull(
 /*===========*/
@@ -321,7 +312,7 @@ mlog_write_ull(
 			mach_write_to_2(log_ptr, page_offset(ptr));
 			log_ptr += 2;
 
-			log_ptr += mach_ull_write_compressed(log_ptr, val);
+			log_ptr += mach_u64_write_compressed(log_ptr, val);
 
 			mlog_close(mtr, log_ptr);
 		}
@@ -332,7 +323,6 @@ mlog_write_ull(
 /********************************************************//**
 Writes a string to a file page buffered in the buffer pool. Writes the
 corresponding log record to the mini-transaction log. */
-UNIV_INTERN
 void
 mlog_write_string(
 /*==============*/
@@ -352,7 +342,6 @@ mlog_write_string(
 /********************************************************//**
 Logs a write of a string to a file page buffered in the buffer pool.
 Writes the corresponding log record to the mini-transaction log. */
-UNIV_INTERN
 void
 mlog_log_string(
 /*============*/
@@ -389,8 +378,7 @@ mlog_log_string(
 
 /********************************************************//**
 Parses a log record written by mlog_write_string.
-@return	parsed record end, NULL if not a complete record */
-UNIV_INTERN
+@return parsed record end, NULL if not a complete record */
 byte*
 mlog_parse_string(
 /*==============*/
@@ -402,7 +390,9 @@ mlog_parse_string(
 	ulint	offset;
 	ulint	len;
 
-	ut_a(!page || !page_zip || fil_page_get_type(page) != FIL_PAGE_INDEX);
+	ut_a(!page || !page_zip
+	     || (fil_page_get_type(page) != FIL_PAGE_INDEX
+		 && fil_page_get_type(page) != FIL_PAGE_RTREE));
 
 	if (end_ptr < ptr + 4) {
 
@@ -414,8 +404,7 @@ mlog_parse_string(
 	len = mach_read_from_2(ptr);
 	ptr += 2;
 
-	if (UNIV_UNLIKELY(offset >= UNIV_PAGE_SIZE)
-	    || UNIV_UNLIKELY(len + offset > UNIV_PAGE_SIZE)) {
+	if (offset >= UNIV_PAGE_SIZE || len + offset > UNIV_PAGE_SIZE) {
 		recv_sys->found_corrupt_log = TRUE;
 
 		return(NULL);
@@ -441,15 +430,14 @@ mlog_parse_string(
 /********************************************************//**
 Opens a buffer for mlog, writes the initial log record and,
 if needed, the field lengths of an index.
-@return	buffer, NULL if log mode MTR_LOG_NONE */
-UNIV_INTERN
+@return buffer, NULL if log mode MTR_LOG_NONE */
 byte*
 mlog_open_and_write_index(
 /*======================*/
 	mtr_t*			mtr,	/*!< in: mtr */
 	const byte*		rec,	/*!< in: index record or page */
 	const dict_index_t*	index,	/*!< in: record descriptor */
-	byte			type,	/*!< in: log item type */
+	mlog_id_t		type,	/*!< in: log item type */
 	ulint			size)	/*!< in: requested buffer size in bytes
 					(if 0, calls mlog_close() and
 					returns NULL) */
@@ -471,25 +459,45 @@ mlog_open_and_write_index(
 	} else {
 		ulint	i;
 		ulint	n	= dict_index_get_n_fields(index);
-		/* total size needed */
 		ulint	total	= 11 + size + (n + 2) * 2;
 		ulint	alloc	= total;
-		/* allocate at most DYN_ARRAY_DATA_SIZE at a time */
-		if (alloc > DYN_ARRAY_DATA_SIZE) {
-			alloc = DYN_ARRAY_DATA_SIZE;
+
+		if (alloc > mtr_buf_t::MAX_DATA_SIZE) {
+			alloc = mtr_buf_t::MAX_DATA_SIZE;
 		}
+
+		/* For spatial index, on non-leaf page, we just keep
+		2 fields, MBR and page no. */
+		if (dict_index_is_spatial(index)
+		    && !page_is_leaf(page_align(rec))) {
+			n = DICT_INDEX_SPATIAL_NODEPTR_SIZE;
+		}
+
 		log_start = log_ptr = mlog_open(mtr, alloc);
+
 		if (!log_ptr) {
 			return(NULL); /* logging is disabled */
 		}
+
 		log_end = log_ptr + alloc;
-		log_ptr = mlog_write_initial_log_record_fast(rec, type,
-							     log_ptr, mtr);
+
+		log_ptr = mlog_write_initial_log_record_fast(
+			rec, type, log_ptr, mtr);
+
 		mach_write_to_2(log_ptr, n);
 		log_ptr += 2;
-		mach_write_to_2(log_ptr,
-				dict_index_get_n_unique_in_tree(index));
+
+		if (page_is_leaf(page_align(rec))) {
+			mach_write_to_2(
+				log_ptr, dict_index_get_n_unique_in_tree(index));
+		} else {
+			mach_write_to_2(
+				log_ptr,
+				dict_index_get_n_unique_in_tree_nonleaf(index));
+		}
+
 		log_ptr += 2;
+
 		for (i = 0; i < n; i++) {
 			dict_field_t*		field;
 			const dict_col_t*	col;
@@ -500,7 +508,7 @@ mlog_open_and_write_index(
 			len = field->fixed_len;
 			ut_ad(len < 0x7fff);
 			if (len == 0
-			    && (col->len > 255 || col->mtype == DATA_BLOB)) {
+			    && (DATA_BIG_COL(col))) {
 				/* variable-length field
 				with maximum length > 255 */
 				len = 0x7fff;
@@ -513,10 +521,13 @@ mlog_open_and_write_index(
 				ut_a(total > (ulint) (log_ptr - log_start));
 				total -= log_ptr - log_start;
 				alloc = total;
-				if (alloc > DYN_ARRAY_DATA_SIZE) {
-					alloc = DYN_ARRAY_DATA_SIZE;
+
+				if (alloc > mtr_buf_t::MAX_DATA_SIZE) {
+					alloc = mtr_buf_t::MAX_DATA_SIZE;
 				}
+
 				log_start = log_ptr = mlog_open(mtr, alloc);
+
 				if (!log_ptr) {
 					return(NULL); /* logging is disabled */
 				}
@@ -539,8 +550,7 @@ mlog_open_and_write_index(
 
 /********************************************************//**
 Parses a log record written by mlog_open_and_write_index.
-@return	parsed record end, NULL if not a complete record */
-UNIV_INTERN
+@return parsed record end, NULL if not a complete record */
 byte*
 mlog_parse_index(
 /*=============*/
@@ -570,7 +580,7 @@ mlog_parse_index(
 	} else {
 		n = n_uniq = 1;
 	}
-	table = dict_mem_table_create("LOG_DUMMY", DICT_HDR_SPACE, n,
+	table = dict_mem_table_create("LOG_DUMMY", DICT_HDR_SPACE, n, 0,
 				      comp ? DICT_TF_COMPACT : 0, 0);
 	ind = dict_mem_index_create("LOG_DUMMY", "LOG_DUMMY",
 				    DICT_HDR_SPACE, 0, n);
diff --git a/storage/innobase/mtr/mtr0mtr.cc b/storage/innobase/mtr/mtr0mtr.cc
index 5843dd80524..520608680bb 100644
--- a/storage/innobase/mtr/mtr0mtr.cc
+++ b/storage/innobase/mtr/mtr0mtr.cc
@@ -25,448 +25,988 @@ Created 11/26/1995 Heikki Tuuri
 
 #include "mtr0mtr.h"
 
-#ifdef UNIV_NONINL
-#include "mtr0mtr.ic"
-#endif
-
 #include "buf0buf.h"
 #include "buf0flu.h"
+#include "fsp0sysspace.h"
 #include "page0types.h"
 #include "mtr0log.h"
 #include "log0log.h"
+#include "row0trunc.h"
 
-#ifndef UNIV_HOTBACKUP
-# include "log0recv.h"
-
-/***************************************************//**
-Checks if a mini-transaction is dirtying a clean page.
-@return TRUE if the mtr is dirtying a clean page. */
-UNIV_INTERN
-ibool
-mtr_block_dirtied(
-/*==============*/
-	const buf_block_t*	block)	/*!< in: block being x-fixed */
-{
-	ut_ad(buf_block_get_state(block) == BUF_BLOCK_FILE_PAGE);
-	ut_ad(block->page.buf_fix_count > 0);
+#include "log0recv.h"
 
-	/* It is OK to read oldest_modification because no
-	other thread can be performing a write of it and it
-	is only during write that the value is reset to 0. */
-	return(block->page.oldest_modification == 0);
-}
+#ifdef UNIV_NONINL
+#include "mtr0mtr.ic"
+#endif /* UNIV_NONINL */
 
-/*****************************************************************//**
-Releases the item in the slot given. */
-static MY_ATTRIBUTE((nonnull))
+/** Iterate over a memo block in reverse. */
+template <typename Functor>
+struct Iterate {
+
+	/** Release specific object */
+	explicit Iterate(Functor& functor)
+		:
+		m_functor(functor)
+	{
+		/* Do nothing */
+	}
+
+	/** @return false if the functor returns false. */
+	bool operator()(mtr_buf_t::block_t* block)
+	{
+		const mtr_memo_slot_t*	start =
+			reinterpret_cast<const mtr_memo_slot_t*>(
+				block->begin());
+
+		mtr_memo_slot_t*	slot =
+			reinterpret_cast<mtr_memo_slot_t*>(
+				block->end());
+
+		ut_ad(!(block->used() % sizeof(*slot)));
+
+		while (slot-- != start) {
+
+			if (!m_functor(slot)) {
+				return(false);
+			}
+		}
+
+		return(true);
+	}
+
+	Functor&	m_functor;
+};
+
+/** Find specific object */
+struct Find {
+
+	/** Constructor */
+	Find(const void* object, ulint type)
+		:
+		m_slot(),
+		m_type(type),
+		m_object(object)
+	{
+		ut_a(object != NULL);
+	}
+
+	/** @return false if the object was found. */
+	bool operator()(mtr_memo_slot_t* slot)
+	{
+		if (m_object == slot->object && m_type == slot->type) {
+			m_slot = slot;
+			return(false);
+		}
+
+		return(true);
+	}
+
+	/** Slot if found */
+	mtr_memo_slot_t*m_slot;
+
+	/** Type of the object to look for */
+	ulint		m_type;
+
+	/** The object instance to look for */
+	const void*	m_object;
+};
+
+/** Release latches and decrement the buffer fix count.
+@param slot	memo slot */
+static
 void
-mtr_memo_slot_release_func(
-/*=======================*/
+memo_slot_release(mtr_memo_slot_t* slot)
+{
+	switch (slot->type) {
+	case MTR_MEMO_BUF_FIX:
+	case MTR_MEMO_PAGE_S_FIX:
+	case MTR_MEMO_PAGE_SX_FIX:
+	case MTR_MEMO_PAGE_X_FIX: {
+
+		buf_block_t*	block;
+
+		block = reinterpret_cast<buf_block_t*>(slot->object);
+
+		buf_block_unfix(block);
+		buf_page_release_latch(block, slot->type);
+		break;
+	}
+
+	case MTR_MEMO_S_LOCK:
+		rw_lock_s_unlock(reinterpret_cast<rw_lock_t*>(slot->object));
+		break;
+
+	case MTR_MEMO_SX_LOCK:
+		rw_lock_sx_unlock(reinterpret_cast<rw_lock_t*>(slot->object));
+		break;
+
+	case MTR_MEMO_X_LOCK:
+		rw_lock_x_unlock(reinterpret_cast<rw_lock_t*>(slot->object));
+		break;
+
 #ifdef UNIV_DEBUG
- 	mtr_t*			mtr,	/*!< in/out: mini-transaction */
+	default:
+		ut_ad(slot->type == MTR_MEMO_MODIFY);
 #endif /* UNIV_DEBUG */
-	mtr_memo_slot_t*	slot)	/*!< in: memo slot */
-{
-	void*	object = slot->object;
-	slot->object = NULL;
+	}
 
-	/* slot release is a local operation for the current mtr.
-	We must not be holding the flush_order mutex while
-	doing this. */
-	ut_ad(!log_flush_order_mutex_own());
+	slot->object = NULL;
+}
 
+/** Unfix a page, do not release the latches on the page.
+@param slot	memo slot */
+static
+void
+memo_block_unfix(mtr_memo_slot_t* slot)
+{
 	switch (slot->type) {
+	case MTR_MEMO_BUF_FIX:
 	case MTR_MEMO_PAGE_S_FIX:
 	case MTR_MEMO_PAGE_X_FIX:
+	case MTR_MEMO_PAGE_SX_FIX: {
+		buf_block_unfix(reinterpret_cast<buf_block_t*>(slot->object));
+		break;
+	}
+
+	case MTR_MEMO_S_LOCK:
+	case MTR_MEMO_X_LOCK:
+	case MTR_MEMO_SX_LOCK:
+		break;
+#ifdef UNIV_DEBUG
+	default:
+#endif /* UNIV_DEBUG */
+		break;
+	}
+}
+/** Release latches represented by a slot.
+@param slot	memo slot */
+static MY_ATTRIBUTE((nonnull))
+void
+memo_latch_release(mtr_memo_slot_t* slot)
+{
+	switch (slot->type) {
 	case MTR_MEMO_BUF_FIX:
-		buf_page_release((buf_block_t*) object, slot->type);
+	case MTR_MEMO_PAGE_S_FIX:
+	case MTR_MEMO_PAGE_SX_FIX:
+	case MTR_MEMO_PAGE_X_FIX: {
+		buf_block_t*	block;
+
+		block = reinterpret_cast<buf_block_t*>(slot->object);
+
+		memo_block_unfix(slot);
+
+		buf_page_release_latch(block, slot->type);
+
+		slot->object = NULL;
 		break;
+	}
+
 	case MTR_MEMO_S_LOCK:
-		rw_lock_s_unlock((rw_lock_t*) object);
+		rw_lock_s_unlock(reinterpret_cast<rw_lock_t*>(slot->object));
+		slot->object = NULL;
 		break;
+
 	case MTR_MEMO_X_LOCK:
-		rw_lock_x_unlock((rw_lock_t*) object);
+		rw_lock_x_unlock(reinterpret_cast<rw_lock_t*>(slot->object));
+		slot->object = NULL;
 		break;
+
+	case MTR_MEMO_SX_LOCK:
+		rw_lock_sx_unlock(reinterpret_cast<rw_lock_t*>(slot->object));
+		slot->object = NULL;
+		break;
+
 #ifdef UNIV_DEBUG
 	default:
 		ut_ad(slot->type == MTR_MEMO_MODIFY);
-		ut_ad(mtr_memo_contains(mtr, object, MTR_MEMO_PAGE_X_FIX));
+
+		slot->object = NULL;
 #endif /* UNIV_DEBUG */
 	}
 }
 
-#ifdef UNIV_DEBUG
-# define mtr_memo_slot_release(mtr, slot) mtr_memo_slot_release_func(mtr, slot)
-#else /* UNIV_DEBUG */
-# define mtr_memo_slot_release(mtr, slot) mtr_memo_slot_release_func(slot)
-#endif /* UNIV_DEBUG */
+/** Release the latches acquired by the mini-transaction. */
+struct ReleaseLatches {
 
-/**********************************************************//**
-Releases the mlocks and other objects stored in an mtr memo.
-They are released in the order opposite to which they were pushed
-to the memo. */
-static MY_ATTRIBUTE((nonnull))
-void
-mtr_memo_pop_all(
-/*=============*/
-	mtr_t*	mtr)	/*!< in/out: mini-transaction */
-{
-	ut_ad(mtr->magic_n == MTR_MAGIC_N);
-	ut_ad(mtr->state == MTR_COMMITTING); /* Currently only used in
-					     commit */
-
-	for (const dyn_block_t* block = dyn_array_get_last_block(&mtr->memo);
-	     block;
-	     block = dyn_array_get_prev_block(&mtr->memo, block)) {
-		const mtr_memo_slot_t*	start
-			= reinterpret_cast<mtr_memo_slot_t*>(
-				dyn_block_get_data(block));
-		mtr_memo_slot_t*	slot
-			= reinterpret_cast<mtr_memo_slot_t*>(
-				dyn_block_get_data(block)
-				+ dyn_block_get_used(block));
-
-		ut_ad(!(dyn_block_get_used(block) % sizeof(mtr_memo_slot_t)));
+	/** @return true always. */
+	bool operator()(mtr_memo_slot_t* slot) const
+	{
+		if (slot->object != NULL) {
+			memo_latch_release(slot);
+		}
 
-		while (slot-- != start) {
-			if (slot->object != NULL) {
-				mtr_memo_slot_release(mtr, slot);
+		return(true);
+	}
+};
+
+/** Release the latches and blocks acquired by the mini-transaction. */
+struct ReleaseAll {
+	/** @return true always. */
+	bool operator()(mtr_memo_slot_t* slot) const
+	{
+		if (slot->object != NULL) {
+			memo_slot_release(slot);
+		}
+
+		return(true);
+	}
+};
+
+/** Check that all slots have been handled. */
+struct DebugCheck {
+	/** @return true always. */
+	bool operator()(const mtr_memo_slot_t* slot) const
+	{
+		ut_a(slot->object == NULL);
+		return(true);
+	}
+};
+
+/** Release a resource acquired by the mini-transaction. */
+struct ReleaseBlocks {
+	/** Release specific object */
+	ReleaseBlocks(lsn_t start_lsn, lsn_t end_lsn, FlushObserver* observer)
+		:
+		m_end_lsn(end_lsn),
+		m_start_lsn(start_lsn),
+		m_flush_observer(observer)
+	{
+		/* Do nothing */
+	}
+
+	/** Add the modified page to the buffer flush list. */
+	void add_dirty_page_to_flush_list(mtr_memo_slot_t* slot) const
+	{
+		ut_ad(m_end_lsn > 0);
+		ut_ad(m_start_lsn > 0);
+
+		buf_block_t*	block;
+
+		block = reinterpret_cast<buf_block_t*>(slot->object);
+
+		buf_flush_note_modification(block, m_start_lsn,
+					    m_end_lsn, m_flush_observer);
+	}
+
+	/** @return true always. */
+	bool operator()(mtr_memo_slot_t* slot) const
+	{
+		if (slot->object != NULL) {
+
+			if (slot->type == MTR_MEMO_PAGE_X_FIX
+			    || slot->type == MTR_MEMO_PAGE_SX_FIX) {
+
+				add_dirty_page_to_flush_list(slot);
+
+			} else if (slot->type == MTR_MEMO_BUF_FIX) {
+
+				buf_block_t*	block;
+				block = reinterpret_cast<buf_block_t*>(
+					slot->object);
+				if (block->made_dirty_with_no_latch) {
+					add_dirty_page_to_flush_list(slot);
+					block->made_dirty_with_no_latch = false;
+				}
 			}
 		}
+
+		return(true);
+	}
+
+	/** Mini-transaction REDO start LSN */
+	lsn_t		m_end_lsn;
+
+	/** Mini-transaction REDO end LSN */
+	lsn_t		m_start_lsn;
+
+	/** Flush observer */
+	FlushObserver*	m_flush_observer;
+};
+
+class mtr_t::Command {
+public:
+	/** Constructor.
+	Takes ownership of the mtr->m_impl, is responsible for deleting it.
+	@param[in,out]	mtr	mini-transaction */
+	explicit Command(mtr_t* mtr)
+		:
+		m_locks_released()
+	{
+		init(mtr);
+	}
+
+	void init(mtr_t* mtr)
+	{
+		m_impl = &mtr->m_impl;
+		m_sync = mtr->m_sync;
+	}
+
+	/** Destructor */
+	~Command()
+	{
+		ut_ad(m_impl == 0);
 	}
+
+	/** Write the redo log record, add dirty pages to the flush list and
+	release the resources. */
+	void execute();
+
+	/** Release the blocks used in this mini-transaction. */
+	void release_blocks();
+
+	/** Release the latches acquired by the mini-transaction. */
+	void release_latches();
+
+	/** Release both the latches and blocks used in the mini-transaction. */
+	void release_all();
+
+	/** Release the resources */
+	void release_resources();
+
+	/** Append the redo log records to the redo log buffer.
+	@param[in]	len	number of bytes to write */
+	void finish_write(ulint len);
+
+private:
+	/** Prepare to write the mini-transaction log to the redo log buffer.
+	@return number of bytes to write in finish_write() */
+	ulint prepare_write();
+
+	/** true if it is a sync mini-transaction. */
+	bool			m_sync;
+
+	/** The mini-transaction state. */
+	mtr_t::Impl*		m_impl;
+
+	/** Set to 1 after the user thread releases the latches. The log
+	writer thread must wait for this to be set to 1. */
+	volatile ulint		m_locks_released;
+
+	/** Start lsn of the possible log entry for this mtr */
+	lsn_t			m_start_lsn;
+
+	/** End lsn of the possible log entry for this mtr */
+	lsn_t			m_end_lsn;
+};
+
+/** Check if a mini-transaction is dirtying a clean page.
+@return true if the mtr is dirtying a clean page. */
+bool
+mtr_t::is_block_dirtied(const buf_block_t* block)
+{
+	ut_ad(buf_block_get_state(block) == BUF_BLOCK_FILE_PAGE);
+	ut_ad(block->page.buf_fix_count > 0);
+
+	/* It is OK to read oldest_modification because no
+	other thread can be performing a write of it and it
+	is only during write that the value is reset to 0. */
+	return(block->page.oldest_modification == 0);
 }
 
-/*****************************************************************//**
-Releases the item in the slot given. */
-static
+/** Write the block contents to the REDO log */
+struct mtr_write_log_t {
+	/** Append a block to the redo log buffer.
+	@return whether the appending should continue */
+	bool operator()(const mtr_buf_t::block_t* block) const
+	{
+		log_write_low(block->begin(), block->used());
+		return(true);
+	}
+};
+
+/** Append records to the system-wide redo log buffer.
+@param[in]	log	redo log records */
 void
-mtr_memo_slot_note_modification(
-/*============================*/
-	mtr_t*			mtr,	/*!< in: mtr */
-	mtr_memo_slot_t*	slot)	/*!< in: memo slot */
+mtr_write_log(
+	const mtr_buf_t*	log)
 {
-	ut_ad(mtr->modifications);
-	ut_ad(!srv_read_only_mode);
-	ut_ad(mtr->magic_n == MTR_MAGIC_N);
+	const ulint	len = log->size();
+	mtr_write_log_t	write_log;
 
-	if (slot->object != NULL && slot->type == MTR_MEMO_PAGE_X_FIX) {
-		buf_block_t*	block = (buf_block_t*) slot->object;
+	DBUG_PRINT("ib_log",
+		   (ULINTPF " extra bytes written at " LSN_PF,
+		    len, log_sys->lsn));
 
-		ut_ad(!mtr->made_dirty || log_flush_order_mutex_own());
-		buf_flush_note_modification(block, mtr);
-	}
+	log_reserve_and_open(len);
+	log->for_each_block(write_log);
+	log_close();
 }
 
-/**********************************************************//**
-Add the modified pages to the buffer flush list. They are released
-in the order opposite to which they were pushed to the memo. NOTE! It is
-essential that the x-rw-lock on a modified buffer page is not released
-before buf_page_note_modification is called for that page! Otherwise,
-some thread might race to modify it, and the flush list sort order on
-lsn would be destroyed. */
-static
+/** Start a mini-transaction.
+@param sync		true if it is a synchronous mini-transaction
+@param read_only	true if read only mini-transaction */
 void
-mtr_memo_note_modifications(
-/*========================*/
-	mtr_t*	mtr)	/*!< in: mtr */
+mtr_t::start(bool sync, bool read_only)
 {
-	ut_ad(!srv_read_only_mode);
-	ut_ad(mtr->magic_n == MTR_MAGIC_N);
-	ut_ad(mtr->state == MTR_COMMITTING); /* Currently only used in
-					     commit */
-
-	for (const dyn_block_t* block = dyn_array_get_last_block(&mtr->memo);
-	     block;
-	     block = dyn_array_get_prev_block(&mtr->memo, block)) {
-		const mtr_memo_slot_t*	start
-			= reinterpret_cast<mtr_memo_slot_t*>(
-				dyn_block_get_data(block));
-		mtr_memo_slot_t*	slot
-			= reinterpret_cast<mtr_memo_slot_t*>(
-				dyn_block_get_data(block)
-				+ dyn_block_get_used(block));
-
-		ut_ad(!(dyn_block_get_used(block) % sizeof(mtr_memo_slot_t)));
+	UNIV_MEM_INVALID(this, sizeof(*this));
 
-		while (slot-- != start) {
-			if (slot->object != NULL) {
-				mtr_memo_slot_note_modification(mtr, slot);
-			}
-		}
-	}
+	UNIV_MEM_INVALID(&m_impl, sizeof(m_impl));
+
+	m_sync =  sync;
+
+	m_commit_lsn = 0;
+
+	new(&m_impl.m_log) mtr_buf_t();
+	new(&m_impl.m_memo) mtr_buf_t();
+
+	m_impl.m_mtr = this;
+	m_impl.m_log_mode = MTR_LOG_ALL;
+	m_impl.m_inside_ibuf = false;
+	m_impl.m_modifications = false;
+	m_impl.m_made_dirty = false;
+	m_impl.m_n_log_recs = 0;
+	m_impl.m_state = MTR_STATE_ACTIVE;
+	ut_d(m_impl.m_user_space_id = TRX_SYS_SPACE);
+	m_impl.m_user_space = NULL;
+	m_impl.m_undo_space = NULL;
+	m_impl.m_sys_space = NULL;
+	m_impl.m_flush_observer = NULL;
+	m_impl.m_trx = NULL;
+
+	ut_d(m_impl.m_magic_n = MTR_MAGIC_N);
 }
 
-/************************************************************//**
-Append the dirty pages to the flush list. */
-static
+/** Start a mini-transaction.
+@param sync		true if it is a synchronous mini-transaction
+@param read_only	true if read only mini-transaction */
 void
-mtr_add_dirtied_pages_to_flush_list(
-/*================================*/
-	mtr_t*	mtr)	/*!< in/out: mtr */
+mtr_t::start(trx_t* trx, bool sync, bool read_only)
 {
-	ut_ad(!srv_read_only_mode);
+	UNIV_MEM_INVALID(this, sizeof(*this));
 
-	/* No need to acquire log_flush_order_mutex if this mtr has
-	not dirtied a clean page. log_flush_order_mutex is used to
-	ensure ordered insertions in the flush_list. We need to
-	insert in the flush_list iff the page in question was clean
-	before modifications. */
-	if (mtr->made_dirty) {
-		log_flush_order_mutex_enter();
-	}
+	UNIV_MEM_INVALID(&m_impl, sizeof(m_impl));
 
-	/* It is now safe to release the log mutex because the
-	flush_order mutex will ensure that we are the first one
-	to insert into the flush list. */
-	log_release();
+	m_sync =  sync;
 
-	if (mtr->modifications) {
-		mtr_memo_note_modifications(mtr);
-	}
+	m_commit_lsn = 0;
 
-	if (mtr->made_dirty) {
-		log_flush_order_mutex_exit();
-	}
+	new(&m_impl.m_log) mtr_buf_t();
+	new(&m_impl.m_memo) mtr_buf_t();
+
+	m_impl.m_mtr = this;
+	m_impl.m_log_mode = MTR_LOG_ALL;
+	m_impl.m_inside_ibuf = false;
+	m_impl.m_modifications = false;
+	m_impl.m_made_dirty = false;
+	m_impl.m_n_log_recs = 0;
+	m_impl.m_state = MTR_STATE_ACTIVE;
+	ut_d(m_impl.m_user_space_id = TRX_SYS_SPACE);
+	m_impl.m_user_space = NULL;
+	m_impl.m_undo_space = NULL;
+	m_impl.m_sys_space = NULL;
+	m_impl.m_flush_observer = NULL;
+	m_impl.m_trx = trx;
+
+	ut_d(m_impl.m_magic_n = MTR_MAGIC_N);
 }
 
-/************************************************************//**
-Writes the contents of a mini-transaction log, if any, to the database log. */
-static
+/** Release the resources */
 void
-mtr_log_reserve_and_write(
-/*======================*/
-	mtr_t*	mtr)	/*!< in/out: mtr */
+mtr_t::Command::release_resources()
 {
-	dyn_array_t*	mlog;
-	ulint		data_size;
-	byte*		first_data;
+	ut_ad(m_impl->m_magic_n == MTR_MAGIC_N);
 
-	ut_ad(!srv_read_only_mode);
+	/* Currently only used in commit */
+	ut_ad(m_impl->m_state == MTR_STATE_COMMITTING);
+
+#ifdef UNIV_DEBUG
+	DebugCheck		release;
+	Iterate<DebugCheck>	iterator(release);
+
+	m_impl->m_memo.for_each_block_in_reverse(iterator);
+#endif /* UNIV_DEBUG */
+
+	/* Reset the mtr buffers */
+	m_impl->m_log.erase();
+
+	m_impl->m_memo.erase();
+
+	m_impl->m_state = MTR_STATE_COMMITTED;
 
-	mlog = &(mtr->log);
+	m_impl = 0;
+}
+
+/** Commit a mini-transaction. */
+void
+mtr_t::commit()
+{
+	ut_ad(is_active());
+	ut_ad(!is_inside_ibuf());
+	ut_ad(m_impl.m_magic_n == MTR_MAGIC_N);
+	m_impl.m_state = MTR_STATE_COMMITTING;
 
-	first_data = dyn_block_get_data(mlog);
+	/* This is a dirty read, for debugging. */
+	ut_ad(!recv_no_log_write);
 
-	if (mtr->n_log_recs > 1) {
-		mlog_catenate_ulint(mtr, MLOG_MULTI_REC_END, MLOG_1BYTE);
+	Command	cmd(this);
+
+	if (m_impl.m_modifications
+	    && (m_impl.m_n_log_recs > 0
+		|| m_impl.m_log_mode == MTR_LOG_NO_REDO)) {
+
+		ut_ad(!srv_read_only_mode
+		      || m_impl.m_log_mode == MTR_LOG_NO_REDO);
+
+		cmd.execute();
 	} else {
-		*first_data = (byte)((ulint)*first_data
-				     | MLOG_SINGLE_REC_FLAG);
+		cmd.release_all();
+		cmd.release_resources();
 	}
+}
 
-	if (mlog->heap == NULL) {
-		ulint	len;
+/** Commit a mini-transaction that did not modify any pages,
+but generated some redo log on a higher level, such as
+MLOG_FILE_NAME records and a MLOG_CHECKPOINT marker.
+The caller must invoke log_mutex_enter() and log_mutex_exit().
+This is to be used at log_checkpoint().
+@param[in]	checkpoint_lsn	the LSN of the log checkpoint  */
+void
+mtr_t::commit_checkpoint(lsn_t checkpoint_lsn)
+{
+	ut_ad(log_mutex_own());
+	ut_ad(is_active());
+	ut_ad(!is_inside_ibuf());
+	ut_ad(m_impl.m_magic_n == MTR_MAGIC_N);
+	ut_ad(get_log_mode() == MTR_LOG_ALL);
+	ut_ad(!m_impl.m_made_dirty);
+	ut_ad(m_impl.m_memo.size() == 0);
+	ut_ad(!srv_read_only_mode);
+	ut_d(m_impl.m_state = MTR_STATE_COMMITTING);
 
-		len = mtr->log_mode != MTR_LOG_NO_REDO
-			? dyn_block_get_used(mlog) : 0;
+	/* This is a dirty read, for debugging. */
+	ut_ad(!recv_no_log_write);
 
-		mtr->end_lsn = log_reserve_and_write_fast(
-			first_data, len, &mtr->start_lsn);
+	switch (m_impl.m_n_log_recs) {
+	case 0:
+		break;
+	case 1:
+		*m_impl.m_log.front()->begin() |= MLOG_SINGLE_REC_FLAG;
+		break;
+	default:
+		mlog_catenate_ulint(
+			&m_impl.m_log, MLOG_MULTI_REC_END, MLOG_1BYTE);
+	}
 
-		if (mtr->end_lsn) {
+	byte*	ptr = m_impl.m_log.push<byte*>(SIZE_OF_MLOG_CHECKPOINT);
+#if SIZE_OF_MLOG_CHECKPOINT != 9
+# error SIZE_OF_MLOG_CHECKPOINT != 9
+#endif
+	*ptr = MLOG_CHECKPOINT;
+	mach_write_to_8(ptr + 1, checkpoint_lsn);
 
-			/* Success. We have the log mutex.
-			Add pages to flush list and exit */
-			mtr_add_dirtied_pages_to_flush_list(mtr);
+	Command	cmd(this);
+	cmd.finish_write(m_impl.m_log.size());
+	cmd.release_resources();
 
-			return;
-		}
+	DBUG_PRINT("ib_log",
+		   ("MLOG_CHECKPOINT(" LSN_PF ") written at " LSN_PF,
+		    checkpoint_lsn, log_sys->lsn));
+}
+
+#ifdef UNIV_DEBUG
+/** Check if a tablespace is associated with the mini-transaction
+(needed for generating a MLOG_FILE_NAME record)
+@param[in]	space	tablespace
+@return whether the mini-transaction is associated with the space */
+bool
+mtr_t::is_named_space(ulint space) const
+{
+	ut_ad(!m_impl.m_sys_space
+	      || m_impl.m_sys_space->id == TRX_SYS_SPACE);
+	ut_ad(!m_impl.m_undo_space
+	      || m_impl.m_undo_space->id != TRX_SYS_SPACE);
+	ut_ad(!m_impl.m_user_space
+	      || m_impl.m_user_space->id != TRX_SYS_SPACE);
+	ut_ad(!m_impl.m_sys_space
+	      || m_impl.m_sys_space != m_impl.m_user_space);
+	ut_ad(!m_impl.m_sys_space
+	      || m_impl.m_sys_space != m_impl.m_undo_space);
+	ut_ad(!m_impl.m_user_space
+	      || m_impl.m_user_space != m_impl.m_undo_space);
+
+	switch (get_log_mode()) {
+	case MTR_LOG_NONE:
+	case MTR_LOG_NO_REDO:
+		return(true);
+	case MTR_LOG_ALL:
+	case MTR_LOG_SHORT_INSERTS:
+		return(m_impl.m_user_space_id == space
+		       || is_predefined_tablespace(space));
 	}
 
-	data_size = dyn_array_get_data_size(mlog);
+	ut_error;
+	return(false);
+}
+#endif /* UNIV_DEBUG */
 
-	/* Open the database log for log_write_low */
-	mtr->start_lsn = log_reserve_and_open(data_size);
+/** Acquire a tablespace X-latch.
+NOTE: use mtr_x_lock_space().
+@param[in]	space_id	tablespace ID
+@param[in]	file		file name from where called
+@param[in]	line		line number in file
+@return the tablespace object (never NULL) */
+fil_space_t*
+mtr_t::x_lock_space(ulint space_id, const char* file, ulint line)
+{
+	fil_space_t*	space;
 
-	if (mtr->log_mode == MTR_LOG_ALL) {
+	ut_ad(m_impl.m_magic_n == MTR_MAGIC_N);
+	ut_ad(is_active());
 
-		for (dyn_block_t* block = mlog;
-		     block != 0;
-		     block = dyn_array_get_next_block(mlog, block)) {
+	if (space_id == TRX_SYS_SPACE) {
+		space = m_impl.m_sys_space;
 
-			log_write_low(
-				dyn_block_get_data(block),
-				dyn_block_get_used(block));
+		if (!space) {
+			space = m_impl.m_sys_space = fil_space_get(space_id);
 		}
-
+	} else if ((space = m_impl.m_user_space) && space_id == space->id) {
+	} else if ((space = m_impl.m_undo_space) && space_id == space->id) {
+	} else if (get_log_mode() == MTR_LOG_NO_REDO) {
+		space = fil_space_get(space_id);
+		ut_ad(space->purpose == FIL_TYPE_TEMPORARY
+		      || space->purpose == FIL_TYPE_IMPORT
+		      || space->redo_skipped_count > 0
+		      || srv_is_tablespace_truncated(space->id));
 	} else {
-		ut_ad(mtr->log_mode == MTR_LOG_NONE
-		      || mtr->log_mode == MTR_LOG_NO_REDO);
-		/* Do nothing */
+		/* called from trx_rseg_create() */
+		space = m_impl.m_undo_space = fil_space_get(space_id);
 	}
 
-	mtr->end_lsn = log_close();
-
-	mtr_add_dirtied_pages_to_flush_list(mtr);
+	ut_ad(space);
+	ut_ad(space->id == space_id);
+	x_lock(&space->latch, file, line);
+	ut_ad(space->purpose == FIL_TYPE_TEMPORARY
+	      || space->purpose == FIL_TYPE_IMPORT
+	      || space->purpose == FIL_TYPE_TABLESPACE);
+	return(space);
 }
-#endif /* !UNIV_HOTBACKUP */
 
-/***************************************************************//**
-Commits a mini-transaction. */
-UNIV_INTERN
+/** Look up the system tablespace. */
 void
-mtr_commit(
-/*=======*/
-	mtr_t*	mtr)	/*!< in: mini-transaction */
+mtr_t::lookup_sys_space()
 {
-	ut_ad(mtr);
-	ut_ad(mtr->magic_n == MTR_MAGIC_N);
-	ut_ad(mtr->state == MTR_ACTIVE);
-	ut_ad(!mtr->inside_ibuf);
-	ut_d(mtr->state = MTR_COMMITTING);
+	ut_ad(!m_impl.m_sys_space);
+	m_impl.m_sys_space = fil_space_get(TRX_SYS_SPACE);
+	ut_ad(m_impl.m_sys_space);
+}
 
-#ifndef UNIV_HOTBACKUP
-	/* This is a dirty read, for debugging. */
-	ut_ad(!recv_no_log_write);
+/** Look up the user tablespace.
+@param[in]	space_id	tablespace ID */
+void
+mtr_t::lookup_user_space(ulint space_id)
+{
+	ut_ad(space_id != TRX_SYS_SPACE);
+	ut_ad(m_impl.m_user_space_id == space_id);
+	ut_ad(!m_impl.m_user_space);
+	m_impl.m_user_space = fil_space_get(space_id);
+	ut_ad(m_impl.m_user_space);
+}
 
-	if (mtr->modifications && mtr->n_log_recs) {
-		ut_ad(!srv_read_only_mode);
-		mtr_log_reserve_and_write(mtr);
+/** Set the tablespace associated with the mini-transaction
+(needed for generating a MLOG_FILE_NAME record)
+@param[in]	space	user or system tablespace */
+void
+mtr_t::set_named_space(fil_space_t* space)
+{
+	ut_ad(m_impl.m_user_space_id == TRX_SYS_SPACE);
+	ut_d(m_impl.m_user_space_id = space->id);
+	if (space->id == TRX_SYS_SPACE) {
+		ut_ad(m_impl.m_sys_space == NULL
+		      || m_impl.m_sys_space == space);
+		m_impl.m_sys_space = space;
+	} else {
+		m_impl.m_user_space = space;
 	}
-
-	mtr_memo_pop_all(mtr);
-#endif /* !UNIV_HOTBACKUP */
-
-	dyn_array_free(&(mtr->memo));
-	dyn_array_free(&(mtr->log));
-#ifdef UNIV_DEBUG_VALGRIND
-	/* Declare everything uninitialized except
-	mtr->start_lsn, mtr->end_lsn and mtr->state. */
-	{
-		lsn_t	start_lsn	= mtr->start_lsn;
-		lsn_t	end_lsn		= mtr->end_lsn;
-		UNIV_MEM_INVALID(mtr, sizeof *mtr);
-		mtr->start_lsn = start_lsn;
-		mtr->end_lsn = end_lsn;
-	}
-#endif /* UNIV_DEBUG_VALGRIND */
-	ut_d(mtr->state = MTR_COMMITTED);
 }
 
-#ifndef UNIV_HOTBACKUP
-/***************************************************//**
-Releases an object in the memo stack.
+/** Release an object in the memo stack.
 @return true if released */
-UNIV_INTERN
 bool
-mtr_memo_release(
-/*=============*/
-	mtr_t*	mtr,	/*!< in/out: mini-transaction */
-	void*	object,	/*!< in: object */
-	ulint	type)	/*!< in: object type: MTR_MEMO_S_LOCK, ... */
+mtr_t::memo_release(const void* object, ulint type)
 {
-	ut_ad(mtr->magic_n == MTR_MAGIC_N);
-	ut_ad(mtr->state == MTR_ACTIVE);
+	ut_ad(m_impl.m_magic_n == MTR_MAGIC_N);
+	ut_ad(is_active());
+
 	/* We cannot release a page that has been written to in the
 	middle of a mini-transaction. */
-	ut_ad(!mtr->modifications || type != MTR_MEMO_PAGE_X_FIX);
-
-	for (const dyn_block_t* block = dyn_array_get_last_block(&mtr->memo);
-	     block;
-	     block = dyn_array_get_prev_block(&mtr->memo, block)) {
-		const mtr_memo_slot_t*	start
-			= reinterpret_cast<mtr_memo_slot_t*>(
-				dyn_block_get_data(block));
-		mtr_memo_slot_t*	slot
-			= reinterpret_cast<mtr_memo_slot_t*>(
-				dyn_block_get_data(block)
-				+ dyn_block_get_used(block));
+	ut_ad(!m_impl.m_modifications || type != MTR_MEMO_PAGE_X_FIX);
 
-		ut_ad(!(dyn_block_get_used(block) % sizeof(mtr_memo_slot_t)));
+	Find		find(object, type);
+	Iterate<Find>	iterator(find);
 
-		while (slot-- != start) {
-			if (object == slot->object && type == slot->type) {
-				mtr_memo_slot_release(mtr, slot);
-				return(true);
-			}
-		}
+	if (!m_impl.m_memo.for_each_block_in_reverse(iterator)) {
+		memo_slot_release(find.m_slot);
+		return(true);
 	}
 
 	return(false);
 }
-#endif /* !UNIV_HOTBACKUP */
 
-/********************************************************//**
-Reads 1 - 4 bytes from a file page buffered in the buffer pool.
-@return	value read */
-UNIV_INTERN
+/** Prepare to write the mini-transaction log to the redo log buffer.
+@return number of bytes to write in finish_write() */
 ulint
-mtr_read_ulint(
-/*===========*/
-	const byte*	ptr,	/*!< in: pointer from where to read */
-	ulint		type,	/*!< in: MLOG_1BYTE, MLOG_2BYTES, MLOG_4BYTES */
-	mtr_t*		mtr MY_ATTRIBUTE((unused)))
-				/*!< in: mini-transaction handle */
+mtr_t::Command::prepare_write()
 {
-	ut_ad(mtr->state == MTR_ACTIVE);
-	ut_ad(mtr_memo_contains_page(mtr, ptr, MTR_MEMO_PAGE_S_FIX)
-	      || mtr_memo_contains_page(mtr, ptr, MTR_MEMO_PAGE_X_FIX));
+	switch (m_impl->m_log_mode) {
+	case MTR_LOG_SHORT_INSERTS:
+		ut_ad(0);
+		/* fall through (write no redo log) */
+	case MTR_LOG_NO_REDO:
+	case MTR_LOG_NONE:
+		ut_ad(m_impl->m_log.size() == 0);
+		log_mutex_enter();
+		m_end_lsn = m_start_lsn = log_sys->lsn;
+		return(0);
+	case MTR_LOG_ALL:
+		break;
+	}
 
-	return(mach_read_ulint(ptr, type));
+	ulint	len	= m_impl->m_log.size();
+	ulint	n_recs	= m_impl->m_n_log_recs;
+	ut_ad(len > 0);
+	ut_ad(n_recs > 0);
+
+	if (len > log_sys->buf_size / 2) {
+		log_buffer_extend((len + 1) * 2);
+	}
+
+	ut_ad(m_impl->m_n_log_recs == n_recs);
+
+	fil_space_t*	space = m_impl->m_user_space;
+
+	if (space != NULL && space->id <= srv_undo_tablespaces_open) {
+		/* Omit MLOG_FILE_NAME for predefined tablespaces. */
+		space = NULL;
+	}
+
+	log_mutex_enter();
+
+	log_margin_checkpoint_age(len);
+
+	if (fil_names_write_if_was_clean(space, m_impl->m_mtr)) {
+		/* This mini-transaction was the first one to modify
+		this tablespace since the latest checkpoint, so
+		some MLOG_FILE_NAME records were appended to m_log. */
+		ut_ad(m_impl->m_n_log_recs > n_recs);
+		mlog_catenate_ulint(
+			&m_impl->m_log, MLOG_MULTI_REC_END, MLOG_1BYTE);
+		len = m_impl->m_log.size();
+	} else {
+		/* This was not the first time of dirtying a
+		tablespace since the latest checkpoint. */
+
+		ut_ad(n_recs == m_impl->m_n_log_recs);
+
+		if (n_recs <= 1) {
+			ut_ad(n_recs == 1);
+
+			/* Flag the single log record as the
+			only record in this mini-transaction. */
+			*m_impl->m_log.front()->begin()
+				|= MLOG_SINGLE_REC_FLAG;
+		} else {
+			/* Because this mini-transaction comprises
+			multiple log records, append MLOG_MULTI_REC_END
+			at the end. */
+
+			mlog_catenate_ulint(
+				&m_impl->m_log, MLOG_MULTI_REC_END,
+				MLOG_1BYTE);
+			len++;
+		}
+	}
+
+	return(len);
 }
 
-#ifdef UNIV_DEBUG
-# ifndef UNIV_HOTBACKUP
-/**********************************************************//**
-Checks if memo contains the given page.
-@return	TRUE if contains */
-UNIV_INTERN
-ibool
-mtr_memo_contains_page(
-/*===================*/
-	mtr_t*		mtr,	/*!< in: mtr */
-	const byte*	ptr,	/*!< in: pointer to buffer frame */
-	ulint		type)	/*!< in: type of object */
+/** Append the redo log records to the redo log buffer
+@param[in] len	number of bytes to write */
+void
+mtr_t::Command::finish_write(
+	ulint	len)
 {
-	return(mtr_memo_contains(mtr, buf_block_align(ptr), type));
+	ut_ad(m_impl->m_log_mode == MTR_LOG_ALL);
+	ut_ad(log_mutex_own());
+	ut_ad(m_impl->m_log.size() == len);
+	ut_ad(len > 0);
+
+	if (m_impl->m_log.is_small()) {
+		const mtr_buf_t::block_t*	front = m_impl->m_log.front();
+		ut_ad(len <= front->used());
+
+		m_end_lsn = log_reserve_and_write_fast(
+			front->begin(), len, &m_start_lsn);
+
+		if (m_end_lsn > 0) {
+			return;
+		}
+	}
+
+	/* Open the database log for log_write_low */
+	m_start_lsn = log_reserve_and_open(len);
+
+	mtr_write_log_t	write_log;
+	m_impl->m_log.for_each_block(write_log);
+
+	m_end_lsn = log_close();
 }
 
-/*********************************************************//**
-Prints info of an mtr handle. */
-UNIV_INTERN
+/** Release the latches and blocks acquired by this mini-transaction */
 void
-mtr_print(
-/*======*/
-	mtr_t*	mtr)	/*!< in: mtr */
+mtr_t::Command::release_all()
 {
-	fprintf(stderr,
-		"Mini-transaction handle: memo size %lu bytes"
-		" log size %lu bytes\n",
-		(ulong) dyn_array_get_data_size(&(mtr->memo)),
-		(ulong) dyn_array_get_data_size(&(mtr->log)));
+	ReleaseAll release;
+	Iterate<ReleaseAll> iterator(release);
+
+	m_impl->m_memo.for_each_block_in_reverse(iterator);
+
+	/* Note that we have released the latches. */
+	m_locks_released = 1;
 }
-# endif /* !UNIV_HOTBACKUP */
-#endif /* UNIV_DEBUG */
 
-/**********************************************************//**
-Releases a buf_page stored in an mtr memo after a
-savepoint. */
-UNIV_INTERN
+/** Release the latches acquired by this mini-transaction */
 void
-mtr_release_buf_page_at_savepoint(
-/*=============================*/
-	mtr_t*		mtr,		/*!< in: mtr */
-	ulint		savepoint,	/*!< in: savepoint */
-	buf_block_t*	block)		/*!< in: block to release */
+mtr_t::Command::release_latches()
 {
-	mtr_memo_slot_t* slot;
-	dyn_array_t*	memo;
+	ReleaseLatches release;
+	Iterate<ReleaseLatches> iterator(release);
 
-	ut_ad(mtr);
-	ut_ad(mtr->magic_n == MTR_MAGIC_N);
-	ut_ad(mtr->state == MTR_ACTIVE);
+	m_impl->m_memo.for_each_block_in_reverse(iterator);
 
-	memo = &(mtr->memo);
+	/* Note that we have released the latches. */
+	m_locks_released = 1;
+}
 
-	ut_ad(dyn_array_get_data_size(memo) > savepoint);
+/** Release the blocks used in this mini-transaction */
+void
+mtr_t::Command::release_blocks()
+{
+	ReleaseBlocks release(m_start_lsn, m_end_lsn, m_impl->m_flush_observer);
+	Iterate<ReleaseBlocks> iterator(release);
 
-	slot = (mtr_memo_slot_t*) dyn_array_get_element(memo, savepoint);
+	m_impl->m_memo.for_each_block_in_reverse(iterator);
+}
 
-	ut_ad(slot->object == block);
-	ut_ad(slot->type == MTR_MEMO_PAGE_S_FIX ||
-	      slot->type == MTR_MEMO_PAGE_X_FIX ||
-	      slot->type == MTR_MEMO_BUF_FIX);
+/** Write the redo log record, add dirty pages to the flush list and release
+the resources. */
+void
+mtr_t::Command::execute()
+{
+	ut_ad(m_impl->m_log_mode != MTR_LOG_NONE);
 
-	buf_page_release((buf_block_t*) slot->object, slot->type);
-	slot->object = NULL;
+	if (const ulint len = prepare_write()) {
+		finish_write(len);
+	}
+
+	if (m_impl->m_made_dirty) {
+		log_flush_order_mutex_enter();
+	}
+
+	/* It is now safe to release the log mutex because the
+	flush_order mutex will ensure that we are the first one
+	to insert into the flush list. */
+	log_mutex_exit();
+
+	m_impl->m_mtr->m_commit_lsn = m_end_lsn;
+
+	release_blocks();
+
+	if (m_impl->m_made_dirty) {
+		log_flush_order_mutex_exit();
+	}
+
+	release_latches();
+
+	release_resources();
+}
+
+#ifdef UNIV_DEBUG
+/** Check if memo contains the given item.
+@return	true if contains */
+bool
+mtr_t::memo_contains(
+	mtr_buf_t*	memo,
+	const void*	object,
+	ulint		type)
+{
+	Find		find(object, type);
+	Iterate<Find>	iterator(find);
+
+	return(!memo->for_each_block_in_reverse(iterator));
+}
+
+/** Check if memo contains the given page.
+@param memo		info
+@param ptr		record
+@param type		type of
+@return	true if contains */
+bool
+mtr_t::memo_contains_page(mtr_buf_t* memo, const byte* ptr, ulint type)
+{
+	return(memo_contains(memo, buf_block_align(ptr), type));
 }
+
+/** Debug check for flags */
+struct FlaggedCheck {
+	FlaggedCheck(const void* ptr, ulint flags)
+		:
+		m_ptr(ptr),
+		m_flags(flags)
+	{
+		// Do nothing
+	}
+
+	bool operator()(const mtr_memo_slot_t* slot) const
+	{
+		if (m_ptr == slot->object && (m_flags & slot->type)) {
+			return(false);
+		}
+
+		return(true);
+	}
+
+	const void*	m_ptr;
+	ulint		m_flags;
+};
+
+/** Check if memo contains the given item.
+@param object		object to search
+@param flags		specify types of object (can be ORred) of
+			MTR_MEMO_PAGE_S_FIX ... values
+@return true if contains */
+bool
+mtr_t::memo_contains_flagged(const void* ptr, ulint flags) const
+{
+	ut_ad(m_impl.m_magic_n == MTR_MAGIC_N);
+	ut_ad(is_committing() || is_active());
+
+	FlaggedCheck		check(ptr, flags);
+	Iterate<FlaggedCheck>	iterator(check);
+
+	return(!m_impl.m_memo.for_each_block_in_reverse(iterator));
+}
+
+/** Check if memo contains the given page.
+@param ptr		buffer frame
+@param flags		specify types of object with OR of
+			MTR_MEMO_PAGE_S_FIX... values
+@return true if contains */
+bool
+mtr_t::memo_contains_page_flagged(
+	const byte*	ptr,
+	ulint		flags) const
+{
+	return(memo_contains_flagged(buf_block_align(ptr), flags));
+}
+
+/** Print info of an mtr handle. */
+void
+mtr_t::print() const
+{
+	ib::info() << "Mini-transaction handle: memo size "
+		<< m_impl.m_memo.size() << " bytes log size "
+		<< get_log()->size() << " bytes";
+}
+
+#endif /* UNIV_DEBUG */
+