MDEV-12353: Replace DELETE_MARK redo log records with MLOG_WRITE_STRING

btr_cur_upd_rec_sys(): Replaces row_upd_rec_sys_fields() and
implements redo logging.

row_upd_rec_sys_fields_in_recovery(): Remove, and merge to the
only remaining caller btr_cur_parse_update_in_place().

btr_cur_del_mark_set_clust_rec_log(),
btr_cur_del_mark_set_sec_rec_log(),
btr_cur_set_deleted_flag_for_ibuf():
Remove, and replace with btr_rec_set_deleted<bool>().

page_zip_rec_set_deleted(): Add the parameter mtr, and write a
MLOG_ZIP_WRITE_STRING record to the log.

storage/innobase/btr/btr0cur.cc

@@ -3979,9 +3979,90 @@ btr_cur_update_in_place_log(
 	row_upd_index_write_log(update, log_ptr, mtr);
 }
 
+/** Update DB_TRX_ID, DB_ROLL_PTR in a clustered index record.
+@param[in,out]	block		clustered index leaf page
+@param[in,out]	rec		clustered index record
+@param[in]	index		clustered index
+@param[in]	offsets		rec_get_offsets(rec, index)
+@param[in]	trx		transaction
+@param[in]	roll_ptr	DB_ROLL_PTR value
+@param[in,out]	mtr		mini-transaction */
+static void btr_cur_upd_rec_sys(buf_block_t *block, rec_t* rec,
+				dict_index_t* index, const offset_t* offsets,
+				const trx_t* trx, roll_ptr_t roll_ptr,
+				mtr_t* mtr)
+{
+	ut_ad(index->is_primary());
+	ut_ad(rec_offs_validate(rec, index, offsets));
+
+	if (UNIV_LIKELY_NULL(block->page.zip.data)) {
+		page_zip_write_trx_id_and_roll_ptr(&block->page.zip,
+						   rec, offsets,
+						   index->db_trx_id(),
+						   trx->id, roll_ptr, mtr);
+	} else {
+		ulint	offset = index->trx_id_offset;
+
+		if (!offset) {
+			offset = row_get_trx_id_offset(index, offsets);
+		}
+
+		compile_time_assert(DATA_TRX_ID + 1 == DATA_ROLL_PTR);
+
+		/* During IMPORT the trx id in the record can be in the
+		future, if the .ibd file is being imported from another
+		instance. During IMPORT roll_ptr will be 0. */
+		ut_ad(roll_ptr == 0
+		      || lock_check_trx_id_sanity(
+			      trx_read_trx_id(rec + offset),
+			      rec, index, offsets));
+
+		trx_write_trx_id(rec + offset, trx->id);
+		trx_write_roll_ptr(rec + offset + DATA_TRX_ID_LEN, roll_ptr);
+		/* MDEV-12353 FIXME: consider emitting MEMMOVE for the
+		DB_TRX_ID if it is found in the preceding record */
+		mtr->memcpy(*block, page_offset(rec + offset),
+			    DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN);
+	}
+}
+
+/*********************************************************************//**
+Parses the log data of system field values.
+@return log data end or NULL */
+static
+byte*
+row_upd_parse_sys_vals(
+/*===================*/
+	const byte*	ptr,	/*!< in: buffer */
+	const byte*	end_ptr,/*!< in: buffer end */
+	ulint*		pos,	/*!< out: TRX_ID position in record */
+	trx_id_t*	trx_id,	/*!< out: trx id */
+	roll_ptr_t*	roll_ptr)/*!< out: roll ptr */
+{
+	*pos = mach_parse_compressed(&ptr, end_ptr);
+
+	if (ptr == NULL) {
+
+		return(NULL);
+	}
+
+	if (end_ptr < ptr + DATA_ROLL_PTR_LEN) {
+
+		return(NULL);
+	}
+
+	*roll_ptr = trx_read_roll_ptr(ptr);
+	ptr += DATA_ROLL_PTR_LEN;
+
+	*trx_id = mach_u64_parse_compressed(&ptr, end_ptr);
+
+	return(const_cast<byte*>(ptr));
+}
+
 /***********************************************************//**
 Parses a redo log record of updating a record in-place.
 @return end of log record or NULL */
+ATTRIBUTE_COLD /* only used when crash-upgrading */
 const byte*
 btr_cur_parse_update_in_place(
 /*==========================*/
@@ -4052,9 +4133,17 @@ btr_cur_parse_update_in_place(
 				  || page_is_leaf(page),
 				  ULINT_UNDEFINED, &heap);
 
-	if (!(flags & BTR_KEEP_SYS_FLAG)) {
+	if (flags & BTR_KEEP_SYS_FLAG) {
-		row_upd_rec_sys_fields_in_recovery(rec, page_zip, offsets,
+	} else if (page_zip) {
-						   pos, trx_id, roll_ptr);
+		page_zip_write_trx_id_and_roll_ptr(
+			page_zip, rec, offsets, pos, trx_id, roll_ptr, NULL);
+	} else {
+		ulint len;
+		byte* field = rec_get_nth_field(rec, offsets, pos, &len);
+		ut_ad(len == DATA_TRX_ID_LEN);
+		compile_time_assert(DATA_TRX_ID + 1 == DATA_ROLL_PTR);
+		trx_write_trx_id(field, trx_id);
+		trx_write_roll_ptr(field + DATA_TRX_ID_LEN, roll_ptr);
 	}
 
 	row_upd_rec_in_place(rec, index, offsets, update, page_zip);
@@ -4235,8 +4324,8 @@ btr_cur_update_in_place(
 	}
 
 	if (!(flags & BTR_KEEP_SYS_FLAG)) {
-		row_upd_rec_sys_fields(rec, NULL, index, offsets,
+		btr_cur_upd_rec_sys(block, rec, index, offsets,
-				       thr_get_trx(thr), roll_ptr);
+				    thr_get_trx(thr), roll_ptr, mtr);
 	}
 
 	was_delete_marked = rec_get_deleted_flag(
@@ -5229,50 +5318,48 @@ return_after_reservations:
 
 /*==================== B-TREE DELETE MARK AND UNMARK ===============*/
 
-/****************************************************************//**
+/** Modify the delete-mark flag of a record.
-Writes the redo log record for delete marking or unmarking of an index
+@tparam         flag    the value of the delete-mark flag
-record. */
+@param[in,out]  block   buffer block
-UNIV_INLINE
+@param[in,out]  rec     record on a physical index page
-void
+@param[in,out]  mtr     mini-transaction  */
-btr_cur_del_mark_set_clust_rec_log(
+template<bool flag>
-/*===============================*/
+void btr_rec_set_deleted(buf_block_t *block, rec_t *rec, mtr_t *mtr)
-	rec_t*		rec,	/*!< in: record */
-	dict_index_t*	index,	/*!< in: index of the record */
-	trx_id_t	trx_id,	/*!< in: transaction id */
-	roll_ptr_t	roll_ptr,/*!< in: roll ptr to the undo log record */
-	mtr_t*		mtr)	/*!< in: mtr */
 {
-	byte*	log_ptr;
+  if (page_rec_is_comp(rec))
-
+  {
-	ut_ad(!!page_rec_is_comp(rec) == dict_table_is_comp(index->table));
+    byte *b= &rec[-REC_NEW_INFO_BITS];
-	ut_ad(mtr->is_named_space(index->table->space));
+    const byte v= flag
-
+      ? (*b | REC_INFO_DELETED_FLAG)
-	log_ptr = mlog_open_and_write_index(mtr, rec, index,
+      : (*b & ~REC_INFO_DELETED_FLAG);
-					    page_rec_is_comp(rec)
+    if (*b == v);
-					    ? MLOG_COMP_REC_CLUST_DELETE_MARK
+    else if (UNIV_LIKELY_NULL(block->page.zip.data))
-					    : MLOG_REC_CLUST_DELETE_MARK,
+    {
-					    1 + 1 + DATA_ROLL_PTR_LEN
+      *b= v;
-					    + 14 + 2);
+      page_zip_rec_set_deleted(&block->page.zip, rec, flag, mtr);
-
+    }
-	if (!log_ptr) {
+    else
-		/* Logging in mtr is switched off during crash recovery */
+      mtr->write<1>(*block, b, v);
-		return;
+  }
-	}
+  else
-
+  {
-	*log_ptr++ = 0;
+    ut_ad(!block->page.zip.data);
-	*log_ptr++ = 1;
+    byte *b= &rec[-REC_OLD_INFO_BITS];
-
+    const byte v = flag
-	log_ptr = btr_cur_log_sys(index, trx_id, roll_ptr, log_ptr);
+      ? (*b | REC_INFO_DELETED_FLAG)
-	mach_write_to_2(log_ptr, page_offset(rec));
+      : (*b & ~REC_INFO_DELETED_FLAG);
-	log_ptr += 2;
+    mtr->write<1,mtr_t::OPT>(*block, b, v);
-
+  }
-	mlog_close(mtr, log_ptr);
 }
 
+template void btr_rec_set_deleted<false>(buf_block_t *, rec_t *, mtr_t *);
+template void btr_rec_set_deleted<true>(buf_block_t *, rec_t *, mtr_t *);
+
 /****************************************************************//**
 Parses the redo log record for delete marking or unmarking of a clustered
 index record.
 @return end of log record or NULL */
+ATTRIBUTE_COLD /* only used when crash-upgrading */
 const byte*
 btr_cur_parse_del_mark_set_clust_rec(
 /*=================================*/
@@ -5280,7 +5367,8 @@ btr_cur_parse_del_mark_set_clust_rec(
 	const byte*	end_ptr,/*!< in: buffer end */
 	page_t*		page,	/*!< in/out: page or NULL */
 	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
-	dict_index_t*	index)	/*!< in: index corresponding to page */
+	dict_index_t*	index,	/*!< in: index corresponding to page */
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 {
 	ulint		flags;
 	ulint		val;
@@ -5331,31 +5419,57 @@ btr_cur_parse_del_mark_set_clust_rec(
 		is only being recovered, and there cannot be a hash index to
 		it. Besides, these fields are being updated in place
 		and the adaptive hash index does not depend on them. */
+		byte* b = rec - (page_is_comp(page)
+				 ? REC_NEW_INFO_BITS
+				 : REC_OLD_INFO_BITS);
+
+		if (val) {
+			*b |= REC_INFO_DELETED_FLAG;
+		} else {
+			*b &= ~REC_INFO_DELETED_FLAG;
+		}
+
+		if (UNIV_LIKELY_NULL(page_zip)) {
+			page_zip_rec_set_deleted(page_zip, rec, val, mtr);
+		}
 
-		btr_rec_set_deleted_flag(rec, page_zip, val);
 		/* pos is the offset of DB_TRX_ID in the clustered index.
 		Debug assertions may also access DB_ROLL_PTR at pos+1.
 		Therefore, we must compute offsets for the first pos+2
 		clustered index fields. */
 		ut_ad(pos <= MAX_REF_PARTS);
 
-		offset_t offsets[REC_OFFS_HEADER_SIZE + MAX_REF_PARTS + 2];
+		offset_t offsets_[REC_OFFS_HEADER_SIZE + MAX_REF_PARTS + 2];
-		rec_offs_init(offsets);
+		rec_offs_init(offsets_);
 		mem_heap_t*	heap	= NULL;
 
 		if (!(flags & BTR_KEEP_SYS_FLAG)) {
-			row_upd_rec_sys_fields_in_recovery(
+			offset_t* offsets = rec_get_offsets(rec, index,
-				rec, page_zip,
+							    offsets_, true,
-				rec_get_offsets(rec, index, offsets, true,
+							    pos + 2, &heap);
-						pos + 2, &heap),
+			if (page_zip) {
-				pos, trx_id, roll_ptr);
+				page_zip_write_trx_id_and_roll_ptr(
+					page_zip, rec, offsets, pos, trx_id,
+					roll_ptr, mtr);
+			} else {
+				ulint	len;
+
+				byte* field = rec_get_nth_field(
+					rec, offsets, pos, &len);
+				ut_ad(len == DATA_TRX_ID_LEN);
+				compile_time_assert(DATA_TRX_ID + 1
+						    == DATA_ROLL_PTR);
+				trx_write_trx_id(field, trx_id);
+				trx_write_roll_ptr(field + DATA_TRX_ID_LEN,
+						   roll_ptr);
+			}
 		} else {
 			/* In delete-marked records, DB_TRX_ID must
 			always refer to an existing undo log record. */
 			ut_ad(memcmp(rec_get_nth_field(
 					     rec,
 					     rec_get_offsets(rec, index,
-							     offsets, true,
+							     offsets_, true,
 							     pos, &heap),
 					     pos, &offset),
 				     field_ref_zero, DATA_TRX_ID_LEN));
@@ -5427,9 +5541,7 @@ btr_cur_del_mark_set_clust_rec(
 	the adaptive hash index does not depend on the delete-mark
 	and the delete-mark is being updated in place. */
 
-	page_zip_des_t* page_zip = buf_block_get_page_zip(block);
+	btr_rec_set_deleted<true>(block, rec, mtr);
-
-	btr_rec_set_deleted_flag(rec, page_zip, TRUE);
 
 	trx = thr_get_trx(thr);
 
@@ -5443,58 +5555,23 @@ btr_cur_del_mark_set_clust_rec(
 		row_log_table_delete(rec, index, offsets, NULL);
 	}
 
-	row_upd_rec_sys_fields(rec, page_zip, index, offsets, trx, roll_ptr);
+	btr_cur_upd_rec_sys(block, rec, index, offsets, trx, roll_ptr, mtr);
-
-	btr_cur_del_mark_set_clust_rec_log(rec, index, trx->id,
-					   roll_ptr, mtr);
-
 	return(err);
 }
 
-/****************************************************************//**
-Writes the redo log record for a delete mark setting of a secondary
-index record. */
-UNIV_INLINE
-void
-btr_cur_del_mark_set_sec_rec_log(
-/*=============================*/
-	rec_t*		rec,	/*!< in: record */
-	ibool		val,	/*!< in: value to set */
-	mtr_t*		mtr)	/*!< in: mtr */
-{
-	byte*	log_ptr;
-	ut_ad(val <= 1);
-
-	log_ptr = mlog_open(mtr, 11 + 1 + 2);
-
-	if (!log_ptr) {
-		/* Logging in mtr is switched off during crash recovery:
-		in that case mlog_open returns NULL */
-		return;
-	}
-
-	log_ptr = mlog_write_initial_log_record_fast(
-		rec, MLOG_REC_SEC_DELETE_MARK, log_ptr, mtr);
-	mach_write_to_1(log_ptr, val);
-	log_ptr++;
-
-	mach_write_to_2(log_ptr, page_offset(rec));
-	log_ptr += 2;
-
-	mlog_close(mtr, log_ptr);
-}
-
 /****************************************************************//**
 Parses the redo log record for delete marking or unmarking of a secondary
 index record.
 @return end of log record or NULL */
+ATTRIBUTE_COLD /* only used when crash-upgrading */
 const byte*
 btr_cur_parse_del_mark_set_sec_rec(
 /*===============================*/
 	const byte*	ptr,	/*!< in: buffer */
 	const byte*	end_ptr,/*!< in: buffer end */
 	page_t*		page,	/*!< in/out: page or NULL */
-	page_zip_des_t*	page_zip)/*!< in/out: compressed page, or NULL */
+	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 {
 	ulint	val;
 	ulint	offset;
@@ -5520,86 +5597,24 @@ btr_cur_parse_del_mark_set_sec_rec(
 		is only being recovered, and there cannot be a hash index to
 		it. Besides, the delete-mark flag is being updated in place
 		and the adaptive hash index does not depend on it. */
+		byte* b = page + offset - (page_is_comp(page)
+					   ? REC_NEW_INFO_BITS
+					   : REC_OLD_INFO_BITS);
 
-		btr_rec_set_deleted_flag(rec, page_zip, val);
+		if (val) {
+			*b |= REC_INFO_DELETED_FLAG;
+		} else {
+			*b &= ~REC_INFO_DELETED_FLAG;
+		}
+
+		if (UNIV_LIKELY_NULL(page_zip)) {
+			page_zip_rec_set_deleted(page_zip, rec, val, mtr);
+		}
 	}
 
 	return(ptr);
 }
 
-/***********************************************************//**
-Sets a secondary index record delete mark to TRUE or FALSE.
-@return DB_SUCCESS, DB_LOCK_WAIT, or error number */
-dberr_t
-btr_cur_del_mark_set_sec_rec(
-/*=========================*/
-	ulint		flags,	/*!< in: locking flag */
-	btr_cur_t*	cursor,	/*!< in: cursor */
-	ibool		val,	/*!< in: value to set */
-	que_thr_t*	thr,	/*!< in: query thread */
-	mtr_t*		mtr)	/*!< in/out: mini-transaction */
-{
-	buf_block_t*	block;
-	rec_t*		rec;
-	dberr_t		err;
-
-	block = btr_cur_get_block(cursor);
-	rec = btr_cur_get_rec(cursor);
-
-	err = lock_sec_rec_modify_check_and_lock(flags,
-						 btr_cur_get_block(cursor),
-						 rec, cursor->index, thr, mtr);
-	if (err != DB_SUCCESS) {
-
-		return(err);
-	}
-
-	ut_ad(!!page_rec_is_comp(rec)
-	      == dict_table_is_comp(cursor->index->table));
-
-	DBUG_PRINT("ib_cur", ("delete-mark=%u sec %u:%u:%u in %s("
-			      IB_ID_FMT ") by " TRX_ID_FMT,
-			      unsigned(val),
-			      block->page.id.space(), block->page.id.page_no(),
-			      unsigned(page_rec_get_heap_no(rec)),
-			      cursor->index->name(), cursor->index->id,
-			      trx_get_id_for_print(thr_get_trx(thr))));
-
-	/* We do not need to reserve search latch, as the
-	delete-mark flag is being updated in place and the adaptive
-	hash index does not depend on it. */
-	btr_rec_set_deleted_flag(rec, buf_block_get_page_zip(block), val);
-
-	btr_cur_del_mark_set_sec_rec_log(rec, val, mtr);
-
-	return(DB_SUCCESS);
-}
-
-/***********************************************************//**
-Sets a secondary index record's delete mark to the given value. This
-function is only used by the insert buffer merge mechanism. */
-void
-btr_cur_set_deleted_flag_for_ibuf(
-/*==============================*/
-	rec_t*		rec,		/*!< in/out: record */
-	page_zip_des_t*	page_zip,	/*!< in/out: compressed page
-					corresponding to rec, or NULL
-					when the tablespace is
-					uncompressed */
-	ibool		val,		/*!< in: value to set */
-	mtr_t*		mtr)		/*!< in/out: mini-transaction */
-{
-	/* We do not need to reserve search latch, as the page
-	has just been read to the buffer pool and there cannot be
-	a hash index to it.  Besides, the delete-mark flag is being
-	updated in place and the adaptive hash index does not depend
-	on it. */
-
-	btr_rec_set_deleted_flag(rec, page_zip, val);
-
-	btr_cur_del_mark_set_sec_rec_log(rec, val, mtr);
-}
-
 /*==================== B-TREE RECORD REMOVE =========================*/
 
 /*************************************************************//**

storage/innobase/gis/gis0rtree.cc

@@ -1442,8 +1442,8 @@ rtr_page_copy_rec_list_end_no_locks(
 					/* We have two identical leaf records,
 					skip copying the undeleted one, and
 					unmark deleted on the current page */
-					btr_rec_set_deleted_flag(
+					btr_rec_set_deleted<false>(
-						cur_rec, NULL, FALSE);
+						new_block, cur_rec, mtr);
 					goto next;
 				}
 			}
@@ -1562,8 +1562,8 @@ rtr_page_copy_rec_list_start_no_locks(
 					/* We have two identical leaf records,
 					skip copying the undeleted one, and
 					unmark deleted on the current page */
-					btr_rec_set_deleted_flag(
+					btr_rec_set_deleted<false>(
-						cur_rec, NULL, FALSE);
+						new_block, cur_rec, mtr);
 					goto next;
 				}
 			}

storage/innobase/ibuf/ibuf0ibuf.cc

@@ -3808,20 +3808,19 @@ dump:
 					  ULINT_UNDEFINED, &heap);
 		update = row_upd_build_sec_rec_difference_binary(
 			rec, index, offsets, entry, heap);
-		page_zip_des_t* page_zip = buf_block_get_page_zip(block);
 
 		if (update->n_fields == 0) {
 			/* The records only differ in the delete-mark.
 			Clear the delete-mark, like we did before
 			Bug #56680 was fixed. */
-			btr_cur_set_deleted_flag_for_ibuf(
+			btr_rec_set_deleted<false>(block, rec, mtr);
-				rec, page_zip, FALSE, mtr);
 			goto updated_in_place;
 		}
 
 		/* Copy the info bits. Clear the delete-mark. */
 		update->info_bits = rec_get_info_bits(rec, page_is_comp(page));
 		update->info_bits &= ~REC_INFO_DELETED_FLAG;
+		page_zip_des_t* page_zip = buf_block_get_page_zip(block);
 
 		/* We cannot invoke btr_cur_optimistic_update() here,
 		because we do not have a btr_cur_t or que_thr_t,
@@ -3919,11 +3918,7 @@ ibuf_set_del_mark(
 	low_match = page_cur_search(block, index, entry, &page_cur);
 
 	if (low_match == dtuple_get_n_fields(entry)) {
-		rec_t*		rec;
+		rec_t* rec = page_cur_get_rec(&page_cur);
-		page_zip_des_t* page_zip;
-
-		rec = page_cur_get_rec(&page_cur);
-		page_zip = page_cur_get_page_zip(&page_cur);
 
 		/* Delete mark the old index record. According to a
 		comment in row_upd_sec_index_entry(), it can already
@@ -3934,8 +3929,7 @@ ibuf_set_del_mark(
 		if (UNIV_LIKELY
 		    (!rec_get_deleted_flag(
 			    rec, dict_table_is_comp(index->table)))) {
-			btr_cur_set_deleted_flag_for_ibuf(rec, page_zip,
+			btr_rec_set_deleted<true>(block, rec, mtr);
-							  TRUE, mtr);
 		}
 	} else {
 		const page_t*		page
@@ -4151,8 +4145,8 @@ bool ibuf_delete_rec(ulint space, ulint page_no, btr_pcur_t* pcur,
 	Delete-mark the record so that it will not be applied again,
 	in case the server crashes before the pessimistic delete is
 	made persistent. */
-	btr_cur_set_deleted_flag_for_ibuf(
+	btr_rec_set_deleted<true>(btr_pcur_get_block(pcur),
-		btr_pcur_get_rec(pcur), NULL, TRUE, mtr);
+				  btr_pcur_get_rec(pcur), mtr);
 
 	btr_pcur_store_position(pcur, mtr);
 	ibuf_btr_pcur_commit_specify_mtr(pcur, mtr);
@@ -4465,10 +4459,9 @@ loop:
 				the server crashes between the following
 				mtr_commit() and the subsequent mtr_commit()
 				of deleting the change buffer record. */
-
+				btr_rec_set_deleted<true>(
-				btr_cur_set_deleted_flag_for_ibuf(
+					btr_pcur_get_block(&pcur),
-					btr_pcur_get_rec(&pcur), NULL,
+					btr_pcur_get_rec(&pcur), &mtr);
-					TRUE, &mtr);
 
 				btr_pcur_store_position(&pcur, &mtr);
 				ibuf_btr_pcur_commit_specify_mtr(&pcur, &mtr);

storage/innobase/include/btr0cur.h

@@ -475,18 +475,6 @@ btr_cur_del_mark_set_clust_rec(
 	const dtuple_t*	entry,	/*!< in: dtuple for the deleting record */
 	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
-/***********************************************************//**
-Sets a secondary index record delete mark to TRUE or FALSE.
-@return DB_SUCCESS, DB_LOCK_WAIT, or error number */
-dberr_t
-btr_cur_del_mark_set_sec_rec(
-/*=========================*/
-	ulint		flags,	/*!< in: locking flag */
-	btr_cur_t*	cursor,	/*!< in: cursor */
-	ibool		val,	/*!< in: value to set */
-	que_thr_t*	thr,	/*!< in: query thread */
-	mtr_t*		mtr)	/*!< in/out: mini-transaction */
-	MY_ATTRIBUTE((nonnull, warn_unused_result));
 /*************************************************************//**
 Tries to compress a page of the tree if it seems useful. It is assumed
 that mtr holds an x-latch on the tree and on the cursor page. To avoid
@@ -567,6 +555,7 @@ void btr_cur_node_ptr_delete(btr_cur_t* parent, mtr_t* mtr)
 /***********************************************************//**
 Parses a redo log record of updating a record in-place.
 @return end of log record or NULL */
+ATTRIBUTE_COLD /* only used when crash-upgrading */
 const byte*
 btr_cur_parse_update_in_place(
 /*==========================*/
@@ -579,6 +568,7 @@ btr_cur_parse_update_in_place(
 Parses the redo log record for delete marking or unmarking of a clustered
 index record.
 @return end of log record or NULL */
+ATTRIBUTE_COLD /* only used when crash-upgrading */
 const byte*
 btr_cur_parse_del_mark_set_clust_rec(
 /*=================================*/
@@ -586,18 +576,21 @@ btr_cur_parse_del_mark_set_clust_rec(
 	const byte*	end_ptr,/*!< in: buffer end */
 	page_t*		page,	/*!< in/out: page or NULL */
 	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
-	dict_index_t*	index);	/*!< in: index corresponding to page */
+	dict_index_t*	index,	/*!< in: index corresponding to page */
+	mtr_t*		mtr);	/*!< in/out: mini-transaction */
 /****************************************************************//**
 Parses the redo log record for delete marking or unmarking of a secondary
 index record.
 @return end of log record or NULL */
+ATTRIBUTE_COLD /* only used when crash-upgrading */
 const byte*
 btr_cur_parse_del_mark_set_sec_rec(
 /*===============================*/
 	const byte*	ptr,	/*!< in: buffer */
 	const byte*	end_ptr,/*!< in: buffer end */
 	page_t*		page,	/*!< in/out: page or NULL */
-	page_zip_des_t*	page_zip);/*!< in/out: compressed page, or NULL */
+	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
+	mtr_t*		mtr);	/*!< in/out: mini-transaction */
 
 /** Estimates the number of rows in a given index range.
 @param[in]	index	index
@@ -784,29 +777,6 @@ btr_rec_copy_externally_stored_field(
 	ulint*			len,
 	mem_heap_t*		heap);
 
-/***********************************************************//**
-Sets a secondary index record's delete mark to the given value. This
-function is only used by the insert buffer merge mechanism. */
-void
-btr_cur_set_deleted_flag_for_ibuf(
-/*==============================*/
-	rec_t*		rec,		/*!< in/out: record */
-	page_zip_des_t*	page_zip,	/*!< in/out: compressed page
-					corresponding to rec, or NULL
-					when the tablespace is uncompressed */
-	ibool		val,		/*!< in: value to set */
-	mtr_t*		mtr);		/*!< in/out: mini-transaction */
-
-/******************************************************//**
-The following function is used to set the deleted bit of a record. */
-UNIV_INLINE
-void
-btr_rec_set_deleted_flag(
-/*=====================*/
-	rec_t*		rec,	/*!< in/out: physical record */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page (or NULL) */
-	ulint		flag);	/*!< in: nonzero if delete marked */
-
 /** Latches the leaf page or pages requested.
 @param[in]	block		leaf page where the search converged
 @param[in]	latch_mode	BTR_SEARCH_LEAF, ...
@@ -967,15 +937,14 @@ struct btr_cur_t {
 	}
 };
 
-/******************************************************//**
+/** Modify the delete-mark flag of a record.
-The following function is used to set the deleted bit of a record. */
+@tparam         flag    the value of the delete-mark flag
-UNIV_INLINE
+@param[in,out]  block   buffer block
-void
+@param[in,out]  rec     record on a physical index page
-btr_rec_set_deleted_flag(
+@param[in,out]  mtr     mini-transaction  */
-/*=====================*/
+template<bool flag>
-	rec_t*		rec,	/*!< in/out: physical record */
+void btr_rec_set_deleted(buf_block_t *block, rec_t *rec, mtr_t *mtr)
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page (or NULL) */
+  MY_ATTRIBUTE((nonnull));
-	ulint		flag);	/*!< in: nonzero if delete marked */
 
 /** If pessimistic delete fails because of lack of file space, there
 is still a good change of success a little later.  Try this many

storage/innobase/include/btr0cur.ic

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2015, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2018, MariaDB Corporation.
+Copyright (c) 2018, 2020, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -209,21 +209,3 @@ btr_blob_op_is_update(
 	ut_ad(0);
 	return(FALSE);
 }
-
-/******************************************************//**
-The following function is used to set the deleted bit of a record. */
-UNIV_INLINE
-void
-btr_rec_set_deleted_flag(
-/*=====================*/
-	rec_t*		rec,	/*!< in/out: physical record */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page (or NULL) */
-	ulint		flag)	/*!< in: nonzero if delete marked */
-{
-	if (page_rec_is_comp(rec)) {
-		rec_set_deleted_flag_new(rec, page_zip, flag);
-	} else {
-		ut_ad(!page_zip);
-		rec_set_deleted_flag_old(rec, flag);
-	}
-}

storage/innobase/include/page0types.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2019, MariaDB Corporation.
+Copyright (c) 2019, 2020, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -144,17 +144,6 @@ extern page_zip_stat_t			page_zip_stat[PAGE_ZIP_SSIZE_MAX];
 /** Statistics on compression, indexed by dict_index_t::id */
 extern page_zip_stat_per_index_t	page_zip_stat_per_index;
 
-/**********************************************************************//**
-Write the "deleted" flag of a record on a compressed page.  The flag must
-already have been written on the uncompressed page. */
-void
-page_zip_rec_set_deleted(
-/*=====================*/
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
-	const byte*	rec,	/*!< in: record on the uncompressed page */
-	ulint		flag)	/*!< in: the deleted flag (nonzero=TRUE) */
-	MY_ATTRIBUTE((nonnull));
-
 /**********************************************************************//**
 Write the "owned" flag of a record on a compressed page.  The n_owned field
 must already have been written on the uncompressed page. */

storage/innobase/include/page0zip.h

@@ -2,7 +2,7 @@
 
 Copyright (c) 2005, 2016, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2020, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -323,7 +323,7 @@ page_zip_write_trx_id_and_roll_ptr(
 	ulint		trx_id_col,
 	trx_id_t	trx_id,
 	roll_ptr_t	roll_ptr,
-	mtr_t*		mtr = NULL)
+	mtr_t*		mtr)
 	MY_ATTRIBUTE((nonnull(1,2,3)));
 
 /** Parse a MLOG_ZIP_WRITE_TRX_ID record.
@@ -350,7 +350,8 @@ page_zip_rec_set_deleted(
 /*=====================*/
 	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
 	const byte*	rec,	/*!< in: record on the uncompressed page */
-	ulint		flag)	/*!< in: the deleted flag (nonzero=TRUE) */
+	ulint		flag,	/*!< in: the deleted flag (nonzero=TRUE) */
+	mtr_t*		mtr)	/*!< in,out: mini-transaction */
 	MY_ATTRIBUTE((nonnull));
 
 /**********************************************************************//**

storage/innobase/include/rem0rec.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2020, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -399,25 +399,6 @@ rec_get_deleted_flag(
 	ulint		comp)	/*!< in: nonzero=compact page format */
 	MY_ATTRIBUTE((warn_unused_result));
 /******************************************************//**
-The following function is used to set the deleted bit. */
-UNIV_INLINE
-void
-rec_set_deleted_flag_old(
-/*=====================*/
-	rec_t*	rec,	/*!< in: old-style physical record */
-	ulint	flag)	/*!< in: nonzero if delete marked */
-	MY_ATTRIBUTE((nonnull));
-/******************************************************//**
-The following function is used to set the deleted bit. */
-UNIV_INLINE
-void
-rec_set_deleted_flag_new(
-/*=====================*/
-	rec_t*		rec,	/*!< in/out: new-style physical record */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
-	ulint		flag)	/*!< in: nonzero if delete marked */
-	MY_ATTRIBUTE((nonnull(1)));
-/******************************************************//**
 The following function tells if a new-style record is a node pointer.
 @return TRUE if node pointer */
 UNIV_INLINE

storage/innobase/include/rem0rec.ic

@@ -648,55 +648,6 @@ rec_get_deleted_flag(
 	}
 }
 
-/******************************************************//**
-The following function is used to set the deleted bit. */
-UNIV_INLINE
-void
-rec_set_deleted_flag_old(
-/*=====================*/
-	rec_t*	rec,	/*!< in: old-style physical record */
-	ulint	flag)	/*!< in: nonzero if delete marked */
-{
-	ulint	val;
-
-	val = rec_get_info_bits(rec, FALSE);
-
-	if (flag) {
-		val |= REC_INFO_DELETED_FLAG;
-	} else {
-		val &= ~REC_INFO_DELETED_FLAG;
-	}
-
-	rec_set_info_bits_old(rec, val);
-}
-
-/******************************************************//**
-The following function is used to set the deleted bit. */
-UNIV_INLINE
-void
-rec_set_deleted_flag_new(
-/*=====================*/
-	rec_t*		rec,	/*!< in/out: new-style physical record */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
-	ulint		flag)	/*!< in: nonzero if delete marked */
-{
-	ulint	val;
-
-	val = rec_get_info_bits(rec, TRUE);
-
-	if (flag) {
-		val |= REC_INFO_DELETED_FLAG;
-	} else {
-		val &= ~REC_INFO_DELETED_FLAG;
-	}
-
-	rec_set_info_bits_new(rec, val);
-
-	if (page_zip) {
-		page_zip_rec_set_deleted(page_zip, rec, flag);
-	}
-}
-
 /******************************************************//**
 The following function tells if a new-style record is a node pointer.
 @return TRUE if node pointer */

storage/innobase/include/row0upd.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2018, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2020, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -101,20 +101,6 @@ upd_get_field_by_field_no(
 	bool		is_virtual) /*!< in: if it is a virtual column */
 	MY_ATTRIBUTE((warn_unused_result));
 /*********************************************************************//**
-Updates the trx id and roll ptr field in a clustered index record when
-a row is updated or marked deleted. */
-UNIV_INLINE
-void
-row_upd_rec_sys_fields(
-/*===================*/
-	rec_t*		rec,	/*!< in/out: record */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page whose
-				uncompressed part will be updated, or NULL */
-	dict_index_t*	index,	/*!< in: clustered index */
-	const offset_t*	offsets,/*!< in: rec_get_offsets(rec, index) */
-	const trx_t*	trx,	/*!< in: transaction */
-	roll_ptr_t	roll_ptr);/*!< in: DB_ROLL_PTR to the undo log */
-/*********************************************************************//**
 Creates an update node for a query graph.
 @return own: update node */
 upd_node_t*
@@ -357,29 +343,6 @@ row_upd_step(
 /*=========*/
 	que_thr_t*	thr);	/*!< in: query thread */
 /*********************************************************************//**
-Parses the log data of system field values.
-@return log data end or NULL */
-byte*
-row_upd_parse_sys_vals(
-/*===================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	ulint*		pos,	/*!< out: TRX_ID position in record */
-	trx_id_t*	trx_id,	/*!< out: trx id */
-	roll_ptr_t*	roll_ptr);/*!< out: roll ptr */
-/*********************************************************************//**
-Updates the trx id and roll ptr field in a clustered index record in database
-recovery. */
-void
-row_upd_rec_sys_fields_in_recovery(
-/*===============================*/
-	rec_t*		rec,	/*!< in/out: record */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
-	const offset_t*	offsets,/*!< in: array returned by rec_get_offsets() */
-	ulint		pos,	/*!< in: TRX_ID position in rec */
-	trx_id_t	trx_id,	/*!< in: transaction id */
-	roll_ptr_t	roll_ptr);/*!< in: roll ptr of the undo log record */
-/*********************************************************************//**
 Parses the log data written by row_upd_index_write_log.
 @return log data end or NULL */
 byte*

storage/innobase/include/row0upd.ic

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2015, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2020, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -151,47 +151,3 @@ upd_get_field_by_field_no(
 
 	return(NULL);
 }
-
-/*********************************************************************//**
-Updates the trx id and roll ptr field in a clustered index record when
-a row is updated or marked deleted. */
-UNIV_INLINE
-void
-row_upd_rec_sys_fields(
-/*===================*/
-	rec_t*		rec,	/*!< in/out: record */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page whose
-				uncompressed part will be updated, or NULL */
-	dict_index_t*	index,	/*!< in: clustered index */
-	const offset_t*	offsets,/*!< in: rec_get_offsets(rec, index) */
-	const trx_t*	trx,	/*!< in: transaction */
-	roll_ptr_t	roll_ptr)/*!< in: DB_ROLL_PTR to the undo log */
-{
-	ut_ad(index->is_primary());
-	ut_ad(rec_offs_validate(rec, index, offsets));
-
-	if (UNIV_LIKELY_NULL(page_zip)) {
-		page_zip_write_trx_id_and_roll_ptr(page_zip, rec, offsets,
-						   index->db_trx_id(),
-						   trx->id, roll_ptr);
-	} else {
-		ulint	offset = index->trx_id_offset;
-
-		if (!offset) {
-			offset = row_get_trx_id_offset(index, offsets);
-		}
-
-		compile_time_assert(DATA_TRX_ID + 1 == DATA_ROLL_PTR);
-
-		/* During IMPORT the trx id in the record can be in the
-		future, if the .ibd file is being imported from another
-		instance. During IMPORT roll_ptr will be 0. */
-		ut_ad(roll_ptr == 0
-		      || lock_check_trx_id_sanity(
-			      trx_read_trx_id(rec + offset),
-			      rec, index, offsets));
-
-		trx_write_trx_id(rec + offset, trx->id);
-		trx_write_roll_ptr(rec + offset + DATA_TRX_ID_LEN, roll_ptr);
-	}
-}

storage/innobase/log/log0recv.cc

@@ -1594,13 +1594,13 @@ parse_log:
 			     || (ibool)!!page_is_comp(page)
 			     == dict_table_is_comp(index->table));
 			ptr = btr_cur_parse_del_mark_set_clust_rec(
-				ptr, end_ptr, page, page_zip, index);
+				ptr, end_ptr, page, page_zip, index, mtr);
 		}
 		break;
 	case MLOG_REC_SEC_DELETE_MARK:
 		ut_ad(!page || fil_page_type_is_index(page_type));
 		ptr = btr_cur_parse_del_mark_set_sec_rec(ptr, end_ptr,
-							 page, page_zip);
+							 page, page_zip, mtr);
 		break;
 	case MLOG_REC_UPDATE_IN_PLACE: case MLOG_COMP_REC_UPDATE_IN_PLACE:
 		ut_ad(!page || fil_page_type_is_index(page_type));

storage/innobase/page/page0zip.cc

@@ -4308,7 +4308,8 @@ page_zip_rec_set_deleted(
 /*=====================*/
 	page_zip_des_t*	page_zip,/*!< in/out: compressed page */
 	const byte*	rec,	/*!< in: record on the uncompressed page */
-	ulint		flag)	/*!< in: the deleted flag (nonzero=TRUE) */
+	ulint		flag,	/*!< in: the deleted flag (nonzero=TRUE) */
+	mtr_t*		mtr)	/*!< in,out: mini-transaction */
 {
 	byte*	slot = page_zip_dir_find(page_zip, page_offset(rec));
 	ut_a(slot);
@@ -4321,6 +4322,14 @@ page_zip_rec_set_deleted(
 #ifdef UNIV_ZIP_DEBUG
 	ut_a(page_zip_validate(page_zip, page_align(rec), NULL));
 #endif /* UNIV_ZIP_DEBUG */
+	if (byte* log_ptr = mlog_open(mtr, 11 + 2 + 2 + 1)) {
+		log_ptr = mlog_write_initial_log_record_fast(
+			rec, MLOG_ZIP_WRITE_STRING, log_ptr, mtr);
+		mach_write_to_2(log_ptr, slot - page_zip->data);
+		mach_write_to_2(log_ptr + 2, 1);
+		log_ptr[4] = *slot;
+		mlog_close(mtr, log_ptr + 5);
+	}
 }
 
 /**********************************************************************//**

storage/innobase/row/row0umod.cc

@@ -627,9 +627,8 @@ row_undo_mod_del_mark_or_remove_sec_low(
 	    || row_vers_old_has_index_entry(
 		    false, btr_pcur_get_rec(&node->pcur),
 		    &mtr_vers, index, entry, 0, 0)) {
-		err = btr_cur_del_mark_set_sec_rec(BTR_NO_LOCKING_FLAG,
+		btr_rec_set_deleted<true>(btr_cur_get_block(btr_cur),
-						   btr_cur, TRUE, thr, &mtr);
+					  btr_cur_get_rec(btr_cur), &mtr);
-		ut_ad(err == DB_SUCCESS);
 	} else {
 		/* Remove the index record */
 
@@ -864,11 +863,8 @@ try_again:
 
 		break;
 	case ROW_FOUND:
-		err = btr_cur_del_mark_set_sec_rec(
+		btr_rec_set_deleted<false>(btr_cur_get_block(btr_cur),
-			BTR_NO_LOCKING_FLAG,
+					   btr_cur_get_rec(btr_cur), &mtr);
-			btr_cur, FALSE, thr, &mtr);
-
-		ut_a(err == DB_SUCCESS);
 		heap = mem_heap_create(
 			sizeof(upd_t)
 			+ dtuple_get_n_fields(entry) * sizeof(upd_field_t));

storage/innobase/row/row0upd.cc

@@ -462,36 +462,6 @@ upd_node_create(
 	return(node);
 }
 
-/*********************************************************************//**
-Updates the trx id and roll ptr field in a clustered index record in database
-recovery. */
-void
-row_upd_rec_sys_fields_in_recovery(
-/*===============================*/
-	rec_t*		rec,	/*!< in/out: record */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
-	const offset_t*	offsets,/*!< in: array returned by rec_get_offsets() */
-	ulint		pos,	/*!< in: TRX_ID position in rec */
-	trx_id_t	trx_id,	/*!< in: transaction id */
-	roll_ptr_t	roll_ptr)/*!< in: roll ptr of the undo log record */
-{
-	ut_ad(rec_offs_validate(rec, NULL, offsets));
-
-	if (page_zip) {
-		page_zip_write_trx_id_and_roll_ptr(
-			page_zip, rec, offsets, pos, trx_id, roll_ptr);
-	} else {
-		byte*	field;
-		ulint	len;
-
-		field = rec_get_nth_field(rec, offsets, pos, &len);
-		ut_ad(len == DATA_TRX_ID_LEN);
-		compile_time_assert(DATA_TRX_ID + 1 == DATA_ROLL_PTR);
-		trx_write_trx_id(field, trx_id);
-		trx_write_roll_ptr(field + DATA_TRX_ID_LEN, roll_ptr);
-	}
-}
-
 /***********************************************************//**
 Returns TRUE if row update changes size of some field in index or if some
 field to be updated is stored externally in rec or update.
@@ -694,38 +664,6 @@ row_upd_rec_in_place(
 	}
 }
 
-/*********************************************************************//**
-Parses the log data of system field values.
-@return log data end or NULL */
-byte*
-row_upd_parse_sys_vals(
-/*===================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	ulint*		pos,	/*!< out: TRX_ID position in record */
-	trx_id_t*	trx_id,	/*!< out: trx id */
-	roll_ptr_t*	roll_ptr)/*!< out: roll ptr */
-{
-	*pos = mach_parse_compressed(&ptr, end_ptr);
-
-	if (ptr == NULL) {
-
-		return(NULL);
-	}
-
-	if (end_ptr < ptr + DATA_ROLL_PTR_LEN) {
-
-		return(NULL);
-	}
-
-	*roll_ptr = trx_read_roll_ptr(ptr);
-	ptr += DATA_ROLL_PTR_LEN;
-
-	*trx_id = mach_u64_parse_compressed(&ptr, end_ptr);
-
-	return(const_cast<byte*>(ptr));
-}
-
 /***********************************************************//**
 Writes to the redo log the new values of the fields occurring in the index. */
 void
@@ -2431,11 +2369,17 @@ row_upd_sec_index_entry(
 		row_ins_sec_index_entry() below */
 		if (!rec_get_deleted_flag(
 			    rec, dict_table_is_comp(index->table))) {
-			err = btr_cur_del_mark_set_sec_rec(
+			err = lock_sec_rec_modify_check_and_lock(
-				flags, btr_cur, TRUE, thr, &mtr);
+				flags,
+				btr_cur_get_block(btr_cur),
+				btr_cur_get_rec(btr_cur), index, thr, &mtr);
 			if (err != DB_SUCCESS) {
 				break;
 			}
+
+			btr_rec_set_deleted<true>(btr_cur_get_block(btr_cur),
+						  btr_cur_get_rec(btr_cur),
+						  &mtr);
 #ifdef WITH_WSREP
 			if (!referenced && foreign
 			    && wsrep_must_process_fk(node, trx)