From dfde6ce004c8bfe165a0a7283887f0715927e8c2 Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilipkumar@localhost.localdomain>
Date: Tue, 16 Apr 2019 16:46:01 +0530
Subject: [PATCH 1/3] Enhance multi-log handling in undo log

As part of this patch UndoLogAllocate will return the current transaction
start header in previous log if there is a log switch during transaction.
This will allow discard worker to fetch the last undo record pointer for
an aborted transaction.  WAL log the undo log switch so that during
recovery we can detect the log switch.

Dilip Kumar
---
 src/backend/access/undo/undolog.c | 76 ++++++++++++++++++++++++++++++++++++++-
 src/include/access/undolog.h      | 19 +++++++++-
 src/include/access/undolog_xlog.h |  9 +++++
 3 files changed, 102 insertions(+), 2 deletions(-)

diff --git a/src/backend/access/undo/undolog.c b/src/backend/access/undo/undolog.c
index 38f8a4f..028b282 100644
--- a/src/backend/access/undo/undolog.c
+++ b/src/backend/access/undo/undolog.c
@@ -650,7 +650,9 @@ UndoLogAllocate(uint16 size,
 				UndoRecPtr try_location,
 				UndoPersistence persistence,
 				bool *need_xact_header,
-				UndoRecPtr *last_xact_start)
+				UndoRecPtr *last_xact_start,
+				UndoRecPtr *prevlog_xact_start,
+				UndoRecPtr *prevlog_insert_urp)
 {
 	UndoLogControl *log = MyUndoLogState.logs[persistence];
 	UndoLogOffset new_insert;
@@ -755,6 +757,11 @@ UndoLogAllocate(uint16 size,
 				 * comments in undorecord.c file header.
 				 */
 				prevlogno = log->logno;
+				*prevlog_xact_start =
+					MakeUndoRecPtr(log->logno,
+								   log->meta.unlogged.this_xact_start);
+				*prevlog_insert_urp =
+					MakeUndoRecPtr(log->logno, log->meta.unlogged.insert);
 			}
 			elog(LOG, "undo log %u is full, switching to a new one", log->logno);
 			log = NULL;
@@ -831,6 +838,8 @@ UndoLogAllocateInRecovery(TransactionId xid,
 						  UndoRecPtr try_location,
 						  bool *need_xact_header,
 						  UndoRecPtr *last_xact_start,
+						  UndoRecPtr *prevlog_xact_start,
+						  UndoRecPtr *prevlog_last_urp,
 						  XLogReaderState *xlog_record)
 {
 	UndoLogControl *log;
@@ -959,6 +968,13 @@ UndoLogAllocateInRecovery(TransactionId xid,
 			*last_xact_start = log->meta.unlogged.last_xact_start;
 			allocate_in_recovery_logno = log->logno;
 
+			/* Read log switch information from meta and reset it. */
+			*prevlog_xact_start = log->meta.unlogged.prevlog_xact_start;
+			*prevlog_last_urp = log->meta.unlogged.prevlog_last_urp;
+
+			log->meta.unlogged.prevlog_xact_start = InvalidUndoRecPtr;
+			log->meta.unlogged.prevlog_last_urp = InvalidUndoRecPtr;
+
 			return MakeUndoRecPtr(log->logno, log->meta.unlogged.insert);
 		}
 		++allocate_in_recovery_block_id;
@@ -1221,6 +1237,43 @@ UndoLogGetFirstValidRecord(UndoLogControl *log, bool *full)
 }
 
 /*
+ * UndoLogSwitchSetPrevLogInfo - Store previous log info on the log switch and
+ * wal log the same.
+ */
+void
+UndoLogSwitchSetPrevLogInfo(UndoLogNumber logno, UndoRecPtr prevlog_xact_start,
+							UndoRecPtr prevlog_last_urp)
+{
+	UndoLogControl *log;
+
+	log = get_undo_log(logno, false);
+
+	/*
+	 * Either we're in recovery, or is a log we are currently attached to, or
+	 * recently detached from because it was full.
+	 */
+	Assert(AmAttachedToUndoLog(log));
+
+	LWLockAcquire(&log->mutex, LW_EXCLUSIVE);
+	log->meta.unlogged.prevlog_xact_start = prevlog_last_urp;
+	log->meta.unlogged.prevlog_last_urp = prevlog_last_urp;
+	LWLockRelease(&log->mutex);
+
+	/* Wal log the log switch. */
+	{
+		xl_undolog_switch xlrec;
+
+		xlrec.logno = logno;
+		xlrec.prevlog_xact_start = prevlog_last_urp;
+		xlrec.prevlog_last_urp = prevlog_xact_start;
+
+		XLogBeginInsert();
+		XLogRegisterData((char *) &xlrec, sizeof(xlrec));
+		XLogInsert(RM_UNDOLOG_ID, XLOG_UNDOLOG_SWITCH);
+	}
+}
+
+/*
  * Return the Next insert location.  This will also validate the input xid
  * if latest insert point is not for the same transaction id then this will
  * return Invalid Undo pointer.
@@ -2552,6 +2605,25 @@ undolog_xlog_rewind(XLogReaderState *record)
 	log->meta.unlogged.prevlen = xlrec->prevlen;
 }
 
+/*
+ * replay the switch of a undo log
+ */
+static void
+undolog_xlog_switch(XLogReaderState *record)
+{
+	xl_undolog_switch *xlrec = (xl_undolog_switch *) XLogRecGetData(record);
+	UndoLogControl *log;
+
+	log = get_undo_log(xlrec->logno, false);
+
+	/*
+	 * Restore the log switch information in the MyUndoLogState this will be
+	 * reset by following UndoLogAllocateDuringRecovery.
+	 */
+	log->meta.unlogged.prevlog_xact_start = xlrec->prevlog_xact_start;
+	log->meta.unlogged.prevlog_last_urp = xlrec->prevlog_last_urp;
+}
+
 void
 undolog_redo(XLogReaderState *record)
 {
@@ -2571,6 +2643,8 @@ undolog_redo(XLogReaderState *record)
 		case XLOG_UNDOLOG_REWIND:
 			undolog_xlog_rewind(record);
 			break;
+		case XLOG_UNDOLOG_SWITCH:
+			undolog_xlog_switch(record);
 		default:
 			elog(PANIC, "undo_redo: unknown op code %u", info);
 	}
diff --git a/src/include/access/undolog.h b/src/include/access/undolog.h
index c1b3a75..2cfce8c 100644
--- a/src/include/access/undolog.h
+++ b/src/include/access/undolog.h
@@ -205,6 +205,16 @@ typedef struct UndoLogUnloggedMetaData
 	uint16		prevlen;		   	/* size of the last record in the log */
 	UndoLogNumber prevlogno;		/* Previous undo log number */
 	TransactionId xid;				/* currently attached/writing xid */
+
+	/*
+	 * Below two variable are used during recovery when transaction's undo
+	 * records are split across undo logs.  Replay of undo do switch wal will
+	 * restore these two undo record pointers which will be reset on next
+	 * allocation during recovery. */
+	UndoRecPtr	prevlog_xact_start; /* Transaction's start undo record pointer
+									 * in the previous log. */
+	UndoRecPtr	prevlog_last_urp;	/* Transaction's last undo record pointer in
+									 * the previous log. */
 } UndoLogUnloggedMetaData;
 
 /*
@@ -343,12 +353,16 @@ extern UndoRecPtr UndoLogAllocate(uint16 size,
 								  UndoRecPtr try_location,
 								  UndoPersistence level,
 								  bool *need_xact_header,
-								  UndoRecPtr *last_xact_start);
+								  UndoRecPtr *last_xact_start,
+								  UndoRecPtr *prevlog_xact_start,
+								  UndoRecPtr *prevlog_insert_urp);
 extern UndoRecPtr UndoLogAllocateInRecovery(TransactionId xid,
 											uint16 size,
 											UndoRecPtr try_location,
 											bool *need_xact_header,
 											UndoRecPtr *last_xact_start,
+											UndoRecPtr *prevlog_xact_start,
+											UndoRecPtr *prevlog_last_urp,
 											XLogReaderState *xlog_record);
 extern void UndoLogAdvance(UndoRecPtr insertion_point, size_t size);
 extern void UndoLogDiscard(UndoRecPtr discard_point, TransactionId xid);
@@ -378,6 +392,9 @@ extern UndoRecPtr UndoLogGetNextInsertPtr(UndoLogNumber logno,
 										  TransactionId xid);
 extern UndoRecPtr UndoLogGetLastRecordPtr(UndoLogNumber,
 										  TransactionId xid);
+extern void UndoLogSwitchSetPrevLogInfo(UndoLogNumber logno,
+										UndoRecPtr prevlog_last_urp,
+										UndoRecPtr prevlog_xact_start);
 extern void UndoLogRewind(UndoRecPtr insert_urp, uint16 prevlen);
 extern uint16 UndoLogGetPrevLen(UndoLogNumber logno);
 extern void UndoLogSetLSN(XLogRecPtr lsn);
diff --git a/src/include/access/undolog_xlog.h b/src/include/access/undolog_xlog.h
index feebf81..802a727 100644
--- a/src/include/access/undolog_xlog.h
+++ b/src/include/access/undolog_xlog.h
@@ -22,6 +22,7 @@
 #define XLOG_UNDOLOG_EXTEND		0x10
 #define XLOG_UNDOLOG_DISCARD	0x20
 #define XLOG_UNDOLOG_REWIND		0x30
+#define XLOG_UNDOLOG_SWITCH		0x40
 
 /* Create a new undo log. */
 typedef struct xl_undolog_create
@@ -56,6 +57,14 @@ typedef struct xl_undolog_rewind
 	uint16		  prevlen;
 } xl_undolog_rewind;
 
+/* Switch undo log. */
+typedef struct xl_undolog_switch
+{
+	UndoLogNumber logno;
+	UndoRecPtr prevlog_xact_start;
+	UndoRecPtr prevlog_last_urp;
+} xl_undolog_switch;
+
 extern void undolog_desc(StringInfo buf,XLogReaderState *record);
 extern const char *undolog_identify(uint8 info);
 
-- 
1.8.3.1

