MDEV-15912: Remove traces of insert_undo

Let us simply refuse an upgrade from earlier versions if the
upgrade procedure was not followed. This simplifies the purge,
commit, and rollback of transactions.

Before upgrading to MariaDB 10.3 or later, a clean shutdown
of the server (with innodb_fast_shutdown=1 or 0) is necessary,
to ensure that any incomplete transactions are rolled back.
The undo log format was changed in MDEV-12288. There is only
one persistent undo log for each transaction.

13 files changed, 248 insertions, 380 deletions

diff --git a/extra/innochecksum.cc b/extra/innochecksum.cc
index 73f62c26dd2..583a0db04f9 100644
--- a/extra/innochecksum.cc
+++ b/extra/innochecksum.cc
@@ -1,6 +1,6 @@
 /*
    Copyright (c) 2005, 2016, Oracle and/or its affiliates. All rights reserved.
-   Copyright (c) 2014, 2019, MariaDB Corporation.
+   Copyright (c) 2014, 2021, 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
@@ -138,13 +138,10 @@ static ulong			write_check;
 struct innodb_page_type {
 	int n_undo_state_active;
 	int n_undo_state_cached;
-	int n_undo_state_to_free;
 	int n_undo_state_to_purge;
 	int n_undo_state_prepared;
 	int n_undo_state_other;
-	int n_undo_insert;
-	int n_undo_update;
-	int n_undo_other;
+	int n_undo;
 	int n_fil_page_index;
 	int n_fil_page_undo_log;
 	int n_fil_page_inode;
@@ -955,21 +952,7 @@ parse_page(
 			fprintf(file, "#::%llu\t\t|\t\tUndo log page\t\t\t|",
 				cur_page_num);
 		}
-		if (undo_page_type == TRX_UNDO_INSERT) {
-			page_type.n_undo_insert++;
-			if (page_type_dump) {
-				fprintf(file, "\t%s",
-					"Insert Undo log page");
-			}
-
-		} else if (undo_page_type == TRX_UNDO_UPDATE) {
-			page_type.n_undo_update++;
-			if (page_type_dump) {
-				fprintf(file, "\t%s",
-					"Update undo log page");
-			}
-		}
-
+		page_type.n_undo++;
 		undo_page_type = mach_read_from_2(page + TRX_UNDO_SEG_HDR +
 						  TRX_UNDO_STATE);
 		switch (undo_page_type) {
@@ -989,14 +972,6 @@ parse_page(
 				}
 				break;
 
-			case TRX_UNDO_TO_FREE:
-				page_type.n_undo_state_to_free++;
-				if (page_type_dump) {
-					fprintf(file, ", %s", "Insert undo "
-						"segment that can be freed");
-				}
-				break;
-
 			case TRX_UNDO_TO_PURGE:
 				page_type.n_undo_state_to_purge++;
 				if (page_type_dump) {
@@ -1220,15 +1195,11 @@ print_summary(
 
 	fprintf(fil_out, "\n===============================================\n");
 	fprintf(fil_out, "Additional information:\n");
-	fprintf(fil_out, "Undo page type: %d insert, %d update, %d other\n",
-		page_type.n_undo_insert,
-		page_type.n_undo_update,
-		page_type.n_undo_other);
-	fprintf(fil_out, "Undo page state: %d active, %d cached, %d to_free, %d"
+	fprintf(fil_out, "Undo page type: %d\n", page_type.n_undo);
+	fprintf(fil_out, "Undo page state: %d active, %d cached, %d"
 		" to_purge, %d prepared, %d other\n",
 		page_type.n_undo_state_active,
 		page_type.n_undo_state_cached,
-		page_type.n_undo_state_to_free,
 		page_type.n_undo_state_to_purge,
 		page_type.n_undo_state_prepared,
 		page_type.n_undo_state_other);
diff --git a/mysql-test/suite/innodb_zip/r/innochecksum_3.result b/mysql-test/suite/innodb_zip/r/innochecksum_3.result
index aaab68b3df9..830888e6b8a 100644
--- a/mysql-test/suite/innodb_zip/r/innochecksum_3.result
+++ b/mysql-test/suite/innodb_zip/r/innochecksum_3.result
@@ -109,8 +109,8 @@ File::tab#.ibd
 
 ===============================================
 Additional information:
-Undo page type: # insert, # update, # other
-Undo page state: # active, # cached, # to_free, # to_purge, # prepared, # other
+Undo page type: #
+Undo page state: # active, # cached, # to_purge, # prepared, # other
 index_id	#pages		#leaf_pages	#recs_per_page	#bytes_per_page
 #		#		#		#		#
 #		#		#		#		#
@@ -144,8 +144,8 @@ File::tab#.ibd
 
 ===============================================
 Additional information:
-Undo page type: # insert, # update, # other
-Undo page state: # active, # cached, # to_free, # to_purge, # prepared, # other
+Undo page type: #
+Undo page state: # active, # cached, # to_purge, # prepared, # other
 index_id	#pages		#leaf_pages	#recs_per_page	#bytes_per_page
 #		#		#		#		#
 #		#		#		#		#
diff --git a/storage/innobase/include/trx0purge.h b/storage/innobase/include/trx0purge.h
index 4bc5aded341..f45bd1410f5 100644
--- a/storage/innobase/include/trx0purge.h
+++ b/storage/innobase/include/trx0purge.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2021, 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
@@ -78,20 +78,16 @@ public:
 	typedef trx_rsegs_t::iterator iterator;
 	typedef trx_rsegs_t::const_iterator const_iterator;
 
-	/** Default constructor */
-	TrxUndoRsegs() {}
+	TrxUndoRsegs() : trx_no(0), m_rsegs() {}
 	/** Constructor */
 	TrxUndoRsegs(trx_rseg_t& rseg)
-		: m_commit(rseg.last_commit), m_rsegs(1, &rseg) {}
+		: trx_no(rseg.last_trx_no()), m_rsegs(1, &rseg) {}
 	/** Constructor */
 	TrxUndoRsegs(trx_id_t trx_no, trx_rseg_t& rseg)
-		: m_commit(trx_no << 1), m_rsegs(1, &rseg) {}
-
-	/** @return the transaction commit identifier */
-	trx_id_t trx_no() const { return m_commit >> 1; }
+		: trx_no(trx_no), m_rsegs(1, &rseg) {}
 
 	bool operator!=(const TrxUndoRsegs& other) const
-	{ return m_commit != other.m_commit; }
+	{ return trx_no != other.trx_no; }
 	bool empty() const { return m_rsegs.empty(); }
 	void erase(iterator& it) { m_rsegs.erase(it); }
 	iterator begin() { return(m_rsegs.begin()); }
@@ -105,14 +101,14 @@ public:
 	@return true if elem1 > elem2 else false.*/
 	bool operator()(const TrxUndoRsegs& lhs, const TrxUndoRsegs& rhs)
 	{
-		return(lhs.m_commit > rhs.m_commit);
+		return(lhs.trx_no > rhs.trx_no);
 	}
 
+	/** Copy of trx_rseg_t::last_trx_no() */
+	trx_id_t trx_no;
 private:
-	/** Copy trx_rseg_t::last_commit */
-	trx_id_t		m_commit;
 	/** Rollback segments of a transaction, scheduled for purge. */
-	trx_rsegs_t		m_rsegs;
+	trx_rsegs_t m_rsegs;
 };
 
 typedef std::priority_queue<
@@ -370,17 +366,13 @@ public:
 	{
 		bool operator<=(const iterator& other) const
 		{
-			if (commit < other.commit) return true;
-			if (commit > other.commit) return false;
+			if (trx_no < other.trx_no) return true;
+			if (trx_no > other.trx_no) return false;
 			return undo_no <= other.undo_no;
 		}
 
-		/** @return the commit number of the transaction */
-		trx_id_t trx_no() const { return commit >> 1; }
-		void reset_trx_no(trx_id_t trx_no) { commit = trx_no << 1; }
-
-		/** 2 * trx_t::no + old_insert of the committed transaction */
-		trx_id_t	commit;
+		/** trx_t::no of the committed transaction */
+		trx_id_t	trx_no;
 		/** The record number within the committed transaction's undo
 		log, increasing, purged from from 0 onwards */
 		undo_no_t	undo_no;
diff --git a/storage/innobase/include/trx0rseg.h b/storage/innobase/include/trx0rseg.h
index d4fdb19a988..687c6fa0e97 100644
--- a/storage/innobase/include/trx0rseg.h
+++ b/storage/innobase/include/trx0rseg.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2021, 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
@@ -82,9 +82,8 @@ trx_rseg_header_create(
 	buf_block_t*	sys_header,
 	mtr_t*		mtr);
 
-/** Initialize the rollback segments in memory at database startup. */
-void
-trx_rseg_array_init();
+/** Initialize or recover the rollback segments at startup. */
+dberr_t trx_rseg_array_init();
 
 /** Free a rollback segment in memory. */
 void
@@ -147,21 +146,13 @@ struct trx_rseg_t {
 	/** List of undo log segments cached for fast reuse */
 	UT_LIST_BASE_NODE_T(trx_undo_t)	undo_cached;
 
-	/** List of recovered old insert_undo logs of incomplete
-	transactions (to roll back or XA COMMIT & purge) */
-	UT_LIST_BASE_NODE_T(trx_undo_t) old_insert_list;
-
 	/*--------------------------------------------------------*/
 
-	/** Page number of the last not yet purged log header in the history
-	list; FIL_NULL if all list purged */
-	ulint				last_page_no;
-
-	/** Byte offset of the last not yet purged log header */
-	ulint				last_offset;
+  /** Last not yet purged undo log header; FIL_NULL if all purged */
+  uint32_t last_page_no;
 
-	/** trx_t::no * 2 + old_insert of the last not yet purged log */
-	trx_id_t			last_commit;
+  /** trx_t::no | last_offset << 48 */
+  uint64_t last_commit_and_offset;
 
 	/** Whether the log segment needs purge */
 	bool				needs_purge;
@@ -173,13 +164,17 @@ struct trx_rseg_t {
 	UNDO-tablespace marked for truncate. */
 	bool				skip_allocation;
 
-	/** @return the commit ID of the last committed transaction */
-	trx_id_t last_trx_no() const { return last_commit >> 1; }
-
-	void set_last_trx_no(trx_id_t trx_no, bool is_update)
-	{
-		last_commit = trx_no << 1 | trx_id_t(is_update);
-	}
+  /** @return the commit ID of the last committed transaction */
+  trx_id_t last_trx_no() const
+  { return last_commit_and_offset & ((1ULL << 48) - 1); }
+  /** @return header offset of the last committed transaction */
+  uint16_t last_offset() const
+  { return static_cast<uint16_t>(last_commit_and_offset >> 48); }
+
+  void set_last_commit(ulint last_offset, trx_id_t trx_no)
+  {
+    last_commit_and_offset= static_cast<uint64_t>(last_offset) << 48 | trx_no;
+  }
 
 	/** @return whether the rollback segment is persistent */
 	bool is_persistent() const
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 7cd11bf5496..b4ed45f3565 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -79,7 +79,7 @@ void trx_free_at_shutdown(trx_t *trx);
 void trx_disconnect_prepared(trx_t *trx);
 
 /** Initialize (resurrect) transactions at startup. */
-void trx_lists_init_at_db_start();
+dberr_t trx_lists_init_at_db_start();
 
 /*************************************************************//**
 Starts the transaction if it is not yet started. */
@@ -698,10 +698,6 @@ struct trx_undo_ptr_t {
 					yet */
 	trx_undo_t*	undo;		/*!< pointer to the undo log, or
 					NULL if nothing logged yet */
-	trx_undo_t*     old_insert;	/*!< pointer to recovered
-					insert undo log, or NULL if no
-					INSERT transactions were
-					recovered from old-format undo logs */
 };
 
 /** An instance of temporary rollback segment. */
@@ -1055,13 +1051,6 @@ public:
 		return(has_logged_persistent() || rsegs.m_noredo.undo);
 	}
 
-	/** @return whether any undo log has been generated or
-	recovered */
-	bool has_logged_or_recovered() const
-	{
-		return(has_logged() || rsegs.m_redo.old_insert);
-	}
-
 	/** @return rollback segment for modifying temporary tables */
 	trx_rseg_t* get_temp_rseg()
 	{
diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index 22420f111b5..b0b7ce2941f 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, 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
@@ -253,13 +253,11 @@ trx_undo_set_state_at_prepare(
 	bool		rollback,
 	mtr_t*		mtr);
 
-/** Free an old insert or temporary undo log after commit or rollback.
+/** Free temporary undo log after commit or rollback.
 The information is not needed after a commit or rollback, therefore
 the data can be discarded.
-@param[in,out]	undo	undo log
-@param[in]	is_temp	whether this is temporary undo log */
-void
-trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp);
+@param undo     temporary undo log */
+void trx_undo_commit_cleanup(trx_undo_t *undo);
 
 /** At shutdown, frees the undo logs of a transaction. */
 void
@@ -302,10 +300,11 @@ trx_undo_parse_page_header(
 @param[in]	id		rollback segment slot
 @param[in]	page_no		undo log segment page number
 @param[in,out]	max_trx_id	the largest observed transaction ID
-@return	size of the undo log in pages */
-ulint
-trx_undo_mem_create_at_db_start(trx_rseg_t* rseg, ulint id, ulint page_no,
-				trx_id_t& max_trx_id);
+@return	the undo log
+@retval nullptr on error */
+trx_undo_t *
+trx_undo_mem_create_at_db_start(trx_rseg_t *rseg, ulint id, uint32_t page_no,
+                                trx_id_t &max_trx_id);
 
 #endif /* !UNIV_INNOCHECKSUM */
 
@@ -319,7 +318,6 @@ trx_undo_mem_create_at_db_start(trx_rseg_t* rseg, ulint id, ulint page_no,
 #define TRX_UNDO_ACTIVE		1	/* contains an undo log of an active
 					transaction */
 #define	TRX_UNDO_CACHED		2	/* cached for quick reuse */
-#define	TRX_UNDO_TO_FREE	3	/* insert undo segment can be freed */
 #define	TRX_UNDO_TO_PURGE	4	/* update undo segment will not be
 					reused: it can be freed in purge when
 					all undo data in it is removed */
@@ -383,7 +381,8 @@ struct trx_undo_t {
 /** Transaction undo log page header offsets */
 /* @{ */
 #define	TRX_UNDO_PAGE_TYPE	0	/*!< unused; 0 (before MariaDB 10.3.1:
-					TRX_UNDO_INSERT or TRX_UNDO_UPDATE) */
+					1=TRX_UNDO_INSERT or
+					2=TRX_UNDO_UPDATE) */
 #define	TRX_UNDO_PAGE_START	2	/*!< Byte offset where the undo log
 					records for the LATEST transaction
 					start on this page (remember that
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index b3086842624..c584de0fa5d 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -4584,7 +4584,7 @@ lock_print_info_summary(
 		"Purge done for trx's n:o < " TRX_ID_FMT
 		" undo n:o < " TRX_ID_FMT " state: %s\n"
 		"History list length %u\n",
-		purge_sys.tail.trx_no(),
+		purge_sys.tail.trx_no,
 		purge_sys.tail.undo_no,
 		purge_sys.enabled()
 		? (purge_sys.running() ? "running"
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index 7314fd60cd6..7d5679e889e 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -1404,6 +1404,11 @@ dberr_t srv_start(bool create_new_db)
 	      || is_mariabackup_restore_or_export());
 
 
+	if (srv_force_recovery) {
+		ib::info() << "!!! innodb_force_recovery is set to "
+			<< srv_force_recovery << " !!!";
+	}
+
 	if (srv_force_recovery == SRV_FORCE_NO_LOG_REDO) {
 		srv_read_only_mode = true;
 	}
@@ -1922,7 +1927,11 @@ files_checked:
 		All the remaining rollback segments will be created later,
 		after the double write buffer has been created. */
 		trx_sys_create_sys_pages();
-		trx_lists_init_at_db_start();
+		err = trx_lists_init_at_db_start();
+
+		if (err != DB_SUCCESS) {
+			return(srv_init_abort(err));
+		}
 
 		err = dict_create();
 
@@ -1986,7 +1995,10 @@ files_checked:
 		case SRV_OPERATION_RESTORE:
 			/* This must precede
 			recv_apply_hashed_log_recs(true). */
-			trx_lists_init_at_db_start();
+			err = trx_lists_init_at_db_start();
+			if (err != DB_SUCCESS) {
+				return srv_init_abort(err);
+			}
 			break;
 		case SRV_OPERATION_RESTORE_DELTA:
 		case SRV_OPERATION_BACKUP:
@@ -2453,11 +2465,6 @@ skip_monitors:
 			   << "; transaction id " << trx_sys.get_max_trx_id();
 	}
 
-	if (srv_force_recovery > 0) {
-		ib::info() << "!!! innodb_force_recovery is set to "
-			<< srv_force_recovery << " !!!";
-	}
-
 	if (srv_force_recovery == 0) {
 		/* In the insert buffer we may have even bigger tablespace
 		id's, because we may have dropped those tablespaces, but
diff --git a/storage/innobase/trx/trx0purge.cc b/storage/innobase/trx/trx0purge.cc
index 02a524d6850..59d60d204d8 100644
--- a/storage/innobase/trx/trx0purge.cc
+++ b/storage/innobase/trx/trx0purge.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, 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
@@ -87,7 +87,7 @@ inline bool TrxUndoRsegsIterator::set_next()
 		number shouldn't increase. Undo the increment of
 		expected commit done by caller assuming rollback
 		segments from given transaction are done. */
-		purge_sys.tail.commit = (*m_iter)->last_commit;
+		purge_sys.tail.trx_no = (*m_iter)->last_trx_no();
 	} else if (!purge_sys.purge_queue.empty()) {
 		m_rsegs = purge_sys.purge_queue.top();
 		purge_sys.purge_queue.pop();
@@ -108,17 +108,17 @@ inline bool TrxUndoRsegsIterator::set_next()
 	mutex_enter(&purge_sys.rseg->mutex);
 
 	ut_a(purge_sys.rseg->last_page_no != FIL_NULL);
-	ut_ad(purge_sys.rseg->last_trx_no() == m_rsegs.trx_no());
+	ut_ad(purge_sys.rseg->last_trx_no() == m_rsegs.trx_no);
 
 	/* We assume in purge of externally stored fields that space id is
 	in the range of UNDO tablespace space ids */
 	ut_ad(purge_sys.rseg->space->id == TRX_SYS_SPACE
 	      || srv_is_undo_tablespace(purge_sys.rseg->space->id));
 
-	ut_a(purge_sys.tail.commit <= purge_sys.rseg->last_commit);
+	ut_a(purge_sys.tail.trx_no <= purge_sys.rseg->last_trx_no());
 
-	purge_sys.tail.commit = purge_sys.rseg->last_commit;
-	purge_sys.hdr_offset = purge_sys.rseg->last_offset;
+	purge_sys.tail.trx_no = purge_sys.rseg->last_trx_no();
+	purge_sys.hdr_offset = purge_sys.rseg->last_offset();
 	purge_sys.hdr_page_no = purge_sys.rseg->last_page_no;
 
 	mutex_exit(&purge_sys.rseg->mutex);
@@ -209,8 +209,7 @@ trx_purge_add_undo_to_history(const trx_t* trx, trx_undo_t*& undo, mtr_t* mtr)
 {
 	DBUG_PRINT("trx", ("commit(" TRX_ID_FMT "," TRX_ID_FMT ")",
 			   trx->id, trx->no));
-	ut_ad(undo == trx->rsegs.m_redo.undo
-	      || undo == trx->rsegs.m_redo.old_insert);
+	ut_ad(undo == trx->rsegs.m_redo.undo);
 	trx_rseg_t*	rseg		= trx->rsegs.m_redo.rseg;
 	ut_ad(undo->rseg == rseg);
 	trx_rsegf_t*	rseg_header	= trx_rsegf_get(
@@ -302,9 +301,8 @@ trx_purge_add_undo_to_history(const trx_t* trx, trx_undo_t*& undo, mtr_t* mtr)
 	}
 
 	if (rseg->last_page_no == FIL_NULL) {
-		rseg->last_page_no = undo->hdr_page_no;
-		rseg->last_offset = undo->hdr_offset;
-		rseg->set_last_trx_no(trx->no, undo == trx->rsegs.m_redo.undo);
+		rseg->last_page_no = static_cast<uint32_t>(undo->hdr_page_no);
+		rseg->set_last_commit(undo->hdr_offset, trx->no);
 		rseg->needs_purge = true;
 	}
 
@@ -460,8 +458,8 @@ func_exit:
 
 	undo_trx_no = mach_read_from_8(log_hdr + TRX_UNDO_TRX_NO);
 
-	if (undo_trx_no >= limit.trx_no()) {
-		if (undo_trx_no == limit.trx_no()) {
+	if (undo_trx_no >= limit.trx_no) {
+		if (undo_trx_no == limit.trx_no) {
 			trx_undo_truncate_start(
 				&rseg, hdr_addr.page,
 				hdr_addr.boffset, limit.undo_no);
@@ -884,7 +882,7 @@ trx_purge_initiate_truncate(
 			     undo != NULL && all_free;
 			     undo = UT_LIST_GET_NEXT(undo_list, undo)) {
 
-				if (limit.trx_no() < undo->trx_id) {
+				if (limit.trx_no < undo->trx_id) {
 					all_free = false;
 				} else {
 					cached_undo_size += undo->size;
@@ -986,7 +984,6 @@ not_found:
 		/* Before re-initialization ensure that we free the existing
 		structure. There can't be any active transactions. */
 		ut_a(UT_LIST_GET_LEN(rseg->undo_list) == 0);
-		ut_a(UT_LIST_GET_LEN(rseg->old_insert_list) == 0);
 
 		trx_undo_t*	next_undo;
 
@@ -1002,7 +999,6 @@ not_found:
 
 		UT_LIST_INIT(rseg->undo_list, &trx_undo_t::undo_list);
 		UT_LIST_INIT(rseg->undo_cached, &trx_undo_t::undo_list);
-		UT_LIST_INIT(rseg->old_insert_list, &trx_undo_t::undo_list);
 
 		/* These were written by trx_rseg_header_create(). */
 		ut_ad(!mach_read_from_4(TRX_RSEG + TRX_RSEG_FORMAT
@@ -1014,8 +1010,7 @@ not_found:
 		rseg->curr_size = 1;
 		rseg->trx_ref_count = 0;
 		rseg->last_page_no = FIL_NULL;
-		rseg->last_offset = 0;
-		rseg->last_commit = 0;
+		rseg->last_commit_and_offset = 0;
 		rseg->needs_purge = false;
 	}
 
@@ -1076,12 +1071,12 @@ function is called, the caller must not have any latches on undo log pages!
 static void trx_purge_truncate_history()
 {
 	ut_ad(purge_sys.head <= purge_sys.tail);
-	purge_sys_t::iterator& head = purge_sys.head.commit
+	purge_sys_t::iterator& head = purge_sys.head.trx_no
 		? purge_sys.head : purge_sys.tail;
 
-	if (head.trx_no() >= purge_sys.view.low_limit_no()) {
+	if (head.trx_no >= purge_sys.view.low_limit_no()) {
 		/* This is sometimes necessary. TODO: find out why. */
-		head.reset_trx_no(purge_sys.view.low_limit_no());
+		head.trx_no = purge_sys.view.low_limit_no();
 		head.undo_no = 0;
 	}
 
@@ -1109,7 +1104,6 @@ static void trx_purge_rseg_get_next_history_log(
 					handled */
 {
 	page_t*		undo_page;
-	trx_ulogf_t*	log_hdr;
 	fil_addr_t	prev_log_addr;
 	trx_id_t	trx_no;
 	mtr_t		mtr;
@@ -1118,7 +1112,7 @@ static void trx_purge_rseg_get_next_history_log(
 
 	ut_a(purge_sys.rseg->last_page_no != FIL_NULL);
 
-	purge_sys.tail.commit = purge_sys.rseg->last_commit + 1;
+	purge_sys.tail.trx_no = purge_sys.rseg->last_trx_no() + 1;
 	purge_sys.tail.undo_no = 0;
 	purge_sys.next_stored = false;
 
@@ -1128,7 +1122,7 @@ static void trx_purge_rseg_get_next_history_log(
 		page_id_t(purge_sys.rseg->space->id,
 			  purge_sys.rseg->last_page_no), &mtr);
 
-	log_hdr = undo_page + purge_sys.rseg->last_offset;
+	const trx_ulogf_t* log_hdr = undo_page + purge_sys.rseg->last_offset();
 
 	/* Increase the purge page count by one for every handled log */
 
@@ -1160,17 +1154,16 @@ static void trx_purge_rseg_get_next_history_log(
 		+ prev_log_addr.boffset;
 
 	trx_no = mach_read_from_8(log_hdr + TRX_UNDO_TRX_NO);
-	unsigned purge = mach_read_from_2(log_hdr + TRX_UNDO_NEEDS_PURGE);
-	ut_ad(purge <= 1);
+	ut_ad(mach_read_from_2(log_hdr + TRX_UNDO_NEEDS_PURGE) <= 1);
 
 	mtr_commit(&mtr);
 
 	mutex_enter(&purge_sys.rseg->mutex);
 
-	purge_sys.rseg->last_page_no = prev_log_addr.page;
-	purge_sys.rseg->last_offset = prev_log_addr.boffset;
-	purge_sys.rseg->set_last_trx_no(trx_no, purge != 0);
-	purge_sys.rseg->needs_purge = purge != 0;
+	purge_sys.rseg->last_page_no = static_cast<uint32_t>(
+		prev_log_addr.page);
+	purge_sys.rseg->set_last_commit(prev_log_addr.boffset, trx_no);
+	purge_sys.rseg->needs_purge = log_hdr[TRX_UNDO_NEEDS_PURGE + 1] != 0;
 
 	/* Purge can also produce events, however these are already ordered
 	in the rollback segment and any user generated event will be greater
@@ -1187,15 +1180,13 @@ static void trx_purge_rseg_get_next_history_log(
 }
 
 /** Position the purge sys "iterator" on the undo record to use for purging. */
-static
-void
-trx_purge_read_undo_rec()
+static void trx_purge_read_undo_rec()
 {
 	ulint		offset;
 	ulint		page_no;
 	ib_uint64_t	undo_no;
 
-	purge_sys.hdr_offset = purge_sys.rseg->last_offset;
+	purge_sys.hdr_offset = purge_sys.rseg->last_offset();
 	page_no = purge_sys.hdr_page_no = purge_sys.rseg->last_page_no;
 
 	if (purge_sys.rseg->needs_purge) {
@@ -1268,7 +1259,7 @@ trx_purge_get_next_rec(
 	mtr_t		mtr;
 
 	ut_ad(purge_sys.next_stored);
-	ut_ad(purge_sys.tail.trx_no() < purge_sys.view.low_limit_no());
+	ut_ad(purge_sys.tail.trx_no < purge_sys.view.low_limit_no());
 
 	space = purge_sys.rseg->space->id;
 	page_no = purge_sys.page_no;
@@ -1361,7 +1352,7 @@ trx_purge_fetch_next_rec(
 		}
 	}
 
-	if (purge_sys.tail.trx_no() >= purge_sys.view.low_limit_no()) {
+	if (purge_sys.tail.trx_no >= purge_sys.view.low_limit_no()) {
 
 		return(NULL);
 	}
diff --git a/storage/innobase/trx/trx0roll.cc b/storage/innobase/trx/trx0roll.cc
index 88ff251548c..bffd168359b 100644
--- a/storage/innobase/trx/trx0roll.cc
+++ b/storage/innobase/trx/trx0roll.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2016, 2020, MariaDB Corporation.
+Copyright (c) 2016, 2021, 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
@@ -70,12 +70,6 @@ static bool trx_rollback_finish(trx_t* trx)
 		ut_a(!srv_undo_sources);
 		ut_ad(srv_fast_shutdown);
 		ut_d(trx->in_rollback = false);
-		if (trx_undo_t*& undo = trx->rsegs.m_redo.old_insert) {
-			UT_LIST_REMOVE(trx->rsegs.m_redo.rseg->old_insert_list,
-				       undo);
-			ut_free(undo);
-			undo = NULL;
-		}
 		if (trx_undo_t*& undo = trx->rsegs.m_redo.undo) {
 			UT_LIST_REMOVE(trx->rsegs.m_redo.rseg->undo_list,
 				       undo);
@@ -124,7 +118,7 @@ trx_rollback_to_savepoint_low(
 
 	trx->error_state = DB_SUCCESS;
 
-	if (trx->has_logged_or_recovered()) {
+	if (trx->has_logged()) {
 
 		ut_ad(trx->rsegs.m_redo.rseg != 0
 		      || trx->rsegs.m_noredo.rseg != 0);
@@ -240,7 +234,7 @@ dberr_t trx_rollback_for_mysql(trx_t* trx)
 	case TRX_STATE_PREPARED:
 	case TRX_STATE_PREPARED_RECOVERED:
 		ut_ad(!trx_is_autocommit_non_locking(trx));
-		if (trx->rsegs.m_redo.undo || trx->rsegs.m_redo.old_insert) {
+		if (trx->rsegs.m_redo.undo) {
 			/* The XA ROLLBACK of a XA PREPARE transaction
 			will consist of multiple mini-transactions.
 
@@ -256,11 +250,7 @@ dberr_t trx_rollback_for_mysql(trx_t* trx)
 			killed, and finally, the transaction would be
 			recovered in XA PREPARE state, with some of
 			the actions already having been rolled back. */
-			ut_ad(!trx->rsegs.m_redo.undo
-			      || trx->rsegs.m_redo.undo->rseg
-			      == trx->rsegs.m_redo.rseg);
-			ut_ad(!trx->rsegs.m_redo.old_insert
-			      || trx->rsegs.m_redo.old_insert->rseg
+			ut_ad(trx->rsegs.m_redo.undo->rseg
 			      == trx->rsegs.m_redo.rseg);
 			mtr_t		mtr;
 			mtr.start();
@@ -269,10 +259,6 @@ dberr_t trx_rollback_for_mysql(trx_t* trx)
 				trx_undo_set_state_at_prepare(trx, undo, true,
 							      &mtr);
 			}
-			if (trx_undo_t* undo = trx->rsegs.m_redo.old_insert) {
-				trx_undo_set_state_at_prepare(trx, undo, true,
-							      &mtr);
-			}
 			mutex_exit(&trx->rsegs.m_redo.rseg->mutex);
 			/* Write the redo log for the XA ROLLBACK
 			state change to the global buffer. It is
@@ -959,23 +945,13 @@ trx_roll_pop_top_rec_of_trx(trx_t* trx, roll_ptr_t* roll_ptr, mem_heap_t* heap)
 	}
 
 	trx_undo_t*	undo	= NULL;
-	trx_undo_t*	insert	= trx->rsegs.m_redo.old_insert;
 	trx_undo_t*	update	= trx->rsegs.m_redo.undo;
 	trx_undo_t*	temp	= trx->rsegs.m_noredo.undo;
 	const undo_no_t	limit	= trx->roll_limit;
 
-	ut_ad(!insert || !update || insert->empty() || update->empty()
-	      || insert->top_undo_no != update->top_undo_no);
-	ut_ad(!insert || !temp || insert->empty() || temp->empty()
-	      || insert->top_undo_no != temp->top_undo_no);
 	ut_ad(!update || !temp || update->empty() || temp->empty()
 	      || update->top_undo_no != temp->top_undo_no);
 
-	if (UNIV_LIKELY_NULL(insert)
-	    && !insert->empty() && limit <= insert->top_undo_no) {
-		undo = insert;
-	}
-
 	if (update && !update->empty() && update->top_undo_no >= limit) {
 		if (!undo) {
 			undo = update;
@@ -1020,18 +996,12 @@ trx_roll_pop_top_rec_of_trx(trx_t* trx, roll_ptr_t* roll_ptr, mem_heap_t* heap)
 		MDEV-12288 removed the insert_undo log. There is no
 		instant ADD COLUMN for temporary tables. Therefore,
 		this record can only be present in the main undo log. */
-		ut_ad(undo == update);
 		/* fall through */
 	case TRX_UNDO_RENAME_TABLE:
-		ut_ad(undo == insert || undo == update);
+		ut_ad(undo == update);
 		/* fall through */
 	case TRX_UNDO_INSERT_REC:
-		ut_ad(undo == insert || undo == update || undo == temp);
 		*roll_ptr |= 1ULL << ROLL_PTR_INSERT_FLAG_POS;
-		break;
-	default:
-		ut_ad(undo == update || undo == temp);
-		break;
 	}
 
 	trx->undo_no = undo_no;
diff --git a/storage/innobase/trx/trx0rseg.cc b/storage/innobase/trx/trx0rseg.cc
index d6720979716..f660b3151e6 100644
--- a/storage/innobase/trx/trx0rseg.cc
+++ b/storage/innobase/trx/trx0rseg.cc
@@ -359,7 +359,6 @@ trx_rseg_mem_free(trx_rseg_t* rseg)
 
 	/* There can't be any active transactions. */
 	ut_a(UT_LIST_GET_LEN(rseg->undo_list) == 0);
-	ut_a(UT_LIST_GET_LEN(rseg->old_insert_list) == 0);
 
 	for (undo = UT_LIST_GET_FIRST(rseg->undo_cached);
 	     undo != NULL;
@@ -399,45 +398,45 @@ trx_rseg_mem_create(ulint id, fil_space_t* space, ulint page_no)
 		     &rseg->mutex);
 
 	UT_LIST_INIT(rseg->undo_list, &trx_undo_t::undo_list);
-	UT_LIST_INIT(rseg->old_insert_list, &trx_undo_t::undo_list);
 	UT_LIST_INIT(rseg->undo_cached, &trx_undo_t::undo_list);
 
 	return(rseg);
 }
 
 /** Read the undo log lists.
-@param[in,out]	rseg		rollback segment
-@param[in,out]	max_trx_id	maximum observed transaction identifier
-@param[in]	rseg_header	rollback segment header
-@return the combined size of undo log segments in pages */
-static
-ulint
-trx_undo_lists_init(trx_rseg_t* rseg, trx_id_t& max_trx_id,
-		    const trx_rsegf_t* rseg_header)
+@param[in,out]  rseg            rollback segment
+@param[in,out]  max_trx_id      maximum observed transaction identifier
+@param[in]      rseg_header     rollback segment header
+@return error code */
+static dberr_t trx_undo_lists_init(trx_rseg_t *rseg, trx_id_t &max_trx_id,
+                                   const trx_rsegf_t *rseg_header)
 {
 	ut_ad(srv_force_recovery < SRV_FORCE_NO_UNDO_LOG_SCAN);
 
-	ulint size = 0;
-
-	for (ulint i = 0; i < TRX_RSEG_N_SLOTS; i++) {
-		ulint	page_no = trx_rsegf_get_nth_undo(rseg_header, i);
-		if (page_no != FIL_NULL) {
-			size += trx_undo_mem_create_at_db_start(
-				rseg, i, page_no, max_trx_id);
-			MONITOR_INC(MONITOR_NUM_UNDO_SLOT_USED);
-		}
-	}
-
-	return(size);
+  for (ulint i= 0; i < TRX_RSEG_N_SLOTS; i++)
+  {
+    uint32_t page_no= trx_rsegf_get_nth_undo(rseg_header, i);
+    if (page_no != FIL_NULL)
+    {
+      const trx_undo_t *undo= trx_undo_mem_create_at_db_start(rseg, i, page_no,
+                                                              max_trx_id);
+      if (!undo)
+        return DB_CORRUPTION;
+      rseg->curr_size+= undo->size;
+      MONITOR_INC(MONITOR_NUM_UNDO_SLOT_USED);
+    }
+  }
+
+  return DB_SUCCESS;
 }
 
 /** Restore the state of a persistent rollback segment.
 @param[in,out]	rseg		persistent rollback segment
 @param[in,out]	max_trx_id	maximum observed transaction identifier
-@param[in,out]	mtr		mini-transaction */
-static
-void
-trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
+@param[in,out]	mtr		mini-transaction
+@return error code */
+static dberr_t trx_rseg_mem_restore(trx_rseg_t *rseg, trx_id_t &max_trx_id,
+                                    mtr_t *mtr)
 {
 	/* This is based on trx_rsegf_get_new().
 	We need to access buf_block_t. */
@@ -484,13 +483,16 @@ trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
 		/* mariabackup --prepare only deals with
 		the redo log and the data files, not with
 		transactions or the data dictionary. */
-		return;
+		return DB_SUCCESS;
 	}
 
 	/* Initialize the undo log lists according to the rseg header */
 
 	rseg->curr_size = mach_read_from_4(rseg_header + TRX_RSEG_HISTORY_SIZE)
-		+ 1 + trx_undo_lists_init(rseg, max_trx_id, rseg_header);
+		+ 1;
+	if (dberr_t err = trx_undo_lists_init(rseg, max_trx_id, rseg_header)) {
+		return err;
+	}
 
 	if (ulint len = flst_get_len(rseg_header + TRX_RSEG_HISTORY)) {
 		trx_sys.history_add(int32(len));
@@ -498,8 +500,7 @@ trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
 		fil_addr_t	node_addr = trx_purge_get_log_from_hist(
 			flst_get_last(rseg_header + TRX_RSEG_HISTORY, mtr));
 
-		rseg->last_page_no = node_addr.page;
-		rseg->last_offset = node_addr.boffset;
+		rseg->last_page_no = static_cast<uint32_t>(node_addr.page);
 
 		const trx_ulogf_t*	undo_log_hdr = trx_undo_page_get(
 			page_id_t(rseg->space->id, node_addr.page), mtr)
@@ -513,10 +514,10 @@ trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
 		if (id > max_trx_id) {
 			max_trx_id = id;
 		}
+		rseg->set_last_commit(node_addr.boffset, id);
 		unsigned purge = mach_read_from_2(
 			undo_log_hdr + TRX_UNDO_NEEDS_PURGE);
 		ut_ad(purge <= 1);
-		rseg->set_last_trx_no(id, purge != 0);
 		rseg->needs_purge = purge != 0;
 
 		if (rseg->last_page_no != FIL_NULL) {
@@ -526,6 +527,8 @@ trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
 			purge_sys.purge_queue.push(*rseg);
 		}
 	}
+
+	return DB_SUCCESS;
 }
 
 /** Read binlog metadata from the TRX_SYS page, in case we are upgrading
@@ -549,9 +552,8 @@ static void trx_rseg_init_binlog_info(const page_t* page)
 #endif
 }
 
-/** Initialize the rollback segments in memory at database startup. */
-void
-trx_rseg_array_init()
+/** Initialize or recover the rollback segments at startup. */
+dberr_t trx_rseg_array_init()
 {
 	trx_id_t max_trx_id = 0;
 
@@ -563,9 +565,9 @@ trx_rseg_array_init()
 	wsrep_sys_xid.null();
 	bool wsrep_xid_in_rseg_found = false;
 #endif
+	mtr_t mtr;
 
 	for (ulint rseg_id = 0; rseg_id < TRX_SYS_N_RSEGS; rseg_id++) {
-		mtr_t mtr;
 		mtr.start();
 		if (const buf_block_t* sys = trx_sysf_get(&mtr, false)) {
 			if (rseg_id == 0) {
@@ -593,7 +595,11 @@ trx_rseg_array_init()
 				ut_ad(rseg->id == rseg_id);
 				ut_ad(!trx_sys.rseg_array[rseg_id]);
 				trx_sys.rseg_array[rseg_id] = rseg;
-				trx_rseg_mem_restore(rseg, max_trx_id, &mtr);
+				if (dberr_t err = trx_rseg_mem_restore(
+					    rseg, max_trx_id, &mtr)) {
+					mtr.commit();
+					return err;
+				}
 #ifdef WITH_WSREP
 				if (!wsrep_sys_xid.is_null() &&
 				    !wsrep_sys_xid.eq(&trx_sys.recovered_wsrep_xid)) {
@@ -620,7 +626,6 @@ trx_rseg_array_init()
 		If no rollback segment has a WSREP XID set,
 		we must copy the XID found in TRX_SYS page
 		to rollback segments. */
-		mtr_t mtr;
 		mtr.start();
 
 		if (!wsrep_xid_in_rseg_found) {
@@ -638,6 +643,7 @@ trx_rseg_array_init()
 #endif
 
 	trx_sys.init_max_trx_id(max_trx_id + 1);
+	return DB_SUCCESS;
 }
 
 /** Create a persistent rollback segment.
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index 2620005269c..c20a1270bb9 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -664,8 +664,7 @@ trx_resurrect_table_locks(
 
 static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
                           time_t start_time, ulonglong start_time_micro,
-                          uint64_t *rows_to_undo,
-                          bool is_old_insert)
+                          uint64_t *rows_to_undo)
 {
   trx_state_t state;
   /*
@@ -688,8 +687,6 @@ static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
     state= TRX_STATE_PREPARED;
     break;
   default:
-    if (is_old_insert && srv_force_recovery < SRV_FORCE_NO_TRX_UNDO)
-      trx_undo_commit_cleanup(undo, false);
     return;
   }
 
@@ -699,11 +696,7 @@ static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
   ut_d(trx->start_line= __LINE__);
   ut_ad(trx->no == TRX_ID_MAX);
 
-  if (is_old_insert)
-    trx->rsegs.m_redo.old_insert= undo;
-  else
-    trx->rsegs.m_redo.undo= undo;
-
+  trx->rsegs.m_redo.undo= undo;
   trx->undo_no= undo->top_undo_no + 1;
   trx->rsegs.m_redo.rseg= rseg;
   /*
@@ -734,8 +727,7 @@ static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
 
 
 /** Initialize (resurrect) transactions at startup. */
-void
-trx_lists_init_at_db_start()
+dberr_t trx_lists_init_at_db_start()
 {
 	ut_a(srv_is_being_started);
 	ut_ad(!srv_was_started);
@@ -744,16 +736,18 @@ trx_lists_init_at_db_start()
 		/* mariabackup --prepare only deals with
 		the redo log and the data files, not with
 		transactions or the data dictionary. */
-		trx_rseg_array_init();
-		return;
+		return trx_rseg_array_init();
 	}
 
 	if (srv_force_recovery >= SRV_FORCE_NO_UNDO_LOG_SCAN) {
-		return;
+		return DB_SUCCESS;
 	}
 
 	purge_sys.create();
-	trx_rseg_array_init();
+	if (dberr_t err = trx_rseg_array_init()) {
+		ib::info() << "Retry with innodb_force_recovery=5";
+		return err;
+	}
 
 	/* Look from the rollback segments if there exist undo logs for
 	transactions. */
@@ -771,17 +765,6 @@ trx_lists_init_at_db_start()
 		if (rseg == NULL) {
 			continue;
 		}
-
-		/* Resurrect transactions that were doing inserts
-		using the old separate insert_undo log. */
-		undo = UT_LIST_GET_FIRST(rseg->old_insert_list);
-		while (undo) {
-			trx_undo_t* next = UT_LIST_GET_NEXT(undo_list, undo);
-			trx_resurrect(undo, rseg, start_time, start_time_micro,
-				      &rows_to_undo, true);
-			undo = next;
-		}
-
 		/* Ressurrect other transactions. */
 		for (undo = UT_LIST_GET_FIRST(rseg->undo_list);
 		     undo != NULL;
@@ -789,8 +772,7 @@ trx_lists_init_at_db_start()
 			trx_t *trx = trx_sys.find(0, undo->trx_id, false);
 			if (!trx) {
 				trx_resurrect(undo, rseg, start_time,
-					      start_time_micro,
-					      &rows_to_undo, false);
+					      start_time_micro, &rows_to_undo);
 			} else {
 				ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE) ||
 				      trx_state_eq(trx, TRX_STATE_PREPARED));
@@ -825,6 +807,7 @@ trx_lists_init_at_db_start()
 		ib::info() << "Trx id counter is " << trx_sys.get_max_trx_id();
 	}
 	trx_sys.clone_oldest_view();
+	return DB_SUCCESS;
 }
 
 /** Assign a persistent rollback segment in a round-robin fashion,
@@ -1121,30 +1104,22 @@ trx_write_serialisation_history(
 	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;
 	}
 
 	trx_undo_t*& undo = trx->rsegs.m_redo.undo;
-	trx_undo_t*& old_insert = trx->rsegs.m_redo.old_insert;
 
-	if (!undo && !old_insert) {
+	if (!undo) {
 		return;
 	}
 
 	ut_ad(!trx->read_only);
 	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);
-
-	if (UNIV_LIKELY_NULL(old_insert)) {
-		UT_LIST_REMOVE(rseg->old_insert_list, old_insert);
-		trx_purge_add_undo_to_history(trx, old_insert, mtr);
-	}
 	if (undo) {
 		UT_LIST_REMOVE(rseg->undo_list, undo);
 		trx_purge_add_undo_to_history(trx, undo, mtr);
@@ -1382,20 +1357,12 @@ trx_commit_in_memory(
 		ut_ad(rseg->trx_ref_count > 0);
 		--rseg->trx_ref_count;
 		mutex_exit(&rseg->mutex);
-
-		if (trx_undo_t*& insert = trx->rsegs.m_redo.old_insert) {
-			ut_ad(insert->rseg == rseg);
-			trx_undo_commit_cleanup(insert, false);
-			insert = NULL;
-		}
 	}
 
-	ut_ad(!trx->rsegs.m_redo.old_insert);
-
 	if (mtr != NULL) {
 		if (trx_undo_t*& undo = trx->rsegs.m_noredo.undo) {
 			ut_ad(undo->rseg == trx->rsegs.m_noredo.rseg);
-			trx_undo_commit_cleanup(undo, true);
+			trx_undo_commit_cleanup(undo);
 			undo = NULL;
 		}
 
@@ -1490,7 +1457,7 @@ void trx_commit_low(trx_t* trx, mtr_t* mtr)
 	ut_ad(!mtr || mtr->is_active());
 	ut_d(bool aborted = trx->in_rollback
 	     && trx->error_state == DB_DEADLOCK);
-	ut_ad(!mtr == (aborted || !trx->has_logged_or_recovered()));
+	ut_ad(!mtr == (aborted || !trx->has_logged()));
 	ut_ad(!mtr || !aborted);
 
 	/* undo_no is non-zero if we're doing the final commit. */
@@ -1577,10 +1544,7 @@ trx_commit(
 	mtr_t*	mtr;
 	mtr_t	local_mtr;
 
-	DBUG_EXECUTE_IF("ib_trx_commit_crash_before_trx_commit_start",
-			DBUG_SUICIDE(););
-
-	if (trx->has_logged_or_recovered()) {
+	if (trx->has_logged()) {
 		mtr = &local_mtr;
 		mtr->start();
 	} else {
@@ -1986,11 +1950,8 @@ trx_weight_ge(
 /** Prepare a transaction.
 @return	log sequence number that makes the XA PREPARE durable
 @retval	0	if no changes needed to be made durable */
-static
-lsn_t
-trx_prepare_low(trx_t* trx)
+static lsn_t trx_prepare_low(trx_t *trx)
 {
-	ut_ad(!trx->rsegs.m_redo.old_insert);
 	ut_ad(!trx->is_recovered);
 
 	mtr_t	mtr;
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index 7254c830cde..dea85c40473 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2014, 2020, MariaDB Corporation.
+Copyright (c) 2014, 2021, 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
@@ -34,6 +34,7 @@ Created 3/26/1996 Heikki Tuuri
 #include "trx0purge.h"
 #include "trx0rec.h"
 #include "trx0rseg.h"
+#include "log.h"
 
 /* How should the old versions in the history list be managed?
    ----------------------------------------------------------
@@ -1011,13 +1012,8 @@ loop:
 }
 
 /** Frees an undo log segment which is not in the history list.
-@param[in]	undo	undo log
-@param[in]	noredo	whether the undo tablespace is redo logged */
-static
-void
-trx_undo_seg_free(
-	const trx_undo_t*	undo,
-	bool			noredo)
+@param undo	temporary undo log */
+static void trx_undo_seg_free(const trx_undo_t *undo)
 {
 	trx_rseg_t*	rseg;
 	fseg_header_t*	file_seg;
@@ -1029,16 +1025,12 @@ trx_undo_seg_free(
 	rseg = undo->rseg;
 
 	do {
-
-		mtr_start(&mtr);
-
-		if (noredo) {
-			mtr.set_log_mode(MTR_LOG_NO_REDO);
-		}
+		mtr.start();
+		mtr.set_log_mode(MTR_LOG_NO_REDO);
 
 		mutex_enter(&(rseg->mutex));
 
-		seg_header = trx_undo_page_get(page_id_t(undo->rseg->space->id,
+		seg_header = trx_undo_page_get(page_id_t(SRV_TMP_SPACE_ID,
 							 undo->hdr_page_no),
 					       &mtr)
 			+ TRX_UNDO_SEG_HDR;
@@ -1069,10 +1061,11 @@ trx_undo_seg_free(
 @param[in]	id		rollback segment slot
 @param[in]	page_no		undo log segment page number
 @param[in,out]	max_trx_id	the largest observed transaction ID
-@return	size of the undo log in pages */
-ulint
-trx_undo_mem_create_at_db_start(trx_rseg_t* rseg, ulint id, ulint page_no,
-				trx_id_t& max_trx_id)
+@return	the undo log
+@retval nullptr on error */
+trx_undo_t *
+trx_undo_mem_create_at_db_start(trx_rseg_t *rseg, ulint id, uint32_t page_no,
+                                trx_id_t &max_trx_id)
 {
 	mtr_t		mtr;
 	XID		xid;
@@ -1082,16 +1075,56 @@ trx_undo_mem_create_at_db_start(trx_rseg_t* rseg, ulint id, ulint page_no,
 	mtr.start();
 	const page_t* undo_page = trx_undo_page_get(
 		page_id_t(rseg->space->id, page_no), &mtr);
-	const ulint type = mach_read_from_2(
-		TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_TYPE + undo_page);
-	ut_ad(type == 0 || type == TRX_UNDO_INSERT || type == TRX_UNDO_UPDATE);
+	const uint16_t type = mach_read_from_2(TRX_UNDO_PAGE_HDR
+					       + TRX_UNDO_PAGE_TYPE
+					       + undo_page);
+	switch (type) {
+	case 0:
+	case 2: /* TRX_UNDO_UPDATE */
+		break;
+	case 1: /* TRX_UNDO_INSERT */
+		sql_print_error("InnoDB: upgrade from older version than"
+				" MariaDB 10.3 requires clean shutdown");
+		goto corrupted;
+	default:
+		sql_print_error("InnoDB: unsupported undo header type %u",
+				type);
+	corrupted:
+		mtr.commit();
+		return NULL;
+	}
 
-	uint state = mach_read_from_2(TRX_UNDO_SEG_HDR + TRX_UNDO_STATE
-				      + undo_page);
-	uint offset = mach_read_from_2(TRX_UNDO_SEG_HDR + TRX_UNDO_LAST_LOG
-				       + undo_page);
+	uint16_t offset = mach_read_from_2(TRX_UNDO_SEG_HDR + TRX_UNDO_LAST_LOG
+					   + undo_page);
+	if (offset < TRX_UNDO_SEG_HDR + TRX_UNDO_SEG_HDR_SIZE ||
+	    offset >= srv_page_size - TRX_UNDO_LOG_OLD_HDR_SIZE) {
+		sql_print_error("InnoDB: invalid undo header offset %u",
+				offset);
+		goto corrupted;
+	}
 
-	const trx_ulogf_t*	undo_header = undo_page + offset;
+	const trx_ulogf_t* const undo_header = undo_page + offset;
+	uint16_t state = mach_read_from_2(TRX_UNDO_SEG_HDR + TRX_UNDO_STATE
+					  + undo_page);
+	switch (state) {
+	case TRX_UNDO_ACTIVE:
+	case TRX_UNDO_PREPARED:
+		break;
+	default:
+		sql_print_error("InnoDB: unsupported undo header state %u",
+				state);
+		goto corrupted;
+	case TRX_UNDO_TO_PURGE:
+	case TRX_UNDO_CACHED:
+		trx_id_t id = mach_read_from_8(TRX_UNDO_TRX_NO + undo_header);
+		if (id >> 48) {
+			sql_print_error("InnoDB: corrupted TRX_NO %llx", id);
+			goto corrupted;
+		}
+		if (id > max_trx_id) {
+			max_trx_id = id;
+		}
+	}
 
 	/* Read X/Open XA transaction identification if it exists, or
 	set it to NULL. */
@@ -1103,6 +1136,10 @@ trx_undo_mem_create_at_db_start(trx_rseg_t* rseg, ulint id, ulint page_no,
 	}
 
 	trx_id_t trx_id = mach_read_from_8(undo_header + TRX_UNDO_TRX_ID);
+	if (trx_id >> 48) {
+		sql_print_error("InnoDB: corrupted TRX_ID %llx", trx_id);
+		goto corrupted;
+	}
 	if (trx_id > max_trx_id) {
 		max_trx_id = trx_id;
 	}
@@ -1111,61 +1148,45 @@ trx_undo_mem_create_at_db_start(trx_rseg_t* rseg, ulint id, ulint page_no,
 	trx_undo_t* undo = trx_undo_mem_create(
 		rseg, id, trx_id, &xid, page_no, offset);
 	mutex_exit(&rseg->mutex);
+	if (!undo) {
+		return undo;
+	}
 
 	undo->dict_operation = undo_header[TRX_UNDO_DICT_TRANS];
 	undo->table_id = mach_read_from_8(undo_header + TRX_UNDO_TABLE_ID);
 	undo->size = flst_get_len(TRX_UNDO_SEG_HDR + TRX_UNDO_PAGE_LIST
 				  + undo_page);
 
-	if (UNIV_UNLIKELY(state == TRX_UNDO_TO_FREE)) {
-		/* This is an old-format insert_undo log segment that
-		is being freed. The page list is inconsistent. */
-		ut_ad(type == TRX_UNDO_INSERT);
-		state = TRX_UNDO_TO_PURGE;
-	} else {
-		if (state == TRX_UNDO_TO_PURGE
-		    || state == TRX_UNDO_CACHED) {
-			trx_id_t id = mach_read_from_8(TRX_UNDO_TRX_NO
-						       + undo_header);
-			if (id > max_trx_id) {
-				max_trx_id = id;
-			}
-		}
+	fil_addr_t	last_addr = flst_get_last(
+		TRX_UNDO_SEG_HDR + TRX_UNDO_PAGE_LIST + undo_page, &mtr);
 
-		fil_addr_t	last_addr = flst_get_last(
-			TRX_UNDO_SEG_HDR + TRX_UNDO_PAGE_LIST + undo_page,
-			&mtr);
-
-		undo->last_page_no = last_addr.page;
-		undo->top_page_no = last_addr.page;
+	undo->last_page_no = last_addr.page;
+	undo->top_page_no = last_addr.page;
 
-		page_t* last_page = trx_undo_page_get(
-			page_id_t(rseg->space->id, undo->last_page_no), &mtr);
+	page_t* last_page = trx_undo_page_get(
+		page_id_t(rseg->space->id, undo->last_page_no), &mtr);
 
-		if (const trx_undo_rec_t* rec = trx_undo_page_get_last_rec(
-			    last_page, page_no, offset)) {
-			undo->top_offset = ulint(rec - last_page);
-			undo->top_undo_no = trx_undo_rec_get_undo_no(rec);
-			ut_ad(!undo->empty());
-		} else {
-			undo->top_undo_no = IB_ID_MAX;
-			ut_ad(undo->empty());
-		}
+	if (const trx_undo_rec_t* rec = trx_undo_page_get_last_rec(
+		    last_page, page_no, offset)) {
+		undo->top_offset = ulint(rec - last_page);
+		undo->top_undo_no = trx_undo_rec_get_undo_no(rec);
+		ut_ad(!undo->empty());
+	} else {
+		undo->top_undo_no = IB_ID_MAX;
+		ut_ad(undo->empty());
 	}
 
 	undo->state = state;
 
 	if (state != TRX_UNDO_CACHED) {
-		UT_LIST_ADD_LAST(type == TRX_UNDO_INSERT
-				 ? rseg->old_insert_list
-				 : rseg->undo_list, undo);
+		UT_LIST_ADD_LAST(rseg->undo_list, undo);
 	} else {
 		UT_LIST_ADD_LAST(rseg->undo_cached, undo);
 		MONITOR_INC(MONITOR_NUM_UNDO_SLOT_CACHED);
 	}
 
 	mtr.commit();
-	return undo->size;
+	return undo;
 }
 
 /********************************************************************//**
@@ -1577,22 +1598,18 @@ trx_undo_set_state_at_prepare(
 	return(undo_page);
 }
 
-/** Free an old insert or temporary undo log after commit or rollback.
+/** Free temporary undo log after commit or rollback.
 The information is not needed after a commit or rollback, therefore
 the data can be discarded.
-@param[in,out]	undo	undo log
-@param[in]	is_temp	whether this is temporary undo log */
-void
-trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp)
+@param undo     temporary undo log */
+void trx_undo_commit_cleanup(trx_undo_t *undo)
 {
 	trx_rseg_t*	rseg	= undo->rseg;
-	ut_ad(is_temp == !rseg->is_persistent());
-	ut_ad(!is_temp || 0 == UT_LIST_GET_LEN(rseg->old_insert_list));
+	ut_ad(rseg->space == fil_system.temp_space);
 
 	mutex_enter(&rseg->mutex);
 
-	UT_LIST_REMOVE(is_temp ? rseg->undo_list : rseg->old_insert_list,
-		       undo);
+	UT_LIST_REMOVE(rseg->undo_list, undo);
 
 	if (undo->state == TRX_UNDO_CACHED) {
 		UT_LIST_ADD_FIRST(rseg->undo_cached, undo);
@@ -1602,7 +1619,7 @@ trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp)
 
 		/* Delete first the undo log segment in the file */
 		mutex_exit(&rseg->mutex);
-		trx_undo_seg_free(undo, is_temp);
+		trx_undo_seg_free(undo);
 		mutex_enter(&rseg->mutex);
 
 		ut_ad(rseg->curr_size > undo->size);
@@ -1615,15 +1632,13 @@ trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp)
 }
 
 /** At shutdown, frees the undo logs of a transaction. */
-void
-trx_undo_free_at_shutdown(trx_t *trx)
+void trx_undo_free_at_shutdown(trx_t *trx)
 {
 	if (trx_undo_t*& undo = trx->rsegs.m_redo.undo) {
 		switch (undo->state) {
 		case TRX_UNDO_PREPARED:
 			break;
 		case TRX_UNDO_CACHED:
-		case TRX_UNDO_TO_FREE:
 		case TRX_UNDO_TO_PURGE:
 			ut_ad(trx_state_eq(trx,
 					   TRX_STATE_COMMITTED_IN_MEMORY));
@@ -1644,34 +1659,6 @@ trx_undo_free_at_shutdown(trx_t *trx)
 		ut_free(undo);
 		undo = NULL;
 	}
-
-	if (trx_undo_t*& undo = trx->rsegs.m_redo.old_insert) {
-		switch (undo->state) {
-		case TRX_UNDO_PREPARED:
-			break;
-		case TRX_UNDO_CACHED:
-		case TRX_UNDO_TO_FREE:
-		case TRX_UNDO_TO_PURGE:
-			ut_ad(trx_state_eq(trx,
-					   TRX_STATE_COMMITTED_IN_MEMORY));
-			/* fall through */
-		case TRX_UNDO_ACTIVE:
-			/* trx_t::commit_state() assigns
-			trx->state = TRX_STATE_COMMITTED_IN_MEMORY. */
-			ut_a(!srv_was_started
-			     || srv_read_only_mode
-			     || srv_force_recovery >= SRV_FORCE_NO_TRX_UNDO
-			     || srv_fast_shutdown);
-			break;
-		default:
-			ut_error;
-		}
-
-		UT_LIST_REMOVE(trx->rsegs.m_redo.rseg->old_insert_list, undo);
-		ut_free(undo);
-		undo = NULL;
-	}
-
 	if (trx_undo_t*& undo = trx->rsegs.m_noredo.undo) {
 		ut_a(undo->state == TRX_UNDO_PREPARED);