database/postgres
database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-04-27 22:56:53 +03:00
Code

Remove WalCompressionMethod in favor of pg_compress_algorithm

Browse Source 
The same structure, with the same set of elements (for none, lz4, gzip
and zstd), exists in compression.h, so let's make use of the centralized
version instead of duplicating things.  Some of the variables used
previously for WalCompressionMethod are renamed to stick better with the
new structure and routine names.

WalCompressionMethod was leading to some confusion in walmethods.c, as
it was sometimes used to refer to some data unrelated to WAL.

Reported-by: Robert Haas
Author: Michael Paquier
Reviewed-by: Robert Haas, Georgios Kokolatos
Discussion: https://postgr.es/m/YlPQGNAAa04raObK@paquier.xyz
This commit is contained in:
Michael Paquier 2022-04-12 17:28:17 +09:00
parent ce4f46fdc8
commit 3603f7c6e6
5 changed files with 84 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -520,7 +520,7 @@ typedef struct
char xlog[MAXPGPATH]; /* directory or tarfile depending on mode */ char xlog[MAXPGPATH]; /* directory or tarfile depending on mode */
char *sysidentifier; char *sysidentifier;
int timeline; int timeline;
WalCompressionMethod wal_compress_method; pg_compress_algorithm wal_compress_algorithm;
int wal_compress_level; int wal_compress_level;
} logstreamer_param; } logstreamer_param;
@ -550,11 +550,11 @@ LogStreamerMain(logstreamer_param *param)
stream.replication_slot = replication_slot; stream.replication_slot = replication_slot;
if (format == 'p') if (format == 'p')
stream.walmethod = CreateWalDirectoryMethod(param->xlog, stream.walmethod = CreateWalDirectoryMethod(param->xlog,
COMPRESSION_NONE, 0, PG_COMPRESSION_NONE, 0,
stream.do_sync); stream.do_sync);
else else
stream.walmethod = CreateWalTarMethod(param->xlog, stream.walmethod = CreateWalTarMethod(param->xlog,
param->wal_compress_method, param->wal_compress_algorithm,
param->wal_compress_level, param->wal_compress_level,
stream.do_sync); stream.do_sync);
@ -602,7 +602,7 @@ LogStreamerMain(logstreamer_param *param)
*/ */
static void static void
StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier, StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier,
WalCompressionMethod wal_compress_method, pg_compress_algorithm wal_compress_algorithm,
int wal_compress_level) int wal_compress_level)
{ {
logstreamer_param *param; logstreamer_param *param;
@ -613,7 +613,7 @@ StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier,
param = pg_malloc0(sizeof(logstreamer_param)); param = pg_malloc0(sizeof(logstreamer_param));
param->timeline = timeline; param->timeline = timeline;
param->sysidentifier = sysidentifier; param->sysidentifier = sysidentifier;
param->wal_compress_method = wal_compress_method; param->wal_compress_algorithm = wal_compress_algorithm;
param->wal_compress_level = wal_compress_level; param->wal_compress_level = wal_compress_level;
/* Convert the starting position */ /* Convert the starting position */
@ -2019,7 +2019,7 @@ BaseBackup(char *compression_algorithm, char *compression_detail,
*/ */
if (includewal == STREAM_WAL) if (includewal == STREAM_WAL)
{ {
WalCompressionMethod wal_compress_method; pg_compress_algorithm wal_compress_algorithm;
int wal_compress_level; int wal_compress_level;
if (verbose) if (verbose)
@ -2027,19 +2027,19 @@ BaseBackup(char *compression_algorithm, char *compression_detail,
if (client_compress->algorithm == PG_COMPRESSION_GZIP) if (client_compress->algorithm == PG_COMPRESSION_GZIP)
{ {
wal_compress_method = COMPRESSION_GZIP; wal_compress_algorithm = PG_COMPRESSION_GZIP;
wal_compress_level = wal_compress_level =
(client_compress->options & PG_COMPRESSION_OPTION_LEVEL) (client_compress->options & PG_COMPRESSION_OPTION_LEVEL)
!= 0 ? client_compress->level : 0; != 0 ? client_compress->level : 0;
} }
else else
{ {
wal_compress_method = COMPRESSION_NONE; wal_compress_algorithm = PG_COMPRESSION_NONE;
wal_compress_level = 0; wal_compress_level = 0;
} }
StartLogStreamer(xlogstart, starttli, sysidentifier, StartLogStreamer(xlogstart, starttli, sysidentifier,
wal_compress_method, wal_compress_level); wal_compress_algorithm, wal_compress_level);
} }
if (serverMajor >= 1500) if (serverMajor >= 1500)

44
src/bin/pg_basebackup/pg_receivewal.c
View File

@ -52,7 +52,7 @@ static bool do_drop_slot = false;
static bool do_sync = true; static bool do_sync = true;
static bool synchronous = false; static bool synchronous = false;
static char *replication_slot = NULL; static char *replication_slot = NULL;
static WalCompressionMethod compression_method = COMPRESSION_NONE; static pg_compress_algorithm compression_algorithm = PG_COMPRESSION_NONE;
static XLogRecPtr endpos = InvalidXLogRecPtr; static XLogRecPtr endpos = InvalidXLogRecPtr;
@ -114,7 +114,7 @@ usage(void)
*/ */
static bool static bool
is_xlogfilename(const char *filename, bool *ispartial, is_xlogfilename(const char *filename, bool *ispartial,
WalCompressionMethod *wal_compression_method) pg_compress_algorithm *wal_compression_algorithm)
{ {
size_t fname_len = strlen(filename); size_t fname_len = strlen(filename);
size_t xlog_pattern_len = strspn(filename, "0123456789ABCDEF"); size_t xlog_pattern_len = strspn(filename, "0123456789ABCDEF");
@ -127,7 +127,7 @@ is_xlogfilename(const char *filename, bool *ispartial,
if (fname_len == XLOG_FNAME_LEN) if (fname_len == XLOG_FNAME_LEN)
{ {
*ispartial = false; *ispartial = false;
*wal_compression_method = COMPRESSION_NONE; *wal_compression_algorithm = PG_COMPRESSION_NONE;
return true; return true;
} }
@ -136,7 +136,7 @@ is_xlogfilename(const char *filename, bool *ispartial,
strcmp(filename + XLOG_FNAME_LEN, ".gz") == 0) strcmp(filename + XLOG_FNAME_LEN, ".gz") == 0)
{ {
*ispartial = false; *ispartial = false;
*wal_compression_method = COMPRESSION_GZIP; *wal_compression_algorithm = PG_COMPRESSION_GZIP;
return true; return true;
} }
@ -145,7 +145,7 @@ is_xlogfilename(const char *filename, bool *ispartial,
strcmp(filename + XLOG_FNAME_LEN, ".lz4") == 0) strcmp(filename + XLOG_FNAME_LEN, ".lz4") == 0)
{ {
*ispartial = false; *ispartial = false;
*wal_compression_method = COMPRESSION_LZ4; *wal_compression_algorithm = PG_COMPRESSION_LZ4;
return true; return true;
} }
@ -154,7 +154,7 @@ is_xlogfilename(const char *filename, bool *ispartial,
strcmp(filename + XLOG_FNAME_LEN, ".partial") == 0) strcmp(filename + XLOG_FNAME_LEN, ".partial") == 0)
{ {
*ispartial = true; *ispartial = true;
*wal_compression_method = COMPRESSION_NONE; *wal_compression_algorithm = PG_COMPRESSION_NONE;
return true; return true;
} }
@ -163,7 +163,7 @@ is_xlogfilename(const char *filename, bool *ispartial,
strcmp(filename + XLOG_FNAME_LEN, ".gz.partial") == 0) strcmp(filename + XLOG_FNAME_LEN, ".gz.partial") == 0)
{ {
*ispartial = true; *ispartial = true;
*wal_compression_method = COMPRESSION_GZIP; *wal_compression_algorithm = PG_COMPRESSION_GZIP;
return true; return true;
} }
@ -172,7 +172,7 @@ is_xlogfilename(const char *filename, bool *ispartial,
strcmp(filename + XLOG_FNAME_LEN, ".lz4.partial") == 0) strcmp(filename + XLOG_FNAME_LEN, ".lz4.partial") == 0)
{ {
*ispartial = true; *ispartial = true;
*wal_compression_method = COMPRESSION_LZ4; *wal_compression_algorithm = PG_COMPRESSION_LZ4;
return true; return true;
} }
@ -279,11 +279,11 @@ FindStreamingStart(uint32 *tli)
{ {
uint32 tli; uint32 tli;
XLogSegNo segno; XLogSegNo segno;
WalCompressionMethod wal_compression_method; pg_compress_algorithm wal_compression_algorithm;
bool ispartial; bool ispartial;
if (!is_xlogfilename(dirent->d_name, if (!is_xlogfilename(dirent->d_name,
&ispartial, &wal_compression_method)) &ispartial, &wal_compression_algorithm))
continue; continue;
/* /*
@ -309,7 +309,7 @@ FindStreamingStart(uint32 *tli)
* where WAL segments could have been compressed by a different source * where WAL segments could have been compressed by a different source
* than pg_receivewal, like an archive_command with lz4. * than pg_receivewal, like an archive_command with lz4.
*/ */
if (!ispartial && wal_compression_method == COMPRESSION_NONE) if (!ispartial && wal_compression_algorithm == PG_COMPRESSION_NONE)
{ {
struct stat statbuf; struct stat statbuf;
char fullpath[MAXPGPATH * 2]; char fullpath[MAXPGPATH * 2];
@ -325,7 +325,7 @@ FindStreamingStart(uint32 *tli)
continue; continue;
} }
} }
else if (!ispartial && wal_compression_method == COMPRESSION_GZIP) else if (!ispartial && wal_compression_algorithm == PG_COMPRESSION_GZIP)
{ {
int fd; int fd;
char buf[4]; char buf[4];
@ -364,7 +364,7 @@ FindStreamingStart(uint32 *tli)
continue; continue;
} }
} }
else if (!ispartial && wal_compression_method == COMPRESSION_LZ4) else if (!ispartial && wal_compression_algorithm == PG_COMPRESSION_LZ4)
{ {
#ifdef USE_LZ4 #ifdef USE_LZ4
#define LZ4_CHUNK_SZ 64 * 1024 /* 64kB as maximum chunk size read */ #define LZ4_CHUNK_SZ 64 * 1024 /* 64kB as maximum chunk size read */
@ -590,7 +590,7 @@ StreamLog(void)
stream.do_sync = do_sync; stream.do_sync = do_sync;
stream.mark_done = false; stream.mark_done = false;
stream.walmethod = CreateWalDirectoryMethod(basedir, stream.walmethod = CreateWalDirectoryMethod(basedir,
compression_method, compression_algorithm,
compresslevel, compresslevel,
stream.do_sync); stream.do_sync);
stream.partial_suffix = ".partial"; stream.partial_suffix = ".partial";
@ -750,11 +750,11 @@ main(int argc, char **argv)
break; break;
case 6: case 6:
if (pg_strcasecmp(optarg, "gzip") == 0) if (pg_strcasecmp(optarg, "gzip") == 0)
compression_method = COMPRESSION_GZIP; compression_algorithm = PG_COMPRESSION_GZIP;
else if (pg_strcasecmp(optarg, "lz4") == 0) else if (pg_strcasecmp(optarg, "lz4") == 0)
compression_method = COMPRESSION_LZ4; compression_algorithm = PG_COMPRESSION_LZ4;
else if (pg_strcasecmp(optarg, "none") == 0) else if (pg_strcasecmp(optarg, "none") == 0)
compression_method = COMPRESSION_NONE; compression_algorithm = PG_COMPRESSION_NONE;
else else
pg_fatal("invalid value \"%s\" for option %s", pg_fatal("invalid value \"%s\" for option %s",
optarg, "--compression-method"); optarg, "--compression-method");
@ -814,9 +814,9 @@ main(int argc, char **argv)
/* /*
* Compression-related options. * Compression-related options.
*/ */
switch (compression_method) switch (compression_algorithm)
{ {
case COMPRESSION_NONE: case PG_COMPRESSION_NONE:
if (compresslevel != 0) if (compresslevel != 0)
{ {
pg_log_error("cannot use --compress with --compression-method=%s", pg_log_error("cannot use --compress with --compression-method=%s",
@ -825,7 +825,7 @@ main(int argc, char **argv)
exit(1); exit(1);
} }
break; break;
case COMPRESSION_GZIP: case PG_COMPRESSION_GZIP:
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (compresslevel == 0) if (compresslevel == 0)
{ {
@ -837,7 +837,7 @@ main(int argc, char **argv)
"gzip"); "gzip");
#endif #endif
break; break;
case COMPRESSION_LZ4: case PG_COMPRESSION_LZ4:
#ifdef USE_LZ4 #ifdef USE_LZ4
if (compresslevel != 0) if (compresslevel != 0)
{ {
@ -851,7 +851,7 @@ main(int argc, char **argv)
"LZ4"); "LZ4");
#endif #endif
break; break;
case COMPRESSION_ZSTD: case PG_COMPRESSION_ZSTD:
pg_fatal("compression with %s is not yet supported", "ZSTD"); pg_fatal("compression with %s is not yet supported", "ZSTD");
break; break;
} }

2
src/bin/pg_basebackup/receivelog.c
View File

@ -114,7 +114,7 @@ open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
* When streaming to tar, no file with this name will exist before, so we * When streaming to tar, no file with this name will exist before, so we
* never have to verify a size. * never have to verify a size.
*/ */
if (stream->walmethod->compression_method() == COMPRESSION_NONE && if (stream->walmethod->compression_algorithm() == PG_COMPRESSION_NONE &&
stream->walmethod->existsfile(fn)) stream->walmethod->existsfile(fn))
{ {
size = stream->walmethod->get_file_size(fn); size = stream->walmethod->get_file_size(fn);

96
src/bin/pg_basebackup/walmethods.c
View File

@ -49,7 +49,7 @@
typedef struct DirectoryMethodData typedef struct DirectoryMethodData
{ {
char *basedir; char *basedir;
WalCompressionMethod compression_method; pg_compress_algorithm compression_algorithm;
int compression_level; int compression_level;
bool sync; bool sync;
const char *lasterrstring; /* if set, takes precedence over lasterrno */ const char *lasterrstring; /* if set, takes precedence over lasterrno */
@ -97,8 +97,8 @@ dir_get_file_name(const char *pathname, const char *temp_suffix)
snprintf(filename, MAXPGPATH, "%s%s%s", snprintf(filename, MAXPGPATH, "%s%s%s",
pathname, pathname,
dir_data->compression_method == COMPRESSION_GZIP ? ".gz" : dir_data->compression_algorithm == PG_COMPRESSION_GZIP ? ".gz" :
dir_data->compression_method == COMPRESSION_LZ4 ? ".lz4" : "", dir_data->compression_algorithm == PG_COMPRESSION_LZ4 ? ".lz4" : "",
temp_suffix ? temp_suffix : ""); temp_suffix ? temp_suffix : "");
return filename; return filename;
@ -141,7 +141,7 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
} }
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (dir_data->compression_method == COMPRESSION_GZIP) if (dir_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
gzfp = gzdopen(fd, "wb"); gzfp = gzdopen(fd, "wb");
if (gzfp == NULL) if (gzfp == NULL)
@ -161,7 +161,7 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
} }
#endif #endif
#ifdef USE_LZ4 #ifdef USE_LZ4
if (dir_data->compression_method == COMPRESSION_LZ4) if (dir_data->compression_algorithm == PG_COMPRESSION_LZ4)
{ {
size_t ctx_out; size_t ctx_out;
size_t header_size; size_t header_size;
@ -202,7 +202,7 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
#endif #endif
/* Do pre-padding on non-compressed files */ /* Do pre-padding on non-compressed files */
if (pad_to_size && dir_data->compression_method == COMPRESSION_NONE) if (pad_to_size && dir_data->compression_algorithm == PG_COMPRESSION_NONE)
{ {
PGAlignedXLogBlock zerobuf; PGAlignedXLogBlock zerobuf;
int bytes; int bytes;
@ -241,12 +241,12 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
{ {
dir_data->lasterrno = errno; dir_data->lasterrno = errno;
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (dir_data->compression_method == COMPRESSION_GZIP) if (dir_data->compression_algorithm == PG_COMPRESSION_GZIP)
gzclose(gzfp); gzclose(gzfp);
else else
#endif #endif
#ifdef USE_LZ4 #ifdef USE_LZ4
if (dir_data->compression_method == COMPRESSION_LZ4) if (dir_data->compression_algorithm == PG_COMPRESSION_LZ4)
{ {
(void) LZ4F_compressEnd(ctx, lz4buf, lz4bufsize, NULL); (void) LZ4F_compressEnd(ctx, lz4buf, lz4bufsize, NULL);
(void) LZ4F_freeCompressionContext(ctx); (void) LZ4F_freeCompressionContext(ctx);
@ -262,11 +262,11 @@ dir_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
f = pg_malloc0(sizeof(DirectoryMethodFile)); f = pg_malloc0(sizeof(DirectoryMethodFile));
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (dir_data->compression_method == COMPRESSION_GZIP) if (dir_data->compression_algorithm == PG_COMPRESSION_GZIP)
f->gzfp = gzfp; f->gzfp = gzfp;
#endif #endif
#ifdef USE_LZ4 #ifdef USE_LZ4
if (dir_data->compression_method == COMPRESSION_LZ4) if (dir_data->compression_algorithm == PG_COMPRESSION_LZ4)
{ {
f->ctx = ctx; f->ctx = ctx;
f->lz4buf = lz4buf; f->lz4buf = lz4buf;
@ -294,7 +294,7 @@ dir_write(Walfile f, const void *buf, size_t count)
dir_clear_error(); dir_clear_error();
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (dir_data->compression_method == COMPRESSION_GZIP) if (dir_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
errno = 0; errno = 0;
r = (ssize_t) gzwrite(df->gzfp, buf, count); r = (ssize_t) gzwrite(df->gzfp, buf, count);
@ -307,7 +307,7 @@ dir_write(Walfile f, const void *buf, size_t count)
else else
#endif #endif
#ifdef USE_LZ4 #ifdef USE_LZ4
if (dir_data->compression_method == COMPRESSION_LZ4) if (dir_data->compression_algorithm == PG_COMPRESSION_LZ4)
{ {
size_t chunk; size_t chunk;
size_t remaining; size_t remaining;
@ -387,7 +387,7 @@ dir_close(Walfile f, WalCloseMethod method)
dir_clear_error(); dir_clear_error();
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (dir_data->compression_method == COMPRESSION_GZIP) if (dir_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
errno = 0; /* in case gzclose() doesn't set it */ errno = 0; /* in case gzclose() doesn't set it */
r = gzclose(df->gzfp); r = gzclose(df->gzfp);
@ -395,7 +395,7 @@ dir_close(Walfile f, WalCloseMethod method)
else else
#endif #endif
#ifdef USE_LZ4 #ifdef USE_LZ4
if (dir_data->compression_method == COMPRESSION_LZ4) if (dir_data->compression_algorithm == PG_COMPRESSION_LZ4)
{ {
size_t compressed; size_t compressed;
@ -514,7 +514,7 @@ dir_sync(Walfile f)
return 0; return 0;
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (dir_data->compression_method == COMPRESSION_GZIP) if (dir_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
if (gzflush(((DirectoryMethodFile *) f)->gzfp, Z_SYNC_FLUSH) != Z_OK) if (gzflush(((DirectoryMethodFile *) f)->gzfp, Z_SYNC_FLUSH) != Z_OK)
{ {
@ -524,7 +524,7 @@ dir_sync(Walfile f)
} }
#endif #endif
#ifdef USE_LZ4 #ifdef USE_LZ4
if (dir_data->compression_method == COMPRESSION_LZ4) if (dir_data->compression_algorithm == PG_COMPRESSION_LZ4)
{ {
DirectoryMethodFile *df = (DirectoryMethodFile *) f; DirectoryMethodFile *df = (DirectoryMethodFile *) f;
size_t compressed; size_t compressed;
@ -571,10 +571,10 @@ dir_get_file_size(const char *pathname)
return statbuf.st_size; return statbuf.st_size;
} }
static WalCompressionMethod static pg_compress_algorithm
dir_compression_method(void) dir_compression_algorithm(void)
{ {
return dir_data->compression_method; return dir_data->compression_algorithm;
} }
static bool static bool
@ -618,7 +618,7 @@ dir_finish(void)
WalWriteMethod * WalWriteMethod *
CreateWalDirectoryMethod(const char *basedir, CreateWalDirectoryMethod(const char *basedir,
WalCompressionMethod compression_method, pg_compress_algorithm compression_algorithm,
int compression_level, bool sync) int compression_level, bool sync)
{ {
WalWriteMethod *method; WalWriteMethod *method;
@ -629,7 +629,7 @@ CreateWalDirectoryMethod(const char *basedir,
method->get_current_pos = dir_get_current_pos; method->get_current_pos = dir_get_current_pos;
method->get_file_size = dir_get_file_size; method->get_file_size = dir_get_file_size;
method->get_file_name = dir_get_file_name; method->get_file_name = dir_get_file_name;
method->compression_method = dir_compression_method; method->compression_algorithm = dir_compression_algorithm;
method->close = dir_close; method->close = dir_close;
method->sync = dir_sync; method->sync = dir_sync;
method->existsfile = dir_existsfile; method->existsfile = dir_existsfile;
@ -637,7 +637,7 @@ CreateWalDirectoryMethod(const char *basedir,
method->getlasterror = dir_getlasterror; method->getlasterror = dir_getlasterror;
dir_data = pg_malloc0(sizeof(DirectoryMethodData)); dir_data = pg_malloc0(sizeof(DirectoryMethodData));
dir_data->compression_method = compression_method; dir_data->compression_algorithm = compression_algorithm;
dir_data->compression_level = compression_level; dir_data->compression_level = compression_level;
dir_data->basedir = pg_strdup(basedir); dir_data->basedir = pg_strdup(basedir);
dir_data->sync = sync; dir_data->sync = sync;
@ -672,7 +672,7 @@ typedef struct TarMethodData
{ {
char *tarfilename; char *tarfilename;
int fd; int fd;
WalCompressionMethod compression_method; pg_compress_algorithm compression_algorithm;
int compression_level; int compression_level;
bool sync; bool sync;
TarMethodFile *currentfile; TarMethodFile *currentfile;
@ -759,7 +759,7 @@ tar_write(Walfile f, const void *buf, size_t count)
tar_clear_error(); tar_clear_error();
/* Tarfile will always be positioned at the end */ /* Tarfile will always be positioned at the end */
if (tar_data->compression_method == COMPRESSION_NONE) if (tar_data->compression_algorithm == PG_COMPRESSION_NONE)
{ {
errno = 0; errno = 0;
r = write(tar_data->fd, buf, count); r = write(tar_data->fd, buf, count);
@ -773,7 +773,7 @@ tar_write(Walfile f, const void *buf, size_t count)
return r; return r;
} }
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
else if (tar_data->compression_method == COMPRESSION_GZIP) else if (tar_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
if (!tar_write_compressed_data(unconstify(void *, buf), count, false)) if (!tar_write_compressed_data(unconstify(void *, buf), count, false))
return -1; return -1;
@ -842,7 +842,7 @@ tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
} }
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (tar_data->compression_method == COMPRESSION_GZIP) if (tar_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
tar_data->zp = (z_streamp) pg_malloc(sizeof(z_stream)); tar_data->zp = (z_streamp) pg_malloc(sizeof(z_stream));
tar_data->zp->zalloc = Z_NULL; tar_data->zp->zalloc = Z_NULL;
@ -893,7 +893,7 @@ tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
pg_free(tmppath); pg_free(tmppath);
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (tar_data->compression_method == COMPRESSION_GZIP) if (tar_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
/* Flush existing data */ /* Flush existing data */
if (!tar_write_compressed_data(NULL, 0, true)) if (!tar_write_compressed_data(NULL, 0, true))
@ -918,7 +918,7 @@ tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
} }
tar_data->currentfile->currpos = 0; tar_data->currentfile->currpos = 0;
if (tar_data->compression_method == COMPRESSION_NONE) if (tar_data->compression_algorithm == PG_COMPRESSION_NONE)
{ {
errno = 0; errno = 0;
if (write(tar_data->fd, tar_data->currentfile->header, if (write(tar_data->fd, tar_data->currentfile->header,
@ -932,7 +932,7 @@ tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
} }
} }
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
else if (tar_data->compression_method == COMPRESSION_GZIP) else if (tar_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
/* Write header through the zlib APIs but with no compression */ /* Write header through the zlib APIs but with no compression */
if (!tar_write_compressed_data(tar_data->currentfile->header, if (!tar_write_compressed_data(tar_data->currentfile->header,
@ -962,7 +962,7 @@ tar_open_for_write(const char *pathname, const char *temp_suffix, size_t pad_to_
if (pad_to_size) if (pad_to_size)
{ {
tar_data->currentfile->pad_to_size = pad_to_size; tar_data->currentfile->pad_to_size = pad_to_size;
if (tar_data->compression_method == COMPRESSION_NONE) if (tar_data->compression_algorithm == PG_COMPRESSION_NONE)
{ {
/* Uncompressed, so pad now */ /* Uncompressed, so pad now */
if (!tar_write_padding_data(tar_data->currentfile, pad_to_size)) if (!tar_write_padding_data(tar_data->currentfile, pad_to_size))
@ -993,10 +993,10 @@ tar_get_file_size(const char *pathname)
return -1; return -1;
} }
static WalCompressionMethod static pg_compress_algorithm
tar_compression_method(void) tar_compression_algorithm(void)
{ {
return tar_data->compression_method; return tar_data->compression_algorithm;
} }
static off_t static off_t
@ -1023,7 +1023,7 @@ tar_sync(Walfile f)
* Always sync the whole tarfile, because that's all we can do. This makes * Always sync the whole tarfile, because that's all we can do. This makes
* no sense on compressed files, so just ignore those. * no sense on compressed files, so just ignore those.
*/ */
if (tar_data->compression_method != COMPRESSION_NONE) if (tar_data->compression_algorithm != PG_COMPRESSION_NONE)
return 0; return 0;
r = fsync(tar_data->fd); r = fsync(tar_data->fd);
@ -1044,7 +1044,7 @@ tar_close(Walfile f, WalCloseMethod method)
if (method == CLOSE_UNLINK) if (method == CLOSE_UNLINK)
{ {
if (tar_data->compression_method != COMPRESSION_NONE) if (tar_data->compression_algorithm != PG_COMPRESSION_NONE)
{ {
tar_set_error("unlink not supported with compression"); tar_set_error("unlink not supported with compression");
return -1; return -1;
@ -1075,7 +1075,7 @@ tar_close(Walfile f, WalCloseMethod method)
*/ */
if (tf->pad_to_size) if (tf->pad_to_size)
{ {
if (tar_data->compression_method == COMPRESSION_GZIP) if (tar_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
/* /*
* A compressed tarfile is padded on close since we cannot know * A compressed tarfile is padded on close since we cannot know
@ -1116,7 +1116,7 @@ tar_close(Walfile f, WalCloseMethod method)
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (tar_data->compression_method == COMPRESSION_GZIP) if (tar_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
/* Flush the current buffer */ /* Flush the current buffer */
if (!tar_write_compressed_data(NULL, 0, true)) if (!tar_write_compressed_data(NULL, 0, true))
@ -1145,7 +1145,7 @@ tar_close(Walfile f, WalCloseMethod method)
tar_data->lasterrno = errno; tar_data->lasterrno = errno;
return -1; return -1;
} }
if (tar_data->compression_method == COMPRESSION_NONE) if (tar_data->compression_algorithm == PG_COMPRESSION_NONE)
{ {
errno = 0; errno = 0;
if (write(tar_data->fd, tf->header, TAR_BLOCK_SIZE) != TAR_BLOCK_SIZE) if (write(tar_data->fd, tf->header, TAR_BLOCK_SIZE) != TAR_BLOCK_SIZE)
@ -1156,7 +1156,7 @@ tar_close(Walfile f, WalCloseMethod method)
} }
} }
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
else if (tar_data->compression_method == COMPRESSION_GZIP) else if (tar_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
/* Turn off compression */ /* Turn off compression */
if (deflateParams(tar_data->zp, 0, 0) != Z_OK) if (deflateParams(tar_data->zp, 0, 0) != Z_OK)
@ -1230,7 +1230,7 @@ tar_finish(void)
/* A tarfile always ends with two empty blocks */ /* A tarfile always ends with two empty blocks */
MemSet(zerobuf, 0, sizeof(zerobuf)); MemSet(zerobuf, 0, sizeof(zerobuf));
if (tar_data->compression_method == COMPRESSION_NONE) if (tar_data->compression_algorithm == PG_COMPRESSION_NONE)
{ {
errno = 0; errno = 0;
if (write(tar_data->fd, zerobuf, sizeof(zerobuf)) != sizeof(zerobuf)) if (write(tar_data->fd, zerobuf, sizeof(zerobuf)) != sizeof(zerobuf))
@ -1241,7 +1241,7 @@ tar_finish(void)
} }
} }
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
else if (tar_data->compression_method == COMPRESSION_GZIP) else if (tar_data->compression_algorithm == PG_COMPRESSION_GZIP)
{ {
if (!tar_write_compressed_data(zerobuf, sizeof(zerobuf), false)) if (!tar_write_compressed_data(zerobuf, sizeof(zerobuf), false))
return false; return false;
@ -1324,18 +1324,18 @@ tar_finish(void)
} }
/* /*
* The argument compression_method is currently ignored. It is in place for * The argument compression_algorithm is currently ignored. It is in place for
* symmetry with CreateWalDirectoryMethod which uses it for distinguishing * symmetry with CreateWalDirectoryMethod which uses it for distinguishing
* between the different compression methods. CreateWalTarMethod and its family * between the different compression methods. CreateWalTarMethod and its family
* of functions handle only zlib compression. * of functions handle only zlib compression.
*/ */
WalWriteMethod * WalWriteMethod *
CreateWalTarMethod(const char *tarbase, CreateWalTarMethod(const char *tarbase,
WalCompressionMethod compression_method, pg_compress_algorithm compression_algorithm,
int compression_level, bool sync) int compression_level, bool sync)
{ {
WalWriteMethod *method; WalWriteMethod *method;
const char *suffix = (compression_method == COMPRESSION_GZIP) ? const char *suffix = (compression_algorithm == PG_COMPRESSION_GZIP) ?
".tar.gz" : ".tar"; ".tar.gz" : ".tar";
method = pg_malloc0(sizeof(WalWriteMethod)); method = pg_malloc0(sizeof(WalWriteMethod));
@ -1344,7 +1344,7 @@ CreateWalTarMethod(const char *tarbase,
method->get_current_pos = tar_get_current_pos; method->get_current_pos = tar_get_current_pos;
method->get_file_size = tar_get_file_size; method->get_file_size = tar_get_file_size;
method->get_file_name = tar_get_file_name; method->get_file_name = tar_get_file_name;
method->compression_method = tar_compression_method; method->compression_algorithm = tar_compression_algorithm;
method->close = tar_close; method->close = tar_close;
method->sync = tar_sync; method->sync = tar_sync;
method->existsfile = tar_existsfile; method->existsfile = tar_existsfile;
@ -1355,11 +1355,11 @@ CreateWalTarMethod(const char *tarbase,
tar_data->tarfilename = pg_malloc0(strlen(tarbase) + strlen(suffix) + 1); tar_data->tarfilename = pg_malloc0(strlen(tarbase) + strlen(suffix) + 1);
sprintf(tar_data->tarfilename, "%s%s", tarbase, suffix); sprintf(tar_data->tarfilename, "%s%s", tarbase, suffix);
tar_data->fd = -1; tar_data->fd = -1;
tar_data->compression_method = compression_method; tar_data->compression_algorithm = compression_algorithm;
tar_data->compression_level = compression_level; tar_data->compression_level = compression_level;
tar_data->sync = sync; tar_data->sync = sync;
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (compression_method == COMPRESSION_GZIP) if (compression_algorithm == PG_COMPRESSION_GZIP)
tar_data->zlibOut = (char *) pg_malloc(ZLIB_OUT_SIZE + 1); tar_data->zlibOut = (char *) pg_malloc(ZLIB_OUT_SIZE + 1);
#endif #endif
@ -1371,7 +1371,7 @@ FreeWalTarMethod(void)
{ {
pg_free(tar_data->tarfilename); pg_free(tar_data->tarfilename);
#ifdef HAVE_LIBZ #ifdef HAVE_LIBZ
if (tar_data->compression_method == COMPRESSION_GZIP) if (tar_data->compression_algorithm == PG_COMPRESSION_GZIP)
pg_free(tar_data->zlibOut); pg_free(tar_data->zlibOut);
#endif #endif
pg_free(tar_data); pg_free(tar_data);

16
src/bin/pg_basebackup/walmethods.h
View File

@ -9,6 +9,7 @@
*------------------------------------------------------------------------- *-------------------------------------------------------------------------
*/ */
#include "common/compression.h"
typedef void *Walfile; typedef void *Walfile;
@ -19,15 +20,6 @@ typedef enum
CLOSE_NO_RENAME CLOSE_NO_RENAME
} WalCloseMethod; } WalCloseMethod;
/* Types of compression supported */
typedef enum
{
COMPRESSION_GZIP,
COMPRESSION_LZ4,
COMPRESSION_ZSTD,
COMPRESSION_NONE
} WalCompressionMethod;
/* /*
* A WalWriteMethod structure represents the different methods used * A WalWriteMethod structure represents the different methods used
* to write the streaming WAL as it's received. * to write the streaming WAL as it's received.
@ -68,7 +60,7 @@ struct WalWriteMethod
char *(*get_file_name) (const char *pathname, const char *temp_suffix); char *(*get_file_name) (const char *pathname, const char *temp_suffix);
/* Returns the compression method */ /* Returns the compression method */
WalCompressionMethod (*compression_method) (void); pg_compress_algorithm (*compression_algorithm) (void);
/* /*
* Write count number of bytes to the file, and return the number of bytes * Write count number of bytes to the file, and return the number of bytes
@ -104,10 +96,10 @@ struct WalWriteMethod
* not all those required for pg_receivewal) * not all those required for pg_receivewal)
*/ */
WalWriteMethod *CreateWalDirectoryMethod(const char *basedir, WalWriteMethod *CreateWalDirectoryMethod(const char *basedir,
WalCompressionMethod compression_method, pg_compress_algorithm compression_algo,
int compression, bool sync); int compression, bool sync);
WalWriteMethod *CreateWalTarMethod(const char *tarbase, WalWriteMethod *CreateWalTarMethod(const char *tarbase,
WalCompressionMethod compression_method, pg_compress_algorithm compression_algo,
int compression, bool sync); int compression, bool sync);
/* Cleanup routines for previously-created methods */ /* Cleanup routines for previously-created methods */