MDEV-19738: Skip doublewrite on MLOG_ZIP_PAGE_COMPRESS

page_zip_compress(), page_zip_compress_write_log(),
page_zip_copy_recs(): Replace the parameters page,page_zip with block,
and set buf_page_t::init_on_flush on success
if innodb_log_optimize_ddl=OFF.

page_zip_parse_compress_no_data(): Merge with the only caller
recv_parse_or_apply_log_rec_body().

storage/innobase/btr/btr0btr.cc

@@ -1507,7 +1507,7 @@ btr_page_reorganize_low(
 	}
 
 	if (page_zip
-	    && !page_zip_compress(page_zip, page, index, z_level, mtr)) {
+	    && !page_zip_compress(block, index, z_level, mtr)) {
 
 		/* Restore the old page and exit. */
 #if defined UNIV_DEBUG || defined UNIV_ZIP_DEBUG
@@ -1951,7 +1951,7 @@ btr_root_raise_and_insert(
 		ut_a(new_page_zip);
 
 		/* Copy the page byte for byte. */
-		page_zip_copy_recs(new_page_zip, new_page,
+		page_zip_copy_recs(new_block,
 				   root_page_zip, root, index, mtr);
 
 		/* Update the lock table and possible hash index. */
@@ -3063,7 +3063,7 @@ btr_page_split_and_insert(
 			as appropriate.  Deleting will always succeed. */
 			ut_a(new_page_zip);
 
-			page_zip_copy_recs(new_page_zip, new_page,
+			page_zip_copy_recs(new_block,
 					   page_zip, page, cursor->index, mtr);
 			page_delete_rec_list_end(move_limit - page + new_page,
 						 new_block, cursor->index,
@@ -3106,7 +3106,7 @@ btr_page_split_and_insert(
 			as appropriate.  Deleting will always succeed. */
 			ut_a(new_page_zip);
 
-			page_zip_copy_recs(new_page_zip, new_page,
+			page_zip_copy_recs(new_block,
 					   page_zip, page, cursor->index, mtr);
 			page_delete_rec_list_start(move_limit - page
 						   + new_page, new_block,
@@ -3396,7 +3396,6 @@ btr_lift_page_up(
 	mtr_t*		mtr)	/*!< in: mtr */
 {
 	buf_block_t*	father_block;
-	page_t*		father_page;
 	ulint		page_level;
 	page_zip_des_t*	father_page_zip;
 	page_t*		page		= buf_block_get_frame(block);
@@ -3433,7 +3432,6 @@ btr_lift_page_up(
 		}
 		father_block = btr_cur_get_block(&cursor);
 		father_page_zip = buf_block_get_page_zip(father_block);
-		father_page = buf_block_get_frame(father_block);
 
 		n_blocks = 0;
 
@@ -3481,7 +3479,6 @@ btr_lift_page_up(
 
 			father_block = blocks[0];
 			father_page_zip = buf_block_get_page_zip(father_block);
-			father_page = buf_block_get_frame(father_block);
 		}
 
 		mem_heap_free(heap);
@@ -3515,7 +3512,7 @@ btr_lift_page_up(
 		ut_a(page_zip);
 
 		/* Copy the page byte for byte. */
-		page_zip_copy_recs(father_page_zip, father_page,
+		page_zip_copy_recs(father_block,
 				   page_zip, page, index, mtr);
 
 		/* Update the lock table and possible hash index. */
@@ -3568,7 +3565,7 @@ btr_lift_page_up(
 	    && !index->table->is_temporary()) {
 		ibuf_reset_free_bits(father_block);
 	}
-	ut_ad(page_validate(father_page, index));
+	ut_ad(page_validate(father_block->frame, index));
 	ut_ad(btr_check_node_ptr(index, father_block, mtr));
 
 	return(lift_father_up ? block_orig : father_block);

storage/innobase/btr/btr0bulk.cc

@@ -373,8 +373,7 @@ PageBulk::compress()
 {
 	ut_ad(m_page_zip != NULL);
 
-	return(page_zip_compress(m_page_zip, m_page, m_index,
-				 page_zip_level, &m_mtr));
+	return page_zip_compress(m_block, m_index, page_zip_level, &m_mtr);
 }
 
 /** Get node pointer

storage/innobase/gis/gis0rtree.cc

@@ -922,7 +922,7 @@ rtr_split_page_move_rec_list(
 	if (new_page_zip) {
 		mtr_set_log_mode(mtr, log_mode);
 
-		if (!page_zip_compress(new_page_zip, new_page, index,
+		if (!page_zip_compress(new_block, index,
 				       page_zip_level, mtr)) {
 			ulint	ret_pos;
 
@@ -1130,7 +1130,7 @@ rtr_page_split_and_insert(
 		as appropriate.  Deleting will always succeed. */
 		ut_a(new_page_zip);
 
-		page_zip_copy_recs(new_page_zip, new_page,
+		page_zip_copy_recs(new_block,
 				   page_zip, page, cursor->index, mtr);
 
 		page_cursor = btr_cur_get_page_cur(cursor);

storage/innobase/include/page0zip.h

@@ -128,17 +128,12 @@ page_zip_set_alloc(
 	void*		stream,		/*!< in/out: zlib stream */
 	mem_heap_t*	heap);		/*!< in: memory heap to use */
 
-/**********************************************************************//**
-Compress a page.
-@return TRUE on success, FALSE on failure; page_zip will be left
-intact on failure. */
-ibool
+/** Attempt to compress a ROW_FORMAT=COMPRESSED page.
+@retval true on success
+@retval false on failure; block->page.zip will be left intact. */
+bool
 page_zip_compress(
-/*==============*/
-	page_zip_des_t*		page_zip,	/*!< in: size; out: data,
-						n_blobs, m_start, m_end,
-						m_nonempty */
-	const page_t*		page,		/*!< in: uncompressed page */
+	buf_block_t*		block,		/*!< in/out: buffer block */
 	dict_index_t*		index,		/*!< in: index of the B-tree
 						node */
 	ulint			level,		/*!< in: commpression level */
@@ -461,11 +456,7 @@ related to the storage of records.  Also copy PAGE_MAX_TRX_ID.
 NOTE: The caller must update the lock table and the adaptive hash index. */
 void
 page_zip_copy_recs(
-/*===============*/
-	page_zip_des_t*		page_zip,	/*!< out: copy of src_zip
-						(n_blobs, m_start, m_end,
-						m_nonempty, data[0..size-1]) */
-	page_t*			page,		/*!< out: copy of src */
+	buf_block_t*		block,		/*!< in/out: buffer block */
 	const page_zip_des_t*	src_zip,	/*!< in: compressed page */
 	const page_t*		src,		/*!< in: page */
 	dict_index_t*		index,		/*!< in: index of the B-tree */
@@ -511,19 +502,6 @@ page_zip_compress_write_log_no_data(
 	const page_t*	page,	/*!< in: page that is compressed */
 	dict_index_t*	index,	/*!< in: index */
 	mtr_t*		mtr);	/*!< in: mtr */
-/**********************************************************************//**
-Parses a log record of compressing an index page without the data.
-@return end of log record or NULL */
-UNIV_INLINE
-byte*
-page_zip_parse_compress_no_data(
-/*============================*/
-	byte*		ptr,		/*!< in: buffer */
-	byte*		end_ptr,	/*!< in: buffer end */
-	page_t*		page,		/*!< in: uncompressed page */
-	page_zip_des_t*	page_zip,	/*!< out: compressed page */
-	dict_index_t*	index)		/*!< in: index */
-	MY_ATTRIBUTE((nonnull(1,2)));
 
 /**********************************************************************//**
 Reset the counters used for filling

storage/innobase/include/page0zip.ic

@@ -379,38 +379,6 @@ page_zip_compress_write_log_no_data(
 	}
 }
 
-/**********************************************************************//**
-Parses a log record of compressing an index page without the data.
-@return end of log record or NULL */
-UNIV_INLINE
-byte*
-page_zip_parse_compress_no_data(
-/*============================*/
-	byte*		ptr,		/*!< in: buffer */
-	byte*		end_ptr,	/*!< in: buffer end */
-	page_t*		page,		/*!< in: uncompressed page */
-	page_zip_des_t*	page_zip,	/*!< out: compressed page */
-	dict_index_t*	index)		/*!< in: index */
-{
-	ulint	level;
-	if (end_ptr == ptr) {
-		return(NULL);
-	}
-
-	level = mach_read_from_1(ptr);
-
-	/* If page compression fails then there must be something wrong
-	because a compress log record is logged only if the compression
-	was successful. Crash in this case. */
-
-	if (page
-	    && !page_zip_compress(page_zip, page, index, level, NULL)) {
-		ut_error;
-	}
-
-	return(ptr + 1);
-}
-
 /**********************************************************************//**
 Reset the counters used for filling
 INFORMATION_SCHEMA.innodb_cmp_per_index. */

storage/innobase/log/log0recv.cc

@@ -1662,8 +1662,15 @@ recv_parse_or_apply_log_rec_body(
 
 			ut_a(!page || ((ibool)!!page_is_comp(page)
 				== dict_table_is_comp(index->table)));
-			ptr = page_zip_parse_compress_no_data(
-				ptr, end_ptr, page, page_zip, index);
+			if (end_ptr == ptr) {
+				ptr = NULL;
+				break;
+			}
+			if (page &&
+			    !page_zip_compress(block, index, *ptr, NULL)) {
+				ut_error;
+			}
+			ptr++;
 		}
 		break;
 	case MLOG_ZIP_WRITE_TRX_ID:

storage/innobase/page/page0cur.cc

@@ -1638,8 +1638,8 @@ page_cur_insert_rec_zip(
 
 			if (!log_compressed) {
 				if (page_zip_compress(
-					    page_zip, page, index,
-					    level, NULL)) {
+					    page_cur_get_block(cursor),
+					    index, level, NULL)) {
 					page_cur_insert_rec_write_log(
 						insert_rec, rec_size,
 						cursor->rec, index, mtr);

storage/innobase/page/page0page.cc

@@ -458,10 +458,9 @@ page_create_zip(
 						handle */
 {
 	page_t*			page;
-	page_zip_des_t*		page_zip = buf_block_get_page_zip(block);
 
 	ut_ad(block);
-	ut_ad(page_zip);
+	ut_ad(buf_block_get_page_zip(block));
 	ut_ad(dict_table_is_comp(index->table));
 
 	/* PAGE_MAX_TRX_ID or PAGE_ROOT_AUTO_INC are always 0 for
@@ -484,7 +483,7 @@ page_create_zip(
 	mach_write_to_2(PAGE_HEADER + PAGE_LEVEL + page, level);
 	mach_write_to_8(PAGE_HEADER + PAGE_MAX_TRX_ID + page, max_trx_id);
 
-	if (!page_zip_compress(page_zip, page, index, page_zip_level, mtr)) {
+	if (!page_zip_compress(block, index, page_zip_level, mtr)) {
 		/* The compression of a newly created
 		page should always succeed. */
 		ut_error;
@@ -705,7 +704,7 @@ page_copy_rec_list_end(
 	if (new_page_zip) {
 		mtr_set_log_mode(mtr, log_mode);
 
-		if (!page_zip_compress(new_page_zip, new_page, index,
+		if (!page_zip_compress(new_block, index,
 				       page_zip_level, mtr)) {
 			/* Before trying to reorganize the page,
 			store the number of preceding records on the page. */
@@ -867,7 +866,7 @@ page_copy_rec_list_start(
 		DBUG_EXECUTE_IF("page_copy_rec_list_start_compress_fail",
 				goto zip_reorganize;);
 
-		if (!page_zip_compress(new_page_zip, new_page, index,
+		if (!page_zip_compress(new_block, index,
 				       page_zip_level, mtr)) {
 			ulint	ret_pos;
 #ifndef DBUG_OFF

storage/innobase/page/page0zip.cc

@@ -365,16 +365,12 @@ page_zip_dir_get(
 				- PAGE_ZIP_DIR_SLOT_SIZE * (slot + 1)));
 }
 
-/**********************************************************************//**
-Write a log record of compressing an index page. */
-static
-void
-page_zip_compress_write_log(
-/*========================*/
-	const page_zip_des_t*	page_zip,/*!< in: compressed page */
-	const page_t*		page,	/*!< in: uncompressed page */
-	dict_index_t*		index,	/*!< in: index of the B-tree node */
-	mtr_t*			mtr)	/*!< in: mini-transaction */
+/** Write a MLOG_ZIP_PAGE_COMPRESS record of compressing an index page.
+@param[in,out]	block	ROW_FORMAT=COMPRESSED index page
+@param[in]	index	the index that the block belongs to
+@param[in,out]	mtr	mini-transaction */
+static void page_zip_compress_write_log(buf_block_t* block,
+					dict_index_t* index, mtr_t* mtr)
 {
 	byte*	log_ptr;
 	ulint	trailer_size;
@@ -388,6 +384,8 @@ page_zip_compress_write_log(
 		return;
 	}
 
+	const page_t* page = block->frame;
+	const page_zip_des_t* page_zip = &block->page.zip;
 	/* Read the number of user records. */
 	trailer_size = ulint(page_dir_get_n_heap(page_zip->data))
 		- PAGE_HEAP_NO_USER_LOW;
@@ -406,9 +404,10 @@ page_zip_compress_write_log(
 	compile_time_assert(FIL_PAGE_DATA <= PAGE_DATA);
 	ut_a(page_zip->m_end + trailer_size <= page_zip_get_size(page_zip));
 
-	log_ptr = mlog_write_initial_log_record_fast((page_t*) page,
-						     MLOG_ZIP_PAGE_COMPRESS,
-						     log_ptr, mtr);
+	log_ptr = mlog_write_initial_log_record_low(MLOG_ZIP_PAGE_COMPRESS,
+						    block->page.id.space(),
+						    block->page.id.page_no(),
+						    log_ptr, mtr);
 	mach_write_to_2(log_ptr, ulint(page_zip->m_end - FIL_PAGE_TYPE));
 	log_ptr += 2;
 	mach_write_to_2(log_ptr, trailer_size);
@@ -425,6 +424,9 @@ page_zip_compress_write_log(
 	/* Write the uncompressed trailer of the compressed page. */
 	mlog_catenate_string(mtr, page_zip->data + page_zip_get_size(page_zip)
 			     - trailer_size, trailer_size);
+	if (!innodb_log_optimize_ddl) {
+		block->page.init_on_flush = true;
+	}
 }
 
 /******************************************************//**
@@ -1225,17 +1227,12 @@ page_zip_compress_clust(
 func_exit:
 	return(err);}
 
-/**********************************************************************//**
-Compress a page.
-@return TRUE on success, FALSE on failure; page_zip will be left
-intact on failure. */
-ibool
+/** Attempt to compress a ROW_FORMAT=COMPRESSED page.
+@retval true on success
+@retval false on failure; block->page.zip will be left intact. */
+bool
 page_zip_compress(
-/*==============*/
-	page_zip_des_t*		page_zip,	/*!< in: size; out: data,
-						n_blobs, m_start, m_end,
-						m_nonempty */
-	const page_t*		page,		/*!< in: uncompressed page */
+	buf_block_t*		block,		/*!< in/out: buffer block */
 	dict_index_t*		index,		/*!< in: index of the B-tree
 						node */
 	ulint			level,		/*!< in: commpression level */
@@ -1268,6 +1265,9 @@ page_zip_compress(
 	my_bool			cmp_per_index_enabled;
 	cmp_per_index_enabled	= srv_cmp_per_index_enabled;
 
+	page_t* page = block->frame;
+	page_zip_des_t* page_zip = &block->page.zip;
+
 	ut_a(page_is_comp(page));
 	ut_a(fil_page_index_page_check(page));
 	ut_ad(page_simple_validate_new((page_t*) page));
@@ -1518,7 +1518,7 @@ page_zip_compress(
 				+= time_diff;
 			mutex_exit(&page_zip_stat_per_index_mutex);
 		}
-		return(FALSE);
+		return false;
 	}
 
 	err = deflateEnd(&c_stream);
@@ -1558,7 +1558,7 @@ page_zip_compress(
 #endif /* UNIV_ZIP_DEBUG */
 
 	if (mtr) {
-		page_zip_compress_write_log(page_zip, page, index, mtr);
+		page_zip_compress_write_log(block, index, mtr);
 	}
 
 	UNIV_MEM_ASSERT_RW(page_zip->data, page_zip_get_size(page_zip));
@@ -1589,7 +1589,7 @@ page_zip_compress(
 		dict_index_zip_success(index);
 	}
 
-	return(TRUE);
+	return true;
 }
 
 /**********************************************************************//**
@@ -4704,7 +4704,6 @@ page_zip_reorganize(
 	mtr_t*		mtr)	/*!< in: mini-transaction */
 {
 	buf_pool_t*	buf_pool	= buf_pool_from_block(block);
-	page_zip_des_t*	page_zip	= buf_block_get_page_zip(block);
 	page_t*		page		= buf_block_get_frame(block);
 	buf_block_t*	temp_block;
 	page_t*		temp_page;
@@ -4715,7 +4714,8 @@ page_zip_reorganize(
 	ut_ad(!index->table->is_temporary());
 	/* Note that page_zip_validate(page_zip, page, index) may fail here. */
 	UNIV_MEM_ASSERT_RW(page, srv_page_size);
-	UNIV_MEM_ASSERT_RW(page_zip->data, page_zip_get_size(page_zip));
+	UNIV_MEM_ASSERT_RW(buf_block_get_page_zip(block)->data,
+			   page_zip_get_size(buf_block_get_page_zip(block)));
 
 	/* Disable logging */
 	mtr_log_t	log_mode = mtr_set_log_mode(mtr, MTR_LOG_NONE);
@@ -4755,7 +4755,7 @@ page_zip_reorganize(
 	/* Restore logging. */
 	mtr_set_log_mode(mtr, log_mode);
 
-	if (!page_zip_compress(page_zip, page, index, page_zip_level, mtr)) {
+	if (!page_zip_compress(block, index, page_zip_level, mtr)) {
 		buf_block_free(temp_block);
 		return(FALSE);
 	}
@@ -4773,16 +4773,15 @@ related to the storage of records.  Also copy PAGE_MAX_TRX_ID.
 NOTE: The caller must update the lock table and the adaptive hash index. */
 void
 page_zip_copy_recs(
-/*===============*/
-	page_zip_des_t*		page_zip,	/*!< out: copy of src_zip
-						(n_blobs, m_start, m_end,
-						m_nonempty, data[0..size-1]) */
-	page_t*			page,		/*!< out: copy of src */
+	buf_block_t*		block,		/*!< in/out: buffer block */
 	const page_zip_des_t*	src_zip,	/*!< in: compressed page */
 	const page_t*		src,		/*!< in: page */
 	dict_index_t*		index,		/*!< in: index of the B-tree */
 	mtr_t*			mtr)		/*!< in: mini-transaction */
 {
+	page_t* page = block->frame;
+	page_zip_des_t* page_zip = &block->page.zip;
+
 	ut_ad(mtr_memo_contains_page(mtr, page, MTR_MEMO_PAGE_X_FIX));
 	ut_ad(mtr_memo_contains_page(mtr, src, MTR_MEMO_PAGE_X_FIX));
 	ut_ad(!dict_index_is_ibuf(index));
@@ -4858,7 +4857,7 @@ page_zip_copy_recs(
 #ifdef UNIV_ZIP_DEBUG
 	ut_a(page_zip_validate(page_zip, page, index));
 #endif /* UNIV_ZIP_DEBUG */
-	page_zip_compress_write_log(page_zip, page, index, mtr);
+	page_zip_compress_write_log(block, index, mtr);
 }
 
 /** Parse and optionally apply MLOG_ZIP_PAGE_COMPRESS.