public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 06/20] pg_dump: zstd compression
6+ messages / 3 participants
[nested] [flat]
* [PATCH 06/20] pg_dump: zstd compression
@ 2020-12-22 07:06 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 6+ messages in thread
From: Justin Pryzby @ 2020-12-22 07:06 UTC (permalink / raw)
document any change in search for .gz?
docs
Maybe compress_io should be split so all the library-specific stuff are in
separate files, like compress_{zlib/zstd}.c
---
configure | 123 ++++++-
configure.ac | 22 ++
src/bin/pg_dump/compress_io.c | 480 ++++++++++++++++++++++++++
src/bin/pg_dump/pg_backup.h | 14 +
src/bin/pg_dump/pg_backup_archiver.h | 4 +
src/bin/pg_dump/pg_backup_directory.c | 8 +-
src/bin/pg_dump/pg_dump.c | 39 +++
src/include/pg_config.h.in | 3 +
src/tools/msvc/Solution.pm | 1 +
9 files changed, 686 insertions(+), 8 deletions(-)
diff --git a/configure b/configure
index 11a4284e5b..fe739879af 100755
--- a/configure
+++ b/configure
@@ -698,6 +698,7 @@ with_gnu_ld
LD
LDFLAGS_SL
LDFLAGS_EX
+with_zstd
with_zlib
with_system_tzdata
with_libxslt
@@ -798,6 +799,7 @@ infodir
docdir
oldincludedir
includedir
+runstatedir
localstatedir
sharedstatedir
sysconfdir
@@ -866,6 +868,7 @@ with_libxml
with_libxslt
with_system_tzdata
with_zlib
+with_zstd
with_gnu_ld
enable_largefile
'
@@ -935,6 +938,7 @@ datadir='${datarootdir}'
sysconfdir='${prefix}/etc'
sharedstatedir='${prefix}/com'
localstatedir='${prefix}/var'
+runstatedir='${localstatedir}/run'
includedir='${prefix}/include'
oldincludedir='/usr/include'
docdir='${datarootdir}/doc/${PACKAGE_TARNAME}'
@@ -1187,6 +1191,15 @@ do
| -silent | --silent | --silen | --sile | --sil)
silent=yes ;;
+ -runstatedir | --runstatedir | --runstatedi | --runstated \
+ | --runstate | --runstat | --runsta | --runst | --runs \
+ | --run | --ru | --r)
+ ac_prev=runstatedir ;;
+ -runstatedir=* | --runstatedir=* | --runstatedi=* | --runstated=* \
+ | --runstate=* | --runstat=* | --runsta=* | --runst=* | --runs=* \
+ | --run=* | --ru=* | --r=*)
+ runstatedir=$ac_optarg ;;
+
-sbindir | --sbindir | --sbindi | --sbind | --sbin | --sbi | --sb)
ac_prev=sbindir ;;
-sbindir=* | --sbindir=* | --sbindi=* | --sbind=* | --sbin=* \
@@ -1324,7 +1337,7 @@ fi
for ac_var in exec_prefix prefix bindir sbindir libexecdir datarootdir \
datadir sysconfdir sharedstatedir localstatedir includedir \
oldincludedir docdir infodir htmldir dvidir pdfdir psdir \
- libdir localedir mandir
+ libdir localedir mandir runstatedir
do
eval ac_val=\$$ac_var
# Remove trailing slashes.
@@ -1477,6 +1490,7 @@ Fine tuning of the installation directories:
--sysconfdir=DIR read-only single-machine data [PREFIX/etc]
--sharedstatedir=DIR modifiable architecture-independent data [PREFIX/com]
--localstatedir=DIR modifiable single-machine data [PREFIX/var]
+ --runstatedir=DIR modifiable per-process data [LOCALSTATEDIR/run]
--libdir=DIR object code libraries [EPREFIX/lib]
--includedir=DIR C header files [PREFIX/include]
--oldincludedir=DIR C header files for non-gcc [/usr/include]
@@ -1570,6 +1584,7 @@ Optional Packages:
--with-system-tzdata=DIR
use system time zone data in DIR
--without-zlib do not use Zlib
+ --with-zstd use Zstd compression library
--with-gnu-ld assume the C compiler uses GNU ld [default=no]
Some influential environment variables:
@@ -8601,6 +8616,35 @@ fi
+#
+# Zstd
+#
+
+
+
+# Check whether --with-zstd was given.
+if test "${with_zstd+set}" = set; then :
+ withval=$with_zstd;
+ case $withval in
+ yes)
+ :
+ ;;
+ no)
+ :
+ ;;
+ *)
+ as_fn_error $? "no argument expected for --with-zstd option" "$LINENO" 5
+ ;;
+ esac
+
+else
+ with_zstd=no
+
+fi
+
+
+
+
#
# Assignments
#
@@ -12092,6 +12136,59 @@ fi
fi
+if test "$with_zstd" = yes; then
+ { $as_echo "$as_me:${as_lineno-$LINENO}: checking for ZSTD_compressStream2 in -lzstd" >&5
+$as_echo_n "checking for ZSTD_compressStream2 in -lzstd... " >&6; }
+if ${ac_cv_lib_zstd_ZSTD_compressStream2+:} false; then :
+ $as_echo_n "(cached) " >&6
+else
+ ac_check_lib_save_LIBS=$LIBS
+LIBS="-lzstd $LIBS"
+cat confdefs.h - <<_ACEOF >conftest.$ac_ext
+/* end confdefs.h. */
+
+/* Override any GCC internal prototype to avoid an error.
+ Use char because int might match the return type of a GCC
+ builtin and then its argument prototype would still apply. */
+#ifdef __cplusplus
+extern "C"
+#endif
+char ZSTD_compressStream2 ();
+int
+main ()
+{
+return ZSTD_compressStream2 ();
+ ;
+ return 0;
+}
+_ACEOF
+if ac_fn_c_try_link "$LINENO"; then :
+ ac_cv_lib_zstd_ZSTD_compressStream2=yes
+else
+ ac_cv_lib_zstd_ZSTD_compressStream2=no
+fi
+rm -f core conftest.err conftest.$ac_objext \
+ conftest$ac_exeext conftest.$ac_ext
+LIBS=$ac_check_lib_save_LIBS
+fi
+{ $as_echo "$as_me:${as_lineno-$LINENO}: result: $ac_cv_lib_zstd_ZSTD_compressStream2" >&5
+$as_echo "$ac_cv_lib_zstd_ZSTD_compressStream2" >&6; }
+if test "x$ac_cv_lib_zstd_ZSTD_compressStream2" = xyes; then :
+ cat >>confdefs.h <<_ACEOF
+#define HAVE_LIBZSTD 1
+_ACEOF
+
+ LIBS="-lzstd $LIBS"
+
+else
+ as_fn_error $? "zstd library not found
+If you have zstd already installed, see config.log for details on the
+failure. It is possible the compiler isn't looking in the proper directory.
+Use --without-zstd to disable zstd support." "$LINENO" 5
+fi
+
+fi
+
if test "$enable_spinlocks" = yes; then
$as_echo "#define HAVE_SPINLOCKS 1" >>confdefs.h
@@ -13295,6 +13392,20 @@ Use --without-zlib to disable zlib support." "$LINENO" 5
fi
+fi
+
+if test "$with_zstd" = yes; then
+ ac_fn_c_check_header_mongrel "$LINENO" "zstd.h" "ac_cv_header_zstd_h" "$ac_includes_default"
+if test "x$ac_cv_header_zstd_h" = xyes; then :
+
+else
+ as_fn_error $? "zstd header not found
+If you have zstd already installed, see config.log for details on the
+failure. It is possible the compiler isn't looking in the proper directory.
+Use --without-zstd to disable zstd support." "$LINENO" 5
+fi
+
+
fi
if test "$with_gssapi" = yes ; then
@@ -14689,7 +14800,7 @@ else
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
@@ -14735,7 +14846,7 @@ else
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
@@ -14759,7 +14870,7 @@ rm -f core conftest.err conftest.$ac_objext conftest.$ac_ext
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
@@ -14804,7 +14915,7 @@ else
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
@@ -14828,7 +14939,7 @@ rm -f core conftest.err conftest.$ac_objext conftest.$ac_ext
We can't simply define LARGE_OFF_T to be 9223372036854775807,
since some C++ compilers masquerading as C compilers
incorrectly reject 9223372036854775807. */
-#define LARGE_OFF_T (((off_t) 1 << 62) - 1 + ((off_t) 1 << 62))
+#define LARGE_OFF_T ((((off_t) 1 << 31) << 31) - 1 + (((off_t) 1 << 31) << 31))
int off_t_is_large[(LARGE_OFF_T % 2147483629 == 721
&& LARGE_OFF_T % 2147483647 == 1)
? 1 : -1];
diff --git a/configure.ac b/configure.ac
index fc523c6aeb..744836ea7f 100644
--- a/configure.ac
+++ b/configure.ac
@@ -999,6 +999,13 @@ PGAC_ARG_BOOL(with, zlib, yes,
[do not use Zlib])
AC_SUBST(with_zlib)
+#
+# Zstd
+#
+PGAC_ARG_BOOL(with, zstd, no,
+ [use Zstd compression library])
+AC_SUBST(with_zstd)
+
#
# Assignments
#
@@ -1186,6 +1193,14 @@ failure. It is possible the compiler isn't looking in the proper directory.
Use --without-zlib to disable zlib support.])])
fi
+if test "$with_zstd" = yes; then
+ AC_CHECK_LIB(zstd, ZSTD_compressStream2, [],
+ [AC_MSG_ERROR([zstd library not found
+If you have zstd already installed, see config.log for details on the
+failure. It is possible the compiler isn't looking in the proper directory.
+Use --without-zstd to disable zstd support.])])
+fi
+
if test "$enable_spinlocks" = yes; then
AC_DEFINE(HAVE_SPINLOCKS, 1, [Define to 1 if you have spinlocks.])
else
@@ -1400,6 +1415,13 @@ failure. It is possible the compiler isn't looking in the proper directory.
Use --without-zlib to disable zlib support.])])
fi
+if test "$with_zstd" = yes; then
+ AC_CHECK_HEADER(zstd.h, [], [AC_MSG_ERROR([zstd header not found
+If you have zstd already installed, see config.log for details on the
+failure. It is possible the compiler isn't looking in the proper directory.
+Use --without-zstd to disable zstd support.])])
+fi
+
if test "$with_gssapi" = yes ; then
AC_CHECK_HEADERS(gssapi/gssapi.h, [],
[AC_CHECK_HEADERS(gssapi.h, [], [AC_MSG_ERROR([gssapi.h header file is required for GSSAPI])])])
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index d66d6f60f5..285f554c1a 100644
--- a/src/bin/pg_dump/compress_io.c
+++ b/src/bin/pg_dump/compress_io.c
@@ -65,6 +65,18 @@ compresslibs[] = {
{ COMPR_ALG_LIBZ, "libz", ".gz", Z_DEFAULT_COMPRESSION },
{ COMPR_ALG_LIBZ, "zlib", ".gz", Z_DEFAULT_COMPRESSION }, /* Alternate name */
+#ifdef HAVE_LIBZSTD
+ /*
+ * ZSTD doesen't have a #define for it, but 0 means "the current default".
+ * Note that ZSTD_CLEVEL_DEFAULT is currently defined to 3.
+ *
+ * Block size should be ZSTD_DStreamOutSize(), but needs to be
+ * constant, so use ZSTD_BLOCKSIZE_MAX (128kB)
+ */
+ { COMPR_ALG_ZSTD, "zst", ".zst", 0 },
+ { COMPR_ALG_ZSTD, "zstd", ".zst", 0 }, /* Alternate name */
+#endif /* HAVE_LIBZSTD */
+
{ 0, NULL, } /* sentinel */
};
@@ -84,6 +96,18 @@ struct CompressorState
char *zlibOut;
size_t zlibOutSize;
#endif
+
+#ifdef HAVE_LIBZSTD
+ union {
+ struct {
+ ZSTD_outBuffer output;
+ ZSTD_inBuffer input;
+ // XXX: use one separate ZSTD_CStream per thread: disable on windows ?
+ ZSTD_CStream *cstream;
+ } zstd;
+ } u;
+#endif
+
};
/* Routines that support zlib compressed data I/O */
@@ -97,6 +121,15 @@ static void WriteDataToArchiveZlib(ArchiveHandle *AH, CompressorState *cs,
static void EndCompressorZlib(ArchiveHandle *AH, CompressorState *cs);
#endif
+#ifdef HAVE_LIBZSTD
+static ZSTD_CStream *ZstdCStreamParams(Compress *compress);
+static void InitCompressorZstd(CompressorState *cs, Compress *compress);
+static void EndCompressorZstd(ArchiveHandle *AH, CompressorState *cs);
+static void WriteDataToArchiveZstd(ArchiveHandle *AH, CompressorState *cs,
+ const char *data, size_t dLen);
+static void ReadDataFromArchiveZstd(ArchiveHandle *AH, ReadFunc readF);
+#endif
+
/* Routines that support uncompressed data I/O */
static void ReadDataFromArchiveNone(ArchiveHandle *AH, ReadFunc readF);
static void WriteDataToArchiveNone(ArchiveHandle *AH, CompressorState *cs,
@@ -125,6 +158,13 @@ AllocateCompressor(Compress *compression, WriteFunc writeF)
InitCompressorZlib(cs, compression);
break;
#endif
+
+#ifdef HAVE_LIBZSTD
+ case COMPR_ALG_ZSTD:
+ InitCompressorZstd(cs, compression);
+ break;
+#endif
+
case COMPR_ALG_NONE:
/* Do nothing */
break;
@@ -153,6 +193,13 @@ ReadDataFromArchive(ArchiveHandle *AH, ReadFunc readF)
ReadDataFromArchiveZlib(AH, readF);
break;
#endif
+
+#ifdef HAVE_LIBZSTD
+ case COMPR_ALG_ZSTD:
+ ReadDataFromArchiveZstd(AH, readF);
+ break;
+#endif
+
default:
/* Should not happen */
fatal("requested compression not available in this installation");
@@ -173,6 +220,12 @@ WriteDataToArchive(ArchiveHandle *AH, CompressorState *cs,
WriteDataToArchiveZlib(AH, cs, data, dLen);
break;
#endif
+
+#ifdef HAVE_LIBZSTD
+ case COMPR_ALG_ZSTD:
+ WriteDataToArchiveZstd(AH, cs, data, dLen);
+ break;
+#endif
case COMPR_ALG_NONE:
WriteDataToArchiveNone(AH, cs, data, dLen);
break;
@@ -193,11 +246,202 @@ EndCompressor(ArchiveHandle *AH, CompressorState *cs)
if (cs->comprAlg == COMPR_ALG_LIBZ)
EndCompressorZlib(AH, cs);
#endif
+
+#ifdef HAVE_LIBZSTD
+ if (cs->comprAlg == COMPR_ALG_ZSTD)
+ EndCompressorZstd(AH, cs);
+#endif
+
free(cs);
}
/* Private routines, specific to each compression method. */
+#ifdef HAVE_LIBZSTD
+
+static void ZSTD_CCtx_setParam_or_die(ZSTD_CStream *cstream,
+ ZSTD_cParameter param, int value)
+
+{
+ size_t res;
+ res = ZSTD_CCtx_setParameter(cstream, param, value);
+ if (ZSTD_isError(res))
+ fatal("could not set compression parameter: %s",
+ ZSTD_getErrorName(res));
+}
+
+/* Return a compression stream with parameters set per argument */
+static ZSTD_CStream*
+ZstdCStreamParams(Compress *compress)
+{
+ ZSTD_CStream *cstream;
+ cstream = ZSTD_createCStream();
+ if (cstream == NULL)
+ fatal("could not initialize compression library");
+
+ if (compress->level != 0) // XXX: ZSTD_CLEVEL_DEFAULT
+ {
+ size_t res;
+ res = ZSTD_CCtx_setParameter(cstream,
+ ZSTD_c_compressionLevel, compress->level);
+ if (ZSTD_isError(res))
+ fatal("could not set compression level: %s",
+ ZSTD_getErrorName(res));
+ }
+
+ if (compress->zstd.longdistance) // XXX: ternary
+ ZSTD_CCtx_setParam_or_die(cstream,
+ ZSTD_c_enableLongDistanceMatching,
+ compress->zstd.longdistance);
+
+ if (compress->zstd.checksum)
+ ZSTD_CCtx_setParam_or_die(cstream, ZSTD_c_checksumFlag,
+ compress->zstd.checksum);
+
+// not supported in my library ?
+ if (compress->zstd.threads)
+ ZSTD_CCtx_setParam_or_die(cstream, ZSTD_c_nbWorkers,
+ compress->zstd.threads);
+
+#if 0
+ /* Still marked as experimental */
+ if (compress->zstd.rsyncable)
+ ZSTD_CCtx_setParam_or_die(cstream, ZSTD_c_rsyncable, 1);
+#endif
+
+ return cstream;
+}
+
+static void
+InitCompressorZstd(CompressorState *cs, Compress *compress)
+{
+ cs->u.zstd.cstream = ZstdCStreamParams(compress);
+ /* XXX: initialize safely like the corresponding zlib "paranoia" */
+ cs->u.zstd.output.size = ZSTD_CStreamOutSize();
+ cs->u.zstd.output.dst = pg_malloc(cs->u.zstd.output.size);
+ cs->u.zstd.output.pos = 0;
+}
+
+static void
+EndCompressorZstd(ArchiveHandle *AH, CompressorState *cs)
+{
+ ZSTD_outBuffer *output = &cs->u.zstd.output;
+
+ for (;;)
+ {
+ size_t res;
+
+ res = ZSTD_compressStream2(cs->u.zstd.cstream, output,
+ &cs->u.zstd.input, ZSTD_e_end);
+
+ if (output->pos > 0)
+ cs->writeF(AH, output->dst, output->pos);
+
+ if (res == 0)
+ break;
+
+ if (ZSTD_isError(res))
+ fatal("could not close compression stream: %s",
+ ZSTD_getErrorName(res));
+ }
+
+ // XXX: retval
+ ZSTD_freeCStream(cs->u.zstd.cstream);
+}
+
+static void
+WriteDataToArchiveZstd(ArchiveHandle *AH, CompressorState *cs,
+ const char *data, size_t dLen)
+{
+ ZSTD_inBuffer *input = &cs->u.zstd.input;
+ ZSTD_outBuffer *output = &cs->u.zstd.output;
+
+ input->src = (void *) unconstify(char *, data);
+ input->size = dLen;
+ input->pos = 0;
+
+ while (input->pos != input->size)
+ {
+ size_t res;
+
+ res = ZSTD_compressStream2(cs->u.zstd.cstream, output,
+ input, ZSTD_e_continue);
+
+ if (output->pos == output->size ||
+ input->pos != input->size)
+ {
+ /*
+ * Extra paranoia: avoid zero-length chunks, since a zero length
+ * chunk is the EOF marker in the custom format. This should never
+ * happen but...
+ */
+ if (output->pos > 0)
+ cs->writeF(AH, output->dst, output->pos);
+
+ output->pos = 0;
+ }
+
+ if (ZSTD_isError(res))
+ fatal("could not compress data: %s", ZSTD_getErrorName(res));
+ }
+}
+
+/* Read data from a compressed zstd archive */
+static void
+ReadDataFromArchiveZstd(ArchiveHandle *AH, ReadFunc readF)
+{
+ ZSTD_DStream *dstream;
+ ZSTD_outBuffer output;
+ ZSTD_inBuffer input;
+ size_t res;
+ size_t input_size;
+
+ dstream = ZSTD_createDStream();
+ if (dstream == NULL)
+ fatal("could not initialize compression library");
+
+ input_size = ZSTD_DStreamInSize();
+ input.src = pg_malloc(input_size);
+
+ output.size = ZSTD_DStreamOutSize();
+ output.dst = pg_malloc(output.size);
+
+ /* read compressed data */
+ for (;;)
+ {
+ size_t cnt;
+
+ input.size = input_size; // XXX: the buffer can grow, we shouldn't keep resetting it to the original value..
+ cnt = readF(AH, (char **)unconstify(void **, &input.src), &input.size);
+ input.pos = 0;
+ input.size = cnt;
+
+ if (cnt == 0)
+ break;
+
+ while (input.pos < input.size)
+ {
+ /* decompress */
+ output.pos = 0;
+ res = ZSTD_decompressStream(dstream, &output, &input);
+
+ if (ZSTD_isError(res))
+ fatal("could not decompress data: %s", ZSTD_getErrorName(res));
+
+ /* write to output handle */
+ ((char *)output.dst)[output.pos] = '\0';
+ ahwrite(output.dst, 1, output.pos, AH);
+ // if (res == 0)
+ // break;
+ }
+ }
+
+ pg_free(unconstify(void *, input.src));
+ pg_free(output.dst);
+}
+
+#endif /* HAVE_LIBZSTD */
+
#ifdef HAVE_LIBZ
/*
* Functions for zlib compressed output.
@@ -411,6 +655,19 @@ struct cfp
#ifdef HAVE_LIBZ
gzFile compressedfp;
#endif
+
+#ifdef HAVE_LIBZSTD // XXX: this should be a union with a CompressionAlgorithm alg?
+ /* This is a normal file to which we read/write compressed data */
+ struct {
+ FILE *fp;
+ // XXX: use one separate ZSTD_CStream per thread: disable on windows ?
+ ZSTD_CStream *cstream;
+ ZSTD_DStream *dstream;
+ ZSTD_outBuffer output;
+ ZSTD_inBuffer input;
+ } zstd;
+#endif
+
};
static int hasSuffix(const char *filename);
@@ -525,6 +782,31 @@ cfopen(const char *path, const char *mode, Compress *compression)
return fp;
#endif
+#ifdef HAVE_LIBZSTD
+ case COMPR_ALG_ZSTD:
+ fp->zstd.fp = fopen(path, mode);
+ if (fp->zstd.fp == NULL)
+ {
+ free_keep_errno(fp);
+ fp = NULL;
+ }
+ else if (mode[0] == 'w' || mode[0] == 'a' ||
+ strchr(mode, '+') != NULL)
+ {
+ fp->zstd.output.size = ZSTD_CStreamOutSize();
+ fp->zstd.output.dst = pg_malloc0(fp->zstd.output.size);
+ fp->zstd.cstream = ZstdCStreamParams(compression);
+ }
+ else if (strchr(mode, 'r'))
+ {
+ fp->zstd.input.src = pg_malloc0(ZSTD_DStreamInSize());
+ fp->zstd.dstream = ZSTD_createDStream();
+ if (fp->zstd.dstream == NULL)
+ fatal("could not initialize compression library");
+ } // XXX else: bad mode
+ return fp;
+#endif
+
case COMPR_ALG_NONE:
fp->uncompressedfp = fopen(path, mode);
if (fp->uncompressedfp == NULL)
@@ -576,6 +858,31 @@ cfdopen(int fd, const char *mode, Compress *compression)
return fp;
#endif
+#ifdef HAVE_LIBZSTD
+ case COMPR_ALG_ZSTD:
+ fp->zstd.fp = fdopen(fd, mode);
+ if (fp->zstd.fp == NULL)
+ {
+ free_keep_errno(fp);
+ fp = NULL;
+ }
+ else if (mode[0] == 'w' || mode[0] == 'a' ||
+ strchr(mode, '+') != NULL)
+ {
+ fp->zstd.output.size = ZSTD_CStreamOutSize();
+ fp->zstd.output.dst = pg_malloc0(fp->zstd.output.size);
+ fp->zstd.cstream = ZstdCStreamParams(compression);
+ }
+ else if (strchr(mode, 'r'))
+ {
+ fp->zstd.input.src = pg_malloc0(ZSTD_DStreamInSize());
+ fp->zstd.dstream = ZSTD_createDStream();
+ if (fp->zstd.dstream == NULL)
+ fatal("could not initialize compression library");
+ } // XXX else: bad mode
+ return fp;
+#endif
+
case COMPR_ALG_NONE:
fp->uncompressedfp = fdopen(fd, mode);
if (fp->uncompressedfp == NULL)
@@ -617,6 +924,68 @@ cfread(void *ptr, int size, cfp *fp)
}
#endif
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ ZSTD_outBuffer *output = &fp->zstd.output;
+ ZSTD_inBuffer *input = &fp->zstd.input;
+ size_t input_size = ZSTD_DStreamInSize();
+ /* input_size is the allocated size */
+ size_t res, cnt;
+
+ output->size = size;
+ output->dst = ptr;
+ output->pos = 0;
+
+ for (;;)
+ {
+ Assert(input->pos <= input->size);
+ Assert(input->size <= input_size);
+
+ /* If the input is completely consumed, start back at the beginning */
+ if (input->pos == input->size)
+ {
+ /* input->size is size produced by "fread" */
+ input->size = 0;
+ /* input->pos is position consumed by decompress */
+ input->pos = 0;
+ }
+
+ /* read compressed data if we must produce more input */
+ if (input->pos == input->size)
+ {
+ cnt = fread(unconstify(void *, input->src), 1, input_size, fp->zstd.fp);
+ input->size = cnt;
+
+ /* If we have no input to consume, we're done */
+ if (cnt == 0)
+ break;
+ }
+
+ Assert(cnt >= 0);
+ Assert(input->size <= input_size);
+
+ /* Now consume as much as possible */
+ for ( ; input->pos < input->size; )
+ {
+ /* decompress */
+ res = ZSTD_decompressStream(fp->zstd.dstream, output, input);
+ if (res == 0)
+ break; /* End of frame */
+ if (output->pos == output->size)
+ break; /* No more room for output */
+ if (ZSTD_isError(res))
+ fatal("could not decompress data: %s", ZSTD_getErrorName(res));
+ }
+
+ if (output->pos == output->size)
+ break; /* We read all the data that fits */
+ }
+
+ return output->pos;
+ }
+#endif
+
ret = fread(ptr, 1, size, fp->uncompressedfp);
if (ret != size && !feof(fp->uncompressedfp))
READ_ERROR_EXIT(fp->uncompressedfp);
@@ -630,6 +999,35 @@ cfwrite(const void *ptr, int size, cfp *fp)
if (fp->compressedfp)
return gzwrite(fp->compressedfp, ptr, size);
#endif
+
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ size_t res, cnt;
+ ZSTD_outBuffer *output = &fp->zstd.output;
+ ZSTD_inBuffer *input = &fp->zstd.input;
+
+ input->src = ptr;
+ input->size = size;
+ input->pos = 0;
+
+ /* Consume all input, and flush later */
+ while (input->pos != input->size)
+ {
+ output->pos = 0;
+ res = ZSTD_compressStream2(fp->zstd.cstream, output, input, ZSTD_e_continue);
+ if (ZSTD_isError(res))
+ fatal("could not compress data: %s", ZSTD_getErrorName(res));
+
+ cnt = fwrite(output->dst, 1, output->pos, fp->zstd.fp);
+ if (cnt != output->pos)
+ fatal("could not write data: %s", strerror(errno));
+ }
+
+ return size;
+ }
+#endif
+
return fwrite(ptr, 1, size, fp->uncompressedfp);
}
@@ -652,6 +1050,21 @@ cfgetc(cfp *fp)
return ret;
}
#endif
+
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ if (cfread(&ret, 1, fp) != 1)
+ {
+ if (feof(fp->zstd.fp))
+ fatal("could not read from input file: end of file");
+ else
+ fatal("could not read from input file: %s", strerror(errno));
+ }
+ return ret;
+ }
+#endif
+
ret = fgetc(fp->uncompressedfp);
if (ret == EOF)
READ_ERROR_EXIT(fp->uncompressedfp);
@@ -665,6 +1078,31 @@ cfgets(cfp *fp, char *buf, int len)
if (fp->compressedfp)
return gzgets(fp->compressedfp, buf, len);
#endif
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ /*
+ * Read one byte at a time until newline or EOF.
+ * This is only used to read the list of blobs, and the I/O is
+ * buffered anyway.
+ */
+ int i, res;
+ for (i = 0; i < len - 1; ++i)
+ {
+ res = cfread(&buf[i], 1, fp);
+ if (res != 1)
+ break;
+ if (buf[i] == '\n')
+ {
+ ++i;
+ break;
+ }
+ }
+ buf[i] = '\0';
+ return i > 0 ? buf : 0;
+ }
+#endif
+
return fgets(buf, len, fp->uncompressedfp);
}
@@ -688,6 +1126,44 @@ cfclose(cfp *fp)
}
#endif
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ {
+ ZSTD_outBuffer *output = &fp->zstd.output;
+ ZSTD_inBuffer *input = &fp->zstd.input;
+ size_t res, cnt;
+
+ if (fp->zstd.cstream)
+ {
+ for (;;)
+ {
+ output->pos = 0;
+ res = ZSTD_compressStream2(fp->zstd.cstream, output, input, ZSTD_e_end);
+ if (ZSTD_isError(res))
+ fatal("could not compress data: %s", ZSTD_getErrorName(res));
+ cnt = fwrite(output->dst, 1, output->pos, fp->zstd.fp);
+ if (cnt != output->pos)
+ fatal("could not write data: %s", strerror(errno));
+ if (res == 0)
+ break;
+ }
+
+ ZSTD_freeCStream(fp->zstd.cstream);
+ pg_free(fp->zstd.output.dst);
+ }
+
+ if (fp->zstd.dstream)
+ {
+ ZSTD_freeDStream(fp->zstd.dstream);
+ pg_free(unconstify(void *, fp->zstd.input.src));
+ }
+
+ result = fclose(fp->zstd.fp);
+ fp->zstd.fp = NULL;
+ return result;
+ }
+#endif
+
result = fclose(fp->uncompressedfp);
fp->uncompressedfp = NULL;
free_keep_errno(fp);
@@ -702,6 +1178,10 @@ cfeof(cfp *fp)
return gzeof(fp->compressedfp);
#endif
+#ifdef HAVE_LIBZSTD
+ if (fp->zstd.fp)
+ return feof(fp->zstd.fp);
+#endif
return feof(fp->uncompressedfp);
}
diff --git a/src/bin/pg_dump/pg_backup.h b/src/bin/pg_dump/pg_backup.h
index f2390b7937..19ff6248d5 100644
--- a/src/bin/pg_dump/pg_backup.h
+++ b/src/bin/pg_dump/pg_backup.h
@@ -77,6 +77,7 @@ typedef enum
COMPR_ALG_DEFAULT = -1,
COMPR_ALG_NONE,
COMPR_ALG_LIBZ,
+ COMPR_ALG_ZSTD,
} CompressionAlgorithm;
/* Should be called "method" or "library" ? */
@@ -88,6 +89,19 @@ typedef struct Compress {
* are all integer, though.
*/
bool level_set;
+
+ /*
+ * This could be a union across all compress algorithms, but
+ * keeping as separate structs allows checking that options are
+ * not specified for a different algorithm than selected.
+ */
+
+ struct {
+ bool longdistance;
+ bool checksum;
+ bool rsyncable;
+ int threads;
+ } zstd;
} Compress;
diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h
index 9f511b49b9..da2eb53277 100644
--- a/src/bin/pg_dump/pg_backup_archiver.h
+++ b/src/bin/pg_dump/pg_backup_archiver.h
@@ -56,6 +56,10 @@ typedef struct _z_stream
typedef z_stream *z_streamp;
#endif
+#ifdef HAVE_LIBZSTD
+#include <zstd.h>
+#endif /* HAVE_LIBZSTD */
+
/* Data block types */
#define BLK_DATA 1
#define BLK_BLOBS 3
diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c
index 75c1bf22e4..b8efeb8ca7 100644
--- a/src/bin/pg_dump/pg_backup_directory.c
+++ b/src/bin/pg_dump/pg_backup_directory.c
@@ -393,8 +393,12 @@ _PrintFileData(ArchiveHandle *AH, char *filename)
if (!cfp)
fatal("could not open input file \"%s\": %m", filename);
- buf = pg_malloc(ZLIB_OUT_SIZE);
- buflen = ZLIB_OUT_SIZE;
+ /*
+ * zstd prefers a 128kB buffer. The allocation cannot happen in
+ * cfread, since the "cfp" is an opaque type.
+ */
+ buf = pg_malloc(128*1024);
+ buflen = 128*1024;
while ((cnt = cfread(buf, buflen, cfp)))
{
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 75985fd4d3..7c2f7a9ca3 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -356,6 +356,12 @@ parse_compression(const char *optarg, Compress *compress)
compress->level = atoi(1+eq);
compress->level_set = true;
}
+ else if (strncmp(optarg, "zstdlong", len) == 0)
+ compress->zstd.longdistance = atoi(1+eq);
+ else if (strncmp(optarg, "checksum", len) == 0)
+ compress->zstd.checksum = atoi(1+eq);
+ else if (strncmp(optarg, "threads", len) == 0)
+ compress->zstd.threads = atoi(1+eq);
else
{
pg_log_error("unknown compression setting: %s", optarg);
@@ -367,11 +373,31 @@ parse_compression(const char *optarg, Compress *compress)
break;
}
+ /* XXX: zstd will check its own compression level later */
+ if (compress->alg != COMPR_ALG_ZSTD)
+ {
+ Compress nullopts = {0};
+
+ if (compress->level < 0 || compress->level > 9)
+ {
+ pg_log_error("compression level must be in range 0..9");
+ exit_nicely(1);
+ }
+
+// XXX: needs to set default alg first
+ if (memcmp(&compress->zstd, &nullopts.zstd, sizeof(nullopts.zstd)) != 0)
+ {
+ pg_log_error("compression option not supported with this algorithm");
+ exit_nicely(1);
+ }
+ }
+
if (!compress->level_set)
{ // XXX
const int default_compress_level[] = {
0, /* COMPR_ALG_NONE */
Z_DEFAULT_COMPRESSION, /* COMPR_ALG_ZLIB */
+ 0, // XXX: ZSTD_CLEVEL_DEFAULT, /* COMPR_ALG_ZSTD */
};
compress->level = default_compress_level[compress->alg];
@@ -764,6 +790,11 @@ main(int argc, char **argv)
compress.alg = COMPR_ALG_LIBZ;
compress.level = Z_DEFAULT_COMPRESSION;
#endif
+
+#ifdef HAVE_LIBZSTD
+ compress.alg = COMPR_ALG_ZSTD; // Set default for testing purposes
+ compress.level = ZSTD_CLEVEL_DEFAULT;
+#endif
}
else
{
@@ -780,6 +811,14 @@ main(int argc, char **argv)
}
#endif
+#ifndef HAVE_LIBZSTD
+ if (compress.alg == COMPR_ALG_ZSTD)
+ {
+ pg_log_warning("requested compression not available in this installation -- archive will be uncompressed");
+ compress.alg = 0;
+ }
+#endif
+
/*
* If emitting an archive format, we always want to emit a DATABASE item,
* in case --create is specified at pg_restore time.
diff --git a/src/include/pg_config.h.in b/src/include/pg_config.h.in
index de8f838e53..da35415c72 100644
--- a/src/include/pg_config.h.in
+++ b/src/include/pg_config.h.in
@@ -346,6 +346,9 @@
/* Define to 1 if you have the `z' library (-lz). */
#undef HAVE_LIBZ
+/* Define to 1 if you have the `zstd' library (-lzstd). */
+#undef HAVE_LIBZSTD
+
/* Define to 1 if you have the `link' function. */
#undef HAVE_LINK
diff --git a/src/tools/msvc/Solution.pm b/src/tools/msvc/Solution.pm
index 22d6abd367..a101366b4c 100644
--- a/src/tools/msvc/Solution.pm
+++ b/src/tools/msvc/Solution.pm
@@ -307,6 +307,7 @@ sub GenerateFiles
HAVE_LIBXML2 => undef,
HAVE_LIBXSLT => undef,
HAVE_LIBZ => $self->{options}->{zlib} ? 1 : undef,
+ HAVE_LIBZSTD => $self->{options}->{zstd} ? 1 : undef,
HAVE_LINK => undef,
HAVE_LOCALE_T => 1,
HAVE_LONG_INT_64 => undef,
--
2.17.0
--GRPZ8SYKNexpdSJ7
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment; filename="0007-fix-comments.patch"
^ permalink raw reply [nested|flat] 6+ messages in thread
* Assert when executing query on partitioned table
@ 2025-02-20 11:13 Dmitry Koval <[email protected]>
0 siblings, 1 reply; 6+ messages in thread
From: Dmitry Koval @ 2025-02-20 11:13 UTC (permalink / raw)
To: pgsql-hackers
Hi!
I got an Assert when executing an "INSERT ... ON CONFLICT ... UPDATE
..." query on partitioned table. Managed to reproduce this situation.
Reproduction order.
-------------------
1) Apply the patch
[v1-0001-Triggering-Assert-on-query-with-ON-CONFLICT.patch] to "master"
branch.
2) Build postgres with key "--enable-injection-points" + build an
extension "injection_points" (src/test/modules/injection_points).
3) Run isolation test onconflict.spec:
make check -C src/test/modules/injection_points
Assert is triggered in postgres with stack, see attached file [stack.txt].
Clarification.
--------------
In the query "INSERT ... ON CONFLICT ... UPDATE ..." when executing
INSERT, a conflict is triggered. But when trying to execute UPDATE, our
tuple has already been moved to another partition and Assert is triggered.
Fixing.
-------
I suggest replace Assert with an error message, see
[v1-0001-draft-of-fix.patch]. This is not a final fix as I am confused
by the comment for Assert: "we don't support an UPDATE of INSERT ON
CONFLICT for a partitioned table".
(Why "don't support an UPDATE"?
It's not forbidden by syntax or errors ...)
--
With best regards,
Dmitry Koval
Postgres Professional: http://postgrespro.com
#0 0x000072313029eb1c in pthread_kill () from /lib/x86_64-linux-gnu/libc.so.6
#1 0x000072313024526e in raise () from /lib/x86_64-linux-gnu/libc.so.6
#2 0x00007231302288ff in abort () from /lib/x86_64-linux-gnu/libc.so.6
#3 0x00005e3fccec9ff8 in ExceptionalCondition (conditionName=0x5e3fcd0b49c0 "!ItemPointerIndicatesMovedPartitions(&tmfd.ctid)", fileName=0x5e3fcd0b3fe0 "nodeModifyTable.c", lineNumber=2797) at assert.c:66
#4 0x00005e3fcca2264f in ExecOnConflictUpdate (context=0x7ffd08eee220, resultRelInfo=0x5e3fce06a310, conflictTid=0x7ffd08eee0cc, excludedSlot=0x5e3fce069640, canSetTag=true, returning=0x7ffd08eee0d8) at nodeModifyTable.c:2797
#5 0x00005e3fcca1fd6f in ExecInsert (context=0x7ffd08eee220, resultRelInfo=0x5e3fce06a310, slot=0x5e3fce069640, canSetTag=true, inserted_tuple=0x0, insert_destrel=0x0) at nodeModifyTable.c:1133
#6 0x00005e3fcca250e3 in ExecModifyTable (pstate=0x5e3fce068cf0) at nodeModifyTable.c:4324
#7 0x00005e3fcc9e0454 in ExecProcNodeFirst (node=0x5e3fce068cf0) at execProcnode.c:469
#8 0x00005e3fcc9d2f5b in ExecProcNode (node=0x5e3fce068cf0) at ../../../src/include/executor/executor.h:272
#9 0x00005e3fcc9d5de8 in ExecutePlan (queryDesc=0x5e3fce10edf0, operation=CMD_INSERT, sendTuples=false, numberTuples=0, direction=ForwardScanDirection, dest=0x5e3fce1149e8) at execMain.c:1675
#10 0x00005e3fcc9d3618 in standard_ExecutorRun (queryDesc=0x5e3fce10edf0, direction=ForwardScanDirection, count=0) at execMain.c:364
#11 0x00005e3fcc9d3479 in ExecutorRun (queryDesc=0x5e3fce10edf0, direction=ForwardScanDirection, count=0) at execMain.c:301
#12 0x00005e3fcccc0ab0 in ProcessQuery (plan=0x5e3fce042778, sourceText=0x5e3fce040ef0 "INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x;", params=0x0, queryEnv=0x0, dest=0x5e3fce1149e8, qc=0x7ffd08eee670) at pquery.c:160
#13 0x00005e3fcccc25e0 in PortalRunMulti (portal=0x5e3fce0bf8a0, isTopLevel=true, setHoldSnapshot=false, dest=0x5e3fce1149e8, altdest=0x5e3fce1149e8, qc=0x7ffd08eee670) at pquery.c:1271
#14 0x00005e3fcccc1aec in PortalRun (portal=0x5e3fce0bf8a0, count=9223372036854775807, isTopLevel=true, dest=0x5e3fce1149e8, altdest=0x5e3fce1149e8, qc=0x7ffd08eee670) at pquery.c:787
#15 0x00005e3fcccba5bf in exec_simple_query (query_string=0x5e3fce040ef0 "INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x;") at postgres.c:1271
#16 0x00005e3fcccbf954 in PostgresMain (dbname=0x5e3fce078fa0 "isolation_regression", username=0x5e3fce078f88 "kovdb") at postgres.c:4691
#17 0x00005e3fcccb5fdf in BackendMain (startup_data=0x7ffd08eee8f0 "", startup_data_len=4) at backend_startup.c:107
#18 0x00005e3fccbc24d7 in postmaster_child_launch (child_type=B_BACKEND, child_slot=2, startup_data=0x7ffd08eee8f0 "", startup_data_len=4, client_sock=0x7ffd08eee950) at launch_backend.c:274
#19 0x00005e3fccbc8ec3 in BackendStartup (client_sock=0x7ffd08eee950) at postmaster.c:3519
#20 0x00005e3fccbc646c in ServerLoop () at postmaster.c:1688
#21 0x00005e3fccbc5d64 in PostmasterMain (argc=8, argv=0x5e3fcdffe970) at postmaster.c:1386
#22 0x00005e3fcca67011 in main (argc=8, argv=0x5e3fcdffe970) at main.c:230
From 02b89799a2f139a3cab41b1df761103cf8627098 Mon Sep 17 00:00:00 2001
From: Koval Dmitry <[email protected]>
Date: Wed, 19 Feb 2025 13:09:33 +0300
Subject: [PATCH v1] Triggering Assert on query with ON CONFLICT
---
src/backend/executor/nodeModifyTable.c | 2 ++
src/test/modules/injection_points/Makefile | 11 +++----
.../injection_points/specs/onconflict.spec | 30 +++++++++++++++++++
3 files changed, 38 insertions(+), 5 deletions(-)
create mode 100644 src/test/modules/injection_points/specs/onconflict.spec
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index b0fe50075ad..01946fdb0d8 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -67,6 +67,7 @@
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/datum.h"
+#include "utils/injection_point.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
@@ -1128,6 +1129,7 @@ ExecInsert(ModifyTableContext *context,
*/
TupleTableSlot *returning = NULL;
+ INJECTION_POINT("before-on-conflict-update");
if (ExecOnConflictUpdate(context, resultRelInfo,
&conflictTid, slot, canSetTag,
&returning))
diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile
index e680991f8d4..b1915800760 100644
--- a/src/test/modules/injection_points/Makefile
+++ b/src/test/modules/injection_points/Makefile
@@ -11,15 +11,16 @@ EXTENSION = injection_points
DATA = injection_points--1.0.sql
PGFILEDESC = "injection_points - facility for injection points"
-REGRESS = injection_points hashagg reindex_conc
-REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress
+#REGRESS = injection_points hashagg reindex_conc
+#REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress
-ISOLATION = basic inplace syscache-update-pruned
+#ISOLATION = basic inplace syscache-update-pruned
+ISOLATION = onconflict
-TAP_TESTS = 1
+#TAP_TESTS = 1
# The injection points are cluster-wide, so disable installcheck
-NO_INSTALLCHECK = 1
+#NO_INSTALLCHECK = 1
export enable_injection_points
diff --git a/src/test/modules/injection_points/specs/onconflict.spec b/src/test/modules/injection_points/specs/onconflict.spec
new file mode 100644
index 00000000000..cdfe7942029
--- /dev/null
+++ b/src/test/modules/injection_points/specs/onconflict.spec
@@ -0,0 +1,30 @@
+setup
+{
+ CREATE EXTENSION injection_points;
+
+ CREATE TABLE t_int (i int PRIMARY KEY, v int, x int) PARTITION BY RANGE (i);
+ CREATE TABLE t_int_1 PARTITION OF t_int FOR VALUES FROM (1) TO (100);
+ CREATE TABLE t_int_2 PARTITION OF t_int FOR VALUES FROM (100) TO (200);
+
+ INSERT INTO t_int VALUES (1, 10, 100);
+}
+
+teardown
+{
+ DROP TABLE t_int;
+ DROP EXTENSION injection_points;
+}
+
+session s1
+step s1 { BEGIN; }
+step s1ipa { SELECT injection_points_attach('before-on-conflict-update', 'wait'); }
+step s1i { INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x; }
+step s1c { COMMIT; }
+
+session s2
+step s2 { BEGIN; }
+step s2ipw { SELECT injection_points_wakeup('before-on-conflict-update'); }
+step s2u { UPDATE t_int SET i = i + 150 WHERE i = 1; }
+step s2c { COMMIT; }
+
+permutation s1 s1ipa s1i(s2c) s2 s2u s2ipw s2c
--
2.34.1
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index b0fe50075ad..4b0ae8bd32c 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -2792,7 +2792,10 @@ ExecOnConflictUpdate(ModifyTableContext *context,
* be lock is moved to another partition due to concurrent update
* of the partition key.
*/
- Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
+ if (ItemPointerIndicatesMovedPartitions(&tmfd.ctid))
+ ereport(ERROR,
+ (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+ errmsg("tuple to be updated was already moved to another partition due to concurrent update")));
/*
* Tell caller to try again from the very start.
Attachments:
[text/plain] stack.txt (3.1K, ../../[email protected]/2-stack.txt)
download | inline:
#0 0x000072313029eb1c in pthread_kill () from /lib/x86_64-linux-gnu/libc.so.6
#1 0x000072313024526e in raise () from /lib/x86_64-linux-gnu/libc.so.6
#2 0x00007231302288ff in abort () from /lib/x86_64-linux-gnu/libc.so.6
#3 0x00005e3fccec9ff8 in ExceptionalCondition (conditionName=0x5e3fcd0b49c0 "!ItemPointerIndicatesMovedPartitions(&tmfd.ctid)", fileName=0x5e3fcd0b3fe0 "nodeModifyTable.c", lineNumber=2797) at assert.c:66
#4 0x00005e3fcca2264f in ExecOnConflictUpdate (context=0x7ffd08eee220, resultRelInfo=0x5e3fce06a310, conflictTid=0x7ffd08eee0cc, excludedSlot=0x5e3fce069640, canSetTag=true, returning=0x7ffd08eee0d8) at nodeModifyTable.c:2797
#5 0x00005e3fcca1fd6f in ExecInsert (context=0x7ffd08eee220, resultRelInfo=0x5e3fce06a310, slot=0x5e3fce069640, canSetTag=true, inserted_tuple=0x0, insert_destrel=0x0) at nodeModifyTable.c:1133
#6 0x00005e3fcca250e3 in ExecModifyTable (pstate=0x5e3fce068cf0) at nodeModifyTable.c:4324
#7 0x00005e3fcc9e0454 in ExecProcNodeFirst (node=0x5e3fce068cf0) at execProcnode.c:469
#8 0x00005e3fcc9d2f5b in ExecProcNode (node=0x5e3fce068cf0) at ../../../src/include/executor/executor.h:272
#9 0x00005e3fcc9d5de8 in ExecutePlan (queryDesc=0x5e3fce10edf0, operation=CMD_INSERT, sendTuples=false, numberTuples=0, direction=ForwardScanDirection, dest=0x5e3fce1149e8) at execMain.c:1675
#10 0x00005e3fcc9d3618 in standard_ExecutorRun (queryDesc=0x5e3fce10edf0, direction=ForwardScanDirection, count=0) at execMain.c:364
#11 0x00005e3fcc9d3479 in ExecutorRun (queryDesc=0x5e3fce10edf0, direction=ForwardScanDirection, count=0) at execMain.c:301
#12 0x00005e3fcccc0ab0 in ProcessQuery (plan=0x5e3fce042778, sourceText=0x5e3fce040ef0 "INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x;", params=0x0, queryEnv=0x0, dest=0x5e3fce1149e8, qc=0x7ffd08eee670) at pquery.c:160
#13 0x00005e3fcccc25e0 in PortalRunMulti (portal=0x5e3fce0bf8a0, isTopLevel=true, setHoldSnapshot=false, dest=0x5e3fce1149e8, altdest=0x5e3fce1149e8, qc=0x7ffd08eee670) at pquery.c:1271
#14 0x00005e3fcccc1aec in PortalRun (portal=0x5e3fce0bf8a0, count=9223372036854775807, isTopLevel=true, dest=0x5e3fce1149e8, altdest=0x5e3fce1149e8, qc=0x7ffd08eee670) at pquery.c:787
#15 0x00005e3fcccba5bf in exec_simple_query (query_string=0x5e3fce040ef0 "INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x;") at postgres.c:1271
#16 0x00005e3fcccbf954 in PostgresMain (dbname=0x5e3fce078fa0 "isolation_regression", username=0x5e3fce078f88 "kovdb") at postgres.c:4691
#17 0x00005e3fcccb5fdf in BackendMain (startup_data=0x7ffd08eee8f0 "", startup_data_len=4) at backend_startup.c:107
#18 0x00005e3fccbc24d7 in postmaster_child_launch (child_type=B_BACKEND, child_slot=2, startup_data=0x7ffd08eee8f0 "", startup_data_len=4, client_sock=0x7ffd08eee950) at launch_backend.c:274
#19 0x00005e3fccbc8ec3 in BackendStartup (client_sock=0x7ffd08eee950) at postmaster.c:3519
#20 0x00005e3fccbc646c in ServerLoop () at postmaster.c:1688
#21 0x00005e3fccbc5d64 in PostmasterMain (argc=8, argv=0x5e3fcdffe970) at postmaster.c:1386
#22 0x00005e3fcca67011 in main (argc=8, argv=0x5e3fcdffe970) at main.c:230
[text/plain] v1-0001-Triggering-Assert-on-query-with-ON-CONFLICT.patch (3.2K, ../../[email protected]/3-v1-0001-Triggering-Assert-on-query-with-ON-CONFLICT.patch)
download | inline diff:
From 02b89799a2f139a3cab41b1df761103cf8627098 Mon Sep 17 00:00:00 2001
From: Koval Dmitry <[email protected]>
Date: Wed, 19 Feb 2025 13:09:33 +0300
Subject: [PATCH v1] Triggering Assert on query with ON CONFLICT
---
src/backend/executor/nodeModifyTable.c | 2 ++
src/test/modules/injection_points/Makefile | 11 +++----
.../injection_points/specs/onconflict.spec | 30 +++++++++++++++++++
3 files changed, 38 insertions(+), 5 deletions(-)
create mode 100644 src/test/modules/injection_points/specs/onconflict.spec
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index b0fe50075ad..01946fdb0d8 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -67,6 +67,7 @@
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/datum.h"
+#include "utils/injection_point.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
@@ -1128,6 +1129,7 @@ ExecInsert(ModifyTableContext *context,
*/
TupleTableSlot *returning = NULL;
+ INJECTION_POINT("before-on-conflict-update");
if (ExecOnConflictUpdate(context, resultRelInfo,
&conflictTid, slot, canSetTag,
&returning))
diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile
index e680991f8d4..b1915800760 100644
--- a/src/test/modules/injection_points/Makefile
+++ b/src/test/modules/injection_points/Makefile
@@ -11,15 +11,16 @@ EXTENSION = injection_points
DATA = injection_points--1.0.sql
PGFILEDESC = "injection_points - facility for injection points"
-REGRESS = injection_points hashagg reindex_conc
-REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress
+#REGRESS = injection_points hashagg reindex_conc
+#REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress
-ISOLATION = basic inplace syscache-update-pruned
+#ISOLATION = basic inplace syscache-update-pruned
+ISOLATION = onconflict
-TAP_TESTS = 1
+#TAP_TESTS = 1
# The injection points are cluster-wide, so disable installcheck
-NO_INSTALLCHECK = 1
+#NO_INSTALLCHECK = 1
export enable_injection_points
diff --git a/src/test/modules/injection_points/specs/onconflict.spec b/src/test/modules/injection_points/specs/onconflict.spec
new file mode 100644
index 00000000000..cdfe7942029
--- /dev/null
+++ b/src/test/modules/injection_points/specs/onconflict.spec
@@ -0,0 +1,30 @@
+setup
+{
+ CREATE EXTENSION injection_points;
+
+ CREATE TABLE t_int (i int PRIMARY KEY, v int, x int) PARTITION BY RANGE (i);
+ CREATE TABLE t_int_1 PARTITION OF t_int FOR VALUES FROM (1) TO (100);
+ CREATE TABLE t_int_2 PARTITION OF t_int FOR VALUES FROM (100) TO (200);
+
+ INSERT INTO t_int VALUES (1, 10, 100);
+}
+
+teardown
+{
+ DROP TABLE t_int;
+ DROP EXTENSION injection_points;
+}
+
+session s1
+step s1 { BEGIN; }
+step s1ipa { SELECT injection_points_attach('before-on-conflict-update', 'wait'); }
+step s1i { INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x; }
+step s1c { COMMIT; }
+
+session s2
+step s2 { BEGIN; }
+step s2ipw { SELECT injection_points_wakeup('before-on-conflict-update'); }
+step s2u { UPDATE t_int SET i = i + 150 WHERE i = 1; }
+step s2c { COMMIT; }
+
+permutation s1 s1ipa s1i(s2c) s2 s2u s2ipw s2c
--
2.34.1
[text/plain] v1-0001-draft-of-fix.patch (759B, ../../[email protected]/4-v1-0001-draft-of-fix.patch)
download | inline diff:
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index b0fe50075ad..4b0ae8bd32c 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -2792,7 +2792,10 @@ ExecOnConflictUpdate(ModifyTableContext *context,
* be lock is moved to another partition due to concurrent update
* of the partition key.
*/
- Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
+ if (ItemPointerIndicatesMovedPartitions(&tmfd.ctid))
+ ereport(ERROR,
+ (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+ errmsg("tuple to be updated was already moved to another partition due to concurrent update")));
/*
* Tell caller to try again from the very start.
^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Assert when executing query on partitioned table
@ 2025-03-08 21:28 Joseph Koshakow <[email protected]>
parent: Dmitry Koval <[email protected]>
0 siblings, 1 reply; 6+ messages in thread
From: Joseph Koshakow @ 2025-03-08 21:28 UTC (permalink / raw)
To: Dmitry Koval <[email protected]>; +Cc: pgsql-hackers
On Thu, Feb 20, 2025 at 6:14 AM Dmitry Koval <[email protected]> wrote:
> I got an Assert when executing an "INSERT ... ON CONFLICT ... UPDATE
> ..." query on partitioned table. Managed to reproduce this situation.
I was able to reproduce the assert with your instructions.
> I suggest replace Assert with an error message, see
> [v1-0001-draft-of-fix.patch]. This is not a final fix as I am confused
> by the comment for Assert: "we don't support an UPDATE of INSERT ON
> CONFLICT for a partitioned table".
> (Why "don't support an UPDATE"?
> It's not forbidden by syntax or errors ...)
The assert was introduced commit f16241 [0]. Specifically it was added as a
result of this discussion [1]
> On Wed, Nov 29, 2017 at 7:51 AM, Amit Kapila
<amit(dot)kapila16(at)gmail(dot)com> wrote:
>> On Tue, Nov 28, 2017 at 5:58 PM, amul sul <sulamul(at)gmail(dot)com>
wrote:
>>> On Sat, Nov 25, 2017 at 11:39 AM, Amit Kapila
<amit(dot)kapila16(at)gmail(dot)com> wrote:
>>>> On Thu, Nov 23, 2017 at 5:18 PM, amul sul <sulamul(at)gmail(dot)com>
wrote:
>>>>> On Sat, Nov 11, 2017 at 1:05 AM, Robert Haas
<robertmhaas(at)gmail(dot)com> wrote:
>>>>>> On Wed, Sep 27, 2017 at 7:07 AM, amul sul <sulamul(at)gmail(dot)com>
wrote:
>>>>>>>
>>>> 1.
>>>> @@ -1480,6 +1493,10 @@ ExecOnConflictUpdate(ModifyTableState *mtstate,
>>>> ereport(ERROR,
>>>> (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
>>>> errmsg("could not serialize access due to concurrent update")));
>>>> + if
(!BlockNumberIsValid(BlockIdGetBlockNumber(&((hufd.ctid).ip_blkid))))
>>>> + ereport(ERROR,
>>>> + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
>>>> + errmsg("tuple to be updated was already moved to an another
>>>> partition due to concurrent update")));
>>>>
>>>> Why do you think we need this check in the OnConflictUpdate path? I
>>>> think we don't it here because we are going to relinquish this version
>>>> of the tuple and will start again and might fetch some other row
>>>> version. Also, we don't support Insert .. On Conflict Update with
>>>> partitioned tables, see[1], which is also an indication that at the
>>>> very least we don't need it now.
>>>>
>>> Agreed, even though this case will never going to be anytime soon
>>> shouldn't we have a check for invalid block id? IMHO, we should have
>>> this check and error report or assert, thoughts?
>>>
>>
>> I feel adding code which can't be hit (even if it is error handling)
>> is not a good idea. I think having an Assert should be okay, but
>> please write comments to explain the reason for adding an Assert.
>>
>
> Agree, updated in the attached patch.
So if I understand correctly, at the time the assert was added, we in
fact did not support UPDATE of INSERT ON CONFLICT for a partitioned
table. However, since then we've added support but nobody removed or
altered the assert.
I'm not very familiar with this code, but from that discussion it
sounds like your solution of converting the assert to an error is
correct.
I don't fully understand why an error is needed though. This specific
case will return false which will signal the caller to retry the
insert. Just as an experiment I tried deleting the assert (attached
patch) and running your test. Everything behaved as expected and
nothing blew up. It also seems to pass a CI run just fine [2]. It also
passes `make check && make check-world` locally.
Hopefully someone from the original thread can shed some light on
whether an error is needed or not.
[0]
https://git.postgresql.org/gitweb/?p=postgresql.git;a=commitdiff;h=f16241bef7cc271bff60e23de2f827a10...
[1]
https://www.postgresql.org/message-id/flat/CAAJ_b95PkwojoYfz0bzXU8OokcTVGzN6vYGCNVUukeUDrnF3dw%40mai...
[2] https://github.com/jkosh44/postgres/runs/38431219796
Thanks,
Joseph Koshakow
Attachments:
[text/x-patch] v1-0001-Remove-assert-for-update-on-conflict.patch (1.3K, ../../CAAvxfHciZfHN22y5kgXAGrd8b5Yn7JVpxJ_qFATiXEk_wV-a-w@mail.gmail.com/3-v1-0001-Remove-assert-for-update-on-conflict.patch)
download | inline diff:
From 3959bc8d53b29d3b7ba79d9cf923f9da65d43d5b Mon Sep 17 00:00:00 2001
From: Joseph Koshakow <[email protected]>
Date: Sat, 8 Mar 2025 15:39:47 -0500
Subject: [PATCH v1] Remove assert for update on conflict
This commit removes an assert that is no longer valid. The assert was
making an assumption that INSERT .. ON CONFLIT UPDATE .. was not
allowed on partitioned tables. This assumption is no longer valid and
the assert is possible to be triggered.
---
src/backend/executor/nodeModifyTable.c | 8 --------
1 file changed, 8 deletions(-)
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index b0fe50075a..f33a273cfa 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -2786,14 +2786,6 @@ ExecOnConflictUpdate(ModifyTableContext *context,
(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
errmsg("could not serialize access due to concurrent update")));
- /*
- * As long as we don't support an UPDATE of INSERT ON CONFLICT for
- * a partitioned table we shouldn't reach to a case where tuple to
- * be lock is moved to another partition due to concurrent update
- * of the partition key.
- */
- Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
-
/*
* Tell caller to try again from the very start.
*
--
2.43.0
^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Assert when executing query on partitioned table
@ 2025-03-15 14:57 Joseph Koshakow <[email protected]>
parent: Joseph Koshakow <[email protected]>
0 siblings, 1 reply; 6+ messages in thread
From: Joseph Koshakow @ 2025-03-15 14:57 UTC (permalink / raw)
To: Dmitry Koval <[email protected]>; +Cc: pgsql-hackers
On Sat, Mar 8, 2025 at 4:28 PM Joseph Koshakow <[email protected]> wrote:
>
> The assert was introduced commit f16241
>
> So if I understand correctly, at the time the assert was added, we in
> fact did not support UPDATE of INSERT ON CONFLICT for a partitioned
> table. However, since then we've added support but nobody removed or
> altered the assert.
I tried to validate this by checking out f16241, and running the following:
psql (11devel)
Type "help" for help.
test=# CREATE TABLE t_int (i int PRIMARY KEY, v int, x int) PARTITION
BY RANGE (i);
CREATE TABLE t_int_1 PARTITION OF t_int FOR VALUES FROM (1) TO (100);
CREATE TABLE t_int_2 PARTITION OF t_int FOR VALUES FROM (100) TO
(200);
INSERT INTO t_int VALUES (1, 10, 100);
CREATE TABLE
CREATE TABLE
CREATE TABLE
INSERT 0 1
test=# INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE
SET x = excluded.x;
INSERT 0 1
So it looks like when that assert was added, we *did* support
INSERT .. ON CONFLICT UPDATE for partitioned tables. So I'm even more
confused about the conversation and assert. You can even update the
partitions directly.
test=# INSERT INTO t_int_1 VALUES (1, 11, 111) ON CONFLICT (i) DO
UPDATE SET x = excluded.x;
INSERT 0 1
test=# INSERT INTO t_int_2 VALUES (150, 11, 111) ON CONFLICT (i) DO
UPDATE SET x = excluded.x;
INSERT 0 1
Thanks,
Joseph Koshakow
^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Assert when executing query on partitioned table
@ 2025-12-18 23:27 Dmitry Koval <[email protected]>
parent: Joseph Koshakow <[email protected]>
0 siblings, 1 reply; 6+ messages in thread
From: Dmitry Koval @ 2025-12-18 23:27 UTC (permalink / raw)
To: pgsql-hackers
Hi!
Unfortunately, the issue is still relevant, so I'm updating the first
post and patches. I got an Assert when executing an "INSERT ... ON
CONFLICT ... UPDATE ..." query on partitioned table.
Reproduction order.
-------------------
1) Apply the patch
[v1-0001-Triggering-Assert-on-query-with-ON-CONFLICT.no_cfbot]
to "master" branch:
git am v1-0001-Triggering-Assert-on-query-with-ON-CONFLICT.no_cfbot
2) Build postgres with "--enable-injection-points --enable-cassert",
for example:
./configure --enable-injection-points --enable-debug --enable-cassert
>/dev/null && make -j4 -s
3) Run isolation test onconflict.spec:
make check -C src/test/modules/injection_points
Assert is triggered in postgres, with part of stack:
...
#3 0x0000624cb170e05f in ExceptionalCondition (
conditionName=conditionName@entry=0x624cb17fcbf0
"!ItemPointerIndicatesMovedPartitions(&tmfd.ctid)",
fileName=fileName@entry=0x624cb1785ea7 "nodeModifyTable.c",
lineNumber=lineNumber@entry=2819) at assert.c:65
#4 0x0000624cb13f1412 in ExecOnConflictUpdate (returning=<synthetic
pointer>, canSetTag=true, excludedSlot=0x624cb2a61750,
conflictTid=0x7fff95cd827a, resultRelInfo=0x624cb2a62420,
context=0x7fff95cd8340) at nodeModifyTable.c:2819
...
Clarification.
--------------
In the query "INSERT ... ON CONFLICT ... UPDATE ..." when executing
INSERT, a conflict is triggered. But when trying to execute UPDATE, our
tuple has already been moved to another partition and Assert is
triggered. I think this is a correct situation and in this case we
should generate an error instead of Assert.
Fixing.
-------
Patch [v1-0001-Replace-invalid-Assert-with-ereport-ERROR.patch].
For testing need to apply
[v1-0001-Replace-invalid-Assert-with-ereport-ERROR.patch]:
git am v1-0001-Replace-invalid-Assert-with-ereport-ERROR.patch
rebuild postgres (2) and run test again (3),
--
With best regards,
Dmitry Koval
Postgres Professional: http://postgrespro.com
From 8780e66fc759872e06746c9c214dec52644b2e93 Mon Sep 17 00:00:00 2001
From: Koval Dmitry <[email protected]>
Date: Wed, 19 Feb 2025 13:09:33 +0300
Subject: [PATCH v1] Triggering Assert on query with ON CONFLICT
---
src/backend/executor/nodeModifyTable.c | 1 +
src/test/modules/injection_points/Makefile | 6 +++-
.../injection_points/expected/onconflict.out | 23 ++++++++++++++
.../injection_points/specs/onconflict.spec | 30 +++++++++++++++++++
4 files changed, 59 insertions(+), 1 deletion(-)
create mode 100644 src/test/modules/injection_points/expected/onconflict.out
create mode 100644 src/test/modules/injection_points/specs/onconflict.spec
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 874b71e660..d131abcd67 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -1148,6 +1148,7 @@ ExecInsert(ModifyTableContext *context,
*/
TupleTableSlot *returning = NULL;
+ INJECTION_POINT("before-on-conflict-update", NULL);
if (ExecOnConflictUpdate(context, resultRelInfo,
&conflictTid, slot, canSetTag,
&returning))
diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile
index bfdb3f5337..5420850b39 100644
--- a/src/test/modules/injection_points/Makefile
+++ b/src/test/modules/injection_points/Makefile
@@ -9,7 +9,8 @@ EXTENSION = injection_points
DATA = injection_points--1.0.sql
PGFILEDESC = "injection_points - facility for injection points"
-REGRESS = injection_points hashagg reindex_conc vacuum
+# Disable regress tests
+#REGRESS = injection_points hashagg reindex_conc vacuum
REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress
ISOLATION = basic \
@@ -24,6 +25,9 @@ ISOLATION = basic \
# reindex-concurrently-upsert-on-constraint \
# reindex-concurrently-upsert-partitioned
+# Enable only one isolation test
+ISOLATION = onconflict
+
# The injection points are cluster-wide, so disable installcheck
NO_INSTALLCHECK = 1
diff --git a/src/test/modules/injection_points/expected/onconflict.out b/src/test/modules/injection_points/expected/onconflict.out
new file mode 100644
index 0000000000..ee89965592
--- /dev/null
+++ b/src/test/modules/injection_points/expected/onconflict.out
@@ -0,0 +1,23 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s1 s1ipa s1i s2 s2u s2ipw s2c s1c
+step s1: BEGIN;
+step s1ipa: SELECT injection_points_attach('before-on-conflict-update', 'wait');
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step s1i: INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x; <waiting ...>
+step s2: BEGIN;
+step s2u: UPDATE t_int SET i = i + 150 WHERE i = 1;
+step s2ipw: SELECT injection_points_wakeup('before-on-conflict-update');
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+step s2c: COMMIT;
+step s1i: <... completed>
+ERROR: tuple to be updated was already moved to another partition due to concurrent update
+step s1c: COMMIT;
diff --git a/src/test/modules/injection_points/specs/onconflict.spec b/src/test/modules/injection_points/specs/onconflict.spec
new file mode 100644
index 0000000000..495449264b
--- /dev/null
+++ b/src/test/modules/injection_points/specs/onconflict.spec
@@ -0,0 +1,30 @@
+setup
+{
+ CREATE EXTENSION injection_points;
+
+ CREATE TABLE t_int (i int PRIMARY KEY, v int, x int) PARTITION BY RANGE (i);
+ CREATE TABLE t_int_1 PARTITION OF t_int FOR VALUES FROM (1) TO (100);
+ CREATE TABLE t_int_2 PARTITION OF t_int FOR VALUES FROM (100) TO (200);
+
+ INSERT INTO t_int VALUES (1, 10, 100);
+}
+
+teardown
+{
+ DROP TABLE t_int;
+ DROP EXTENSION injection_points;
+}
+
+session s1
+step s1 { BEGIN; }
+step s1ipa { SELECT injection_points_attach('before-on-conflict-update', 'wait'); }
+step s1i { INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x; }
+step s1c { COMMIT; }
+
+session s2
+step s2 { BEGIN; }
+step s2ipw { SELECT injection_points_wakeup('before-on-conflict-update'); }
+step s2u { UPDATE t_int SET i = i + 150 WHERE i = 1; }
+step s2c { COMMIT; }
+
+permutation s1 s1ipa s1i(s2c) s2 s2u s2ipw s2c s1c
--
2.43.0
From 50fe893680665226982ffa67bbde6c2cf6017260 Mon Sep 17 00:00:00 2001
From: Koval Dmitry <[email protected]>
Date: Fri, 19 Dec 2025 01:51:07 +0300
Subject: [PATCH v1] Replace invalid Assert with ereport(ERROR, ...)
At the time Assert was added, the UPDATE of INSERT ON CONFLICT operation
was not supported for partitioned tables. However, support has been
added, so need to replace Assert with ereport(ERROR, ...).
---
src/backend/executor/nodeModifyTable.c | 11 ++++-------
1 file changed, 4 insertions(+), 7 deletions(-)
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index d131abcd67..f51ffbfc1e 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -2810,13 +2810,10 @@ ExecOnConflictUpdate(ModifyTableContext *context,
(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
errmsg("could not serialize access due to concurrent update")));
- /*
- * As long as we don't support an UPDATE of INSERT ON CONFLICT for
- * a partitioned table we shouldn't reach to a case where tuple to
- * be lock is moved to another partition due to concurrent update
- * of the partition key.
- */
- Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
+ if (ItemPointerIndicatesMovedPartitions(&tmfd.ctid))
+ ereport(ERROR,
+ errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+ errmsg("tuple to be updated was already moved to another partition due to concurrent update"));
/*
* Tell caller to try again from the very start.
--
2.43.0
Attachments:
[text/plain] v1-0001-Triggering-Assert-on-query-with-ON-CONFLICT.no_cfbot (4.2K, ../../[email protected]/2-v1-0001-Triggering-Assert-on-query-with-ON-CONFLICT.no_cfbot)
download | inline diff:
From 8780e66fc759872e06746c9c214dec52644b2e93 Mon Sep 17 00:00:00 2001
From: Koval Dmitry <[email protected]>
Date: Wed, 19 Feb 2025 13:09:33 +0300
Subject: [PATCH v1] Triggering Assert on query with ON CONFLICT
---
src/backend/executor/nodeModifyTable.c | 1 +
src/test/modules/injection_points/Makefile | 6 +++-
.../injection_points/expected/onconflict.out | 23 ++++++++++++++
.../injection_points/specs/onconflict.spec | 30 +++++++++++++++++++
4 files changed, 59 insertions(+), 1 deletion(-)
create mode 100644 src/test/modules/injection_points/expected/onconflict.out
create mode 100644 src/test/modules/injection_points/specs/onconflict.spec
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 874b71e660..d131abcd67 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -1148,6 +1148,7 @@ ExecInsert(ModifyTableContext *context,
*/
TupleTableSlot *returning = NULL;
+ INJECTION_POINT("before-on-conflict-update", NULL);
if (ExecOnConflictUpdate(context, resultRelInfo,
&conflictTid, slot, canSetTag,
&returning))
diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile
index bfdb3f5337..5420850b39 100644
--- a/src/test/modules/injection_points/Makefile
+++ b/src/test/modules/injection_points/Makefile
@@ -9,7 +9,8 @@ EXTENSION = injection_points
DATA = injection_points--1.0.sql
PGFILEDESC = "injection_points - facility for injection points"
-REGRESS = injection_points hashagg reindex_conc vacuum
+# Disable regress tests
+#REGRESS = injection_points hashagg reindex_conc vacuum
REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress
ISOLATION = basic \
@@ -24,6 +25,9 @@ ISOLATION = basic \
# reindex-concurrently-upsert-on-constraint \
# reindex-concurrently-upsert-partitioned
+# Enable only one isolation test
+ISOLATION = onconflict
+
# The injection points are cluster-wide, so disable installcheck
NO_INSTALLCHECK = 1
diff --git a/src/test/modules/injection_points/expected/onconflict.out b/src/test/modules/injection_points/expected/onconflict.out
new file mode 100644
index 0000000000..ee89965592
--- /dev/null
+++ b/src/test/modules/injection_points/expected/onconflict.out
@@ -0,0 +1,23 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s1 s1ipa s1i s2 s2u s2ipw s2c s1c
+step s1: BEGIN;
+step s1ipa: SELECT injection_points_attach('before-on-conflict-update', 'wait');
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step s1i: INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x; <waiting ...>
+step s2: BEGIN;
+step s2u: UPDATE t_int SET i = i + 150 WHERE i = 1;
+step s2ipw: SELECT injection_points_wakeup('before-on-conflict-update');
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+step s2c: COMMIT;
+step s1i: <... completed>
+ERROR: tuple to be updated was already moved to another partition due to concurrent update
+step s1c: COMMIT;
diff --git a/src/test/modules/injection_points/specs/onconflict.spec b/src/test/modules/injection_points/specs/onconflict.spec
new file mode 100644
index 0000000000..495449264b
--- /dev/null
+++ b/src/test/modules/injection_points/specs/onconflict.spec
@@ -0,0 +1,30 @@
+setup
+{
+ CREATE EXTENSION injection_points;
+
+ CREATE TABLE t_int (i int PRIMARY KEY, v int, x int) PARTITION BY RANGE (i);
+ CREATE TABLE t_int_1 PARTITION OF t_int FOR VALUES FROM (1) TO (100);
+ CREATE TABLE t_int_2 PARTITION OF t_int FOR VALUES FROM (100) TO (200);
+
+ INSERT INTO t_int VALUES (1, 10, 100);
+}
+
+teardown
+{
+ DROP TABLE t_int;
+ DROP EXTENSION injection_points;
+}
+
+session s1
+step s1 { BEGIN; }
+step s1ipa { SELECT injection_points_attach('before-on-conflict-update', 'wait'); }
+step s1i { INSERT INTO t_int VALUES (1, 11, 111) ON CONFLICT (i) DO UPDATE SET x = excluded.x; }
+step s1c { COMMIT; }
+
+session s2
+step s2 { BEGIN; }
+step s2ipw { SELECT injection_points_wakeup('before-on-conflict-update'); }
+step s2u { UPDATE t_int SET i = i + 150 WHERE i = 1; }
+step s2c { COMMIT; }
+
+permutation s1 s1ipa s1i(s2c) s2 s2u s2ipw s2c s1c
--
2.43.0
[text/plain] v1-0001-Replace-invalid-Assert-with-ereport-ERROR.patch (1.5K, ../../[email protected]/3-v1-0001-Replace-invalid-Assert-with-ereport-ERROR.patch)
download | inline diff:
From 50fe893680665226982ffa67bbde6c2cf6017260 Mon Sep 17 00:00:00 2001
From: Koval Dmitry <[email protected]>
Date: Fri, 19 Dec 2025 01:51:07 +0300
Subject: [PATCH v1] Replace invalid Assert with ereport(ERROR, ...)
At the time Assert was added, the UPDATE of INSERT ON CONFLICT operation
was not supported for partitioned tables. However, support has been
added, so need to replace Assert with ereport(ERROR, ...).
---
src/backend/executor/nodeModifyTable.c | 11 ++++-------
1 file changed, 4 insertions(+), 7 deletions(-)
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index d131abcd67..f51ffbfc1e 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -2810,13 +2810,10 @@ ExecOnConflictUpdate(ModifyTableContext *context,
(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
errmsg("could not serialize access due to concurrent update")));
- /*
- * As long as we don't support an UPDATE of INSERT ON CONFLICT for
- * a partitioned table we shouldn't reach to a case where tuple to
- * be lock is moved to another partition due to concurrent update
- * of the partition key.
- */
- Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
+ if (ItemPointerIndicatesMovedPartitions(&tmfd.ctid))
+ ereport(ERROR,
+ errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+ errmsg("tuple to be updated was already moved to another partition due to concurrent update"));
/*
* Tell caller to try again from the very start.
--
2.43.0
^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Assert when executing query on partitioned table
@ 2025-12-29 08:43 Dmitry Koval <[email protected]>
parent: Dmitry Koval <[email protected]>
0 siblings, 0 replies; 6+ messages in thread
From: Dmitry Koval @ 2025-12-29 08:43 UTC (permalink / raw)
To: pgsql-hackers
Hi!
I want to send the patch to Commitfest, so the version changed:
[1] -> [2] (for cfbot).
--
With best regards,
Dmitry Koval
Postgres Professional: http://postgrespro.com
From a450b0c085bde7020c9f8056bd45adb2340afc6a Mon Sep 17 00:00:00 2001
From: Koval Dmitry <[email protected]>
Date: Fri, 19 Dec 2025 01:51:07 +0300
Subject: [PATCH v2] Replace invalid Assert with ereport(ERROR, ...)
At the time Assert was added, the UPDATE of INSERT ON CONFLICT operation
was not supported for partitioned tables. However, support has been
added, so need to replace Assert with ereport(ERROR, ...).
---
src/backend/executor/nodeModifyTable.c | 11 ++++-------
1 file changed, 4 insertions(+), 7 deletions(-)
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index d131abcd67..f51ffbfc1e 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -2810,13 +2810,10 @@ ExecOnConflictUpdate(ModifyTableContext *context,
(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
errmsg("could not serialize access due to concurrent update")));
- /*
- * As long as we don't support an UPDATE of INSERT ON CONFLICT for
- * a partitioned table we shouldn't reach to a case where tuple to
- * be lock is moved to another partition due to concurrent update
- * of the partition key.
- */
- Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
+ if (ItemPointerIndicatesMovedPartitions(&tmfd.ctid))
+ ereport(ERROR,
+ errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+ errmsg("tuple to be updated was already moved to another partition due to concurrent update"));
/*
* Tell caller to try again from the very start.
--
2.43.0
Attachments:
[text/plain] v2-0001-Replace-invalid-Assert-with-ereport-ERROR.patch (1.5K, ../../[email protected]/2-v2-0001-Replace-invalid-Assert-with-ereport-ERROR.patch)
download | inline diff:
From a450b0c085bde7020c9f8056bd45adb2340afc6a Mon Sep 17 00:00:00 2001
From: Koval Dmitry <[email protected]>
Date: Fri, 19 Dec 2025 01:51:07 +0300
Subject: [PATCH v2] Replace invalid Assert with ereport(ERROR, ...)
At the time Assert was added, the UPDATE of INSERT ON CONFLICT operation
was not supported for partitioned tables. However, support has been
added, so need to replace Assert with ereport(ERROR, ...).
---
src/backend/executor/nodeModifyTable.c | 11 ++++-------
1 file changed, 4 insertions(+), 7 deletions(-)
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index d131abcd67..f51ffbfc1e 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -2810,13 +2810,10 @@ ExecOnConflictUpdate(ModifyTableContext *context,
(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
errmsg("could not serialize access due to concurrent update")));
- /*
- * As long as we don't support an UPDATE of INSERT ON CONFLICT for
- * a partitioned table we shouldn't reach to a case where tuple to
- * be lock is moved to another partition due to concurrent update
- * of the partition key.
- */
- Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
+ if (ItemPointerIndicatesMovedPartitions(&tmfd.ctid))
+ ereport(ERROR,
+ errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+ errmsg("tuple to be updated was already moved to another partition due to concurrent update"));
/*
* Tell caller to try again from the very start.
--
2.43.0
^ permalink raw reply [nested|flat] 6+ messages in thread
end of thread, other threads:[~2025-12-29 08:43 UTC | newest]
Thread overview: 6+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2020-12-22 07:06 [PATCH 06/20] pg_dump: zstd compression Justin Pryzby <[email protected]>
2025-02-20 11:13 Assert when executing query on partitioned table Dmitry Koval <[email protected]>
2025-03-08 21:28 ` Re: Assert when executing query on partitioned table Joseph Koshakow <[email protected]>
2025-03-15 14:57 ` Re: Assert when executing query on partitioned table Joseph Koshakow <[email protected]>
2025-12-18 23:27 ` Re: Assert when executing query on partitioned table Dmitry Koval <[email protected]>
2025-12-29 08:43 ` Re: Assert when executing query on partitioned table Dmitry Koval <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox