MDEV-13495 Crash in rollback of a recovered INSERT transaction after upgrade

If the server is upgraded from a database that was created
before MDEV-12288, and if the undo logs in the database contain
an incomplete transaction that performed an INSERT operation,
the server would crash when rolling back that transaction.

trx_commit_low(): Relax a too strict transaction. This function
will also be called after completing the rollback of a recovered
transaction.

trx_purge_add_undo_to_history(): Merged from the functions
trx_purge_add_update_undo_to_history() and trx_undo_update_cleanup(),
which are removed. Remove the parameter undo_page, and instead call
trx_undo_set_state_at_finish() to obtain it.

trx_write_serialisation_history(): Treat undo and old_insert equally.
That is, after the rollback (or XA COMMIT) of a recovered transaction
before upgrade, move all logs (both insert_undo and update_undo) to
the purge queue.

storage/innobase/include/trx0purge.h

@@ -51,16 +51,13 @@ trx_purge_get_log_from_hist(
 /*========================*/
 	fil_addr_t	node_addr);	/*!< in: file address of the history
 					list node of the log */
-/************************************************************************
-Adds the update undo log as the first log in the history list. Removes the
-update undo log segment from the rseg slot if it is too big for reuse. */
+/** Prepend the history list with an undo log.
+Remove the undo log segment from the rseg slot if it is too big for reuse.
+@param[in]	trx		transaction
+@param[in,out]	undo		undo log
+@param[in,out]	mtr		mini-transaction */
 void
-trx_purge_add_update_undo_to_history(
-/*=================================*/
-	trx_t*		trx,		/*!< in: transaction */
-	page_t*		undo_page,	/*!< in: update undo log header page,
-					x-latched */
-	mtr_t*		mtr);		/*!< in: mtr */
+trx_purge_add_undo_to_history(const trx_t* trx, trx_undo_t*& undo, mtr_t* mtr);
 /*******************************************************************//**
 This function runs a purge batch.
 @return number of undo log pages handled in the batch */

storage/innobase/include/trx0undo.h

@@ -287,19 +287,6 @@ trx_undo_set_state_at_prepare(
 	bool		rollback,
 	mtr_t*		mtr);
 
-/**********************************************************************//**
-Adds the update undo log header as the first in the history list, and
-frees the memory object, or puts it to the list of cached update undo log
-segments. */
-void
-trx_undo_update_cleanup(
-/*====================*/
-	trx_t*		trx,		/*!< in: trx owning the update
-					undo log */
-	page_t*		undo_page,	/*!< in: update undo log header page,
-					x-latched */
-	mtr_t*		mtr);		/*!< in: mtr */
-
 /** Free an old insert or temporary undo log after commit or rollback.
 The information is not needed after a commit or rollback, therefore
 the data can be discarded.

storage/innobase/trx/trx0purge.cc

@@ -237,21 +237,22 @@ purge_sys_t::~purge_sys_t()
 
 /*================ UNDO LOG HISTORY LIST =============================*/
 
-/********************************************************************//**
-Adds the update undo log as the first log in the history list. Removes the
-update undo log segment from the rseg slot if it is too big for reuse. */
+/** Prepend the history list with an undo log.
+Remove the undo log segment from the rseg slot if it is too big for reuse.
+@param[in]	trx		transaction
+@param[in,out]	undo		undo log
+@param[in,out]	mtr		mini-transaction */
 void
-trx_purge_add_update_undo_to_history(
-/*=================================*/
-	trx_t*		trx,		/*!< in: transaction */
-	page_t*		undo_page,	/*!< in: update undo log header page,
-					x-latched */
-	mtr_t*		mtr)		/*!< in: mtr */
+trx_purge_add_undo_to_history(const trx_t* trx, trx_undo_t*& undo, mtr_t* mtr)
 {
-	trx_undo_t*	undo		= trx->rsegs.m_redo.undo;
-	trx_rseg_t*	rseg		= undo->rseg;
+	ut_ad(undo == trx->rsegs.m_redo.undo
+	      || undo == trx->rsegs.m_redo.old_insert);
+	trx_rseg_t*	rseg		= trx->rsegs.m_redo.rseg;
+	ut_ad(undo->rseg == rseg);
 	trx_rsegf_t*	rseg_header	= trx_rsegf_get(
 		rseg->space, rseg->page_no, mtr);
+	page_t*		undo_page	= trx_undo_set_state_at_finish(
+		undo, mtr);
 	trx_ulogf_t*	undo_header	= undo_page + undo->hdr_offset;
 
 	if (undo->state != TRX_UNDO_CACHED) {
@@ -326,6 +327,16 @@ trx_purge_add_update_undo_to_history(
 		rseg->last_trx_no = trx->no;
 		rseg->last_del_marks = undo->del_marks;
 	}
+
+	if (undo->state == TRX_UNDO_CACHED) {
+		UT_LIST_ADD_FIRST(rseg->undo_cached, undo);
+		MONITOR_INC(MONITOR_NUM_UNDO_SLOT_CACHED);
+	} else {
+		ut_ad(undo->state == TRX_UNDO_TO_PURGE);
+		trx_undo_mem_free(undo);
+	}
+
+	undo = NULL;
 }
 
 /** Remove undo log header from the history list.

storage/innobase/trx/trx0trx.cc

@@ -1420,13 +1420,14 @@ trx_write_serialisation_history(
 		temp_mtr.commit();
 	}
 
-	if (!trx->rsegs.m_redo.rseg) {
+	trx_rseg_t*	rseg = trx->rsegs.m_redo.rseg;
+	if (!rseg) {
 		ut_ad(!trx->rsegs.m_redo.undo);
 		ut_ad(!trx->rsegs.m_redo.old_insert);
 		return false;
 	}
 
-	trx_undo_t* undo = trx->rsegs.m_redo.undo;
+	trx_undo_t*& undo = trx->rsegs.m_redo.undo;
 	trx_undo_t*& old_insert = trx->rsegs.m_redo.old_insert;
 
 	if (!undo && !old_insert) {
@@ -1434,40 +1435,29 @@ trx_write_serialisation_history(
 	}
 
 	ut_ad(!trx->read_only);
-	trx_rseg_t*	undo_rseg = undo ? undo->rseg : NULL;
-	ut_ad(!undo || undo->rseg == trx->rsegs.m_redo.rseg);
-	mutex_enter(&trx->rsegs.m_redo.rseg->mutex);
+	trx_rseg_t*	undo_rseg
+		= undo ? undo->rseg : old_insert ? old_insert->rseg : NULL;
+	ut_ad(!undo || undo->rseg == rseg);
+	ut_ad(!old_insert || old_insert->rseg == rseg);
+	mutex_enter(&rseg->mutex);
 
 	/* Assign the transaction serialisation number and add any
 	undo log to the purge queue. */
 	trx_serialise(trx, undo_rseg);
 
 	/* It is not necessary to acquire trx->undo_mutex here because
-	only a single OS thread is allowed to commit this transaction. */
+	only a single OS thread is allowed to commit this transaction.
+	The undo logs will be processed and purged later. */
 	if (UNIV_LIKELY_NULL(old_insert)) {
-		page_t* undo_hdr_page = trx_undo_set_state_at_finish(
-			old_insert, mtr);
-		trx_rseg_t* rseg = trx->rsegs.m_redo.rseg;
-		trx_purge_add_update_undo_to_history(trx, undo_hdr_page, mtr);
 		UT_LIST_REMOVE(rseg->old_insert_list, old_insert);
-
-		if (old_insert->state == TRX_UNDO_CACHED) {
-			UT_LIST_ADD_FIRST(rseg->undo_cached, old_insert);
-			MONITOR_INC(MONITOR_NUM_UNDO_SLOT_CACHED);
-		} else {
-			ut_ad(old_insert->state == TRX_UNDO_TO_PURGE);
-			trx_undo_mem_free(old_insert);
-		}
-		old_insert = NULL;
+		trx_purge_add_undo_to_history(trx, old_insert, mtr);
 	}
 	if (undo) {
-		/* The undo logs will be processed and purged later. */
-		page_t*	undo_hdr_page = trx_undo_set_state_at_finish(
-			undo, mtr);
-		trx_undo_update_cleanup(trx, undo_hdr_page, mtr);
+		UT_LIST_REMOVE(rseg->undo_list, undo);
+		trx_purge_add_undo_to_history(trx, undo, mtr);
 	}
 
-	mutex_exit(&trx->rsegs.m_redo.rseg->mutex);
+	mutex_exit(&rseg->mutex);
 
 	MONITOR_INC(MONITOR_TRX_COMMIT_UNDO);
 
@@ -1915,7 +1905,7 @@ trx_commit_low(
 	assert_trx_nonlocking_or_in_list(trx);
 	ut_ad(!trx_state_eq(trx, TRX_STATE_COMMITTED_IN_MEMORY));
 	ut_ad(!mtr || mtr->is_active());
-	ut_ad(!mtr == !trx->has_logged());
+	ut_ad(!mtr == !trx->has_logged_or_recovered());
 
 	/* undo_no is non-zero if we're doing the final commit. */
 	if (trx->fts_trx != NULL && trx->undo_no != 0) {

storage/innobase/trx/trx0undo.cc

@@ -1623,42 +1623,6 @@ trx_undo_set_state_at_prepare(
 	return(undo_page);
 }
 
-/**********************************************************************//**
-Adds the update undo log header as the first in the history list, and
-frees the memory object, or puts it to the list of cached update undo log
-segments. */
-void
-trx_undo_update_cleanup(
-/*====================*/
-	trx_t*		trx,		/*!< in: trx owning the update
-					undo log */
-	page_t*		undo_page,	/*!< in: update undo log header page,
-					x-latched */
-	mtr_t*		mtr)		/*!< in: mtr */
-{
-	trx_undo_t*	undo	= trx->rsegs.m_redo.undo;
-	trx_rseg_t*	rseg	= undo->rseg;
-
-	ut_ad(mutex_own(&rseg->mutex));
-
-	trx_purge_add_update_undo_to_history(trx, undo_page, mtr);
-
-	UT_LIST_REMOVE(rseg->undo_list, undo);
-
-	trx->rsegs.m_redo.undo = NULL;
-
-	if (undo->state == TRX_UNDO_CACHED) {
-
-		UT_LIST_ADD_FIRST(rseg->undo_cached, undo);
-
-		MONITOR_INC(MONITOR_NUM_UNDO_SLOT_CACHED);
-	} else {
-		ut_ad(undo->state == TRX_UNDO_TO_PURGE);
-
-		trx_undo_mem_free(undo);
-	}
-}
-
 /** Free an old insert or temporary undo log after commit or rollback.
 The information is not needed after a commit or rollback, therefore
 the data can be discarded.