MDEV-19348 MariaBackup prepare fails with corruption on diskbb-10.2-MDEV-19348

Problem:
========
 For alter table.. add index command, MLOG_INDEX_LOAD log ensures that only
index pages are flushed to disk. It doesn't ensure about page 0, page 1, page 2.
During recovery, redo of page0 applies after any index page read. It leads to
failure of decryption of index page.

Solution:
=========
- While parsing of redo log, store the space ids of failed crypt_data initialization
and ignore the pages of tablespace if the key version and space->crypt_data doesn't
match.
- Try to apply the redo log for ignored page after initializing the crypt_data
for the tablespace.

5 files changed, 108 insertions, 4 deletions

diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index e3c6605652f..31e97536e4a 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -6068,6 +6068,14 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
 
 		} else if (read_space_id == 0 && read_page_no == 0) {
 			/* This is likely an uninitialized page. */
+		} else if (key_version != 0 && !space->crypt_data
+			   && read_page_no != 0
+			   && recv_recover_ignore_crypt_page(bpage->id)) {
+			/* Page is encrypted but space->crypt_data
+			is not yet updated during recovery */
+			buf_corrupt_page_release(bpage, space);
+			fil_space_release_for_io(space);
+			return DB_SUCCESS;
 		} else if ((bpage->id.space() != TRX_SYS_SPACE
 			    && bpage->id.space() != read_space_id)
 			   || bpage->id.page_no() != read_page_no) {
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 9b8373a4d55..718dc847617 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -423,6 +423,8 @@ Parse a MLOG_FILE_WRITE_CRYPT_DATA log entry
 @param[in]	ptr		Log entry start
 @param[in]	end_ptr		Log entry end
 @param[in]	block		buffer block
+@param[out]	init_crypt	Initialize the crypt_data
+				for the space
 @return position on log buffer */
 UNIV_INTERN
 byte*
@@ -430,7 +432,8 @@ fil_parse_write_crypt_data(
 	byte*			ptr,
 	const byte*		end_ptr,
 	const buf_block_t*	block,
-	dberr_t*		err)
+	dberr_t*		err,
+	bool*			init_crypt)
 {
 	/* check that redo log entry is complete */
 	uint entry_size =
@@ -481,6 +484,7 @@ fil_parse_write_crypt_data(
 	fil_space_t* space = fil_space_get_by_id(space_id);
 
 	if (!space) {
+		*init_crypt = false;
 		mutex_exit(&fil_system->mutex);
 		return ptr + len;
 	}
@@ -502,6 +506,7 @@ fil_parse_write_crypt_data(
 		space->crypt_data = crypt_data;
 	}
 
+	*init_crypt = false;
 	mutex_exit(&fil_system->mutex);
 
 	if (crypt_data->should_encrypt() && !crypt_data->is_key_found()) {
diff --git a/storage/innobase/include/fil0crypt.h b/storage/innobase/include/fil0crypt.h
index 9b9f0da6fdd..cba68982b2d 100644
--- a/storage/innobase/include/fil0crypt.h
+++ b/storage/innobase/include/fil0crypt.h
@@ -298,6 +298,8 @@ Parse a MLOG_FILE_WRITE_CRYPT_DATA log entry
 @param[in]	end_ptr		Log entry end
 @param[in]	block		buffer block
 @param[out]	err		DB_SUCCESS or DB_DECRYPTION_FAILED
+@param[out]	init_crypt	Initialize the crypt data for
+				the space
 @return position on log buffer */
 UNIV_INTERN
 byte*
@@ -305,7 +307,8 @@ fil_parse_write_crypt_data(
 	byte*			ptr,
 	const byte*		end_ptr,
 	const buf_block_t*	block,
-	dberr_t*		err)
+	dberr_t*		err,
+	bool*			init_crypt)
 	MY_ATTRIBUTE((warn_unused_result));
 
 /** Encrypt a buffer.
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 74aa4fcb517..ce3cf67852a 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -49,6 +49,13 @@ dberr_t
 recv_find_max_checkpoint(ulint* max_field)
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
+/** Ignore the page because key version doesn't exist in page 0 and it is
+present in the given page id.
+@param[in]	page_id		Given page id is to be ignored
+@return ignore the page to check corruption */
+bool
+recv_recover_ignore_crypt_page(const page_id_t page_id);
+
 /** Reduces recv_sys->n_addrs for the corrupted page.
 This function should called when srv_force_recovery > 0.
 @param[in]	page_id page id of the corrupted page */
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 2df1a8950d8..3e4aa796977 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -182,6 +182,13 @@ typedef std::map<
 
 static recv_spaces_t	recv_spaces;
 
+/** Store the list of tablespace ids which was failed to initialize
+crypt data while parsing MLOG_FILE_WRITE_CRYPT_DATA. */
+static std::set<ulint>	recv_mlog_crypt_ids;
+
+/** Store the page ids for crypt_data mismatch happened. */
+static std::set<page_id_t> recv_ignore_pages;
+
 /** States of recv_addr_t */
 enum recv_addr_state {
 	/** not yet processed */
@@ -1829,12 +1836,20 @@ parse_log:
 		}
 		break;
 	case MLOG_FILE_WRITE_CRYPT_DATA:
-		dberr_t err;
-		ptr = const_cast<byte*>(fil_parse_write_crypt_data(ptr, end_ptr, block, &err));
+		dberr_t	err;
+		bool	init_crypt;
+		ptr = const_cast<byte*>(fil_parse_write_crypt_data(
+					ptr, end_ptr, block, &err,
+					&init_crypt));
 
 		if (err != DB_SUCCESS) {
 			recv_sys->found_corrupt_log = TRUE;
 		}
+
+		if (!init_crypt) {
+			recv_mlog_crypt_ids.insert(space_id);
+		}
+
 		break;
 	default:
 		ptr = NULL;
@@ -2212,6 +2227,27 @@ skip_log:
 	}
 }
 
+/** Ignore the page because key version doesn't exist in page 0 and it is
+present in the given page id.
+@param[in]	page_id		Given page id is to be ignored
+@return ignore the page to check corruption */
+bool recv_recover_ignore_crypt_page(const page_id_t page_id)
+{
+	if (!recv_recovery_is_on()) {
+		return false;
+	}
+
+	mutex_enter(&recv_sys->mutex);
+
+	bool ignore = (recv_mlog_crypt_ids.find(page_id.space())
+		       != recv_mlog_crypt_ids.end());
+
+	recv_ignore_pages.insert(page_id);
+
+	mutex_exit(&recv_sys->mutex);
+	return ignore;
+}
+
 /** Reduces recv_sys->n_addrs for the corrupted page.
 This function should called when srv_force_recovery > 0.
 @param[in]	page_id	page id of the corrupted page */
@@ -2308,6 +2344,47 @@ static void recv_read_in_area(const page_id_t page_id)
 	mutex_enter(&recv_sys->mutex);
 }
 
+/** Apply the redo log for the ignored pages. */
+static void recv_apply_for_ignored_pages()
+{
+	ut_ad(mutex_own(&recv_sys->mutex));
+
+	for (std::set<page_id_t>::iterator it = recv_ignore_pages.begin();
+	     it != recv_ignore_pages.end(); it++) {
+
+		if (recv_mlog_crypt_ids.find(it->space())
+		    != recv_mlog_crypt_ids.end()) {
+
+			recv_addr_t* recv_addr = recv_get_fil_addr_struct(
+							it->space(), 0);
+
+			if (recv_addr->state != RECV_BEING_PROCESSED
+			    && recv_addr->state != RECV_PROCESSED) {
+				continue;
+			}
+
+			recv_mlog_crypt_ids.erase(it->space());
+		}
+
+		recv_addr_t* recv_addr = recv_get_fil_addr_struct(
+				it->space(), it->page_no());
+		mtr_t	mtr;
+
+		if (recv_addr->state != RECV_BEING_PROCESSED
+		    && recv_addr->state != RECV_PROCESSED) {
+
+			mutex_exit(&recv_sys->mutex);
+			mtr.start();
+			mtr.set_log_mode(MTR_LOG_NONE);
+			buf_page_get_gen(*it, univ_page_size,
+				RW_X_LATCH, NULL, BUF_GET,
+				__FILE__, __LINE__, &mtr, NULL);
+			mtr.commit();
+			mutex_enter(&recv_sys->mutex);
+		}
+	}
+}
+
 /** Apply the hash table of stored log records to persistent data pages.
 @param[in]	last_batch	whether the change buffer merge will be
 				performed as part of the operation */
@@ -2508,6 +2585,10 @@ do_read:
 		os_thread_sleep(500000);
 
 		mutex_enter(&(recv_sys->mutex));
+
+		if (recv_ignore_pages.size()) {
+			recv_apply_for_ignored_pages();
+		}
 	}
 
 	if (!last_batch) {