Assert that DB_TRX_ID must be set on delete-marked records

This is preparation for MDEV-12288, which would set DB_TRX_ID=0
when purging history. Also with that change in place, delete-marked
records must always refer to an undo log record via a nonzero
DB_TRX_ID column. (The DB_TRX_ID is only present in clustered index
leaf page records.)

btr_cur_parse_del_mark_set_clust_rec(), rec_get_trx_id():
Statically allocate the offsets
(should never use the heap). Add some debug assertions.

Replace some use of rec_get_trx_id() with row_get_rec_trx_id().

trx_undo_report_row_operation(): Add some sanity checks that are
common for all operations that produce undo log.

12 files changed, 130 insertions, 46 deletions

diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 3bef25e945c..a672f451ea9 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -3679,6 +3679,11 @@ btr_cur_update_in_place(
 
 	was_delete_marked = rec_get_deleted_flag(
 		rec, page_is_comp(buf_block_get_frame(block)));
+	/* In delete-marked records, DB_TRX_ID must always refer to an
+	existing undo log record. */
+	ut_ad(!was_delete_marked
+	      || !dict_index_is_clust(index)
+	      || row_get_rec_trx_id(rec, index, offsets));
 
 #ifdef BTR_CUR_HASH_ADAPT
 	if (block->index) {
@@ -4321,6 +4326,10 @@ btr_cur_pessimistic_update(
 			stored fields */
 			btr_cur_unmark_extern_fields(
 				page_zip, rec, index, *offsets, mtr);
+		} else {
+			/* In delete-marked records, DB_TRX_ID must
+			always refer to an existing undo log record. */
+			ut_ad(row_get_rec_trx_id(rec, index, *offsets));
 		}
 
 		bool adjust = big_rec_vec && (flags & BTR_KEEP_POS_FLAG);
@@ -4448,6 +4457,10 @@ btr_cur_pessimistic_update(
 
 		btr_cur_unmark_extern_fields(page_zip,
 					     rec, index, *offsets, mtr);
+	} else {
+		/* In delete-marked records, DB_TRX_ID must
+		always refer to an existing undo log record. */
+		ut_ad(row_get_rec_trx_id(rec, index, *offsets));
 	}
 
 	if (!dict_table_is_locking_disabled(index->table)) {
@@ -4573,6 +4586,10 @@ btr_cur_parse_del_mark_set_clust_rec(
 
 	ut_a(offset <= UNIV_PAGE_SIZE);
 
+	/* In delete-marked records, DB_TRX_ID must
+	always refer to an existing undo log record. */
+	ut_ad(trx_id || (flags & BTR_KEEP_SYS_FLAG));
+
 	if (page) {
 		rec = page + offset;
 
@@ -4582,20 +4599,33 @@ btr_cur_parse_del_mark_set_clust_rec(
 		and the adaptive hash index does not depend on them. */
 
 		btr_rec_set_deleted_flag(rec, page_zip, val);
+		ut_ad(pos <= MAX_REF_PARTS);
 
-		if (!(flags & BTR_KEEP_SYS_FLAG)) {
-			mem_heap_t*	heap		= NULL;
-			ulint		offsets_[REC_OFFS_NORMAL_SIZE];
-			rec_offs_init(offsets_);
+		ulint offsets[REC_OFFS_HEADER_SIZE + MAX_REF_PARTS + 2];
+		rec_offs_init(offsets);
+		mem_heap_t*	heap	= NULL;
 
+		if (!(flags & BTR_KEEP_SYS_FLAG)) {
 			row_upd_rec_sys_fields_in_recovery(
 				rec, page_zip,
-				rec_get_offsets(rec, index, offsets_,
-						ULINT_UNDEFINED, &heap),
+				rec_get_offsets(rec, index, offsets,
+						pos + 1, &heap),
 				pos, trx_id, roll_ptr);
-			if (UNIV_LIKELY_NULL(heap)) {
-				mem_heap_free(heap);
-			}
+		} 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, pos,
+							     &heap),
+					     pos, &offset),
+				     field_ref_zero, DATA_TRX_ID_LEN));
+			ut_ad(offset == DATA_TRX_ID_LEN);
+		}
+
+		if (UNIV_LIKELY_NULL(heap)) {
+			mem_heap_free(heap);
 		}
 	}
 
@@ -4633,8 +4663,10 @@ btr_cur_del_mark_set_clust_rec(
 	ut_ad(mtr->is_named_space(index->space));
 
 	if (rec_get_deleted_flag(rec, rec_offs_comp(offsets))) {
-		/* While cascading delete operations, this becomes possible. */
-		ut_ad(rec_get_trx_id(rec, index) == thr_get_trx(thr)->id);
+		/* We may already have delete-marked this record
+		when executing an ON DELETE CASCADE operation. */
+		ut_ad(row_get_rec_trx_id(rec, index, offsets)
+		      == thr_get_trx(thr)->id);
 		return(DB_SUCCESS);
 	}
 
@@ -4663,10 +4695,6 @@ btr_cur_del_mark_set_clust_rec(
 	btr_rec_set_deleted_flag(rec, page_zip, TRUE);
 
 	trx = thr_get_trx(thr);
-	/* This function must not be invoked during rollback
-	(of a TRX_STATE_PREPARE transaction or otherwise). */
-	ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE));
-	ut_ad(!trx->in_rollback);
 
 	DBUG_LOG("ib_cur",
 		 "delete-mark clust " << index->table->name
diff --git a/storage/innobase/include/rem0rec.h b/storage/innobase/include/rem0rec.h
index 916222d5fdb..96d3bdfab8f 100644
--- a/storage/innobase/include/rem0rec.h
+++ b/storage/innobase/include/rem0rec.h
@@ -1080,14 +1080,14 @@ private:
 # endif /* !DBUG_OFF */
 
 # ifdef UNIV_DEBUG
-/************************************************************//**
-Reads the DB_TRX_ID of a clustered index record.
+/** Read the DB_TRX_ID of a clustered index record.
+@param[in]	rec	clustered index record
+@param[in]	index	clustered index
 @return the value of DB_TRX_ID */
 trx_id_t
 rec_get_trx_id(
-/*===========*/
-	const rec_t*		rec,	/*!< in: record */
-	const dict_index_t*	index)	/*!< in: clustered index */
+	const rec_t*		rec,
+	const dict_index_t*	index)
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 # endif /* UNIV_DEBUG */
 
diff --git a/storage/innobase/page/page0zip.cc b/storage/innobase/page/page0zip.cc
index 59022dbe301..f556c887520 100644
--- a/storage/innobase/page/page0zip.cc
+++ b/storage/innobase/page/page0zip.cc
@@ -3711,6 +3711,10 @@ page_zip_write_rec(
 	ut_a(slot);
 	/* Copy the delete mark. */
 	if (rec_get_deleted_flag(rec, TRUE)) {
+		/* In delete-marked records, DB_TRX_ID must
+		always refer to an existing undo log record. */
+		ut_ad(!dict_index_is_clust(index)
+		      || row_get_rec_trx_id(rec, index, offsets));
 		*slot |= PAGE_ZIP_DIR_SLOT_DEL >> 8;
 	} else {
 		*slot &= ~(PAGE_ZIP_DIR_SLOT_DEL >> 8);
diff --git a/storage/innobase/rem/rem0rec.cc b/storage/innobase/rem/rem0rec.cc
index c78df489179..b7d7a30f4dc 100644
--- a/storage/innobase/rem/rem0rec.cc
+++ b/storage/innobase/rem/rem0rec.cc
@@ -2106,8 +2106,6 @@ rec_print_mbr_rec(
 
 /***************************************************************//**
 Prints a physical record. */
-/***************************************************************//**
-Prints a physical record. */
 void
 rec_print_new(
 /*==========*/
@@ -2254,14 +2252,14 @@ operator<<(std::ostream& o, const rec_offsets_print& r)
 }
 
 #ifdef UNIV_DEBUG
-/************************************************************//**
-Reads the DB_TRX_ID of a clustered index record.
+/** Read the DB_TRX_ID of a clustered index record.
+@param[in]	rec	clustered index record
+@param[in]	index	clustered index
 @return the value of DB_TRX_ID */
 trx_id_t
 rec_get_trx_id(
-/*===========*/
-	const rec_t*		rec,	/*!< in: record */
-	const dict_index_t*	index)	/*!< in: clustered index */
+	const rec_t*		rec,
+	const dict_index_t*	index)
 {
 	const page_t*	page
 		= page_align(rec);
@@ -2270,10 +2268,11 @@ rec_get_trx_id(
 	const byte*	trx_id;
 	ulint		len;
 	mem_heap_t*	heap		= NULL;
-	ulint		offsets_[REC_OFFS_NORMAL_SIZE];
-	ulint*		offsets		= offsets_;
+	ulint offsets_[REC_OFFS_HEADER_SIZE + MAX_REF_PARTS + 2];
 	rec_offs_init(offsets_);
+	ulint* offsets = offsets_;
 
+	ut_ad(trx_id_col <= MAX_REF_PARTS);
 	ut_ad(fil_page_index_page_check(page));
 	ut_ad(mach_read_from_8(page + PAGE_HEADER + PAGE_INDEX_ID)
 	      == index->id);
@@ -2287,7 +2286,7 @@ rec_get_trx_id(
 
 	ut_ad(len == DATA_TRX_ID_LEN);
 
-	if (heap) {
+	if (UNIV_LIKELY_NULL(heap)) {
 		mem_heap_free(heap);
 	}
 
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 4d13051c985..439b253e80d 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -346,6 +346,9 @@ row_ins_clust_index_entry_by_modify(
 
 	ut_ad(rec_get_deleted_flag(rec,
 				   dict_table_is_comp(cursor->index->table)));
+	/* In delete-marked records, DB_TRX_ID must
+	always refer to an existing undo log record. */
+	ut_ad(rec_get_trx_id(rec, cursor->index));
 
 	/* Build an update vector containing all the fields to be modified;
 	NOTE that this vector may NOT contain system columns trx_id or
@@ -1266,6 +1269,9 @@ row_ins_foreign_check_on_constraint(
 	}
 
 	if (rec_get_deleted_flag(clust_rec, dict_table_is_comp(table))) {
+		/* In delete-marked records, DB_TRX_ID must
+		always refer to an existing undo log record. */
+		ut_ad(rec_get_trx_id(clust_rec, clust_index));
 		/* This can happen if there is a circular reference of
 		rows such that cascading delete comes to delete a row
 		already in the process of being delete marked */
@@ -1274,7 +1280,6 @@ row_ins_foreign_check_on_constraint(
 		goto nonstandard_exit_func;
 	}
 
-
 	if (table->fts) {
 		doc_id = fts_get_doc_id_from_rec(table, clust_rec,
 						 clust_index, tmp_heap);
@@ -1758,6 +1763,12 @@ row_ins_check_foreign_constraint(
 
 			if (rec_get_deleted_flag(rec,
 						 rec_offs_comp(offsets))) {
+				/* In delete-marked records, DB_TRX_ID must
+				always refer to an existing undo log record. */
+				ut_ad(!dict_index_is_clust(check_index)
+				      || row_get_rec_trx_id(rec, check_index,
+							    offsets));
+
 				err = row_ins_set_shared_rec_lock(
 					lock_type, block,
 					rec, check_index, offsets, thr);
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index ea332adfdc3..4250634baec 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -2130,10 +2130,10 @@ end_of_index:
 
 		rec = page_cur_get_rec(cur);
 
-		offsets = rec_get_offsets(rec, clust_index, NULL,
-					  ULINT_UNDEFINED, &row_heap);
-
 		if (online) {
+			offsets = rec_get_offsets(rec, clust_index, NULL,
+						  ULINT_UNDEFINED, &row_heap);
+
 			/* Perform a REPEATABLE READ.
 
 			When rebuilding the table online,
@@ -2176,6 +2176,10 @@ end_of_index:
 			if (rec_get_deleted_flag(
 				    rec,
 				    dict_table_is_comp(old_table))) {
+				/* In delete-marked records, DB_TRX_ID must
+				always refer to an existing undo log record. */
+				ut_ad(row_get_rec_trx_id(rec, clust_index,
+							 offsets));
 				/* This record was deleted in the latest
 				committed version, or it was deleted and
 				then reinserted-by-update before purge
@@ -2186,6 +2190,9 @@ end_of_index:
 			ut_ad(!rec_offs_any_null_extern(rec, offsets));
 		} else if (rec_get_deleted_flag(
 				   rec, dict_table_is_comp(old_table))) {
+			/* In delete-marked records, DB_TRX_ID must
+			always refer to an existing undo log record. */
+			ut_ad(rec_get_trx_id(rec, clust_index));
 			/* Skip delete-marked records.
 
 			Skipping delete-marked records will make the
@@ -2195,6 +2202,9 @@ end_of_index:
 			would make it tricky to detect duplicate
 			keys. */
 			continue;
+		} else {
+			offsets = rec_get_offsets(rec, clust_index, NULL,
+						  ULINT_UNDEFINED, &row_heap);
 		}
 
 		/* When !online, we are holding a lock on old_table, preventing
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index 86a9e1259ac..cecd127e71e 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -160,6 +160,9 @@ row_purge_remove_clust_if_poss_low(
 	}
 
 	ut_ad(rec_get_deleted_flag(rec, rec_offs_comp(offsets)));
+	/* In delete-marked records, DB_TRX_ID must
+	always refer to an existing undo log record. */
+	ut_ad(row_get_rec_trx_id(rec, index, offsets));
 
 	if (mode == BTR_MODIFY_LEAF) {
 		success = btr_cur_optimistic_delete(
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index 52af4bd1828..a8f250bf6dd 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -183,6 +183,9 @@ row_sel_sec_rec_is_for_clust_rec(
 
 	if (rec_get_deleted_flag(clust_rec,
 				 dict_table_is_comp(clust_index->table))) {
+		/* In delete-marked records, DB_TRX_ID must
+		always refer to an existing undo log record. */
+		ut_ad(rec_get_trx_id(clust_rec, clust_index));
 
 		/* The clustered index record is delete-marked;
 		it is not visible in the read view.  Besides,
@@ -2042,6 +2045,11 @@ skip_lock:
 
 		if (rec_get_deleted_flag(clust_rec,
 					 dict_table_is_comp(plan->table))) {
+			/* In delete-marked records, DB_TRX_ID must
+			always refer to an existing update_undo log record. */
+			ut_ad(rec_get_trx_id(clust_rec,
+					     dict_table_get_first_index(
+						     plan->table)));
 
 			/* The record is delete marked: we can skip it */
 
@@ -3934,6 +3942,9 @@ row_sel_try_search_shortcut_for_mysql(
 	}
 
 	if (rec_get_deleted_flag(rec, dict_table_is_comp(index->table))) {
+		/* In delete-marked records, DB_TRX_ID must
+		always refer to an existing undo log record. */
+		ut_ad(row_get_rec_trx_id(rec, index, *offsets));
 
 		return(SEL_EXHAUSTED);
 	}
@@ -5119,6 +5130,10 @@ locks_ok:
 	page_rec_is_comp() cannot be used! */
 
 	if (rec_get_deleted_flag(rec, comp)) {
+		/* In delete-marked records, DB_TRX_ID must
+		always refer to an existing undo log record. */
+		ut_ad(index != clust_index
+		      || row_get_rec_trx_id(rec, index, offsets));
 
 		/* The record is delete-marked: we can skip it */
 
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 934b5ad5a7a..ef0c3ed9cc3 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -230,6 +230,14 @@ row_undo_ins_remove_sec_low(
 	case ROW_NOT_FOUND:
 		goto func_exit;
 	case ROW_FOUND:
+		if (dict_index_is_spatial(index)
+		    && rec_get_deleted_flag(
+			    btr_pcur_get_rec(&pcur),
+			    dict_table_is_comp(index->table))) {
+			ib::error() << "Record found in index " << index->name
+				<< " is deleted marked on insert rollback.";
+			ut_ad(0);
+		}
 		break;
 
 	case ROW_BUFFERED:
@@ -240,15 +248,6 @@ row_undo_ins_remove_sec_low(
 		ut_error;
 	}
 
-	if (search_result == ROW_FOUND && dict_index_is_spatial(index)) {
-		rec_t*	rec = btr_pcur_get_rec(&pcur);
-		if (rec_get_deleted_flag(rec,
-					 dict_table_is_comp(index->table))) {
-			ib::error() << "Record found in index " << index->name
-				<< " is deleted marked on insert rollback.";
-		}
-	}
-
 	btr_cur = btr_pcur_get_btr_cur(&pcur);
 
 	if (modify_leaf) {
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index 8fa5c2ccbff..a0c69fb97f9 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -215,6 +215,9 @@ row_undo_mod_remove_clust_low(
 	than the rolling-back one. */
 	ut_ad(rec_get_deleted_flag(btr_cur_get_rec(btr_cur),
 				   dict_table_is_comp(node->table)));
+	/* In delete-marked records, DB_TRX_ID must
+	always refer to an existing update_undo log record. */
+	ut_ad(rec_get_trx_id(btr_cur_get_rec(btr_cur), btr_cur->index));
 
 	if (mode == BTR_MODIFY_LEAF) {
 		err = btr_cur_optimistic_delete(btr_cur, 0, mtr)
@@ -350,8 +353,9 @@ row_undo_mod_clust(
 	*   it can be reallocated at any time after this mtr-commits
 	*   which is just below
 	*/
-	ut_ad(srv_immediate_scrub_data_uncompressed ||
-	      rec_get_trx_id(btr_pcur_get_rec(pcur), index) == node->new_trx_id);
+	ut_ad(srv_immediate_scrub_data_uncompressed
+	      || row_get_rec_trx_id(btr_pcur_get_rec(pcur), index, offsets)
+	      == node->new_trx_id);
 
 	btr_pcur_commit_specify_mtr(pcur, &mtr);
 
@@ -515,6 +519,7 @@ row_undo_mod_del_mark_or_remove_sec_low(
 				ib::error() << "Record found in index "
 					<< index->name << " is deleted marked"
 					" on rollback update.";
+				ut_ad(0);
 			}
 		}
 
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 0ca5d14084c..4f57136db0b 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -2663,7 +2663,8 @@ row_upd_clust_rec_by_insert(
 			marked, then we are here after a DB_LOCK_WAIT.
 			Skip delete marking clustered index and disowning
 			its blobs. */
-			ut_ad(rec_get_trx_id(rec, index) == trx->id);
+			ut_ad(row_get_rec_trx_id(rec, index, offsets)
+			      == trx->id);
 			ut_ad(!trx_undo_roll_ptr_is_insert(
 			              row_get_rec_roll_ptr(rec, index,
 							   offsets)));
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index 6d3ade289bb..02aee500e02 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -906,6 +906,10 @@ trx_undo_page_report_modify(
 		ut_ad(!rec_get_deleted_flag(rec, dict_table_is_comp(table)));
 		type_cmpl = TRX_UNDO_DEL_MARK_REC;
 	} else if (rec_get_deleted_flag(rec, dict_table_is_comp(table))) {
+		/* In delete-marked records, DB_TRX_ID must
+		always refer to an existing update_undo log record. */
+		ut_ad(row_get_rec_trx_id(rec, index, offsets));
+
 		type_cmpl = TRX_UNDO_UPD_DEL_REC;
 		/* We are about to update a delete marked record.
 		We don't typically need the prefix in this case unless
@@ -1875,6 +1879,10 @@ trx_undo_report_row_operation(
 	ut_ad(!srv_read_only_mode);
 
 	trx = thr_get_trx(thr);
+	/* This function must not be invoked during rollback
+	(of a TRX_STATE_PREPARE transaction or otherwise). */
+	ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE));
+	ut_ad(!trx->in_rollback);
 
 	mtr.start();
 	trx_undo_t**	pundo;
@@ -1888,6 +1896,7 @@ trx_undo_report_row_operation(
 		pundo = &trx->rsegs.m_noredo.undo;
 	} else {
 		ut_ad(!trx->read_only);
+		ut_ad(trx->id);
 		/* Keep INFORMATION_SCHEMA.TABLES.UPDATE_TIME
 		up-to-date for persistent tables. Temporary tables are
 		not listed there. */