<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>
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>
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 is selected, but will not be compressed if any other
+ compression algorithm is selected, or if server-side compression
+ is selected.
</para>
</listitem>
</varlistentry>
basebackup_copy.o \
basebackup_gzip.o \
basebackup_lz4.o \
+ basebackup_zstd.o \
basebackup_progress.o \
basebackup_server.o \
basebackup_sink.o \
{
BACKUP_COMPRESSION_NONE,
BACKUP_COMPRESSION_GZIP,
- BACKUP_COMPRESSION_LZ4
+ BACKUP_COMPRESSION_LZ4,
+ BACKUP_COMPRESSION_ZSTD
} basebackup_compression_type;
typedef struct
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),
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);
--- /dev/null
+/*-------------------------------------------------------------------------
+ *
+ * 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 max_needed = 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 < max_needed)
+ {
+ 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 max_needed = 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 < max_needed)
+ {
+ 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 any compression context that
+ * got allocated, so that we don't leak memory.
+ */
+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
bbstreamer_gzip.o \
bbstreamer_inject.o \
bbstreamer_lz4.o \
+ bbstreamer_zstd.o \
bbstreamer_tar.o
all: pg_basebackup pg_receivewal pg_recvlogical
extern bbstreamer *bbstreamer_lz4_compressor_new(bbstreamer *next,
int compresslevel);
extern bbstreamer *bbstreamer_lz4_decompressor_new(bbstreamer *next);
+extern bbstreamer *bbstreamer_zstd_compressor_new(bbstreamer *next,
+ int compresslevel);
+extern bbstreamer *bbstreamer_zstd_decompressor_new(bbstreamer *next);
extern bbstreamer *bbstreamer_tar_parser_new(bbstreamer *next);
extern bbstreamer *bbstreamer_tar_terminator_new(bbstreamer *next);
extern bbstreamer *bbstreamer_tar_archiver_new(bbstreamer *next);
--- /dev/null
+/*-------------------------------------------------------------------------
+ *
+ * bbstreamer_zstd.c
+ *
+ * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/bin/pg_basebackup/bbstreamer_zstd.c
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres_fe.h"
+
+#include <unistd.h>
+
+#ifdef HAVE_LIBZSTD
+#include <zstd.h>
+#endif
+
+#include "bbstreamer.h"
+#include "common/logging.h"
+
+#ifdef HAVE_LIBZSTD
+
+typedef struct bbstreamer_zstd_frame
+{
+ bbstreamer base;
+
+ ZSTD_CCtx *cctx;
+ ZSTD_DCtx *dctx;
+ ZSTD_outBuffer zstd_outBuf;
+} bbstreamer_zstd_frame;
+
+static void bbstreamer_zstd_compressor_content(bbstreamer *streamer,
+ bbstreamer_member *member,
+ const char *data, int len,
+ bbstreamer_archive_context context);
+static void bbstreamer_zstd_compressor_finalize(bbstreamer *streamer);
+static void bbstreamer_zstd_compressor_free(bbstreamer *streamer);
+
+const bbstreamer_ops bbstreamer_zstd_compressor_ops = {
+ .content = bbstreamer_zstd_compressor_content,
+ .finalize = bbstreamer_zstd_compressor_finalize,
+ .free = bbstreamer_zstd_compressor_free
+};
+
+static void bbstreamer_zstd_decompressor_content(bbstreamer *streamer,
+ bbstreamer_member *member,
+ const char *data, int len,
+ bbstreamer_archive_context context);
+static void bbstreamer_zstd_decompressor_finalize(bbstreamer *streamer);
+static void bbstreamer_zstd_decompressor_free(bbstreamer *streamer);
+
+const bbstreamer_ops bbstreamer_zstd_decompressor_ops = {
+ .content = bbstreamer_zstd_decompressor_content,
+ .finalize = bbstreamer_zstd_decompressor_finalize,
+ .free = bbstreamer_zstd_decompressor_free
+};
+#endif
+
+/*
+ * Create a new base backup streamer that performs zstd compression of tar
+ * blocks.
+ */
+bbstreamer *
+bbstreamer_zstd_compressor_new(bbstreamer *next, int compresslevel)
+{
+#ifdef HAVE_LIBZSTD
+ bbstreamer_zstd_frame *streamer;
+
+ Assert(next != NULL);
+
+ streamer = palloc0(sizeof(bbstreamer_zstd_frame));
+
+ *((const bbstreamer_ops **) &streamer->base.bbs_ops) =
+ &bbstreamer_zstd_compressor_ops;
+
+ streamer->base.bbs_next = next;
+ initStringInfo(&streamer->base.bbs_buffer);
+ enlargeStringInfo(&streamer->base.bbs_buffer, ZSTD_DStreamOutSize());
+
+ streamer->cctx = ZSTD_createCCtx();
+ if (!streamer->cctx)
+ pg_log_error("could not create zstd compression context");
+
+ /* Initialize stream compression preferences */
+ ZSTD_CCtx_setParameter(streamer->cctx, ZSTD_c_compressionLevel,
+ compresslevel);
+
+ /* Initialize the ZSTD output buffer. */
+ streamer->zstd_outBuf.dst = streamer->base.bbs_buffer.data;
+ streamer->zstd_outBuf.size = streamer->base.bbs_buffer.maxlen;
+ streamer->zstd_outBuf.pos = 0;
+
+ return &streamer->base;
+#else
+ pg_log_error("this build does not support zstd compression");
+ exit(1);
+#endif
+}
+
+#ifdef HAVE_LIBZSTD
+/*
+ * Compress the input data to output buffer.
+ *
+ * Find out the compression bound based on input data length for each
+ * invocation to make sure that output buffer has enough capacity to
+ * accommodate the compressed data. In case if the output buffer
+ * capacity falls short of compression bound then forward the content
+ * of output buffer to next streamer and empty the buffer.
+ */
+static void
+bbstreamer_zstd_compressor_content(bbstreamer *streamer,
+ bbstreamer_member *member,
+ const char *data, int len,
+ bbstreamer_archive_context context)
+{
+ bbstreamer_zstd_frame *mystreamer = (bbstreamer_zstd_frame *) streamer;
+ ZSTD_inBuffer inBuf = {data, len, 0};
+
+ while (inBuf.pos < inBuf.size)
+ {
+ size_t yet_to_flush;
+ size_t max_needed = ZSTD_compressBound(inBuf.size - inBuf.pos);
+
+ /*
+ * If the output buffer is not left with enough space, send the
+ * compressed bytes to the next streamer, and empty the buffer.
+ */
+ if (mystreamer->zstd_outBuf.size - mystreamer->zstd_outBuf.pos <
+ max_needed)
+ {
+ bbstreamer_content(mystreamer->base.bbs_next, member,
+ mystreamer->zstd_outBuf.dst,
+ mystreamer->zstd_outBuf.pos,
+ context);
+
+ /* Reset the ZSTD output buffer. */
+ mystreamer->zstd_outBuf.dst = mystreamer->base.bbs_buffer.data;
+ mystreamer->zstd_outBuf.size = mystreamer->base.bbs_buffer.maxlen;
+ mystreamer->zstd_outBuf.pos = 0;
+ }
+
+ yet_to_flush =
+ ZSTD_compressStream2(mystreamer->cctx, &mystreamer->zstd_outBuf,
+ &inBuf, ZSTD_e_continue);
+
+ if (ZSTD_isError(yet_to_flush))
+ pg_log_error("could not compress data: %s",
+ ZSTD_getErrorName(yet_to_flush));
+ }
+}
+
+/*
+ * End-of-stream processing.
+ */
+static void
+bbstreamer_zstd_compressor_finalize(bbstreamer *streamer)
+{
+ bbstreamer_zstd_frame *mystreamer = (bbstreamer_zstd_frame *) streamer;
+ size_t yet_to_flush;
+
+ do
+ {
+ ZSTD_inBuffer in = {NULL, 0, 0};
+ size_t max_needed = ZSTD_compressBound(0);
+
+ /*
+ * If the output buffer is not left with enough space, send the
+ * compressed bytes to the next streamer, and empty the buffer.
+ */
+ if (mystreamer->zstd_outBuf.size - mystreamer->zstd_outBuf.pos <
+ max_needed)
+ {
+ bbstreamer_content(mystreamer->base.bbs_next, NULL,
+ mystreamer->zstd_outBuf.dst,
+ mystreamer->zstd_outBuf.pos,
+ BBSTREAMER_UNKNOWN);
+
+ /* Reset the ZSTD output buffer. */
+ mystreamer->zstd_outBuf.dst = mystreamer->base.bbs_buffer.data;
+ mystreamer->zstd_outBuf.size = mystreamer->base.bbs_buffer.maxlen;
+ mystreamer->zstd_outBuf.pos = 0;
+ }
+
+ yet_to_flush = ZSTD_compressStream2(mystreamer->cctx,
+ &mystreamer->zstd_outBuf,
+ &in, ZSTD_e_end);
+
+ if (ZSTD_isError(yet_to_flush))
+ pg_log_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 streamer. */
+ if (mystreamer->zstd_outBuf.pos > 0)
+ bbstreamer_content(mystreamer->base.bbs_next, NULL,
+ mystreamer->zstd_outBuf.dst,
+ mystreamer->zstd_outBuf.pos,
+ BBSTREAMER_UNKNOWN);
+
+ bbstreamer_finalize(mystreamer->base.bbs_next);
+}
+
+/*
+ * Free memory.
+ */
+static void
+bbstreamer_zstd_compressor_free(bbstreamer *streamer)
+{
+ bbstreamer_zstd_frame *mystreamer = (bbstreamer_zstd_frame *) streamer;
+
+ bbstreamer_free(streamer->bbs_next);
+ ZSTD_freeCCtx(mystreamer->cctx);
+ pfree(streamer->bbs_buffer.data);
+ pfree(streamer);
+}
+#endif
+
+/*
+ * Create a new base backup streamer that performs decompression of zstd
+ * compressed blocks.
+ */
+bbstreamer *
+bbstreamer_zstd_decompressor_new(bbstreamer *next)
+{
+#ifdef HAVE_LIBZSTD
+ bbstreamer_zstd_frame *streamer;
+
+ Assert(next != NULL);
+
+ streamer = palloc0(sizeof(bbstreamer_zstd_frame));
+ *((const bbstreamer_ops **) &streamer->base.bbs_ops) =
+ &bbstreamer_zstd_decompressor_ops;
+
+ streamer->base.bbs_next = next;
+ initStringInfo(&streamer->base.bbs_buffer);
+ enlargeStringInfo(&streamer->base.bbs_buffer, ZSTD_DStreamOutSize());
+
+ streamer->dctx = ZSTD_createDCtx();
+ if (!streamer->dctx)
+ {
+ pg_log_error("could not create zstd decompression context");
+ exit(1);
+ }
+
+ /* Initialize the ZSTD output buffer. */
+ streamer->zstd_outBuf.dst = streamer->base.bbs_buffer.data;
+ streamer->zstd_outBuf.size = streamer->base.bbs_buffer.maxlen;
+ streamer->zstd_outBuf.pos = 0;
+
+ return &streamer->base;
+#else
+ pg_log_error("this build does not support compression");
+ exit(1);
+#endif
+}
+
+#ifdef HAVE_LIBZSTD
+/*
+ * Decompress the input data to output buffer until we run out of input
+ * data. Each time the output buffer is full, pass on the decompressed data
+ * to the next streamer.
+ */
+static void
+bbstreamer_zstd_decompressor_content(bbstreamer *streamer,
+ bbstreamer_member *member,
+ const char *data, int len,
+ bbstreamer_archive_context context)
+{
+ bbstreamer_zstd_frame *mystreamer = (bbstreamer_zstd_frame *) streamer;
+ ZSTD_inBuffer inBuf = {data, len, 0};
+
+ while (inBuf.pos < inBuf.size)
+ {
+ size_t ret;
+
+ /*
+ * If output buffer is full then forward the content to next streamer
+ * and update the output buffer.
+ */
+ if (mystreamer->zstd_outBuf.pos >= mystreamer->zstd_outBuf.size)
+ {
+ bbstreamer_content(mystreamer->base.bbs_next, member,
+ mystreamer->zstd_outBuf.dst,
+ mystreamer->zstd_outBuf.pos,
+ context);
+
+ /* Reset the ZSTD output buffer. */
+ mystreamer->zstd_outBuf.dst = mystreamer->base.bbs_buffer.data;
+ mystreamer->zstd_outBuf.size = mystreamer->base.bbs_buffer.maxlen;
+ mystreamer->zstd_outBuf.pos = 0;
+ }
+
+ ret = ZSTD_decompressStream(mystreamer->dctx,
+ &mystreamer->zstd_outBuf, &inBuf);
+
+ if (ZSTD_isError(ret))
+ pg_log_error("could not decompress data: %s", ZSTD_getErrorName(ret));
+ }
+}
+
+/*
+ * End-of-stream processing.
+ */
+static void
+bbstreamer_zstd_decompressor_finalize(bbstreamer *streamer)
+{
+ bbstreamer_zstd_frame *mystreamer = (bbstreamer_zstd_frame *) streamer;
+
+ /*
+ * End of the stream, if there is some pending data in output buffers then
+ * we must forward it to next streamer.
+ */
+ if (mystreamer->zstd_outBuf.pos > 0)
+ bbstreamer_content(mystreamer->base.bbs_next, NULL,
+ mystreamer->base.bbs_buffer.data,
+ mystreamer->base.bbs_buffer.maxlen,
+ BBSTREAMER_UNKNOWN);
+
+ bbstreamer_finalize(mystreamer->base.bbs_next);
+}
+
+/*
+ * Free memory.
+ */
+static void
+bbstreamer_zstd_decompressor_free(bbstreamer *streamer)
+{
+ bbstreamer_zstd_frame *mystreamer = (bbstreamer_zstd_frame *) streamer;
+
+ bbstreamer_free(streamer->bbs_next);
+ ZSTD_freeDCtx(mystreamer->dctx);
+ pfree(streamer->bbs_buffer.data);
+ pfree(streamer);
+}
+#endif
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"));
*methodres = COMPRESSION_LZ4;
*locationres = COMPRESS_LOCATION_SERVER;
}
+ else if (pg_strcasecmp(firstpart, "zstd") == 0)
+ {
+ *methodres = COMPRESSION_ZSTD;
+ *locationres = COMPRESS_LOCATION_UNSPECIFIED;
+ }
+ else if (pg_strcasecmp(firstpart, "client-zstd") == 0)
+ {
+ *methodres = COMPRESSION_ZSTD;
+ *locationres = COMPRESS_LOCATION_CLIENT;
+ }
+ 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;
bool inject_manifest;
bool is_tar,
is_tar_gz,
- is_tar_lz4;
+ is_tar_lz4,
+ is_tar_zstd;
bool must_parse_archive;
int archive_name_len = strlen(archive_name);
is_tar_lz4 = (archive_name_len > 8 &&
strcmp(archive_name + archive_name_len - 4, ".lz4") == 0);
+ /* Is this a ZSTD archive? */
+ is_tar_zstd = (archive_name_len > 8 &&
+ strcmp(archive_name + archive_name_len - 4, ".zst") == 0);
+
/*
* We have to parse the archive if (1) we're suppose to extract it, or if
* (2) we need to inject backup_manifest or recovery configuration into it.
(spclocation == NULL && writerecoveryconf));
/* At present, we only know how to parse tar archives. */
- if (must_parse_archive && !is_tar && !is_tar_gz && !is_tar_lz4)
+ if (must_parse_archive && !is_tar && !is_tar_gz && !is_tar_lz4
+ && !is_tar_zstd)
{
pg_log_error("unable to parse archive: %s", archive_name);
pg_log_info("only tar archives can be parsed");
streamer = bbstreamer_lz4_compressor_new(streamer,
compresslevel);
}
+ else if (compressmethod == COMPRESSION_ZSTD)
+ {
+ strlcat(archive_filename, ".zst", sizeof(archive_filename));
+ streamer = bbstreamer_plain_writer_new(archive_filename,
+ archive_file);
+ streamer = bbstreamer_zstd_compressor_new(streamer,
+ compresslevel);
+ }
else
{
Assert(false); /* not reachable */
streamer = bbstreamer_gzip_decompressor_new(streamer);
else if (compressmethod == COMPRESSION_LZ4)
streamer = bbstreamer_lz4_decompressor_new(streamer);
+ else if (compressmethod == COMPRESSION_ZSTD)
+ streamer = bbstreamer_zstd_decompressor_new(streamer);
}
/* Return the results. */
case COMPRESSION_LZ4:
compressmethodstr = "lz4";
break;
+ case COMPRESSION_ZSTD:
+ compressmethodstr = "zstd";
+ break;
default:
Assert(false);
break;
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;
}
/*
exit(1);
#endif
break;
+ case COMPRESSION_ZSTD:
+ pg_log_error("compression with %s is not yet supported", "ZSTD");
+ exit(1);
+
}
{
COMPRESSION_GZIP,
COMPRESSION_LZ4,
+ COMPRESSION_ZSTD,
COMPRESSION_NONE
} WalCompressionMethod;
# name.
export GZIP_PROGRAM=$(GZIP)
export LZ4=$(LZ4)
+export ZSTD=$(ZSTD)
subdir = src/bin/pg_verifybackup
top_builddir = ../../..
'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")
}
);
# Cleanup.
unlink($backup_path . '/backup_manifest');
unlink($backup_path . '/base.tar');
+ unlink($backup_path . '/' . $tc->{'backup_archive'});
rmtree($extract_path);
}
}
'compression_method' => 'lz4',
'backup_flags' => ['--compress', 'server-lz4:5'],
'enabled' => check_pg_config("#define HAVE_LIBLZ4 1")
+ },
+ {
+ 'compression_method' => 'zstd',
+ 'backup_flags' => ['--compress', 'server-zstd:5'],
+ 'enabled' => check_pg_config("#define HAVE_LIBZSTD 1")
}
);
'decompress_flags' => [ '-d' ],
'output_file' => 'base.tar',
'enabled' => check_pg_config("#define HAVE_LIBLZ4 1")
+ },
+ {
+ 'compression_method' => 'zstd',
+ 'backup_flags' => ['--compress', 'client-zstd:5'],
+ 'backup_archive' => 'base.tar.zst',
+ 'decompress_program' => $ENV{'ZSTD'},
+ 'decompress_flags' => [ '-d' ],
+ 'enabled' => check_pg_config("#define HAVE_LIBZSTD 1")
}
);
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);
$pgbasebackup->AddFile('src/bin/pg_basebackup/bbstreamer_gzip.c');
$pgbasebackup->AddFile('src/bin/pg_basebackup/bbstreamer_inject.c');
$pgbasebackup->AddFile('src/bin/pg_basebackup/bbstreamer_lz4.c');
+ $pgbasebackup->AddFile('src/bin/pg_basebackup/bbstreamer_zstd.c');
$pgbasebackup->AddFile('src/bin/pg_basebackup/bbstreamer_tar.c');
$pgbasebackup->AddLibrary('ws2_32.lib');
projects / users / c2main / postgres.git / commitdiff