v1-0006-WIP-pg_waldump-Remove-the-restriction-on-the-orde.patch

From 7469b7b6bf3dd84d092fd86f69bf5ab574ee4f85 Mon Sep 17 00:00:00 2001
From: Amul Sul <sulamul@gmail.com>
Date: Thu, 7 Aug 2025 17:37:23 +0530
Subject: [PATCH v1 6/9] WIP-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.

TODO:
  Timeline switching is not handled correctly, especially when a
  timeline change occurs on the next WAL file that was previously
  written to a temporary location.
---
 doc/src/sgml/ref/pg_waldump.sgml       |   8 +-
 src/bin/pg_waldump/astreamer_waldump.c | 188 +++++++++++++++++++++----
 src/bin/pg_waldump/pg_waldump.c        |  99 ++++++++++++-
 src/bin/pg_waldump/pg_waldump.h        |   1 +
 src/bin/pg_waldump/t/001_basic.pl      |  40 +++++-
 5 files changed, 301 insertions(+), 35 deletions(-)

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d004bb0f67e..8a28b4f0f91 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -149,8 +149,12 @@ 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 to a temporary directory
+        named <filename>pg_waldump_tmp_dir/</filename>. This directory will be
+        created inside the directory specified by the <envar>TMPDIR</envar>
+        environment variable if it is set; otherwise, it will be created within
+        the same directory as the tar archive.
        </para>
       </listitem>
      </varlistentry>
diff --git a/src/bin/pg_waldump/astreamer_waldump.c b/src/bin/pg_waldump/astreamer_waldump.c
index d0ac903c54e..a088c33b16f 100644
--- a/src/bin/pg_waldump/astreamer_waldump.c
+++ b/src/bin/pg_waldump/astreamer_waldump.c
@@ -18,6 +18,7 @@
 
 #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"
@@ -37,6 +38,9 @@ typedef struct astreamer_waldump
 
 	/* These fields change with archive member. */
 	bool		skipThisSeg;
+	bool		writeThisSeg;
+	FILE	   *segFp;
+	SimpleStringList exportedSegList;	/* Temporary exported segment list */
 	XLogSegNo	nextSegNo;		/* Next expected segment to stream */
 } astreamer_waldump;
 
@@ -53,8 +57,11 @@ static bool member_is_relevant_wal(astreamer_member *member,
 								   XLogSegNo startSegNo,
 								   XLogSegNo endSegNo,
 								   XLogSegNo nextSegNo,
+								   char **curFname,
 								   XLogSegNo *curSegNo,
 								   TimeLineID *curSegTimeline);
+static bool member_needs_temp_write(astreamer_waldump *mystreamer,
+									const char *fname);
 
 static const astreamer_ops astreamer_waldump_ops = {
 	.content = astreamer_waldump_content,
@@ -189,17 +196,8 @@ astreamer_waldump_content_new(astreamer *next, XLogRecPtr startptr,
 	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
@@ -228,19 +226,21 @@ astreamer_waldump_content(astreamer *streamer, astreamer_member *member,
 	{
 		case ASTREAMER_MEMBER_HEADER:
 			{
+				char	   *fname;
 				XLogSegNo	segNo;
 				TimeLineID	timeline;
 
 				pg_log_debug("pg_waldump: reading \"%s\"", member->pathname);
 
 				mystreamer->skipThisSeg = false;
+				mystreamer->writeThisSeg = false;
 
 				if (!member_is_relevant_wal(member,
 											privateInfo->timeline,
 											mystreamer->startSegNo,
 											mystreamer->endSegNo,
 											mystreamer->nextSegNo,
-											&segNo, &timeline))
+											&fname, &segNo, &timeline))
 				{
 					mystreamer->skipThisSeg = true;
 					break;
@@ -254,24 +254,37 @@ astreamer_waldump_content(astreamer *streamer, astreamer_member *member,
 				if (mystreamer->nextSegNo == 0)
 					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 != segNo &&
+					member_needs_temp_write(mystreamer, fname))
 				{
-					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;
+					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.
+				 * 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.
+				 * Also, if there's any existing data in the buffer, the next
+				 * WAL record should logically follow it.
 				 */
 #ifdef USE_ASSERT_CHECKING
-				if (mystreamer->nextSegNo != 0)
+				Assert(!mystreamer->skipThisSeg);
+				Assert(!mystreamer->writeThisSeg);
+
+				if (privateInfo->archive_streamer_buf->len != 0)
 				{
 					XLogRecPtr	recPtr;
 
@@ -280,6 +293,13 @@ astreamer_waldump_content(astreamer *streamer, astreamer_member *member,
 				}
 #endif
 
+				/*
+				 * Initialized to the beginning of the current segment being
+				 * streamed through the buffer.
+				 */
+				XLogSegNoOffsetToRecPtr(segNo, 0, WalSegSz,
+										privateInfo->archive_streamer_read_ptr);
+
 				/* Save the timeline */
 				privateInfo->timeline = timeline;
 
@@ -293,12 +313,44 @@ 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/%s\": %m",
+							 privateInfo->tmpdir, 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;
+			}
 			break;
 
 		case ASTREAMER_ARCHIVE_TRAILER:
@@ -325,8 +377,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);
 }
@@ -339,8 +397,8 @@ astreamer_waldump_free(astreamer *streamer)
 static bool
 member_is_relevant_wal(astreamer_member *member, TimeLineID startTimeLineID,
 					   XLogSegNo startSegNo, XLogSegNo endSegNo,
-					   XLogSegNo nextSegNo, XLogSegNo *curSegNo,
-					   TimeLineID *curSegTimeline)
+					   XLogSegNo nextSegNo, char **curFname,
+					   XLogSegNo *curSegNo, TimeLineID *curSegTimeline)
 {
 	int			pathlen;
 	XLogSegNo	segNo;
@@ -371,8 +429,90 @@ member_is_relevant_wal(astreamer_member *member, TimeLineID startTimeLineID,
 	if (startSegNo > segNo || endSegNo < segNo)
 		return false;
 
+	/*
+	 * A corner case where we've already streamed the contents of an archived
+	 * WAL segment with a similar name, so ignoring this duplicate.
+	 */
+	if (nextSegNo > segNo)
+		return false;
+
+	*curFname = fname;
 	*curSegNo = segNo;
 	*curSegTimeline = timeline;
 
 	return true;
 }
+
+/*
+ * Returns true and creates a temporary file if the given WAL segment needs to
+ * be written to temporary space. This is required when the segment is not the
+ * one currently being decoded. Conversely, if a temporary file for the
+ * preceding segment already exists and the current segment is its direct
+ * successor, then writing to temporary space is not necessary, and false is
+ * returned.
+ */
+static bool
+member_needs_temp_write(astreamer_waldump *mystreamer, const char *fname)
+{
+	bool		exists;
+	XLogSegNo	segNo;
+	TimeLineID	timeline;
+	XLogDumpPrivate *privateInfo = mystreamer->privateInfo;
+
+	/* Parse position from file */
+	XLogFromFileName(fname, &timeline, &segNo, WalSegSz);
+
+	/*
+	 * 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 the current segment number matches the required WAL
+	 * segment (i.e. nextSegNo). If it does, we allow it to stream normally
+	 * through the buffer. Otherwise, we write it to the temporary space, from
+	 * where the caller is expected to read it directly.
+	 */
+	do
+	{
+		char		segName[MAXFNAMELEN];
+
+		XLogFileName(segName, timeline, mystreamer->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(&mystreamer->exportedSegList, segName))
+		{
+			mystreamer->nextSegNo += 1;
+			exists = true;
+		}
+	} while (exists);
+
+	/*
+	 * Need to export this segment to disk; create an empty placeholder file
+	 * to be written once its content is received.
+	 */
+	if (mystreamer->nextSegNo != segNo)
+	{
+		char		fpath[MAXPGPATH];
+
+		snprintf(fpath, MAXPGPATH, "%s/%s", privateInfo->tmpdir, fname);
+
+		mystreamer->segFp = fopen(fpath, PG_BINARY_W);
+		if (mystreamer->segFp == 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
+
+		/* Record this segment's export to temporary space */
+		simple_string_list_append(&mystreamer->exportedSegList, fname);
+		return true;
+	}
+
+	return false;
+}
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 64f3a65b735..54a3b2dacda 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -325,6 +325,51 @@ identify_target_directory(char *directory, char *fname)
 	return NULL;				/* not reached */
 }
 
+/*
+ * Set up a temporary directory to temporarily store WAL segments.
+ */
+static char *
+setup_tmp_dir(char *waldir)
+{
+	char	   *tmpdir = waldir != NULL ? pstrdup(waldir) :  pstrdup(".");
+
+	canonicalize_path(tmpdir);
+	tmpdir = psprintf("%s/pg_waldump_tmp_dir",
+					  getenv("TMPDIR") ? getenv("TMPDIR") : tmpdir);
+
+	create_directory(tmpdir);
+
+	return tmpdir;
+}
+
+/*
+ * Removes a directory along with its contents, if any.
+ */
+static void
+remove_tmp_dir(char *tmpdir)
+{
+	DIR		   *dir;
+	struct dirent *de;
+
+	dir = opendir(tmpdir);
+	while ((de = readdir(dir)) != NULL)
+	{
+		char		path[MAXPGPATH];
+
+		if (strcmp(de->d_name, ".") == 0 ||
+			strcmp(de->d_name, "..") == 0)
+			continue;
+
+		snprintf(path, MAXPGPATH, "%s/%s", tmpdir, de->d_name);
+		unlink(path);
+	}
+	closedir(dir);
+
+	if (rmdir(tmpdir) < 0)
+		pg_log_error("could not remove directory \"%s\": %m",
+					 tmpdir);
+}
+
 /*
  * Returns true if the given file is a tar archive and outputs its compression
  * algorithm.
@@ -559,7 +604,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)
@@ -618,12 +663,53 @@ 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.
+	 *
+	 * XXX: Timeline change is not handled.
+	 */
+	nextSegNo = state->seg.ws_segno;
+	if (!XLByteInSeg(targetPagePtr, nextSegNo, WalSegSz))
+	{
+		char		fname[MAXPGPATH];
+
+		if (state->seg.ws_file >= 0)
+		{
+			char		fpath[MAXPGPATH];
+
+			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);
+			snprintf(fpath, MAXPGPATH, "%s/%s", private->tmpdir, fname);
+			unlink(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);
+		state->seg.ws_file = open_file_in_directory(private->tmpdir, fname);
+	}
+
+	/* Continue reading from the open WAL segment, if any */
+	if (state->seg.ws_file >= 0)
+		return WALDumpReadPage(state, targetPagePtr, reqLen, targetPtr,
+							   readBuff);
+
+	/* Otherwise, read the WAL page from the archive streamer */
 	return astreamer_wal_read(readBuff, targetPagePtr, count, private);
 }
 
@@ -1435,6 +1521,9 @@ main(int argc, char **argv)
 		/* Set up for reading tar file */
 		init_tar_archive_reader(&private, waldir, compression);
 
+		/* Create temporary space for writing WAL segments. */
+		private.tmpdir = setup_tmp_dir(waldir);
+
 		/* Routine to decode WAL files in tar archive */
 		routine = XL_ROUTINE(.page_read = TarWALDumpReadPage,
 							 .segment_open = TarWALDumpOpenSegment,
@@ -1549,6 +1638,10 @@ main(int argc, char **argv)
 	if (config.stats == true && !config.quiet)
 		XLogDumpDisplayStats(&config, &stats);
 
+	/* Remove temporary directory if any */
+	if (private.tmpdir != NULL)
+		remove_tmp_dir(private.tmpdir);
+
 	if (time_to_stop)
 		exit(0);
 
diff --git a/src/bin/pg_waldump/pg_waldump.h b/src/bin/pg_waldump/pg_waldump.h
index d2c2307d6c2..2644d847b47 100644
--- a/src/bin/pg_waldump/pg_waldump.h
+++ b/src/bin/pg_waldump/pg_waldump.h
@@ -33,6 +33,7 @@ typedef struct XLogDumpPrivate
 	StringInfo	archive_streamer_buf;	/* Buffer for receiving WAL data */
 	XLogRecPtr	archive_streamer_read_ptr; /* Populate the buffer with records
 											  until this record pointer */
+	char	   *tmpdir;
 } XLogDumpPrivate;
 
 
diff --git a/src/bin/pg_waldump/t/001_basic.pl b/src/bin/pg_waldump/t/001_basic.pl
index 80298d2a51d..a3bf950db97 100644
--- a/src/bin/pg_waldump/t/001_basic.pl
+++ b/src/bin/pg_waldump/t/001_basic.pl
@@ -7,6 +7,8 @@ use Cwd;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
 use Test::More;
+use File::Path qw(rmtree);
+use List::Util qw(shuffle);
 
 my $tar = $ENV{TAR};
 
@@ -258,6 +260,32 @@ sub test_pg_waldump
 	return @lines;
 }
 
+# Create a tar archive, shuffling the file order
+sub generate_archive
+{
+	my ($archive, $directory, $compression_flags) = @_;
+
+	my @files;
+	opendir my $dh, $directory or die "opendir: $!";
+	while (my $entry = readdir $dh) {
+		# Skip '.' and '..'
+		next if $entry eq '.' || $entry eq '..';
+		push @files, $entry;
+	}
+	closedir $dh;
+
+	@files = shuffle @files;
+
+	# move into the WAL directory before archiving files
+	my $cwd = getcwd;
+	chdir($directory) || die "chdir: $!";
+	command_ok([$tar, $compression_flags, $archive, @files]);
+	chdir($cwd) || die "chdir: $!";
+
+	# give necessary permission
+	chmod(0755, $archive) || die "chmod $archive: $!";
+}
+
 my $tmp_dir = PostgreSQL::Test::Utils::tempdir_short();
 
 my @scenario = (
@@ -291,16 +319,16 @@ for my $scenario (@scenario)
 		  if !defined $tar;
 		skip "$scenario->{'compression_method'} compression not supported by this build", 3
 		  if !$scenario->{'enabled'} && $scenario->{'is_archive'};
+		skip "unix-style permissions not supported on Windows", 3
+		  if ($scenario->{'is_archive'}
+			&& ($windows_os || $Config::Config{osname} eq 'cygwin'));
 
 		  # create pg_wal archive
 		  if ($scenario->{'is_archive'})
 		  {
-			  # move into the WAL directory before archiving files
-			  my $cwd = getcwd;
-			  chdir($node->data_dir . '/pg_wal/') || die "chdir: $!";
-			  command_ok(
-				  [ $tar, $scenario->{'compression_flags'}, $path , '.' ]);
-			  chdir($cwd) || die "chdir: $!";
+			  generate_archive($path,
+				  $node->data_dir . '/pg_wal',
+				  $scenario->{'compression_flags'});
 		  }
 
 		command_fails_like(
-- 
2.47.1