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

- Addressed marko's review comments.

4 files changed, 23 insertions, 22 deletions

diff --git a/storage/innobase/buf/buf0dblwr.cc b/storage/innobase/buf/buf0dblwr.cc
index 458b6148c2d..cbbb93b443b 100644
--- a/storage/innobase/buf/buf0dblwr.cc
+++ b/storage/innobase/buf/buf0dblwr.cc
@@ -530,6 +530,7 @@ buf_dblwr_init_or_load_pages(
 void
 buf_dblwr_process()
 {
+	ut_ad(recv_sys.parse_start_lsn);
 	ulint		page_no_dblwr	= 0;
 	byte*		read_buf;
 	recv_dblwr_t&	recv_dblwr	= recv_sys.dblwr;
@@ -548,11 +549,16 @@ buf_dblwr_process()
 		byte*	page		= *i;
 		ulint	space_id	= page_get_space_id(page);
 
-		/* Ignore the dblwr pages if dblwr page shouldn't
-		be lesser than checkpoint lsn or greater than
-		last scanned lsn. */
+		if (!page_get_page_no(page)) {
+			/* page 0 should have been recovered
+			already via Datafile::restore_from_doublewrite() */
+			continue;
+		}
+
 		lsn_t lsn = mach_read_from_8(page + FIL_PAGE_LSN);
 		if (recv_sys.parse_start_lsn > lsn) {
+			/* Pages written before the checkpoint are
+			not useful for recovery. */
 			continue;
 		}
 
@@ -587,7 +593,7 @@ buf_dblwr_process()
 					<< space->name
 					<< " (" << space->size << " pages)";
 			}
-next_process:
+next_page:
 			space->release_for_io();
 			continue;
 		}
@@ -627,7 +633,7 @@ next_process:
 			records to initialize it). */
 		} else if (recv_dblwr.validate_page(
 				page_id, read_buf, space, buf)) {
-			goto next_process;
+			goto next_page;
 		} else {
 			/* We intentionally skip this message for
 			is_all_zero pages. */
@@ -639,7 +645,7 @@ next_process:
 		page = recv_dblwr.find_page(space_id, page_no, space, buf);
 
 		if (!page) {
-			goto next_process;
+			goto next_page;
 		}
 
 		/* Write the good page from the doublewrite buffer to
@@ -653,7 +659,7 @@ next_process:
 		ib::info() << "Recovered page " << page_id
 			<< " from the doublewrite buffer.";
 
-		goto next_process;
+		goto next_page;
 	}
 
 	recv_dblwr.pages.clear();
diff --git a/storage/innobase/fsp/fsp0file.cc b/storage/innobase/fsp/fsp0file.cc
index ccb34c5215a..ba8b2e37ac0 100644
--- a/storage/innobase/fsp/fsp0file.cc
+++ b/storage/innobase/fsp/fsp0file.cc
@@ -799,7 +799,6 @@ Datafile::restore_from_doublewrite()
 
 	if (!fil_space_t::is_valid_flags(flags, m_space_id)) {
 		flags = fsp_flags_convert_from_101(flags);
-		/* The flags on the page should be converted later. */
 	}
 
 	ulint physical_size = fil_space_t::physical_size(flags);
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 8b7ff0706c6..f350adebaa2 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -178,10 +178,8 @@ struct recv_dblwr_t {
 	@param[in,out]	tmp_buf		temporary buffer to decrypt and
 					decompress the page
 	@return true if success */
-	bool validate_page(const page_id_t page_id,
-                           const byte *page,
-                           const fil_space_t *space,
-                           byte *tmp_buf);
+	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
@@ -193,8 +191,8 @@ struct recv_dblwr_t {
 	@return	page frame
 	@retval NULL if no page was found */
 	byte* find_page(ulint space_id, ulint page_no,
-                        const fil_space_t *space=nullptr,
-                        byte *tmp_buf=nullptr);
+			const fil_space_t *space=nullptr,
+			byte *tmp_buf=nullptr);
 
 	typedef std::deque<byte*, ut_allocator<byte*> > list;
 
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 10c29760918..c303c4a1a52 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -3757,10 +3757,8 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 			rescan = true;
 		}
 
+		recv_sys.parse_start_lsn = checkpoint_lsn;
 		if (srv_operation == SRV_OPERATION_NORMAL) {
-			/* Assign the checkpoint lsn in parse_start_lsn
-			to avoid the recovery of oldest dblwr pages */
-			recv_sys.parse_start_lsn = checkpoint_lsn;
 			buf_dblwr_process();
 		}
 
@@ -3951,7 +3949,6 @@ bool recv_dblwr_t::validate_page(const page_id_t page_id,
   }
 
   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);
@@ -3968,7 +3965,7 @@ bool recv_dblwr_t::validate_page(const page_id_t page_id,
       return false;
     if (page_type != FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED)
       return true;
-    if (space->is_compressed())
+    if (space->zip_size())
       return false;
     memcpy(tmp_page, page, space->physical_size());
     if (!fil_space_decrypt(space, tmp_frame, tmp_page))
@@ -3981,9 +3978,10 @@ bool recv_dblwr_t::validate_page(const page_id_t page_id,
     memcpy(tmp_page, page, space->physical_size());
     /* fall through */
   case FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED:
-    return !space->is_compressed() &&
-      fil_page_decompress(tmp_frame, tmp_page, space->flags) == srv_page_size &&
-      !buf_page_is_corrupted(true, tmp_page, space->flags);
+    ulint decomp= fil_page_decompress(tmp_frame, tmp_page, space->flags);
+    if (!decomp || decomp == srv_page_size || space->zip_size())
+      return false;
+    return !buf_page_is_corrupted(true, tmp_page, space->flags);
   }
 
   return !buf_page_is_corrupted(true, page, space->flags);