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: Thu, 15 Jan 2026 08:00:31 +0800
Message-ID: <CAEoWx2mk4AS6H39PRs8JNS4ZaCMfZbbKqddConBmTcOGnLGtdA@mail.gmail.com> (raw)
In-Reply-To: <CAD21AoAsXVhLmbGdOpAHHsjXh+XxJ0HOz7x6wxX_n4aFCuas3A@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>
<CAEoWx2nCwUa2mKCY7ZyYqTxZ8L1=arVosaCjcQPNkzuOztS5mw@mail.gmail.com>
<CAD21AoAsXVhLmbGdOpAHHsjXh+XxJ0HOz7x6wxX_n4aFCuas3A@mail.gmail.com>
On Thu, Jan 15, 2026 at 2:56 AM Masahiko Sawada <[email protected]>
wrote:
> On Sun, Jan 11, 2026 at 5:41 PM Chao Li <[email protected]> wrote:
> >
> > ---
> > 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.
>
> Thank you for updating the patch.
>
> I'm not even sure that we need to have setter functions like
> replorigin_xact_set_{origin,lsn_timestamp} given that
> replorigin_xact_state is exposed. While the reset helper function
> helps us as it removes duplicated codes and some potential accidents
> like wrongly setting -1 as an invalid timestamp etc. these setter
> functions don't so much. So I think we can have the patch just
> consolidating the separated variables. What do you think?
>
>
No problem. I reverted the two helpers in v10.
Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/
Attachments:
[application/octet-stream] v10-0002-Consolidate-replication-origin-session-globals-i.patch (21.5K, ../CAEoWx2mk4AS6H39PRs8JNS4ZaCMfZbbKqddConBmTcOGnLGtdA@mail.gmail.com/3-v10-0002-Consolidate-replication-origin-session-globals-i.patch)
download | inline diff:
From a04172df30f423cf01183630456deac0f22b4c77 Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Tue, 30 Dec 2025 12:55:31 +0800
Subject: [PATCH v10 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 | 26 ++++++-----
src/backend/replication/logical/tablesync.c | 4 +-
src/backend/replication/logical/worker.c | 30 ++++++-------
src/include/replication/origin.h | 12 +++--
src/tools/pgindent/typedefs.list | 1 +
9 files changed, 93 insertions(+), 74 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..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 bb65e4d36d3..884b5d1f698 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -162,10 +162,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
@@ -902,7 +904,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).
*
@@ -1349,10 +1351,10 @@ 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;
}
@@ -1462,7 +1464,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);
@@ -1492,7 +1494,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);
}
@@ -1536,8 +1538,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();
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 73a0d2838cf..864476b4a72 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_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 c1aceabbdf5..833d59f2284 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;
@@ -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();
@@ -5869,7 +5869,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..081a32cc12c 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,7 @@ 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_clear(bool clear_origin);
/* Checkpoint/Startup integration */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 14dec2d49c1..28ff7542888 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2570,6 +2570,7 @@ ReorderBufferTupleCidKey
ReorderBufferUpdateProgressTxnCB
ReorderTuple
RepOriginId
+RepOriginXactState
ReparameterizeForeignPathByChild_function
ReplaceVarsFromTargetList_context
ReplaceVarsNoMatchOption
--
2.39.5 (Apple Git-154)
[application/octet-stream] v10-0001-Refactor-replication-origin-state-reset-helpers.patch (5.2K, ../CAEoWx2mk4AS6H39PRs8JNS4ZaCMfZbbKqddConBmTcOGnLGtdA@mail.gmail.com/4-v10-0001-Refactor-replication-origin-state-reset-helpers.patch)
download | inline diff:
From 94f4555ed96af3bf5b046568150777bd13e70fd4 Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Wed, 24 Dec 2025 09:17:27 +0800
Subject: [PATCH v10 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 13808a4674b..bb65e4d36d3 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -1341,6 +1341,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;
+}
/* ---------------------------------------------------------------------------
@@ -1466,9 +1479,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();
}
@@ -1536,8 +1547,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)
view thread (10+ messages)
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: <CAEoWx2mk4AS6H39PRs8JNS4ZaCMfZbbKqddConBmTcOGnLGtdA@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