2022-03-07 21:08:45 +01:00
|
|
|
/*-------------------------------------------------------------------------
|
|
|
|
*
|
|
|
|
* basebackup_zstd.c
|
|
|
|
* Basebackup sink implementing zstd compression.
|
|
|
|
*
|
2023-01-02 21:00:37 +01:00
|
|
|
* Portions Copyright (c) 2010-2023, PostgreSQL Global Development Group
|
2022-03-07 21:08:45 +01:00
|
|
|
*
|
|
|
|
* IDENTIFICATION
|
2022-08-10 20:03:23 +02:00
|
|
|
* src/backend/backup/basebackup_zstd.c
|
2022-03-07 21:08:45 +01:00
|
|
|
*
|
|
|
|
*-------------------------------------------------------------------------
|
|
|
|
*/
|
|
|
|
#include "postgres.h"
|
|
|
|
|
2022-03-15 18:06:25 +01:00
|
|
|
#ifdef USE_ZSTD
|
2022-03-07 21:08:45 +01:00
|
|
|
#include <zstd.h>
|
|
|
|
#endif
|
|
|
|
|
2022-08-10 20:03:23 +02:00
|
|
|
#include "backup/basebackup_sink.h"
|
2022-03-07 21:08:45 +01:00
|
|
|
|
2022-03-15 18:06:25 +01:00
|
|
|
#ifdef USE_ZSTD
|
2022-03-07 21:08:45 +01:00
|
|
|
|
|
|
|
typedef struct bbsink_zstd
|
|
|
|
{
|
|
|
|
/* Common information for all types of sink. */
|
|
|
|
bbsink base;
|
|
|
|
|
2022-03-30 15:35:14 +02:00
|
|
|
/* Compression options */
|
Rename backup_compression.{c,h} to compression.{c,h}
Compression option handling (level, algorithm or even workers) can be
used across several parts of the system and not only base backups.
Structures, objects and routines are renamed in consequence, to remove
the concept of base backups from this part of the code making this
change straight-forward.
pg_receivewal, that has gained support for LZ4 since babbbb5, will make
use of this infrastructure for its set of compression options, bringing
more consistency with pg_basebackup. This cleanup needs to be done
before releasing a beta of 15. pg_dump is a potential future target, as
well, and adding more compression options to it may happen in 16~.
Author: Michael Paquier
Reviewed-by: Robert Haas, Georgios Kokolatos
Discussion: https://postgr.es/m/YlPQGNAAa04raObK@paquier.xyz
2022-04-12 06:38:54 +02:00
|
|
|
pg_compress_specification *compress;
|
2022-03-07 21:08:45 +01:00
|
|
|
|
|
|
|
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);
|
|
|
|
|
2022-05-12 18:10:03 +02:00
|
|
|
static const bbsink_ops bbsink_zstd_ops = {
|
2022-03-07 21:08:45 +01:00
|
|
|
.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
|
|
|
|
|
|
|
|
/*
|
2022-03-23 14:19:14 +01:00
|
|
|
* Create a new basebackup sink that performs zstd compression.
|
2022-03-07 21:08:45 +01:00
|
|
|
*/
|
|
|
|
bbsink *
|
Rename backup_compression.{c,h} to compression.{c,h}
Compression option handling (level, algorithm or even workers) can be
used across several parts of the system and not only base backups.
Structures, objects and routines are renamed in consequence, to remove
the concept of base backups from this part of the code making this
change straight-forward.
pg_receivewal, that has gained support for LZ4 since babbbb5, will make
use of this infrastructure for its set of compression options, bringing
more consistency with pg_basebackup. This cleanup needs to be done
before releasing a beta of 15. pg_dump is a potential future target, as
well, and adding more compression options to it may happen in 16~.
Author: Michael Paquier
Reviewed-by: Robert Haas, Georgios Kokolatos
Discussion: https://postgr.es/m/YlPQGNAAa04raObK@paquier.xyz
2022-04-12 06:38:54 +02:00
|
|
|
bbsink_zstd_new(bbsink *next, pg_compress_specification *compress)
|
2022-03-07 21:08:45 +01:00
|
|
|
{
|
2022-03-15 18:06:25 +01:00
|
|
|
#ifndef USE_ZSTD
|
2022-03-07 21:08:45 +01:00
|
|
|
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);
|
|
|
|
|
|
|
|
sink = palloc0(sizeof(bbsink_zstd));
|
|
|
|
*((const bbsink_ops **) &sink->base.bbs_ops) = &bbsink_zstd_ops;
|
|
|
|
sink->base.bbs_next = next;
|
2022-03-30 15:35:14 +02:00
|
|
|
sink->compress = compress;
|
2022-03-07 21:08:45 +01:00
|
|
|
|
|
|
|
return &sink->base;
|
|
|
|
#endif
|
|
|
|
}
|
|
|
|
|
2022-03-15 18:06:25 +01:00
|
|
|
#ifdef USE_ZSTD
|
2022-03-07 21:08:45 +01:00
|
|
|
|
|
|
|
/*
|
|
|
|
* Begin backup.
|
|
|
|
*/
|
|
|
|
static void
|
|
|
|
bbsink_zstd_begin_backup(bbsink *sink)
|
|
|
|
{
|
|
|
|
bbsink_zstd *mysink = (bbsink_zstd *) sink;
|
|
|
|
size_t output_buffer_bound;
|
2022-03-28 18:19:05 +02:00
|
|
|
size_t ret;
|
Rename backup_compression.{c,h} to compression.{c,h}
Compression option handling (level, algorithm or even workers) can be
used across several parts of the system and not only base backups.
Structures, objects and routines are renamed in consequence, to remove
the concept of base backups from this part of the code making this
change straight-forward.
pg_receivewal, that has gained support for LZ4 since babbbb5, will make
use of this infrastructure for its set of compression options, bringing
more consistency with pg_basebackup. This cleanup needs to be done
before releasing a beta of 15. pg_dump is a potential future target, as
well, and adding more compression options to it may happen in 16~.
Author: Michael Paquier
Reviewed-by: Robert Haas, Georgios Kokolatos
Discussion: https://postgr.es/m/YlPQGNAAa04raObK@paquier.xyz
2022-04-12 06:38:54 +02:00
|
|
|
pg_compress_specification *compress = mysink->compress;
|
2022-03-07 21:08:45 +01:00
|
|
|
|
|
|
|
mysink->cctx = ZSTD_createCCtx();
|
|
|
|
if (!mysink->cctx)
|
|
|
|
elog(ERROR, "could not create zstd compression context");
|
|
|
|
|
Simplify handling of compression level with compression specifications
PG_COMPRESSION_OPTION_LEVEL is removed from the compression
specification logic, and instead the compression level is always
assigned with each library's default if nothing is directly given. This
centralizes the checks on the compression methods supported by a given
build, and always assigns a default compression level when parsing a
compression specification. This results in complaining at an earlier
stage than previously if a build supports a compression method or not,
aka when parsing a specification in the backend or the frontend, and not
when processing it. zstd, lz4 and zlib are able to handle in their
respective routines setting up the compression level the case of a
default value, hence the backend or frontend code (pg_receivewal or
pg_basebackup) has now no need to know what the default compression
level should be if nothing is specified: the logic is now done so as the
specification parsing assigns it. It can also be enforced by passing
down a "level" set to the default value, that the backend will accept
(the replication protocol is for example able to handle a command like
BASE_BACKUP (COMPRESSION_DETAIL 'gzip:level=-1')).
This code simplification fixes an issue with pg_basebackup --gzip
introduced by ffd5365, where the tarball of the streamed WAL segments
would be created as of pg_wal.tar.gz with uncompressed contents, while
the intention is to compress the segments with gzip at a default level.
The origin of the confusion comes from the handling of the default
compression level of gzip (-1 or Z_DEFAULT_COMPRESSION) and the value of
0 was getting assigned, which is what walmethods.c would consider
as equivalent to no compression when streaming WAL segments with its tar
methods. Assigning always the compression level removes the confusion
of some code paths considering a value of 0 set in a specification as
either no compression or a default compression level.
Note that 010_pg_basebackup.pl has to be adjusted to skip a few tests
where the shape of the compression detail string for client and
server-side compression was checked using gzip. This is a result of the
code simplification, as gzip specifications cannot be used if a build
does not support it.
Reported-by: Tom Lane
Reviewed-by: Tom Lane
Discussion: https://postgr.es/m/1400032.1662217889@sss.pgh.pa.us
Backpatch-through: 15
2022-09-14 05:16:57 +02:00
|
|
|
ret = ZSTD_CCtx_setParameter(mysink->cctx, ZSTD_c_compressionLevel,
|
|
|
|
compress->level);
|
|
|
|
if (ZSTD_isError(ret))
|
|
|
|
elog(ERROR, "could not set zstd compression level to %d: %s",
|
|
|
|
compress->level, ZSTD_getErrorName(ret));
|
2022-03-30 15:35:14 +02:00
|
|
|
|
Rename backup_compression.{c,h} to compression.{c,h}
Compression option handling (level, algorithm or even workers) can be
used across several parts of the system and not only base backups.
Structures, objects and routines are renamed in consequence, to remove
the concept of base backups from this part of the code making this
change straight-forward.
pg_receivewal, that has gained support for LZ4 since babbbb5, will make
use of this infrastructure for its set of compression options, bringing
more consistency with pg_basebackup. This cleanup needs to be done
before releasing a beta of 15. pg_dump is a potential future target, as
well, and adding more compression options to it may happen in 16~.
Author: Michael Paquier
Reviewed-by: Robert Haas, Georgios Kokolatos
Discussion: https://postgr.es/m/YlPQGNAAa04raObK@paquier.xyz
2022-04-12 06:38:54 +02:00
|
|
|
if ((compress->options & PG_COMPRESSION_OPTION_WORKERS) != 0)
|
2022-03-30 15:35:14 +02:00
|
|
|
{
|
|
|
|
/*
|
|
|
|
* On older versions of libzstd, this option does not exist, and
|
|
|
|
* trying to set it will fail. Similarly for newer versions if they
|
|
|
|
* are compiled without threading support.
|
|
|
|
*/
|
|
|
|
ret = ZSTD_CCtx_setParameter(mysink->cctx, ZSTD_c_nbWorkers,
|
|
|
|
compress->workers);
|
|
|
|
if (ZSTD_isError(ret))
|
|
|
|
ereport(ERROR,
|
|
|
|
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
|
|
|
|
errmsg("could not set compression worker count to %d: %s",
|
|
|
|
compress->workers, ZSTD_getErrorName(ret)));
|
|
|
|
}
|
2022-03-07 21:08:45 +01:00
|
|
|
|
2023-04-06 17:18:38 +02:00
|
|
|
if ((compress->options & PG_COMPRESSION_OPTION_LONG_DISTANCE) != 0)
|
|
|
|
{
|
|
|
|
ret = ZSTD_CCtx_setParameter(mysink->cctx,
|
|
|
|
ZSTD_c_enableLongDistanceMatching,
|
|
|
|
compress->long_distance);
|
|
|
|
if (ZSTD_isError(ret))
|
|
|
|
ereport(ERROR,
|
|
|
|
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
|
|
|
|
errmsg("could not set compression flag for %s: %s",
|
|
|
|
"long", ZSTD_getErrorName(ret)));
|
|
|
|
}
|
|
|
|
|
2022-03-07 21:08:45 +01:00
|
|
|
/*
|
|
|
|
* 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
|