InnoDB: Write status bits to MLOG_COMP_REC_INSERT entries.

It is not safe to infer the status bits from the B-tree page
level, because after MLOG_COMP_LIST_END_COPY_CREATED, the
level will not be initialized before the records have been inserted.
(Bug #7973)


innobase/btr/btr0cur.c:
  Add parameter "offsets" to page_cur_insert_rec_low()
innobase/include/page0cur.h:
  page_cur_rec_insert(), page_cur_insert_rec_low(): Add param "offsets"
innobase/include/page0cur.ic:
  page_cur_rec_insert(), page_cur_insert_rec_low(): Add param "offsets"
innobase/include/rem0rec.h:
  Add rec_get_info_and_status_bits() and rec_set_info_and_status_bits()
innobase/include/rem0rec.ic:
  Add rec_get_info_and_status_bits() and rec_set_info_and_status_bits()
innobase/page/page0cur.c:
  page_cur_insert_rec_write_log(), page_cur_parse_insert_rec():
  write the status bits of the record to the log (Bug #7973)
  page_cur_insert_rec_low(): add parameter "offsets"
  page_copy_rec_list_end_to_created_page(): remove unnecessary call
  to mem_heap_create()
innobase/page/page0page.c:
  page_copy_rec_list_end_no_locks(), page_copy_rec_list_start():
  compute offsets and pass them to page_cur_rec_insert()

innobase/btr/btr0cur.c

@@ -1022,7 +1022,8 @@ btr_cur_optimistic_insert(
 
 	/* Now, try the insert */
 
-	*rec = page_cur_insert_rec_low(page_cursor, entry, index, NULL, mtr);
+	*rec = page_cur_insert_rec_low(page_cursor, entry, index,
+							NULL, NULL, mtr);
 	if (!(*rec)) {
 		/* If the record did not fit, reorganize */
 		btr_page_reorganize(page, index, mtr);

innobase/include/page0cur.h

@@ -144,6 +144,7 @@ page_cur_rec_insert(
 	page_cur_t*	cursor,	/* in: a page cursor */
 	rec_t*		rec,	/* in: record to insert */
 	dict_index_t*	index,	/* in: record descriptor */
+	ulint*		offsets,/* in: rec_get_offsets(rec, index) */
 	mtr_t*		mtr);	/* in: mini-transaction handle */
 /***************************************************************
 Inserts a record next to page cursor. Returns pointer to inserted record if
@@ -160,6 +161,7 @@ page_cur_insert_rec_low(
 	dtuple_t*	tuple,	/* in: pointer to a data tuple or NULL */
 	dict_index_t*	index,	/* in: record descriptor */
 	rec_t*		rec,	/* in: pointer to a physical record or NULL */
+	ulint*		offsets,/* in: rec_get_offsets(rec, index) or NULL */
 	mtr_t*		mtr);	/* in: mini-transaction handle */
 /*****************************************************************
 Copies records from page to a newly created page, from a given record onward,

innobase/include/page0cur.ic

@@ -195,7 +195,7 @@ page_cur_tuple_insert(
 	dict_index_t*	index,	/* in: record descriptor */
 	mtr_t*		mtr)	/* in: mini-transaction handle */
 {
-	return(page_cur_insert_rec_low(cursor, tuple, index, NULL, mtr));
+	return(page_cur_insert_rec_low(cursor, tuple, index, NULL, NULL, mtr));
 }
 
 /***************************************************************
@@ -211,8 +211,10 @@ page_cur_rec_insert(
 	page_cur_t*	cursor,	/* in: a page cursor */
 	rec_t*		rec,	/* in: record to insert */
 	dict_index_t*	index,	/* in: record descriptor */
+	ulint*		offsets,/* in: rec_get_offsets(rec, index) */
 	mtr_t*		mtr)	/* in: mini-transaction handle */
 {
-	return(page_cur_insert_rec_low(cursor, NULL, index, rec, mtr));
+	return(page_cur_insert_rec_low(cursor, NULL, index, rec,
+				offsets, mtr));
 }
 

innobase/include/rem0rec.h

@@ -133,6 +133,27 @@ rec_set_status(
 	rec_t*	rec,	/* in: physical record */
 	ulint	bits);	/* in: info bits */
 
+/**********************************************************
+The following function is used to retrieve the info and status
+bits of a record.  (Only compact records have status bits.) */
+UNIV_INLINE
+ulint
+rec_get_info_and_status_bits(
+/*==============*/
+			/* out: info bits */
+	rec_t*	rec,	/* in: physical record */
+	ibool	comp);	/* in: TRUE=compact page format */
+/**********************************************************
+The following function is used to set the info and status
+bits of a record.  (Only compact records have status bits.) */
+UNIV_INLINE
+void
+rec_set_info_and_status_bits(
+/*==============*/
+	rec_t*	rec,	/* in: physical record */
+	ibool	comp,	/* in: TRUE=compact page format */
+	ulint	bits);	/* in: info bits */
+
 /**********************************************************
 The following function tells if record is delete marked. */
 UNIV_INLINE

innobase/include/rem0rec.ic

@@ -520,6 +520,53 @@ rec_set_status(
 				REC_NEW_STATUS_MASK, REC_NEW_STATUS_SHIFT);
 }
 
+/**********************************************************
+The following function is used to retrieve the info and status
+bits of a record.  (Only compact records have status bits.) */
+UNIV_INLINE
+ulint
+rec_get_info_and_status_bits(
+/*==============*/
+			/* out: info bits */
+	rec_t*	rec,	/* in: physical record */
+	ibool	comp)	/* in: TRUE=compact page format */
+{
+	ulint	bits;
+#if (REC_NEW_STATUS_MASK >> REC_NEW_STATUS_SHIFT) \
+& (REC_INFO_BITS_MASK >> REC_INFO_BITS_SHIFT)
+# error "REC_NEW_STATUS_MASK and REC_INFO_BITS_MASK overlap"
+#endif
+	if (comp) {
+		bits = rec_get_info_bits(rec, TRUE) | rec_get_status(rec);
+	} else {
+		bits = rec_get_info_bits(rec, FALSE);
+		ut_ad(!(bits & ~(REC_INFO_BITS_MASK >> REC_INFO_BITS_SHIFT)));
+	}
+	return(bits);
+}
+/**********************************************************
+The following function is used to set the info and status
+bits of a record.  (Only compact records have status bits.) */
+UNIV_INLINE
+void
+rec_set_info_and_status_bits(
+/*==============*/
+	rec_t*	rec,	/* in: physical record */
+	ibool	comp,	/* in: TRUE=compact page format */
+	ulint	bits)	/* in: info bits */
+{
+#if (REC_NEW_STATUS_MASK >> REC_NEW_STATUS_SHIFT) \
+& (REC_INFO_BITS_MASK >> REC_INFO_BITS_SHIFT)
+# error "REC_NEW_STATUS_MASK and REC_INFO_BITS_MASK overlap"
+#endif
+	if (comp) {
+		rec_set_status(rec, bits & REC_NEW_STATUS_MASK);
+	} else {
+		ut_ad(!(bits & ~(REC_INFO_BITS_MASK >> REC_INFO_BITS_SHIFT)));
+	}
+	rec_set_info_bits(rec, bits & ~REC_NEW_STATUS_MASK, comp);
+}
+
 /**********************************************************
 The following function tells if record is delete marked. */
 UNIV_INLINE

innobase/page/page0cur.c

@@ -605,8 +605,8 @@ page_cur_insert_rec_write_log(
 		log_end = &log_ptr[5 + 1 + 5 + 5 + MLOG_BUF_MARGIN];
 	}
 
-	if ((rec_get_info_bits(insert_rec, index->table->comp) !=
-	     rec_get_info_bits(cursor_rec, index->table->comp))
+	if ((rec_get_info_and_status_bits(insert_rec, index->table->comp) !=
+	     rec_get_info_and_status_bits(cursor_rec, index->table->comp))
 	    || (extra_size != cur_extra_size)
 	    || (rec_size != cur_rec_size)) {
 
@@ -622,7 +622,8 @@ page_cur_insert_rec_write_log(
 	if (extra_info_yes) {
 		/* Write the info bits */
 		mach_write_to_1(log_ptr,
-			rec_get_info_bits(insert_rec, index->table->comp));
+			rec_get_info_and_status_bits(insert_rec,
+							index->table->comp));
 		log_ptr++;
 
 		/* Write the record origin offset */
@@ -673,7 +674,7 @@ page_cur_parse_insert_rec(
 	byte	buf1[1024];
 	byte*	buf;
 	byte*   ptr2 = ptr;
-	ulint	info_bits = 0; /* remove warning */
+	ulint	info_and_status_bits = 0; /* remove warning */
 	page_cur_t cursor;
 	mem_heap_t*	heap		= NULL;
 	ulint		offsets_[100]	= { 100, };
@@ -723,7 +724,7 @@ page_cur_parse_insert_rec(
 			return(NULL);
 		}
 		
-		info_bits = mach_read_from_1(ptr);
+		info_and_status_bits = mach_read_from_1(ptr);
 		ptr++;
 
 		ptr = mach_parse_compressed(ptr, end_ptr, &origin_offset);
@@ -768,7 +769,8 @@ page_cur_parse_insert_rec(
 						ULINT_UNDEFINED, &heap);
 
 	if (extra_info_yes == 0) {
-		info_bits = rec_get_info_bits(cursor_rec, index->table->comp);
+		info_and_status_bits = rec_get_info_and_status_bits(
+					cursor_rec, index->table->comp);
 		origin_offset = rec_offs_extra_size(offsets);
 		mismatch_index = rec_offs_size(offsets) - end_seg_len;
 	}
@@ -783,11 +785,12 @@ page_cur_parse_insert_rec(
 	
         if (mismatch_index >= UNIV_PAGE_SIZE) {
 		fprintf(stderr,
-			"Is short %lu, info_bits %lu, offset %lu, "
+			"Is short %lu, info_and_status_bits %lu, offset %lu, "
 			"o_offset %lu\n"
                     "mismatch index %lu, end_seg_len %lu\n"
                     "parsed len %lu\n",
-		    (ulong) is_short, (ulong) info_bits, (ulong) offset,
+		    (ulong) is_short, (ulong) info_and_status_bits,
+		    (ulong) offset,
 		    (ulong) origin_offset,
 		    (ulong) mismatch_index, (ulong) end_seg_len,
 		    (ulong) (ptr - ptr2));
@@ -803,21 +806,14 @@ page_cur_parse_insert_rec(
 	ut_memcpy(buf, rec_get_start(cursor_rec, offsets), mismatch_index);
 	ut_memcpy(buf + mismatch_index, ptr, end_seg_len);
 
-	rec_set_info_bits(buf + origin_offset, index->table->comp, info_bits);
-
-	/* Set the status bits for new-style records. */
-	if (index->table->comp) {
-		/* Leaf pages (level 0) contain ordinary records;
-		non-leaf pages contain node pointer records. */
-		ulint	level	= page_header_get_field(
-				buf_frame_align(cursor_rec), PAGE_LEVEL);
-		rec_set_status(buf + origin_offset,
-			level ? REC_STATUS_NODE_PTR : REC_STATUS_ORDINARY);
-	}
+	rec_set_info_and_status_bits(buf + origin_offset, index->table->comp,
+							info_and_status_bits);
 
 	page_cur_position(cursor_rec, &cursor);
 
-	page_cur_rec_insert(&cursor, buf + origin_offset, index, mtr);
+	offsets = rec_get_offsets(buf + origin_offset, index, offsets,
+						ULINT_UNDEFINED, &heap);
+	page_cur_rec_insert(&cursor, buf + origin_offset, index, offsets, mtr);
 
 	if (buf != buf1) {
 
@@ -846,6 +842,7 @@ page_cur_insert_rec_low(
 	dtuple_t*	tuple,	/* in: pointer to a data tuple or NULL */
 	dict_index_t*	index,	/* in: record descriptor */
 	rec_t*		rec,	/* in: pointer to a physical record or NULL */
+	ulint*		offsets,/* in: rec_get_offsets(rec, index) or NULL */
 	mtr_t*		mtr)	/* in: mini-transaction handle */
 {
 	byte*		insert_buf	= NULL;
@@ -863,8 +860,6 @@ page_cur_insert_rec_low(
 	rec_t*		owner_rec;
 	ulint		n_owned;
 	mem_heap_t*	heap		= NULL;
-	ulint		offsets_[100]	= { 100, };
-	ulint*		offsets		= offsets_;
 	ibool		comp		= index->table->comp;
 
 	ut_ad(cursor && mtr);
@@ -882,8 +877,11 @@ page_cur_insert_rec_low(
 	if (tuple != NULL) {
 		rec_size = rec_get_converted_size(index, tuple);
 	} else {
-		offsets = rec_get_offsets(rec, index, offsets,
+		if (!offsets) {
+			offsets = rec_get_offsets(rec, index, offsets,
 						ULINT_UNDEFINED, &heap);
+		}
+		ut_ad(rec_offs_validate(rec, index, offsets));
 		rec_size = rec_offs_size(offsets);
 	}
 
@@ -1131,8 +1129,6 @@ page_copy_rec_list_end_to_created_page(
 	slot_index = 0;
 	n_recs = 0;
 
-	heap = mem_heap_create(100);
-
 	/* should be do ... until, comment by Jani */
 	while (rec != page_get_supremum_rec(page)) {
 		offsets = rec_get_offsets(rec, index, offsets,

innobase/page/page0page.c

@@ -463,6 +463,9 @@ page_copy_rec_list_end_no_locks(
 	page_cur_t	cur1;
 	page_cur_t	cur2;
 	rec_t*		sup;
+	mem_heap_t*	heap		= NULL;
+	ulint		offsets_[100]	= { 100, };
+	ulint*		offsets		= offsets_;
 
 	page_cur_position(rec, &cur1);
 
@@ -483,8 +486,11 @@ page_copy_rec_list_end_no_locks(
 	sup = page_get_supremum_rec(page);
 	
 	while (sup != page_cur_get_rec(&cur1)) {
-		if (!page_cur_rec_insert(&cur2,
-					page_cur_get_rec(&cur1), index, mtr)) {
+		rec_t*	cur1_rec = page_cur_get_rec(&cur1);
+		offsets = rec_get_offsets(cur1_rec, index, offsets,
+					ULINT_UNDEFINED, &heap);
+		if (!page_cur_rec_insert(&cur2, cur1_rec, index,
+							offsets, mtr)) {
 			/* Track an assertion failure reported on the mailing
 			list on June 18th, 2003 */
 
@@ -503,7 +509,11 @@ page_copy_rec_list_end_no_locks(
 		page_cur_move_to_next(&cur1);
 		page_cur_move_to_next(&cur2);
 	}
-}	
+
+	if (heap) {
+		mem_heap_free(heap);
+	}
+}
 
 /*****************************************************************
 Copies records from page to new_page, from a given record onward,
@@ -553,6 +563,9 @@ page_copy_rec_list_start(
 	page_cur_t	cur1;
 	page_cur_t	cur2;
 	rec_t*		old_end;
+	mem_heap_t*	heap		= NULL;
+	ulint		offsets_[100]	= { 100, };
+	ulint*		offsets		= offsets_;
 
 	page_cur_set_before_first(page, &cur1);
 
@@ -570,8 +583,13 @@ page_copy_rec_list_start(
 	/* Copy records from the original page to the new page */	
 
 	while (page_cur_get_rec(&cur1) != rec) {
-		ut_a(page_cur_rec_insert(&cur2,
-				page_cur_get_rec(&cur1), index, mtr));
+		rec_t*	ins_rec;
+		rec_t*	cur1_rec = page_cur_get_rec(&cur1);
+		offsets = rec_get_offsets(cur1_rec, index, offsets,
+					ULINT_UNDEFINED, &heap);
+		ins_rec = page_cur_rec_insert(&cur2, cur1_rec, index,
+						offsets, mtr);
+		ut_a(ins_rec);
 
 		page_cur_move_to_next(&cur1);
 		page_cur_move_to_next(&cur2);
@@ -584,7 +602,11 @@ page_copy_rec_list_start(
 	page_update_max_trx_id(new_page, page_get_max_trx_id(page));
 
 	btr_search_move_or_delete_hash_entries(new_page, page, index);
-}	
+
+	if (heap) {
+		mem_heap_free(heap);
+	}
+}
 
 /**************************************************************
 Writes a log record of a record list end or start deletion. */