wal_compression=zstd

From 9253013c789ffb121272bfeeaa9dcdebbef79ced Mon Sep 17 00:00:00 2001
From: Justin Pryzby <pryzbyj@telsasoft.com>
Date: Fri, 18 Feb 2022 22:54:18 -0600
Subject: [PATCH 1/6] add wal_compression=zstd

---
 doc/src/sgml/config.sgml                      |  9 +++---
 doc/src/sgml/installation.sgml                | 10 +++++++
 src/backend/access/transam/xloginsert.c       | 30 ++++++++++++++++++-
 src/backend/access/transam/xlogreader.c       | 20 +++++++++++++
 src/backend/utils/misc/guc.c                  |  3 ++
 src/backend/utils/misc/postgresql.conf.sample |  2 +-
 src/bin/pg_waldump/pg_waldump.c               |  2 ++
 src/include/access/xlog.h                     |  3 +-
 src/include/access/xlogrecord.h               |  5 +++-
 9 files changed, 76 insertions(+), 8 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 7ed8c82a9d..97740c7b66 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -3154,10 +3154,11 @@ include_dir 'conf.d'
         server compresses full page images written to WAL when
         <xref linkend="guc-full-page-writes"/> is on or during a base backup.
         A compressed page image will be decompressed during WAL replay.
-        The supported methods are <literal>pglz</literal> and
-        <literal>lz4</literal> (if <productname>PostgreSQL</productname> was
-        compiled with <option>--with-lz4</option>). The default value is
-        <literal>off</literal>. Only superusers can change this setting.
+        The supported methods are <literal>pglz</literal>, and
+        (if configured when <productname>PostgreSQL</productname> was built) 
+        <literal>lz4</literal> and zstd.
+        The default value is <literal>off</literal>.
+        Only superusers can change this setting.
        </para>
 
        <para>
diff --git a/doc/src/sgml/installation.sgml b/doc/src/sgml/installation.sgml
index 0f74252590..d32767b229 100644
--- a/doc/src/sgml/installation.sgml
+++ b/doc/src/sgml/installation.sgml
@@ -271,6 +271,14 @@ su - postgres
      </para>
     </listitem>
 
+    <listitem>
+     <para>
+      The <productname>ZSTD</productname> library can be used to enable
+      compression using that method; see
+      <xref linkend="guc-wal-compression"/>.
+     </para>
+    </listitem>
+
     <listitem>
      <para>
       To build the <productname>PostgreSQL</productname> documentation,
@@ -988,6 +996,8 @@ build-postgresql:
        <listitem>
         <para>
          Build with <productname>ZSTD</productname> compression support.
+         This enables use of <productname>ZSTD</productname> for
+         compression of WAL data.
         </para>
        </listitem>
       </varlistentry>
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index c260310c4c..847ce0c3fc 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -44,9 +44,17 @@
 #define LZ4_MAX_BLCKSZ		0
 #endif
 
+#ifdef USE_ZSTD
+#include <zstd.h>
+#define ZSTD_MAX_BLCKSZ		ZSTD_COMPRESSBOUND(BLCKSZ)
+#else
+#define ZSTD_MAX_BLCKSZ		0
+#endif
+
+/* Buffer size required to store a compressed version of backup block image */
 #define PGLZ_MAX_BLCKSZ		PGLZ_MAX_OUTPUT(BLCKSZ)
 
-#define COMPRESS_BUFSIZE	Max(PGLZ_MAX_BLCKSZ, LZ4_MAX_BLCKSZ)
+#define COMPRESS_BUFSIZE	Max(Max(PGLZ_MAX_BLCKSZ, LZ4_MAX_BLCKSZ), ZSTD_MAX_BLCKSZ)
 
 /*
  * For each block reference registered with XLogRegisterBuffer, we fill in
@@ -695,6 +703,14 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 #endif
 						break;
 
+					case WAL_COMPRESSION_ZSTD:
+#ifdef USE_ZSTD
+						bimg.bimg_info |= BKPIMAGE_COMPRESS_ZSTD;
+#else
+						elog(ERROR, "ZSTD is not supported by this build");
+#endif
+						break;
+
 					case WAL_COMPRESSION_NONE:
 						Assert(false);	/* cannot happen */
 						break;
@@ -903,6 +919,18 @@ XLogCompressBackupBlock(char *page, uint16 hole_offset, uint16 hole_length,
 #endif
 			break;
 
+		case WAL_COMPRESSION_ZSTD:
+#ifdef USE_ZSTD
+			/* Uses level=1, not ZSTD_CLEVEL_DEFAULT */
+			len = ZSTD_compress(dest, COMPRESS_BUFSIZE, source, orig_len,
+								1);
+			if (ZSTD_isError(len))
+				len = -1;
+#else
+			elog(ERROR, "ZSTD is not supported by this build");
+#endif
+			break;
+
 		case WAL_COMPRESSION_NONE:
 			Assert(false);		/* cannot happen */
 			break;
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 35029cf97d..d60e4cbaf6 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -21,6 +21,9 @@
 #ifdef USE_LZ4
 #include <lz4.h>
 #endif
+#ifdef USE_ZSTD
+#include <zstd.h>
+#endif
 
 #include "access/transam.h"
 #include "access/xlog_internal.h"
@@ -1618,6 +1621,23 @@ RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page)
 								  "LZ4",
 								  block_id);
 			return false;
+#endif
+		}
+		else if ((bkpb->bimg_info & BKPIMAGE_COMPRESS_ZSTD) != 0)
+		{
+#ifdef USE_ZSTD
+			size_t decomp_result = ZSTD_decompress(tmp.data,
+												   BLCKSZ-bkpb->hole_length,
+												   ptr, bkpb->bimg_len);
+
+			if (ZSTD_isError(decomp_result))
+				decomp_success = false;
+#else
+			report_invalid_record(record, "image at %X/%X compressed with %s not supported by build, block %d",
+								  LSN_FORMAT_ARGS(record->ReadRecPtr),
+								  "zstd",
+								  block_id);
+			return false;
 #endif
 		}
 		else
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index e4afd07bfe..66e6d664b0 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -550,6 +550,9 @@ static const struct config_enum_entry wal_compression_options[] = {
 	{"pglz", WAL_COMPRESSION_PGLZ, false},
 #ifdef USE_LZ4
 	{"lz4", WAL_COMPRESSION_LZ4, false},
+#endif
+#ifdef USE_ZSTD
+	{"zstd", WAL_COMPRESSION_ZSTD, false},
 #endif
 	{"on", WAL_COMPRESSION_PGLZ, false},
 	{"off", WAL_COMPRESSION_NONE, false},
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 4a094bb38b..4cf5b26a36 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -220,7 +220,7 @@
 #wal_log_hints = off			# also do full page writes of non-critical updates
 					# (change requires restart)
 #wal_compression = off			# enables compression of full-page writes;
-					# off, pglz, lz4, or on
+					# off, pglz, lz4, zstd, or on
 #wal_init_zero = on			# zero-fill new WAL files
 #wal_recycle = on			# recycle WAL files
 #wal_buffers = -1			# min 32kB, -1 sets based on shared_buffers
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index a6251e1a96..c448658ea2 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -565,6 +565,8 @@ XLogDumpDisplayRecord(XLogDumpConfig *config, XLogReaderState *record)
 						method = "pglz";
 					else if ((bimg_info & BKPIMAGE_COMPRESS_LZ4) != 0)
 						method = "lz4";
+					else if ((bimg_info & BKPIMAGE_COMPRESS_ZSTD) != 0)
+						method = "zstd";
 					else
 						method = "unknown";
 
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 4b45ac64db..09f6464331 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -75,7 +75,8 @@ typedef enum WalCompression
 {
 	WAL_COMPRESSION_NONE = 0,
 	WAL_COMPRESSION_PGLZ,
-	WAL_COMPRESSION_LZ4
+	WAL_COMPRESSION_LZ4,
+	WAL_COMPRESSION_ZSTD
 } WalCompression;
 
 /* Recovery states */
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index c1b1137aa7..052ac6817a 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -149,8 +149,11 @@ typedef struct XLogRecordBlockImageHeader
 /* compression methods supported */
 #define BKPIMAGE_COMPRESS_PGLZ	0x04
 #define BKPIMAGE_COMPRESS_LZ4	0x08
+#define BKPIMAGE_COMPRESS_ZSTD	0x10
+
 #define	BKPIMAGE_COMPRESSED(info) \
-	((info & (BKPIMAGE_COMPRESS_PGLZ | BKPIMAGE_COMPRESS_LZ4)) != 0)
+	((info & (BKPIMAGE_COMPRESS_PGLZ | BKPIMAGE_COMPRESS_LZ4 | \
+			  BKPIMAGE_COMPRESS_ZSTD)) != 0)
 
 /*
  * Extra header information used when page image has "hole" and
-- 
2.17.1

From 75a3fa4e68235f956f8761e43adf25be2fc9147c Mon Sep 17 00:00:00 2001
From: Justin Pryzby <pryzbyj@telsasoft.com>
Date: Sat, 23 Oct 2021 10:41:19 -0500
Subject: [PATCH 2/6] Save bits in xlog FPI record..

2 bits can support 4 methods (including 'none'), and 3 bits can support 8

See also:
https://www.postgresql.org/message-id/20220131222800.GY23027@telsasoft.com
---
 src/backend/access/transam/xloginsert.c | 29 +++----------------------
 src/backend/access/transam/xlogreader.c |  7 +++---
 src/bin/pg_waldump/pg_waldump.c         | 16 +++++---------
 src/include/access/xlogrecord.h         | 17 ++++++++++-----
 4 files changed, 23 insertions(+), 46 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 847ce0c3fc..5f9d07156a 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -689,33 +689,10 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				bimg.length = compressed_len;
 
 				/* Set the compression method used for this block */
-				switch ((WalCompression) wal_compression)
-				{
-					case WAL_COMPRESSION_PGLZ:
-						bimg.bimg_info |= BKPIMAGE_COMPRESS_PGLZ;
-						break;
-
-					case WAL_COMPRESSION_LZ4:
-#ifdef USE_LZ4
-						bimg.bimg_info |= BKPIMAGE_COMPRESS_LZ4;
-#else
-						elog(ERROR, "LZ4 is not supported by this build");
-#endif
-						break;
+				Assert(wal_compression < (1 << BKPIMAGE_COMPRESS_BITS));
 
-					case WAL_COMPRESSION_ZSTD:
-#ifdef USE_ZSTD
-						bimg.bimg_info |= BKPIMAGE_COMPRESS_ZSTD;
-#else
-						elog(ERROR, "ZSTD is not supported by this build");
-#endif
-						break;
-
-					case WAL_COMPRESSION_NONE:
-						Assert(false);	/* cannot happen */
-						break;
-						/* no default case, so that compiler will warn */
-				}
+				bimg.bimg_info |=
+					wal_compression << BKPIMAGE_COMPRESS_OFFSET_BITS;
 
 				rdt_datas_last->data = regbuf->compressed_page;
 				rdt_datas_last->len = compressed_len;
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index d60e4cbaf6..d2d882d0d1 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -1602,14 +1602,15 @@ RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page)
 	{
 		/* If a backup block image is compressed, decompress it */
 		bool		decomp_success = true;
+		int			compressid = BKPIMAGE_COMPRESSION(bkpb->bimg_info);
 
-		if ((bkpb->bimg_info & BKPIMAGE_COMPRESS_PGLZ) != 0)
+		if (compressid == WAL_COMPRESSION_PGLZ)
 		{
 			if (pglz_decompress(ptr, bkpb->bimg_len, tmp.data,
 								BLCKSZ - bkpb->hole_length, true) < 0)
 				decomp_success = false;
 		}
-		else if ((bkpb->bimg_info & BKPIMAGE_COMPRESS_LZ4) != 0)
+		else if (compressid == WAL_COMPRESSION_LZ4)
 		{
 #ifdef USE_LZ4
 			if (LZ4_decompress_safe(ptr, tmp.data,
@@ -1623,7 +1624,7 @@ RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page)
 			return false;
 #endif
 		}
-		else if ((bkpb->bimg_info & BKPIMAGE_COMPRESS_ZSTD) != 0)
+		else if (compressid == WAL_COMPRESSION_ZSTD)
 		{
 #ifdef USE_ZSTD
 			size_t decomp_result = ZSTD_decompress(tmp.data,
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index c448658ea2..bec3fda2aa 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -18,6 +18,7 @@
 #include <unistd.h>
 
 #include "access/transam.h"
+#include "access/xlog.h"
 #include "access/xlog_internal.h"
 #include "access/xlogreader.h"
 #include "access/xlogrecord.h"
@@ -559,17 +560,10 @@ XLogDumpDisplayRecord(XLogDumpConfig *config, XLogReaderState *record)
 
 				if (BKPIMAGE_COMPRESSED(bimg_info))
 				{
-					const char *method;
-
-					if ((bimg_info & BKPIMAGE_COMPRESS_PGLZ) != 0)
-						method = "pglz";
-					else if ((bimg_info & BKPIMAGE_COMPRESS_LZ4) != 0)
-						method = "lz4";
-					else if ((bimg_info & BKPIMAGE_COMPRESS_ZSTD) != 0)
-						method = "zstd";
-					else
-						method = "unknown";
+					int compressid = BKPIMAGE_COMPRESSION(bimg_info);
+					const char *methods[] = {"none", "pglz", "lz4", "zstd"};
 
+					Assert(compressid < lengthof(methods));
 					printf(" (FPW%s); hole: offset: %u, length: %u, "
 						   "compression saved: %u, method: %s",
 						   XLogRecBlockImageApply(record, block_id) ?
@@ -579,7 +573,7 @@ XLogDumpDisplayRecord(XLogDumpConfig *config, XLogReaderState *record)
 						   BLCKSZ -
 						   record->blocks[block_id].hole_length -
 						   record->blocks[block_id].bimg_len,
-						   method);
+						   methods[compressid]);
 				}
 				else
 				{
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 052ac6817a..2080fb707f 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -147,13 +147,18 @@ typedef struct XLogRecordBlockImageHeader
 #define BKPIMAGE_APPLY			0x02	/* page image should be restored
 										 * during replay */
 /* compression methods supported */
-#define BKPIMAGE_COMPRESS_PGLZ	0x04
-#define BKPIMAGE_COMPRESS_LZ4	0x08
-#define BKPIMAGE_COMPRESS_ZSTD	0x10
-
+#define BKPIMAGE_COMPRESS_METHOD1	0x04	/* bits to encode compression method */
+#define BKPIMAGE_COMPRESS_METHOD2	0x08	/* 0=none, 1=pglz, 2=lz4, 3=zstd */
+
+/* How many bits to shift to extract compression */
+#define	BKPIMAGE_COMPRESS_OFFSET_BITS	2
+/* How many bits are for compression */
+#define	BKPIMAGE_COMPRESS_BITS			2
+/* Extract the compression from the bimg_info */
+#define	BKPIMAGE_COMPRESSION(info)		((info >> BKPIMAGE_COMPRESS_OFFSET_BITS) & ((1<<BKPIMAGE_COMPRESS_BITS) - 1))
+// #define	BKPIMAGE_COMPRESSED(info)	(BKPIMAGE_COMPRESSION(info) != 0)
 #define	BKPIMAGE_COMPRESSED(info) \
-	((info & (BKPIMAGE_COMPRESS_PGLZ | BKPIMAGE_COMPRESS_LZ4 | \
-			  BKPIMAGE_COMPRESS_ZSTD)) != 0)
+	((info & (BKPIMAGE_COMPRESS_METHOD1 | BKPIMAGE_COMPRESS_METHOD2)) != 0)
 
 /*
  * Extra header information used when page image has "hole" and
-- 
2.17.1

From 118dbf303c39412d38c4105bc323927bffa97ca1 Mon Sep 17 00:00:00 2001
From: Justin Pryzby <pryzbyj@telsasoft.com>
Date: Fri, 12 Mar 2021 15:35:53 -0600
Subject: [PATCH 3/6] Default to zstd..

for CI, not for merge
---
 .cirrus.yml                          | 14 ++++++++++++++
 configure                            |  6 ++++--
 configure.ac                         |  2 +-
 src/backend/access/transam/xlog.c    |  2 +-
 src/backend/utils/misc/guc.c         |  2 +-
 src/tools/ci/windows_build_config.pl |  1 +
 6 files changed, 22 insertions(+), 5 deletions(-)

diff --git a/.cirrus.yml b/.cirrus.yml
index d10b0a82f9..4735c67608 100644
--- a/.cirrus.yml
+++ b/.cirrus.yml
@@ -77,6 +77,7 @@ task:
     mkdir -m 770 /tmp/cores
     chown root:postgres /tmp/cores
     sysctl kern.corefile='/tmp/cores/%N.%P.core'
+    pkg install -y zstd
 
   # NB: Intentionally build without --with-llvm. The freebsd image size is
   # already large enough to make VM startup slow, and even without llvm
@@ -184,6 +185,8 @@ task:
     mkdir -m 770 /tmp/cores
     chown root:postgres /tmp/cores
     sysctl kernel.core_pattern='/tmp/cores/%e-%s-%p.core'
+    apt-get update
+    apt-get -y install libzstd-dev
 
   configure_script: |
     su postgres <<-EOF
@@ -265,6 +268,7 @@ task:
       openldap \
       openssl \
       python \
+      zstd \
       tcl-tk
 
     brew cleanup -s # to reduce cache size
@@ -385,6 +389,12 @@ task:
     set
 
   configure_script:
+    # XXX: not working
+    - choco install -y cmake
+    - curl -L -o zstd.zip https://github.com/facebook/zstd/archive/refs/tags/v1.5.2.zip
+    - unzip zstd.zip
+    - cd zstd-1.5.2 && vcvarsall x64 && build\VS_scripts\build.VS2017.cmd && cd ..
+    #- cp zstd-1.5.2/build/VS_scripts/build.VS2017.cmd
     # copy errors out when using forward slashes
     - copy src\tools\ci\windows_build_config.pl src\tools\msvc\config.pl
     - vcvarsall x64
@@ -467,6 +477,10 @@ task:
     image: $CONTAINER_REPO/linux_debian_bullseye_ci:latest
     cpu: $CPUS
 
+  setup_os_script: |
+    apt-get update
+    apt-get -y install libzstd-dev
+
   sysinfo_script: |
     id
     uname -a
diff --git a/configure b/configure
index f3cb5c2b51..b88917fcb5 100755
--- a/configure
+++ b/configure
@@ -1584,7 +1584,7 @@ Optional Packages:
                           use system time zone data in DIR
   --without-zlib          do not use Zlib
   --with-lz4              build with LZ4 support
-  --with-zstd             build with ZSTD support
+  --without-zstd          build without Zstd support
   --with-gnu-ld           assume the C compiler uses GNU ld [default=no]
   --with-ssl=LIB          use LIB for SSL/TLS support (openssl)
   --with-openssl          obsolete spelling of --with-ssl=openssl
@@ -9070,7 +9070,9 @@ $as_echo "#define USE_ZSTD 1" >>confdefs.h
   esac
 
 else
-  with_zstd=no
+  with_zstd=yes
+
+$as_echo "#define USE_ZSTD 1" >>confdefs.h
 
 fi
 
diff --git a/configure.ac b/configure.ac
index 19d1a80367..4ae681d8bf 100644
--- a/configure.ac
+++ b/configure.ac
@@ -1060,7 +1060,7 @@ fi
 # ZSTD
 #
 AC_MSG_CHECKING([whether to build with ZSTD support])
-PGAC_ARG_BOOL(with, zstd, no, [build with ZSTD support],
+PGAC_ARG_BOOL(with, zstd, yes, [build with ZSTD support],
               [AC_DEFINE([USE_ZSTD], 1, [Define to 1 to build with ZSTD support. (--with-zstd)])])
 AC_MSG_RESULT([$with_zstd])
 AC_SUBST(with_zstd)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 0d2bd7a357..9aa4d3eaea 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -120,7 +120,7 @@ char	   *XLogArchiveCommand = NULL;
 bool		EnableHotStandby = false;
 bool		fullPageWrites = true;
 bool		wal_log_hints = false;
-int			wal_compression = WAL_COMPRESSION_NONE;
+int			wal_compression = WAL_COMPRESSION_ZSTD;
 char	   *wal_consistency_checking_string = NULL;
 bool	   *wal_consistency_checking = NULL;
 bool		wal_init_zero = true;
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 66e6d664b0..96854b6a2b 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -4900,7 +4900,7 @@ static struct config_enum ConfigureNamesEnum[] =
 			NULL
 		},
 		&wal_compression,
-		WAL_COMPRESSION_NONE, wal_compression_options,
+		WAL_COMPRESSION_ZSTD, wal_compression_options,
 		NULL, NULL, NULL
 	},
 
diff --git a/src/tools/ci/windows_build_config.pl b/src/tools/ci/windows_build_config.pl
index b0d4360c74..97ce79ea7f 100644
--- a/src/tools/ci/windows_build_config.pl
+++ b/src/tools/ci/windows_build_config.pl
@@ -9,5 +9,6 @@ $config->{"asserts"} = 1;
 $config->{"openssl"} = "c:/openssl/1.1/";
 $config->{"perl"} = "c:/strawberry/$ENV{DEFAULT_PERL_VERSION}/perl/";
 $config->{"python"} = "c:/python/";
+$config->{"zstd"} = "c:/zstd/";
 
 1;
-- 
2.17.1

From dc145ccbe47146bdf9dc86910d3135e7e64873ea Mon Sep 17 00:00:00 2001
From: Justin Pryzby <pryzbyj@telsasoft.com>
Date: Sun, 14 Mar 2021 17:12:07 -0500
Subject: [PATCH 4/6] Use GUC hooks to support compression 'level'..

..which is useful for zstd and zlib, but less so for lz4.
---
 src/backend/access/transam/xlog.c       |  20 ++++
 src/backend/access/transam/xloginsert.c |   7 +-
 src/backend/access/transam/xlogreader.c | 129 ++++++++++++++++++++++++
 src/backend/utils/misc/guc.c            |  39 ++-----
 src/include/access/xlog.h               |  11 ++
 src/include/access/xlogreader.h         |   2 +
 6 files changed, 175 insertions(+), 33 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 9aa4d3eaea..198e6906ee 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -121,6 +121,8 @@ bool		EnableHotStandby = false;
 bool		fullPageWrites = true;
 bool		wal_log_hints = false;
 int			wal_compression = WAL_COMPRESSION_ZSTD;
+char		*wal_compression_string = ""; /* Overwritten by GUC */
+int			wal_compression_level = 1;
 char	   *wal_consistency_checking_string = NULL;
 bool	   *wal_consistency_checking = NULL;
 bool		wal_init_zero = true;
@@ -7923,6 +7925,24 @@ assign_xlog_sync_method(int new_sync_method, void *extra)
 	}
 }
 
+bool
+check_wal_compression(char **newval, void **extra, GucSource source)
+{
+	int tmp;
+	if (get_compression_level(*newval, &tmp) != -1)
+		return true;
+
+	return false;
+}
+
+/* Parse the GUC into integers for wal_compression and wal_compression_level */
+void
+assign_wal_compression(const char *newval, void *extra)
+{
+	wal_compression = get_compression_level(newval, &wal_compression_level);
+	Assert(wal_compression >= 0);
+}
+
 
 /*
  * Issue appropriate kind of fsync (if any) for an XLOG output file.
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 5f9d07156a..0e0cfc199c 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -887,8 +887,8 @@ XLogCompressBackupBlock(char *page, uint16 hole_offset, uint16 hole_length,
 
 		case WAL_COMPRESSION_LZ4:
 #ifdef USE_LZ4
-			len = LZ4_compress_default(source, dest, orig_len,
-									   COMPRESS_BUFSIZE);
+			len = LZ4_compress_fast(source, dest, orig_len,
+									   COMPRESS_BUFSIZE, wal_compression_level);
 			if (len <= 0)
 				len = -1;		/* failure */
 #else
@@ -898,9 +898,8 @@ XLogCompressBackupBlock(char *page, uint16 hole_offset, uint16 hole_length,
 
 		case WAL_COMPRESSION_ZSTD:
 #ifdef USE_ZSTD
-			/* Uses level=1, not ZSTD_CLEVEL_DEFAULT */
 			len = ZSTD_compress(dest, COMPRESS_BUFSIZE, source, orig_len,
-								1);
+								wal_compression_level);
 			if (ZSTD_isError(len))
 				len = -1;
 #else
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index d2d882d0d1..c548898eb0 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -18,6 +18,8 @@
 #include "postgres.h"
 
 #include <unistd.h>
+#include <limits.h>
+
 #ifdef USE_LZ4
 #include <lz4.h>
 #endif
@@ -53,6 +55,36 @@ static void ResetDecoder(XLogReaderState *state);
 static void WALOpenSegmentInit(WALOpenSegment *seg, WALSegmentContext *segcxt,
 							   int segsize, const char *waldir);
 
+static const struct {
+	char *name;
+	enum WalCompression compress_id; /* The internal ID */
+	bool has_level; /* If it accepts a numeric "level" */
+	int min_level, dfl_level, max_level;
+} wal_compression_options[] = {
+	{"pglz", WAL_COMPRESSION_PGLZ, false},
+
+#ifdef USE_LZ4
+	{"lz4", WAL_COMPRESSION_LZ4, false}, // XXX
+#endif
+
+#ifdef USE_ZSTD
+	/* XXX: the minimum level depends on the version */
+	/* Must be first */
+	{"zstd-fast", WAL_COMPRESSION_ZSTD, true, -7, -1, 0},
+
+	{"zstd", WAL_COMPRESSION_ZSTD, true, -7, 1, 22},
+#endif
+
+	{"on", WAL_COMPRESSION_PGLZ, false},
+	{"off", WAL_COMPRESSION_NONE, false},
+	{"true", WAL_COMPRESSION_PGLZ, false},
+	{"false", WAL_COMPRESSION_NONE, false},
+	{"yes", WAL_COMPRESSION_PGLZ, false},
+	{"no", WAL_COMPRESSION_NONE, false},
+	{"1", WAL_COMPRESSION_PGLZ, false},
+	{"0", WAL_COMPRESSION_NONE, false},
+};
+
 /* size of the buffer allocated for error message. */
 #define MAX_ERRORMSG_LEN 1000
 
@@ -1578,6 +1610,103 @@ XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
 	}
 }
 
+/*
+ * Return the wal compression ID, or -1 if the input is
+ * invalid/unrecognized/unsupported.
+ * The compression level is stored in *level.
+ */
+int
+get_compression_level(const char *in, int *level)
+{
+	for (int idx=0; idx < lengthof(wal_compression_options); ++idx)
+	{
+		int len;
+		long tmp;
+		char *end;
+
+		if (strcmp(in, wal_compression_options[idx].name) == 0)
+		{
+			/* it has no -level suffix */
+			*level = wal_compression_options[idx].dfl_level;
+			return wal_compression_options[idx].compress_id;
+		}
+
+		len = strlen(wal_compression_options[idx].name);
+		if (strncmp(in, wal_compression_options[idx].name, len) != 0)
+			continue;
+		if (in[len] != '-')
+			continue;
+
+		/* it has a -level suffix, but level is not allowed */
+		if (!wal_compression_options[idx].has_level)
+		{
+#ifndef FRONTEND
+			GUC_check_errdetail("Compression method does not accept a compression level");
+#endif
+			return -1;
+		}
+
+		in += len + 1;
+		len = strlen(in);
+		errno = 0;
+		/* pg_strtoint16 throws an error, which we don't want */
+		/* option_parse_int is frontend only */
+		tmp = strtol(in, &end, 0);
+		if (end != in+len || end == in ||
+				(errno != 0 && tmp == 0) ||
+				(errno == ERANGE && (tmp == LONG_MIN || tmp == LONG_MAX)))
+		{
+#ifndef FRONTEND
+			GUC_check_errdetail("Could not parse compression level: %s", in);
+#endif
+			return -1;
+		}
+
+		/*
+		 * For convenience, allow specification of zstd-fast-N, which is
+		 * interpretted as a negative compression level.
+		 */
+		if (strncmp(wal_compression_options[idx].name, "zstd-fast", 9) == 0 &&
+			tmp > 0)
+			tmp = -tmp;
+
+		if (tmp < wal_compression_options[idx].min_level ||
+				tmp > wal_compression_options[idx].max_level)
+		{
+#ifndef FRONTEND
+			GUC_check_errdetail("Compression level is outside of allowed range: %d...%d",
+					wal_compression_options[idx].min_level,
+					wal_compression_options[idx].max_level);
+#endif
+			return -1;
+		}
+
+		*level = tmp;
+		return wal_compression_options[idx].compress_id;
+	}
+
+#ifndef FRONTEND
+	// XXX: this is trying to distinguish between invalid an unsupported algorithms?
+	if (strcmp(in, "zlib") == 0 && false)
+		GUC_check_errdetail("Compression method is not supported by this build.");
+	else {
+		StringInfoData all_methods;
+
+		initStringInfo(&all_methods);
+		for (int idx=0; idx < lengthof(wal_compression_options); ++idx)
+		{
+			if (idx > 0)
+				appendStringInfoString(&all_methods, ", ");
+			appendStringInfoString(&all_methods, wal_compression_options[idx].name);
+		}
+
+		GUC_check_errdetail("Supported compression methods are: %s", all_methods.data);
+		pfree(all_methods.data);
+	}
+#endif
+	return -1;
+}
+
 /*
  * Restore a full-page image from a backup block attached to an XLOG record.
  *
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 96854b6a2b..9d3fb67fad 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -546,25 +546,6 @@ static struct config_enum_entry default_toast_compression_options[] = {
 	{NULL, 0, false}
 };
 
-static const struct config_enum_entry wal_compression_options[] = {
-	{"pglz", WAL_COMPRESSION_PGLZ, false},
-#ifdef USE_LZ4
-	{"lz4", WAL_COMPRESSION_LZ4, false},
-#endif
-#ifdef USE_ZSTD
-	{"zstd", WAL_COMPRESSION_ZSTD, false},
-#endif
-	{"on", WAL_COMPRESSION_PGLZ, false},
-	{"off", WAL_COMPRESSION_NONE, false},
-	{"true", WAL_COMPRESSION_PGLZ, true},
-	{"false", WAL_COMPRESSION_NONE, true},
-	{"yes", WAL_COMPRESSION_PGLZ, true},
-	{"no", WAL_COMPRESSION_NONE, true},
-	{"1", WAL_COMPRESSION_PGLZ, true},
-	{"0", WAL_COMPRESSION_NONE, true},
-	{NULL, 0, false}
-};
-
 /*
  * Options for enum values stored in other modules
  */
@@ -4643,6 +4624,16 @@ static struct config_string ConfigureNamesString[] =
 		check_wal_consistency_checking, assign_wal_consistency_checking, NULL
 	},
 
+	{
+		{"wal_compression", PGC_SUSET, WAL_SETTINGS,
+			gettext_noop("Compresses full-page writes written in WAL file with specified method."),
+			NULL
+		},
+		&wal_compression_string,
+		"zstd-6",
+		check_wal_compression, assign_wal_compression, NULL
+	},
+
 	{
 		{"jit_provider", PGC_POSTMASTER, CLIENT_CONN_PRELOAD,
 			gettext_noop("JIT provider to use."),
@@ -4894,16 +4885,6 @@ static struct config_enum ConfigureNamesEnum[] =
 		NULL, NULL, NULL
 	},
 
-	{
-		{"wal_compression", PGC_SUSET, WAL_SETTINGS,
-			gettext_noop("Compresses full-page writes written in WAL file with specified method."),
-			NULL
-		},
-		&wal_compression,
-		WAL_COMPRESSION_ZSTD, wal_compression_options,
-		NULL, NULL, NULL
-	},
-
 	{
 		{"wal_level", PGC_POSTMASTER, WAL_SETTINGS,
 			gettext_noop("Sets the level of information written to the WAL."),
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 09f6464331..521aac2ce4 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -16,6 +16,8 @@
 #include "datatype/timestamp.h"
 #include "lib/stringinfo.h"
 #include "nodes/pg_list.h"
+#include "storage/fd.h"
+#include "utils/guc.h"
 
 
 /* Sync methods */
@@ -51,6 +53,10 @@ extern char *wal_consistency_checking_string;
 extern bool log_checkpoints;
 extern bool track_wal_io_timing;
 
+extern char *wal_compression_string;
+extern int wal_compression;
+extern int wal_compression_level;
+
 extern int	CheckPointSegments;
 
 /* Archive modes */
@@ -248,6 +254,11 @@ extern void SetWalWriterSleeping(bool sleeping);
 extern void assign_max_wal_size(int newval, void *extra);
 extern void assign_checkpoint_completion_target(double newval, void *extra);
 
+/* GUC */
+extern bool check_wal_compression(char **newval, void **extra, GucSource source);
+extern void assign_wal_compression(const char *newval, void *extra);
+
+
 /*
  * Routines used by xlogrecovery.c to call back into xlog.c during recovery.
  */
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 477f0efe26..dd6c117ef7 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -337,4 +337,6 @@ extern bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id,
 							   RelFileNode *rnode, ForkNumber *forknum,
 							   BlockNumber *blknum);
 
+extern int get_compression_level(const char *in, int *level);
+
 #endif							/* XLOGREADER_H */
-- 
2.17.1

From a238e5b6b4868d256b58fb631beace2e9c1f2fc6 Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Date: Mon, 8 Mar 2021 15:32:30 +0900
Subject: [PATCH 5/6] Run 011_crash_recovery.pl with wal_level=minimal

The test doesn't need that feature and pg_current_xact_id() is better
exercised by turning off the feature.

Copied from: https://www.postgresql.org/message-id/20210308.173242.463790587797836129.horikyota.ntt%40gmail.com
---
 src/test/recovery/t/011_crash_recovery.pl | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/test/recovery/t/011_crash_recovery.pl b/src/test/recovery/t/011_crash_recovery.pl
index 14154d1ce0..4b11e40544 100644
--- a/src/test/recovery/t/011_crash_recovery.pl
+++ b/src/test/recovery/t/011_crash_recovery.pl
@@ -12,7 +12,7 @@ use Test::More;
 use Config;
 
 my $node = PostgreSQL::Test::Cluster->new('primary');
-$node->init(allows_streaming => 1);
+$node->init(allows_streaming => 0);
 $node->start;
 
 my ($stdin, $stdout, $stderr) = ('', '', '');
-- 
2.17.1

From 81044d628a20f5b2ee0d4d36b013d59d0bc33643 Mon Sep 17 00:00:00 2001
From: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Date: Mon, 8 Mar 2021 15:43:01 +0900
Subject: [PATCH 6/6] Make sure published XIDs are persistent

pg_xact_status() premises that XIDs obtained by
pg_current_xact_id(_if_assigned)() are persistent beyond a crash. But
XIDs are not guaranteed to go beyond WAL buffers before commit and
thus XIDs may vanish if server crashes before commit. This patch
guarantees the XID shown by the functions to be flushed out to disk.

Copied from: https://www.postgresql.org/message-id/20210308.173242.463790587797836129.horikyota.ntt%40gmail.com
---
 src/backend/access/transam/xact.c | 56 +++++++++++++++++++++++++------
 src/backend/access/transam/xlog.c |  2 +-
 src/backend/utils/adt/xid8funcs.c | 12 ++++++-
 src/include/access/xact.h         |  3 +-
 4 files changed, 60 insertions(+), 13 deletions(-)

diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index bb1f106946..f0ba9c33ed 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -204,7 +204,7 @@ typedef struct TransactionStateData
 	int			prevSecContext; /* previous SecurityRestrictionContext */
 	bool		prevXactReadOnly;	/* entry-time xact r/o state */
 	bool		startedInRecovery;	/* did we start in recovery? */
-	bool		didLogXid;		/* has xid been included in WAL record? */
+	XLogRecPtr 	minLSN;			/* LSN needed to reach to record the xid */
 	int			parallelModeLevel;	/* Enter/ExitParallelMode counter */
 	bool		chain;			/* start a new block after this one */
 	bool		topXidLogged;	/* for a subxact: is top-level XID logged? */
@@ -523,12 +523,16 @@ GetCurrentFullTransactionIdIfAny(void)
  *	MarkCurrentTransactionIdLoggedIfAny
  *
  * Remember that the current xid - if it is assigned - now has been wal logged.
+ *
+ * upto is the LSN up to which we need to flush WAL to ensure the current xid
+ * to be persistent. See EnsureCurrentTransactionIdLogged().
  */
 void
-MarkCurrentTransactionIdLoggedIfAny(void)
+MarkCurrentTransactionIdLoggedIfAny(XLogRecPtr upto)
 {
-	if (FullTransactionIdIsValid(CurrentTransactionState->fullTransactionId))
-		CurrentTransactionState->didLogXid = true;
+	if (FullTransactionIdIsValid(CurrentTransactionState->fullTransactionId) &&
+		XLogRecPtrIsInvalid(CurrentTransactionState->minLSN))
+		CurrentTransactionState->minLSN = upto;
 }
 
 /*
@@ -582,6 +586,35 @@ MarkSubxactTopXidLogged(void)
 	CurrentTransactionState->topXidLogged = true;
 }
 
+/*
+ *	EnsureCurrentTransactionIdLogged
+ *
+ * Make sure that the current top XID is WAL-logged.
+ */
+void
+EnsureTopTransactionIdLogged(void)
+{
+	/*
+	 * We need at least one WAL record for the current top transaction to be
+	 * flushed out.  Write one if we don't have one yet.
+	 */
+	if (XLogRecPtrIsInvalid(TopTransactionStateData.minLSN))
+	{
+		xl_xact_assignment xlrec;
+
+		xlrec.xtop = XidFromFullTransactionId(XactTopFullTransactionId);
+		Assert(TransactionIdIsValid(xlrec.xtop));
+		xlrec.nsubxacts = 0;
+
+		XLogBeginInsert();
+		XLogRegisterData((char *) &xlrec, MinSizeOfXactAssignment);
+		TopTransactionStateData.minLSN =
+			XLogInsert(RM_XACT_ID, XLOG_XACT_ASSIGNMENT);
+	}
+
+	XLogFlush(TopTransactionStateData.minLSN);
+}
+
 /*
  *	GetStableLatestTransactionId
  *
@@ -669,14 +702,14 @@ AssignTransactionId(TransactionState s)
 	 * When wal_level=logical, guarantee that a subtransaction's xid can only
 	 * be seen in the WAL stream if its toplevel xid has been logged before.
 	 * If necessary we log an xact_assignment record with fewer than
-	 * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set
+	 * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if minLSN isn't set
 	 * for a transaction even though it appears in a WAL record, we just might
 	 * superfluously log something. That can happen when an xid is included
 	 * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in
 	 * xl_standby_locks.
 	 */
 	if (isSubXact && XLogLogicalInfoActive() &&
-		!TopTransactionStateData.didLogXid)
+		XLogRecPtrIsInvalid(TopTransactionStateData.minLSN))
 		log_unknown_top = true;
 
 	/*
@@ -746,6 +779,7 @@ AssignTransactionId(TransactionState s)
 			log_unknown_top)
 		{
 			xl_xact_assignment xlrec;
+			XLogRecPtr		   endptr;
 
 			/*
 			 * xtop is always set by now because we recurse up transaction
@@ -760,11 +794,13 @@ AssignTransactionId(TransactionState s)
 			XLogRegisterData((char *) unreportedXids,
 							 nUnreportedXids * sizeof(TransactionId));
 
-			(void) XLogInsert(RM_XACT_ID, XLOG_XACT_ASSIGNMENT);
+			endptr = XLogInsert(RM_XACT_ID, XLOG_XACT_ASSIGNMENT);
 
 			nUnreportedXids = 0;
-			/* mark top, not current xact as having been logged */
-			TopTransactionStateData.didLogXid = true;
+
+			/* set minLSN of top, not of current xact if not yet */
+			if (XLogRecPtrIsInvalid(TopTransactionStateData.minLSN))
+				TopTransactionStateData.minLSN = endptr;
 		}
 	}
 }
@@ -2049,7 +2085,7 @@ StartTransaction(void)
 	 * initialize reported xid accounting
 	 */
 	nUnreportedXids = 0;
-	s->didLogXid = false;
+	s->minLSN = InvalidXLogRecPtr;
 
 	/*
 	 * must initialize resource-management stuff first
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 198e6906ee..7c7bea768c 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -915,7 +915,7 @@ XLogInsertRecord(XLogRecData *rdata,
 
 	END_CRIT_SECTION();
 
-	MarkCurrentTransactionIdLoggedIfAny();
+	MarkCurrentTransactionIdLoggedIfAny(EndPos);
 
 	/*
 	 * Mark top transaction id is logged (if needed) so that we should not try
diff --git a/src/backend/utils/adt/xid8funcs.c b/src/backend/utils/adt/xid8funcs.c
index 60e57876de..47bae59898 100644
--- a/src/backend/utils/adt/xid8funcs.c
+++ b/src/backend/utils/adt/xid8funcs.c
@@ -357,6 +357,8 @@ bad_format:
 Datum
 pg_current_xact_id(PG_FUNCTION_ARGS)
 {
+	FullTransactionId xid;
+
 	/*
 	 * Must prevent during recovery because if an xid is not assigned we try
 	 * to assign one, which would fail. Programs already rely on this function
@@ -365,7 +367,12 @@ pg_current_xact_id(PG_FUNCTION_ARGS)
 	 */
 	PreventCommandDuringRecovery("pg_current_xact_id()");
 
-	PG_RETURN_FULLTRANSACTIONID(GetTopFullTransactionId());
+	xid = GetTopFullTransactionId();
+
+	/* the XID is going to be published, make sure it is psersistent */
+	EnsureTopTransactionIdLogged();
+
+	PG_RETURN_FULLTRANSACTIONID(xid);
 }
 
 /*
@@ -380,6 +387,9 @@ pg_current_xact_id_if_assigned(PG_FUNCTION_ARGS)
 	if (!FullTransactionIdIsValid(topfxid))
 		PG_RETURN_NULL();
 
+	/* the XID is going to be published, make sure it is psersistent */
+	EnsureTopTransactionIdLogged();
+
 	PG_RETURN_FULLTRANSACTIONID(topfxid);
 }
 
diff --git a/src/include/access/xact.h b/src/include/access/xact.h
index 17a6fa4abd..33d03e1913 100644
--- a/src/include/access/xact.h
+++ b/src/include/access/xact.h
@@ -386,7 +386,8 @@ extern FullTransactionId GetTopFullTransactionId(void);
 extern FullTransactionId GetTopFullTransactionIdIfAny(void);
 extern FullTransactionId GetCurrentFullTransactionId(void);
 extern FullTransactionId GetCurrentFullTransactionIdIfAny(void);
-extern void MarkCurrentTransactionIdLoggedIfAny(void);
+extern void MarkCurrentTransactionIdLoggedIfAny(XLogRecPtr upto);
+extern void EnsureTopTransactionIdLogged(void);
 extern bool SubTransactionIsActive(SubTransactionId subxid);
 extern CommandId GetCurrentCommandId(bool used);
 extern void SetParallelStartTimestamps(TimestampTz xact_ts, TimestampTz stmt_ts);
-- 
2.17.1