MDEV-12353 preparation: Do not write MLOG_FILE_WRITE_CRYPT_DATA

The MLOG_FILE_WRITE_CRYPT_DATA record was completely redundant.
It can be replaced with a single MLOG_WRITE_STRING record.

To facilitate upgrade from older versions, we will retain
fil_parse_write_crypt_data().

fil_crypt_parse(): Recover fil_space_crypt_t::write_page0().

fil_space_crypt_t::write_page0(): Write everything in a single
MLOG_WRITE_STRING for easy parsing.

fil_space_crypt_t::page0_offset: Remove.

storage/innobase/fil/fil0crypt.cc

@@ -321,7 +321,6 @@ fil_space_crypt_t* fil_space_read_crypt_data(ulint zip_size, const byte* page)
 	members */
 	crypt_data->type = type;
 	crypt_data->min_key_version = min_key_version;
-	crypt_data->page0_offset = offset;
 	memcpy(crypt_data->iv, page + offset + MAGIC_SZ + 2, iv_length);
 
 	return crypt_data;
@@ -354,6 +353,34 @@ fil_space_destroy_crypt_data(
 	}
 }
 
+/** Amend encryption information from redo log.
+@param[in]	space	tablespace
+@param[in]	data	encryption metadata */
+void fil_crypt_parse(fil_space_t* space, const byte* data)
+{
+	ut_ad(data[1] == MY_AES_BLOCK_SIZE);
+	if (void* buf = ut_zalloc_nokey(sizeof(fil_space_crypt_t))) {
+		fil_space_crypt_t* crypt_data = new(buf)
+			fil_space_crypt_t(
+				data[0],
+				mach_read_from_4(&data[2 + MY_AES_BLOCK_SIZE]),
+				mach_read_from_4(&data[6 + MY_AES_BLOCK_SIZE]),
+				static_cast<fil_encryption_t>
+				(data[10 + MY_AES_BLOCK_SIZE]));
+		memcpy(crypt_data->iv, data + 2, MY_AES_BLOCK_SIZE);
+		mutex_enter(&fil_system.mutex);
+		if (space->crypt_data) {
+			fil_space_merge_crypt_data(space->crypt_data,
+						   crypt_data);
+			fil_space_destroy_crypt_data(&crypt_data);
+			crypt_data = space->crypt_data;
+		} else {
+			space->crypt_data = crypt_data;
+		}
+		mutex_exit(&fil_system.mutex);
+	}
+}
+
 /** Fill crypt data information to the give page.
 It should be called during ibd file creation.
 @param[in]	flags	tablespace flags
@@ -367,7 +394,6 @@ fil_space_crypt_t::fill_page0(
 	const ulint offset = FSP_HEADER_OFFSET
 		+ fsp_header_get_encryption_offset(
 			fil_space_t::zip_size(flags));
-	page0_offset = offset;
 
 	memcpy(page + offset, CRYPT_MAGIC, MAGIC_SZ);
 	mach_write_to_1(page + offset + MAGIC_SZ, type);
@@ -395,53 +421,29 @@ fil_space_crypt_t::write_page0(
 	mtr_t*			mtr)
 {
 	ut_ad(this == space->crypt_data);
-	const uint len = sizeof(iv);
 	const ulint offset = FSP_HEADER_OFFSET
 		+ fsp_header_get_encryption_offset(space->zip_size());
-	page0_offset = offset;
-
-	/*
-	redo log this as bytewise updates to page 0
-	followed by an MLOG_FILE_WRITE_CRYPT_DATA
-	(that will during recovery update fil_space_t)
-	*/
-	mlog_write_string(page + offset, CRYPT_MAGIC, MAGIC_SZ, mtr);
-	mlog_write_ulint(page + offset + MAGIC_SZ + 0, type, MLOG_1BYTE, mtr);
-	mlog_write_ulint(page + offset + MAGIC_SZ + 1, len, MLOG_1BYTE, mtr);
-	mlog_write_string(page + offset + MAGIC_SZ + 2, iv, len,
-			  mtr);
-	mlog_write_ulint(page + offset + MAGIC_SZ + 2 + len, min_key_version,
-			 MLOG_4BYTES, mtr);
-	mlog_write_ulint(page + offset + MAGIC_SZ + 2 + len + 4, key_id,
-			 MLOG_4BYTES, mtr);
-	mlog_write_ulint(page + offset + MAGIC_SZ + 2 + len + 8, encryption,
-		MLOG_1BYTE, mtr);
-
-	byte* log_ptr = mlog_open(mtr, 11 + 17 + len);
-
-	if (log_ptr != NULL) {
-		log_ptr = mlog_write_initial_log_record_fast(
-			page,
-			MLOG_FILE_WRITE_CRYPT_DATA,
-			log_ptr, mtr);
-		mach_write_to_4(log_ptr, space->id);
-		log_ptr += 4;
-		mach_write_to_2(log_ptr, offset);
-		log_ptr += 2;
-		mach_write_to_1(log_ptr, type);
-		log_ptr += 1;
-		mach_write_to_1(log_ptr, len);
-		log_ptr += 1;
-		mach_write_to_4(log_ptr, min_key_version);
-		log_ptr += 4;
-		mach_write_to_4(log_ptr, key_id);
-		log_ptr += 4;
-		mach_write_to_1(log_ptr, encryption);
-		log_ptr += 1;
-		mlog_close(mtr, log_ptr);
-
-		mlog_catenate_string(mtr, iv, len);
-	}
+	byte* b = page + offset;
+
+	if (memcmp(b, CRYPT_MAGIC, MAGIC_SZ)) {
+		mlog_write_string(b, CRYPT_MAGIC, MAGIC_SZ, mtr);
+	}
+
+	b += MAGIC_SZ;
+	byte* const start = b;
+	*b++ = static_cast<byte>(type);
+	compile_time_assert(sizeof iv == MY_AES_BLOCK_SIZE);
+	compile_time_assert(sizeof iv == CRYPT_SCHEME_1_IV_LEN);
+	*b++ = sizeof iv;
+	memcpy(b, iv, sizeof iv);
+	b += sizeof iv;
+	mach_write_to_4(b, min_key_version);
+	b += 4;
+	mach_write_to_4(b, key_id);
+	b += 4;
+	*b++ = byte(encryption);
+	ut_ad(b - start == 11 + MY_AES_BLOCK_SIZE);
+	mlog_log_string(start, b - start, mtr);
 }
 
 /******************************************************************
@@ -475,7 +477,7 @@ fil_parse_write_crypt_data(
 
 	ulint space_id = mach_read_from_4(ptr);
 	ptr += 4;
-	uint offset = mach_read_from_2(ptr);
+	// uint offset = mach_read_from_2(ptr);
 	ptr += 2;
 	uint type = mach_read_from_1(ptr);
 	ptr += 1;
@@ -513,7 +515,6 @@ fil_parse_write_crypt_data(
 	fil_space_crypt_t* crypt_data = fil_space_create_crypt_data(
 		encryption, key_id);
 
-	crypt_data->page0_offset = offset;
 	crypt_data->min_key_version = min_key_version;
 	crypt_data->type = type;
 	memcpy(crypt_data->iv, ptr, len);

storage/innobase/include/fil0crypt.h

@@ -115,7 +115,6 @@ struct fil_space_crypt_t : st_encryption_scheme
 		fil_encryption_t new_encryption)
 		: st_encryption_scheme(),
 		min_key_version(new_min_key_version),
-		page0_offset(0),
 		encryption(new_encryption),
 		key_found(0),
 		rotate_state()
@@ -191,7 +190,6 @@ struct fil_space_crypt_t : st_encryption_scheme
 	void write_page0(const fil_space_t* space, byte* page0, mtr_t* mtr);
 
 	uint min_key_version; // min key version for this space
-	ulint page0_offset;   // byte offset on page 0 for crypt data
 	fil_encryption_t encryption; // Encryption setup
 
 	ib_mutex_t mutex;   // mutex protecting following variables
@@ -308,6 +306,11 @@ fil_parse_write_crypt_data(
 	dberr_t*		err)
 	MY_ATTRIBUTE((warn_unused_result));
 
+/** Amend encryption information from redo log.
+@param[in]	space	tablespace
+@param[in]	data	encryption metadata */
+void fil_crypt_parse(fil_space_t* space, const byte* data);
+
 /** Encrypt a buffer.
 @param[in,out]		crypt_data		Crypt data
 @param[in]		space			space_id

storage/innobase/include/mtr0log.ic

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, MariaDB Corporation.
+Copyright (c) 2017, 2019, MariaDB Corporation.
 
 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
@@ -181,14 +181,13 @@ mlog_write_initial_log_record_low(
 	byte*		log_ptr,
 	mtr_t*		mtr)
 {
-	ut_ad(type <= MLOG_BIGGEST_TYPE || EXTRA_CHECK_MLOG_NUMBER(type));
+	ut_ad(type <= MLOG_BIGGEST_TYPE);
 	ut_ad(type == MLOG_FILE_NAME
 	      || type == MLOG_FILE_DELETE
 	      || type == MLOG_FILE_CREATE2
 	      || type == MLOG_FILE_RENAME2
 	      || type == MLOG_INDEX_LOAD
 	      || type == MLOG_TRUNCATE
-	      || type == MLOG_FILE_WRITE_CRYPT_DATA
 	      || mtr->is_named_space(space_id));
 
 	mach_write_to_1(log_ptr, type);

storage/innobase/log/log0recv.cc

@@ -1462,10 +1462,8 @@ recv_parse_or_apply_log_rec_body(
 		}
 #endif /* UNIV_DEBUG */
 		ptr = mlog_parse_nbytes(type, ptr, end_ptr, page, page_zip);
-		if (ptr != NULL && page != NULL
-		    && page_no == 0 && type == MLOG_4BYTES) {
-			ulint	offs = mach_read_from_2(old_ptr);
-			switch (offs) {
+		if (ptr && page && !page_no && type == MLOG_4BYTES) {
+			switch (ulint offs = mach_read_from_2(old_ptr)) {
 				fil_space_t*	space;
 				ulint		val;
 			default:
@@ -1662,6 +1660,25 @@ recv_parse_or_apply_log_rec_body(
 		this record yet. */
 		break;
 	case MLOG_WRITE_STRING:
+		if (page_no || mach_read_from_2(ptr + 2)
+		    != 11 + MY_AES_BLOCK_SIZE) {
+			/* Not writing crypt_info */
+		} else if (fil_space_t* space
+			   = fil_space_acquire_silent(space_id)) {
+			if (mach_read_from_2(ptr)
+			    == FSP_HEADER_OFFSET + XDES_ARR_OFFSET + MAGIC_SZ
+			    + space->physical_size() * XDES_SIZE
+			    / FSP_EXTENT_SIZE
+			    && (ptr[4] == CRYPT_SCHEME_UNENCRYPTED
+				|| ptr[4] == CRYPT_SCHEME_1)
+			    && ptr[5] == MY_AES_BLOCK_SIZE
+			    && ptr[6 + MY_AES_BLOCK_SIZE + 4 + 4]
+			    <= FIL_ENCRYPTION_OFF) {
+				/* from fil_space_crypt_t::write_page0() */
+				fil_crypt_parse(space, ptr + 4);
+			}
+			space->release();
+		}
 		ptr = mlog_parse_string(ptr, end_ptr, page, page_zip);
 		break;
 	case MLOG_ZIP_WRITE_NODE_PTR:

storage/innobase/mtr/mtr0log.cc

@@ -64,7 +64,7 @@ mlog_write_initial_log_record(
 {
 	byte*	log_ptr;
 
-	ut_ad(type <= MLOG_BIGGEST_TYPE || EXTRA_CHECK_MLOG_NUMBER(type));
+	ut_ad(type <= MLOG_BIGGEST_TYPE);
 	ut_ad(type > MLOG_8BYTES);
 
 	log_ptr = mlog_open(mtr, 11);