public inbox for [email protected]
help / color / mirror / Atom feedFrom: Chao Li <[email protected]>
To: Masahiko Sawada <[email protected]>
Cc: Ashutosh Bapat <[email protected]>
Cc: Álvaro Herrera <[email protected]>
Cc: Postgres hackers <[email protected]>
Subject: Re: Refactor replication origin state reset helpers
Date: Mon, 12 Jan 2026 09:41:45 +0800
Message-ID: <CAEoWx2nCwUa2mKCY7ZyYqTxZ8L1=arVosaCjcQPNkzuOztS5mw@mail.gmail.com> (raw)
In-Reply-To: <CAD21AoALhGyb=5b+j4ZNQpdMER5XDR92j9Nbouz3M0Hc=X2mpA@mail.gmail.com>
References: <CAExHW5usKxNkgGdM+6cw-RiDN55MLSmoUC1LBi6tQ3eYMuZm0Q@mail.gmail.com>
<[email protected]>
<CAExHW5vYPkRW2CeeQX+z230LNb8kqKgxbFxmbSMNiHVQdvEP8Q@mail.gmail.com>
<CAEoWx2mT+gu8H-KhaeNodk4T7csX-_wwqRew=-t3zo6b6jXfaQ@mail.gmail.com>
<CAEoWx2=7xcxQPJgnPfWsKj2SzCNiBaxOXa1u9fSKME1Yd=r=fQ@mail.gmail.com>
<CAD21AoD1HCtzN1WKXOPpifVTprzC=fkTPqPDuO9Z6oAZLXSYLA@mail.gmail.com>
<CAEoWx2nsZeL1A_tbzoBCRXC1R2uBpYuQTN3dPy=dDPhpvvsRNA@mail.gmail.com>
<[email protected]>
<CAEoWx2moawAzQLaHu=C9PEkro7tmXXq792ODM0=bM9rtoE_E2g@mail.gmail.com>
<CAD21AoDo_JGZPooDmbXheJrCBbEHh_WTgi31B_qNXr3-=JV93A@mail.gmail.com>
<CAExHW5v9WYiTScYjXbqW-ttatRwUwhiaKw-n5YSBykc3hGyQng@mail.gmail.com>
<CAEoWx2mL=s4ROe6fjOr2=YbUrrZOVQVVnqb8tK2C7sCgvnGUHg@mail.gmail.com>
<CAD21AoALhGyb=5b+j4ZNQpdMER5XDR92j9Nbouz3M0Hc=X2mpA@mail.gmail.com>
On Jan 10, 2026, at 06:08, Masahiko Sawada <[email protected]> wrote:
On Thu, Jan 8, 2026 at 10:50 PM Chao Li <[email protected]> wrote:
On Jan 8, 2026, at 17:44, Ashutosh Bapat <[email protected]>
wrote:
Hi Masahiko,
Thanks for updating the patches. Here are some more comments.
On Thu, Jan 8, 2026 at 7:17 AM Masahiko Sawada <[email protected]>
wrote:
On Wed, Jan 7, 2026 at 5:15 PM Chao Li <[email protected]> wrote:
I've made some cosmetic changes to both patches (comments and the
commit messages). Please review them and let me know what you think.
0001
-------
+/*
+ * Clear session replication origin state.
+ *
+ * replorigin_session_origin is also cleared if clear_origin is set.
+ */
+void
+replorigin_session_clear(bool clear_origin)
+{
+ replorigin_session_origin_lsn = InvalidXLogRecPtr;
+ replorigin_session_origin_timestamp = 0;
+ if (clear_origin)
+ replorigin_session_origin = InvalidRepOriginId;
+}
All the other replorigin_session_* functions deal with
session_replication_state, but this function does not deal with it. I
see that in the next patch this function has been renamed as
replorigin_xact_clear() which seems more appropriate. Do you intend to
squash these two patches when committing?
Renamed the help function to the same name as in 0002.
@@ -1482,8 +1493,8 @@ pg_replication_origin_xact_reset(PG_FUNCTION_ARGS)
{
replorigin_check_prerequisites(true, false);
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ /* Clear only origin_lsn and origin_timestamp */
+ replorigin_session_clear(false);
The comment can explain why we are not clearing
replorigin_session_origin here. Something like "This function is
cancel the effects of pg_replication_origin_xact_setup(), which only
sets origin_lsn and origin_timestamp, so we only clear those two
fields here.".
Updated the comment in 0001.
Next comment does not apply to this patch, but the inconsistency I am
speaking about becomes apparent now. This function resets the state
setup by pg_replication_origin_xact_setup(), which checks for
session_replication_state being non-NULL. So I expected
pg_replication_origin_xact_reset() also to check for the same
condition or at least Assert it. Why is it not doing so?
Added the same check in 0002.
0002
-------
- replorigin = (replorigin_session_origin != InvalidRepOriginId &&
- replorigin_session_origin != DoNotReplicateId);
+ replorigin = (replorigin_xact_state.origin != InvalidRepOriginId &&
+ replorigin_xact_state.origin != DoNotReplicateId);
There's another small deduplication opportunity here. Define function
replorigin_xact_origin_isvalid() to check these two conditions and use
that function here and in other places like
RecordTransactionCommitPrepared(). I would go as far as making the
function static inline, and use it instead of replorigin variable,
whose name is certainly misleading - it doesn't talk about transaction
at all. With static inline there, optimizer may be able to eliminate
the multiple function call overhead.
Good point. Added replorigin_xact_origin_isvalid() in 0002 as you suggested.
/*
* 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.
suggestion "Record commit timestamp. Use one in replorigin_xact_state
if set, otherwise use plain commit timestamp.". This reads better and
is closer to the code.". If you agree, please change other similar
comments too.
Agreed. Updated the comment in both places.
@@ -5928,12 +5928,12 @@ XactLogCommitRecord(TimestampTz commit_time,
}
/* dump transaction origin information */
- if (replorigin_session_origin != InvalidRepOriginId)
+ if (replorigin_xact_state.origin != InvalidRepOriginId)
{
* 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)
{
/* followed by the record's origin, if any */
if ((curinsert_flags & XLOG_INCLUDE_ORIGIN) &&
- replorigin_session_origin != InvalidRepOriginId)
+ replorigin_xact_state.origin != InvalidRepOriginId)
Not a change in this patch but the refactoring makes it more visible.
What about the case of replorigin_session_origin == DoNotReplicateId?
Should there be a comment explaining why it's excluded here?
Added a comment to the 3 places.
replorigin_session_setup(originid, MyLogicalRepWorker->leader_pid);
- replorigin_session_origin = originid;
+ replorigin_xact_state.origin = originid;
replorigin_session_setup() is always followed by
replorigin_xact_state.origin assignment. abort/commit handlers set the
other two members. Do we want to create replorigin_xact_set_origin()
and replorigin_xact_set_lsn_timestamp() functions to encapsulate these
assignments? Those two will serve as counterparts to
replorigin_xact_clear(). Or would that be an overkill?
Actually, when I added the new structure, I had the idea of adding
replorigin_xact_set_origin() and replorigin_xact_set_lsn_timestamp() as I
saw a lot of duplicate code for the assignment. But I hesitated to do that
because I worried people might consider that’s too much.
Given you raised the same idea, let me add the helpers as static inline. We
now have 3 static inline helpers, I guess we have no reason to leave
replorigin_xact_clear() alone as external. So I made it static inline as
well.
* 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
I think just replorigin_xact_state should suffice for the sake of brevity.
Updated the comment to eliminate the duplication.
static void
on_exit_clear_state(int code, Datum arg)
{
- replorigin_session_clear(true);
+ replorigin_xact_clear(true);
The prologue still states clear origin, but it should mention
transaction state instead.
}
Good catch. Updated the header comment.
All comments are addressed in v8.
Thank you for updating the patch!
The 0001 patch looks good to me.
I have some comments on the 0002 patch:
Thanks for the review.
- replorigin = (replorigin_session_origin != InvalidRepOriginId &&
- replorigin_session_origin != DoNotReplicateId);
+ replorigin = replorigin_xact_origin_isvalid();
I'm not sure it's a good refactoring TBH. While it can remove some
lines, we still need to check if replorigin_xact_state.origin !=
InvalidRepOriginId at several places. For example,
Datum
pg_replication_origin_session_is_setup(PG_FUNCTION_ARGS)
{
replorigin_check_prerequisites(false, false);
PG_RETURN_BOOL(replorigin_xact_state.origin != InvalidRepOriginId);
}
and
/*
* Dump transaction origin information. We need this during recovery to
* update the replication origin progress.
*
* Note that DoNotReplicateId is intentionally excluded here.
*/
if (replorigin_xact_state.origin != InvalidRepOriginId)
The readers might be confused why we cannot use
replorigin_xact_origin_isvalid() here in spite of its function name,
replorigin_xact_origin_isvalid, sounds quite suitable. The comment
mentions the fact that DoNotReplicated is excluded but doesn't mention
why.
Also, even if we want this kind of change, it should be implemented in
a separate patch because it's not related to the main idea of
consolidating the replication origin global variables.
Fair enough. Let me revert the change related to
replorigin_xact_origin_isvalid.
---
- * 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.
+ * Record commit timestamp. Use one in replorigin_xact_state if set,
+ * otherwise use plain commit timestamp.
and
- * 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.
+ * Record commit timestamp. Use one in replorigin_xact_state if set,
+ * otherwise use plain commit timestamp.
I think the previous comments are clearer, so I'm not sure we need to
change these comments by this patch. Can we just change
replorigin_session_origin_timestamp to
replorigin_xact_state.origin_timestamp?
Reverted to the old comments and updated the variable name in the comments.
---
+ if (session_replication_state == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("no replication origin is configured")));
I'm not sure what the original author intended but given that
pg_replication_origin_xact_reset() just clears the local state I'm not
sure that we should strictly require the session_replication_state to
be set up nor it improves the behavior. If we raise an error in that
case, it might be more user-friendly but we can live even without it.
Fair point. I reverted that. If we really want to add the check, let’s do
that in a separate thread.
---
In origin.h:
+/*
+ * Clear the per-transaction replication origin state.
+ *
+ * replorigin_session_origin is also cleared if clear_origin is set.
+ */
+static inline void
+replorigin_xact_clear(bool clear_origin)
+{
+ replorigin_xact_state.origin_lsn = InvalidXLogRecPtr;
+ replorigin_xact_state.origin_timestamp = 0;
+ if (clear_origin)
+ replorigin_xact_state.origin = InvalidRepOriginId;
+}
Why does this function need to move to origin.h from origin.c?
That’s because, per Ashutosh’s suggestion, I added two static inline
helpers replorigin_xact_set_origin()
and replorigin_xact_set_lsn_timestamp(), and I
thought replorigin_xact_clear() should stay close with them.
But looks like they don’t have to be inline as they are not on hot paths.
So I moved them all to origin.c and only extern them.
PFA v9.
Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/
Attachments:
[application/octet-stream] v9-0001-Refactor-replication-origin-state-reset-helpers.patch (5.2K, ../CAEoWx2nCwUa2mKCY7ZyYqTxZ8L1=arVosaCjcQPNkzuOztS5mw@mail.gmail.com/3-v9-0001-Refactor-replication-origin-state-reset-helpers.patch)
download | inline diff:
From c94c6598f8b6a5cad37a963319d316f4de50e728 Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Wed, 24 Dec 2025 09:17:27 +0800
Subject: [PATCH v9 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 replorigin_session_* variables
into a dedicated helper function, replorigin_xact_clear().
This removes duplicated assignments of these variables across multiple
call sites, and makes the intended scope of each reset explicit.
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 | 21 ++++++++++++++++-----
src/backend/replication/logical/tablesync.c | 4 +---
src/backend/replication/logical/worker.c | 14 ++++++--------
src/include/replication/origin.h | 2 ++
4 files changed, 25 insertions(+), 16 deletions(-)
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index 04bc704a332..09616641903 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 the per-transaction replication origin state.
+ *
+ * replorigin_session_origin is also cleared if clear_origin is set.
+ */
+void
+replorigin_xact_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_xact_clear(true);
PG_RETURN_VOID();
}
@@ -1482,8 +1493,8 @@ pg_replication_origin_xact_reset(PG_FUNCTION_ARGS)
{
replorigin_check_prerequisites(true, false);
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ /* Do not clear the session origin */
+ replorigin_xact_clear(false);
PG_RETURN_VOID();
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 67e57520386..73a0d2838cf 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_xact_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..c1aceabbdf5 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_xact_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_xact_clear(true);
}
/*
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 1da77363955..1eaabacde03 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -67,6 +67,8 @@ extern void replorigin_session_setup(RepOriginId node, int acquired_by);
extern void replorigin_session_reset(void);
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);
--
2.39.5 (Apple Git-154)
[application/octet-stream] v9-0002-Consolidate-replication-origin-session-globals-in.patch (22.2K, ../CAEoWx2nCwUa2mKCY7ZyYqTxZ8L1=arVosaCjcQPNkzuOztS5mw@mail.gmail.com/4-v9-0002-Consolidate-replication-origin-session-globals-in.patch)
download | inline diff:
From 3e83a7e99b7e4d5abed63dac79112b797b1d2a84 Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Tue, 30 Dec 2025 12:55:31 +0800
Subject: [PATCH v9 2/2] Consolidate replication origin session globals into a
single struct.
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
This commit moves the separate global variables for replication origin
state into a single RepOriginXactState struct. This groups logically
related variables, which improves code readability and simplifies
state management (e.g., resetting the state) by handling them as a
unit.
Author: Chao Li <[email protected]>
Suggested-by: Álvaro Herrera <[email protected]
Reviewed-by: Masahiko Sawada <[email protected]>
Reviewed-by: Ashutosh Bapat <[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 | 44 ++++++++++--------
src/backend/access/transam/xloginsert.c | 12 +++--
.../replication/logical/applyparallelworker.c | 6 +--
src/backend/replication/logical/origin.c | 46 +++++++++++++------
src/backend/replication/logical/tablesync.c | 4 +-
src/backend/replication/logical/worker.c | 28 ++++++-----
src/include/replication/origin.h | 15 ++++--
src/tools/pgindent/typedefs.list | 1 +
9 files changed, 113 insertions(+), 75 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..c5c3f21c9a7 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);
/*
@@ -5927,13 +5927,17 @@ XactLogCommitRecord(TimestampTz commit_time,
xl_xinfo.xinfo |= XACT_XINFO_HAS_GID;
}
- /* dump transaction origin information */
- if (replorigin_session_origin != InvalidRepOriginId)
+ /*
+ * Dump transaction origin information
+ *
+ * Note that DoNotReplicateId is intentionally excluded here.
+ */
+ 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)
@@ -6080,13 +6084,15 @@ XactLogAbortRecord(TimestampTz abort_time,
/*
* Dump transaction origin information. We need this during recovery to
* update the replication origin progress.
+ *
+ * Note that DoNotReplicateId is intentionally excluded here.
*/
- 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..64534e45216 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -859,13 +859,17 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
scratch += sizeof(BlockNumber);
}
- /* followed by the record's origin, if any */
+ /*
+ * followed by the record's origin, if any
+ *
+ * DoNotReplicateId is intentionally excluded here
+ */
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..232f0447b30 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_set_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_set_lsn_timestamp(abort_data->abort_lsn,
+ 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 09616641903..1dcc20b00db 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -159,10 +159,12 @@ typedef struct ReplicationStateCtl
ReplicationState states[FLEXIBLE_ARRAY_MEMBER];
} ReplicationStateCtl;
-/* external variables */
-RepOriginId replorigin_session_origin = InvalidRepOriginId; /* assumed identity */
-XLogRecPtr replorigin_session_origin_lsn = InvalidXLogRecPtr;
-TimestampTz replorigin_session_origin_timestamp = 0;
+/* Global variable for per-transaction replication origin state */
+RepOriginXactState replorigin_xact_state = {
+ .origin = InvalidRepOriginId, /* assumed identity */
+ .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, .origin_timestamp}
* that ensures we won't lose knowledge about that after a crash if the
* transaction had a persistent effect (think of asynchronous commits).
*
@@ -1287,6 +1289,26 @@ replorigin_session_get_progress(bool flush)
return remote_lsn;
}
+/*
+ * Set the per-transaction replication origin state.
+ */
+void
+replorigin_xact_set_origin(RepOriginId origin)
+{
+ replorigin_xact_state.origin = origin;
+}
+
+/*
+ * Set the per-transaction replication origin LSN and timestamp.
+ */
+void
+replorigin_xact_set_lsn_timestamp(XLogRecPtr origin_lsn,
+ TimestampTz origin_timestamp)
+{
+ replorigin_xact_state.origin_lsn = origin_lsn;
+ replorigin_xact_state.origin_timestamp = origin_timestamp;
+}
+
/*
* Clear the per-transaction replication origin state.
*
@@ -1295,13 +1317,12 @@ replorigin_session_get_progress(bool flush)
void
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;
}
-
/* ---------------------------------------------------------------------------
* SQL functions for working with replication origin.
*
@@ -1408,7 +1429,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_set_origin(origin);
pfree(name);
@@ -1438,7 +1459,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 +1503,7 @@ 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_set_lsn_timestamp(location, PG_GETARG_TIMESTAMPTZ(1));
PG_RETURN_VOID();
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 73a0d2838cf..e34377a559c 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -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_set_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_set_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 c1aceabbdf5..d5e23854250 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1318,8 +1318,7 @@ 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_set_lsn_timestamp(prepare_data->end_lsn, prepare_data->prepare_time);
PrepareTransactionBlock(gid);
}
@@ -1421,8 +1420,7 @@ 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_set_lsn_timestamp(prepare_data.end_lsn, prepare_data.commit_time);
FinishPreparedTransaction(gid, true);
end_replication_step();
@@ -1479,8 +1477,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_set_lsn_timestamp(rollback_data.rollback_end_lsn,
+ rollback_data.rollback_time);
/* There is no transaction when ABORT/ROLLBACK PREPARED is called */
begin_replication_step();
@@ -2526,8 +2524,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_set_lsn_timestamp(commit_data->end_lsn,
+ commit_data->committime);
CommitTransactionCommand();
@@ -2940,7 +2938,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 +2980,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 +3133,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 +3475,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 +3501,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;
@@ -5652,7 +5650,7 @@ run_apply_worker(void)
if (!OidIsValid(originid))
originid = replorigin_create(originname);
replorigin_session_setup(originid, 0);
- replorigin_session_origin = originid;
+ replorigin_xact_set_origin(originid);
origin_startpos = replorigin_session_get_progress(false);
CommitTransactionCommand();
@@ -5869,7 +5867,7 @@ InitializeLogRepWorker(void)
}
/*
- * Callback on exit to reset the origin state.
+ * Callback on exit to clear transaction-level replication origin state.
*/
static void
on_exit_clear_state(int code, Datum arg)
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 1eaabacde03..c953785f6a4 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;
@@ -67,6 +72,10 @@ extern void replorigin_session_setup(RepOriginId node, int acquired_by);
extern void replorigin_session_reset(void);
extern XLogRecPtr replorigin_session_get_progress(bool flush);
+/* Per-transaction replication origin state manipulation */
+extern void replorigin_xact_set_origin(RepOriginId origin);
+extern void replorigin_xact_set_lsn_timestamp(XLogRecPtr origin_lsn,
+ TimestampTz origin_timestamp);
extern void replorigin_xact_clear(bool clear_origin);
/* Checkpoint/Startup integration */
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)
view thread (10+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Refactor replication origin state reset helpers
In-Reply-To: <CAEoWx2nCwUa2mKCY7ZyYqTxZ8L1=arVosaCjcQPNkzuOztS5mw@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox