23 typedef struct bbsink_zstd
32 ZSTD_outBuffer zstd_outBuf;
35 static void bbsink_zstd_begin_backup(
bbsink *sink);
36 static void bbsink_zstd_begin_archive(
bbsink *sink,
const char *archive_name);
37 static void bbsink_zstd_archive_contents(
bbsink *sink,
size_t avail_in);
38 static void bbsink_zstd_manifest_contents(
bbsink *sink,
size_t len);
39 static void bbsink_zstd_end_archive(
bbsink *sink);
40 static void bbsink_zstd_cleanup(
bbsink *sink);
46 .begin_archive = bbsink_zstd_begin_archive,
47 .archive_contents = bbsink_zstd_archive_contents,
48 .end_archive = bbsink_zstd_end_archive,
50 .manifest_contents = bbsink_zstd_manifest_contents,
52 .end_backup = bbsink_zstd_end_backup,
53 .cleanup = bbsink_zstd_cleanup
65 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
66 errmsg(
"zstd compression is not supported by this build")));
73 sink =
palloc0(
sizeof(bbsink_zstd));
74 *((
const bbsink_ops **) &sink->base.bbs_ops) = &bbsink_zstd_ops;
75 sink->base.bbs_next =
next;
76 sink->compress = compress;
88 bbsink_zstd_begin_backup(
bbsink *sink)
90 bbsink_zstd *mysink = (bbsink_zstd *) sink;
91 size_t output_buffer_bound;
95 mysink->cctx = ZSTD_createCCtx();
97 elog(
ERROR,
"could not create zstd compression context");
99 ret = ZSTD_CCtx_setParameter(mysink->cctx, ZSTD_c_compressionLevel,
101 if (ZSTD_isError(ret))
102 elog(
ERROR,
"could not set zstd compression level to %d: %s",
103 compress->
level, ZSTD_getErrorName(ret));
112 ret = ZSTD_CCtx_setParameter(mysink->cctx, ZSTD_c_nbWorkers,
114 if (ZSTD_isError(ret))
116 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
117 errmsg(
"could not set compression worker count to %d: %s",
118 compress->
workers, ZSTD_getErrorName(ret)));
125 mysink->base.bbs_buffer =
palloc(mysink->base.bbs_buffer_length);
131 output_buffer_bound = ZSTD_compressBound(mysink->base.bbs_buffer_length);
136 output_buffer_bound = output_buffer_bound + BLCKSZ -
137 (output_buffer_bound % BLCKSZ);
146 bbsink_zstd_begin_archive(
bbsink *sink,
const char *archive_name)
148 bbsink_zstd *mysink = (bbsink_zstd *) sink;
149 char *zstd_archive_name;
156 ZSTD_CCtx_reset(mysink->cctx, ZSTD_reset_session_only);
158 mysink->zstd_outBuf.dst = mysink->base.bbs_next->bbs_buffer;
159 mysink->zstd_outBuf.size = mysink->base.bbs_next->bbs_buffer_length;
160 mysink->zstd_outBuf.pos = 0;
163 zstd_archive_name =
psprintf(
"%s.zst", archive_name);
166 pfree(zstd_archive_name);
181 bbsink_zstd_archive_contents(
bbsink *sink,
size_t len)
183 bbsink_zstd *mysink = (bbsink_zstd *) sink;
184 ZSTD_inBuffer inBuf = {mysink->base.bbs_buffer,
len, 0};
186 while (inBuf.pos < inBuf.size)
189 size_t max_needed = ZSTD_compressBound(inBuf.size - inBuf.pos);
195 if (mysink->zstd_outBuf.size - mysink->zstd_outBuf.pos < max_needed)
198 mysink->zstd_outBuf.pos);
199 mysink->zstd_outBuf.dst = mysink->base.bbs_next->bbs_buffer;
200 mysink->zstd_outBuf.size =
201 mysink->base.bbs_next->bbs_buffer_length;
202 mysink->zstd_outBuf.pos = 0;
205 yet_to_flush = ZSTD_compressStream2(mysink->cctx, &mysink->zstd_outBuf,
206 &inBuf, ZSTD_e_continue);
208 if (ZSTD_isError(yet_to_flush))
210 "could not compress data: %s",
211 ZSTD_getErrorName(yet_to_flush));
223 bbsink_zstd_end_archive(
bbsink *sink)
225 bbsink_zstd *mysink = (bbsink_zstd *) sink;
230 ZSTD_inBuffer in = {NULL, 0, 0};
231 size_t max_needed = ZSTD_compressBound(0);
237 if (mysink->zstd_outBuf.size - mysink->zstd_outBuf.pos < max_needed)
240 mysink->zstd_outBuf.pos);
241 mysink->zstd_outBuf.dst = mysink->base.bbs_next->bbs_buffer;
242 mysink->zstd_outBuf.size =
243 mysink->base.bbs_next->bbs_buffer_length;
244 mysink->zstd_outBuf.pos = 0;
247 yet_to_flush = ZSTD_compressStream2(mysink->cctx,
248 &mysink->zstd_outBuf,
251 if (ZSTD_isError(yet_to_flush))
252 elog(
ERROR,
"could not compress data: %s",
253 ZSTD_getErrorName(yet_to_flush));
255 }
while (yet_to_flush > 0);
258 if (mysink->zstd_outBuf.pos > 0)
260 mysink->zstd_outBuf.pos);
273 bbsink_zstd *mysink = (bbsink_zstd *) sink;
278 ZSTD_freeCCtx(mysink->cctx);
290 bbsink_zstd_manifest_contents(
bbsink *sink,
size_t len)
301 bbsink_zstd_cleanup(
bbsink *sink)
303 bbsink_zstd *mysink = (bbsink_zstd *) sink;
308 ZSTD_freeCCtx(mysink->cctx);
void bbsink_forward_begin_manifest(bbsink *sink)
void bbsink_forward_end_backup(bbsink *sink, XLogRecPtr endptr, TimeLineID endtli)
void bbsink_forward_end_archive(bbsink *sink)
void bbsink_forward_end_manifest(bbsink *sink)
static void bbsink_begin_backup(bbsink *sink, bbsink_state *state, int buffer_length)
static void bbsink_begin_archive(bbsink *sink, const char *archive_name)
static void bbsink_archive_contents(bbsink *sink, size_t len)
static void bbsink_manifest_contents(bbsink *sink, size_t len)
bbsink * bbsink_zstd_new(bbsink *next, pg_compress_specification *compress)
#define PG_COMPRESSION_OPTION_WORKERS
elog(ERROR, "%s: %s", p2, msg)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
Assert(fmt[strlen(fmt) - 1] !='\n')
void pfree(void *pointer)
void * palloc0(Size size)
char * psprintf(const char *fmt,...)
void(* begin_backup)(bbsink *sink)