agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Justin Pryzby <[email protected]>
Subject: [PATCH 3/7] pg_dump: zstd compression
Date: Fri, 11 Dec 2020 15:01:25 -0600
---
configure | 123 ++++++-
configure.ac | 22 ++
src/bin/pg_dump/compress_io.c | 475 ++++++++++++++++++++++++--
src/bin/pg_dump/compress_io.h | 5 +-
src/bin/pg_dump/pg_backup_archiver.h | 5 +
src/bin/pg_dump/pg_backup_directory.c | 6 +-
src/bin/pg_dump/pg_dump.c | 3 +-
src/include/pg_config.h.in | 3 +
8 files changed, 597 insertions(+), 45 deletions(-)
diff --git a/configure b/configure
index 11a4284e5b..240e536e04 100755
--- a/configure
+++ b/configure
@@ -698,6 +698,7 @@ with_gnu_ld
LD
LDFLAGS_SL
LDFLAGS_EX
+with_zstd
with_zlib
with_system_tzdata
with_libxslt
@@ -798,6 +799,7 @@ infodir
docdir
oldincludedir
includedir
+runstatedir
localstatedir
sharedstatedir
sysconfdir
@@ -866,6 +868,7 @@ with_libxml
with_libxslt
with_system_tzdata
with_zlib
+with_zstd
with_gnu_ld
enable_largefile
'
@@ -935,6 +938,7 @@ datadir='${datarootdir}'
sysconfdir='${prefix}/etc'
sharedstatedir='${prefix}/com'
localstatedir='${prefix}/var'
+runstatedir='${localstatedir}/run'
includedir='${prefix}/include'
oldincludedir='/usr/include'
docdir='${datarootdir}/doc/${PACKAGE_TARNAME}'
@@ -1187,6 +1191,15 @@ do
| -silent | --silent | --silen | --sile | --sil)
silent=yes ;;
+ -runstatedir | --runstatedir | --runstatedi | --runstated \
+ | --runstate | --runstat | --runsta | --runst | --runs \
+ | --run | --ru | --r)
+ ac_prev=runstatedir ;;
+ -runstatedir=* | --runstatedir=* | --runstatedi=* | --runstated=* \
+ | --runstate=* | --runstat=* | --runsta=* | --runst=* | --runs=* \
+ | --run=* | --ru=* | --r=*)
+ runstatedir=$ac_optarg ;;
+
-sbindir | --sbindir | --sbindi | --sbind | --sbin | --sbi | --sb)
ac_prev=sbindir ;;
-sbindir=* | --sbindir=* | --sbindi=* | --sbind=* | --sbin=* \
@@ -1324,7 +1337,7 @@ fi
for ac_var in exec_prefix prefix bindir sbindir libexecdir datarootdir \
datadir sysconfdir sharedstatedir localstatedir includedir \
oldincludedir docdir infodir htmldir dvidir pdfdir psdir \
- libdir localedir mandir
+ libdir localedir mandir runstatedir
do
eval ac_val=\$$ac_var
# Remove trailing slashes.
@@ -1477,6 +1490,7 @@ Fine tuning of the installation directories:
--sysconfdir=DIR read-only single-machine data [PREFIX/etc]
--sharedstatedir=DIR modifiable architecture-independent data [PREFIX/com]
--localstatedir=DIR modifiable single-machine data [PREFIX/var]
+ --runstatedir=DIR modifiable per-process data [LOCALSTATEDIR/run]
--libdir=DIR object code libraries [EPREFIX/lib]
--includedir=DIR C header files [PREFIX/include]
--oldincludedir=DIR C header files for non-gcc [/usr/include]
@@ -1570,6 +1584,7 @@ Optional Packages:
--with-system-tzdata=DIR
use system time zone data in DIR
--without-zlib do not use Zlib
+ --without-zstd do not use Zstd
--with-gnu-ld assume the C compiler uses GNU ld [default=no]
Some influential environment variables:
@@ -8601,6 +8616,35 @@ fi
+#
+# Zstd
+#
+
+
+
+# Check whether --with-zstd was given.
+if test "${with_zstd+set}" = set; then :
+ withval=$with_zstd;
+ case $withval in
+ yes)
+ :
+ ;;
+ no)
+ :
+ ;;
+ *)
+ as_fn_error $? "no argument expected for --with-zstd option" "$LINENO" 5
+ ;;
+ esac
+
+else
+ with_zstd=yes
+
+fi
+
+
+
+
#
# Assignments
#
@@ -12092,6 +12136,59 @@ fi
fi
+if test "$with_zstd" = yes; then
+ { $as_echo "$as_me:${as_lineno-$LINENO}: checking for ZSTD_compressStream2 in -lzstd" >&5
+$as_echo_n "checking for ZSTD_compressStream2 in -lzstd... " >&6; }
+if ${ac_cv_lib_zstd_ZSTD_compressStream2+:} false; then :
+ $as_echo_n "(cached) " >&6
+else
+ ac_check_lib_save_LIBS=$LIBS
+LIBS="-lzstd $LIBS"
+cat confdefs.h - <<_ACEOF >conftest.$ac_ext
+/* end confdefs.h. */
+
+/* Override any GCC internal prototype to avoid an error.
+ Use char because int might match the return type of a GCC
+ builtin and then its argument prototype would still apply. */
+#ifdef __cplusplus
+extern "C"
+#endif
+char ZSTD_compressStream2 ();
+int
+main ()
+{
+return ZSTD_compressStream2 ();
+ ;
+ return 0;
+}
+_ACEOF
+if ac_fn_c_try_link "$LINENO"; then :
+ ac_cv_lib_zstd_ZSTD_compressStream2=yes
+else
+ ac_cv_lib_zstd_ZSTD_compressStream2=no
+fi
+rm -f core conftest.err conftest.$ac_objext \
+ conftest$ac_exeext conftest.$ac_ext
+LIBS=$ac_check_lib_save_LIBS
+fi
+{ $as_echo "$as_me:${as_lineno-$LINENO}: result: $ac_cv_lib_zstd_ZSTD_compressStream2" >&5
+$as_echo "$ac_cv_lib_zstd_ZSTD_compressStream2" >&6; }
+if test "x$ac_cv_lib_zstd_ZSTD_compressStream2" = xyes; then :
+ cat >>confdefs.h <<_ACEOF
+#define HAVE_LIBZSTD 1
+_ACEOF
+
+ LIBS="-lzstd $LIBS"
+
+else
+ as_fn_error $? "zstd library not found
+If you have zstd already installed, see config.log for details on the
+failure. It is possible the compiler isn't looking in the proper directory.
+Use --without-zstd to disable zstd support." "$LINENO" 5
+fi
+
+fi
+
if test "$enable_spinlocks" = yes; then
$as_echo "#define HAVE_SPINLOCKS 1" >>confdefs.h
@@ -13295,6 +13392,20 @@ Use --without-zlib to disable zlib support." "$LINENO" 5
fi
+fi
+
+if test "$with_zstd" = yes; then
+ ac_fn_c_check_header_mongrel "$LINENO" "zstd.h" "ac_cv_header_zstd_h" "$ac_includes_default"
+if test "x$ac_cv_header_zstd_h" = xyes; then :
+
+else
+ as_fn_error $? "zstd header not found
+If you have zstd already installed, see config.log for details on the
+failure. It is possible the compiler isn't looking in the proper directory.
+Use --without-zstd to disable zstd support." "$LINENO" 5
+fi
+
+
fi
if test "$with_gssapi" = yes ; then
@@ -14689,7 +14800,7 @@ else
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
@@ -14735,7 +14846,7 @@ else
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
@@ -14759,7 +14870,7 @@ rm -f core conftest.err conftest.$ac_objext conftest.$ac_ext
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
@@ -14804,7 +14915,7 @@ else
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
@@ -14828,7 +14939,7 @@ rm -f core conftest.err conftest.$ac_objext conftest.$ac_ext
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
diff --git a/configure.ac b/configure.ac
index fc523c6aeb..7f7222159a 100644
--- a/configure.ac
+++ b/configure.ac
@@ -999,6 +999,13 @@ PGAC_ARG_BOOL(with, zlib, yes,
[do not use Zlib])
AC_SUBST(with_zlib)
+#
+# Zstd
+#
+PGAC_ARG_BOOL(with, zstd, yes,
+ [do not use Zstd])
+AC_SUBST(with_zstd)
+
#
# Assignments
#
@@ -1186,6 +1193,14 @@ failure. It is possible the compiler isn't looking in the proper directory.
Use --without-zlib to disable zlib support.])])
fi
+if test "$with_zstd" = yes; then
+ AC_CHECK_LIB(zstd, ZSTD_compressStream2, [],
+ [AC_MSG_ERROR([zstd library not found
+If you have zstd already installed, see config.log for details on the
+failure. It is possible the compiler isn't looking in the proper directory.
+Use --without-zstd to disable zstd support.])])
+fi
+
if test "$enable_spinlocks" = yes; then
AC_DEFINE(HAVE_SPINLOCKS, 1, [Define to 1 if you have spinlocks.])
else
@@ -1400,6 +1415,13 @@ failure. It is possible the compiler isn't looking in the proper directory.
Use --without-zlib to disable zlib support.])])
fi
+if test "$with_zstd" = yes; then
+ AC_CHECK_HEADER(zstd.h, [], [AC_MSG_ERROR([zstd header not found
+If you have zstd already installed, see config.log for details on the
+failure. It is possible the compiler isn't looking in the proper directory.
+Use --without-zstd to disable zstd support.])])
+fi
+
if test "$with_gssapi" = yes ; then
AC_CHECK_HEADERS(gssapi/gssapi.h, [],
[AC_CHECK_HEADERS(gssapi.h, [], [AC_MSG_ERROR([gssapi.h header file is required for GSSAPI])])])
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index 1417401086..b51ba680a2 100644
--- a/src/bin/pg_dump/compress_io.c
+++ b/src/bin/pg_dump/compress_io.c
@@ -13,7 +13,7 @@
* friends, providing an interface similar to those, but abstracts away
* the possible compression. Both APIs use libz for the compression, but
* the second API uses gzip headers, so the resulting files can be easily
- * manipulated with the gzip utility.
+ * manipulated with the gzip utility. XXX
*
* Compressor API
* --------------
@@ -41,7 +41,7 @@
* libz's gzopen() APIs. It allows you to use the same functions for
* compressed and uncompressed streams. cfopen_read() first tries to open
* the file with given name, and if it fails, it tries to open the same
- * file with the .gz suffix. cfopen_write() opens a file for writing, an
+ * file with the .gz suffix. cfopen_write() opens a file for writing, an XXX
* extra argument specifies if the file should be compressed, and adds the
* .gz suffix to the filename if so. This allows you to easily handle both
* compressed and uncompressed files.
@@ -72,6 +72,18 @@ struct CompressorState
char *zlibOut;
size_t zlibOutSize;
#endif
+
+#ifdef HAVE_LIBZSTD
+ union {
+ struct {
+ ZSTD_outBuffer output;
+ ZSTD_inBuffer input;
+ // XXX: use one separate ZSTD_CStream per thread: disable on windows ?
+ ZSTD_CStream *cstream;
+ } zstd;
+ } u;
+#endif
+
};
static void ParseCompressionOption(int compression, CompressionAlgorithm *alg,
@@ -88,6 +100,15 @@ static void WriteDataToArchiveZlib(ArchiveHandle *AH, CompressorState *cs,
static void EndCompressorZlib(ArchiveHandle *AH, CompressorState *cs);
#endif
+#ifdef HAVE_LIBZSTD
+static void InitCompressorZstd(CompressorState *cs, int level);
+static void EndCompressorZstd(ArchiveHandle *AH, CompressorState *cs);
+static void DeflateCompressorZstd(ArchiveHandle *AH, CompressorState *cs);
+static void WriteDataToArchiveZstd(ArchiveHandle *AH, CompressorState *cs,
+ const char *data, size_t dLen);
+static void ReadDataFromArchiveZstd(ArchiveHandle *AH, ReadFunc readF);
+#endif
+
/* Routines that support uncompressed data I/O */
static void ReadDataFromArchiveNone(ArchiveHandle *AH, ReadFunc readF);
static void WriteDataToArchiveNone(ArchiveHandle *AH, CompressorState *cs,
@@ -101,15 +122,25 @@ static void WriteDataToArchiveNone(ArchiveHandle *AH, CompressorState *cs,
static void
ParseCompressionOption(int compression, CompressionAlgorithm *alg, int *level)
{
- if (compression == Z_DEFAULT_COMPRESSION ||
- (compression > 0 && compression <= 9))
- *alg = COMPR_ALG_LIBZ;
- else if (compression == 0)
- *alg = COMPR_ALG_NONE;
- else
+ switch (compression)
{
- fatal("invalid compression code: %d", compression);
- *alg = COMPR_ALG_NONE; /* keep compiler quiet */
+#ifdef HAVE_LIBZSTD
+ case ZSTD_COMPRESSION:
+ *alg = COMPR_ALG_ZSTD;
+ break;
+#endif
+#ifdef HAVE_ZLIB
+ case Z_DEFAULT_COMPRESSION:
+ case 1..9:
+ *alg = COMPR_ALG_LIBZ;
+ break;
+#endif
+ case 0:
+ *alg = COMPR_ALG_NONE;
+ break;
+ default:
+ fatal("invalid compression code: %d", compression);
+ *alg = COMPR_ALG_NONE; /* keep compiler quiet */
}
/* The level is just the passed-in value. */
@@ -141,10 +172,23 @@ AllocateCompressor(int compression, WriteFunc writeF)
/*
* Perform compression algorithm specific initialization.
*/
+ switch (alg)
+ {
#ifdef HAVE_LIBZ
- if (alg == COMPR_ALG_LIBZ)
+ case COMPR_ALG_LIBZ:
InitCompressorZlib(cs, level);
+ break;
+#endif
+#ifdef HAVE_LIBZSTD
+ case COMPR_ALG_ZSTD:
+ InitCompressorZstd(cs, level);
+ break;
#endif
+ case COMPR_ALG_NONE:
+ /* Do nothing */
+ break;
+ // default:
+ }
return cs;
}
@@ -162,12 +206,20 @@ ReadDataFromArchive(ArchiveHandle *AH, int compression, ReadFunc readF)
if (alg == COMPR_ALG_NONE)
ReadDataFromArchiveNone(AH, readF);
- if (alg == COMPR_ALG_LIBZ)
+ else if (alg == COMPR_ALG_LIBZ)
{
#ifdef HAVE_LIBZ
ReadDataFromArchiveZlib(AH, readF);
#else
fatal("not built with zlib support");
+#endif
+ }
+ else if (alg == COMPR_ALG_ZSTD)
+ {
+#ifdef HAVE_LIBZSTD
+ ReadDataFromArchiveZstd(AH, readF);
+#else
+ fatal("not built with zstd support");
#endif
}
}
@@ -188,6 +240,15 @@ WriteDataToArchive(ArchiveHandle *AH, CompressorState *cs,
fatal("not built with zlib support");
#endif
break;
+
+ case COMPR_ALG_ZSTD:
+#ifdef HAVE_LIBZSTD
+ WriteDataToArchiveZstd(AH, cs, data, dLen);
+#else
+ fatal("not built with zstd support");
+#endif
+ break;
+
case COMPR_ALG_NONE:
WriteDataToArchiveNone(AH, cs, data, dLen);
break;
@@ -204,12 +265,172 @@ EndCompressor(ArchiveHandle *AH, CompressorState *cs)
if (cs->comprAlg == COMPR_ALG_LIBZ)
EndCompressorZlib(AH, cs);
#endif
+#ifdef HAVE_LIBZSTD
+ if (cs->comprAlg == COMPR_ALG_ZSTD)
+ EndCompressorZstd(AH, cs);
+#endif
+
free(cs);
}
/* Private routines, specific to each compression method. */
+// XXX: put in separate files ?
+
+#ifdef HAVE_LIBZSTD
+static void
+InitCompressorZstd(CompressorState *cs, int level)
+{
+ cs->u.zstd.cstream = ZSTD_createCStream();
+ if (cs->u.zstd.cstream == NULL)
+ fatal("could not initialize compression library");
+
+ /* XXX: initialize safely like the corresponding zlib "paranoia" */
+ cs->u.zstd.output.size = ZSTD_CStreamOutSize();
+ cs->u.zstd.output.dst = pg_malloc(cs->u.zstd.output.size);
+ cs->u.zstd.output.pos = 0;
+}
+
+static void
+EndCompressorZstd(ArchiveHandle *AH, CompressorState *cs)
+{
+ ZSTD_outBuffer *output = &cs->u.zstd.output;
+
+ for (;;)
+ {
+ size_t res;
+
+ res = ZSTD_compressStream2(cs->u.zstd.cstream, output,
+ &cs->u.zstd.input, ZSTD_e_end);
+
+ if (output->pos > 0)
+ cs->writeF(AH, output->dst, output->pos);
+
+ if (res == 0)
+ break;
+
+ if (ZSTD_isError(res))
+ fatal("could not close compression stream: %s",
+ ZSTD_getErrorName(res));
+ }
+
+ // XXX: retval
+ ZSTD_freeCStream(cs->u.zstd.cstream);
+}
+
+static void
+DeflateCompressorZstd(ArchiveHandle *AH, CompressorState *cs)
+{
+ ZSTD_inBuffer *input = &cs->u.zstd.input;
+ ZSTD_outBuffer *output = &cs->u.zstd.output;
+
+ while (input->pos != input->size)
+ {
+ size_t res;
+
+ res = ZSTD_compressStream2(cs->u.zstd.cstream, output,
+ input, ZSTD_e_continue);
+
+ if (output->pos == output->size ||
+ input->pos != input->size)
+ {
+ /*
+ * 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 (ZSTD_isError(res))
+ fatal("could not compress data: %s", ZSTD_getErrorName(res));
+ }
+}
+
+static void
+WriteDataToArchiveZstd(ArchiveHandle *AH, CompressorState *cs,
+ const char *data, size_t dLen)
+{
+ cs->u.zstd.input.src = (void *) unconstify(char *, data);
+ cs->u.zstd.input.size = dLen;
+ cs->u.zstd.input.pos = 0;
+ DeflateCompressorZstd(AH, cs);
+}
+
+/* Read data from a compressed zstd archive */
+static void
+ReadDataFromArchiveZstd(ArchiveHandle *AH, ReadFunc readF)
+{
+ ZSTD_DStream *dstream;
+ ZSTD_outBuffer output;
+ ZSTD_inBuffer input;
+ size_t res;
+ size_t input_size;
+
+ dstream = ZSTD_createDStream();
+ if (dstream == NULL)
+ 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;
+
+ input.size = input_size; // XXX: the buffer can grow, we shouldn't keep resetting it to the original value..
+ cnt = readF(AH, (char **)unconstify(void **, &input.src), &input.size);
+ input.pos = 0;
+ input.size = cnt;
+
+ if (cnt == 0)
+ break;
+
+ while (input.pos < input.size)
+ {
+ /* decompress */
+ output.pos = 0;
+ res = ZSTD_decompressStream(dstream, &output, &input);
+
+ if (ZSTD_isError(res))
+ 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);
+ }
+ }
+
+ /* write any remainder to output handle */
+ /* XXX: is it needed? */
+ /* If `input.pos < input.size`, some input has not been consumed. */
+ /* But if `output.pos == output.size`, there might be some data left within internal buffers., */
+#if 0
+ while (input.pos < input.size || output.pos == output.size)
+ {
+ output.pos = 0;
+ res = ZSTD_decompressStream(dstream, &output, &input);
+ if (ZSTD_isError(res))
+ fatal("could not decompress data: %s", ZSTD_getErrorName(res));
+ ((char *)output.dst)[output.pos] = '\0';
+ ahwrite(output.dst, 1, output.pos, AH);
+ }
+#endif
+
+ pg_free(unconstify(void *, input.src));
+ pg_free(output.dst);
+}
+
+#endif /* HAVE_LIBZSTD */
#ifdef HAVE_LIBZ
+
/*
* Functions for zlib compressed output.
*/
@@ -422,6 +643,19 @@ struct cfp
#ifdef HAVE_LIBZ
gzFile compressedfp;
#endif
+
+#ifdef HAVE_LIBZSTD // XXX: this should be a union with a CompressionAlgorithm alg?
+ /* This is a normal file to which we read/write compressed data */
+ struct {
+ 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;
+ } zstd;
+#endif
+
};
#ifdef HAVE_LIBZ
@@ -449,24 +683,25 @@ free_keep_errno(void *p)
* On failure, return NULL with an error code in errno.
*/
cfp *
-cfopen_read(const char *path, const char *mode)
+cfopen_read(const char *path, const char *mode, int compression)
{
cfp *fp;
#ifdef HAVE_LIBZ
- if (hasSuffix(path, ".gz"))
- fp = cfopen(path, mode, 1);
+ if (hasSuffix(path, ".gz") || hasSuffix(path, ".zst"))
+ fp = cfopen(path, mode, compression);
else
#endif
{
- fp = cfopen(path, mode, 0);
+ fp = cfopen(path, mode, compression);
#ifdef HAVE_LIBZ
if (fp == NULL)
{
char *fname;
+ char *suffix = compression == ZSTD_COMPRESSION ? "zst" : "gz";
- fname = psprintf("%s.gz", path);
- fp = cfopen(fname, mode, 1);
+ fname = psprintf("%s.%s", path, suffix);
+ fp = cfopen(fname, mode, compression);
free_keep_errno(fname);
}
#endif
@@ -491,13 +726,14 @@ cfopen_write(const char *path, const char *mode, int compression)
cfp *fp;
if (compression == 0)
- fp = cfopen(path, mode, 0);
+ fp = cfopen(path, mode, compression);
else
{
-#ifdef HAVE_LIBZ
+#ifdef HAVE_LIBZ // XXX
char *fname;
+ char *suffix = compression == ZSTD_COMPRESSION ? "zst" : "gz";
- fname = psprintf("%s.gz", path);
+ fname = psprintf("%s.%s", path, suffix);
fp = cfopen(fname, mode, compression);
free_keep_errno(fname);
#else
@@ -505,11 +741,12 @@ cfopen_write(const char *path, const char *mode, int compression)
fp = NULL; /* keep compiler quiet */
#endif
}
+
return fp;
}
/*
- * Opens file 'path' in 'mode'. If 'compression' is non-zero, the file
+ * Opens file 'path' in 'mode'. If 'alg' is COMPR_ALG_ZLIB, the file
* is opened with libz gzopen(), otherwise with plain fopen().
*
* On failure, return NULL with an error code in errno.
@@ -519,9 +756,19 @@ cfopen(const char *path, const char *mode, int compression)
{
cfp *fp = pg_malloc(sizeof(cfp));
- if (compression != 0)
+ fp->uncompressedfp = NULL;
+#ifdef HAVE_LIBZ
+ fp->compressedfp = NULL;
+#endif
+#ifdef HAVE_LIBZSTD
+ fp->zstd.fp = NULL;
+#endif
+
+ switch (compression)
{
#ifdef HAVE_LIBZ
+ case 1 ... 9: // XXX: nonportable
+ case Z_DEFAULT_COMPRESSION:
if (compression != Z_DEFAULT_COMPRESSION)
{
/* user has specified a compression level, so tell zlib to use it */
@@ -537,30 +784,57 @@ cfopen(const char *path, const char *mode, int compression)
fp->compressedfp = gzopen(path, mode);
}
- fp->uncompressedfp = NULL;
if (fp->compressedfp == NULL)
{
free_keep_errno(fp);
fp = NULL;
}
-#else
- fatal("not built with zlib support");
+
+ return fp;
+ break;
#endif
- }
- else
- {
-#ifdef HAVE_LIBZ
- fp->compressedfp = NULL;
+
+#ifdef HAVE_LIBZSTD
+ case ZSTD_COMPRESSION:
+ fp->zstd.fp = fopen(path, mode);
+ // XXX: save the compression params
+ if (fp->zstd.fp == NULL)
+ {
+ free_keep_errno(fp);
+ fp = NULL;
+ }
+ else if (strchr(mode, 'w'))
+ {
+ fp->zstd.dstream = NULL;
+ fp->zstd.output.size = ZSTD_CStreamOutSize(); // XXX
+ fp->zstd.output.dst = pg_malloc0(fp->zstd.output.size);
+ fp->zstd.cstream = ZSTD_createCStream();
+ if (fp->zstd.cstream == NULL)
+ fatal("could not initialize compression library");
+ }
+ else if (strchr(mode, 'r'))
+ {
+ fp->zstd.cstream = NULL;
+ fp->zstd.input.size = ZSTD_DStreamOutSize(); // XXX
+ fp->zstd.input.src = pg_malloc0(fp->zstd.input.size);
+ fp->zstd.dstream = ZSTD_createDStream();
+ if (fp->zstd.dstream == NULL)
+ fatal("could not initialize compression library");
+ } // XXX else: bad mode
+ return fp;
+ break;
#endif
+
+ default:
fp->uncompressedfp = fopen(path, mode);
if (fp->uncompressedfp == NULL)
{
free_keep_errno(fp);
fp = NULL;
}
- }
- return fp;
+ return fp;
+ }
}
@@ -587,6 +861,44 @@ cfread(void *ptr, int size, cfp *fp)
}
else
#endif
+
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ ZSTD_outBuffer *output = &fp->zstd.output;
+ ZSTD_inBuffer *input = &fp->zstd.input;
+ size_t input_size = ZSTD_DStreamInSize();
+ size_t res, cnt;
+
+ output->size = size;
+ output->dst = ptr;
+ output->pos = 0;
+
+ /* read compressed data */
+ while ((cnt = fread(unconstify(void *, input->src), 1, input_size, fp->zstd.fp)))
+ {
+ input->size = cnt;
+ input->pos = 0;
+
+ for ( ; input->pos < input->size; )
+ {
+ /* decompress */
+ res = ZSTD_decompressStream(fp->zstd.dstream, output, input);
+ if (res == 0 || output->pos == output->size)
+ break;
+ if (ZSTD_isError(res))
+ fatal("could not decompress data: %s", ZSTD_getErrorName(res));
+ }
+
+ if (output->pos == output->size)
+ break; /* We read all the data that fits */
+ }
+
+ ret = output->pos;
+ }
+ else
+#endif
+
{
ret = fread(ptr, 1, size, fp->uncompressedfp);
if (ret != size && !feof(fp->uncompressedfp))
@@ -603,6 +915,35 @@ cfwrite(const void *ptr, int size, cfp *fp)
return gzwrite(fp->compressedfp, ptr, size);
else
#endif
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ size_t res, cnt;
+ ZSTD_outBuffer *output = &fp->zstd.output;
+ ZSTD_inBuffer *input = &fp->zstd.input;
+
+ input->src = ptr;
+ input->size = size;
+ input->pos = 0;
+
+ /* Consume all input, and flush later */
+ while (input->pos != input->size)
+ {
+ output->pos = 0;
+ res = ZSTD_compressStream2(fp->zstd.cstream, output, input, ZSTD_e_continue);
+ if (ZSTD_isError(res))
+ fatal("could not compress data: %s", ZSTD_getErrorName(res));
+
+ cnt = fwrite(output->dst, 1, output->pos, fp->zstd.fp);
+ if (cnt != output->pos)
+ fatal("could not write data: %s", strerror(errno));
+ }
+
+ return size;
+ }
+ else
+#endif
+
return fwrite(ptr, 1, size, fp->uncompressedfp);
}
@@ -625,6 +966,20 @@ cfgetc(cfp *fp)
}
else
#endif
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ if (cfread(&ret, 1, fp) != 1)
+ {
+ if (feof(fp->zstd.fp))
+ fatal("could not read from input file: end of file");
+ else
+ fatal("could not read from input file: %s", strerror(errno));
+ }
+fprintf(stderr, "cfgetc %d\n", ret);
+ }
+#endif
+
{
ret = fgetc(fp->uncompressedfp);
if (ret == EOF)
@@ -641,6 +996,18 @@ cfgets(cfp *fp, char *buf, int len)
if (fp->compressedfp)
return gzgets(fp->compressedfp, buf, len);
else
+#endif
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ int res;
+ res = cfread(buf, len, fp);
+ buf[res] = 0;
+ if (strchr(buf, '\n'))
+ *strchr(buf, '\n') = '\0';
+ return buf;
+ }
+ else
#endif
return fgets(buf, len, fp->uncompressedfp);
}
@@ -662,6 +1029,43 @@ cfclose(cfp *fp)
fp->compressedfp = NULL;
}
else
+#endif
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ ZSTD_outBuffer *output = &fp->zstd.output;
+ ZSTD_inBuffer *input = &fp->zstd.input;
+ size_t res, cnt;
+
+ if (fp->zstd.cstream)
+ {
+ for (;;)
+ {
+ output->pos = 0;
+ res = ZSTD_compressStream2(fp->zstd.cstream, output, input, ZSTD_e_end);
+ if (ZSTD_isError(res))
+ fatal("could not compress data: %s", ZSTD_getErrorName(res));
+ cnt = fwrite(output->dst, 1, output->pos, fp->zstd.fp);
+ if (cnt != output->pos)
+ fatal("could not write data: %s", strerror(errno));
+ if (res == 0)
+ break;
+ }
+
+ ZSTD_freeCStream(fp->zstd.cstream);
+ pg_free(fp->zstd.output.dst);
+ }
+
+ if (fp->zstd.dstream)
+ {
+ ZSTD_freeDStream(fp->zstd.dstream);
+ pg_free(unconstify(void *, fp->zstd.input.src));
+ }
+
+ result = fclose(fp->zstd.fp);
+ fp->zstd.fp = NULL;
+ }
+ else
#endif
{
result = fclose(fp->uncompressedfp);
@@ -679,6 +1083,11 @@ cfeof(cfp *fp)
if (fp->compressedfp)
return gzeof(fp->compressedfp);
else
+#endif
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ return feof(fp->zstd.fp);
+ else
#endif
return feof(fp->uncompressedfp);
}
diff --git a/src/bin/pg_dump/compress_io.h b/src/bin/pg_dump/compress_io.h
index d2e6e1b854..f0ce06f176 100644
--- a/src/bin/pg_dump/compress_io.h
+++ b/src/bin/pg_dump/compress_io.h
@@ -24,7 +24,8 @@
typedef enum
{
COMPR_ALG_NONE,
- COMPR_ALG_LIBZ
+ COMPR_ALG_LIBZ,
+ COMPR_ALG_ZSTD,
} CompressionAlgorithm;
/* Prototype for callback function to WriteDataToArchive() */
@@ -57,7 +58,7 @@ extern void EndCompressor(ArchiveHandle *AH, CompressorState *cs);
typedef struct cfp cfp;
extern cfp *cfopen(const char *path, const char *mode, int compression);
-extern cfp *cfopen_read(const char *path, const char *mode);
+extern cfp *cfopen_read(const char *path, const char *mode, int compression);
extern cfp *cfopen_write(const char *path, const char *mode, int compression);
extern int cfread(void *ptr, int size, cfp *fp);
extern int cfwrite(const void *ptr, int size, cfp *fp);
diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h
index 6a5a22637b..1a229ebedb 100644
--- a/src/bin/pg_dump/pg_backup_archiver.h
+++ b/src/bin/pg_dump/pg_backup_archiver.h
@@ -60,6 +60,11 @@ typedef struct _z_stream
typedef z_stream *z_streamp;
#endif
+#ifdef HAVE_LIBZSTD
+#include <zstd.h>
+#define ZSTD_COMPRESSION -2
+#endif /* HAVE_LIBZSTD */
+
/* Data block types */
#define BLK_DATA 1
#define BLK_BLOBS 3
diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c
index 650b542fce..6e55cb425e 100644
--- a/src/bin/pg_dump/pg_backup_directory.c
+++ b/src/bin/pg_dump/pg_backup_directory.c
@@ -202,7 +202,7 @@ InitArchiveFmt_Directory(ArchiveHandle *AH)
setFilePath(AH, fname, "toc.dat");
- tocFH = cfopen_read(fname, PG_BINARY_R);
+ tocFH = cfopen_read(fname, PG_BINARY_R, AH->compression);
if (tocFH == NULL)
fatal("could not open input file \"%s\": %m", fname);
@@ -388,7 +388,7 @@ _PrintFileData(ArchiveHandle *AH, char *filename)
if (!filename)
return;
- cfp = cfopen_read(filename, PG_BINARY_R);
+ cfp = cfopen_read(filename, PG_BINARY_R, AH->compression);
if (!cfp)
fatal("could not open input file \"%s\": %m", filename);
@@ -440,7 +440,7 @@ _LoadBlobs(ArchiveHandle *AH)
setFilePath(AH, fname, "blobs.toc");
- ctx->blobsTocFH = cfopen_read(fname, PG_BINARY_R);
+ ctx->blobsTocFH = cfopen_read(fname, PG_BINARY_R, AH->compression);
if (ctx->blobsTocFH == NULL)
fatal("could not open large object TOC file \"%s\" for input: %m",
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 8b1e5cc2b5..f21eb021c7 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -533,7 +533,8 @@ main(int argc, char **argv)
case 'Z': /* Compression Level */
compressLevel = atoi(optarg);
- if (compressLevel < 0 || compressLevel > 9)
+ if ((compressLevel < 0 || compressLevel > 9) &&
+ compressLevel != -2)
{
pg_log_error("compression level must be in range 0..9");
exit_nicely(1);
diff --git a/src/include/pg_config.h.in b/src/include/pg_config.h.in
index de8f838e53..da35415c72 100644
--- a/src/include/pg_config.h.in
+++ b/src/include/pg_config.h.in
@@ -346,6 +346,9 @@
/* Define to 1 if you have the `z' library (-lz). */
#undef HAVE_LIBZ
+/* Define to 1 if you have the `zstd' library (-lzstd). */
+#undef HAVE_LIBZSTD
+
/* Define to 1 if you have the `link' function. */
#undef HAVE_LINK
--
2.17.0
--bgLLobvf7eP6VP5c
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="0004-cmdline-parser-for-compression-alg-level-opts.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 3/7] pg_dump: zstd compression
In-Reply-To: <no-message-id-175480@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