From d79a505af9532baae675557de0efb1bcc35d72f5 Mon Sep 17 00:00:00 2001
From: Amul Sul <sulamul@gmail.com>
Date: Mon, 25 Aug 2025 17:26:29 +0530
Subject: [PATCH v4 5/8] pg_waldump: Remove the restriction on the order of
 archived WAL files.

With previous patch, pg_waldump would stop decoding if WAL files were
not in the required sequence. With this patch, decoding will now
continue.  Any WAL file that is out of order will be written to a
temporary location, from which it will be read later. Once a temporary
file has been read, it will be removed.
---
 doc/src/sgml/ref/pg_waldump.sgml       |   7 +-
 src/bin/pg_waldump/astreamer_waldump.c | 214 +++++++++++++++++++++----
 src/bin/pg_waldump/pg_waldump.c        | 112 ++++++++++++-
 src/bin/pg_waldump/pg_waldump.h        |  30 +++-
 src/bin/pg_waldump/t/001_basic.pl      |   3 +-
 5 files changed, 323 insertions(+), 43 deletions(-)

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d004bb0f67e..c1afb4097b5 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -149,8 +149,11 @@ PostgreSQL documentation
         of <envar>PGDATA</envar>.
        </para>
        <para>
-        If a tar archive is provided, its WAL segment files must be in
-        sequential order; otherwise, an error will be reported.
+        If a tar archive is provided and its WAL segment files are not in
+        sequential order, those files will be written temporarily. These files
+        will be created inside the directory specified by the <envar>TMPDIR</envar>
+        environment variable if it is set; otherwise, the temporary files will
+        be created within the same directory as the tar archive itself.
        </para>
       </listitem>
      </varlistentry>
diff --git a/src/bin/pg_waldump/astreamer_waldump.c b/src/bin/pg_waldump/astreamer_waldump.c
index caf7da6ccb8..40876c77f6c 100644
--- a/src/bin/pg_waldump/astreamer_waldump.c
+++ b/src/bin/pg_waldump/astreamer_waldump.c
@@ -18,8 +18,8 @@
 
 #include "access/xlog_internal.h"
 #include "access/xlogdefs.h"
+#include "common/file_perm.h"
 #include "common/logging.h"
-#include "fe_utils/simple_list.h"
 #include "pg_waldump.h"
 
 /*
@@ -42,10 +42,11 @@ typedef struct astreamer_waldump
 
 	/* These fields change with archive member. */
 	bool		skipThisSeg;
+	bool		writeThisSeg;
+	FILE	   *segFp;
 	XLogSegNo	nextSegNo;		/* Next expected segment to stream */
 } astreamer_waldump;
 
-static int	astreamer_archive_read(XLogDumpPrivate *privateInfo);
 static void astreamer_waldump_content(astreamer *streamer,
 									  astreamer_member *member,
 									  const char *data, int len,
@@ -56,7 +57,12 @@ static void astreamer_waldump_free(astreamer *streamer);
 static bool member_is_relevant_wal(astreamer_waldump *mystreamer,
 								   astreamer_member *member,
 								   TimeLineID startTimeLineID,
+								   char **curFname,
 								   XLogSegNo *curSegNo);
+static FILE *member_prepare_tmp_write(XLogSegNo curSegNo,
+									  const char *fname);
+static XLogSegNo member_next_segno(XLogSegNo curSegNo,
+								   TimeLineID timeline);
 
 static const astreamer_ops astreamer_waldump_ops = {
 	.content = astreamer_waldump_content,
@@ -164,7 +170,7 @@ astreamer_wal_read(char *readBuff, XLogRecPtr targetPagePtr, Size count,
 /*
  * Reads the archive and passes it to the archive streamer for decompression.
  */
-static int
+int
 astreamer_archive_read(XLogDumpPrivate *privateInfo)
 {
 	int			rc;
@@ -208,17 +214,8 @@ astreamer_waldump_new(XLogRecPtr startptr, XLogRecPtr endPtr,
 	if (XLogRecPtrIsInvalid(startptr))
 		streamer->startSegNo = 0;
 	else
-	{
 		XLByteToSeg(startptr, streamer->startSegNo, WalSegSz);
 
-		/*
-		 * Initialize the record pointer to the beginning of the first
-		 * segment; this pointer will track the WAL record reading status.
-		 */
-		XLogSegNoOffsetToRecPtr(streamer->startSegNo, 0, WalSegSz,
-								privateInfo->archive_streamer_read_ptr);
-	}
-
 	if (XLogRecPtrIsInvalid(endPtr))
 		streamer->endSegNo = UINT64_MAX;
 	else
@@ -247,14 +244,16 @@ astreamer_waldump_content(astreamer *streamer, astreamer_member *member,
 	{
 		case ASTREAMER_MEMBER_HEADER:
 			{
-				XLogSegNo	segNo;
+				char	   *fname;
 
 				pg_log_debug("pg_waldump: reading \"%s\"", member->pathname);
 
 				mystreamer->skipThisSeg = false;
+				mystreamer->writeThisSeg = false;
 
 				if (!member_is_relevant_wal(mystreamer, member,
-											privateInfo->timeline, &segNo))
+											privateInfo->timeline,
+											&fname, &privateInfo->curSegNo))
 				{
 					mystreamer->skipThisSeg = true;
 					break;
@@ -267,33 +266,67 @@ astreamer_waldump_content(astreamer *streamer, astreamer_member *member,
 				if (READ_ANY_WAL(mystreamer))
 					break;
 
-				/* WAL segments must be archived in order */
-				if (mystreamer->nextSegNo != segNo)
+				/*
+				 * When WAL segments are not archived sequentially, it becomes
+				 * necessary to write out (or preserve) segments that might be
+				 * required at a later point.
+				 */
+				if (mystreamer->nextSegNo != privateInfo->curSegNo)
 				{
-					pg_log_error("WAL files are not archived in sequential order");
-					pg_log_error_detail("Expecting segment number " UINT64_FORMAT " but found " UINT64_FORMAT ".",
-										mystreamer->nextSegNo, segNo);
-					exit(1);
+					mystreamer->writeThisSeg = true;
+					mystreamer->segFp =
+						member_prepare_tmp_write(privateInfo->curSegNo, fname);
+					break;
 				}
 
 				/*
-				 * We track the reading of WAL segment records using a pointer
-				 * that's continuously incremented by the length of the
-				 * received data. This pointer is crucial for serving WAL page
-				 * requests from the WAL decoding routine, so it must be
-				 * accurate.
+				 * If the buffer contains data, the next WAL record must
+				 * logically follow it. Otherwise, this file isn't the one we
+				 * need, and we must export it.
 				 */
-#ifdef USE_ASSERT_CHECKING
-				if (mystreamer->nextSegNo != 0)
+				else if (privateInfo->archive_streamer_buf->len != 0)
 				{
 					XLogRecPtr	recPtr;
 
-					XLogSegNoOffsetToRecPtr(segNo, 0, WalSegSz, recPtr);
-					Assert(privateInfo->archive_streamer_read_ptr == recPtr);
+					XLogSegNoOffsetToRecPtr(privateInfo->curSegNo, 0, WalSegSz,
+											recPtr);
+
+					if (privateInfo->archive_streamer_read_ptr != recPtr)
+					{
+						mystreamer->writeThisSeg = true;
+						mystreamer->segFp =
+							member_prepare_tmp_write(privateInfo->curSegNo, fname);
+
+						/* Update the next expected segment number after this */
+						mystreamer->nextSegNo =
+							member_next_segno(privateInfo->curSegNo + 1,
+											  privateInfo->timeline);
+						break;
+					}
 				}
-#endif
+
+				Assert(!mystreamer->skipThisSeg);
+				Assert(!mystreamer->writeThisSeg);
+
+				/*
+				 * We are now streaming segment containt.
+				 *
+				 * We need to track the reading of WAL segment records using a
+				 * pointer that's typically incremented by the length of the
+				 * data read. However, we sometimes export the WAL file to
+				 * temporary storage, allowing the decoding routine to read
+				 * directly from there. This makes continuous pointer
+				 * incrementing challenging, as file reads can occur from any
+				 * offset, leading to potential errors. Therefore, we now
+				 * reset the pointer when reading from a file for streaming.
+				 */
+				XLogSegNoOffsetToRecPtr(privateInfo->curSegNo, 0, WalSegSz,
+										privateInfo->archive_streamer_read_ptr);
+
 				/* Update the next expected segment number */
-				mystreamer->nextSegNo += 1;
+				mystreamer->nextSegNo =
+					member_next_segno(privateInfo->curSegNo,
+									  privateInfo->timeline);
 			}
 			break;
 
@@ -302,12 +335,45 @@ astreamer_waldump_content(astreamer *streamer, astreamer_member *member,
 			if (mystreamer->skipThisSeg)
 				break;
 
+			/* Or, write contents to file */
+			if (mystreamer->writeThisSeg)
+			{
+				Assert(mystreamer->segFp != NULL);
+
+				errno = 0;
+				if (len > 0 && fwrite(data, len, 1, mystreamer->segFp) != 1)
+				{
+					char	   *fname;
+					int			pathlen = strlen(member->pathname);
+
+					Assert(pathlen >= XLOG_FNAME_LEN);
+
+					fname = member->pathname + (pathlen - XLOG_FNAME_LEN);
+
+					/*
+					 * If write didn't set errno, assume problem is no disk
+					 * space
+					 */
+					if (errno == 0)
+						errno = ENOSPC;
+					pg_fatal("could not write to file \"%s\": %m",
+							 get_tmp_wal_file_path(fname));
+				}
+				break;
+			}
+
 			/* Or, copy contents to buffer */
 			privateInfo->archive_streamer_read_ptr += len;
 			astreamer_buffer_bytes(streamer, &data, &len, len);
 			break;
 
 		case ASTREAMER_MEMBER_TRAILER:
+			if (mystreamer->segFp != NULL)
+			{
+				fclose(mystreamer->segFp);
+				mystreamer->segFp = NULL;
+			}
+			privateInfo->curSegNo = 0;
 			break;
 
 		case ASTREAMER_ARCHIVE_TRAILER:
@@ -334,8 +400,14 @@ astreamer_waldump_finalize(astreamer *streamer)
 static void
 astreamer_waldump_free(astreamer *streamer)
 {
+	astreamer_waldump *mystreamer;
+
 	Assert(streamer->bbs_next == NULL);
 
+	mystreamer = (astreamer_waldump *) streamer;
+	if (mystreamer->segFp != NULL)
+		fclose(mystreamer->segFp);
+
 	pfree(streamer->bbs_buffer.data);
 	pfree(streamer);
 }
@@ -347,7 +419,8 @@ astreamer_waldump_free(astreamer *streamer)
  */
 static bool
 member_is_relevant_wal(astreamer_waldump *mystreamer, astreamer_member *member,
-					   TimeLineID startTimeLineID, XLogSegNo *curSegNo)
+					   TimeLineID startTimeLineID, char **curFname,
+					   XLogSegNo *curSegNo)
 {
 	int			pathlen;
 	XLogSegNo	segNo;
@@ -382,7 +455,84 @@ member_is_relevant_wal(astreamer_waldump *mystreamer, astreamer_member *member,
 			return false;
 	}
 
+	*curFname = fname;
 	*curSegNo = segNo;
 
 	return true;
 }
+
+/*
+ * Create an empty placeholder file and return its handle.  The file is also
+ * added to an exported list for future management, e.g.  access, deletion, and
+ * existence checks.
+ */
+static FILE *
+member_prepare_tmp_write(XLogSegNo curSegNo, const char *fname)
+{
+	FILE	   *file;
+	char	   *fpath = get_tmp_wal_file_path(fname);
+
+	/* Create an empty placeholder */
+	file = fopen(fpath, PG_BINARY_W);
+	if (file == NULL)
+		pg_fatal("could not create file \"%s\": %m", fpath);
+
+#ifndef WIN32
+	if (chmod(fpath, pg_file_create_mode))
+		pg_fatal("could not set permissions on file \"%s\": %m",
+				 fpath);
+#endif
+
+	pg_log_info("temporarily exporting file \"%s\"", fpath);
+
+	/* Record this segment's export */
+	simple_string_list_append(&TmpWalSegList, fname);
+	pfree(fpath);
+
+	return file;
+}
+
+/*
+ * Get next WAL segment that needs to be retrieved from the archive.
+ *
+ * The function checks for the presence of a previously read and extracted WAL
+ * segment in the temporary storage. If a temporary file is found for that
+ * segment, it indicates the segment has already been successfully retrieved
+ * from the archive. In this case, the function increments the segment number
+ * and repeats the check. This process continues until a segment that has not
+ * yet been retrieved is found, at which point the function returns its number.
+ */
+static XLogSegNo
+member_next_segno(XLogSegNo curSegNo, TimeLineID timeline)
+{
+	XLogSegNo	nextSegNo = curSegNo + 1;
+	bool		exists;
+
+	/*
+	 * If we find a file that was previously written to the temporary space,
+	 * it indicates that the corresponding WAL segment request has already
+	 * been fulfilled. In that case, we increment the nextSegNo counter and
+	 * check again whether that segment number again. if found above steps
+	 * will be return if not then we return that segment number which would be
+	 * needed from the archive.
+	 */
+	do
+	{
+		char		fname[MAXFNAMELEN];
+
+		XLogFileName(fname, timeline, nextSegNo, WalSegSz);
+
+		/*
+		 * If the WAL segment has already been exported, increment the counter
+		 * and check for the next segment.
+		 */
+		exists = false;
+		if (simple_string_list_member(&TmpWalSegList, fname))
+		{
+			nextSegNo += 1;
+			exists = true;
+		}
+	} while (exists);
+
+	return nextSegNo;
+}
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 393d6bfa9ef..615227b691c 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -43,6 +43,10 @@ static const char *progname;
 int			WalSegSz = DEFAULT_XLOG_SEG_SIZE;
 static volatile sig_atomic_t time_to_stop = false;
 
+/* Temporary exported WAL file directory and the list */
+char	   *TmpWalSegDir = NULL;
+SimpleStringList TmpWalSegList = {NULL, NULL};
+
 static const RelFileLocator emptyRelFileLocator = {0, 0, 0};
 
 typedef struct XLogDumpConfig
@@ -360,6 +364,41 @@ is_tar_file(const char *fname, pg_compress_algorithm *compression)
 	return true;
 }
 
+/*
+ * Set up a temporary directory to temporarily store WAL segments.
+ */
+static void
+setup_tmp_walseg_dir(const char *waldir)
+{
+	/*
+	 * Use the directory specified by the TEMDIR environment variable. If it's
+	 * not set, use the provided WAL directory.
+	 */
+	TmpWalSegDir = getenv("TMPDIR") ?
+		pg_strdup(getenv("TMPDIR")) : pg_strdup(waldir);
+	canonicalize_path(TmpWalSegDir);
+}
+
+/*
+ * Removes the temporarily store WAL segments, if any at exiting.
+ */
+static void
+remove_tmp_walseg_dir_atexit(void)
+{
+	SimpleStringListCell *cell;
+
+	/* Clear out any existing temporary files */
+	for (cell = TmpWalSegList.head; cell; cell = cell->next)
+	{
+		char	   *fpath = get_tmp_wal_file_path(cell->val);
+
+		if (unlink(fpath) == 0)
+			pg_log_info("removed file \"%s\"", fpath);
+		pfree(fpath);
+	}
+}
+
+
 /*
  * Initializes the tar archive reader and a temporary directory for WAL files.
  */
@@ -404,6 +443,7 @@ init_tar_archive_reader(XLogDumpPrivate *private, const char *waldir,
 		streamer = astreamer_zstd_decompressor_new(streamer);
 
 	private->archive_streamer = streamer;
+	private->curSegNo = 0;
 }
 
 /*
@@ -548,7 +588,7 @@ WALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
 				XLogRecPtr targetPtr, char *readBuff)
 {
 	XLogDumpPrivate *private = state->private_data;
-	int			count = required_read_len(private, targetPtr, reqLen);
+	int			count = required_read_len(private, targetPagePtr, reqLen);
 	WALReadError errinfo;
 
 	if (private->endptr_reached)
@@ -607,12 +647,70 @@ TarWALDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
 				   XLogRecPtr targetPtr, char *readBuff)
 {
 	XLogDumpPrivate *private = state->private_data;
-	int			count = required_read_len(private, targetPtr, reqLen);
+	int			count = required_read_len(private, targetPagePtr, reqLen);
+	XLogSegNo	nextSegNo;
 
 	if (private->endptr_reached)
 		return -1;
 
-	/* Read the WAL page from the archive streamer */
+	/*
+	 * If the target page is in a different segment, first check for the WAL
+	 * segment's physical existence in the temporary directory.
+	 */
+	nextSegNo = state->seg.ws_segno;
+	if (!XLByteInSeg(targetPagePtr, nextSegNo, WalSegSz))
+	{
+		char		fname[MAXFNAMELEN];
+		char	   *fpath;
+
+		if (state->seg.ws_file >= 0)
+		{
+			close(state->seg.ws_file);
+			state->seg.ws_file = -1;
+
+			/* Remove this file, as it is no longer needed. */
+			XLogFileName(fname, state->seg.ws_tli, nextSegNo, WalSegSz);
+			fpath = get_tmp_wal_file_path(fname);
+			pg_log_info("removing file \"%s\"", fpath);
+			unlink(fpath);
+			pfree(fpath);
+		}
+
+		XLByteToSeg(targetPagePtr, nextSegNo, WalSegSz);
+		state->seg.ws_tli = private->timeline;
+		state->seg.ws_segno = nextSegNo;
+
+		/*
+		 * If the next segment exists, open it and continue reading from there
+		 */
+		XLogFileName(fname, private->timeline, nextSegNo, WalSegSz);
+		if (simple_string_list_member(&TmpWalSegList, fname))
+		{
+			fpath = get_tmp_wal_file_path(fname);
+			state->seg.ws_file = open(fpath, O_RDONLY | PG_BINARY, 0);
+
+			if (state->seg.ws_file < 0)
+				pg_fatal("could not open file \"%s\": %m", fpath);
+			pfree(fpath);
+		}
+	}
+
+	/* Continue reading from the open WAL segment, if any */
+	if (state->seg.ws_file >= 0)
+	{
+		/*
+		 * To prevent a race condition where the archive streamer is still
+		 * exporting a file that we are trying to read, we invoke the streamer
+		 * to ensure enough data is available.
+		 */
+		if (private->curSegNo == state->seg.ws_segno)
+			astreamer_archive_read(private);
+
+		return WALDumpReadPage(state, targetPagePtr, reqLen, targetPtr,
+							   readBuff);
+	}
+
+	/* Otherwise, read the WAL page from the archive streamer */
 	return astreamer_wal_read(readBuff, targetPagePtr, count, private);
 }
 
@@ -1340,7 +1438,6 @@ main(int argc, char **argv)
 		if (fname != NULL && is_tar_file(fname, &compression))
 		{
 			private.archive_name = fname;
-			waldir = walpath ? pg_strdup(walpath) : pg_strdup(".");
 			is_tar = true;
 		}
 		else
@@ -1434,6 +1531,13 @@ main(int argc, char **argv)
 		init_tar_archive_reader(&private, waldir, private.startptr,
 								private.endptr, compression);
 
+		/*
+		 * Setup temporary directory to store WAL segments and set up an exit
+		 * callback to remove it upon completion.
+		 */
+		setup_tmp_walseg_dir(waldir);
+		atexit(remove_tmp_walseg_dir_atexit);
+
 		/* Routine to decode WAL files in tar archive */
 		xlogreader_state =
 			XLogReaderAllocate(WalSegSz, waldir,
diff --git a/src/bin/pg_waldump/pg_waldump.h b/src/bin/pg_waldump/pg_waldump.h
index 4205e0ef597..1a1cf35e6f3 100644
--- a/src/bin/pg_waldump/pg_waldump.h
+++ b/src/bin/pg_waldump/pg_waldump.h
@@ -13,9 +13,14 @@
 
 #include "access/xlogdefs.h"
 #include "fe_utils/astreamer.h"
+#include "fe_utils/simple_list.h"
 #include "lib/stringinfo.h"
 
+#define TEMP_FILE_EXT	"waldump.tmp"
+
 extern int	WalSegSz;
+extern char *TmpWalSegDir;
+extern SimpleStringList TmpWalSegList;
 
 /* Contains the necessary information to drive WAL decoding */
 typedef struct XLogDumpPrivate
@@ -31,15 +36,32 @@ typedef struct XLogDumpPrivate
 
 	astreamer  *archive_streamer;
 	StringInfo	archive_streamer_buf;	/* Buffer for receiving WAL data */
-	XLogRecPtr	archive_streamer_read_ptr; /* Populate the buffer with records
-											  until this record pointer */
+	XLogRecPtr	archive_streamer_read_ptr;	/* Populate the buffer with
+											 * records until this record
+											 * pointer */
+	XLogSegNo	curSegNo;		/* Current segment being read */
 } XLogDumpPrivate;
 
+/*
+ * Generate the temporary WAL file path.
+ *
+ * Note that the caller is responsible to pfree it.
+ */
+static inline char *
+get_tmp_wal_file_path(const char *fname)
+{
+	char	   *fpath = (char *) palloc(MAXPGPATH);
 
-extern astreamer *astreamer_waldump_new(XLogRecPtr startptr,
-										XLogRecPtr endptr,
+	snprintf(fpath, MAXPGPATH, "%s/%s.%s", TmpWalSegDir, fname,
+			 TEMP_FILE_EXT);
+
+	return fpath;
+}
+
+extern astreamer *astreamer_waldump_new(XLogRecPtr startptr, XLogRecPtr endptr,
 										XLogDumpPrivate *privateInfo);
 extern int astreamer_wal_read(char *readBuff, XLogRecPtr startptr, Size count,
 							  XLogDumpPrivate *privateInfo);
+extern int astreamer_archive_read(XLogDumpPrivate *privateInfo);
 
 #endif							/* end of PG_WALDUMP_H */
diff --git a/src/bin/pg_waldump/t/001_basic.pl b/src/bin/pg_waldump/t/001_basic.pl
index 443126a9ce6..d5fa1f6d28d 100644
--- a/src/bin/pg_waldump/t/001_basic.pl
+++ b/src/bin/pg_waldump/t/001_basic.pl
@@ -7,6 +7,7 @@ use Cwd;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
 use Test::More;
+use List::Util qw(shuffle);
 
 my $tar = $ENV{TAR};
 
@@ -272,7 +273,7 @@ sub generate_archive
 	}
 	closedir $dh;
 
-	@files = sort @files;
+	@files = shuffle @files;
 
 	# move into the WAL directory before archiving files
 	my $cwd = getcwd;
-- 
2.47.1

