MDEV-15060 Assertion in row_log_table_apply_op after instant ADD when the table is emptied during subsequent ALTER TABLE

During an online table rebuild, a table could be emptied and converted
from 'instant ADD' format to plain (pre-10.3) format. All online_log
records for rebuilding the table must be written and parsed in the
format of the table that existed at the start of the operation.

row_log_t::n_core_fields: A new field for recording index->n_core_fields
when online ALTER is initiated in row_log_allocate().

row_log_t::is_instant(): Determine if the log is in the instant format.
Only invoked by the row_log_table_ family of functions.

dict_index_t::get_n_nullable(): Remove is_instant() debug assertions.
Because a table can be converted to non-instant format during a
table-rebuilding ALTER TABLE, these assertions would be bogus when
executing row_log_table_apply().

rec_init_offsets_temp(): Add the parameter n_core for passing the
original index->n_core_fields.

rec_init_offsets_temp(): Add a 3-parameter variant.

rec_init_offsets_comp_ordinary(): Add the parameter n_core for
passing the index->n_core_fields.

6 files changed, 160 insertions, 31 deletions

diff --git a/mysql-test/suite/innodb/r/instant_alter_debug.result b/mysql-test/suite/innodb/r/instant_alter_debug.result
index bd7d538678a..bb7c414e9ef 100644
--- a/mysql-test/suite/innodb/r/instant_alter_debug.result
+++ b/mysql-test/suite/innodb/r/instant_alter_debug.result
@@ -164,4 +164,28 @@ INSERT INTO t11 () VALUES ();
 UPDATE t11 SET c22 = 1;
 InnoDB		0 transactions not purged
 DROP TABLE t1,t2,t3,t4,t5,t6,t7,t8,t9,t10,t11;
+#
+# MDEV-15060 Assertion in row_log_table_apply_op after instant ADD
+# when the table is emptied during subsequent ALTER TABLE
+#
+CREATE TABLE t1 (a INT) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (NULL);
+ALTER TABLE t1 ADD COLUMN b INT NOT NULL;
+connect stop_purge,localhost,root;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connect ddl,localhost,root,,test;
+DELETE FROM t1;
+SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL copied WAIT_FOR logged';
+ALTER TABLE t1 FORCE;
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR copied';
+BEGIN;
+INSERT INTO t1 SET b=1;
+ROLLBACK;
+disconnect stop_purge;
+InnoDB		2 transactions not purged
+SET DEBUG_SYNC='now SIGNAL logged';
+disconnect ddl;
+DROP TABLE t1;
+SET DEBUG_SYNC='RESET';
 SET GLOBAL innodb_purge_rseg_truncate_frequency = @save_frequency;
diff --git a/mysql-test/suite/innodb/t/instant_alter_debug.test b/mysql-test/suite/innodb/t/instant_alter_debug.test
index 69aab6e2fc1..bcf69628010 100644
--- a/mysql-test/suite/innodb/t/instant_alter_debug.test
+++ b/mysql-test/suite/innodb/t/instant_alter_debug.test
@@ -1,7 +1,7 @@
 --source include/have_innodb.inc
 --source include/have_debug.inc
 --source include/have_debug_sync.inc
---source include/have_innodb.inc
+
 SET @save_frequency= @@GLOBAL.innodb_purge_rseg_truncate_frequency;
 SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
 
@@ -178,4 +178,52 @@ UPDATE t11 SET c22 = 1;
 --source include/wait_all_purged.inc
 DROP TABLE t1,t2,t3,t4,t5,t6,t7,t8,t9,t10,t11;
 
+--echo #
+--echo # MDEV-15060 Assertion in row_log_table_apply_op after instant ADD
+--echo # when the table is emptied during subsequent ALTER TABLE
+--echo #
+
+CREATE TABLE t1 (a INT) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (NULL);
+ALTER TABLE t1 ADD COLUMN b INT NOT NULL;
+connect stop_purge,localhost,root;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connect ddl,localhost,root,,test;
+DELETE FROM t1;
+SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL copied WAIT_FOR logged';
+send ALTER TABLE t1 FORCE;
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR copied';
+
+BEGIN;
+INSERT INTO t1 SET b=1;
+ROLLBACK;
+disconnect stop_purge;
+
+# Wait for purge to empty the table.
+# (This is based on wait_all_purged.inc, but there are 2 transactions
+# from the pending ALTER TABLE t1 FORCE.)
+
+let $wait_counter= 300;
+while ($wait_counter)
+{
+    --replace_regex /.*History list length ([0-9]+).*/\1/
+    let $remaining= `SHOW ENGINE INNODB STATUS`;
+    if ($remaining == 'InnoDB		2')
+    {
+        let $wait_counter= 0;
+    }
+    if ($wait_counter)
+    {
+        real_sleep 0.1;
+        dec $wait_counter;
+    }
+}
+echo $remaining transactions not purged;
+
+SET DEBUG_SYNC='now SIGNAL logged';
+disconnect ddl;
+DROP TABLE t1;
+SET DEBUG_SYNC='RESET';
+
 SET GLOBAL innodb_purge_rseg_truncate_frequency = @save_frequency;
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 163671f854d..d4f879a6c77 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -1050,13 +1050,11 @@ struct dict_index_t{
 	@return	number of fields 0..n_prefix-1 that can be set NULL */
 	unsigned get_n_nullable(ulint n_prefix) const
 	{
-		DBUG_ASSERT(is_instant());
 		DBUG_ASSERT(n_prefix > 0);
 		DBUG_ASSERT(n_prefix <= n_fields);
 		unsigned n = n_nullable;
 		for (; n_prefix < n_fields; n_prefix++) {
 			const dict_col_t* col = fields[n_prefix].col;
-			DBUG_ASSERT(is_dummy || col->is_instant());
 			DBUG_ASSERT(!col->is_virtual());
 			n -= col->is_nullable();
 		}
diff --git a/storage/innobase/include/rem0rec.h b/storage/innobase/include/rem0rec.h
index 9c4151a28a5..cc8b92e99b1 100644
--- a/storage/innobase/include/rem0rec.h
+++ b/storage/innobase/include/rem0rec.h
@@ -961,15 +961,27 @@ rec_get_converted_size_temp(
 @param[in]	rec	temporary file record
 @param[in]	index	index of that the record belongs to
 @param[in,out]	offsets	offsets to the fields; in: rec_offs_n_fields(offsets)
-@param[in]	status	REC_STATUS_ORDINARY or REC_STATUS_COLUMNS_ADDED
-*/
+@param[in]	n_core	number of core fields (index->n_core_fields)
+@param[in]	status	REC_STATUS_ORDINARY or REC_STATUS_COLUMNS_ADDED */
 void
 rec_init_offsets_temp(
 	const rec_t*		rec,
 	const dict_index_t*	index,
 	ulint*			offsets,
+	ulint			n_core,
 	rec_comp_status_t	status = REC_STATUS_ORDINARY)
 	MY_ATTRIBUTE((nonnull));
+/** Determine the offset to each field in temporary file.
+@param[in]	rec	temporary file record
+@param[in]	index	index of that the record belongs to
+@param[in,out]	offsets	offsets to the fields; in: rec_offs_n_fields(offsets)
+*/
+void
+rec_init_offsets_temp(
+	const rec_t*		rec,
+	const dict_index_t*	index,
+	ulint*			offsets)
+	MY_ATTRIBUTE((nonnull));
 
 /** Convert a data tuple prefix to the temporary file format.
 @param[out]	rec		record in temporary file format
diff --git a/storage/innobase/rem/rem0rec.cc b/storage/innobase/rem/rem0rec.cc
index 6da2866b0c1..7d1a35d82e4 100644
--- a/storage/innobase/rem/rem0rec.cc
+++ b/storage/innobase/rem/rem0rec.cc
@@ -298,6 +298,7 @@ in ROW_FORMAT=COMPACT,DYNAMIC,COMPRESSED.
 This is a special case of rec_init_offsets() and rec_get_offsets_func().
 @param[in]	rec	leaf-page record
 @param[in]	index	the index that the record belongs in
+@param[in]	n_core	number of core fields (index->n_core_fields)
 @param[in,out]	offsets	offsets, with valid rec_offs_n_fields(offsets)
 @param[in]	format	record format */
 static inline
@@ -306,17 +307,19 @@ rec_init_offsets_comp_ordinary(
 	const rec_t*		rec,
 	const dict_index_t*	index,
 	ulint*			offsets,
+	ulint			n_core,
 	rec_leaf_format		format)
 {
 	ulint		offs		= 0;
 	ulint		any		= 0;
 	const byte*	nulls		= rec;
 	const byte*	lens		= NULL;
-	ulint		n_fields	= index->n_core_fields;
+	ulint		n_fields	= n_core;
 	ulint		null_mask	= 1;
 
-	ut_ad(index->n_core_fields > 0);
-	ut_ad(index->n_fields >= index->n_core_fields);
+	ut_ad(index->n_core_fields >= n_core);
+	ut_ad(n_core > 0);
+	ut_ad(index->n_fields >= n_core);
 	ut_ad(index->n_core_null_bytes <= UT_BITS_IN_BYTES(index->n_nullable));
 	ut_ad(format == REC_LEAF_TEMP || format == REC_LEAF_TEMP_COLUMNS_ADDED
 	      || dict_table_is_comp(index->table));
@@ -344,17 +347,17 @@ ordinary:
 		/* We would have !index->is_instant() when rolling back
 		an instant ADD COLUMN operation. */
 		nulls -= REC_N_NEW_EXTRA_BYTES;
+		ut_ad(index->is_instant());
 		/* fall through */
 	case REC_LEAF_TEMP_COLUMNS_ADDED:
-		ut_ad(index->is_instant());
-		n_fields = unsigned(index->n_core_fields) + 1
-			+ rec_get_n_add_field(nulls);
+		n_fields = n_core + 1 + rec_get_n_add_field(nulls);
 		ut_ad(n_fields <= index->n_fields);
 		const ulint n_nullable = index->get_n_nullable(n_fields);
 		const ulint n_null_bytes = UT_BITS_IN_BYTES(n_nullable);
 		ut_d(n_null = n_nullable);
 		ut_ad(n_null <= index->n_nullable);
-		ut_ad(n_null_bytes >= index->n_core_null_bytes);
+		ut_ad(n_null_bytes >= index->n_core_null_bytes
+		      || n_core < index->n_core_fields);
 		lens = --nulls - n_null_bytes;
 	}
 
@@ -614,11 +617,13 @@ rec_init_offsets(
 		case REC_STATUS_COLUMNS_ADDED:
 			ut_ad(leaf);
 			rec_init_offsets_comp_ordinary(rec, index, offsets,
+						       index->n_core_fields,
 						       REC_LEAF_COLUMNS_ADDED);
 			return;
 		case REC_STATUS_ORDINARY:
 			ut_ad(leaf);
 			rec_init_offsets_comp_ordinary(rec, index, offsets,
+						       index->n_core_fields,
 						       REC_LEAF_ORDINARY);
 			return;
 		}
@@ -1689,25 +1694,44 @@ rec_get_converted_size_temp(
 @param[in]	rec	temporary file record
 @param[in]	index	index of that the record belongs to
 @param[in,out]	offsets	offsets to the fields; in: rec_offs_n_fields(offsets)
-@param[in]	status	REC_STATUS_ORDINARY or REC_STATUS_COLUMNS_ADDED
-*/
+@param[in]	n_core	number of core fields (index->n_core_fields)
+@param[in]	status	REC_STATUS_ORDINARY or REC_STATUS_COLUMNS_ADDED */
 void
 rec_init_offsets_temp(
 	const rec_t*		rec,
 	const dict_index_t*	index,
 	ulint*			offsets,
+	ulint			n_core,
 	rec_comp_status_t	status)
 {
 	ut_ad(status == REC_STATUS_ORDINARY
 	      || status == REC_STATUS_COLUMNS_ADDED);
-	ut_ad(status == REC_STATUS_ORDINARY || index->is_instant());
-
-	rec_init_offsets_comp_ordinary(rec, index, offsets,
+	/* The table may have been converted to plain format
+	if it was emptied during an ALTER TABLE operation. */
+	ut_ad(index->n_core_fields == n_core || !index->is_instant());
+	ut_ad(index->n_core_fields >= n_core);
+	rec_init_offsets_comp_ordinary(rec, index, offsets, n_core,
 				       status == REC_STATUS_COLUMNS_ADDED
 				       ? REC_LEAF_TEMP_COLUMNS_ADDED
 				       : REC_LEAF_TEMP);
 }
 
+/** Determine the offset to each field in temporary file.
+@param[in]	rec	temporary file record
+@param[in]	index	index of that the record belongs to
+@param[in,out]	offsets	offsets to the fields; in: rec_offs_n_fields(offsets)
+*/
+void
+rec_init_offsets_temp(
+	const rec_t*		rec,
+	const dict_index_t*	index,
+	ulint*			offsets)
+{
+	ut_ad(!index->is_instant());
+	rec_init_offsets_comp_ordinary(rec, index, offsets,
+				       index->n_core_fields, REC_LEAF_TEMP);
+}
+
 /** Convert a data tuple prefix to the temporary file format.
 @param[out]	rec		record in temporary file format
 @param[in]	index		clustered or secondary index
diff --git a/storage/innobase/row/row0log.cc b/storage/innobase/row/row0log.cc
index be562334630..1357217ee4e 100644
--- a/storage/innobase/row/row0log.cc
+++ b/storage/innobase/row/row0log.cc
@@ -221,9 +221,24 @@ struct row_log_t {
 				decryption or NULL */
 	const char*	path;	/*!< where to create temporary file during
 				log operation */
+	/** the number of core fields in the clustered index of the
+	source table; before row_log_table_apply() completes, the
+	table could be emptied, so that table->is_instant() no longer holds,
+	but all log records must be in the "instant" format. */
+	unsigned	n_core_fields;
 	bool		ignore; /*!< Whether the alter ignore is being used;
 				if not, NULL values will not be converted to
 				defaults */
+
+	/** Determine whether the log should be in the 'instant ADD' format
+	@param[in]	index	the clustered index of the source table
+	@return	whether to use the 'instant ADD COLUMN' format */
+	bool is_instant(const dict_index_t* index) const
+	{
+		ut_ad(table);
+		ut_ad(n_core_fields <= index->n_fields);
+		return n_core_fields != index->n_fields;
+	}
 };
 
 /** Create the file or online log if it does not exist.
@@ -872,12 +887,13 @@ row_log_table_low_redundant(
 				DATA_ROLL_PTR_LEN);
 	}
 
-	rec_comp_status_t status = index->is_instant()
+	const bool is_instant = index->online_log->is_instant(index);
+	rec_comp_status_t status = is_instant
 		? REC_STATUS_COLUMNS_ADDED : REC_STATUS_ORDINARY;
 
 	size = rec_get_converted_size_temp(
 		index, tuple->fields, tuple->n_fields, &extra_size, status);
-	if (index->is_instant()) {
+	if (is_instant) {
 		size++;
 		extra_size++;
 	}
@@ -928,8 +944,8 @@ row_log_table_low_redundant(
 		}
 
 		if (status == REC_STATUS_COLUMNS_ADDED) {
-			ut_ad(index->is_instant());
-			if (n_fields <= index->n_core_fields) {
+			ut_ad(is_instant);
+			if (n_fields <= index->online_log->n_core_fields) {
 				status = REC_STATUS_ORDINARY;
 			}
 			*b = status;
@@ -1019,11 +1035,12 @@ row_log_table_low(
 	const ulint omit_size = REC_N_NEW_EXTRA_BYTES;
 
 	const ulint rec_extra_size = rec_offs_extra_size(offsets) - omit_size;
-	extra_size = rec_extra_size + index->is_instant();
+	const bool is_instant = index->online_log->is_instant(index);
+	extra_size = rec_extra_size + is_instant;
 
 	mrec_size = ROW_LOG_HEADER_SIZE
 		+ (extra_size >= 0x80) + rec_offs_size(offsets) - omit_size
-		+ index->is_instant();
+		+ is_instant;
 
 	if (insert || index->online_log->same_pk) {
 		ut_ad(!old_pk);
@@ -1068,7 +1085,7 @@ row_log_table_low(
 			*b++ = static_cast<byte>(extra_size);
 		}
 
-		if (index->is_instant()) {
+		if (is_instant) {
 			*b++ = rec_get_status(rec);
 		} else {
 			ut_ad(rec_get_status(rec) == REC_STATUS_ORDINARY);
@@ -2424,6 +2441,7 @@ row_log_table_apply_op(
 		return(NULL);
 	}
 
+	const bool is_instant = log->is_instant(dup->index);
 	const mrec_t* const mrec_start = mrec;
 
 	switch (*mrec++) {
@@ -2443,7 +2461,7 @@ row_log_table_apply_op(
 
 		mrec += extra_size;
 
-		ut_ad(extra_size || !dup->index->is_instant());
+		ut_ad(extra_size || !is_instant);
 
 		if (mrec > mrec_end) {
 			return(NULL);
@@ -2451,7 +2469,8 @@ row_log_table_apply_op(
 
 		rec_offs_set_n_fields(offsets, dup->index->n_fields);
 		rec_init_offsets_temp(mrec, dup->index, offsets,
-				      dup->index->is_instant()
+				      log->n_core_fields,
+				      is_instant
 				      ? static_cast<rec_comp_status_t>(
 					      *(mrec - extra_size))
 				      : REC_STATUS_ORDINARY);
@@ -2535,7 +2554,7 @@ row_log_table_apply_op(
 		is not changed, the log will only contain
 		DB_TRX_ID,new_row. */
 
-		if (dup->index->online_log->same_pk) {
+		if (log->same_pk) {
 			ut_ad(new_index->n_uniq == dup->index->n_uniq);
 
 			extra_size = *mrec++;
@@ -2549,7 +2568,7 @@ row_log_table_apply_op(
 
 			mrec += extra_size;
 
-			ut_ad(extra_size || !dup->index->is_instant());
+			ut_ad(extra_size || !is_instant);
 
 			if (mrec > mrec_end) {
 				return(NULL);
@@ -2557,7 +2576,8 @@ row_log_table_apply_op(
 
 			rec_offs_set_n_fields(offsets, dup->index->n_fields);
 			rec_init_offsets_temp(mrec, dup->index, offsets,
-					      dup->index->is_instant()
+					      log->n_core_fields,
+					      is_instant
 					      ? static_cast<rec_comp_status_t>(
 						      *(mrec - extra_size))
 					      : REC_STATUS_ORDINARY);
@@ -2649,7 +2669,7 @@ row_log_table_apply_op(
 
 			mrec += extra_size;
 
-			ut_ad(extra_size || !dup->index->is_instant());
+			ut_ad(extra_size || !is_instant);
 
 			if (mrec > mrec_end) {
 				return(NULL);
@@ -2657,7 +2677,8 @@ row_log_table_apply_op(
 
 			rec_offs_set_n_fields(offsets, dup->index->n_fields);
 			rec_init_offsets_temp(mrec, dup->index, offsets,
-					      dup->index->is_instant()
+					      log->n_core_fields,
+					      is_instant
 					      ? static_cast<rec_comp_status_t>(
 						      *(mrec - extra_size))
 					      : REC_STATUS_ORDINARY);
@@ -3211,6 +3232,8 @@ row_log_allocate(
 	log->head.blocks = log->head.bytes = 0;
 	log->head.total = 0;
 	log->path = path;
+	log->n_core_fields = index->n_core_fields;
+	ut_ad(!table || log->is_instant(index) == index->is_instant());
 	log->ignore=ignore;
 
 	dict_index_set_online_status(index, ONLINE_INDEX_CREATION);