MDEV-27014 InnoDB fails to restore page 0 from the doublewrite buffer

This patch reverts the commit cab8f4b552.
InnoDB fails to restore page0 from doublewrite buffer when the
tablespace is being deferred. In that case, InnoDB doesn't find
INIT_PAGE redo log record for page0 and it leads to failure.
InnoDB should recovery page0 from doublewrite buffer for the
deferred tablespace before applying the redo log records.

Added deferred_dblwr() to restore page0 of deferred tablespace
from doublewrite buffer

storage/innobase/fil/fil0fil.cc

@@ -2576,8 +2576,7 @@ fil_ibd_load(
 
 	/* Read and validate the first page of the tablespace.
 	Assign a tablespace name based on the tablespace type. */
-	switch (file.validate_for_recovery(
-			static_cast<uint32_t>(space_id))) {
+	switch (file.validate_for_recovery()) {
 		os_offset_t	minimum_size;
 	case DB_SUCCESS:
 		deferred_space = file.m_defer;

storage/innobase/fsp/fsp0file.cc

@@ -387,11 +387,10 @@ exist and be successfully opened. We initially open it in read-only mode
 because we just want to read the SpaceID.  However, if the first page is
 corrupt and needs to be restored from the doublewrite buffer, we will
 reopen it in write mode and ry to restore that page.
-@param	space_id	space id to validate for recovery
 @retval DB_SUCCESS if tablespace is valid, DB_ERROR if not.
 m_is_valid is also set true on success, else false. */
 dberr_t
-Datafile::validate_for_recovery(uint32_t space_id)
+Datafile::validate_for_recovery()
 {
 	dberr_t err;
 
@@ -434,25 +433,15 @@ Datafile::validate_for_recovery(uint32_t space_id)
 			}
 		}
 
-		bool empty_tablespace= (m_space_id == ULINT_UNDEFINED);
-		if (empty_tablespace && space_id) {
-			/* Set space id to find out whether
-			the page exist in double write buffer */
-			m_space_id = space_id;
+		if (m_space_id == ULINT_UNDEFINED) {
+			return DB_SUCCESS; /* empty file */
 		}
 
 		if (restore_from_doublewrite()) {
-			if (!m_defer) {
-				return DB_CORRUPTION;
-			} else if (!empty_tablespace) {
+			if (m_defer) {
 				return err;
-			} else {
-				/* Reset the space id. InnoDB
-				could rebuild the page0
-				from redo log */
-				m_space_id = ULINT_UNDEFINED;
-				return DB_SUCCESS; /* empty file */
 			}
+			return(DB_CORRUPTION);
 		}
 
 		/* Free the previously read first page and then re-validate. */
@@ -780,14 +769,10 @@ Datafile::restore_from_doublewrite()
 		in the doublewrite buffer, then the recovery is going to fail
 		now. Hence this is treated as an error. */
 
-		if (!m_defer) {
-			ib::error()
-				<< "Corrupted page " << page_id
-				<< " of datafile '" << m_filepath
-				<< "' could not be found in the "
-				<< "doublewrite buffer.";
-		}
-
+		ib::error()
+			<< "Corrupted page " << page_id
+			<< " of datafile '" << m_filepath
+			<< "' could not be found in the doublewrite buffer.";
 		return(true);
 	}
 

storage/innobase/include/fsp0file.h

@@ -207,10 +207,9 @@ public:
 	However, if the first page is corrupt and needs to be restored
 	from the doublewrite buffer, we will reopen it in write mode and
 	ry to restore that page.
-	@param	space_id	space id to validate for recovery
 	@retval DB_SUCCESS if tablespace is valid, DB_ERROR if not.
 	m_is_valid is also set true on success, else false. */
-	dberr_t validate_for_recovery(uint32_t space_id=0)
+	dberr_t validate_for_recovery()
 		MY_ATTRIBUTE((warn_unused_result));
 
 	/** Checks the consistency of the first page of a datafile when the

storage/innobase/log/log0recv.cc

@@ -785,6 +785,59 @@ processed:
     space->size_in_header= size;
     return space;
   }
+
+  /* Recover the page0 of deferred tablespace from doublewrite buffer.*/
+  void deferred_dblwr()
+  {
+    for (auto d= defers.begin(); d != defers.end(); )
+    {
+      if (d->second.deleted)
+      {
+      next_item:
+        d++;
+        continue;
+      }
+      const page_id_t page_id{d->first, 0};
+      const byte *page= recv_sys.dblwr.find_page(page_id);
+      if (!page)
+        goto next_item;
+      const uint32_t space_id= mach_read_from_4(page + FIL_PAGE_SPACE_ID);
+      const uint32_t flags= fsp_header_get_flags(page);
+      const uint32_t page_no= mach_read_from_4(page + FIL_PAGE_OFFSET);
+      const uint32_t size= fsp_header_get_field(page, FSP_SIZE);
+
+      if (page_no == 0 && space_id == d->first && size >= 4 &&
+          fil_space_t::is_valid_flags(flags, space_id) &&
+          fil_space_t::logical_size(flags) == srv_page_size)
+      {
+        recv_spaces_t::iterator it {recv_spaces.find(d->first)};
+        ut_ad(it != recv_spaces.end());
+
+        fil_space_t *space= create(
+          it, d->second.file_name.c_str(), flags,
+          fil_space_read_crypt_data(fil_space_t::zip_size(flags), page),
+          size);
+
+        space->free_limit= fsp_header_get_field(page, FSP_FREE_LIMIT);
+        space->free_len= flst_get_len(FSP_HEADER_OFFSET + FSP_FREE + page);
+        fil_node_t *node= UT_LIST_GET_FIRST(space->chain);
+	if (!space->acquire())
+          goto next_item;
+        if (os_file_write(IORequestWrite, node->name, node->handle,
+                      page, 0, fil_space_t::physical_size(flags) !=
+            DB_SUCCESS))
+        {
+          space->release();
+          goto next_item;
+        }
+        space->release();
+        it->second.space= space;
+        defers.erase(d++);
+        continue;
+      }
+      goto next_item;
+    }
+  }
 }
 deferred_spaces;
 
@@ -4068,6 +4121,7 @@ completed:
 		recv_sys.parse_start_lsn = checkpoint_lsn;
 
 		if (srv_operation == SRV_OPERATION_NORMAL) {
+			deferred_spaces.deferred_dblwr();
 			buf_dblwr.recover();
 		}