Fix btr_pcur store and restore

rec_copy_prefix_to_buf(): If the page is in the flexible format,
convert the record to the original format, so that
btr_pcur_restore_position() will only have to deal with one format.
Merge the code from rec_copy_prefix_to_buf_old().

btr_pcur_restore_position_func(): Interpret a special value of
btr_cur_t::old_rec that points to the metadata record.

4 files changed, 135 insertions, 82 deletions

diff --git a/storage/innobase/btr/btr0pcur.cc b/storage/innobase/btr/btr0pcur.cc
index 6b6158d5892..5a052e58e86 100644
--- a/storage/innobase/btr/btr0pcur.cc
+++ b/storage/innobase/btr/btr0pcur.cc
@@ -29,6 +29,9 @@ Created 2/23/1996 Heikki Tuuri
 #include "rem0cmp.h"
 #include "trx0trx.h"
 
+/* FIXME: move this to a proper .h file */
+extern const dtuple_t trx_undo_metadata;
+
 /**************************************************************//**
 Allocates memory for a persistent cursor object and initializes the cursor.
 @return own: persistent cursor */
@@ -177,9 +180,8 @@ btr_pcur_store_position(
 	if (index->is_ibuf()) {
 		ut_ad(!index->table->not_redundant());
 		cursor->old_n_fields = rec_get_n_fields_old(rec);
-	} else if (page_rec_is_leaf(rec)) {
-		cursor->old_n_fields = dict_index_get_n_unique_in_tree(index);
-	} else if (index->is_spatial()) {
+	} else if (UNIV_UNLIKELY(!page_rec_is_leaf(rec)
+				 && index->is_spatial())) {
 		ut_ad(dict_index_get_n_unique_in_tree_nonleaf(index)
 		      == DICT_INDEX_SPATIAL_NODEPTR_SIZE);
 		/* For R-tree, we have to compare
@@ -252,10 +254,8 @@ btr_pcur_restore_position_func(
 	mtr_t*		mtr)		/*!< in: mtr */
 {
 	dict_index_t*	index;
-	dtuple_t*	tuple;
 	page_cur_mode_t	mode;
 	page_cur_mode_t	old_mode;
-	mem_heap_t*	heap;
 
 	ut_ad(mtr->is_active());
 	//ut_ad(cursor->old_stored);
@@ -326,7 +326,7 @@ btr_pcur_restore_position_func(
 				const ulint*	offsets2;
 				rec = btr_pcur_get_rec(cursor);
 
-				heap = mem_heap_create(256);
+				mem_heap_t* heap = mem_heap_create(256);
 				offsets1 = rec_get_offsets(
 					cursor->old_rec, index, NULL,
 					REC_FMT_LEAF,
@@ -355,12 +355,34 @@ btr_pcur_restore_position_func(
 	}
 
 	/* If optimistic restoration did not succeed, open the cursor anew */
+	if (UNIV_UNLIKELY(reinterpret_cast<const byte*>(&trx_undo_metadata)
+			  == cursor->old_rec)) {
+		btr_pcur_open_at_index_side(true, index, cursor->latch_mode,
+					    cursor, true, 0, mtr);
+		ut_ad(btr_pcur_is_before_first_on_page(cursor));
+		btr_pcur_move_to_next_on_page(cursor);
+		/* We have to store the NEW value for the modify clock,
+		since the cursor can now be on a different page! */
+		if (!page_rec_is_metadata(btr_pcur_get_rec(cursor))) {
+not_found:
+			btr_pcur_store_position(cursor, mtr);
+			return(FALSE);
+		}
+found:
+		/* We can retain the value of old_rec */
+		cursor->block_when_stored = btr_pcur_get_block(cursor);
+		cursor->modify_clock = buf_block_get_modify_clock(
+			cursor->block_when_stored);
+		cursor->old_stored = true;
+		cursor->withdraw_clock = buf_withdraw_clock;
+		return(TRUE);
+	}
 
-	heap = mem_heap_create(256);
+	mem_heap_t* heap = mem_heap_create(256);
 
-	tuple = dict_index_build_data_tuple(cursor->old_rec, index,
-					    REC_FMT_LEAF,
-					    cursor->old_n_fields, heap);
+	const dtuple_t* tuple = dict_index_build_data_tuple(
+		cursor->old_rec, index, REC_FMT_LEAF,
+		cursor->old_n_fields, heap);
 
 	/* Save the old search mode of the cursor */
 	old_mode = cursor->search_mode;
@@ -404,31 +426,12 @@ btr_pcur_restore_position_func(
 					       ? REC_FMT_LEAF
 					       : REC_FMT_LEAF_FLEXIBLE,
 					       ULINT_UNDEFINED, &heap))) {
-
-		/* We have to store the NEW value for the modify clock,
-		since the cursor can now be on a different page!
-		But we can retain the value of old_rec */
-
-		cursor->block_when_stored = btr_pcur_get_block(cursor);
-		cursor->modify_clock = buf_block_get_modify_clock(
-						cursor->block_when_stored);
-		cursor->old_stored = true;
-		cursor->withdraw_clock = buf_withdraw_clock;
-
 		mem_heap_free(heap);
-
-		return(TRUE);
+		goto found;
 	}
 
 	mem_heap_free(heap);
-
-	/* We have to store new position information, modify_clock etc.,
-	to the cursor because it can now be on a different page, the record
-	under it may have been removed, etc. */
-
-	btr_pcur_store_position(cursor, mtr);
-
-	return(FALSE);
+	goto not_found;
 }
 
 /*********************************************************//**
diff --git a/storage/innobase/include/btr0pcur.h b/storage/innobase/include/btr0pcur.h
index b69d07811f3..a703272ae23 100644
--- a/storage/innobase/include/btr0pcur.h
+++ b/storage/innobase/include/btr0pcur.h
@@ -503,7 +503,7 @@ struct btr_pcur_t{
 	bool		old_stored;
 	/** if cursor position is stored, contains an initial segment of the
 	latest record cursor was positioned either on, before or after */
-	rec_t*		old_rec;
+	const rec_t*	old_rec;
 	/** number of fields in old_rec */
 	ulint		old_n_fields;
 	/** BTR_PCUR_ON, BTR_PCUR_BEFORE, or BTR_PCUR_AFTER, depending on
diff --git a/storage/innobase/include/rem0rec.h b/storage/innobase/include/rem0rec.h
index e020d25f926..9ec560f1c9a 100644
--- a/storage/innobase/include/rem0rec.h
+++ b/storage/innobase/include/rem0rec.h
@@ -1102,7 +1102,7 @@ rec_convert_dtuple_to_temp(
 Copies the first n fields of a physical record to a new physical record in
 a buffer.
 @return own: copied record */
-rec_t*
+const rec_t*
 rec_copy_prefix_to_buf(
 /*===================*/
 	const rec_t*		rec,		/*!< in: physical record */
diff --git a/storage/innobase/rem/rem0rec.cc b/storage/innobase/rem/rem0rec.cc
index 2eea8e8df15..d1a8a786dff 100644
--- a/storage/innobase/rem/rem0rec.cc
+++ b/storage/innobase/rem/rem0rec.cc
@@ -30,6 +30,9 @@ Created 5/30/1994 Heikki Tuuri
 #include "fts0fts.h"
 #include "trx0sys.h"
 
+/* FIXME: move this to a proper .h file */
+extern const dtuple_t trx_undo_metadata;
+
 /*			PHYSICAL RECORD (OLD STYLE)
 			===========================
 
@@ -1895,52 +1898,10 @@ rec_copy_prefix_to_dtuple(
 }
 
 /**************************************************************//**
-Copies the first n fields of an old-style physical record
-to a new physical record in a buffer.
-@return own: copied record */
-static
-rec_t*
-rec_copy_prefix_to_buf_old(
-/*=======================*/
-	const rec_t*	rec,		/*!< in: physical record */
-	ulint		n_fields,	/*!< in: number of fields to copy */
-	ulint		area_end,	/*!< in: end of the prefix data */
-	byte**		buf,		/*!< in/out: memory buffer for
-					the copied prefix, or NULL */
-	ulint*		buf_size)	/*!< in/out: buffer size */
-{
-	rec_t*	copy_rec;
-	ulint	area_start;
-	ulint	prefix_len;
-
-	if (rec_get_1byte_offs_flag(rec)) {
-		area_start = REC_N_OLD_EXTRA_BYTES + n_fields;
-	} else {
-		area_start = REC_N_OLD_EXTRA_BYTES + 2 * n_fields;
-	}
-
-	prefix_len = area_start + area_end;
-
-	if ((*buf == NULL) || (*buf_size < prefix_len)) {
-		ut_free(*buf);
-		*buf_size = prefix_len;
-		*buf = static_cast<byte*>(ut_malloc_nokey(prefix_len));
-	}
-
-	ut_memcpy(*buf, rec - area_start, prefix_len);
-
-	copy_rec = *buf + area_start;
-
-	rec_set_n_fields_old(copy_rec, n_fields);
-
-	return(copy_rec);
-}
-
-/**************************************************************//**
 Copies the first n fields of a physical record to a new physical record in
 a buffer.
 @return own: copied record */
-rec_t*
+const rec_t*
 rec_copy_prefix_to_buf(
 /*===================*/
 	const rec_t*		rec,		/*!< in: physical record */
@@ -1952,16 +1913,105 @@ rec_copy_prefix_to_buf(
 						or NULL */
 	ulint*			buf_size)	/*!< in/out: buffer size */
 {
-	ut_ad(n_fields <= index->n_fields || dict_index_is_ibuf(index));
+	ut_ad(n_fields <= index->n_fields || index->is_ibuf());
 	ut_ad(index->n_core_null_bytes <= UT_BITS_IN_BYTES(index->n_nullable));
 	UNIV_PREFETCH_RW(*buf);
+	ut_ad(!!page_rec_is_comp(rec) == index->table->not_redundant()
+	      || index->dual_format());
 
-	if (!dict_table_is_comp(index->table)) {
+	if (!page_rec_is_comp(rec)) {
 		ut_ad(rec_validate_old(rec));
-		return(rec_copy_prefix_to_buf_old(
-			       rec, n_fields,
-			       rec_get_field_start_offs(rec, n_fields),
-			       buf, buf_size));
+		const ulint data_size = rec_get_field_start_offs(rec, n_fields);
+		ulint extra_size;
+		const bool convert = index->table->not_redundant();
+		if (convert) {
+			ut_ad(n_fields <= index->first_user_field());
+			if (UNIV_UNLIKELY(rec_is_metadata(rec, false))
+			    && page_rec_is_leaf(rec)) {
+				return reinterpret_cast<const rec_t*>(
+					&trx_undo_metadata);
+			}
+			extra_size = (REC_N_NEW_EXTRA_BYTES + 1)
+				+ index->n_core_null_bytes;
+			if (rec_get_1byte_offs_flag(rec)) {
+				for (unsigned i = index->n_uniq; i--; ) {
+					if (!index->fields[i].fixed_len) {
+						extra_size++;
+					}
+				}
+			} else {
+				for (unsigned i = index->n_uniq; i--; ) {
+					const auto& f = index->fields[i];
+					if (f.fixed_len) {
+						continue;
+					}
+					if (!DATA_BIG_COL(f.col)
+					    || rec_get_nth_field_size(rec, i)
+					    < 128) {
+						extra_size++;
+					} else {
+						extra_size += 2;
+					}
+				}
+			}
+		} else {
+			extra_size = rec_get_1byte_offs_flag(rec)
+				? REC_N_OLD_EXTRA_BYTES + n_fields
+				: REC_N_OLD_EXTRA_BYTES + 2 * n_fields;
+		}
+		const ulint size = extra_size + data_size;
+		if (!*buf || *buf_size < size) {
+			ut_free(*buf);
+			*buf_size = size;
+			*buf = static_cast<byte*>(ut_malloc_nokey(size));
+		}
+		byte* copy_rec = *buf + extra_size;
+		if (convert) {
+			memcpy(*buf, rec, data_size);
+			byte* lens = copy_rec - (REC_N_NEW_EXTRA_BYTES + 2)
+				- index->n_core_null_bytes;
+			memset(lens, 0, copy_rec - lens);
+			compile_time_assert(REC_STATUS_ORDINARY == 0);
+
+			if (rec_get_1byte_offs_flag(rec)) {
+				for (unsigned i = 0; i < n_fields; i++) {
+					auto e = rec_1_get_prev_field_end_info(
+						rec, i);
+					/* PRIMARY KEY is NOT NULL */
+					ut_ad(!(e & REC_1BYTE_SQL_NULL_MASK));
+					ut_ad(e < 128);
+					if (!index->fields[i].fixed_len) {
+						*lens-- = e;
+					}
+				}
+			} else {
+				for (unsigned i = index->n_uniq; i--; ) {
+					const auto& f = index->fields[i];
+					if (f.fixed_len) {
+						continue;
+					}
+					auto e = rec_2_get_prev_field_end_info(
+						rec, i);
+					/* PRIMARY KEY is NOT NULL and
+					always stored locally */
+					ut_ad(!(e
+						& (REC_2BYTE_SQL_NULL_MASK
+						   | REC_2BYTE_EXTERN_MASK)));
+					ut_ad(e < 16384);
+					if (e < 128 || !DATA_BIG_COL(f.col)) {
+						*lens-- = e;
+					} else {
+						*lens-- = byte(e >> 8) | 0x80U;
+						*lens-- = byte(e & 0xff);
+					}
+				}
+			}
+			ut_ad(lens + 1 == *buf);
+		} else {
+			memcpy(*buf, rec - extra_size, size);
+			rec_set_n_fields_old(copy_rec, n_fields);
+		}
+		return *buf + extra_size;
 	}
 
 	ulint		prefix_len	= 0;