Introduce dict_table_t::instant (DROP or reorder columns)

storage/innobase/btr/btr0cur.cc

@@ -4370,7 +4370,7 @@ btr_cur_optimistic_update(
 		return(DB_OVERFLOW);
 	}
 
-	if (rec_is_metadata(rec, index) && index->table->n_dropped_cols) {
+	if (rec_is_metadata(rec, index) && index->table->instant) {
 		goto any_extern;
 	}
 

storage/innobase/dict/dict0mem.cc

@@ -1195,8 +1195,8 @@ operator<< (std::ostream& out, const dict_foreign_set& fk_set)
 fields. */
 void dict_index_t::reconstruct_fields()
 {
-	n_fields += table->n_dropped_cols;
-	n_def += table->n_dropped_cols;
+	n_fields += table->instant->n_dropped;
+	n_def += table->instant->n_dropped;
 
 	unsigned	n_pk_fields = unsigned(n_uniq + DATA_ROLL_PTR);
 
@@ -1213,7 +1213,7 @@ void dict_index_t::reconstruct_fields()
 		ulint	col_no = table->non_pk_col_map[i - n_pk_fields];
 		if (col_no == 0) {
 			/* Dropped Column */
-			temp_fields[i].col = &table->dropped_cols[j++];
+			temp_fields[i].col = &table->instant->dropped[j++];
 			ut_ad(i == unsigned(temp_fields[i].col->ind));
 		} else {
 			field = fields[old_field_no++];
@@ -1239,8 +1239,9 @@ void dict_index_t::remove_instant()
 		return;
 	}
 
-	if (table->n_dropped_cols == 0) {
-
+	const unsigned n_dropped = table->n_dropped();
+	if (!n_dropped) {
+		/* FIXME: reorder columns too! */
 		for (unsigned i = n_core_fields; i < n_fields; i++) {
 			fields[i].col->add_to_core();
 		}
@@ -1252,8 +1253,8 @@ void dict_index_t::remove_instant()
 	}
 
 	ulint	old_n_fields = n_fields;
-	n_fields -= table->n_dropped_cols;
-	n_def -= table->n_dropped_cols;
+	n_fields -= n_dropped;
+	n_def -= n_dropped;
 
 	unsigned	n_null = 0;
 	unsigned	new_field = 0;
@@ -1290,8 +1291,7 @@ void dict_index_t::remove_instant()
 	n_core_fields = n_fields;
 	n_nullable = n_null;
 	n_core_null_bytes = UT_BITS_IN_BYTES(n_null);
-	table->dropped_cols = NULL;
-	table->n_dropped_cols = 0;
+	table->instant = NULL;
 	table->non_pk_col_map = NULL;
 }
 
@@ -1319,8 +1319,8 @@ inline void dict_index_t::instant_op_field(
 
 	ulint	old_n_fields = n_fields;
 
-	n_fields = instant.n_fields + table->n_dropped_cols;
-	n_def = instant.n_def + table->n_dropped_cols;
+	n_fields = instant.n_fields + table->n_dropped();
+	n_def = instant.n_def + table->n_dropped();
 
 	unsigned	n_null = 0;
 	unsigned	n_non_drop_null = 0;
@@ -1337,9 +1337,10 @@ inline void dict_index_t::instant_op_field(
 		bool		is_dropped = false;
 		ulint		new_col_offset = 0;
 
-		for (unsigned j = 0; j < table->n_dropped_cols; j++) {
-			if (table->dropped_cols[j].ind == i) {
-				temp_field.col = &table->dropped_cols[j];
+		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;
 			}
@@ -1352,7 +1353,8 @@ inline void dict_index_t::instant_op_field(
 
 			if (i >= old_n_fields) {
 
-				new_col_offset = i - (DATA_N_SYS_COLS + table->n_dropped_cols);
+				new_col_offset = i - DATA_N_SYS_COLS
+					- table->n_dropped();
 
 				if (!dict_index_is_auto_gen_clust(this)) {
 					new_col_offset += 1;
@@ -1471,9 +1473,9 @@ byte* dict_table_t::construct_metadata_blob(
 		unsigned col_no = non_pk_col_map[i];
 		col_no <<= INSTANT_FIELD_COL_NO_SHIFT;
 
-		for (ulint j = 0; col_no == 0 && j < n_dropped_cols;
+		for (ulint j = 0; col_no == 0 && j < instant->n_dropped;
 		     j++) {
-			if (dropped_cols[j].ind == i + num_pk_fields) {
+			if (instant->dropped[j].ind == i + num_pk_fields) {
 				col_no |= INSTANT_DROP_COL_FIXED;
 				break;
 			}
@@ -1491,6 +1493,7 @@ blob data.
 @param[in]	data	metadata blob data */
 void dict_table_t::construct_dropped_columns(const byte* data)
 {
+	ut_ad(!instant);
 	unsigned num_non_pk_fields = mach_read_from_4(data);
 	dict_index_t*	clust_index = dict_table_get_first_index(this);
 
@@ -1499,9 +1502,9 @@ void dict_table_t::construct_dropped_columns(const byte* data)
 
 	const byte*	field_data = data + INSTANT_NON_PK_FIELDS_LEN;
 	std::vector<ulint>	fixed_dcols;
+	unsigned n_dropped_cols = 0;
 
-	for (ulint i = 0; i < num_non_pk_fields; i++) {
-
+	for (unsigned i = 0; i < num_non_pk_fields; i++) {
 		unsigned col_no = mach_read_from_2(field_data);
 		bool is_fixed = col_no & INSTANT_DROP_COL_FIXED;
 		col_no >>= INSTANT_FIELD_COL_NO_SHIFT;
@@ -1517,24 +1520,23 @@ void dict_table_t::construct_dropped_columns(const byte* data)
 		field_data += INSTANT_FIELD_LEN;
 	}
 
-	dropped_cols = static_cast<dict_col_t*>(mem_heap_zalloc(
+	dict_col_t* dropped_cols = static_cast<dict_col_t*>(mem_heap_zalloc(
 		heap, n_dropped_cols * sizeof(dict_col_t)));
+	instant = new (mem_heap_alloc(heap, sizeof instant)) dict_instant_t();
+	instant->n_dropped = n_dropped_cols;
+	instant->dropped = dropped_cols;
 
-	ulint j = 0;
-	for (ulint i = 0; i < n_dropped_cols; i++) {
+	unsigned j = 0;
+	for (unsigned i = 0; i < n_dropped_cols; i++) {
 		dict_col_t&	drop_col = dropped_cols[i];
 		bool		is_fixed = false;
 		drop_col.dropped = true;
 
 		while (j < num_non_pk_fields) {
-			if (non_pk_col_map[j] == 0) {
-			    drop_col.ind = unsigned(
-				j + clust_index->n_uniq + DATA_ROLL_PTR);
-				j++;
+			if (non_pk_col_map[j++] == 0) {
+				drop_col.ind = j + clust_index->n_uniq + 1;
 				break;
 			}
-
-			j++;
 		}
 
 		is_fixed = (std::find(fixed_dcols.begin(), fixed_dcols.end(),
@@ -1551,22 +1553,24 @@ void dict_table_t::construct_dropped_columns(const byte* data)
 @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 */
-void dict_table_t::fill_dropped_column(
+inline void dict_table_t::fill_dropped_column(
 	const dict_table_t&	table,
 	const ulint*		col_map,
 	ulint			n_newly_drop)
 {
-	ulint	n_total_drop_col = n_dropped_cols + 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)));
 
-	memcpy(temp_drop_cols, dropped_cols, n_dropped_cols * sizeof(dict_col_t));
-
-	int j = n_dropped_cols;
+	if (j) {
+		memcpy(temp_drop_cols, instant->dropped, j
+		       * sizeof *instant->dropped);
+	}
 
-	for (int i = 0; i < n_def; i++) {
+	for (unsigned i = 0; i < n_def; i++) {
 		if (col_map[i] != ULINT_UNDEFINED) {
 			continue;
 		}
@@ -1597,9 +1601,14 @@ void dict_table_t::fill_dropped_column(
 	n_def -= n_newly_drop;
 	n_cols -= n_newly_drop;
 	n_t_cols -= n_newly_drop;
-	n_dropped_cols += n_newly_drop;
 
-	dropped_cols = temp_drop_cols;
+	if (!instant) {
+		instant = new (mem_heap_alloc(heap, sizeof *instant))
+			dict_instant_t();
+	}
+
+	instant->n_dropped = n_total_drop_col;
+	instant->dropped = temp_drop_cols;
 }
 
 /** Adjust table metadata for instant operation.

storage/innobase/handler/handler0alter.cc

@@ -4363,7 +4363,8 @@ innobase_op_instant_try(
 
 	dict_table_t* user_table = ctx->old_table;
 
-	ulint	n_old_drop_cols = ctx->old_table->n_dropped_cols;
+	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;
 
 	user_table->instant_op_column(
@@ -4502,7 +4503,7 @@ innobase_op_instant_try(
 
 	dtuple_t* entry;
 
-	if (index->is_drop_field_exist()) {
+	if (index->table->instant) {
 		entry = row_build_clust_default_entry(row, index, ctx->heap);
 		entry->info_bits = REC_INFO_METADATA_ALTER;
 	} else {
@@ -4535,20 +4536,14 @@ innobase_op_instant_try(
 			goto empty_table;
 		}
 
-		unsigned n = 0;
-
-		if (index->is_drop_field_exist()) {
-			n = ctx->n_instant_add_cols;
-			n += ctx->n_instant_drop_cols;
-		} else {
-			/* Extend the record with the instantly added columns. */
-			n = user_table->n_cols - ctx->old_n_cols;
-		}
+		/* 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;
 		/* 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->n_dropped_cols > 0
-				   || ctx->n_instant_drop_cols;
+		const unsigned f = user_table->instant != NULL;
 		upd_t* update = upd_create(index->n_fields, ctx->heap);
 		update->n_fields = n;
 		update->info_bits = f
@@ -4561,8 +4556,8 @@ innobase_op_instant_try(
 
 		for (unsigned i = 0; i < ctx->old_n_cols; i++) {
 			if (ctx->col_map[i] == ULINT_UNDEFINED) {
-				unsigned field_no = user_table->dropped_cols[
-					drop_cols_offset++].ind;
+				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;

storage/innobase/include/dict0mem.h

@@ -1075,9 +1075,6 @@ struct dict_index_t {
 	/** @return whether the index is corrupted */
 	inline bool is_corrupted() const;
 
-	/** Whether the index has dropped fields. */
-	bool is_drop_field_exist() const;
-
 	/** Detach the columns from the index that is to be freed. */
 	void detach_columns()
 	{
@@ -1134,16 +1131,12 @@ struct dict_index_t {
 	@return number of non-drop nullable fields in the index. */
 	unsigned get_n_non_drop_nullable_fields() const
 	{
+		DBUG_ASSERT(n_fields <= n_def);
 		unsigned n = 0;
-		for (ulint i = 0; i < n_fields; i++) {
-			const dict_col_t* col = fields[i].col;
-
-			if (!col->is_nullable()) {
-				n++;
-			}
+		for (unsigned i = 0; i < n_fields; i++) {
+			n += !fields[i].col->is_nullable();
 		}
 
-		DBUG_ASSERT(n <= n_def);
 		return n;
 	}
 
@@ -1574,6 +1567,15 @@ struct dict_vcol_templ_t {
 	dict_vcol_templ_t() : vtempl(0), mysql_table_query_id(~0ULL) {}
 };
 
+/** Instantly dropped or reordered columns */
+struct dict_instant_t
+{
+	/** Number of dropped columns */
+	unsigned n_dropped;
+	/** Dropped columns */
+	dict_col_t* dropped;
+};
+
 /** These are used when MySQL FRM and InnoDB data dictionary are
 in inconsistent state. */
 typedef enum {
@@ -1639,6 +1641,9 @@ struct dict_table_t {
 		return(!(flags & DICT_TF_MASK_ZIP_SSIZE));
 	}
 
+	/** @return the number of instantly dropped columns */
+	unsigned n_dropped() const { return instant ? instant->n_dropped : 0; }
+
 	/** Read the metadata blob and fill the non primary fields,
 	non-drop nullable fields and fill the drop columns in the
 	vector.
@@ -1670,7 +1675,7 @@ struct dict_table_t {
 	@param[in]	col_map		mapping of cols from old table
 					to new table
 	@param[in]	n_newly_drop	number of newly drop column */
-	void fill_dropped_column(
+	inline void fill_dropped_column(
 		const dict_table_t&	table,
 		const ulint*		col_map,
 		ulint			n_newly_drop);
@@ -1844,11 +1849,8 @@ struct dict_table_t {
 	reason s_cols is a part of dict_table_t */
 	dict_s_col_list*			s_cols;
 
-	/** List of dropped columns. */
-	dict_col_t*				dropped_cols;
-
-	/** Number of dropped columns. */
-	unsigned				n_dropped_cols:10;
+	/** Instantly dropped or reordered columns, or NULL if none */
+	dict_instant_t*				instant;
 
 	/** Column names packed in a character string
 	"name1\0name2\0...nameN\0". Until the string contains n_cols, it will
@@ -2140,14 +2142,13 @@ inline bool dict_index_t::is_readable() const { return table->is_readable(); }
 inline bool dict_index_t::is_instant() const
 {
 	ut_ad(n_core_fields > 0);
-	ut_ad(n_core_fields <= n_fields || is_drop_field_exist());
+	ut_ad(n_core_fields <= n_fields || table->n_dropped());
 	ut_ad(n_core_fields == n_fields
 	      || (type & ~(DICT_UNIQUE | DICT_CORRUPT)) == DICT_CLUSTERED);
 	ut_ad(n_core_fields == n_fields || table->supports_instant());
 	ut_ad(n_core_fields == n_fields || !table->is_temporary());
 
-	/* MDEV-15562 FIXME: reordered columns */
-	return n_core_fields != n_fields || is_drop_field_exist();
+	return n_core_fields != n_fields || table->instant;
 }
 
 inline bool dict_index_t::is_corrupted() const
@@ -2157,14 +2158,6 @@ inline bool dict_index_t::is_corrupted() const
 			     || (table && table->corrupted));
 }
 
-
-/** Whether the index has dropped fields. */
-inline bool dict_index_t::is_drop_field_exist() const
-{
-	ut_ad(!table->n_dropped_cols || is_primary());
-	return table->n_dropped_cols > 0;
-}
-
 /*******************************************************************//**
 Initialise the table lock list. */
 void

storage/innobase/rem/rem0rec.cc

@@ -328,14 +328,9 @@ rec_init_offsets_comp_ordinary(
 		ut_ad(index->is_instant());
 		/* fall through */
 	case REC_LEAF_TEMP_INSTANT:
-
-		if (index->is_drop_field_exist()
-		    && !rec_is_metadata(rec, index)) {
-			n_fields = n_core + rec_get_n_add_field(nulls);
-		} else {
-			n_fields = n_core + 1 + rec_get_n_add_field(nulls);
-		}
-
+		n_fields = n_core + rec_get_n_add_field(nulls)
+			+ (!index->table->instant
+			   || rec_is_metadata(rec, index));
 		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);
@@ -1121,7 +1116,8 @@ 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->n_dropped_cols));
+			  + (index->table->instant
+			     ? index->table->instant->n_dropped : 0)));
 
 		if (is_user_rec && leaf && n < index->n_fields) {
 			ut_ad(!index->is_dummy);
@@ -1350,7 +1346,7 @@ rec_get_metadata_converted_size(
 	ulint*			extra)
 {
 	ut_ad(dict_index_is_clust(clust_index));
-	ut_ad(clust_index);
+	ut_ad(clust_index->table->instant);
 	ut_ad(tuple);
 	ut_ad(dtuple_check_typed(tuple));
 
@@ -1366,7 +1362,7 @@ rec_get_metadata_converted_size(
 	and number of dropped fields. */
 	int	n_null_bytes = UT_BITS_IN_BYTES(
 				clust_index->get_n_non_drop_nullable_fields()
-				+ clust_index->table->n_dropped_cols);
+				+ clust_index->table->instant->n_dropped);
 
 	extra_size += n_null_bytes;
 
@@ -1462,9 +1458,9 @@ rec_get_converted_size_comp_prefix_low(
 		extra_size += UT_BITS_IN_BYTES(
 				index->get_n_nullable(n_fields));
 
-		if (index->is_drop_field_exist()) {
+		if (index->table->instant) {
 			extra_size += rec_get_n_add_field_len(
-					index->get_n_instant_cols(n_fields));
+				index->get_n_instant_cols(n_fields));
 		} else {
 			extra_size += rec_get_n_add_field_len(
 				n_fields - 1 - index->n_core_fields);
@@ -1839,9 +1835,9 @@ rec_convert_dtuple_to_default_rec_comp(
 	ulint		null_mask	= 1;
 
 	/* Nullable non-dropped fields + number of dropped fields. */
-	int		n_null_bytes = UT_BITS_IN_BYTES(
-				(clust_index->get_n_non_drop_nullable_fields()
-				 + clust_index->table->n_dropped_cols));
+	unsigned	n_null_bytes = UT_BITS_IN_BYTES(
+		(clust_index->get_n_non_drop_nullable_fields()
+				 + clust_index->table->instant->n_dropped));
 
 	rec_set_n_add_field(nulls, n_null_bytes);
 
@@ -1969,7 +1965,7 @@ rec_convert_dtuple_to_rec_comp(
 	case REC_STATUS_INSTANT:
 		ut_ad(index->is_instant());
 
-		if (index->is_drop_field_exist()) {
+		if (index->table->instant) {
 			rec_set_n_add_field(
 				nulls, index->get_n_instant_cols(n_fields));
 		} else {
@@ -2446,12 +2442,12 @@ rec_copy_prefix_to_buf(
 		if (UNIV_UNLIKELY(rec_is_alter_metadata(rec, index))) {
 			ulint	n_null_bytes = UT_BITS_IN_BYTES(
 				index->get_n_non_drop_nullable_fields()
-				+ index->table->n_dropped_cols);
+				+ index->table->instant->n_dropped);
 			nb += rec_get_n_add_field_len(n_null_bytes);
 
 			nb += n_null_bytes;
 		} else {
-			if (index->is_drop_field_exist()) {
+			if (index->table->instant) {
 				n_rec = ulint(index->n_core_fields)
 					+ rec_get_n_add_field(nulls);
 			} else {

storage/innobase/row/row0mysql.cc

@@ -830,7 +830,7 @@ row_create_prebuilt(
 
 	/* Make sure that search_tuple is long enough for clustered index */
 	ut_a(2 * table->n_cols >= clust_index->n_fields
-	     - clust_index->table->n_dropped_cols);
+	     - clust_index->table->n_dropped());
 
 	ref_len = dict_index_get_n_unique(clust_index);
 

storage/innobase/row/row0row.cc

@@ -998,7 +998,7 @@ row_rec_to_index_entry(
 
 	if (rec_is_alter_metadata(copy_rec, index)
 	    || (rec_is_metadata(copy_rec, index)
-		&& index->table->n_dropped_cols > 0)) {
+		&& index->table->instant)) {
 
 		entry = row_def_rec_to_index_entry_impl(
 			copy_rec, index, offsets, n_ext, heap);