Revert MDEV-23484 Rollback unnecessarily acquires dict_operation_lock

In row_undo_ins_remove_clust_rec() and similar places,
an assertion !node->trx->dict_operation_lock_mode could fail,
because an online ALTER is not allowed to run at the same time
while DDL operations on the table are being rolled back.

This race condition would be fixed by always acquiring an InnoDB
table lock in ha_innobase::prepare_inplace_alter_table() or
prepare_inplace_alter_table_dict(), or by ensuring that recovered
transactions are protected by MDL that would block concurrent DDL
until the rollback has been completed.

This reverts commit 15093639
and commit 22c4a751.

storage/innobase/handler/handler0alter.cc

@@ -8273,11 +8273,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 meta-data lock (MDL)
+		change the table definition. The MySQL meta-data lock
 		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 MDL. */
+		holding InnoDB locks only, not MySQL locks. */
 
 		dberr_t error = row_merge_lock_table(
 			m_prebuilt->trx, ctx->old_table, LOCK_X);

storage/innobase/row/row0uins.cc

@@ -81,7 +81,6 @@ row_undo_ins_remove_clust_rec(
 		mtr.set_log_mode(MTR_LOG_NO_REDO);
 	} else {
 		mtr.set_named_space(index->space);
-		ut_ad(lock_table_has_locks(index->table));
 	}
 
 	/* This is similar to row_undo_mod_clust(). The DDL thread may
@@ -92,7 +91,8 @@ row_undo_ins_remove_clust_rec(
 
 	online = dict_index_is_online_ddl(index);
 	if (online) {
-		ut_ad(!node->trx->dict_operation_lock_mode);
+		ut_ad(node->trx->dict_operation_lock_mode
+		      != RW_X_LATCH);
 		ut_ad(node->table->id != DICT_INDEXES_ID);
 		mtr_s_lock(dict_index_get_lock(index), &mtr);
 	}
@@ -491,9 +491,6 @@ 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

@@ -264,7 +264,10 @@ 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_operation_lock,
+				  RW_LOCK_FLAG_X | RW_LOCK_FLAG_S));
 
 	log_free_check();
 	pcur = &node->pcur;
@@ -275,12 +278,11 @@ row_undo_mod_clust(
 		mtr.set_log_mode(MTR_LOG_NO_REDO);
 	} else {
 		mtr.set_named_space(index->space);
-		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);
+		ut_ad(node->trx->dict_operation_lock_mode != RW_X_LATCH);
 		mtr_s_lock(dict_index_get_lock(index), &mtr);
 	}
 
@@ -319,7 +321,17 @@ row_undo_mod_clust(
 		ut_ad(err == DB_SUCCESS || err == DB_OUT_OF_FILE_SPACE);
 	}
 
-	if (err == DB_SUCCESS && online && dict_index_is_online_ddl(index)) {
+	/* Online rebuild cannot be initiated while we are holding
+	dict_operation_lock 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));
+
 		switch (node->rec_type) {
 		case TRX_UNDO_DEL_MARK_REC:
 			row_log_table_insert(
@@ -794,6 +806,37 @@ 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.*/
+		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 */
@@ -907,7 +950,8 @@ row_undo_mod_del_mark_sec(
 		}
 
 		if (err == DB_DUPLICATE_KEY) {
-			index->type |= DICT_CORRUPT;
+			row_undo_mod_sec_flag_corrupted(
+				thr_get_trx(thr), index);
 			err = DB_SUCCESS;
 			/* Do not return any error to the caller. The
 			duplicate will be reported by ALTER TABLE or
@@ -1053,7 +1097,8 @@ row_undo_mod_upd_exist_sec(
 		}
 
 		if (err == DB_DUPLICATE_KEY) {
-			index->type |= DICT_CORRUPT;
+			row_undo_mod_sec_flag_corrupted(
+				thr_get_trx(thr), index);
 			err = DB_SUCCESS;
 		} else if (err != DB_SUCCESS) {
 			break;
@@ -1205,8 +1250,6 @@ 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));
 	/* Skip the clustered index (the first index) */

storage/innobase/row/row0undo.cc

@@ -279,16 +279,15 @@ row_undo(
 			? UNDO_NODE_INSERT : UNDO_NODE_MODIFY;
 	}
 
-	/* 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.) */
-	const bool locked_data_dict = UNIV_UNLIKELY(trx->is_recovered)
-		&& !trx->dict_operation_lock_mode;
-
-	if (UNIV_UNLIKELY(locked_data_dict)) {
+	/* 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_operation_lock locked in x-mode. Do not
+	try to lock again, because that would cause a hang. */
+
+	const bool locked_data_dict = (trx->dict_operation_lock_mode == 0);
+
+	if (locked_data_dict) {
+
 		row_mysql_freeze_data_dictionary(trx);
 	}
 
@@ -304,7 +303,8 @@ row_undo(
 		err = row_undo_mod(node, thr);
 	}
 
-	if (UNIV_UNLIKELY(locked_data_dict)) {
+	if (locked_data_dict) {
+
 		row_mysql_unfreeze_data_dictionary(trx);
 	}