Merge 10.3 into 10.4

19 files changed, 273 insertions, 184 deletions

diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index 7ff0820dd13..1f104e59db8 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -5325,7 +5325,6 @@ btr_validate_index(
 
 		if (!btr_validate_level(index, trx, n - i, lockout)) {
 			err = DB_CORRUPTION;
-			break;
 		}
 	}
 
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 0dae73c56d7..ac88ea26d39 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -3497,7 +3497,7 @@ fail:
 
 		/* prefetch siblings of the leaf for the pessimistic
 		operation, if the page is leaf. */
-		if (page_is_leaf(page)) {
+		if (page_is_leaf(page) && !index->is_ibuf()) {
 			btr_cur_prefetch_siblings(block);
 		}
 fail_err:
@@ -4585,6 +4585,7 @@ btr_cur_optimistic_update(
 
 	if (rec_offs_any_extern(*offsets)) {
 any_extern:
+		ut_ad(!index->is_ibuf());
 		/* Externally stored fields are treated in pessimistic
 		update */
 
@@ -4781,7 +4782,7 @@ func_exit:
 		}
 	}
 
-	if (err != DB_SUCCESS) {
+	if (err != DB_SUCCESS && !index->is_ibuf()) {
 		/* prefetch siblings of the leaf for the pessimistic
 		operation. */
 		btr_cur_prefetch_siblings(block);
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 905c04fa811..51b67237f22 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -5615,9 +5615,21 @@ buf_page_create(
 			 && block->index);
 
 		if (drop_hash_entry) {
-			mutex_enter(&block->mutex);
-			buf_page_set_sticky(&block->page);
-			mutex_exit(&block->mutex);
+			/* Avoid a hang if I/O is going on. Release
+			the buffer pool mutex and page hash lock
+			and wait for I/O to complete */
+			while (buf_block_get_io_fix(block) != BUF_IO_NONE) {
+				block->fix();
+				buf_pool_mutex_exit(buf_pool);
+				rw_lock_x_unlock(hash_lock);
+
+				os_thread_yield();
+
+				buf_pool_mutex_enter(buf_pool);
+				rw_lock_x_lock(hash_lock);
+				block->unfix();
+			}
+			rw_lock_x_lock(&block->lock);
 		}
 #endif
 		/* Page can be found in buf_pool */
@@ -5628,11 +5640,7 @@ buf_page_create(
 #ifdef BTR_CUR_HASH_ADAPT
 		if (drop_hash_entry) {
 			btr_search_drop_page_hash_index(block);
-			buf_pool_mutex_enter(buf_pool);
-			mutex_enter(&block->mutex);
-			buf_page_unset_sticky(&block->page);
-			mutex_exit(&block->mutex);
-			buf_pool_mutex_exit(buf_pool);
+			rw_lock_x_unlock(&block->lock);
 		}
 #endif /* BTR_CUR_HASH_ADAPT */
 
diff --git a/storage/innobase/buf/buf0flu.cc b/storage/innobase/buf/buf0flu.cc
index 9b49b9570e2..45c9b661d6f 100644
--- a/storage/innobase/buf/buf0flu.cc
+++ b/storage/innobase/buf/buf0flu.cc
@@ -3079,7 +3079,7 @@ DECLARE_THREAD(buf_flush_page_cleaner_coordinator)(void*)
 		/* The page_cleaner skips sleep if the server is
 		idle and there are no pending IOs in the buffer pool
 		and there is work to do. */
-		if (srv_check_activity(&last_activity)
+		if (srv_check_activity(last_activity)
 		    || buf_get_n_pending_read_ios()
 		    || n_flushed == 0) {
 
@@ -3171,7 +3171,7 @@ DECLARE_THREAD(buf_flush_page_cleaner_coordinator)(void*)
 
 			n_flushed = n_flushed_lru + n_flushed_list;
 
-		} else if (srv_check_activity(&last_activity)) {
+		} else if (srv_check_activity(last_activity)) {
 			ulint	n_to_flush;
 			lsn_t	lsn_limit = 0;
 
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 38429878b65..c088a5c7dc1 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -431,6 +431,14 @@ TYPELIB innodb_flush_method_typelib = {
 	NULL
 };
 
+/* The following counter is used to convey information to InnoDB
+about server activity: in case of normal DML ops it is not
+sensible to call srv_active_wake_master_thread after each
+operation, we only do it every INNOBASE_WAKE_INTERVAL'th step. */
+
+#define INNOBASE_WAKE_INTERVAL	32
+static ulong	innobase_active_counter	= 0;
+
 /** Allowed values of innodb_change_buffering */
 static const char* innodb_change_buffering_names[] = {
 	"none",		/* IBUF_USE_NONE */
@@ -1830,6 +1838,23 @@ static int innobase_wsrep_set_checkpoint(handlerton* hton, const XID* xid);
 static int innobase_wsrep_get_checkpoint(handlerton* hton, XID* xid);
 #endif /* WITH_WSREP */
 /********************************************************************//**
+Increments innobase_active_counter and every INNOBASE_WAKE_INTERVALth
+time calls srv_active_wake_master_thread. This function should be used
+when a single database operation may introduce a small need for
+server utility activity, like checkpointing. */
+inline
+void
+innobase_active_small(void)
+/*=======================*/
+{
+	innobase_active_counter++;
+
+	if ((innobase_active_counter % INNOBASE_WAKE_INTERVAL) == 0) {
+		srv_active_wake_master_thread();
+	}
+}
+
+/********************************************************************//**
 Converts an InnoDB error code to a MySQL error code and also tells to MySQL
 about a possible transaction rollback inside InnoDB caused by a lock wait
 timeout or a deadlock.
@@ -3478,8 +3503,7 @@ static int innodb_init_abort()
 	DBUG_RETURN(1);
 }
 
-/** Update log_checksum_algorithm_ptr with a pointer to the function
-corresponding to whether checksums are enabled.
+/** If applicable, emit a message that log checksums cannot be disabled.
 @param[in,out]	thd	client session, or NULL if at startup
 @param[in]	check	whether redo log block checksums are enabled
 @return whether redo log block checksums are enabled */
@@ -3487,34 +3511,21 @@ static inline
 bool
 innodb_log_checksums_func_update(THD* thd, bool check)
 {
-	static const char msg[] = "innodb_encrypt_log implies"
-		" innodb_log_checksums";
+	static const char msg[] = "innodb_log_checksums is deprecated"
+		" and has no effect outside recovery";
 
 	ut_ad(!thd == !srv_was_started);
 
 	if (!check) {
-		check = srv_encrypt_log;
-		if (!check) {
-		} else if (thd) {
+		if (thd) {
 			push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
 					    HA_ERR_UNSUPPORTED, msg);
+			check = true;
 		} else {
 			sql_print_warning(msg);
 		}
 	}
 
-	if (thd) {
-		log_mutex_enter();
-		log_checksum_algorithm_ptr = check
-			? log_block_calc_checksum_crc32
-			: log_block_calc_checksum_none;
-		log_mutex_exit();
-	} else {
-		log_checksum_algorithm_ptr = check
-			? log_block_calc_checksum_crc32
-			: log_block_calc_checksum_none;
-	}
-
 	return(check);
 }
 
@@ -6453,6 +6464,11 @@ ha_innobase::close()
 
 	MONITOR_INC(MONITOR_TABLE_CLOSE);
 
+	/* Tell InnoDB server that there might be work for
+	utility threads: */
+
+	srv_active_wake_master_thread();
+
 	DBUG_RETURN(0);
 }
 
@@ -8150,6 +8166,8 @@ report_error:
 	}
 
 func_exit:
+	innobase_active_small();
+
 	DBUG_RETURN(error_result);
 }
 
@@ -8829,6 +8847,11 @@ func_exit:
 			error, m_prebuilt->table->flags, m_user_thd);
 	}
 
+	/* Tell InnoDB server that there might be work for
+	utility threads: */
+
+	innobase_active_small();
+
 #ifdef WITH_WSREP
 	if (error == DB_SUCCESS && trx->is_wsrep()
 	    && wsrep_thd_is_local(m_user_thd)
@@ -8890,6 +8913,11 @@ ha_innobase::delete_row(
 
 	innobase_srv_conc_exit_innodb(m_prebuilt);
 
+	/* Tell the InnoDB server that there might be work for
+	utility threads: */
+
+	innobase_active_small();
+
 #ifdef WITH_WSREP
 	if (error == DB_SUCCESS && trx->is_wsrep()
 	    && wsrep_thd_is_local(m_user_thd)
@@ -12705,6 +12733,7 @@ create_table_info_t::create_table_update_dict()
 	if (m_flags2 & DICT_TF2_FTS) {
 		if (!innobase_fts_load_stopword(innobase_table, NULL, m_thd)) {
 			dict_table_close(innobase_table, FALSE, FALSE);
+			srv_active_wake_master_thread();
 			trx_free(m_trx);
 			DBUG_RETURN(-1);
 		}
@@ -12854,6 +12883,11 @@ ha_innobase::create(
 
 	error = info.create_table_update_dict();
 
+	/* Tell the InnoDB server that there might be work for
+	utility threads: */
+
+	srv_active_wake_master_thread();
+
 	DBUG_RETURN(error);
 }
 
@@ -18782,7 +18816,7 @@ static MYSQL_SYSVAR_ENUM(checksum_algorithm, srv_checksum_algorithm,
 
 static MYSQL_SYSVAR_BOOL(log_checksums, innodb_log_checksums,
   PLUGIN_VAR_RQCMDARG,
-  "Whether to compute and require checksums for InnoDB redo log blocks",
+  "DEPRECATED. Whether to require checksums for InnoDB redo log blocks.",
   NULL, innodb_log_checksums_update, TRUE);
 
 static MYSQL_SYSVAR_BOOL(checksums, innobase_use_checksums,
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index df42f20c55d..2b74d2a4b4c 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -10863,11 +10863,11 @@ ha_innobase::commit_inplace_alter_table(
 
 		/* Exclusively lock the table, to ensure that no other
 		transaction is holding locks on the table while we
-		change the table definition. The MySQL meta-data lock
+		change the table definition. The meta-data lock (MDL)
 		should normally guarantee that no conflicting locks
 		exist. However, FOREIGN KEY constraints checks and any
 		transactions collected during crash recovery could be
-		holding InnoDB locks only, not MySQL locks. */
+		holding InnoDB locks only, not MDL. */
 
 		dberr_t error = row_merge_lock_table(
 			m_prebuilt->trx, ctx->old_table, LOCK_X);
@@ -11244,6 +11244,11 @@ foreign_fail:
 
 	log_append_on_checkpoint(NULL);
 
+	/* Tell the InnoDB server that there might be work for
+	utility threads: */
+
+	srv_active_wake_master_thread();
+
 	if (fail) {
 		for (inplace_alter_handler_ctx** pctx = ctx_array;
 		     *pctx; pctx++) {
diff --git a/storage/innobase/include/log0log.h b/storage/innobase/include/log0log.h
index 399319537c8..368d912f07f 100644
--- a/storage/innobase/include/log0log.h
+++ b/storage/innobase/include/log0log.h
@@ -2,7 +2,7 @@
 
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All rights reserved.
 Copyright (c) 2009, Google Inc.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2020, MariaDB Corporation.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -56,12 +56,6 @@ step which modifies the database, is started */
 #define LOG_CHECKPOINT_FREE_PER_THREAD	(4U << srv_page_size_shift)
 #define LOG_CHECKPOINT_EXTRA_FREE	(8U << srv_page_size_shift)
 
-typedef ulint (*log_checksum_func_t)(const byte* log_block);
-
-/** Pointer to the log checksum calculation function. Protected with
-log_sys.mutex. */
-extern log_checksum_func_t log_checksum_algorithm_ptr;
-
 /** Append a string to the log.
 @param[in]	str		string
 @param[in]	len		string length
@@ -263,14 +257,6 @@ log_block_set_data_len(
 /*===================*/
 	byte*	log_block,	/*!< in/out: log block */
 	ulint	len);		/*!< in: data length */
-/************************************************************//**
-Calculates the checksum for a log block.
-@return checksum */
-UNIV_INLINE
-ulint
-log_block_calc_checksum(
-/*====================*/
-	const byte*	block);	/*!< in: log block */
 
 /** Calculates the checksum for a log block using the CRC32 algorithm.
 @param[in]	block	log block
@@ -280,12 +266,6 @@ ulint
 log_block_calc_checksum_crc32(
 	const byte*	block);
 
-/** Calculates the checksum for a log block using the "no-op" algorithm.
-@return		the calculated checksum value */
-UNIV_INLINE
-ulint
-log_block_calc_checksum_none(const byte*);
-
 /************************************************************//**
 Gets a log block checksum field value.
 @return checksum */
@@ -362,7 +342,7 @@ void
 log_refresh_stats(void);
 /*===================*/
 
-/** Whether to generate and require checksums on the redo log pages */
+/** Whether to require checksums on the redo log pages */
 extern my_bool	innodb_log_checksums;
 
 /* Values used as flags */
diff --git a/storage/innobase/include/log0log.ic b/storage/innobase/include/log0log.ic
index 7dfa7c0db68..8dfd86d3078 100644
--- a/storage/innobase/include/log0log.ic
+++ b/storage/innobase/include/log0log.ic
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2015, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2020, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -190,18 +190,6 @@ log_block_convert_lsn_to_no(
 			0xFUL, 0x3FFFFFFFUL)) + 1);
 }
 
-/************************************************************//**
-Calculates the checksum for a log block.
-@return checksum */
-UNIV_INLINE
-ulint
-log_block_calc_checksum(
-/*====================*/
-	const byte*	block)	/*!< in: log block */
-{
-	return(log_checksum_algorithm_ptr(block));
-}
-
 /** Calculate the checksum for a log block using the pre-5.7.9 algorithm.
 @param[in]	block	log block
 @return		checksum */
@@ -242,15 +230,6 @@ log_block_calc_checksum_crc32(
 	return ut_crc32(block, OS_FILE_LOG_BLOCK_SIZE - LOG_BLOCK_CHECKSUM);
 }
 
-/** Calculates the checksum for a log block using the "no-op" algorithm.
-@return        checksum */
-UNIV_INLINE
-ulint
-log_block_calc_checksum_none(const byte*)
-{
-	return(LOG_NO_CHECKSUM_MAGIC);
-}
-
 /************************************************************//**
 Gets a log block checksum field value.
 @return checksum */
diff --git a/storage/innobase/include/srv0srv.h b/storage/innobase/include/srv0srv.h
index 961d6aa1073..e1d37613dc9 100644
--- a/storage/innobase/include/srv0srv.h
+++ b/storage/innobase/include/srv0srv.h
@@ -784,6 +784,19 @@ srv_reset_io_thread_op_info();
 /** Wake up the purge threads if there is work to do. */
 void
 srv_wake_purge_thread_if_not_active();
+/** Wake up the InnoDB master thread if it was suspended (not sleeping). */
+void
+srv_active_wake_master_thread_low();
+
+#define srv_active_wake_master_thread()					\
+	do {								\
+		if (!srv_read_only_mode) {				\
+			srv_active_wake_master_thread_low();		\
+		}							\
+	} while (0)
+/** Wake up the master thread if it is suspended or being suspended. */
+void
+srv_wake_master_thread();
 
 /******************************************************************//**
 Outputs to a file the output of the InnoDB Monitor.
@@ -812,13 +825,13 @@ reading this value as it is only used in heuristics.
 ulint
 srv_get_activity_count(void);
 /*========================*/
-
-/** Check if there has been any activity.
-@param[in,out]  activity_count  recent activity count to be returned
-if there is a change
+/*******************************************************************//**
+Check if there has been any activity.
 @return FALSE if no change in activity counter. */
-bool srv_check_activity(ulint *activity_count);
-
+ibool
+srv_check_activity(
+/*===============*/
+	ulint		old_activity_count);	/*!< old activity count */
 /******************************************************************//**
 Increment the server activity counter. */
 void
diff --git a/storage/innobase/log/log0log.cc b/storage/innobase/log/log0log.cc
index 8e0fe582c84..130c830858b 100644
--- a/storage/innobase/log/log0log.cc
+++ b/storage/innobase/log/log0log.cc
@@ -2,7 +2,7 @@
 
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2009, Google Inc.
-Copyright (c) 2014, 2019, MariaDB Corporation.
+Copyright (c) 2014, 2020, MariaDB Corporation.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -83,12 +83,9 @@ reduce the size of the log.
 /** Redo log system */
 log_t	log_sys;
 
-/** Whether to generate and require checksums on the redo log pages */
+/** Whether to require checksums on the redo log pages */
 my_bool	innodb_log_checksums;
 
-/** Pointer to the log checksum calculation function */
-log_checksum_func_t log_checksum_algorithm_ptr;
-
 /* Next log block number to do dummy record filling if no log records written
 for a while */
 static ulint		next_lbn_to_pad = 0;
@@ -674,7 +671,7 @@ log_block_store_checksum(
 /*=====================*/
 	byte*	block)	/*!< in/out: pointer to a log block */
 {
-	log_block_set_checksum(block, log_block_calc_checksum(block));
+	log_block_set_checksum(block, log_block_calc_checksum_crc32(block));
 }
 
 /******************************************************//**
diff --git a/storage/innobase/page/page0page.cc b/storage/innobase/page/page0page.cc
index 86c86bd084d..ae2cf1870e1 100644
--- a/storage/innobase/page/page0page.cc
+++ b/storage/innobase/page/page0page.cc
@@ -2455,6 +2455,7 @@ wrong_page_type:
 	/* Validate the record list in a loop checking also that
 	it is consistent with the directory. */
 	ulint count = 0, data_size = 0, own_count = 1, slot_no = 0;
+	ulint info_bits;
 	slot_no = 0;
 	slot = page_dir_get_nth_slot(page, slot_no);
 
@@ -2478,9 +2479,16 @@ wrong_page_type:
 			goto next_rec;
 		}
 
+		info_bits = rec_get_info_bits(rec, page_is_comp(page));
+		if (info_bits
+		    & ~(REC_INFO_MIN_REC_FLAG | REC_INFO_DELETED_FLAG)) {
+			ib::error() << "info_bits has an incorrect value "
+				    << info_bits;
+			ret = false;
+		}
+
 		if (rec == first_rec) {
-			if ((rec_get_info_bits(rec, page_is_comp(page))
-			     & REC_INFO_MIN_REC_FLAG)) {
+			if (info_bits & REC_INFO_MIN_REC_FLAG) {
 				if (page_has_prev(page)) {
 					ib::error() << "REC_INFO_MIN_REC_FLAG "
 						"is set on non-left page";
@@ -2491,8 +2499,7 @@ wrong_page_type:
 					ib::error() << "REC_INFO_MIN_REC_FLAG "
 						"is set in a leaf-page record";
 					ret = false;
-				} else if (!rec_get_deleted_flag(
-						   rec, page_is_comp(page))
+				} else if (!(info_bits & REC_INFO_DELETED_FLAG)
 					   != !index->table->instant) {
 					ib::error() << (index->table->instant
 							? "Metadata record "
@@ -2506,13 +2513,51 @@ wrong_page_type:
 				ib::error() << "Metadata record is missing";
 				ret = false;
 			}
-		} else if (rec_get_info_bits(rec, page_is_comp(page))
-			   & REC_INFO_MIN_REC_FLAG) {
+		} else if (info_bits & REC_INFO_MIN_REC_FLAG) {
 			ib::error() << "REC_INFO_MIN_REC_FLAG record is not "
 				       "first in page";
 			ret = false;
 		}
 
+		if (page_is_comp(page)) {
+			const rec_comp_status_t status = rec_get_status(rec);
+			if (status != REC_STATUS_ORDINARY
+			    && status != REC_STATUS_NODE_PTR
+			    && status != REC_STATUS_INFIMUM
+			    && status != REC_STATUS_SUPREMUM
+			    && status != REC_STATUS_INSTANT) {
+				ib::error() << "impossible record status "
+					    << status;
+				ret = false;
+			} else if (page_rec_is_infimum(rec)) {
+				if (status != REC_STATUS_INFIMUM) {
+					ib::error()
+						<< "infimum record has status "
+						<< status;
+					ret = false;
+				}
+			} else if (page_rec_is_supremum(rec)) {
+				if (status != REC_STATUS_SUPREMUM) {
+					ib::error() << "supremum record has "
+						       "status "
+						    << status;
+					ret = false;
+				}
+			} else if (!page_is_leaf(page)) {
+				if (status != REC_STATUS_NODE_PTR) {
+					ib::error() << "node ptr record has "
+						       "status "
+						    << status;
+					ret = false;
+				}
+			} else if (!index->is_instant()
+				   && status == REC_STATUS_INSTANT) {
+				ib::error() << "instantly added record in a "
+					       "non-instant index";
+				ret = false;
+			}
+		}
+
 		/* Check that the records are in the ascending order */
 		if (count >= PAGE_HEAP_NO_USER_LOW
 		    && !page_rec_is_supremum(rec)) {
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index ac5986243d6..86cdf1995db 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -3811,7 +3811,7 @@ funct_exit_all_freed:
 
 	trx->op_info = "";
 
-	srv_inc_activity_count();
+	srv_wake_master_thread();
 
 	DBUG_RETURN(err);
 }
@@ -4522,12 +4522,20 @@ end:
 		if (err != DB_SUCCESS) {
 
 			if (old_is_tmp) {
-				ib::error() << "In ALTER TABLE "
+				/* In case of copy alter, ignore the
+				loading of foreign key constraint
+				when foreign_key_check is disabled */
+				ib::error_or_warn(trx->check_foreigns)
+					<< "In ALTER TABLE "
 					<< ut_get_name(trx, new_name)
 					<< " has or is referenced in foreign"
 					" key constraints which are not"
 					" compatible with the new table"
 					" definition.";
+				if (!trx->check_foreigns) {
+					err = DB_SUCCESS;
+					goto funct_exit;
+				}
 			} else {
 				ib::error() << "In RENAME TABLE table "
 					<< ut_get_name(trx, new_name)
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 2f66f3636ff..2f4cb90e265 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -84,11 +84,11 @@ row_undo_ins_remove_clust_rec(
 		online = false;
 	} else {
 		index->set_modified(mtr);
+		ut_ad(lock_table_has_locks(index->table));
 		online = dict_index_is_online_ddl(index);
 		if (online) {
 			ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
-			ut_ad(node->trx->dict_operation_lock_mode
-			      != RW_X_LATCH);
+			ut_ad(!node->trx->dict_operation_lock_mode);
 			ut_ad(node->table->id != DICT_INDEXES_ID);
 			ut_ad(node->table->id != DICT_COLUMNS_ID);
 			mtr_s_lock_index(index, &mtr);
@@ -551,6 +551,9 @@ row_undo_ins(
 		return DB_SUCCESS;
 	}
 
+	ut_ad(node->table->is_temporary()
+	      || lock_table_has_locks(node->table));
+
 	/* Iterate over all the indexes and undo the insert.*/
 
 	node->index = dict_table_get_first_index(node->table);
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index e9551d33c73..e4654354068 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -269,10 +269,7 @@ row_undo_mod_clust(
 	bool		online;
 
 	ut_ad(thr_get_trx(thr) == node->trx);
-	ut_ad(node->trx->dict_operation_lock_mode);
 	ut_ad(node->trx->in_rollback);
-	ut_ad(rw_lock_own_flagged(&dict_sys.latch,
-				  RW_LOCK_FLAG_X | RW_LOCK_FLAG_S));
 
 	log_free_check();
 	pcur = &node->pcur;
@@ -284,11 +281,12 @@ row_undo_mod_clust(
 		mtr.set_log_mode(MTR_LOG_NO_REDO);
 	} else {
 		index->set_modified(mtr);
+		ut_ad(lock_table_has_locks(index->table));
 	}
 
 	online = dict_index_is_online_ddl(index);
 	if (online) {
-		ut_ad(node->trx->dict_operation_lock_mode != RW_X_LATCH);
+		ut_ad(!node->trx->dict_operation_lock_mode);
 		mtr_s_lock_index(index, &mtr);
 	}
 
@@ -327,17 +325,7 @@ row_undo_mod_clust(
 		ut_ad(err == DB_SUCCESS || err == DB_OUT_OF_FILE_SPACE);
 	}
 
-	/* Online rebuild cannot be initiated while we are holding
-	dict_sys.latch and index->lock. (It can be aborted.) */
-	ut_ad(online || !dict_index_is_online_ddl(index));
-
-	if (err == DB_SUCCESS && online) {
-
-		ut_ad(rw_lock_own_flagged(
-				&index->lock,
-				RW_LOCK_FLAG_S | RW_LOCK_FLAG_X
-				| RW_LOCK_FLAG_SX));
-
+	if (err == DB_SUCCESS && online && dict_index_is_online_ddl(index)) {
 		switch (node->rec_type) {
 		case TRX_UNDO_DEL_MARK_REC:
 			row_log_table_insert(
@@ -911,37 +899,6 @@ func_exit_no_pcur:
 }
 
 /***********************************************************//**
-Flags a secondary index corrupted. */
-static MY_ATTRIBUTE((nonnull))
-void
-row_undo_mod_sec_flag_corrupted(
-/*============================*/
-	trx_t*		trx,	/*!< in/out: transaction */
-	dict_index_t*	index)	/*!< in: secondary index */
-{
-	ut_ad(!dict_index_is_clust(index));
-
-	switch (trx->dict_operation_lock_mode) {
-	case RW_S_LATCH:
-		/* Because row_undo() is holding an S-latch
-		on the data dictionary during normal rollback,
-		we can only mark the index corrupted in the
-		data dictionary cache. TODO: fix this somehow.*/
-		mutex_enter(&dict_sys.mutex);
-		dict_set_corrupted_index_cache_only(index);
-		mutex_exit(&dict_sys.mutex);
-		break;
-	default:
-		ut_ad(0);
-		/* fall through */
-	case RW_X_LATCH:
-		/* This should be the rollback of a data dictionary
-		transaction. */
-		dict_set_corrupted(index, trx, "rollback");
-	}
-}
-
-/***********************************************************//**
 Undoes a modify in secondary indexes when undo record type is UPD_DEL.
 @return DB_SUCCESS or DB_OUT_OF_FILE_SPACE */
 static MY_ATTRIBUTE((nonnull, warn_unused_result))
@@ -1054,8 +1011,7 @@ row_undo_mod_del_mark_sec(
 		}
 
 		if (err == DB_DUPLICATE_KEY) {
-			row_undo_mod_sec_flag_corrupted(
-				thr_get_trx(thr), index);
+			index->type |= DICT_CORRUPT;
 			err = DB_SUCCESS;
 			/* Do not return any error to the caller. The
 			duplicate will be reported by ALTER TABLE or
@@ -1200,8 +1156,7 @@ row_undo_mod_upd_exist_sec(
 		}
 
 		if (err == DB_DUPLICATE_KEY) {
-			row_undo_mod_sec_flag_corrupted(
-				thr_get_trx(thr), index);
+			index->type |= DICT_CORRUPT;
 			err = DB_SUCCESS;
 		} else if (err != DB_SUCCESS) {
 			break;
@@ -1364,6 +1319,8 @@ row_undo_mod(
 		return DB_SUCCESS;
 	}
 
+	ut_ad(node->table->is_temporary()
+	      || lock_table_has_locks(node->table));
 	node->index = dict_table_get_first_index(node->table);
 	ut_ad(dict_index_is_clust(node->index));
 
diff --git a/storage/innobase/row/row0undo.cc b/storage/innobase/row/row0undo.cc
index a07258910b4..789bcc441bc 100644
--- a/storage/innobase/row/row0undo.cc
+++ b/storage/innobase/row/row0undo.cc
@@ -418,15 +418,17 @@ row_undo(
 		return DB_SUCCESS;
 	}
 
-	/* Prevent DROP TABLE etc. while we are rolling back this row.
-	If we are doing a TABLE CREATE or some other dictionary operation,
-	then we already have dict_sys.latch locked in x-mode. Do not
-	try to lock again, because that would cause a hang. */
-
+	/* Prevent prepare_inplace_alter_table_dict() from adding
+	dict_table_t::indexes while we are processing the record.
+	Recovered transactions are not protected by MDL, and the
+	secondary index creation is not protected by table locks
+	for online operation. (A table lock would only be acquired
+	when committing the ALTER TABLE operation.) */
 	trx_t* trx = node->trx;
-	const bool locked_data_dict = (trx->dict_operation_lock_mode == 0);
+	const bool locked_data_dict = UNIV_UNLIKELY(trx->is_recovered)
+		&& !trx->dict_operation_lock_mode;
 
-	if (locked_data_dict) {
+	if (UNIV_UNLIKELY(locked_data_dict)) {
 		row_mysql_freeze_data_dictionary(trx);
 	}
 
@@ -446,7 +448,7 @@ row_undo(
 		err = DB_CORRUPTION;
 	}
 
-	if (locked_data_dict) {
+	if (UNIV_UNLIKELY(locked_data_dict)) {
 		row_mysql_unfreeze_data_dictionary(trx);
 	}
 
diff --git a/storage/innobase/srv/srv0srv.cc b/storage/innobase/srv/srv0srv.cc
index 0bd52cb486d..a5305684935 100644
--- a/storage/innobase/srv/srv0srv.cc
+++ b/storage/innobase/srv/srv0srv.cc
@@ -1910,6 +1910,33 @@ srv_get_active_thread_type(void)
 	return(ret);
 }
 
+/** Wake up the InnoDB master thread if it was suspended (not sleeping). */
+void
+srv_active_wake_master_thread_low()
+{
+	ut_ad(!srv_read_only_mode);
+	ut_ad(!mutex_own(&srv_sys.mutex));
+
+	srv_inc_activity_count();
+
+	if (srv_sys.n_threads_active[SRV_MASTER] == 0) {
+		srv_slot_t*	slot;
+
+		srv_sys_mutex_enter();
+
+		slot = &srv_sys.sys_threads[SRV_MASTER_SLOT];
+
+		/* Only if the master thread has been started. */
+
+		if (slot->in_use) {
+			ut_a(srv_slot_get_type(slot) == SRV_MASTER);
+			os_event_set(slot->event);
+		}
+
+		srv_sys_mutex_exit();
+	}
+}
+
 /** Wake up the purge threads if there is work to do. */
 void
 srv_wake_purge_thread_if_not_active()
@@ -1925,6 +1952,14 @@ srv_wake_purge_thread_if_not_active()
 	}
 }
 
+/** Wake up the master thread if it is suspended or being suspended. */
+void
+srv_wake_master_thread()
+{
+	srv_inc_activity_count();
+	srv_release_threads(SRV_MASTER, 1);
+}
+
 /*******************************************************************//**
 Get current server activity count. We don't hold srv_sys::mutex while
 reading this value as it is only used in heuristics.
@@ -1936,20 +1971,15 @@ srv_get_activity_count(void)
 	return(srv_sys.activity_count);
 }
 
-/** Check if there has been any activity.
-@param[in,out]  activity_count  recent activity count to be returned
-if there is a change
+/*******************************************************************//**
+Check if there has been any activity.
 @return FALSE if no change in activity counter. */
-bool srv_check_activity(ulint  *activity_count)
+ibool
+srv_check_activity(
+/*===============*/
+	ulint		old_activity_count)	/*!< in: old activity count */
 {
-  ulint new_activity_count= srv_sys.activity_count;
-  if (new_activity_count != *activity_count)
-  {
-    *activity_count= new_activity_count;
-    return true;
-  }
-
-  return false;
+	return(srv_sys.activity_count != old_activity_count);
 }
 
 /********************************************************************//**
@@ -2346,30 +2376,48 @@ DECLARE_THREAD(srv_master_thread)(
 	slot = srv_reserve_slot(SRV_MASTER);
 	ut_a(slot == srv_sys.sys_threads);
 
+loop:
 	while (srv_shutdown_state <= SRV_SHUTDOWN_INITIATED) {
 		srv_master_sleep();
 
 		MONITOR_INC(MONITOR_MASTER_THREAD_SLEEP);
 
-		if (srv_check_activity(&old_activity_count)) {
+		if (srv_check_activity(old_activity_count)) {
+			old_activity_count = srv_get_activity_count();
 			srv_master_do_active_tasks();
 		} else {
 			srv_master_do_idle_tasks();
 		}
 	}
 
-        ut_ad(srv_shutdown_state == SRV_SHUTDOWN_EXIT_THREADS
-              || srv_shutdown_state == SRV_SHUTDOWN_CLEANUP);
-
-	if (srv_shutdown_state == SRV_SHUTDOWN_CLEANUP
-            && srv_fast_shutdown < 2) {
-	        srv_shutdown(srv_fast_shutdown == 0);
+	switch (srv_shutdown_state) {
+	case SRV_SHUTDOWN_NONE:
+	case SRV_SHUTDOWN_INITIATED:
+		break;
+	case SRV_SHUTDOWN_FLUSH_PHASE:
+	case SRV_SHUTDOWN_LAST_PHASE:
+		ut_ad(0);
+		/* fall through */
+	case SRV_SHUTDOWN_EXIT_THREADS:
+		/* srv_init_abort() must have been invoked */
+	case SRV_SHUTDOWN_CLEANUP:
+		if (srv_shutdown_state == SRV_SHUTDOWN_CLEANUP
+		    && srv_fast_shutdown < 2) {
+			srv_shutdown(srv_fast_shutdown == 0);
+		}
+		srv_suspend_thread(slot);
+		my_thread_end();
+		os_thread_exit();
 	}
 
+	srv_main_thread_op_info = "suspending";
+
 	srv_suspend_thread(slot);
-	my_thread_end();
-	os_thread_exit();
-	OS_THREAD_DUMMY_RETURN;
+
+	srv_main_thread_op_info = "waiting for server activity";
+
+	srv_resume_thread(slot);
+	goto loop;
 }
 
 /** @return whether purge should exit due to shutdown */
@@ -2537,13 +2585,15 @@ static uint32_t srv_do_purge(ulint* n_total_purged
 				++n_use_threads;
 			}
 
-		} else if (srv_check_activity(&old_activity_count)
+		} else if (srv_check_activity(old_activity_count)
 			   && n_use_threads > 1) {
 
 			/* History length same or smaller since last snapshot,
 			use fewer threads. */
 
 			--n_use_threads;
+
+			old_activity_count = srv_get_activity_count();
 		}
 
 		/* Ensure that the purge threads are less than what
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index 42d757d1a84..746935f5794 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -1088,7 +1088,7 @@ srv_shutdown_all_bg_threads()
 			if (srv_start_state_is_set(SRV_START_STATE_MASTER)) {
 				/* c. We wake the master thread so that
 				it exits */
-				srv_inc_activity_count();
+				srv_wake_master_thread();
 			}
 
 			if (srv_start_state_is_set(SRV_START_STATE_PURGE)) {
@@ -2420,7 +2420,7 @@ void srv_shutdown_bg_undo_sources()
 		fts_optimize_shutdown();
 		dict_stats_shutdown();
 		while (row_get_background_drop_list_len_low()) {
-			srv_inc_activity_count();
+			srv_wake_master_thread();
 			os_thread_yield();
 		}
 		srv_undo_sources = false;
diff --git a/storage/innobase/trx/trx0roll.cc b/storage/innobase/trx/trx0roll.cc
index 917274d5f0d..32b053b2bab 100644
--- a/storage/innobase/trx/trx0roll.cc
+++ b/storage/innobase/trx/trx0roll.cc
@@ -160,6 +160,9 @@ trx_rollback_to_savepoint_low(
 
 	mem_heap_free(heap);
 
+	/* There might be work for utility threads.*/
+	srv_active_wake_master_thread();
+
 	MONITOR_DEC(MONITOR_TRX_ACTIVE);
 }
 
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index d63d7db4e32..7da5b2bc738 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -1485,6 +1485,11 @@ inline void trx_t::commit_in_memory(const mtr_t *mtr)
       must_flush_log_later= true;
     else if (srv_flush_log_at_trx_commit)
       trx_flush_log_if_needed(commit_lsn, this);
+
+    /* Tell server some activity has happened, since the trx does
+    changes something. Background utility threads like master thread,
+    purge thread or page_cleaner thread might have some work to do. */
+    srv_active_wake_master_thread();
   }
 
   ut_ad(!rsegs.m_noredo.undo);