From 37ef234284d0fb95df02677f3b4a7927c72186a5 Mon Sep 17 00:00:00 2001
From: Julien Rouhaud <julien.rouhaud@free.fr>
Date: Thu, 9 Feb 2023 17:33:17 +0800
Subject: [PATCH] Add a --save-records=PATH option to pg_waldump

Ignore anything that's not a HEAP / INSERT[+INIT] on a user table.

Note that this option will extract (and reformat) xl_heap_insert records from
full-page images.

Author: Julien Rouhaud
---
 src/bin/pg_waldump/pg_waldump.c | 192 ++++++++++++++++++++++++++++++++
 1 file changed, 192 insertions(+)

diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index a3535bdfa9..b23ceb1e4c 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -18,10 +18,13 @@
 #include <sys/stat.h>
 #include <unistd.h>
 
+#include "access/heapam_xlog.h"
+#include "access/htup_details.h"
 #include "access/transam.h"
 #include "access/xlog_internal.h"
 #include "access/xlogreader.h"
 #include "access/xlogrecord.h"
+#include "access/xlogutils.h"
 #include "access/xlogstats.h"
 #include "common/fe_memutils.h"
 #include "common/file_perm.h"
@@ -78,6 +81,7 @@ typedef struct XLogDumpConfig
 
 	/* save options */
 	char	   *save_fullpage_path;
+	char	   *save_record_path;
 } XLogDumpConfig;
 
 
@@ -539,6 +543,182 @@ XLogRecordSaveFPWs(XLogReaderState *record, const char *savepath)
 	}
 }
 
+/*
+ * Extract the record, ie. the chunk at the given offset, for a given FPI of a
+ * HEAP record, with the exact same format as it it were a normal
+ * xl_heap_insert record.
+ */
+static char *
+extract_heap_record_from_fpw(XLogReaderState *record, const char *data,
+							 Size *len)
+{
+	char	   *rec = XLogRecGetData(record);
+	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
+	xl_heap_insert *xlrec;
+	OffsetNumber lineoff;
+	Page		dp;
+	int			lines;
+	ItemId		lpp;
+	char	   *extracted;
+	int			fullsize;
+	HeapTupleHeader htup;
+	xl_heap_header *xlhdr;
+
+	/* see heap_desc() for how to get the offset */
+	info &= XLOG_HEAP_OPMASK;
+	if (info != XLOG_HEAP_INSERT)
+		pg_fatal("Unsupportes HEAP operation: 0x%02x\n", info);
+
+	xlrec = (xl_heap_insert *) rec;
+	lineoff = xlrec->offnum;
+
+	/* see heapgetpage() / heapgettup() for how to get the data chunk */
+	dp = (Page) data;
+	lines = PageGetMaxOffsetNumber(dp);
+
+	if (lineoff > lines)
+		pg_fatal("off %u > max value %d", lineoff, lines);
+
+	lpp = PageGetItemId(dp, lineoff);
+
+	if (!ItemIdIsNormal(lpp))
+		pg_fatal("ItemId %u:%u:%u for off %u is not LP_NORMAL",
+					lpp->lp_off, lpp->lp_flags, lpp->lp_len, lineoff);
+
+	if (ItemIdGetLength(lpp) == 0)
+		pg_fatal("Item does not have data\n");
+	if (ItemIdGetLength(lpp) >= MaxHeapTupleSize)
+		pg_fatal("Item length seems wrong\n");
+	if (ItemIdGetOffset(lpp) + ItemIdGetLength(lpp) >= BLCKSZ)
+		pg_fatal("Item points to data outside of the page\n");
+
+	/*
+	 * We want to generate the same data as a normal HEAP INSERT record, which
+	 * means a xl_heap_header followed by a HeapTuple (without its header).
+	 */
+	fullsize = SizeOfHeapHeader + ItemIdGetLength(lpp) - SizeofHeapTupleHeader;
+	extracted = malloc(sizeof(char) * fullsize);
+
+	/* Generate the expected xl_heap_header */
+	memset(extracted, 0, sizeof(char) + SizeOfHeapHeader);
+	xlhdr = (xl_heap_header *) extracted;
+	htup = (HeapTupleHeader) PageGetItem((Page) dp, lpp);
+	xlhdr->t_infomask2 = htup->t_infomask2;
+	xlhdr->t_infomask = htup->t_infomask;
+	xlhdr->t_hoff = htup->t_hoff;
+
+	/* And append the actual data, minus the HeapTupleHeader */
+	memcpy(extracted + SizeOfHeapHeader,
+		   PageGetItem((Page) dp, lpp) + SizeofHeapTupleHeader,
+		   sizeof(char) * (ItemIdGetLength(lpp) - SizeofHeapTupleHeader));
+	*len = fullsize;
+
+	return extracted;
+}
+
+/*
+ * Function to externally save the record data stored in the given WAL record.
+ * For now, ignore anything that's not a HEAP / INSERT record on a user table.
+ */
+static void
+XLogRecordSaveRecord(XLogReaderState *record, const char *savepath)
+{
+	const char *id;
+	const RmgrDescData *desc = GetRmgrDesc(XLogRecGetRmid(record));
+	uint8		info = XLogRecGetInfo(record);
+	int			rmgr_id = XLogRecGetRmid(record);
+	int			block_id;
+
+	if (rmgr_id != RM_HEAP_ID)
+		return;
+
+	id = desc->rm_identify(info);
+
+	/* we want both "INSERT" and "INSERT+INIT" */
+	if (!id || (strncmp(id, "INSERT", strlen("INSERT")) != 0))
+		return;
+
+	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
+	{
+		char	   *data;
+		Size		len;
+		char		filename[MAXPGPATH];
+		char		forkname[FORKNAMECHARS + 2];	/* _ + terminating zero */
+		FILE	   *file;
+		BlockNumber blk;
+		RelFileLocator rnode;
+		ForkNumber	fork;
+		PGAlignedXLogBlock page;
+		bool		freeit = false;
+
+		/* hack, for heap_insert we only want data from block 0 */
+		if (block_id != 0)
+			continue;
+
+		if (!XLogRecHasBlockRef(record, block_id))
+			continue;
+
+		data = XLogRecGetBlockData(record, block_id, &len);
+		if (len == 0)
+		{
+			if (!RestoreBlockImage(record, block_id, page.data))
+				continue;
+
+			data = extract_heap_record_from_fpw(record, page.data, &len);
+
+			if (len == 0)
+			{
+				printf("could not extract record\n");
+				continue;
+			}
+			else
+			{
+				/*
+				 * we malloc'ed some data, need to free it from that point
+				 * before looping.
+				 */
+				freeit = true;
+			}
+		}
+
+		(void) XLogRecGetBlockTagExtended(record, block_id,
+										  &rnode, &fork, &blk, NULL);
+
+		/*
+		 * Ignore activity on system tables, assuming a relfilenode/oid mapping
+		 */
+		if (rnode.relNumber < FirstNormalObjectId)
+			goto may_free_continue;
+
+		/* Ignore activity on shared tables */
+		if (rnode.dbOid == InvalidOid)
+			goto may_free_continue;
+
+		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(data, len, 1, file) != 1)
+			pg_fatal("could not write file \"%s\": %m", filename);
+
+		if (fclose(file) != 0)
+			pg_fatal("could not close file \"%s\": %m", filename);
+
+may_free_continue:
+		if (freeit)
+			free(data);
+	}
+}
+
 /*
  * Print a record to stdout
  */
@@ -783,6 +963,7 @@ usage(void)
 	printf(_("  -z, --stats[=record]   show statistics instead of records\n"
 			 "                         (optionally, show per-record statistics)\n"));
 	printf(_("  --save-fullpage=DIR    save full page images to DIR\n"));
+	printf(_("  --save-records=DIR     save all records to DIR\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);
@@ -821,6 +1002,7 @@ main(int argc, char **argv)
 		{"version", no_argument, NULL, 'V'},
 		{"stats", optional_argument, NULL, 'z'},
 		{"save-fullpage", required_argument, NULL, 1},
+		{"save-records", required_argument, NULL, 2},
 		{NULL, 0, NULL, 0}
 	};
 
@@ -873,6 +1055,7 @@ main(int argc, char **argv)
 	config.filter_by_relation_forknum = InvalidForkNumber;
 	config.filter_by_fpw = false;
 	config.save_fullpage_path = NULL;
+	config.save_record_path = NULL;
 	config.stats = false;
 	config.stats_per_record = false;
 
@@ -1076,6 +1259,9 @@ main(int argc, char **argv)
 			case 1:
 				config.save_fullpage_path = pg_strdup(optarg);
 				break;
+			case 2:
+				config.save_record_path = pg_strdup(optarg);
+				break;
 			default:
 				goto bad_argument;
 		}
@@ -1108,6 +1294,8 @@ main(int argc, char **argv)
 
 	if (config.save_fullpage_path != NULL)
 		create_fullpage_directory(config.save_fullpage_path);
+	if (config.save_record_path != NULL)
+		create_fullpage_directory(config.save_record_path);
 
 	/* parse files as start/end boundaries, extract path if not specified */
 	if (optind < argc)
@@ -1295,6 +1483,10 @@ main(int argc, char **argv)
 		if (config.save_fullpage_path != NULL)
 			XLogRecordSaveFPWs(xlogreader_state, config.save_fullpage_path);
 
+		/* save record if requested */
+		if (config.save_record_path != NULL)
+			XLogRecordSaveRecord(xlogreader_state, config.save_record_path);
+
 		/* check whether we printed enough */
 		config.already_displayed_records++;
 		if (config.stop_after_records > 0 &&
-- 
2.37.0

