Show record information in pg_get_wal_block_info.

Expand the output parameters in pg_walinspect's pg_get_wal_block_info
function to return additional information that was previously only
available from pg_walinspect's pg_get_wal_records_info function.  Some
of the details are attributed to individual block references, rather
than aggregated into whole-record values, since the function returns one
row per block reference per WAL record (unlike pg_get_wal_records_info,
which always returns one row per WAL record).

This structure is much easier to work with when writing queries that
track how individual blocks changed over time, or when attributing costs
to individual blocks (not WAL records) is useful.

This is the second time that pg_get_wal_block_info has been enhanced in
recent weeks.  Commit 9ecb134a expanded on the original version of the
function added in commit c31cf1c0 (where it first appeared under the
name pg_get_wal_fpi_info).  There still hasn't been a stable release
since commit c31cf1c0, so no bump in the pg_walinspect extension
version.

Author: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Author: Peter Geoghegan <pg@bowt.ie>
Reviewed-By: Peter Geoghegan <pg@bowt.ie>
Reviewed-By: Kyotaro HORIGUCHI <horikyota.ntt@gmail.com>
Discussion: https://postgr.es/m/CALj2ACVRK5=Z+2ZVsjgTTSkfEnQzCuwny7iigpG7g1btk4Ws2A@mail.gmail.com

4 files changed, 140 insertions, 88 deletions

diff --git a/contrib/pg_walinspect/expected/pg_walinspect.out b/contrib/pg_walinspect/expected/pg_walinspect.out
index b9e2bb994c..950f0e9217 100644
--- a/contrib/pg_walinspect/expected/pg_walinspect.out
+++ b/contrib/pg_walinspect/expected/pg_walinspect.out
@@ -121,7 +121,7 @@ UPDATE sample_tbl SET col1 = col1 + 1 WHERE col1 = 1;
 SELECT pg_current_wal_lsn() AS wal_lsn4 \gset
 -- Check if we get block data from WAL record.
 SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_block_info(:'wal_lsn3', :'wal_lsn4')
-  WHERE relfilenode = :'sample_tbl_oid' AND blockdata IS NOT NULL;
+  WHERE relfilenode = :'sample_tbl_oid' AND block_data IS NOT NULL;
  ok 
 ----
  t
@@ -134,7 +134,7 @@ UPDATE sample_tbl SET col1 = col1 + 1 WHERE col1 = 2;
 SELECT pg_current_wal_lsn() AS wal_lsn6 \gset
 -- Check if we get FPI from WAL record.
 SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_block_info(:'wal_lsn5', :'wal_lsn6')
-  WHERE relfilenode = :'sample_tbl_oid' AND fpi IS NOT NULL;
+  WHERE relfilenode = :'sample_tbl_oid' AND block_fpi_data IS NOT NULL;
  ok 
 ----
  t
diff --git a/contrib/pg_walinspect/pg_walinspect--1.0--1.1.sql b/contrib/pg_walinspect/pg_walinspect--1.0--1.1.sql
index 586c3b4467..fdd75e8a72 100644
--- a/contrib/pg_walinspect/pg_walinspect--1.0--1.1.sql
+++ b/contrib/pg_walinspect/pg_walinspect--1.0--1.1.sql
@@ -12,17 +12,26 @@ DROP FUNCTION pg_get_wal_stats_till_end_of_wal(pg_lsn, boolean);
 --
 CREATE FUNCTION pg_get_wal_block_info(IN start_lsn pg_lsn,
 	IN end_lsn pg_lsn,
-	OUT lsn pg_lsn,
-	OUT blockid int2,
+	OUT start_lsn pg_lsn,
+	OUT end_lsn pg_lsn,
+	OUT prev_lsn pg_lsn,
+	OUT block_id int2,
 	OUT reltablespace oid,
 	OUT reldatabase oid,
 	OUT relfilenode oid,
+	OUT relforknumber int2,
 	OUT relblocknumber int8,
-	OUT forkname text,
-	OUT blockdata bytea,
-	OUT fpi bytea,
-	OUT fpilen int4,
-	OUT fpiinfo text[]
+	OUT xid xid,
+	OUT resource_manager text,
+	OUT record_type text,
+	OUT record_length int4,
+	OUT main_data_length int4,
+	OUT block_data_length int4,
+	OUT block_fpi_length int4,
+	OUT block_fpi_info text[],
+	OUT description text,
+	OUT block_data bytea,
+	OUT block_fpi_data bytea
 )
 RETURNS SETOF record
 AS 'MODULE_PATHNAME', 'pg_get_wal_block_info'
diff --git a/contrib/pg_walinspect/pg_walinspect.c b/contrib/pg_walinspect/pg_walinspect.c
index 062e90dbce..1ff53c5abc 100644
--- a/contrib/pg_walinspect/pg_walinspect.c
+++ b/contrib/pg_walinspect/pg_walinspect.c
@@ -176,13 +176,18 @@ ReadNextXLogRecord(XLogReaderState *xlogreader)
 }
 
 /*
- * Get a single WAL record info.
+ * Output values that make up a row describing caller's WAL record.
+ *
+ * This function leaks memory.  Caller may need to use its own custom memory
+ * context.
+ *
+ * Keep this in sync with GetWALBlockInfo.
  */
 static void
 GetWALRecordInfo(XLogReaderState *record, Datum *values,
 				 bool *nulls, uint32 ncols)
 {
-	const char *id;
+	const char *record_type;
 	RmgrData	desc;
 	uint32		fpi_len = 0;
 	StringInfoData rec_desc;
@@ -190,10 +195,10 @@ GetWALRecordInfo(XLogReaderState *record, Datum *values,
 	int			i = 0;
 
 	desc = GetRmgr(XLogRecGetRmid(record));
-	id = desc.rm_identify(XLogRecGetInfo(record));
+	record_type = desc.rm_identify(XLogRecGetInfo(record));
 
-	if (id == NULL)
-		id = psprintf("UNKNOWN (%x)", XLogRecGetInfo(record) & ~XLR_INFO_MASK);
+	if (record_type == NULL)
+		record_type = psprintf("UNKNOWN (%x)", XLogRecGetInfo(record) & ~XLR_INFO_MASK);
 
 	initStringInfo(&rec_desc);
 	desc.rm_desc(&rec_desc, record);
@@ -209,7 +214,7 @@ GetWALRecordInfo(XLogReaderState *record, Datum *values,
 	values[i++] = LSNGetDatum(XLogRecGetPrev(record));
 	values[i++] = TransactionIdGetDatum(XLogRecGetXid(record));
 	values[i++] = CStringGetTextDatum(desc.rm_name);
-	values[i++] = CStringGetTextDatum(id);
+	values[i++] = CStringGetTextDatum(record_type);
 	values[i++] = UInt32GetDatum(XLogRecGetTotalLen(record));
 	values[i++] = UInt32GetDatum(XLogRecGetDataLen(record));
 	values[i++] = UInt32GetDatum(fpi_len);
@@ -229,24 +234,48 @@ GetWALRecordInfo(XLogReaderState *record, Datum *values,
 
 
 /*
- * Store a set of block information from a single record (FPI and block
- * information).
+ * Output one or more rows in rsinfo tuple store, each describing a single
+ * block reference from caller's WAL record. (Should only be called with
+ * records that have block references.)
+ *
+ * This function leaks memory.  Caller may need to use its own custom memory
+ * context.
+ *
+ * Keep this in sync with GetWALRecordInfo.
  */
 static void
 GetWALBlockInfo(FunctionCallInfo fcinfo, XLogReaderState *record)
 {
-#define PG_GET_WAL_BLOCK_INFO_COLS 11
+#define PG_GET_WAL_BLOCK_INFO_COLS 20
 	int			block_id;
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+	RmgrData	desc;
+	const char	*record_type;
+	StringInfoData	rec_desc;
+
+	Assert(XLogRecHasAnyBlockRefs(record));
+
+	desc = GetRmgr(XLogRecGetRmid(record));
+	record_type = desc.rm_identify(XLogRecGetInfo(record));
+
+	if (record_type == NULL)
+		record_type = psprintf("UNKNOWN (%x)",
+							   XLogRecGetInfo(record) & ~XLR_INFO_MASK);
+
+	initStringInfo(&rec_desc);
+	desc.rm_desc(&rec_desc, record);
 
 	for (block_id = 0; block_id <= XLogRecMaxBlockId(record); block_id++)
 	{
 		DecodedBkpBlock *blk;
 		BlockNumber blkno;
 		RelFileLocator rnode;
-		ForkNumber	fork;
+		ForkNumber	forknum;
 		Datum		values[PG_GET_WAL_BLOCK_INFO_COLS] = {0};
 		bool		nulls[PG_GET_WAL_BLOCK_INFO_COLS] = {0};
+		uint32		block_data_len = 0,
+					block_fpi_len = 0;
+		ArrayType  *block_fpi_info = NULL;
 		int			i = 0;
 
 		if (!XLogRecHasBlockRef(record, block_id))
@@ -255,99 +284,117 @@ GetWALBlockInfo(FunctionCallInfo fcinfo, XLogReaderState *record)
 		blk = XLogRecGetBlock(record, block_id);
 
 		(void) XLogRecGetBlockTagExtended(record, block_id,
-										  &rnode, &fork, &blkno, NULL);
+										  &rnode, &forknum, &blkno, NULL);
 
+		/* Save block_data_len */
+		if (blk->has_data)
+			block_data_len = blk->data_len;
+
+		if (blk->has_image)
+		{
+			/* Block reference has an FPI, so prepare relevant output */
+			int			bitcnt;
+			int			cnt = 0;
+			Datum	   *flags;
+
+			/* Save block_fpi_len */
+			block_fpi_len = blk->bimg_len;
+
+			/* Construct and save block_fpi_info */
+			bitcnt = pg_popcount((const char *) &blk->bimg_info,
+								 sizeof(uint8));
+			flags = (Datum *) palloc0(sizeof(Datum) * bitcnt);
+			if ((blk->bimg_info & BKPIMAGE_HAS_HOLE) != 0)
+				flags[cnt++] = CStringGetTextDatum("HAS_HOLE");
+			if (blk->apply_image)
+				flags[cnt++] = CStringGetTextDatum("APPLY");
+			if ((blk->bimg_info & BKPIMAGE_COMPRESS_PGLZ) != 0)
+				flags[cnt++] = CStringGetTextDatum("COMPRESS_PGLZ");
+			if ((blk->bimg_info & BKPIMAGE_COMPRESS_LZ4) != 0)
+				flags[cnt++] = CStringGetTextDatum("COMPRESS_LZ4");
+			if ((blk->bimg_info & BKPIMAGE_COMPRESS_ZSTD) != 0)
+				flags[cnt++] = CStringGetTextDatum("COMPRESS_ZSTD");
+
+			Assert(cnt <= bitcnt);
+			block_fpi_info = construct_array_builtin(flags, cnt, TEXTOID);
+		}
+
+		/* start_lsn, end_lsn, prev_lsn, and blockid outputs */
 		values[i++] = LSNGetDatum(record->ReadRecPtr);
+		values[i++] = LSNGetDatum(record->EndRecPtr);
+		values[i++] = LSNGetDatum(XLogRecGetPrev(record));
 		values[i++] = Int16GetDatum(block_id);
+
+		/* relfile and block related outputs */
 		values[i++] = ObjectIdGetDatum(blk->rlocator.spcOid);
 		values[i++] = ObjectIdGetDatum(blk->rlocator.dbOid);
 		values[i++] = ObjectIdGetDatum(blk->rlocator.relNumber);
+		values[i++] = Int16GetDatum(forknum);
 		values[i++] = Int64GetDatum((int64) blkno);
 
-		if (fork >= 0 && fork <= MAX_FORKNUM)
-			values[i++] = CStringGetTextDatum(forkNames[fork]);
+		/* xid, resource_manager, and record_type outputs */
+		values[i++] = TransactionIdGetDatum(XLogRecGetXid(record));
+		values[i++] = CStringGetTextDatum(desc.rm_name);
+		values[i++] = CStringGetTextDatum(record_type);
+
+		/*
+		 * record_length, main_data_length, block_data_len, and
+		 * block_fpi_length outputs
+		 */
+		values[i++] = UInt32GetDatum(XLogRecGetTotalLen(record));
+		values[i++] = UInt32GetDatum(XLogRecGetDataLen(record));
+		values[i++] = UInt32GetDatum(block_data_len);
+		values[i++] = UInt32GetDatum(block_fpi_len);
+
+		/* block_fpi_info (text array) output */
+		if (block_fpi_info)
+			values[i++] = PointerGetDatum(block_fpi_info);
 		else
-			ereport(ERROR,
-					(errcode(ERRCODE_INTERNAL_ERROR),
-					 errmsg_internal("invalid fork number: %u", fork)));
+			nulls[i++] = true;
 
-		/* Block data */
+		/* description output (describes WAL record) */
+		if (rec_desc.len > 0)
+			values[i++] = CStringGetTextDatum(rec_desc.data);
+		else
+			nulls[i++] = true;
+
+		/* block_data output */
 		if (blk->has_data)
 		{
-			bytea	   *raw_data;
+			bytea	   *block_data;
 
-			/* Initialize bytea buffer to copy the data to */
-			raw_data = (bytea *) palloc(blk->data_len + VARHDRSZ);
-			SET_VARSIZE(raw_data, blk->data_len + VARHDRSZ);
-
-			/* Copy the data */
-			memcpy(VARDATA(raw_data), blk->data, blk->data_len);
-			values[i++] = PointerGetDatum(raw_data);
+			block_data = (bytea *) palloc(block_data_len + VARHDRSZ);
+			SET_VARSIZE(block_data, block_data_len + VARHDRSZ);
+			memcpy(VARDATA(block_data), blk->data, block_data_len);
+			values[i++] = PointerGetDatum(block_data);
 		}
 		else
-		{
-			/* No data, so set this field to NULL */
 			nulls[i++] = true;
-		}
 
+		/* block_fpi_data output */
 		if (blk->has_image)
 		{
 			PGAlignedBlock buf;
 			Page		page;
-			bytea	   *raw_page;
-			int			bitcnt;
-			int			cnt = 0;
-			Datum	   *flags;
-			ArrayType  *a;
+			bytea	   *block_fpi_data;
 
 			page = (Page) buf.data;
-
-			/* Full page image exists, so let's save it */
 			if (!RestoreBlockImage(record, block_id, page))
 				ereport(ERROR,
 						(errcode(ERRCODE_INTERNAL_ERROR),
 						 errmsg_internal("%s", record->errormsg_buf)));
 
-			/* Initialize bytea buffer to copy the FPI to */
-			raw_page = (bytea *) palloc(BLCKSZ + VARHDRSZ);
-			SET_VARSIZE(raw_page, BLCKSZ + VARHDRSZ);
-
-			/* Take a verbatim copy of the FPI */
-			memcpy(VARDATA(raw_page), page, BLCKSZ);
-
-			values[i++] = PointerGetDatum(raw_page);
-			values[i++] = UInt32GetDatum(blk->bimg_len);
-
-			/* FPI flags */
-			bitcnt = pg_popcount((const char *) &blk->bimg_info,
-								 sizeof(uint8));
-			/* Build set of raw flags */
-			flags = (Datum *) palloc0(sizeof(Datum) * bitcnt);
-
-			if ((blk->bimg_info & BKPIMAGE_HAS_HOLE) != 0)
-				flags[cnt++] = CStringGetTextDatum("HAS_HOLE");
-			if (blk->apply_image)
-				flags[cnt++] = CStringGetTextDatum("APPLY");
-			if ((blk->bimg_info & BKPIMAGE_COMPRESS_PGLZ) != 0)
-				flags[cnt++] = CStringGetTextDatum("COMPRESS_PGLZ");
-			if ((blk->bimg_info & BKPIMAGE_COMPRESS_LZ4) != 0)
-				flags[cnt++] = CStringGetTextDatum("COMPRESS_LZ4");
-			if ((blk->bimg_info & BKPIMAGE_COMPRESS_ZSTD) != 0)
-				flags[cnt++] = CStringGetTextDatum("COMPRESS_ZSTD");
-
-			Assert(cnt <= bitcnt);
-			a = construct_array_builtin(flags, cnt, TEXTOID);
-			values[i++] = PointerGetDatum(a);
+			block_fpi_data = (bytea *) palloc(BLCKSZ + VARHDRSZ);
+			SET_VARSIZE(block_fpi_data, BLCKSZ + VARHDRSZ);
+			memcpy(VARDATA(block_fpi_data), page, BLCKSZ);
+			values[i++] = PointerGetDatum(block_fpi_data);
 		}
 		else
-		{
-			/* No full page image, so store NULLs for all its fields */
-			memset(&nulls[i], true, 3 * sizeof(bool));
-			i += 3;
-		}
+			nulls[i++] = true;
 
 		Assert(i == PG_GET_WAL_BLOCK_INFO_COLS);
 
+		/* Store a tuple for this block reference */
 		tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
 							 values, nulls);
 	}
@@ -356,11 +403,7 @@ GetWALBlockInfo(FunctionCallInfo fcinfo, XLogReaderState *record)
 }
 
 /*
- * Get information about all the blocks saved in WAL records between start
- * and end LSNs. This produces information about the full page images with
- * their relation information, and the data saved in each block associated
- * to a record. Decompression is applied to the full page images, if
- * necessary.
+ * Get WAL record info, unnested by block reference
  */
 Datum
 pg_get_wal_block_info(PG_FUNCTION_ARGS)
@@ -484,7 +527,7 @@ ValidateInputLSNs(XLogRecPtr start_lsn, XLogRecPtr *end_lsn)
 }
 
 /*
- * Get info and data of all WAL records between start LSN and end LSN.
+ * Get info of all WAL records between start LSN and end LSN.
  */
 static void
 GetWALRecordsInfo(FunctionCallInfo fcinfo, XLogRecPtr start_lsn,
@@ -536,7 +579,7 @@ GetWALRecordsInfo(FunctionCallInfo fcinfo, XLogRecPtr start_lsn,
 }
 
 /*
- * Get info and data of all WAL records between start LSN and end LSN.
+ * Get info of all WAL records between start LSN and end LSN.
  */
 Datum
 pg_get_wal_records_info(PG_FUNCTION_ARGS)
diff --git a/contrib/pg_walinspect/sql/pg_walinspect.sql b/contrib/pg_walinspect/sql/pg_walinspect.sql
index 2f0a909c18..0541e5fbf1 100644
--- a/contrib/pg_walinspect/sql/pg_walinspect.sql
+++ b/contrib/pg_walinspect/sql/pg_walinspect.sql
@@ -78,7 +78,7 @@ UPDATE sample_tbl SET col1 = col1 + 1 WHERE col1 = 1;
 SELECT pg_current_wal_lsn() AS wal_lsn4 \gset
 -- Check if we get block data from WAL record.
 SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_block_info(:'wal_lsn3', :'wal_lsn4')
-  WHERE relfilenode = :'sample_tbl_oid' AND blockdata IS NOT NULL;
+  WHERE relfilenode = :'sample_tbl_oid' AND block_data IS NOT NULL;
 
 -- Force full-page image on the next update.
 SELECT pg_current_wal_lsn() AS wal_lsn5 \gset
@@ -87,7 +87,7 @@ UPDATE sample_tbl SET col1 = col1 + 1 WHERE col1 = 2;
 SELECT pg_current_wal_lsn() AS wal_lsn6 \gset
 -- Check if we get FPI from WAL record.
 SELECT COUNT(*) >= 1 AS ok FROM pg_get_wal_block_info(:'wal_lsn5', :'wal_lsn6')
-  WHERE relfilenode = :'sample_tbl_oid' AND fpi IS NOT NULL;
+  WHERE relfilenode = :'sample_tbl_oid' AND block_fpi_data IS NOT NULL;
 
 -- ===================================================================
 -- Tests for permissions