Bug #22963951 REDUCE THE TIME OF LOOKING FOR

                MLOG_CHECKPOINT DURING CRASH RECOVERY
Analysis:
=========
In mysql-5.7, there can be maximum three scans are possible.
Simultaneous scan and apply is possible only in the third scan.

First scan: Scan all the redo logs from checkpoint lsn and process
only MLOG_FILE_* records during first scan. It scans till the
last MLOG_CHECKPOINT.

Second scan: Scan all redo logs from checkpoint lsn and add log
records to hash table. It verifies whether space id is having
corresponding MLOG_FILE_NAME record. If the hash table heap memory
is reached the threshold then stop adding records to hash but it
continues to scan till end of the redo log file.

Third scan: Scan all redo logs from checkpoint lsn and add log records
to hash table only if the tablespace exists. If the heap memory reached
the threshold then simultaneous scan and apply will happen.

Note: Third scan only happens if hash table memory reaches the
threshold during second scan.

Fix:
====

Merge first scan and second scan and make it a single scan. Single scan
if the hash table doesn't reach the threshold limit.
In worst case scenario, it does two scan.

Reviewed-by: Debarun Banerjee <debarun.banerjee@oracle.com>
RB: 14725

Author: Thirunarayanan

storage/innobase/log/log0recv.cc

@@ -387,7 +387,6 @@ fil_name_process(
 @param[in]	first_page_no	first page number in the file
 @param[in]	type		MLOG_FILE_NAME or MLOG_FILE_DELETE
 or MLOG_FILE_CREATE2 or MLOG_FILE_RENAME2
-@param[in]	apply		whether to apply the record
 @return pointer to next redo log record
 @retval NULL if this log record was truncated */
 static
@@ -397,8 +396,7 @@ fil_name_parse(
 	const byte*	end,
 	ulint		space_id,
 	ulint		first_page_no,
-	mlog_id_t	type,
-	bool		apply)
+	mlog_id_t	type)
 {
 	if (type == MLOG_FILE_CREATE2) {
 		if (end < ptr + 4) {
@@ -488,9 +486,6 @@ fil_name_parse(
 			reinterpret_cast<char*>(new_name), new_len,
 			space_id, false);
 
-		if (!apply) {
-			break;
-		}
 		if (!fil_op_replay_rename(
 			    space_id, first_page_no,
 			    reinterpret_cast<const char*>(ptr),
@@ -583,7 +578,6 @@ fil_name_process(
 @param[in]	first_page_no	first page number in the file
 @param[in]	type		MLOG_FILE_NAME or MLOG_FILE_DELETE
 or MLOG_FILE_CREATE2 or MLOG_FILE_RENAME2
-@param[in]	apply		whether to apply the record
 @retval	pointer to next redo log record
 @retval	NULL if this log record was truncated */
 static
@@ -593,8 +587,7 @@ fil_name_parse(
 	const byte*	end,
 	ulint		space_id,
 	ulint		first_page_no,
-	mlog_id_t	type,
-	bool		apply)
+	mlog_id_t	type)
 {
 
 	ulint flags = mach_read_from_4(ptr);
@@ -674,7 +667,7 @@ fil_name_parse(
 		fil_name_process(
 			name, len, space_id, true);
 
-		if (apply && recv_replay_file_ops
+		if (recv_replay_file_ops
 			&& fil_space_get(space_id)) {
 			dberr_t	err = fil_delete_tablespace(
 				space_id, BUF_REMOVE_FLUSH_NO_WRITE);
@@ -1668,7 +1661,6 @@ specified.
 @param[in]	end_ptr		end of buffer
 @param[in]	space_id	tablespace identifier
 @param[in]	page_no		page number
-@param[in]	apply		whether to apply the record
 @param[in,out]	block		buffer block, or NULL if
 a page log record should not be applied
 or if it is a MLOG_FILE_ operation
@@ -1683,14 +1675,10 @@ recv_parse_or_apply_log_rec_body(
 	byte*		end_ptr,
 	ulint		space_id,
 	ulint		page_no,
-	bool		apply,
 	buf_block_t*	block,
 	mtr_t*		mtr)
 {
 	ut_ad(!block == !mtr);
-#ifndef UNIV_HOTBACKUP
-	ut_ad(!apply || recv_sys->mlog_checkpoint_lsn != 0);
-#endif /* !UNIV_HOTBACKUP */
 
 	switch (type) {
 	case MLOG_FILE_NAME:
@@ -1700,8 +1688,7 @@ recv_parse_or_apply_log_rec_body(
 		ut_ad(block == NULL);
 		/* Collect the file names when parsing the log,
 		before applying any log records. */
-		return(fil_name_parse(ptr, end_ptr, space_id, page_no, type,
-				      apply));
+		return(fil_name_parse(ptr, end_ptr, space_id, page_no, type));
 	case MLOG_INDEX_LOAD:
 #ifdef UNIV_HOTBACKUP
 		/* While scaning redo logs during  backup phase a
@@ -1774,8 +1761,7 @@ recv_parse_or_apply_log_rec_body(
 		encryption key information before the page 0 is recovered.
 	        Otherwise, redo will not find the key to decrypt
 		the data pages. */
-		if (page_no == 0 && !is_system_tablespace(space_id)
-		    && !apply) {
+		if (page_no == 0 && !is_system_tablespace(space_id)) {
 			return(fil_write_encryption_parse(ptr,
 							  end_ptr,
 							  space_id));
@@ -1795,18 +1781,9 @@ recv_parse_or_apply_log_rec_body(
 
 	if (block) {
 		/* Applying a page log record. */
-		ut_ad(apply);
 		page = block->frame;
 		page_zip = buf_block_get_page_zip(block);
 		ut_d(page_type = fil_page_get_type(page));
-	} else if (apply
-		   && !is_predefined_tablespace(space_id)
-		   && recv_spaces.find(space_id) == recv_spaces.end()) {
-		ib::fatal() << "Missing MLOG_FILE_NAME or MLOG_FILE_DELETE"
-			" for redo log record " << type << " (page "
-			<< space_id << ":" << page_no << ") at "
-			<< recv_sys->recovered_lsn << ".";
-		return(NULL);
 	} else {
 		/* Parsing a page log record. */
 		page = NULL;
@@ -2520,7 +2497,7 @@ recv_recover_page_func(
 			recv_parse_or_apply_log_rec_body(
 				recv->type, buf, buf + recv->len,
 				recv_addr->space, recv_addr->page_no,
-				true, block, &mtr);
+				block, &mtr);
 
 			end_lsn = recv->start_lsn + recv->len;
 			mach_write_to_8(FIL_PAGE_LSN + page, end_lsn);
@@ -3052,7 +3029,7 @@ recv_parse_log_rec(
 	}
 
 	new_ptr = recv_parse_or_apply_log_rec_body(
-		*type, new_ptr, end_ptr, *space, *page_no, apply, NULL, NULL);
+		*type, new_ptr, end_ptr, *space, *page_no, NULL, NULL);
 
 	if (UNIV_UNLIKELY(new_ptr == NULL)) {
 
@@ -3159,15 +3136,13 @@ enum store_t {
 hash table to wait merging to file pages.
 @param[in]	checkpoint_lsn	the LSN of the latest checkpoint
 @param[in]	store		whether to store page operations
-@param[in]	apply		whether to apply the records
 @return whether MLOG_CHECKPOINT record was seen the first time,
 or corruption was noticed */
 static MY_ATTRIBUTE((warn_unused_result))
 bool
 recv_parse_log_recs(
 	lsn_t		checkpoint_lsn,
-	store_t		store,
-	bool		apply)
+	store_t		store)
 {
 	byte*		ptr;
 	byte*		end_ptr;
@@ -3213,7 +3188,7 @@ recv_parse_log_recs(
 		page no, and a pointer to the body of the log record */
 
 		len = recv_parse_log_rec(&type, ptr, end_ptr, &space,
-					 &page_no, apply, &body);
+					 &page_no, true, &body);
 
 		if (len == 0) {
 			return(false);
@@ -3279,9 +3254,6 @@ recv_parse_log_recs(
 				}
 				recv_sys->mlog_checkpoint_lsn
 					= recv_sys->recovered_lsn;
-#ifndef UNIV_HOTBACKUP
-				return(true);
-#endif /* !UNIV_HOTBACKUP */
 			}
 			break;
 		case MLOG_FILE_NAME:
@@ -3425,7 +3397,7 @@ recv_parse_log_recs(
 			completely recovered (until MLOG_MULTI_REC_END). */
 			len = recv_parse_log_rec(
 				&type, ptr, end_ptr, &space, &page_no,
-				apply, &body);
+				true, &body);
 
 			if (recv_sys->found_corrupt_log
 			    && !recv_report_corrupt_log(
@@ -3614,7 +3586,6 @@ recv_scan_log_recs(
 	bool		finished	= false;
 	ulint		data_len;
 	bool		more_data	= false;
-	bool		apply		= recv_sys->mlog_checkpoint_lsn != 0;
 	ulint		recv_parsing_buf_size = RECV_PARSING_BUF_SIZE;
 
 	ut_ad(start_lsn % OS_FILE_LOG_BLOCK_SIZE == 0);
@@ -3781,7 +3752,7 @@ recv_scan_log_recs(
 		/* Try to parse more log records */
 
 		if (recv_parse_log_recs(checkpoint_lsn,
-					*store_to_hash, apply)) {
+					*store_to_hash)) {
 			ut_ad(recv_sys->found_corrupt_log
 			      || recv_sys->found_corrupt_fs
 			      || recv_sys->mlog_checkpoint_lsn
@@ -3843,8 +3814,7 @@ recv_group_scan_log_recs(
 	lsn_t	checkpoint_lsn	= *contiguous_lsn;
 	lsn_t	start_lsn;
 	lsn_t	end_lsn;
-	store_t	store_to_hash	= recv_sys->mlog_checkpoint_lsn == 0
-		? STORE_NO : (last_phase ? STORE_IF_EXISTS : STORE_YES);
+	store_t	store_to_hash	= last_phase ? STORE_IF_EXISTS : STORE_YES;
 	ulint	available_mem	= UNIV_PAGE_SIZE
 		* (buf_pool_get_n_pages()
 		   - (recv_n_pool_free_frames * srv_buf_pool_instances));
@@ -3926,6 +3896,25 @@ recv_init_missing_space(dberr_t err, const recv_spaces_t::const_iterator& i)
 	return(err);
 }
 
+/** Report a missing mlog_file_name or mlog_file_delete record for
+the tablespace.
+@param[in]	recv_addr	Hashed page file address. */
+static
+void
+recv_init_missing_mlog(
+	recv_addr_t*	recv_addr)
+{
+	ulint	space_id = recv_addr->space;
+	ulint	page_no = recv_addr->page_no;
+	ulint	type = UT_LIST_GET_FIRST(recv_addr->rec_list)->type;
+	ulint	start_lsn = UT_LIST_GET_FIRST(recv_addr->rec_list)->start_lsn;
+
+	ib::fatal() << "Missing MLOG_FILE_NAME or MLOG_FILE_DELETE "
+		"for redo log record " << type << " (page "
+		<< space_id << ":" << page_no << ") at "
+		<< start_lsn;
+}
+
 /** Check if all tablespaces were found for crash recovery.
 @return error code or DB_SUCCESS */
 static MY_ATTRIBUTE((warn_unused_result))
@@ -3981,7 +3970,12 @@ recv_init_crash_recovery_spaces(void)
 
 				recv_spaces_t::iterator i
 					= recv_spaces.find(space);
-				ut_ad(i != recv_spaces.end());
+				
+				if (i != recv_spaces.end()) {
+					recv_init_missing_mlog(recv_addr);
+					recv_addr->state = RECV_DISCARDED;
+					continue;
+				}
 
 				if (i->second.deleted) {
 					ut_ad(missing_spaces.find(space)
@@ -4153,13 +4147,13 @@ recv_recovery_from_checkpoint_start(
 		return(DB_ERROR);
 	}
 
-	/* Look for MLOG_CHECKPOINT. */
-	recv_group_scan_log_recs(group, &contiguous_lsn, false);
-	/* The first scan should not have stored or applied any records. */
-	ut_ad(recv_sys->n_addrs == 0);
-	ut_ad(!recv_sys->found_corrupt_fs);
+	/** Scan the redo log from checkpoint lsn and redo log to
+	the hash table. */
+	rescan = recv_group_scan_log_recs(group, &contiguous_lsn, false);
 
-	if (recv_sys->found_corrupt_log && !srv_force_recovery) {
+
+	if ((recv_sys->found_corrupt_log && !srv_force_recovery)
+	    || recv_sys->found_corrupt_fs) {
 		log_mutex_exit();
 		return(DB_ERROR);
 	}
@@ -4179,16 +4173,6 @@ recv_recovery_from_checkpoint_start(
 
 		group->scanned_lsn = checkpoint_lsn;
 		rescan = false;
-	} else {
-		contiguous_lsn = checkpoint_lsn;
-		rescan = recv_group_scan_log_recs(
-			group, &contiguous_lsn, false);
-
-		if ((recv_sys->found_corrupt_log && !srv_force_recovery)
-		    || recv_sys->found_corrupt_fs) {
-			log_mutex_exit();
-			return(DB_ERROR);
-		}
 	}
 
 	/* NOTE: we always do a 'recovery' at startup, but only if