MDEV-11799 InnoDB can abort if the doublewrite buffer contains a bad and a good copybb-10.2-MDEV-11799

- Addressed marko's review comments.

4 files changed, 43 insertions, 53 deletions

diff --git a/storage/innobase/buf/buf0dblwr.cc b/storage/innobase/buf/buf0dblwr.cc
index 475e31ce305..02edc8e80ce 100644
--- a/storage/innobase/buf/buf0dblwr.cc
+++ b/storage/innobase/buf/buf0dblwr.cc
@@ -551,12 +551,11 @@ buf_dblwr_process()
 		byte*	page		= *i;
 		ulint	space_id	= page_get_space_id(page);
 
-		/* Ignore the dblwr pages for the following:
-		1) find_page could have set lsn to zero for oldest page.
-		2) dblwr page shouldn't be lesser than checkpoint lsn or
-		greater than last scanned lsn. */
+		/* Ignore the dblwr pages if dblwr page shouldn't
+		be lesser than checkpoint lsn or greater than
+		last scanned lsn. */
 		ulint lsn= mach_read_from_8(page + FIL_PAGE_LSN);
-		if (!lsn || recv_sys->parse_start_lsn > lsn) {
+		if (recv_sys->parse_start_lsn > lsn) {
 			continue;
 		}
 
@@ -633,7 +632,7 @@ next_process:
 			ignore this page (there should be redo log
 			records to initialize it). */
 		} else if (recv_dblwr.validate_page(
-				space, page_no, read_buf, false, buf)) {
+				page_id, read_buf, space, buf)) {
 			goto next_process;
 		} else {
 			/* We intentionally skip this message for
@@ -650,18 +649,18 @@ next_process:
 			goto next_process;
 		}
 
-		fil_space_release_for_io(space);
 		/* Write the good page from the doublewrite buffer to
 		the intended position. */
 
 		IORequest	write_request(IORequest::WRITE);
 
 		fil_io(write_request, true, page_id, page_size,
-		       0, page_size.physical(),
-                       const_cast<byte*>(page), NULL);
+		       0, page_size.physical(), page, NULL);
 
 		ib::info() << "Recovered page " << page_id
 			<< " from the doublewrite buffer.";
+
+		goto next_process;
 	}
 
 	recv_dblwr.pages.clear();
diff --git a/storage/innobase/fsp/fsp0file.cc b/storage/innobase/fsp/fsp0file.cc
index 5abdf22d939..26836330671 100644
--- a/storage/innobase/fsp/fsp0file.cc
+++ b/storage/innobase/fsp/fsp0file.cc
@@ -788,15 +788,7 @@ Datafile::restore_from_doublewrite()
 		FSP_HEADER_OFFSET + FSP_SPACE_FLAGS + page);
 
 	if (!fsp_flags_is_valid(flags, m_space_id)) {
-		ulint cflags = fsp_flags_convert_from_101(flags);
-		if (cflags == ULINT_UNDEFINED) {
-			ib::warn()
-				<< "Ignoring a doublewrite copy of page "
-				<< page_id
-				<< " due to invalid flags " << ib::hex(flags);
-			return(true);
-		}
-		flags = cflags;
+		flags = fsp_flags_convert_from_101(flags);
 		/* The flags on the page should be converted later. */
 	}
 
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index e154d853b33..d17c7148d44 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -202,29 +202,29 @@ struct recv_dblwr_t {
 	}
 
 	/** Validate the page and check the metadata also
-	@param[in]	space		tablespace of page to be validated
-	@param[in]	page_no		page number to be validated
+	@param[in]	page_id		page id to be validated
 	@param[in]	page		page to be validated
-	@param[in]	dblwr_page	variable to indicate doublewrite
-					buffer page
-	@param[in]	tmp_buf		temporary buffer to decrypt and
+	@param[in]	space		tablespace of the page
+	@param[in,out]	tmp_buf		temporary buffer to decrypt and
 					decompress the page
 	@return true if success */
-	bool validate_page(fil_space_t *space, ulint page_no,
-                           const byte *page, bool dblwr_page=false,
-                           byte *tmp_buf=nullptr);
+	bool validate_page(const page_id_t page_id,
+			   const byte *page,
+			   const fil_space_t *space,
+			   byte *tmp_buf);
 
 	/** Find a doublewrite copy of a page.
 	@param[in]	space_id	tablespace identifier
 	@param[in]	page_no		page number
 	@param[in]	space		tablespace object to do
                                         dblwr page validation
-	@param[in]	tmp_buf		temporary buffer to decrypt and
+	@param[in,out]	tmp_buf		temporary buffer to decrypt and
 					decompress the page
 	@return	page frame
 	@retval NULL if no page was found */
         byte* find_page(ulint space_id, ulint page_no,
-                        fil_space_t *space=nullptr, byte *tmp_buf=nullptr);
+                        const fil_space_t *space=NULL,
+                        byte *tmp_buf=NULL);
 
 	typedef std::list<byte*, ut_allocator<byte*> >	list;
 
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 524f359e8d0..9562e50cb1c 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -4088,33 +4088,38 @@ recv_recovery_rollback_active(void)
 	}
 }
 
-bool recv_dblwr_t::validate_page(fil_space_t* space, ulint page_no,
-                                 const byte* page, bool dblwr_page,
-                                 byte* tmp_buf)
+bool recv_dblwr_t::validate_page(const page_id_t page_id,
+                                 const byte *page,
+                                 const fil_space_t *space,
+                                 byte *tmp_buf)
 {
-  ut_ad(tmp_buf);
-  byte *tmp_frame = tmp_buf;
-  byte *tmp_page = tmp_buf + srv_page_size;
-  const uint16_t page_type= mach_read_from_2(page + FIL_PAGE_TYPE);
-  const page_size_t page_size(space->flags);
-  const ulint phy_size= page_size.physical();
-  const bool expect_encrypted= space->crypt_data &&
-    space->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED;
-
-  if (dblwr_page)
+  ut_ad(space || !page_id.page_no());
+  if (page_id.page_no() == 0)
   {
     ulint flags= fsp_header_get_flags(page);
-    if (page_no == 0 && !fsp_flags_is_valid(flags, space->id) &&
+    if (!fsp_flags_is_valid(flags, page_id.space()) &&
         fsp_flags_convert_from_101(flags) == ULINT_UNDEFINED)
     {
       ib::warn() << "Ignoring a doublewrite copy of page "
-	         << page_id_t(space->id, page_no)
+	         << page_id
 		 << "due to invalid flags " << ib::hex(flags);
       return false;
     }
+
+    return !buf_page_is_corrupted(true, page, page_size_t(flags), NULL);
   }
 
-  if (page_no && expect_encrypted &&
+  ut_ad(tmp_buf);
+  ut_ad(space);
+  byte *tmp_frame = tmp_buf;
+  byte *tmp_page = tmp_buf + srv_page_size;
+  const uint16_t page_type= mach_read_from_2(page + FIL_PAGE_TYPE);
+  const page_size_t page_size(space->flags);
+  const ulint phy_size= page_size.physical();
+  const bool expect_encrypted= space->crypt_data &&
+    space->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED;
+
+  if (expect_encrypted &&
       mach_read_from_4(page + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION))
   {
     if (!fil_space_verify_crypt_checksum(page, page_size))
@@ -4143,15 +4148,9 @@ bool recv_dblwr_t::validate_page(fil_space_t* space, ulint page_no,
   return !buf_page_is_corrupted(true, page, page_size, space);
 }
 
-/** Find a doublewrite copy of a page.
-@param[in]	space_id	tablespace identifier
-@param[in]	page_no		page number
-@param[in]	validate	validate the doublewrite buffer page
-@return	page frame
-@retval NULL if no page was found */
 byte*
 recv_dblwr_t::find_page(ulint space_id, ulint page_no,
-                        fil_space_t *space, byte *tmp_buf)
+                        const fil_space_t *space, byte *tmp_buf)
 {
   byte *result= NULL;
   lsn_t max_lsn= 0;
@@ -4164,9 +4163,9 @@ recv_dblwr_t::find_page(ulint space_id, ulint page_no,
       continue;
     const lsn_t lsn= mach_read_from_8(page + FIL_PAGE_LSN);
     if (lsn <= max_lsn ||
-        (space && !validate_page(space, page_no, page, true, tmp_buf)))
+        !validate_page(page_id_t(space_id, page_no), page, space, tmp_buf))
     {
-      /* Reset the PAGE_LSN to 0 for the older dblwr pages */
+      /* Mark processed for subsequent iterations in buf_dblwr_process() */
       memset(page + FIL_PAGE_LSN, 0, 8);
       continue;
     }