</listitem>
</varlistentry>
+ <varlistentry>
+ <term><option>--compression-method=<replaceable class="parameter">level</replaceable></option></term>
+ <listitem>
+ <para>
+ Enables compression of write-ahead logs using the specified method.
+ Supported values <literal>gzip</literal>, and
+ <literal>none</literal>.
+ </para>
+
+ <para>
+ The suffix <filename>.gz</filename> will automatically be added to
+ all filenames when using <literal>gzip</literal>
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term><option>-Z <replaceable class="parameter">level</replaceable></option></term>
<term><option>--compress=<replaceable class="parameter">level</replaceable></option></term>
<listitem>
<para>
- Enables gzip compression of write-ahead logs, and specifies the
- compression level (0 through 9, 0 being no compression and 9 being best
- compression). The suffix <filename>.gz</filename> will
- automatically be added to all filenames.
+ Specifies the compression level (<literal>1</literal> through
+ <literal>9</literal>, <literal>1</literal> being worst compression
+ and <literal>9</literal> being best compression) for WAL segments
+ compressed with <application>gzip</application>.
+ </para>
+
+ <para>
+ This option requires <option>--compression-method</option> to be
+ specified with <literal>gzip</literal>.
</para>
</listitem>
</varlistentry>
stream.replication_slot = replication_slot;
if (format == 'p')
- stream.walmethod = CreateWalDirectoryMethod(param->xlog, 0,
+ stream.walmethod = CreateWalDirectoryMethod(param->xlog,
+ COMPRESSION_NONE, 0,
stream.do_sync);
else
- stream.walmethod = CreateWalTarMethod(param->xlog, compresslevel,
+ stream.walmethod = CreateWalTarMethod(param->xlog,
+ COMPRESSION_NONE, /* ignored */
+ compresslevel,
stream.do_sync);
if (!ReceiveXlogStream(param->bgconn, &stream))
#include <signal.h>
#include <sys/stat.h>
#include <unistd.h>
+#ifdef HAVE_LIBZ
+#include <zlib.h>
+#endif
#include "access/xlog_internal.h"
#include "common/file_perm.h"
static bool do_sync = true;
static bool synchronous = false;
static char *replication_slot = NULL;
+static WalCompressionMethod compression_method = COMPRESSION_NONE;
static XLogRecPtr endpos = InvalidXLogRecPtr;
PQfinish(conn);
}
-/* Routines to evaluate segment file format */
-#define IsCompressXLogFileName(fname) \
- (strlen(fname) == XLOG_FNAME_LEN + strlen(".gz") && \
- strspn(fname, "0123456789ABCDEF") == XLOG_FNAME_LEN && \
- strcmp((fname) + XLOG_FNAME_LEN, ".gz") == 0)
-#define IsPartialCompressXLogFileName(fname) \
- (strlen(fname) == XLOG_FNAME_LEN + strlen(".gz.partial") && \
- strspn(fname, "0123456789ABCDEF") == XLOG_FNAME_LEN && \
- strcmp((fname) + XLOG_FNAME_LEN, ".gz.partial") == 0)
-
static void
usage(void)
{
printf(_(" --synchronous flush write-ahead log immediately after writing\n"));
printf(_(" -v, --verbose output verbose messages\n"));
printf(_(" -V, --version output version information, then exit\n"));
- printf(_(" -Z, --compress=0-9 compress logs with given compression level\n"));
+ printf(_(" --compression-method=METHOD\n"
+ " method to compress logs\n"));
+ printf(_(" -Z, --compress=1-9 compress logs with given compression level\n"));
printf(_(" -?, --help show this help, then exit\n"));
printf(_("\nConnection options:\n"));
printf(_(" -d, --dbname=CONNSTR connection string\n"));
printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
}
+/*
+ * Check if the filename looks like a WAL file, letting caller know if this
+ * WAL segment is partial and/or compressed.
+ */
+static bool
+is_xlogfilename(const char *filename, bool *ispartial,
+ WalCompressionMethod *wal_compression_method)
+{
+ size_t fname_len = strlen(filename);
+ size_t xlog_pattern_len = strspn(filename, "0123456789ABCDEF");
+
+ /* File does not look like a WAL file */
+ if (xlog_pattern_len != XLOG_FNAME_LEN)
+ return false;
+
+ /* File looks like a completed uncompressed WAL file */
+ if (fname_len == XLOG_FNAME_LEN)
+ {
+ *ispartial = false;
+ *wal_compression_method = COMPRESSION_NONE;
+ return true;
+ }
+
+ /* File looks like a completed gzip-compressed WAL file */
+ if (fname_len == XLOG_FNAME_LEN + strlen(".gz") &&
+ strcmp(filename + XLOG_FNAME_LEN, ".gz") == 0)
+ {
+ *ispartial = false;
+ *wal_compression_method = COMPRESSION_GZIP;
+ return true;
+ }
+
+ /* File looks like a partial uncompressed WAL file */
+ if (fname_len == XLOG_FNAME_LEN + strlen(".partial") &&
+ strcmp(filename + XLOG_FNAME_LEN, ".partial") == 0)
+ {
+ *ispartial = true;
+ *wal_compression_method = COMPRESSION_NONE;
+ return true;
+ }
+
+ /* File looks like a partial gzip-compressed WAL file */
+ if (fname_len == XLOG_FNAME_LEN + strlen(".gz.partial") &&
+ strcmp(filename + XLOG_FNAME_LEN, ".gz.partial") == 0)
+ {
+ *ispartial = true;
+ *wal_compression_method = COMPRESSION_GZIP;
+ return true;
+ }
+
+ /* File does not look like something we know */
+ return false;
+}
+
static bool
stop_streaming(XLogRecPtr xlogpos, uint32 timeline, bool segment_finished)
{
{
uint32 tli;
XLogSegNo segno;
+ WalCompressionMethod wal_compression_method;
bool ispartial;
- bool iscompress;
- /*
- * Check if the filename looks like an xlog file, or a .partial file.
- */
- if (IsXLogFileName(dirent->d_name))
- {
- ispartial = false;
- iscompress = false;
- }
- else if (IsPartialXLogFileName(dirent->d_name))
- {
- ispartial = true;
- iscompress = false;
- }
- else if (IsCompressXLogFileName(dirent->d_name))
- {
- ispartial = false;
- iscompress = true;
- }
- else if (IsPartialCompressXLogFileName(dirent->d_name))
- {
- ispartial = true;
- iscompress = true;
- }
- else
+ if (!is_xlogfilename(dirent->d_name,
+ &ispartial, &wal_compression_method))
continue;
/*
/*
* Check that the segment has the right size, if it's supposed to be
* completed. For non-compressed segments just check the on-disk size
- * and see if it matches a completed segment. For compressed segments,
- * look at the last 4 bytes of the compressed file, which is where the
- * uncompressed size is located for gz files with a size lower than
- * 4GB, and then compare it to the size of a completed segment. The 4
- * last bytes correspond to the ISIZE member according to
+ * and see if it matches a completed segment. For gzip-compressed
+ * segments, look at the last 4 bytes of the compressed file, which is
+ * where the uncompressed size is located for files with a size lower
+ * than 4GB, and then compare it to the size of a completed segment.
+ * The 4 last bytes correspond to the ISIZE member according to
* http://www.zlib.org/rfc-gzip.html.
*/
- if (!ispartial && !iscompress)
+ if (!ispartial && wal_compression_method == COMPRESSION_NONE)
{
struct stat statbuf;
char fullpath[MAXPGPATH * 2];
continue;
}
}
- else if (!ispartial && iscompress)
+ else if (!ispartial && wal_compression_method == COMPRESSION_GZIP)
{
int fd;
char buf[4];
stream.synchronous = synchronous;
stream.do_sync = do_sync;
stream.mark_done = false;
- stream.walmethod = CreateWalDirectoryMethod(basedir, compresslevel,
+ stream.walmethod = CreateWalDirectoryMethod(basedir,
+ compression_method,
+ compresslevel,
stream.do_sync);
stream.partial_suffix = ".partial";
stream.replication_slot = replication_slot;
{"status-interval", required_argument, NULL, 's'},
{"slot", required_argument, NULL, 'S'},
{"verbose", no_argument, NULL, 'v'},
+ {"compression-method", required_argument, NULL, 'I'},
{"compress", required_argument, NULL, 'Z'},
/* action */
{"create-slot", no_argument, NULL, 1},
case 'v':
verbose++;
break;
+ case 'I':
+ if (pg_strcasecmp(optarg, "gzip") == 0)
+ compression_method = COMPRESSION_GZIP;
+ else if (pg_strcasecmp(optarg, "none") == 0)
+ compression_method = COMPRESSION_NONE;
+ else
+ {
+ pg_log_error("invalid value \"%s\" for option %s",
+ optarg, "--compress-method");
+ exit(1);
+ }
+ break;
case 'Z':
- if (!option_parse_int(optarg, "-Z/--compress", 0, 9,
+ if (!option_parse_int(optarg, "-Z/--compress", 1, 9,
&compresslevel))
exit(1);
break;
exit(1);
}
-#ifndef HAVE_LIBZ
- if (compresslevel != 0)
+
+ /*
+ * Compression-related options.
+ */
+ switch (compression_method)
{
- pg_log_error("this build does not support compression");
- exit(1);
- }
+ case COMPRESSION_NONE:
+ if (compresslevel != 0)
+ {
+ pg_log_error("cannot use --compress with --compression-method=%s",
+ "none");
+ fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+ progname);
+ exit(1);
+ }
+ break;
+ case COMPRESSION_GZIP:
+#ifdef HAVE_LIBZ
+ if (compresslevel == 0)
+ {
+ pg_log_info("no value specified for --compress, switching to default");
+ compresslevel = Z_DEFAULT_COMPRESSION;
+ }
+#else
+ pg_log_error("this build does not support compression with %s",
+ "gzip");
+ exit(1);
#endif
+ break;
+ }
+
/*
* Check existence of destination folder.
* When streaming to tar, no file with this name will exist before, so we
* never have to verify a size.
*/
- if (stream->walmethod->compression() == 0 &&
+ if (stream->walmethod->compression_method() == COMPRESSION_NONE &&
stream->walmethod->existsfile(fn))
{
size = stream->walmethod->get_file_size(fn);
use warnings;
use PostgreSQL::Test::Utils;
use PostgreSQL::Test::Cluster;
-use Test::More tests => 35;
+use Test::More tests => 37;
program_help_ok('pg_receivewal');
program_version_ok('pg_receivewal');
$primary->command_fails(
[ 'pg_receivewal', '-D', $stream_dir, '--synchronous', '--no-sync' ],
'failure if --synchronous specified with --no-sync');
+$primary->command_fails_like(
+ [
+ 'pg_receivewal', '-D', $stream_dir, '--compression-method', 'none',
+ '--compress', '1'
+ ],
+ qr/\Qpg_receivewal: error: cannot use --compress with --compression-method=none/,
+ 'failure if --compress spwcified with --compression-method=none');
# Slot creation and drop
my $slot_name = 'test';
# a valid value.
$primary->command_ok(
[
- 'pg_receivewal', '-D', $stream_dir, '--verbose',
- '--endpos', $nextlsn, '--compress', '1 ',
+ 'pg_receivewal', '-D',
+ $stream_dir, '--verbose',
+ '--endpos', $nextlsn,
+ '--compression-method', 'gzip',
+ '--compress', '1 ',
'--no-loop'
],
"streaming some WAL using ZLIB compression");
typedef struct DirectoryMethodData
{
char *basedir;
- int compression;
+ WalCompressionMethod compression_method;
+ int compression_level;
bool sync;
} DirectoryMethodData;
static DirectoryMethodData *dir_data = NULL;
char *filename = pg_malloc0(MAXPGPATH * sizeof(char));
snprintf(filename, MAXPGPATH, "%s%s%s",
- pathname, dir_data->compression > 0 ? ".gz" : "",
+ pathname,
+ dir_data->compression_method == COMPRESSION_GZIP ? ".gz" : "",
temp_suffix ? temp_suffix : "");
return filename;
return NULL;
#ifdef HAVE_LIBZ
- if (dir_data->compression > 0)
+ if (dir_data->compression_method == COMPRESSION_GZIP)
{
gzfp = gzdopen(fd, "wb");
if (gzfp == NULL)
return NULL;
}
- if (gzsetparams(gzfp, dir_data->compression,
+ if (gzsetparams(gzfp, dir_data->compression_level,
Z_DEFAULT_STRATEGY) != Z_OK)
{
gzclose(gzfp);
#endif
/* Do pre-padding on non-compressed files */
- if (pad_to_size && dir_data->compression == 0)
+ if (pad_to_size && dir_data->compression_method == COMPRESSION_NONE)
{
PGAlignedXLogBlock zerobuf;
int bytes;
fsync_parent_path(tmppath) != 0)
{
#ifdef HAVE_LIBZ
- if (dir_data->compression > 0)
+ if (dir_data->compression_method == COMPRESSION_GZIP)
gzclose(gzfp);
else
#endif
f = pg_malloc0(sizeof(DirectoryMethodFile));
#ifdef HAVE_LIBZ
- if (dir_data->compression > 0)
+ if (dir_data->compression_method == COMPRESSION_GZIP)
f->gzfp = gzfp;
#endif
f->fd = fd;
Assert(f != NULL);
#ifdef HAVE_LIBZ
- if (dir_data->compression > 0)
+ if (dir_data->compression_method == COMPRESSION_GZIP)
r = (ssize_t) gzwrite(df->gzfp, buf, count);
else
#endif
Assert(f != NULL);
#ifdef HAVE_LIBZ
- if (dir_data->compression > 0)
+ if (dir_data->compression_method == COMPRESSION_GZIP)
r = gzclose(df->gzfp);
else
#endif
return 0;
#ifdef HAVE_LIBZ
- if (dir_data->compression > 0)
+ if (dir_data->compression_method == COMPRESSION_GZIP)
{
if (gzflush(((DirectoryMethodFile *) f)->gzfp, Z_SYNC_FLUSH) != Z_OK)
return -1;
return statbuf.st_size;
}
-static int
-dir_compression(void)
+static WalCompressionMethod
+dir_compression_method(void)
{
- return dir_data->compression;
+ return dir_data->compression_method;
}
static bool
WalWriteMethod *
-CreateWalDirectoryMethod(const char *basedir, int compression, bool sync)
+CreateWalDirectoryMethod(const char *basedir,
+ WalCompressionMethod compression_method,
+ int compression_level, bool sync)
{
WalWriteMethod *method;
method->get_current_pos = dir_get_current_pos;
method->get_file_size = dir_get_file_size;
method->get_file_name = dir_get_file_name;
- method->compression = dir_compression;
+ method->compression_method = dir_compression_method;
method->close = dir_close;
method->sync = dir_sync;
method->existsfile = dir_existsfile;
method->getlasterror = dir_getlasterror;
dir_data = pg_malloc0(sizeof(DirectoryMethodData));
- dir_data->compression = compression;
+ dir_data->compression_method = compression_method;
+ dir_data->compression_level = compression_level;
dir_data->basedir = pg_strdup(basedir);
dir_data->sync = sync;
{
char *tarfilename;
int fd;
- int compression;
+ WalCompressionMethod compression_method;
+ int compression_level;
bool sync;
TarMethodFile *currentfile;
char lasterror[1024];
tar_clear_error();
/* Tarfile will always be positioned at the end */
- if (!tar_data->compression)
+ if (!tar_data->compression_level)
{
r = write(tar_data->fd, buf, count);
if (r > 0)
return NULL;
#ifdef HAVE_LIBZ
- if (tar_data->compression)
+ if (tar_data->compression_level)
{
tar_data->zp = (z_streamp) pg_malloc(sizeof(z_stream));
tar_data->zp->zalloc = Z_NULL;
* default 15 for the windowBits parameter makes the output be
* gzip instead of zlib.
*/
- if (deflateInit2(tar_data->zp, tar_data->compression, Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY) != Z_OK)
+ if (deflateInit2(tar_data->zp, tar_data->compression_level,
+ Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY) != Z_OK)
{
pg_free(tar_data->zp);
tar_data->zp = NULL;
pg_free(tmppath);
#ifdef HAVE_LIBZ
- if (tar_data->compression)
+ if (tar_data->compression_level)
{
/* Flush existing data */
if (!tar_write_compressed_data(NULL, 0, true))
}
tar_data->currentfile->currpos = 0;
- if (!tar_data->compression)
+ if (!tar_data->compression_level)
{
errno = 0;
if (write(tar_data->fd, tar_data->currentfile->header,
return NULL;
/* Re-enable compression for the rest of the file */
- if (deflateParams(tar_data->zp, tar_data->compression, 0) != Z_OK)
+ if (deflateParams(tar_data->zp, tar_data->compression_level, 0) != Z_OK)
{
tar_set_error("could not change compression parameters");
return NULL;
if (pad_to_size)
{
tar_data->currentfile->pad_to_size = pad_to_size;
- if (!tar_data->compression)
+ if (!tar_data->compression_level)
{
/* Uncompressed, so pad now */
tar_write_padding_data(tar_data->currentfile, pad_to_size);
return -1;
}
-static int
-tar_compression(void)
+static WalCompressionMethod
+tar_compression_method(void)
{
- return tar_data->compression;
+ return tar_data->compression_method;
}
static off_t
* Always sync the whole tarfile, because that's all we can do. This makes
* no sense on compressed files, so just ignore those.
*/
- if (tar_data->compression)
+ if (tar_data->compression_level)
return 0;
return fsync(tar_data->fd);
if (method == CLOSE_UNLINK)
{
- if (tar_data->compression)
+ if (tar_data->compression_level)
{
tar_set_error("unlink not supported with compression");
return -1;
*/
if (tf->pad_to_size)
{
- if (tar_data->compression)
+ if (tar_data->compression_level)
{
/*
* A compressed tarfile is padded on close since we cannot know
#ifdef HAVE_LIBZ
- if (tar_data->compression)
+ if (tar_data->compression_level)
{
/* Flush the current buffer */
if (!tar_write_compressed_data(NULL, 0, true))
print_tar_number(&(tf->header[148]), 8, tarChecksum(((TarMethodFile *) f)->header));
if (lseek(tar_data->fd, tf->ofs_start, SEEK_SET) != ((TarMethodFile *) f)->ofs_start)
return -1;
- if (!tar_data->compression)
+ if (!tar_data->compression_level)
{
errno = 0;
if (write(tar_data->fd, tf->header, TAR_BLOCK_SIZE) != TAR_BLOCK_SIZE)
return -1;
/* Turn compression back on */
- if (deflateParams(tar_data->zp, tar_data->compression, 0) != Z_OK)
+ if (deflateParams(tar_data->zp, tar_data->compression_level, 0) != Z_OK)
{
tar_set_error("could not change compression parameters");
return -1;
/* A tarfile always ends with two empty blocks */
MemSet(zerobuf, 0, sizeof(zerobuf));
- if (!tar_data->compression)
+ if (!tar_data->compression_level)
{
errno = 0;
if (write(tar_data->fd, zerobuf, sizeof(zerobuf)) != sizeof(zerobuf))
return true;
}
+/*
+ * The argument compression_method is currently ignored. It is in place for
+ * symmetry with CreateWalDirectoryMethod which uses it for distinguishing
+ * between the different compression methods. CreateWalTarMethod and its family
+ * of functions handle only zlib compression.
+ */
WalWriteMethod *
-CreateWalTarMethod(const char *tarbase, int compression, bool sync)
+CreateWalTarMethod(const char *tarbase,
+ WalCompressionMethod compression_method,
+ int compression_level, bool sync)
{
WalWriteMethod *method;
- const char *suffix = (compression != 0) ? ".tar.gz" : ".tar";
+ const char *suffix = (compression_level != 0) ? ".tar.gz" : ".tar";
method = pg_malloc0(sizeof(WalWriteMethod));
method->open_for_write = tar_open_for_write;
method->get_current_pos = tar_get_current_pos;
method->get_file_size = tar_get_file_size;
method->get_file_name = tar_get_file_name;
- method->compression = tar_compression;
+ method->compression_method = tar_compression_method;
method->close = tar_close;
method->sync = tar_sync;
method->existsfile = tar_existsfile;
tar_data->tarfilename = pg_malloc0(strlen(tarbase) + strlen(suffix) + 1);
sprintf(tar_data->tarfilename, "%s%s", tarbase, suffix);
tar_data->fd = -1;
- tar_data->compression = compression;
+ tar_data->compression_method = compression_method;
+ tar_data->compression_level = compression_level;
tar_data->sync = sync;
#ifdef HAVE_LIBZ
- if (compression)
+ if (compression_level)
tar_data->zlibOut = (char *) pg_malloc(ZLIB_OUT_SIZE + 1);
#endif
{
pg_free(tar_data->tarfilename);
#ifdef HAVE_LIBZ
- if (tar_data->compression)
+ if (tar_data->compression_level)
pg_free(tar_data->zlibOut);
#endif
pg_free(tar_data);
CLOSE_NO_RENAME
} WalCloseMethod;
+/* Types of compression supported */
+typedef enum
+{
+ COMPRESSION_GZIP,
+ COMPRESSION_NONE
+} WalCompressionMethod;
+
/*
* A WalWriteMethod structure represents the different methods used
* to write the streaming WAL as it's received.
*/
char *(*get_file_name) (const char *pathname, const char *temp_suffix);
- /* Return the level of compression */
- int (*compression) (void);
+ /* Returns the compression method */
+ WalCompressionMethod (*compression_method) (void);
/*
* Write count number of bytes to the file, and return the number of bytes
* not all those required for pg_receivewal)
*/
WalWriteMethod *CreateWalDirectoryMethod(const char *basedir,
+ WalCompressionMethod compression_method,
int compression, bool sync);
-WalWriteMethod *CreateWalTarMethod(const char *tarbase, int compression, bool sync);
+WalWriteMethod *CreateWalTarMethod(const char *tarbase,
+ WalCompressionMethod compression_method,
+ int compression, bool sync);
/* Cleanup routines for previously-created methods */
void FreeWalDirectoryMethod(void);
WaitPMResult
WalCloseMethod
WalCompression
+WalCompressionMethod
WalLevel
WalRcvData
WalRcvExecResult
projects / users / gsingh / postgres.git / commitdiff