Bug #24793413 LOG PARSING BUFFER OVERFLOW

Problem:
========
During checkpoint, we are writing all MLOG_FILE_NAME records in one mtr
and parse buffer can't be processed till MLOG_MULTI_REC_END. Eventually parse
buffer exceeds the RECV_PARSING_BUF_SIZE and eventually it overflows.

Fix:
===
1) Break the large mtr if it exceeds LOG_CHECKPOINT_FREE_PER_THREAD into multiple mtr during checkpoint.
2) Move the parsing buffer if we are encountering only MLOG_FILE_NAME
records. So that it will never exceed the RECV_PARSING_BUF_SIZE.
Reviewed-by: Debarun Bannerjee <debarun.bannerjee@oracle.com>
Reviewed-by: Rahul M Malik <rahul.m.malik@oracle.com>
RB: 14743

storage/innobase/fil/fil0fil.cc

@@ -40,6 +40,7 @@ Created 10/25/1995 Heikki Tuuri
 #include "fsp0space.h"
 #include "fsp0sysspace.h"
 #include "hash0hash.h"
+#include "log0log.h"
 #include "log0recv.h"
 #include "mach0data.h"
 #include "mem0mem.h"
@@ -6536,6 +6537,12 @@ fil_names_clear(
 	bool	do_write)
 {
 	mtr_t	mtr;
+	ulint	mtr_checkpoint_size = LOG_CHECKPOINT_FREE_PER_THREAD;
+
+	DBUG_EXECUTE_IF(
+		"increase_mtr_checkpoint_size",
+		mtr_checkpoint_size = 75 * 1024;
+		);
 
 	ut_ad(log_mutex_own());
 
@@ -6569,11 +6576,24 @@ fil_names_clear(
 		fil_names_write(space, &mtr);
 		do_write = true;
 
+		const mtr_buf_t* mtr_log = mtr_get_log(&mtr);
+
+		/** If the mtr buffer size exceeds the size of
+		LOG_CHECKPOINT_FREE_PER_THREAD then commit the multi record
+		mini-transaction, start the new mini-transaction to
+		avoid the parsing buffer overflow error during recovery. */
+
+		if (mtr_log->size() > mtr_checkpoint_size) {
+			ut_ad(mtr_log->size() < (RECV_PARSING_BUF_SIZE / 2));
+			mtr.commit_checkpoint(lsn, false);
+			mtr.start();
+		}
+
 		space = next;
 	}
 
 	if (do_write) {
-		mtr.commit_checkpoint(lsn);
+		mtr.commit_checkpoint(lsn, true);
 	} else {
 		ut_ad(!mtr.has_modifications());
 	}

storage/innobase/include/log0log.h

 /*****************************************************************************
 
-Copyright (c) 1995, 2016, Oracle and/or its affiliates. All rights reserved.
+Copyright (c) 1995, 2017, Oracle and/or its affiliates. All rights reserved.
 Copyright (c) 2009, Google Inc.
 Copyright (c) 2017, MariaDB Corporation. All Rights Reserved.
 
@@ -46,6 +46,12 @@ struct log_group_t;
 /** Magic value to use instead of log checksums when they are disabled */
 #define LOG_NO_CHECKSUM_MAGIC 0xDEADBEEFUL
 
+/* Margin for the free space in the smallest log group, before a new query
+step which modifies the database, is started */
+
+#define LOG_CHECKPOINT_FREE_PER_THREAD	(4 * UNIV_PAGE_SIZE)
+#define LOG_CHECKPOINT_EXTRA_FREE	(8 * UNIV_PAGE_SIZE)
+
 typedef ulint (*log_checksum_func_t)(const byte* log_block);
 
 /** Pointer to the log checksum calculation function. Protected with

storage/innobase/include/mtr0mtr.h

 /*****************************************************************************
 
-Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
 Copyright (c) 2013, 2017, MariaDB Corporation
 
@@ -269,8 +269,12 @@ struct mtr_t {
 	MLOG_FILE_NAME records and a MLOG_CHECKPOINT marker.
 	The caller must invoke log_mutex_enter() and log_mutex_exit().
 	This is to be used at log_checkpoint().
-	@param[in]	checkpoint_lsn	the LSN of the log checkpoint  */
-	void commit_checkpoint(lsn_t checkpoint_lsn);
+	@param[in]	checkpoint_lsn		the LSN of the log checkpoint
+	@param[in]	write_mlog_checkpoint	Write MLOG_CHECKPOINT marker
+						if it is enabled. */
+	void commit_checkpoint(
+		lsn_t	checkpoint_lsn,
+		bool	write_mlog_checkpoint);
 
 	/** Return current size of the buffer.
 	@return	savepoint */

storage/innobase/log/log0log.cc

 /*****************************************************************************
 
-Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2009, Google Inc.
 Copyright (c) 2014, 2017, MariaDB Corporation.
 
@@ -106,12 +106,6 @@ static time_t	log_last_margine_warning_time;
 #define LOG_BUF_FLUSH_RATIO	2
 #define LOG_BUF_FLUSH_MARGIN	(LOG_BUF_WRITE_MARGIN + 4 * UNIV_PAGE_SIZE)
 
-/* Margin for the free space in the smallest log group, before a new query
-step which modifies the database, is started */
-
-#define LOG_CHECKPOINT_FREE_PER_THREAD	(4 * UNIV_PAGE_SIZE)
-#define LOG_CHECKPOINT_EXTRA_FREE	(8 * UNIV_PAGE_SIZE)
-
 /* This parameter controls asynchronous making of a new checkpoint; the value
 should be bigger than LOG_POOL_PREFLUSH_RATIO_SYNC */
 

storage/innobase/log/log0recv.cc

@@ -2482,6 +2482,8 @@ recv_parse_log_recs(
 
 		ulint	total_len	= 0;
 		ulint	n_recs		= 0;
+		bool	only_mlog_file	= true;
+		ulint	mlog_rec_len	= 0;
 
 		for (;;) {
 			len = recv_parse_log_rec(
@@ -2510,6 +2512,22 @@ recv_parse_log_recs(
 				= recv_sys->recovered_offset + total_len;
 			recv_previous_parsed_rec_is_multi = 1;
 
+			/* MLOG_FILE_NAME redo log records doesn't make changes
+			to persistent data. If only MLOG_FILE_NAME redo
+			log record exists then reset the parsing buffer pointer
+			by changing recovered_lsn and recovered_offset. */
+			if (type != MLOG_FILE_NAME && only_mlog_file == true) {
+				only_mlog_file = false;
+			}
+
+			if (only_mlog_file) {
+				new_recovered_lsn = recv_calc_lsn_on_data_add(
+					recv_sys->recovered_lsn, len);
+				mlog_rec_len += len;
+				recv_sys->recovered_offset += len;
+				recv_sys->recovered_lsn = new_recovered_lsn;
+			}
+
 			total_len += len;
 			n_recs++;
 
@@ -2523,6 +2541,7 @@ recv_parse_log_recs(
 					    " n=" ULINTPF,
 					    recv_sys->recovered_lsn,
 					    total_len, n_recs));
+				total_len -= mlog_rec_len;
 				break;
 			}
 
@@ -2742,6 +2761,7 @@ recv_scan_log_recs(
 	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);
 	ut_ad(end_lsn % OS_FILE_LOG_BLOCK_SIZE == 0);
@@ -2814,8 +2834,14 @@ recv_scan_log_recs(
 			parsing buffer if parse_start_lsn is already
 			non-zero */
 
+			DBUG_EXECUTE_IF(
+				"reduce_recv_parsing_buf",
+				recv_parsing_buf_size
+					= (70 * 1024);
+				);
+
 			if (recv_sys->len + 4 * OS_FILE_LOG_BLOCK_SIZE
-			    >= RECV_PARSING_BUF_SIZE) {
+			    >= recv_parsing_buf_size) {
 				ib::error() << "Log parsing buffer overflow."
 					" Recovery may have failed!";
 
@@ -2865,7 +2891,7 @@ recv_scan_log_recs(
 			*store_to_hash = STORE_NO;
 		}
 
-		if (recv_sys->recovered_offset > RECV_PARSING_BUF_SIZE / 4) {
+		if (recv_sys->recovered_offset > recv_parsing_buf_size / 4) {
 			/* Move parsing buffer data to the buffer start */
 
 			recv_sys_justify_left_parsing_buf();

storage/innobase/mtr/mtr0mtr.cc

 /*****************************************************************************
 
-Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2017, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
@@ -580,9 +580,13 @@ but generated some redo log on a higher level, such as
 MLOG_FILE_NAME records and a MLOG_CHECKPOINT marker.
 The caller must invoke log_mutex_enter() and log_mutex_exit().
 This is to be used at log_checkpoint().
-@param[in]	checkpoint_lsn	the LSN of the log checkpoint  */
+@param[in]	checkpoint_lsn		the LSN of the log checkpoint
+@param[in]	write_mlog_checkpoint	Write MLOG_CHECKPOINT marker
+					if it is enabled. */
 void
-mtr_t::commit_checkpoint(lsn_t checkpoint_lsn)
+mtr_t::commit_checkpoint(
+	lsn_t	checkpoint_lsn,
+	bool	write_mlog_checkpoint)
 {
 	ut_ad(log_mutex_own());
 	ut_ad(is_active());
@@ -593,6 +597,7 @@ mtr_t::commit_checkpoint(lsn_t checkpoint_lsn)
 	ut_ad(m_impl.m_memo.size() == 0);
 	ut_ad(!srv_read_only_mode);
 	ut_d(m_impl.m_state = MTR_STATE_COMMITTING);
+	ut_ad(write_mlog_checkpoint || m_impl.m_n_log_recs > 1);
 
 	/* This is a dirty read, for debugging. */
 	ut_ad(!recv_no_log_write);
@@ -608,20 +613,24 @@ mtr_t::commit_checkpoint(lsn_t checkpoint_lsn)
 			&m_impl.m_log, MLOG_MULTI_REC_END, MLOG_1BYTE);
 	}
 
+	if (write_mlog_checkpoint) {
 		byte*	ptr = m_impl.m_log.push<byte*>(SIZE_OF_MLOG_CHECKPOINT);
 #if SIZE_OF_MLOG_CHECKPOINT != 9
 # error SIZE_OF_MLOG_CHECKPOINT != 9
 #endif
 		*ptr = MLOG_CHECKPOINT;
 		mach_write_to_8(ptr + 1, checkpoint_lsn);
+	}
 
 	Command	cmd(this);
 	cmd.finish_write(m_impl.m_log.size());
 	cmd.release_resources();
 
+	if (write_mlog_checkpoint) {
 		DBUG_PRINT("ib_log",
 			   ("MLOG_CHECKPOINT(" LSN_PF ") written at " LSN_PF,
 			    checkpoint_lsn, log_sys->lsn));
+	}
 }
 
 #ifdef UNIV_DEBUG