1 files changed, 181 insertions, 236 deletions

diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index eb7d0967901..52130b50eb1 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -52,96 +52,6 @@ const dtuple_t trx_undo_metadata = {
 
 /*=========== UNDO LOG RECORD CREATION AND DECODING ====================*/
 
-/** Write redo log of writing an undo log record.
-@param[in]	undo_block	undo log page
-@param[in]	old_free	start offset of the undo log record
-@param[in]	new_free	end offset of the undo log record
-@param[in,out]	mtr		mini-transaction */
-static void trx_undof_page_add_undo_rec_log(const buf_block_t* undo_block,
-					    ulint old_free, ulint new_free,
-					    mtr_t* mtr)
-{
-	ut_ad(old_free >= TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE);
-	ut_ad(new_free >= old_free);
-	ut_ad(new_free < srv_page_size);
-	ut_ad(mach_read_from_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE
-			       + undo_block->frame)
-	      == new_free);
-	mtr->set_modified();
-	switch (mtr->get_log_mode()) {
-	case MTR_LOG_NONE:
-	case MTR_LOG_NO_REDO:
-		return;
-	case MTR_LOG_SHORT_INSERTS:
-		ut_ad(0);
-		/* fall through */
-	case MTR_LOG_ALL:
-		break;
-	}
-
-	const uint32_t
-		len = uint32_t(new_free - old_free - 4),
-		reserved = std::min<uint32_t>(11 + 13 + len,
-					      mtr->get_log()->MAX_DATA_SIZE);
-	byte* log_ptr = mtr->get_log()->open(reserved);
-	const byte* log_end = log_ptr + reserved;
-	log_ptr = mlog_write_initial_log_record_low(
-		MLOG_UNDO_INSERT,
-		undo_block->page.id.space(), undo_block->page.id.page_no(),
-		log_ptr, mtr);
-	mach_write_to_2(log_ptr, len);
-	if (log_ptr + 2 + len <= log_end) {
-		memcpy(log_ptr + 2, undo_block->frame + old_free + 2, len);
-		mlog_close(mtr, log_ptr + 2 + len);
-	} else {
-		mlog_close(mtr, log_ptr + 2);
-		mtr->get_log()->push(undo_block->frame + old_free + 2, len);
-	}
-}
-
-/** Parse MLOG_UNDO_INSERT.
-@param[in]	ptr	log record
-@param[in]	end_ptr	end of log record buffer
-@param[in,out]	page	page or NULL
-@return	end of log record
-@retval	NULL	if the log record is incomplete */
-byte*
-trx_undo_parse_add_undo_rec(
-	const byte*	ptr,
-	const byte*	end_ptr,
-	page_t*		page)
-{
-	ulint	len;
-
-	if (end_ptr < ptr + 2) {
-
-		return(NULL);
-	}
-
-	len = mach_read_from_2(ptr);
-	ptr += 2;
-
-	if (end_ptr < ptr + len) {
-
-		return(NULL);
-	}
-
-	if (page) {
-		ulint first_free = mach_read_from_2(page + TRX_UNDO_PAGE_HDR
-						    + TRX_UNDO_PAGE_FREE);
-		byte* rec = page + first_free;
-
-		mach_write_to_2(rec, first_free + 4 + len);
-		mach_write_to_2(rec + 2 + len, first_free);
-
-		mach_write_to_2(page + TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE,
-				first_free + 4 + len);
-		memcpy(rec + 2, ptr, len);
-	}
-
-	return(const_cast<byte*>(ptr + len));
-}
-
 /** Calculate the free space left for extending an undo log record.
 @param[in]	undo_block	undo log page
 @param[in]	ptr		current end of the undo page
@@ -160,7 +70,7 @@ that was written to ptr. Update the first free value by the number of bytes
 written for this undo record.
 @return offset of the inserted entry on the page if succeeded, 0 if fail */
 static
-ulint
+uint16_t
 trx_undo_page_set_next_prev_and_add(
 /*================================*/
 	buf_block_t*	undo_block,	/*!< in/out: undo log page */
@@ -168,42 +78,31 @@ trx_undo_page_set_next_prev_and_add(
 					written on this undo page. */
 	mtr_t*		mtr)		/*!< in: mtr */
 {
-	ulint		first_free;	/*!< offset within undo_page */
-	ulint		end_of_rec;	/*!< offset within undo_page */
-	byte*		ptr_to_first_free;
-					/* pointer within undo_page
-					that points to the next free
-					offset value within undo_page.*/
+  ut_ad(page_align(ptr) == undo_block->frame);
 
-	ut_ad(ptr > undo_block->frame);
-	ut_ad(ptr < undo_block->frame + srv_page_size);
+  if (UNIV_UNLIKELY(trx_undo_left(undo_block, ptr) < 2))
+    return 0;
 
-	if (UNIV_UNLIKELY(trx_undo_left(undo_block, ptr) < 2)) {
-		return(0);
-	}
-
-	ptr_to_first_free = TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE
-		+ undo_block->frame;
+  byte *ptr_to_first_free= my_assume_aligned<2>(TRX_UNDO_PAGE_HDR +
+						TRX_UNDO_PAGE_FREE +
+						undo_block->frame);
 
-	first_free = mach_read_from_2(ptr_to_first_free);
+  const uint16_t first_free= mach_read_from_2(ptr_to_first_free);
 
-	/* Write offset of the previous undo log record */
-	mach_write_to_2(ptr, first_free);
-	ptr += 2;
+  /* Write offset of the previous undo log record */
+  memcpy(ptr, ptr_to_first_free, 2);
+  ptr += 2;
 
-	end_of_rec = ulint(ptr - undo_block->frame);
+  const uint16_t end_of_rec= static_cast<uint16_t>(ptr - undo_block->frame);
 
-	/* Write offset of the next undo log record */
-	mach_write_to_2(undo_block->frame + first_free, end_of_rec);
+  /* Update the offset to first free undo record */
+  mach_write_to_2(ptr_to_first_free, end_of_rec);
+  /* Write offset of the next undo log record */
+  memcpy(undo_block->frame + first_free, ptr_to_first_free, 2);
+  const byte *start= undo_block->frame + first_free + 2;
 
-	/* Update the offset to first free undo record */
-	mach_write_to_2(ptr_to_first_free, end_of_rec);
-
-	/* Write this log entry to the UNDO log */
-	trx_undof_page_add_undo_rec_log(undo_block, first_free,
-					end_of_rec, mtr);
-
-	return(first_free);
+  mtr->undo_append(*undo_block, start, ptr - start - 2);
+  return first_free;
 }
 
 /** Virtual column undo log version. To distinguish it from a length value
@@ -289,7 +188,7 @@ trx_undo_log_v_idx(
 indexed, and return its position
 @param[in]	table		the table
 @param[in]	ptr		undo log pointer
-@param[out]	col_pos		the column number or ULINT_UNDEFINED
+@param[out]	col_pos		the column number or FIL_NULL
 				if the column is not indexed any more
 @return remaining part of undo log record after reading these values */
 static
@@ -297,12 +196,12 @@ const byte*
 trx_undo_read_v_idx_low(
 	const dict_table_t*	table,
 	const byte*		ptr,
-	ulint*			col_pos)
+	uint32_t*		col_pos)
 {
 	ulint		len = mach_read_from_2(ptr);
 	const byte*	old_ptr = ptr;
 
-	*col_pos = ULINT_UNDEFINED;
+	*col_pos = FIL_NULL;
 
 	ptr += 2;
 
@@ -350,7 +249,7 @@ still indexed, and output its position
 				check to see if this is undo log. When
 				first_v_col is true, is_undo_log is output,
 				when first_v_col is false, is_undo_log is input
-@param[in,out]	field_no	the column number
+@param[out]	field_no	the column number, or FIL_NULL if not indexed
 @return remaining part of undo log record after reading these values */
 const byte*
 trx_undo_read_v_idx(
@@ -358,7 +257,7 @@ trx_undo_read_v_idx(
 	const byte*		ptr,
 	bool			first_v_col,
 	bool*			is_undo_log,
-	ulint*			field_no)
+	uint32_t*		field_no)
 {
 	/* Version marker only put on the first virtual column */
 	if (first_v_col) {
@@ -471,7 +370,7 @@ trx_undo_report_insert_virtual(
 Reports in the undo log of an insert of a clustered index record.
 @return offset of the inserted entry on the page if succeed, 0 if fail */
 static
-ulint
+uint16_t
 trx_undo_page_report_insert(
 /*========================*/
 	buf_block_t*	undo_block,	/*!< in: undo log page */
@@ -481,11 +380,7 @@ trx_undo_page_report_insert(
 					inserted to the clustered index */
 	mtr_t*		mtr)		/*!< in: mtr */
 {
-	ulint		first_free;
-	byte*		ptr;
-	ulint		i;
-
-	ut_ad(dict_index_is_clust(index));
+	ut_ad(index->is_primary());
 	/* MariaDB 10.3.1+ in trx_undo_page_init() always initializes
 	TRX_UNDO_PAGE_TYPE as 0, but previous versions wrote
 	TRX_UNDO_INSERT == 1 into insert_undo pages,
@@ -493,11 +388,14 @@ trx_undo_page_report_insert(
 	ut_ad(mach_read_from_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_TYPE
 			       + undo_block->frame) <= 2);
 
-	first_free = mach_read_from_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE
-				      + undo_block->frame);
-	ptr = undo_block->frame + first_free;
+	uint16_t first_free = mach_read_from_2(my_assume_aligned<2>
+					       (TRX_UNDO_PAGE_HDR
+						+ TRX_UNDO_PAGE_FREE
+						+ undo_block->frame));
+	byte* ptr = undo_block->frame + first_free;
 
-	ut_ad(first_free <= srv_page_size);
+	ut_ad(first_free >= TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE);
+	ut_ad(first_free <= srv_page_size - FIL_PAGE_DATA_END);
 
 	if (trx_undo_left(undo_block, ptr) < 2 + 1 + 11 + 11) {
 		/* Not enough space for writing the general parameters */
@@ -523,7 +421,7 @@ trx_undo_page_report_insert(
 		goto done;
 	}
 
-	for (i = 0; i < dict_index_get_n_unique(index); i++) {
+	for (unsigned i = 0; i < dict_index_get_n_unique(index); i++) {
 
 		const dfield_t*	field	= dtuple_get_nth_field(clust_entry, i);
 		ulint		flen	= dfield_get_len(field);
@@ -586,12 +484,14 @@ trx_undo_rec_get_pars(
 
 	*updated_extern = !!(type_cmpl & TRX_UNDO_UPD_EXTERN);
 	type_cmpl &= ~TRX_UNDO_UPD_EXTERN;
-
 	*type = type_cmpl & (TRX_UNDO_CMPL_INFO_MULT - 1);
+	ut_ad(*type >= TRX_UNDO_RENAME_TABLE);
+	ut_ad(*type <= TRX_UNDO_DEL_MARK_REC);
 	*cmpl_info = type_cmpl / TRX_UNDO_CMPL_INFO_MULT;
 
 	*undo_no = mach_read_next_much_compressed(&ptr);
 	*table_id = mach_read_next_much_compressed(&ptr);
+	ut_ad(*table_id);
 
 	return(const_cast<byte*>(ptr));
 }
@@ -607,8 +507,8 @@ byte*
 trx_undo_rec_get_col_val(
 	const byte*	ptr,
 	const byte**	field,
-	ulint*		len,
-	ulint*		orig_len)
+	uint32_t*	len,
+	uint32_t*	orig_len)
 {
 	*len = mach_read_next_compressed(&ptr);
 	*orig_len = 0;
@@ -681,8 +581,7 @@ trx_undo_rec_get_row_ref(
 
 	for (i = 0; i < ref_len; i++) {
 		const byte*	field;
-		ulint		len;
-		ulint		orig_len;
+		uint32_t	len, orig_len;
 
 		dfield_t* dfield = dtuple_get_nth_field(tuple, i);
 
@@ -715,8 +614,7 @@ trx_undo_rec_skip_row_ref(
 
 	for (i = 0; i < ref_len; i++) {
 		const byte*	field;
-		ulint		len;
-		ulint		orig_len;
+		uint32_t len, orig_len;
 
 		ptr = trx_undo_rec_get_col_val(ptr, &field, &len, &orig_len);
 	}
@@ -870,7 +768,7 @@ record.
 @return byte offset of the inserted undo log entry on the page if
 succeed, 0 if fail */
 static
-ulint
+uint16_t
 trx_undo_page_report_modify(
 /*========================*/
 	buf_block_t*	undo_block,	/*!< in: undo log page */
@@ -889,9 +787,6 @@ trx_undo_page_report_modify(
 					virtual column info */
 	mtr_t*		mtr)		/*!< in: mtr */
 {
-	ulint		first_free;
-	byte*		ptr;
-
 	ut_ad(index->is_primary());
 	ut_ad(rec_offs_validate(rec, index, offsets));
 	/* MariaDB 10.3.1+ in trx_undo_page_init() always initializes
@@ -901,11 +796,15 @@ trx_undo_page_report_modify(
 	ut_ad(mach_read_from_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_TYPE
 			       + undo_block->frame) <= 2);
 
-	first_free = mach_read_from_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE
-				      + undo_block->frame);
-	ptr = undo_block->frame + first_free;
+	byte* ptr_to_first_free = my_assume_aligned<2>(TRX_UNDO_PAGE_HDR
+						       + TRX_UNDO_PAGE_FREE
+						       + undo_block->frame);
 
-	ut_ad(first_free <= srv_page_size);
+	const uint16_t first_free = mach_read_from_2(ptr_to_first_free);
+	byte *ptr = undo_block->frame + first_free;
+
+	ut_ad(first_free >= TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE);
+	ut_ad(first_free <= srv_page_size - FIL_PAGE_DATA_END);
 
 	if (trx_undo_left(undo_block, ptr) < 50) {
 		/* NOTE: the value 50 must be big enough so that the general
@@ -1493,15 +1392,14 @@ already_logged:
 	}
 
 	mach_write_to_2(ptr, first_free);
-	ptr += 2;
-	const ulint new_free = ulint(ptr - undo_block->frame);
+	const uint16_t new_free = static_cast<uint16_t>(
+		ptr + 2 - undo_block->frame);
 	mach_write_to_2(undo_block->frame + first_free, new_free);
 
-	mach_write_to_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE
-			+ undo_block->frame, new_free);
+	mach_write_to_2(ptr_to_first_free, new_free);
 
-	/* Write to the REDO log about this change in the UNDO log */
-	trx_undof_page_add_undo_rec_log(undo_block, first_free, new_free, mtr);
+	const byte* start = &undo_block->frame[first_free + 2];
+	mtr->undo_append(*undo_block, start, ptr - start);
 	return(first_free);
 }
 
@@ -1517,11 +1415,10 @@ trx_undo_update_rec_get_sys_cols(
 					general parameters */
 	trx_id_t*	trx_id,		/*!< out: trx id */
 	roll_ptr_t*	roll_ptr,	/*!< out: roll ptr */
-	ulint*		info_bits)	/*!< out: info bits state */
+	byte*		info_bits)	/*!< out: info bits state */
 {
 	/* Read the state of the info bits */
-	*info_bits = mach_read_from_1(ptr);
-	ptr += 1;
+	*info_bits = *ptr++;
 
 	/* Read the values of the system columns */
 
@@ -1552,7 +1449,7 @@ trx_undo_update_rec_get_update(
 				the update vector */
 	trx_id_t	trx_id,	/*!< in: transaction id from this undo record */
 	roll_ptr_t	roll_ptr,/*!< in: roll pointer from this undo record */
-	ulint		info_bits,/*!< in: info bits from this undo record */
+	byte		info_bits,/*!< in: info bits from this undo record */
 	mem_heap_t*	heap,	/*!< in: memory heap from which the memory
 				needed is allocated */
 	upd_t**		upd)	/*!< out, own: update vector */
@@ -1600,12 +1497,11 @@ trx_undo_update_rec_get_update(
 	/* Store then the updated ordinary columns to the update vector */
 
 	for (ulint i = 0; i < n_fields; i++) {
-		const byte*	field;
-		ulint		len;
-		ulint		orig_len;
+		const byte* field;
+		uint32_t len, orig_len;
 
 		upd_field = upd_get_nth_field(update, i);
-		ulint field_no = mach_read_next_compressed(&ptr);
+		uint32_t field_no = mach_read_next_compressed(&ptr);
 
 		const bool is_virtual = (field_no >= REC_MAX_N_FIELDS);
 
@@ -1617,7 +1513,7 @@ trx_undo_update_rec_get_update(
 				&field_no);
 			first_v_col = false;
 			/* This column could be dropped or no longer indexed */
-			if (field_no == ULINT_UNDEFINED) {
+			if (field_no >= index->n_fields) {
 				/* Mark this is no longer needed */
 				upd_field->field_no = REC_MAX_N_FIELDS;
 
@@ -1629,12 +1525,14 @@ trx_undo_update_rec_get_update(
 				continue;
 			}
 
-			upd_field_set_v_field_no(upd_field, field_no, index);
+			upd_field_set_v_field_no(
+				upd_field, static_cast<uint16_t>(field_no),
+				index);
 		} else if (UNIV_UNLIKELY((update->info_bits
 					  & ~REC_INFO_DELETED_FLAG)
 					 == REC_INFO_MIN_REC_FLAG)) {
 			ut_ad(type == TRX_UNDO_UPD_EXIST_REC);
-			const ulint uf = index->first_user_field();
+			const uint32_t uf = index->first_user_field();
 			ut_ad(field_no >= uf);
 
 			if (update->info_bits != REC_INFO_MIN_REC_FLAG) {
@@ -1684,9 +1582,12 @@ trx_undo_update_rec_get_update(
 							       field_no),
 					&upd_field->new_val.type);
 			}
-			upd_field->field_no = field_no;
+			upd_field->field_no = field_no
+				& dict_index_t::MAX_N_FIELDS;
 		} else if (field_no < index->n_fields) {
-			upd_field_set_field_no(upd_field, field_no, index);
+			upd_field_set_field_no(upd_field,
+					       static_cast<uint16_t>(field_no),
+					       index);
 		} else {
 			ib::error() << "Trying to access update undo rec"
 				" field " << field_no
@@ -1706,7 +1607,7 @@ trx_undo_update_rec_get_update(
 
 		ptr = trx_undo_rec_get_col_val(ptr, &field, &len, &orig_len);
 
-		upd_field->orig_len = orig_len;
+		upd_field->orig_len = static_cast<uint16_t>(orig_len);
 
 		if (len == UNIV_SQL_NULL) {
 			dfield_set_null(&upd_field->new_val);
@@ -1826,15 +1727,13 @@ trx_undo_rec_get_partial_row(
 	while (ptr != end_ptr) {
 		dfield_t*	dfield;
 		const byte*	field;
-		ulint		field_no;
+		uint32_t	field_no;
 		const dict_col_t* col;
-		ulint		len;
-		ulint		orig_len;
-		bool		is_virtual;
+		uint32_t len, orig_len;
 
 		field_no = mach_read_next_compressed(&ptr);
 
-		is_virtual = (field_no >= REC_MAX_N_FIELDS);
+		const bool is_virtual = (field_no >= REC_MAX_N_FIELDS);
 
 		if (is_virtual) {
 			ptr = trx_undo_read_v_idx(
@@ -1846,7 +1745,7 @@ trx_undo_rec_get_partial_row(
 		ptr = trx_undo_rec_get_col_val(ptr, &field, &len, &orig_len);
 
 		/* This column could be dropped or no longer indexed */
-		if (field_no == ULINT_UNDEFINED) {
+		if (field_no == FIL_NULL) {
 			ut_ad(is_virtual);
 			continue;
 		}
@@ -1943,22 +1842,6 @@ trx_undo_rec_get_partial_row(
 	return(const_cast<byte*>(ptr));
 }
 
-/** Erase the unused undo log page end.
-@param[in,out]	undo_page	undo log page
-@return whether the page contained something */
-bool
-trx_undo_erase_page_end(page_t* undo_page)
-{
-	ulint	first_free;
-
-	first_free = mach_read_from_2(undo_page + TRX_UNDO_PAGE_HDR
-				      + TRX_UNDO_PAGE_FREE);
-	memset(undo_page + first_free, 0,
-	       (srv_page_size - FIL_PAGE_DATA_END) - first_free);
-
-	return(first_free != TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE);
-}
-
 /** Report a RENAME TABLE operation.
 @param[in,out]	trx	transaction
 @param[in]	table	table that is being renamed
@@ -1967,16 +1850,17 @@ trx_undo_erase_page_end(page_t* undo_page)
 @return	byte offset of the undo log record
 @retval	0	in case of failure */
 static
-ulint
+uint16_t
 trx_undo_page_report_rename(trx_t* trx, const dict_table_t* table,
 			    buf_block_t* block, mtr_t* mtr)
 {
-	byte*	ptr_first_free  = TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE
-		+ block->frame;
-	ulint	first_free = mach_read_from_2(ptr_first_free);
+	byte*	ptr_first_free  = my_assume_aligned<2>(TRX_UNDO_PAGE_HDR
+						       + TRX_UNDO_PAGE_FREE
+						       + block->frame);
+	const uint16_t first_free = mach_read_from_2(ptr_first_free);
 	ut_ad(first_free >= TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE);
-	ut_ad(first_free <= srv_page_size);
-	byte* start = block->frame + first_free;
+	ut_ad(first_free <= srv_page_size - FIL_PAGE_DATA_END);
+	byte* const start = block->frame + first_free;
 	size_t len = strlen(table->name.m_name);
 	const size_t fixed = 2 + 1 + 11 + 11 + 2;
 	ut_ad(len <= NAME_LEN * 2 + 1);
@@ -1998,12 +1882,9 @@ trx_undo_page_report_rename(trx_t* trx, const dict_table_t* table,
 	memcpy(ptr, table->name.m_name, len);
 	ptr += len;
 	mach_write_to_2(ptr, first_free);
-	ptr += 2;
-	ulint offset = page_offset(ptr);
-	mach_write_to_2(start, offset);
-	mach_write_to_2(ptr_first_free, offset);
-
-	trx_undof_page_add_undo_rec_log(block, first_free, offset, mtr);
+	mach_write_to_2(ptr_first_free, ptr + 2 - block->frame);
+	memcpy(start, ptr_first_free, 2);
+	mtr->undo_append(*block, start + 2, ptr - start - 2);
 	return first_free;
 }
 
@@ -2026,9 +1907,10 @@ dberr_t trx_undo_report_rename(trx_t* trx, const dict_table_t* table)
 		ut_ad(undo);
 		for (ut_d(int loop_count = 0);;) {
 			ut_ad(loop_count++ < 2);
-			ut_ad(undo->last_page_no == block->page.id.page_no());
+			ut_ad(undo->last_page_no
+			      == block->page.id().page_no());
 
-			if (ulint offset = trx_undo_page_report_rename(
+			if (uint16_t offset = trx_undo_page_report_rename(
 				    trx, table, block, &mtr)) {
 				undo->top_page_no = undo->last_page_no;
 				undo->top_offset  = offset;
@@ -2130,7 +2012,7 @@ trx_undo_report_row_operation(
 	ut_ad(undo != NULL);
 
 	do {
-		ulint	offset = !rec
+		uint16_t offset = !rec
 			? trx_undo_page_report_insert(
 				undo_block, trx, index, clust_entry, &mtr)
 			: trx_undo_page_report_modify(
@@ -2138,7 +2020,15 @@ trx_undo_report_row_operation(
 				cmpl_info, clust_entry, &mtr);
 
 		if (UNIV_UNLIKELY(offset == 0)) {
-			if (!trx_undo_erase_page_end(undo_block->frame)) {
+			const uint16_t first_free = mach_read_from_2(
+				TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE
+				+ undo_block->frame);
+			memset(undo_block->frame + first_free, 0,
+			       (srv_page_size - FIL_PAGE_DATA_END)
+			       - first_free);
+
+			if (first_free
+			    == TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE) {
 				/* The record did not fit on an empty
 				undo page. Discard the freshly allocated
 				page and return an error. */
@@ -2164,6 +2054,21 @@ trx_undo_report_row_operation(
 
 				err = DB_UNDO_RECORD_TOO_BIG;
 				goto err_exit;
+			} else {
+				/* Write log for clearing the unused
+				tail of the undo page. It might
+				contain some garbage from a previously
+				written record, and mtr_t::write()
+				will optimize away writes of unchanged
+				bytes. Failure to write this caused a
+				recovery failure when we avoided
+				reading the undo log page from the
+				data file and initialized it based on
+				redo log records (which included the
+				write of the previous garbage). */
+				mtr.memset(*undo_block, first_free,
+					   srv_page_size - first_free
+					   - FIL_PAGE_DATA_END, 0);
 			}
 
 			mtr_commit(&mtr);
@@ -2171,7 +2076,7 @@ trx_undo_report_row_operation(
 			/* Success */
 			mtr_commit(&mtr);
 
-			undo->top_page_no = undo_block->page.id.page_no();
+			undo->top_page_no = undo_block->page.id().page_no();
 			undo->top_offset  = offset;
 			undo->top_undo_no = trx->undo_no++;
 			undo->guess_block = undo_block;
@@ -2203,7 +2108,7 @@ trx_undo_report_row_operation(
 			return(DB_SUCCESS);
 		}
 
-		ut_ad(undo_block->page.id.page_no() == undo->last_page_no);
+		ut_ad(undo_block->page.id().page_no() == undo->last_page_no);
 
 		/* We have to extend the undo log by one page */
 
@@ -2251,11 +2156,10 @@ trx_undo_get_undo_rec_low(
 {
 	trx_undo_rec_t*	undo_rec;
 	ulint		rseg_id;
-	ulint		page_no;
-	ulint		offset;
-	const page_t*	undo_page;
+	uint32_t	page_no;
+	uint16_t	offset;
 	trx_rseg_t*	rseg;
-	ibool		is_insert;
+	bool		is_insert;
 	mtr_t		mtr;
 
 	trx_undo_decode_roll_ptr(roll_ptr, &is_insert, &rseg_id, &page_no,
@@ -2265,14 +2169,14 @@ trx_undo_get_undo_rec_low(
 	rseg = trx_sys.rseg_array[rseg_id];
 	ut_ad(rseg->is_persistent());
 
-	mtr_start(&mtr);
+	mtr.start();
 
-	undo_page = trx_undo_page_get_s_latched(
+	buf_block_t* undo_page = trx_undo_page_get_s_latched(
 		page_id_t(rseg->space->id, page_no), &mtr);
 
-	undo_rec = trx_undo_rec_copy(undo_page + offset, heap);
+	undo_rec = trx_undo_rec_copy(undo_page->frame + offset, heap);
 
-	mtr_commit(&mtr);
+	mtr.commit();
 
 	return(undo_rec);
 }
@@ -2298,11 +2202,9 @@ trx_undo_get_undo_rec(
 	const table_name_t&	name,
 	trx_undo_rec_t**	undo_rec)
 {
-	bool		missing_history;
-
 	rw_lock_s_lock(&purge_sys.latch);
 
-	missing_history = purge_sys.view.changes_visible(trx_id, name);
+	bool missing_history = purge_sys.changes_visible(trx_id, name);
 	if (!missing_history) {
 		*undo_rec = trx_undo_get_undo_rec_low(roll_ptr, heap);
 	}
@@ -2364,16 +2266,16 @@ trx_undo_prev_version_build(
 	roll_ptr_t	roll_ptr;
 	upd_t*		update;
 	byte*		ptr;
-	ulint		info_bits;
+	byte		info_bits;
 	ulint		cmpl_info;
 	bool		dummy_extern;
 	byte*		buf;
 
 	ut_ad(!index->table->is_temporary());
 	ut_ad(!rw_lock_own(&purge_sys.latch, RW_LOCK_S));
-	ut_ad(mtr_memo_contains_page_flagged(index_mtr, index_rec,
-					     MTR_MEMO_PAGE_S_FIX
-					     | MTR_MEMO_PAGE_X_FIX));
+	ut_ad(index_mtr->memo_contains_page_flagged(index_rec,
+						    MTR_MEMO_PAGE_S_FIX
+						    | MTR_MEMO_PAGE_X_FIX));
 	ut_ad(rec_offs_validate(rec, index, offsets));
 	ut_a(index->is_primary());
 
@@ -2468,7 +2370,7 @@ trx_undo_prev_version_build(
 
 			rw_lock_s_lock(&purge_sys.latch);
 
-			missing_extern = purge_sys.view.changes_visible(
+			missing_extern = purge_sys.changes_visible(
 				trx_id,	index->table->name);
 
 			rw_lock_s_unlock(&purge_sys.latch);
@@ -2511,7 +2413,53 @@ trx_undo_prev_version_build(
 
 		*old_vers = rec_copy(buf, rec, offsets);
 		rec_offs_make_valid(*old_vers, index, true, offsets);
-		row_upd_rec_in_place(*old_vers, index, offsets, update, NULL);
+		rec_set_bit_field_1(*old_vers, update->info_bits,
+				    rec_offs_comp(offsets)
+				    ? REC_NEW_INFO_BITS : REC_OLD_INFO_BITS,
+				    REC_INFO_BITS_MASK, REC_INFO_BITS_SHIFT);
+		for (ulint i = 0; i < update->n_fields; i++) {
+			const upd_field_t* uf = upd_get_nth_field(update, i);
+			if (upd_fld_is_virtual_col(uf)) {
+				/* There are no virtual columns in
+				a clustered index record. */
+				continue;
+			}
+			const ulint n = uf->field_no;
+			ut_ad(!dfield_is_ext(&uf->new_val)
+			      == !rec_offs_nth_extern(offsets, n));
+			ut_ad(!rec_offs_nth_default(offsets, n));
+
+			if (UNIV_UNLIKELY(dfield_is_null(&uf->new_val))) {
+				if (rec_offs_nth_sql_null(offsets, n)) {
+					ut_ad(index->table->is_instant());
+					ut_ad(n >= index->n_core_fields);
+					continue;
+				}
+				ut_ad(!index->table->not_redundant());
+				ulint l = rec_get_1byte_offs_flag(*old_vers)
+					? (n + 1) : (n + 1) * 2;
+				byte* b = *old_vers - REC_N_OLD_EXTRA_BYTES
+					- l;
+				*b= byte(*b | REC_1BYTE_SQL_NULL_MASK);
+				compile_time_assert(REC_1BYTE_SQL_NULL_MASK << 8
+						    == REC_2BYTE_SQL_NULL_MASK);
+				continue;
+			}
+
+			ulint len;
+			memcpy(rec_get_nth_field(*old_vers, offsets, n, &len),
+			       uf->new_val.data, uf->new_val.len);
+			if (UNIV_UNLIKELY(len != uf->new_val.len)) {
+				ut_ad(len == UNIV_SQL_NULL);
+				ut_ad(!rec_offs_comp(offsets));
+				ut_ad(uf->new_val.len
+				      == rec_get_nth_field_size(rec, n));
+				ulint l = rec_get_1byte_offs_flag(*old_vers)
+					? (n + 1) : (n + 1) * 2;
+				(*old_vers)[-REC_N_OLD_EXTRA_BYTES - l]
+					&= byte(~REC_1BYTE_SQL_NULL_MASK);
+			}
+		}
 	}
 
 	/* Set the old value (which is the after image of an update) in the
@@ -2565,17 +2513,14 @@ trx_undo_read_v_cols(
 	end_ptr = ptr + mach_read_from_2(ptr);
 	ptr += 2;
 	while (ptr < end_ptr) {
-		dfield_t*	dfield;
-		const byte*	field;
-		ulint		field_no;
-		ulint		len;
-		ulint		orig_len;
-		bool		is_virtual;
+		dfield_t* dfield;
+		const byte* field;
+		uint32_t field_no, len, orig_len;
 
 		field_no = mach_read_next_compressed(
 				const_cast<const byte**>(&ptr));
 
-		is_virtual = (field_no >= REC_MAX_N_FIELDS);
+		const bool is_virtual = (field_no >= REC_MAX_N_FIELDS);
 
 		if (is_virtual) {
 			ptr = trx_undo_read_v_idx(
@@ -2590,7 +2535,7 @@ trx_undo_read_v_cols(
 		/* The virtual column is no longer indexed or does not exist.
 		This needs to put after trx_undo_rec_get_col_val() so the
 		undo ptr advances */
-		if (field_no == ULINT_UNDEFINED) {
+		if (field_no == FIL_NULL) {
 			ut_ad(is_virtual);
 			continue;
 		}