MDEV-27028 [ERROR] [FATAL] InnoDB: SYS_VIRTUAL.TABLE_ID mismatch

If the server is killed during any DDL operation that is about to
delete an .ibd file, recovery could crash when attempting to load
the table definition of the being-dropped table. By design of
commit 1bd681c8 (MDEV-25506 part 3),
a table whose name starts with #sql-ib in the data dictionary may
belong to an uncommitted transaction. So, we must ignore any missing
SYS_COLUMNS, SYS_FIELDS, and SYS_VIRTUAL records for such tables.

The "ID mismatch" error messages were misleading; they really mean
"record not found".

storage/innobase/dict/dict0load.cc

@@ -951,7 +951,8 @@ void dict_check_tablespaces_and_store_max_id()
 }
 
 /** Error message for a delete-marked record in dict_load_column_low() */
-static const char* dict_load_column_del = "delete-marked record in SYS_COLUMN";
+static const char *dict_load_column_del= "delete-marked record in SYS_COLUMNS";
+static const char *dict_load_column_none= "SYS_COLUMNS record not found";
 
 /** Load a table column definition from a SYS_COLUMNS record to dict_table_t.
 @return	error message
@@ -1003,7 +1004,7 @@ dict_load_column_low(
 	if (table_id) {
 		*table_id = mach_read_from_8(field);
 	} else if (table->id != mach_read_from_8(field)) {
-		return("SYS_COLUMNS.TABLE_ID mismatch");
+		return dict_load_column_none;
 	}
 
 	field = rec_get_nth_field_old(
@@ -1126,7 +1127,8 @@ dict_load_column_low(
 }
 
 /** Error message for a delete-marked record in dict_load_virtual_low() */
-static const char* dict_load_virtual_del = "delete-marked record in SYS_VIRTUAL";
+static const char *dict_load_virtual_del= "delete-marked record in SYS_VIRTUAL";
+static const char *dict_load_virtual_none= "SYS_VIRTUAL record not found";
 
 /** Load a virtual column "mapping" (to base columns) information
 from a SYS_VIRTUAL record
@@ -1170,7 +1172,7 @@ dict_load_virtual_low(
 	if (table_id != NULL) {
 		*table_id = mach_read_from_8(field);
 	} else if (table->id != mach_read_from_8(field)) {
-		return("SYS_VIRTUAL.TABLE_ID mismatch");
+		return dict_load_virtual_none;
 	}
 
 	field = rec_get_nth_field_old(
@@ -1269,16 +1271,21 @@ dict_load_columns(
 
 		rec = btr_pcur_get_rec(&pcur);
 
-		ut_a(btr_pcur_is_on_user_rec(&pcur));
-
-		err_msg = dict_load_column_low(table, heap, NULL, NULL,
-					       &name, rec, &nth_v_col);
+		err_msg = btr_pcur_is_on_user_rec(&pcur)
+			? dict_load_column_low(table, heap, NULL, NULL,
+					       &name, rec, &nth_v_col)
+			: dict_load_column_none;
 
-		if (err_msg == dict_load_column_del) {
+		if (!err_msg) {
+		} else if (err_msg == dict_load_column_del) {
 			n_skipped++;
 			goto next_rec;
-		} else if (err_msg) {
-			ib::fatal() << err_msg;
+		} else if (err_msg == dict_load_column_none
+			&& strstr(table->name.m_name,
+				  "/" TEMP_FILE_PREFIX_INNODB)) {
+			break;
+		} else {
+			ib::fatal() << err_msg << " for table " << table->name;
 		}
 
 		/* Note: Currently we have one DOC_ID column that is
@@ -1341,7 +1348,6 @@ dict_load_virtual_one_col(
 	btr_pcur_t	pcur;
 	dtuple_t*	tuple;
 	dfield_t*	dfield;
-	const rec_t*	rec;
 	byte*		buf;
 	ulint		i = 0;
 	mtr_t		mtr;
@@ -1386,28 +1392,26 @@ dict_load_virtual_one_col(
 				  BTR_SEARCH_LEAF, &pcur, &mtr);
 
 	for (i = 0; i < unsigned{v_col->num_base} + skipped; i++) {
-		const char*	err_msg;
 		ulint		pos;
-
-		ut_ad(btr_pcur_is_on_user_rec(&pcur));
-
-		rec = btr_pcur_get_rec(&pcur);
-
-		ut_a(btr_pcur_is_on_user_rec(&pcur));
-
-		err_msg = dict_load_virtual_low(table,
+		const char*	err_msg
+			= btr_pcur_is_on_user_rec(&pcur)
+			? dict_load_virtual_low(table,
 						&v_col->base_col[i - skipped],
 						NULL,
-					        &pos, NULL, rec);
+					        &pos, NULL,
+						btr_pcur_get_rec(&pcur))
+			: dict_load_virtual_none;
 
-		if (err_msg) {
-			if (err_msg != dict_load_virtual_del) {
-				ib::fatal() << err_msg;
-			} else {
-				skipped++;
-			}
-		} else {
+		if (!err_msg) {
 			ut_ad(pos == vcol_pos);
+		} else if (err_msg == dict_load_virtual_del) {
+			skipped++;
+		} else if (err_msg == dict_load_virtual_none
+			   && strstr(table->name.m_name,
+				     "/" TEMP_FILE_PREFIX_INNODB)) {
+			break;
+		} else {
+			ib::fatal() << err_msg << " for table " << table->name;
 		}
 
 		btr_pcur_move_to_next_user_rec(&pcur, &mtr);
@@ -1435,7 +1439,9 @@ dict_load_virtual(
 }
 
 /** Error message for a delete-marked record in dict_load_field_low() */
-static const char* dict_load_field_del = "delete-marked record in SYS_FIELDS";
+static const char *dict_load_field_del= "delete-marked record in SYS_FIELDS";
+
+static const char *dict_load_field_none= "SYS_FIELDS record not found";
 
 /** Load an index field definition from a SYS_FIELDS record to dict_index_t.
 @return	error message
@@ -1490,7 +1496,7 @@ dict_load_field_low(
 	} else {
 		first_field = (index->n_def == 0);
 		if (memcmp(field, index_id, 8)) {
-			return("SYS_FIELDS.INDEX_ID mismatch");
+			return dict_load_field_none;
 		}
 	}
 
@@ -1571,7 +1577,6 @@ dict_load_fields(
 	btr_pcur_t	pcur;
 	dtuple_t*	tuple;
 	dfield_t*	dfield;
-	const rec_t*	rec;
 	byte*		buf;
 	ulint		i;
 	mtr_t		mtr;
@@ -1598,27 +1603,29 @@ dict_load_fields(
 	btr_pcur_open_on_user_rec(sys_index, tuple, PAGE_CUR_GE,
 				  BTR_SEARCH_LEAF, &pcur, &mtr);
 	for (i = 0; i < index->n_fields; i++) {
-		const char* err_msg;
+		const char *err_msg = btr_pcur_is_on_user_rec(&pcur)
+			? dict_load_field_low(buf, index, NULL, NULL, NULL,
+					      heap, btr_pcur_get_rec(&pcur))
+			: dict_load_field_none;
 
-		rec = btr_pcur_get_rec(&pcur);
-
-		ut_a(btr_pcur_is_on_user_rec(&pcur));
-
-		err_msg = dict_load_field_low(buf, index, NULL, NULL, NULL,
-					      heap, rec);
-
-		if (err_msg == dict_load_field_del) {
+		if (!err_msg) {
+		} else if (err_msg == dict_load_field_del) {
 			/* There could be delete marked records in
 			SYS_FIELDS because SYS_FIELDS.INDEX_ID can be
 			updated by ALTER TABLE ADD INDEX. */
-
-			goto next_rec;
-		} else if (err_msg) {
-			ib::error() << err_msg;
+		} else {
+			if (err_msg != dict_load_field_none
+			    || strstr(index->table->name.m_name,
+				      "/" TEMP_FILE_PREFIX_INNODB)) {
+				ib::error() << err_msg << " for index "
+					    << index->name
+					    << " of table "
+					    << index->table->name;
+			}
 			error = DB_CORRUPTION;
 			goto func_exit;
 		}
-next_rec:
+
 		btr_pcur_move_to_next_user_rec(&pcur, &mtr);
 	}
 
@@ -1630,11 +1637,11 @@ dict_load_fields(
 }
 
 /** Error message for a delete-marked record in dict_load_index_low() */
-static const char* dict_load_index_del = "delete-marked record in SYS_INDEXES";
+static const char *dict_load_index_del= "delete-marked record in SYS_INDEXES";
 /** Error message for table->id mismatch in dict_load_index_low() */
-static const char* dict_load_index_id_err = "SYS_INDEXES.TABLE_ID mismatch";
+static const char *dict_load_index_none= "SYS_INDEXES record not found";
 /** Error message for SYS_TABLES flags mismatch in dict_load_table_low() */
-static const char* dict_load_table_flags = "incorrect flags in SYS_TABLES";
+static const char *dict_load_table_flags= "incorrect flags in SYS_TABLES";
 
 /** Load an index definition from a SYS_INDEXES record to dict_index_t.
 If allocate=TRUE, we will create a dict_index_t structure and fill it
@@ -1707,7 +1714,7 @@ dict_load_index_low(
 	} else if (memcmp(field, table_id, 8)) {
 		/* Caller supplied table_id, verify it is the same
 		id as on the index record */
-		return(dict_load_index_id_err);
+		return dict_load_index_none;
 	}
 
 	field = rec_get_nth_field_old(
@@ -1860,7 +1867,7 @@ dict_load_indexes(
 		err_msg = dict_load_index_low(buf, heap, rec, TRUE, &index);
 		ut_ad(!index == !!err_msg);
 
-		if (err_msg == dict_load_index_id_err) {
+		if (err_msg == dict_load_index_none) {
 			/* We have ran out of index definitions for
 			the table. */
 			break;
@@ -1964,8 +1971,8 @@ dict_load_indexes(
 			of the database server */
 			dict_mem_index_free(index);
 		} else {
-			dict_load_fields(index, heap);
 			index->table = table;
+			dict_load_fields(index, heap);
 
 			/* The data dictionary tables should never contain
 			invalid index definitions.  If we ignored this error