Add contrib/pg_logicalinspect.

This module provides SQL functions that allow to inspect logical
decoding components.

It currently allows to inspect the contents of serialized logical
snapshots of a running database cluster, which is useful for debugging
or educational purposes.

Author: Bertrand Drouvot
Reviewed-by: Amit Kapila, Shveta Malik, Peter Smith, Peter Eisentraut
Reviewed-by: David G. Johnston
Discussion: https://postgr.es/m/ZscuZ92uGh3wm4tW%40ip-10-97-1-34.eu-west-3.compute.internal

contrib/Makefile

@@ -32,6 +32,7 @@ SUBDIRS = \
 		passwordcheck	\
 		pg_buffercache	\
 		pg_freespacemap \
+		pg_logicalinspect \
 		pg_prewarm	\
 		pg_stat_statements \
 		pg_surgery	\

contrib/meson.build

@@ -46,6 +46,7 @@ subdir('passwordcheck')
 subdir('pg_buffercache')
 subdir('pgcrypto')
 subdir('pg_freespacemap')
+subdir('pg_logicalinspect')
 subdir('pg_prewarm')
 subdir('pgrowlocks')
 subdir('pg_stat_statements')

contrib/pg_logicalinspect/.gitignore

@@ -0,0 +1,6 @@
+# Generated subdirectories
+/log/
+/results/
+/output_iso/
+/tmp_check/
+/tmp_check_iso/

contrib/pg_logicalinspect/Makefile

@@ -0,0 +1,31 @@
+# contrib/pg_logicalinspect/Makefile
+
+MODULE_big = pg_logicalinspect
+OBJS = \
+	$(WIN32RES) \
+	pg_logicalinspect.o
+PGFILEDESC = "pg_logicalinspect - functions to inspect logical decoding components"
+
+EXTENSION = pg_logicalinspect
+DATA = pg_logicalinspect--1.0.sql
+
+EXTRA_INSTALL = contrib/test_decoding
+
+ISOLATION = logical_inspect
+
+ISOLATION_OPTS = --temp-config $(top_srcdir)/contrib/pg_logicalinspect/logicalinspect.conf
+
+# Disabled because these tests require "wal_level=logical", which
+# some installcheck users do not have (e.g. buildfarm clients).
+NO_INSTALLCHECK = 1
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/pg_logicalinspect
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif

contrib/pg_logicalinspect/expected/logical_inspect.out

@@ -0,0 +1,52 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s0_init s0_begin s0_savepoint s0_truncate s1_checkpoint s1_get_changes s0_commit s0_begin s0_insert s1_checkpoint s1_get_changes s0_commit s1_get_changes s1_get_logical_snapshot_info s1_get_logical_snapshot_meta
+step s0_init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding');
+?column?
+--------
+init    
+(1 row)
+
+step s0_begin: BEGIN;
+step s0_savepoint: SAVEPOINT sp1;
+step s0_truncate: TRUNCATE tbl1;
+step s1_checkpoint: CHECKPOINT;
+step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0');
+data
+----
+(0 rows)
+
+step s0_commit: COMMIT;
+step s0_begin: BEGIN;
+step s0_insert: INSERT INTO tbl1 VALUES (1);
+step s1_checkpoint: CHECKPOINT;
+step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0');
+data                                   
+---------------------------------------
+BEGIN                                  
+table public.tbl1: TRUNCATE: (no-flags)
+COMMIT                                 
+(3 rows)
+
+step s0_commit: COMMIT;
+step s1_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0');
+data                                                         
+-------------------------------------------------------------
+BEGIN                                                        
+table public.tbl1: INSERT: val1[integer]:1 val2[integer]:null
+COMMIT                                                       
+(3 rows)
+
+step s1_get_logical_snapshot_info: SELECT info.state, info.catchange_count, array_length(info.catchange_xip,1) AS catchange_array_length, info.committed_count, array_length(info.committed_xip,1) AS committed_array_length FROM pg_ls_logicalsnapdir(), pg_get_logical_snapshot_info(name) AS info ORDER BY 2;
+state     |catchange_count|catchange_array_length|committed_count|committed_array_length
+----------+---------------+----------------------+---------------+----------------------
+consistent|              0|                      |              2|                     2
+consistent|              2|                     2|              0|                      
+(2 rows)
+
+step s1_get_logical_snapshot_meta: SELECT COUNT(meta.*) from pg_ls_logicalsnapdir(), pg_get_logical_snapshot_meta(name) as meta;
+count
+-----
+    2
+(1 row)
+

contrib/pg_logicalinspect/logicalinspect.conf

@@ -0,0 +1 @@
+wal_level = logical

contrib/pg_logicalinspect/meson.build

@@ -0,0 +1,39 @@
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+pg_logicalinspect_sources = files('pg_logicalinspect.c')
+
+if host_system == 'windows'
+  pg_logicalinspect_sources += rc_lib_gen.process(win32ver_rc, extra_args: [
+    '--NAME', 'pg_logicalinspect',
+    '--FILEDESC', 'pg_logicalinspect - functions to inspect logical decoding components',])
+endif
+
+pg_logicalinspect = shared_module('pg_logicalinspect',
+  pg_logicalinspect_sources,
+  kwargs: contrib_mod_args + {
+      'dependencies': contrib_mod_args['dependencies'],
+  },
+)
+contrib_targets += pg_logicalinspect
+
+install_data(
+  'pg_logicalinspect.control',
+  'pg_logicalinspect--1.0.sql',
+  kwargs: contrib_data_args,
+)
+
+tests += {
+  'name': 'pg_logicalinspect',
+  'sd': meson.current_source_dir(),
+  'bd': meson.current_build_dir(),
+  'isolation': {
+    'specs': [
+      'logical_inspect',
+    ],
+    'regress_args': [
+      '--temp-config', files('logicalinspect.conf'),
+    ],
+    # see above
+    'runningcheck': false,
+  },
+}

contrib/pg_logicalinspect/pg_logicalinspect--1.0.sql

@@ -0,0 +1,43 @@
+/* contrib/pg_logicalinspect/pg_logicalinspect--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION pg_logicalinspect" to load this file. \quit
+
+--
+-- pg_get_logical_snapshot_meta()
+--
+CREATE FUNCTION pg_get_logical_snapshot_meta(IN filename text,
+    OUT magic int4,
+    OUT checksum int8,
+    OUT version int4
+)
+AS 'MODULE_PATHNAME', 'pg_get_logical_snapshot_meta'
+LANGUAGE C STRICT PARALLEL SAFE;
+
+REVOKE EXECUTE ON FUNCTION pg_get_logical_snapshot_meta(text) FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_get_logical_snapshot_meta(text) TO pg_read_server_files;
+
+--
+-- pg_get_logical_snapshot_info()
+--
+CREATE FUNCTION pg_get_logical_snapshot_info(IN filename text,
+    OUT state text,
+    OUT xmin xid,
+    OUT xmax xid,
+    OUT start_decoding_at pg_lsn,
+    OUT two_phase_at pg_lsn,
+    OUT initial_xmin_horizon xid,
+    OUT building_full_snapshot boolean,
+    OUT in_slot_creation boolean,
+    OUT last_serialized_snapshot pg_lsn,
+    OUT next_phase_at xid,
+    OUT committed_count int4,
+    OUT committed_xip xid[],
+    OUT catchange_count int4,
+    OUT catchange_xip xid[]
+)
+AS 'MODULE_PATHNAME', 'pg_get_logical_snapshot_info'
+LANGUAGE C STRICT PARALLEL SAFE;
+
+REVOKE EXECUTE ON FUNCTION pg_get_logical_snapshot_info(text) FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_get_logical_snapshot_info(text) TO pg_read_server_files;

contrib/pg_logicalinspect/pg_logicalinspect.c

@@ -0,0 +1,167 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_logicalinspect.c
+ *		  Functions to inspect contents of PostgreSQL logical snapshots
+ *
+ * Copyright (c) 2024, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		  contrib/pg_logicalinspect/pg_logicalinspect.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "funcapi.h"
+#include "replication/snapbuild_internal.h"
+#include "utils/array.h"
+#include "utils/builtins.h"
+#include "utils/pg_lsn.h"
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(pg_get_logical_snapshot_meta);
+PG_FUNCTION_INFO_V1(pg_get_logical_snapshot_info);
+
+/* Return the description of SnapBuildState */
+static const char *
+get_snapbuild_state_desc(SnapBuildState state)
+{
+	const char *stateDesc = "unknown state";
+
+	switch (state)
+	{
+		case SNAPBUILD_START:
+			stateDesc = "start";
+			break;
+		case SNAPBUILD_BUILDING_SNAPSHOT:
+			stateDesc = "building";
+			break;
+		case SNAPBUILD_FULL_SNAPSHOT:
+			stateDesc = "full";
+			break;
+		case SNAPBUILD_CONSISTENT:
+			stateDesc = "consistent";
+			break;
+	}
+
+	return stateDesc;
+}
+
+/*
+ * Retrieve the logical snapshot file metadata.
+ */
+Datum
+pg_get_logical_snapshot_meta(PG_FUNCTION_ARGS)
+{
+#define PG_GET_LOGICAL_SNAPSHOT_META_COLS 3
+	SnapBuildOnDisk ondisk;
+	HeapTuple	tuple;
+	Datum		values[PG_GET_LOGICAL_SNAPSHOT_META_COLS] = {0};
+	bool		nulls[PG_GET_LOGICAL_SNAPSHOT_META_COLS] = {0};
+	TupleDesc	tupdesc;
+	char		path[MAXPGPATH];
+	int			i = 0;
+	text	   *filename_t = PG_GETARG_TEXT_PP(0);
+
+	/* Build a tuple descriptor for our result type */
+	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+		elog(ERROR, "return type must be a row type");
+
+	sprintf(path, "%s/%s",
+			PG_LOGICAL_SNAPSHOTS_DIR,
+			text_to_cstring(filename_t));
+
+	/* Validate and restore the snapshot to 'ondisk' */
+	SnapBuildRestoreSnapshot(&ondisk, path, CurrentMemoryContext, false);
+
+	values[i++] = UInt32GetDatum(ondisk.magic);
+	values[i++] = Int64GetDatum((int64) ondisk.checksum);
+	values[i++] = UInt32GetDatum(ondisk.version);
+
+	Assert(i == PG_GET_LOGICAL_SNAPSHOT_META_COLS);
+
+	tuple = heap_form_tuple(tupdesc, values, nulls);
+
+	PG_RETURN_DATUM(HeapTupleGetDatum(tuple));
+
+#undef PG_GET_LOGICAL_SNAPSHOT_META_COLS
+}
+
+Datum
+pg_get_logical_snapshot_info(PG_FUNCTION_ARGS)
+{
+#define PG_GET_LOGICAL_SNAPSHOT_INFO_COLS 14
+	SnapBuildOnDisk ondisk;
+	HeapTuple	tuple;
+	Datum		values[PG_GET_LOGICAL_SNAPSHOT_INFO_COLS] = {0};
+	bool		nulls[PG_GET_LOGICAL_SNAPSHOT_INFO_COLS] = {0};
+	TupleDesc	tupdesc;
+	char		path[MAXPGPATH];
+	int			i = 0;
+	text	   *filename_t = PG_GETARG_TEXT_PP(0);
+
+	/* Build a tuple descriptor for our result type */
+	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+		elog(ERROR, "return type must be a row type");
+
+	sprintf(path, "%s/%s",
+			PG_LOGICAL_SNAPSHOTS_DIR,
+			text_to_cstring(filename_t));
+
+	/* Validate and restore the snapshot to 'ondisk' */
+	SnapBuildRestoreSnapshot(&ondisk, path, CurrentMemoryContext, false);
+
+	values[i++] = CStringGetTextDatum(get_snapbuild_state_desc(ondisk.builder.state));
+	values[i++] = TransactionIdGetDatum(ondisk.builder.xmin);
+	values[i++] = TransactionIdGetDatum(ondisk.builder.xmax);
+	values[i++] = LSNGetDatum(ondisk.builder.start_decoding_at);
+	values[i++] = LSNGetDatum(ondisk.builder.two_phase_at);
+	values[i++] = TransactionIdGetDatum(ondisk.builder.initial_xmin_horizon);
+	values[i++] = BoolGetDatum(ondisk.builder.building_full_snapshot);
+	values[i++] = BoolGetDatum(ondisk.builder.in_slot_creation);
+	values[i++] = LSNGetDatum(ondisk.builder.last_serialized_snapshot);
+	values[i++] = TransactionIdGetDatum(ondisk.builder.next_phase_at);
+
+	values[i++] = UInt32GetDatum(ondisk.builder.committed.xcnt);
+	if (ondisk.builder.committed.xcnt > 0)
+	{
+		Datum	   *arrayelems;
+
+		arrayelems = (Datum *) palloc(ondisk.builder.committed.xcnt * sizeof(Datum));
+
+		for (int j = 0; j < ondisk.builder.committed.xcnt; j++)
+			arrayelems[j] = TransactionIdGetDatum(ondisk.builder.committed.xip[j]);
+
+		values[i++] = PointerGetDatum(construct_array_builtin(arrayelems,
+															  ondisk.builder.committed.xcnt,
+															  XIDOID));
+	}
+	else
+		nulls[i++] = true;
+
+	values[i++] = UInt32GetDatum(ondisk.builder.catchange.xcnt);
+	if (ondisk.builder.catchange.xcnt > 0)
+	{
+		Datum	   *arrayelems;
+
+		arrayelems = (Datum *) palloc(ondisk.builder.catchange.xcnt * sizeof(Datum));
+
+		for (int j = 0; j < ondisk.builder.catchange.xcnt; j++)
+			arrayelems[j] = TransactionIdGetDatum(ondisk.builder.catchange.xip[j]);
+
+		values[i++] = PointerGetDatum(construct_array_builtin(arrayelems,
+															  ondisk.builder.catchange.xcnt,
+															  XIDOID));
+	}
+	else
+		nulls[i++] = true;
+
+	Assert(i == PG_GET_LOGICAL_SNAPSHOT_INFO_COLS);
+
+	tuple = heap_form_tuple(tupdesc, values, nulls);
+
+	PG_RETURN_DATUM(HeapTupleGetDatum(tuple));
+
+#undef PG_GET_LOGICAL_SNAPSHOT_INFO_COLS
+}

contrib/pg_logicalinspect/pg_logicalinspect.control

@@ -0,0 +1,5 @@
+# pg_logicalinspect extension
+comment = 'functions to inspect logical decoding components'
+default_version = '1.0'
+module_pathname = '$libdir/pg_logicalinspect'
+relocatable = true

contrib/pg_logicalinspect/specs/logical_inspect.spec

@@ -0,0 +1,34 @@
+# Test the pg_logicalinspect functions: that needs some permutation to
+# ensure that we are creating multiple logical snapshots and that one of them
+# contains ongoing catalogs changes.
+setup
+{
+    DROP TABLE IF EXISTS tbl1;
+    CREATE TABLE tbl1 (val1 integer, val2 integer);
+    CREATE EXTENSION pg_logicalinspect;
+}
+
+teardown
+{
+    DROP TABLE tbl1;
+    SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot');
+    DROP EXTENSION pg_logicalinspect;
+}
+
+session "s0"
+setup { SET synchronous_commit=on; }
+step "s0_init" { SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); }
+step "s0_begin" { BEGIN; }
+step "s0_savepoint" { SAVEPOINT sp1; }
+step "s0_truncate" { TRUNCATE tbl1; }
+step "s0_insert" { INSERT INTO tbl1 VALUES (1); }
+step "s0_commit" { COMMIT; }
+
+session "s1"
+setup { SET synchronous_commit=on; }
+step "s1_checkpoint" { CHECKPOINT; }
+step "s1_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0'); }
+step "s1_get_logical_snapshot_meta" { SELECT COUNT(meta.*) from pg_ls_logicalsnapdir(), pg_get_logical_snapshot_meta(name) as meta;}
+step "s1_get_logical_snapshot_info" { SELECT info.state, info.catchange_count, array_length(info.catchange_xip,1) AS catchange_array_length, info.committed_count, array_length(info.committed_xip,1) AS committed_array_length FROM pg_ls_logicalsnapdir(), pg_get_logical_snapshot_info(name) AS info ORDER BY 2; }
+
+permutation "s0_init" "s0_begin" "s0_savepoint" "s0_truncate" "s1_checkpoint" "s1_get_changes" "s0_commit" "s0_begin" "s0_insert" "s1_checkpoint" "s1_get_changes" "s0_commit" "s1_get_changes" "s1_get_logical_snapshot_info" "s1_get_logical_snapshot_meta"

doc/src/sgml/contrib.sgml

@@ -154,6 +154,7 @@ CREATE EXTENSION <replaceable>extension_name</replaceable>;
  &pgbuffercache;
  &pgcrypto;
  &pgfreespacemap;
+ &pglogicalinspect;
  &pgprewarm;
  &pgrowlocks;
  &pgstatstatements;

doc/src/sgml/filelist.sgml

@@ -143,6 +143,7 @@
 <!ENTITY pgbuffercache   SYSTEM "pgbuffercache.sgml">
 <!ENTITY pgcrypto        SYSTEM "pgcrypto.sgml">
 <!ENTITY pgfreespacemap  SYSTEM "pgfreespacemap.sgml">
+<!ENTITY pglogicalinspect  SYSTEM "pglogicalinspect.sgml">
 <!ENTITY pgprewarm       SYSTEM "pgprewarm.sgml">
 <!ENTITY pgrowlocks      SYSTEM "pgrowlocks.sgml">
 <!ENTITY pgstatstatements SYSTEM "pgstatstatements.sgml">

doc/src/sgml/pglogicalinspect.sgml

@@ -0,0 +1,143 @@
+<!-- doc/src/sgml/pglogicalinspect.sgml -->
+
+<sect1 id="pglogicalinspect" xreflabel="pg_logicalinspect">
+ <title>pg_logicalinspect &mdash; logical decoding components inspection</title>
+
+ <indexterm zone="pglogicalinspect">
+  <primary>pg_logicalinspect</primary>
+ </indexterm>
+
+ <para>
+  The <filename>pg_logicalinspect</filename> module provides SQL functions
+  that allow you to inspect the contents of logical decoding components. It
+  allows the inspection of serialized logical snapshots of a running
+  <productname>PostgreSQL</productname> database cluster, which is useful
+  for debugging or educational purposes.
+ </para>
+
+ <para>
+  By default, use of these functions is restricted to superusers and members of
+  the <literal>pg_read_server_files</literal> role. Access may be granted by
+  superusers to others using <command>GRANT</command>.
+ </para>
+
+ <sect2 id="pglogicalinspect-funcs">
+  <title>Functions</title>
+
+  <variablelist>
+   <varlistentry id="pglogicalinspect-funcs-pg-get-logical-snapshot-meta">
+    <term>
+     <function>pg_get_logical_snapshot_meta(filename text) returns record</function>
+    </term>
+
+    <listitem>
+     <para>
+      Gets logical snapshot metadata about a snapshot file that is located in
+      the server's <filename>pg_logical/snapshots</filename> directory.
+      The <replaceable>filename</replaceable> argument represents the snapshot
+      file name.
+      For example:
+<screen>
+postgres=# SELECT * FROM pg_ls_logicalsnapdir();
+-[ RECORD 1 ]+-----------------------
+name         | 0-40796E18.snap
+size         | 152
+modification | 2024-08-14 16:36:32+00
+
+postgres=# SELECT * FROM pg_get_logical_snapshot_meta('0-40796E18.snap');
+-[ RECORD 1 ]--------
+magic    | 1369563137
+checksum | 1028045905
+version  | 6
+
+postgres=# SELECT ss.name, meta.* FROM pg_ls_logicalsnapdir() AS ss,
+pg_get_logical_snapshot_meta(ss.name) AS meta;
+-[ RECORD 1 ]-------------
+name     | 0-40796E18.snap
+magic    | 1369563137
+checksum | 1028045905
+version  | 6
+</screen>
+     </para>
+     <para>
+      If <replaceable>filename</replaceable> does not match a snapshot file, the
+      function raises an error.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry id="pglogicalinspect-funcs-pg-get-logical-snapshot-info">
+    <term>
+     <function>pg_get_logical_snapshot_info(filename text) returns record</function>
+    </term>
+
+    <listitem>
+     <para>
+      Gets logical snapshot information about a snapshot file that is located in
+      the server's <filename>pg_logical/snapshots</filename> directory.
+      The <replaceable>filename</replaceable> argument represents the snapshot
+      file name.
+      For example:
+<screen>
+postgres=# SELECT * FROM pg_ls_logicalsnapdir();
+-[ RECORD 1 ]+-----------------------
+name         | 0-40796E18.snap
+size         | 152
+modification | 2024-08-14 16:36:32+00
+
+postgres=# SELECT * FROM pg_get_logical_snapshot_info('0-40796E18.snap');
+-[ RECORD 1 ]------------+-----------
+state                    | consistent
+xmin                     | 751
+xmax                     | 751
+start_decoding_at        | 0/40796AF8
+two_phase_at             | 0/40796AF8
+initial_xmin_horizon     | 0
+building_full_snapshot   | f
+in_slot_creation         | f
+last_serialized_snapshot | 0/0
+next_phase_at            | 0
+committed_count          | 0
+committed_xip            |
+catchange_count          | 2
+catchange_xip            | {751,752}
+
+postgres=# SELECT ss.name, info.* FROM pg_ls_logicalsnapdir() AS ss,
+pg_get_logical_snapshot_info(ss.name) AS info;
+-[ RECORD 1 ]------------+----------------
+name                     | 0-40796E18.snap
+state                    | consistent
+xmin                     | 751
+xmax                     | 751
+start_decoding_at        | 0/40796AF8
+two_phase_at             | 0/40796AF8
+initial_xmin_horizon     | 0
+building_full_snapshot   | f
+in_slot_creation         | f
+last_serialized_snapshot | 0/0
+next_phase_at            | 0
+committed_count          | 0
+committed_xip            |
+catchange_count          | 2
+catchange_xip            | {751,752}
+</screen>
+     </para>
+     <para>
+      If <replaceable>filename</replaceable> does not match a snapshot file, the
+      function raises an error.
+     </para>
+    </listitem>
+   </varlistentry>
+
+  </variablelist>
+ </sect2>
+
+ <sect2 id="pglogicalinspect-author">
+  <title>Author</title>
+
+  <para>
+   Bertrand Drouvot <email>bertranddrouvot.pg@gmail.com</email>
+  </para>
+ </sect2>
+
+</sect1>

src/backend/replication/logical/snapbuild.c

@@ -1684,34 +1684,31 @@ out:
 }
 
 /*
- * Restore a snapshot into 'builder' if previously one has been stored at the
+ * Restore the logical snapshot file contents to 'ondisk'.
- * location indicated by 'lsn'. Returns true if successful, false otherwise.
+ *
+ * 'context' is the memory context where the catalog modifying/committed xid
+ * will live.
+ * If 'missing_ok' is true, will not throw an error if the file is not found.
  */
-static bool
+bool
-SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
+SnapBuildRestoreSnapshot(SnapBuildOnDisk *ondisk, const char *path,
+						 MemoryContext context, bool missing_ok)
 {
-	SnapBuildOnDisk ondisk;
 	int			fd;
-	char		path[MAXPGPATH];
-	Size		sz;
 	pg_crc32c	checksum;
-
+	Size		sz;
-	/* no point in loading a snapshot if we're already there */
-	if (builder->state == SNAPBUILD_CONSISTENT)
-		return false;
-
-	sprintf(path, "%s/%X-%X.snap",
-			PG_LOGICAL_SNAPSHOTS_DIR,
-			LSN_FORMAT_ARGS(lsn));
 
 	fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
 
-	if (fd < 0 && errno == ENOENT)
+	if (fd < 0)
+	{
+		if (missing_ok && errno == ENOENT)
 			return false;
-	else if (fd < 0)
+
 		ereport(ERROR,
 				(errcode_for_file_access(),
 				 errmsg("could not open file \"%s\": %m", path)));
+	}
 
 	/* ----
 	 * Make sure the snapshot had been stored safely to disk, that's normally
@@ -1724,47 +1721,46 @@ SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
 	fsync_fname(path, false);
 	fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
 
-
 	/* read statically sized portion of snapshot */
-	SnapBuildRestoreContents(fd, (char *) &ondisk, SnapBuildOnDiskConstantSize, path);
+	SnapBuildRestoreContents(fd, (char *) ondisk, SnapBuildOnDiskConstantSize, path);
 
-	if (ondisk.magic != SNAPBUILD_MAGIC)
+	if (ondisk->magic != SNAPBUILD_MAGIC)
 		ereport(ERROR,
 				(errcode(ERRCODE_DATA_CORRUPTED),
 				 errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
-						path, ondisk.magic, SNAPBUILD_MAGIC)));
+						path, ondisk->magic, SNAPBUILD_MAGIC)));
 
-	if (ondisk.version != SNAPBUILD_VERSION)
+	if (ondisk->version != SNAPBUILD_VERSION)
 		ereport(ERROR,
 				(errcode(ERRCODE_DATA_CORRUPTED),
 				 errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
-						path, ondisk.version, SNAPBUILD_VERSION)));
+						path, ondisk->version, SNAPBUILD_VERSION)));
 
 	INIT_CRC32C(checksum);
 	COMP_CRC32C(checksum,
-				((char *) &ondisk) + SnapBuildOnDiskNotChecksummedSize,
+				((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
 				SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
 
 	/* read SnapBuild */
-	SnapBuildRestoreContents(fd, (char *) &ondisk.builder, sizeof(SnapBuild), path);
+	SnapBuildRestoreContents(fd, (char *) &ondisk->builder, sizeof(SnapBuild), path);
-	COMP_CRC32C(checksum, &ondisk.builder, sizeof(SnapBuild));
+	COMP_CRC32C(checksum, &ondisk->builder, sizeof(SnapBuild));
 
 	/* restore committed xacts information */
-	if (ondisk.builder.committed.xcnt > 0)
+	if (ondisk->builder.committed.xcnt > 0)
 	{
-		sz = sizeof(TransactionId) * ondisk.builder.committed.xcnt;
+		sz = sizeof(TransactionId) * ondisk->builder.committed.xcnt;
-		ondisk.builder.committed.xip = MemoryContextAllocZero(builder->context, sz);
+		ondisk->builder.committed.xip = MemoryContextAllocZero(context, sz);
-		SnapBuildRestoreContents(fd, (char *) ondisk.builder.committed.xip, sz, path);
+		SnapBuildRestoreContents(fd, (char *) ondisk->builder.committed.xip, sz, path);
-		COMP_CRC32C(checksum, ondisk.builder.committed.xip, sz);
+		COMP_CRC32C(checksum, ondisk->builder.committed.xip, sz);
 	}
 
 	/* restore catalog modifying xacts information */
-	if (ondisk.builder.catchange.xcnt > 0)
+	if (ondisk->builder.catchange.xcnt > 0)
 	{
-		sz = sizeof(TransactionId) * ondisk.builder.catchange.xcnt;
+		sz = sizeof(TransactionId) * ondisk->builder.catchange.xcnt;
-		ondisk.builder.catchange.xip = MemoryContextAllocZero(builder->context, sz);
+		ondisk->builder.catchange.xip = MemoryContextAllocZero(context, sz);
-		SnapBuildRestoreContents(fd, (char *) ondisk.builder.catchange.xip, sz, path);
+		SnapBuildRestoreContents(fd, (char *) ondisk->builder.catchange.xip, sz, path);
-		COMP_CRC32C(checksum, ondisk.builder.catchange.xip, sz);
+		COMP_CRC32C(checksum, ondisk->builder.catchange.xip, sz);
 	}
 
 	if (CloseTransientFile(fd) != 0)
@@ -1775,11 +1771,36 @@ SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
 	FIN_CRC32C(checksum);
 
 	/* verify checksum of what we've read */
-	if (!EQ_CRC32C(checksum, ondisk.checksum))
+	if (!EQ_CRC32C(checksum, ondisk->checksum))
 		ereport(ERROR,
 				(errcode(ERRCODE_DATA_CORRUPTED),
 				 errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
-						path, checksum, ondisk.checksum)));
+						path, checksum, ondisk->checksum)));
+
+	return true;
+}
+
+/*
+ * Restore a snapshot into 'builder' if previously one has been stored at the
+ * location indicated by 'lsn'. Returns true if successful, false otherwise.
+ */
+static bool
+SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
+{
+	SnapBuildOnDisk ondisk;
+	char		path[MAXPGPATH];
+
+	/* no point in loading a snapshot if we're already there */
+	if (builder->state == SNAPBUILD_CONSISTENT)
+		return false;
+
+	sprintf(path, "%s/%X-%X.snap",
+			PG_LOGICAL_SNAPSHOTS_DIR,
+			LSN_FORMAT_ARGS(lsn));
+
+	/* validate and restore the snapshot to 'ondisk' */
+	if (!SnapBuildRestoreSnapshot(&ondisk, path, builder->context, true))
+		return false;
 
 	/*
 	 * ok, we now have a sensible snapshot here, figure out if it has more

src/backend/utils/adt/arrayfuncs.c

@@ -3453,6 +3453,12 @@ construct_array_builtin(Datum *elems, int nelems, Oid elmtype)
 			elmalign = TYPALIGN_SHORT;
 			break;
 
+		case XIDOID:
+			elmlen = sizeof(TransactionId);
+			elmbyval = true;
+			elmalign = TYPALIGN_INT;
+			break;
+
 		default:
 			elog(ERROR, "type %u not supported by construct_array_builtin()", elmtype);
 			/* keep compiler quiet */

src/include/replication/snapbuild.h

@@ -15,6 +15,10 @@
 #include "access/xlogdefs.h"
 #include "utils/snapmgr.h"
 
+/*
+ * Please keep get_snapbuild_state_desc() (located in the pg_logicalinspect
+ * module) updated if a change needs to be made to SnapBuildState.
+ */
 typedef enum
 {
 	/*

src/include/replication/snapbuild_internal.h

@@ -193,4 +193,7 @@ typedef struct SnapBuildOnDisk
 	/* variable amount of TransactionIds follows */
 } SnapBuildOnDisk;
 
+extern bool SnapBuildRestoreSnapshot(SnapBuildOnDisk *ondisk, const char *path,
+									 MemoryContext context, bool missing_ok);
+
 #endif							/* SNAPBUILD_INTERNAL_H */