From f9576c60b6f14defd05c72d753435563553ac908 Mon Sep 17 00:00:00 2001 From: Georgios Kokolatos Date: Wed, 23 Feb 2022 14:40:33 +0000 Subject: [PATCH v1 2/2] Add LZ4 compression in pg_{dump|restore} Within compress_io.{c,h} there are two distinct APIs exposed, the streaming API and a file API. The first one, is aimed at inlined use cases and thus simple lz4.h calls can be used directly. The second one is generating output, or is parsing input, which can be read/generated via the lz4 utility. In the later case, the API is using an opaque wrapper around a file stream, which aquired via fopen() or gzopen() respectively. It would then provide wrappers around fread(), fwrite(), fgets(), fgetc(), feof(), and fclose(); or their gz equivallents. However the LZ4F api does not provide this functionality. So this has been implemented localy. In order to maintain the API compatibility a new structure LZ4File is introduced. It is responsible for keeping state and any yet unused generated content. The later is required when the generated decompressed output, exceeds the caller's buffer capacity. Custom compressed archives need to now store the compression method in their header. This requires a bump in the version number. The level of compression is still stored in the dump, though admittedly is of no apparent use. --- doc/src/sgml/ref/pg_dump.sgml | 23 +- src/bin/pg_dump/Makefile | 1 + src/bin/pg_dump/compress_io.c | 738 ++++++++++++++++++++++++--- src/bin/pg_dump/pg_backup.h | 1 + src/bin/pg_dump/pg_backup_archiver.c | 71 ++- src/bin/pg_dump/pg_backup_archiver.h | 3 +- src/bin/pg_dump/pg_dump.c | 12 +- src/bin/pg_dump/t/001_basic.pl | 4 +- src/bin/pg_dump/t/002_pg_dump.pl | 141 ++++- 9 files changed, 868 insertions(+), 126 deletions(-) diff --git a/doc/src/sgml/ref/pg_dump.sgml b/doc/src/sgml/ref/pg_dump.sgml index 992b7312df..68a7c6a3bf 100644 --- a/doc/src/sgml/ref/pg_dump.sgml +++ b/doc/src/sgml/ref/pg_dump.sgml @@ -328,9 +328,10 @@ PostgreSQL documentation machine-readable format that pg_restore 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 gzip tool. - This format is compressed by default and also supports parallel - dumps. + can be compressed with the gzip or + lz4tool. + This format is compressed by default using gzip + and also supports parallel dumps. @@ -652,12 +653,12 @@ PostgreSQL documentation Specify the compression method and/or the compression level to use. The compression method can be set to gzip or - none for no compression. A compression level can - be optionally specified, by appending the level number after a colon - (:). If no level is specified, the default compression - level will be used for the specified method. If only a level is - specified without mentioning a method, gzip compression - will be used. + lz4 or none for no compression. A + compression level can be optionally specified, by appending the level + number after a colon (:). If no level is specified, + the default compression level will be used for the specified method. If + only a level is specified without mentioning a method, + gzip compression willbe used. @@ -665,8 +666,8 @@ PostgreSQL documentation individual table-data segments, and the default is to compress using gzip at a moderate level. For plain text output, setting a nonzero compression level causes the entire output file to be compressed, - as though it had been fed through gzip; but the default - is not to compress. + as though it had been fed through gzip or + lz4; but the default is not to compress. The tar archive format currently does not support compression at all. diff --git a/src/bin/pg_dump/Makefile b/src/bin/pg_dump/Makefile index 56e041c9b3..625c00b123 100644 --- a/src/bin/pg_dump/Makefile +++ b/src/bin/pg_dump/Makefile @@ -17,6 +17,7 @@ top_builddir = ../../.. include $(top_builddir)/src/Makefile.global export GZIP_PROGRAM=$(GZIP) +export LZ4 override CPPFLAGS := -I$(libpq_srcdir) $(CPPFLAGS) LDFLAGS_INTERNAL += -L$(top_builddir)/src/fe_utils -lpgfeutils $(libpq_pgport) diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c index 630f9e4b18..790f225ec4 100644 --- a/src/bin/pg_dump/compress_io.c +++ b/src/bin/pg_dump/compress_io.c @@ -38,13 +38,15 @@ * ---------------------- * * The compressed stream API is a wrapper around the C standard fopen() and - * 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 - * 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. + * libz's gzopen() APIs and custom LZ4 calls which provide similar + * functionality. 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, + * failing that it tries to open the same file with the .lz4 suffix. + * cfopen_write() opens a file for writing, an extra argument specifies the + * method to use should the file be compressed, and adds the appropriate + * suffix, .gz or .lz4, to the filename if so. This allows you to easily handle + * both compressed and uncompressed files. * * IDENTIFICATION * src/bin/pg_dump/compress_io.c @@ -56,6 +58,14 @@ #include "compress_io.h" #include "pg_backup_utils.h" +#ifdef HAVE_LIBLZ4 +#include "lz4.h" +#include "lz4frame.h" + +#define LZ4_OUT_SIZE (4 * 1024) +#define LZ4_IN_SIZE (16 * 1024) +#endif + /*---------------------- * Compressor API *---------------------- @@ -69,9 +79,9 @@ struct CompressorState #ifdef HAVE_LIBZ z_streamp zp; - char *zlibOut; - size_t zlibOutSize; #endif + void *outbuf; + size_t outsize; }; /* Routines that support zlib compressed data I/O */ @@ -85,6 +95,15 @@ static void WriteDataToArchiveZlib(ArchiveHandle *AH, CompressorState *cs, static void EndCompressorZlib(ArchiveHandle *AH, CompressorState *cs); #endif +/* Routines that support LZ4 compressed data I/O */ +#ifdef HAVE_LIBLZ4 +static void InitCompressorLZ4(CompressorState *cs); +static void ReadDataFromArchiveLZ4(ArchiveHandle *AH, ReadFunc readF); +static void WriteDataToArchiveLZ4(ArchiveHandle *AH, CompressorState *cs, + const char *data, size_t dLen); +static void EndCompressorLZ4(ArchiveHandle *AH, CompressorState *cs); +#endif + /* Routines that support uncompressed data I/O */ static void ReadDataFromArchiveNone(ArchiveHandle *AH, ReadFunc readF); static void WriteDataToArchiveNone(ArchiveHandle *AH, CompressorState *cs, @@ -103,6 +122,11 @@ AllocateCompressor(CompressionMethod compressionMethod, if (compressionMethod == COMPRESSION_GZIP) fatal("not built with zlib support"); #endif +#ifndef HAVE_LIBLZ4 + if (compressionMethod == COMPRESSION_LZ4) + fatal("not built with LZ4 support"); +#endif + cs = (CompressorState *) pg_malloc0(sizeof(CompressorState)); cs->writeF = writeF; @@ -115,6 +139,10 @@ AllocateCompressor(CompressionMethod compressionMethod, if (compressionMethod == COMPRESSION_GZIP) InitCompressorZlib(cs, compressionLevel); #endif +#ifdef HAVE_LIBLZ4 + if (compressionMethod == COMPRESSION_LZ4) + InitCompressorLZ4(cs); +#endif return cs; } @@ -137,6 +165,13 @@ ReadDataFromArchive(ArchiveHandle *AH, CompressionMethod compressionMethod, ReadDataFromArchiveZlib(AH, readF); #else fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + ReadDataFromArchiveLZ4(AH, readF); +#else + fatal("not built with lz4 support"); #endif break; default: @@ -159,6 +194,13 @@ WriteDataToArchive(ArchiveHandle *AH, CompressorState *cs, WriteDataToArchiveZlib(AH, cs, data, dLen); #else fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + WriteDataToArchiveLZ4(AH, cs, data, dLen); +#else + fatal("not built with lz4 support"); #endif break; case COMPRESSION_NONE: @@ -183,6 +225,13 @@ EndCompressor(ArchiveHandle *AH, CompressorState *cs) EndCompressorZlib(AH, cs); #else fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + EndCompressorLZ4(AH, cs); +#else + fatal("not built with lz4 support"); #endif break; case COMPRESSION_NONE: @@ -213,20 +262,20 @@ InitCompressorZlib(CompressorState *cs, int level) zp->opaque = Z_NULL; /* - * zlibOutSize is the buffer size we tell zlib it can output to. We + * outsize is the buffer size we tell zlib it can output to. We * actually allocate one extra byte because some routines want to append a * trailing zero byte to the zlib output. */ - cs->zlibOut = (char *) pg_malloc(ZLIB_OUT_SIZE + 1); - cs->zlibOutSize = ZLIB_OUT_SIZE; + cs->outbuf = pg_malloc(ZLIB_OUT_SIZE + 1); + cs->outsize = ZLIB_OUT_SIZE; if (deflateInit(zp, level) != Z_OK) fatal("could not initialize compression library: %s", zp->msg); /* Just be paranoid - maybe End is called after Start, with no Write */ - zp->next_out = (void *) cs->zlibOut; - zp->avail_out = cs->zlibOutSize; + zp->next_out = cs->outbuf; + zp->avail_out = cs->outsize; } static void @@ -243,7 +292,7 @@ EndCompressorZlib(ArchiveHandle *AH, CompressorState *cs) if (deflateEnd(zp) != Z_OK) fatal("could not close compression stream: %s", zp->msg); - free(cs->zlibOut); + free(cs->outbuf); free(cs->zp); } @@ -251,7 +300,7 @@ static void DeflateCompressorZlib(ArchiveHandle *AH, CompressorState *cs, bool flush) { z_streamp zp = cs->zp; - char *out = cs->zlibOut; + void *out = cs->outbuf; int res = Z_OK; while (cs->zp->avail_in != 0 || flush) @@ -259,7 +308,7 @@ DeflateCompressorZlib(ArchiveHandle *AH, CompressorState *cs, bool flush) res = deflate(zp, flush ? Z_FINISH : Z_NO_FLUSH); if (res == Z_STREAM_ERROR) fatal("could not compress data: %s", zp->msg); - if ((flush && (zp->avail_out < cs->zlibOutSize)) + if ((flush && (zp->avail_out < cs->outsize)) || (zp->avail_out == 0) || (zp->avail_in != 0) ) @@ -269,18 +318,18 @@ DeflateCompressorZlib(ArchiveHandle *AH, CompressorState *cs, bool flush) * chunk is the EOF marker in the custom format. This should never * happen but... */ - if (zp->avail_out < cs->zlibOutSize) + if (zp->avail_out < cs->outsize) { /* * Any write function should do its own error checking but to * make sure we do a check here as well... */ - size_t len = cs->zlibOutSize - zp->avail_out; + size_t len = cs->outsize - zp->avail_out; - cs->writeF(AH, out, len); + cs->writeF(AH, (char *)out, len); } - zp->next_out = (void *) out; - zp->avail_out = cs->zlibOutSize; + zp->next_out = out; + zp->avail_out = cs->outsize; } if (res == Z_STREAM_END) @@ -364,6 +413,71 @@ ReadDataFromArchiveZlib(ArchiveHandle *AH, ReadFunc readF) } #endif /* HAVE_LIBZ */ +#ifdef HAVE_LIBLZ4 +static void +InitCompressorLZ4(CompressorState *cs) +{ + /* Will be lazy init'd */ + cs->outbuf = NULL; + cs->outsize = 0; +} + +static void +EndCompressorLZ4(ArchiveHandle *AH, CompressorState *cs) +{ + pg_free(cs->outbuf); + + cs->outbuf = NULL; + cs->outsize = 0; +} + +static void +ReadDataFromArchiveLZ4(ArchiveHandle *AH, ReadFunc readF) +{ + LZ4_streamDecode_t lz4StreamDecode; + char *buf; + char *decbuf; + size_t buflen; + size_t cnt; + + buflen = (4 * 1024) + 1; + buf = pg_malloc(buflen); + decbuf = pg_malloc(buflen); + + LZ4_setStreamDecode(&lz4StreamDecode, NULL, 0); + + while ((cnt = readF(AH, &buf, &buflen))) + { + int decBytes = LZ4_decompress_safe_continue(&lz4StreamDecode, + buf, decbuf, + cnt, buflen); + + ahwrite(decbuf, 1, decBytes, AH); + } +} + +static void +WriteDataToArchiveLZ4(ArchiveHandle *AH, CompressorState *cs, + const char *data, size_t dLen) +{ + size_t compressed; + size_t requiredsize = LZ4_compressBound(dLen); + + if (requiredsize > cs->outsize) + { + cs->outbuf = pg_realloc(cs->outbuf, requiredsize); + cs->outsize = requiredsize; + } + + compressed = LZ4_compress_default(data, cs->outbuf, + dLen, cs->outsize); + + if (compressed <= 0) + fatal("failed to LZ4 compress data"); + + cs->writeF(AH, cs->outbuf, compressed); +} +#endif /* HAVE_LIBLZ4 */ /* * Functions for uncompressed output. @@ -400,9 +514,36 @@ WriteDataToArchiveNone(ArchiveHandle *AH, CompressorState *cs, *---------------------- */ +#ifdef HAVE_LIBLZ4 /* - * cfp represents an open stream, wrapping the underlying FILE or gzFile - * pointer. This is opaque to the callers. + * State needed for LZ4 (de)compression using the cfp API. + */ +typedef struct LZ4File +{ + FILE *fp; + + LZ4F_preferences_t prefs; + + LZ4F_compressionContext_t ctx; + LZ4F_decompressionContext_t dtx; + + bool inited; + bool compressing; + + size_t buflen; + char *buffer; + + size_t overflowalloclen; + size_t overflowlen; + char *overflowbuf; + + size_t errcode; +} LZ4File; +#endif + +/* + * cfp represents an open stream, wrapping the underlying FILE, gzFile + * pointer, or LZ4File pointer. This is opaque to the callers. */ struct cfp { @@ -410,9 +551,7 @@ struct cfp void *fp; }; -#ifdef HAVE_LIBZ static int hasSuffix(const char *filename, const char *suffix); -#endif /* free() without changing errno; useful in several places below */ static void @@ -424,26 +563,380 @@ free_keep_errno(void *p) errno = save_errno; } +#ifdef HAVE_LIBLZ4 +/* + * LZ4 equivalent to feof() or gzeof(). The end of file + * is reached if there is no decompressed output in the + * overflow buffer and the end of the file is reached. + */ +static int +LZ4File_eof(LZ4File *fs) +{ + return fs->overflowlen == 0 && feof(fs->fp); +} + +static const char * +LZ4File_error(LZ4File *fs) +{ + const char *errmsg; + + if (LZ4F_isError(fs->errcode)) + errmsg = LZ4F_getErrorName(fs->errcode); + else + errmsg = strerror(errno); + + return errmsg; +} + +/* + * Prepare an already alloc'ed LZ4File struct for subsequent calls. + * + * It creates the nessary contexts for the operations. When compressing, + * it additionally writes the LZ4 header in the output stream. + */ +static int +LZ4File_init(LZ4File *fs, int size, bool compressing) +{ + size_t status; + + if (fs->inited) + return 0; + + fs->compressing = compressing; + fs->inited = true; + + if (fs->compressing) + { + fs->buflen = LZ4F_compressBound(LZ4_IN_SIZE, &fs->prefs); + if (fs->buflen < LZ4F_HEADER_SIZE_MAX) + fs->buflen = LZ4F_HEADER_SIZE_MAX; + + status = LZ4F_createCompressionContext(&fs->ctx, LZ4F_VERSION); + if (LZ4F_isError(status)) + { + fs->errcode = status; + return 1; + } + + fs->buffer = pg_malloc(fs->buflen); + status = LZ4F_compressBegin(fs->ctx, fs->buffer, fs->buflen, &fs->prefs); + + if (LZ4F_isError(status)) + { + fs->errcode = status; + return 1; + } + + if (fwrite(fs->buffer, 1, status, fs->fp) != status) + { + errno = errno ? : ENOSPC; + return 1; + } + } + else + { + status = LZ4F_createDecompressionContext(&fs->dtx, LZ4F_VERSION); + if (LZ4F_isError(status)) + { + fs->errcode = status; + return 1; + } + + fs->buflen = size > LZ4_OUT_SIZE ? size : LZ4_OUT_SIZE; + fs->buffer = pg_malloc(fs->buflen); + + fs->overflowalloclen = fs->buflen; + fs->overflowbuf = pg_malloc(fs->overflowalloclen); + fs->overflowlen = 0; + } + + return 0; +} + +/* + * Read already decompressed content from the overflow buffer into 'ptr' up to + * 'size' bytes, if available. If the eol_flag is set, then stop at the first + * occurance of the new line char prior to 'size' bytes. + * + * Any unread content in the overflow buffer, is moved to the beginning. + */ +static int +LZ4File_read_overflow(LZ4File *fs, void *ptr, int size, bool eol_flag) +{ + char *p; + int readlen = 0; + + if (fs->overflowlen == 0) + return 0; + + if (fs->overflowlen >= size) + readlen = size; + else + readlen = fs->overflowlen; + + if (eol_flag && (p = memchr(fs->overflowbuf, '\n', readlen))) + readlen = p - fs->overflowbuf + 1; /* Include the line terminating char */ + + memcpy(ptr, fs->overflowbuf, readlen); + fs->overflowlen -= readlen; + + if (fs->overflowlen > 0) + memmove(fs->overflowbuf, fs->overflowbuf + readlen, fs->overflowlen); + + return readlen; +} + +/* + * The workhorse for reading decompressed content out of an LZ4 compressed + * stream. + * + * It will read up to 'ptrsize' decompressed content, or up to the new line char + * if found first when the eol_flag is set. It is possible that the decompressed + * output generated by reading any compressed input via the LZ4F API, exceeds + * 'ptrsize'. Any exceeding decompressed content is stored at an overflow + * buffer within LZ4File. Of course, when the function is called, it will first + * try to consume any decompressed content already present in the overflow + * buffer, before decompressing new content. + */ +static int +LZ4File_read(LZ4File *fs, void *ptr, int ptrsize, bool eol_flag) +{ + size_t dsize = 0; + size_t rsize; + size_t size = ptrsize; + bool eol_found = false; + + void *readbuf; + + /* Lazy init */ + if (!fs->inited && LZ4File_init(fs, size, false /* decompressing */)) + return -1; + + /* Verfiy that there is enough space in the outbuf */ + if (size > fs->buflen) + { + fs->buflen = size; + fs->buffer = pg_realloc(fs->buffer, size); + } + + /* use already decompressed content if available */ + dsize = LZ4File_read_overflow(fs, ptr, size, eol_flag); + if (dsize == size || (eol_flag && memchr(ptr, '\n', dsize))) + return dsize; + + readbuf = pg_malloc(size); + + do + { + char *rp; + char *rend; + + rsize = fread(readbuf, 1, size, fs->fp); + if (rsize < size && !feof(fs->fp)) + return -1; + + rp = (char *)readbuf; + rend = (char *)readbuf + rsize; + + while (rp < rend) + { + size_t status; + size_t outlen = fs->buflen; + size_t read_remain = rend - rp; + + memset(fs->buffer, 0, outlen); + status = LZ4F_decompress(fs->dtx, fs->buffer, &outlen, + rp, &read_remain, NULL); + if (LZ4F_isError(status)) + { + fs->errcode = status; + return -1; + } + + rp += read_remain; + + /* + * fill in what space is available in ptr + * if the eol flag is set, either skip if one already found or fill up to EOL + * if present in the outbuf + */ + if (outlen > 0 && dsize < size && eol_found == false) + { + char *p; + size_t lib = (eol_flag == 0) ? size - dsize : size -1 -dsize; + size_t len = outlen < lib ? outlen : lib; + + if (eol_flag == true && (p = memchr(fs->buffer, '\n', outlen)) && + (size_t)(p - fs->buffer + 1) <= len) + { + len = p - fs->buffer + 1; + eol_found = true; + } + + memcpy((char *)ptr + dsize, fs->buffer, len); + dsize += len; + + /* move what did not fit, if any, at the begining of the buf */ + if (len < outlen) + memmove(fs->buffer, fs->buffer + len, outlen - len); + outlen -= len; + } + + /* if there is available output, save it */ + if (outlen > 0) + { + while (fs->overflowlen + outlen > fs->overflowalloclen) + { + fs->overflowalloclen *= 2; + fs->overflowbuf = pg_realloc(fs->overflowbuf, fs->overflowalloclen); + } + + memcpy(fs->overflowbuf + fs->overflowlen, fs->buffer, outlen); + fs->overflowlen += outlen; + } + } + } while (rsize == size && dsize < size && eol_found == 0); + + pg_free(readbuf); + + return (int)dsize; +} + +/* + * Compress size bytes from ptr and write them to the stream. + */ +static int +LZ4File_write(LZ4File *fs, const void *ptr, int size) +{ + size_t status; + int remaining = size; + + if (!fs->inited && LZ4File_init(fs, size, true)) + return -1; + + while (remaining > 0) + { + int chunk = remaining < LZ4_IN_SIZE ? remaining : LZ4_IN_SIZE; + remaining -= chunk; + + status = LZ4F_compressUpdate(fs->ctx, fs->buffer, fs->buflen, + ptr, chunk, NULL); + if (LZ4F_isError(status)) + { + fs->errcode = status; + return -1; + } + + if (fwrite(fs->buffer, 1, status, fs->fp) != status) + { + errno = errno ? : ENOSPC; + return 1; + } + } + + return size; +} + +/* + * fgetc() and gzgetc() equivalent implementation for LZ4 compressed files. + */ +static int +LZ4File_getc(LZ4File *fs) +{ + unsigned char c; + + if (LZ4File_read(fs, &c, 1, false) != 1) + return EOF; + + return c; +} + +/* + * fgets() and gzgets() equivalent implementation for LZ4 compressed files. + */ +static char * +LZ4File_gets(LZ4File *fs, char *buf, int len) +{ + size_t dsize; + + dsize = LZ4File_read(fs, buf, len, true); + if (dsize < 0) + fatal("failed to read from archive %s", LZ4File_error(fs)); + + /* Done reading */ + if (dsize == 0) + return NULL; + + return buf; +} + +/* + * Finalize (de)compression of a stream. When compressing it will write any + * remaining content and/or generated footer from the LZ4 API. + */ +static int +LZ4File_close(LZ4File *fs) +{ + FILE *fp; + size_t status; + int ret; + + fp = fs->fp; + if (fs->inited) + { + if (fs->compressing) + { + status = LZ4F_compressEnd(fs->ctx, fs->buffer, fs->buflen, NULL); + if (LZ4F_isError(status)) + fatal("failed to end compression: %s", LZ4F_getErrorName(status)); + else if ((ret = fwrite(fs->buffer, 1, status, fs->fp)) != status) + { + errno = errno ? : ENOSPC; + WRITE_ERROR_EXIT; + } + + status = LZ4F_freeCompressionContext(fs->ctx); + if (LZ4F_isError(status)) + fatal("failed to end compression: %s", LZ4F_getErrorName(status)); + } + else + { + status = LZ4F_freeDecompressionContext(fs->dtx); + if (LZ4F_isError(status)) + fatal("failed to end decompression: %s", LZ4F_getErrorName(status)); + pg_free(fs->overflowbuf); + } + + pg_free(fs->buffer); + } + + pg_free(fs); + + return fclose(fp); +} +#endif + /* * Open a file for reading. 'path' is the file to open, and 'mode' should * be either "r" or "rb". * - * If the file at 'path' does not exist, we append the ".gz" suffix (if 'path' - * doesn't already have it) and try again. So if you pass "foo" as 'path', - * this will open either "foo" or "foo.gz". + * If the file at 'path' does not exist, we append the "{.gz,.lz4}" suffix (if + * 'path' doesn't already have it) and try again. So if you pass "foo" as 'path', + * this will open either "foo" or "foo.gz" or "foo.lz4", trying in that order. * * On failure, return NULL with an error code in errno. + * */ cfp * cfopen_read(const char *path, const char *mode) { - cfp *fp; + cfp *fp = NULL; -#ifdef HAVE_LIBZ if (hasSuffix(path, ".gz")) fp = cfopen(path, mode, COMPRESSION_GZIP, 0); + else if (hasSuffix(path, ".lz4")) + fp = cfopen(path, mode, COMPRESSION_LZ4, 0); else -#endif { fp = cfopen(path, mode, COMPRESSION_NONE, 0); #ifdef HAVE_LIBZ @@ -455,8 +948,19 @@ cfopen_read(const char *path, const char *mode) fp = cfopen(fname, mode, COMPRESSION_GZIP, 0); free_keep_errno(fname); } +#endif +#ifdef HAVE_LIBLZ4 + if (fp == NULL) + { + char *fname; + + fname = psprintf("%s.lz4", path); + fp = cfopen(fname, mode, COMPRESSION_LZ4, 0); + free_keep_errno(fname); + } #endif } + return fp; } @@ -465,9 +969,13 @@ cfopen_read(const char *path, const char *mode) * be a filemode as accepted by fopen() and gzopen() that indicates writing * ("w", "wb", "a", or "ab"). * - * If 'compression' is non-zero, a gzip compressed stream is opened, and - * 'compression' indicates the compression level used. The ".gz" suffix - * is automatically added to 'path' in that case. + * When 'compressionMethod' indicates gzip, a gzip compressed stream is opened, + * and 'compressionLevel' is used. The ".gz" suffix is automatically added to + * 'path' in that case. The same applies when 'compressionMethod' indicates lz4, + * but then the ".lz4" suffix is added instead. + * + * It is the caller's responsibility to verify that the requested + * 'compressionMethod' is supported by the build. * * On failure, return NULL with an error code in errno. */ @@ -476,23 +984,44 @@ cfopen_write(const char *path, const char *mode, CompressionMethod compressionMethod, int compressionLevel) { - cfp *fp; + cfp *fp = NULL; - if (compressionMethod == COMPRESSION_NONE) - fp = cfopen(path, mode, compressionMethod, 0); - else + switch (compressionMethod) { + case COMPRESSION_NONE: + fp = cfopen(path, mode, compressionMethod, 0); + break; + case COMPRESSION_GZIP: #ifdef HAVE_LIBZ - char *fname; + { + char *fname; - fname = psprintf("%s.gz", path); - fp = cfopen(fname, mode, compressionMethod, compressionLevel); - free_keep_errno(fname); + fname = psprintf("%s.gz", path); + fp = cfopen(fname, mode, compressionMethod, compressionLevel); + free_keep_errno(fname); + } #else - fatal("not built with zlib support"); - fp = NULL; /* keep compiler quiet */ + fatal("not built with zlib support"); #endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + { + char *fname; + + fname = psprintf("%s.lz4", path); + fp = cfopen(fname, mode, compressionMethod, compressionLevel); + free_keep_errno(fname); + } +#else + fatal("not built with LZ4 support"); +#endif + break; + default: + fatal("invalid compression method"); + break; } + return fp; } @@ -509,7 +1038,7 @@ static cfp * cfopen_internal(const char *path, int fd, const char *mode, CompressionMethod compressionMethod, int compressionLevel) { - cfp *fp = pg_malloc(sizeof(cfp)); + cfp *fp = pg_malloc0(sizeof(cfp)); fp->compressionMethod = compressionMethod; @@ -560,6 +1089,27 @@ cfopen_internal(const char *path, int fd, const char *mode, } #else fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + { + LZ4File *lz4fp = pg_malloc0(sizeof(*lz4fp)); + if (fd >= 0) + lz4fp->fp = fdopen(fd, mode); + else + lz4fp->fp = fopen(path, mode); + if (lz4fp->fp == NULL) + { + free_keep_errno(lz4fp); + fp = NULL; + } + if (compressionLevel >= 0) + lz4fp->prefs.compressionLevel = compressionLevel; + fp->fp = lz4fp; + } +#else + fatal("not built with LZ4 support"); #endif break; default: @@ -580,8 +1130,8 @@ cfopen(const char *path, const char *mode, cfp * cfdopen(int fd, const char *mode, - CompressionMethod compressionMethod, - int compressionLevel) + CompressionMethod compressionMethod, + int compressionLevel) { return cfopen_internal(NULL, fd, mode, compressionMethod, compressionLevel); } @@ -617,7 +1167,16 @@ cfread(void *ptr, int size, cfp *fp) fatal("not built with zlib support"); #endif break; - + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + ret = LZ4File_read(fp->fp, ptr, size, false); + if (ret != size && !LZ4File_eof(fp->fp)) + fatal("could not read from input file: %s", + LZ4File_error(fp->fp)); +#else + fatal("not built with LZ4 support"); +#endif + break; default: fatal("invalid compression method"); break; @@ -641,6 +1200,13 @@ cfwrite(const void *ptr, int size, cfp *fp) ret = gzwrite(fp->fp, ptr, size); #else fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + ret = LZ4File_write(fp->fp, ptr, size); +#else + fatal("not built with LZ4 support"); #endif break; default: @@ -676,6 +1242,20 @@ cfgetc(cfp *fp) } #else fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + ret = LZ4File_getc(fp->fp); + if (ret == EOF) + { + if (!LZ4File_eof(fp->fp)) + fatal("could not read from input file: %s", strerror(errno)); + else + fatal("could not read from input file: end of file"); + } +#else + fatal("not built with LZ4 support"); #endif break; default: @@ -695,13 +1275,19 @@ cfgets(cfp *fp, char *buf, int len) { case COMPRESSION_NONE: ret = fgets(buf, len, fp->fp); - break; case COMPRESSION_GZIP: #ifdef HAVE_LIBZ ret = gzgets(fp->fp, buf, len); #else fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + ret = LZ4File_gets(fp->fp, buf, len); +#else + fatal("not built with LZ4 support"); #endif break; default: @@ -736,6 +1322,14 @@ cfclose(cfp *fp) fp->fp = NULL; #else fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + ret = LZ4File_close(fp->fp); + fp->fp = NULL; +#else + fatal("not built with LZ4 support"); #endif break; default: @@ -764,6 +1358,13 @@ cfeof(cfp *fp) ret = gzeof(fp->fp); #else fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + ret = LZ4File_eof(fp->fp); +#else + fatal("not built with LZ4 support"); #endif break; default: @@ -777,23 +1378,42 @@ cfeof(cfp *fp) const char * get_cfp_error(cfp *fp) { - if (fp->compressionMethod == COMPRESSION_GZIP) + const char *errmsg = NULL; + + switch(fp->compressionMethod) { + case COMPRESSION_NONE: + errmsg = strerror(errno); + + break; + case COMPRESSION_GZIP: #ifdef HAVE_LIBZ - int errnum; - const char *errmsg = gzerror(fp->fp, &errnum); + { + int errnum; + errmsg = gzerror(fp->fp, &errnum); - if (errnum != Z_ERRNO) - return errmsg; + if (errnum == Z_ERRNO) + errmsg = strerror(errno); + } #else - fatal("not built with zlib support"); + fatal("not built with zlib support"); +#endif + break; + case COMPRESSION_LZ4: +#ifdef HAVE_LIBLZ4 + errmsg = LZ4File_error(fp->fp); +#else + fatal("not built with LZ4 support"); #endif + break; + default: + fatal("invalid compression method"); + break; } - return strerror(errno); + return errmsg; } -#ifdef HAVE_LIBZ static int hasSuffix(const char *filename, const char *suffix) { @@ -807,5 +1427,3 @@ hasSuffix(const char *filename, const char *suffix) suffix, suffixlen) == 0; } - -#endif diff --git a/src/bin/pg_dump/pg_backup.h b/src/bin/pg_dump/pg_backup.h index 741810b066..95b02dac3b 100644 --- a/src/bin/pg_dump/pg_backup.h +++ b/src/bin/pg_dump/pg_backup.h @@ -78,6 +78,7 @@ enum _dumpPreparedQueries typedef enum _compressionMethod { COMPRESSION_GZIP, + COMPRESSION_LZ4, COMPRESSION_NONE } CompressionMethod; diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 7d96446f1a..e6f727534b 100644 --- a/src/bin/pg_dump/pg_backup_archiver.c +++ b/src/bin/pg_dump/pg_backup_archiver.c @@ -353,6 +353,7 @@ RestoreArchive(Archive *AHX) ArchiveHandle *AH = (ArchiveHandle *) AHX; RestoreOptions *ropt = AH->public.ropt; bool parallel_mode; + bool supports_compression; TocEntry *te; cfp *sav; @@ -382,17 +383,27 @@ RestoreArchive(Archive *AHX) /* * Make sure we won't need (de)compression we haven't got */ -#ifndef HAVE_LIBZ - if (AH->compressionMethod == COMPRESSION_GZIP && + supports_compression = true; + if (AH->compressionMethod != COMPRESSION_NONE && AH->PrintTocDataPtr != NULL) { for (te = AH->toc->next; te != AH->toc; te = te->next) { if (te->hadDumper && (te->reqs & REQ_DATA) != 0) - fatal("cannot restore from compressed archive (compression not supported in this installation)"); + { +#ifndef HAVE_LIBZ + if (AH->compressionMethod == COMPRESSION_GZIP) + supports_compression = false; +#endif +#ifndef HAVE_LIBLZ4 + if (AH->compressionMethod == COMPRESSION_LZ4) + supports_compression = false; +#endif + if (supports_compression == false) + fatal("cannot restore from compressed archive (compression not supported in this installation)"); + } } } -#endif /* * Prepare index arrays, so we can assume we have them throughout restore. @@ -2019,6 +2030,18 @@ ReadStr(ArchiveHandle *AH) return buf; } +static bool +_fileExistsInDirectory(const char *dir, const char *filename) +{ + struct stat st; + char buf[MAXPGPATH]; + + if (snprintf(buf, MAXPGPATH, "%s/%s", dir, filename) >= MAXPGPATH) + fatal("directory name too long: \"%s\"", dir); + + return (stat(buf, &st) == 0 && S_ISREG(st.st_mode)); +} + static int _discoverArchiveFormat(ArchiveHandle *AH) { @@ -2046,30 +2069,21 @@ _discoverArchiveFormat(ArchiveHandle *AH) /* * Check if the specified archive is a directory. If so, check if - * there's a "toc.dat" (or "toc.dat.gz") file in it. + * there's a "toc.dat" (or "toc.dat.{gz,lz4}") file in it. */ if (stat(AH->fSpec, &st) == 0 && S_ISDIR(st.st_mode)) { - char buf[MAXPGPATH]; - if (snprintf(buf, MAXPGPATH, "%s/toc.dat", AH->fSpec) >= MAXPGPATH) - fatal("directory name too long: \"%s\"", - AH->fSpec); - if (stat(buf, &st) == 0 && S_ISREG(st.st_mode)) - { - AH->format = archDirectory; + AH->format = archDirectory; + if (_fileExistsInDirectory(AH->fSpec, "toc.dat")) return AH->format; - } - #ifdef HAVE_LIBZ - if (snprintf(buf, MAXPGPATH, "%s/toc.dat.gz", AH->fSpec) >= MAXPGPATH) - fatal("directory name too long: \"%s\"", - AH->fSpec); - if (stat(buf, &st) == 0 && S_ISREG(st.st_mode)) - { - AH->format = archDirectory; + if (_fileExistsInDirectory(AH->fSpec, "toc.dat.gz")) + return AH->format; +#endif +#ifdef HAVE_LIBLZ4 + if (_fileExistsInDirectory(AH->fSpec, "toc.dat.lz4")) return AH->format; - } #endif fatal("directory \"%s\" does not appear to be a valid archive (\"toc.dat\" does not exist)", AH->fSpec); @@ -3681,6 +3695,7 @@ WriteHead(ArchiveHandle *AH) AH->WriteBytePtr(AH, AH->offSize); AH->WriteBytePtr(AH, AH->format); WriteInt(AH, AH->compressionLevel); + AH->WriteBytePtr(AH, AH->compressionMethod); crtm = *localtime(&AH->createDate); WriteInt(AH, crtm.tm_sec); WriteInt(AH, crtm.tm_min); @@ -3761,14 +3776,20 @@ ReadHead(ArchiveHandle *AH) else AH->compressionLevel = Z_DEFAULT_COMPRESSION; - if (AH->compressionLevel != INT_MIN) + if (AH->version >= K_VERS_1_15) + AH->compressionMethod = AH->ReadBytePtr(AH); + else if (AH->compressionLevel != 0) + AH->compressionMethod = COMPRESSION_GZIP; + #ifndef HAVE_LIBZ + if (AH->compressionMethod == COMPRESSION_GZIP) + { pg_log_warning("archive is compressed, but this installation does not support compression -- no data will be available"); -#else - AH->compressionMethod = COMPRESSION_GZIP; + AH->compressionMethod = COMPRESSION_NONE; + AH->compressionLevel = 0; + } #endif - if (AH->version >= K_VERS_1_4) { struct tm crtm; diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h index 837e9d73f5..037bfcf913 100644 --- a/src/bin/pg_dump/pg_backup_archiver.h +++ b/src/bin/pg_dump/pg_backup_archiver.h @@ -65,10 +65,11 @@ #define K_VERS_1_13 MAKE_ARCHIVE_VERSION(1, 13, 0) /* change search_path * behavior */ #define K_VERS_1_14 MAKE_ARCHIVE_VERSION(1, 14, 0) /* add tableam */ +#define K_VERS_1_15 MAKE_ARCHIVE_VERSION(1, 15, 0) /* add compressionMethod in header */ /* Current archive version number (the format we can output) */ #define K_VERS_MAJOR 1 -#define K_VERS_MINOR 14 +#define K_VERS_MINOR 15 #define K_VERS_REV 0 #define K_VERS_SELF MAKE_ARCHIVE_VERSION(K_VERS_MAJOR, K_VERS_MINOR, K_VERS_REV) diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 7477d5112e..437a25b8d9 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -999,7 +999,7 @@ help(const char *progname) printf(_(" -j, --jobs=NUM use this many parallel jobs to dump\n")); printf(_(" -v, --verbose verbose mode\n")); printf(_(" -V, --version output version information, then exit\n")); - printf(_(" -Z, --compress=[gzip,none][:LEVEL] or [LEVEL]\n" + printf(_(" -Z, --compress=[gzip,lz4,none][:LEVEL] or [LEVEL]\n" " compress output with given method or level\n")); printf(_(" --lock-wait-timeout=TIMEOUT fail after waiting TIMEOUT for a table lock\n")); printf(_(" --no-sync do not wait for changes to be written safely to disk\n")); @@ -1268,11 +1268,13 @@ parse_compression_method(const char *method, if (pg_strcasecmp(method, "gzip") == 0) *compressionMethod = COMPRESSION_GZIP; + else if (pg_strcasecmp(method, "lz4") == 0) + *compressionMethod = COMPRESSION_LZ4; else if (pg_strcasecmp(method, "none") == 0) *compressionMethod = COMPRESSION_NONE; else { - pg_log_error("invalid compression method \"%s\" (gzip, none)", method); + pg_log_error("invalid compression method \"%s\" (gzip, lz4, none)", method); res = false; } @@ -1343,10 +1345,10 @@ parse_compression_option(const char *opt, CompressionMethod *compressionMethod, if (!res) return res; - /* one can set level when method is gzip */ - if (*compressionMethod != COMPRESSION_GZIP && *compressLevel != INT_MIN) + /* one can set level when a compression method is set */ + if (*compressionMethod == COMPRESSION_NONE && *compressLevel != INT_MIN) { - pg_log_error("can only specify -Z/--compress [LEVEL] when method is gzip"); + pg_log_error("can only specify -Z/--compress [LEVEL] when method is set"); return false; } diff --git a/src/bin/pg_dump/t/001_basic.pl b/src/bin/pg_dump/t/001_basic.pl index 840c3ac899..4185c19e8b 100644 --- a/src/bin/pg_dump/t/001_basic.pl +++ b/src/bin/pg_dump/t/001_basic.pl @@ -128,12 +128,12 @@ command_fails_like( command_fails_like( [ 'pg_dump', '--compress', 'garbage' ], - qr/\Qpg_dump: error: invalid compression method "garbage" (gzip, none)\E/, + qr/\Qpg_dump: error: invalid compression method "garbage" (gzip, lz4, none)\E/, 'pg_dump: invalid --compress'); command_fails_like( [ 'pg_dump', '--compress', 'none:1' ], - qr/\Qpg_dump: error: can only specify -Z\/--compress [LEVEL] when method is gzip\E/, + qr/\Qpg_dump: error: can only specify -Z\/--compress [LEVEL] when method is set\E/, 'pg_dump: can only specify -Z/--compress [LEVEL] when method is gzip'); command_fails_like( diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl index d30e96d705..81e18f5a87 100644 --- a/src/bin/pg_dump/t/002_pg_dump.pl +++ b/src/bin/pg_dump/t/002_pg_dump.pl @@ -82,11 +82,14 @@ my %pgdump_runs = ( 'postgres', ], # Give coverage for manually compressed blob.toc files during restore. - compress_cmd => [ - $ENV{'GZIP_PROGRAM'}, - '-f', - "$tempdir/compression_gzip_directory_format/blobs.toc", - ], + compression => { + method => 'gzip', + cmd => [ + $ENV{'GZIP_PROGRAM'}, + '-f', + "$tempdir/compression_gzip_directory_format/blobs.toc", + ], + }, restore_cmd => [ 'pg_restore', "--file=$tempdir/compression_gzip_directory_format.sql", @@ -102,12 +105,15 @@ my %pgdump_runs = ( "--file=$tempdir/compression_gzip_directory_format_parallel", 'postgres', ], - # Give coverage for manually compressed blob.toc files during restore. - compress_cmd => [ - $ENV{'GZIP_PROGRAM'}, - '-f', - "$tempdir/compression_gzip_directory_format_parallel/blobs.toc", - ], + # Give coverage for manually compressed toc.dat files during restore. + compression => { + method => 'gzip', + cmd => [ + $ENV{'GZIP_PROGRAM'}, + '-f', + "$tempdir/compression_gzip_directory_format_parallel/toc.dat", + ], + }, restore_cmd => [ 'pg_restore', '--jobs=3', @@ -124,11 +130,95 @@ my %pgdump_runs = ( "--file=$tempdir/compression_gzip_plain_format.sql.gz", 'postgres', ], - compress_cmd => [ - $ENV{'GZIP_PROGRAM'}, - '-k', '-d', - "$tempdir/compression_gzip_plain_format.sql.gz", + compression => { + method => 'gzip', + cmd => [ + $ENV{'GZIP_PROGRAM'}, + '-k', '-d', + "$tempdir/compression_gzip_plain_format.sql.gz", + ], + }, + }, + compression_lz4_custom_format => { + test_key => 'compression', + dump_cmd => [ + 'pg_dump', '--no-sync', + '--format=custom', '--compress=lz4:9', + "--file=$tempdir/compression_lz4_custom_format.dump", + 'postgres', + ], + restore_cmd => [ + 'pg_restore', + "--file=$tempdir/compression_lz4_custom_format.sql", + "$tempdir/compression_lz4_custom_format.dump", + ], + }, + compression_lz4_directory_format => { + test_key => 'compression', + dump_cmd => [ + 'pg_dump', '--no-sync', + '--format=directory', '--compress=lz4', + "--file=$tempdir/compression_lz4_directory_format", + 'postgres', + ], + # Give coverage for manually compressed toc.dat files during restore. + compression => { + method => 'lz4', + cmd => [ + $ENV{'LZ4'}, + '-z', '-f', '--rm', + "$tempdir/compression_lz4_directory_format/toc.dat", + "$tempdir/compression_lz4_directory_format/toc.dat.lz4", + ], + }, + restore_cmd => [ + 'pg_restore', + "--file=$tempdir/compression_lz4_directory_format.sql", + "$tempdir/compression_lz4_directory_format", + ], + }, + compression_lz4_directory_format_parallel => { + test_key => 'compression', + dump_cmd => [ + 'pg_dump', '--no-sync', '--jobs=2', + '--format=directory', '--compress=lz4:9', + "--file=$tempdir/compression_lz4_directory_format_parallel", + 'postgres', + ], + # Give coverage for manually compressed blob.toc files during restore. + compression => { + method => 'lz4', + cmd => [ + $ENV{'LZ4'}, + '-z', '-f', '--rm', + "$tempdir/compression_lz4_directory_format_parallel/blobs.toc", + "$tempdir/compression_lz4_directory_format_parallel/blobs.toc.lz4", + ], + }, + restore_cmd => [ + 'pg_restore', '--jobs=2', + "--file=$tempdir/compression_lz4_directory_format_parallel.sql", + "$tempdir/compression_lz4_directory_format_parallel", + ], + }, + # Check that the output is valid lz4 + compression_lz4_plain_format => { + test_key => 'compression', + dump_cmd => [ + 'pg_dump', + '--no-sync', + '--format=plain', '--compress=lz4:1', + "--file=$tempdir/compression_lz4_plain_format.sql.lz4", + 'postgres', ], + compression => { + method => 'lz4', + cmd => [ + $ENV{LZ4}, '-d', '-f', + "$tempdir/compression_lz4_plain_format.sql.lz4", + "$tempdir/compression_lz4_plain_format.sql", + ], + } }, clean => { dump_cmd => [ @@ -3899,21 +3989,28 @@ foreach my $run (sort keys %pgdump_runs) my $test_key = $run; my $run_db = 'postgres'; - my $supports_compression = check_pg_config("#define HAVE_LIBZ 1"); - my $compress_program = $ENV{GZIP_PROGRAM}; + my $gzip_program = defined($ENV{GZIP_PROGRAM}) && $ENV{GZIP_PROGRAM} ne ''; + my $lz4_program = defined($ENV{LZ4}) && $ENV{LZ4} ne ''; + my $supports_gzip = check_pg_config("#define HAVE_LIBZ 1"); + my $supports_compression = check_pg_config("#define HAVE_LIBZ 1") || + $supports_lz4; $node->command_ok(\@{ $pgdump_runs{$run}->{dump_cmd} }, "$run: pg_dump runs"); - if (defined($pgdump_runs{$run}->{compress_cmd})) + if (defined($pgdump_runs{$run}->{'compression'})) { # Skip compression_cmd tests when compression is not supported, # as the result is uncompressed or the utility program does not # exist - next if !$supports_compression || !defined($compress_program) - || $compress_program eq ''; - command_ok( \@{ $pgdump_runs{$run}->{compress_cmd} }, - "$run: compression commands"); + next if !$supports_compression; + + my ($compression) = $pgdump_runs{$run}->{'compression'}; + + next if ${compression}->{method} eq 'gzip' && (!$gzip_program || !$supports_gzip); + next if ${compression}->{method} eq 'lz4' && (!$lz4_program || !$supports_lz4); + + command_ok( \@{ $compression->{cmd} }, "$run: compression commands"); } if ($pgdump_runs{$run}->{restore_cmd}) -- 2.32.0