MDEV-21174: Remove some mlog_write_initial_log_record_fast()

Pass buf_block_t* to more functions that write redo log.

page_zip_write_node_ptr(), page_zip_write_blob_ptr(),
page_zip_compress_write_log_no_data():
Take buf_block_t* as parameter, and do not tolerate mtr=NULL.

page_zip_compress(): Do not tolerate mtr=NULL.

page_zip_dir_insert(): Take page_cur_t* as parameter.

mlog_write_initial_log_record(): Remove. This function was unused.

RecIterator::remove(): Remove the redundant page_zip parameter.

PageConverter::m_page_zip_ptr: Remove.

9 files changed, 98 insertions, 153 deletions

diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index c02e8766a0d..d6755e75fb8 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -780,7 +780,7 @@ inline void btr_node_ptr_set_child_page_no(buf_block_t *block,
         REC_NODE_PTR_SIZE);
 
   if (UNIV_LIKELY_NULL(block->page.zip.data))
-    page_zip_write_node_ptr(&block->page.zip, rec, offs, page_no, mtr);
+    page_zip_write_node_ptr(block, rec, offs, page_no, mtr);
   else
     mtr->write<4>(*block, rec + offs - REC_NODE_PTR_SIZE, page_no);
 }
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index cf3d0839e86..ff316e575db 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -7153,8 +7153,7 @@ btr_cur_set_ownership_of_extern_field(
 
 	if (UNIV_LIKELY_NULL(block->page.zip.data)) {
 		mach_write_to_1(data + local_len + BTR_EXTERN_LEN, byte_val);
-		page_zip_write_blob_ptr(&block->page.zip, rec, index, offsets,
-					i, mtr);
+		page_zip_write_blob_ptr(block, rec, index, offsets, i, mtr);
 	} else {
 		mtr->write<1,mtr_t::OPT>(*block,
 					 data + local_len + BTR_EXTERN_LEN,
@@ -7826,7 +7825,7 @@ btr_store_big_rec_extern_fields(
 				/* We compress a page when finish bulk insert.*/
 				if (UNIV_LIKELY(op != BTR_STORE_INSERT_BULK)) {
 					page_zip_write_blob_ptr(
-						page_zip, rec, index, offsets,
+						rec_block, rec, index, offsets,
 						field_no, &mtr);
 				}
 
@@ -8103,8 +8102,7 @@ btr_free_externally_stored_field(
 				mach_write_to_4(field_ref + BTR_EXTERN_PAGE_NO,
 						next_page_no);
 				memset(field_ref + BTR_EXTERN_LEN, 0, 4);
-				page_zip_write_blob_ptr(&block->page.zip,
-							rec, index,
+				page_zip_write_blob_ptr(block, rec, index,
 							offsets, i, &mtr);
 			} else {
 				mtr.write<4>(*block,
diff --git a/storage/innobase/include/page0zip.h b/storage/innobase/include/page0zip.h
index 4dade7d7efe..88125538fbf 100644
--- a/storage/innobase/include/page0zip.h
+++ b/storage/innobase/include/page0zip.h
@@ -133,12 +133,11 @@ page_zip_set_alloc(
 @retval false on failure; block->page.zip will be left intact. */
 bool
 page_zip_compress(
-	buf_block_t*		block,		/*!< in/out: buffer block */
-	dict_index_t*		index,		/*!< in: index of the B-tree
-						node */
-	ulint			level,		/*!< in: commpression level */
-	mtr_t*			mtr);		/*!< in/out: mini-transaction,
-						or NULL */
+	buf_block_t*		block,	/*!< in/out: buffer block */
+	dict_index_t*		index,	/*!< in: index of the B-tree node */
+	ulint			level,	/*!< in: commpression level */
+	mtr_t*			mtr)	/*!< in/out: mini-transaction */
+	MY_ATTRIBUTE((nonnull));
 
 /**********************************************************************//**
 Write the index information for the compressed page.
@@ -278,14 +277,14 @@ The information must already have been updated on the uncompressed page. */
 void
 page_zip_write_blob_ptr(
 /*====================*/
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
+	buf_block_t*	block,	/*!< in/out: ROW_FORMAT=COMPRESSED page */
 	const byte*	rec,	/*!< in/out: record whose data is being
 				written */
 	dict_index_t*	index,	/*!< in: index of the page */
 	const ulint*	offsets,/*!< in: rec_get_offsets(rec, index) */
 	ulint		n,	/*!< in: column index */
-	mtr_t*		mtr);	/*!< in: mini-transaction handle,
-				or NULL if no logging is needed */
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
+	MY_ATTRIBUTE((nonnull));
 
 /***********************************************************//**
 Parses a log record of writing the node pointer of a record.
@@ -303,11 +302,12 @@ Write the node pointer of a record on a non-leaf compressed page. */
 void
 page_zip_write_node_ptr(
 /*====================*/
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
+	buf_block_t*	block,	/*!< in/out: compressed page */
 	byte*		rec,	/*!< in/out: record */
 	ulint		size,	/*!< in: data size of rec */
 	ulint		ptr,	/*!< in: node pointer */
-	mtr_t*		mtr);	/*!< in: mini-transaction, or NULL */
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
+	MY_ATTRIBUTE((nonnull));
 
 /** Write the DB_TRX_ID,DB_ROLL_PTR into a clustered index leaf page record.
 @param[in,out]	page_zip	compressed page
@@ -370,8 +370,7 @@ Insert a record to the dense page directory. */
 void
 page_zip_dir_insert(
 /*================*/
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
-	const byte*	prev_rec,/*!< in: record after which to insert */
+	page_cur_t*	cursor,	/*!< in/out: page cursor */
 	const byte*	free_rec,/*!< in: record from which rec was
 				allocated, or NULL */
 	byte*		rec);	/*!< in: record to insert */
@@ -412,22 +411,6 @@ page_zip_parse_write_header(
 	page_zip_des_t*	page_zip);/*!< in/out: compressed page */
 
 /**********************************************************************//**
-Write data to the uncompressed header portion of a page.  The data must
-already have been written to the uncompressed page.
-However, the data portion of the uncompressed page may differ from
-the compressed page when a record is being inserted in
-page_cur_insert_rec_low(). */
-UNIV_INLINE
-void
-page_zip_write_header(
-/*==================*/
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
-	const byte*	str,	/*!< in: address on the uncompressed page */
-	ulint		length,	/*!< in: length of the data */
-	mtr_t*		mtr)	/*!< in: mini-transaction, or NULL */
-	MY_ATTRIBUTE((nonnull(1,2)));
-
-/**********************************************************************//**
 Reorganize and compress a page.  This is a low-level operation for
 compressed pages, to be used when page_zip_compress() fails.
 On success, a redo log entry MLOG_ZIP_PAGE_COMPRESS will be written.
@@ -499,9 +482,10 @@ void
 page_zip_compress_write_log_no_data(
 /*================================*/
 	ulint		level,	/*!< in: compression level */
-	const page_t*	page,	/*!< in: page that is compressed */
+	buf_block_t*	block,	/*!< in: ROW_FORMAT=COMPRESSED index page */
 	dict_index_t*	index,	/*!< in: index */
-	mtr_t*		mtr);	/*!< in: mtr */
+	mtr_t*		mtr)	/*!< in: mtr */
+	MY_ATTRIBUTE((nonnull));
 
 /**********************************************************************//**
 Reset the counters used for filling
diff --git a/storage/innobase/include/page0zip.ic b/storage/innobase/include/page0zip.ic
index 23a14b5947c..c7e6737c0d8 100644
--- a/storage/innobase/include/page0zip.ic
+++ b/storage/innobase/include/page0zip.ic
@@ -369,14 +369,15 @@ void
 page_zip_compress_write_log_no_data(
 /*================================*/
 	ulint		level,	/*!< in: compression level */
-	const page_t*	page,	/*!< in: page that is compressed */
+	buf_block_t*	block,	/*!< in: ROW_FORMAT=COMPRESSED index page */
 	dict_index_t*	index,	/*!< in: index */
 	mtr_t*		mtr)	/*!< in: mtr */
 {
-	byte* log_ptr = mlog_open_and_write_index(
-		mtr, page, index, MLOG_ZIP_PAGE_COMPRESS_NO_DATA, 1);
-
-	if (log_ptr) {
+	if (byte* log_ptr = mlog_open(mtr, 11 + 1)) {
+		log_ptr = mlog_write_initial_log_record_low(
+			MLOG_ZIP_PAGE_COMPRESS_NO_DATA,
+			block->page.id.space(), block->page.id.page_no(),
+			log_ptr, mtr);
 		mach_write_to_1(log_ptr, level);
 		mlog_close(mtr, log_ptr + 1);
 	}
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 5d41ff1fb03..8e3a55f4197 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -1704,7 +1704,7 @@ parse_log:
 				break;
 			}
 			if (page &&
-			    !page_zip_compress(block, index, *ptr, NULL)) {
+			    !page_zip_compress(block, index, *ptr, mtr)) {
 				ut_error;
 			}
 			ptr++;
diff --git a/storage/innobase/mtr/mtr0log.cc b/storage/innobase/mtr/mtr0log.cc
index 60ed8830593..8563c82fe89 100644
--- a/storage/innobase/mtr/mtr0log.cc
+++ b/storage/innobase/mtr/mtr0log.cc
@@ -50,38 +50,7 @@ mlog_catenate_string(
 }
 
 /********************************************************//**
-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. */
-void
-mlog_write_initial_log_record(
-/*==========================*/
-	const byte*	ptr,	/*!< in: pointer to (inside) a buffer
-				frame holding the file page where
-				modification is made */
-	mlog_id_t	type,	/*!< in: log item type: MLOG_1BYTE, ... */
-	mtr_t*		mtr)	/*!< in: mini-transaction handle */
-{
-	byte*	log_ptr;
-
-	ut_ad(type <= MLOG_BIGGEST_TYPE);
-	ut_ad(type > MLOG_8BYTES);
-
-	log_ptr = mlog_open(mtr, 11);
-
-	/* If no logging is requested, we may return now */
-	if (log_ptr == NULL) {
-
-		return;
-	}
-
-	log_ptr = mlog_write_initial_log_record_fast(ptr, type, log_ptr, mtr);
-
-	mlog_close(mtr, log_ptr);
-}
-
-/********************************************************//**
-Parses an initial log record written by mlog_write_initial_log_record.
+Parses an initial log record written by mlog_write_initial_log_record_low().
 @return parsed record end, NULL if not a complete record */
 const byte*
 mlog_parse_initial_log_record(
diff --git a/storage/innobase/page/page0cur.cc b/storage/innobase/page/page0cur.cc
index 72473f45980..457dc64f2bc 100644
--- a/storage/innobase/page/page0cur.cc
+++ b/storage/innobase/page/page0cur.cc
@@ -1772,14 +1772,20 @@ page_cur_insert_rec_zip(
 			ut_ad(pos > 0);
 
 			if (!log_compressed) {
-				if (page_zip_compress(
-					    page_cur_get_block(cursor),
-					    index, level, NULL)) {
+				const mtr_log_t log_mode = mtr->set_log_mode(
+					MTR_LOG_NONE);
+				const bool ok = page_zip_compress(
+					page_cur_get_block(cursor),
+					index, level, mtr);
+				mtr->set_log_mode(log_mode);
+				if (ok) {
 					page_cur_insert_rec_write_log(
 						insert_rec, rec_size,
 						cursor->rec, index, mtr);
 					page_zip_compress_write_log_no_data(
-						level, page, index, mtr);
+						level,
+						page_cur_get_block(cursor),
+						index, mtr);
 
 					rec_offs_make_valid(
 						insert_rec, index,
@@ -1991,7 +1997,7 @@ use_heap:
 	UNIV_MEM_ASSERT_RW(rec_get_start(insert_rec, offsets),
 			   rec_offs_size(offsets));
 
-	page_zip_dir_insert(page_zip, cursor->rec, free_rec, insert_rec);
+	page_zip_dir_insert(cursor, free_rec, insert_rec);
 
 	/* 6. Update the last insertion info in page header */
 
diff --git a/storage/innobase/page/page0zip.cc b/storage/innobase/page/page0zip.cc
index f862cf8dce0..8a5a580433e 100644
--- a/storage/innobase/page/page0zip.cc
+++ b/storage/innobase/page/page0zip.cc
@@ -1232,12 +1232,10 @@ func_exit:
 @retval false on failure; block->page.zip will be left intact. */
 bool
 page_zip_compress(
-	buf_block_t*		block,		/*!< in/out: buffer block */
-	dict_index_t*		index,		/*!< in: index of the B-tree
-						node */
-	ulint			level,		/*!< in: commpression level */
-	mtr_t*			mtr)		/*!< in/out: mini-transaction,
-						or NULL */
+	buf_block_t*		block,	/*!< in/out: buffer block */
+	dict_index_t*		index,	/*!< in: index of the B-tree node */
+	ulint			level,	/*!< in: commpression level */
+	mtr_t*			mtr)	/*!< in/out: mini-transaction */
 {
 	z_stream		c_stream;
 	int			err;
@@ -1505,7 +1503,7 @@ err_exit:
 			fclose(logfile);
 		}
 #endif /* PAGE_ZIP_COMPRESS_DBG */
-		if (page_is_leaf(page) && index) {
+		if (page_is_leaf(page)) {
 			dict_index_zip_failure(index);
 		}
 
@@ -1558,9 +1556,7 @@ err_exit:
 	ut_a(page_zip_validate(page_zip, page, index));
 #endif /* UNIV_ZIP_DEBUG */
 
-	if (mtr) {
-		page_zip_compress_write_log(block, index, mtr);
-	}
+	page_zip_compress_write_log(block, index, mtr);
 
 	UNIV_MEM_ASSERT_RW(page_zip->data, page_zip_get_size(page_zip));
 
@@ -3876,23 +3872,22 @@ The information must already have been updated on the uncompressed page. */
 void
 page_zip_write_blob_ptr(
 /*====================*/
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
+	buf_block_t*	block,	/*!< in/out: ROW_FORMAT=COMPRESSED page */
 	const byte*	rec,	/*!< in/out: record whose data is being
 				written */
 	dict_index_t*	index,	/*!< in: index of the page */
 	const ulint*	offsets,/*!< in: rec_get_offsets(rec, index) */
 	ulint		n,	/*!< in: column index */
-	mtr_t*		mtr)	/*!< in: mini-transaction handle,
-				or NULL if no logging is needed */
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 {
 	const byte*	field;
 	byte*		externs;
-	const page_t*	page	= page_align(rec);
+	const page_t* const page = block->frame;
+	page_zip_des_t* const page_zip = &block->page.zip;
 	ulint		blob_no;
 	ulint		len;
 
-	ut_ad(page_zip != NULL);
-	ut_ad(rec != NULL);
+	ut_ad(page_align(rec) == page);
 	ut_ad(index != NULL);
 	ut_ad(offsets != NULL);
 	ut_ad(page_simple_validate_new((page_t*) page));
@@ -3934,15 +3929,11 @@ page_zip_write_blob_ptr(
 	ut_a(page_zip_validate(page_zip, page, index));
 #endif /* UNIV_ZIP_DEBUG */
 
-	if (mtr) {
-		byte*	log_ptr	= mlog_open(
-			mtr, 11 + 2 + 2 + BTR_EXTERN_FIELD_REF_SIZE);
-		if (UNIV_UNLIKELY(!log_ptr)) {
-			return;
-		}
-
-		log_ptr = mlog_write_initial_log_record_fast(
-			(byte*) field, MLOG_ZIP_WRITE_BLOB_PTR, log_ptr, mtr);
+	if (byte* log_ptr = mlog_open(mtr, 11 + 2 + 2 + FIELD_REF_SIZE)) {
+		log_ptr = mlog_write_initial_log_record_low(
+			MLOG_ZIP_WRITE_BLOB_PTR,
+			block->page.id.space(), block->page.id.page_no(),
+			log_ptr, mtr);
 		mach_write_to_2(log_ptr, page_offset(field));
 		log_ptr += 2;
 		mach_write_to_2(log_ptr, ulint(externs - page_zip->data));
@@ -4033,18 +4024,17 @@ Write the node pointer of a record on a non-leaf compressed page. */
 void
 page_zip_write_node_ptr(
 /*====================*/
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
+	buf_block_t*	block,	/*!< in/out: compressed page */
 	byte*		rec,	/*!< in/out: record */
 	ulint		size,	/*!< in: data size of rec */
 	ulint		ptr,	/*!< in: node pointer */
-	mtr_t*		mtr)	/*!< in: mini-transaction, or NULL */
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 {
 	byte*	field;
 	byte*	storage;
-#ifdef UNIV_DEBUG
-	page_t*	page	= page_align(rec);
-#endif /* UNIV_DEBUG */
+	page_zip_des_t* const page_zip = &block->page.zip;
 
+	ut_d(const page_t* const page = block->frame);
 	ut_ad(page_simple_validate_new(page));
 	ut_ad(page_zip_simple_validate(page_zip));
 	ut_ad(page_zip_get_size(page_zip)
@@ -4070,15 +4060,11 @@ page_zip_write_node_ptr(
 	mach_write_to_4(field, ptr);
 	memcpy(storage, field, REC_NODE_PTR_SIZE);
 
-	if (mtr) {
-		byte*	log_ptr	= mlog_open(mtr,
-					    11 + 2 + 2 + REC_NODE_PTR_SIZE);
-		if (UNIV_UNLIKELY(!log_ptr)) {
-			return;
-		}
-
-		log_ptr = mlog_write_initial_log_record_fast(
-			field, MLOG_ZIP_WRITE_NODE_PTR, log_ptr, mtr);
+	if (byte* log_ptr = mlog_open(mtr, 11 + 2 + 2 + REC_NODE_PTR_SIZE)) {
+		log_ptr = mlog_write_initial_log_record_low(
+			MLOG_ZIP_WRITE_NODE_PTR,
+			block->page.id.space(), block->page.id.page_no(),
+			log_ptr, mtr);
 		mach_write_to_2(log_ptr, page_offset(field));
 		log_ptr += 2;
 		mach_write_to_2(log_ptr, ulint(storage - page_zip->data));
@@ -4366,23 +4352,26 @@ Insert a record to the dense page directory. */
 void
 page_zip_dir_insert(
 /*================*/
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
-	const byte*	prev_rec,/*!< in: record after which to insert */
+	page_cur_t*	cursor,	/*!< in/out: page cursor */
 	const byte*	free_rec,/*!< in: record from which rec was
 				allocated, or NULL */
 	byte*		rec)	/*!< in: record to insert */
 {
+	ut_ad(page_align(cursor->rec) == cursor->block->frame);
+	ut_ad(page_align(rec) == cursor->block->frame);
+	page_zip_des_t *const page_zip= &cursor->block->page.zip;
+
 	ulint	n_dense;
 	byte*	slot_rec;
 	byte*	slot_free;
 
-	ut_ad(prev_rec != rec);
-	ut_ad(page_rec_get_next((rec_t*) prev_rec) == rec);
+	ut_ad(cursor->rec != rec);
+	ut_ad(page_rec_get_next_const(cursor->rec) == rec);
 	ut_ad(page_zip_simple_validate(page_zip));
 
 	UNIV_MEM_ASSERT_RW(page_zip->data, page_zip_get_size(page_zip));
 
-	if (page_rec_is_infimum(prev_rec)) {
+	if (page_rec_is_infimum(cursor->rec)) {
 		/* Use the first slot. */
 		slot_rec = page_zip->data + page_zip_get_size(page_zip);
 	} else {
@@ -4398,7 +4387,7 @@ page_zip_dir_insert(
 		}
 
 		slot_rec = page_zip_dir_find_low(start, end,
-						 page_offset(prev_rec));
+						 page_offset(cursor->rec));
 		ut_a(slot_rec);
 	}
 
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 3b754bb1072..fc25f9711e5 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -256,6 +256,8 @@ public:
 		return(page_cur_get_rec(&m_cur));
 	}
 
+	buf_block_t* current_block() const { return m_cur.block; }
+
 	/**
 	@return true if cursor is at the end */
 	bool	end() UNIV_NOTHROW
@@ -267,7 +269,6 @@ public:
 	@return true on success */
 	bool remove(
 		const dict_index_t*	index,
-		page_zip_des_t*		page_zip,
 		ulint*			offsets) UNIV_NOTHROW
 	{
 		ut_ad(page_is_leaf(m_cur.block->frame));
@@ -287,6 +288,7 @@ public:
 		}
 
 #ifdef UNIV_ZIP_DEBUG
+		page_zip_des_t* page_zip = buf_block_get_page_zip(m_cur.block);
 		ut_a(!page_zip || page_zip_validate(
 			     page_zip, m_cur.block->frame, index));
 #endif /* UNIV_ZIP_DEBUG */
@@ -303,6 +305,7 @@ public:
 
 private:
 	page_cur_t	m_cur;
+public:
 	mtr_t		m_mtr;
 };
 
@@ -823,7 +826,6 @@ public:
 		AbstractCallback(trx, space_id),
 		m_cfg(cfg),
 		m_index(cfg->m_indexes),
-		m_page_zip_ptr(0),
 		m_rec_iter(),
 		m_offsets_(), m_offsets(m_offsets_),
 		m_heap(0),
@@ -931,9 +933,6 @@ private:
 	/** Current index whose pages are being imported */
 	row_index_t*		m_index;
 
-	/** Alias for m_page_zip, only set for compressed pages. */
-	page_zip_des_t*		m_page_zip_ptr;
-
 	/** Iterator over records in a block */
 	RecIterator		m_rec_iter;
 
@@ -1616,9 +1615,10 @@ PageConverter::adjust_cluster_index_blob_column(
 
 	mach_write_to_4(field, get_space_id());
 
-	if (m_page_zip_ptr) {
+	if (UNIV_LIKELY_NULL(m_rec_iter.current_block()->page.zip.data)) {
 		page_zip_write_blob_ptr(
-			m_page_zip_ptr, rec, m_cluster_index, offsets, i, 0);
+			m_rec_iter.current_block(), rec, m_cluster_index,
+			offsets, i, &m_rec_iter.m_mtr);
 	}
 
 	return(DB_SUCCESS);
@@ -1689,7 +1689,7 @@ inline bool PageConverter::purge() UNIV_NOTHROW
 	const dict_index_t*	index = m_index->m_srv_index;
 
 	/* We can't have a page that is empty and not root. */
-	if (m_rec_iter.remove(index, m_page_zip_ptr, m_offsets)) {
+	if (m_rec_iter.remove(index, m_offsets)) {
 
 		++m_index->m_stats.m_n_purged;
 
@@ -1720,11 +1720,13 @@ PageConverter::adjust_cluster_record(
 		record. */
 		ulint	trx_id_pos = m_cluster_index->n_uniq
 			? m_cluster_index->n_uniq : 1;
-		if (m_page_zip_ptr) {
+		if (UNIV_LIKELY_NULL(m_rec_iter.current_block()
+				     ->page.zip.data)) {
 			page_zip_write_trx_id_and_roll_ptr(
-				m_page_zip_ptr, rec, m_offsets, trx_id_pos,
+				&m_rec_iter.current_block()->page.zip,
+				rec, m_offsets, trx_id_pos,
 				0, roll_ptr_t(1) << ROLL_PTR_INSERT_FLAG_POS,
-				NULL);
+				&m_rec_iter.m_mtr);
 		} else {
 			ulint	len;
 			byte*	ptr = rec_get_nth_field(
@@ -1843,8 +1845,8 @@ PageConverter::update_index_page(
 	the current index id. */
 	mach_write_to_8(page + (PAGE_HEADER + PAGE_INDEX_ID),
 			m_index->m_srv_index->id);
-	if (m_page_zip_ptr) {
-		memcpy(&m_page_zip_ptr->data[PAGE_HEADER + PAGE_INDEX_ID],
+	if (UNIV_LIKELY_NULL(block->page.zip.data)) {
+		memcpy(&block->page.zip.data[PAGE_HEADER + PAGE_INDEX_ID],
 		       &block->frame[PAGE_HEADER + PAGE_INDEX_ID], 8);
 	}
 
@@ -1891,11 +1893,11 @@ PageConverter::update_index_page(
 		/* Set PAGE_MAX_TRX_ID on secondary index leaf pages. */
 		mach_write_to_8(&block->frame[PAGE_HEADER + PAGE_MAX_TRX_ID],
 				m_trx->id);
-		if (m_page_zip_ptr) {
-			memcpy(&m_page_zip_ptr
-			       ->data[PAGE_HEADER + PAGE_MAX_TRX_ID],
-			       &block->frame[PAGE_HEADER + PAGE_MAX_TRX_ID],
-			       8);
+		if (UNIV_LIKELY_NULL(block->page.zip.data)) {
+			memcpy_aligned<8>(&block->page.zip.data
+					  [PAGE_HEADER + PAGE_MAX_TRX_ID],
+					  &block->frame
+					  [PAGE_HEADER + PAGE_MAX_TRX_ID], 8);
 		}
 	} else {
 clear_page_max_trx_id:
@@ -1904,10 +1906,12 @@ clear_page_max_trx_id:
 		in MySQL 5.6, 5.7 and MariaDB 10.0 and 10.1
 		would set the field to the transaction ID even
 		on clustered index pages. */
-		memset(&block->frame[PAGE_HEADER + PAGE_MAX_TRX_ID], 0, 8);
-		if (m_page_zip_ptr) {
-			memset(&m_page_zip_ptr
-			       ->data[PAGE_HEADER + PAGE_MAX_TRX_ID], 0, 8);
+		memset_aligned<8>(&block->frame[PAGE_HEADER + PAGE_MAX_TRX_ID],
+				  0, 8);
+		if (UNIV_LIKELY_NULL(block->page.zip.data)) {
+			memset_aligned<8>(&block->page.zip.data
+					  [PAGE_HEADER + PAGE_MAX_TRX_ID],
+					  0, 8);
 		}
 	}
 
@@ -1963,12 +1967,6 @@ PageConverter::update_page(buf_block_t* block, uint16_t& page_type)
 
 	ut_ad(!block->page.zip.data == !is_compressed_table());
 
-	if (block->page.zip.data) {
-		m_page_zip_ptr = &block->page.zip;
-	} else {
-		ut_ad(!m_page_zip_ptr);
-	}
-
 	switch (page_type = fil_page_get_type(get_frame(block))) {
 	case FIL_PAGE_TYPE_FSP_HDR:
 		ut_a(block->page.id.page_no() == 0);