MDEV-17899 Assertion failures on rollback of instant ADD/DROP

row_undo_ins_remove_clust_rec(): When rolling back the insert
of a metadata record, also reset the root page state within the
same mini-transaction.

dict_table_t::rollback_instant(): Do not allow index->n_core_fields
to be larger than index->n_fields. This logic exists in the 10.3
branch but was incorrectly removed as part of MDEV-15562 in 10.4.0.

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

 SET @save_frequency= @@GLOBAL.innodb_purge_rseg_truncate_frequency;
 SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
+SET @old_instant=
+(SELECT variable_value FROM information_schema.global_status
+WHERE variable_name = 'innodb_instant_alter_column');
 CREATE TABLE t1 (
 pk INT AUTO_INCREMENT PRIMARY KEY,
 c1 INT,
@@ -260,4 +263,31 @@ a	b	c	d
 1	2	NULL	1
 2	3	4	1
 DROP TABLE t1;
+#
+# MDEV-17899 Assertion failures on rollback of instant ADD/DROP
+#
+SET @save_dbug = @@SESSION.debug_dbug;
+SET debug_dbug='+d,ib_commit_inplace_fail_1';
+CREATE TABLE t1 (a int, b int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1,2);
+ALTER TABLE t1 DROP COLUMN b;
+ERROR HY000: Internal error: Injected error!
+ALTER TABLE t1 DROP COLUMN b;
+ERROR HY000: Internal error: Injected error!
+ALTER TABLE t1 ADD COLUMN c INT;
+ERROR HY000: Internal error: Injected error!
+SELECT * FROM t1;
+a	b
+1	2
+DROP TABLE t1;
+CREATE TABLE t1 (a int, b int) ENGINE=InnoDB;
+ALTER TABLE t1 ADD COLUMN c INT;
+ERROR HY000: Internal error: Injected error!
+DROP TABLE t1;
+SET debug_dbug = @save_dbug;
+SELECT variable_value-@old_instant instants
+FROM information_schema.global_status
+WHERE variable_name = 'innodb_instant_alter_column';
+instants
+20
 SET GLOBAL innodb_purge_rseg_truncate_frequency = @save_frequency;

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

@@ -5,6 +5,10 @@
 SET @save_frequency= @@GLOBAL.innodb_purge_rseg_truncate_frequency;
 SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
 
+SET @old_instant=
+(SELECT variable_value FROM information_schema.global_status
+WHERE variable_name = 'innodb_instant_alter_column');
+
 CREATE TABLE t1 (
  pk INT AUTO_INCREMENT PRIMARY KEY,
  c1 INT,
@@ -296,4 +300,32 @@ SET DEBUG_SYNC = RESET;
 SELECT * FROM t1;
 DROP TABLE t1;
 
+--echo #
+--echo # MDEV-17899 Assertion failures on rollback of instant ADD/DROP
+--echo #
+
+SET @save_dbug = @@SESSION.debug_dbug;
+SET debug_dbug='+d,ib_commit_inplace_fail_1';
+CREATE TABLE t1 (a int, b int) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1,2);
+--error ER_INTERNAL_ERROR
+ALTER TABLE t1 DROP COLUMN b;
+--error ER_INTERNAL_ERROR
+ALTER TABLE t1 DROP COLUMN b;
+--error ER_INTERNAL_ERROR
+ALTER TABLE t1 ADD COLUMN c INT;
+SELECT * FROM t1;
+DROP TABLE t1;
+
+CREATE TABLE t1 (a int, b int) ENGINE=InnoDB;
+--error ER_INTERNAL_ERROR
+ALTER TABLE t1 ADD COLUMN c INT;
+DROP TABLE t1;
+
+SET debug_dbug = @save_dbug;
+
+SELECT variable_value-@old_instant instants
+FROM information_schema.global_status
+WHERE variable_name = 'innodb_instant_alter_column';
+
 SET GLOBAL innodb_purge_rseg_truncate_frequency = @save_frequency;

storage/innobase/handler/handler0alter.cc

@@ -680,7 +680,6 @@ inline void dict_table_t::rollback_instant(
 	DBUG_ASSERT(old_n_cols >= DATA_N_SYS_COLS);
 	DBUG_ASSERT(n_cols == n_def);
 	DBUG_ASSERT(index->n_def == index->n_fields);
-	DBUG_ASSERT(index->n_core_fields <= old_n_fields);
 	DBUG_ASSERT(index->n_core_fields <= index->n_fields);
 	DBUG_ASSERT(instant || !old_instant);
 
@@ -699,6 +698,10 @@ inline void dict_table_t::rollback_instant(
 	}
 
 	index->n_def = index->n_fields = old_n_fields;
+	if (index->n_core_fields > old_n_fields) {
+		index->n_core_fields = old_n_fields;
+		index->n_core_null_bytes = UT_BITS_IN_BYTES(index->n_nullable);
+	}
 
 	const dict_col_t* const new_cols = cols;
 	const dict_col_t* const new_cols_end = cols + n_cols;

storage/innobase/row/row0uins.cc

@@ -65,7 +65,6 @@ row_undo_ins_remove_clust_rec(
 /*==========================*/
 	undo_node_t*	node)	/*!< in: undo node */
 {
-	btr_cur_t*	btr_cur;
 	ibool		success;
 	dberr_t		err;
 	ulint		n_tries	= 0;
@@ -73,7 +72,7 @@ row_undo_ins_remove_clust_rec(
 	dict_index_t*	index	= node->pcur.btr_cur.index;
 	bool		online;
 
-	ut_ad(dict_index_is_clust(index));
+	ut_ad(index->is_primary());
 	ut_ad(node->trx->in_rollback);
 
 	mtr.start();
@@ -87,6 +86,7 @@ row_undo_ins_remove_clust_rec(
 		index->set_modified(mtr);
 		online = dict_index_is_online_ddl(index);
 		if (online) {
+			ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
 			ut_ad(node->trx->dict_operation_lock_mode
 			      != RW_X_LATCH);
 			ut_ad(node->table->id != DICT_INDEXES_ID);
@@ -104,19 +104,19 @@ row_undo_ins_remove_clust_rec(
 	success = btr_pcur_restore_position(
 		online
 		? BTR_MODIFY_LEAF | BTR_ALREADY_S_LATCHED
-		: BTR_MODIFY_LEAF, &node->pcur, &mtr);
+		: (node->rec_type == TRX_UNDO_INSERT_METADATA)
+		? BTR_MODIFY_TREE : BTR_MODIFY_LEAF, &node->pcur, &mtr);
 	ut_a(success);
 
-	btr_cur = btr_pcur_get_btr_cur(&node->pcur);
+	rec_t* rec = btr_pcur_get_rec(&node->pcur);
 
-	ut_ad(rec_get_trx_id(btr_cur_get_rec(btr_cur), btr_cur->index)
-	      == node->trx->id);
-	ut_ad(!rec_get_deleted_flag(
-		      btr_cur_get_rec(btr_cur),
-		      dict_table_is_comp(btr_cur->index->table)));
+	ut_ad(rec_get_trx_id(rec, index) == node->trx->id);
+	ut_ad(!rec_get_deleted_flag(rec, index->table->not_redundant())
+	      || rec_is_alter_metadata(rec, index->table->not_redundant()));
+	ut_ad(rec_is_metadata(rec, index->table->not_redundant())
+	      == (node->rec_type == TRX_UNDO_INSERT_METADATA));
 
 	if (online && dict_index_is_online_ddl(index)) {
-		const rec_t*	rec	= btr_cur_get_rec(btr_cur);
 		mem_heap_t*	heap	= NULL;
 		const ulint*	offsets	= rec_get_offsets(
 			rec, index, NULL, true, ULINT_UNDEFINED, &heap);
@@ -130,8 +130,7 @@ row_undo_ins_remove_clust_rec(
 			      == RW_X_LATCH);
 			ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
 
-			dict_drop_index_tree(btr_pcur_get_rec(&node->pcur),
-					     &node->pcur, &mtr);
+			dict_drop_index_tree(rec, &node->pcur, &mtr);
 			mtr.commit();
 
 			mtr.start();
@@ -150,7 +149,6 @@ row_undo_ins_remove_clust_rec(
 			ut_ad(node->trx->dict_operation_lock_mode
 			      == RW_X_LATCH);
 			ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
-			const rec_t* rec = btr_pcur_get_rec(&node->pcur);
 			if (rec_get_n_fields_old(rec)
 			    != DICT_NUM_FIELDS__SYS_COLUMNS) {
 				break;
@@ -165,7 +163,7 @@ row_undo_ins_remove_clust_rec(
 		}
 	}
 
-	if (btr_cur_optimistic_delete(btr_cur, 0, &mtr)) {
+	if (btr_cur_optimistic_delete(&node->pcur.btr_cur, 0, &mtr)) {
 		err = DB_SUCCESS;
 		goto func_exit;
 	}
@@ -185,7 +183,8 @@ row_undo_ins_remove_clust_rec(
 			&node->pcur, &mtr);
 	ut_a(success);
 
-	btr_cur_pessimistic_delete(&err, FALSE, btr_cur, 0, true, &mtr);
+	btr_cur_pessimistic_delete(&err, FALSE, &node->pcur.btr_cur, 0, true,
+				   &mtr);
 
 	/* The delete operation may fail if we have little
 	file space left: TODO: easiest to crash the database
@@ -204,29 +203,34 @@ row_undo_ins_remove_clust_rec(
 	}
 
 func_exit:
-	btr_pcur_commit_specify_mtr(&node->pcur, &mtr);
 	if (err == DB_SUCCESS && node->rec_type == TRX_UNDO_INSERT_METADATA) {
 		/* When rolling back the very first instant ADD COLUMN
 		operation, reset the root page to the basic state. */
 		ut_ad(!index->table->is_temporary());
-		mtr.start();
 		if (page_t* root = btr_root_get(index, &mtr)) {
 			byte* page_type = root + FIL_PAGE_TYPE;
 			ut_ad(mach_read_from_2(page_type)
 			      == FIL_PAGE_TYPE_INSTANT
 			      || mach_read_from_2(page_type)
 			      == FIL_PAGE_INDEX);
-			index->set_modified(mtr);
 			mlog_write_ulint(page_type, FIL_PAGE_INDEX,
 					 MLOG_2BYTES, &mtr);
 			byte* instant = PAGE_INSTANT + PAGE_HEADER + root;
 			mlog_write_ulint(instant,
 					 page_ptr_get_direction(instant + 1),
 					 MLOG_2BYTES, &mtr);
+			rec_t* infimum = page_get_infimum_rec(root);
+			rec_t* supremum = page_get_supremum_rec(root);
+			static const byte str[8 + 8] = "supremuminfimum";
+			if (memcmp(infimum, str + 8, 8)
+			    || memcmp(supremum, str, 8)) {
+				mlog_write_string(infimum, str + 8, 8, &mtr);
+				mlog_write_string(supremum, str, 8, &mtr);
+			}
 		}
-		mtr.commit();
 	}
 
+	btr_pcur_commit_specify_mtr(&node->pcur, &mtr);
 	return(err);
 }