agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Justin Pryzby <[email protected]>
Subject: [PATCH 2/4] pg_dump: zstd compression
Date: Sat, 7 Jan 2023 15:45:06 -0600
Previously proposed at: [email protected]
---
doc/src/sgml/ref/pg_dump.sgml | 10 +-
src/bin/pg_dump/Makefile | 2 +
src/bin/pg_dump/compress_io.c | 53 +--
src/bin/pg_dump/compress_zstd.c | 505 ++++++++++++++++++++++++++
src/bin/pg_dump/compress_zstd.h | 9 +
src/bin/pg_dump/meson.build | 1 +
src/bin/pg_dump/pg_backup_archiver.c | 9 +-
src/bin/pg_dump/pg_backup_directory.c | 2 +
src/bin/pg_dump/pg_dump.c | 13 -
src/bin/pg_dump/t/002_pg_dump.pl | 79 +++-
src/tools/pginclude/cpluspluscheck | 1 +
11 files changed, 637 insertions(+), 47 deletions(-)
create mode 100644 src/bin/pg_dump/compress_zstd.c
create mode 100644 src/bin/pg_dump/compress_zstd.h
diff --git a/doc/src/sgml/ref/pg_dump.sgml b/doc/src/sgml/ref/pg_dump.sgml
index 6fbe49f7ede..0df86a245f0 100644
--- a/doc/src/sgml/ref/pg_dump.sgml
+++ b/doc/src/sgml/ref/pg_dump.sgml
@@ -330,8 +330,9 @@ PostgreSQL documentation
machine-readable format that <application>pg_restore</application>
can read. A directory format archive can be manipulated with
standard Unix tools; for example, files in an uncompressed archive
- can be compressed with the <application>gzip</application> or
- <application>lz4</application> tools.
+ can be compressed with the <application>gzip</application>,
+ <application>lz4</application>, or
+ <application>zstd</application> tools.
This format is compressed by default using <literal>gzip</literal>
and also supports parallel dumps.
</para>
@@ -654,8 +655,9 @@ PostgreSQL documentation
<listitem>
<para>
Specify the compression method and/or the compression level to use.
- The compression method can be set to <literal>gzip</literal> or
- <literal>lz4</literal>, or <literal>none</literal> for no compression.
+ The compression method can be set to <literal>gzip</literal>,
+ <literal>lz4</literal> or <literal>zstd</literal>,
+ or <literal>none</literal> for no compression.
A compression detail string can optionally be specified. If the
detail string is an integer, it specifies the compression level.
Otherwise, it should be a comma-separated list of items, each of the
diff --git a/src/bin/pg_dump/Makefile b/src/bin/pg_dump/Makefile
index eb8f59459a1..bf540fee2ba 100644
--- a/src/bin/pg_dump/Makefile
+++ b/src/bin/pg_dump/Makefile
@@ -18,6 +18,7 @@ include $(top_builddir)/src/Makefile.global
export GZIP_PROGRAM=$(GZIP)
export LZ4
+#XXX export ZSTD
export with_icu
override CPPFLAGS := -I$(libpq_srcdir) $(CPPFLAGS)
@@ -29,6 +30,7 @@ OBJS = \
compress_io.o \
compress_lz4.o \
compress_none.o \
+ compress_zstd.o \
dumputils.o \
parallel.o \
pg_backup_archiver.o \
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index 9239dbb2755..121c69ed48f 100644
--- a/src/bin/pg_dump/compress_io.c
+++ b/src/bin/pg_dump/compress_io.c
@@ -52,8 +52,8 @@
*
* InitDiscoverCompressFileHandle tries to infer the compression by the
* filename suffix. If the suffix is not yet known then it tries to simply
- * open the file and if it fails, it tries to open the same file with the .gz
- * suffix, and then again with the .lz4 suffix.
+ * open the file and if it fails, it tries to open the same file with
+ * compressed suffixes.
*
* IDENTIFICATION
* src/bin/pg_dump/compress_io.c
@@ -69,6 +69,7 @@
#include "compress_io.h"
#include "compress_lz4.h"
#include "compress_none.h"
+#include "compress_zstd.h"
#include "pg_backup_utils.h"
/*----------------------
@@ -98,6 +99,10 @@ pgdump_supports_compression(const pg_compress_specification compression_spec)
if (algorithm == PG_COMPRESSION_LZ4)
supported = true;
#endif
+#ifdef USE_ZSTD
+ if (algorithm == PG_COMPRESSION_ZSTD)
+ supported = true;
+#endif
if (!supported)
return psprintf("this build does not support compression with %s",
@@ -130,6 +135,8 @@ AllocateCompressor(const pg_compress_specification compression_spec,
InitCompressorGzip(cs, compression_spec);
else if (compression_spec.algorithm == PG_COMPRESSION_LZ4)
InitCompressorLZ4(cs, compression_spec);
+ else if (compression_spec.algorithm == PG_COMPRESSION_ZSTD)
+ InitCompressorZstd(cs, compression_spec);
return cs;
}
@@ -196,20 +203,30 @@ InitCompressFileHandle(const pg_compress_specification compression_spec)
InitCompressFileHandleGzip(CFH, compression_spec);
else if (compression_spec.algorithm == PG_COMPRESSION_LZ4)
InitCompressFileHandleLZ4(CFH, compression_spec);
+ else if (compression_spec.algorithm == PG_COMPRESSION_ZSTD)
+ InitCompressFileHandleZstd(CFH, compression_spec);
return CFH;
}
+static bool
+check_compressed_file(const char *path, char **fname, char *ext)
+{
+ free_keep_errno(*fname);
+ *fname = psprintf("%s.%s", path, ext);
+ return (access(*fname, F_OK) == 0);
+}
+
/*
* Open a file for reading. 'path' is the file to open, and 'mode' should
* be either "r" or "rb".
*
* If the file at 'path' contains the suffix of a supported compression method,
- * currently this includes ".gz" and ".lz4", then this compression will be used
+ * currently this includes ".gz", ".lz4" and ".zst", then this compression will be used
* throughout. Otherwise the compression will be inferred by iteratively trying
* to open the file at 'path', first as is, then by appending known compression
* suffixes. So if you pass "foo" as 'path', this will open either "foo" or
- * "foo.gz" or "foo.lz4", trying in that order.
+ * "foo.{gz,lz4,zst}", trying in that order.
*
* On failure, return NULL with an error code in errno.
*/
@@ -237,28 +254,12 @@ InitDiscoverCompressFileHandle(const char *path, const char *mode)
/* avoid unused warning if it is not built with compression */
if (exists)
compression_spec.algorithm = PG_COMPRESSION_NONE;
-#ifdef HAVE_LIBZ
- if (!exists)
- {
- free_keep_errno(fname);
- fname = psprintf("%s.gz", path);
- exists = (stat(fname, &st) == 0);
-
- if (exists)
- compression_spec.algorithm = PG_COMPRESSION_GZIP;
- }
-#endif
-#ifdef USE_LZ4
- if (!exists)
- {
- free_keep_errno(fname);
- fname = psprintf("%s.lz4", path);
- exists = (stat(fname, &st) == 0);
-
- if (exists)
- compression_spec.algorithm = PG_COMPRESSION_LZ4;
- }
-#endif
+ else if (check_compressed_file(path, &fname, "gz"))
+ compression_spec.algorithm = PG_COMPRESSION_GZIP;
+ else if (check_compressed_file(path, &fname, "lz4"))
+ compression_spec.algorithm = PG_COMPRESSION_LZ4;
+ else if (check_compressed_file(path, &fname, "zst"))
+ compression_spec.algorithm = PG_COMPRESSION_ZSTD;
}
CFH = InitCompressFileHandle(compression_spec);
diff --git a/src/bin/pg_dump/compress_zstd.c b/src/bin/pg_dump/compress_zstd.c
new file mode 100644
index 00000000000..5d2f2bf2a2d
--- /dev/null
+++ b/src/bin/pg_dump/compress_zstd.c
@@ -0,0 +1,505 @@
+#include "postgres_fe.h"
+
+#include "pg_backup_utils.h"
+#include "compress_zstd.h"
+
+#ifndef USE_ZSTD
+
+void
+InitCompressorZstd(CompressorState *cs, const pg_compress_specification compression_spec)
+{
+ pg_fatal("this build does not support compression with %s", "ZSTD");
+}
+
+void
+InitCompressFileHandleZstd(CompressFileHandle *CFH, const pg_compress_specification compression_spec)
+{
+ pg_fatal("this build does not support compression with %s", "ZSTD");
+}
+
+#else
+
+#include <zstd.h>
+
+typedef struct ZstdCompressorState
+{
+ /* This is a normal file to which we read/write compressed data */
+ FILE *fp;
+ /* XXX: use one separate ZSTD_CStream per thread: disable on windows ? */
+ ZSTD_CStream *cstream;
+ ZSTD_DStream *dstream;
+ ZSTD_outBuffer output;
+ ZSTD_inBuffer input;
+} ZstdCompressorState;
+
+static ZSTD_CStream *ZstdCStreamParams(pg_compress_specification compress);
+static void EndCompressorZstd(ArchiveHandle *AH, CompressorState *cs);
+static void WriteDataToArchiveZstd(ArchiveHandle *AH, CompressorState *cs,
+ const void *data, size_t dLen);
+static void ReadDataFromArchiveZstd(ArchiveHandle *AH, CompressorState *cs);
+
+static void
+ZSTD_CCtx_setParam_or_die(ZSTD_CStream * cstream,
+ ZSTD_cParameter param, int value)
+{
+ size_t res;
+
+ res = ZSTD_CCtx_setParameter(cstream, param, value);
+ if (ZSTD_isError(res))
+ pg_fatal("could not set compression parameter: %s",
+ ZSTD_getErrorName(res));
+}
+
+/* Return a compression stream with parameters set per argument */
+static ZSTD_CStream *
+ZstdCStreamParams(pg_compress_specification compress)
+{
+ ZSTD_CStream *cstream;
+
+ cstream = ZSTD_createCStream();
+ if (cstream == NULL)
+ pg_fatal("could not initialize compression library");
+
+ ZSTD_CCtx_setParam_or_die(cstream, ZSTD_c_compressionLevel,
+ compress.level);
+
+ if (compress.options & PG_COMPRESSION_OPTION_WORKERS)
+ ZSTD_CCtx_setParam_or_die(cstream, ZSTD_c_nbWorkers,
+ compress.workers);
+
+#if 0
+ if (compress.options & PG_COMPRESSION_OPTION_CHECKSUM)
+ ZSTD_CCtx_setParam_or_die(cstream, ZSTD_c_checksumFlag,
+ compress.checksum);
+
+ /* Still marked as experimental */
+ if (compress.options & PG_COMPRESSION_OPTION_RSYNCABLE)
+ ZSTD_CCtx_setParam_or_die(cstream, ZSTD_c_rsyncable, 1);
+#endif
+
+ return cstream;
+}
+
+/* Helper function for EndCompressorZstd and WriteDataToArchiveZstd */
+static void
+ZstdWriteCommon(ArchiveHandle *AH, CompressorState *cs, bool flush)
+{
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) cs->private_data;
+ ZSTD_inBuffer *input = &zstdcs->input;
+ ZSTD_outBuffer *output = &zstdcs->output;
+
+ /* Loop while there's any input or until flushed */
+ while (input->pos != input->size || flush)
+ {
+ size_t res;
+
+ res = ZSTD_compressStream2(zstdcs->cstream, output,
+ input, flush ? ZSTD_e_end : ZSTD_e_continue);
+
+ if (ZSTD_isError(res))
+ pg_fatal("could not compress data: %s", ZSTD_getErrorName(res));
+
+ /*
+ * Extra paranoia: avoid zero-length chunks, since a zero length
+ * chunk is the EOF marker in the custom format. This should never
+ * happen but...
+ */
+ if (output->pos > 0)
+ cs->writeF(AH, output->dst, output->pos);
+
+ output->pos = 0;
+
+ if (res == 0)
+ break;
+ }
+}
+
+void
+EndCompressorZstd(ArchiveHandle *AH, CompressorState *cs)
+{
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) cs->private_data;
+
+ if (zstdcs->cstream)
+ {
+ ZstdWriteCommon(AH, cs, true);
+ ZSTD_freeCStream(zstdcs->cstream);
+ pg_free(zstdcs->output.dst);
+ zstdcs->cstream = NULL;
+ }
+ pg_free(zstdcs);
+}
+
+static void
+WriteDataToArchiveZstd(ArchiveHandle *AH, CompressorState *cs,
+ const void *data, size_t dLen)
+{
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) cs->private_data;
+
+ if (zstdcs->cstream == NULL)
+ {
+ zstdcs->cstream = ZstdCStreamParams(cs->compression_spec);
+
+ zstdcs->output.size = ZSTD_CStreamOutSize();
+ zstdcs->output.dst = pg_malloc(zstdcs->output.size);
+ zstdcs->output.pos = 0;
+ }
+
+ zstdcs->input.src = data;
+ zstdcs->input.size = dLen;
+ zstdcs->input.pos = 0;
+
+ ZstdWriteCommon(AH, cs, false);
+}
+
+/* Read data from a compressed zstd archive */
+static void
+ReadDataFromArchiveZstd(ArchiveHandle *AH, CompressorState *cs)
+{
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) cs->private_data;
+ ZSTD_outBuffer *output = &zstdcs->output;
+ ZSTD_inBuffer *input = &zstdcs->input;
+ size_t res;
+
+ if (zstdcs->dstream == NULL)
+ {
+ zstdcs->dstream = ZSTD_createDStream();
+ if (zstdcs->dstream == NULL)
+ pg_fatal("could not initialize compression library");
+
+ input->size = ZSTD_DStreamInSize();
+ input->src = pg_malloc(input->size);
+
+ output->size = ZSTD_DStreamOutSize();
+ output->dst = pg_malloc(output->size);
+ }
+
+ /* read compressed data */
+ for (;;)
+ {
+ size_t cnt;
+
+ /*
+ * XXX: the buffer can grow, we shouldn't keep resetting it to the
+ * original value..
+ */
+ input->size = ZSTD_DStreamInSize();
+
+ cnt = cs->readF(AH, (char **) unconstify(void **, &input->src), &input->size);
+ input->pos = 0;
+ input->size = cnt;
+
+ if (cnt == 0)
+ break;
+
+ /* Now consume as much as possible */
+ while (input->pos < input->size)
+ {
+ /* decompress */
+ output->pos = 0;
+ res = ZSTD_decompressStream(zstdcs->dstream, output, input);
+ if (ZSTD_isError(res))
+ pg_fatal("could not decompress data: %s", ZSTD_getErrorName(res));
+
+ /* write to output handle */
+ ((char *) output->dst)[output->pos] = '\0';
+ ahwrite(output->dst, 1, output->pos, AH);
+
+ if (res == 0)
+ break; /* End of frame */
+ }
+ }
+
+ pg_free(unconstify(void *, input->src));
+ pg_free(output->dst);
+}
+
+/* Public routines that support Zstd compressed data I/O */
+void
+InitCompressorZstd(CompressorState *cs, const pg_compress_specification compression_spec)
+{
+ cs->readData = ReadDataFromArchiveZstd;
+ cs->writeData = WriteDataToArchiveZstd;
+ cs->end = EndCompressorZstd;
+
+ cs->compression_spec = compression_spec;
+
+ cs->private_data = pg_malloc0(sizeof(ZstdCompressorState));
+}
+
+/*----------------------
+ * Compressed stream API
+ *----------------------
+ */
+
+static size_t
+Zstd_read(void *ptr, size_t size, CompressFileHandle *CFH)
+{
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data;
+ ZSTD_inBuffer *input = &zstdcs->input;
+ ZSTD_outBuffer *output = &zstdcs->output;
+ /* input_size is the allocated size */
+ size_t input_size = ZSTD_DStreamInSize();
+ size_t res,
+ cnt;
+
+ output->size = size;
+ output->dst = ptr;
+ output->pos = 0;
+
+ for (;;)
+ {
+ Assert(input->pos <= input->size);
+ Assert(input->size <= input_size);
+
+ /* If the input is completely consumed, start back at the beginning */
+ if (input->pos == input->size)
+ {
+ /* input->size is size produced by "fread" */
+ input->size = 0;
+ /* input->pos is position consumed by decompress */
+ input->pos = 0;
+ }
+
+ /* read compressed data if we must produce more input */
+ if (input->pos == input->size)
+ {
+ cnt = fread(unconstify(void *, input->src), 1, input_size, zstdcs->fp);
+ input->size = cnt;
+
+ /* If we have no input to consume, we're done */
+ if (cnt == 0)
+ break;
+ }
+
+ Assert(cnt >= 0);
+ Assert(input->size <= input_size);
+
+ while (input->pos < input->size)
+ {
+ /* decompress */
+ res = ZSTD_decompressStream(zstdcs->dstream, output, input);
+
+ if (ZSTD_isError(res))
+ pg_fatal("could not decompress data: %s", ZSTD_getErrorName(res));
+
+ if (output->pos == output->size)
+ break; /* No more room for output */
+
+ if (res == 0)
+ break; /* End of frame */
+ }
+
+ if (output->pos == output->size)
+ break; /* We read all the data that fits */
+ }
+
+ return output->pos;
+}
+
+static size_t
+Zstd_write(const void *ptr, size_t size, CompressFileHandle *CFH)
+{
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data;
+ ZSTD_inBuffer *input = &zstdcs->input;
+ ZSTD_outBuffer *output = &zstdcs->output;
+ size_t res,
+ cnt;
+
+ input->src = ptr;
+ input->size = size;
+ input->pos = 0;
+
+#if 0
+ ZSTD_CCtx_reset(fp->zstd.cstream, ZSTD_reset_session_only);
+ res = ZSTD_CCtx_setPledgedSrcSize(fp->zstd.cstream, size);
+ if (ZSTD_isError(res))
+ pg_fatal("could not compress data: %s", ZSTD_getErrorName(res));
+#endif
+
+ /* Consume all input, and flush later */
+ while (input->pos != input->size)
+ {
+ output->pos = 0;
+ res = ZSTD_compressStream2(zstdcs->cstream, output, input, ZSTD_e_continue);
+ if (ZSTD_isError(res))
+ pg_fatal("could not compress data: %s", ZSTD_getErrorName(res));
+
+ cnt = fwrite(output->dst, 1, output->pos, zstdcs->fp);
+ if (cnt != output->pos)
+ pg_fatal("could not write data: %m");
+ }
+
+ return size;
+}
+
+static int
+Zstd_getc(CompressFileHandle *CFH)
+{
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data;
+ int ret;
+
+ if (CFH->read_func(&ret, 1, CFH) != 1)
+ {
+ if (feof(zstdcs->fp))
+ pg_fatal("could not read from input file: end of file");
+ else
+ pg_fatal("could not read from input file: %m");
+ }
+ return ret;
+}
+
+static char *
+Zstd_gets(char *buf, int len, CompressFileHandle *CFH)
+{
+ int i,
+ res;
+
+ /*
+ * Read one byte at a time until newline or EOF. This is only used to read
+ * the list of LOs, and the I/O is buffered anyway.
+ */
+ for (i = 0; i < len - 1; ++i)
+ {
+ res = CFH->read_func(&buf[i], 1, CFH);
+ if (res != 1)
+ break;
+ if (buf[i] == '\n')
+ {
+ ++i;
+ break;
+ }
+ }
+ buf[i] = '\0';
+ return i > 0 ? buf : NULL;
+}
+
+static int
+Zstd_close(CompressFileHandle *CFH)
+{
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data;
+ int result;
+
+ if (zstdcs->cstream)
+ {
+ size_t res,
+ cnt;
+ ZSTD_inBuffer *input = &zstdcs->input;
+ ZSTD_outBuffer *output = &zstdcs->output;
+
+ for (;;)
+ {
+ output->pos = 0;
+ res = ZSTD_compressStream2(zstdcs->cstream, output, input, ZSTD_e_end);
+ if (ZSTD_isError(res))
+ pg_fatal("could not compress data: %s", ZSTD_getErrorName(res));
+
+ cnt = fwrite(output->dst, 1, output->pos, zstdcs->fp);
+ if (cnt != output->pos)
+ pg_fatal("could not write data: %m");
+
+ if (res == 0)
+ break;
+ }
+
+ ZSTD_freeCStream(zstdcs->cstream);
+ pg_free(zstdcs->output.dst);
+ }
+
+ if (zstdcs->dstream)
+ {
+ ZSTD_freeDStream(zstdcs->dstream);
+ pg_free(unconstify(void *, zstdcs->input.src));
+ }
+
+ result = fclose(zstdcs->fp);
+ pg_free(zstdcs);
+ return result;
+}
+
+static int
+Zstd_eof(CompressFileHandle *CFH)
+{
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data;
+
+ return feof(zstdcs->fp);
+}
+
+static int
+Zstd_open(const char *path, int fd, const char *mode,
+ CompressFileHandle *CFH)
+{
+ FILE *fp;
+ ZstdCompressorState *zstdcs;
+
+ if (fd >= 0)
+ fp = fdopen(fd, mode);
+ else
+ fp = fopen(path, mode);
+
+ if (fp == NULL)
+ return 1;
+
+ CFH->private_data = pg_malloc0(sizeof(ZstdCompressorState));
+ zstdcs = (ZstdCompressorState *) CFH->private_data;
+ zstdcs->fp = fp;
+
+ if (mode[0] == 'w' || mode[0] == 'a')
+ {
+ zstdcs->output.size = ZSTD_CStreamOutSize();
+ zstdcs->output.dst = pg_malloc0(zstdcs->output.size);
+ zstdcs->cstream = ZstdCStreamParams(CFH->compression_spec);
+ }
+ else if (strchr(mode, 'r'))
+ {
+ zstdcs->input.src = pg_malloc0(ZSTD_DStreamInSize());
+ zstdcs->dstream = ZSTD_createDStream();
+ if (zstdcs->dstream == NULL)
+ pg_fatal("could not initialize compression library");
+ }
+ /* XXX else: bad mode */
+
+ return 0;
+}
+
+static int
+Zstd_open_write(const char *path, const char *mode, CompressFileHandle *CFH)
+{
+ char fname[MAXPGPATH];
+
+ sprintf(fname, "%s.zst", path);
+ return CFH->open_func(fname, -1, mode, CFH);
+}
+
+static const char *
+Zstd_get_error(CompressFileHandle *CFH)
+{
+#if 0
+ ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data;
+
+ if (ZSTD_isError(res))
+ return ZSTD_getErrorName(res)
+ else
+#endif
+
+ return strerror(errno);
+}
+
+void
+InitCompressFileHandleZstd(CompressFileHandle *CFH, const pg_compress_specification compression_spec)
+{
+ CFH->open_func = Zstd_open;
+ CFH->open_write_func = Zstd_open_write;
+ CFH->read_func = Zstd_read;
+ CFH->write_func = Zstd_write;
+ CFH->gets_func = Zstd_gets;
+ CFH->getc_func = Zstd_getc;
+ CFH->close_func = Zstd_close;
+ CFH->eof_func = Zstd_eof;
+ CFH->get_error_func = Zstd_get_error;
+
+ CFH->compression_spec = compression_spec;
+
+ CFH->private_data = NULL;
+}
+
+#endif /* USE_ZSTD */
diff --git a/src/bin/pg_dump/compress_zstd.h b/src/bin/pg_dump/compress_zstd.h
new file mode 100644
index 00000000000..f36698b4c26
--- /dev/null
+++ b/src/bin/pg_dump/compress_zstd.h
@@ -0,0 +1,9 @@
+#ifndef COMPRESS_ZSTD_H
+#define COMPRESS_ZSTD_H
+
+#include "compress_io.h"
+
+extern void InitCompressorZstd(CompressorState *cs, const pg_compress_specification compression_spec);
+extern void InitCompressFileHandleZstd(CompressFileHandle *CFH, const pg_compress_specification compression_spec);
+
+#endif /* COMPRESS_ZSTD_H */
diff --git a/src/bin/pg_dump/meson.build b/src/bin/pg_dump/meson.build
index 0da476a4c34..334d449091d 100644
--- a/src/bin/pg_dump/meson.build
+++ b/src/bin/pg_dump/meson.build
@@ -5,6 +5,7 @@ pg_dump_common_sources = files(
'compress_io.c',
'compress_lz4.c',
'compress_none.c',
+ 'compress_zstd.c',
'dumputils.c',
'parallel.c',
'pg_backup_archiver.c',
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 2063d6f239d..fc739db4355 100644
--- a/src/bin/pg_dump/pg_backup_archiver.c
+++ b/src/bin/pg_dump/pg_backup_archiver.c
@@ -2075,7 +2075,7 @@ _discoverArchiveFormat(ArchiveHandle *AH)
/*
* Check if the specified archive is a directory. If so, check if
- * there's a "toc.dat" (or "toc.dat.{gz,lz4}") file in it.
+ * there's a "toc.dat" (or "toc.dat.{gz,lz4,zst}") file in it.
*/
if (stat(AH->fSpec, &st) == 0 && S_ISDIR(st.st_mode))
{
@@ -2086,10 +2086,17 @@ _discoverArchiveFormat(ArchiveHandle *AH)
if (_fileExistsInDirectory(AH->fSpec, "toc.dat.gz"))
return AH->format;
#endif
+
#ifdef USE_LZ4
if (_fileExistsInDirectory(AH->fSpec, "toc.dat.lz4"))
return AH->format;
#endif
+
+#ifdef USE_ZSTD
+ if (_fileExistsInDirectory(AH->fSpec, "toc.dat.zst"))
+ return AH->format;
+#endif
+
pg_fatal("directory \"%s\" does not appear to be a valid archive (\"toc.dat\" does not exist)",
AH->fSpec);
fh = NULL; /* keep compiler quiet */
diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c
index 41c2b733e3e..29845340859 100644
--- a/src/bin/pg_dump/pg_backup_directory.c
+++ b/src/bin/pg_dump/pg_backup_directory.c
@@ -785,6 +785,8 @@ _PrepParallelRestore(ArchiveHandle *AH)
strlcat(fname, ".gz", sizeof(fname));
else if (AH->compression_spec.algorithm == PG_COMPRESSION_LZ4)
strlcat(fname, ".lz4", sizeof(fname));
+ else if (AH->compression_spec.algorithm == PG_COMPRESSION_ZSTD)
+ strlcat(fname, ".zst", sizeof(fname));
if (stat(fname, &st) == 0)
te->dataLength = st.st_size;
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index ce2242195f3..6495a88ff99 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -710,19 +710,6 @@ main(int argc, char **argv)
pg_fatal("invalid compression specification: %s",
error_detail);
- switch (compression_algorithm)
- {
- case PG_COMPRESSION_NONE:
- /* fallthrough */
- case PG_COMPRESSION_GZIP:
- /* fallthrough */
- case PG_COMPRESSION_LZ4:
- break;
- case PG_COMPRESSION_ZSTD:
- pg_fatal("compression with %s is not yet supported", "ZSTD");
- break;
- }
-
/*
* Custom and directory formats are compressed by default with gzip when
* available, not the others.
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index ad7bc5c194b..eb5e09f55c8 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -54,8 +54,9 @@ my $tempdir = PostgreSQL::Test::Utils::tempdir;
# those lines) to validate that part of the process.
my $supports_icu = ($ENV{with_icu} eq 'yes');
-my $supports_lz4 = check_pg_config("#define USE_LZ4 1");
my $supports_gzip = check_pg_config("#define HAVE_LIBZ 1");
+my $supports_lz4 = check_pg_config("#define USE_LZ4 1");
+my $supports_zstd = check_pg_config("#define USE_ZSTD 1");
my %pgdump_runs = (
binary_upgrade => {
@@ -213,6 +214,77 @@ my %pgdump_runs = (
},
},
+ compression_zstd_custom => {
+ test_key => 'compression',
+ compile_option => 'zstd',
+ dump_cmd => [
+ 'pg_dump', '--format=custom',
+ '--compress=zstd', "--file=$tempdir/compression_zstd_custom.dump",
+ 'postgres',
+ ],
+ restore_cmd => [
+ 'pg_restore',
+ "--file=$tempdir/compression_zstd_custom.sql",
+ "$tempdir/compression_zstd_custom.dump",
+ ],
+ command_like => {
+ command => [
+ 'pg_restore',
+ '-l', "$tempdir/compression_zstd_custom.dump",
+ ],
+ expected => qr/Compression: zstd/,
+ name => 'data content is zstd compressed'
+ },
+ },
+
+ compression_zstd_dir => {
+ test_key => 'compression',
+ compile_option => 'zstd',
+ dump_cmd => [
+ 'pg_dump', '--jobs=2',
+ '--format=directory', '--compress=zstd:1',
+ "--file=$tempdir/compression_zstd_dir", 'postgres',
+ ],
+ # Give coverage for manually compressed blob.toc files during
+ # restore.
+ compress_cmd => {
+ program => $ENV{'ZSTD'},
+ args => [
+ '-z', '-f', '--rm',
+ "$tempdir/compression_zstd_dir/blobs.toc",
+ "$tempdir/compression_zstd_dir/blobs.toc.zst",
+ ],
+ },
+ # Verify that data files were compressed
+ glob_patterns => [
+ "$tempdir/compression_zstd_dir/toc.dat",
+ "$tempdir/compression_zstd_dir/*.dat.zst",
+ ],
+ restore_cmd => [
+ 'pg_restore', '--jobs=2',
+ "--file=$tempdir/compression_zstd_dir.sql",
+ "$tempdir/compression_zstd_dir",
+ ],
+ },
+
+ compression_zstd_plain => {
+ test_key => 'compression',
+ compile_option => 'zstd',
+ dump_cmd => [
+ 'pg_dump', '--format=plain', '--compress=zstd',
+ "--file=$tempdir/compression_zstd_plain.sql.zst", 'postgres',
+ ],
+ # Decompress the generated file to run through the tests.
+ compress_cmd => {
+ program => $ENV{'ZSTD'},
+ args => [
+ '-d', '-f',
+ "$tempdir/compression_zstd_plain.sql.zst",
+ "$tempdir/compression_zstd_plain.sql",
+ ],
+ },
+ },
+
clean => {
dump_cmd => [
'pg_dump',
@@ -4248,10 +4320,11 @@ foreach my $run (sort keys %pgdump_runs)
my $test_key = $run;
my $run_db = 'postgres';
- # Skip command-level tests for gzip/lz4 if they're not supported.
+ # Skip command-level tests for gzip/lz4/zstd if they're not supported.
if ($pgdump_runs{$run}->{compile_option} &&
(($pgdump_runs{$run}->{compile_option} eq 'gzip' && !$supports_gzip) ||
- ($pgdump_runs{$run}->{compile_option} eq 'lz4' && !$supports_lz4)))
+ ($pgdump_runs{$run}->{compile_option} eq 'lz4' && !$supports_lz4) ||
+ ($pgdump_runs{$run}->{compile_option} eq 'zstd' && !$supports_zstd)))
{
note "$run: skipped due to no $pgdump_runs{$run}->{compile_option} support";
next;
diff --git a/src/tools/pginclude/cpluspluscheck b/src/tools/pginclude/cpluspluscheck
index 135633fb19b..71d19683817 100755
--- a/src/tools/pginclude/cpluspluscheck
+++ b/src/tools/pginclude/cpluspluscheck
@@ -154,6 +154,7 @@ do
test "$f" = src/bin/pg_dump/compress_io.h && continue
test "$f" = src/bin/pg_dump/compress_lz4.h && continue
test "$f" = src/bin/pg_dump/compress_none.h && continue
+ test "$f" = src/bin/pg_dump/compress_zstd.h && continue
test "$f" = src/bin/pg_dump/parallel.h && continue
test "$f" = src/bin/pg_dump/pg_backup_archiver.h && continue
test "$f" = src/bin/pg_dump/pg_dump.h && continue
--
2.34.1
--VbXIs2ae+j0NhctA
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="0003-TMP-pg_dump-use-Zstd-by-default-for-CI-only.patch"
view thread (10+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH 2/4] pg_dump: zstd compression
In-Reply-To: <no-message-id-103502@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox