Preserve ALTER TABLE metadata even when the table becomes empty

For instant ADD COLUMN…LAST it was possible to remove the metadata
whenever the table becomes empty. For the generic ALTER TABLE,
we must preserve the metadata on dropped or reordered columns,
because concurrent transactions are assuming that this metadata
cannot change.

rec_is_add_metadata(): Check if a record is ADD COLUMN metadata
record (not more generic instant ALTER TABLE).

btr_discard_only_page_on_level(): Preserve the generic instant
ALTER TABLE metadata.

btr_cur_optimistic_delete_func(), btr_cur_pessimistic_delete():
When the last user record is deleted, do not delete the
generic instant ALTER TABLE metadata record. Only delete
instant ADD COLUMN metadata records.

storage/innobase/btr/btr0btr.cc

@@ -4290,15 +4290,42 @@ btr_discard_only_page_on_level(
 	}
 #endif /* UNIV_BTR_DEBUG */
 
+	mem_heap_t* heap = NULL;
+	const rec_t* rec = NULL;
+	ulint* offsets = NULL;
+	if (index->table->instant) {
+		const rec_t* r = page_rec_get_next(page_get_infimum_rec(
+							   block->frame));
+		ut_ad(rec_is_metadata(r, *index) == index->is_instant());
+		if (rec_is_alter_metadata(r, *index)) {
+			heap = mem_heap_create(srv_page_size);
+			offsets = rec_get_offsets(r, index, NULL, true,
+						  ULINT_UNDEFINED, &heap);
+			rec = rec_copy(mem_heap_alloc(heap,
+						      rec_offs_size(offsets)),
+				       r, offsets);
+			rec_offs_make_valid(rec, index, true, offsets);
+		}
+	}
+
 	btr_page_empty(block, buf_block_get_page_zip(block), index, 0, mtr);
 	ut_ad(page_is_leaf(buf_block_get_frame(block)));
 	/* btr_page_empty() is supposed to zero-initialize the field. */
 	ut_ad(!page_get_instant(block->frame));
 
 	if (index->is_primary()) {
-		/* Concurrent access is prevented by the root_block->lock
-		X-latch, so this should be safe. */
-		index->remove_instant();
+		if (rec) {
+			DBUG_ASSERT(index->table->instant);
+			DBUG_ASSERT(rec_is_alter_metadata(rec, *index));
+			btr_set_instant(block, *index, mtr);
+			rec = page_cur_insert_rec_low(
+				page_get_infimum_rec(block->frame),
+				index, rec, offsets, mtr);
+			ut_ad(rec);
+			mem_heap_free(heap);
+		} else if (index->is_instant()) {
+			index->clear_instant_add();
+		}
 	} else if (!index->table->is_temporary()) {
 		/* We play it safe and reset the free bits for the root */
 		ibuf_reset_free_bits(block);

storage/innobase/btr/btr0cur.cc

@@ -5546,37 +5546,35 @@ btr_cur_optimistic_delete_func(
 		/* The whole index (and table) becomes logically empty.
 		Empty the whole page. That is, if we are deleting the
 		only user record, also delete the metadata record
-		if one exists (it exists if and only if is_instant()).
+		if one exists for instant ADD COLUMN (not generic ALTER TABLE).
 		If we are deleting the metadata record and the
 		table becomes empty, clean up the whole page. */
 		dict_index_t* index = cursor->index;
+		const rec_t* first_rec = page_rec_get_next_const(
+			page_get_infimum_rec(block->frame));
 		ut_ad(!index->is_instant()
-		      || rec_is_metadata(
-			      page_rec_get_next_const(
-				      page_get_infimum_rec(block->frame)),
-			      *index));
-		if (UNIV_UNLIKELY(rec_get_info_bits(rec, page_rec_is_comp(rec))
-				  & REC_INFO_MIN_REC_FLAG)) {
-			/* This should be rolling back instant ADD COLUMN.
-			If this is a recovered transaction, then
-			index->is_instant() will hold until the
-			insert into SYS_COLUMNS is rolled back. */
-			ut_ad(index->table->supports_instant());
-			ut_ad(index->is_primary());
-		} else {
+		      || rec_is_metadata(first_rec, *index));
+		const bool is_metadata = rec_is_metadata(rec, *index);
+		/* We can remove the metadata when rolling back an
+		instant ALTER TABLE operation, or when deleting the
+		last user record on the page such that only metadata for
+		instant ADD COLUMN (not generic ALTER TABLE) remains. */
+		const bool empty_table = is_metadata
+			|| !index->is_instant()
+			|| (first_rec != rec
+			    && rec_is_add_metadata(first_rec, *index));
+		if (UNIV_LIKELY(!is_metadata)) {
 			lock_update_delete(block, rec);
 		}
+		if (UNIV_LIKELY(empty_table)) {
 			btr_page_empty(block, buf_block_get_page_zip(block),
 				       index, 0, mtr);
-		page_cur_set_after_last(block, btr_cur_get_page_cur(cursor));
-
-		if (index->is_primary()) {
-			/* Concurrent access is prevented by
-			root_block->lock X-latch, so this should be
-			safe. */
-			index->remove_instant();
+			if (index->is_instant()) {
+				/* FIXME: free metadata BLOBs */
+				index->clear_instant_alter();
 			}
-
+		}
+		page_cur_set_after_last(block, btr_cur_get_page_cur(cursor));
 		return true;
 	}
 
@@ -5757,10 +5755,10 @@ btr_cur_pessimistic_delete(
 	}
 
 	if (page_is_leaf(page)) {
-		const bool is_metadata = rec_get_info_bits(
-			rec, page_rec_is_comp(rec)) & REC_INFO_MIN_REC_FLAG;
+		const bool is_metadata = rec_is_metadata(
+			rec, page_rec_is_comp(rec));
 		if (UNIV_UNLIKELY(is_metadata)) {
-			/* This should be rolling back instant ADD COLUMN.
+			/* This should be rolling back instant ALTER TABLE.
 			If this is a recovered transaction, then
 			index->is_instant() will hold until the
 			insert into SYS_COLUMNS is rolled back. */
@@ -5776,28 +5774,30 @@ btr_cur_pessimistic_delete(
 				goto discard_page;
 			}
 		} else if (page_get_n_recs(page) == 1
-			   + (index->is_instant()
-			      && !rec_is_metadata(rec, *index))) {
+			   + (index->is_instant() && !is_metadata)) {
 			/* The whole index (and table) becomes logically empty.
 			Empty the whole page. That is, if we are deleting the
 			only user record, also delete the metadata record
-			if one exists (it exists if and only if is_instant()).
+			if one exists for instant ADD COLUMN
+			(not generic ALTER TABLE).
 			If we are deleting the metadata record and the
 			table becomes empty, clean up the whole page. */
+
+			const rec_t* first_rec = page_rec_get_next_const(
+				page_get_infimum_rec(page));
 			ut_ad(!index->is_instant()
-			      || rec_is_metadata(
-				      page_rec_get_next_const(
-					      page_get_infimum_rec(page)),
-				      *index));
+			      || rec_is_metadata(first_rec, *index));
+			if (is_metadata || !index->is_instant()
+			    || (first_rec != rec
+				&& rec_is_add_metadata(first_rec, *index))) {
 				btr_page_empty(block, page_zip, index, 0, mtr);
+				if (index->is_instant()) {
+					/* FIXME: free metadata BLOBs */
+					index->clear_instant_alter();
+				}
+			}
 			page_cur_set_after_last(block,
 						btr_cur_get_page_cur(cursor));
-			if (index->is_primary()) {
-				/* Concurrent access is prevented by
-				index->lock and root_block->lock
-				X-latch, so this should be safe. */
-				index->remove_instant();
-			}
 			ret = TRUE;
 			goto return_after_reservations;
 		}

storage/innobase/dict/dict0mem.cc

@@ -1235,69 +1235,6 @@ inline void dict_index_t::reconstruct_fields()
 	n_core_null_bytes = UT_BITS_IN_BYTES(n_core_null);
 }
 
-/** Remove instant ALTER TABLE metadata.
-Protected by index root page x-latch or table X-lock. */
-void dict_index_t::remove_instant()
-{
-	DBUG_ASSERT(is_primary());
-
-	if (!is_instant()) {
-		return;
-	}
-
-	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->clear_instant();
-		}
-
-		n_core_fields = n_fields;
-		n_core_null_bytes = UT_BITS_IN_BYTES(n_nullable);
-		table->instant = NULL;
-		return;
-	}
-
-	ulint	old_n_fields = n_fields;
-	n_fields -= n_dropped;
-	n_def -= n_dropped;
-
-	unsigned	n_null = 0;
-	unsigned	new_field = 0;
-	dict_field_t*	temp_fields = static_cast<dict_field_t*>(mem_heap_zalloc(
-				heap, n_fields * sizeof *temp_fields));
-
-	for (unsigned i = 0; i < old_n_fields; i++) {
-		dict_field_t&	temp_field = temp_fields[new_field];
-		dict_field_t	field = fields[i];
-
-		if (field.col->is_dropped()) {
-			continue;
-		}
-
-		new_field++;
-
-		temp_field.prefix_len = field.prefix_len;
-		temp_field.fixed_len = field.fixed_len;
-		temp_field.name = dict_table_get_col_name(
-				table, dict_col_get_no(field.col));
-		temp_field.col = &table->cols[field.col->ind];
-
-		temp_field.col->clear_instant();
-
-		if (temp_field.col->is_nullable()) {
-			n_null++;
-		}
-	}
-
-	ut_ad(n_fields == new_field);
-	fields = temp_fields;
-	n_core_fields = n_fields;
-	n_nullable = n_null;
-	n_core_null_bytes = UT_BITS_IN_BYTES(n_null);
-	table->instant = NULL;
-}
-
 /** 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)

storage/innobase/handler/handler0alter.cc

@@ -5075,7 +5075,7 @@ static bool innobase_instant_try(
 		/* The table is empty. */
 		ut_ad(page_is_root(block->frame));
 		btr_page_empty(block, NULL, index, 0, &mtr);
-		index->remove_instant();
+		index->clear_instant_alter();
 		err = DB_SUCCESS;
 		goto func_exit;
 	}

storage/innobase/include/dict0mem.h

@@ -1126,6 +1126,10 @@ struct dict_index_t {
 	/** 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 ADD COLUMN metadata. */
+	inline void clear_instant_add();
+	/** Remove instant ALTER TABLE metadata. */
+	inline void clear_instant_alter();
 
 	/** Construct the metadata record for instant ALTER TABLE.
 	@param[in]	row	dummy or default values for existing columns
@@ -1134,10 +1138,6 @@ struct dict_index_t {
 	inline dtuple_t*
 	instant_metadata(const dtuple_t& row, mem_heap_t* heap) const;
 
-	/** Remove instant ALTER TABLE metadata.
-	Protected by index root page x-latch or table X-lock. */
-	void remove_instant();
-
 	/** Check if record in clustered index is historical row.
 	@param[in]	rec	clustered row
 	@param[in]	offsets	offsets
@@ -2079,6 +2079,62 @@ inline bool dict_index_t::is_corrupted() const
 			     || (table && table->corrupted));
 }
 
+inline void dict_index_t::clear_instant_add()
+{
+	DBUG_ASSERT(is_primary());
+	DBUG_ASSERT(is_instant());
+	DBUG_ASSERT(!table->instant);
+	for (unsigned i = n_core_fields; i < n_fields; i++) {
+		fields[i].col->clear_instant();
+	}
+	n_core_fields = n_fields;
+	n_core_null_bytes = UT_BITS_IN_BYTES(unsigned(n_nullable));
+}
+
+inline void dict_index_t::clear_instant_alter()
+{
+	DBUG_ASSERT(is_primary());
+
+	if (!table->instant) {
+		if (is_instant()) {
+			clear_instant_add();
+		}
+		return;
+	}
+
+	const unsigned n_dropped = table->n_dropped();
+
+	unsigned old_n_fields = n_fields;
+	n_fields -= n_dropped;
+	n_def -= n_dropped;
+
+	unsigned        n_null = 0;
+	unsigned        new_field = 0;
+	dict_field_t*   temp_fields = static_cast<dict_field_t*>(
+		mem_heap_alloc(heap, n_fields * sizeof *temp_fields));
+
+	for (unsigned i = 0; i < old_n_fields; i++) {
+		dict_field_t    field = fields[i];
+
+		if (field.col->is_dropped()) {
+			continue;
+		}
+
+		dict_field_t& f = temp_fields[new_field++];
+		f = field;
+		f.col->clear_instant();
+
+		n_null += f.col->is_nullable();
+	}
+
+	ut_ad(n_fields == new_field);
+	fields = temp_fields;
+	n_core_fields = n_fields;
+	n_nullable = n_null;
+	n_core_null_bytes = UT_BITS_IN_BYTES(n_null);
+	table->instant = NULL;
+}
+
 /** @return whether the column was instantly dropped
 @param[in] index	the clustered index */
 inline bool dict_col_t::is_dropped(const dict_index_t& index) const

storage/innobase/include/rem0rec.h

@@ -775,6 +775,30 @@ inline bool rec_is_metadata(const rec_t* rec, const dict_index_t& index)
 	return is;
 }
 
+/** Determine if the record is the metadata pseudo-record
+in the clustered index for instant ADD COLUMN (not other ALTER TABLE).
+@param[in]	rec	leaf page record
+@param[in]	comp	0 if ROW_FORMAT=REDUNDANT, else nonzero
+@return	whether the record is the metadata pseudo-record */
+inline bool rec_is_add_metadata(const rec_t* rec, ulint comp)
+{
+	bool is = rec_get_info_bits(rec, comp) == REC_INFO_MIN_REC_FLAG;
+	ut_ad(!is || !comp || rec_get_status(rec) == REC_STATUS_INSTANT);
+	return is;
+}
+
+/** Determine if the record is the metadata pseudo-record
+in the clustered index for instant ADD COLUMN (not other ALTER TABLE).
+@param[in]	rec	leaf page record
+@param[in]	index	index of the record
+@return	whether the record is the metadata pseudo-record */
+inline bool rec_is_add_metadata(const rec_t* rec, const dict_index_t& index)
+{
+	bool is = rec_is_add_metadata(rec, dict_table_is_comp(index.table));
+	ut_ad(!is || index.is_instant());
+	return is;
+}
+
 /** Determine if the record is the metadata pseudo-record
 in the clustered index for instant ALTER TABLE (not plain ADD COLUMN).
 @param[in]	rec	leaf page record