Include dropped columns in ctx->instant_table and dict_index_t

row_metadata_to_tuple(): Convert a metadata record to a data tuple,
based on the new info_bits of the metadata record.

btr_cur_pessimistic_update(): Invoke row_metadata_to_tuple() if needed.

dict_index_t::instant_add_field(): Append fields at the end, if any.
The fields for dropped columns are preserved.

dict_table_t::instant_column(): Renamed from instant_add_column().
Add the parameter col_map so that columns can be reordered.
FIXME: Also support ADD/DROP/reorder/rename of VIRTUAL COLUMN.

ha_innobase_inplace_ctx::prepare_instant(): Preserve any instantly
dropped columns in instant_table and its first index.
FIXME: Support instant ALTER TABLE even if hidden FTS_DOC_ID exists.

innobase_instant_try(): Simplify the logic for detecting dropped
or reordered columns. If any drop or reorder is needed, construct
an update vector with the metadata blob.

dtype_t::metadata_blob_init(): Initialize the metadata BLOB data type.

dict_table_t::find(): Find an old column based on a new column number.

rec_convert_dtuple_to_metadata_comp(): Convert an alter metadata tuple
into a record.

row_rec_to_index_entry_impl(): Add the template parameter mblob
and the optional parameter info_bits for specifying the desired new
info bits. For the metadata tuple, allow conversion between the original
format (ADD COLUMN only) and the generic format (with hidden BLOB).

row_upd_index_replace_metadata(): Apply an update vector to an
alter_metadata tuple.

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

@@ -182,8 +182,8 @@ affected rows: 0
 info: Records: 0  Duplicates: 0  Warnings: 0
 INSERT INTO t1 SET id=9;
 ALTER TABLE t1 DROP c3;
-affected rows: 9
-info: Records: 9  Duplicates: 0  Warnings: 0
+affected rows: 0
+info: Records: 0  Duplicates: 0  Warnings: 0
 SHOW CREATE TABLE t1;
 Table	Create Table
 t1	CREATE TABLE `t1` (
@@ -599,8 +599,8 @@ affected rows: 0
 info: Records: 0  Duplicates: 0  Warnings: 0
 INSERT INTO t1 SET id=9;
 ALTER TABLE t1 DROP c3;
-affected rows: 9
-info: Records: 9  Duplicates: 0  Warnings: 0
+affected rows: 0
+info: Records: 0  Duplicates: 0  Warnings: 0
 SHOW CREATE TABLE t1;
 Table	Create Table
 t1	CREATE TABLE `t1` (
@@ -1016,8 +1016,8 @@ affected rows: 0
 info: Records: 0  Duplicates: 0  Warnings: 0
 INSERT INTO t1 SET id=9;
 ALTER TABLE t1 DROP c3;
-affected rows: 9
-info: Records: 9  Duplicates: 0  Warnings: 0
+affected rows: 0
+info: Records: 0  Duplicates: 0  Warnings: 0
 SHOW CREATE TABLE t1;
 Table	Create Table
 t1	CREATE TABLE `t1` (
@@ -1310,5 +1310,5 @@ SELECT variable_value-@old_instant instants
 FROM information_schema.global_status
 WHERE variable_name = 'innodb_instant_alter_column';
 instants
-54
+57
 SET GLOBAL innodb_purge_rseg_truncate_frequency= @saved_frequency;

storage/innobase/btr/btr0cur.cc

@@ -4726,8 +4726,14 @@ btr_cur_pessimistic_update(
 		rec, index, *offsets, page_is_leaf(page),
 		ULINT_UNDEFINED, offsets_heap);
 
-	dtuple_t*	new_entry = row_rec_to_index_entry(
-		rec, index, *offsets, &n_ext, entry_heap);
+	const bool is_metadata = rec_is_metadata(rec, index);
+	ut_ad(!is_metadata || (flags & BTR_NO_LOCKING_FLAG));
+
+	dtuple_t* new_entry = is_metadata
+		? row_metadata_to_tuple(rec, index, *offsets, &n_ext,
+					entry_heap, update->info_bits)
+		: row_rec_to_index_entry(rec, index, *offsets, &n_ext,
+					 entry_heap);
 
 	/* The page containing the clustered index record
 	corresponding to new_entry is latched in mtr.  If the
@@ -4739,9 +4745,6 @@ btr_cur_pessimistic_update(
 						     entry_heap);
 	btr_cur_trim(new_entry, index, update, thr);
 
-	const bool is_metadata = new_entry->info_bits
-		& REC_INFO_MIN_REC_FLAG;
-
 	/* We have to set appropriate extern storage bits in the new
 	record to be inserted: we have to remember which fields were such */
 
@@ -4769,10 +4772,10 @@ btr_cur_pessimistic_update(
 	}
 
 	if (page_zip_rec_needs_ext(
-			rec_get_converted_size(index, new_entry, n_ext),
-			page_is_comp(page),
-			dict_index_get_n_fields(index),
-			block->page.size)
+		    rec_get_converted_size(index, new_entry, n_ext),
+		    page_is_comp(page),
+		    dict_index_get_n_fields(index),
+		    block->page.size)
 	    || UNIV_UNLIKELY(rec_is_alter_metadata(rec, index))) {
 
 		big_rec_vec = dtuple_convert_big_rec(index, update, new_entry, &n_ext);
@@ -4838,7 +4841,7 @@ btr_cur_pessimistic_update(
 		ut_ad(new_entry->is_metadata());
 		ut_ad(index->is_instant());
 		/* This can be innobase_add_instant_try() performing a
-		subsequent instant ADD COLUMN, or its rollback by
+		subsequent instant ALTER TABLE, or its rollback by
 		row_undo_mod_clust_low(). */
 		ut_ad(flags & BTR_NO_LOCKING_FLAG);
 	} else {

storage/innobase/dict/dict0mem.cc

@@ -1291,96 +1291,197 @@ void dict_index_t::remove_instant()
 	table->instant = NULL;
 }
 
-/** Adjust clustered index metadata for instant ADD/DROP COLUMN.
-@param[in]	instant		clustered index definition after instant ADD COLUMN
-@param[in]	n_newly_add	number of newly added columns
-@param[in]	n_newly_drop	number of newly dropped columns
-@param[in]	col_map		mapping of old table cols to new table cols */
-inline void dict_index_t::instant_op_field(
-	const dict_index_t&	instant,
-	ulint			n_newly_add,
-	ulint			n_newly_drop,
-	const ulint*		col_map)
+/** Adjust index metadata for instant ADD/DROP/reorder COLUMN.
+@param[in]	clustered index definition after instant ALTER TABLE */
+inline void dict_index_t::instant_add_field(const dict_index_t& instant)
 {
 	DBUG_ASSERT(is_primary());
 	DBUG_ASSERT(instant.is_primary());
-	DBUG_ASSERT(!instant.is_instant());
+	DBUG_ASSERT(!has_virtual());
+	DBUG_ASSERT(!instant.has_virtual());
+	DBUG_ASSERT(instant.n_core_fields == instant.n_fields);
 	DBUG_ASSERT(n_def == n_fields);
 	DBUG_ASSERT(instant.n_def == instant.n_fields);
-
 	DBUG_ASSERT(type == instant.type);
 	DBUG_ASSERT(trx_id_offset == instant.trx_id_offset);
 	DBUG_ASSERT(n_user_defined_cols == instant.n_user_defined_cols);
 	DBUG_ASSERT(n_uniq == instant.n_uniq);
+	DBUG_ASSERT(instant.n_fields >= n_fields);
+	DBUG_ASSERT(instant.n_nullable >= n_nullable);
+	DBUG_ASSERT(instant.n_core_fields >= n_core_fields);
+	DBUG_ASSERT(instant.n_core_null_bytes >= n_core_null_bytes);
+
+	/* instant will have all fields (including ones for columns
+	that have been or are being instantly dropped) in the same position
+	as this index. Fields for any added columns are appended at the end. */
+#ifndef DBUG_OFF
+	for (unsigned i = 0; i < n_fields; i++) {
+		DBUG_ASSERT(fields[i].same(instant.fields[i]));
+		DBUG_ASSERT(fields[i].col->is_nullable()
+			    == instant.fields[i].col->is_nullable());
+	}
+#endif
+	n_fields = instant.n_fields;
+	n_def = instant.n_def;
+	n_nullable = instant.n_nullable;
+	fields = static_cast<dict_field_t*>(
+		mem_heap_dup(heap, instant.fields, n_fields * sizeof *fields));
 
-	ulint	old_n_fields = n_fields;
+	ut_d(unsigned n_null = 0);
+	ut_d(unsigned n_dropped = 0);
 
-	n_fields = instant.n_fields + table->n_dropped();
-	n_def = instant.n_def + table->n_dropped();
+	for (unsigned i = 0; i < n_fields; i++) {
+		const dict_col_t* icol = instant.fields[i].col;
+		dict_field_t& f = fields[i];
+		ut_d(n_null += icol->is_nullable());
+		DBUG_ASSERT(!icol->is_virtual());
+		if (icol->is_dropped()) {
+			ut_d(n_dropped++);
+			f.col->dropped = true;
+			f.col->ind = 0;
+			f.name = NULL;
+		} else {
+			f.col = &table->cols[icol - instant.table->cols];
+			f.name = f.col->name(*table);
+		}
+	}
 
-	unsigned	n_null = 0;
-	ulint		old_field_no = 0;
-	ulint		new_field_no = 0;
+	ut_ad(n_null == n_nullable);
+	ut_ad(n_dropped == instant.table->n_dropped());
+}
 
-	dict_field_t*	temp_fields = static_cast<dict_field_t*>(mem_heap_zalloc(
-				heap, n_fields * sizeof *temp_fields));
+/** Adjust table metadata for instant ADD/DROP/reorder COLUMN.
+@param[in]	table		altered table (with dropped columns)
+@param[in]	map		mapping from cols[] to table.cols[] */
+void dict_table_t::instant_column(const dict_table_t& table, const ulint* map)
+{
+	DBUG_ASSERT(!table.cached);
+	DBUG_ASSERT(table.n_def == table.n_cols);
+	DBUG_ASSERT(table.n_t_def == table.n_t_cols);
+	DBUG_ASSERT(n_def == n_cols);
+	DBUG_ASSERT(n_t_def == n_t_cols);
+	DBUG_ASSERT(n_v_def == n_v_cols);
+	DBUG_ASSERT(table.n_v_def == table.n_v_cols);
+	DBUG_ASSERT(table.n_cols + table.n_dropped() >= n_cols + n_dropped());
+	ut_ad(mutex_own(&dict_sys->mutex));
 
-	for (unsigned i = 0; i < n_fields; i++) {
-		const dict_field_t* field = &fields[old_field_no];
-		const dict_field_t& instant_field
-			= instant.fields[new_field_no];
-		dict_field_t&	temp_field = temp_fields[i];
-		bool		is_dropped = false;
-		ulint		new_col_offset = 0;
-
-		for (unsigned j = 0; j < table->n_dropped(); j++) {
-			dict_col_t* col = &table->instant->dropped[j];
-			if (col->ind == i) {
-				temp_field.col = col;
-				is_dropped = true;
-				break;
-			}
+	const char* end = table.col_names;
+	for (unsigned i = table.n_cols; i--; ) end += strlen(end) + 1;
+
+	col_names = static_cast<char*>(
+		mem_heap_dup(heap, table.col_names,
+			     ulint(end - table.col_names)));
+	const dict_col_t* const old_cols = cols;
+	const dict_col_t* const old_cols_end = cols + n_cols;
+	cols = static_cast<dict_col_t*>(mem_heap_dup(heap, table.cols,
+						     table.n_cols
+						     * sizeof *cols));
+
+	/* Preserve the default values of previously instantly added
+	columns, or copy the new default values to this->heap. */
+	for (ulint i = 0; i < ulint(table.n_cols); i++) {
+		dict_col_t& c = cols[i];
+
+		if (const dict_col_t* old = dict_table_t::find(old_cols, map,
+							       n_cols, i)) {
+			c.def_val = old->def_val;
+			continue;
 		}
 
-		if (is_dropped) {
-			old_field_no++;
-			temp_field.fixed_len = temp_field.col->len;
+		DBUG_ASSERT(c.is_added());
+		if (c.def_val.len <= sizeof field_ref_zero
+		    && !memcmp(c.def_val.data, field_ref_zero,
+			       c.def_val.len)) {
+			c.def_val.data = field_ref_zero;
+		} else if (const void*& d = c.def_val.data) {
+			d = mem_heap_dup(heap, d, c.def_val.len);
 		} else {
+			DBUG_ASSERT(c.def_val.len == UNIV_SQL_NULL);
+		}
+	}
 
-			if (i >= old_n_fields) {
+	n_t_def += table.n_cols - n_cols;
+	n_t_cols += table.n_cols - n_cols;
+	n_def = table.n_cols;
 
-				new_col_offset = i - DATA_N_SYS_COLS
-					- table->n_dropped();
+	/* FIXME: copy v_col_names[] and v_cols[] from table to this
+	(that is, support ADD/DROP/reorder/rename of VIRTUAL COLUMN).
+	Here, we wrongly assume that the virtual columns were unchanged. */
+	DBUG_ASSERT(n_v_def == table.n_v_def);
 
-				if (!dict_index_is_auto_gen_clust(this)) {
-					new_col_offset += 1;
-				}
+	for (unsigned i = 0; i < n_v_def; i++) {
+		dict_v_col_t& v = v_cols[i];
+		v.m_col.ind = (n_def - DATA_N_SYS_COLS) + i;
 
-				field = &instant_field;
+		for (ulint n = v.num_base; n--; ) {
+			dict_col_t*& base = v.base_col[n];
+			if (base->is_virtual()) {
+			} else if (base >= table.cols
+				   && base <= table.cols + table.n_cols) {
+				/* The base column was instantly added. */
+				base += cols - table.cols;
 			} else {
-				new_col_offset = col_map[field->col->ind];
-				old_field_no++;
+				DBUG_ASSERT(base >= old_cols);
+				DBUG_ASSERT(base + DATA_N_SYS_COLS
+					    <= old_cols_end);
+				size_t n = map[base - old_cols];
+				DBUG_ASSERT(n + DATA_N_SYS_COLS < n_cols);
+				base = &cols[n];
 			}
+		}
+	}
 
-			new_field_no++;
+	n_cols = table.n_cols;
+
+	dict_index_t* index = dict_table_get_first_index(this);
 
-			temp_fields[i].prefix_len = field->prefix_len;
-			temp_fields[i].fixed_len = field->fixed_len;
+	index->instant_add_field(*dict_table_get_first_index(&table));
 
-			const char* end = table->col_names;
-			for (unsigned col = 0; col < new_col_offset; col++) {
-				end += strlen(end) + 1;
+	if (instant || table.instant) {
+		unsigned num_non_pk = index->n_fields - (index->n_uniq + 2);
+		unsigned* non_pk_col_map = static_cast<unsigned*>(
+			mem_heap_zalloc(heap, num_non_pk
+					* sizeof *non_pk_col_map));
+		const uint u = index->n_uniq + 2;
+		for (unsigned i = 0; i + u < index->n_fields; i++) {
+			const dict_col_t* col = dict_index_get_nth_col(
+				index, i + u);
+			if (!col->is_dropped()) {
+				non_pk_col_map[i] = col->ind + 1;
 			}
+		}
 
-			temp_fields[i].name = end;
-			temp_fields[i].col = &table->cols[new_col_offset];
+		if (!instant) {
+			instant = new (mem_heap_zalloc(heap, sizeof *instant))
+				dict_instant_t();
 		}
 
-		n_null += temp_fields[i].col->is_nullable();
+		instant->non_pk_col_map = non_pk_col_map;
 	}
 
-	fields = temp_fields;
-	n_nullable = n_null;
+	while ((index = dict_table_get_next_index(index)) != NULL) {
+		if (index->to_be_dropped) {
+			continue;
+		}
+		for (unsigned i = 0; i < index->n_fields; i++) {
+			dict_field_t& field = index->fields[i];
+			if (field.col >= table.cols
+			    && field.col <= table.cols + n_cols) {
+				/* This is an instantly added column
+				in a newly added index. */
+				DBUG_ASSERT(!field.col->is_virtual());
+				field.col += cols - table.cols;
+				field.name = field.col->name(*this);
+			} else if (field.col < old_cols
+				 || field.col >= old_cols_end) {
+				DBUG_ASSERT(field.col->is_virtual());
+			} else {
+				field.col = &cols[map[field.col - old_cols]];
+				DBUG_ASSERT(!field.col->is_virtual());
+				field.name = field.col->name(*this);
+			}
+		}
+	}
 }
 
 /** Read the metadata blob and fill the non primary fields,
@@ -1514,206 +1615,6 @@ void dict_table_t::construct_dropped_columns(const byte* data)
 	}
 }
 
-/** Adjust table metadata for instant drop operation.
-@param[in]	table		instant table
-@param[in]	col_map		mapping of old table columns to new table
-@param[in]	n_newly_drop	number of instant drop column */
-inline void dict_table_t::fill_dropped_column(
-	const dict_table_t&	table,
-	const ulint*		col_map,
-	ulint			n_newly_drop)
-{
-	unsigned j = instant ? instant->n_dropped : 0;
-	ulint	n_total_drop_col = j + n_newly_drop;
-	dict_col_t*	temp_drop_cols;
-
-	temp_drop_cols = static_cast<dict_col_t*>(mem_heap_zalloc(
-		heap, n_total_drop_col * sizeof(dict_col_t)));
-
-	if (j) {
-		memcpy(temp_drop_cols, instant->dropped, j
-		       * sizeof *instant->dropped);
-	}
-
-	for (unsigned i = 0; i < n_def; i++) {
-		if (col_map[i] != ULINT_UNDEFINED) {
-			continue;
-		}
-
-		dict_col_t&	drop_col = temp_drop_cols[j++];
-		dict_col_t	col = cols[i];
-
-		drop_col.dropped = true;
-
-		drop_col.ind = dict_col_get_clust_pos(
-				&cols[i], dict_table_get_first_index(this));
-
-		if (col.prtype & DATA_NOT_NULL) {
-			drop_col.prtype = DATA_NOT_NULL;
-		}
-
-		drop_col.len = dict_col_get_fixed_size(
-				&col, dict_table_is_comp(this));
-
-		if (drop_col.len == 0) {
-			drop_col.mtype = DATA_BINARY;
-		} else {
-			drop_col.mtype = DATA_FIXBINARY;
-		}
-	}
-
-	n_t_def -= n_newly_drop;
-	n_def -= n_newly_drop;
-	n_cols -= n_newly_drop;
-	n_t_cols -= n_newly_drop;
-
-	if (!instant) {
-		instant = new (mem_heap_zalloc(heap, sizeof *instant))
-			dict_instant_t();
-	}
-
-	instant->n_dropped = n_total_drop_col;
-	instant->dropped = temp_drop_cols;
-}
-
-/** Adjust table metadata for instant operation.
-@param[in]	table		instant table
-@param[in]	col_map		mapping of old table cols to
-				new table cols
-@param[in]	n_newly_add	number of newly added column
-@param[in]	n_newly_drop	number of newly dropped column */
-void dict_table_t::instant_op_column(
-	const dict_table_t&	table,
-	const ulint*		col_map,
-	ulint			n_newly_add,
-	ulint			n_newly_drop)
-{
-	DBUG_ASSERT(!table.cached);
-	DBUG_ASSERT(table.n_def == table.n_cols);
-	DBUG_ASSERT(table.n_t_def == table.n_t_cols);
-	DBUG_ASSERT(n_def == n_cols);
-	DBUG_ASSERT(n_t_def == n_t_cols);
-	ut_ad(mutex_own(&dict_sys->mutex));
-
-	ulint	old_n_cols = n_cols;
-
-	if (n_newly_drop > 0) {
-		fill_dropped_column(table, col_map, n_newly_drop);
-	}
-
-	const char* end = table.col_names;
-	for (unsigned i = table.n_cols; i--; ) end += strlen(end) + 1;
-
-	col_names = static_cast<char*>(
-		mem_heap_dup(heap, table.col_names,
-			     ulint(end - table.col_names)));
-	const dict_col_t* const old_cols = cols;
-	const dict_col_t* const old_cols_end = cols + n_cols;
-	cols = static_cast<dict_col_t*>(mem_heap_dup(heap, table.cols,
-						     table.n_cols
-						     * sizeof *cols));
-
-	/* Preserve the default values of previously instantly
-	added columns. */
-	for (unsigned i = unsigned(old_n_cols - DATA_N_SYS_COLS); i--; ) {
-		if (col_map[i] == ULINT_UNDEFINED) {
-			continue;
-		}
-
-		ulint	new_offset = col_map[i];
-		cols[new_offset].def_val = old_cols[i].def_val;
-	}
-
-	/* Copy the new default values to this->heap. */
-	for (unsigned i = n_cols; i < table.n_cols; i++) {
-		dict_col_t& c = cols[i - DATA_N_SYS_COLS];
-		/* FIXME: Allow adding columns other than LAST,
-		and take col_map[] into account. */
-		DBUG_ASSERT(c.is_added());
-		if (c.def_val.len == 0) {
-			c.def_val.data = field_ref_zero;
-		} else if (const void*& d = c.def_val.data) {
-			d = mem_heap_dup(heap, d, c.def_val.len);
-		} else {
-			DBUG_ASSERT(c.def_val.len == UNIV_SQL_NULL);
-		}
-	}
-
-	const unsigned n_add = unsigned(table.n_cols -
-					(old_n_cols - n_newly_drop));
-
-	n_t_def += n_add;
-	n_t_cols += n_add;
-	n_cols = table.n_cols;
-	n_def = n_cols;
-
-	for (unsigned i = 0; i < n_v_def; i++) {
-		dict_v_col_t& v = v_cols[i];
-		v.m_col.ind = (n_def - DATA_N_SYS_COLS) + i;
-
-		for (ulint n = v.num_base; n--; ) {
-			dict_col_t*& base = v.base_col[n];
-			if (!base->is_virtual()) {
-				DBUG_ASSERT(base >= old_cols);
-				size_t n = col_map[(base - old_cols)];
-				DBUG_ASSERT(n + DATA_N_SYS_COLS < old_n_cols);
-				base = &cols[n];
-			}
-		}
-	}
-
-	dict_index_t* index = dict_table_get_first_index(this);
-
-	index->instant_op_field(*dict_table_get_first_index(&table),
-				n_newly_add, n_newly_drop, col_map);
-
-	if (instant || n_newly_drop || n_dropped()) {
-		unsigned num_non_pk = index->n_fields - (index->n_uniq + 2);
-		unsigned* non_pk_col_map = static_cast<unsigned*>(
-			mem_heap_zalloc(heap, num_non_pk
-					* sizeof *non_pk_col_map));
-		for (unsigned i = index->n_uniq + 2, j = 0;
-		     i < index->n_fields; i++) {
-			if (!dict_index_get_nth_col(index, i)->is_dropped()) {
-				non_pk_col_map[j] =
-					dict_index_get_nth_col_no(index, i) + 1;
-			}
-
-			j++;
-		}
-
-		if (!instant) {
-			instant = new (mem_heap_zalloc(heap, sizeof *instant))
-				dict_instant_t();
-		}
-
-		instant->non_pk_col_map = non_pk_col_map;
-	}
-
-	while ((index = dict_table_get_next_index(index)) != NULL) {
-		if (index->to_be_dropped) {
-			continue;
-		}
-		for (unsigned i = 0; i < index->n_fields; i++) {
-			dict_field_t& field = index->fields[i];
-			if (field.col < old_cols
-			    || field.col >= old_cols_end) {
-				DBUG_ASSERT(field.col->is_virtual());
-			} else {
-				/* Secondary indexes may contain user
-				columns and DB_ROW_ID (if there is
-				GEN_CLUST_INDEX instead of PRIMARY KEY),
-				but not DB_TRX_ID,DB_ROLL_PTR. */
-				DBUG_ASSERT(field.col >= old_cols);
-				ulint n = col_map[field.col - old_cols];
-				field.col = &cols[n];
-				DBUG_ASSERT(!field.col->is_virtual());
-				field.name = field.col->name(*this);
-			}
-		}
-	}
-}
-
 /** Find the old column number for the given new column position.
 @param[in]	col_map	column map from old column to new column
 @param[in]	pos	new column position
@@ -1734,7 +1635,7 @@ ulint find_old_col_no(
 	return ULINT_UNDEFINED;
 }
 
-/** Roll back instant_op_column().
+/** Roll back instant_column().
 @param[in]	old_n_cols		original n_cols
 @param[in]	old_cols		original cols
 @param[in]	old_col_names		original col_names

storage/innobase/handler/handler0alter.cc

@@ -171,7 +171,7 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 	dict_table_t*	old_table;
 	/** table where the indexes are being created or dropped */
 	dict_table_t*	new_table;
-	/** table definition for instant ADD COLUMN */
+	/** table definition for instant ADD/DROP/reorder COLUMN */
 	dict_table_t*	instant_table;
 	/** mapping of old column numbers to new ones, or NULL */
 	const ulint*	col_map;
@@ -219,11 +219,6 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 	/** The page_compression_level attribute, or 0 */
 	const uint	page_compression_level;
 
-	/** Number of instant add columns */
-	ulint		n_instant_add_cols;
-	/** Number of instant drop columns */
-	ulint		n_instant_drop_cols;
-
 	ha_innobase_inplace_ctx(row_prebuilt_t*& prebuilt_arg,
 				dict_index_t** drop_arg,
 				ulint num_to_drop_arg,
@@ -277,9 +272,7 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 				       ? (page_compression_level_arg
 					  ? uint(page_compression_level_arg)
 					  : page_zip_level)
-				       : 0),
-		n_instant_add_cols(0),
-		n_instant_drop_cols(0)
+				       : 0)
 	{
 		ut_ad(old_n_cols >= DATA_N_SYS_COLS);
 		ut_ad(page_compression_level <= 9);
@@ -330,13 +323,136 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 	{
 		DBUG_ASSERT(need_rebuild());
 		DBUG_ASSERT(!is_instant());
+		DBUG_ASSERT(!new_table->is_instant());
+		DBUG_ASSERT(new_table->n_dropped() == 0);
 		DBUG_ASSERT(old_table->n_cols == old_table->n_def);
 		DBUG_ASSERT(new_table->n_cols == new_table->n_def);
 		DBUG_ASSERT(old_table->n_cols == old_n_cols);
-		instant_table = new_table;
 
+		instant_table = new_table;
 		new_table = old_table;
 		export_vars.innodb_instant_alter_column++;
+
+		if (old_table->instant) {
+add_metadata:
+			const unsigned n_old_drop = old_table->n_dropped();
+			unsigned n_drop = n_old_drop;
+			for (unsigned i = old_table->n_cols; i--; ) {
+				if (col_map[i] == ULINT_UNDEFINED) {
+					DBUG_ASSERT(i + DATA_N_SYS_COLS
+						    < uint(old_table->n_cols));
+					n_drop++;
+				}
+			}
+
+			instant_table->instant = new
+				(mem_heap_alloc(instant_table->heap,
+						sizeof(dict_instant_t)))
+				dict_instant_t();
+			instant_table->instant->n_dropped = n_drop;
+			instant_table->instant->dropped
+				= static_cast<dict_col_t*>(
+					mem_heap_alloc(instant_table->heap,
+						       n_drop
+						       * sizeof(dict_col_t)));
+			if (n_old_drop) {
+				memcpy(instant_table->instant->dropped,
+				       old_table->instant->dropped,
+				       n_old_drop * sizeof(dict_col_t));
+			}
+
+			unsigned d = n_old_drop;
+
+			for (unsigned i = 0; i < old_table->n_cols; i++) {
+				if (col_map[i] == ULINT_UNDEFINED) {
+					dict_col_t* drop = new
+						(&instant_table->instant
+						 ->dropped[d++])
+						dict_col_t(old_table->cols[i]);
+					drop->dropped = true;
+					drop->ind = 0;
+				}
+			}
+#ifndef DBUG_OFF
+			for (unsigned i = 0; i < n_drop; i++) {
+				DBUG_ASSERT(instant_table->instant
+					    ->dropped[i].is_dropped());
+			}
+#endif
+			DBUG_ASSERT(d == n_drop);
+			dict_index_t* old = UT_LIST_GET_FIRST(
+				old_table->indexes);
+			dict_index_t* instant = UT_LIST_GET_FIRST(
+				instant_table->indexes);
+			const uint n_fields = instant->n_fields
+				+ instant_table->n_dropped();
+
+			DBUG_ASSERT(n_fields
+				    >= old->n_fields + old_table->n_dropped());
+			dict_field_t* fields = static_cast<dict_field_t*>(
+				mem_heap_zalloc(instant_table->heap,
+						n_fields * sizeof *fields));
+			d = n_old_drop;
+			uint j = 0;
+			for (uint i = 0; i < n_fields; i++) {
+				DBUG_ASSERT(j <= i);
+				if (i >= old->n_fields) {
+existing_field:
+					fields[i] = instant->fields[j++];
+					DBUG_ASSERT(!fields[i].col
+						    ->is_dropped());
+					DBUG_ASSERT(fields[i].name
+						    == fields[i].col->name(
+							    *instant_table));
+					continue;
+				}
+
+				dict_field_t&f = fields[i] = old->fields[i];
+				if (f.col->is_dropped()) {
+					/* The column has been instantly
+					dropped earlier. */
+					DBUG_ASSERT(f.col >= old_table->instant
+						    ->dropped);
+					DBUG_ASSERT(f.col < old_table->instant
+						    ->dropped + n_old_drop);
+					f.col += instant_table->instant->dropped
+						- old_table->instant->dropped;
+					f.name = f.col->name(*instant_table);
+					continue;
+				}
+
+				if (col_map[f.col->ind] != ULINT_UNDEFINED) {
+					DBUG_ASSERT(col_map[f.col->ind]
+						    == instant->fields[j].col
+						    ->ind);
+					goto existing_field;
+				}
+
+				/* This column is being dropped. */
+				DBUG_ASSERT(d < n_drop);
+				f.col = &instant_table->instant->dropped[d++];
+				f.name = NULL;
+			}
+
+			DBUG_ASSERT(d == n_drop);
+			DBUG_ASSERT(j == instant->n_fields);
+			instant->n_fields = instant->n_core_fields =
+				instant->n_def = n_fields;
+			instant->fields = fields;
+		} else {
+			for (unsigned i = old_table->n_cols - DATA_N_SYS_COLS;
+			     i--; ) {
+				if (col_map[i] != i) {
+					goto add_metadata;
+				}
+			}
+		}
+
+		DBUG_ASSERT(instant_table->n_cols
+			    + instant_table->n_dropped()
+			    >= old_table->n_cols + old_table->n_dropped());
+		DBUG_ASSERT(instant_table->n_dropped()
+			    >= old_table->n_dropped());
 	}
 
 	/** Revert prepare_instant() if the transaction is rolled back. */
@@ -4541,15 +4657,14 @@ static bool innobase_instant_try(
 
 	dict_table_t* user_table = ctx->old_table;
 
-	unsigned n_old_drop_cols = user_table->instant
-		? user_table->instant->n_dropped : 0;
-	ulint	n_old_fields = dict_table_get_first_index(user_table)->n_fields;
+	dict_index_t* index = dict_table_get_first_index(user_table);
+	uint n_old_fields = index->n_fields;
+	const dict_col_t* old_cols = user_table->cols;
+	DBUG_ASSERT(user_table->n_cols == ctx->old_n_cols);
 
-	user_table->instant_op_column(
-			*ctx->instant_table, ctx->col_map,
-			ctx->n_instant_add_cols, ctx->n_instant_drop_cols);
+	user_table->instant_column(*ctx->instant_table, ctx->col_map);
 
-	dict_index_t* index = dict_table_get_first_index(user_table);
+	DBUG_ASSERT(index->n_fields >= n_old_fields);
 	/* The table may have been emptied and may have lost its
 	'instantness' during this ALTER TABLE. */
 
@@ -4558,60 +4673,36 @@ static bool innobase_instant_try(
 	dict_table_copy_types(row, user_table);
 	Field** af = altered_table->field;
 	Field** const end = altered_table->field + altered_table->s->fields;
-	List_iterator_fast<Create_field> cf_it(
-		ha_alter_info->alter_info->create_list);
+	ut_d(List_iterator_fast<Create_field> cf_it(
+		     ha_alter_info->alter_info->create_list));
 	bool rebuild_metadata = false;
-	for (uint i = 0, dropped = 0;;) {
-		if (af < end && !(*af)->stored_in_db()) {
-			af++; cf_it++;
+	for (uint i = 0; af < end; af++) {
+		if (!(*af)->stored_in_db()) {
+			ut_d(cf_it++);
 			continue;
 		}
 
-		if (i + ctx->n_instant_add_cols >= user_table->n_cols) {
-			/* Avoid out-of-bounds access of ctx->col_map[].
-			It is only defined for the pre-existing columns
-			of the table, excluding any ADD COLUMN of this
-			ALTER TABLE operation. */
-		} else if (ctx->col_map[i] != i) {
-			const bool not_last = user_table->n_cols
-				+ ctx->n_instant_drop_cols > i
-				+ ctx->n_instant_add_cols + DATA_N_SYS_COLS;
-			/* System columns cannot be dropped. */
-			DBUG_ASSERT(not_last
-				    || ctx->col_map[i] != ULINT_UNDEFINED);
-			if (!rebuild_metadata && not_last) {
-				rebuild_metadata = true;
-				if (innobase_instant_drop_cols(user_table->id,
-							       i, trx)) {
-					return true;
-				}
-			}
-
-			if (ctx->col_map[i] == ULINT_UNDEFINED) {
-				DBUG_ASSERT(dropped
-					    < user_table->instant->n_dropped);
-				DBUG_ASSERT(user_table->instant->dropped[
-						    dropped].is_dropped());
-				dropped++;
-				i++;
-				continue;
+		const dict_col_t* old = dict_table_t::find(old_cols,
+							   ctx->col_map,
+							   ctx->old_n_cols, i);
+		if (old && i < ctx->old_n_cols - DATA_N_SYS_COLS
+		    && old->ind != i) {
+			if (!rebuild_metadata
+			    && innobase_instant_drop_cols(user_table->id,
+							  i, trx)) {
+				return true;
 			}
+			rebuild_metadata = true;
 		}
 
-		const uint c = i - dropped;
-		dfield_t* d = dtuple_get_nth_field(row, c);
-		const dict_col_t* col = dict_table_get_nth_col(user_table, c);
+		dfield_t* d = dtuple_get_nth_field(row, i);
+		const dict_col_t* col = dict_table_get_nth_col(user_table, i);
 		DBUG_ASSERT(!col->is_virtual());
 		DBUG_ASSERT(!col->is_dropped());
-
-		if (col->mtype == DATA_SYS) {
-			DBUG_ASSERT(!col->is_added());
-			break;
-		}
-
-		DBUG_ASSERT(af < end);
+		DBUG_ASSERT(col->mtype != DATA_SYS);
 		DBUG_ASSERT(!strcmp((*af)->field_name.str,
-				    dict_table_get_col_name(user_table, c)));
+				    dict_table_get_col_name(user_table, i)));
+		DBUG_ASSERT(old || col->is_added());
 
 		if (col->is_added()) {
 			dfield_set_data(d, col->def_val.data,
@@ -4649,22 +4740,20 @@ static bool innobase_instant_try(
 			}
 		}
 
-		const Create_field* new_field = cf_it++;
+		ut_d(const Create_field* new_field = cf_it++);
 		/* new_field->field would point to an existing column.
 		If it is NULL, the column was added by this ALTER TABLE. */
-		DBUG_ASSERT(new_field->field || col->is_added());
+		ut_ad(!new_field->field == !old);
 
-		if (!rebuild_metadata && new_field->field) {
+		if (!rebuild_metadata && old) {
 			/* The record is already present in SYS_COLUMNS. */
-			DBUG_ASSERT(!col->is_dropped());
-		} else if (innobase_instant_add_col(user_table->id, c,
+		} else if (innobase_instant_add_col(user_table->id, i,
 						    (*af)->field_name.str,
 						    d->type, trx)) {
 			return true;
 		}
 
 		i++;
-		af++;
 	}
 
 	if (innodb_update_cols(user_table, dict_table_encode_n_col(
@@ -4748,46 +4837,38 @@ static bool innobase_instant_try(
 			goto empty_table;
 		}
 
-		/* Extend the record with the instantly added columns. */
-		unsigned n = user_table->instant
-			? ctx->n_instant_add_cols + ctx->n_instant_drop_cols
-			: user_table->n_cols - ctx->old_n_cols;
+		/* Extend the record with any added columns. */
+		uint n = uint(index->n_fields) - n_old_fields;
 		/* Reserve room for DB_TRX_ID,DB_ROLL_PTR and any
 		non-updated off-page columns in case they are moved off
 		page as a result of the update. */
 		const unsigned f = user_table->instant != NULL;
-		upd_t* update = upd_create(index->n_fields, ctx->heap);
-		update->n_fields = n;
+		upd_t* update = upd_create(index->n_fields + f, ctx->heap);
+		update->n_fields = n + f;
 		update->info_bits = f
 			? REC_INFO_METADATA_ALTER
 			: REC_INFO_METADATA_ADD;
+		if (f) {
+			upd_field_t* uf = upd_get_nth_field(update, 0);
+			uf->field_no = index->n_uniq + 2;
+			uf->new_val = entry->fields[uf->field_no];
+			DBUG_ASSERT(dfield_is_ext(&uf->new_val));
+			DBUG_ASSERT(dfield_get_len(&uf->new_val)
+				    == BTR_EXTERN_FIELD_REF_SIZE);
+		}
 
 		/* Add the default values for instantly added columns */
-		unsigned j = 0;
-		unsigned drop_cols_offset = n_old_drop_cols;
-
-		for (unsigned i = 0; i < ctx->old_n_cols; i++) {
-			if (ctx->col_map[i] == ULINT_UNDEFINED) {
-				unsigned field_no = user_table->instant
-					->dropped[drop_cols_offset++].ind;
-				upd_field_t* uf = upd_get_nth_field(
-					update, j++);
-				uf->field_no = field_no;
-				uf->new_val = entry->fields[field_no + f];
-			}
-
-			ut_ad(j <= n);
-		}
+		unsigned j = f;
 
 		for (unsigned k = n_old_fields; k < index->n_fields; k++) {
 			upd_field_t* uf = upd_get_nth_field(update, j++);
-			uf->field_no = k;
+			uf->field_no = k + f;
 			uf->new_val = entry->fields[k + f];
 
-			ut_ad(j <= n);
+			ut_ad(j <= n + f);
 		}
 
-		ut_ad(j == n);
+		ut_ad(j == n + f);
 
 		ulint* offsets = NULL;
 		mem_heap_t* offsets_heap = NULL;
@@ -5461,23 +5542,7 @@ prepare_inplace_alter_table_dict(
 		if (!instant_alter_column_possible(ha_alter_info, old_table)
 		    || (innobase_fulltext_exist(old_table)
 			&& !innobase_fulltext_exist(altered_table))) {
-			goto not_instant_add_column;
-		}
-
-		for (int i = 0; i < ctx->old_table->n_cols; i++) {
-			if (ctx->col_map[i] == ULINT_UNDEFINED) {
-				ctx->n_instant_drop_cols++;
-			}
-		}
-
-		int n_exist_old_col = user_table->n_cols + user_table->n_v_cols
-					- ctx->n_instant_drop_cols;
-
-		for (int i = n_exist_old_col;
-		     i < ctx->new_table->n_cols + ctx->new_table->n_v_cols;
-		     i++) {
-			ut_ad(ctx->col_map[i] != ULINT_UNDEFINED);
-			ctx->n_instant_add_cols++;
+			goto not_instant_column;
 		}
 
 		for (uint a = 0; a < ctx->num_to_add_index; a++) {
@@ -5621,7 +5686,7 @@ prepare_inplace_alter_table_dict(
 	}
 
 	if (ctx->need_rebuild()) {
-not_instant_add_column:
+not_instant_column:
 		DBUG_ASSERT(ctx->need_rebuild());
 		DBUG_ASSERT(!ctx->is_instant());
 		DBUG_ASSERT(num_fts_index <= 1);

storage/innobase/include/data0data.h

@@ -646,9 +646,11 @@ struct dtuple_t {
 	@param[in]	index	index possibly with instantly added columns */
 	void trim(const dict_index_t& index);
 
-	/** @return whether this is a hidden metadata record
-	for instant ALTER TABLE (not only ADD COLUMN) */
-	bool is_alter_metadata() const
+	/**
+	@param info_bits	the info_bits of a data tuple
+	@return whether this is a hidden metadata record
+	for instant ADD COLUMN or ALTER TABLE */
+	static bool is_alter_metadata(ulint info_bits)
 	{
 		return UNIV_UNLIKELY(info_bits == REC_INFO_METADATA_ALTER);
 	}
@@ -663,6 +665,10 @@ struct dtuple_t {
 				     == REC_INFO_METADATA_ADD);
 	}
 
+	/** @return whether this is a hidden metadata record
+	for instant ALTER TABLE (not only ADD COLUMN) */
+	bool is_alter_metadata() const { return is_alter_metadata(info_bits); }
+
 	/** @return whether this is a hidden metadata record
 	for instant ADD COLUMN or ALTER TABLE */
 	bool is_metadata() const { return is_metadata(info_bits); }

storage/innobase/include/data0type.h

@@ -554,6 +554,16 @@ struct dtype_t{
 	{
 		return (prtype & DATA_VERSIONED) == DATA_VERS_END;
 	}
+
+	/** Set the type of the BLOB in the hidden metadata record. */
+	void metadata_blob_init()
+	{
+		prtype = DATA_NOT_NULL;
+		mtype = DATA_BLOB;
+		len = 0;
+		mbminlen = 0;
+		mbmaxlen = 0;
+	}
 };
 
 /** The DB_TRX_ID,DB_ROLL_PTR values for "no history is available" */

storage/innobase/include/dict0mem.h

@@ -1156,18 +1156,9 @@ struct dict_index_t {
 		return fields[n].col->instant_value(len);
 	}
 
-	/** Adjust clustered index metadata for instant ADD/DROP COLUMN.
-	@param[in]	instant	clustered index definition
-					after instant ADD/DROP COLUM
-	@param[in]	n_newly_add	number of newly added columns
-	@param[in]	n_newly_drop	number of newly dropped columns
-	@param[in]	col_map	mapping of old table cols
-					to new table cols */
-	inline void instant_op_field(
-		const dict_index_t&	instant,
-		ulint			n_newly_add,
-		ulint			n_newly_drop,
-		const ulint*		col_map);
+	/** Adjust index metadata for instant ADD/DROP/reorder COLUMN.
+	@param[in]	clustered index definition after instant ALTER TABLE */
+	inline void instant_add_field(const dict_index_t& instant);
 
 	/** Remove instant ALTER TABLE metadata.
 	Protected by index root page x-latch or table X-lock. */
@@ -1649,6 +1640,25 @@ struct dict_table_t {
 	/** @return the number of instantly dropped columns */
 	unsigned n_dropped() const { return instant ? instant->n_dropped : 0; }
 
+	/** Look up an old column.
+	@param[in]	cols	the old columns of the table
+	@param[in]	col_map	map from old table columns to altered ones
+	@param[in]	n_cols	number of old columns
+	@param[in]	i	the number of the new column
+	@return	old column
+	@retval	NULL	if column i was added to the table */
+	static const dict_col_t* find(const dict_col_t* cols,
+				      const ulint* col_map, ulint n_cols,
+				      ulint i)
+	{
+		for (ulint o = n_cols; o--; ) {
+			if (col_map[o] == i) {
+				return &cols[o];
+			}
+		}
+		return NULL;
+	}
+
 	/** Read the metadata blob and fill the non primary fields,
 	non-drop nullable fields and fill the drop columns in the
 	vector.
@@ -1675,29 +1685,12 @@ struct dict_table_t {
 	@param[in]	data	metadata blob data. */
 	void construct_dropped_columns(const byte* data);
 
-	/** Fill the dropped column in dropped_cols
-	@param[in]	table		instant table
-	@param[in]	col_map		mapping of cols from old table
-					to new table
-	@param[in]	n_newly_drop	number of newly drop column */
-	inline void fill_dropped_column(
-		const dict_table_t&	table,
-		const ulint*		col_map,
-		ulint			n_newly_drop);
-
-	/** Adjust table metadat for instant operation.
-	@param[in]	table		instant table
-	@param[in]	col_map		mapping of cols from old table
-					to new table
-	@param[in]	n_newly_add	number of newly added column
-	@param[in]	n_newly_drop	number of newly drop column */
-	void instant_op_column(
-		const dict_table_t&	table,
-		const ulint*		col_map,
-		ulint			n_newly_add,
-		ulint			n_newly_drop);
-
-	/** Roll back instant_op_column().
+	/** Adjust table metadata for instant ADD/DROP/reorder COLUMN.
+	@param[in]	table		altered table (with dropped columns)
+	@param[in]	map		mapping from cols[] to table.cols[] */
+	void instant_column(const dict_table_t& table, const ulint* map);
+
+	/** Roll back instant_column().
 	@param[in]	old_n_cols		original n_cols
 	@param[in]	old_cols		original cols
 	@param[in]	old_col_names		original col_names

storage/innobase/include/row0row.h

@@ -243,6 +243,24 @@ row_rec_to_index_entry(
 	mem_heap_t*		heap)	/*!< in: memory heap from which
 					the memory needed is allocated */
 	MY_ATTRIBUTE((warn_unused_result));
+
+/** Convert a metadata record to a data tuple.
+@param[in]	rec		metadata record
+@param[in]	index		clustered index after instant ALTER TABLE
+@param[in]	offsets		rec_get_offsets(rec)
+@param[out]	n_ext		number of externally stored fields
+@param[in,out]	heap		memory heap for allocations
+@param[in]	info_bits	the info_bits after an update */
+dtuple_t*
+row_metadata_to_tuple(
+	const rec_t*		rec,
+	const dict_index_t*	index,
+	const ulint*		offsets,
+	ulint*			n_ext,
+	mem_heap_t*		heap,
+	ulint			info_bits)
+	MY_ATTRIBUTE((nonnull,warn_unused_result));
+
 /*******************************************************************//**
 Builds from a secondary index record a row reference with which we can
 search the clustered index record.

storage/innobase/include/row0upd.h

@@ -493,6 +493,10 @@ struct upd_t{
 	/** @return whether this is for a hidden metadata record
 	for instant ALTER TABLE */
 	bool is_metadata() const { return dtuple_t::is_metadata(info_bits); }
+	/** @return whether this is for a hidden metadata record
+	for instant ALTER TABLE (not only ADD COLUMN) */
+	bool is_alter_metadata() const
+	{ return dtuple_t::is_alter_metadata(info_bits); }
 
 #ifdef UNIV_DEBUG
         bool validate() const

storage/innobase/rem/rem0rec.cc

@@ -1087,8 +1087,7 @@ rec_get_offsets_func(
 		      || n == n_fields /* btr_pcur_restore_position() */
 		      || (n + (index->id == DICT_INDEXES_ID)
 			  >= index->n_core_fields && n <= index->n_fields
-			  + (index->table->instant
-			     ? index->table->instant->n_dropped : 0)));
+			  + unsigned(rec_is_alter_metadata(rec, false))));
 
 		if (is_user_rec && leaf && n < index->n_fields) {
 			ut_ad(!index->is_dummy);
@@ -1782,33 +1781,30 @@ rec_convert_dtuple_to_rec_old(
 	return(rec);
 }
 
-/** Convert a data tuple into a default ROW_FORMAT=COMPACT record
+/** Convert a data tuple into a ROW_FORMAT=COMPACT metadata record
 for instant alter operation.
 @param[out]	rec		converted record
 @param[in]	clust_index	clustered index
 @param[in]	fields		data fields to convert
 @param[in]	n_fields	number of data fields */
-static
+ATTRIBUTE_COLD static
 void
-rec_convert_dtuple_to_default_rec_comp(
+rec_convert_dtuple_to_metadata_comp(
 	rec_t*			rec,
 	const dict_index_t*	clust_index,
 	const dfield_t*		fields,
 	ulint			n_fields)
 {
-	const dfield_t*	field;
-	const dtype_t*	type;
 	byte*		end;
 	byte*		nulls = rec - (REC_N_NEW_EXTRA_BYTES + 1);
 	byte*		UNINIT_VAR(lens);
-	ulint		len;
-	ulint		fixed_len;
 	ulint		null_mask	= 1;
 
 	/* Nullable non-dropped fields + number of dropped fields. */
 	unsigned	n_null_bytes = UT_BITS_IN_BYTES(
 		(clust_index->get_n_non_drop_nullable_fields()
 				 + clust_index->table->instant->n_dropped));
+	ut_ad(n_fields == ulint(clust_index->n_fields) + 1);
 
 	rec_set_n_add_field(nulls, n_null_bytes);
 
@@ -1822,41 +1818,24 @@ rec_convert_dtuple_to_default_rec_comp(
 	/* clear the SQL-null flags */
 	memset(lens + 1, 0, ulint(nulls - lens));
 
-	ulint 	i;
-	ulint	field_no = 0;
 	/* Store the data and the offsets */
-	for (i = 0; i < n_fields; i++) {
-		const dict_field_t*	ifield;
-		dict_col_t*		col = NULL;
-		bool			is_dropped = false;
-		bool			is_nullable = false;
+	for (ulint i = 0, j = 0; i < n_fields; i++) {
+		const dfield_t*	field = &fields[i];
+		ulint len = dfield_get_len(field);
+		ulint fixed_len;
 
-		field = &fields[i];
-		type = dfield_get_type(field);
-		len = dfield_get_len(field);
-
-		if (i == unsigned(clust_index->n_uniq + DATA_ROLL_PTR)) {
-			/* Default row blob. */
+		if (i == unsigned(clust_index->n_uniq + 2)) {
+			/* Metadata blob */
 			fixed_len = 0;
+			ut_ad(dfield_is_ext(field));
+			ut_ad(dtype_get_prtype(&field->type) & DATA_NOT_NULL);
 			goto data_write;
 		} else {
-			ifield = dict_index_get_nth_field(clust_index, field_no++);
-			fixed_len = ifield->fixed_len;
-			col = ifield->col;
-			is_dropped = col->is_dropped();
+			fixed_len = dict_index_get_nth_field(
+				clust_index, j++)->fixed_len;
 		}
 
-		is_nullable = !(dtype_get_prtype(type) & DATA_NOT_NULL);
-
-		if (is_dropped) {
-			if (is_nullable) {
-				/* Set the nullable flag for
-				dropped column */
-				*nulls |= null_mask;
-			}
-
-			null_mask <<= 1;
-		} else if (is_nullable) {
+		if (!(dtype_get_prtype(&field->type) & DATA_NOT_NULL)) {
 			if (dfield_is_null(field)) {
 				*nulls |= null_mask;
 				null_mask <<= 1;
@@ -1866,19 +1845,13 @@ rec_convert_dtuple_to_default_rec_comp(
 			null_mask <<= 1;
 		}
 
-		ut_ad(!dfield_is_null(field) || is_dropped);
-
-		/* Set the length for the dropped column. */
-		if (fixed_len && !is_dropped) {
-			ut_ad(len <= fixed_len);
-			ut_ad(!dfield_is_ext(field));
+		if (fixed_len) {
 		} else if (dfield_is_ext(field)) {
 			*lens-- = (byte) (len >> 8) | 0xc0;
 			*lens-- = (byte) len;
 		} else {
 			if (len < 128 || !DATA_BIG_LEN_MTYPE(
-				dtype_get_len(type), dtype_get_mtype(type))) {
-
+				    field->type.len, field->type.mtype)) {
 				*lens-- = (byte) len;
 			} else {
 				ut_ad(len < 16384);
@@ -1888,8 +1861,9 @@ rec_convert_dtuple_to_default_rec_comp(
 		}
 
 data_write:
-		/* Store the record only non-dropped column. */
-		if (len && !is_dropped) {
+		ut_ad(!fixed_len || len <= fixed_len);
+		ut_ad(!fixed_len || !dfield_is_ext(field));
+		if (len) {
 			memcpy(end, dfield_get_data(field), len);
 			end += len;
 		}
@@ -2113,7 +2087,7 @@ rec_convert_dtuple_to_rec_new(
 			  && dict_table_is_comp(index->table))) {
 		rec_get_metadata_converted_size(index, dtuple, &extra_size);
 		rec = buf + extra_size;
-		rec_convert_dtuple_to_default_rec_comp(
+		rec_convert_dtuple_to_metadata_comp(
 			rec, index, dtuple->fields, dtuple->n_fields);
 		rec_set_info_bits_new(
 			rec, dtuple->info_bits & ~REC_NEW_STATUS_MASK);

storage/innobase/row/row0row.cc

@@ -77,14 +77,13 @@ row_build_clust_default_entry(
 
 		dfield = dtuple_get_nth_field(entry, field_no);
 
-		if (field_no == clust_index->n_uniq + 2) {
+		if (field_no == ulint(clust_index->n_uniq) + 2) {
 			dfield_set_data(dfield, mem_heap_zalloc(
 						heap,
 						BTR_EXTERN_FIELD_REF_SIZE),
 					BTR_EXTERN_FIELD_REF_SIZE);
 			dfield_set_ext(dfield);
-			dfield->type.mtype = DATA_BLOB;
-			dfield->type.prtype = DATA_NOT_NULL;
+			dfield->type.metadata_blob_init();
 			continue;
 		}
 
@@ -776,115 +775,19 @@ row_build_w_add_vcol(
 			     defaults, add_v, col_map, ext, heap));
 }
 
-/** Convert the rec to new metadata with drop column info.
-@param[in]	rec	index record
-@param[in]	index	index
-@param[in]	offsets	rec_get_offsets(rec, index)
-@param[out]	n_ext	number of externally stored columns
-@param[in,out]	heap	memory heap for allocations
-@return index entry built; does not set info_bits, and the data fields
-in the entry will point directly to rec */
-static inline
-dtuple_t*
-row_def_rec_to_index_entry_impl(
-	const rec_t*		rec,
-	const dict_index_t*	index,
-	const ulint*		offsets,
-	ulint*			n_ext,
-	mem_heap_t*		heap)
-{
-	dtuple_t*		entry;
-	dfield_t*		dfield;
-	ulint			i;
-	const byte*		field;
-	ulint			len;
-	ulint			rec_len;
-	const dict_field_t*	ifield;
-	const dict_col_t*	col;
-
-	ut_ad(rec != NULL);
-	ut_ad(heap != NULL);
-	ut_ad(index != NULL);
-
-	/* Because this function may be invoked by row0merge.cc
-	on a record whose header is in different format, the check
-	rec_offs_validate(rec, index, offsets) must be avoided here. */
-	ut_ad(n_ext);
-	*n_ext = 0;
-
-	rec_len = index->n_fields + 1;
-
-	entry = dtuple_create(heap, rec_len);
-
-	dtuple_set_n_fields_cmp(entry,
-				dict_index_get_n_unique_in_tree(index));
-	ulint	field_no = 0;
-	const bool is_alter_metadata = rec_is_alter_metadata(rec, index);
-
-	for (i = 0; i < rec_len; i++) {
-
-		dfield = dtuple_get_nth_field(entry, i);
-		if (i == unsigned(index->n_uniq + DATA_ROLL_PTR)) {
-			dfield->type.mtype = DATA_BLOB;
-			dfield->type.prtype = DATA_NOT_NULL;
-
-			if (!is_alter_metadata) {
-				dfield_set_data(dfield, field_ref_zero,
-						FIELD_REF_SIZE);
-				dfield->len = FIELD_REF_SIZE;
-				dfield_set_ext(dfield);
-				(*n_ext)++;
-				continue;
-			}
-
-			goto fetch_rec;
-		}
-
-		ifield = dict_index_get_nth_field(index, field_no++);
-		col = ifield->col;
-
-		dict_col_copy_type(col, dfield_get_type(dfield));
-fetch_rec:
-		ulint field_no = i;
-
-		if (is_alter_metadata) {
-			field = rec_get_nth_def_field(
-				rec, index, offsets, field_no, &len);
-		} else {
-			if (i > unsigned(index->n_uniq + DATA_ROLL_PTR)) {
-				field_no = i - 1;
-			}
-
-			field = rec_get_nth_def_field(
-					rec, index, offsets, field_no, &len);
-		}
-
-		if (col->is_dropped()) {
-			dfield_set_data(dfield, field_ref_zero, len);
-		} else {
-			dfield_set_data(dfield, field, len);
-		}
-
-		if (rec_offs_nth_extern(offsets, field_no)) {
-			dfield_set_ext(dfield);
-			(*n_ext)++;
-		}
-	}
-
-	ut_ad(dtuple_check_typed(entry));
-	return(entry);
-}
-
 /** Convert an index record to a data tuple.
-@tparam def whether the index->instant_field_value() needs to be accessed
-@param[in]	rec	index record
-@param[in]	index	index
-@param[in]	offsets	rec_get_offsets(rec, index)
-@param[out]	n_ext	number of externally stored columns
-@param[in,out]	heap	memory heap for allocations
+@tparam metadata whether the index->instant_field_value() needs to be accessed
+@tparam mblob 1 if rec_is_alter_metadata();
+2 if we want converted metadata corresponding to info_bits
+@param[in]	rec		index record
+@param[in]	index		index
+@param[in]	offsets		rec_get_offsets(rec, index)
+@param[out]	n_ext		number of externally stored columns
+@param[in,out]	heap		memory heap for allocations
+@param[in]	info_bits	(only used if mblob=2)
 @return index entry built; does not set info_bits, and the data fields
 in the entry will point directly to rec */
-template<bool def>
+template<bool metadata, int mblob = 0>
 static inline
 dtuple_t*
 row_rec_to_index_entry_impl(
@@ -892,44 +795,57 @@ row_rec_to_index_entry_impl(
 	const dict_index_t*	index,
 	const ulint*		offsets,
 	ulint*			n_ext,
-	mem_heap_t*		heap)
+	mem_heap_t*		heap,
+	ulint			info_bits = 0)
 {
-	dtuple_t*	entry;
-	dfield_t*	dfield;
-	ulint		i;
-	const byte*	field;
-	ulint		len;
-	ulint		rec_len;
-
 	ut_ad(rec != NULL);
 	ut_ad(heap != NULL);
 	ut_ad(index != NULL);
-	ut_ad(def || !rec_offs_any_default(offsets));
-
+	ut_ad(!mblob || metadata);
+	ut_ad(!mblob || index->is_primary());
+	ut_ad(!mblob || !dict_index_is_spatial(index));
+	compile_time_assert(mblob <= 2);
+	ut_ad(mblob != 2 || dtuple_t::is_metadata(info_bits));
+	ut_ad(mblob == 2 || info_bits == 0);
 	/* Because this function may be invoked by row0merge.cc
 	on a record whose header is in different format, the check
 	rec_offs_validate(rec, index, offsets) must be avoided here. */
 	ut_ad(n_ext);
 	*n_ext = 0;
 
-	rec_len = rec_offs_n_fields(offsets);
+	ut_ad(mblob != 2
+	      || rec_offs_n_fields(offsets)
+	      == ulint(index->n_fields + rec_is_alter_metadata(rec, index)));
+
+	ulint rec_len = mblob == 2
+		? ulint(index->n_fields
+			+ (info_bits == REC_INFO_METADATA_ALTER))
+		: rec_offs_n_fields(offsets);
 
-	entry = dtuple_create(heap, rec_len);
+	dtuple_t* entry = dtuple_create(heap, rec_len);
+	dfield_t* dfield = entry->fields;
 
 	dtuple_set_n_fields_cmp(entry,
 				dict_index_get_n_unique_in_tree(index));
-	ut_ad(rec_len == dict_index_get_n_fields(index)
+	ut_ad(mblob == 2
+	      || rec_len == dict_index_get_n_fields(index) + uint(mblob == 1)
 	      /* a record for older SYS_INDEXES table
 	      (missing merge_threshold column) is acceptable. */
 	      || (index->table->id == DICT_INDEXES_ID
 		  && rec_len == dict_index_get_n_fields(index) - 1));
 
-	dict_index_copy_types(entry, index, rec_len);
-
-	for (i = 0; i < rec_len; i++) {
+	ulint i;
+	for (i = 0; i < (mblob ? index->n_uniq + 2 : rec_len); i++, dfield++) {
+		dict_col_copy_type(dict_index_get_nth_col(index, i),
+				   &dfield->type);
+		if (!mblob
+		    && dict_index_is_spatial(index)
+		    && DATA_GEOMETRY_MTYPE(dfield->type.mtype)) {
+			dfield->type.prtype |= DATA_GIS_MBR;
+		}
 
-		dfield = dtuple_get_nth_field(entry, i);
-		field = def
+		ulint len;
+		const byte* field = metadata
 			? rec_get_nth_cfield(rec, index, offsets, i, &len)
 			: rec_get_nth_field(rec, offsets, i, &len);
 
@@ -937,12 +853,68 @@ row_rec_to_index_entry_impl(
 
 		if (rec_offs_nth_extern(offsets, i)) {
 			dfield_set_ext(dfield);
-			(*n_ext)++;
+			++*n_ext;
+		}
+	}
+
+	if (mblob) {
+		ulint len;
+		const byte* field;
+		ulint j = i;
+
+		if (mblob == 2) {
+			const bool got = rec_is_alter_metadata(rec, index);
+			const bool want = info_bits == REC_INFO_METADATA_ALTER;
+			if (got == want) {
+				if (got) {
+					goto copy_metadata;
+				}
+			} else {
+				if (want) {
+					/* Allocate a placeholder for
+					adding metadata in an update. */
+					len = FIELD_REF_SIZE;
+					field = static_cast<byte*>(
+						mem_heap_zalloc(heap, len));
+					/* In reality there is one fewer
+					field present in the record. */
+					rec_len--;
+					goto init_metadata;
+				}
+
+				/* Skip the undesired metadata blob
+				(for example, when rolling back an
+				instant ALTER TABLE). */
+				i++;
+			}
+			goto copy_user_fields;
+		}
+copy_metadata:
+		ut_ad(rec_offs_nth_extern(offsets, i));
+		field = rec_get_nth_field(rec, offsets, i++, &len);
+init_metadata:
+		dfield->type.metadata_blob_init();
+		ut_ad(len == FIELD_REF_SIZE);
+		dfield_set_data(dfield, field, len);
+		dfield_set_ext(dfield++);
+		++*n_ext;
+copy_user_fields:
+		for (; i < rec_len; i++, dfield++) {
+			dict_col_copy_type(dict_index_get_nth_col(index, j++),
+					   &dfield->type);
+			field = rec_get_nth_field(rec, offsets, i, &len);
+			dfield_set_data(dfield, field, len);
+
+			if (rec_offs_nth_extern(offsets, i)) {
+				dfield_set_ext(dfield);
+				++*n_ext;
+			}
 		}
 	}
 
+	ut_ad(dfield == entry->fields + entry->n_fields);
 	ut_ad(dtuple_check_typed(entry));
-	return(entry);
+	return entry;
 }
 
 /** Convert an index record to a data tuple.
@@ -978,34 +950,25 @@ row_rec_to_index_entry(
 	mem_heap_t*		heap)	/*!< in: memory heap from which
 					the memory needed is allocated */
 {
-	dtuple_t*	entry;
-	byte*		buf;
-	const rec_t*	copy_rec;
-
 	ut_ad(rec != NULL);
 	ut_ad(heap != NULL);
 	ut_ad(index != NULL);
 	ut_ad(rec_offs_validate(rec, index, offsets));
 
 	/* Take a copy of rec to heap */
-	buf = static_cast<byte*>(
-		mem_heap_alloc(heap, rec_offs_size(offsets)));
-
-	copy_rec = rec_copy(buf, rec, offsets);
+	const rec_t* copy_rec = rec_copy(
+		static_cast<byte*>(mem_heap_alloc(heap,
+						  rec_offs_size(offsets))),
+		rec, offsets);
 
 	rec_offs_make_valid(copy_rec, index, true,
 			    const_cast<ulint*>(offsets));
 
-	if (rec_is_alter_metadata(copy_rec, index)
-	    || (rec_is_metadata(copy_rec, index)
-		&& index->table->instant)) {
-
-		entry = row_def_rec_to_index_entry_impl(
+	dtuple_t* entry = rec_is_alter_metadata(copy_rec, index)
+		? row_rec_to_index_entry_impl<true,1>(
+			copy_rec, index, offsets, n_ext, heap)
+		: row_rec_to_index_entry_impl<true>(
 			copy_rec, index, offsets, n_ext, heap);
-	} else {
-		entry = row_rec_to_index_entry_impl<true>(
-			copy_rec, index, offsets, n_ext, heap);
-	}
 
 	rec_offs_make_valid(rec, index, true,
 			    const_cast<ulint*>(offsets));
@@ -1016,6 +979,49 @@ row_rec_to_index_entry(
 	return(entry);
 }
 
+/** Convert a metadata record to a data tuple.
+@param[in]	rec		metadata record
+@param[in]	index		clustered index after instant ALTER TABLE
+@param[in]	offsets		rec_get_offsets(rec)
+@param[out]	n_ext		number of externally stored fields
+@param[in,out]	heap		memory heap for allocations
+@param[in]	info_bits	the info_bits after an update */
+dtuple_t*
+row_metadata_to_tuple(
+	const rec_t*		rec,
+	const dict_index_t*	index,
+	const ulint*		offsets,
+	ulint*			n_ext,
+	mem_heap_t*		heap,
+	ulint			info_bits)
+{
+	ut_ad(info_bits == REC_INFO_METADATA_ALTER
+	      || info_bits == REC_INFO_METADATA_ADD);
+	ut_ad(rec_is_metadata(rec, index));
+	ut_ad(rec_offs_validate(rec, index, offsets));
+
+	const rec_t* copy_rec = rec_copy(
+		static_cast<byte*>(mem_heap_alloc(heap,
+						  rec_offs_size(offsets))),
+		rec, offsets);
+
+	rec_offs_make_valid(copy_rec, index, true,
+			    const_cast<ulint*>(offsets));
+
+	dtuple_t* entry = info_bits == REC_INFO_METADATA_ALTER
+		|| rec_is_alter_metadata(copy_rec, index)
+		? row_rec_to_index_entry_impl<true,2>(
+			copy_rec, index, offsets, n_ext, heap, info_bits)
+		: row_rec_to_index_entry_impl<true>(
+			copy_rec, index, offsets, n_ext, heap);
+
+	rec_offs_make_valid(rec, index, true,
+			    const_cast<ulint*>(offsets));
+
+	dtuple_set_info_bits(entry, info_bits);
+	return entry;
+}
+
 /*******************************************************************//**
 Builds from a secondary index record a row reference with which we can
 search the clustered index record.

storage/innobase/row/row0upd.cc

@@ -1323,43 +1323,46 @@ row_upd_index_replace_new_col_val(
 }
 
 /** Apply an update vector to an metadata entry.
-@param[in,out]	entry	index entry to be updated; the clustered index default
-			row record
+@param[in,out]	entry	clustered index metadata record to be updated
 @param[in]	index	index of the entry
 @param[in]	update	update vector built for the entry
 @param[in,out]	heap	memory heap for copying off-page columns */
 static
 void
-row_upd_index_replace_metadata_pos(
+row_upd_index_replace_metadata(
 	dtuple_t*		entry,
 	const dict_index_t*	index,
 	const upd_t*		update,
 	mem_heap_t*		heap)
 {
 	ut_ad(!index->table->skip_alter_undo);
-
-	const page_size_t&	page_size = dict_table_page_size(index->table);
-
-	dtuple_set_info_bits(entry, update->info_bits);
-
-	for (unsigned i = entry->n_fields;
-	     i >= unsigned(index->n_uniq + DATA_ROLL_PTR); i--) {
-		const dict_field_t*	field = NULL;
-		const dict_col_t*	col = NULL;
-		const upd_field_t*	uf;
-
-		uf = upd_get_field_by_field_no(update, i -1, false);
-
-		if (uf) {
-			if (i > unsigned(index->n_uniq + DATA_ROLL_PTR)) {
-				field = dict_index_get_nth_field(index, i - 1);
-				col = dict_field_get_col(field);
-			}
-
-			row_upd_index_replace_new_col_val(
-				dtuple_get_nth_field(entry, i),
-				field, col, uf, heap, page_size);
+	ut_ad(update->is_alter_metadata());
+	ut_ad(entry->info_bits == update->info_bits);
+	ut_ad(entry->n_fields == ulint(index->n_fields) + 1);
+	const page_size_t& page_size = dict_table_page_size(index->table);
+	const ulint first = index->n_uniq + 2;
+
+	for (ulint i = upd_get_n_fields(update); i--; ) {
+		const upd_field_t* uf = upd_get_nth_field(update, i);
+		ut_ad(!upd_fld_is_virtual_col(uf));
+		ut_ad(uf->field_no >= first);
+		ulint f = uf->field_no;
+		dfield_t* dfield = dtuple_get_nth_field(entry, f);
+
+		if (f-- == first) {
+			ut_ad(dfield_is_ext(&uf->new_val));
+			ut_ad(dfield_get_len(&uf->new_val) == FIELD_REF_SIZE);
+			ut_ad(!dfield_is_null(&uf->new_val));
+			ut_ad(dfield_is_ext(dfield));
+			ut_ad(dfield_get_len(dfield) == FIELD_REF_SIZE);
+			ut_ad(!dfield_is_null(dfield));
+			dfield->data = uf->new_val.data;
+			continue;
 		}
+
+		const dict_field_t* field = dict_index_get_nth_field(index, f);
+		row_upd_index_replace_new_col_val(dfield, field, field->col,
+						  uf, heap, page_size);
 	}
 }
 
@@ -1378,16 +1381,17 @@ row_upd_index_replace_new_col_vals_index_pos(
 	mem_heap_t*		heap)
 {
 	ut_ad(!index->table->skip_alter_undo);
-
-	const page_size_t&	page_size = dict_table_page_size(index->table);
-
-	dtuple_set_info_bits(entry, update->info_bits);
+	ut_ad(!entry->is_metadata() || entry->info_bits == update->info_bits);
 
 	if (UNIV_UNLIKELY(entry->is_alter_metadata())) {
-		row_upd_index_replace_metadata_pos(entry, index, update, heap);
+		row_upd_index_replace_metadata(entry, index, update, heap);
 		return;
 	}
 
+	const page_size_t&	page_size = dict_table_page_size(index->table);
+
+	dtuple_set_info_bits(entry, update->info_bits);
+
 	for (unsigned i = index->n_fields; i--; ) {
 		const dict_field_t*	field;
 		const dict_col_t*	col;