MDEV-24589 DROP TABLE is not crash-safe

row_upd_clust_step(): Remove the "trigger" on DELETE SYS_INDEXES
that would invoke dict_drop_index_tree(). Let us do it on purge.

row_purge_remove_clust_if_poss_low(): Invoke
dict_drop_index_tree() when purging a delete-marked SYS_INDEXES record.

4 files changed, 20 insertions, 34 deletions

diff --git a/storage/innobase/dict/dict0crea.cc b/storage/innobase/dict/dict0crea.cc
index 55e3191c228..b36f3c2d6ee 100644
--- a/storage/innobase/dict/dict0crea.cc
+++ b/storage/innobase/dict/dict0crea.cc
@@ -910,7 +910,7 @@ void dict_drop_index_tree(btr_pcur_t* pcur, trx_t* trx, mtr_t* mtr)
 	byte*	ptr;
 	ulint	len;
 
-	ut_ad(mutex_own(&dict_sys.mutex));
+	ut_ad(!trx || mutex_own(&dict_sys.mutex));
 	ut_a(!dict_table_is_comp(dict_sys.sys_indexes));
 
 	ptr = rec_get_nth_field_old(rec, DICT_FLD__SYS_INDEXES__PAGE_NO, &len);
@@ -936,7 +936,7 @@ void dict_drop_index_tree(btr_pcur_t* pcur, trx_t* trx, mtr_t* mtr)
 
 	const uint32_t space_id = mach_read_from_4(ptr);
 	ut_ad(space_id < SRV_TMP_SPACE_ID);
-	if (space_id != TRX_SYS_SPACE
+	if (space_id != TRX_SYS_SPACE && trx
 	    && trx_get_dict_operation(trx) == TRX_DICT_OP_TABLE) {
 		/* We are about to delete the entire .ibd file;
 		do not bother to free pages inside it. */
diff --git a/storage/innobase/include/dict0crea.h b/storage/innobase/include/dict0crea.h
index 13706d6bfad..3f18abed4ef 100644
--- a/storage/innobase/include/dict0crea.h
+++ b/storage/innobase/include/dict0crea.h
@@ -101,7 +101,7 @@ dict_create_index_tree(
 @param[in,out]	trx	dictionary transaction
 @param[in,out]	mtr	mini-transaction */
 void dict_drop_index_tree(btr_pcur_t* pcur, trx_t* trx, mtr_t* mtr)
-	MY_ATTRIBUTE((nonnull));
+	MY_ATTRIBUTE((nonnull(1,3)));
 
 /***************************************************************//**
 Creates an index tree for the index if it is not a member of a cluster.
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index 5579e53b6c1..cad70a2c6eb 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -27,6 +27,7 @@ Created 3/14/1997 Heikki Tuuri
 #include "row0purge.h"
 #include "fsp0fsp.h"
 #include "mach0data.h"
+#include "dict0crea.h"
 #include "dict0stats.h"
 #include "trx0rseg.h"
 #include "trx0trx.h"
@@ -116,6 +117,21 @@ row_purge_remove_clust_if_poss_low(
 		return true;
 	}
 
+	if (node->table->id == DICT_INDEXES_ID) {
+		/* 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 */
+		dict_drop_index_tree(&node->pcur, nullptr, &mtr);
+		mtr.commit();
+		mtr.start();
+		index->set_modified(mtr);
+
+		if (!row_purge_reposition_pcur(mode, node, &mtr)) {
+			mtr.commit();
+			return true;
+		}
+	}
+
 	rec_t* rec = btr_pcur_get_rec(&node->pcur);
 	rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
 	rec_offs_init(offsets_);
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 84f5bcd4a4e..efc62365193 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -2764,7 +2764,6 @@ row_upd_clust_step(
 {
 	dict_index_t*	index;
 	btr_pcur_t*	pcur;
-	ibool		success;
 	dberr_t		err;
 	mtr_t		mtr;
 	rec_t*		rec;
@@ -2826,9 +2825,7 @@ row_upd_clust_step(
 		mode = BTR_MODIFY_LEAF;
 	}
 
-	success = btr_pcur_restore_position(mode, pcur, &mtr);
-
-	if (!success) {
+	if (!btr_pcur_restore_position(mode, pcur, &mtr)) {
 		err = DB_RECORD_NOT_FOUND;
 
 		mtr_commit(&mtr);
@@ -2836,33 +2833,6 @@ row_upd_clust_step(
 		return(err);
 	}
 
-	/* If this is a row in SYS_INDEXES table of the data dictionary,
-	then we have to free the file segments of the index tree associated
-	with the index */
-
-	if (node->is_delete == PLAIN_DELETE
-	    && node->table->id == DICT_INDEXES_ID) {
-
-		ut_ad(!dict_index_is_online_ddl(index));
-
-		dict_drop_index_tree(pcur, trx, &mtr);
-
-		mtr.commit();
-
-		mtr.start();
-		index->set_modified(mtr);
-
-		success = btr_pcur_restore_position(BTR_MODIFY_LEAF, pcur,
-						    &mtr);
-		if (!success) {
-			err = DB_ERROR;
-
-			mtr.commit();
-
-			return(err);
-		}
-	}
-
 	rec = btr_pcur_get_rec(pcur);
 	offsets = rec_get_offsets(rec, index, offsets_, true,
 				  ULINT_UNDEFINED, &heap);