public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 05/10] (re)add wal_compression_method: lz4
3+ messages / 2 participants
[nested] [flat]
* [PATCH 05/10] (re)add wal_compression_method: lz4
@ 2021-03-12 20:32 Justin Pryzby <[email protected]>
0 siblings, 0 replies; 3+ messages in thread
From: Justin Pryzby @ 2021-03-12 20:32 UTC (permalink / raw)
---
configure | 176 ++++++++++++++++++
configure.ac | 26 +++
doc/src/sgml/config.sgml | 3 +-
src/backend/access/transam/xlog.c | 3 +
src/backend/access/transam/xloginsert.c | 17 +-
src/backend/access/transam/xlogreader.c | 12 ++
src/backend/utils/misc/postgresql.conf.sample | 2 +-
src/include/access/xlog_internal.h | 1 +
src/include/pg_config.h.in | 3 +
src/tools/msvc/Solution.pm | 1 +
10 files changed, 241 insertions(+), 3 deletions(-)
diff --git a/configure b/configure
index 3fd4cecbeb..fed440adcf 100755
--- a/configure
+++ b/configure
@@ -699,6 +699,9 @@ with_gnu_ld
LD
LDFLAGS_SL
LDFLAGS_EX
+LZ4_LIBS
+LZ4_CFLAGS
+with_lz4
with_zlib
with_system_tzdata
with_libxslt
@@ -864,6 +867,7 @@ with_libxml
with_libxslt
with_system_tzdata
with_zlib
+with_lz4
with_gnu_ld
with_ssl
with_openssl
@@ -891,6 +895,8 @@ ICU_LIBS
XML2_CONFIG
XML2_CFLAGS
XML2_LIBS
+LZ4_CFLAGS
+LZ4_LIBS
LDFLAGS_EX
LDFLAGS_SL
PERL
@@ -1569,6 +1575,7 @@ Optional Packages:
--with-system-tzdata=DIR
use system time zone data in DIR
--without-zlib do not use Zlib
+ --with-lz4 build with LZ4 support
--with-gnu-ld assume the C compiler uses GNU ld [default=no]
--with-ssl=LIB use LIB for SSL/TLS support (openssl)
--with-openssl obsolete spelling of --with-ssl=openssl
@@ -1596,6 +1603,8 @@ Some influential environment variables:
XML2_CONFIG path to xml2-config utility
XML2_CFLAGS C compiler flags for XML2, overriding pkg-config
XML2_LIBS linker flags for XML2, overriding pkg-config
+ LZ4_CFLAGS C compiler flags for LZ4, overriding pkg-config
+ LZ4_LIBS linker flags for LZ4, overriding pkg-config
LDFLAGS_EX extra linker flags for linking executables only
LDFLAGS_SL extra linker flags for linking shared libraries only
PERL Perl program
@@ -8563,6 +8572,137 @@ fi
+#
+# LZ4
+#
+{ $as_echo "$as_me:${as_lineno-$LINENO}: checking whether to build with LZ4 support" >&5
+$as_echo_n "checking whether to build with LZ4 support... " >&6; }
+
+
+
+# Check whether --with-lz4 was given.
+if test "${with_lz4+set}" = set; then :
+ withval=$with_lz4;
+ case $withval in
+ yes)
+
+$as_echo "#define USE_LZ4 1" >>confdefs.h
+
+ ;;
+ no)
+ :
+ ;;
+ *)
+ as_fn_error $? "no argument expected for --with-lz4 option" "$LINENO" 5
+ ;;
+ esac
+
+else
+ with_lz4=no
+
+fi
+
+
+{ $as_echo "$as_me:${as_lineno-$LINENO}: result: $with_lz4" >&5
+$as_echo "$with_lz4" >&6; }
+
+
+if test "$with_lz4" = yes; then
+
+pkg_failed=no
+{ $as_echo "$as_me:${as_lineno-$LINENO}: checking for liblz4" >&5
+$as_echo_n "checking for liblz4... " >&6; }
+
+if test -n "$LZ4_CFLAGS"; then
+ pkg_cv_LZ4_CFLAGS="$LZ4_CFLAGS"
+ elif test -n "$PKG_CONFIG"; then
+ if test -n "$PKG_CONFIG" && \
+ { { $as_echo "$as_me:${as_lineno-$LINENO}: \$PKG_CONFIG --exists --print-errors \"liblz4\""; } >&5
+ ($PKG_CONFIG --exists --print-errors "liblz4") 2>&5
+ ac_status=$?
+ $as_echo "$as_me:${as_lineno-$LINENO}: \$? = $ac_status" >&5
+ test $ac_status = 0; }; then
+ pkg_cv_LZ4_CFLAGS=`$PKG_CONFIG --cflags "liblz4" 2>/dev/null`
+ test "x$?" != "x0" && pkg_failed=yes
+else
+ pkg_failed=yes
+fi
+ else
+ pkg_failed=untried
+fi
+if test -n "$LZ4_LIBS"; then
+ pkg_cv_LZ4_LIBS="$LZ4_LIBS"
+ elif test -n "$PKG_CONFIG"; then
+ if test -n "$PKG_CONFIG" && \
+ { { $as_echo "$as_me:${as_lineno-$LINENO}: \$PKG_CONFIG --exists --print-errors \"liblz4\""; } >&5
+ ($PKG_CONFIG --exists --print-errors "liblz4") 2>&5
+ ac_status=$?
+ $as_echo "$as_me:${as_lineno-$LINENO}: \$? = $ac_status" >&5
+ test $ac_status = 0; }; then
+ pkg_cv_LZ4_LIBS=`$PKG_CONFIG --libs "liblz4" 2>/dev/null`
+ test "x$?" != "x0" && pkg_failed=yes
+else
+ pkg_failed=yes
+fi
+ else
+ pkg_failed=untried
+fi
+
+
+
+if test $pkg_failed = yes; then
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: no" >&5
+$as_echo "no" >&6; }
+
+if $PKG_CONFIG --atleast-pkgconfig-version 0.20; then
+ _pkg_short_errors_supported=yes
+else
+ _pkg_short_errors_supported=no
+fi
+ if test $_pkg_short_errors_supported = yes; then
+ LZ4_PKG_ERRORS=`$PKG_CONFIG --short-errors --print-errors --cflags --libs "liblz4" 2>&1`
+ else
+ LZ4_PKG_ERRORS=`$PKG_CONFIG --print-errors --cflags --libs "liblz4" 2>&1`
+ fi
+ # Put the nasty error message in config.log where it belongs
+ echo "$LZ4_PKG_ERRORS" >&5
+
+ as_fn_error $? "Package requirements (liblz4) were not met:
+
+$LZ4_PKG_ERRORS
+
+Consider adjusting the PKG_CONFIG_PATH environment variable if you
+installed software in a non-standard prefix.
+
+Alternatively, you may set the environment variables LZ4_CFLAGS
+and LZ4_LIBS to avoid the need to call pkg-config.
+See the pkg-config man page for more details." "$LINENO" 5
+elif test $pkg_failed = untried; then
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: no" >&5
+$as_echo "no" >&6; }
+ { { $as_echo "$as_me:${as_lineno-$LINENO}: error: in \`$ac_pwd':" >&5
+$as_echo "$as_me: error: in \`$ac_pwd':" >&2;}
+as_fn_error $? "The pkg-config script could not be found or is too old. Make sure it
+is in your PATH or set the PKG_CONFIG environment variable to the full
+path to pkg-config.
+
+Alternatively, you may set the environment variables LZ4_CFLAGS
+and LZ4_LIBS to avoid the need to call pkg-config.
+See the pkg-config man page for more details.
+
+To get pkg-config, see <http://pkg-config.freedesktop.org/;.
+See \`config.log' for more details" "$LINENO" 5; }
+else
+ LZ4_CFLAGS=$pkg_cv_LZ4_CFLAGS
+ LZ4_LIBS=$pkg_cv_LZ4_LIBS
+ { $as_echo "$as_me:${as_lineno-$LINENO}: result: yes" >&5
+$as_echo "yes" >&6; }
+
+fi
+ LIBS="$LZ4_LIBS $LIBS"
+ CFLAGS="$LZ4_CFLAGS $CFLAGS"
+fi
+
#
# Assignments
#
@@ -13381,6 +13521,42 @@ fi
fi
+if test "$with_lz4" = yes; then
+ ac_save_CPPFLAGS=$CPPFLAGS
+ CPPFLAGS="$LZ4_CFLAGS $CPPFLAGS"
+
+ # Verify we have LZ4's header files
+ for ac_header in lz4/lz4.h
+do :
+ ac_fn_c_check_header_mongrel "$LINENO" "lz4/lz4.h" "ac_cv_header_lz4_lz4_h" "$ac_includes_default"
+if test "x$ac_cv_header_lz4_lz4_h" = xyes; then :
+ cat >>confdefs.h <<_ACEOF
+#define HAVE_LZ4_LZ4_H 1
+_ACEOF
+
+else
+ for ac_header in lz4.h
+do :
+ ac_fn_c_check_header_mongrel "$LINENO" "lz4.h" "ac_cv_header_lz4_h" "$ac_includes_default"
+if test "x$ac_cv_header_lz4_h" = xyes; then :
+ cat >>confdefs.h <<_ACEOF
+#define HAVE_LZ4_H 1
+_ACEOF
+
+else
+ as_fn_error $? "lz4.h header file is required for LZ4" "$LINENO" 5
+fi
+
+done
+
+fi
+
+done
+
+
+ CPPFLAGS=$ac_save_CPPFLAGS
+fi
+
if test "$with_gssapi" = yes ; then
for ac_header in gssapi/gssapi.h
do :
diff --git a/configure.ac b/configure.ac
index 2f1585adc0..8c454128bb 100644
--- a/configure.ac
+++ b/configure.ac
@@ -986,6 +986,21 @@ PGAC_ARG_BOOL(with, zlib, yes,
[do not use Zlib])
AC_SUBST(with_zlib)
+#
+# LZ4
+#
+AC_MSG_CHECKING([whether to build with LZ4 support])
+PGAC_ARG_BOOL(with, lz4, no, [build with LZ4 support],
+ [AC_DEFINE([USE_LZ4], 1, [Define to 1 to build with LZ4 support. (--with-lz4)])])
+AC_MSG_RESULT([$with_lz4])
+AC_SUBST(with_lz4)
+
+if test "$with_lz4" = yes; then
+ PKG_CHECK_MODULES(LZ4, liblz4)
+ LIBS="$LZ4_LIBS $LIBS"
+ CFLAGS="$LZ4_CFLAGS $CFLAGS"
+fi
+
#
# Assignments
#
@@ -1410,6 +1425,17 @@ failure. It is possible the compiler isn't looking in the proper directory.
Use --without-zlib to disable zlib support.])])
fi
+if test "$with_lz4" = yes; then
+ ac_save_CPPFLAGS=$CPPFLAGS
+ CPPFLAGS="$LZ4_CFLAGS $CPPFLAGS"
+
+ # Verify we have LZ4's header files
+ AC_CHECK_HEADERS(lz4/lz4.h, [],
+ [AC_CHECK_HEADERS(lz4.h, [], [AC_MSG_ERROR([lz4.h header file is required for LZ4])])])
+
+ CPPFLAGS=$ac_save_CPPFLAGS
+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/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 7fb2a84626..257775c83b 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -3082,7 +3082,8 @@ include_dir 'conf.d'
<para>
This parameter selects the compression method used to compress WAL when
<varname>wal_compression</varname> is enabled.
- The supported methods are pglz and zlib.
+ The supported methods are pglz, zlib, and (if configured when
+ <productname>PostgreSQL</productname> was built) lz4.
The default value is <literal>pglz</literal>.
Only superusers can change this setting.
</para>
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f5d4450654..984ce39cc7 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -186,6 +186,9 @@ const struct config_enum_entry wal_compression_options[] = {
{"pglz", WAL_COMPRESSION_PGLZ, false},
#ifdef HAVE_LIBZ
{"zlib", WAL_COMPRESSION_ZLIB, false},
+#endif
+#ifdef USE_LZ4
+ {"lz4", WAL_COMPRESSION_LZ4, false},
#endif
{NULL, 0, false}
};
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index a93b33464f..3c15286bd8 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -41,10 +41,17 @@
#define ZLIB_MAX_BLCKSZ 0
#endif
+#ifdef USE_LZ4
+#include "lz4.h"
+#define LZ4_MAX_BLCKSZ LZ4_COMPRESSBOUND(BLCKSZ)
+#else
+#define LZ4_MAX_BLCKSZ 0
+#endif
+
/* Buffer size required to store a compressed version of backup block image */
#define PGLZ_MAX_BLCKSZ PGLZ_MAX_OUTPUT(BLCKSZ)
-#define COMPRESS_BUFSIZE Max(PGLZ_MAX_BLCKSZ, ZLIB_MAX_BLCKSZ)
+#define COMPRESS_BUFSIZE Max(Max(PGLZ_MAX_BLCKSZ, ZLIB_MAX_BLCKSZ), LZ4_MAX_BLCKSZ)
/*
* For each block reference registered with XLogRegisterBuffer, we fill in
@@ -891,6 +898,14 @@ XLogCompressBackupBlock(char *page, uint16 hole_offset, uint16 hole_length,
}
#endif
+#ifdef USE_LZ4
+ case WAL_COMPRESSION_LZ4:
+ len = LZ4_compress_fast(source, dest, orig_len, COMPRESS_BUFSIZE, 1);
+ if (len == 0)
+ len = -1;
+ break;
+#endif
+
default:
/*
* It should be impossible to get here for unsupported algorithms,
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 0d8830fc50..97165f1bb1 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -37,6 +37,10 @@
#include <zlib.h>
#endif
+#ifdef USE_LZ4
+#include "lz4.h"
+#endif
+
static void report_invalid_record(XLogReaderState *state, const char *fmt,...)
pg_attribute_printf(2, 3);
static bool allocate_recordbuf(XLogReaderState *state, uint32 reclength);
@@ -1544,6 +1548,7 @@ XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
struct walcompression walmethods[] = {
{"pglz", WAL_COMPRESSION_PGLZ},
{"zlib", WAL_COMPRESSION_ZLIB},
+ {"lz4", WAL_COMPRESSION_LZ4},
};
/*
@@ -1609,6 +1614,13 @@ RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page)
}
#endif
+#ifdef USE_LZ4
+ case WAL_COMPRESSION_LZ4:
+ decomp_result = LZ4_decompress_safe(ptr, tmp.data,
+ bkpb->bimg_len, BLCKSZ-bkpb->hole_length);
+ break;
+#endif
+
default:
report_invalid_record(record, "image at %X/%X is compressed with unsupported codec, block %d (%d/%s)",
(uint32) (record->ReadRecPtr >> 32),
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index ef69a94492..76f494cb9b 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -213,7 +213,7 @@
# open_sync
#full_page_writes = on # recover from partial page writes
#wal_compression = off # enable compression of full-page writes
-#wal_compression_method = pglz # pglz, zlib
+#wal_compression_method = pglz # pglz, zlib, lz4
#wal_log_hints = off # also do full page writes of non-critical updates
# (change requires restart)
#wal_init_zero = on # zero-fill new WAL files
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index b000a21557..b908fa48de 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -336,6 +336,7 @@ typedef enum WalCompression
{
WAL_COMPRESSION_PGLZ,
WAL_COMPRESSION_ZLIB,
+ WAL_COMPRESSION_LZ4,
} WalCompression;
extern const char *wal_compression_name(WalCompression compression);
diff --git a/src/include/pg_config.h.in b/src/include/pg_config.h.in
index 7a7cc21d8d..0a6422da4f 100644
--- a/src/include/pg_config.h.in
+++ b/src/include/pg_config.h.in
@@ -899,6 +899,9 @@
/* Define to 1 to build with LLVM based JIT support. (--with-llvm) */
#undef USE_LLVM
+/* Define to 1 to build with LZ4 support (--with-lz4) */
+#undef USE_LZ4
+
/* Define to select named POSIX semaphores. */
#undef USE_NAMED_POSIX_SEMAPHORES
diff --git a/src/tools/msvc/Solution.pm b/src/tools/msvc/Solution.pm
index a4f5cc4bdb..14605371bb 100644
--- a/src/tools/msvc/Solution.pm
+++ b/src/tools/msvc/Solution.pm
@@ -485,6 +485,7 @@ sub GenerateFiles
USE_ICU => $self->{options}->{icu} ? 1 : undef,
USE_LIBXML => undef,
USE_LIBXSLT => undef,
+ USE_LZ4 => undef,
USE_LDAP => $self->{options}->{ldap} ? 1 : undef,
USE_LLVM => undef,
USE_NAMED_POSIX_SEMAPHORES => undef,
--
2.17.0
--jozmn01XJZjDjM3N
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment; filename="0006-Default-to-LZ4.patch"
^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Refactor replication origin state reset helpers
@ 2026-01-07 08:49 Chao Li <[email protected]>
2026-01-08 01:15 ` Re: Refactor replication origin state reset helpers Chao Li <[email protected]>
0 siblings, 1 reply; 3+ messages in thread
From: Chao Li @ 2026-01-07 08:49 UTC (permalink / raw)
To: Masahiko Sawada <[email protected]>; +Cc: Ashutosh Bapat <[email protected]>; Álvaro Herrera <[email protected]>; Postgres hackers <[email protected]>
> On Jan 7, 2026, at 15:21, Chao Li <[email protected]> wrote:
>
>
> On Wed, Jan 7, 2026 at 8:30 AM Masahiko Sawada <[email protected]> wrote:
> On Mon, Dec 29, 2025 at 11:17 PM Chao Li <[email protected]> wrote:
> >
> > On Tue, Dec 30, 2025 at 1:07 PM Chao Li <[email protected]> wrote:
> >>
> >>
> >> On Tue, Dec 30, 2025 at 12:48 PM Ashutosh Bapat <[email protected]> wrote:
> >>>
> >>> On Mon, Dec 29, 2025 at 8:14 PM Álvaro Herrera <[email protected]> wrote:
> >>> >
> >>> > On 2025-Dec-24, Ashutosh Bapat wrote:
> >>> >
> >>> > > If we go this route, we at least need to declare the new functions as
> >>> > > static inline and move them to a header file instead of .c file.
> >>> >
> >>> > Hmm, why would we make them static inline instead of standard (extern)
> >>> > functions? We use static inline functions when we want to avoid the
> >>> > overhead of a function call in a hot code path, but I doubt that's the
> >>> > case here. Am I mistaken on this?
> >>> >
> >>>
> >>> I wasn't aware that we are using static inline only in hot code paths.
> >>> Looking around I see most of the static inline functions are from
> >>> modules which are used in hot code paths. So, yeah that seems to be
> >>> the convention. I also see some exceptions like those in
> >>> basebackup_sink.h - I don't think all of those are used in hot code
> >>> paths.
> >>>
> >>> In this case, we are moving three assignments into their own
> >>> functions. CPU instructions to call extern functions will be
> >>> significant compared to CPU instructions for those assignments. static
> >>> inline functions, OTOH, would have similar performance as the existing
> >>> code while providing modularization. If you feel that's not a good
> >>> enough reason, I am ok keeping them extern.
> >>>
> >>> > > Further, does it make sense to put together all the state variables
> >>> > > into a single structure?
> >>> >
> >>> > Yeah -- keeping the threaded-backend project in mind, moving them to a
> >>> > single struct seems to make sense. I think it's a separate patch though
> >>> > because it'd be more invasive than Chao's initial patch, as those
> >>> > variables are used in many places.
> >>> >
> >>
> >>
> >> Attached v3 patch set. Comparing to v2, the changes are:
> >>
> >> 0001:
> >> * Combine the two cleanup functions into one and control them by a bool flag.
> >> * Change the helper function to be extern.
> >> * Move out cleanup from reset function.
> >>
> >> 0002: Consolidate replication origin session globals into a single state struct.
> >
> >
> > Fixed a bug in v4.
> >
>
> I've reviewed both patches. Here are some comments:
>
> Thank you very much for the review.
>
> 0001 patch:
>
> +/*
> + * Clear session replication origin state.
> + *
> + * If xact_only is true, only clear the per-transaction state.
> + */
> +void
> +replorigin_session_clear_state(bool xact_only)
> +{
> + replorigin_session_origin_lsn = InvalidXLogRecPtr;
> + replorigin_session_origin_timestamp = 0;
> + if (!xact_only)
> + replorigin_session_origin = InvalidRepOriginId;
> +}
>
> Given that we already have session_replication_state, I am concerned
> that the name replorigin_session_clear_state creates ambiguity. Could
> we rename it to something like replorigin_session_clear()?
>
> Agreed.
> Additionally, I feel that the term "per-transaction state" in the
> comments does not accurately describe these two fields. How about
> renaming the xact_only parameter to clear_origin? This would make it
> explicit that setting the flag to true clears
> replorigin_session_origin as well.
>
> Fixed.
>
> 0002 patch:
>
> +typedef struct RepOriginSessionState
> +{
> + RepOriginId origin;
> + XLogRecPtr origin_lsn;
> + TimestampTz origin_timestamp;
> +} RepOriginSessionState;
> +
> +extern PGDLLIMPORT RepOriginSessionState replorigin_session_state;
>
> replorigin_session_state is quite confusable with the existing
> session_replication_state. Given that these values are used to add the
> additional information to the transaction, how about the name
> something like "replorigin_xact_state" or "replorigin_xact_origin"?
>
>
> Okay, I take replorigin_xact_state, and I renamed the structure name as well.
> Given replorigin_session_clear() only sets members of RepOriginXactState, I also renamed it to replorigin_xact_clear().
>
> --
> +RepOriginSessionState replorigin_session_state = {
> + InvalidRepOriginId, InvalidXLogRecPtr, 0
> +};
>
> I think using designated initializers here would be better for
> readability and robustness against future struct changes.
>
> Fixed.
>
>
> Best regards,
> --
> Chao Li (Evan)
> HighGo Software Co., Ltd.
> https://www.highgo.com/ <v5-0001-Refactor-replication-origin-state-reset-helpers.patch><v5-0002-Consolidate-replication-origin-session-globals-in.patch>
The CF CI failed a test case, but I don’t think that’s mine. I just did a self-review for v5 and couldn’t find any logic change. So I would guess the CI failure was intermittent. I don’t know how to rerun the CI test other than bumping the patch version and re-posting.
Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/
^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Refactor replication origin state reset helpers
2026-01-07 08:49 Re: Refactor replication origin state reset helpers Chao Li <[email protected]>
@ 2026-01-08 01:15 ` Chao Li <[email protected]>
0 siblings, 0 replies; 3+ messages in thread
From: Chao Li @ 2026-01-08 01:15 UTC (permalink / raw)
To: Masahiko Sawada <[email protected]>; +Cc: Ashutosh Bapat <[email protected]>; Álvaro Herrera <[email protected]>; Postgres hackers <[email protected]>
On Wed, Jan 7, 2026 at 4:49 PM Chao Li <[email protected]> wrote:
>
>
> <v5-0001-Refactor-replication-origin-state-reset-helpers.patch><v5-0002-Consolidate-replication-origin-session-globals-in.patch>
>
> The CF CI failed a test case, but I don’t think that’s mine. I just did a
> self-review for v5 and couldn’t find any logic change. So I would guess the
> CI failure was intermittent. I don’t know how to rerun the CI test other
> than bumping the patch version and re-posting.
V6 is only a rebase, nothing new.
Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/
Attachments:
[application/octet-stream] v6-0001-Refactor-replication-origin-state-reset-helpers.patch (5.4K, ../../CAEoWx2moawAzQLaHu=C9PEkro7tmXXq792ODM0=bM9rtoE_E2g@mail.gmail.com/3-v6-0001-Refactor-replication-origin-state-reset-helpers.patch)
download | inline diff:
From 630cb849639e312527e255dc30cfd58ef21639d9 Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Wed, 24 Dec 2025 09:17:27 +0800
Subject: [PATCH v6 1/2] Refactor replication origin state reset helpers
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Factor out common logic for clearing per-transaction and per-session
replication origin state into a dedicated helper function.
This removes duplicated assignments of replorigin_session_origin,
replorigin_session_origin_lsn, and replorigin_session_origin_timestamp
across multiple call sites, and makes the intended scope of each reset
(clear per-transaction state vs. clear per-session state) explicit.
No functional change intended.
Author: Chao Li <[email protected]>
Reviewed-by: Masahiko Sawada <[email protected]>
Reviewed-by: Ashutosh Bapat <[email protected]>
Reviewed-by: Álvaro Herrera <[email protected]>
Discussion: https://postgr.es/m/CAEoWx2=pYvfRthXHTzSrOsf5_FfyY4zJyK4zV2v4W=yjUij1cA@mail.gmail.com
---
src/backend/replication/logical/origin.c | 20 +++++++++++++++-----
src/backend/replication/logical/tablesync.c | 4 +---
src/backend/replication/logical/worker.c | 14 ++++++--------
src/include/replication/origin.h | 1 +
4 files changed, 23 insertions(+), 16 deletions(-)
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index 7268d7b5e6c..c27e790c3a8 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -1287,6 +1287,19 @@ replorigin_session_get_progress(bool flush)
return remote_lsn;
}
+/*
+ * Clear session replication origin state.
+ *
+ * Only when clear_origin is true, the origin ID is also cleared.
+ */
+void
+replorigin_session_clear(bool clear_origin)
+{
+ replorigin_session_origin_lsn = InvalidXLogRecPtr;
+ replorigin_session_origin_timestamp = 0;
+ if (clear_origin)
+ replorigin_session_origin = InvalidRepOriginId;
+}
/* ---------------------------------------------------------------------------
@@ -1412,9 +1425,7 @@ pg_replication_origin_session_reset(PG_FUNCTION_ARGS)
replorigin_session_reset();
- replorigin_session_origin = InvalidRepOriginId;
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ replorigin_session_clear(true);
PG_RETURN_VOID();
}
@@ -1482,8 +1493,7 @@ pg_replication_origin_xact_reset(PG_FUNCTION_ARGS)
{
replorigin_check_prerequisites(true, false);
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ replorigin_session_clear(false);
PG_RETURN_VOID();
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 67e57520386..72e45c2de19 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -323,9 +323,7 @@ ProcessSyncingTablesForSync(XLogRecPtr current_lsn)
* This is needed to allow the origin to be dropped.
*/
replorigin_session_reset();
- replorigin_session_origin = InvalidRepOriginId;
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ replorigin_session_clear(true);
/*
* Drop the tablesync's origin tracking if exists.
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index ad281e7069b..6ad99cc5afd 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -627,7 +627,7 @@ static inline void reset_apply_error_context_info(void);
static TransApplyAction get_transaction_apply_action(TransactionId xid,
ParallelApplyWorkerInfo **winfo);
-static void replorigin_reset(int code, Datum arg);
+static void on_exit_clear_state(int code, Datum arg);
/*
* Form the origin name for the subscription.
@@ -5594,7 +5594,7 @@ start_apply(XLogRecPtr origin_startpos)
* transaction loss as that transaction won't be sent again by the
* server.
*/
- replorigin_reset(0, (Datum) 0);
+ replorigin_session_clear(true);
if (MySubscription->disableonerr)
DisableSubscriptionAndExit();
@@ -5865,18 +5865,16 @@ InitializeLogRepWorker(void)
* replication workers that set up origins and apply remote transactions
* are protected.
*/
- before_shmem_exit(replorigin_reset, (Datum) 0);
+ before_shmem_exit(on_exit_clear_state, (Datum) 0);
}
/*
- * Reset the origin state.
+ * Callback on exit to reset the origin state.
*/
static void
-replorigin_reset(int code, Datum arg)
+on_exit_clear_state(int code, Datum arg)
{
- replorigin_session_origin = InvalidRepOriginId;
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ replorigin_session_clear(true);
}
/*
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 1da77363955..309b800bd5f 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -65,6 +65,7 @@ extern void replorigin_session_advance(XLogRecPtr remote_commit,
XLogRecPtr local_commit);
extern void replorigin_session_setup(RepOriginId node, int acquired_by);
extern void replorigin_session_reset(void);
+extern void replorigin_session_clear(bool clear_origin);
extern XLogRecPtr replorigin_session_get_progress(bool flush);
/* Checkpoint/Startup integration */
--
2.39.5 (Apple Git-154)
[application/octet-stream] v6-0002-Consolidate-replication-origin-session-globals-in.patch (22.4K, ../../CAEoWx2moawAzQLaHu=C9PEkro7tmXXq792ODM0=bM9rtoE_E2g@mail.gmail.com/4-v6-0002-Consolidate-replication-origin-session-globals-in.patch)
download | inline diff:
From 1d8fdb8d65559284e72b78faadec510713bf4266 Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Tue, 30 Dec 2025 12:55:31 +0800
Subject: [PATCH v6 2/2] Consolidate replication origin session globals into a
single state struct
Replace the separate global variables used to track replication origin
session state (origin ID, origin LSN, and origin timestamp) with a single
RepOriginXactState struct.
This refactoring groups logically related session state into one object,
reducing the proliferation of loosely related globals and making the code
easier to read and reason about. All existing users are updated to refer
to replorigin_xact_state.{origin, origin_lsn, origin_timestamp}.
There is no intended behavior change; this is a mechanical cleanup to
improve clarity and maintainability of replication origin handling.
Author: Chao Li <[email protected]>
Reviewed-by: Masahiko Sawada <[email protected]>
Discussion: https://postgr.es/m/CAEoWx2=pYvfRthXHTzSrOsf5_FfyY4zJyK4zV2v4W=yjUij1cA@mail.gmail.com
---
src/backend/access/transam/twophase.c | 32 ++++++++---------
src/backend/access/transam/xact.c | 36 +++++++++----------
src/backend/access/transam/xloginsert.c | 6 ++--
.../replication/logical/applyparallelworker.c | 6 ++--
src/backend/replication/logical/origin.c | 32 +++++++++--------
src/backend/replication/logical/tablesync.c | 6 ++--
src/backend/replication/logical/worker.c | 32 ++++++++---------
src/include/replication/origin.h | 14 +++++---
src/tools/pgindent/typedefs.list | 1 +
9 files changed, 87 insertions(+), 78 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index e50abb331cc..329c6bbf3c7 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -1157,13 +1157,13 @@ EndPrepare(GlobalTransaction gxact)
Assert(hdr->magic == TWOPHASE_MAGIC);
hdr->total_len = records.total_len + sizeof(pg_crc32c);
- replorigin = (replorigin_session_origin != InvalidRepOriginId &&
- replorigin_session_origin != DoNotReplicateId);
+ replorigin = (replorigin_xact_state.origin != InvalidRepOriginId &&
+ replorigin_xact_state.origin != DoNotReplicateId);
if (replorigin)
{
- hdr->origin_lsn = replorigin_session_origin_lsn;
- hdr->origin_timestamp = replorigin_session_origin_timestamp;
+ hdr->origin_lsn = replorigin_xact_state.origin_lsn;
+ hdr->origin_timestamp = replorigin_xact_state.origin_timestamp;
}
/*
@@ -1211,7 +1211,7 @@ EndPrepare(GlobalTransaction gxact)
if (replorigin)
{
/* Move LSNs forward for this replication origin */
- replorigin_session_advance(replorigin_session_origin_lsn,
+ replorigin_session_advance(replorigin_xact_state.origin_lsn,
gxact->prepare_end_lsn);
}
@@ -2330,8 +2330,8 @@ RecordTransactionCommitPrepared(TransactionId xid,
* Are we using the replication origins feature? Or, in other words, are
* we replaying remote actions?
*/
- replorigin = (replorigin_session_origin != InvalidRepOriginId &&
- replorigin_session_origin != DoNotReplicateId);
+ replorigin = (replorigin_xact_state.origin != InvalidRepOriginId &&
+ replorigin_xact_state.origin != DoNotReplicateId);
/* Load the injection point before entering the critical section */
INJECTION_POINT_LOAD("commit-after-delay-checkpoint");
@@ -2376,23 +2376,23 @@ RecordTransactionCommitPrepared(TransactionId xid,
if (replorigin)
/* Move LSNs forward for this replication origin */
- replorigin_session_advance(replorigin_session_origin_lsn,
+ replorigin_session_advance(replorigin_xact_state.origin_lsn,
XactLastRecEnd);
/*
* Record commit timestamp. The value comes from plain commit timestamp
* if replorigin is not enabled, or replorigin already set a value for us
- * in replorigin_session_origin_timestamp otherwise.
+ * in replorigin_xact_state.origin_timestamp otherwise.
*
* We don't need to WAL-log anything here, as the commit record written
* above already contains the data.
*/
- if (!replorigin || replorigin_session_origin_timestamp == 0)
- replorigin_session_origin_timestamp = committs;
+ if (!replorigin || replorigin_xact_state.origin_timestamp == 0)
+ replorigin_xact_state.origin_timestamp = committs;
TransactionTreeSetCommitTsData(xid, nchildren, children,
- replorigin_session_origin_timestamp,
- replorigin_session_origin);
+ replorigin_xact_state.origin_timestamp,
+ replorigin_xact_state.origin);
/*
* We don't currently try to sleep before flush here ... nor is there any
@@ -2445,8 +2445,8 @@ RecordTransactionAbortPrepared(TransactionId xid,
* Are we using the replication origins feature? Or, in other words, are
* we replaying remote actions?
*/
- replorigin = (replorigin_session_origin != InvalidRepOriginId &&
- replorigin_session_origin != DoNotReplicateId);
+ replorigin = (replorigin_xact_state.origin != InvalidRepOriginId &&
+ replorigin_xact_state.origin != DoNotReplicateId);
/*
* Catch the scenario where we aborted partway through
@@ -2472,7 +2472,7 @@ RecordTransactionAbortPrepared(TransactionId xid,
if (replorigin)
/* Move LSNs forward for this replication origin */
- replorigin_session_advance(replorigin_session_origin_lsn,
+ replorigin_session_advance(replorigin_xact_state.origin_lsn,
XactLastRecEnd);
/* Always flush, since we're about to remove the 2PC state file */
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index c857e23552f..c779ce71f07 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -1413,8 +1413,8 @@ RecordTransactionCommit(void)
* Are we using the replication origins feature? Or, in other words,
* are we replaying remote actions?
*/
- replorigin = (replorigin_session_origin != InvalidRepOriginId &&
- replorigin_session_origin != DoNotReplicateId);
+ replorigin = (replorigin_xact_state.origin != InvalidRepOriginId &&
+ replorigin_xact_state.origin != DoNotReplicateId);
/*
* Mark ourselves as within our "commit critical section". This
@@ -1462,25 +1462,25 @@ RecordTransactionCommit(void)
if (replorigin)
/* Move LSNs forward for this replication origin */
- replorigin_session_advance(replorigin_session_origin_lsn,
+ replorigin_session_advance(replorigin_xact_state.origin_lsn,
XactLastRecEnd);
/*
* Record commit timestamp. The value comes from plain commit
* timestamp if there's no replication origin; otherwise, the
- * timestamp was already set in replorigin_session_origin_timestamp by
- * replication.
+ * timestamp was already set in replorigin_xact_state.origin_timestamp
+ * by replication.
*
* We don't need to WAL-log anything here, as the commit record
* written above already contains the data.
*/
- if (!replorigin || replorigin_session_origin_timestamp == 0)
- replorigin_session_origin_timestamp = GetCurrentTransactionStopTimestamp();
+ if (!replorigin || replorigin_xact_state.origin_timestamp == 0)
+ replorigin_xact_state.origin_timestamp = GetCurrentTransactionStopTimestamp();
TransactionTreeSetCommitTsData(xid, nchildren, children,
- replorigin_session_origin_timestamp,
- replorigin_session_origin);
+ replorigin_xact_state.origin_timestamp,
+ replorigin_xact_state.origin);
}
/*
@@ -1810,8 +1810,8 @@ RecordTransactionAbort(bool isSubXact)
* Are we using the replication origins feature? Or, in other words, are
* we replaying remote actions?
*/
- replorigin = (replorigin_session_origin != InvalidRepOriginId &&
- replorigin_session_origin != DoNotReplicateId);
+ replorigin = (replorigin_xact_state.origin != InvalidRepOriginId &&
+ replorigin_xact_state.origin != DoNotReplicateId);
/* Fetch the data we need for the abort record */
nrels = smgrGetPendingDeletes(false, &rels);
@@ -1838,7 +1838,7 @@ RecordTransactionAbort(bool isSubXact)
if (replorigin)
/* Move LSNs forward for this replication origin */
- replorigin_session_advance(replorigin_session_origin_lsn,
+ replorigin_session_advance(replorigin_xact_state.origin_lsn,
XactLastRecEnd);
/*
@@ -5928,12 +5928,12 @@ XactLogCommitRecord(TimestampTz commit_time,
}
/* dump transaction origin information */
- if (replorigin_session_origin != InvalidRepOriginId)
+ if (replorigin_xact_state.origin != InvalidRepOriginId)
{
xl_xinfo.xinfo |= XACT_XINFO_HAS_ORIGIN;
- xl_origin.origin_lsn = replorigin_session_origin_lsn;
- xl_origin.origin_timestamp = replorigin_session_origin_timestamp;
+ xl_origin.origin_lsn = replorigin_xact_state.origin_lsn;
+ xl_origin.origin_timestamp = replorigin_xact_state.origin_timestamp;
}
if (xl_xinfo.xinfo != 0)
@@ -6081,12 +6081,12 @@ XactLogAbortRecord(TimestampTz abort_time,
* Dump transaction origin information. We need this during recovery to
* update the replication origin progress.
*/
- if (replorigin_session_origin != InvalidRepOriginId)
+ if (replorigin_xact_state.origin != InvalidRepOriginId)
{
xl_xinfo.xinfo |= XACT_XINFO_HAS_ORIGIN;
- xl_origin.origin_lsn = replorigin_session_origin_lsn;
- xl_origin.origin_timestamp = replorigin_session_origin_timestamp;
+ xl_origin.origin_lsn = replorigin_xact_state.origin_lsn;
+ xl_origin.origin_timestamp = replorigin_xact_state.origin_timestamp;
}
if (xl_xinfo.xinfo != 0)
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 92c48e768c3..7f9485e08e7 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -861,11 +861,11 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
/* followed by the record's origin, if any */
if ((curinsert_flags & XLOG_INCLUDE_ORIGIN) &&
- replorigin_session_origin != InvalidRepOriginId)
+ replorigin_xact_state.origin != InvalidRepOriginId)
{
*(scratch++) = (char) XLR_BLOCK_ID_ORIGIN;
- memcpy(scratch, &replorigin_session_origin, sizeof(replorigin_session_origin));
- scratch += sizeof(replorigin_session_origin);
+ memcpy(scratch, &replorigin_xact_state.origin, sizeof(replorigin_xact_state.origin));
+ scratch += sizeof(replorigin_xact_state.origin);
}
/* followed by toplevel XID, if not already included in previous record */
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 5ebd2353fed..fbc113b099e 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -962,7 +962,7 @@ ParallelApplyWorkerMain(Datum main_arg)
* origin which was already acquired by its leader process.
*/
replorigin_session_setup(originid, MyLogicalRepWorker->leader_pid);
- replorigin_session_origin = originid;
+ replorigin_xact_state.origin = originid;
CommitTransactionCommand();
/*
@@ -1430,8 +1430,8 @@ pa_stream_abort(LogicalRepStreamAbortData *abort_data)
* Update origin state so we can restart streaming from correct position
* in case of crash.
*/
- replorigin_session_origin_lsn = abort_data->abort_lsn;
- replorigin_session_origin_timestamp = abort_data->abort_time;
+ replorigin_xact_state.origin_lsn = abort_data->abort_lsn;
+ replorigin_xact_state.origin_timestamp = abort_data->abort_time;
/*
* If the two XIDs are the same, it's in fact abort of toplevel xact, so
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index c27e790c3a8..97c8f8ddb87 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -160,9 +160,11 @@ typedef struct ReplicationStateCtl
} ReplicationStateCtl;
/* external variables */
-RepOriginId replorigin_session_origin = InvalidRepOriginId; /* assumed identity */
-XLogRecPtr replorigin_session_origin_lsn = InvalidXLogRecPtr;
-TimestampTz replorigin_session_origin_timestamp = 0;
+RepOriginXactState replorigin_xact_state = {
+ .origin = InvalidRepOriginId,
+ .origin_lsn = InvalidXLogRecPtr,
+ .origin_timestamp = 0
+};
/*
* Base address into a shared memory array of replication states of size
@@ -896,7 +898,7 @@ replorigin_redo(XLogReaderState *record)
* Tell the replication origin progress machinery that a commit from 'node'
* that originated at the LSN remote_commit on the remote node was replayed
* successfully and that we don't need to do so again. In combination with
- * setting up replorigin_session_origin_lsn and replorigin_session_origin
+ * setting up replorigin_xact_state.origin_lsn and replorigin_xact_state.origin
* that ensures we won't lose knowledge about that after a crash if the
* transaction had a persistent effect (think of asynchronous commits).
*
@@ -1288,17 +1290,17 @@ replorigin_session_get_progress(bool flush)
}
/*
- * Clear session replication origin state.
+ * Clear the transaction replication origin state.
*
* Only when clear_origin is true, the origin ID is also cleared.
*/
void
-replorigin_session_clear(bool clear_origin)
+replorigin_xact_clear(bool clear_origin)
{
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ replorigin_xact_state.origin_lsn = InvalidXLogRecPtr;
+ replorigin_xact_state.origin_timestamp = 0;
if (clear_origin)
- replorigin_session_origin = InvalidRepOriginId;
+ replorigin_xact_state.origin = InvalidRepOriginId;
}
@@ -1408,7 +1410,7 @@ pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
pid = PG_GETARG_INT32(1);
replorigin_session_setup(origin, pid);
- replorigin_session_origin = origin;
+ replorigin_xact_state.origin = origin;
pfree(name);
@@ -1425,7 +1427,7 @@ pg_replication_origin_session_reset(PG_FUNCTION_ARGS)
replorigin_session_reset();
- replorigin_session_clear(true);
+ replorigin_xact_clear(true);
PG_RETURN_VOID();
}
@@ -1438,7 +1440,7 @@ pg_replication_origin_session_is_setup(PG_FUNCTION_ARGS)
{
replorigin_check_prerequisites(false, false);
- PG_RETURN_BOOL(replorigin_session_origin != InvalidRepOriginId);
+ PG_RETURN_BOOL(replorigin_xact_state.origin != InvalidRepOriginId);
}
@@ -1482,8 +1484,8 @@ pg_replication_origin_xact_setup(PG_FUNCTION_ARGS)
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("no replication origin is configured")));
- replorigin_session_origin_lsn = location;
- replorigin_session_origin_timestamp = PG_GETARG_TIMESTAMPTZ(1);
+ replorigin_xact_state.origin_lsn = location;
+ replorigin_xact_state.origin_timestamp = PG_GETARG_TIMESTAMPTZ(1);
PG_RETURN_VOID();
}
@@ -1493,7 +1495,7 @@ pg_replication_origin_xact_reset(PG_FUNCTION_ARGS)
{
replorigin_check_prerequisites(true, false);
- replorigin_session_clear(false);
+ replorigin_xact_clear(false);
PG_RETURN_VOID();
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 72e45c2de19..864476b4a72 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -323,7 +323,7 @@ ProcessSyncingTablesForSync(XLogRecPtr current_lsn)
* This is needed to allow the origin to be dropped.
*/
replorigin_session_reset();
- replorigin_session_clear(true);
+ replorigin_xact_clear(true);
/*
* Drop the tablesync's origin tracking if exists.
@@ -1318,7 +1318,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
*/
originid = replorigin_by_name(originname, false);
replorigin_session_setup(originid, 0);
- replorigin_session_origin = originid;
+ replorigin_xact_state.origin = originid;
*origin_startpos = replorigin_session_get_progress(false);
CommitTransactionCommand();
@@ -1405,7 +1405,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
replorigin_session_setup(originid, 0);
- replorigin_session_origin = originid;
+ replorigin_xact_state.origin = originid;
/*
* If the user did not opt to run as the owner of the subscription
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 6ad99cc5afd..1f21b56ace6 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1318,8 +1318,8 @@ apply_handle_prepare_internal(LogicalRepPreparedTxnData *prepare_data)
* Update origin state so we can restart streaming from correct position
* in case of crash.
*/
- replorigin_session_origin_lsn = prepare_data->end_lsn;
- replorigin_session_origin_timestamp = prepare_data->prepare_time;
+ replorigin_xact_state.origin_lsn = prepare_data->end_lsn;
+ replorigin_xact_state.origin_timestamp = prepare_data->prepare_time;
PrepareTransactionBlock(gid);
}
@@ -1421,8 +1421,8 @@ apply_handle_commit_prepared(StringInfo s)
* Update origin state so we can restart streaming from correct position
* in case of crash.
*/
- replorigin_session_origin_lsn = prepare_data.end_lsn;
- replorigin_session_origin_timestamp = prepare_data.commit_time;
+ replorigin_xact_state.origin_lsn = prepare_data.end_lsn;
+ replorigin_xact_state.origin_timestamp = prepare_data.commit_time;
FinishPreparedTransaction(gid, true);
end_replication_step();
@@ -1479,8 +1479,8 @@ apply_handle_rollback_prepared(StringInfo s)
* Update origin state so we can restart streaming from correct
* position in case of crash.
*/
- replorigin_session_origin_lsn = rollback_data.rollback_end_lsn;
- replorigin_session_origin_timestamp = rollback_data.rollback_time;
+ replorigin_xact_state.origin_lsn = rollback_data.rollback_end_lsn;
+ replorigin_xact_state.origin_timestamp = rollback_data.rollback_time;
/* There is no transaction when ABORT/ROLLBACK PREPARED is called */
begin_replication_step();
@@ -2526,8 +2526,8 @@ apply_handle_commit_internal(LogicalRepCommitData *commit_data)
* Update origin state so we can restart streaming from correct
* position in case of crash.
*/
- replorigin_session_origin_lsn = commit_data->end_lsn;
- replorigin_session_origin_timestamp = commit_data->committime;
+ replorigin_xact_state.origin_lsn = commit_data->end_lsn;
+ replorigin_xact_state.origin_timestamp = commit_data->committime;
CommitTransactionCommand();
@@ -2940,7 +2940,7 @@ apply_handle_update_internal(ApplyExecutionData *edata,
*/
if (GetTupleTransactionInfo(localslot, &conflicttuple.xmin,
&conflicttuple.origin, &conflicttuple.ts) &&
- conflicttuple.origin != replorigin_session_origin)
+ conflicttuple.origin != replorigin_xact_state.origin)
{
TupleTableSlot *newslot;
@@ -2982,7 +2982,7 @@ apply_handle_update_internal(ApplyExecutionData *edata,
&conflicttuple.xmin,
&conflicttuple.origin,
&conflicttuple.ts) &&
- conflicttuple.origin != replorigin_session_origin)
+ conflicttuple.origin != replorigin_xact_state.origin)
type = CT_UPDATE_DELETED;
else
type = CT_UPDATE_MISSING;
@@ -3135,7 +3135,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
*/
if (GetTupleTransactionInfo(localslot, &conflicttuple.xmin,
&conflicttuple.origin, &conflicttuple.ts) &&
- conflicttuple.origin != replorigin_session_origin)
+ conflicttuple.origin != replorigin_xact_state.origin)
{
conflicttuple.slot = localslot;
ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_ORIGIN_DIFFERS,
@@ -3477,7 +3477,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
&conflicttuple.xmin,
&conflicttuple.origin,
&conflicttuple.ts) &&
- conflicttuple.origin != replorigin_session_origin)
+ conflicttuple.origin != replorigin_xact_state.origin)
type = CT_UPDATE_DELETED;
else
type = CT_UPDATE_MISSING;
@@ -3503,7 +3503,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
if (GetTupleTransactionInfo(localslot, &conflicttuple.xmin,
&conflicttuple.origin,
&conflicttuple.ts) &&
- conflicttuple.origin != replorigin_session_origin)
+ conflicttuple.origin != replorigin_xact_state.origin)
{
TupleTableSlot *newslot;
@@ -5594,7 +5594,7 @@ start_apply(XLogRecPtr origin_startpos)
* transaction loss as that transaction won't be sent again by the
* server.
*/
- replorigin_session_clear(true);
+ replorigin_xact_clear(true);
if (MySubscription->disableonerr)
DisableSubscriptionAndExit();
@@ -5652,7 +5652,7 @@ run_apply_worker(void)
if (!OidIsValid(originid))
originid = replorigin_create(originname);
replorigin_session_setup(originid, 0);
- replorigin_session_origin = originid;
+ replorigin_xact_state.origin = originid;
origin_startpos = replorigin_session_get_progress(false);
CommitTransactionCommand();
@@ -5874,7 +5874,7 @@ InitializeLogRepWorker(void)
static void
on_exit_clear_state(int code, Datum arg)
{
- replorigin_session_clear(true);
+ replorigin_xact_clear(true);
}
/*
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 309b800bd5f..6d58fb20463 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -40,9 +40,14 @@ typedef struct xl_replorigin_drop
*/
#define MAX_RONAME_LEN 512
-extern PGDLLIMPORT RepOriginId replorigin_session_origin;
-extern PGDLLIMPORT XLogRecPtr replorigin_session_origin_lsn;
-extern PGDLLIMPORT TimestampTz replorigin_session_origin_timestamp;
+typedef struct RepOriginXactState
+{
+ RepOriginId origin;
+ XLogRecPtr origin_lsn;
+ TimestampTz origin_timestamp;
+} RepOriginXactState;
+
+extern PGDLLIMPORT RepOriginXactState replorigin_xact_state;
/* GUCs */
extern PGDLLIMPORT int max_active_replication_origins;
@@ -65,9 +70,10 @@ extern void replorigin_session_advance(XLogRecPtr remote_commit,
XLogRecPtr local_commit);
extern void replorigin_session_setup(RepOriginId node, int acquired_by);
extern void replorigin_session_reset(void);
-extern void replorigin_session_clear(bool clear_origin);
extern XLogRecPtr replorigin_session_get_progress(bool flush);
+extern void replorigin_xact_clear(bool clear_origin);
+
/* Checkpoint/Startup integration */
extern void CheckPointReplicationOrigin(void);
extern void StartupReplicationOrigin(void);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 09e7f1d420e..94a1dbed466 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2569,6 +2569,7 @@ ReorderBufferTupleCidKey
ReorderBufferUpdateProgressTxnCB
ReorderTuple
RepOriginId
+RepOriginXactState
ReparameterizeForeignPathByChild_function
ReplaceVarsFromTargetList_context
ReplaceVarsNoMatchOption
--
2.39.5 (Apple Git-154)
^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2026-01-08 01:15 UTC | newest]
Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2021-03-12 20:32 [PATCH 05/10] (re)add wal_compression_method: lz4 Justin Pryzby <[email protected]>
2026-01-07 08:49 Re: Refactor replication origin state reset helpers Chao Li <[email protected]>
2026-01-08 01:15 ` Re: Refactor replication origin state reset helpers Chao Li <[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