MDEV-24626 Remove synchronous write of page0 and flushing file during file creation10.6-MDEV-24626

InnoDB holds dict_sys mutex while doing file i/o operation
during creation of table. InnoDB remove the page0 write and
flushing the file then InnoDB should handle this scenario
during recovery. InnoDB may encounter zero filled page0 while
processing FILE redo log records. In that case, InnoDB should
defer the tablespace initialization. After parsing all redo log
records, InnoDB should initialize the tablespace based on
redo log of page0.

9 files changed, 259 insertions, 131 deletions

diff --git a/mysql-test/suite/innodb/disabled.def b/mysql-test/suite/innodb/disabled.def
index 35c941f8af7..adc9aa27fc1 100644
--- a/mysql-test/suite/innodb/disabled.def
+++ b/mysql-test/suite/innodb/disabled.def
@@ -11,3 +11,4 @@
 ##############################################################################
 
 create-index-debug : MDEV-13680 InnoDB may crash when btr_page_alloc() fails
+log_file_name : MDEV-24626 Remove synchronous write of page0 and flushing file during file creation
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 416e8be746c..594c7dcc314 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -3267,11 +3267,12 @@ buf_block_t*
 buf_page_create(fil_space_t *space, uint32_t offset,
                 ulint zip_size, mtr_t *mtr, buf_block_t *free_block)
 {
-  page_id_t page_id(space->id, offset);
+  page_id_t page_id (space ? space->id : offset, space ? offset : 0);
   ut_ad(mtr->is_active());
   ut_ad(page_id.space() != 0 || !zip_size);
 
-  space->free_page(offset, false);
+  if (space)
+    space->free_page(offset, false);
   free_block->initialise(page_id, zip_size, 1);
 
   const ulint fold= page_id.fold();
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 7cd619827fa..ef511a0d419 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -347,8 +347,9 @@ fil_node_t* fil_space_t::add(const char* name, pfs_os_file_t handle,
 
 /** Open a tablespace file.
 @param node  data file
+@param validate validate the page0
 @return whether the file was successfully opened */
-static bool fil_node_open_file_low(fil_node_t *node)
+static bool fil_node_open_file_low(fil_node_t *node, bool validate=true)
 {
   ut_ad(!node->is_open());
   ut_ad(node->space->is_closing());
@@ -383,7 +384,7 @@ static bool fil_node_open_file_low(fil_node_t *node)
   }
 
   if (node->size);
-  else if (!node->read_page0() || !fil_comp_algo_validate(node->space))
+  else if (!node->read_page0(validate) || !fil_comp_algo_validate(node->space))
   {
     os_file_close(node->handle);
     node->handle= OS_FILE_CLOSED;
@@ -406,8 +407,9 @@ static bool fil_node_open_file_low(fil_node_t *node)
 
 /** Open a tablespace file.
 @param node  data file
+@param validate_page0 validate the page0
 @return whether the file was successfully opened */
-static bool fil_node_open_file(fil_node_t *node)
+bool fil_node_open_file(fil_node_t *node, bool validate_page0)
 {
   mysql_mutex_assert_owner(&fil_system.mutex);
   ut_ad(!node->is_open());
@@ -439,7 +441,7 @@ static bool fil_node_open_file(fil_node_t *node)
     }
   }
 
-  return fil_node_open_file_low(node);
+  return fil_node_open_file_low(node, validate_page0);
 }
 
 /** Close the file handle. */
@@ -638,7 +640,7 @@ fil_space_extend_must_retry(
 }
 
 /** @return whether the file is usable for io() */
-ATTRIBUTE_COLD bool fil_space_t::prepare(bool have_mutex)
+ATTRIBUTE_COLD bool fil_space_t::prepare(bool have_mutex, bool validate_page0)
 {
   ut_ad(referenced());
   if (!have_mutex)
@@ -648,7 +650,8 @@ ATTRIBUTE_COLD bool fil_space_t::prepare(bool have_mutex)
   ut_ad(!id || purpose == FIL_TYPE_TEMPORARY ||
         node == UT_LIST_GET_FIRST(chain));
 
-  const bool is_open= node && (node->is_open() || fil_node_open_file(node));
+  const bool is_open= node && (node->is_open()
+                               || fil_node_open_file(node, validate_page0));
 
   if (!is_open)
     release();
@@ -1446,7 +1449,7 @@ fil_write_flushed_lsn(
 @param id      tablespace identifier
 @return tablespace
 @retval nullptr if the tablespace is missing or inaccessible */
-fil_space_t *fil_space_t::get(ulint id)
+fil_space_t *fil_space_t::get(ulint id, bool validate_page0)
 {
   mysql_mutex_lock(&fil_system.mutex);
   fil_space_t *space= fil_space_get_by_id(id);
@@ -1456,8 +1459,9 @@ fil_space_t *fil_space_t::get(ulint id)
   if (n & STOPPING)
     space= nullptr;
 
-  if ((n & CLOSING) && !space->prepare())
+  if ((n & CLOSING) && !space->prepare(false, validate_page0)) {
     space= nullptr;
+  }
 
   return space;
 }
@@ -2309,45 +2313,8 @@ err_exit:
 		crypt_data->fill_page0(flags, page);
 	}
 
-	if (ulint zip_size = fil_space_t::zip_size(flags)) {
-		page_zip_des_t	page_zip;
-		page_zip_set_size(&page_zip, zip_size);
-		page_zip.data = page + srv_page_size;
-#ifdef UNIV_DEBUG
-		page_zip.m_start = 0;
-#endif /* UNIV_DEBUG */
-		page_zip.m_end = 0;
-		page_zip.m_nonempty = 0;
-		page_zip.n_blobs = 0;
-
-		buf_flush_init_for_writing(NULL, page, &page_zip, false);
-
-		*err = os_file_write(IORequestWrite, path, file,
-				     page_zip.data, 0, zip_size);
-	} else {
-		buf_flush_init_for_writing(NULL, page, NULL,
-					   fil_space_t::full_crc32(flags));
-
-		*err = os_file_write(IORequestWrite, path, file,
-				     page, 0, srv_page_size);
-	}
-
 	aligned_free(page);
 
-	if (*err != DB_SUCCESS) {
-		ib::error()
-			<< "Could not write the first page to"
-			<< " tablespace '" << path << "'";
-		goto err_exit;
-	}
-
-	if (!os_file_flush(file)) {
-		ib::error() << "File flush of tablespace '"
-			<< path << "' failed";
-		*err = DB_ERROR;
-		goto err_exit;
-	}
-
 	if (has_data_dir) {
 		/* Make the ISL file if the IBD file is not
 		in the default location. */
@@ -2900,7 +2867,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()) {
+	switch (dberr_t err= file.validate_for_recovery()) {
 		os_offset_t	minimum_size;
 	case DB_SUCCESS:
 		if (file.space_id() != space_id) {
@@ -2913,6 +2880,8 @@ fil_ibd_load(
 				<< space_id << ".";
 			return(FIL_LOAD_ID_CHANGED);
 		}
+		/* fall through */
+	case DB_DEFER_TABLESPACE:
 		/* Get and test the file size. */
 		size = os_file_get_size(file.handle());
 
@@ -2933,16 +2902,16 @@ fil_ibd_load(
 				<< file.filepath() << "' is only " << size
 				<< ", should be at least " << minimum_size
 				<< "!";
+		} else if (err == DB_DEFER_TABLESPACE) {
+			return FIL_LOAD_DEFER;
 		} else {
 			/* Everything is fine so far. */
 			break;
 		}
 
 		/* fall through */
-
 	case DB_TABLESPACE_EXISTS:
 		return(FIL_LOAD_INVALID);
-
 	default:
 		return(FIL_LOAD_NOT_FOUND);
 	}
diff --git a/storage/innobase/fsp/fsp0file.cc b/storage/innobase/fsp/fsp0file.cc
index 57164113647..7e455d3501a 100644
--- a/storage/innobase/fsp/fsp0file.cc
+++ b/storage/innobase/fsp/fsp0file.cc
@@ -447,6 +447,7 @@ Datafile::validate_for_recovery()
 	switch (err) {
 	case DB_SUCCESS:
 	case DB_TABLESPACE_EXISTS:
+	case DB_DEFER_TABLESPACE:
 		break;
 
 	default:
@@ -535,6 +536,11 @@ err_exit:
 		}
 
 		if (nonzero_bytes == 0) {
+			if (recv_recovery_is_on()) {
+				free_first_page();
+				return DB_DEFER_TABLESPACE;
+			}
+
 			error_txt = "Header page consists of zero bytes";
 			goto err_exit;
 		}
diff --git a/storage/innobase/include/db0err.h b/storage/innobase/include/db0err.h
index 6cfc63f4a9e..41835451a30 100644
--- a/storage/innobase/include/db0err.h
+++ b/storage/innobase/include/db0err.h
@@ -173,6 +173,7 @@ enum dberr_t {
 	DB_END_OF_INDEX,
 	DB_NOT_FOUND,			/*!< Generic error code for "Not found"
 					type of errors */
+	DB_DEFER_TABLESPACE,
 };
 
 #endif
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index a5c5a22f8ca..e30d4eed4d9 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -917,7 +917,7 @@ public:
   @param id      tablespace identifier
   @return tablespace
   @retval nullptr if the tablespace is missing or inaccessible */
-  static fil_space_t *get(ulint id);
+  static fil_space_t *get(ulint id, bool validate_page0=true);
 
   /** Add/remove the free page in the freed ranges list.
   @param[in] offset     page number to be added
@@ -1030,7 +1030,8 @@ public:
 
 private:
   /** @return whether the file is usable for io() */
-  ATTRIBUTE_COLD bool prepare(bool have_mutex= false);
+  ATTRIBUTE_COLD bool prepare(
+	bool have_mutex= false, bool validate_page0=true);
 #endif /*!UNIV_INNOCHECKSUM */
 };
 
@@ -1082,7 +1083,7 @@ struct fil_node_t final
 
 	/** Read the first page of a data file.
 	@return	whether the page was found valid */
-	bool read_page0();
+	bool read_page0(bool validate=true);
 
 	/** Determine some file metadata when creating or reading the file.
 	@param	file	the file that is being created, or OS_FILE_CLOSED */
@@ -1518,6 +1519,7 @@ inline uint32_t fil_space_t::get_size()
   return size;
 }
 
+bool fil_node_open_file(fil_node_t *node, bool validate_page0=true);
 #include "fil0crypt.h"
 
 /*******************************************************************//**
@@ -1678,7 +1680,9 @@ enum fil_load_status {
 	/** The file(s) were not found */
 	FIL_LOAD_NOT_FOUND,
 	/** The file(s) were not valid */
-	FIL_LOAD_INVALID
+	FIL_LOAD_INVALID,
+	/** The tablespace file was defered to open. */
+	FIL_LOAD_DEFER
 };
 
 /** Open a single-file tablespace and add it to the InnoDB data structures.
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 2b8c9e64b47..16a1c9617c7 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -587,6 +587,8 @@ static recv_spaces_t	recv_spaces;
 /** The last parsed FILE_RENAME records */
 static std::map<uint32_t,std::string> renamed_spaces;
 
+static std::set<uint32_t> defer_spaces;
+
 /** Report an operation to create, delete, or rename a file during backup.
 @param[in]	space_id	tablespace identifier
 @param[in]	create		whether the file is being created
@@ -824,6 +826,8 @@ fil_name_process(char* name, ulint len, ulint space_id, bool deleted)
 				fil_space_free(space_id, false);
 				f.space = NULL;
 			}
+
+			defer_spaces.erase((uint32_t)space_id);
 		}
 
 		ut_ad(f.space == NULL);
@@ -885,7 +889,10 @@ same_space:
 					<< " for tablespace " << space_id;
 			}
 			break;
-
+		case FIL_LOAD_DEFER:
+			if (!p.second) f.name= fname.name;
+			defer_spaces.emplace(space_id);
+			break;
 		case FIL_LOAD_INVALID:
 			ut_ad(space == NULL);
 			if (srv_force_recovery == 0) {
@@ -2083,6 +2090,8 @@ same_page:
       const bool is_init= (b & 0x70) <= INIT_PAGE;
       switch (*store) {
       case STORE_IF_EXISTS:
+        if (defer_spaces.find(space_id) != defer_spaces.end())
+          goto store_defer;
         if (fil_space_t *space= fil_space_t::get(space_id))
         {
           const auto size= space->get_size();
@@ -2094,6 +2103,7 @@ same_page:
           continue;
         /* fall through */
       case STORE_YES:
+store_defer:
         if (!mlog_init.will_avoid_read(id, start_lsn))
           add(id, start_lsn, end_lsn, recs,
               static_cast<size_t>(l + rlen - recs));
@@ -2590,41 +2600,56 @@ inline buf_block_t *recv_sys_t::recover_low(const page_id_t page_id,
   ut_ad(recs.state == page_recv_t::RECV_WILL_NOT_READ);
   buf_block_t* block= nullptr;
   mlog_init_t::init &i= mlog_init.last(page_id);
+  bool first_page= (page_id.page_no() == 0);
   const lsn_t end_lsn = recs.log.last()->lsn;
+
   if (end_lsn < i.lsn)
     DBUG_LOG("ib_log", "skip log for page " << page_id
              << " LSN " << end_lsn << " < " << i.lsn);
-  else if (fil_space_t *space= fil_space_t::get(page_id.space()))
+  fil_space_t *space= fil_space_t::get(page_id.space());
+  if (!space && !first_page)
+    return block;
+
+  mtr.start();
+  mtr.set_log_mode(MTR_LOG_NO_REDO);
+
+  ulint zip_size= space ? space->zip_size() : 0;
+  if (!space)
   {
-    mtr.start();
-    mtr.set_log_mode(MTR_LOG_NO_REDO);
-    block= buf_page_create(space, page_id.page_no(), space->zip_size(), &mtr,
-                           b);
-    if (UNIV_UNLIKELY(block != b))
-    {
-      /* The page happened to exist in the buffer pool, or it was just
-      being read in. Before buf_page_get_with_no_latch() returned to
-      buf_page_create(), all changes must have been applied to the
-      page already. */
-      ut_ad(pages.find(page_id) == pages.end());
-      mtr.commit();
-      block= nullptr;
-    }
-    else
-    {
-      ut_ad(&recs == &pages.find(page_id)->second);
-      i.created= true;
-      recv_recover_page(block, mtr, p, space, &i);
-      ut_ad(mtr.has_committed());
-      recs.log.clear();
-      map::iterator r= p++;
-      pages.erase(r);
-      if (pages.empty())
-        pthread_cond_signal(&cond);
-    }
-    space->release();
+    auto it= recv_spaces.find(page_id.space());
+    ut_ad(it != recv_spaces.end());
+    uint32_t flags= it->second.flags;
+    zip_size= fil_space_t::zip_size(flags);
   }
 
+  block= buf_page_create(
+     space, space ? page_id.page_no() : page_id.space(), zip_size, &mtr, b);
+  if (UNIV_UNLIKELY(block != b))
+  {
+    /* The page happened to exist in the buffer pool, or it was just
+    being read in. Before buf_page_get_with_no_latch() returned to
+    buf_page_create(), all changes must have been applied to the
+    page already. */
+    ut_ad(pages.find(page_id) == pages.end());
+    mtr.commit();
+    block= nullptr;
+  }
+  else
+  {
+    ut_ad(&recs == &pages.find(page_id)->second);
+    i.created= true;
+    recv_recover_page(block, mtr, p, space, &i);
+    ut_ad(mtr.has_committed());
+    recs.log.clear();
+    map::iterator r= p++;
+    pages.erase(r);
+    if (pages.empty())
+      pthread_cond_signal(&cond);
+  }
+
+  if (space)
+    space->release();
+
   return block;
 }
 
@@ -2652,6 +2677,122 @@ buf_block_t *recv_sys_t::recover_low(const page_id_t page_id)
   return block;
 }
 
+/** Report a missing tablespace for which page-redo log exists.
+@param[in]	err	previous error code
+@param[in]	i	tablespace descriptor
+@return new error code */
+static
+dberr_t
+recv_init_missing_space(dberr_t err, const recv_spaces_t::const_iterator& i)
+{
+	if (srv_operation == SRV_OPERATION_RESTORE
+	    || srv_operation == SRV_OPERATION_RESTORE_EXPORT) {
+		if (i->second.name.find(TEMP_TABLE_PATH_PREFIX)
+		    != std::string::npos) {
+			ib::warn() << "Tablespace " << i->first << " was not"
+				" found at " << i->second.name << " when"
+				" restoring a (partial?) backup. All redo log"
+				" for this file will be ignored!";
+		}
+		return(err);
+	}
+
+	if (srv_force_recovery == 0) {
+		ib::error() << "Tablespace " << i->first << " was not"
+			" found at " << i->second.name << ".";
+
+		if (err == DB_SUCCESS) {
+			ib::error() << "Set innodb_force_recovery=1 to"
+				" ignore this and to permanently lose"
+				" all changes to the tablespace.";
+			err = DB_TABLESPACE_NOT_FOUND;
+		}
+	} else {
+		ib::warn() << "Tablespace " << i->first << " was not"
+			" found at " << i->second.name << ", and"
+			" innodb_force_recovery was set. All redo log"
+			" for this tablespace will be ignored!";
+	}
+
+	return(err);
+}
+
+static dberr_t
+recv_validate_deferred_first_page(buf_block_t *first_block)
+{
+  ut_ad(first_block);
+  auto it= recv_spaces.find(first_block->page.id().space());
+  ut_ad(it != recv_spaces.end());
+  byte *first_page= UNIV_LIKELY_NULL(first_block->page.zip.data)
+	  ? first_block->page.zip.data
+	  : first_block->frame;
+
+  if (buf_is_zeroes(span<const byte> (first_page, srv_page_size)))
+err_exit:
+    return DB_CORRUPTION;
+  uint32_t space_id= mach_read_from_4(first_page + FIL_PAGE_SPACE_ID);
+  uint32_t flags= fsp_header_get_flags(first_page);
+  uint32_t page_no= mach_read_from_4(first_page + FIL_PAGE_OFFSET);
+  if (space_id >= SRV_SPACE_ID_UPPER_BOUND
+      || page_no > 0
+      || flags != it->second.flags
+      || !fil_space_t::is_valid_flags(flags, space_id)
+      || fil_space_t::logical_size(flags) != srv_page_size)
+    goto err_exit;
+  return DB_SUCCESS;
+}
+
+static dberr_t recv_create_deferred_space(buf_block_t *first_block)
+{
+  auto it= recv_spaces.find(first_block->page.id().space());
+  ut_ad(it != recv_spaces.end());
+  byte *page= UNIV_LIKELY_NULL(first_block->page.zip.data)
+	  ? first_block->page.zip.data
+	  : first_block->frame;
+  char *space_name= fil_path_to_space_name(it->second.name.c_str());
+  const uint32_t size = fsp_header_get_field(page, FSP_SIZE);
+  const uint32_t free_limit = fsp_header_get_field(
+	page, FSP_FREE_LIMIT);
+  const uint32_t free_len = flst_get_len(FSP_HEADER_OFFSET + FSP_FREE
+                                         + page);
+
+  fil_space_crypt_t *crypt_data= fil_space_read_crypt_data(
+     fil_space_t::zip_size(it->second.flags), page);
+  fil_space_t *space= fil_space_t::create(
+    space_name, it->first, it->second.flags, FIL_TYPE_TABLESPACE,
+    crypt_data);
+  if (!space)
+    return DB_TABLESPACE_NOT_FOUND;
+  space->add(it->second.name.c_str(), OS_FILE_CLOSED, 0, false, false);
+
+  space->recv_size= it->second.size;
+  space->size_in_header = size;
+  space->free_limit = free_limit;
+  space->free_len = free_len;
+  space= fil_space_t::get(space->id, false);
+  space->release();
+
+  return DB_SUCCESS;
+}
+
+static dberr_t recv_init_deferred_space(uint32_t space)
+{
+  page_id_t defer_first_page(space, 0);
+  buf_block_t *first_block= recv_sys.recover(defer_first_page);
+  if (first_block)
+  {
+    dberr_t err= recv_validate_deferred_first_page(first_block);
+    if (err == DB_SUCCESS)
+    {
+      err= recv_create_deferred_space(first_block);
+      defer_spaces.erase(space);
+    }
+    return err;
+  }
+
+  return DB_CORRUPTION;
+}
+
 /** Apply buffered log to persistent data pages.
 @param last_batch     whether it is possible to write more redo log */
 void recv_sys_t::apply(bool last_batch)
@@ -2723,6 +2864,24 @@ void recv_sys_t::apply(bool last_batch)
       page_recv_t &recs= p->second;
       ut_ad(!recs.log.empty());
 
+      auto it= defer_spaces.find(page_id.space());
+
+      if (it != defer_spaces.end())
+      {
+        mysql_mutex_unlock(&mutex);
+	dberr_t err= recv_init_deferred_space(page_id.space());
+	if (err != DB_SUCCESS)
+	{
+          auto i= recv_spaces.find(page_id.space());
+	  err= recv_init_missing_space(DB_SUCCESS, i);
+	  recv_sys.set_corrupt_log();
+	  return;
+	}
+	mysql_mutex_lock(&mutex);
+        p= pages.lower_bound(page_id);
+	continue;
+      }
+
       switch (recs.state) {
       case page_recv_t::RECV_BEING_READ:
       case page_recv_t::RECV_BEING_PROCESSED:
@@ -3255,46 +3414,6 @@ recv_group_scan_log_recs(
 	DBUG_RETURN(store == STORE_NO);
 }
 
-/** Report a missing tablespace for which page-redo log exists.
-@param[in]	err	previous error code
-@param[in]	i	tablespace descriptor
-@return new error code */
-static
-dberr_t
-recv_init_missing_space(dberr_t err, const recv_spaces_t::const_iterator& i)
-{
-	if (srv_operation == SRV_OPERATION_RESTORE
-	    || srv_operation == SRV_OPERATION_RESTORE_EXPORT) {
-		if (i->second.name.find(TEMP_TABLE_PATH_PREFIX)
-		    != std::string::npos) {
-			ib::warn() << "Tablespace " << i->first << " was not"
-				" found at " << i->second.name << " when"
-				" restoring a (partial?) backup. All redo log"
-				" for this file will be ignored!";
-		}
-		return(err);
-	}
-
-	if (srv_force_recovery == 0) {
-		ib::error() << "Tablespace " << i->first << " was not"
-			" found at " << i->second.name << ".";
-
-		if (err == DB_SUCCESS) {
-			ib::error() << "Set innodb_force_recovery=1 to"
-				" ignore this and to permanently lose"
-				" all changes to the tablespace.";
-			err = DB_TABLESPACE_NOT_FOUND;
-		}
-	} else {
-		ib::warn() << "Tablespace " << i->first << " was not"
-			" found at " << i->second.name << ", and"
-			" innodb_force_recovery was set. All redo log"
-			" for this tablespace will be ignored!";
-	}
-
-	return(err);
-}
-
 /** Report the missing tablespace and discard the redo logs for the deleted
 tablespace.
 @param[in]	rescan			rescan of redo logs is needed
@@ -3322,6 +3441,10 @@ next:
 		recv_spaces_t::iterator i = recv_spaces.find(space);
 		ut_ad(i != recv_spaces.end());
 
+		if (defer_spaces.find((uint32_t)space) != defer_spaces.end()) {
+			goto next;
+		}
+
 		switch (i->second.status) {
 		case file_name_t::NORMAL:
 			goto next;
@@ -3352,6 +3475,9 @@ func_exit:
 			continue;
 		}
 
+		if (defer_spaces.find((uint32_t)rs.first) != defer_spaces.end())
+		  continue;
+
 		missing_tablespace = true;
 
 		if (srv_force_recovery > 0) {
@@ -3742,6 +3868,18 @@ completed:
 
 	recv_lsn_checks_on = true;
 
+	while(defer_spaces.size()) {
+		uint32_t space_id = *(defer_spaces.begin());
+		dberr_t err= recv_init_deferred_space(space_id);
+		if (err != DB_SUCCESS) {
+			err= recv_init_missing_space(
+				DB_SUCCESS,
+				recv_spaces.find(space_id));
+			recv_sys.set_corrupt_log();
+			return err;
+		}
+	}
+
 	/* The database is now ready to start almost normal processing of user
 	transactions: transaction rollbacks and the application of the log
 	records in the hash table can be run in background. */
diff --git a/storage/innobase/os/os0file.cc b/storage/innobase/os/os0file.cc
index 605c77b577e..3fe115710b9 100644
--- a/storage/innobase/os/os0file.cc
+++ b/storage/innobase/os/os0file.cc
@@ -4475,8 +4475,9 @@ void fil_node_t::find_metadata(os_file_t file
 }
 
 /** Read the first page of a data file.
+@param validate	validate the page0
 @return	whether the page was found valid */
-bool fil_node_t::read_page0()
+bool fil_node_t::read_page0(bool validate)
 {
 	mysql_mutex_assert_owner(&fil_system.mutex);
 	const unsigned psize = space->physical_size();
@@ -4500,6 +4501,10 @@ bool fil_node_t::read_page0()
 		return false;
 	}
 
+	if (!validate)
+	  goto skip_page0;
+
+	{
 	page_t *page= static_cast<byte*>(aligned_malloc(psize, psize));
 	if (os_file_read(IORequestRead, handle, page, 0, psize)
 	    != DB_SUCCESS) {
@@ -4558,6 +4563,14 @@ invalid:
 		return false;
 	}
 
+	space->flags = (space->flags & FSP_FLAGS_MEM_MASK) | flags;
+	ut_ad(space->free_limit == 0 || space->free_limit == free_limit);
+	ut_ad(space->free_len == 0 || space->free_len == free_len);
+	space->size_in_header = size;
+	space->free_limit = free_limit;
+	space->free_len = free_len;
+	}
+skip_page0:
 #ifdef UNIV_LINUX
 	find_metadata(handle, &statbuf);
 #else
@@ -4573,16 +4586,9 @@ invalid:
 		size_bytes &= ~os_offset_t(mask);
 	}
 
-	space->flags = (space->flags & FSP_FLAGS_MEM_MASK) | flags;
-
 	space->punch_hole = space->is_compressed();
 	this->size = uint32_t(size_bytes / psize);
 	space->set_sizes(this->size);
-	ut_ad(space->free_limit == 0 || space->free_limit == free_limit);
-	ut_ad(space->free_len == 0 || space->free_len == free_len);
-	space->size_in_header = size;
-	space->free_limit = free_limit;
-	space->free_len = free_len;
 	return true;
 }
 
diff --git a/storage/innobase/ut/ut0ut.cc b/storage/innobase/ut/ut0ut.cc
index fa04af6de13..0a896aa15cb 100644
--- a/storage/innobase/ut/ut0ut.cc
+++ b/storage/innobase/ut/ut0ut.cc
@@ -468,6 +468,8 @@ ut_strerr(
 		return ("File system does not support punch hole (trim) operation.");
 	case DB_PAGE_CORRUPTED:
 		return("Page read from tablespace is corrupted.");
+	case DB_DEFER_TABLESPACE:
+		return("Deferring the tablespace to load till InnoDB recovers page0");
 
 	/* do not add default: in order to produce a warning if new code
 	is added to the enum but not added here */