MDEV-25522: Purge of aborted ADD INDEX leaves orphan locks behind

lock_discard_for_index(): New function, to discard locks for an
index whose index tree has been purged. By definition, such indexes
must be ones for which the MDL upgrade failed in inplace ALTER TABLE
and the ADD INDEX operation was never committed.
Note: Because we do not support online ADD SPATIAL INDEX, we only
have to traverse the lock_sys.rec_hash for B-trees and not the
hash tables for R-trees.

row_purge_remove_clust_if_poss_low(): Invoke lock_discard_for_index()
if necessary before dropping a B-tree for a SYS_INDEXES record.

storage/innobase/include/lock0lock.h

@@ -61,6 +61,10 @@ ulint
 lock_get_min_heap_no(
 /*=================*/
 	const buf_block_t*	block);	/*!< in: buffer block */
+
+/** Discard locks for an index */
+void lock_discard_for_index(const dict_index_t &index);
+
 /*************************************************************//**
 Updates the lock table when we have reorganized a page. NOTE: we copy
 also the locks set on the infimum of the page; the infimum may carry

storage/innobase/lock/lock0lock.cc

@@ -2019,6 +2019,31 @@ lock_rec_free_all_from_discard_page(page_id_t id, const hash_cell_t &cell,
   }
 }
 
+/** Discard locks for an index */
+void lock_discard_for_index(const dict_index_t &index)
+{
+  ut_ad(!index.is_committed());
+  lock_sys.wr_lock(SRW_LOCK_CALL);
+  const ulint n= lock_sys.rec_hash.pad(lock_sys.rec_hash.n_cells);
+  for (ulint i= 0; i < n; i++)
+  {
+    for (lock_t *lock= static_cast<lock_t*>(lock_sys.rec_hash.array[i].node);
+         lock; )
+    {
+      ut_ad(!lock->is_table());
+      if (lock->index == &index)
+      {
+        ut_ad(!lock->is_waiting());
+        lock_rec_discard(lock_sys.rec_hash, lock);
+        lock= static_cast<lock_t*>(lock_sys.rec_hash.array[i].node);
+      }
+      else
+        lock= lock->hash;
+    }
+  }
+  lock_sys.wr_unlock();
+}
+
 /*============= RECORD LOCK MOVING AND INHERITING ===================*/
 
 /*************************************************************//**

storage/innobase/row/row0purge.cc

@@ -105,9 +105,29 @@ row_purge_remove_clust_if_poss_low(
 	ulint		mode)	/*!< in: BTR_MODIFY_LEAF or BTR_MODIFY_TREE */
 {
 	dict_index_t* index = dict_table_get_first_index(node->table);
-
+	table_id_t table_id = 0;
+	index_id_t index_id = 0;
+retry:
+	if (table_id) {
+		MDL_ticket* mdl_ticket = nullptr;
+		if (dict_table_t *table = dict_table_open_on_id(
+			    table_id, false,
+			    DICT_TABLE_OP_OPEN_ONLY_IF_CACHED,
+			    node->purge_thd, &mdl_ticket)) {
+			if (table->n_rec_locks) {
+				for (dict_index_t* ind = UT_LIST_GET_FIRST(
+					     table->indexes); ind;
+				     ind = UT_LIST_GET_NEXT(indexes, ind)) {
+					if (ind->id == index_id) {
+						lock_discard_for_index(*ind);
+					}
+				}
+			}
+			dict_table_close(table, false, false,
+					 node->purge_thd, mdl_ticket);
+		}
+	}
 	log_free_check();
-
 	mtr_t mtr;
 	mtr.start();
 	index->set_modified(mtr);
@@ -122,6 +142,17 @@ row_purge_remove_clust_if_poss_low(
 		/* If this is a record of the SYS_INDEXES table, then
 		we have to free the file segments of the index tree
 		associated with the index */
+		if (!table_id) {
+			const rec_t* rec = btr_pcur_get_rec(&node->pcur);
+
+			table_id = mach_read_from_8(rec);
+			index_id = mach_read_from_8(rec + 8);
+			if (table_id) {
+				mtr.commit();
+				goto retry;
+			}
+			ut_ad("corrupted SYS_INDEXES record" == 0);
+		}
 		dict_drop_index_tree(&node->pcur, nullptr, &mtr);
 		mtr.commit();
 		mtr.start();