1 files changed, 489 insertions, 159 deletions

diff --git a/storage/innobase/trx/trx0rseg.cc b/storage/innobase/trx/trx0rseg.cc
index e76695b7e43..6cdc704e5e2 100644
--- a/storage/innobase/trx/trx0rseg.cc
+++ b/storage/innobase/trx/trx0rseg.cc
@@ -33,28 +33,251 @@ Created 3/26/1996 Heikki Tuuri
 
 #include <algorithm>
 
-/** Creates a rollback segment header.
-This function is called only when a new rollback segment is created in
-the database.
-@param[in]	space		space id
-@param[in]	max_size	max size in pages
-@param[in]	rseg_slot_no	rseg id == slot number in trx sys
+#ifdef WITH_WSREP
+#include <mysql/service_wsrep.h>
+
+#ifdef UNIV_DEBUG
+/** The latest known WSREP XID sequence number */
+static long long wsrep_seqno = -1;
+#endif /* UNIV_DEBUG */
+/** The latest known WSREP XID UUID */
+static unsigned char wsrep_uuid[16];
+
+/** Update the WSREP XID information in rollback segment header.
+@param[in,out]	rseg_header	rollback segment header
+@param[in]	xid		WSREP XID
+@param[in,out]	mtr		mini-transaction */
+void
+trx_rseg_update_wsrep_checkpoint(
+	trx_rsegf_t*	rseg_header,
+	const XID*	xid,
+	mtr_t*		mtr)
+{
+	ut_ad(wsrep_is_wsrep_xid(xid));
+
+#ifdef UNIV_DEBUG
+	/* Check that seqno is monotonically increasing */
+	long long xid_seqno = wsrep_xid_seqno(xid);
+	const byte* xid_uuid = wsrep_xid_uuid(xid);
+
+	if (!memcmp(xid_uuid, wsrep_uuid, sizeof wsrep_uuid)) {
+		ut_ad(xid_seqno > wsrep_seqno);
+	} else {
+		memcpy(wsrep_uuid, xid_uuid, sizeof wsrep_uuid);
+	}
+	wsrep_seqno = xid_seqno;
+#endif /* UNIV_DEBUG */
+
+	mlog_write_ulint(TRX_RSEG_WSREP_XID_FORMAT + rseg_header,
+			 uint32_t(xid->formatID),
+			 MLOG_4BYTES, mtr);
+
+	mlog_write_ulint(TRX_RSEG_WSREP_XID_GTRID_LEN + rseg_header,
+			 uint32_t(xid->gtrid_length),
+			 MLOG_4BYTES, mtr);
+
+	mlog_write_ulint(TRX_RSEG_WSREP_XID_BQUAL_LEN + rseg_header,
+			 uint32_t(xid->bqual_length),
+			 MLOG_4BYTES, mtr);
+
+	mlog_write_string(TRX_RSEG_WSREP_XID_DATA + rseg_header,
+			  reinterpret_cast<const byte*>(xid->data),
+			  XIDDATASIZE, mtr);
+}
+
+/** Update WSREP checkpoint XID in first rollback segment header
+as part of wsrep_set_SE_checkpoint() when it is guaranteed that there
+are no wsrep transactions committing.
+If the UUID part of the WSREP XID does not match to the UUIDs of XIDs already
+stored into rollback segments, the WSREP XID in all the remaining rollback
+segments will be reset.
+@param[in]	xid		WSREP XID */
+void trx_rseg_update_wsrep_checkpoint(const XID* xid)
+{
+	mtr_t	mtr;
+	mtr.start();
+
+	const trx_rseg_t* rseg = trx_sys.rseg_array[0];
+
+	trx_rsegf_t* rseg_header = trx_rsegf_get(rseg->space, rseg->page_no,
+						 &mtr);
+	if (UNIV_UNLIKELY(mach_read_from_4(rseg_header + TRX_RSEG_FORMAT))) {
+		trx_rseg_format_upgrade(rseg_header, &mtr);
+	}
+
+	trx_rseg_update_wsrep_checkpoint(rseg_header, xid, &mtr);
+
+	const byte* xid_uuid = wsrep_xid_uuid(xid);
+	if (memcmp(wsrep_uuid, xid_uuid, sizeof wsrep_uuid)) {
+		memcpy(wsrep_uuid, xid_uuid, sizeof wsrep_uuid);
+
+		/* Because the UUID part of the WSREP XID differed
+		from current_xid_uuid, the WSREP group UUID was
+		changed, and we must reset the XID in all rollback
+		segment headers. */
+		for (ulint rseg_id = 1; rseg_id < TRX_SYS_N_RSEGS; ++rseg_id) {
+			if (const trx_rseg_t* rseg =
+			    trx_sys.rseg_array[rseg_id]) {
+				trx_rseg_update_wsrep_checkpoint(
+					trx_rsegf_get(rseg->space,
+						      rseg->page_no, &mtr),
+					xid, &mtr);
+			}
+		}
+	}
+
+	mtr.commit();
+}
+
+/** Read the WSREP XID information in rollback segment header.
+@param[in]	rseg_header	Rollback segment header
+@param[out]	xid		Transaction XID
+@return	whether the WSREP XID was present */
+static
+bool trx_rseg_read_wsrep_checkpoint(const trx_rsegf_t* rseg_header, XID& xid)
+{
+	int formatID = static_cast<int>(
+		mach_read_from_4(
+			TRX_RSEG_WSREP_XID_FORMAT + rseg_header));
+	if (formatID == 0) {
+		return false;
+	}
+
+	xid.formatID = formatID;
+	xid.gtrid_length = static_cast<int>(
+		mach_read_from_4(
+			TRX_RSEG_WSREP_XID_GTRID_LEN + rseg_header));
+
+	xid.bqual_length = static_cast<int>(
+		mach_read_from_4(
+			TRX_RSEG_WSREP_XID_BQUAL_LEN + rseg_header));
+
+	memcpy(xid.data, TRX_RSEG_WSREP_XID_DATA + rseg_header, XIDDATASIZE);
+
+	return true;
+}
+
+/** Read the WSREP XID from the TRX_SYS page (in case of upgrade).
+@param[in]	page	TRX_SYS page
+@param[out]	xid	WSREP XID (if present)
+@return	whether the WSREP XID is present */
+static bool trx_rseg_init_wsrep_xid(const page_t* page, XID& xid)
+{
+	if (mach_read_from_4(TRX_SYS + TRX_SYS_WSREP_XID_INFO
+			     + TRX_SYS_WSREP_XID_MAGIC_N_FLD
+			     + page)
+	    != TRX_SYS_WSREP_XID_MAGIC_N) {
+		return false;
+	}
+
+	xid.formatID = static_cast<int>(
+		mach_read_from_4(
+			TRX_SYS + TRX_SYS_WSREP_XID_INFO
+			+ TRX_SYS_WSREP_XID_FORMAT + page));
+	xid.gtrid_length = static_cast<int>(
+		mach_read_from_4(
+			TRX_SYS + TRX_SYS_WSREP_XID_INFO
+			+ TRX_SYS_WSREP_XID_GTRID_LEN + page));
+	xid.bqual_length = static_cast<int>(
+		mach_read_from_4(
+			TRX_SYS + TRX_SYS_WSREP_XID_INFO
+			+ TRX_SYS_WSREP_XID_BQUAL_LEN + page));
+	memcpy(xid.data,
+	       TRX_SYS + TRX_SYS_WSREP_XID_INFO
+	       + TRX_SYS_WSREP_XID_DATA + page, XIDDATASIZE);
+	return true;
+}
+
+/** Recover the latest WSREP checkpoint XID.
+@param[out]	xid	WSREP XID
+@return	whether the WSREP XID was found */
+bool trx_rseg_read_wsrep_checkpoint(XID& xid)
+{
+	mtr_t		mtr;
+	long long       max_xid_seqno = -1;
+	bool		found = false;
+
+	for (ulint rseg_id = 0; rseg_id < TRX_SYS_N_RSEGS;
+	     rseg_id++, mtr.commit()) {
+		mtr.start();
+		const buf_block_t* sys = trx_sysf_get(&mtr, false);
+		if (rseg_id == 0) {
+			found = trx_rseg_init_wsrep_xid(sys->frame, xid);
+			ut_ad(!found || xid.formatID == 1);
+			if (found) {
+				max_xid_seqno = wsrep_xid_seqno(&xid);
+				memcpy(wsrep_uuid, wsrep_xid_uuid(&xid),
+				       sizeof wsrep_uuid);
+			}
+		}
+
+		const uint32_t page_no = trx_sysf_rseg_get_page_no(
+			sys, rseg_id);
+
+		if (page_no == FIL_NULL) {
+			continue;
+		}
+
+		const trx_rsegf_t* rseg_header = trx_rsegf_get_new(
+			trx_sysf_rseg_get_space(sys, rseg_id), page_no, &mtr);
+
+		if (mach_read_from_4(rseg_header + TRX_RSEG_FORMAT)) {
+			continue;
+		}
+
+		XID tmp_xid;
+		long long tmp_seqno = 0;
+		if (trx_rseg_read_wsrep_checkpoint(rseg_header, tmp_xid)
+		    && (tmp_seqno = wsrep_xid_seqno(&tmp_xid))
+		    > max_xid_seqno) {
+			found = true;
+			max_xid_seqno = tmp_seqno;
+			xid = tmp_xid;
+			memcpy(wsrep_uuid, wsrep_xid_uuid(&tmp_xid),
+			       sizeof wsrep_uuid);
+		}
+	}
+
+	return found;
+}
+#endif /* WITH_WSREP */
+
+/** Upgrade a rollback segment header page to MariaDB 10.3 format.
+@param[in,out]	rseg_header	rollback segment header page
+@param[in,out]	mtr		mini-transaction */
+void trx_rseg_format_upgrade(trx_rsegf_t* rseg_header, mtr_t* mtr)
+{
+	ut_ad(page_offset(rseg_header) == TRX_RSEG);
+	byte* rseg_format = TRX_RSEG_FORMAT + rseg_header;
+	mlog_write_ulint(rseg_format, 0, MLOG_4BYTES, mtr);
+	/* Clear also possible garbage at the end of the page. Old
+	InnoDB versions did not initialize unused parts of pages. */
+	byte* b = rseg_header + TRX_RSEG_MAX_TRX_ID + 8;
+	ulint len = srv_page_size
+		- (FIL_PAGE_DATA_END
+		   + TRX_RSEG + TRX_RSEG_MAX_TRX_ID + 8);
+	memset(b, 0, len);
+	mlog_log_string(b, len, mtr);
+}
+
+/** Create a rollback segment header.
+@param[in,out]	space		system, undo, or temporary tablespace
+@param[in]	rseg_id		rollback segment identifier
+@param[in,out]	sys_header	the TRX_SYS page (NULL for temporary rseg)
 @param[in,out]	mtr		mini-transaction
 @return the created rollback segment
 @retval	NULL	on failure */
 buf_block_t*
 trx_rseg_header_create(
-	ulint			space,
-	ulint			max_size,
-	ulint			rseg_slot_no,
-	mtr_t*			mtr)
+	fil_space_t*	space,
+	ulint		rseg_id,
+	buf_block_t*	sys_header,
+	mtr_t*		mtr)
 {
-	trx_sysf_t*	sys_header;
 	buf_block_t*	block;
 
-	ut_ad(mtr);
-	ut_ad(mtr_memo_contains(mtr, fil_space_get_latch(space, NULL),
-				MTR_MEMO_X_LOCK));
+	ut_ad(mtr_memo_contains(mtr, &space->latch, MTR_MEMO_X_LOCK));
+	ut_ad(!sys_header == (space == fil_system.temp_space));
 
 	/* Allocate a new file segment for the rollback segment */
 	block = fseg_create(space, 0, TRX_RSEG + TRX_RSEG_FSEG_HEADER, mtr);
@@ -66,9 +289,8 @@ trx_rseg_header_create(
 
 	buf_block_dbg_add_level(block, SYNC_RSEG_HEADER_NEW);
 
-	/* Initialize max size field */
-	mlog_write_ulint(TRX_RSEG + TRX_RSEG_MAX_SIZE + block->frame,
-			 max_size, MLOG_4BYTES, mtr);
+	mlog_write_ulint(TRX_RSEG + TRX_RSEG_FORMAT + block->frame, 0,
+			 MLOG_4BYTES, mtr);
 
 	/* Initialize the history list */
 
@@ -84,17 +306,20 @@ trx_rseg_header_create(
 		trx_rsegf_set_nth_undo(rsegf, i, FIL_NULL, mtr);
 	}
 
-	if (space != SRV_TMP_SPACE_ID) {
+	if (sys_header) {
 		/* Add the rollback segment info to the free slot in
 		the trx system header */
 
-		sys_header = trx_sysf_get(mtr);
-
-		trx_sysf_rseg_set_space(sys_header, rseg_slot_no, space, mtr);
-
-		trx_sysf_rseg_set_page_no(
-			sys_header, rseg_slot_no,
-			block->page.id.page_no(), mtr);
+		mlog_write_ulint(TRX_SYS + TRX_SYS_RSEGS
+				 + TRX_SYS_RSEG_SPACE
+				 + rseg_id * TRX_SYS_RSEG_SLOT_SIZE
+				 + sys_header->frame,
+				 space->id, MLOG_4BYTES, mtr);
+		mlog_write_ulint(TRX_SYS + TRX_SYS_RSEGS
+				 + TRX_SYS_RSEG_PAGE_NO
+				 + rseg_id * TRX_SYS_RSEG_SLOT_SIZE
+				 + sys_header->frame,
+				 block->page.id.page_no(), MLOG_4BYTES, mtr);
 	}
 
 	return block;
@@ -110,33 +335,20 @@ trx_rseg_mem_free(trx_rseg_t* rseg)
 	mutex_free(&rseg->mutex);
 
 	/* There can't be any active transactions. */
-	ut_a(UT_LIST_GET_LEN(rseg->update_undo_list) == 0);
-	ut_a(UT_LIST_GET_LEN(rseg->insert_undo_list) == 0);
+	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->update_undo_cached);
+	for (undo = UT_LIST_GET_FIRST(rseg->undo_cached);
 	     undo != NULL;
 	     undo = next_undo) {
 
 		next_undo = UT_LIST_GET_NEXT(undo_list, undo);
 
-		UT_LIST_REMOVE(rseg->update_undo_cached, undo);
+		UT_LIST_REMOVE(rseg->undo_cached, undo);
 
 		MONITOR_DEC(MONITOR_NUM_UNDO_SLOT_CACHED);
 
-		trx_undo_mem_free(undo);
-	}
-
-	for (undo = UT_LIST_GET_FIRST(rseg->insert_undo_cached);
-	     undo != NULL;
-	     undo = next_undo) {
-
-		next_undo = UT_LIST_GET_NEXT(undo_list, undo);
-
-		UT_LIST_REMOVE(rseg->insert_undo_cached, undo);
-
-		MONITOR_DEC(MONITOR_NUM_UNDO_SLOT_CACHED);
-
-		trx_undo_mem_free(undo);
+		ut_free(undo);
 	}
 
 	ut_free(rseg);
@@ -148,7 +360,7 @@ trx_rseg_mem_free(trx_rseg_t* rseg)
 @param[in]	page_no		page number of the segment header */
 static
 trx_rseg_t*
-trx_rseg_mem_create(ulint id, ulint space, ulint page_no)
+trx_rseg_mem_create(ulint id, fil_space_t* space, ulint page_no)
 {
 	trx_rseg_t* rseg = static_cast<trx_rseg_t*>(
 		ut_zalloc_nokey(sizeof *rseg));
@@ -157,104 +369,215 @@ trx_rseg_mem_create(ulint id, ulint space, ulint page_no)
 	rseg->space = space;
 	rseg->page_no = page_no;
 	rseg->last_page_no = FIL_NULL;
+	rseg->curr_size = 1;
 
 	mutex_create(rseg->is_persistent()
 		     ? LATCH_ID_REDO_RSEG : LATCH_ID_NOREDO_RSEG,
 		     &rseg->mutex);
 
-	UT_LIST_INIT(rseg->update_undo_list, &trx_undo_t::undo_list);
-	UT_LIST_INIT(rseg->update_undo_cached, &trx_undo_t::undo_list);
-	UT_LIST_INIT(rseg->insert_undo_list, &trx_undo_t::undo_list);
-	UT_LIST_INIT(rseg->insert_undo_cached, &trx_undo_t::undo_list);
+	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)
+{
+	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);
+}
+
 /** Restore the state of a persistent rollback segment.
-@param[in,out]	rseg	persistent rollback segment
-@param[in,out]	mtr	mini-transaction */
+@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, mtr_t* mtr)
+trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
 {
-	ulint		len;
-	fil_addr_t	node_addr;
-	trx_rsegf_t*	rseg_header;
-	trx_ulogf_t*	undo_log_hdr;
-	ulint		sum_of_undo_sizes;
+	trx_rsegf_t*	rseg_header = trx_rsegf_get_new(
+		rseg->space->id, rseg->page_no, mtr);
 
-	rseg_header = trx_rsegf_get_new(rseg->space, rseg->page_no, mtr);
+	if (mach_read_from_4(rseg_header + TRX_RSEG_FORMAT) == 0) {
+		trx_id_t id = mach_read_from_8(rseg_header
+					       + TRX_RSEG_MAX_TRX_ID);
 
-	rseg->max_size = mtr_read_ulint(
-		rseg_header + TRX_RSEG_MAX_SIZE, MLOG_4BYTES, mtr);
+		if (id > max_trx_id) {
+			max_trx_id = id;
+		}
 
-	/* Initialize the undo log lists according to the rseg header */
+		if (rseg_header[TRX_RSEG_BINLOG_NAME]) {
+			const char* binlog_name = reinterpret_cast<const char*>
+				(rseg_header) + TRX_RSEG_BINLOG_NAME;
+			compile_time_assert(TRX_RSEG_BINLOG_NAME_LEN == sizeof
+					    trx_sys.recovered_binlog_filename);
+
+			int cmp = *trx_sys.recovered_binlog_filename
+				? strncmp(binlog_name,
+					  trx_sys.recovered_binlog_filename,
+					  TRX_RSEG_BINLOG_NAME_LEN)
+				: 1;
+
+			if (cmp >= 0) {
+				uint64_t binlog_offset = mach_read_from_8(
+					rseg_header + TRX_RSEG_BINLOG_OFFSET);
+				if (cmp) {
+					memcpy(trx_sys.
+					       recovered_binlog_filename,
+					       binlog_name,
+					       TRX_RSEG_BINLOG_NAME_LEN);
+					trx_sys.recovered_binlog_offset
+						= binlog_offset;
+				} else if (binlog_offset >
+					   trx_sys.recovered_binlog_offset) {
+					trx_sys.recovered_binlog_offset
+						= binlog_offset;
+				}
+			}
+
+#ifdef WITH_WSREP
+			trx_rseg_read_wsrep_checkpoint(
+				rseg_header, trx_sys.recovered_wsrep_xid);
+#endif
+		}
+	}
 
-	sum_of_undo_sizes = trx_undo_lists_init(rseg);
+	if (srv_operation == SRV_OPERATION_RESTORE) {
+		/* mariabackup --prepare only deals with
+		the redo log and the data files, not with
+		transactions or the data dictionary. */
+		return;
+	}
 
-	rseg->curr_size = mtr_read_ulint(
-		rseg_header + TRX_RSEG_HISTORY_SIZE, MLOG_4BYTES, mtr)
-		+ 1 + sum_of_undo_sizes;
+	/* Initialize the undo log lists according to the rseg header */
 
-	len = flst_get_len(rseg_header + TRX_RSEG_HISTORY);
+	rseg->curr_size = mach_read_from_4(rseg_header + TRX_RSEG_HISTORY_SIZE)
+		+ 1 + trx_undo_lists_init(rseg, max_trx_id, rseg_header);
 
-	if (len > 0) {
-		my_atomic_addlint(&trx_sys->rseg_history_len, len);
+	if (ulint len = flst_get_len(rseg_header + TRX_RSEG_HISTORY)) {
+		trx_sys.history_add(int32(len));
 
-		node_addr = trx_purge_get_log_from_hist(
+		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;
 
-		undo_log_hdr = trx_undo_page_get(
-			page_id_t(rseg->space, node_addr.page), mtr)
+		const trx_ulogf_t*	undo_log_hdr = trx_undo_page_get(
+			page_id_t(rseg->space->id, node_addr.page), mtr)
 			+ node_addr.boffset;
 
-		rseg->last_trx_no = mach_read_from_8(
-			undo_log_hdr + TRX_UNDO_TRX_NO);
-
-		rseg->last_del_marks = mtr_read_ulint(
-			undo_log_hdr + TRX_UNDO_DEL_MARKS, MLOG_2BYTES, mtr);
-
-		TrxUndoRsegs elem(rseg->last_trx_no);
-		elem.push_back(rseg);
+		trx_id_t id = mach_read_from_8(undo_log_hdr + TRX_UNDO_TRX_ID);
+		if (id > max_trx_id) {
+			max_trx_id = id;
+		}
+		id = mach_read_from_8(undo_log_hdr + TRX_UNDO_TRX_NO);
+		if (id > max_trx_id) {
+			max_trx_id = 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) {
 
 			/* There is no need to cover this operation by the purge
 			mutex because we are still bootstrapping. */
-
-			purge_sys->purge_queue.push(elem);
+			purge_sys.purge_queue.push(*rseg);
 		}
 	}
 }
 
+/** Read binlog metadata from the TRX_SYS page, in case we are upgrading
+from MySQL or a MariaDB version older than 10.3.5. */
+static void trx_rseg_init_binlog_info(const page_t* page)
+{
+	if (mach_read_from_4(TRX_SYS + TRX_SYS_MYSQL_LOG_INFO
+			     + TRX_SYS_MYSQL_LOG_MAGIC_N_FLD
+			     + page)
+	    == TRX_SYS_MYSQL_LOG_MAGIC_N) {
+		memcpy(trx_sys.recovered_binlog_filename,
+		       TRX_SYS_MYSQL_LOG_INFO + TRX_SYS_MYSQL_LOG_NAME
+		       + TRX_SYS + page, TRX_SYS_MYSQL_LOG_NAME_LEN);
+		trx_sys.recovered_binlog_offset = mach_read_from_8(
+			TRX_SYS_MYSQL_LOG_INFO + TRX_SYS_MYSQL_LOG_OFFSET
+			+ TRX_SYS + page);
+	}
+
+#ifdef WITH_WSREP
+	trx_rseg_init_wsrep_xid(page, trx_sys.recovered_wsrep_xid);
+#endif
+}
+
 /** Initialize the rollback segments in memory at database startup. */
 void
 trx_rseg_array_init()
 {
-	mtr_t	mtr;
+	trx_id_t max_trx_id = 0;
 
-	for (ulint i = 0; i < TRX_SYS_N_RSEGS; i++) {
+	*trx_sys.recovered_binlog_filename = '\0';
+	trx_sys.recovered_binlog_offset = 0;
+#ifdef WITH_WSREP
+	trx_sys.recovered_wsrep_xid.null();
+#endif
+
+	for (ulint rseg_id = 0; rseg_id < TRX_SYS_N_RSEGS; rseg_id++) {
+		mtr_t mtr;
 		mtr.start();
-		trx_sysf_t*	sys_header = trx_sysf_get(&mtr);
-		ulint		page_no = trx_sysf_rseg_get_page_no(
-			sys_header, i, &mtr);
+		if (const buf_block_t* sys = trx_sysf_get(&mtr, false)) {
+			if (rseg_id == 0) {
+				/* In case this is an upgrade from
+				before MariaDB 10.3.5, fetch the base
+				information from the TRX_SYS page. */
+				max_trx_id = mach_read_from_8(
+					TRX_SYS + TRX_SYS_TRX_ID_STORE
+					+ sys->frame);
+				trx_rseg_init_binlog_info(sys->frame);
+			}
 
-		if (page_no != FIL_NULL) {
-			trx_rseg_t* rseg = trx_rseg_mem_create(
-				i,
-				trx_sysf_rseg_get_space(sys_header, i, &mtr),
-				page_no);
-			ut_ad(rseg->is_persistent());
-			ut_ad(!trx_sys->rseg_array[rseg->id]);
-			trx_sys->rseg_array[rseg->id] = rseg;
-			trx_rseg_mem_restore(rseg, &mtr);
+			const uint32_t	page_no = trx_sysf_rseg_get_page_no(
+				sys, rseg_id);
+			if (page_no != FIL_NULL) {
+				trx_rseg_t* rseg = trx_rseg_mem_create(
+					rseg_id,
+					fil_space_get(trx_sysf_rseg_get_space(
+							      sys, rseg_id)),
+					page_no);
+				ut_ad(rseg->is_persistent());
+				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);
+			}
 		}
 
 		mtr.commit();
 	}
+
+	trx_sys.init_max_trx_id(max_trx_id + 1);
 }
 
 /** Create a persistent rollback segment.
@@ -270,29 +593,25 @@ trx_rseg_create(ulint space_id)
 	mtr.start();
 
 	/* To obey the latching order, acquire the file space
-	x-latch before the trx_sys->mutex. */
-#ifdef UNIV_DEBUG
-	const fil_space_t*	space =
-#endif /* UNIV_DEBUG */
-		mtr_x_lock_space(space_id, &mtr);
+	x-latch before the trx_sys.mutex. */
+	fil_space_t*	space = mtr_x_lock_space(space_id, &mtr);
 	ut_ad(space->purpose == FIL_TYPE_TABLESPACE);
 
-	ulint	slot_no = trx_sysf_rseg_find_free(&mtr);
-	if (buf_block_t* block = slot_no == ULINT_UNDEFINED
-	    ? NULL
-	    : trx_rseg_header_create(space_id, ULINT_MAX, slot_no, &mtr)) {
-		trx_sysf_t*	sys_header = trx_sysf_get(&mtr);
-
-		ulint		id = trx_sysf_rseg_get_space(
-			sys_header, slot_no, &mtr);
-		ut_a(id == space_id);
-
-		rseg = trx_rseg_mem_create(slot_no, space_id,
-					   block->page.id.page_no());
-		ut_ad(rseg->is_persistent());
-		ut_ad(!trx_sys->rseg_array[rseg->id]);
-		trx_sys->rseg_array[rseg->id] = rseg;
-		trx_rseg_mem_restore(rseg, &mtr);
+	if (buf_block_t* sys_header = trx_sysf_get(&mtr)) {
+		ulint	rseg_id = trx_sys_rseg_find_free(sys_header);
+		if (buf_block_t* rblock = rseg_id == ULINT_UNDEFINED
+		    ? NULL
+		    : trx_rseg_header_create(space, rseg_id, sys_header,
+					     &mtr)) {
+			ut_ad(trx_sysf_rseg_get_space(sys_header, rseg_id)
+			      == space_id);
+			rseg = trx_rseg_mem_create(rseg_id, space,
+						   rblock->page.id.page_no());
+			ut_ad(rseg->id == rseg_id);
+			ut_ad(rseg->is_persistent());
+			ut_ad(!trx_sys.rseg_array[rseg->id]);
+			trx_sys.rseg_array[rseg->id] = rseg;
+		}
 	}
 
 	mtr.commit();
@@ -309,20 +628,15 @@ trx_temp_rseg_create()
 	for (ulong i = 0; i < TRX_SYS_N_RSEGS; i++) {
 		mtr.start();
 		mtr.set_log_mode(MTR_LOG_NO_REDO);
-#ifdef UNIV_DEBUG
-		const fil_space_t*	space =
-#endif /* UNIV_DEBUG */
-			mtr_x_lock_space(SRV_TMP_SPACE_ID, &mtr);
-		ut_ad(space->purpose == FIL_TYPE_TEMPORARY);
+		mtr_x_lock(&fil_system.temp_space->latch, &mtr);
 
-		buf_block_t* block = trx_rseg_header_create(
-			SRV_TMP_SPACE_ID, ULINT_MAX, i, &mtr);
+		buf_block_t* rblock = trx_rseg_header_create(
+			fil_system.temp_space, i, NULL, &mtr);
 		trx_rseg_t* rseg = trx_rseg_mem_create(
-			i, SRV_TMP_SPACE_ID, block->page.id.page_no());
+			i, fil_system.temp_space, rblock->page.id.page_no());
 		ut_ad(!rseg->is_persistent());
-		ut_ad(!trx_sys->temp_rsegs[i]);
-		trx_sys->temp_rsegs[i] = rseg;
-		trx_rseg_mem_restore(rseg, &mtr);
+		ut_ad(!trx_sys.temp_rsegs[i]);
+		trx_sys.temp_rsegs[i] = rseg;
 		mtr.commit();
 	}
 }
@@ -339,54 +653,70 @@ trx_rseg_get_n_undo_tablespaces(
 	ulint*		space_ids)	/*!< out: array of space ids of
 					UNDO tablespaces */
 {
-	ulint		i;
-	mtr_t		mtr;
-	trx_sysf_t*	sys_header;
-	ulint		n_undo_tablespaces = 0;
-
-	mtr_start(&mtr);
+	mtr_t mtr;
+	mtr.start();
 
-	sys_header = trx_sysf_get(&mtr);
+	buf_block_t* sys_header = trx_sysf_get(&mtr, false);
+	if (!sys_header) {
+		mtr.commit();
+		return 0;
+	}
 
-	for (i = 0; i < TRX_SYS_N_RSEGS; i++) {
-		ulint	page_no;
-		ulint	space;
+	ulint* end = space_ids;
 
-		page_no = trx_sysf_rseg_get_page_no(sys_header, i, &mtr);
+	for (ulint rseg_id = 0; rseg_id < TRX_SYS_N_RSEGS; rseg_id++) {
+		uint32_t page_no = trx_sysf_rseg_get_page_no(sys_header,
+							     rseg_id);
 
 		if (page_no == FIL_NULL) {
 			continue;
 		}
 
-		space = trx_sysf_rseg_get_space(sys_header, i, &mtr);
-
-		if (space != 0) {
-			ulint	j;
-			ibool	found = FALSE;
-
-			for (j = 0; j < n_undo_tablespaces; ++j) {
-				if (space_ids[j] == space) {
-					found = TRUE;
-					break;
-				}
-			}
-
-			if (!found) {
-				ut_a(n_undo_tablespaces <= i);
-				space_ids[n_undo_tablespaces++] = space;
+		if (ulint space = trx_sysf_rseg_get_space(sys_header,
+							  rseg_id)) {
+			if (std::find(space_ids, end, space) == end) {
+				*end++ = space;
 			}
 		}
 	}
 
-	mtr_commit(&mtr);
+	mtr.commit();
+
+	ut_a(end - space_ids <= TRX_SYS_N_RSEGS);
+	*end = ULINT_UNDEFINED;
+
+	std::sort(space_ids, end);
 
-	ut_a(n_undo_tablespaces <= TRX_SYS_N_RSEGS);
+	return ulint(end - space_ids);
+}
 
-	space_ids[n_undo_tablespaces] = ULINT_UNDEFINED;
+/** Update the offset information about the end of the binlog entry
+which corresponds to the transaction just being committed.
+In a replication slave, this updates the master binlog position
+up to which replication has proceeded.
+@param[in,out]	rseg_header	rollback segment header
+@param[in]	trx		committing transaction
+@param[in,out]	mtr		mini-transaction */
+void
+trx_rseg_update_binlog_offset(byte* rseg_header, const trx_t* trx, mtr_t* mtr)
+{
+	DBUG_LOG("trx", "trx_mysql_binlog_offset: " << trx->mysql_log_offset);
 
-	if (n_undo_tablespaces > 0) {
-		std::sort(space_ids, space_ids + n_undo_tablespaces);
+	const size_t len = strlen(trx->mysql_log_file_name) + 1;
+
+	ut_ad(len > 1);
+
+	if (UNIV_UNLIKELY(len > TRX_RSEG_BINLOG_NAME_LEN)) {
+		return;
 	}
 
-	return(n_undo_tablespaces);
+	mlog_write_ull(rseg_header + TRX_RSEG_BINLOG_OFFSET,
+		       trx->mysql_log_offset, mtr);
+	byte* p = rseg_header + TRX_RSEG_BINLOG_NAME;
+	const byte* binlog_name = reinterpret_cast<const byte*>
+		(trx->mysql_log_file_name);
+
+	if (memcmp(binlog_name, p, len)) {
+		mlog_write_string(p, binlog_name, len, mtr);
+	}
 }