26 typedef struct bbstreamer_lz4_frame
30 LZ4F_compressionContext_t cctx;
31 LZ4F_decompressionContext_t dctx;
32 LZ4F_preferences_t prefs;
36 } bbstreamer_lz4_frame;
38 static void bbstreamer_lz4_compressor_content(
bbstreamer *streamer,
42 static void bbstreamer_lz4_compressor_finalize(
bbstreamer *streamer);
43 static void bbstreamer_lz4_compressor_free(
bbstreamer *streamer);
46 .
content = bbstreamer_lz4_compressor_content,
47 .finalize = bbstreamer_lz4_compressor_finalize,
48 .free = bbstreamer_lz4_compressor_free
51 static void bbstreamer_lz4_decompressor_content(
bbstreamer *streamer,
55 static void bbstreamer_lz4_decompressor_finalize(
bbstreamer *streamer);
56 static void bbstreamer_lz4_decompressor_free(
bbstreamer *streamer);
59 .
content = bbstreamer_lz4_decompressor_content,
60 .finalize = bbstreamer_lz4_decompressor_finalize,
61 .free = bbstreamer_lz4_decompressor_free
73 bbstreamer_lz4_frame *streamer;
74 LZ4F_errorCode_t ctxError;
75 LZ4F_preferences_t *prefs;
79 streamer =
palloc0(
sizeof(bbstreamer_lz4_frame));
81 &bbstreamer_lz4_compressor_ops;
83 streamer->base.bbs_next =
next;
85 streamer->header_written =
false;
88 prefs = &streamer->prefs;
89 memset(prefs, 0,
sizeof(LZ4F_preferences_t));
90 prefs->frameInfo.blockSizeID = LZ4F_max256KB;
91 prefs->compressionLevel = compress->
level;
93 ctxError = LZ4F_createCompressionContext(&streamer->cctx, LZ4F_VERSION);
94 if (LZ4F_isError(ctxError))
95 pg_log_error(
"could not create lz4 compression context: %s",
96 LZ4F_getErrorName(ctxError));
98 return &streamer->base;
100 pg_fatal(
"this build does not support compression with %s",
"LZ4");
116 bbstreamer_lz4_compressor_content(
bbstreamer *streamer,
121 bbstreamer_lz4_frame *mystreamer;
128 mystreamer = (bbstreamer_lz4_frame *) streamer;
132 if (!mystreamer->header_written)
134 compressed_size = LZ4F_compressBegin(mystreamer->cctx,
135 (
uint8 *) mystreamer->base.bbs_buffer.data,
136 mystreamer->base.bbs_buffer.maxlen,
139 if (LZ4F_isError(compressed_size))
141 LZ4F_getErrorName(compressed_size));
143 mystreamer->bytes_written += compressed_size;
144 mystreamer->header_written =
true;
151 next_out = (
uint8 *) mystreamer->base.bbs_buffer.data + mystreamer->bytes_written;
152 avail_out = mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
159 out_bound = LZ4F_compressBound(
len, &mystreamer->prefs);
160 if (avail_out < out_bound)
163 mystreamer->base.bbs_buffer.data,
164 mystreamer->bytes_written,
168 if (mystreamer->base.bbs_buffer.maxlen < out_bound)
171 avail_out = mystreamer->base.bbs_buffer.maxlen;
172 mystreamer->bytes_written = 0;
173 next_out = (
uint8 *) mystreamer->base.bbs_buffer.data;
184 compressed_size = LZ4F_compressUpdate(mystreamer->cctx,
188 if (LZ4F_isError(compressed_size))
190 LZ4F_getErrorName(compressed_size));
192 mystreamer->bytes_written += compressed_size;
199 bbstreamer_lz4_compressor_finalize(
bbstreamer *streamer)
201 bbstreamer_lz4_frame *mystreamer;
207 mystreamer = (bbstreamer_lz4_frame *) streamer;
210 footer_bound = LZ4F_compressBound(0, &mystreamer->prefs);
211 if ((mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written) <
215 mystreamer->base.bbs_buffer.data,
216 mystreamer->bytes_written,
220 if (mystreamer->base.bbs_buffer.maxlen < footer_bound)
223 avail_out = mystreamer->base.bbs_buffer.maxlen;
224 mystreamer->bytes_written = 0;
225 next_out = (
uint8 *) mystreamer->base.bbs_buffer.data;
229 next_out = (
uint8 *) mystreamer->base.bbs_buffer.data + mystreamer->bytes_written;
230 avail_out = mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
237 compressed_size = LZ4F_compressEnd(mystreamer->cctx,
238 next_out, avail_out, NULL);
240 if (LZ4F_isError(compressed_size))
242 LZ4F_getErrorName(compressed_size));
244 mystreamer->bytes_written += compressed_size;
247 mystreamer->base.bbs_buffer.data,
248 mystreamer->bytes_written,
258 bbstreamer_lz4_compressor_free(
bbstreamer *streamer)
260 bbstreamer_lz4_frame *mystreamer;
262 mystreamer = (bbstreamer_lz4_frame *) streamer;
264 LZ4F_freeCompressionContext(mystreamer->cctx);
278 bbstreamer_lz4_frame *streamer;
279 LZ4F_errorCode_t ctxError;
283 streamer =
palloc0(
sizeof(bbstreamer_lz4_frame));
285 &bbstreamer_lz4_decompressor_ops;
287 streamer->base.bbs_next =
next;
291 ctxError = LZ4F_createDecompressionContext(&streamer->dctx, LZ4F_VERSION);
292 if (LZ4F_isError(ctxError))
293 pg_fatal(
"could not initialize compression library: %s",
294 LZ4F_getErrorName(ctxError));
296 return &streamer->base;
298 pg_fatal(
"this build does not support compression with %s",
"LZ4");
310 bbstreamer_lz4_decompressor_content(
bbstreamer *streamer,
315 bbstreamer_lz4_frame *mystreamer;
321 mystreamer = (bbstreamer_lz4_frame *) streamer;
323 next_out = (
uint8 *) mystreamer->base.bbs_buffer.data;
325 avail_out = mystreamer->base.bbs_buffer.maxlen;
333 read_size = avail_in;
334 out_size = avail_out;
349 ret = LZ4F_decompress(mystreamer->dctx,
351 next_in, &read_size, NULL);
353 if (LZ4F_isError(ret))
355 LZ4F_getErrorName(ret));
358 avail_in -= read_size;
359 next_in += read_size;
361 mystreamer->bytes_written += out_size;
367 if (mystreamer->bytes_written >= mystreamer->base.bbs_buffer.maxlen)
370 mystreamer->base.bbs_buffer.data,
371 mystreamer->base.bbs_buffer.maxlen,
374 avail_out = mystreamer->base.bbs_buffer.maxlen;
375 mystreamer->bytes_written = 0;
376 next_out = (
uint8 *) mystreamer->base.bbs_buffer.data;
380 avail_out = mystreamer->base.bbs_buffer.maxlen - mystreamer->bytes_written;
381 next_out += mystreamer->bytes_written;
390 bbstreamer_lz4_decompressor_finalize(
bbstreamer *streamer)
392 bbstreamer_lz4_frame *mystreamer;
394 mystreamer = (bbstreamer_lz4_frame *) streamer;
401 mystreamer->base.bbs_buffer.data,
402 mystreamer->base.bbs_buffer.maxlen,
412 bbstreamer_lz4_decompressor_free(
bbstreamer *streamer)
414 bbstreamer_lz4_frame *mystreamer;
416 mystreamer = (bbstreamer_lz4_frame *) streamer;
418 LZ4F_freeDecompressionContext(mystreamer->dctx);
static void bbstreamer_content(bbstreamer *streamer, bbstreamer_member *member, const char *data, int len, bbstreamer_archive_context context)
static void bbstreamer_finalize(bbstreamer *streamer)
bbstreamer_archive_context
static void bbstreamer_free(bbstreamer *streamer)
bbstreamer * bbstreamer_lz4_compressor_new(bbstreamer *next, pg_compress_specification *compress)
bbstreamer * bbstreamer_lz4_decompressor_new(bbstreamer *next)
Assert(fmt[strlen(fmt) - 1] !='\n')
#define pg_log_error(...)
void pfree(void *pointer)
void * palloc0(Size size)
void enlargeStringInfo(StringInfo str, int needed)
void initStringInfo(StringInfo str)
void(* content)(bbstreamer *streamer, bbstreamer_member *member, const char *data, int len, bbstreamer_archive_context context)
StringInfoData bbs_buffer