Try to ensure that rollback of inserting metadata record worksbb-10.4-MDEV-17520-2

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.

5 files changed, 35 insertions, 28 deletions

diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 1c26ac0b17a..cf0f100fe5b 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -3481,7 +3481,11 @@ fail_err:
 		 << 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)) {
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index aed67b95c68..426d5a92df0 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/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 {
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 3aeefb8b601..77fe2d24b22 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/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;
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 5d6d3dcb8ae..3579a49bb59 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/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);
 			}
 
diff --git a/storage/innobase/rem/rem0rec.cc b/storage/innobase/rem/rem0rec.cc
index fbbed7b528f..b52b42bc646 100644
--- a/storage/innobase/rem/rem0rec.cc
+++ b/storage/innobase/rem/rem0rec.cc
@@ -1610,7 +1610,8 @@ start:
 			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 @@ start:
 
 		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;