[PATCH] Teach pg_waldump to extract FPIs from the WAL

From a016be601767956a26f8bc52c198b719e635d9a3 Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Wed, 20 Apr 2022 19:59:35 -0500
Subject: [PATCH] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a target directory, which must be empty or not
exist.  These images are subject to the same filtering rules as normal display in pg_waldump, which
means that you can isolate the full page writes to a target relation, among other things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk> with formatting to make things
somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0
00000000-01000150.1664.0.6115.0
00000000-010001E0.1664.0.6114.0
00000000-01000270.1663.1.6116.0
00000000-01000300.1663.1.6113.0
00000000-01000390.1663.1.6112.0
00000000-01000420.1663.1.8903.0
00000000-010004B0.1663.1.8902.0
00000000-01000540.1663.1.6111.0
00000000-010005D0.1663.1.6110.0

If the FPI comes from a fork other than the main fork, the fork name will be appended on the output
file name; e.g.:

00000000-014A4758.1663.1.12864.0_vm

It's noteworthy that the raw images do not have the current LSN stored with them in the WAL
stream (as would be true for on-heap versions of the blocks), nor would the checksum be valid in
them (though WAL itself has checksums, so there is some protection there).  This patch chooses to
place the LSN and calculate the proper checksum (if non-zero in the source image) in the outputted
block.  (This could perhaps be a targetted flag if we decide we don't always want this.)

These images could be loaded/inspected via `pg_read_binary_file()` and used in the `pageinspect`
suite of tools to perform detailed analysis on the pages in question, based on historical
information, and may come in handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml |  65 ++++++++++++++
 src/bin/pg_waldump/pg_waldump.c  | 144 ++++++++++++++++++++++++++++++-
 2 files changed, 208 insertions(+), 1 deletion(-)

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index 57746d9421..bd96798cef 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,71 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--raw-fpi=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        given <replaceable>save_path</replaceable>, which should not exist or
+        should be empty.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>TSOID</replaceable>.<replaceable>DBOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <table id="pgwaldump-save-fpi-components">
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+            as <literal>%08x-%08X</literal> instead of the
+            conventional <literal>%X/%X</literal> due to filesystem naming
+            limits</entry>
+           </row>
+           <row>
+            <entry>TSOID</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+           <row>
+            <entry>DBOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             if coming from the main fork, will be empty, otherwise will be
+             one of <literal>_fsm</literal>, <literal>_vm</literal>,
+             or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </table>        
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 4f265ef546..fccc07ee8a 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -23,9 +23,13 @@
 #include "access/xlogrecord.h"
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
+#include "common/relpath.h"
 #include "common/logging.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
+#include "storage/checksum.h"
+#include "storage/checksum_impl.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,10 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* output options */
+	bool        save_fpw;
+	char        *save_fpw_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +447,79 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPWs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		/* we will now extract the fullpage image from the XLogRecord and save
+		 * it to a calculated filename */
+
+		if (XLogRecHasBlockImage(record, block_id))
+		{
+			char page[BLCKSZ];
+
+			memset(page, 0, BLCKSZ);
+
+			if (RestoreBlockImage(record, block_id, page))
+			{
+				/* we have our extracted FPI, let's save it now */
+				char filename[MAXPGPATH];
+				char forkname[FORKNAMECHARS + 2]; /* _ + \0 */
+				FILE *OPF;
+				BlockNumber blk;
+				RelFileNode rnode;
+				ForkNumber  fork;
+
+				XLogRecGetBlockTagExtended(record, block_id,
+										   &rnode, &fork, &blk, NULL);
+
+				/*
+				 * The page may be uninitialized. If so, we can't set the LSN because
+				 * that would corrupt the page.
+				 */
+				if (!PageIsNew(page))
+				{
+					PageSetLSN(page, record->ReadRecPtr);
+					/* if checksum field is non-zero then we have checksums enabled,
+					 * so recalculate the checksum with new LSN (yes, this is a hack)
+					 */
+					if (((PageHeader) page)->pd_checksum)
+						((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blk);
+				}
+
+				if (fork >= 0 && fork <= MAX_FORKNUM)
+				{
+					if (fork)
+						sprintf(forkname, "_%s", forkNames[fork]);
+					else
+						forkname[0] = 0;
+				}
+				else
+					pg_fatal("Found invalid fork number: %u", fork);
+
+				snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+						 LSN_FORMAT_ARGS(record->ReadRecPtr),
+						 rnode.spcNode, rnode.dbNode, rnode.relNode, blk, forkname);
+
+				OPF = fopen(filename, PG_BINARY_W);
+				if (!OPF)
+					pg_fatal("Couldn't open file for output: %s", filename);
+
+				if (pg_pwrite(fileno(OPF), page, BLCKSZ, 0) != BLCKSZ)
+					pg_fatal("Couldn't write out complete FPI to file: %s", filename);
+
+				fclose(OPF);
+			}
+		}
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +760,7 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --raw-fpi=path     save found full page images to given path\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +794,7 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"raw-fpi", required_argument, NULL, 'W'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +855,8 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpw = false;
+	config.save_fpw_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +867,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -919,6 +1004,10 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+				config.save_fpw = true;
+				config.save_fpw_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1061,54 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpw_path != NULL)
+	{
+		struct stat st;
+		bool		is_empty = false;
+
+		/* we accept an empty existing directory */
+		if (stat(config.save_fpw_path, &st) == 0 && S_ISDIR(st.st_mode))
+		{
+			DIR		   *dir = opendir(config.save_fpw_path);
+
+			if (dir)
+			{
+				struct dirent *d;
+
+				is_empty = true;
+				while (errno = 0, (d = readdir(dir)))
+				{
+					if (strcmp(d->d_name, ".") != 0 && strcmp(d->d_name, "..") != 0)
+					{
+						is_empty = false;
+						break;
+					}
+				}
+
+				if (errno)
+				{
+					pg_log_error("could not read raw-fpi directory \"%s\": %m",
+								 config.save_fpw_path);
+					goto bad_argument;
+				}
+
+				if (closedir(dir))
+				{
+					pg_log_error("could not close raw-fpi directory \"%s\": %m",
+								 config.save_fpw_path);
+					goto bad_argument;
+				}
+			}
+		}
+		if (!is_empty && mkdir(config.save_fpw_path, 0700) < 0)
+		{
+			pg_fatal("could not create raw-fpi output directory \"%s\": %m",
+					 config.save_fpw_path);
+
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1287,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpw)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPWs(xlogreader_state, config.save_fpw_path);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
-- 
2.32.0 (Apple Git-132)

From d85ccf8c89a55cf1761149ad9ebb9f836877fc24 Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Wed, 20 Apr 2022 19:59:35 -0500
Subject: [PATCH] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a target directory,
which must be empty or not exist.  These images are subject to the same
filtering rules as normal display in pg_waldump, which means that you
can isolate the full page writes to a target relation, among other
things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0
00000000-01000150.1664.0.6115.0
00000000-010001E0.1664.0.6114.0
00000000-01000270.1663.1.6116.0
00000000-01000300.1663.1.6113.0
00000000-01000390.1663.1.6112.0
00000000-01000420.1663.1.8903.0
00000000-010004B0.1663.1.8902.0
00000000-01000540.1663.1.6111.0
00000000-010005D0.1663.1.6110.0

If the FPI comes from a fork other than the main fork, the fork name
will be appended on the output file name; e.g.:

00000000-014A4758.1663.1.12864.0_vm

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).  As such there are two versions of this functionality, one which
returns the raw page as it appears in the WAL (--save-fullpage) and one
which applies the updated pd_lsn and pd_checksum (--save-fullpage-fixup).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          |  73 ++++++++
 src/bin/pg_waldump/pg_waldump.c           | 199 +++++++++++++++++++---
 src/bin/pg_waldump/t/002_save_fullpage.pl | 132 ++++++++++++++
 3 files changed, 379 insertions(+), 25 deletions(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index 57746d9421..a4e1e6388d 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,79 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fullpage=<replaceable>save_path</replaceable></option></term>
+       <term><option>-X <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fullpage-fixup=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        If invoked using
+        the <literal>-X</literal>/<literal>--save-fullpage-fixup</literal>
+        option, this page image will include the <literal>pd_lsn</literal> of
+        the replayed record rather than the raw page image; as well,
+        the <literal>pd_checksum</literal> field will be updated if it had
+        previously existed.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>TSOID</replaceable>.<replaceable>DBOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <table id="pgwaldump-save-fpi-components">
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+            as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+           <row>
+            <entry>TSOID</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+           <row>
+            <entry>DBOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             if coming from the main fork, will be empty, otherwise will be
+             one of <literal>_fsm</literal>, <literal>_vm</literal>,
+             or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </table>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 4f265ef546..dd1035e593 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -24,8 +24,12 @@
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
+#include "storage/checksum.h"
+#include "storage/checksum_impl.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,11 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	bool		save_fpw;
+	bool		fixup_fpw;
+	char	   *save_fpw_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +448,109 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPWs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath, bool fixup)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (XLogRecHasBlockImage(record, block_id))
+		{
+			char		page[BLCKSZ];
+
+			memset(page, 0, BLCKSZ);
+
+			if (RestoreBlockImage(record, block_id, page))
+			{
+				/* we have our extracted FPI, let's save it now */
+				char		filename[MAXPGPATH];
+				char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+				FILE	   *OPF;
+				BlockNumber blk;
+				RelFileNode rnode;
+				ForkNumber	fork;
+
+				XLogRecGetBlockTagExtended(record, block_id,
+										   &rnode, &fork, &blk, NULL);
+
+				/*
+				 * The raw page as stored in the WAL record includes the LSN
+				 * of the block as it appeared when it was originally written,
+				 * however this differs than the effects of replaying this
+				 * same FPI in recovery, as recovery calls RestoreBlockImage()
+				 * and then sets the LSN as part of one action.  What this
+				 * means is that a page as recovered from WAL and the version
+				 * of the page saved here will differ by the LSN and the
+				 * checksum (if enabled).
+				 *
+				 * There are potentially use-cases for both versions (with and
+				 * without mentioned fixups), so allow this to be
+				 * user-selected, unless the restored page was empty, in which
+				 * case we leave it alone.
+				 */
+
+				if (fixup && !PageIsNew(page))
+				{
+					PageSetLSN(page, record->ReadRecPtr);
+
+					/*
+					 * If checksum field is non-zero then we have checksums
+					 * enabled, so recalculate the checksum with new LSN
+					 * (whether this is considered a hack or heuristics is an
+					 * exercise for the reader).
+					 *
+					 * We make this choice to allow pages saved by this
+					 * function to work as expected with the checksum-related
+					 * functions in pageinspect without having to worry about
+					 * zero_damaged_pages or other considerations.
+					 *
+					 * FPIs in WAL do not have the checksum field updated in
+					 * the page image; in a checksums-enabled cluster, this
+					 * task is handled by FlushBuffer() when a dirty buffer is
+					 * written out to disk.  Since we are running outside of
+					 * Postmaster that won't work in this case, so we handle
+					 * ourselves.
+					 */
+
+					if (((PageHeader) page)->pd_checksum)
+						((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blk);
+				}
+
+				if (fork >= 0 && fork <= MAX_FORKNUM)
+				{
+					if (fork)
+						sprintf(forkname, "_%s", forkNames[fork]);
+					else
+						forkname[0] = 0;
+				}
+				else
+					pg_fatal("found invalid fork number: %u", fork);
+
+				snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+						 LSN_FORMAT_ARGS(record->ReadRecPtr),
+						 rnode.spcNode, rnode.dbNode, rnode.relNode, blk, forkname);
+
+				OPF = fopen(filename, PG_BINARY_W);
+				if (!OPF)
+					pg_fatal("couldn't open file for output: %s", filename);
+
+				if (pg_pwrite(fileno(OPF), page, BLCKSZ, 0) != BLCKSZ)
+					pg_fatal("couldn't write out complete fullpage image to file: %s", filename);
+
+				fsync(fileno(OPF));
+				fclose(OPF);
+			}
+		}
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -660,29 +772,32 @@ usage(void)
 	printf(_("Usage:\n"));
 	printf(_("  %s [OPTION]... [STARTSEG [ENDSEG]]\n"), progname);
 	printf(_("\nOptions:\n"));
-	printf(_("  -b, --bkp-details      output detailed information about backup blocks\n"));
-	printf(_("  -B, --block=N          with --relation, only show records that modify block N\n"));
-	printf(_("  -e, --end=RECPTR       stop reading at WAL location RECPTR\n"));
-	printf(_("  -f, --follow           keep retrying after reaching end of WAL\n"));
-	printf(_("  -F, --fork=FORK        only show records that modify blocks in fork FORK;\n"
-			 "                         valid names are main, fsm, vm, init\n"));
-	printf(_("  -n, --limit=N          number of records to display\n"));
-	printf(_("  -p, --path=PATH        directory in which to find log segment files or a\n"
-			 "                         directory with a ./pg_wal that contains such files\n"
-			 "                         (default: current directory, ./pg_wal, $PGDATA/pg_wal)\n"));
-	printf(_("  -q, --quiet            do not print any output, except for errors\n"));
-	printf(_("  -r, --rmgr=RMGR        only show records generated by resource manager RMGR;\n"
-			 "                         use --rmgr=list to list valid resource manager names\n"));
-	printf(_("  -R, --relation=T/D/R   only show records that modify blocks in relation T/D/R\n"));
-	printf(_("  -s, --start=RECPTR     start reading at WAL location RECPTR\n"));
-	printf(_("  -t, --timeline=TLI     timeline from which to read log records\n"
-			 "                         (default: 1 or the value used in STARTSEG)\n"));
-	printf(_("  -V, --version          output version information, then exit\n"));
-	printf(_("  -w, --fullpage         only show records with a full page write\n"));
-	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
-	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
-			 "                         (optionally, show per-record statistics)\n"));
-	printf(_("  -?, --help             show this help, then exit\n"));
+	printf(_("  -b, --bkp-details               output detailed information about backup blocks\n"));
+	printf(_("  -B, --block=N                   with --relation, only show records that modify block N\n"));
+	printf(_("  -e, --end=RECPTR                stop reading at WAL location RECPTR\n"));
+	printf(_("  -f, --follow                    keep retrying after reaching end of WAL\n"));
+	printf(_("  -F, --fork=FORK                 only show records that modify blocks in fork FORK;\n"
+			 "                                  valid names are main, fsm, vm, init\n"));
+	printf(_("  -n, --limit=N                   number of records to display\n"));
+	printf(_("  -p, --path=PATH                 directory in which to find log segment files or a\n"
+			 "                                  directory with a ./pg_wal that contains such files\n"
+			 "                                  (default: current directory, ./pg_wal, $PGDATA/pg_wal)\n"));
+	printf(_("  -q, --quiet                     do not print any output, except for errors\n"));
+	printf(_("  -r, --rmgr=RMGR                 only show records generated by resource manager RMGR;\n"
+			 "                                  use --rmgr=list to list valid resource manager names\n"));
+	printf(_("  -R, --relation=T/D/R            only show records that modify blocks in relation T/D/R\n"));
+	printf(_("  -s, --start=RECPTR              start reading at WAL location RECPTR\n"));
+	printf(_("  -t, --timeline=TLI              timeline from which to read log records\n"
+			 "                                  (default: 1 or the value used in STARTSEG)\n"));
+	printf(_("  -V, --version                   output version information, then exit\n"));
+	printf(_("  -w, --fullpage                  only show records with a full page write\n"));
+	printf(_("  -W, --save-fullpage=path        save found full page images to given path\n"
+			 "                                  as LSN.T.D.R.B_F\n"));
+	printf(_("  -X, --save-fullpage-fixup=path  like -W, but include page-level fixups in saved page\n"));
+	printf(_("  -x, --xid=XID                   only show records with transaction ID XID\n"));
+	printf(_("  -z, --stats[=record]            show statistics instead of records\n"
+			 "                                  (optionally, show per-record statistics)\n"));
+	printf(_("  -?, --help                      show this help, then exit\n"));
 	printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
 	printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
 }
@@ -695,7 +810,7 @@ main(int argc, char **argv)
 	XLogReaderState *xlogreader_state;
 	XLogDumpPrivate private;
 	XLogDumpConfig config;
-	XLogStats stats;
+	XLogStats	stats;
 	XLogRecord *record;
 	XLogRecPtr	first_record;
 	char	   *waldir = NULL;
@@ -712,6 +827,8 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fullpage", required_argument, NULL, 'W'},
+		{"save-fullpage-fixup", required_argument, NULL, 'X'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +889,9 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpw = false;
+	config.fixup_fpw = false;
+	config.save_fpw_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +902,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:X:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -919,6 +1039,12 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+			case 'X':
+				config.save_fpw = true;
+				config.fixup_fpw = (option == 'X');
+				config.save_fpw_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1098,24 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpw_path != NULL)
+	{
+		int			dir_status = pg_check_dir(config.save_fpw_path);
+
+		if (dir_status < 0)
+		{
+			pg_log_error("could not access output directory: %s", config.save_fpw_path);
+			goto bad_argument;
+		}
+
+		if (!dir_status && mkdir(config.save_fpw_path, 0700) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpw_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1294,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpw)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPWs(xlogreader_state, config.save_fpw_path, config.fixup_fpw);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..55aa9afb22
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,132 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Utils;
+use PostgreSQL::Test::Cluster;
+use Test::More;
+use File::Basename;
+
+# routine to extract the LSN and checksum from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if 8192 != read $fh, $block, 8192;
+	my ($lsn_hi, $lsn_lo, $checksum) = unpack('VVv', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo, $checksum);
+}
+
+# Set umask so test directories and files are created with default permissions
+umask(0077);
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init('-k');
+$primary->adjust_conf('postgresql.conf', 'max_wal_size', '100MB');
+$primary->start;
+
+# Sanity checks for command line options.
+$primary->command_fails(
+	[ 'pg_waldump', '--save-fullpage' ],
+	'--save-fullpage fails without path');
+$primary->command_fails(
+	[ 'pg_waldump', '--save-fullpage-fixup' ],
+	'--save-fullpage-fixup fails without path');
+
+# generate data/wal to examine
+$primary->safe_psql('postgres', q(CREATE DATABASE db1));
+$primary->safe_psql('db1',      <<EOF);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT;
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+EOF
+
+# get the relation node, etc for the new table
+my $relation = $primary->safe_psql('db1',
+	q{SELECT CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END || '/' || pg_database.oid || '/' || relfilenode FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
+);
+
+diag $relation;
+
+$primary->stop;
+my $waldir = $primary->basedir . '/pgdata/pg_wal';
+system("rsync -av " . $primary->basedir . " /tmp");
+diag "using waldir: $waldir";
+diag "waldir content: @{[qx/ls -l $waldir/]}";
+#my $walfile = [glob("$waldir/*")]->[0];
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+#diag "using walfile: $walfile";
+
+# # extract files
+system(
+	"pg_waldump --save-fullpage $tmp_folder/raw --relation $relation $waldir/000* || true"
+);
+system(
+	"pg_waldump --save-fullpage-fixup $tmp_folder/fixup --relation $relation $waldir/000* || true"
+);
+system("rsync -av " . $tmp_folder . " /tmp/pgtesttmp/");
+
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm)?$/;
+
+my %checksums;
+my %files;
+
+# verify filename formats matches w/--save-fullpage
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since no fixup, verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+
+	# stash checksum for later comparisons
+	$checksums{$file} = $checksum;
+}
+
+# verify filename formats matches w/--save-fullpage-fixup
+for my $fullpath (glob "$tmp_folder/fixup/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since fixup, verify the lsn in the block equals file lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn eq $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN is the same as block-based one');
+
+	# verify checksum change; XXX: there could be valid clashes here,
+	# just validate that the page matches the expected checksum instead?
+	ok( $checksum == 0 || $checksums{$file} != $checksum,
+		'check for checksum change or no checksum');
+}
+
+# validate that we ended up with some files output and they were the same
+ok(%files > 0, 'verify we processed some files');
+ok((grep { $_ != 2 } values %files) == 0,
+	'ensure raw and fixup had same number of files');
+
+done_testing();
-- 
2.32.0 (Apple Git-132)

From acfd94c3db2497a8f690ed0c700d0dd4b65ead99 Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Wed, 20 Apr 2022 19:59:35 -0500
Subject: [PATCH] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a target directory,
which must be empty or not exist.  These images are subject to the same
filtering rules as normal display in pg_waldump, which means that you
can isolate the full page writes to a target relation, among other
things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0
00000000-01000150.1664.0.6115.0
00000000-010001E0.1664.0.6114.0
00000000-01000270.1663.1.6116.0
00000000-01000300.1663.1.6113.0
00000000-01000390.1663.1.6112.0
00000000-01000420.1663.1.8903.0
00000000-010004B0.1663.1.8902.0
00000000-01000540.1663.1.6111.0
00000000-010005D0.1663.1.6110.0

If the FPI comes from a fork other than the main fork, the fork name
will be appended on the output file name; e.g.:

00000000-014A4758.1663.1.12864.0_vm

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).  As such there are two versions of this functionality, one which
returns the raw page as it appears in the WAL (--save-fullpage) and one
which applies the updated pd_lsn and pd_checksum (--save-fullpage-fixup).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          |  79 +++++++++
 src/bin/pg_waldump/pg_waldump.c           | 199 +++++++++++++++++++---
 src/bin/pg_waldump/t/002_save_fullpage.pl | 132 ++++++++++++++
 3 files changed, 385 insertions(+), 25 deletions(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index 57746d9421..2bd6062069 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,85 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fullpage=<replaceable>save_path</replaceable></option></term>
+       <term><option>-X <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fullpage-fixup=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        If invoked using
+        the <literal>-X</literal>/<literal>--save-fullpage-fixup</literal>
+        option, this page image will include the <literal>pd_lsn</literal> of
+        the replayed record rather than the raw page image; as well,
+        the <literal>pd_checksum</literal> field will be updated if it had
+        previously existed.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>TSOID</replaceable>.<replaceable>DBOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>TSOID</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DBOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             if coming from the main fork, will be empty, otherwise will be
+             one of <literal>_fsm</literal>, <literal>_vm</literal>,
+             or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 4f265ef546..dd1035e593 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -24,8 +24,12 @@
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
+#include "storage/checksum.h"
+#include "storage/checksum_impl.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,11 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	bool		save_fpw;
+	bool		fixup_fpw;
+	char	   *save_fpw_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +448,109 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPWs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath, bool fixup)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (XLogRecHasBlockImage(record, block_id))
+		{
+			char		page[BLCKSZ];
+
+			memset(page, 0, BLCKSZ);
+
+			if (RestoreBlockImage(record, block_id, page))
+			{
+				/* we have our extracted FPI, let's save it now */
+				char		filename[MAXPGPATH];
+				char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+				FILE	   *OPF;
+				BlockNumber blk;
+				RelFileNode rnode;
+				ForkNumber	fork;
+
+				XLogRecGetBlockTagExtended(record, block_id,
+										   &rnode, &fork, &blk, NULL);
+
+				/*
+				 * The raw page as stored in the WAL record includes the LSN
+				 * of the block as it appeared when it was originally written,
+				 * however this differs than the effects of replaying this
+				 * same FPI in recovery, as recovery calls RestoreBlockImage()
+				 * and then sets the LSN as part of one action.  What this
+				 * means is that a page as recovered from WAL and the version
+				 * of the page saved here will differ by the LSN and the
+				 * checksum (if enabled).
+				 *
+				 * There are potentially use-cases for both versions (with and
+				 * without mentioned fixups), so allow this to be
+				 * user-selected, unless the restored page was empty, in which
+				 * case we leave it alone.
+				 */
+
+				if (fixup && !PageIsNew(page))
+				{
+					PageSetLSN(page, record->ReadRecPtr);
+
+					/*
+					 * If checksum field is non-zero then we have checksums
+					 * enabled, so recalculate the checksum with new LSN
+					 * (whether this is considered a hack or heuristics is an
+					 * exercise for the reader).
+					 *
+					 * We make this choice to allow pages saved by this
+					 * function to work as expected with the checksum-related
+					 * functions in pageinspect without having to worry about
+					 * zero_damaged_pages or other considerations.
+					 *
+					 * FPIs in WAL do not have the checksum field updated in
+					 * the page image; in a checksums-enabled cluster, this
+					 * task is handled by FlushBuffer() when a dirty buffer is
+					 * written out to disk.  Since we are running outside of
+					 * Postmaster that won't work in this case, so we handle
+					 * ourselves.
+					 */
+
+					if (((PageHeader) page)->pd_checksum)
+						((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blk);
+				}
+
+				if (fork >= 0 && fork <= MAX_FORKNUM)
+				{
+					if (fork)
+						sprintf(forkname, "_%s", forkNames[fork]);
+					else
+						forkname[0] = 0;
+				}
+				else
+					pg_fatal("found invalid fork number: %u", fork);
+
+				snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+						 LSN_FORMAT_ARGS(record->ReadRecPtr),
+						 rnode.spcNode, rnode.dbNode, rnode.relNode, blk, forkname);
+
+				OPF = fopen(filename, PG_BINARY_W);
+				if (!OPF)
+					pg_fatal("couldn't open file for output: %s", filename);
+
+				if (pg_pwrite(fileno(OPF), page, BLCKSZ, 0) != BLCKSZ)
+					pg_fatal("couldn't write out complete fullpage image to file: %s", filename);
+
+				fsync(fileno(OPF));
+				fclose(OPF);
+			}
+		}
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -660,29 +772,32 @@ usage(void)
 	printf(_("Usage:\n"));
 	printf(_("  %s [OPTION]... [STARTSEG [ENDSEG]]\n"), progname);
 	printf(_("\nOptions:\n"));
-	printf(_("  -b, --bkp-details      output detailed information about backup blocks\n"));
-	printf(_("  -B, --block=N          with --relation, only show records that modify block N\n"));
-	printf(_("  -e, --end=RECPTR       stop reading at WAL location RECPTR\n"));
-	printf(_("  -f, --follow           keep retrying after reaching end of WAL\n"));
-	printf(_("  -F, --fork=FORK        only show records that modify blocks in fork FORK;\n"
-			 "                         valid names are main, fsm, vm, init\n"));
-	printf(_("  -n, --limit=N          number of records to display\n"));
-	printf(_("  -p, --path=PATH        directory in which to find log segment files or a\n"
-			 "                         directory with a ./pg_wal that contains such files\n"
-			 "                         (default: current directory, ./pg_wal, $PGDATA/pg_wal)\n"));
-	printf(_("  -q, --quiet            do not print any output, except for errors\n"));
-	printf(_("  -r, --rmgr=RMGR        only show records generated by resource manager RMGR;\n"
-			 "                         use --rmgr=list to list valid resource manager names\n"));
-	printf(_("  -R, --relation=T/D/R   only show records that modify blocks in relation T/D/R\n"));
-	printf(_("  -s, --start=RECPTR     start reading at WAL location RECPTR\n"));
-	printf(_("  -t, --timeline=TLI     timeline from which to read log records\n"
-			 "                         (default: 1 or the value used in STARTSEG)\n"));
-	printf(_("  -V, --version          output version information, then exit\n"));
-	printf(_("  -w, --fullpage         only show records with a full page write\n"));
-	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
-	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
-			 "                         (optionally, show per-record statistics)\n"));
-	printf(_("  -?, --help             show this help, then exit\n"));
+	printf(_("  -b, --bkp-details               output detailed information about backup blocks\n"));
+	printf(_("  -B, --block=N                   with --relation, only show records that modify block N\n"));
+	printf(_("  -e, --end=RECPTR                stop reading at WAL location RECPTR\n"));
+	printf(_("  -f, --follow                    keep retrying after reaching end of WAL\n"));
+	printf(_("  -F, --fork=FORK                 only show records that modify blocks in fork FORK;\n"
+			 "                                  valid names are main, fsm, vm, init\n"));
+	printf(_("  -n, --limit=N                   number of records to display\n"));
+	printf(_("  -p, --path=PATH                 directory in which to find log segment files or a\n"
+			 "                                  directory with a ./pg_wal that contains such files\n"
+			 "                                  (default: current directory, ./pg_wal, $PGDATA/pg_wal)\n"));
+	printf(_("  -q, --quiet                     do not print any output, except for errors\n"));
+	printf(_("  -r, --rmgr=RMGR                 only show records generated by resource manager RMGR;\n"
+			 "                                  use --rmgr=list to list valid resource manager names\n"));
+	printf(_("  -R, --relation=T/D/R            only show records that modify blocks in relation T/D/R\n"));
+	printf(_("  -s, --start=RECPTR              start reading at WAL location RECPTR\n"));
+	printf(_("  -t, --timeline=TLI              timeline from which to read log records\n"
+			 "                                  (default: 1 or the value used in STARTSEG)\n"));
+	printf(_("  -V, --version                   output version information, then exit\n"));
+	printf(_("  -w, --fullpage                  only show records with a full page write\n"));
+	printf(_("  -W, --save-fullpage=path        save found full page images to given path\n"
+			 "                                  as LSN.T.D.R.B_F\n"));
+	printf(_("  -X, --save-fullpage-fixup=path  like -W, but include page-level fixups in saved page\n"));
+	printf(_("  -x, --xid=XID                   only show records with transaction ID XID\n"));
+	printf(_("  -z, --stats[=record]            show statistics instead of records\n"
+			 "                                  (optionally, show per-record statistics)\n"));
+	printf(_("  -?, --help                      show this help, then exit\n"));
 	printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
 	printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
 }
@@ -695,7 +810,7 @@ main(int argc, char **argv)
 	XLogReaderState *xlogreader_state;
 	XLogDumpPrivate private;
 	XLogDumpConfig config;
-	XLogStats stats;
+	XLogStats	stats;
 	XLogRecord *record;
 	XLogRecPtr	first_record;
 	char	   *waldir = NULL;
@@ -712,6 +827,8 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fullpage", required_argument, NULL, 'W'},
+		{"save-fullpage-fixup", required_argument, NULL, 'X'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +889,9 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpw = false;
+	config.fixup_fpw = false;
+	config.save_fpw_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +902,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:X:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -919,6 +1039,12 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+			case 'X':
+				config.save_fpw = true;
+				config.fixup_fpw = (option == 'X');
+				config.save_fpw_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1098,24 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpw_path != NULL)
+	{
+		int			dir_status = pg_check_dir(config.save_fpw_path);
+
+		if (dir_status < 0)
+		{
+			pg_log_error("could not access output directory: %s", config.save_fpw_path);
+			goto bad_argument;
+		}
+
+		if (!dir_status && mkdir(config.save_fpw_path, 0700) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpw_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1294,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpw)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPWs(xlogreader_state, config.save_fpw_path, config.fixup_fpw);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..bf2389abf8
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,132 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Utils;
+use PostgreSQL::Test::Cluster;
+use Test::More;
+use File::Basename;
+
+# routine to extract the LSN and checksum from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if 8192 != read $fh, $block, 8192;
+	my ($lsn_hi, $lsn_lo, $checksum) = unpack('VVv', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo, $checksum);
+}
+
+# Set umask so test directories and files are created with default permissions
+umask(0077);
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init('-k');
+$primary->adjust_conf('postgresql.conf', 'max_wal_size', '100MB');
+$primary->start;
+
+# Sanity checks for command line options.
+$primary->command_fails(
+	[ 'pg_waldump', '--save-fullpage' ],
+	'--save-fullpage fails without path');
+$primary->command_fails(
+	[ 'pg_waldump', '--save-fullpage-fixup' ],
+	'--save-fullpage-fixup fails without path');
+
+# generate data/wal to examine
+$primary->safe_psql('postgres', q(CREATE DATABASE db1));
+$primary->safe_psql('db1',      <<EOF);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT;
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+EOF
+
+# get the relation node, etc for the new table
+my $relation = $primary->safe_psql('db1',
+	q{SELECT CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END || '/' || pg_database.oid || '/' || relfilenode FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
+);
+
+diag $relation;
+
+$primary->stop;
+my $waldir = $primary->basedir . '/pgdata/pg_wal';
+system("rsync -av " . $primary->basedir . " /tmp");
+diag "using waldir: $waldir";
+diag "waldir content: @{[qx/ls -l $waldir/]}";
+#my $walfile = [glob("$waldir/*")]->[0];
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+#diag "using walfile: $walfile";
+
+# # extract files
+system(
+	"pg_waldump --save-fullpage $tmp_folder/raw --relation $relation $waldir/000* || true"
+);
+system(
+	"pg_waldump --save-fullpage-fixup $tmp_folder/fixup --relation $relation $waldir/000* || true"
+);
+system("rsync -av " . $tmp_folder . " /tmp/pgtesttmp/");
+
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm)?$/;
+
+my %checksums;
+my %files;
+
+# verify filename formats matches w/--save-fullpage
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since no fixup, verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+
+	# stash checksum for later comparisons
+	$checksums{$file} = $checksum;
+}
+
+# verify filename formats matches w/--save-fullpage-fixup
+for my $fullpath (glob "$tmp_folder/fixup/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since fixup, verify the lsn in the block equals file lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn eq $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN is the same as block-based one');
+
+	# verify checksum change; XXX: there could be valid clashes here,
+	# just validate that the page matches the expected checksum instead?
+	ok( $checksum == 0 || $checksums{$file} != $checksum,
+		'check for checksum change or no checksum');
+}
+
+# validate that we ended up with some files output and they were the same
+ok(keys %files > 0, 'verify we processed some files');
+ok((grep { $_ != 2 } values %files) == 0,
+	'ensure raw and fixup had same number of files');
+
+done_testing();
-- 
2.32.0 (Apple Git-132)

From 92ecf6e3f2b2f5ae5224daa063b784444c40f906 Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Wed, 20 Apr 2022 19:59:35 -0500
Subject: [PATCH v5] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a target directory,
which must be empty or not exist.  These images are subject to the same
filtering rules as normal display in pg_waldump, which means that you
can isolate the full page writes to a target relation, among other
things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0
00000000-01000150.1664.0.6115.0
00000000-010001E0.1664.0.6114.0
00000000-01000270.1663.1.6116.0
00000000-01000300.1663.1.6113.0
00000000-01000390.1663.1.6112.0
00000000-01000420.1663.1.8903.0
00000000-010004B0.1663.1.8902.0
00000000-01000540.1663.1.6111.0
00000000-010005D0.1663.1.6110.0

If the FPI comes from a fork other than the main fork, the fork name
will be appended on the output file name; e.g.:

00000000-014A4758.1663.1.12864.0_vm

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).  As such there are two versions of this functionality, one which
returns the raw page as it appears in the WAL (--save-fpi) and one
which applies the updated pd_lsn and pd_checksum (--fixup-fpi).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          |  79 +++++++++++
 src/bin/pg_waldump/pg_waldump.c           | 151 +++++++++++++++++++++-
 src/bin/pg_waldump/t/002_save_fullpage.pl | 137 ++++++++++++++++++++
 3 files changed, 366 insertions(+), 1 deletion(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5..cc3ce91890 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,85 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fpi=<replaceable>save_path</replaceable></option></term>
+       <term><option>-X <replaceable>save_path</replaceable></option></term>
+       <term><option>--fixup-fpi=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        If invoked using
+        the <literal>-X</literal>/<literal>--fixup-fpi</literal>
+        option, this page image will include the <literal>pd_lsn</literal> of
+        the replayed record rather than the raw page image; as well,
+        the <literal>pd_checksum</literal> field will be updated if it had
+        previously existed.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>TSOID</replaceable>.<replaceable>DBOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>TSOID</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DBOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             if coming from the main fork, will be empty, otherwise will be
+             one of <literal>_fsm</literal>, <literal>_vm</literal>,
+             or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..2b6f772dc7 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -24,8 +24,12 @@
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
+#include "storage/checksum.h"
+#include "storage/checksum_impl.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,11 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	bool		save_fpw;
+	bool		fixup_fpw;
+	char	   *save_fpw_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +448,109 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPWs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath, bool fixup)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (XLogRecHasBlockImage(record, block_id))
+		{
+			char		page[BLCKSZ];
+
+			memset(page, 0, BLCKSZ);
+
+			if (RestoreBlockImage(record, block_id, page))
+			{
+				/* we have our extracted FPI, let's save it now */
+				char		filename[MAXPGPATH];
+				char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+				FILE	   *OPF;
+				BlockNumber blk;
+				RelFileLocator rnode;
+				ForkNumber	fork;
+
+				XLogRecGetBlockTagExtended(record, block_id,
+										   &rnode, &fork, &blk, NULL);
+
+				/*
+				 * The raw page as stored in the WAL record includes the LSN
+				 * of the block as it appeared when it was originally written,
+				 * however this differs than the effects of replaying this
+				 * same FPI in recovery, as recovery calls RestoreBlockImage()
+				 * and then sets the LSN as part of one action.  What this
+				 * means is that a page as recovered from WAL and the version
+				 * of the page saved here will differ by the LSN and the
+				 * checksum (if enabled).
+				 *
+				 * There are potentially use-cases for both versions (with and
+				 * without mentioned fixups), so allow this to be
+				 * user-selected, unless the restored page was empty, in which
+				 * case we leave it alone.
+				 */
+
+				if (fixup && !PageIsNew(page))
+				{
+					PageSetLSN(page, record->ReadRecPtr);
+
+					/*
+					 * If checksum field is non-zero then we have checksums
+					 * enabled, so recalculate the checksum with new LSN
+					 * (whether this is considered a hack or heuristics is an
+					 * exercise for the reader).
+					 *
+					 * We make this choice to allow pages saved by this
+					 * function to work as expected with the checksum-related
+					 * functions in pageinspect without having to worry about
+					 * zero_damaged_pages or other considerations.
+					 *
+					 * FPIs in WAL do not have the checksum field updated in
+					 * the page image; in a checksums-enabled cluster, this
+					 * task is handled by FlushBuffer() when a dirty buffer is
+					 * written out to disk.  Since we are running outside of
+					 * Postmaster that won't work in this case, so we handle
+					 * ourselves.
+					 */
+
+					if (((PageHeader) page)->pd_checksum)
+						((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blk);
+				}
+
+				if (fork >= 0 && fork <= MAX_FORKNUM)
+				{
+					if (fork)
+						sprintf(forkname, "_%s", forkNames[fork]);
+					else
+						forkname[0] = 0;
+				}
+				else
+					pg_fatal("found invalid fork number: %u", fork);
+
+				snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+						 LSN_FORMAT_ARGS(record->ReadRecPtr),
+						 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+				OPF = fopen(filename, PG_BINARY_W);
+				if (!OPF)
+					pg_fatal("couldn't open file for output: %s", filename);
+
+				if (pg_pwrite(fileno(OPF), page, BLCKSZ, 0) != BLCKSZ)
+					pg_fatal("couldn't write out complete fullpage image to file: %s", filename);
+
+				fsync(fileno(OPF));
+				fclose(OPF);
+			}
+		}
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +791,9 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --save-fpi=path    save found full page images to given path\n"
+			 "                         as LSN.T.D.R.B_F\n"));
+	printf(_("  -X, --fixup-fpi=path   like --save-fpi but apply LSN fixups to \n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +827,8 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fpi", required_argument, NULL, 'W'},
+		{"fixup-fpi", required_argument, NULL, 'X'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +889,9 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpw = false;
+	config.fixup_fpw = false;
+	config.save_fpw_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +902,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:X:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -918,6 +1038,12 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+			case 'X':
+				config.save_fpw = true;
+				config.fixup_fpw = (option == 'X');
+				config.save_fpw_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1098,24 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpw_path != NULL)
+	{
+		int			dir_status = pg_check_dir(config.save_fpw_path);
+
+		if (dir_status < 0)
+		{
+			pg_log_error("could not access output directory: %s", config.save_fpw_path);
+			goto bad_argument;
+		}
+
+		if (!dir_status && mkdir(config.save_fpw_path, 0700) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpw_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1294,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpw)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPWs(xlogreader_state, config.save_fpw_path, config.fixup_fpw);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..f605b4f6ad
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,137 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# routine to extract the LSN and checksum from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if 8192 != read $fh, $block, 8192;
+	my ($lsn_hi, $lsn_lo, $checksum) = unpack('VVv', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo, $checksum);
+}
+
+# Set umask so test directories and files are created with default permissions
+umask(0077);
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init('-k');
+$primary->append_conf('postgresql.conf', "max_wal_size='100MB'");
+$primary->append_conf('postgresql.conf', "wal_level='replica'");
+$primary->append_conf('postgresql.conf', 'archive_mode=on');
+$primary->append_conf('postgresql.conf', "archive_command='/bin/false'");
+$primary->start;
+
+# Sanity checks for command line options.
+$primary->command_fails(
+	[ 'pg_waldump', '--save-fpi' ],
+	'--save-fpi fails without path');
+$primary->command_fails(
+	[ 'pg_waldump', '--fixup-fpi' ],
+	'--fixup-fpi fails without path');
+
+# generate data/wal to examine
+$primary->safe_psql('postgres', q(CREATE DATABASE db1));
+$primary->safe_psql('db1',      <<EOF);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT;
+SELECT pg_switch_wal();
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+SELECT pg_switch_wal();
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+SELECT pg_switch_wal();
+EOF
+
+# get the relation node, etc for the new table
+my $relation = $primary->safe_psql('db1',
+	q{SELECT CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END || '/' || pg_database.oid || '/' || relfilenode FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
+);
+
+diag $relation;
+
+$primary->stop;
+my $waldir = $primary->basedir . '/pgdata/pg_wal';
+my $walfile = [glob("$waldir/*")]->[2]; # we want the 00000002 file
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+diag "using walfile: $walfile";
+
+# # # extract files
+# PostgreSQL::Test::RecursiveCopy::copypath(
+# 	$primary->data_dir . "/pg_wal",
+# 	$primary->tmp_folder);
+
+$primary->command_ok(['pg_waldump', '--save-fpi', "$tmp_folder/raw", '--relation', $relation, $walfile]);
+$primary->command_ok(['pg_waldump', '--fixup-fpi', "$tmp_folder/fixup", '--relation', $relation, $walfile]);
+
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm)?$/;
+
+my %checksums;
+my %files;
+
+# verify filename formats matches w/--save-fpi
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since no fixup, verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+
+	# stash checksum for later comparisons
+	$checksums{$file} = $checksum;
+}
+
+# verify filename formats matches w/--fixup-fpi
+for my $fullpath (glob "$tmp_folder/fixup/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since fixup, verify the lsn in the block equals file lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn eq $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN is the same as block-based one');
+
+	# verify checksum change; XXX: there could be valid clashes here,
+	# just validate that the page matches the expected checksum instead?
+	ok( $checksum == 0 || $checksums{$file} != $checksum,
+		'check for checksum change or no checksum');
+}
+
+# validate that we ended up with some files output and they were the same
+ok(keys %files > 0, 'verify we processed some files');
+ok((grep { $_ != 2 } values %files) == 0,
+	'ensure raw and fixup had same number of files');
+
+done_testing();
-- 
2.37.1 (Apple Git-137.1)

From d4cb23bc2f0edcc65b0bef75fd2e8d6e5aeda21f Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Wed, 20 Apr 2022 19:59:35 -0500
Subject: [PATCH 1/2] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a target directory,
which must be empty or not exist.  These images are subject to the same
filtering rules as normal display in pg_waldump, which means that you
can isolate the full page writes to a target relation, among other
things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0
00000000-01000150.1664.0.6115.0
00000000-010001E0.1664.0.6114.0
00000000-01000270.1663.1.6116.0
00000000-01000300.1663.1.6113.0
00000000-01000390.1663.1.6112.0
00000000-01000420.1663.1.8903.0
00000000-010004B0.1663.1.8902.0
00000000-01000540.1663.1.6111.0
00000000-010005D0.1663.1.6110.0

If the FPI comes from a fork other than the main fork, the fork name
will be appended on the output file name; e.g.:

00000000-014A4758.1663.1.12864.0_vm

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).  As such there are two versions of this functionality, one which
returns the raw page as it appears in the WAL (--save-fpi) and one
which applies the updated pd_lsn and pd_checksum (--fixup-fpi).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          |  79 +++++++++++
 src/bin/pg_waldump/pg_waldump.c           | 151 +++++++++++++++++++++-
 src/bin/pg_waldump/t/002_save_fullpage.pl | 137 ++++++++++++++++++++
 3 files changed, 366 insertions(+), 1 deletion(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5f..cc3ce918905 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,85 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fpi=<replaceable>save_path</replaceable></option></term>
+       <term><option>-X <replaceable>save_path</replaceable></option></term>
+       <term><option>--fixup-fpi=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        If invoked using
+        the <literal>-X</literal>/<literal>--fixup-fpi</literal>
+        option, this page image will include the <literal>pd_lsn</literal> of
+        the replayed record rather than the raw page image; as well,
+        the <literal>pd_checksum</literal> field will be updated if it had
+        previously existed.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>TSOID</replaceable>.<replaceable>DBOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>TSOID</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DBOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             if coming from the main fork, will be empty, otherwise will be
+             one of <literal>_fsm</literal>, <literal>_vm</literal>,
+             or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca58..2b6f772dc71 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -24,8 +24,12 @@
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
+#include "storage/checksum.h"
+#include "storage/checksum_impl.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,11 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	bool		save_fpw;
+	bool		fixup_fpw;
+	char	   *save_fpw_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +448,109 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPWs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath, bool fixup)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (XLogRecHasBlockImage(record, block_id))
+		{
+			char		page[BLCKSZ];
+
+			memset(page, 0, BLCKSZ);
+
+			if (RestoreBlockImage(record, block_id, page))
+			{
+				/* we have our extracted FPI, let's save it now */
+				char		filename[MAXPGPATH];
+				char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+				FILE	   *OPF;
+				BlockNumber blk;
+				RelFileLocator rnode;
+				ForkNumber	fork;
+
+				XLogRecGetBlockTagExtended(record, block_id,
+										   &rnode, &fork, &blk, NULL);
+
+				/*
+				 * The raw page as stored in the WAL record includes the LSN
+				 * of the block as it appeared when it was originally written,
+				 * however this differs than the effects of replaying this
+				 * same FPI in recovery, as recovery calls RestoreBlockImage()
+				 * and then sets the LSN as part of one action.  What this
+				 * means is that a page as recovered from WAL and the version
+				 * of the page saved here will differ by the LSN and the
+				 * checksum (if enabled).
+				 *
+				 * There are potentially use-cases for both versions (with and
+				 * without mentioned fixups), so allow this to be
+				 * user-selected, unless the restored page was empty, in which
+				 * case we leave it alone.
+				 */
+
+				if (fixup && !PageIsNew(page))
+				{
+					PageSetLSN(page, record->ReadRecPtr);
+
+					/*
+					 * If checksum field is non-zero then we have checksums
+					 * enabled, so recalculate the checksum with new LSN
+					 * (whether this is considered a hack or heuristics is an
+					 * exercise for the reader).
+					 *
+					 * We make this choice to allow pages saved by this
+					 * function to work as expected with the checksum-related
+					 * functions in pageinspect without having to worry about
+					 * zero_damaged_pages or other considerations.
+					 *
+					 * FPIs in WAL do not have the checksum field updated in
+					 * the page image; in a checksums-enabled cluster, this
+					 * task is handled by FlushBuffer() when a dirty buffer is
+					 * written out to disk.  Since we are running outside of
+					 * Postmaster that won't work in this case, so we handle
+					 * ourselves.
+					 */
+
+					if (((PageHeader) page)->pd_checksum)
+						((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blk);
+				}
+
+				if (fork >= 0 && fork <= MAX_FORKNUM)
+				{
+					if (fork)
+						sprintf(forkname, "_%s", forkNames[fork]);
+					else
+						forkname[0] = 0;
+				}
+				else
+					pg_fatal("found invalid fork number: %u", fork);
+
+				snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+						 LSN_FORMAT_ARGS(record->ReadRecPtr),
+						 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+				OPF = fopen(filename, PG_BINARY_W);
+				if (!OPF)
+					pg_fatal("couldn't open file for output: %s", filename);
+
+				if (pg_pwrite(fileno(OPF), page, BLCKSZ, 0) != BLCKSZ)
+					pg_fatal("couldn't write out complete fullpage image to file: %s", filename);
+
+				fsync(fileno(OPF));
+				fclose(OPF);
+			}
+		}
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +791,9 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --save-fpi=path    save found full page images to given path\n"
+			 "                         as LSN.T.D.R.B_F\n"));
+	printf(_("  -X, --fixup-fpi=path   like --save-fpi but apply LSN fixups to \n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +827,8 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fpi", required_argument, NULL, 'W'},
+		{"fixup-fpi", required_argument, NULL, 'X'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +889,9 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpw = false;
+	config.fixup_fpw = false;
+	config.save_fpw_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +902,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:X:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -918,6 +1038,12 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+			case 'X':
+				config.save_fpw = true;
+				config.fixup_fpw = (option == 'X');
+				config.save_fpw_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1098,24 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpw_path != NULL)
+	{
+		int			dir_status = pg_check_dir(config.save_fpw_path);
+
+		if (dir_status < 0)
+		{
+			pg_log_error("could not access output directory: %s", config.save_fpw_path);
+			goto bad_argument;
+		}
+
+		if (!dir_status && mkdir(config.save_fpw_path, 0700) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpw_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1294,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpw)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPWs(xlogreader_state, config.save_fpw_path, config.fixup_fpw);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 00000000000..f605b4f6adb
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,137 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# routine to extract the LSN and checksum from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if 8192 != read $fh, $block, 8192;
+	my ($lsn_hi, $lsn_lo, $checksum) = unpack('VVv', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo, $checksum);
+}
+
+# Set umask so test directories and files are created with default permissions
+umask(0077);
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init('-k');
+$primary->append_conf('postgresql.conf', "max_wal_size='100MB'");
+$primary->append_conf('postgresql.conf', "wal_level='replica'");
+$primary->append_conf('postgresql.conf', 'archive_mode=on');
+$primary->append_conf('postgresql.conf', "archive_command='/bin/false'");
+$primary->start;
+
+# Sanity checks for command line options.
+$primary->command_fails(
+	[ 'pg_waldump', '--save-fpi' ],
+	'--save-fpi fails without path');
+$primary->command_fails(
+	[ 'pg_waldump', '--fixup-fpi' ],
+	'--fixup-fpi fails without path');
+
+# generate data/wal to examine
+$primary->safe_psql('postgres', q(CREATE DATABASE db1));
+$primary->safe_psql('db1',      <<EOF);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT;
+SELECT pg_switch_wal();
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+SELECT pg_switch_wal();
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+SELECT pg_switch_wal();
+EOF
+
+# get the relation node, etc for the new table
+my $relation = $primary->safe_psql('db1',
+	q{SELECT CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END || '/' || pg_database.oid || '/' || relfilenode FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
+);
+
+diag $relation;
+
+$primary->stop;
+my $waldir = $primary->basedir . '/pgdata/pg_wal';
+my $walfile = [glob("$waldir/*")]->[2]; # we want the 00000002 file
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+diag "using walfile: $walfile";
+
+# # # extract files
+# PostgreSQL::Test::RecursiveCopy::copypath(
+# 	$primary->data_dir . "/pg_wal",
+# 	$primary->tmp_folder);
+
+$primary->command_ok(['pg_waldump', '--save-fpi', "$tmp_folder/raw", '--relation', $relation, $walfile]);
+$primary->command_ok(['pg_waldump', '--fixup-fpi', "$tmp_folder/fixup", '--relation', $relation, $walfile]);
+
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm)?$/;
+
+my %checksums;
+my %files;
+
+# verify filename formats matches w/--save-fpi
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since no fixup, verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+
+	# stash checksum for later comparisons
+	$checksums{$file} = $checksum;
+}
+
+# verify filename formats matches w/--fixup-fpi
+for my $fullpath (glob "$tmp_folder/fixup/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since fixup, verify the lsn in the block equals file lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn eq $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN is the same as block-based one');
+
+	# verify checksum change; XXX: there could be valid clashes here,
+	# just validate that the page matches the expected checksum instead?
+	ok( $checksum == 0 || $checksums{$file} != $checksum,
+		'check for checksum change or no checksum');
+}
+
+# validate that we ended up with some files output and they were the same
+ok(keys %files > 0, 'verify we processed some files');
+ok((grep { $_ != 2 } values %files) == 0,
+	'ensure raw and fixup had same number of files');
+
+done_testing();
-- 
2.25.1

From 3c2c369c495760d8f96d3cdeab623c4f0372792a Mon Sep 17 00:00:00 2001
From: Justin Pryzby <pryzbyj@telsasoft.com>
Date: Thu, 15 Sep 2022 13:39:53 -0500
Subject: [PATCH 2/2] f!justin

ci-os-only: freebsd
---
 src/bin/pg_waldump/meson.build            |   1 +
 src/bin/pg_waldump/pg_waldump.c           | 172 +++++++++++-----------
 src/bin/pg_waldump/t/002_save_fullpage.pl |   2 +-
 3 files changed, 86 insertions(+), 89 deletions(-)

diff --git a/src/bin/pg_waldump/meson.build b/src/bin/pg_waldump/meson.build
index 9605976870d..34e37bffc36 100644
--- a/src/bin/pg_waldump/meson.build
+++ b/src/bin/pg_waldump/meson.build
@@ -29,6 +29,7 @@ tests += {
   'tap': {
     'tests': [
       't/001_basic.pl',
+      't/002_save_fullpage.pl',
     ],
   },
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 2b6f772dc71..e35d15132f4 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -76,7 +76,6 @@ typedef struct XLogDumpConfig
 	bool		filter_by_fpw;
 
 	/* save options */
-	bool		save_fpw;
 	bool		fixup_fpw;
 	char	   *save_fpw_path;
 } XLogDumpConfig;
@@ -458,96 +457,94 @@ XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath, bool fixup)
 
 	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
 	{
+		char		page[BLCKSZ] = {0};
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+		FILE	   *OPF;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+
 		if (!XLogRecHasBlockRef(record, block_id))
 			continue;
 
-		if (XLogRecHasBlockImage(record, block_id))
-		{
-			char		page[BLCKSZ];
-
-			memset(page, 0, BLCKSZ);
-
-			if (RestoreBlockImage(record, block_id, page))
-			{
-				/* we have our extracted FPI, let's save it now */
-				char		filename[MAXPGPATH];
-				char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
-				FILE	   *OPF;
-				BlockNumber blk;
-				RelFileLocator rnode;
-				ForkNumber	fork;
-
-				XLogRecGetBlockTagExtended(record, block_id,
-										   &rnode, &fork, &blk, NULL);
-
-				/*
-				 * The raw page as stored in the WAL record includes the LSN
-				 * of the block as it appeared when it was originally written,
-				 * however this differs than the effects of replaying this
-				 * same FPI in recovery, as recovery calls RestoreBlockImage()
-				 * and then sets the LSN as part of one action.  What this
-				 * means is that a page as recovered from WAL and the version
-				 * of the page saved here will differ by the LSN and the
-				 * checksum (if enabled).
-				 *
-				 * There are potentially use-cases for both versions (with and
-				 * without mentioned fixups), so allow this to be
-				 * user-selected, unless the restored page was empty, in which
-				 * case we leave it alone.
-				 */
-
-				if (fixup && !PageIsNew(page))
-				{
-					PageSetLSN(page, record->ReadRecPtr);
+		if (!XLogRecHasBlockImage(record, block_id))
+			continue;
 
-					/*
-					 * If checksum field is non-zero then we have checksums
-					 * enabled, so recalculate the checksum with new LSN
-					 * (whether this is considered a hack or heuristics is an
-					 * exercise for the reader).
-					 *
-					 * We make this choice to allow pages saved by this
-					 * function to work as expected with the checksum-related
-					 * functions in pageinspect without having to worry about
-					 * zero_damaged_pages or other considerations.
-					 *
-					 * FPIs in WAL do not have the checksum field updated in
-					 * the page image; in a checksums-enabled cluster, this
-					 * task is handled by FlushBuffer() when a dirty buffer is
-					 * written out to disk.  Since we are running outside of
-					 * Postmaster that won't work in this case, so we handle
-					 * ourselves.
-					 */
+		if (!RestoreBlockImage(record, block_id, page))
+			continue;
 
-					if (((PageHeader) page)->pd_checksum)
-						((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blk);
-				}
+		/* we have our extracted FPI, let's save it now */
+
+		XLogRecGetBlockTagExtended(record, block_id,
+								   &rnode, &fork, &blk, NULL);
+
+		/*
+		 * The raw page as stored in the WAL record includes the LSN
+		 * of the block as it appeared when it was originally written,
+		 * however this differs than the effects of replaying this
+		 * same FPI in recovery, as recovery calls RestoreBlockImage()
+		 * and then sets the LSN as part of one action.  What this
+		 * means is that a page as recovered from WAL and the version
+		 * of the page saved here will differ by the LSN and the
+		 * checksum (if enabled).
+		 *
+		 * There are potentially use-cases for both versions (with and
+		 * without mentioned fixups), so allow this to be
+		 * user-selected, unless the restored page was empty, in which
+		 * case we leave it alone.
+		 */
+
+		if (fixup && !PageIsNew(page))
+		{
+			PageSetLSN(page, record->ReadRecPtr);
+
+			/*
+			 * If checksum field is non-zero then we have checksums
+			 * enabled, so recalculate the checksum with new LSN
+			 * (whether this is considered a hack or heuristics is an
+			 * exercise for the reader).
+			 *
+			 * We make this choice to allow pages saved by this
+			 * function to work as expected with the checksum-related
+			 * functions in pageinspect without having to worry about
+			 * zero_damaged_pages or other considerations.
+			 *
+			 * FPIs in WAL do not have the checksum field updated in
+			 * the page image; in a checksums-enabled cluster, this
+			 * task is handled by FlushBuffer() when a dirty buffer is
+			 * written out to disk.  Since we are running outside of
+			 * Postmaster that won't work in this case, so we handle
+			 * ourselves.
+			 */
+
+			if (((PageHeader) page)->pd_checksum)
+				((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blk);
+		}
 
-				if (fork >= 0 && fork <= MAX_FORKNUM)
-				{
-					if (fork)
-						sprintf(forkname, "_%s", forkNames[fork]);
-					else
-						forkname[0] = 0;
-				}
-				else
-					pg_fatal("found invalid fork number: %u", fork);
+		if (fork >= 0 && fork <= MAX_FORKNUM)
+		{
+			if (fork)
+				sprintf(forkname, "_%s", forkNames[fork]);
+			else
+				forkname[0] = 0;
+		}
+		else
+			pg_fatal("found invalid fork number: %u", fork);
 
-				snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
-						 LSN_FORMAT_ARGS(record->ReadRecPtr),
-						 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+		snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+				 LSN_FORMAT_ARGS(record->ReadRecPtr),
+				 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
 
-				OPF = fopen(filename, PG_BINARY_W);
-				if (!OPF)
-					pg_fatal("couldn't open file for output: %s", filename);
+		OPF = fopen(filename, PG_BINARY_W);
+		if (!OPF)
+			pg_fatal("couldn't open file for output: %s", filename);
 
-				if (pg_pwrite(fileno(OPF), page, BLCKSZ, 0) != BLCKSZ)
-					pg_fatal("couldn't write out complete fullpage image to file: %s", filename);
+		if (pg_pwrite(fileno(OPF), page, BLCKSZ, 0) != BLCKSZ)
+			pg_fatal("couldn't write out complete fullpage image to file: %s", filename);
 
-				fsync(fileno(OPF));
-				fclose(OPF);
-			}
-		}
+		fsync(fileno(OPF));
+		fclose(OPF);
 	}
 }
 
@@ -791,9 +788,9 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
-	printf(_("  -W, --save-fpi=path    save found full page images to given path\n"
-			 "                         as LSN.T.D.R.B_F\n"));
-	printf(_("  -X, --fixup-fpi=path   like --save-fpi but apply LSN fixups to \n"));
+	printf(_("  -W, --save-fpi=path    save full page images to given path as\n"
+			 "                         LSN.T.D.R.B_F\n"));
+	printf(_("  -X, --fixup-fpi=path   like --save-fpi but apply LSN fixups to saved page\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -889,7 +886,6 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
-	config.save_fpw = false;
 	config.fixup_fpw = false;
 	config.save_fpw_path = NULL;
 
@@ -1040,7 +1036,6 @@ main(int argc, char **argv)
 				break;
 			case 'W':
 			case 'X':
-				config.save_fpw = true;
 				config.fixup_fpw = (option == 'X');
 				config.save_fpw_path = pg_strdup(optarg);
 				break;
@@ -1108,7 +1103,8 @@ main(int argc, char **argv)
 			goto bad_argument;
 		}
 
-		if (!dir_status && mkdir(config.save_fpw_path, 0700) < 0)
+		/* Create the dir if it doesn't exist */
+		if (dir_status == 0 && mkdir(config.save_fpw_path, 0700) < 0)
 		{
 			pg_log_error("could not create output directory \"%s\": %m",
 						 config.save_fpw_path);
@@ -1294,7 +1290,7 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
-			else if (config.save_fpw)
+			else if (config.save_fpw_path)
 			{
 				if (XLogRecordHasFPW(xlogreader_state))
 					XLogRecordSaveFPWs(xlogreader_state, config.save_fpw_path, config.fixup_fpw);
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
index f605b4f6adb..6df670c39ba 100644
--- a/src/bin/pg_waldump/t/002_save_fullpage.pl
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -60,7 +60,7 @@ EOF
 
 # get the relation node, etc for the new table
 my $relation = $primary->safe_psql('db1',
-	q{SELECT CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END || '/' || pg_database.oid || '/' || relfilenode FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
+	q{SELECT format('%s/%s/%s', CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END, pg_database.oid, relfilenode) FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
 );
 
 diag $relation;
-- 
2.25.1

From 02e9c308cd435e266dda95186127932bc2f50965 Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Wed, 20 Apr 2022 19:59:35 -0500
Subject: [PATCH v6] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a target directory,
which must be empty or not exist.  These images are subject to the same
filtering rules as normal display in pg_waldump, which means that you
can isolate the full page writes to a target relation, among other
things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0
00000000-01000150.1664.0.6115.0
00000000-010001E0.1664.0.6114.0
00000000-01000270.1663.1.6116.0
00000000-01000300.1663.1.6113.0
00000000-01000390.1663.1.6112.0
00000000-01000420.1663.1.8903.0
00000000-010004B0.1663.1.8902.0
00000000-01000540.1663.1.6111.0
00000000-010005D0.1663.1.6110.0

If the FPI comes from a fork other than the main fork, the fork name
will be appended on the output file name; e.g.:

00000000-014A4758.1663.1.12864.0_vm

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).  As such there are two versions of this functionality, one which
returns the raw page as it appears in the WAL (--save-fpi) and one
which applies the updated pd_lsn and pd_checksum (--fixup-fpi).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          |  79 ++++++++++++
 src/bin/pg_waldump/meson.build            |   1 +
 src/bin/pg_waldump/pg_waldump.c           | 147 +++++++++++++++++++++-
 src/bin/pg_waldump/t/002_save_fullpage.pl | 142 +++++++++++++++++++++
 4 files changed, 368 insertions(+), 1 deletion(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5..ecc973e86b 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,85 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fpi=<replaceable>save_path</replaceable></option></term>
+       <term><option>-X <replaceable>save_path</replaceable></option></term>
+       <term><option>--fixup-fpi=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        If invoked using
+        the <literal>-X</literal>/<literal>--fixup-fpi</literal>
+        option, this page image will include the <literal>pd_lsn</literal> of
+        the replayed record rather than the raw page image; as well,
+        the <literal>pd_checksum</literal> field will be updated if it had
+        previously existed.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>RELTABLESPACE</replaceable>.<replaceable>DATOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>RELTABLESPACE</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DATOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             if coming from the main fork, will be empty, otherwise will be
+             one of <literal>_fsm</literal>, <literal>_vm</literal>,
+             or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/meson.build b/src/bin/pg_waldump/meson.build
index 9605976870..34e37bffc3 100644
--- a/src/bin/pg_waldump/meson.build
+++ b/src/bin/pg_waldump/meson.build
@@ -29,6 +29,7 @@ tests += {
   'tap': {
     'tests': [
       't/001_basic.pl',
+      't/002_save_fullpage.pl',
     ],
   },
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..e35d15132f 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -24,8 +24,12 @@
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
+#include "storage/checksum.h"
+#include "storage/checksum_impl.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,10 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	bool		fixup_fpw;
+	char	   *save_fpw_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +447,107 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPWs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath, bool fixup)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		char		page[BLCKSZ] = {0};
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+		FILE	   *OPF;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (!XLogRecHasBlockImage(record, block_id))
+			continue;
+
+		if (!RestoreBlockImage(record, block_id, page))
+			continue;
+
+		/* we have our extracted FPI, let's save it now */
+
+		XLogRecGetBlockTagExtended(record, block_id,
+								   &rnode, &fork, &blk, NULL);
+
+		/*
+		 * The raw page as stored in the WAL record includes the LSN
+		 * of the block as it appeared when it was originally written,
+		 * however this differs than the effects of replaying this
+		 * same FPI in recovery, as recovery calls RestoreBlockImage()
+		 * and then sets the LSN as part of one action.  What this
+		 * means is that a page as recovered from WAL and the version
+		 * of the page saved here will differ by the LSN and the
+		 * checksum (if enabled).
+		 *
+		 * There are potentially use-cases for both versions (with and
+		 * without mentioned fixups), so allow this to be
+		 * user-selected, unless the restored page was empty, in which
+		 * case we leave it alone.
+		 */
+
+		if (fixup && !PageIsNew(page))
+		{
+			PageSetLSN(page, record->ReadRecPtr);
+
+			/*
+			 * If checksum field is non-zero then we have checksums
+			 * enabled, so recalculate the checksum with new LSN
+			 * (whether this is considered a hack or heuristics is an
+			 * exercise for the reader).
+			 *
+			 * We make this choice to allow pages saved by this
+			 * function to work as expected with the checksum-related
+			 * functions in pageinspect without having to worry about
+			 * zero_damaged_pages or other considerations.
+			 *
+			 * FPIs in WAL do not have the checksum field updated in
+			 * the page image; in a checksums-enabled cluster, this
+			 * task is handled by FlushBuffer() when a dirty buffer is
+			 * written out to disk.  Since we are running outside of
+			 * Postmaster that won't work in this case, so we handle
+			 * ourselves.
+			 */
+
+			if (((PageHeader) page)->pd_checksum)
+				((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blk);
+		}
+
+		if (fork >= 0 && fork <= MAX_FORKNUM)
+		{
+			if (fork)
+				sprintf(forkname, "_%s", forkNames[fork]);
+			else
+				forkname[0] = 0;
+		}
+		else
+			pg_fatal("found invalid fork number: %u", fork);
+
+		snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+				 LSN_FORMAT_ARGS(record->ReadRecPtr),
+				 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+		OPF = fopen(filename, PG_BINARY_W);
+		if (!OPF)
+			pg_fatal("couldn't open file for output: %s", filename);
+
+		if (pg_pwrite(fileno(OPF), page, BLCKSZ, 0) != BLCKSZ)
+			pg_fatal("couldn't write out complete fullpage image to file: %s", filename);
+
+		fsync(fileno(OPF));
+		fclose(OPF);
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +788,9 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --save-fpi=path    save full page images to given path as\n"
+			 "                         LSN.T.D.R.B_F\n"));
+	printf(_("  -X, --fixup-fpi=path   like --save-fpi but apply LSN fixups to saved page\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +824,8 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fpi", required_argument, NULL, 'W'},
+		{"fixup-fpi", required_argument, NULL, 'X'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +886,8 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.fixup_fpw = false;
+	config.save_fpw_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +898,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:X:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -918,6 +1034,11 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+			case 'X':
+				config.fixup_fpw = (option == 'X');
+				config.save_fpw_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1093,25 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpw_path != NULL)
+	{
+		int			dir_status = pg_check_dir(config.save_fpw_path);
+
+		if (dir_status < 0)
+		{
+			pg_log_error("could not access output directory: %s", config.save_fpw_path);
+			goto bad_argument;
+		}
+
+		/* Create the dir if it doesn't exist */
+		if (dir_status == 0 && mkdir(config.save_fpw_path, 0700) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpw_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1290,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpw_path)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPWs(xlogreader_state, config.save_fpw_path, config.fixup_fpw);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..c98100bf21
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,142 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my $blocksize = 8192;
+
+# routine to extract the LSN and checksum from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if $blocksize != read $fh, $block, $blocksize;
+	my ($lsn_hi, $lsn_lo, $checksum) = unpack('VVv', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo, $checksum);
+}
+
+# Set umask so test directories and files are created with default permissions
+umask(0077);
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init('-k');
+$primary->append_conf('postgresql.conf', "max_wal_size='100MB'");
+$primary->append_conf('postgresql.conf', "wal_level='replica'");
+$primary->append_conf('postgresql.conf', 'archive_mode=on');
+$primary->append_conf('postgresql.conf', "archive_command='/bin/false'");
+$primary->start;
+
+# Sanity checks for command line options.
+$primary->command_fails(
+	[ 'pg_waldump', '--save-fpi' ],
+	'--save-fpi fails without path');
+$primary->command_fails(
+	[ 'pg_waldump', '--fixup-fpi' ],
+	'--fixup-fpi fails without path');
+
+# generate data/wal to examine
+$primary->safe_psql('postgres', q(CREATE DATABASE db1));
+$primary->safe_psql('db1',      <<EOF);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT;
+SELECT pg_switch_wal();
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+SELECT pg_switch_wal();
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+SELECT pg_switch_wal();
+EOF
+
+# get our blocksize from the db
+$blocksize = $primary->safe_psql('postgres', "SELECT current_setting('block_size')");
+
+# get the relation node, etc for the new table
+my $relation = $primary->safe_psql('db1',
+	q{SELECT format('%s/%s/%s', CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END, pg_database.oid, pg_relation_filenode(pg_class.oid)) FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
+);
+
+diag $relation;
+
+$primary->stop;
+my $waldir = $primary->basedir . '/pgdata/pg_wal';
+my $walfile = [glob("$waldir/*")]->[2]; # we want the 00000002 file
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+diag "using walfile: $walfile";
+
+# # # extract files
+# PostgreSQL::Test::RecursiveCopy::copypath(
+# 	$primary->data_dir . "/pg_wal",
+# 	$primary->tmp_folder);
+
+$primary->command_ok(['pg_waldump', '--save-fpi', "$tmp_folder/raw", '--relation', $relation, $walfile]);
+$primary->command_ok(['pg_waldump', '--fixup-fpi', "$tmp_folder/fixup", '--relation', $relation, $walfile]);
+
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm)?$/;
+
+my %checksums;
+my %files;
+
+# verify filename formats matches w/--save-fpi
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since no fixup, verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+
+	# stash checksum for later comparisons
+	$checksums{$file} = $checksum;
+}
+
+# verify filename formats matches w/--fixup-fpi
+for my $fullpath (glob "$tmp_folder/fixup/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since fixup, verify the lsn in the block equals file lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn eq $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN is the same as block-based one');
+
+	# verify checksum change; XXX: there could be valid clashes here,
+	# just validate that the page matches the expected checksum instead?
+	ok( $checksum == 0 || $checksums{$file} != $checksum,
+		'check for checksum change or no checksum');
+}
+
+# validate that we ended up with some files output and they were the same
+ok(keys %files > 0, 'verify we processed some files');
+ok((grep { $_ != 2 } values %files) == 0,
+	'ensure raw and fixup had same number of files');
+
+done_testing();
-- 
2.37.1 (Apple Git-137.1)

From b61a55343d896813c64be07b2269b8c0235615ff Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Wed, 20 Apr 2022 19:59:35 -0500
Subject: [PATCH v7] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a target directory,
which must be empty or not exist.  These images are subject to the same
filtering rules as normal display in pg_waldump, which means that you
can isolate the full page writes to a target relation, among other
things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0
00000000-01000150.1664.0.6115.0
00000000-010001E0.1664.0.6114.0
00000000-01000270.1663.1.6116.0
00000000-01000300.1663.1.6113.0
00000000-01000390.1663.1.6112.0
00000000-01000420.1663.1.8903.0
00000000-010004B0.1663.1.8902.0
00000000-01000540.1663.1.6111.0
00000000-010005D0.1663.1.6110.0

If the FPI comes from a fork other than the main fork, the fork name
will be appended on the output file name; e.g.:

00000000-014A4758.1663.1.12864.0_vm

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).  As such there are two versions of this functionality, one which
returns the raw page as it appears in the WAL (--save-fpi) and one
which applies the updated pd_lsn and pd_checksum (--fixup-fpi).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          |  79 ++++++++++++
 src/bin/pg_waldump/meson.build            |   1 +
 src/bin/pg_waldump/pg_waldump.c           | 138 ++++++++++++++++++++-
 src/bin/pg_waldump/t/002_save_fullpage.pl | 142 ++++++++++++++++++++++
 4 files changed, 359 insertions(+), 1 deletion(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5..ecc973e86b 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,85 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fpi=<replaceable>save_path</replaceable></option></term>
+       <term><option>-X <replaceable>save_path</replaceable></option></term>
+       <term><option>--fixup-fpi=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        If invoked using
+        the <literal>-X</literal>/<literal>--fixup-fpi</literal>
+        option, this page image will include the <literal>pd_lsn</literal> of
+        the replayed record rather than the raw page image; as well,
+        the <literal>pd_checksum</literal> field will be updated if it had
+        previously existed.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>RELTABLESPACE</replaceable>.<replaceable>DATOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>RELTABLESPACE</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DATOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             if coming from the main fork, will be empty, otherwise will be
+             one of <literal>_fsm</literal>, <literal>_vm</literal>,
+             or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/meson.build b/src/bin/pg_waldump/meson.build
index 9605976870..34e37bffc3 100644
--- a/src/bin/pg_waldump/meson.build
+++ b/src/bin/pg_waldump/meson.build
@@ -29,6 +29,7 @@ tests += {
   'tap': {
     'tests': [
       't/001_basic.pl',
+      't/002_save_fullpage.pl',
     ],
   },
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..6769d21182 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -23,9 +23,14 @@
 #include "access/xlogrecord.h"
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
+#include "common/file_perm.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
+#include "storage/checksum.h"
+#include "storage/checksum_impl.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +75,10 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	bool		fixup_fpw;
+	char	   *save_fpw_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +448,105 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPWs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath, bool fixup)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		PGAlignedBlock	zerobuf;
+		Page		page;
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+		FILE	   *OPF;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+
+		page = (Page) zerobuf.data;
+
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (!XLogRecHasBlockImage(record, block_id))
+			continue;
+
+		if (!RestoreBlockImage(record, block_id, page))
+			continue;
+
+		/* we have our extracted FPI, let's save it now */
+
+		XLogRecGetBlockTagExtended(record, block_id,
+								   &rnode, &fork, &blk, NULL);
+
+		/*
+		 * The raw page as stored in the WAL record includes the LSN
+		 * of the block as it appeared when it was originally written,
+		 * however this differs than the effects of replaying this
+		 * same FPI in recovery, as recovery calls RestoreBlockImage()
+		 * and then sets the LSN as part of one action.  What this
+		 * means is that a page as recovered from WAL and the version
+		 * of the page saved here will differ by the LSN and the
+		 * checksum (if enabled).
+		 *
+		 * There are potentially use-cases for both versions (with and
+		 * without mentioned fixups), so allow this to be
+		 * user-selected, unless the restored page was empty, in which
+		 * case we leave it alone.
+		 */
+
+		if (fixup && !PageIsNew(page))
+		{
+			PageSetLSN(page, record->ReadRecPtr);
+
+			/*
+			 * If checksum field is non-zero then we have checksums
+			 * enabled, so recalculate the checksum with new LSN
+			 * (whether this is considered a hack or heuristics is an
+			 * exercise for the reader).
+			 *
+			 * We make this choice to allow pages saved by this
+			 * function to work as expected with the checksum-related
+			 * functions in pageinspect without having to worry about
+			 * zero_damaged_pages or other considerations.
+			 *
+			 * FPIs in WAL do not have the checksum field updated in
+			 * the page image; in a checksums-enabled cluster, this
+			 * task is handled by FlushBuffer() when a dirty buffer is
+			 * written out to disk.  Since we are running outside of
+			 * Postmaster that won't work in this case, so we handle
+			 * ourselves.
+			 */
+
+			if (((PageHeader) page)->pd_checksum)
+				((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blk);
+		}
+
+		if (fork >= 0 && fork <= MAX_FORKNUM)
+			sprintf(forkname, "_%s", forkNames[fork]);
+		else
+			pg_fatal("found invalid fork number: %u", fork);
+
+		snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+				 LSN_FORMAT_ARGS(record->ReadRecPtr),
+				 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+		OPF = fopen(filename, PG_BINARY_W);
+		if (!OPF)
+			pg_fatal("couldn't open file for output: %s", filename);
+
+		if (pg_pwrite(fileno(OPF), page, BLCKSZ, 0) != BLCKSZ)
+			pg_fatal("couldn't write out complete fullpage image to file: %s", filename);
+
+		fsync(fileno(OPF));
+		fclose(OPF);
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +787,9 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --save-fpi=path    save full page images to given path as\n"
+			 "                         LSN.T.D.R.B_F\n"));
+	printf(_("  -X, --fixup-fpi=path   like --save-fpi but apply LSN fixups to saved page\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +823,8 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fpi", required_argument, NULL, 'W'},
+		{"fixup-fpi", required_argument, NULL, 'X'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +885,8 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.fixup_fpw = false;
+	config.save_fpw_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +897,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:X:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -918,6 +1033,11 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+			case 'X':
+				config.fixup_fpw = (option == 'X');
+				config.save_fpw_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1092,17 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpw_path != NULL)
+	{
+		/* Create the dir if it doesn't exist */
+		if (pg_mkdir_p(config.save_fpw_path, pg_dir_create_mode) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpw_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1281,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpw_path)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPWs(xlogreader_state, config.save_fpw_path, config.fixup_fpw);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..8131ff2d93
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,142 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my $blocksize = 8192;
+
+# routine to extract the LSN and checksum from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if $blocksize != read $fh, $block, $blocksize;
+	my ($lsn_hi, $lsn_lo, $checksum) = unpack('VVv', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo, $checksum);
+}
+
+# Set umask so test directories and files are created with default permissions
+umask(0077);
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init('-k');
+$primary->append_conf('postgresql.conf', "max_wal_size='100MB'");
+$primary->append_conf('postgresql.conf', "wal_level='replica'");
+$primary->append_conf('postgresql.conf', 'archive_mode=on');
+$primary->append_conf('postgresql.conf', "archive_command='/bin/false'");
+$primary->start;
+
+# Sanity checks for command line options.
+$primary->command_fails(
+	[ 'pg_waldump', '--save-fpi' ],
+	'--save-fpi fails without path');
+$primary->command_fails(
+	[ 'pg_waldump', '--fixup-fpi' ],
+	'--fixup-fpi fails without path');
+
+# generate data/wal to examine
+$primary->safe_psql('postgres', q(CREATE DATABASE db1));
+$primary->safe_psql('db1',      <<EOF);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT;
+SELECT pg_switch_wal();
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+SELECT pg_switch_wal();
+UPDATE test_table SET a = a + 1;
+CHECKPOINT;
+SELECT pg_switch_wal();
+EOF
+
+# get our blocksize from the db
+$blocksize = $primary->safe_psql('postgres', "SELECT current_setting('block_size')");
+
+# get the relation node, etc for the new table
+my $relation = $primary->safe_psql('db1',
+	q{SELECT format('%s/%s/%s', CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END, pg_database.oid, pg_relation_filenode(pg_class.oid)) FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
+);
+
+diag $relation;
+
+$primary->stop;
+my $waldir = $primary->basedir . '/pgdata/pg_wal';
+my $walfile = [glob("$waldir/*")]->[2]; # we want the 00000002 file
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+diag "using walfile: $walfile";
+
+# # # extract files
+# PostgreSQL::Test::RecursiveCopy::copypath(
+# 	$primary->data_dir . "/pg_wal",
+# 	$primary->tmp_folder);
+
+$primary->command_ok(['pg_waldump', '--save-fpi', "$tmp_folder/raw", '--relation', $relation, $walfile]);
+$primary->command_ok(['pg_waldump', '--fixup-fpi', "$tmp_folder/fixup", '--relation', $relation, $walfile]);
+
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm|_main)?$/;
+
+my %checksums;
+my %files;
+
+# verify filename formats matches w/--save-fpi
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since no fixup, verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+
+	# stash checksum for later comparisons
+	$checksums{$file} = $checksum;
+}
+
+# verify filename formats matches w/--fixup-fpi
+for my $fullpath (glob "$tmp_folder/fixup/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# since fixup, verify the lsn in the block equals file lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn eq $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN is the same as block-based one');
+
+	# verify checksum change; XXX: there could be valid clashes here,
+	# just validate that the page matches the expected checksum instead?
+	ok( $checksum == 0 || $checksums{$file} != $checksum,
+		'check for checksum change or no checksum');
+}
+
+# validate that we ended up with some files output and they were the same
+ok(keys %files > 0, 'verify we processed some files');
+ok((grep { $_ != 2 } values %files) == 0,
+	'ensure raw and fixup had same number of files');
+
+done_testing();
-- 
2.37.1 (Apple Git-137.1)

From 82b5cc8356b49ca4a781e066f9abc703aead15c5 Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Thu, 10 Nov 2022 10:29:38 -0600
Subject: [PATCH v8] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a given target directory.
These images are subject to the same filtering rules as normal display in
pg_waldump, which means that you can isolate the full page writes to a
target relation, among other things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk>_<fork> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0_main
00000000-01000150.1664.0.6115.0_main
00000000-010001E0.1664.0.6114.0_main
00000000-01000270.1663.1.6116.0_main
00000000-01000300.1663.1.6113.0_main
00000000-01000390.1663.1.6112.0_main
00000000-01000420.1663.1.8903.0_main
00000000-010004B0.1663.1.8902.0_main
00000000-01000540.1663.1.6111.0_main
00000000-010005D0.1663.1.6110.0_main

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          |  69 ++++++++++++++
 src/bin/pg_waldump/meson.build            |   1 +
 src/bin/pg_waldump/pg_waldump.c           |  95 +++++++++++++++++++-
 src/bin/pg_waldump/t/002_save_fullpage.pl | 104 ++++++++++++++++++++++
 4 files changed, 268 insertions(+), 1 deletion(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5..3c5bd9eb3f 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,75 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fpi=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>RELTABLESPACE</replaceable>.<replaceable>DATOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>RELTABLESPACE</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DATOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             The name of the fork the full page image came from.  One
+             of <literal>_main</literal>, <literal>_fsm</literal>,
+             <literal>_vm</literal>, or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/meson.build b/src/bin/pg_waldump/meson.build
index 9605976870..34e37bffc3 100644
--- a/src/bin/pg_waldump/meson.build
+++ b/src/bin/pg_waldump/meson.build
@@ -29,6 +29,7 @@ tests += {
   'tap': {
     'tests': [
       't/001_basic.pl',
+      't/002_save_fullpage.pl',
     ],
   },
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..b79f09ad4f 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -23,9 +23,15 @@
 #include "access/xlogrecord.h"
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
+#include "common/file_perm.h"
+#include "common/file_utils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
+#include "storage/checksum.h"
+#include "storage/checksum_impl.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +76,9 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	char	   *save_fpw_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +448,61 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPWs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		PGAlignedBlock buf;
+		Page		page;
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+		FILE	   *OPF;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+
+		page = (Page) buf.data;
+
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (!XLogRecHasBlockImage(record, block_id))
+			continue;
+
+		if (!RestoreBlockImage(record, block_id, page))
+			continue;
+
+		/* we have our extracted FPI, let's save it now */
+
+		XLogRecGetBlockTagExtended(record, block_id,
+								   &rnode, &fork, &blk, NULL);
+
+		if (fork >= 0 && fork <= MAX_FORKNUM)
+			sprintf(forkname, "_%s", forkNames[fork]);
+		else
+			pg_fatal("found invalid fork number: %u", fork);
+
+		snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+				 LSN_FORMAT_ARGS(record->ReadRecPtr),
+				 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+		OPF = fopen(filename, PG_BINARY_W);
+		if (!OPF)
+			pg_fatal("couldn't open file for output: %s", filename);
+
+		if (fwrite(page, BLCKSZ, 1, OPF) != 1)
+			pg_fatal("couldn't write out complete full page image to file: %s", filename);
+
+		fclose(OPF);
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +743,8 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --save-fpi=path    save full page images to given path as\n"
+			 "                         LSN.T.D.R.B_F\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +778,7 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fpi", required_argument, NULL, 'W'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +839,7 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpw_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +850,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:X:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -918,6 +986,9 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+				config.save_fpw_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1043,17 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpw_path != NULL)
+	{
+		/* Create the dir if it doesn't exist */
+		if (pg_mkdir_p(config.save_fpw_path, pg_dir_create_mode) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpw_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1232,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpw_path)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPWs(xlogreader_state, config.save_fpw_path);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
@@ -1167,6 +1254,12 @@ main(int argc, char **argv)
 	if (time_to_stop)
 		exit(0);
 
+	if (config.save_fpw_path != NULL)
+	{
+		/* Fsync our output directory */
+		fsync_fname(config.save_fpw_path, true);
+	}
+
 	if (errormsg)
 		pg_fatal("error in WAL record at %X/%X: %s",
 				 LSN_FORMAT_ARGS(xlogreader_state->ReadRecPtr),
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..c9a2ac733a
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,104 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my $blocksize;
+
+# routine to extract the LSN and checksum from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if $blocksize != read $fh, $block, $blocksize;
+	my ($lsn_hi, $lsn_lo, $checksum) = unpack('VVv', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo, $checksum);
+}
+
+# Set umask so test directories and files are created with default permissions
+umask(0077);
+
+my $node =  PostgreSQL::Test::Cluster->new('primary');
+$node->init(extra => ['-k'], allows_streaming => 1);
+$node->start;
+
+# Sanity checks for command line options.
+$node->command_fails(
+	[ 'pg_waldump', '--save-fpi' ],
+	'--save-fpi fails without path');
+
+# generate data/wal to examine that will have FPIs in them
+$node->safe_psql('postgres', <<EOF);
+SELECT 'init' FROM pg_create_physical_replication_slot('regress_pg_waldump_slot', true, false);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT;
+SELECT pg_switch_wal();
+UPDATE test_table SET a = a + 1;
+SELECT pg_switch_wal();
+EOF
+
+# get our blocksize from the db
+$blocksize = $node->safe_psql('postgres', "SELECT current_setting('block_size')");
+
+# get the relation node, etc for the new table
+my $relation = $node->safe_psql('postgres',
+	q{SELECT format('%s/%s/%s', CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END, pg_database.oid, pg_relation_filenode(pg_class.oid)) FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
+);
+
+diag $relation;
+
+my $waldir = $node->basedir . '/pgdata/pg_wal';
+my $walfile = [sort { $a <=> $b } glob("$waldir/00*")]->[1]; # we want the second WAL file, which will be a complete WAL file with full-page writes for our specific relation.
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+diag "using walfile: $walfile";
+
+ok($walfile, "Got a WAL file");
+
+$node->command_ok(['pg_waldump', '--save-fpi', "$tmp_folder/raw", '--relation', $relation, $walfile]);
+
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm|_main)?$/;
+
+my %checksums;
+my %files;
+
+# verify filename formats matches w/--save-fpi
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk, $checksum) = get_block_info($fullpath);
+
+	# verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+
+	# stash checksum for later comparisons
+	$checksums{$file} = $checksum;
+}
+
+# validate that we ended up with some files output and they were the same
+ok(keys %files > 0, 'verify we processed some files');
+
+$node->safe_psql('postgres', <<EOQ);
+SELECT pg_drop_replication_slot('regress_pg_waldump_slot');
+EOQ
+done_testing();
-- 
2.37.1 (Apple Git-137.1)

From e0fe44626cea657f5dd6a6e035406989c529a02a Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Thu, 10 Nov 2022 10:29:38 -0600
Subject: [PATCH v9] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a given target directory.
These images are subject to the same filtering rules as normal display in
pg_waldump, which means that you can isolate the full page writes to a
target relation, among other things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk>_<fork> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0_main
00000000-01000150.1664.0.6115.0_main
00000000-010001E0.1664.0.6114.0_main
00000000-01000270.1663.1.6116.0_main
00000000-01000300.1663.1.6113.0_main
00000000-01000390.1663.1.6112.0_main
00000000-01000420.1663.1.8903.0_main
00000000-010004B0.1663.1.8902.0_main
00000000-01000540.1663.1.6111.0_main
00000000-010005D0.1663.1.6110.0_main

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          |  69 +++++++++++++++
 src/bin/pg_waldump/meson.build            |   1 +
 src/bin/pg_waldump/pg_waldump.c           | 101 +++++++++++++++++++++-
 src/bin/pg_waldump/t/002_save_fullpage.pl |  96 ++++++++++++++++++++
 4 files changed, 264 insertions(+), 3 deletions(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5..3c5bd9eb3f 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,75 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fpi=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>RELTABLESPACE</replaceable>.<replaceable>DATOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>RELTABLESPACE</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DATOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             The name of the fork the full page image came from.  One
+             of <literal>_main</literal>, <literal>_fsm</literal>,
+             <literal>_vm</literal>, or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/meson.build b/src/bin/pg_waldump/meson.build
index 9605976870..34e37bffc3 100644
--- a/src/bin/pg_waldump/meson.build
+++ b/src/bin/pg_waldump/meson.build
@@ -29,6 +29,7 @@ tests += {
   'tap': {
     'tests': [
       't/001_basic.pl',
+      't/002_save_fullpage.pl',
     ],
   },
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..cbfafebe8a 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -23,9 +23,13 @@
 #include "access/xlogrecord.h"
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
+#include "common/file_perm.h"
+#include "common/file_utils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,9 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	char	   *save_fpi_path;
 } XLogDumpConfig;
 
 
@@ -423,7 +430,7 @@ XLogRecordMatchesRelationBlock(XLogReaderState *record,
  * Boolean to return whether the given WAL record contains a full page write.
  */
 static bool
-XLogRecordHasFPW(XLogReaderState *record)
+XLogRecordHasFPI(XLogReaderState *record)
 {
 	int			block_id;
 
@@ -439,6 +446,61 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPIs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPIs(XLogReaderState *record, const char *savepath)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		PGAlignedBlock buf;
+		Page		page;
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+		FILE	   *OPF;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+
+		page = (Page) buf.data;
+
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (!XLogRecHasBlockImage(record, block_id))
+			continue;
+
+		if (!RestoreBlockImage(record, block_id, page))
+			continue;
+
+		/* we have our extracted FPI, let's save it now */
+
+		XLogRecGetBlockTagExtended(record, block_id,
+								   &rnode, &fork, &blk, NULL);
+
+		if (fork >= 0 && fork <= MAX_FORKNUM)
+			sprintf(forkname, "_%s", forkNames[fork]);
+		else
+			pg_fatal("found invalid fork number: %u", fork);
+
+		snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+				 LSN_FORMAT_ARGS(record->ReadRecPtr),
+				 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+		OPF = fopen(filename, PG_BINARY_W);
+		if (!OPF)
+			pg_fatal("couldn't open file for output: %s", filename);
+
+		if (fwrite(page, BLCKSZ, 1, OPF) != 1)
+			pg_fatal("couldn't write out complete full page image to file: %s", filename);
+
+		fclose(OPF);
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +741,8 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --save-fpi=path    save full page images to given path as\n"
+			 "                         LSN.T.D.R.B_F\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +776,7 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fpi", required_argument, NULL, 'W'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +837,7 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpi_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +848,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:X:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -918,6 +984,9 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+				config.save_fpi_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1041,17 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpi_path != NULL)
+	{
+		/* Create the dir if it doesn't exist */
+		if (pg_mkdir_p(config.save_fpi_path, pg_dir_create_mode) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpi_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1139,7 +1219,7 @@ main(int argc, char **argv)
 											config.filter_by_relation_forknum))
 			continue;
 
-		if (config.filter_by_fpw && !XLogRecordHasFPW(xlogreader_state))
+		if (config.filter_by_fpw && !XLogRecordHasFPI(xlogreader_state))
 			continue;
 
 		/* perform any per-record work */
@@ -1150,6 +1230,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpi_path != NULL)
+			{
+				if (XLogRecordHasFPI(xlogreader_state))
+					XLogRecordSaveFPIs(xlogreader_state, config.save_fpi_path);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
@@ -1164,6 +1249,16 @@ main(int argc, char **argv)
 	if (config.stats == true && !config.quiet)
 		XLogDumpDisplayStats(&config, &stats);
 
+	if (config.save_fpi_path != NULL)
+	{
+		/* We fsync our output directory only; since these files are not part
+		 * of the production database we do not require the performance hit
+		 * that fsyncing every FPI would entail, so are doing this as a
+		 * compromise. */
+
+		fsync_fname(config.save_fpi_path, true);
+	}
+
 	if (time_to_stop)
 		exit(0);
 
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..ba0d6c7ad6
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,96 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my ($blocksize, $walfile_name);
+
+# routine to extract the LSN from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if $blocksize != read $fh, $block, $blocksize;
+	my ($lsn_hi, $lsn_lo) = unpack('VV', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo);
+}
+
+# Set umask so test directories and files are created with default permissions
+umask(0077);
+
+my $node =  PostgreSQL::Test::Cluster->new('primary');
+$node->init(extra => ['-k'], allows_streaming => 1);
+$node->start;
+
+# Sanity checks for command line options.
+$node->command_fails(
+	[ 'pg_waldump', '--save-fpi' ],
+	'--save-fpi fails without path');
+
+# generate data/wal to examine that will have FPIs in them
+$node->safe_psql('postgres', <<EOF);
+SELECT 'init' FROM pg_create_physical_replication_slot('regress_pg_waldump_slot', true, false);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT; -- required to force FPI for next writes
+UPDATE test_table SET a = a + 1;
+EOF
+
+($walfile_name, $blocksize) = split '\|' => $node->safe_psql('postgres',"SELECT pg_walfile_name(pg_switch_wal()), current_setting('block_size')");
+
+# get the relation node, etc for the new table
+my $relation = $node->safe_psql('postgres',
+	q{SELECT format('%s/%s/%s', CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END, pg_database.oid, pg_relation_filenode(pg_class.oid)) FROM pg_class, pg_database WHERE relname = 'test_table' AND datname = current_database()}
+);
+
+diag $relation;
+
+my $walfile = $node->basedir . '/pgdata/pg_wal/' . $walfile_name;
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+
+diag "using walfile: $walfile";
+
+ok(-f $walfile, "Got a WAL file");
+
+$node->command_ok(['pg_waldump', '--save-fpi', "$tmp_folder/raw", '--relation', $relation, $walfile]);
+
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm|_main)?$/;
+
+my %files;
+
+# verify filename formats matches w/--save-fpi
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+
+	# save filename for later verification
+	$files{$file}++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk) = get_block_info($fullpath);
+
+	# verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+}
+
+# validate that we ended up with some FPIs saved
+ok(keys %files > 0, 'verify we processed some files');
+
+$node->safe_psql('postgres', "SELECT pg_drop_replication_slot('regress_pg_waldump_slot')");
+
+done_testing();
-- 
2.37.1 (Apple Git-137.1)

From 2858c2ff9e17faca516cd782c606bce5028bc51e Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Thu, 15 Dec 2022 18:14:57 -0500
Subject: [PATCH v10] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a given target directory.
These images are subject to the same filtering rules as normal display in
pg_waldump, which means that you can isolate the full page writes to a
target relation, among other things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk>_<fork> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0_main
00000000-01000150.1664.0.6115.0_main
00000000-010001E0.1664.0.6114.0_main
00000000-01000270.1663.1.6116.0_main
00000000-01000300.1663.1.6113.0_main
00000000-01000390.1663.1.6112.0_main
00000000-01000420.1663.1.8903.0_main
00000000-010004B0.1663.1.8902.0_main
00000000-01000540.1663.1.6111.0_main
00000000-010005D0.1663.1.6110.0_main

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          | 69 ++++++++++++++++
 src/bin/pg_waldump/meson.build            |  1 +
 src/bin/pg_waldump/pg_waldump.c           | 87 +++++++++++++++++++-
 src/bin/pg_waldump/t/002_save_fullpage.pl | 99 +++++++++++++++++++++++
 4 files changed, 255 insertions(+), 1 deletion(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5..3c5bd9eb3f 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,75 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fpi=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>RELTABLESPACE</replaceable>.<replaceable>DATOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>RELTABLESPACE</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DATOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             The name of the fork the full page image came from.  One
+             of <literal>_main</literal>, <literal>_fsm</literal>,
+             <literal>_vm</literal>, or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/meson.build b/src/bin/pg_waldump/meson.build
index 9605976870..34e37bffc3 100644
--- a/src/bin/pg_waldump/meson.build
+++ b/src/bin/pg_waldump/meson.build
@@ -29,6 +29,7 @@ tests += {
   'tap': {
     'tests': [
       't/001_basic.pl',
+      't/002_save_fullpage.pl',
     ],
   },
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..90f8d48e5f 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -23,9 +23,13 @@
 #include "access/xlogrecord.h"
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
+#include "common/file_perm.h"
+#include "common/file_utils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,9 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	char	   *save_fpi_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +446,61 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPIs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPIs(XLogReaderState *record, const char *savepath)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		PGAlignedBlock buf;
+		Page		page;
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+		FILE	   *OPF;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+
+		page = (Page) buf.data;
+
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (!XLogRecHasBlockImage(record, block_id))
+			continue;
+
+		if (!RestoreBlockImage(record, block_id, page))
+			continue;
+
+		/* we have our extracted FPI, let's save it now */
+
+		XLogRecGetBlockTagExtended(record, block_id,
+								   &rnode, &fork, &blk, NULL);
+
+		if (fork >= 0 && fork <= MAX_FORKNUM)
+			sprintf(forkname, "_%s", forkNames[fork]);
+		else
+			pg_fatal("found invalid fork number: %u", fork);
+
+		snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+				 LSN_FORMAT_ARGS(record->ReadRecPtr),
+				 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+		OPF = fopen(filename, PG_BINARY_W);
+		if (!OPF)
+			pg_fatal("couldn't open file for output: %s", filename);
+
+		if (fwrite(page, BLCKSZ, 1, OPF) != 1)
+			pg_fatal("couldn't write out complete full page image to file: %s", filename);
+
+		fclose(OPF);
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +741,8 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --save-fpi=path    save full page images to given path as\n"
+			 "                         LSN.T.D.R.B_F\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +776,7 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fpi", required_argument, NULL, 'W'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +837,7 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpi_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +848,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -918,6 +984,9 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+				config.save_fpi_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1041,17 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpi_path != NULL)
+	{
+		/* Create the dir if it doesn't exist */
+		if (pg_mkdir_p(config.save_fpi_path, pg_dir_create_mode) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpi_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1230,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpi_path != NULL)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPIs(xlogreader_state, config.save_fpi_path);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..2fe2821eb6
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,99 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my ($blocksize, $walfile_name);
+
+# routine to extract the LSN from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if $blocksize != read $fh, $block, $blocksize;
+	my ($lsn_hi, $lsn_lo) = unpack('VV', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo);
+}
+
+my $node =  PostgreSQL::Test::Cluster->new('primary');
+$node->init(extra => ['-k'], allows_streaming => 1);
+$node->start;
+
+# Sanity checks for command line options.
+$node->command_fails(
+	[ 'pg_waldump', '--save-fpi' ],
+	'--save-fpi fails without path');
+
+# generate data/wal to examine that will have FPIs in them
+$node->safe_psql('postgres', "SELECT 'init' FROM pg_create_physical_replication_slot('regress_pg_waldump_slot', true, false);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT; -- required to force FPI for next writes
+UPDATE test_table SET a = a + 1;
+");
+
+($walfile_name, $blocksize) = split '\|' => $node->safe_psql('postgres',"SELECT pg_walfile_name(pg_switch_wal()), current_setting('block_size')");
+
+# get the relation node, etc for the new table
+my $relation = $node->safe_psql('postgres',
+	q{SELECT format(
+        '%s/%s/%s',
+        CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END,
+        pg_database.oid,
+        pg_relation_filenode(pg_class.oid))
+    FROM pg_class, pg_database
+    WHERE relname = 'test_table' AND
+        datname = current_database()}
+);
+
+my $walfile = $node->data_dir . '/pg_wal/' . $walfile_name;
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+
+ok(-f $walfile, "Got a WAL file");
+
+$node->command_ok(['pg_waldump', '--save-fpi', "$tmp_folder/raw", '--relation', $relation, $walfile]);
+
+# This regexp will match filenames of the format XXXXXXXX-XXXXXXXX.dd.dd.dd.dd_fork
+# with the components being:
+# - WAL LSN in hex format,
+# - Decimal database oid
+# - Decimal tablespace oid or 0 for global
+# - Decimal relfilenumber
+# - Decimal block number
+# - fork this block came from (one of vm, init, fsm, or main)
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm|_main)?$/;
+
+my $file_count = 0;
+
+# verify filename formats matches w/--save-fpi
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+	$file_count++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk) = get_block_info($fullpath);
+
+	# verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+}
+
+# validate that we ended up with some FPIs saved
+ok($file_count > 0, 'verify we processed some files');
+
+done_testing();
-- 
2.37.1 (Apple Git-137.1)

From 573fa21454d032faef41b69c91d338fffbeda43c Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Fri, 23 Dec 2022 13:24:58 -0600
Subject: [PATCH v11] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a given target directory.
These images are subject to the same filtering rules as normal display in
pg_waldump, which means that you can isolate the full page writes to a
target relation, among other things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk>_<fork> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0_main
00000000-01000150.1664.0.6115.0_main
00000000-010001E0.1664.0.6114.0_main
00000000-01000270.1663.1.6116.0_main
00000000-01000300.1663.1.6113.0_main
00000000-01000390.1663.1.6112.0_main
00000000-01000420.1663.1.8903.0_main
00000000-010004B0.1663.1.8902.0_main
00000000-01000540.1663.1.6111.0_main
00000000-010005D0.1663.1.6110.0_main

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml          | 69 ++++++++++++++++
 src/bin/pg_waldump/meson.build            |  1 +
 src/bin/pg_waldump/pg_waldump.c           | 87 +++++++++++++++++++-
 src/bin/pg_waldump/t/002_save_fullpage.pl | 99 +++++++++++++++++++++++
 4 files changed, 255 insertions(+), 1 deletion(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5..090205e122 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,75 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fullpage=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>RELTABLESPACE</replaceable>.<replaceable>DATOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>RELTABLESPACE</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DATOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             The name of the fork the full page image came from.  One
+             of <literal>_main</literal>, <literal>_fsm</literal>,
+             <literal>_vm</literal>, or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/meson.build b/src/bin/pg_waldump/meson.build
index 3fa1b53e71..0428998350 100644
--- a/src/bin/pg_waldump/meson.build
+++ b/src/bin/pg_waldump/meson.build
@@ -31,6 +31,7 @@ tests += {
   'tap': {
     'tests': [
       't/001_basic.pl',
+      't/002_save_fullpage.pl',
     ],
   },
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..81aeef3b97 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -23,9 +23,13 @@
 #include "access/xlogrecord.h"
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
+#include "common/file_perm.h"
+#include "common/file_utils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,9 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	char	   *save_fpi_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +446,61 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPIs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPIs(XLogReaderState *record, const char *savepath)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		PGAlignedBlock buf;
+		Page		page;
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+		FILE	   *OPF;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (!XLogRecHasBlockImage(record, block_id))
+			continue;
+
+		page = (Page) buf.data;
+
+		if (!RestoreBlockImage(record, block_id, page))
+			continue;
+
+		/* we have our extracted FPI, let's save it now */
+
+		XLogRecGetBlockTagExtended(record, block_id,
+								   &rnode, &fork, &blk, NULL);
+
+		if (fork >= 0 && fork <= MAX_FORKNUM)
+			sprintf(forkname, "_%s", forkNames[fork]);
+		else
+			pg_fatal("found invalid fork number: %u", fork);
+
+		snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+				 LSN_FORMAT_ARGS(record->ReadRecPtr),
+				 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+		OPF = fopen(filename, PG_BINARY_W);
+		if (!OPF)
+			pg_fatal("could not open file \"%s\"", filename);
+
+		if (fwrite(page, BLCKSZ, 1, OPF) != 1)
+			pg_fatal("could not write file \"%s\"", filename);
+
+		fclose(OPF);
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +741,8 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --save-fullpage=path    save full page images to given path as\n"
+			 "                              LSN.T.D.R.B_F\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +776,7 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fullpage", required_argument, NULL, 'W'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +837,7 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpi_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +848,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -918,6 +984,9 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+				config.save_fpi_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1041,17 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpi_path != NULL)
+	{
+		/* Create the dir if it doesn't exist */
+		if (pg_mkdir_p(config.save_fpi_path, pg_dir_create_mode) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpi_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1230,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpi_path != NULL)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPIs(xlogreader_state, config.save_fpi_path);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..35bb12e067
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,99 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my ($blocksize, $walfile_name);
+
+# routine to extract the LSN from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if $blocksize != read $fh, $block, $blocksize;
+	my ($lsn_hi, $lsn_lo) = unpack('VV', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo);
+}
+
+my $node =  PostgreSQL::Test::Cluster->new('node_a');
+$node->init(allows_streaming => 1);
+$node->start;
+
+# Sanity checks for command line options.
+$node->command_fails(
+	[ 'pg_waldump', '--save-fullpage' ],
+	'--save-fullpage fails without path');
+
+# generate data/wal to examine that will have FPIs in them
+$node->safe_psql('postgres', "SELECT 'init' FROM pg_create_physical_replication_slot('regress_pg_waldump_slot', true, false);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT; -- required to force FPI for next writes
+UPDATE test_table SET a = a + 1;
+");
+
+($walfile_name, $blocksize) = split '\|' => $node->safe_psql('postgres',"SELECT pg_walfile_name(pg_switch_wal()), current_setting('block_size')");
+
+# get the relation node, etc for the new table
+my $relation = $node->safe_psql('postgres',
+	q{SELECT format(
+        '%s/%s/%s',
+        CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END,
+        pg_database.oid,
+        pg_relation_filenode(pg_class.oid))
+    FROM pg_class, pg_database
+    WHERE relname = 'test_table' AND
+        datname = current_database()}
+);
+
+my $walfile = $node->data_dir . '/pg_wal/' . $walfile_name;
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+
+ok(-f $walfile, "Got a WAL file");
+
+$node->command_ok(['pg_waldump', '--save-fullpage', "$tmp_folder/raw", '--relation', $relation, $walfile]);
+
+# This regexp will match filenames of the format XXXXXXXX-XXXXXXXX.dd.dd.dd.dd_fork
+# with the components being:
+# - WAL LSN in hex format,
+# - Decimal database oid
+# - Decimal tablespace oid or 0 for global
+# - Decimal relfilenumber
+# - Decimal block number
+# - fork this block came from (one of vm, init, fsm, or main)
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm|_main)?$/;
+
+my $file_count = 0;
+
+# verify filename formats matches w/--save-fullpage
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+	$file_count++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk) = get_block_info($fullpath);
+
+	# verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+}
+
+# validate that we ended up with some FPIs saved
+ok($file_count > 0, 'verify we processed some files');
+
+done_testing();
-- 
2.37.1 (Apple Git-137.1)

From cee0c0017342fc8847e4b099358245fac6046a98 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Date: Sat, 24 Dec 2022 12:39:12 +0000
Subject: [PATCH v12] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a given target directory.
These images are subject to the same filtering rules as normal display in
pg_waldump, which means that you can isolate the full page writes to a
target relation, among other things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk>_<fork> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0_main
00000000-01000150.1664.0.6115.0_main
00000000-010001E0.1664.0.6114.0_main
00000000-01000270.1663.1.6116.0_main
00000000-01000300.1663.1.6113.0_main
00000000-01000390.1663.1.6112.0_main
00000000-01000420.1663.1.8903.0_main
00000000-010004B0.1663.1.8902.0_main
00000000-01000540.1663.1.6111.0_main
00000000-010005D0.1663.1.6110.0_main

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 doc/src/sgml/ref/pg_waldump.sgml     |  69 ++++++++++++++++++
 src/bin/pg_waldump/meson.build       |   1 +
 src/bin/pg_waldump/pg_waldump.c      |  87 +++++++++++++++++++++-
 src/bin/pg_waldump/t/002_save_fpi.pl | 104 +++++++++++++++++++++++++++
 4 files changed, 260 insertions(+), 1 deletion(-)
 create mode 100644 src/bin/pg_waldump/t/002_save_fpi.pl

diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5..090205e122 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,75 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>-W <replaceable>save_path</replaceable></option></term>
+       <term><option>--save-fullpage=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images seen in the WAL stream to the
+        <replaceable>save_path</replaceable> directory, which will be created
+        if it does not exist.  The images saved will be subject to the same
+        filtering and limiting criteria as display records, but in this
+        mode <application>pg_waldump</application> will not output any other
+        information.
+       </para>
+       <para>
+        The page images will be saved with the file
+        format: <literal><replaceable>LSN</replaceable>.<replaceable>RELTABLESPACE</replaceable>.<replaceable>DATOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The dot-separated components are (in order):
+
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The LSN of the record with this block, formatted
+             as two 8-character hexadecimal numbers <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>RELTABLESPACE</entry>
+            <entry>tablespace OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>DATOID</entry>
+            <entry>database OID for the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>relnode id for the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>the block number of this block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             The name of the fork the full page image came from.  One
+             of <literal>_main</literal>, <literal>_fsm</literal>,
+             <literal>_vm</literal>, or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
diff --git a/src/bin/pg_waldump/meson.build b/src/bin/pg_waldump/meson.build
index 3fa1b53e71..98c6de1a80 100644
--- a/src/bin/pg_waldump/meson.build
+++ b/src/bin/pg_waldump/meson.build
@@ -31,6 +31,7 @@ tests += {
   'tap': {
     'tests': [
       't/001_basic.pl',
+      't/002_save_fpi.pl',
     ],
   },
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..81aeef3b97 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -23,9 +23,13 @@
 #include "access/xlogrecord.h"
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
+#include "common/file_perm.h"
+#include "common/file_utils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,9 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	char	   *save_fpi_path;
 } XLogDumpConfig;
 
 
@@ -439,6 +446,61 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPIs stored in the given WAL record
+ */
+static void
+XLogRecordSaveFPIs(XLogReaderState *record, const char *savepath)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		PGAlignedBlock buf;
+		Page		page;
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + \0 */
+		FILE	   *OPF;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (!XLogRecHasBlockImage(record, block_id))
+			continue;
+
+		page = (Page) buf.data;
+
+		if (!RestoreBlockImage(record, block_id, page))
+			continue;
+
+		/* we have our extracted FPI, let's save it now */
+
+		XLogRecGetBlockTagExtended(record, block_id,
+								   &rnode, &fork, &blk, NULL);
+
+		if (fork >= 0 && fork <= MAX_FORKNUM)
+			sprintf(forkname, "_%s", forkNames[fork]);
+		else
+			pg_fatal("found invalid fork number: %u", fork);
+
+		snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+				 LSN_FORMAT_ARGS(record->ReadRecPtr),
+				 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+		OPF = fopen(filename, PG_BINARY_W);
+		if (!OPF)
+			pg_fatal("could not open file \"%s\"", filename);
+
+		if (fwrite(page, BLCKSZ, 1, OPF) != 1)
+			pg_fatal("could not write file \"%s\"", filename);
+
+		fclose(OPF);
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +741,8 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("  -W, --save-fullpage=path    save full page images to given path as\n"
+			 "                              LSN.T.D.R.B_F\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -712,6 +776,7 @@ main(int argc, char **argv)
 		{"limit", required_argument, NULL, 'n'},
 		{"path", required_argument, NULL, 'p'},
 		{"quiet", no_argument, NULL, 'q'},
+		{"save-fullpage", required_argument, NULL, 'W'},
 		{"relation", required_argument, NULL, 'R'},
 		{"rmgr", required_argument, NULL, 'r'},
 		{"start", required_argument, NULL, 's'},
@@ -772,6 +837,7 @@ main(int argc, char **argv)
 	config.filter_by_fpw = false;
 	config.stats = false;
 	config.stats_per_record = false;
+	config.save_fpi_path = NULL;
 
 	stats.startptr = InvalidXLogRecPtr;
 	stats.endptr = InvalidXLogRecPtr;
@@ -782,7 +848,7 @@ main(int argc, char **argv)
 		goto bad_argument;
 	}
 
-	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wx:z",
+	while ((option = getopt_long(argc, argv, "bB:e:fF:n:p:qr:R:s:t:wW:x:z",
 								 long_options, &optindex)) != -1)
 	{
 		switch (option)
@@ -918,6 +984,9 @@ main(int argc, char **argv)
 			case 'w':
 				config.filter_by_fpw = true;
 				break;
+			case 'W':
+				config.save_fpi_path = pg_strdup(optarg);
+				break;
 			case 'x':
 				if (sscanf(optarg, "%u", &config.filter_by_xid) != 1)
 				{
@@ -972,6 +1041,17 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fpi_path != NULL)
+	{
+		/* Create the dir if it doesn't exist */
+		if (pg_mkdir_p(config.save_fpi_path, pg_dir_create_mode) < 0)
+		{
+			pg_log_error("could not create output directory \"%s\": %m",
+						 config.save_fpi_path);
+			goto bad_argument;
+		}
+	}
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1150,6 +1230,11 @@ main(int argc, char **argv)
 				XLogRecStoreStats(&stats, xlogreader_state);
 				stats.endptr = xlogreader_state->EndRecPtr;
 			}
+			else if (config.save_fpi_path != NULL)
+			{
+				if (XLogRecordHasFPW(xlogreader_state))
+					XLogRecordSaveFPIs(xlogreader_state, config.save_fpi_path);
+			}
 			else
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
diff --git a/src/bin/pg_waldump/t/002_save_fpi.pl b/src/bin/pg_waldump/t/002_save_fpi.pl
new file mode 100644
index 0000000000..244692320f
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fpi.pl
@@ -0,0 +1,104 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my ($blocksize, $walfile_name);
+
+# Function to extract the LSN from the given block structure
+sub get_block_info
+{
+	my $path = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "couldn't read full block\n" if $blocksize != read $fh, $block, $blocksize;
+	my ($lsn_hi, $lsn_lo) = unpack('VV', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo);
+}
+
+my $node =  PostgreSQL::Test::Cluster->new('main');
+$node->init;
+$node->append_conf(
+	'postgresql.conf', q{
+wal_level = 'replica'
+max_wal_senders = 4
+});
+$node->start;
+
+# Sanity checks for command line options.
+$node->command_fails(
+	[ 'pg_waldump', '--save-fullpage' ],
+	'--save-fullpage fails without path');
+
+# Generate data/WAL to examine that will have FPIs in them
+$node->safe_psql('postgres', "SELECT 'init' FROM pg_create_physical_replication_slot('regress_pg_waldump_slot', true, false);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+CHECKPOINT; -- required to force FPI for next writes
+UPDATE test_table SET a = a + 1;
+");
+
+($walfile_name, $blocksize) = split '\|' => $node->safe_psql('postgres',"SELECT pg_walfile_name(pg_switch_wal()), current_setting('block_size')");
+
+# Get the relation node, etc for the new table
+my $relation = $node->safe_psql('postgres',
+	q{SELECT format(
+        '%s/%s/%s',
+        CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END,
+        pg_database.oid,
+        pg_relation_filenode(pg_class.oid))
+    FROM pg_class, pg_database
+    WHERE relname = 'test_table' AND
+        datname = current_database()}
+);
+
+my $walfile = $node->data_dir . '/pg_wal/' . $walfile_name;
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+
+ok(-f $walfile, "Got a WAL file");
+
+$node->command_ok(['pg_waldump', '--save-fullpage', "$tmp_folder/raw", '--relation', $relation, $walfile]);
+
+# This regexp will match filenames of the format XXXXXXXX-XXXXXXXX.dd.dd.dd.dd_fork
+# with the components being:
+# - WAL LSN in hex format,
+# - Decimal database oid
+# - Decimal tablespace oid or 0 for global
+# - Decimal relfilenumber
+# - Decimal block number
+# - fork this block came from (one of vm, init, fsm, or main)
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm|_main)?$/;
+
+my $file_count = 0;
+
+# Verify filename formats matches w/--save-fullpage
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+	$file_count++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk) = get_block_info($fullpath);
+
+	# Verify the lsn in the block comes before the file's lsn
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'verify file-based LSN precedes block-based one');
+}
+
+# Validate that we ended up with some FPIs saved
+ok($file_count > 0, 'verify we processed some files');
+
+done_testing();
-- 
2.34.1

From 66eedbb96858a4f6804509900b116d8a63b39925 Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@paquier.xyz>
Date: Mon, 26 Dec 2022 16:28:00 +0900
Subject: [PATCH v13] Teach pg_waldump to extract FPIs from the WAL stream

Extracts full-page images from the WAL stream into a given target directory.
These images are subject to the same filtering rules as normal display in
pg_waldump, which means that you can isolate the full page writes to a
target relation, among other things.

Files are saved with the filename: <lsn>.<ts>.<db>.<rel>.<blk>_<fork> with
formatting to make things somewhat sortable; for instance:

00000000-010000C0.1663.1.6117.0_main
00000000-01000150.1664.0.6115.0_main
00000000-010001E0.1664.0.6114.0_main
00000000-01000270.1663.1.6116.0_main
00000000-01000300.1663.1.6113.0_main
00000000-01000390.1663.1.6112.0_main
00000000-01000420.1663.1.8903.0_main
00000000-010004B0.1663.1.8902.0_main
00000000-01000540.1663.1.6111.0_main
00000000-010005D0.1663.1.6110.0_main

It's noteworthy that the raw block images do not have the current LSN
stored with them in the WAL stream (as would be true for on-heap
versions of the blocks), nor would the checksum be updated in
them (though WAL itself has checksums, so there is some protection
there).

These images could be loaded/inspected via `pg_read_binary_file()` and
used in the `pageinspect` suite of tools to perform detailed analysis on
the pages in question, based on historical information, and may come in
handy for forensics work.
---
 src/bin/pg_waldump/meson.build            |   1 +
 src/bin/pg_waldump/pg_waldump.c           | 107 +++++++++++++++++++++
 src/bin/pg_waldump/t/002_save_fullpage.pl | 111 ++++++++++++++++++++++
 doc/src/sgml/ref/pg_waldump.sgml          |  66 +++++++++++++
 4 files changed, 285 insertions(+)
 create mode 100644 src/bin/pg_waldump/t/002_save_fullpage.pl

diff --git a/src/bin/pg_waldump/meson.build b/src/bin/pg_waldump/meson.build
index 3fa1b53e71..0428998350 100644
--- a/src/bin/pg_waldump/meson.build
+++ b/src/bin/pg_waldump/meson.build
@@ -31,6 +31,7 @@ tests += {
   'tap': {
     'tests': [
       't/001_basic.pl',
+      't/002_save_fullpage.pl',
     ],
   },
 }
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 9993378ca5..d90a142c68 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -23,9 +23,13 @@
 #include "access/xlogrecord.h"
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
+#include "common/file_perm.h"
+#include "common/file_utils.h"
 #include "common/logging.h"
+#include "common/relpath.h"
 #include "getopt_long.h"
 #include "rmgrdesc.h"
+#include "storage/bufpage.h"
 
 /*
  * NOTE: For any code change or issue fix here, it is highly recommended to
@@ -70,6 +74,9 @@ typedef struct XLogDumpConfig
 	bool		filter_by_relation_block_enabled;
 	ForkNumber	filter_by_relation_forknum;
 	bool		filter_by_fpw;
+
+	/* save options */
+	char	   *save_fullpage_path;
 } XLogDumpConfig;
 
 
@@ -112,6 +119,37 @@ verify_directory(const char *directory)
 	return true;
 }
 
+/*
+ * Create if necessary the directory storing the full-page images extracted
+ * from the WAL records read.
+ */
+static void
+create_fullpage_directory(char *path)
+{
+	int			ret;
+
+	switch ((ret = pg_check_dir(path)))
+	{
+		case 0:
+			/* Does not exist, so create it */
+			if (pg_mkdir_p(path, pg_dir_create_mode) < 0)
+				pg_fatal("could not create directory \"%s\": %m", path);
+			break;
+		case 1:
+			/* Present and empty, so do nothing */
+			break;
+		case 2:
+		case 3:
+		case 4:
+			/* Exists and not empty */
+			pg_fatal("directory \"%s\" exists but is not empty", path);
+			break;
+		default:
+			/* Trouble accessing directory */
+			pg_fatal("could not access directory \"%s\": %m", path);
+	}
+}
+
 /*
  * Split a pathname as dirname(1) and basename(1) would.
  *
@@ -439,6 +477,61 @@ XLogRecordHasFPW(XLogReaderState *record)
 	return false;
 }
 
+/*
+ * Function to externally save all FPWs stored in the given WAL record.
+ * Decompression is applied to all the blocks saved, if necessary.
+ */
+static void
+XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath)
+{
+	int			block_id;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		PGAlignedBlock buf;
+		Page		page;
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + terminating zero */
+		FILE	   *file;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		if (!XLogRecHasBlockImage(record, block_id))
+			continue;
+
+		page = (Page) buf.data;
+
+		/* Full page exists, so let's save it */
+		if (!RestoreBlockImage(record, block_id, page))
+			pg_fatal("%s", record->errormsg_buf);
+
+		(void) XLogRecGetBlockTagExtended(record, block_id,
+										  &rnode, &fork, &blk, NULL);
+
+		if (fork >= 0 && fork <= MAX_FORKNUM)
+			sprintf(forkname, "_%s", forkNames[fork]);
+		else
+			pg_fatal("invalid fork number: %u", fork);
+
+		snprintf(filename, MAXPGPATH, "%s/%08X-%08X.%u.%u.%u.%u%s", savepath,
+				 LSN_FORMAT_ARGS(record->ReadRecPtr),
+				 rnode.spcOid, rnode.dbOid, rnode.relNumber, blk, forkname);
+
+		file = fopen(filename, PG_BINARY_W);
+		if (!file)
+			pg_fatal("could not open file \"%s\": %m", filename);
+
+		if (fwrite(page, BLCKSZ, 1, file) != 1)
+			pg_fatal("could not write file \"%s\": %m", filename);
+
+		fclose(file);
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -679,6 +772,8 @@ usage(void)
 			 "                         (default: 1 or the value used in STARTSEG)\n"));
 	printf(_("  -V, --version          output version information, then exit\n"));
 	printf(_("  -w, --fullpage         only show records with a full page write\n"));
+	printf(_("      --save-fullpage=PATH\n"
+			 "                         save full page images\n"));
 	printf(_("  -x, --xid=XID          only show records with transaction ID XID\n"));
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
@@ -719,6 +814,7 @@ main(int argc, char **argv)
 		{"xid", required_argument, NULL, 'x'},
 		{"version", no_argument, NULL, 'V'},
 		{"stats", optional_argument, NULL, 'z'},
+		{"save-fullpage", required_argument, NULL, 1},
 		{NULL, 0, NULL, 0}
 	};
 
@@ -770,6 +866,7 @@ main(int argc, char **argv)
 	config.filter_by_relation_block_enabled = false;
 	config.filter_by_relation_forknum = InvalidForkNumber;
 	config.filter_by_fpw = false;
+	config.save_fullpage_path = NULL;
 	config.stats = false;
 	config.stats_per_record = false;
 
@@ -942,6 +1039,9 @@ main(int argc, char **argv)
 					}
 				}
 				break;
+			case 1:
+				config.save_fullpage_path = pg_strdup(optarg);
+				break;
 			default:
 				goto bad_argument;
 		}
@@ -972,6 +1072,9 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (config.save_fullpage_path != NULL)
+		create_fullpage_directory(config.save_fullpage_path);
+
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
 	{
@@ -1154,6 +1257,10 @@ main(int argc, char **argv)
 				XLogDumpDisplayRecord(&config, xlogreader_state);
 		}
 
+		/* save full pages if requested */
+		if (config.save_fullpage_path != NULL)
+			XLogRecordSaveFPWs(xlogreader_state, config.save_fullpage_path);
+
 		/* check whether we printed enough */
 		config.already_displayed_records++;
 		if (config.stop_after_records > 0 &&
diff --git a/src/bin/pg_waldump/t/002_save_fullpage.pl b/src/bin/pg_waldump/t/002_save_fullpage.pl
new file mode 100644
index 0000000000..eac3c25a4e
--- /dev/null
+++ b/src/bin/pg_waldump/t/002_save_fullpage.pl
@@ -0,0 +1,111 @@
+
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use File::Basename;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::RecursiveCopy;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my ($blocksize, $walfile_name);
+
+# Function to extract the LSN from the given block structure
+sub get_block_info
+{
+	my $path      = shift;
+	my $blocksize = shift;
+	my $block;
+
+	open my $fh, '<', $path or die "couldn't open file: $path\n";
+	die "could not read block\n"
+	  if $blocksize != read($fh, $block, $blocksize);
+	my ($lsn_hi, $lsn_lo) = unpack('VV', $block);
+
+	$lsn_hi = sprintf('%08X', $lsn_hi);
+	$lsn_lo = sprintf('%08X', $lsn_lo);
+
+	return ($lsn_hi, $lsn_lo);
+}
+
+my $node = PostgreSQL::Test::Cluster->new('main');
+$node->init;
+$node->append_conf(
+	'postgresql.conf', q{
+wal_level = 'replica'
+max_wal_senders = 4
+});
+$node->start;
+
+# Generate data/WAL to examine that will have full pages in them.
+$node->safe_psql(
+	'postgres',
+	"SELECT 'init' FROM pg_create_physical_replication_slot('regress_pg_waldump_slot', true, false);
+CREATE TABLE test_table AS SELECT generate_series(1,100) a;
+-- Force FPWs on the next writes.
+CHECKPOINT;
+UPDATE test_table SET a = a + 1;
+");
+
+($walfile_name, $blocksize) = split '\|' => $node->safe_psql('postgres',
+	"SELECT pg_walfile_name(pg_switch_wal()), current_setting('block_size')");
+
+# Get the relation node, etc for the new table
+my $relation = $node->safe_psql(
+	'postgres',
+	q{SELECT format(
+        '%s/%s/%s',
+        CASE WHEN reltablespace = 0 THEN dattablespace ELSE reltablespace END,
+        pg_database.oid,
+        pg_relation_filenode(pg_class.oid))
+    FROM pg_class, pg_database
+    WHERE relname = 'test_table' AND
+        datname = current_database()}
+);
+
+my $walfile    = $node->data_dir . '/pg_wal/' . $walfile_name;
+my $tmp_folder = PostgreSQL::Test::Utils::tempdir;
+
+ok(-f $walfile, "Got a WAL file");
+
+$node->command_ok(
+	[
+		'pg_waldump',      '--quiet',
+		'--save-fullpage', "$tmp_folder/raw",
+		'--relation',      $relation,
+		$walfile
+	]);
+
+# This regexp will match filenames formatted as:
+# XXXXXXXX-XXXXXXXX.dd.dd.dd.dd_fork with the components being:
+# - WAL LSN in hex format,
+# - Decimal database OID.
+# - Decimal tablespace OID or 0 for global.
+# - Decimal relfilenode.
+# - Decimal block number.
+# - Fork this block came from (vm, init, fsm, or main)
+my $file_re =
+  qr/^([0-9A-F]{8})-([0-9A-F]{8})[.][0-9]+[.][0-9]+[.][0-9]+[.][0-9]+(?:_vm|_init|_fsm|_main)?$/;
+
+my $file_count = 0;
+
+# Verify filename formats matches --save-fullpage.
+for my $fullpath (glob "$tmp_folder/raw/*")
+{
+	my $file = File::Basename::basename($fullpath);
+
+	like($file, $file_re, "verify filename format for file $file");
+	$file_count++;
+
+	my ($hi_lsn_fn, $lo_lsn_fn) = ($file =~ $file_re);
+	my ($hi_lsn_bk, $lo_lsn_bk) = get_block_info($fullpath, $blocksize);
+
+	# The LSN on the block comes before the file's LSN.
+	ok( $hi_lsn_fn . $lo_lsn_fn gt $hi_lsn_bk . $lo_lsn_bk,
+		'LSN stored in the file precedes the one stored in the block');
+}
+
+ok($file_count > 0, 'verify that at least one block has been saved');
+
+done_testing();
diff --git a/doc/src/sgml/ref/pg_waldump.sgml b/doc/src/sgml/ref/pg_waldump.sgml
index d559f091e5..343f0482a9 100644
--- a/doc/src/sgml/ref/pg_waldump.sgml
+++ b/doc/src/sgml/ref/pg_waldump.sgml
@@ -240,6 +240,72 @@ PostgreSQL documentation
        </listitem>
      </varlistentry>
 
+     <varlistentry>
+       <term><option>--save-fullpage=<replaceable>save_path</replaceable></option></term>
+       <listitem>
+       <para>
+        Save full page images found in the WAL records to the
+        <replaceable>save_path</replaceable> directory. The images saved
+        are subject to the same filtering and limiting criteria as the
+        records displayed.
+       </para>
+       <para>
+        The full page images are saved with the following file name format:
+        <literal><replaceable>LSN</replaceable>.<replaceable>RELTABLESPACE</replaceable>.<replaceable>DATOID</replaceable>.<replaceable>RELNODE</replaceable>.<replaceable>BLKNO</replaceable><replaceable>FORK</replaceable></literal>
+
+        The file names are composed of the following parts:
+        <informaltable>
+         <tgroup cols="2">
+          <thead>
+           <row>
+            <entry>Component</entry>
+            <entry>Description</entry>
+           </row>
+          </thead>
+
+          <tbody>
+           <row>
+            <entry>LSN</entry>
+            <entry>The <acronym>LSN</acronym> of the record with this image,
+             formatted as two 8-character hexadecimal numbers
+             <literal>%08X-%08X</literal></entry>
+           </row>
+
+           <row>
+            <entry>RELTABLESPACE</entry>
+            <entry>tablespace OID of the block</entry>
+           </row>
+
+           <row>
+            <entry>DATOID</entry>
+            <entry>database OID of the block</entry>
+           </row>
+
+           <row>
+            <entry>RELNODE</entry>
+            <entry>filenode of the block</entry>
+           </row>
+
+           <row>
+            <entry>BLKNO</entry>
+            <entry>block number of the block</entry>
+           </row>
+
+           <row>
+            <entry>FORK</entry>
+            <entry>
+             The name of the fork the full page image came from, as of
+             <literal>_main</literal>, <literal>_fsm</literal>,
+             <literal>_vm</literal>, or <literal>_init</literal>.
+            </entry>
+           </row>
+          </tbody>
+         </tgroup>
+        </informaltable>
+       </para>
+       </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>-x <replaceable>xid</replaceable></option></term>
       <term><option>--xid=<replaceable>xid</replaceable></option></term>
-- 
2.39.0