v13-0002-ZSTD-add-server-side-compression-support.patch
application/octet-stream
Filename: v13-0002-ZSTD-add-server-side-compression-support.patch
Type: application/octet-stream
Part: 2
Message:
Re: refactoring basebackup.c
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: format-patch
Series: patch v13-0002
Subject: ZSTD: add server-side compression support.
| File | + | − |
|---|---|---|
| doc/src/sgml/protocol.sgml | 4 | 3 |
| doc/src/sgml/ref/pg_basebackup.sgml | 20 | 18 |
| src/backend/replication/basebackup.c | 6 | 1 |
| src/backend/replication/basebackup_zstd.c | 294 | 0 |
| src/backend/replication/Makefile | 1 | 0 |
| src/bin/pg_basebackup/pg_basebackup.c | 18 | 1 |
| src/bin/pg_basebackup/pg_receivewal.c | 4 | 0 |
| src/bin/pg_basebackup/walmethods.h | 1 | 0 |
| src/bin/pg_verifybackup/Makefile | 1 | 0 |
| src/bin/pg_verifybackup/t/008_untar.pl | 9 | 0 |
| src/include/replication/basebackup_sink.h | 1 | 0 |
From 0f0770583989fc55afa8003046ea9b85af2142b3 Mon Sep 17 00:00:00 2001
From: Jeevan Ladhe <jeevan.ladhe@enterprisedb.com>
Date: Thu, 17 Feb 2022 06:55:53 +0530
Subject: [PATCH 2/4] ZSTD: add server-side compression support.
This patch introduces --compress=server-zstd[:LEVEL]
Add tap test.
Add config option --with-zstd.
Add documentation for ZSTD option.
Add pg_basebackup help for ZSTD option.
Example:
pg_basebackup -t server:/tmp/data_test -Xnone --compress=server-zstd:4
---
doc/src/sgml/protocol.sgml | 7 +-
doc/src/sgml/ref/pg_basebackup.sgml | 38 +--
src/backend/replication/Makefile | 1 +
src/backend/replication/basebackup.c | 7 +-
src/backend/replication/basebackup_zstd.c | 294 ++++++++++++++++++++++
src/bin/pg_basebackup/pg_basebackup.c | 19 +-
src/bin/pg_basebackup/pg_receivewal.c | 4 +
src/bin/pg_basebackup/walmethods.h | 1 +
src/bin/pg_verifybackup/Makefile | 1 +
src/bin/pg_verifybackup/t/008_untar.pl | 9 +
src/include/replication/basebackup_sink.h | 1 +
11 files changed, 359 insertions(+), 23 deletions(-)
create mode 100644 src/backend/replication/basebackup_zstd.c
mode change 100644 => 100755 src/bin/pg_verifybackup/t/008_untar.pl
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 1c5ab00879..8fe638767d 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2724,8 +2724,8 @@ The commands accepted in replication mode are:
<listitem>
<para>
Instructs the server to compress the backup using the specified
- method. Currently, the supported methods are <literal>gzip</literal>
- and <literal>lz4</literal>.
+ method. Currently, the supported methods are <literal>gzip</literal>,
+ <literal>lz4</literal>, and <literal>zstd</literal>.
</para>
</listitem>
</varlistentry>
@@ -2737,7 +2737,8 @@ The commands accepted in replication mode are:
Specifies the compression level to be used. This should only be
used in conjunction with the <literal>COMPRESSION</literal> option.
For <literal>gzip</literal> the value should be an integer between 1
- and 9, and for <literal>lz4</literal> it should be between 1 and 12.
+ and 9, for <literal>lz4</literal> between 1 and 12, and for
+ <literal>zstd</literal> it should be between 1 and 22.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/pg_basebackup.sgml b/doc/src/sgml/ref/pg_basebackup.sgml
index 53aa40dcd1..4cf28a2a61 100644
--- a/doc/src/sgml/ref/pg_basebackup.sgml
+++ b/doc/src/sgml/ref/pg_basebackup.sgml
@@ -417,30 +417,32 @@ PostgreSQL documentation
specify <literal>-Xfetch</literal>.
</para>
<para>
- The compression method can be set to <literal>gzip</literal> or
- <literal>lz4</literal>, or <literal>none</literal> for no
- compression. A compression level can be optionally specified, by
- appending the level number after a colon (<literal>:</literal>). If no
- level is specified, the default compression level will be used. If
- only a level is specified without mentioning an algorithm,
- <literal>gzip</literal> compression will be used if the level is
- greater than 0, and no compression will be used if the level is 0.
- </para>
- <para>
- When the tar format is used with <literal>gzip</literal> or
- <literal>lz4</literal>, the suffix <filename>.gz</filename> or
- <filename>.lz4</filename> will automatically be added to all tar
- filenames. When the plain format is used, client-side compression may
- not be specified, but it is still possible to request server-side
- compression. If this is done, the server will compress the backup for
- transmission, and the client will decompress and extract it.
+ The compression method can be set to <literal>gzip</literal>,
+ <literal>lz4</literal>, <literal>zstd</literal>, or
+ <literal>none</literal> for no compression. A compression level can
+ optionally be specified, by appending the level number after a colon
+ (<literal>:</literal>). If no level is specified, the default
+ compression level will be used. If only a level is specified without
+ mentioning an algorithm, <literal>gzip</literal> compression will be
+ used if the level is greater than 0, and no compression will be used if
+ the level is 0.
+ </para>
+ <para>
+ When the tar format is used with <literal>gzip</literal>,
+ <literal>lz4</literal>, or <literal>zstd</literal>, the suffix
+ <filename>.gz</filename>, <filename>.lz4</filename>, or
+ <filename>.zst</filename> respectively will be automatically added to
+ all tar filenames. When the plain format is used, client-side
+ compression may not be specified, but it is still possible to request
+ server-side compression. If this is done, the server will compress the
+ backup for transmission, and the client will decompress and extract it.
</para>
<para>
When this option is used in combination with
<literal>-Xstream</literal>, <literal>pg_wal.tar</literal> will
be compressed using <literal>gzip</literal> if client-side gzip
compression is selected, but will not be compressed if server-side
- compresion or LZ4 compresion is selected.
+ compression, LZ4, or ZSTD compression is selected.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/replication/Makefile b/src/backend/replication/Makefile
index 74043ff331..2e6de7007f 100644
--- a/src/backend/replication/Makefile
+++ b/src/backend/replication/Makefile
@@ -20,6 +20,7 @@ OBJS = \
basebackup_copy.o \
basebackup_gzip.o \
basebackup_lz4.o \
+ basebackup_zstd.o \
basebackup_progress.o \
basebackup_server.o \
basebackup_sink.o \
diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c
index 0bf28b55d7..2378ce5c5e 100644
--- a/src/backend/replication/basebackup.c
+++ b/src/backend/replication/basebackup.c
@@ -64,7 +64,8 @@ typedef enum
{
BACKUP_COMPRESSION_NONE,
BACKUP_COMPRESSION_GZIP,
- BACKUP_COMPRESSION_LZ4
+ BACKUP_COMPRESSION_LZ4,
+ BACKUP_COMPRESSION_ZSTD
} basebackup_compression_type;
typedef struct
@@ -906,6 +907,8 @@ parse_basebackup_options(List *options, basebackup_options *opt)
opt->compression = BACKUP_COMPRESSION_GZIP;
else if (strcmp(optval, "lz4") == 0)
opt->compression = BACKUP_COMPRESSION_LZ4;
+ else if (strcmp(optval, "zstd") == 0)
+ opt->compression = BACKUP_COMPRESSION_ZSTD;
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -1026,6 +1029,8 @@ SendBaseBackup(BaseBackupCmd *cmd)
sink = bbsink_gzip_new(sink, opt.compression_level);
else if (opt.compression == BACKUP_COMPRESSION_LZ4)
sink = bbsink_lz4_new(sink, opt.compression_level);
+ else if (opt.compression == BACKUP_COMPRESSION_ZSTD)
+ sink = bbsink_zstd_new(sink, opt.compression_level);
/* Set up progress reporting. */
sink = bbsink_progress_new(sink, opt.progress);
diff --git a/src/backend/replication/basebackup_zstd.c b/src/backend/replication/basebackup_zstd.c
new file mode 100644
index 0000000000..24993a5bb6
--- /dev/null
+++ b/src/backend/replication/basebackup_zstd.c
@@ -0,0 +1,294 @@
+/*-------------------------------------------------------------------------
+ *
+ * basebackup_zstd.c
+ * Basebackup sink implementing zstd compression.
+ *
+ * Portions Copyright (c) 2010-2020, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/replication/basebackup_zstd.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#ifdef HAVE_LIBZSTD
+#include <zstd.h>
+#endif
+
+#include "replication/basebackup_sink.h"
+
+#ifdef HAVE_LIBZSTD
+
+typedef struct bbsink_zstd
+{
+ /* Common information for all types of sink. */
+ bbsink base;
+
+ /* Compression level */
+ int compresslevel;
+
+ ZSTD_CCtx *cctx;
+ ZSTD_outBuffer zstd_outBuf;
+} bbsink_zstd;
+
+static void bbsink_zstd_begin_backup(bbsink *sink);
+static void bbsink_zstd_begin_archive(bbsink *sink, const char *archive_name);
+static void bbsink_zstd_archive_contents(bbsink *sink, size_t avail_in);
+static void bbsink_zstd_manifest_contents(bbsink *sink, size_t len);
+static void bbsink_zstd_end_archive(bbsink *sink);
+static void bbsink_zstd_cleanup(bbsink *sink);
+static void bbsink_zstd_end_backup(bbsink *sink, XLogRecPtr endptr,
+ TimeLineID endtli);
+
+const bbsink_ops bbsink_zstd_ops = {
+ .begin_backup = bbsink_zstd_begin_backup,
+ .begin_archive = bbsink_zstd_begin_archive,
+ .archive_contents = bbsink_zstd_archive_contents,
+ .end_archive = bbsink_zstd_end_archive,
+ .begin_manifest = bbsink_forward_begin_manifest,
+ .manifest_contents = bbsink_zstd_manifest_contents,
+ .end_manifest = bbsink_forward_end_manifest,
+ .end_backup = bbsink_zstd_end_backup,
+ .cleanup = bbsink_zstd_cleanup
+};
+#endif
+
+/*
+ * Create a new basebackup sink that performs zstd compression using the
+ * designated compression level.
+ */
+bbsink *
+bbsink_zstd_new(bbsink *next, int compresslevel)
+{
+#ifndef HAVE_LIBZSTD
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("zstd compression is not supported by this build")));
+ return NULL; /* keep compiler quiet */
+#else
+ bbsink_zstd *sink;
+
+ Assert(next != NULL);
+
+ if (compresslevel < 0 || compresslevel > 22)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("zstd compression level %d is out of range",
+ compresslevel)));
+
+ sink = palloc0(sizeof(bbsink_zstd));
+ *((const bbsink_ops **) &sink->base.bbs_ops) = &bbsink_zstd_ops;
+ sink->base.bbs_next = next;
+ sink->compresslevel = compresslevel;
+
+ return &sink->base;
+#endif
+}
+
+#ifdef HAVE_LIBZSTD
+
+/*
+ * Begin backup.
+ */
+static void
+bbsink_zstd_begin_backup(bbsink *sink)
+{
+ bbsink_zstd *mysink = (bbsink_zstd *) sink;
+ size_t output_buffer_bound;
+
+ mysink->cctx = ZSTD_createCCtx();
+ if (!mysink->cctx)
+ elog(ERROR, "could not create zstd compression context");
+
+ ZSTD_CCtx_setParameter(mysink->cctx, ZSTD_c_compressionLevel,
+ mysink->compresslevel);
+
+ /*
+ * We need our own buffer, because we're going to pass different data to
+ * the next sink than what gets passed to us.
+ */
+ mysink->base.bbs_buffer = palloc(mysink->base.bbs_buffer_length);
+
+ /*
+ * Make sure that the next sink's bbs_buffer is big enough to accommodate
+ * the compressed input buffer.
+ */
+ output_buffer_bound = ZSTD_compressBound(mysink->base.bbs_buffer_length);
+
+ /*
+ * The buffer length is expected to be a multiple of BLCKSZ, so round up.
+ */
+ output_buffer_bound = output_buffer_bound + BLCKSZ -
+ (output_buffer_bound % BLCKSZ);
+
+ bbsink_begin_backup(sink->bbs_next, sink->bbs_state, output_buffer_bound);
+}
+
+/*
+ * Prepare to compress the next archive.
+ */
+static void
+bbsink_zstd_begin_archive(bbsink *sink, const char *archive_name)
+{
+ bbsink_zstd *mysink = (bbsink_zstd *) sink;
+ char *zstd_archive_name;
+
+ /*
+ * At the start of each archive we reset the state to start a new
+ * compression operation. The parameters are sticky and they will stick
+ * around as we are resetting with option ZSTD_reset_session_only.
+ */
+ ZSTD_CCtx_reset(mysink->cctx, ZSTD_reset_session_only);
+
+ mysink->zstd_outBuf.dst = mysink->base.bbs_next->bbs_buffer;
+ mysink->zstd_outBuf.size = mysink->base.bbs_next->bbs_buffer_length;
+ mysink->zstd_outBuf.pos = 0;
+
+ /* Add ".zst" to the archive name. */
+ zstd_archive_name = psprintf("%s.zst", archive_name);
+ Assert(sink->bbs_next != NULL);
+ bbsink_begin_archive(sink->bbs_next, zstd_archive_name);
+ pfree(zstd_archive_name);
+}
+
+/*
+ * Compress the input data to the output buffer until we run out of input
+ * data. Each time the output buffer falls below the compression bound for
+ * the input buffer, invoke the archive_contents() method for the next sink.
+ *
+ * Note that since we're compressing the input, it may very commonly happen
+ * that we consume all the input data without filling the output buffer. In
+ * that case, the compressed representation of the current input data won't
+ * actually be sent to the next bbsink until a later call to this function,
+ * or perhaps even not until bbsink_zstd_end_archive() is invoked.
+ */
+static void
+bbsink_zstd_archive_contents(bbsink *sink, size_t len)
+{
+ bbsink_zstd *mysink = (bbsink_zstd *) sink;
+ ZSTD_inBuffer inBuf = {mysink->base.bbs_buffer, len, 0};
+
+ while (inBuf.pos < inBuf.size)
+ {
+ size_t yet_to_flush;
+ size_t required_outBuf_bound = ZSTD_compressBound(inBuf.size - inBuf.pos);
+
+ /*
+ * If the out buffer is not left with enough space, send the output
+ * buffer to the next sink, and reset it.
+ */
+ if ((mysink->zstd_outBuf.size - mysink->zstd_outBuf.pos) <=
+ required_outBuf_bound)
+ {
+ bbsink_archive_contents(mysink->base.bbs_next, mysink->zstd_outBuf.pos);
+ mysink->zstd_outBuf.dst = mysink->base.bbs_next->bbs_buffer;
+ mysink->zstd_outBuf.size = mysink->base.bbs_next->bbs_buffer_length;
+ mysink->zstd_outBuf.pos = 0;
+ }
+
+ yet_to_flush = ZSTD_compressStream2(mysink->cctx, &mysink->zstd_outBuf,
+ &inBuf, ZSTD_e_continue);
+
+ if (ZSTD_isError(yet_to_flush))
+ elog(ERROR, "could not compress data: %s", ZSTD_getErrorName(yet_to_flush));
+ }
+}
+
+/*
+ * There might be some data inside zstd's internal buffers; we need to get that
+ * flushed out, also end the zstd frame and then get that forwarded to the
+ * successor sink as archive content.
+ *
+ * Then we can end processing for this archive.
+ */
+static void
+bbsink_zstd_end_archive(bbsink *sink)
+{
+ bbsink_zstd *mysink = (bbsink_zstd *) sink;
+ size_t yet_to_flush;
+
+ do
+ {
+ ZSTD_inBuffer in = {NULL, 0, 0};
+ size_t required_outBuf_bound = ZSTD_compressBound(0);
+
+ /*
+ * If the out buffer is not left with enough space, send the output
+ * buffer to the next sink, and reset it.
+ */
+ if ((mysink->zstd_outBuf.size - mysink->zstd_outBuf.pos) <=
+ required_outBuf_bound)
+ {
+ bbsink_archive_contents(mysink->base.bbs_next, mysink->zstd_outBuf.pos);
+ mysink->zstd_outBuf.dst = mysink->base.bbs_next->bbs_buffer;
+ mysink->zstd_outBuf.size = mysink->base.bbs_next->bbs_buffer_length;
+ mysink->zstd_outBuf.pos = 0;
+ }
+
+ yet_to_flush = ZSTD_compressStream2(mysink->cctx,
+ &mysink->zstd_outBuf,
+ &in, ZSTD_e_end);
+
+ if (ZSTD_isError(yet_to_flush))
+ elog(ERROR, "could not compress data: %s",
+ ZSTD_getErrorName(yet_to_flush));
+
+ } while (yet_to_flush > 0);
+
+ /* Make sure to pass any remaining bytes to the next sink. */
+ if (mysink->zstd_outBuf.pos > 0)
+ bbsink_archive_contents(mysink->base.bbs_next, mysink->zstd_outBuf.pos);
+
+ /* Pass on the information that this archive has ended. */
+ bbsink_forward_end_archive(sink);
+}
+
+/*
+ * Free the resources and context.
+ */
+static void
+bbsink_zstd_end_backup(bbsink *sink, XLogRecPtr endptr,
+ TimeLineID endtli)
+{
+ bbsink_zstd *mysink = (bbsink_zstd *) sink;
+
+ /* Release the context. */
+ if (mysink->cctx)
+ {
+ ZSTD_freeCCtx(mysink->cctx);
+ mysink->cctx = NULL;
+ }
+
+ bbsink_forward_end_backup(sink, endptr, endtli);
+}
+
+/*
+ * Manifest contents are not compressed, but we do need to copy them into
+ * the successor sink's buffer, because we have our own.
+ */
+static void
+bbsink_zstd_manifest_contents(bbsink *sink, size_t len)
+{
+ memcpy(sink->bbs_next->bbs_buffer, sink->bbs_buffer, len);
+ bbsink_manifest_contents(sink->bbs_next, len);
+}
+
+/*
+ * In case the backup fails, make sure we free the compression context by
+ * calling ZSTD_freeCCtx if needed to avoid memory leak.
+ */
+static void
+bbsink_zstd_cleanup(bbsink *sink)
+{
+ bbsink_zstd *mysink = (bbsink_zstd *) sink;
+
+ /* Release the context if not already released. */
+ if (mysink->cctx)
+ {
+ ZSTD_freeCCtx(mysink->cctx);
+ mysink->cctx = NULL;
+ }
+}
+
+#endif
diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c
index 0003b59615..304d510220 100644
--- a/src/bin/pg_basebackup/pg_basebackup.c
+++ b/src/bin/pg_basebackup/pg_basebackup.c
@@ -391,8 +391,9 @@ usage(void)
printf(_(" -X, --wal-method=none|fetch|stream\n"
" include required WAL files with specified method\n"));
printf(_(" -z, --gzip compress tar output\n"));
- printf(_(" -Z, --compress={[{client,server}-]gzip,lz4,none}[:LEVEL] or [LEVEL]\n"
+ printf(_(" -Z, --compress=[{client|server}-]{gzip|lz4|zstd}[:LEVEL]\n"
" compress tar output with given compression method or level\n"));
+ printf(_(" -Z, --compress=none do not compress tar output\n"));
printf(_("\nGeneral options:\n"));
printf(_(" -c, --checkpoint=fast|spread\n"
" set fast or spread checkpointing\n"));
@@ -1023,6 +1024,11 @@ parse_compress_options(char *src, WalCompressionMethod *methodres,
*methodres = COMPRESSION_LZ4;
*locationres = COMPRESS_LOCATION_SERVER;
}
+ else if (pg_strcasecmp(firstpart, "server-zstd") == 0)
+ {
+ *methodres = COMPRESSION_ZSTD;
+ *locationres = COMPRESS_LOCATION_SERVER;
+ }
else if (pg_strcasecmp(firstpart, "none") == 0)
{
*methodres = COMPRESSION_NONE;
@@ -1970,6 +1976,9 @@ BaseBackup(void)
case COMPRESSION_LZ4:
compressmethodstr = "lz4";
break;
+ case COMPRESSION_ZSTD:
+ compressmethodstr = "zstd";
+ break;
default:
Assert(false);
break;
@@ -2819,6 +2828,14 @@ main(int argc, char **argv)
exit(1);
}
break;
+ case COMPRESSION_ZSTD:
+ if (compresslevel > 22)
+ {
+ pg_log_error("compression level %d of method %s higher than maximum of 22",
+ compresslevel, "zstd");
+ exit(1);
+ }
+ break;
}
/*
diff --git a/src/bin/pg_basebackup/pg_receivewal.c b/src/bin/pg_basebackup/pg_receivewal.c
index ccb215c398..9b7656c692 100644
--- a/src/bin/pg_basebackup/pg_receivewal.c
+++ b/src/bin/pg_basebackup/pg_receivewal.c
@@ -904,6 +904,10 @@ main(int argc, char **argv)
exit(1);
#endif
break;
+ case COMPRESSION_ZSTD:
+ pg_log_error("compression with %s is not yet supported", "ZSTD");
+ exit(1);
+
}
diff --git a/src/bin/pg_basebackup/walmethods.h b/src/bin/pg_basebackup/walmethods.h
index 2dfb353baa..ec54019cfc 100644
--- a/src/bin/pg_basebackup/walmethods.h
+++ b/src/bin/pg_basebackup/walmethods.h
@@ -24,6 +24,7 @@ typedef enum
{
COMPRESSION_GZIP,
COMPRESSION_LZ4,
+ COMPRESSION_ZSTD,
COMPRESSION_NONE
} WalCompressionMethod;
diff --git a/src/bin/pg_verifybackup/Makefile b/src/bin/pg_verifybackup/Makefile
index 851233a6e0..596df15118 100644
--- a/src/bin/pg_verifybackup/Makefile
+++ b/src/bin/pg_verifybackup/Makefile
@@ -10,6 +10,7 @@ export TAR
# name.
export GZIP_PROGRAM=$(GZIP)
export LZ4=$(LZ4)
+export ZSTD=$(ZSTD)
subdir = src/bin/pg_verifybackup
top_builddir = ../../..
diff --git a/src/bin/pg_verifybackup/t/008_untar.pl b/src/bin/pg_verifybackup/t/008_untar.pl
old mode 100644
new mode 100755
index 6927ca4c74..1ccc6cb9df
--- a/src/bin/pg_verifybackup/t/008_untar.pl
+++ b/src/bin/pg_verifybackup/t/008_untar.pl
@@ -43,6 +43,14 @@ my @test_configuration = (
'decompress_program' => $ENV{'LZ4'},
'decompress_flags' => [ '-d', '-m'],
'enabled' => check_pg_config("#define HAVE_LIBLZ4 1")
+ },
+ {
+ 'compression_method' => 'zstd',
+ 'backup_flags' => ['--compress', 'server-zstd'],
+ 'backup_archive' => 'base.tar.zst',
+ 'decompress_program' => $ENV{'ZSTD'},
+ 'decompress_flags' => [ '-d' ],
+ 'enabled' => check_pg_config("#define HAVE_LIBZSTD 1")
}
);
@@ -108,6 +116,7 @@ for my $tc (@test_configuration)
# Cleanup.
unlink($backup_path . '/backup_manifest');
unlink($backup_path . '/base.tar');
+ unlink($backup_path . '/' . $tc->{'backup_archive'});
rmtree($extract_path);
}
}
diff --git a/src/include/replication/basebackup_sink.h b/src/include/replication/basebackup_sink.h
index a3f8d37258..a7f16758a4 100644
--- a/src/include/replication/basebackup_sink.h
+++ b/src/include/replication/basebackup_sink.h
@@ -285,6 +285,7 @@ extern void bbsink_forward_cleanup(bbsink *sink);
extern bbsink *bbsink_copystream_new(bool send_to_client);
extern bbsink *bbsink_gzip_new(bbsink *next, int compresslevel);
extern bbsink *bbsink_lz4_new(bbsink *next, int compresslevel);
+extern bbsink *bbsink_zstd_new(bbsink *next, int compresslevel);
extern bbsink *bbsink_progress_new(bbsink *next, bool estimate_backup_size);
extern bbsink *bbsink_server_new(bbsink *next, char *pathname);
extern bbsink *bbsink_throttle_new(bbsink *next, uint32 maxrate);
--
2.25.1