Revert "WIP: Add row_log_t::instant"

This reverts commit 26897301
and adds some changes.

Only a table on which the 10.3 ADD COLUMN...LAST was performed
can "lose its instantness" when it becomes empty. After
instant DROP COLUMN or reordering columns, we cannot simply
convert the table to the canonical format, because the data
dictionary cache and all possibly existing references to it
from other client connection threads would have to be adjusted.

rec_init_offsets_comp_ordinary(): Remove a FIXME comment.

row_log_allocate(): Replace dict_index_t::is_instant()
with a more appropriate condition that ignores dict_table_t::instant.

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

@@ -197,7 +197,8 @@ DELETE FROM t1;
 connection ddl;
 SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL copied WAIT_FOR logged';
 ALTER TABLE t1 FORCE;
-disconnect stop_purge;
+connection stop_purge;
+COMMIT;
 connection default;
 SET DEBUG_SYNC = 'now WAIT_FOR copied';
 InnoDB		1 transactions not purged
@@ -211,6 +212,28 @@ SELECT * FROM t1;
 a	b	c
 1	2	NULL
 2	3	4
+ALTER TABLE t1 DROP b, ALGORITHM=INSTANT;
+connection stop_purge;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+DELETE FROM t1;
+connection ddl;
+SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL copied WAIT_FOR logged';
+ALTER TABLE t1 FORCE;
+disconnect stop_purge;
+connection default;
+SET DEBUG_SYNC = 'now WAIT_FOR copied';
+InnoDB		1 transactions not purged
+INSERT INTO t1 SET a=1;
+INSERT INTO t1 SET a=2,c=4;
+SET DEBUG_SYNC = 'now SIGNAL logged';
+connection ddl;
+connection default;
+SET DEBUG_SYNC = RESET;
+SELECT * FROM t1;
+a	c
+1	NULL
+2	4
 #
 # MDEV-15872 Crash in online ALTER TABLE...ADD PRIMARY KEY
 # after instant ADD COLUMN ... NULL
@@ -232,8 +255,8 @@ disconnect ddl;
 connection default;
 SET DEBUG_SYNC = RESET;
 SELECT * FROM t1;
-a	b	c	d
-1	2	NULL	1
-2	3	4	1
+a	c	d
+1	NULL	1
+2	4	1
 DROP TABLE t1;
 SET GLOBAL innodb_purge_rseg_truncate_frequency = @save_frequency;

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

@@ -222,7 +222,8 @@ connection ddl;
 SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL copied WAIT_FOR logged';
 send ALTER TABLE t1 FORCE;
 
-disconnect stop_purge;
+connection stop_purge;
+COMMIT;
 
 connection default;
 SET DEBUG_SYNC = 'now WAIT_FOR copied';
@@ -235,6 +236,33 @@ SET DEBUG_SYNC = 'now SIGNAL logged';
 connection ddl;
 reap;
 
+connection default;
+SET DEBUG_SYNC = RESET;
+SELECT * FROM t1;
+ALTER TABLE t1 DROP b, ALGORITHM=INSTANT;
+connection stop_purge;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+
+connection default;
+DELETE FROM t1;
+
+connection ddl;
+SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL copied WAIT_FOR logged';
+send ALTER TABLE t1 FORCE;
+
+disconnect stop_purge;
+
+connection default;
+SET DEBUG_SYNC = 'now WAIT_FOR copied';
+let $wait_all_purged = 1;
+--source include/wait_all_purged.inc
+INSERT INTO t1 SET a=1;
+INSERT INTO t1 SET a=2,c=4;
+SET DEBUG_SYNC = 'now SIGNAL logged';
+
+connection ddl;
+reap;
+
 connection default;
 SET DEBUG_SYNC = RESET;
 SELECT * FROM t1;

storage/innobase/rem/rem0rec.cc

@@ -387,13 +387,6 @@ rec_init_offsets_comp_ordinary(
 		} else if (i < n_fields) {
 			/* The field is present, and will be covered below. */
 		} else if (!mblob && def_val) {
-			/* FIXME: Fix the MDEV-16131 scenario when
-			instant DROP COLUMN was followed by
-			ALTER TABLE...FORCE, LOCK=NONE and the
-			table was emptied during row_log_table_apply().
-			Currently, def_val!=NULL only signals that
-			instant ADD had been used, but it is not
-			telling which columns were dropped. */
 			const dict_col_t::def_t& d = def_val[i - n_core];
 			if (!d.data) {
 				len = offs | REC_OFFS_SQL_NULL;
@@ -472,7 +465,8 @@ rec_init_offsets_comp_ordinary(
 		} else {
 			len = offs += field->fixed_len;
 		}
-	} while (field++, rec_offs_base(offsets)[++i] = len, i < rec_offs_n_fields(offsets));
+	} while (field++, rec_offs_base(offsets)[++i] = len,
+		 i < rec_offs_n_fields(offsets));
 
 	*rec_offs_base(offsets)
 		= ulint(rec - (lens + 1)) | REC_OFFS_COMPACT | any;

storage/innobase/row/row0log.cc

@@ -222,11 +222,6 @@ struct row_log_t {
 				decryption or NULL */
 	const char*	path;	/*!< where to create temporary file during
 				log operation */
-	/** the instantly dropped columns 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. */
-	dict_instant_t*	instant;
 	/** 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,
@@ -241,15 +236,14 @@ struct row_log_t {
 	const TABLE*	old_table; /*< Use old table in case of error. */
 
 	uint64_t	n_rows; /*< Number of rows read from the table */
-	/** Determine whether the log should be in the 'instant ALTER TABLE'
-	format
+	/** 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 ALTER TABLE' format */
+	@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 || instant);
-		return instant || n_core_fields != index->n_fields;
+		ut_ad(n_core_fields <= index->n_fields);
+		return n_core_fields != index->n_fields;
 	}
 
 	const byte* instant_field_value(ulint n, ulint* len) const
@@ -2536,7 +2530,6 @@ row_log_table_apply_op(
 			}
 
 			rec_offs_set_n_fields(offsets, dup->index->n_fields);
-			/* FIXME: pass log->instant as well */
 			rec_init_offsets_temp(mrec, dup->index, offsets,
 					      log->n_core_fields,
 					      log->non_core_fields,
@@ -3212,10 +3205,8 @@ row_log_allocate(
 	log->head.total = 0;
 	log->path = path;
 	log->n_core_fields = index->n_core_fields;
-#if 1 /* FIXME: when we add instant->heap, this must be a copy! */
-	log->instant = index->table->instant;
-#endif
-	ut_ad(!table || log->is_instant(index) == index->is_instant());
+	ut_ad(!table || log->is_instant(index)
+	      == (index->n_core_fields < index->n_fields));
 	log->allow_not_null = allow_not_null;
 	log->old_table = old_table;
 	log->n_rows = 0;