diff --git a/Makefile.in b/Makefile.in index 9e97c88d..e71e534a 100644 --- a/Makefile.in +++ b/Makefile.in @@ -19,6 +19,7 @@ EXTRA_CLEAN = src/pgactive_init_copy$(X) src/pgactive_init_copy.o include/pgacti test/regression.diffs test/regression.out \ pgactive_init_copy_postgres.log home tmp_test_* \ pgactive_dump$(X) $(pgactive_DUMP_OBJS) \ + src/*.o src/compat/*/pg_dump/*.o # When in development add -Werror. PG_CPPFLAGS = -I$(srcdir)/include -I$(srcdir)/src/$(pgactive_PGVERCOMPAT_INCDIR) -I$(libpq_srcdir) -Wall -Wmissing-prototypes -Wmissing-declarations $(EXTRA_CFLAGS) diff --git a/src/compat/18/pg_dump.patch b/src/compat/18/pg_dump.patch index a0907050..069809dc 100644 --- a/src/compat/18/pg_dump.patch +++ b/src/compat/18/pg_dump.patch @@ -11,10 +11,10 @@ index d9041da..4dd06e8 100644 /* various user-settable parameters */ int dumpSections; /* bitmask of chosen sections */ diff --git a/src/compat/18/pg_dump/pg_dump.c b/src/compat/18/pg_dump/pg_dump.c -index 2b70da8..2fd19c6 100644 +index 8f6332d..5b024e1 100644 --- a/src/compat/18/pg_dump/pg_dump.c +++ b/src/compat/18/pg_dump/pg_dump.c -@@ -487,6 +487,7 @@ main(int argc, char **argv) +@@ -488,6 +488,7 @@ main(int argc, char **argv) */ {"attribute-inserts", no_argument, &dopt.column_inserts, 1}, {"binary-upgrade", no_argument, &dopt.binary_upgrade, 1}, @@ -22,7 +22,7 @@ index 2b70da8..2fd19c6 100644 {"column-inserts", no_argument, &dopt.column_inserts, 1}, {"disable-dollar-quoting", no_argument, &dopt.disable_dollar_quoting, 1}, {"disable-triggers", no_argument, &dopt.disable_triggers, 1}, -@@ -9948,6 +9949,8 @@ shouldPrintColumn(const DumpOptions *dopt, const TableInfo *tbinfo, int colno) +@@ -9965,6 +9966,8 @@ shouldPrintColumn(const DumpOptions *dopt, const TableInfo *tbinfo, int colno) { if (dopt->binary_upgrade) return true; @@ -31,7 +31,7 @@ index 2b70da8..2fd19c6 100644 if (tbinfo->attisdropped[colno]) return false; return (tbinfo->attislocal[colno] || tbinfo->ispartition); -@@ -15957,7 +15960,7 @@ dumpForeignServer(Archive *fout, const ForeignServerInfo *srvinfo) +@@ -15974,7 +15977,7 @@ dumpForeignServer(Archive *fout, const ForeignServerInfo *srvinfo) res = ExecuteSqlQueryForSingleRow(fout, query->data); fdwname = PQgetvalue(res, 0, 0); @@ -40,7 +40,7 @@ index 2b70da8..2fd19c6 100644 if (srvinfo->srvtype && strlen(srvinfo->srvtype) > 0) { appendPQExpBufferStr(q, " TYPE "); -@@ -16085,7 +16088,7 @@ dumpUserMappings(Archive *fout, +@@ -16102,7 +16105,7 @@ dumpUserMappings(Archive *fout, umoptions = PQgetvalue(res, i, i_umoptions); resetPQExpBuffer(q); @@ -49,7 +49,7 @@ index 2b70da8..2fd19c6 100644 appendPQExpBuffer(q, " SERVER %s", fmtId(servername)); if (umoptions && strlen(umoptions) > 0) -@@ -17593,6 +17596,35 @@ dumpTableSchema(Archive *fout, const TableInfo *tbinfo) +@@ -17613,6 +17616,35 @@ dumpTableSchema(Archive *fout, const TableInfo *tbinfo) } } diff --git a/src/compat/18/pg_dump/compress_gzip.c b/src/compat/18/pg_dump/compress_gzip.c index 5a30ebf9..be31209f 100644 --- a/src/compat/18/pg_dump/compress_gzip.c +++ b/src/compat/18/pg_dump/compress_gzip.c @@ -20,6 +20,15 @@ #ifdef HAVE_LIBZ #include +/* + * We don't use the gzgetc() macro, because zlib's configuration logic is not + * robust enough to guarantee that the macro will have the same ideas about + * struct field layout as the library itself does; see for example + * https://gnats.netbsd.org/cgi-bin/query-pr-single.pl?number=59711 + * Instead, #undef the macro and fall back to the underlying function. + */ +#undef gzgetc + /*---------------------- * Compressor API *---------------------- @@ -251,34 +260,53 @@ InitCompressorGzip(CompressorState *cs, *---------------------- */ -static bool -Gzip_read(void *ptr, size_t size, size_t *rsize, CompressFileHandle *CFH) +static size_t +Gzip_read(void *ptr, size_t size, CompressFileHandle *CFH) { gzFile gzfp = (gzFile) CFH->private_data; int gzret; + /* Reading zero bytes must be a no-op */ + if (size == 0) + return 0; + gzret = gzread(gzfp, ptr, size); - if (gzret <= 0 && !gzeof(gzfp)) + + /* + * gzread returns zero on EOF as well as some error conditions, and less + * than zero on other error conditions, so we need to inspect for EOF on + * zero. + */ + if (gzret <= 0) { int errnum; - const char *errmsg = gzerror(gzfp, &errnum); + const char *errmsg; + + if (gzret == 0 && gzeof(gzfp)) + return 0; + + errmsg = gzerror(gzfp, &errnum); pg_fatal("could not read from input file: %s", errnum == Z_ERRNO ? strerror(errno) : errmsg); } - if (rsize) - *rsize = (size_t) gzret; - - return true; + return (size_t) gzret; } -static bool +static void Gzip_write(const void *ptr, size_t size, CompressFileHandle *CFH) { gzFile gzfp = (gzFile) CFH->private_data; + int errnum; + const char *errmsg; - return gzwrite(gzfp, ptr, size) > 0; + if (gzwrite(gzfp, ptr, size) != size) + { + errmsg = gzerror(gzfp, &errnum); + pg_fatal("could not write to file: %s", + errnum == Z_ERRNO ? strerror(errno) : errmsg); + } } static int diff --git a/src/compat/18/pg_dump/compress_io.c b/src/compat/18/pg_dump/compress_io.c index 8c3d9c91..9cadc6f2 100644 --- a/src/compat/18/pg_dump/compress_io.c +++ b/src/compat/18/pg_dump/compress_io.c @@ -269,6 +269,7 @@ InitDiscoverCompressFileHandle(const char *path, const char *mode) } CFH = InitCompressFileHandle(compression_spec); + errno = 0; if (!CFH->open_func(fname, -1, mode, CFH)) { free_keep_errno(CFH); @@ -289,6 +290,7 @@ EndCompressFileHandle(CompressFileHandle *CFH) { bool ret = false; + errno = 0; if (CFH->private_data) ret = CFH->close_func(CFH); diff --git a/src/compat/18/pg_dump/compress_io.h b/src/compat/18/pg_dump/compress_io.h index db9b3874..25a7bf09 100644 --- a/src/compat/18/pg_dump/compress_io.h +++ b/src/compat/18/pg_dump/compress_io.h @@ -123,21 +123,22 @@ struct CompressFileHandle CompressFileHandle *CFH); /* - * Read 'size' bytes of data from the file and store them into 'ptr'. - * Optionally it will store the number of bytes read in 'rsize'. + * Read up to 'size' bytes of data from the file and store them into + * 'ptr'. * - * Returns true on success and throws an internal error otherwise. + * Returns number of bytes read (this might be less than 'size' if EOF was + * reached). Exits via pg_fatal for all error conditions. */ - bool (*read_func) (void *ptr, size_t size, size_t *rsize, + size_t (*read_func) (void *ptr, size_t size, CompressFileHandle *CFH); /* * Write 'size' bytes of data into the file from 'ptr'. * - * Returns true on success and false on error. + * Returns nothing, exits via pg_fatal for all error conditions. */ - bool (*write_func) (const void *ptr, size_t size, - struct CompressFileHandle *CFH); + void (*write_func) (const void *ptr, size_t size, + CompressFileHandle *CFH); /* * Read at most size - 1 characters from the compress file handle into diff --git a/src/compat/18/pg_dump/compress_lz4.c b/src/compat/18/pg_dump/compress_lz4.c index e99f0cad..442ae4e2 100644 --- a/src/compat/18/pg_dump/compress_lz4.c +++ b/src/compat/18/pg_dump/compress_lz4.c @@ -12,6 +12,7 @@ *------------------------------------------------------------------------- */ #include "postgres_fe.h" +#include #include "compress_lz4.h" #include "pg_backup_utils.h" @@ -358,7 +359,6 @@ LZ4Stream_init(LZ4State *state, int size, bool compressing) return true; state->compressing = compressing; - state->inited = true; /* When compressing, write LZ4 header to the output stream. */ if (state->compressing) @@ -367,6 +367,7 @@ LZ4Stream_init(LZ4State *state, int size, bool compressing) if (!LZ4State_compression_init(state)) return false; + errno = 0; if (fwrite(state->buffer, 1, state->compressedlen, state->fp) != state->compressedlen) { errno = (errno) ? errno : ENOSPC; @@ -390,6 +391,7 @@ LZ4Stream_init(LZ4State *state, int size, bool compressing) state->overflowlen = 0; } + state->inited = true; return true; } @@ -457,7 +459,11 @@ LZ4Stream_read_internal(LZ4State *state, void *ptr, int ptrsize, bool eol_flag) /* Lazy init */ if (!LZ4Stream_init(state, size, false /* decompressing */ )) + { + pg_log_error("unable to initialize LZ4 library: %s", + LZ4F_getErrorName(state->errcode)); return -1; + } /* No work needs to be done for a zero-sized output buffer */ if (size <= 0) @@ -484,7 +490,10 @@ LZ4Stream_read_internal(LZ4State *state, void *ptr, int ptrsize, bool eol_flag) rsize = fread(readbuf, 1, size, state->fp); if (rsize < size && !feof(state->fp)) + { + pg_log_error("could not read from input file: %m"); return -1; + } rp = (char *) readbuf; rend = (char *) readbuf + rsize; @@ -501,6 +510,8 @@ LZ4Stream_read_internal(LZ4State *state, void *ptr, int ptrsize, bool eol_flag) if (LZ4F_isError(status)) { state->errcode = status; + pg_log_error("could not read from input file: %s", + LZ4F_getErrorName(state->errcode)); return -1; } @@ -558,7 +569,7 @@ LZ4Stream_read_internal(LZ4State *state, void *ptr, int ptrsize, bool eol_flag) /* * Compress size bytes from ptr and write them to the stream. */ -static bool +static void LZ4Stream_write(const void *ptr, size_t size, CompressFileHandle *CFH) { LZ4State *state = (LZ4State *) CFH->private_data; @@ -567,7 +578,8 @@ LZ4Stream_write(const void *ptr, size_t size, CompressFileHandle *CFH) /* Lazy init */ if (!LZ4Stream_init(state, size, true)) - return false; + pg_fatal("unable to initialize LZ4 library: %s", + LZ4F_getErrorName(state->errcode)); while (remaining > 0) { @@ -578,28 +590,24 @@ LZ4Stream_write(const void *ptr, size_t size, CompressFileHandle *CFH) status = LZ4F_compressUpdate(state->ctx, state->buffer, state->buflen, ptr, chunk, NULL); if (LZ4F_isError(status)) - { - state->errcode = status; - return false; - } + pg_fatal("error during writing: %s", LZ4F_getErrorName(status)); + errno = 0; if (fwrite(state->buffer, 1, status, state->fp) != status) { errno = (errno) ? errno : ENOSPC; - return false; + pg_fatal("error during writing: %m"); } ptr = ((const char *) ptr) + chunk; } - - return true; } /* * fread() equivalent implementation for LZ4 compressed files. */ -static bool -LZ4Stream_read(void *ptr, size_t size, size_t *rsize, CompressFileHandle *CFH) +static size_t +LZ4Stream_read(void *ptr, size_t size, CompressFileHandle *CFH) { LZ4State *state = (LZ4State *) CFH->private_data; int ret; @@ -607,10 +615,7 @@ LZ4Stream_read(void *ptr, size_t size, size_t *rsize, CompressFileHandle *CFH) if ((ret = LZ4Stream_read_internal(state, ptr, size, false)) < 0) pg_fatal("could not read from input file: %s", LZ4Stream_get_error(CFH)); - if (rsize) - *rsize = (size_t) ret; - - return true; + return (size_t) ret; } /* @@ -643,11 +648,13 @@ LZ4Stream_gets(char *ptr, int size, CompressFileHandle *CFH) int ret; ret = LZ4Stream_read_internal(state, ptr, size - 1, true); - if (ret < 0 || (ret == 0 && !LZ4Stream_eof(CFH))) - pg_fatal("could not read from input file: %s", LZ4Stream_get_error(CFH)); - /* Done reading */ - if (ret == 0) + /* + * LZ4Stream_read_internal returning 0 or -1 means that it was either an + * EOF or an error, but gets_func is defined to return NULL in either case + * so we can treat both the same here. + */ + if (ret <= 0) return NULL; /* @@ -669,6 +676,8 @@ LZ4Stream_close(CompressFileHandle *CFH) FILE *fp; LZ4State *state = (LZ4State *) CFH->private_data; size_t status; + int ret; + bool success = true; fp = state->fp; if (state->inited) @@ -677,25 +686,39 @@ LZ4Stream_close(CompressFileHandle *CFH) { status = LZ4F_compressEnd(state->ctx, state->buffer, state->buflen, NULL); if (LZ4F_isError(status)) - pg_fatal("could not end compression: %s", - LZ4F_getErrorName(status)); - else if (fwrite(state->buffer, 1, status, state->fp) != status) { - errno = (errno) ? errno : ENOSPC; - WRITE_ERROR_EXIT; + pg_log_error("could not end compression: %s", + LZ4F_getErrorName(status)); + success = false; + } + else + { + errno = 0; + if (fwrite(state->buffer, 1, status, state->fp) != status) + { + errno = (errno) ? errno : ENOSPC; + pg_log_error("could not write to output file: %m"); + success = false; + } } status = LZ4F_freeCompressionContext(state->ctx); if (LZ4F_isError(status)) - pg_fatal("could not end compression: %s", - LZ4F_getErrorName(status)); + { + pg_log_error("could not end compression: %s", + LZ4F_getErrorName(status)); + success = false; + } } else { status = LZ4F_freeDecompressionContext(state->dtx); if (LZ4F_isError(status)) - pg_fatal("could not end decompression: %s", - LZ4F_getErrorName(status)); + { + pg_log_error("could not end decompression: %s", + LZ4F_getErrorName(status)); + success = false; + } pg_free(state->overflowbuf); } @@ -703,29 +726,35 @@ LZ4Stream_close(CompressFileHandle *CFH) } pg_free(state); + CFH->private_data = NULL; - return fclose(fp) == 0; + errno = 0; + ret = fclose(fp); + if (ret != 0) + { + pg_log_error("could not close file: %m"); + success = false; + } + + return success; } static bool LZ4Stream_open(const char *path, int fd, const char *mode, CompressFileHandle *CFH) { - FILE *fp; LZ4State *state = (LZ4State *) CFH->private_data; if (fd >= 0) - fp = fdopen(fd, mode); + state->fp = fdopen(dup(fd), mode); else - fp = fopen(path, mode); - if (fp == NULL) + state->fp = fopen(path, mode); + if (state->fp == NULL) { state->errcode = errno; return false; } - state->fp = fp; - return true; } diff --git a/src/compat/18/pg_dump/compress_none.c b/src/compat/18/pg_dump/compress_none.c index 3fc89c99..4abb2e95 100644 --- a/src/compat/18/pg_dump/compress_none.c +++ b/src/compat/18/pg_dump/compress_none.c @@ -83,35 +83,31 @@ InitCompressorNone(CompressorState *cs, * Private routines */ -static bool -read_none(void *ptr, size_t size, size_t *rsize, CompressFileHandle *CFH) +static size_t +read_none(void *ptr, size_t size, CompressFileHandle *CFH) { FILE *fp = (FILE *) CFH->private_data; size_t ret; - if (size == 0) - return true; - ret = fread(ptr, 1, size, fp); - if (ret != size && !feof(fp)) + if (ferror(fp)) pg_fatal("could not read from input file: %m"); - if (rsize) - *rsize = ret; - - return true; + return ret; } -static bool +static void write_none(const void *ptr, size_t size, CompressFileHandle *CFH) { size_t ret; + errno = 0; ret = fwrite(ptr, 1, size, (FILE *) CFH->private_data); if (ret != size) - return false; - - return true; + { + errno = (errno) ? errno : ENOSPC; + pg_fatal("could not write to file: %m"); + } } static const char * @@ -153,7 +149,12 @@ close_none(CompressFileHandle *CFH) CFH->private_data = NULL; if (fp) + { + errno = 0; ret = fclose(fp); + if (ret != 0) + pg_log_error("could not close file: %m"); + } return ret == 0; } diff --git a/src/compat/18/pg_dump/compress_zstd.c b/src/compat/18/pg_dump/compress_zstd.c index cb595b10..e24d45e1 100644 --- a/src/compat/18/pg_dump/compress_zstd.c +++ b/src/compat/18/pg_dump/compress_zstd.c @@ -13,6 +13,7 @@ */ #include "postgres_fe.h" +#include #include "compress_zstd.h" #include "pg_backup_utils.h" @@ -258,8 +259,8 @@ InitCompressorZstd(CompressorState *cs, * Compressed stream API */ -static bool -Zstd_read(void *ptr, size_t size, size_t *rdsize, CompressFileHandle *CFH) +static size_t +Zstd_read_internal(void *ptr, size_t size, CompressFileHandle *CFH, bool exit_on_error) { ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data; ZSTD_inBuffer *input = &zstdcs->input; @@ -268,6 +269,22 @@ Zstd_read(void *ptr, size_t size, size_t *rdsize, CompressFileHandle *CFH) size_t res, cnt; + /* + * If this is the first call to the reading function, initialize the + * required datastructures. + */ + if (zstdcs->dstream == NULL) + { + zstdcs->input.src = pg_malloc0(input_allocated_size); + zstdcs->dstream = ZSTD_createDStream(); + if (zstdcs->dstream == NULL) + { + if (exit_on_error) + pg_fatal("could not initialize compression library"); + return -1; + } + } + output->size = size; output->dst = ptr; output->pos = 0; @@ -292,6 +309,13 @@ Zstd_read(void *ptr, size_t size, size_t *rdsize, CompressFileHandle *CFH) if (input->pos == input->size) { cnt = fread(unconstify(void *, input->src), 1, input_allocated_size, zstdcs->fp); + if (ferror(zstdcs->fp)) + { + if (exit_on_error) + pg_fatal("could not read from input file: %m"); + return -1; + } + input->size = cnt; Assert(cnt <= input_allocated_size); @@ -307,7 +331,11 @@ Zstd_read(void *ptr, size_t size, size_t *rdsize, CompressFileHandle *CFH) res = ZSTD_decompressStream(zstdcs->dstream, output, input); if (ZSTD_isError(res)) - pg_fatal("could not decompress data: %s", ZSTD_getErrorName(res)); + { + if (exit_on_error) + pg_fatal("could not decompress data: %s", ZSTD_getErrorName(res)); + return -1; + } if (output->pos == output->size) break; /* No more room for output */ @@ -320,13 +348,10 @@ Zstd_read(void *ptr, size_t size, size_t *rdsize, CompressFileHandle *CFH) break; /* We read all the data that fits */ } - if (rdsize != NULL) - *rdsize = output->pos; - - return true; + return output->pos; } -static bool +static void Zstd_write(const void *ptr, size_t size, CompressFileHandle *CFH) { ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data; @@ -339,41 +364,40 @@ Zstd_write(const void *ptr, size_t size, CompressFileHandle *CFH) input->size = size; input->pos = 0; + if (zstdcs->cstream == NULL) + { + zstdcs->output.size = ZSTD_CStreamOutSize(); + zstdcs->output.dst = pg_malloc0(zstdcs->output.size); + zstdcs->cstream = _ZstdCStreamParams(CFH->compression_spec); + if (zstdcs->cstream == NULL) + pg_fatal("could not initialize compression library"); + } + /* Consume all input, to be flushed later */ while (input->pos != input->size) { output->pos = 0; res = ZSTD_compressStream2(zstdcs->cstream, output, input, ZSTD_e_continue); if (ZSTD_isError(res)) - { - zstdcs->zstderror = ZSTD_getErrorName(res); - return false; - } + pg_fatal("could not write to file: %s", ZSTD_getErrorName(res)); + errno = 0; cnt = fwrite(output->dst, 1, output->pos, zstdcs->fp); if (cnt != output->pos) { - zstdcs->zstderror = strerror(errno); - return false; + errno = (errno) ? errno : ENOSPC; + pg_fatal("could not write to file: %m"); } } - - return size; } static int Zstd_getc(CompressFileHandle *CFH) { - ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data; - int ret; + unsigned char ret; - if (CFH->read_func(&ret, 1, NULL, CFH) != 1) - { - if (feof(zstdcs->fp)) - pg_fatal("could not read from input file: end of file"); - else - pg_fatal("could not read from input file: %m"); - } + if (CFH->read_func(&ret, 1, CFH) != 1) + pg_fatal("could not read from input file: end of file"); return ret; } @@ -390,11 +414,7 @@ Zstd_gets(char *buf, int len, CompressFileHandle *CFH) */ for (i = 0; i < len - 1; ++i) { - size_t readsz; - - if (!CFH->read_func(&buf[i], 1, &readsz, CFH)) - break; - if (readsz != 1) + if (Zstd_read_internal(&buf[i], 1, CFH, false) != 1) break; if (buf[i] == '\n') { @@ -406,10 +426,17 @@ Zstd_gets(char *buf, int len, CompressFileHandle *CFH) return i > 0 ? buf : NULL; } +static size_t +Zstd_read(void *ptr, size_t size, CompressFileHandle *CFH) +{ + return Zstd_read_internal(ptr, size, CFH, true); +} + static bool Zstd_close(CompressFileHandle *CFH) { ZstdCompressorState *zstdcs = (ZstdCompressorState *) CFH->private_data; + bool success = true; if (zstdcs->cstream) { @@ -426,14 +453,18 @@ Zstd_close(CompressFileHandle *CFH) if (ZSTD_isError(res)) { zstdcs->zstderror = ZSTD_getErrorName(res); - return false; + success = false; + break; } + errno = 0; cnt = fwrite(output->dst, 1, output->pos, zstdcs->fp); if (cnt != output->pos) { + errno = (errno) ? errno : ENOSPC; zstdcs->zstderror = strerror(errno); - return false; + success = false; + break; } if (res == 0) @@ -450,11 +481,16 @@ Zstd_close(CompressFileHandle *CFH) pg_free(unconstify(void *, zstdcs->input.src)); } + errno = 0; if (fclose(zstdcs->fp) != 0) - return false; + { + zstdcs->zstderror = strerror(errno); + success = false; + } pg_free(zstdcs); - return true; + CFH->private_data = NULL; + return success; } static bool @@ -472,35 +508,33 @@ Zstd_open(const char *path, int fd, const char *mode, FILE *fp; ZstdCompressorState *zstdcs; + /* + * Clear state storage to avoid having the fd point to non-NULL memory on + * error return. + */ + CFH->private_data = NULL; + + zstdcs = (ZstdCompressorState *) pg_malloc_extended(sizeof(*zstdcs), + MCXT_ALLOC_NO_OOM | MCXT_ALLOC_ZERO); + if (!zstdcs) + { + errno = ENOMEM; + return false; + } + if (fd >= 0) - fp = fdopen(fd, mode); + fp = fdopen(dup(fd), mode); else fp = fopen(path, mode); if (fp == NULL) + { + pg_free(zstdcs); return false; + } - zstdcs = (ZstdCompressorState *) pg_malloc0(sizeof(*zstdcs)); - CFH->private_data = zstdcs; zstdcs->fp = fp; - - if (mode[0] == 'r') - { - zstdcs->input.src = pg_malloc0(ZSTD_DStreamInSize()); - zstdcs->dstream = ZSTD_createDStream(); - if (zstdcs->dstream == NULL) - pg_fatal("could not initialize compression library"); - } - else if (mode[0] == 'w' || mode[0] == 'a') - { - zstdcs->output.size = ZSTD_CStreamOutSize(); - zstdcs->output.dst = pg_malloc0(zstdcs->output.size); - zstdcs->cstream = _ZstdCStreamParams(CFH->compression_spec); - if (zstdcs->cstream == NULL) - pg_fatal("could not initialize compression library"); - } - else - pg_fatal("unhandled mode \"%s\"", mode); + CFH->private_data = zstdcs; return true; } diff --git a/src/compat/18/pg_dump/dumputils.c b/src/compat/18/pg_dump/dumputils.c index 8945bdd4..dde16590 100644 --- a/src/compat/18/pg_dump/dumputils.c +++ b/src/compat/18/pg_dump/dumputils.c @@ -730,6 +730,7 @@ bool variable_is_guc_list_quote(const char *name) { if (pg_strcasecmp(name, "local_preload_libraries") == 0 || + pg_strcasecmp(name, "oauth_validator_libraries") == 0 || pg_strcasecmp(name, "search_path") == 0 || pg_strcasecmp(name, "session_preload_libraries") == 0 || pg_strcasecmp(name, "shared_preload_libraries") == 0 || diff --git a/src/compat/18/pg_dump/pg_backup_archiver.c b/src/compat/18/pg_dump/pg_backup_archiver.c index bb50a170..4293e20b 100644 --- a/src/compat/18/pg_dump/pg_backup_archiver.c +++ b/src/compat/18/pg_dump/pg_backup_archiver.c @@ -42,6 +42,7 @@ #include "pg_backup_archiver.h" #include "pg_backup_db.h" #include "pg_backup_utils.h" +#include "pgtar.h" #define TEXT_DUMP_HEADER "--\n-- PostgreSQL database dump\n--\n\n" #define TEXT_DUMPALL_HEADER "--\n-- PostgreSQL database cluster dump\n--\n\n" @@ -1881,8 +1882,8 @@ ahwrite(const void *ptr, size_t size, size_t nmemb, ArchiveHandle *AH) { CompressFileHandle *CFH = (CompressFileHandle *) AH->OF; - if (CFH->write_func(ptr, size * nmemb, CFH)) - bytes_written = size * nmemb; + CFH->write_func(ptr, size * nmemb, CFH); + bytes_written = size * nmemb; } if (bytes_written != size * nmemb) @@ -2349,7 +2350,7 @@ _discoverArchiveFormat(ArchiveHandle *AH) } if (!isValidTarHeader(AH->lookahead)) - pg_fatal("input file does not appear to be a valid archive"); + pg_fatal("input file does not appear to be a valid tar archive"); AH->format = archTar; } @@ -3033,6 +3034,25 @@ _tocEntryRequired(TocEntry *te, teSection curSection, ArchiveHandle *AH) strcmp(te->desc, "ROW SECURITY") == 0)) return 0; + /* + * If it's a comment on a policy, a publication, or a subscription, maybe + * ignore it. + */ + if (strcmp(te->desc, "COMMENT") == 0) + { + if (ropt->no_policies && + strncmp(te->tag, "POLICY", strlen("POLICY")) == 0) + return 0; + + if (ropt->no_publications && + strncmp(te->tag, "PUBLICATION", strlen("PUBLICATION")) == 0) + return 0; + + if (ropt->no_subscriptions && + strncmp(te->tag, "SUBSCRIPTION", strlen("SUBSCRIPTION")) == 0) + return 0; + } + /* * If it's a publication or a table part of a publication, maybe ignore * it. @@ -3047,6 +3067,21 @@ _tocEntryRequired(TocEntry *te, teSection curSection, ArchiveHandle *AH) if (ropt->no_security_labels && strcmp(te->desc, "SECURITY LABEL") == 0) return 0; + /* + * If it's a security label on a publication or a subscription, maybe + * ignore it. + */ + if (strcmp(te->desc, "SECURITY LABEL") == 0) + { + if (ropt->no_publications && + strncmp(te->tag, "PUBLICATION", strlen("PUBLICATION")) == 0) + return 0; + + if (ropt->no_subscriptions && + strncmp(te->tag, "SUBSCRIPTION", strlen("SUBSCRIPTION")) == 0) + return 0; + } + /* If it's a subscription, maybe ignore it */ if (ropt->no_subscriptions && strcmp(te->desc, "SUBSCRIPTION") == 0) return 0; @@ -3291,12 +3326,14 @@ _tocEntryRestorePass(TocEntry *te) return RESTORE_PASS_POST_ACL; /* - * Comments need to be emitted in the same pass as their parent objects. - * ACLs haven't got comments, and neither do matview data objects, but - * event triggers do. (Fortunately, event triggers haven't got ACLs, or - * we'd need yet another weird special case.) + * Comments and security labels need to be emitted in the same pass as + * their parent objects. ACLs haven't got comments and security labels, + * and neither do matview data objects, but event triggers do. + * (Fortunately, event triggers haven't got ACLs, or we'd need yet another + * weird special case.) */ - if (strcmp(te->desc, "COMMENT") == 0 && + if ((strcmp(te->desc, "COMMENT") == 0 || + strcmp(te->desc, "SECURITY LABEL") == 0) && strncmp(te->tag, "EVENT TRIGGER ", 14) == 0) return RESTORE_PASS_POST_ACL; diff --git a/src/compat/18/pg_dump/pg_backup_archiver.h b/src/compat/18/pg_dump/pg_backup_archiver.h index 325b53fc..c01d4506 100644 --- a/src/compat/18/pg_dump/pg_backup_archiver.h +++ b/src/compat/18/pg_dump/pg_backup_archiver.h @@ -464,8 +464,6 @@ extern void InitArchiveFmt_Null(ArchiveHandle *AH); extern void InitArchiveFmt_Directory(ArchiveHandle *AH); extern void InitArchiveFmt_Tar(ArchiveHandle *AH); -extern bool isValidTarHeader(char *header); - extern void ReconnectToServer(ArchiveHandle *AH, const char *dbname); extern void IssueCommandPerBlob(ArchiveHandle *AH, TocEntry *te, const char *cmdBegin, const char *cmdEnd); diff --git a/src/compat/18/pg_dump/pg_backup_directory.c b/src/compat/18/pg_dump/pg_backup_directory.c index bc2a2fb4..94d401d8 100644 --- a/src/compat/18/pg_dump/pg_backup_directory.c +++ b/src/compat/18/pg_dump/pg_backup_directory.c @@ -316,15 +316,9 @@ _WriteData(ArchiveHandle *AH, const void *data, size_t dLen) lclContext *ctx = (lclContext *) AH->formatData; CompressFileHandle *CFH = ctx->dataFH; - errno = 0; - if (dLen > 0 && !CFH->write_func(data, dLen, CFH)) - { - /* if write didn't set errno, assume problem is no disk space */ - if (errno == 0) - errno = ENOSPC; - pg_fatal("could not write to output file: %s", - CFH->get_error_func(CFH)); - } + if (dLen <= 0) + return; + CFH->write_func(data, dLen, CFH); } /* @@ -351,7 +345,7 @@ _EndData(ArchiveHandle *AH, TocEntry *te) static void _PrintFileData(ArchiveHandle *AH, char *filename) { - size_t cnt = 0; + size_t cnt; char *buf; size_t buflen; CompressFileHandle *CFH; @@ -366,7 +360,7 @@ _PrintFileData(ArchiveHandle *AH, char *filename) buflen = DEFAULT_IO_BUFFER_SIZE; buf = pg_malloc(buflen); - while (CFH->read_func(buf, buflen, &cnt, CFH) && cnt > 0) + while ((cnt = CFH->read_func(buf, buflen, CFH)) > 0) { ahwrite(buf, 1, cnt, AH); } @@ -470,16 +464,7 @@ _WriteByte(ArchiveHandle *AH, const int i) lclContext *ctx = (lclContext *) AH->formatData; CompressFileHandle *CFH = ctx->dataFH; - errno = 0; - if (!CFH->write_func(&c, 1, CFH)) - { - /* if write didn't set errno, assume problem is no disk space */ - if (errno == 0) - errno = ENOSPC; - pg_fatal("could not write to output file: %s", - CFH->get_error_func(CFH)); - } - + CFH->write_func(&c, 1, CFH); return 1; } @@ -508,15 +493,7 @@ _WriteBuf(ArchiveHandle *AH, const void *buf, size_t len) lclContext *ctx = (lclContext *) AH->formatData; CompressFileHandle *CFH = ctx->dataFH; - errno = 0; - if (!CFH->write_func(buf, len, CFH)) - { - /* if write didn't set errno, assume problem is no disk space */ - if (errno == 0) - errno = ENOSPC; - pg_fatal("could not write to output file: %s", - CFH->get_error_func(CFH)); - } + CFH->write_func(buf, len, CFH); } /* @@ -531,10 +508,10 @@ _ReadBuf(ArchiveHandle *AH, void *buf, size_t len) CompressFileHandle *CFH = ctx->dataFH; /* - * If there was an I/O error, we already exited in readF(), so here we - * exit on short reads. + * We do not expect a short read, so fail if we get one. The read_func + * already dealt with any outright I/O error. */ - if (!CFH->read_func(buf, len, NULL, CFH)) + if (CFH->read_func(buf, len, CFH) != len) pg_fatal("could not read from input file: end of file"); } @@ -677,14 +654,7 @@ _EndLO(ArchiveHandle *AH, TocEntry *te, Oid oid) /* register the LO in blobs_NNN.toc */ len = snprintf(buf, sizeof(buf), "%u blob_%u.dat\n", oid, oid); - if (!CFH->write_func(buf, len, CFH)) - { - /* if write didn't set errno, assume problem is no disk space */ - if (errno == 0) - errno = ENOSPC; - pg_fatal("could not write to LOs TOC file: %s", - CFH->get_error_func(CFH)); - } + CFH->write_func(buf, len, CFH); } /* diff --git a/src/compat/18/pg_dump/pg_backup_tar.c b/src/compat/18/pg_dump/pg_backup_tar.c index b5ba3b46..ec42a2cb 100644 --- a/src/compat/18/pg_dump/pg_backup_tar.c +++ b/src/compat/18/pg_dump/pg_backup_tar.c @@ -984,31 +984,6 @@ tarPrintf(TAR_MEMBER *th, const char *fmt,...) return (int) cnt; } -bool -isValidTarHeader(char *header) -{ - int sum; - int chk = tarChecksum(header); - - sum = read_tar_number(&header[TAR_OFFSET_CHECKSUM], 8); - - if (sum != chk) - return false; - - /* POSIX tar format */ - if (memcmp(&header[TAR_OFFSET_MAGIC], "ustar\0", 6) == 0 && - memcmp(&header[TAR_OFFSET_VERSION], "00", 2) == 0) - return true; - /* GNU tar format */ - if (memcmp(&header[TAR_OFFSET_MAGIC], "ustar \0", 8) == 0) - return true; - /* not-quite-POSIX format written by pre-9.3 pg_dump */ - if (memcmp(&header[TAR_OFFSET_MAGIC], "ustar00\0", 8) == 0) - return true; - - return false; -} - /* Given the member, write the TAR header & copy the file */ static void _tarAddFile(ArchiveHandle *AH, TAR_MEMBER *th) diff --git a/src/compat/18/pg_dump/pg_dump.c b/src/compat/18/pg_dump/pg_dump.c index 2fd19c66..5b024e1e 100644 --- a/src/compat/18/pg_dump/pg_dump.c +++ b/src/compat/18/pg_dump/pg_dump.c @@ -135,6 +135,7 @@ typedef struct int64 cache; /* cache size */ int64 last_value; /* last value of sequence */ bool is_called; /* whether nextval advances before returning */ + bool null_seqtuple; /* did pg_get_sequence_data return nulls? */ } SequenceItem; typedef enum OidOptions @@ -824,23 +825,30 @@ main(int argc, char **argv) /* reject conflicting "-only" options */ if (data_only && schema_only) - pg_fatal("options -s/--schema-only and -a/--data-only cannot be used together"); + pg_fatal("options %s and %s cannot be used together", + "-s/--schema-only", "-a/--data-only"); if (schema_only && statistics_only) - pg_fatal("options -s/--schema-only and --statistics-only cannot be used together"); + pg_fatal("options %s and %s cannot be used together", + "-s/--schema-only", "--statistics-only"); if (data_only && statistics_only) - pg_fatal("options -a/--data-only and --statistics-only cannot be used together"); + pg_fatal("options %s and %s cannot be used together", + "-a/--data-only", "--statistics-only"); /* reject conflicting "-only" and "no-" options */ if (data_only && no_data) - pg_fatal("options -a/--data-only and --no-data cannot be used together"); + pg_fatal("options %s and %s cannot be used together", + "-a/--data-only", "--no-data"); if (schema_only && no_schema) - pg_fatal("options -s/--schema-only and --no-schema cannot be used together"); + pg_fatal("options %s and %s cannot be used together", + "-s/--schema-only", "--no-schema"); if (statistics_only && no_statistics) - pg_fatal("options --statistics-only and --no-statistics cannot be used together"); + pg_fatal("options %s and %s cannot be used together", + "--statistics-only", "--no-statistics"); /* reject conflicting "no-" options */ if (with_statistics && no_statistics) - pg_fatal("options --statistics and --no-statistics cannot be used together"); + pg_fatal("options %s and %s cannot be used together", + "--statistics", "--no-statistics"); /* reject conflicting "-only" options */ if (data_only && with_statistics) @@ -851,16 +859,20 @@ main(int argc, char **argv) "-s/--schema-only", "--statistics"); if (schema_only && foreign_servers_include_patterns.head != NULL) - pg_fatal("options -s/--schema-only and --include-foreign-data cannot be used together"); + pg_fatal("options %s and %s cannot be used together", + "-s/--schema-only", "--include-foreign-data"); if (numWorkers > 1 && foreign_servers_include_patterns.head != NULL) - pg_fatal("option --include-foreign-data is not supported with parallel backup"); + pg_fatal("option %s is not supported with parallel backup", + "--include-foreign-data"); if (data_only && dopt.outputClean) - pg_fatal("options -c/--clean and -a/--data-only cannot be used together"); + pg_fatal("options %s and %s cannot be used together", + "-c/--clean", "-a/--data-only"); if (dopt.if_exists && !dopt.outputClean) - pg_fatal("option --if-exists requires option -c/--clean"); + pg_fatal("option %s requires option %s", + "--if-exists", "-c/--clean"); /* * Set derivative flags. Ambiguous or nonsensical combinations, e.g. @@ -880,7 +892,9 @@ main(int argc, char **argv) * --rows-per-insert were specified. */ if (dopt.do_nothing && dopt.dump_inserts == 0) - pg_fatal("option --on-conflict-do-nothing requires option --inserts, --rows-per-insert, or --column-inserts"); + pg_fatal("option %s requires option %s, %s, or %s", + "--on-conflict-do-nothing", + "--inserts", "--rows-per-insert", "--column-inserts"); /* Identify archive format to emit */ archiveFormat = parseArchiveFormat(format, &archiveMode); @@ -901,7 +915,8 @@ main(int argc, char **argv) pg_fatal("invalid restrict key"); } else if (dopt.restrict_key) - pg_fatal("option --restrict-key can only be used with --format=plain"); + pg_fatal("option %s can only be used with %s", + "--restrict-key", "--format=plain"); /* * Custom and directory formats are compressed by default with gzip when @@ -5219,7 +5234,9 @@ getSubscriptionTables(Archive *fout) subrinfo[i].dobj.catId.tableoid = relid; subrinfo[i].dobj.catId.oid = cur_srsubid; AssignDumpId(&subrinfo[i].dobj); - subrinfo[i].dobj.name = pg_strdup(subinfo->dobj.name); + subrinfo[i].dobj.namespace = tblinfo->dobj.namespace; + subrinfo[i].dobj.name = tblinfo->dobj.name; + subrinfo[i].subinfo = subinfo; subrinfo[i].tblinfo = tblinfo; subrinfo[i].srsubstate = PQgetvalue(res, i, i_srsubstate)[0]; if (PQgetisnull(res, i, i_srsublsn)) @@ -5227,8 +5244,6 @@ getSubscriptionTables(Archive *fout) else subrinfo[i].srsublsn = pg_strdup(PQgetvalue(res, i, i_srsublsn)); - subrinfo[i].subinfo = subinfo; - /* Decide whether we want to dump it */ selectDumpableObject(&(subrinfo[i].dobj), fout); } @@ -5256,7 +5271,7 @@ dumpSubscriptionTable(Archive *fout, const SubRelInfo *subrinfo) Assert(fout->dopt->binary_upgrade && fout->remoteVersion >= 170000); - tag = psprintf("%s %s", subinfo->dobj.name, subrinfo->dobj.name); + tag = psprintf("%s %s", subinfo->dobj.name, subrinfo->tblinfo->dobj.name); query = createPQExpBuffer(); @@ -5271,7 +5286,7 @@ dumpSubscriptionTable(Archive *fout, const SubRelInfo *subrinfo) "\n-- For binary upgrade, must preserve the subscriber table.\n"); appendPQExpBufferStr(query, "SELECT pg_catalog.binary_upgrade_add_sub_rel_state("); - appendStringLiteralAH(query, subrinfo->dobj.name, fout); + appendStringLiteralAH(query, subinfo->dobj.name, fout); appendPQExpBuffer(query, ", %u, '%c'", subrinfo->tblinfo->dobj.catId.oid, @@ -9180,8 +9195,7 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables) * * We track in notnull_islocal whether the constraint was defined directly * in this table or via an ancestor, for binary upgrade. flagInhAttrs - * might modify this later; that routine is also in charge of determining - * the correct inhcount. + * might modify this later. */ if (fout->remoteVersion >= 180000) appendPQExpBufferStr(q, @@ -9198,7 +9212,10 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables) "NULL AS notnull_comment,\n" "NULL AS notnull_invalidoid,\n" "false AS notnull_noinherit,\n" - "a.attislocal AS notnull_islocal,\n"); + "CASE WHEN a.attislocal THEN true\n" + " WHEN a.attnotnull AND NOT a.attislocal THEN true\n" + " ELSE false\n" + "END AS notnull_islocal,\n"); if (fout->remoteVersion >= 140000) appendPQExpBufferStr(q, @@ -9900,7 +9917,7 @@ determineNotNullFlags(Archive *fout, PGresult *res, int r, */ if ((dopt->binary_upgrade && !tbinfo->ispartition && - !tbinfo->notnull_islocal) || + !tbinfo->notnull_islocal[j]) || !PQgetisnull(res, r, i_notnull_comment)) { tbinfo->notnull_constrs[j] = @@ -16604,7 +16621,7 @@ collectSecLabels(Archive *fout) appendPQExpBufferStr(query, "SELECT label, provider, classoid, objoid, objsubid " - "FROM pg_catalog.pg_seclabel " + "FROM pg_catalog.pg_seclabels " "ORDER BY classoid, objoid, objsubid"); res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK); @@ -17216,6 +17233,9 @@ dumpTableSchema(Archive *fout, const TableInfo *tbinfo) appendPQExpBuffer(q, "CONSTRAINT %s NOT NULL %s", tbinfo->notnull_constrs[j], fmtId(tbinfo->attnames[j])); + + if (tbinfo->notnull_noinh[j]) + appendPQExpBufferStr(q, " NO INHERIT"); } } @@ -18666,7 +18686,7 @@ dumpConstraint(Archive *fout, const ConstraintInfo *coninfo) dumpComment(fout, conprefix->data, qtypname, tyinfo->dobj.namespace->dobj.name, tyinfo->rolname, - coninfo->dobj.catId, 0, tyinfo->dobj.dumpId); + coninfo->dobj.catId, 0, coninfo->dobj.dumpId); destroyPQExpBuffer(conprefix); free(qtypname); } @@ -18801,6 +18821,7 @@ collectSequences(Archive *fout) sequences[i].cycled = (strcmp(PQgetvalue(res, i, 7), "t") == 0); sequences[i].last_value = strtoi64(PQgetvalue(res, i, 8), NULL, 10); sequences[i].is_called = (strcmp(PQgetvalue(res, i, 9), "t") == 0); + sequences[i].null_seqtuple = (PQgetisnull(res, i, 8) || PQgetisnull(res, i, 9)); } PQclear(res); @@ -19070,7 +19091,13 @@ dumpSequenceData(Archive *fout, const TableDataInfo *tdinfo) TableInfo *tbinfo = tdinfo->tdtable; int64 last; bool called; - PQExpBuffer query = createPQExpBuffer(); + PQExpBuffer query; + + /* needn't bother if not dumping sequence data */ + if (!fout->dopt->dumpData && !fout->dopt->sequence_data) + return; + + query = createPQExpBuffer(); /* * For versions >= 18, the sequence information is gathered in the sorted @@ -19113,6 +19140,12 @@ dumpSequenceData(Archive *fout, const TableDataInfo *tdinfo) entry = bsearch(&key, sequences, nsequences, sizeof(SequenceItem), SequenceItemCmp); + if (entry->null_seqtuple) + pg_fatal("failed to get data for sequence \"%s\"; user may lack " + "SELECT privilege on the sequence or the sequence may " + "have been concurrently dropped", + tbinfo->dobj.name); + last = entry->last_value; called = entry->is_called; } @@ -19343,6 +19376,11 @@ dumpEventTrigger(Archive *fout, const EventTriggerInfo *evtinfo) NULL, evtinfo->evtowner, evtinfo->dobj.catId, 0, evtinfo->dobj.dumpId); + if (evtinfo->dobj.dump & DUMP_COMPONENT_SECLABEL) + dumpSecLabel(fout, "EVENT TRIGGER", qevtname, + NULL, evtinfo->evtowner, + evtinfo->dobj.catId, 0, evtinfo->dobj.dumpId); + destroyPQExpBuffer(query); destroyPQExpBuffer(delqry); free(qevtname); diff --git a/src/compat/18/pg_dump/pg_dump_sort.c b/src/compat/18/pg_dump/pg_dump_sort.c index ff3aa024..297b4dfa 100644 --- a/src/compat/18/pg_dump/pg_dump_sort.c +++ b/src/compat/18/pg_dump/pg_dump_sort.c @@ -385,7 +385,8 @@ DOTypeNameCompare(const void *p1, const void *p2) if (cmpval != 0) return cmpval; } - else if (obj1->objType == DO_CONSTRAINT) + else if (obj1->objType == DO_CONSTRAINT || + obj1->objType == DO_FK_CONSTRAINT) { ConstraintInfo *robj1 = *(ConstraintInfo *const *) p1; ConstraintInfo *robj2 = *(ConstraintInfo *const *) p2; @@ -418,6 +419,19 @@ DOTypeNameCompare(const void *p1, const void *p2) return cmpval; } } + else if (obj1->objType == DO_DEFAULT_ACL) + { + DefaultACLInfo *daclobj1 = *(DefaultACLInfo *const *) p1; + DefaultACLInfo *daclobj2 = *(DefaultACLInfo *const *) p2; + + /* + * Sort by defaclrole, per pg_default_acl_role_nsp_obj_index. The + * (namespace, name) match (defaclnamespace, defaclobjtype). + */ + cmpval = strcmp(daclobj1->defaclrole, daclobj2->defaclrole); + if (cmpval != 0) + return cmpval; + } else if (obj1->objType == DO_PUBLICATION_REL) { PublicationRelInfo *probj1 = *(PublicationRelInfo *const *) p1; @@ -440,6 +454,17 @@ DOTypeNameCompare(const void *p1, const void *p2) if (cmpval != 0) return cmpval; } + else if (obj1->objType == DO_SUBSCRIPTION_REL) + { + SubRelInfo *srobj1 = *(SubRelInfo *const *) p1; + SubRelInfo *srobj2 = *(SubRelInfo *const *) p2; + + /* Sort by subscription name, since (namespace, name) match the rel */ + cmpval = strcmp(srobj1->subinfo->dobj.name, + srobj2->subinfo->dobj.name); + if (cmpval != 0) + return cmpval; + } /* * Shouldn't get here except after catalog corruption, but if we do, sort diff --git a/src/compat/README.pg_dump b/src/compat/README.pg_dump index b15c9842..d85f5390 100644 --- a/src/compat/README.pg_dump +++ b/src/compat/README.pg_dump @@ -38,5 +38,5 @@ Following table shows commit hash of pg_dump source | 15 | 0ab43b548237b3791261480d6a023f6b95b53942 | | 16 | 327bd6111aede2a41f8891d2cd2501c773bc047e | | 17 | e81a5031d51f9cd68adb7bdccc424880bac6838e | -| 18 | 2102667b3c1ba6dbc65fcbe687cccb578772a5e9 | +| 18 | ca938ec213d954aaa9eba60bbbfbfa924fac0d7a | diff --git a/test/t/051_node_loss_desync_pause.pl b/test/t/051_node_loss_desync_pause.pl index 1f38a119..c816f54d 100644 --- a/test/t/051_node_loss_desync_pause.pl +++ b/test/t/051_node_loss_desync_pause.pl @@ -14,6 +14,7 @@ use Config; use PostgreSQL::Test::Cluster; use PostgreSQL::Test::Utils; +use Time::HiRes qw(usleep); use threads; use Test::More; use utils::nodemanagement; @@ -46,8 +47,10 @@ # Check changes from node_2 are replayed on node_1 and not on node_0 wait_for_apply($node_2,$node_1); +sleep(1); is($node_1->safe_psql($pgactive_test_dbname,"SELECT id FROM $test_table"), $value,"Changes replayed to node_1"); +sleep(1); is($node_0->safe_psql($pgactive_test_dbname,"SELECT id FROM $test_table"), '',"Changes not replayed to node_0 due to apply pause"); diff --git a/test/t/utils/nodemanagement.pm b/test/t/utils/nodemanagement.pm index f14f1a77..4b1a50a0 100644 --- a/test/t/utils/nodemanagement.pm +++ b/test/t/utils/nodemanagement.pm @@ -505,6 +505,7 @@ sub pgactive_detach_nodes { $upstream_node->safe_psql( $pgactive_test_dbname, "SELECT pgactive.pgactive_detach_nodes($nodelist)" ); + sleep(5); # We can tell a detach has taken effect when the downstream's slot vanishes # on the upstream. for my $detach_node (@{$pgactive_detach_nodes}) { @@ -558,6 +559,7 @@ sub check_detach_status { qq($detach_node_name status on local node after detach is 'k' or 'r') ); + sleep(5); # The downstream's slot on the upstream MUST be gone is( $upstream_node->safe_psql($pgactive_test_dbname, qq[SELECT EXISTS (SELECT 1 FROM pgactive.pgactive_get_replication_lag_info() WHERE active and node_name = '$detach_node_name')]),