Try to ensure that rollback of inserting metadata record works

We must use the original page format for inserting metadata records.
This is because in case the instant ALTER TABLE operation needs to
be rolled back (due to crash recovery), all pages must remain in
the original format.

btr_cur_optimistic_insert(): Avoid reorganize for inserting the
metadata record.

btr_cur_optimistic_update(): Initialize rec,offsets correctly
after possible page format conversion.

innobase_instant_try(): Do not store NULL values for those
metadata record columns that were originally declared NOT NULL.

dict_index_t::get_n_nullable(): Refer to dict_col_t::was_not_null()
in order to keep the original format intact.

lock_move_reorganize_page(): Allow the two pages to be in
different formats.

rec_convert_dtuple_to_rec_comp(), rec_convert_dtuple_to_rec_comp():
Also observe the DATA_WAS_NOT_NULL flag.

storage/innobase/btr/btr0cur.cc

@@ -3481,7 +3481,11 @@ btr_cur_optimistic_insert(
 		 << ib::hex(thr ? thr->graph->trx->id : 0)
 		 << ' ' << rec_printer(entry).str());
 
-	bool reorg = leaf && page_is_comp(page) && index->dual_format();
+	/* When inserting a metadata record that introduces
+	index->dual_format(), we must not convert the page into
+	flexible format, because we want to be able to roll back. */
+	bool reorg = leaf && page_is_comp(page) && index->dual_format()
+		&& !entry->is_alter_metadata();
 	DBUG_EXECUTE_IF("do_page_reorganize", reorg = true; );
 	if (reorg && !btr_page_reorganize(page_cursor, index, mtr)) {
 		goto fail;
@@ -4514,11 +4518,9 @@ btr_cur_optimistic_update(
 
 	block = btr_cur_get_block(cursor);
 	page = buf_block_get_frame(block);
-	rec = btr_cur_get_rec(cursor);
 	index = cursor->index;
+
 	ut_ad(trx_id > 0 || (flags & BTR_KEEP_SYS_FLAG));
-	ut_ad(!!page_rec_is_comp(rec) == index->table->not_redundant()
-	      || index->dual_format());
 	ut_ad(mtr_memo_contains(mtr, block, MTR_MEMO_PAGE_X_FIX));
 	/* This is intended only for leaf page updates */
 	ut_ad(page_is_leaf(page));
@@ -4533,6 +4535,17 @@ btr_cur_optimistic_update(
 	ut_ad(fil_page_index_page_check(page));
 	ut_ad(btr_page_get_index_id(page) == index->id);
 
+	if (page_is_comp(page) && index->dual_format()
+	    && !btr_page_reorganize(btr_cur_get_page_cur(cursor), index,
+				    mtr)) {
+		/* Conversion to flexible format failed.
+		We must split the page in a pessimistic operation. */
+		return DB_OVERFLOW;
+	}
+
+	rec = btr_cur_get_rec(cursor);
+	ut_ad(!!page_rec_is_comp(rec) == index->table->not_redundant()
+	      || index->dual_format());
 	*offsets = rec_get_offsets(rec, index, *offsets,
 				   page_is_comp(page)
 				   ? REC_FMT_LEAF : REC_FMT_LEAF_FLEXIBLE,
@@ -4542,14 +4555,6 @@ btr_cur_optimistic_update(
 	     || trx_is_recv(thr_get_trx(thr)));
 #endif /* UNIV_DEBUG || UNIV_BLOB_LIGHT_DEBUG */
 
-	if (page_is_comp(page) && index->dual_format()
-	    && !btr_page_reorganize(btr_cur_get_page_cur(cursor), index,
-				    mtr)) {
-		/* Conversion to flexible format failed.
-		We must split the page in a pessimistic operation. */
-		return DB_OVERFLOW;
-	}
-
 	if (UNIV_LIKELY(!update->is_metadata())
 	    && !row_upd_changes_field_size_or_external(index, *offsets,
 						       update)) {

storage/innobase/handler/handler0alter.cc

@@ -5537,11 +5537,12 @@ static bool innobase_instant_try(
 		DBUG_ASSERT(!strcmp((*af)->field_name.str,
 				    dict_table_get_col_name(user_table, i)));
 		DBUG_ASSERT(old || col->is_added());
+		DBUG_ASSERT(col->is_nullable() == (*af)->real_maybe_null());
 
 		if (col->is_added()) {
 			dfield_set_data(d, col->def_val.data,
 					col->def_val.len);
-		} else if ((*af)->real_maybe_null()) {
+		} else if (!col->was_not_null()) {
 			/* Store NULL for nullable 'core' columns. */
 			dfield_set_null(d);
 		} else {

storage/innobase/include/dict0mem.h

@@ -1175,7 +1175,7 @@ struct dict_index_t {
 		for (; n_prefix < n_fields; n_prefix++) {
 			const dict_col_t* col = fields[n_prefix].col;
 			DBUG_ASSERT(!col->is_virtual());
-			n -= col->is_nullable();
+			n -= !col->was_not_null();
 		}
 		DBUG_ASSERT(n < n_def);
 		return n;

storage/innobase/lock/lock0lock.cc

@@ -2629,7 +2629,6 @@ lock_move_reorganize_page(
 	lock_t*		lock;
 	UT_LIST_BASE_NODE_T(lock_t)	old_locks;
 	mem_heap_t*	heap		= NULL;
-	ulint		comp;
 
 	lock_mutex_enter();
 
@@ -2667,9 +2666,6 @@ lock_move_reorganize_page(
 		lock = lock_rec_get_next_on_page(lock);
 	} while (lock != NULL);
 
-	comp = page_is_comp(block->frame);
-	ut_ad(comp == page_is_comp(oblock->frame));
-
 	lock_move_granted_locks_to_front(old_locks);
 
 	DBUG_EXECUTE_IF("do_lock_reverse_page_reorganize",
@@ -2695,19 +2691,22 @@ lock_move_reorganize_page(
 			ut_ad(page_rec_is_metadata(rec1)
 			      == page_rec_is_metadata(rec2));
 
-			if (comp) {
-				old_heap_no = rec_get_heap_no_new(rec2);
+			if (page_is_comp(block->frame)) {
 				new_heap_no = rec_get_heap_no_new(rec1);
-
 				rec1 = page_rec_get_next_low(rec1, TRUE);
-				rec2 = page_rec_get_next_low(rec2, TRUE);
 			} else {
-				old_heap_no = rec_get_heap_no_old(rec2);
 				new_heap_no = rec_get_heap_no_old(rec1);
-				ut_ad(!memcmp(rec1, rec2,
-					      rec_get_data_size_old(rec2)));
-
+				ut_ad(page_is_comp(oblock->frame)
+				      || !memcmp(rec1, rec2,
+						 rec_get_data_size_old(rec2)));
 				rec1 = page_rec_get_next_low(rec1, FALSE);
+			}
+
+			if (page_is_comp(oblock->frame)) {
+				old_heap_no = rec_get_heap_no_new(rec2);
+				rec2 = page_rec_get_next_low(rec2, TRUE);
+			} else {
+				old_heap_no = rec_get_heap_no_old(rec2);
 				rec2 = page_rec_get_next_low(rec2, FALSE);
 			}
 

storage/innobase/rem/rem0rec.cc

@@ -1610,7 +1610,8 @@ rec_convert_dtuple_to_rec_comp(
 			break;
 		}
 
-		if (!(field->type.prtype & DATA_NOT_NULL)) {
+		if (!(field->type.prtype
+		      & (DATA_NOT_NULL | DATA_WAS_NOT_NULL))) {
 			/* nullable field */
 			ut_ad(n_null--);
 
@@ -1635,7 +1636,8 @@ rec_convert_dtuple_to_rec_comp(
 
 		const dict_field_t* ifield
 			= dict_index_get_nth_field(index, i);
-		ut_ad(!(field->type.prtype & DATA_NOT_NULL)
+		ut_ad(!(field->type.prtype
+			& (DATA_NOT_NULL | DATA_WAS_NOT_NULL))
 		      == !index->was_not_null(i));
 		ulint fixed_len = ifield->fixed_len;