MDEV-26217 Failing assertion: list.count > 0 in ut_list_remove or Assertion `lock->trx == this' failed in dberr_t trx_t::drop_table

This follows up the previous fix in
commit c3c53926c467c95386ae98d61ada87294bd61478 (MDEV-26554).

ha_innobase::delete_table(): Work around the insufficient
metadata locking (MDL) during DML operations by acquiring exclusive
InnoDB table locks on all child tables. Previously, this was only
done on TRUNCATE and ALTER.

ibuf_delete_rec(), btr_cur_optimistic_delete(): Do not invoke
lock_update_delete() during change buffer operations.
The revised trx_t::commit(std::vector<pfs_os_file_t>&) will
hold exclusive lock_sys.latch while invoking fil_delete_tablespace(),
which in turn may invoke ibuf_delete_rec().

dict_index_t::has_locking(): A new predicate, replacing the dummy
!dict_table_is_locking_disabled(index->table). Used for skipping lock
operations during ibuf_delete_rec().

trx_t::commit(std::vector<pfs_os_file_t>&): Release the locks
and remove the table from the cache while holding exclusive
lock_sys.latch.

trx_t::commit_in_memory(): Skip release_locks() if dict_operation holds.

trx_t::commit(): Reset dict_operation before invoking commit_in_memory()
via commit_persist().

lock_release_on_drop(): Release locks while lock_sys.latch is
exclusively locked.

lock_table(): Add a parameter for a pointer to the table.
We must not dereference the table before a lock_sys.latch has
been acquired. If the pointer to the table does not match the table
at that point, the table is invalid and DB_DEADLOCK will be returned.

row_ins_foreign_check_on_constraint(): Improve the checks.
Remove a bogus DB_LOCK_WAIT_TIMEOUT return that was needed
before commit c5fd9aa562fb15e8d6ededceccbec0c9792a3243 (MDEV-25919).

row_upd_check_references_constraints(),
wsrep_row_upd_check_foreign_constraints(): Simplify checks.

17 files changed, 270 insertions, 213 deletions

diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index 2d83f16b537..7fdecf9e4ef 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -1378,7 +1378,7 @@ static void btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
                 block->page.frame + PAGE_MAX_TRX_ID + PAGE_HEADER,
                 PAGE_DATA - (PAGE_MAX_TRX_ID + PAGE_HEADER)));
 
-  if (!dict_table_is_locking_disabled(index->table))
+  if (index->has_locking())
     lock_move_reorganize_page(block, old);
 
   /* Write log for the changes, if needed. */
@@ -1849,8 +1849,11 @@ btr_root_raise_and_insert(
 				   root->page.frame, index, mtr);
 
 		/* Update the lock table and possible hash index. */
-		lock_move_rec_list_end(new_block, root,
-				       page_get_infimum_rec(root->page.frame));
+		if (index->has_locking()) {
+			lock_move_rec_list_end(
+				new_block, root,
+				page_get_infimum_rec(root->page.frame));
+		}
 
 		/* Move any existing predicate locks */
 		if (dict_index_is_spatial(index)) {
@@ -1898,7 +1901,7 @@ btr_root_raise_and_insert(
 	information of the record to be inserted on the infimum of the
 	root page: we cannot discard the lock structs on the root page */
 
-	if (!dict_table_is_locking_disabled(index->table)) {
+	if (index->has_locking()) {
 		lock_update_root_raise(*new_block, root_id);
 	}
 
@@ -2594,7 +2597,7 @@ btr_insert_into_right_sibling(
 	max_size = page_get_max_insert_size_after_reorganize(next_page, 1);
 
 	/* Extends gap lock for the next page */
-	if (!dict_table_is_locking_disabled(cursor->index->table)) {
+	if (cursor->index->has_locking()) {
 		lock_update_split_left(next_block, block);
 	}
 
@@ -2907,9 +2910,11 @@ insert_empty:
 						 ULINT_UNDEFINED, mtr);
 
 			/* Update the lock table and possible hash index. */
-			lock_move_rec_list_start(
-				new_block, block, move_limit,
-				new_page + PAGE_NEW_INFIMUM);
+			if (cursor->index->has_locking()) {
+				lock_move_rec_list_start(
+					new_block, block, move_limit,
+					new_page + PAGE_NEW_INFIMUM);
+			}
 
 			btr_search_move_or_delete_hash_entries(
 				new_block, block);
@@ -2923,7 +2928,7 @@ insert_empty:
 		left_block = new_block;
 		right_block = block;
 
-		if (!dict_table_is_locking_disabled(cursor->index->table)) {
+		if (cursor->index->has_locking()) {
 			lock_update_split_left(right_block, left_block);
 		}
 	} else {
@@ -2949,7 +2954,10 @@ insert_empty:
 						   cursor->index, mtr);
 
 			/* Update the lock table and possible hash index. */
-			lock_move_rec_list_end(new_block, block, move_limit);
+			if (cursor->index->has_locking()) {
+				lock_move_rec_list_end(new_block, block,
+						       move_limit);
+			}
 
 			btr_search_move_or_delete_hash_entries(
 				new_block, block);
@@ -2965,7 +2973,7 @@ insert_empty:
 		left_block = block;
 		right_block = new_block;
 
-		if (!dict_table_is_locking_disabled(cursor->index->table)) {
+		if (cursor->index->has_locking()) {
 			lock_update_split_right(right_block, left_block);
 		}
 	}
@@ -3265,8 +3273,10 @@ btr_lift_page_up(
 
 		/* Update the lock table and possible hash index. */
 
-		lock_move_rec_list_end(father_block, block,
-				       page_get_infimum_rec(page));
+		if (index->has_locking()) {
+			lock_move_rec_list_end(father_block, block,
+					       page_get_infimum_rec(page));
+		}
 
 		/* Also update the predicate locks */
 		if (dict_index_is_spatial(index)) {
@@ -3277,7 +3287,7 @@ btr_lift_page_up(
 		}
 	}
 
-	if (!dict_table_is_locking_disabled(index->table)) {
+	if (index->has_locking()) {
 		const page_id_t id{block->page.id()};
 		/* Free predicate page locks on the block */
 		if (index->is_spatial()) {
@@ -3541,7 +3551,7 @@ retry:
 			lock_sys.prdt_page_free_from_discard(id);
 		} else {
 			btr_cur_node_ptr_delete(&father_cursor, mtr);
-			if (!dict_table_is_locking_disabled(index->table)) {
+			if (index->has_locking()) {
 				lock_update_merge_left(
 					*merge_block, orig_pred, id);
 			}
@@ -3706,7 +3716,7 @@ retry:
 							   mtr);
 			}
 
-			if (!dict_table_is_locking_disabled(index->table)) {
+			if (index->has_locking()) {
 				lock_update_merge_right(
 					merge_block, orig_succ, block);
 			}
@@ -3848,7 +3858,7 @@ btr_discard_only_page_on_level(
 		}
 		father = btr_cur_get_block(&cursor);
 
-		if (!dict_table_is_locking_disabled(index->table)) {
+		if (index->has_locking()) {
 			lock_update_discard(
 				father, PAGE_HEAP_NO_SUPREMUM, block);
 		}
@@ -4026,7 +4036,7 @@ btr_discard_page(
 	}
 #endif /* UNIV_ZIP_DEBUG */
 
-	if (!dict_table_is_locking_disabled(index->table)) {
+	if (index->has_locking()) {
 		if (left_page_no != FIL_NULL) {
 			lock_update_discard(merge_block, PAGE_HEAP_NO_SUPREMUM,
 					    block);
@@ -4035,10 +4045,10 @@ btr_discard_page(
 					    lock_get_min_heap_no(merge_block),
 					    block);
 		}
-	}
 
-	if (dict_index_is_spatial(index)) {
-		rtr_check_discard_page(index, cursor, block);
+		if (index->is_spatial()) {
+			rtr_check_discard_page(index, cursor, block);
+		}
 	}
 
 	/* Free the file page */
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index f12561c3126..1778a655511 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -4254,6 +4254,7 @@ btr_cur_update_in_place(
 	ut_ad(page_is_leaf(cursor->page_cur.block->page.frame));
 	rec = btr_cur_get_rec(cursor);
 	index = cursor->index;
+	ut_ad(!index->is_ibuf());
 	ut_ad(rec_offs_validate(rec, index, offsets));
 	ut_ad(!!page_rec_is_comp(rec) == dict_table_is_comp(index->table));
 	ut_ad(trx_id > 0 || (flags & BTR_KEEP_SYS_FLAG)
@@ -4554,6 +4555,7 @@ btr_cur_optimistic_update(
 	page = buf_block_get_frame(block);
 	rec = btr_cur_get_rec(cursor);
 	index = cursor->index;
+	ut_ad(index->has_locking());
 	ut_ad(trx_id > 0 || (flags & BTR_KEEP_SYS_FLAG)
 	      || index->table->is_temporary());
 	ut_ad(!!page_rec_is_comp(rec) == dict_table_is_comp(index->table));
@@ -4737,7 +4739,7 @@ any_extern:
 	/* Ok, we may do the replacement. Store on the page infimum the
 	explicit locks on rec, before deleting rec (see the comment in
 	btr_cur_pessimistic_update). */
-	if (!dict_table_is_locking_disabled(index->table)) {
+	if (index->has_locking()) {
 		lock_rec_store_on_page_infimum(block, rec);
 	}
 
@@ -4771,7 +4773,7 @@ any_extern:
 		would have too many fields, and we would be unable to
 		know the size of the freed record. */
 		btr_page_reorganize(page_cursor, index, mtr);
-	} else if (!dict_table_is_locking_disabled(index->table)) {
+	} else {
 		/* Restore the old explicit lock state on the record */
 		lock_rec_restore_from_page_infimum(*block, rec,
 						   block->page.id());
@@ -4901,6 +4903,7 @@ btr_cur_pessimistic_update(
 	block = btr_cur_get_block(cursor);
 	page_zip = buf_block_get_page_zip(block);
 	index = cursor->index;
+	ut_ad(index->has_locking());
 
 	ut_ad(mtr->memo_contains_flagged(&index->lock, MTR_MEMO_X_LOCK |
 					 MTR_MEMO_SX_LOCK));
@@ -5097,9 +5100,7 @@ btr_cur_pessimistic_update(
 		in the lock system delete the lock structs set on the
 		root page even if the root page carries just node
 		pointers. */
-		if (!dict_table_is_locking_disabled(index->table)) {
-			lock_rec_store_on_page_infimum(block, rec);
-		}
+		lock_rec_store_on_page_infimum(block, rec);
 	}
 
 #ifdef UNIV_ZIP_DEBUG
@@ -5131,7 +5132,7 @@ btr_cur_pessimistic_update(
 				btr_set_instant(page_cursor->block, *index,
 						mtr);
 			}
-		} else if (!dict_table_is_locking_disabled(index->table)) {
+		} else {
 			lock_rec_restore_from_page_infimum(
 				*btr_cur_get_block(cursor), rec,
 				block->page.id());
@@ -5277,7 +5278,7 @@ btr_cur_pessimistic_update(
 		know the size of the freed record. */
 		btr_page_reorganize(page_cursor, index, mtr);
 		rec = page_cursor->rec;
-	} else if (!dict_table_is_locking_disabled(index->table)) {
+	} else {
 		lock_rec_restore_from_page_infimum(
 			*btr_cur_get_block(cursor), rec, block->page.id());
 	}
@@ -5287,7 +5288,7 @@ btr_cur_pessimistic_update(
 	record was nonexistent, the supremum might have inherited its locks
 	from a wrong record. */
 
-	if (!was_first && !dict_table_is_locking_disabled(index->table)) {
+	if (!was_first) {
 		btr_cur_pess_upd_restore_supremum(btr_cur_get_block(cursor),
 						  rec, mtr);
 	}
@@ -5451,15 +5452,13 @@ It is assumed that the mtr has an x-latch on the page where the cursor is
 positioned, but no latch on the whole tree.
 @return TRUE if success, i.e., the page did not become too empty */
 ibool
-btr_cur_optimistic_delete_func(
-/*===========================*/
+btr_cur_optimistic_delete(
+/*======================*/
 	btr_cur_t*	cursor,	/*!< in: cursor on leaf page, on the record to
 				delete; cursor stays valid: if deletion
 				succeeds, on function exit it points to the
 				successor of the deleted record */
-#ifdef UNIV_DEBUG
 	ulint		flags,	/*!< in: BTR_CREATE_FLAG or 0 */
-#endif /* UNIV_DEBUG */
 	mtr_t*		mtr)	/*!< in: mtr; if this function returns
 				TRUE on a leaf page of a secondary
 				index, the mtr must be committed
@@ -5531,7 +5530,7 @@ btr_cur_optimistic_delete_func(
 			|| (first_rec != rec
 			    && rec_is_add_metadata(first_rec, *index));
 		if (UNIV_LIKELY(empty_table)) {
-			if (UNIV_LIKELY(!is_metadata)) {
+			if (UNIV_LIKELY(!is_metadata && !flags)) {
 				lock_update_delete(block, rec);
 			}
 			btr_page_empty(block, buf_block_get_page_zip(block),
@@ -5570,7 +5569,9 @@ btr_cur_optimistic_delete_func(
 					    cursor->index, mtr);
 			goto func_exit;
 		} else {
-			lock_update_delete(block, rec);
+			if (!flags) {
+				lock_update_delete(block, rec);
+			}
 
 			btr_search_update_hash_on_delete(cursor);
 		}
diff --git a/storage/innobase/dict/drop.cc b/storage/innobase/dict/drop.cc
index 0263c08118e..ff5ceee5e43 100644
--- a/storage/innobase/dict/drop.cc
+++ b/storage/innobase/dict/drop.cc
@@ -68,6 +68,7 @@ before transaction commit and must be rolled back explicitly are as follows:
 
 #include "dict0defrag_bg.h"
 #include "btr0defragment.h"
+#include "lock0lock.h"
 
 #include "que0que.h"
 #include "pars0pars.h"
@@ -234,6 +235,24 @@ void trx_t::commit(std::vector<pfs_os_file_t> &deleted)
   if (dict_operation)
   {
     ut_ad(dict_sys.locked());
+    lock_sys.wr_lock(SRW_LOCK_CALL);
+    mutex_lock();
+    lock_release_on_drop(this);
+    ut_ad(UT_LIST_GET_LEN(lock.trx_locks) == 0);
+    ut_ad(ib_vector_is_empty(autoinc_locks));
+    mem_heap_empty(lock.lock_heap);
+    lock.table_locks.clear();
+    lock.was_chosen_as_deadlock_victim= false;
+    lock.n_rec_locks= 0;
+    while (dict_table_t *table= UT_LIST_GET_FIRST(lock.evicted_tables))
+    {
+      UT_LIST_REMOVE(lock.evicted_tables, table);
+      dict_mem_table_free(table);
+    }
+    dict_operation= false;
+    id= 0;
+    mutex_unlock();
+
     for (const auto &p : mod_tables)
     {
       if (p.second.is_dropped())
@@ -255,6 +274,12 @@ void trx_t::commit(std::vector<pfs_os_file_t> &deleted)
         }
       }
     }
+
+    lock_sys.wr_unlock();
+
+    mysql_mutex_lock(&lock_sys.wait_mutex);
+    lock_sys.deadlock_check();
+    mysql_mutex_unlock(&lock_sys.wait_mutex);
   }
   commit_cleanup();
 }
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 8e8191f6f68..2ce02ca018c 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -13490,6 +13490,7 @@ int ha_innobase::delete_table(const char *name)
   dict_sys.unlock();
 
   trx_t *trx= parent_trx;
+  dberr_t err= DB_SUCCESS;
   if (!trx->lock.table_locks.empty() &&
       thd_ddl_options(trx->mysql_thd)->is_create_select())
   {
@@ -13509,11 +13510,28 @@ int ha_innobase::delete_table(const char *name)
   {
     trx= innobase_trx_allocate(thd);
     trx_start_for_ddl(trx);
+
+    if (table->name.is_temporary())
+      /* There is no need to lock any FOREIGN KEY child tables. */;
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+    else if (table->name.part())
+      /* FOREIGN KEY constraints cannot exist on partitioned tables. */;
+#endif
+    else
+    {
+      dict_sys.freeze(SRW_LOCK_CALL);
+      for (const dict_foreign_t* f : table->referenced_set)
+        if (dict_table_t* child= f->foreign_table)
+          if ((err= lock_table_for_trx(child, trx, LOCK_X)) != DB_SUCCESS)
+            break;
+      dict_sys.unfreeze();
+    }
   }
 
   dict_table_t *table_stats= nullptr, *index_stats= nullptr;
   MDL_ticket *mdl_table= nullptr, *mdl_index= nullptr;
-  dberr_t err= lock_table_for_trx(table, trx, LOCK_X);
+  if (err == DB_SUCCESS)
+    err= lock_table_for_trx(table, trx, LOCK_X);
 
   const bool fts= err == DB_SUCCESS &&
     (table->flags2 & (DICT_TF2_FTS_HAS_DOC_ID | DICT_TF2_FTS));
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index a4469910766..f4e9b75dc7d 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -4030,7 +4030,6 @@ static MY_ATTRIBUTE((warn_unused_result, nonnull))
 bool ibuf_delete_rec(const page_id_t page_id, btr_pcur_t* pcur,
 		     const dtuple_t* search_tuple, mtr_t* mtr)
 {
-	ibool		success;
 	page_t*		root;
 	dberr_t		err;
 
@@ -4041,10 +4040,8 @@ bool ibuf_delete_rec(const page_id_t page_id, btr_pcur_t* pcur,
 	ut_ad(ibuf_rec_get_space(mtr, btr_pcur_get_rec(pcur))
 	      == page_id.space());
 
-	success = btr_cur_optimistic_delete(btr_pcur_get_btr_cur(pcur),
-					    0, mtr);
-
-	if (success) {
+	if (btr_cur_optimistic_delete(btr_pcur_get_btr_cur(pcur),
+				      BTR_CREATE_FLAG, mtr)) {
 		if (page_is_empty(btr_pcur_get_page(pcur))) {
 			/* If a B-tree page is empty, it must be the root page
 			and the whole B-tree must be empty. InnoDB does not
@@ -4088,8 +4085,8 @@ bool ibuf_delete_rec(const page_id_t page_id, btr_pcur_t* pcur,
 
 	root = ibuf_tree_root_get(mtr)->page.frame;
 
-	btr_cur_pessimistic_delete(&err, TRUE, btr_pcur_get_btr_cur(pcur), 0,
-				   false, mtr);
+	btr_cur_pessimistic_delete(&err, TRUE, btr_pcur_get_btr_cur(pcur),
+				   BTR_CREATE_FLAG, false, mtr);
 	ut_a(err == DB_SUCCESS);
 
 	ibuf_size_update(root);
diff --git a/storage/innobase/include/btr0cur.h b/storage/innobase/include/btr0cur.h
index f5f1c972957..6bcc71702d3 100644
--- a/storage/innobase/include/btr0cur.h
+++ b/storage/innobase/include/btr0cur.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2019, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -459,27 +459,18 @@ that the mtr has an x-latch on the page where the cursor is positioned,
 but no latch on the whole tree.
 @return TRUE if success, i.e., the page did not become too empty */
 ibool
-btr_cur_optimistic_delete_func(
+btr_cur_optimistic_delete(
 /*===========================*/
 	btr_cur_t*	cursor,	/*!< in: cursor on the record to delete;
 				cursor stays valid: if deletion succeeds,
 				on function exit it points to the successor
 				of the deleted record */
-# ifdef UNIV_DEBUG
 	ulint		flags,	/*!< in: BTR_CREATE_FLAG or 0 */
-# endif /* UNIV_DEBUG */
 	mtr_t*		mtr)	/*!< in: mtr; if this function returns
 				TRUE on a leaf page of a secondary
 				index, the mtr must be committed
 				before latching any further pages */
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
-# ifdef UNIV_DEBUG
-#  define btr_cur_optimistic_delete(cursor, flags, mtr)		\
-	btr_cur_optimistic_delete_func(cursor, flags, mtr)
-# else /* UNIV_DEBUG */
-#  define btr_cur_optimistic_delete(cursor, flags, mtr)		\
-	btr_cur_optimistic_delete_func(cursor, mtr)
-# endif /* UNIV_DEBUG */
 /*************************************************************//**
 Removes the record on which the tree cursor is positioned. Tries
 to compress the page if its fillfactor drops below a threshold
diff --git a/storage/innobase/include/data0type.h b/storage/innobase/include/data0type.h
index 40d9412d27f..ae688bf85fe 100644
--- a/storage/innobase/include/data0type.h
+++ b/storage/innobase/include/data0type.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -24,9 +24,7 @@ Data types
 Created 1/16/1996 Heikki Tuuri
 *******************************************************/
 
-#ifndef data0type_h
-#define data0type_h
-
+#pragma once
 #include "univ.i"
 
 /** Special length indicating a missing instantly added column */
@@ -196,9 +194,6 @@ constexpr uint8_t DATA_MBR_LEN= uint8_t(SPDIMS * 2 * sizeof(double));
 /** system-versioned user data column */
 #define DATA_VERSIONED (DATA_VERS_START|DATA_VERS_END)
 
-/** Check whether locking is disabled (never). */
-#define dict_table_is_locking_disabled(table) false
-
 /*-------------------------------------------*/
 
 /* This many bytes we need to store the type information affecting the
@@ -588,5 +583,3 @@ static const byte REC_INFO_METADATA_ALTER
 	= REC_INFO_METADATA_ADD | REC_INFO_DELETED_FLAG;
 
 #include "data0type.inl"
-
-#endif
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index a7efa5a6431..622a453c136 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -1184,6 +1184,9 @@ public:
 	/** @return whether this is the change buffer */
 	bool is_ibuf() const { return UNIV_UNLIKELY(type & DICT_IBUF); }
 
+	/** @return whether this index requires locking */
+	bool has_locking() const { return !is_ibuf(); }
+
 	/** @return whether the index includes virtual columns */
 	bool has_virtual() const { return type & DICT_VIRTUAL; }
 
diff --git a/storage/innobase/include/lock0lock.h b/storage/innobase/include/lock0lock.h
index 776d9ec9f99..a11bc60e7a0 100644
--- a/storage/innobase/include/lock0lock.h
+++ b/storage/innobase/include/lock0lock.h
@@ -374,18 +374,18 @@ lock_clust_rec_read_check_and_lock_alt(
 					LOCK_REC_NOT_GAP */
 	que_thr_t*		thr)	/*!< in: query thread */
 	MY_ATTRIBUTE((warn_unused_result));
-/*********************************************************************//**
-Locks the specified database table in the mode given. If the lock cannot
-be granted immediately, the query thread is put to wait.
-@return DB_SUCCESS, DB_LOCK_WAIT, or DB_DEADLOCK */
-dberr_t
-lock_table(
-/*=======*/
-	dict_table_t*	table,	/*!< in/out: database table
-				in dictionary cache */
-	lock_mode	mode,	/*!< in: lock mode */
-	que_thr_t*	thr)	/*!< in: query thread */
-	MY_ATTRIBUTE((warn_unused_result));
+
+/** Acquire a table lock.
+@param table   table to be locked
+@param fktable pointer to table, in case of a FOREIGN key check
+@param mode    lock mode
+@param thr     SQL execution thread
+@retval DB_SUCCESS    if the lock was acquired
+@retval DB_DEADLOCK   if a deadlock occurred, or fktable && *fktable != table
+@retval DB_LOCK_WAIT  if lock_wait() must be invoked */
+dberr_t lock_table(dict_table_t *table, dict_table_t *const*fktable,
+                   lock_mode mode, que_thr_t *thr)
+  MY_ATTRIBUTE((warn_unused_result));
 
 /** Create a table lock object for a resurrected transaction.
 @param table    table to be X-locked
@@ -426,6 +426,11 @@ lock_rec_unlock(
 and release possible other transactions waiting because of these locks. */
 void lock_release(trx_t* trx);
 
+/** Release the explicit locks of a committing transaction while
+dict_sys.latch is exclusively locked,
+and release possible other transactions waiting because of these locks. */
+void lock_release_on_drop(trx_t *trx);
+
 /** Release non-exclusive locks on XA PREPARE,
 and release possible other transactions waiting because of these locks. */
 void lock_release_on_prepare(trx_t *trx);
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 23bedfaea43..a45bbac0334 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -766,8 +766,12 @@ public:
 					flush the log in
 					trx_commit_complete_for_mysql() */
 	ulint		duplicates;	/*!< TRX_DUP_IGNORE | TRX_DUP_REPLACE */
-	bool		dict_operation;	/**< whether this modifies InnoDB
-					data dictionary */
+  /** whether this modifies InnoDB dictionary tables */
+  bool dict_operation;
+#ifdef UNIV_DEBUG
+  /** copy of dict_operation during commit() */
+  bool was_dict_operation;
+#endif
 	/** whether dict_sys.latch is held exclusively; protected by
 	dict_sys.latch */
 	bool dict_operation_lock_mode;
@@ -942,7 +946,8 @@ private:
   ATTRIBUTE_COLD void apply_log();
   /** Process tables that were modified by the committing transaction. */
   inline void commit_tables();
-  /** Mark a transaction committed in the main memory data structures. */
+  /** Mark a transaction committed in the main memory data structures.
+  @param mtr  mini-transaction (if there are any persistent modifications) */
   inline void commit_in_memory(const mtr_t *mtr);
   /** Write log for committing the transaction. */
   void commit_persist();
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 617a055b3be..2d27a674275 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -3460,59 +3460,53 @@ static dberr_t lock_table_wsrep(dict_table_t *table, lock_mode mode,
 }
 #endif
 
-/*********************************************************************//**
-Locks the specified database table in the mode given. If the lock cannot
-be granted immediately, the query thread is put to wait.
-@return DB_SUCCESS, DB_LOCK_WAIT, or DB_DEADLOCK */
-dberr_t
-lock_table(
-/*=======*/
-	dict_table_t*	table,	/*!< in/out: database table
-				in dictionary cache */
-	lock_mode	mode,	/*!< in: lock mode */
-	que_thr_t*	thr)	/*!< in: query thread */
-{
-	trx_t*		trx;
-
-	if (table->is_temporary()) {
-		return DB_SUCCESS;
-	}
-
-	trx = thr_get_trx(thr);
+/** Acquire a table lock.
+@param table   table to be locked
+@param fktable pointer to table, in case of a FOREIGN key check
+@param mode    lock mode
+@param thr     SQL execution thread
+@retval DB_SUCCESS    if the lock was acquired
+@retval DB_DEADLOCK   if a deadlock occurred, or fktable && *fktable != table
+@retval DB_LOCK_WAIT  if lock_wait() must be invoked */
+dberr_t lock_table(dict_table_t *table, dict_table_t *const*fktable,
+                   lock_mode mode, que_thr_t *thr)
+{
+  ut_ad(table);
+
+  if (!fktable && table->is_temporary())
+    return DB_SUCCESS;
 
-	/* Look for equal or stronger locks the same trx already
-	has on the table. No need to acquire LockMutexGuard here
-	because only this transaction can add/access table locks
-	to/from trx_t::table_locks. */
+  ut_ad(fktable || table->get_ref_count() || !table->can_be_evicted);
 
-	if (lock_table_has(trx, table, mode) || srv_read_only_mode) {
-		return(DB_SUCCESS);
-	}
-
-	/* Read only transactions can write to temp tables, we don't want
-	to promote them to RW transactions. Their updates cannot be visible
-	to other transactions. Therefore we can keep them out
-	of the read views. */
+  trx_t *trx= thr_get_trx(thr);
 
-	if ((mode == LOCK_IX || mode == LOCK_X)
-	    && !trx->read_only
-	    && trx->rsegs.m_redo.rseg == 0) {
+  /* Look for equal or stronger locks the same trx already has on the
+  table. No need to acquire LockMutexGuard here because only the
+  thread that is executing a transaction can access trx_t::table_locks. */
+  if (lock_table_has(trx, table, mode) || srv_read_only_mode)
+    return DB_SUCCESS;
 
-		trx_set_rw_mode(trx);
-	}
+  if ((mode == LOCK_IX || mode == LOCK_X) &&
+      !trx->read_only && !trx->rsegs.m_redo.rseg)
+    trx_set_rw_mode(trx);
 
 #ifdef WITH_WSREP
-	if (trx->is_wsrep()) {
-		return lock_table_wsrep(table, mode, thr, trx);
-	}
+  if (trx->is_wsrep())
+    return lock_table_wsrep(table, mode, thr, trx);
 #endif
-	lock_sys.rd_lock(SRW_LOCK_CALL);
-	table->lock_mutex_lock();
-	dberr_t err = lock_table_low(table, mode, thr, trx);
-	table->lock_mutex_unlock();
-	lock_sys.rd_unlock();
+  lock_sys.rd_lock(SRW_LOCK_CALL);
+  dberr_t err;
+  if (fktable != nullptr && *fktable != table)
+    err= DB_DEADLOCK;
+  else
+  {
+    table->lock_mutex_lock();
+    err= lock_table_low(table, mode, thr, trx);
+    table->lock_mutex_unlock();
+  }
+  lock_sys.rd_unlock();
 
-	return err;
+  return err;
 }
 
 /** Create a table lock object for a resurrected transaction.
@@ -3649,7 +3643,7 @@ dberr_t lock_table_for_trx(dict_table_t *table, trx_t *trx, lock_mode mode,
 run_again:
   thr->run_node= thr;
   thr->prev_node= thr->common.parent;
-  dberr_t err= lock_table(table, mode, thr);
+  dberr_t err= lock_table(table, nullptr, mode, thr);
 
   switch (err) {
   case DB_SUCCESS:
@@ -3820,7 +3814,7 @@ restart:
       ut_ad(!lock->index->table->is_temporary());
       ut_ad(lock->mode() != LOCK_X ||
             lock->index->table->id >= DICT_HDR_FIRST_ID ||
-            trx->dict_operation);
+            trx->dict_operation || trx->was_dict_operation);
       auto &lock_hash= lock_sys.hash_get(lock->type_mode);
       auto cell= lock_hash.cell_get(lock->un_member.rec_lock.page_id.fold());
       auto latch= lock_sys_t::hash_table::latch(cell);
@@ -3838,7 +3832,7 @@ restart:
       ut_ad(!table->is_temporary());
       ut_ad(table->id >= DICT_HDR_FIRST_ID ||
             (lock->mode() != LOCK_IX && lock->mode() != LOCK_X) ||
-            trx->dict_operation);
+            trx->dict_operation || trx->was_dict_operation);
       if (!table->lock_mutex_trylock())
         all_released= false;
       else
@@ -3895,7 +3889,7 @@ restart:
       ut_ad(!lock->index->table->is_temporary());
       ut_ad(lock->mode() != LOCK_X ||
             lock->index->table->id >= DICT_HDR_FIRST_ID ||
-            trx->dict_operation);
+            trx->dict_operation || trx->was_dict_operation);
       lock_rec_dequeue_from_page(lock, false);
     }
     else
@@ -3904,7 +3898,7 @@ restart:
       ut_ad(!table->is_temporary());
       ut_ad(table->id >= DICT_HDR_FIRST_ID ||
             (lock->mode() != LOCK_IX && lock->mode() != LOCK_X) ||
-            trx->dict_operation);
+            trx->dict_operation || trx->was_dict_operation);
       lock_table_dequeue(lock, false);
     }
 
@@ -3941,6 +3935,38 @@ released:
 #endif
 }
 
+/** Release the explicit locks of a committing transaction while
+dict_sys.latch is exclusively locked,
+and release possible other transactions waiting because of these locks. */
+void lock_release_on_drop(trx_t *trx)
+{
+  ut_ad(lock_sys.is_writer());
+  ut_ad(trx->mutex_is_owner());
+  ut_ad(trx->dict_operation);
+
+  while (lock_t *lock= UT_LIST_GET_LAST(trx->lock.trx_locks))
+  {
+    ut_ad(lock->trx == trx);
+    if (!lock->is_table())
+    {
+      ut_ad(!lock->index->table->is_temporary());
+      ut_ad(lock->mode() != LOCK_X ||
+            lock->index->table->id >= DICT_HDR_FIRST_ID ||
+            trx->dict_operation);
+      lock_rec_dequeue_from_page(lock, false);
+    }
+    else
+    {
+      ut_d(dict_table_t *table= lock->un_member.tab_lock.table);
+      ut_ad(!table->is_temporary());
+      ut_ad(table->id >= DICT_HDR_FIRST_ID ||
+            (lock->mode() != LOCK_IX && lock->mode() != LOCK_X) ||
+            trx->dict_operation);
+      lock_table_dequeue(lock, false);
+    }
+  }
+}
+
 /** Release non-exclusive locks on XA PREPARE,
 and wake up possible other transactions waiting because of these locks.
 @param trx   transaction in XA PREPARE state
diff --git a/storage/innobase/page/page0page.cc b/storage/innobase/page/page0page.cc
index 2f85ef94233..27aab0c475a 100644
--- a/storage/innobase/page/page0page.cc
+++ b/storage/innobase/page/page0page.cc
@@ -2,7 +2,7 @@
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -654,7 +654,7 @@ page_copy_rec_list_end(
 
 	/* Update the lock table and possible hash index */
 
-	if (dict_table_is_locking_disabled(index->table)) {
+	if (!index->has_locking()) {
 	} else if (rec_move && dict_index_is_spatial(index)) {
 		lock_rtr_move_rec_list(new_block, block, rec_move, num_moved);
 	} else {
@@ -821,7 +821,7 @@ zip_reorganize:
 
 	/* Update the lock table and possible hash index */
 
-	if (dict_table_is_locking_disabled(index->table)) {
+	if (!index->has_locking()) {
 	} else if (dict_index_is_spatial(index)) {
 		lock_rtr_move_rec_list(new_block, block, rec_move, num_moved);
 	} else {
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 66cbbd24b9a..d319ee1d6c1 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -1000,7 +1000,8 @@ row_ins_foreign_check_on_constraint(
 {
 	upd_node_t*	node;
 	upd_node_t*	cascade;
-	dict_table_t*	table		= foreign->foreign_table;
+	dict_table_t*const*const fktable = &foreign->foreign_table;
+	dict_table_t*	table = *fktable;
 	dict_index_t*	index;
 	dict_index_t*	clust_index;
 	dtuple_t*	ref;
@@ -1168,7 +1169,7 @@ row_ins_foreign_check_on_constraint(
 
 	/* Set an X-lock on the row to delete or update in the child table */
 
-	err = lock_table(table, LOCK_IX, thr);
+	err = lock_table(table, fktable, LOCK_IX, thr);
 
 	if (err == DB_SUCCESS) {
 		/* Here it suffices to use a LOCK_REC_NOT_GAP type lock;
@@ -1454,10 +1455,7 @@ row_ins_check_foreign_constraint(
 	dtuple_t*	entry,	/*!< in: index entry for index */
 	que_thr_t*	thr)	/*!< in: query thread */
 {
-	dberr_t		err;
 	upd_node_t*	upd_node;
-	dict_table_t*	check_table;
-	dict_index_t*	check_index;
 	ulint		n_fields_cmp;
 	btr_pcur_t	pcur;
 	int		cmp;
@@ -1479,12 +1477,10 @@ row_ins_check_foreign_constraint(
 	upd_node= NULL;
 #endif /* WITH_WSREP */
 
-	err = DB_SUCCESS;
-
-	if (trx->check_foreigns == FALSE) {
+	if (!trx->check_foreigns) {
 		/* The user has suppressed foreign key checks currently for
 		this session */
-		goto exit_func;
+		DBUG_RETURN(DB_SUCCESS);
 	}
 
 	/* If any of the foreign key fields in entry is SQL NULL, we
@@ -1493,12 +1489,12 @@ row_ins_check_foreign_constraint(
 	for (ulint i = 0; i < entry->n_fields; i++) {
 		dfield_t* field = dtuple_get_nth_field(entry, i);
 		if (i < foreign->n_fields && dfield_is_null(field)) {
-			goto exit_func;
+			DBUG_RETURN(DB_SUCCESS);
 		}
 		/* System Versioning: if row_end != Inf, we
 		suppress the foreign key check */
 		if (field->type.vers_sys_end() && field->vers_history_row()) {
-			goto exit_func;
+			DBUG_RETURN(DB_SUCCESS);
 		}
 	}
 
@@ -1523,7 +1519,7 @@ row_ins_check_foreign_constraint(
 			another, and the user has problems predicting in
 			which order they are performed. */
 
-			goto exit_func;
+			DBUG_RETURN(DB_SUCCESS);
 		}
 	}
 
@@ -1535,23 +1531,32 @@ row_ins_check_foreign_constraint(
 			dfield_t* row_end = dtuple_get_nth_field(
 				insert_node->row, table->vers_end);
 			if (row_end->vers_history_row()) {
-				goto exit_func;
+				DBUG_RETURN(DB_SUCCESS);
 			}
 		}
 	}
 
-	if (check_ref) {
-		check_table = foreign->referenced_table;
-		check_index = foreign->referenced_index;
-	} else {
-		check_table = foreign->foreign_table;
-		check_index = foreign->foreign_index;
+	dict_table_t *check_table;
+	dict_index_t *check_index;
+	dberr_t err = DB_SUCCESS;
+
+	{
+		dict_table_t*& fktable = check_ref
+			? foreign->referenced_table : foreign->foreign_table;
+		check_table = fktable;
+		if (check_table) {
+			err = lock_table(check_table, &fktable, LOCK_IS, thr);
+			if (err != DB_SUCCESS) {
+				goto do_possible_lock_wait;
+			}
+		}
+		check_table = fktable;
 	}
 
-	if (check_table == NULL
-	    || !check_table->is_readable()
-	    || check_index == NULL) {
+	check_index = check_ref
+		? foreign->referenced_index : foreign->foreign_index;
 
+	if (!check_table || !check_table->is_readable() || !check_index) {
 		FILE*	ef = dict_foreign_err_file;
 		std::string fk_str;
 
@@ -1600,18 +1605,6 @@ row_ins_check_foreign_constraint(
 		goto exit_func;
 	}
 
-	if (check_table != table) {
-		/* We already have a LOCK_IX on table, but not necessarily
-		on check_table */
-
-		err = lock_table(check_table, LOCK_IS, thr);
-
-		if (err != DB_SUCCESS) {
-
-			goto do_possible_lock_wait;
-		}
-	}
-
 	mtr_start(&mtr);
 
 	/* Store old value on n_fields_cmp */
@@ -1824,10 +1817,7 @@ do_possible_lock_wait:
 
 		thr->lock_state = QUE_THR_LOCK_NOLOCK;
 
-		if (err != DB_SUCCESS) {
-		} else if (check_table->name.is_temporary()) {
-			err = DB_LOCK_WAIT_TIMEOUT;
-		} else {
+		if (err == DB_SUCCESS) {
 			err = DB_LOCK_WAIT;
 		}
 	}
@@ -2632,7 +2622,7 @@ commit_exit:
 		DEBUG_SYNC_C("empty_root_page_insert");
 
 		if (!index->table->is_temporary()) {
-			err = lock_table(index->table, LOCK_X, thr);
+			err = lock_table(index->table, NULL, LOCK_X, thr);
 
 			if (err != DB_SUCCESS) {
 				trx->error_state = err;
@@ -3694,7 +3684,7 @@ row_ins_step(
 			goto same_trx;
 		}
 
-		err = lock_table(node->table, LOCK_IX, thr);
+		err = lock_table(node->table, NULL, LOCK_IX, thr);
 
 		DBUG_EXECUTE_IF("ib_row_ins_ix_lock_wait",
 				err = DB_LOCK_WAIT;);
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 2d5ef06fbb9..143365f02a7 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -1124,7 +1124,7 @@ row_lock_table_autoinc_for_mysql(
 
 		trx_start_if_not_started_xa(trx, true);
 
-		err = lock_table(prebuilt->table, LOCK_AUTO_INC, thr);
+		err = lock_table(prebuilt->table, NULL, LOCK_AUTO_INC, thr);
 
 		trx->error_state = err;
 	} while (err != DB_SUCCESS
@@ -1166,7 +1166,7 @@ row_lock_table(row_prebuilt_t* prebuilt)
 
 		trx_start_if_not_started_xa(trx, false);
 
-		err = lock_table(prebuilt->table, static_cast<lock_mode>(
+		err = lock_table(prebuilt->table, NULL, static_cast<lock_mode>(
 					 prebuilt->select_lock_type), thr);
 		trx->error_state = err;
 	} while (err != DB_SUCCESS
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index e01637d22cc..265646ddb93 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -2,7 +2,7 @@
 
 Copyright (c) 1997, 2017, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2008, Google Inc.
-Copyright (c) 2015, 2021, MariaDB Corporation.
+Copyright (c) 2015, 2022, MariaDB Corporation.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -2357,7 +2357,8 @@ row_sel_step(
 					que_node_get_next(table_node))) {
 
 				dberr_t	err = lock_table(
-					table_node->table, i_lock_mode, thr);
+					table_node->table, nullptr,
+					i_lock_mode, thr);
 
 				if (err != DB_SUCCESS) {
 					trx_t*	trx;
@@ -4672,7 +4673,7 @@ aborted:
 			trx->read_view.open(trx);
 		} else {
 wait_table_again:
-			err = lock_table(prebuilt->table,
+			err = lock_table(prebuilt->table, nullptr,
 					 prebuilt->select_lock_type == LOCK_S
 					 ? LOCK_IS : LOCK_IX, thr);
 
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 5b89582e23b..b51133aaa0d 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -241,12 +241,9 @@ row_upd_check_references_constraints(
 			|| row_upd_changes_first_fields_binary(
 				entry, index, node->update,
 				foreign->n_fields))) {
-			dict_table_t*	foreign_table = foreign->foreign_table;
-
-			dict_table_t*	ref_table = NULL;
-
-			if (foreign_table == NULL) {
+			dict_table_t*	ref_table = nullptr;
 
+			if (!foreign->foreign_table) {
 				ref_table = dict_table_open_on_name(
 					foreign->foreign_table_name_lookup,
 					false, DICT_ERR_IGNORE_NONE);
@@ -293,7 +290,6 @@ wsrep_row_upd_check_foreign_constraints(
 	dtuple_t*	entry;
 	const rec_t*	rec;
 	dberr_t		err;
-	ibool		opened     	= FALSE;
 
 	if (table->foreign_set.empty()) {
 		return(DB_SUCCESS);
@@ -328,27 +324,21 @@ wsrep_row_upd_check_foreign_constraints(
 				entry, index, node->update,
 				foreign->n_fields))) {
 
-			if (foreign->referenced_table == NULL) {
+			dict_table_t *opened = nullptr;
+
+			if (!foreign->referenced_table) {
 				foreign->referenced_table =
 					dict_table_open_on_name(
 					  foreign->referenced_table_name_lookup,
 					  false, DICT_ERR_IGNORE_NONE);
-				opened = (foreign->referenced_table) ? TRUE : FALSE;
+				opened = foreign->referenced_table;
 			}
 
-			/* NOTE that if the thread ends up waiting for a lock
-			we will release dict_sys.latch temporarily!
-			But the counter on the table protects 'foreign' from
-			being dropped while the check is running. */
-
 			err = row_ins_check_foreign_constraint(
 				TRUE, foreign, table, entry, thr);
 
-			if (foreign->referenced_table) {
-				if (opened) {
-					dict_table_close(foreign->referenced_table);
-					opened = FALSE;
-				}
+			if (opened) {
+				dict_table_close(opened);
 			}
 
 			if (err != DB_SUCCESS) {
@@ -2893,7 +2883,7 @@ row_upd_step(
 			/* It may be that the current session has not yet
 			started its transaction, or it has been committed: */
 
-			err = lock_table(node->table, LOCK_IX, thr);
+			err = lock_table(node->table, nullptr, LOCK_IX, thr);
 
 			if (err != DB_SUCCESS) {
 
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index 64883fef38d..ba0809dc444 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -497,6 +497,13 @@ inline void trx_t::release_locks()
   }
 
   lock.table_locks.clear();
+  id= 0;
+  while (dict_table_t *table= UT_LIST_GET_FIRST(lock.evicted_tables))
+  {
+    UT_LIST_REMOVE(lock.evicted_tables, table);
+    dict_mem_table_free(table);
+  }
+  DEBUG_SYNC_C("after_trx_committed_in_memory");
 }
 
 /** At shutdown, frees a transaction object. */
@@ -526,7 +533,6 @@ TRANSACTIONAL_TARGET void trx_free_at_shutdown(trx_t *trx)
 	DBUG_LOG("trx", "Free prepared: " << trx);
 	trx->state = TRX_STATE_NOT_STARTED;
 	ut_ad(!UT_LIST_GET_LEN(trx->lock.trx_locks));
-	trx->id = 0;
 	trx->free();
 }
 
@@ -1229,9 +1235,10 @@ void trx_t::evict_table(table_id_t table_id, bool reset_only)
 	}
 }
 
-/** Mark a transaction committed in the main memory data structures. */
 TRANSACTIONAL_INLINE inline void trx_t::commit_in_memory(const mtr_t *mtr)
 {
+  /* We already detached from rseg in trx_write_serialisation_history() */
+  ut_ad(!rsegs.m_redo.undo);
   must_flush_log_later= false;
   read_view.close();
 
@@ -1242,12 +1249,14 @@ TRANSACTIONAL_INLINE inline void trx_t::commit_in_memory(const mtr_t *mtr)
     ut_ad(!will_lock);
     ut_a(!is_recovered);
     ut_ad(!rsegs.m_redo.rseg);
+    ut_ad(!rsegs.m_redo.undo);
     ut_ad(mysql_thd);
     ut_ad(state == TRX_STATE_ACTIVE);
 
     /* Note: We do not have to hold any lock_sys latch here, because
     this is a non-locking transaction. */
     ut_a(UT_LIST_GET_LEN(lock.trx_locks) == 0);
+    ut_ad(UT_LIST_GET_LEN(lock.evicted_tables) == 0);
 
     /* This state change is not protected by any mutex, therefore
     there is an inherent race here around state transition during
@@ -1293,21 +1302,10 @@ TRANSACTIONAL_INLINE inline void trx_t::commit_in_memory(const mtr_t *mtr)
       is_recovered= false;
     }
 
-    release_locks();
-    id= 0;
-    DEBUG_SYNC_C("after_trx_committed_in_memory");
-
-    while (dict_table_t *table= UT_LIST_GET_FIRST(lock.evicted_tables))
-    {
-      UT_LIST_REMOVE(lock.evicted_tables, table);
-      dict_mem_table_free(table);
-    }
+    if (UNIV_LIKELY(!dict_operation))
+      release_locks();
   }
 
-  /* We already detached from rseg in trx_write_serialisation_history() */
-  ut_ad(!rsegs.m_redo.undo);
-  ut_ad(UT_LIST_GET_LEN(lock.evicted_tables) == 0);
-
   if (trx_rseg_t *rseg= rsegs.m_redo.rseg)
     /* This is safe due to us having detached the persistent undo log. */
     rseg->release();
@@ -1381,10 +1379,10 @@ TRANSACTIONAL_INLINE inline void trx_t::commit_in_memory(const mtr_t *mtr)
 
 void trx_t::commit_cleanup()
 {
-  mutex.wr_lock();
-  dict_operation= false;
+  ut_ad(!dict_operation);
+  ut_ad(!was_dict_operation);
 
-  DBUG_LOG("trx", "Commit in memory: " << this);
+  mutex.wr_lock();
   state= TRX_STATE_NOT_STARTED;
   mod_tables.clear();
 
@@ -1470,7 +1468,11 @@ void trx_t::commit_persist()
 
 void trx_t::commit()
 {
+  ut_ad(!was_dict_operation);
+  ut_d(was_dict_operation= dict_operation);
+  dict_operation= false;
   commit_persist();
+  ut_d(was_dict_operation= false);
   ut_d(for (const auto &p : mod_tables) ut_ad(!p.second.is_dropped()));
   commit_cleanup();
 }