MDEV-17793 Crash in purge after instant DROP and emptying the table

There was a race condition between ALTER TABLE and purge.

If a table turns out to be logically empty when instant ALTER TABLE
is executing, we will convert the table to the canonical format,
to avoid overhead during subsequent accesses, and to allow the
data file to be imported into older versions of MariaDB.

It could happen that at the time the table is logically empty,
there still exists an undo log record for updating the hidden
metadata record for an earlier instant ALTER TABLE operation.
If the table was converted to the canonical format before
purge processes this undo log record, the undo log record
could be referring to index fields that no longer exist,
causing a crash.

To prevent the race condition, we must delete the old undo log records.
We do this lazily by assigning a new table ID, so that the table lookup
for the old undo log records will fail.

dict_table_t::reassign_id(): Reassign the table_id to
effectively lazily delete old undo log records.

innobase_instant_try(): Invoke index->table->reassign_id() before
index->clear_instant_alter().

mysql-test/suite/innodb/r/instant_alter_purge.result

+SET @saved_frequency = @@GLOBAL.innodb_purge_rseg_truncate_frequency;
+SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
+#
+# MDEV-17793 Crash in purge after instant DROP and emptying the table
+#
+connect  prevent_purge,localhost,root;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+CREATE TABLE t1 (f1 INT, f2 INT) ENGINE=InnoDB;
+INSERT INTO t1 () VALUES ();
+ALTER TABLE t1 DROP f2, ADD COLUMN f2 INT;
+ALTER TABLE t1 DROP f1;
+DELETE FROM t1;
+connection prevent_purge;
+COMMIT;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+ALTER TABLE t1 ADD COLUMN extra TINYINT UNSIGNED NOT NULL DEFAULT 42;
+InnoDB		1 transactions not purged
+ALTER TABLE t1 DROP extra;
+disconnect prevent_purge;
+InnoDB		0 transactions not purged
+DROP TABLE t1;
+SET GLOBAL innodb_purge_rseg_truncate_frequency = @saved_frequency;

mysql-test/suite/innodb/t/instant_alter_purge.test

+--source include/have_innodb.inc
+
+SET @saved_frequency = @@GLOBAL.innodb_purge_rseg_truncate_frequency;
+SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
+
+--echo #
+--echo # MDEV-17793 Crash in purge after instant DROP and emptying the table
+--echo #
+
+connect (prevent_purge,localhost,root);
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+
+connection default;
+CREATE TABLE t1 (f1 INT, f2 INT) ENGINE=InnoDB;
+INSERT INTO t1 () VALUES ();
+ALTER TABLE t1 DROP f2, ADD COLUMN f2 INT;
+ALTER TABLE t1 DROP f1;
+DELETE FROM t1;
+
+connection prevent_purge;
+COMMIT;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+
+ALTER TABLE t1 ADD COLUMN extra TINYINT UNSIGNED NOT NULL DEFAULT 42;
+let $wait_all_purged= 1;
+--source include/wait_all_purged.inc
+ALTER TABLE t1 DROP extra;
+disconnect prevent_purge;
+let $wait_all_purged= 0;
+--source include/wait_all_purged.inc
+DROP TABLE t1;
+SET GLOBAL innodb_purge_rseg_truncate_frequency = @saved_frequency;

storage/innobase/handler/handler0alter.cc

@@ -5217,6 +5217,51 @@ dict_index_t::instant_metadata(const dtuple_t& row, mem_heap_t* heap) const
 	return entry;
 }
 
+/** Assign a new id to invalidate old undo log records, so
+that purge will be unable to refer to fields that used to be
+instantly added to the end of the index. This is only to be
+used during ALTER TABLE when the table is empty, before
+invoking dict_index_t::clear_instant_alter().
+@param[in,out] trx	dictionary transaction
+@return	error code */
+inline dberr_t dict_table_t::reassign_id(trx_t* trx)
+{
+	DBUG_ASSERT(instant);
+	ut_ad(magic_n == DICT_TABLE_MAGIC_N);
+
+	table_id_t new_id;
+	dict_hdr_get_new_id(&new_id, NULL, NULL, NULL, false);
+	pars_info_t*	pinfo = pars_info_create();
+
+	pars_info_add_ull_literal(pinfo, "old", id);
+	pars_info_add_ull_literal(pinfo, "new", new_id);
+
+	ut_ad(mutex_own(&dict_sys->mutex));
+	ut_ad(rw_lock_own(dict_operation_lock, RW_LOCK_X));
+	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
+
+	dberr_t err = que_eval_sql(
+		pinfo,
+		"PROCEDURE RENUMBER_TABLE_ID_PROC () IS\n"
+		"BEGIN\n"
+		"UPDATE SYS_TABLES SET ID=:new WHERE ID=:old;\n"
+		"UPDATE SYS_COLUMNS SET TABLE_ID=:new WHERE TABLE_ID=:old;\n"
+		"UPDATE SYS_INDEXES SET TABLE_ID=:new WHERE TABLE_ID=:old;\n"
+		"END;\n"
+		, FALSE, trx);
+	if (err == DB_SUCCESS) {
+		auto fold = ut_fold_ull(id);
+		HASH_DELETE(dict_table_t, id_hash, dict_sys->table_id_hash,
+			    fold, this);
+		id = new_id;
+		fold = ut_fold_ull(id);
+		HASH_INSERT(dict_table_t, id_hash, dict_sys->table_id_hash,
+			    fold, this);
+	}
+
+	return err;
+}
+
 /** Insert or update SYS_COLUMNS and the hidden metadata record
 for instant ALTER TABLE.
 @param[in]	ha_alter_info	ALTER TABLE context
@@ -5495,6 +5540,20 @@ static bool innobase_instant_try(
 empty_table:
 		/* The table is empty. */
 		ut_ad(page_is_root(block->frame));
+		if (index->table->instant) {
+			/* Assign a new dict_table_t::id
+			to invalidate old undo log records in purge,
+			so that they cannot refer to fields that were
+			instantly added to the end of the index,
+			instead of using the canonical positions
+			that will be replaced below
+			by index->clear_instant_alter(). */
+			err = index->table->reassign_id(trx);
+			if (err != DB_SUCCESS) {
+				goto func_exit;
+			}
+		}
+		/* MDEV-17383: free metadata BLOBs! */
 		btr_page_empty(block, NULL, index, 0, &mtr);
 		index->clear_instant_alter();
 		err = DB_SUCCESS;

storage/innobase/include/dict0mem.h

@@ -1677,6 +1677,15 @@ struct dict_table_t {
 		const char*	old_v_col_names,
 		const ulint*	col_map);
 
+	/** Assign a new id to invalidate old undo log records, so
+	that purge will be unable to refer to fields that used to be
+	instantly added to the end of the index. This is only to be
+	used during ALTER TABLE when the table is empty, before
+	invoking dict_index_t::clear_instant_alter().
+	@param[in,out] trx	dictionary transaction
+	@return	error code */
+	inline dberr_t reassign_id(trx_t* trx);
+
 	/** Add the table definition to the data dictionary cache */
 	void add_to_cache();