($INBOX_DIR/description missing)
help / color / mirror / Atom feedFrom: Justin Pryzby <[email protected]>
Subject: [PATCH 4/4] WIP: +change to use LZ4 frame API
Date: Sun, 26 Feb 2023 11:41:47 -0600
This uses buffering to avoid flushing and writing a block header for
each row, for an additional improvement in compressed size.
XXX: update archive version since this changes the meaning of "lz4" in
the header
---
src/bin/pg_dump/compress_lz4.c | 102 +++++++++++++++++++++++++--------
1 file changed, 79 insertions(+), 23 deletions(-)
diff --git a/src/bin/pg_dump/compress_lz4.c b/src/bin/pg_dump/compress_lz4.c
index 32a8f668907..88a819c553b 100644
--- a/src/bin/pg_dump/compress_lz4.c
+++ b/src/bin/pg_dump/compress_lz4.c
@@ -40,7 +40,7 @@ typedef struct LZ4CompressorState
{
char *outbuf;
size_t outsize;
- LZ4_stream_t *stream;
+ LZ4F_compressionContext_t ctx;
} LZ4CompressorState;
/* Private routines that support LZ4 compressed data I/O */
@@ -52,29 +52,59 @@ static void EndCompressorLZ4(ArchiveHandle *AH, CompressorState *cs);
static void
ReadDataFromArchiveLZ4(ArchiveHandle *AH, CompressorState *cs)
{
- LZ4_streamDecode_t lz4StreamDecode;
+ size_t res;
char *buf;
char *decbuf;
size_t buflen;
+ size_t decbuflen;
size_t cnt;
+ LZ4F_decompressOptions_t opts = {.stableDst = 0}; /* not stable */
+ LZ4F_decompressionContext_t dtx;
+
+ res = LZ4F_createDecompressionContext(&dtx, LZ4F_VERSION);
+ if (LZ4F_isError(res))
+ pg_fatal("failed to LZ4F_createDecompressionContext: %s",
+ LZ4F_getErrorName(res));
+
buflen = LZ4_IN_SIZE;
buf = pg_malloc(buflen);
- decbuf = pg_malloc(buflen);
- LZ4_setStreamDecode(&lz4StreamDecode, NULL, 0);
+ decbuflen = LZ4_IN_SIZE;
+ decbuf = pg_malloc(LZ4_IN_SIZE);
while ((cnt = cs->readF(AH, &buf, &buflen)))
{
- int decBytes = LZ4_decompress_safe_continue(&lz4StreamDecode,
- buf, decbuf,
- cnt, buflen);
+ char *end = buf + cnt;
- ahwrite(decbuf, 1, decBytes, AH);
+ for (char *ptr = buf; ptr != end;)
+ {
+ size_t decBytes = decbuflen;
+ size_t srcBytes = cnt;
+
+ res = LZ4F_decompress(dtx,
+ decbuf, &decBytes,
+ ptr, &srcBytes, &opts);
+ if (LZ4F_isError(res))
+ pg_fatal("failed to LZ4F_decompress: %s", LZ4F_getErrorName(res));
+
+ ptr += srcBytes;
+ cnt -= srcBytes;
+ if (decBytes > 0)
+ ahwrite(decbuf, 1, decBytes, AH);
+
+ if (decbuflen < res)
+ {
+ /* resize the buffer to the expected size */
+ decbuf = pg_realloc(decbuf, res);
+ decbuflen = res;
+ }
+ }
}
pg_free(buf);
pg_free(decbuf);
+ LZ4F_freeDecompressionContext(dtx);
}
static void
@@ -82,40 +112,66 @@ WriteDataToArchiveLZ4(ArchiveHandle *AH, CompressorState *cs,
const void *data, size_t dLen)
{
LZ4CompressorState *LZ4cs = (LZ4CompressorState *) cs->private_data;
- size_t compressed;
- size_t requiredsize = LZ4_compressBound(dLen);
+ size_t requiredsize = LZ4F_compressBound(dLen, NULL);
+ size_t res;
+ char *oldout = LZ4cs->outbuf;
if (requiredsize > LZ4cs->outsize)
{
LZ4cs->outbuf = pg_realloc(LZ4cs->outbuf, requiredsize);
LZ4cs->outsize = requiredsize;
}
- if (LZ4cs->stream == NULL)
- LZ4cs->stream = LZ4_createStream();
-
- compressed = LZ4_compress_fast_continue(LZ4cs->stream, data, LZ4cs->outbuf,
- dLen, LZ4cs->outsize,
- AH->compression_spec.level);
+ if (oldout == NULL)
+ {
+ LZ4F_preferences_t prefs = {
+ .compressionLevel = cs->compression_spec.level
+ };
+
+ res = LZ4F_createCompressionContext(&LZ4cs->ctx, LZ4F_VERSION);
+ if (LZ4F_isError(res))
+ pg_fatal("failed to LZ4F_createCompressionContext: %s", LZ4F_getErrorName(res));
+
+ res = LZ4F_compressBegin(LZ4cs->ctx, LZ4cs->outbuf, LZ4cs->outsize, &prefs);
+ if (LZ4F_isError(res))
+ pg_fatal("failed to LZ4F_compressBegin: %s", LZ4F_getErrorName(res));
+ cs->writeF(AH, LZ4cs->outbuf, res);
+ }
- if (compressed <= 0)
- pg_fatal("failed to LZ4 compress data");
+ res = LZ4F_compressUpdate(LZ4cs->ctx, LZ4cs->outbuf, LZ4cs->outsize,
+ data, dLen, NULL);
+ if (LZ4F_isError(res))
+ pg_fatal("failed to LZ4F_compressUpdate: %s", LZ4F_getErrorName(res));
- cs->writeF(AH, LZ4cs->outbuf, compressed);
+ if (res > 0)
+ cs->writeF(AH, LZ4cs->outbuf, res);
}
static void
EndCompressorLZ4(ArchiveHandle *AH, CompressorState *cs)
{
LZ4CompressorState *LZ4cs;
+ size_t res;
LZ4cs = (LZ4CompressorState *) cs->private_data;
- if (LZ4cs->stream != NULL)
- LZ4_freeStream(LZ4cs->stream);
+ if (LZ4cs->outbuf != NULL)
+ {
+ res = LZ4F_compressEnd(LZ4cs->ctx, LZ4cs->outbuf, LZ4cs->outsize, NULL);
+ if (LZ4F_isError(res))
+ pg_fatal("failed to LZ4F_compressEnd: %s", LZ4F_getErrorName(res));
+
+ if (res > 0)
+ cs->writeF(AH, LZ4cs->outbuf, res);
- pg_free(LZ4cs->outbuf);
- pg_free(LZ4cs);
+ LZ4F_freeCompressionContext(LZ4cs->ctx);
+ if (LZ4F_isError(res))
+ pg_fatal("failed to LZ4F_freeCompressionContext: %s", LZ4F_getErrorName(res));
+
+ pg_free(LZ4cs->outbuf);
+ LZ4cs->outbuf = NULL;
+ pg_free(LZ4cs);
+ }
}
--
2.34.1
--Fw50BJRL4GwAnGn1--
view thread (43+ 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 4/4] WIP: +change to use LZ4 frame API
In-Reply-To: <no-message-id-1857669@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