mirror/postgresql
2
0
Fork 0
mirror of https://git.postgresql.org/git/postgresql.git synced 2025-03-07 19:47:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

Allow server-side compression to be used with -Fp.

Browse Source 
If you have a low-bandwidth connection between the client and the
server, it's reasonable to want to compress on the server side but
then decompress and extract the backup on the client side. This
commit allows you do to do just that.

Dipesh Pandit, with minor and mostly cosmetic changes by me.

Discussion: http://postgr.es/m/CAN1g5_HiSh8ajUMd4ePtGyCXo89iKZTzaNyzP_qv1eJbi4YHXA@mail.gmail.com
This commit is contained in:
Robert Haas 2022-01-28 08:41:25 -05:00
parent 43f33dc018
commit d45099425e
7 changed files with 463 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
doc/src/sgml/ref/pg_basebackup.sgml
View File

@ -428,8 +428,11 @@ PostgreSQL documentation
</para>
<para>
When the tar format is used, the suffix <filename>.gz</filename> will
automatically be added to all tar filenames. Compression is not
available in plain format.
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.
</para>
</listitem>
</varlistentry>

1
src/bin/pg_basebackup/Makefile
View File

@ -38,6 +38,7 @@ OBJS = \
BBOBJS = \
pg_basebackup.o \
bbstreamer_file.o \
bbstreamer_gzip.o \
bbstreamer_inject.o \
bbstreamer_tar.o

1
src/bin/pg_basebackup/bbstreamer.h
View File

@ -205,6 +205,7 @@ extern bbstreamer *bbstreamer_extractor_new(const char *basepath,
const char *(*link_map) (const char *),
void (*report_output_file) (const char *));
extern bbstreamer *bbstreamer_gzip_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);

182
src/bin/pg_basebackup/bbstreamer_file.c
View File

@ -11,10 +11,6 @@
#include "postgres_fe.h"
#ifdef HAVE_LIBZ
#include <zlib.h>
#endif
#include <unistd.h>
#include "bbstreamer.h"
@ -30,15 +26,6 @@ typedef struct bbstreamer_plain_writer
bool should_close_file;
} bbstreamer_plain_writer;
#ifdef HAVE_LIBZ
typedef struct bbstreamer_gzip_writer
{
bbstreamer base;
char *pathname;
gzFile gzfile;
} bbstreamer_gzip_writer;
#endif
typedef struct bbstreamer_extractor
{
bbstreamer base;
@ -62,22 +49,6 @@ const bbstreamer_ops bbstreamer_plain_writer_ops = {
.free = bbstreamer_plain_writer_free
};
#ifdef HAVE_LIBZ
static void bbstreamer_gzip_writer_content(bbstreamer *streamer,
bbstreamer_member *member,
const char *data, int len,
bbstreamer_archive_context context);
static void bbstreamer_gzip_writer_finalize(bbstreamer *streamer);
static void bbstreamer_gzip_writer_free(bbstreamer *streamer);
static const char *get_gz_error(gzFile gzf);
const bbstreamer_ops bbstreamer_gzip_writer_ops = {
.content = bbstreamer_gzip_writer_content,
.finalize = bbstreamer_gzip_writer_finalize,
.free = bbstreamer_gzip_writer_free
};
#endif
static void bbstreamer_extractor_content(bbstreamer *streamer,
bbstreamer_member *member,
const char *data, int len,
@ -195,159 +166,6 @@ bbstreamer_plain_writer_free(bbstreamer *streamer)
pfree(mystreamer);
}
/*
* Create a bbstreamer that just compresses data using gzip, and then writes
* it to a file.
*
* As in the case of bbstreamer_plain_writer_new, pathname is always used
* for error reporting purposes; if file is NULL, it is also the opened and
* closed so that the data may be written there.
*/
bbstreamer *
bbstreamer_gzip_writer_new(char *pathname, FILE *file, int compresslevel)
{
#ifdef HAVE_LIBZ
bbstreamer_gzip_writer *streamer;
streamer = palloc0(sizeof(bbstreamer_gzip_writer));
*((const bbstreamer_ops **) &streamer->base.bbs_ops) =
&bbstreamer_gzip_writer_ops;
streamer->pathname = pstrdup(pathname);
if (file == NULL)
{
streamer->gzfile = gzopen(pathname, "wb");
if (streamer->gzfile == NULL)
{
pg_log_error("could not create compressed file \"%s\": %m",
pathname);
exit(1);
}
}
else
{
int fd = dup(fileno(file));
if (fd < 0)
{
pg_log_error("could not duplicate stdout: %m");
exit(1);
}
streamer->gzfile = gzdopen(fd, "wb");
if (streamer->gzfile == NULL)
{
pg_log_error("could not open output file: %m");
exit(1);
}
}
if (gzsetparams(streamer->gzfile, compresslevel,
Z_DEFAULT_STRATEGY) != Z_OK)
{
pg_log_error("could not set compression level %d: %s",
compresslevel, get_gz_error(streamer->gzfile));
exit(1);
}
return &streamer->base;
#else
pg_log_error("this build does not support compression");
exit(1);
#endif
}
#ifdef HAVE_LIBZ
/*
* Write archive content to gzip file.
*/
static void
bbstreamer_gzip_writer_content(bbstreamer *streamer,
bbstreamer_member *member, const char *data,
int len, bbstreamer_archive_context context)
{
bbstreamer_gzip_writer *mystreamer;
mystreamer = (bbstreamer_gzip_writer *) streamer;
if (len == 0)
return;
errno = 0;
if (gzwrite(mystreamer->gzfile, data, len) != len)
{
/* if write didn't set errno, assume problem is no disk space */
if (errno == 0)
errno = ENOSPC;
pg_log_error("could not write to compressed file \"%s\": %s",
mystreamer->pathname, get_gz_error(mystreamer->gzfile));
exit(1);
}
}
/*
* End-of-archive processing when writing to a gzip file consists of just
* calling gzclose.
*
* It makes no difference whether we opened the file or the caller did it,
* because libz provides no way of avoiding a close on the underling file
* handle. Notice, however, that bbstreamer_gzip_writer_new() uses dup() to
* work around this issue, so that the behavior from the caller's viewpoint
* is the same as for bbstreamer_plain_writer.
*/
static void
bbstreamer_gzip_writer_finalize(bbstreamer *streamer)
{
bbstreamer_gzip_writer *mystreamer;
mystreamer = (bbstreamer_gzip_writer *) streamer;
errno = 0; /* in case gzclose() doesn't set it */
if (gzclose(mystreamer->gzfile) != 0)
{
pg_log_error("could not close compressed file \"%s\": %m",
mystreamer->pathname);
exit(1);
}
mystreamer->gzfile = NULL;
}
/*
* Free memory associated with this bbstreamer.
*/
static void
bbstreamer_gzip_writer_free(bbstreamer *streamer)
{
bbstreamer_gzip_writer *mystreamer;
mystreamer = (bbstreamer_gzip_writer *) streamer;
Assert(mystreamer->base.bbs_next == NULL);
Assert(mystreamer->gzfile == NULL);
pfree(mystreamer->pathname);
pfree(mystreamer);
}
/*
* Helper function for libz error reporting.
*/
static const char *
get_gz_error(gzFile gzf)
{
int errnum;
const char *errmsg;
errmsg = gzerror(gzf, &errnum);
if (errnum == Z_ERRNO)
return strerror(errno);
else
return errmsg;
}
#endif
/*
* Create a bbstreamer that extracts an archive.
*

380
src/bin/pg_basebackup/bbstreamer_gzip.c Normal file
View File

@ -0,0 +1,380 @@
/*-------------------------------------------------------------------------
*
* bbstreamer_gzip.c
*
* Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
*
* IDENTIFICATION
* src/bin/pg_basebackup/bbstreamer_gzip.c
*-------------------------------------------------------------------------
*/
#include "postgres_fe.h"
#ifdef HAVE_LIBZ
#include <zlib.h>
#endif
#include "bbstreamer.h"
#include "common/logging.h"
#include "common/file_perm.h"
#include "common/string.h"
#ifdef HAVE_LIBZ
typedef struct bbstreamer_gzip_writer
{
bbstreamer base;
char *pathname;
gzFile gzfile;
} bbstreamer_gzip_writer;
typedef struct bbstreamer_gzip_decompressor
{
bbstreamer base;
z_stream zstream;
size_t bytes_written;
} bbstreamer_gzip_decompressor;
static void bbstreamer_gzip_writer_content(bbstreamer *streamer,
bbstreamer_member *member,
const char *data, int len,
bbstreamer_archive_context context);
static void bbstreamer_gzip_writer_finalize(bbstreamer *streamer);
static void bbstreamer_gzip_writer_free(bbstreamer *streamer);
static const char *get_gz_error(gzFile gzf);
const bbstreamer_ops bbstreamer_gzip_writer_ops = {
.content = bbstreamer_gzip_writer_content,
.finalize = bbstreamer_gzip_writer_finalize,
.free = bbstreamer_gzip_writer_free
};
static void bbstreamer_gzip_decompressor_content(bbstreamer *streamer,
bbstreamer_member *member,
const char *data, int len,
bbstreamer_archive_context context);
static void bbstreamer_gzip_decompressor_finalize(bbstreamer *streamer);
static void bbstreamer_gzip_decompressor_free(bbstreamer *streamer);
static void *gzip_palloc(void *opaque, unsigned items, unsigned size);
static void gzip_pfree(void *opaque, void *address);
const bbstreamer_ops bbstreamer_gzip_decompressor_ops = {
.content = bbstreamer_gzip_decompressor_content,
.finalize = bbstreamer_gzip_decompressor_finalize,
.free = bbstreamer_gzip_decompressor_free
};
#endif
/*
* Create a bbstreamer that just compresses data using gzip, and then writes
* it to a file.
*
* As in the case of bbstreamer_plain_writer_new, pathname is always used
* for error reporting purposes; if file is NULL, it is also the opened and
* closed so that the data may be written there.
*/
bbstreamer *
bbstreamer_gzip_writer_new(char *pathname, FILE *file, int compresslevel)
{
#ifdef HAVE_LIBZ
bbstreamer_gzip_writer *streamer;
streamer = palloc0(sizeof(bbstreamer_gzip_writer));
*((const bbstreamer_ops **) &streamer->base.bbs_ops) =
&bbstreamer_gzip_writer_ops;
streamer->pathname = pstrdup(pathname);
if (file == NULL)
{
streamer->gzfile = gzopen(pathname, "wb");
if (streamer->gzfile == NULL)
{
pg_log_error("could not create compressed file \"%s\": %m",
pathname);
exit(1);
}
}
else
{
int fd = dup(fileno(file));
if (fd < 0)
{
pg_log_error("could not duplicate stdout: %m");
exit(1);
}
streamer->gzfile = gzdopen(fd, "wb");
if (streamer->gzfile == NULL)
{
pg_log_error("could not open output file: %m");
exit(1);
}
}
if (gzsetparams(streamer->gzfile, compresslevel,
Z_DEFAULT_STRATEGY) != Z_OK)
{
pg_log_error("could not set compression level %d: %s",
compresslevel, get_gz_error(streamer->gzfile));
exit(1);
}
return &streamer->base;
#else
pg_log_error("this build does not support compression");
exit(1);
#endif
}
#ifdef HAVE_LIBZ
/*
* Write archive content to gzip file.
*/
static void
bbstreamer_gzip_writer_content(bbstreamer *streamer,
bbstreamer_member *member, const char *data,
int len, bbstreamer_archive_context context)
{
bbstreamer_gzip_writer *mystreamer;
mystreamer = (bbstreamer_gzip_writer *) streamer;
if (len == 0)
return;
errno = 0;
if (gzwrite(mystreamer->gzfile, data, len) != len)
{
/* if write didn't set errno, assume problem is no disk space */
if (errno == 0)
errno = ENOSPC;
pg_log_error("could not write to compressed file \"%s\": %s",
mystreamer->pathname, get_gz_error(mystreamer->gzfile));
exit(1);
}
}
/*
* End-of-archive processing when writing to a gzip file consists of just
* calling gzclose.
*
* It makes no difference whether we opened the file or the caller did it,
* because libz provides no way of avoiding a close on the underling file
* handle. Notice, however, that bbstreamer_gzip_writer_new() uses dup() to
* work around this issue, so that the behavior from the caller's viewpoint
* is the same as for bbstreamer_plain_writer.
*/
static void
bbstreamer_gzip_writer_finalize(bbstreamer *streamer)
{
bbstreamer_gzip_writer *mystreamer;
mystreamer = (bbstreamer_gzip_writer *) streamer;
errno = 0; /* in case gzclose() doesn't set it */
if (gzclose(mystreamer->gzfile) != 0)
{
pg_log_error("could not close compressed file \"%s\": %m",
mystreamer->pathname);
exit(1);
}
mystreamer->gzfile = NULL;
}
/*
* Free memory associated with this bbstreamer.
*/
static void
bbstreamer_gzip_writer_free(bbstreamer *streamer)
{
bbstreamer_gzip_writer *mystreamer;
mystreamer = (bbstreamer_gzip_writer *) streamer;
Assert(mystreamer->base.bbs_next == NULL);
Assert(mystreamer->gzfile == NULL);
pfree(mystreamer->pathname);
pfree(mystreamer);
}
/*
* Helper function for libz error reporting.
*/
static const char *
get_gz_error(gzFile gzf)
{
int errnum;
const char *errmsg;
errmsg = gzerror(gzf, &errnum);
if (errnum == Z_ERRNO)
return strerror(errno);
else
return errmsg;
}
#endif
/*
* Create a new base backup streamer that performs decompression of gzip
* compressed blocks.
*/
bbstreamer *
bbstreamer_gzip_decompressor_new(bbstreamer *next)
{
#ifdef HAVE_LIBZ
bbstreamer_gzip_decompressor *streamer;
z_stream *zs;
Assert(next != NULL);
streamer = palloc0(sizeof(bbstreamer_gzip_decompressor));
*((const bbstreamer_ops **) &streamer->base.bbs_ops) =
&bbstreamer_gzip_decompressor_ops;
streamer->base.bbs_next = next;
initStringInfo(&streamer->base.bbs_buffer);
/* Initialize internal stream state for decompression */
zs = &streamer->zstream;
zs->zalloc = gzip_palloc;
zs->zfree = gzip_pfree;
zs->next_out = (uint8 *) streamer->base.bbs_buffer.data;
zs->avail_out = streamer->base.bbs_buffer.maxlen;
/*
* Data compression was initialized using deflateInit2 to request a gzip
* header. Similarly, we are using inflateInit2 to initialize data
* decompression.
*
* Per the documentation for inflateInit2, the second argument is
* "windowBits" and its value must be greater than or equal to the value
* provided while compressing the data, so we are using the maximum
* possible value for safety.
*/
if (inflateInit2(zs, 15 + 16) != Z_OK)
{
pg_log_error("could not initialize compression library");
exit(1);
}
return &streamer->base;
#else
pg_log_error("this build does not support compression");
exit(1);
#endif
}
#ifdef HAVE_LIBZ
/*
* 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_gzip_decompressor_content(bbstreamer *streamer,
bbstreamer_member *member,
const char *data, int len,
bbstreamer_archive_context context)
{
bbstreamer_gzip_decompressor *mystreamer;
z_stream *zs;
mystreamer = (bbstreamer_gzip_decompressor *) streamer;
zs = &mystreamer->zstream;
zs->next_in = (uint8 *) data;
zs->avail_in = len;
/* Process the current chunk */
while (zs->avail_in > 0)
{
int res;
Assert(mystreamer->bytes_written < mystreamer->base.bbs_buffer.maxlen);
zs->next_out = (uint8 *)
mystreamer->base.bbs_buffer.data + mystreamer->bytes_written;
zs->avail_out =
mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
/*
* This call decompresses data starting at zs->next_in and updates
* zs->next_in * and zs->avail_in. It generates output data starting at
* zs->next_out and updates zs->next_out and zs->avail_out accordingly.
*/
res = inflate(zs, Z_NO_FLUSH);
if (res == Z_STREAM_ERROR)
pg_log_error("could not decompress data: %s", zs->msg);
mystreamer->bytes_written =
mystreamer->base.bbs_buffer.maxlen - zs->avail_out;
/* If output buffer is full then pass data to next streamer */
if (mystreamer->bytes_written >= mystreamer->base.bbs_buffer.maxlen)
{
bbstreamer_content(mystreamer->base.bbs_next, member,
mystreamer->base.bbs_buffer.data,
mystreamer->base.bbs_buffer.maxlen, context);
mystreamer->bytes_written = 0;
}
}
}
/*
* End-of-stream processing.
*/
static void
bbstreamer_gzip_decompressor_finalize(bbstreamer *streamer)
{
bbstreamer_gzip_decompressor *mystreamer;
mystreamer = (bbstreamer_gzip_decompressor *) streamer;
/*
* End of the stream, if there is some pending data in output buffers then
* we must forward it to next streamer.
*/
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_gzip_decompressor_free(bbstreamer *streamer)
{
bbstreamer_free(streamer->bbs_next);
pfree(streamer->bbs_buffer.data);
pfree(streamer);
}
/*
* Wrapper function to adjust the signature of palloc to match what libz
* expects.
*/
static void *
gzip_palloc(void *opaque, unsigned items, unsigned size)
{
return palloc(items * size);
}
/*
* Wrapper function to adjust the signature of pfree to match what libz
* expects.
*/
static void
gzip_pfree(void *opaque, void *address)
{
pfree(address);
}
#endif

19
src/bin/pg_basebackup/pg_basebackup.c
View File

@ -1113,7 +1113,8 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
bbstreamer *streamer = NULL;
bbstreamer *manifest_inject_streamer = NULL;
bool inject_manifest;
bool is_tar;
bool is_tar,
is_tar_gz;
bool must_parse_archive;
int archive_name_len = strlen(archive_name);
@ -1128,6 +1129,10 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
is_tar = (archive_name_len > 4 &&
strcmp(archive_name + archive_name_len - 4, ".tar") == 0);
/* Is this a gzip archive? */
is_tar_gz = (archive_name_len > 8 &&
strcmp(archive_name + archive_name_len - 3, ".gz") == 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.
@ -1137,7 +1142,7 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
(spclocation == NULL && writerecoveryconf));
/* At present, we only know how to parse tar archives. */
if (must_parse_archive && !is_tar)
if (must_parse_archive && !is_tar && !is_tar_gz)
{
pg_log_error("unable to parse archive: %s", archive_name);
pg_log_info("only tar archives can be parsed");
@ -1194,7 +1199,6 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
compressloc != COMPRESS_LOCATION_CLIENT)
streamer = bbstreamer_plain_writer_new(archive_filename,
archive_file);
#ifdef HAVE_LIBZ
else if (compressmethod == COMPRESSION_GZIP)
{
strlcat(archive_filename, ".gz", sizeof(archive_filename));
@ -1202,7 +1206,6 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
archive_file,
compresslevel);
}
#endif
else
{
Assert(false); /* not reachable */
@ -1251,6 +1254,14 @@ CreateBackupStreamer(char *archive_name, char *spclocation,
else if (expect_unterminated_tarfile)
streamer = bbstreamer_tar_terminator_new(streamer);
/*
* If the user has requested a server compressed archive along with archive
* extraction at client then we need to decompress it.
*/
if (format == 'p' && compressmethod == COMPRESSION_GZIP &&
compressloc == COMPRESS_LOCATION_SERVER)
streamer = bbstreamer_gzip_decompressor_new(streamer);
/* Return the results. */
*manifest_inject_streamer_p = manifest_inject_streamer;
return streamer;

61
src/bin/pg_verifybackup/t/009_extract.pl Normal file
View File

@ -0,0 +1,61 @@
# Copyright (c) 2021-2022, PostgreSQL Global Development Group
# This test aims to verify that the client can decompress and extract
# a backup which was compressed by the server.
use strict;
use warnings;
use Cwd;
use Config;
use File::Path qw(rmtree);
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More tests => 4;
my $primary = PostgreSQL::Test::Cluster->new('primary');
$primary->init(allows_streaming => 1);
$primary->start;
my @test_configuration = (
{
'compression_method' => 'none',
'backup_flags' => [],
'enabled' => 1
},
{
'compression_method' => 'gzip',
'backup_flags' => ['--compress', 'server-gzip:5'],
'enabled' => check_pg_config("#define HAVE_LIBZ 1")
}
);
for my $tc (@test_configuration)
{
my $backup_path = $primary->backup_dir . '/' . 'extract_backup';
my $method = $tc->{'compression_method'};
SKIP: {
skip "$method compression not supported by this build", 3
if ! $tc->{'enabled'};
# Take backup with server compression enabled.
my @backup = (
'pg_basebackup', '-D', $backup_path,
'-Xfetch', '--no-sync', '-cfast', '-Fp');
push @backup, @{$tc->{'backup_flags'}};
my @verify = ('pg_verifybackup', '-e', $backup_path);
# A backup with a valid compression method should work.
$primary->command_ok(\@backup,
"backup done, compression method \"$method\"");
# Make sure that it verifies OK.
$primary->command_ok(\@verify,
"backup verified, compression method \"$method\"");
}
# Remove backup immediately to save disk space.
rmtree($backup_path);
}