MDEV-23484 Rollback unnecessarily acquires dict_sys.latch

row_undo(): Remove the unnecessary acquisition and release of
dict_sys.latch. This was supposed to prevent the table from being
dropped while the undo log record is being rolled back. But, thanks to
trx_resurrect_table_locks() that was introduced in
mysql/mysql-server@935ba09
and commit c291ddf
as well as commit 1bd681c
(MDEV-25506 part 3) tables will be protected against dropping
due to table locks.

This reverts commit 0049d5b
(which had reverted a previous attempt of fixing this) and addresses
an earlier race condition with the following:

prepare_inplace_alter_table_dict(): If recovered transactions hold
locks on the table while we are executing online ADD INDEX, acquire
a table lock so that the rollback of those recovered transactions will
not interfere with the ADD INDEX.

storage/innobase/handler/handler0alter.cc

@@ -6158,7 +6158,7 @@ prepare_inplace_alter_table_dict(
 	dict_table_t*		user_table;
 	dict_index_t*		fts_index	= NULL;
 	bool			new_clustered	= false;
-	dberr_t			error;
+	dberr_t			error		= DB_SUCCESS;
 	ulint			num_fts_index;
 	dict_add_v_col_t*	add_v = NULL;
 	ha_innobase_inplace_ctx*ctx;
@@ -6282,11 +6282,27 @@ prepare_inplace_alter_table_dict(
 	/* Acquire a lock on the table before creating any indexes. */
 	bool table_lock_failed = false;
 
-	if (ctx->online) {
-		error = DB_SUCCESS;
-	} else {
+	if (!ctx->online) {
+acquire_lock:
 		ctx->prebuilt->trx->op_info = "acquiring table lock";
-		error = lock_table_for_trx(ctx->new_table, ctx->trx, LOCK_S);
+		error = lock_table_for_trx(user_table, ctx->trx, LOCK_S);
+	} else if (add_key_nums) {
+		/* FIXME: trx_resurrect_table_locks() will not resurrect
+		MDL for any recovered transactions that may hold locks on
+		the table. We will prevent race conditions by "unnecessarily"
+		acquiring an InnoDB table lock even for online operation,
+		to ensure that the rollback of recovered transactions will
+		not run concurrently with online ADD INDEX. */
+		user_table->lock_mutex_lock();
+		for (lock_t *lock = UT_LIST_GET_FIRST(user_table->locks);
+		     lock;
+		     lock = UT_LIST_GET_NEXT(un_member.tab_lock.locks, lock)) {
+			if (lock->trx->is_recovered) {
+				user_table->lock_mutex_unlock();
+				goto acquire_lock;
+			}
+		}
+		user_table->lock_mutex_unlock();
 	}
 
 	if (fts_exist) {

storage/innobase/row/row0uins.cc

@@ -101,8 +101,7 @@ row_undo_ins_remove_clust_rec(
 		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);
@@ -617,6 +616,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);

storage/innobase/row/row0umod.cc

@@ -267,7 +267,6 @@ 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);
 
 	log_free_check();
@@ -285,7 +284,7 @@ row_undo_mod_clust(
 
 	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);
 	}
 
@@ -324,13 +323,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(index->lock.have_any());
-
+	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(
@@ -920,37 +913,6 @@ row_undo_mod_del_unmark_sec_and_undo_update(
 	return(err);
 }
 
-/***********************************************************//**
-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.*/
-		dict_sys.mutex_lock();
-		dict_set_corrupted_index_cache_only(index);
-		dict_sys.mutex_unlock();
-		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 */
@@ -1064,8 +1026,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
@@ -1210,8 +1171,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;
@@ -1374,6 +1334,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));
 

storage/innobase/row/row0undo.cc

@@ -401,19 +401,6 @@ row_undo(
 		return DB_SUCCESS;
 	}
 
-	/* 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;
-
-	if (UNIV_UNLIKELY(locked_data_dict)) {
-		row_mysql_freeze_data_dictionary(trx);
-	}
-
 	dberr_t err;
 
 	switch (node->state) {
@@ -430,11 +417,6 @@ row_undo(
 		err = DB_CORRUPTION;
 	}
 
-	if (locked_data_dict) {
-
-		row_mysql_unfreeze_data_dictionary(trx);
-	}
-
 	node->state = UNDO_NODE_FETCH_NEXT;
 	btr_pcur_close(&(node->pcur));